Claude Code: Implement session archival strategy for Cain's agent system:

.openclaw/agents/logs/session-archive.jsonl

@@ -0,0 +1,2 @@
+{"timestamp": "2026-03-14T04:45:10.399081+00:00", "threshold_days": 7, "dry_run": true, "action": "archive_sessions", "total_files": 0, "archived_count": 0, "skipped_count": 0, "error_count": 0, "status": "no_sessions_found"}
+{"timestamp": "2026-03-14T04:45:28.188361+00:00", "threshold_days": 7, "dry_run": true, "action": "archive_sessions", "total_files": 0, "archived_count": 0, "skipped_count": 0, "error_count": 0, "status": "no_sessions_found"}

.openclaw/agents/main/sessions/archive_manager.py

@@ -0,0 +1,373 @@
+#!/usr/bin/env python3
+"""
+Session Archive Manager for Cain's Agent System
+
+Manages archival and restoration of agent sessions:
+- Automatically archives sessions older than a configurable threshold
+- Maintains an index of archived sessions
+- Provides restore functionality for recovering archived sessions
+"""
+import json
+import os
+import shutil
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Dict, List, Optional, Any
+
+
+# Configuration
+SESSIONS_DIR = Path(__file__).parent
+ARCHIVED_DIR = SESSIONS_DIR / "archived"
+SESSIONS_INDEX = SESSIONS_DIR / "sessions.json"
+ARCHIVED_INDEX = ARCHIVED_DIR / "archived_sessions.json"
+LOG_FILE = Path(__file__).parent.parent.parent / "logs" / "session-archive.jsonl"
+
+# Default archival threshold (7 days)
+DEFAULT_ARCHIVE_THRESHOLD_DAYS = 7
+
+
+class SessionArchiveManager:
+    """Manages session archival and restoration"""
+
+    def __init__(self, threshold_days: int = DEFAULT_ARCHIVE_THRESHOLD_DAYS, dry_run: bool = False):
+        """
+        Initialize the archive manager.
+
+        Args:
+            threshold_days: Number of days after which a session should be archived
+            dry_run: If True, simulate archival without making changes
+        """
+        self.threshold_days = threshold_days
+        self.dry_run = dry_run
+        self.cutoff_date = datetime.now(timezone.utc) - timedelta(days=threshold_days)
+
+    def log(self, event_data: Dict[str, Any]):
+        """Log an archival event"""
+        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
+
+        event = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "threshold_days": self.threshold_days,
+            "dry_run": self.dry_run,
+            **event_data
+        }
+
+        with open(LOG_FILE, "a") as f:
+            f.write(json.dumps(event) + "\n")
+
+    def load_sessions_index(self) -> Dict[str, Any]:
+        """Load the main sessions index file"""
+        if not SESSIONS_INDEX.exists():
+            return {"sessions": [], "metadata": {"created_at": datetime.now(timezone.utc).isoformat()}}
+
+        with open(SESSIONS_INDEX) as f:
+            return json.load(f)
+
+    def save_sessions_index(self, data: Dict[str, Any]):
+        """Save the main sessions index file (unless dry run)"""
+        if self.dry_run:
+            return
+
+        SESSIONS_INDEX.parent.mkdir(parents=True, exist_ok=True)
+        with open(SESSIONS_INDEX, "w") as f:
+            json.dump(data, f, indent=2)
+
+    def load_archived_index(self) -> Dict[str, Any]:
+        """Load the archived sessions index file"""
+        if not ARCHIVED_INDEX.exists():
+            return {"archived_sessions": [], "metadata": {"created_at": datetime.now(timezone.utc).isoformat()}}
+
+        with open(ARCHIVED_INDEX) as f:
+            return json.load(f)
+
+    def save_archived_index(self, data: Dict[str, Any]):
+        """Save the archived sessions index file (unless dry run)"""
+        if self.dry_run:
+            return
+
+        ARCHIVED_DIR.mkdir(parents=True, exist_ok=True)
+        with open(ARCHIVED_INDEX, "w") as f:
+            json.dump(data, f, indent=2)
+
+    def get_session_files(self) -> List[Path]:
+        """Get all session .jsonl files in the sessions directory"""
+        return list(SESSIONS_DIR.glob("*.jsonl"))
+
+    def parse_session_timestamp(self, session_file: Path) -> Optional[datetime]:
+        """
+        Parse the timestamp from a session file.
+        First tries to read the first line to get the session timestamp,
+        otherwise falls back to file modification time.
+        """
+        try:
+            with open(session_file) as f:
+                first_line = f.readline().strip()
+                if first_line:
+                    session_data = json.loads(first_line)
+                    if "timestamp" in session_data:
+                        return datetime.fromisoformat(session_data["timestamp"].replace("Z", "+00:00"))
+        except (json.JSONDecodeError, KeyError, ValueError):
+            pass
+
+        # Fall back to file modification time
+        return datetime.fromtimestamp(session_file.stat().st_mtime, tz=timezone.utc)
+
+    def archive_session(self, session_file: Path) -> Optional[Dict[str, Any]]:
+        """
+        Archive a single session file.
+
+        Returns:
+            Archive info dict if successful, None otherwise
+        """
+        try:
+            session_id = session_file.stem
+            timestamp = self.parse_session_timestamp(session_file)
+
+            if timestamp and timestamp >= self.cutoff_date:
+                return None  # Session is too recent, don't archive
+
+            archive_path = ARCHIVED_DIR / session_file.name
+
+            # Check if already archived
+            if archive_path.exists():
+                return {
+                    "session_id": session_id,
+                    "status": "already_archived",
+                    "timestamp": timestamp.isoformat() if timestamp else "unknown"
+                }
+
+            if not self.dry_run:
+                shutil.move(str(session_file), str(archive_path))
+
+            return {
+                "session_id": session_id,
+                "status": "archived",
+                "timestamp": timestamp.isoformat() if timestamp else "unknown",
+                "original_path": str(session_file),
+                "archive_path": str(archive_path)
+            }
+
+        except Exception as e:
+            return {
+                "session_id": session_file.stem,
+                "status": "error",
+                "error": str(e)
+            }
+
+    def update_main_index(self, archived_sessions: List[Dict[str, Any]]):
+        """Remove archived sessions from the main index"""
+        index = self.load_sessions_index()
+        archived_ids = {s["session_id"] for s in archived_sessions if s.get("status") == "archived"}
+
+        original_count = len(index.get("sessions", []))
+        index["sessions"] = [
+            s for s in index.get("sessions", [])
+            if s.get("id") not in archived_ids and s.get("session_id") not in archived_ids
+        ]
+
+        # Add metadata about archival
+        index.setdefault("metadata", {})["last_archived_at"] = datetime.now(timezone.utc).isoformat()
+        index["metadata"]["archived_count"] = len(archived_sessions)
+
+        self.save_sessions_index(index)
+
+        return original_count - len(index.get("sessions", []))
+
+    def update_archived_index(self, archived_sessions: List[Dict[str, Any]]):
+        """Add newly archived sessions to the archived index"""
+        index = self.load_archived_index()
+
+        for session_info in archived_sessions:
+            if session_info.get("status") == "archived":
+                index.setdefault("archived_sessions", []).append({
+                    "session_id": session_info["session_id"],
+                    "timestamp": session_info["timestamp"],
+                    "archived_at": datetime.now(timezone.utc).isoformat(),
+                    "archive_path": session_info.get("archive_path")
+                })
+
+        index.setdefault("metadata", {})["last_updated"] = datetime.now(timezone.utc).isoformat()
+
+        self.save_archived_index(index)
+
+    def archive_sessions(self) -> Dict[str, Any]:
+        """
+        Main archival function - finds and archives old sessions.
+
+        Returns:
+            Summary of archival operation
+        """
+        session_files = self.get_session_files()
+
+        if not session_files:
+            result = {
+                "action": "archive_sessions",
+                "total_files": 0,
+                "archived_count": 0,
+                "skipped_count": 0,
+                "error_count": 0,
+                "status": "no_sessions_found"
+            }
+            self.log(result)
+            return result
+
+        archived_sessions = []
+        skipped = 0
+        errors = 0
+
+        for session_file in session_files:
+            result = self.archive_session(session_file)
+            if result:
+                if result.get("status") == "archived":
+                    archived_sessions.append(result)
+                elif result.get("status") == "already_archived":
+                    skipped += 1
+                elif result.get("status") == "error":
+                    errors += 1
+
+        # Update indices
+        if archived_sessions:
+            removed_from_index = self.update_main_index(archived_sessions)
+            self.update_archived_index(archived_sessions)
+
+        result = {
+            "action": "archive_sessions",
+            "total_files": len(session_files),
+            "archived_count": len(archived_sessions),
+            "skipped_count": skipped,
+            "error_count": errors,
+            "removed_from_index": removed_from_index if archived_sessions else 0,
+            "status": "success" if archived_sessions or skipped > 0 else "no_sessions_to_archive"
+        }
+
+        self.log(result)
+        return result
+
+    def restore_session(self, session_id: str) -> Dict[str, Any]:
+        """
+        Restore an archived session back to the main sessions directory.
+
+        Args:
+            session_id: The ID of the session to restore (without .jsonl extension)
+
+        Returns:
+            Result of the restore operation
+        """
+        archive_file = ARCHIVED_DIR / f"{session_id}.jsonl"
+        target_file = SESSIONS_DIR / f"{session_id}.jsonl"
+
+        if not archive_file.exists():
+            result = {
+                "action": "restore_session",
+                "session_id": session_id,
+                "status": "not_found",
+                "error": f"Archived session {session_id} not found"
+            }
+            self.log(result)
+            return result
+
+        if target_file.exists():
+            result = {
+                "action": "restore_session",
+                "session_id": session_id,
+                "status": "conflict",
+                "error": f"Session file {session_id}.jsonl already exists in main directory"
+            }
+            self.log(result)
+            return result
+
+        if not self.dry_run:
+            shutil.copy(str(archive_file), str(target_file))
+
+        # Update archived index to mark as restored
+        index = self.load_archived_index()
+        for session in index.get("archived_sessions", []):
+            if session.get("session_id") == session_id:
+                session["restored_at"] = datetime.now(timezone.utc).isoformat()
+                session["status"] = "restored"
+                break
+        self.save_archived_index(index)
+
+        result = {
+            "action": "restore_session",
+            "session_id": session_id,
+            "status": "success",
+            "restored_from": str(archive_file),
+            "restored_to": str(target_file)
+        }
+        self.log(result)
+        return result
+
+    def list_archived_sessions(self) -> List[Dict[str, Any]]:
+        """List all archived sessions"""
+        index = self.load_archived_index()
+        return index.get("archived_sessions", [])
+
+    def get_stats(self) -> Dict[str, Any]:
+        """Get statistics about session archival"""
+        session_files = self.get_session_files()
+        archived_files = list(ARCHIVED_DIR.glob("*.jsonl")) if ARCHIVED_DIR.exists() else []
+
+        # Count active vs archived
+        active_count = len(session_files)
+        archived_count = len(archived_files)
+
+        # Get index stats
+        main_index = self.load_sessions_index()
+        archived_index = self.load_archived_index()
+
+        return {
+            "active_sessions": active_count,
+            "archived_sessions": archived_count,
+            "main_index_entries": len(main_index.get("sessions", [])),
+            "archived_index_entries": len(archived_index.get("archived_sessions", [])),
+            "threshold_days": self.threshold_days,
+            "cutoff_date": self.cutoff_date.isoformat(),
+            "sessions_dir": str(SESSIONS_DIR),
+            "archived_dir": str(ARCHIVED_DIR)
+        }
+
+
+def main():
+    """CLI entry point for session archival"""
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Cain Session Archive Manager")
+    parser.add_argument("--archive", action="store_true", help="Run archival process")
+    parser.add_argument("--restore", metavar="SESSION_ID", help="Restore a session by ID")
+    parser.add_argument("--list", action="store_true", help="List archived sessions")
+    parser.add_argument("--stats", action="store_true", help="Show archival statistics")
+    parser.add_argument("--dry-run", action="store_true", help="Simulate without making changes")
+    parser.add_argument("--threshold", type=int, default=DEFAULT_ARCHIVE_THRESHOLD_DAYS,
+                       help=f"Archive threshold in days (default: {DEFAULT_ARCHIVE_THRESHOLD_DAYS})")
+
+    args = parser.parse_args()
+
+    manager = SessionArchiveManager(threshold_days=args.threshold, dry_run=args.dry_run)
+
+    if args.archive:
+        print(f"Running session archival (threshold: {args.threshold} days, dry_run: {args.dry_run})")
+        result = manager.archive_sessions()
+        print(json.dumps(result, indent=2))
+
+    elif args.restore:
+        print(f"Restoring session: {args.restore} (dry_run: {args.dry_run})")
+        result = manager.restore_session(args.restore)
+        print(json.dumps(result, indent=2))
+
+    elif args.list:
+        sessions = manager.list_archived_sessions()
+        print(f"Archived sessions: {len(sessions)}")
+        for session in sessions:
+            print(f"  - {session.get('session_id')}: {session.get('timestamp')}")
+
+    elif args.stats:
+        stats = manager.get_stats()
+        print(json.dumps(stats, indent=2))
+
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()

.openclaw/cron/README.md

@@ -37,6 +37,82 @@ Located in `.openclaw/cron/jobs.json`, following the OpenClaw cron schema:
 - **Auto-recovery**: Restarts space if in RUNTIME_ERROR or BUILDING state for >10 minutes
 - **Logs**: `.openclaw/logs/health-check.jsonl`
 
+### session-archive
+- **Schedule**: Weekly on Sunday at 2:00 AM
+- **Purpose**: Automatically archive sessions older than 7 days
+- **Tool**: `session_archive`
+- **Parameters**: `threshold_days: 7`, `dry_run: false`
+- **Logs**: `.openclaw/logs/session-archive.jsonl`
+
+## Session Archival
+
+Cain includes an automatic session archival system to keep the active sessions directory clean and performant.
+
+### How It Works
+
+1. **Automatic Archival**: Sessions older than 7 days (configurable) are automatically moved to the archived directory
+2. **Scheduled Job**: Runs weekly by default (Sunday at 2:00 AM)
+3. **Index Updates**: Both the main and archived session indices are updated
+4. **Restore Capability**: Archived sessions can be restored if needed
+
+### Directory Structure
+
+```
+.openclaw/agents/main/sessions/
+├── sessions.json              # Main sessions index (active sessions only)
+├── session_*.jsonl            # Active session files
+└── archived/
+    ├── archived_sessions.json # Index of archived sessions
+    └── session_*.jsonl        # Archived session files
+```
+
+### Manual Archival Operations
+
+Run the archive manager directly:
+
+```bash
+# Archive sessions (dry-run first)
+python3 .openclaw/agents/main/sessions/archive_manager.py --archive --dry-run
+
+# Actually archive sessions
+python3 .openclaw/agents/main/sessions/archive_manager.py --archive
+
+# List archived sessions
+python3 .openclaw/agents/main/sessions/archive_manager.py --list
+
+# Restore a session
+python3 .openclaw/agents/main/sessions/archive_manager.py --restore SESSION_ID
+
+# Show statistics
+python3 .openclaw/agents/main/sessions/archive_manager.py --stats
+
+# Custom threshold (e.g., 14 days)
+python3 .openclaw/agents/main/sessions/archive_manager.py --archive --threshold 14
+```
+
+### Archival Log Format
+
+```json
+{"timestamp":"2026-03-14T02:00:00Z","action":"archive_sessions","threshold_days":7,"dry_run":false,"total_files":50,"archived_count":35,"skipped_count":10,"error_count":0,"status":"success"}
+{"timestamp":"2026-03-14T02:00:01Z","action":"restore_session","session_id":"session_abc123","status":"success","restored_from":"/path/to/archived/session_abc123.jsonl"}
+```
+
+### Configuration
+
+To modify the archival behavior, edit `.openclaw/cron/jobs/session-archive.json`:
+
+```json
+{
+  "id": "session-archive",
+  "schedule": "0 2 * * 0",  // Adjust cron schedule
+  "enabled": true,
+  "params": {
+    "threshold_days": 7,     // Days before archival
+    "dry_run": false         // Set true for testing
+  }
+}
+```
+
 ## Log Format
 
 Health checks are logged in JSONL format:

.openclaw/cron/executor.py

@@ -17,12 +17,25 @@ except ImportError:
     print("Error: huggingface_hub not installed. Run: pip install huggingface_hub")
     sys.exit(1)
 
+# Add archive manager module path
+sys.path.insert(0, str(Path(__file__).parent.parent / "agents" / "main" / "sessions"))
+
+try:
+    from archive_manager import SessionArchiveManager
+except ImportError:
+    SessionArchiveManager = None
+    print("Warning: archive_manager not found. Session archival will be skipped.")
+
 JOBS_FILE = Path(__file__).parent / "jobs.json"
+JOBS_DIR = Path(__file__).parent / "jobs"
 LOGS_DIR = Path(__file__).parent.parent / "logs"
 
 # Space configuration
 SPACE_ID = "tao-shen/HuggingClaw-Cain"
 
+# Session archive configuration
+ARCHIVE_MANAGER_PATH = Path(__file__).parent.parent / "agents" / "main" / "sessions" / "archive_manager.py"
+
 
 def get_hf_api():
     """Get HuggingFace API instance"""
@@ -87,6 +100,52 @@ def hf_restart_space(space_id: str = SPACE_ID) -> dict:
         }
 
 
+def run_session_archive(threshold_days: int = 7, dry_run: bool = False) -> dict:
+    """Run session archival process"""
+    if SessionArchiveManager is None:
+        return {
+            "success": False,
+            "detail": "SessionArchiveManager not available"
+        }
+
+    try:
+        manager = SessionArchiveManager(threshold_days=threshold_days, dry_run=dry_run)
+        result = manager.archive_sessions()
+
+        return {
+            "success": result.get("status") in ["success", "no_sessions_to_archive", "no_sessions_found"],
+            "detail": result
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "detail": str(e)
+        }
+
+
+def restore_session(session_id: str) -> dict:
+    """Restore an archived session"""
+    if SessionArchiveManager is None:
+        return {
+            "success": False,
+            "detail": "SessionArchiveManager not available"
+        }
+
+    try:
+        manager = SessionArchiveManager(dry_run=False)
+        result = manager.restore_session(session_id)
+
+        return {
+            "success": result.get("status") == "success",
+            "detail": result
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "detail": str(e)
+        }
+
+
 def log_event(job_id: str, event_data: dict):
     """Log an event to the appropriate log file"""
     log_file = LOGS_DIR / f"{job_id}.jsonl"
@@ -103,10 +162,26 @@ def log_event(job_id: str, event_data: dict):
 
 
 def load_jobs() -> list:
-    """Load jobs from configuration file"""
-    with open(JOBS_FILE) as f:
-        config = json.load(f)
-    return config.get("jobs", [])
+    """Load jobs from configuration files (both jobs.json and jobs/*.json)"""
+    jobs = []
+
+    # Load from main jobs.json
+    if JOBS_FILE.exists():
+        with open(JOBS_FILE) as f:
+            config = json.load(f)
+        jobs.extend(config.get("jobs", []))
+
+    # Load from jobs/ directory
+    if JOBS_DIR.exists():
+        for job_file in JOBS_DIR.glob("*.json"):
+            try:
+                with open(job_file) as f:
+                    job_config = json.load(f)
+                jobs.append(job_config)
+            except Exception as e:
+                print(f"Warning: Failed to load job from {job_file}: {e}")
+
+    return jobs
 
 
 def check_condition(status: str, condition: str) -> bool:
@@ -128,49 +203,78 @@ def run_job(job: dict):
     print(f"[{datetime.utcnow().isoformat()}] Running job: {job_id}")
 
     try:
-        # Get health status
-        result = hf_space_status(space_id="tao-shen/HuggingClaw-Cain")
-
-        status = result.get("stage", "UNKNOWN")
-        detail = result.get("detail", "")
+        # Check job type and execute accordingly
+        tool = job.get("tool")
+
+        if tool == "session_archive":
+            # Session archival job
+            params = job.get("params", {})
+            threshold_days = params.get("threshold_days", 7)
+            dry_run = params.get("dry_run", False)
+
+            print(f"  Archiving sessions older than {threshold_days} days (dry_run: {dry_run})")
+            result = run_session_archive(threshold_days=threshold_days, dry_run=dry_run)
+
+            log_event(job_id, {
+                "action": "session_archive",
+                "success": result.get("success", False),
+                "detail": result.get("detail", {})
+            })
+
+            print(f"  Result: {result.get('detail', {}).get('status', 'unknown')}")
+
+            if result.get("success"):
+                detail = result.get("detail", {})
+                print(f"  Archived: {detail.get('archived_count', 0)}, "
+                      f"Skipped: {detail.get('skipped_count', 0)}, "
+                      f"Errors: {detail.get('error_count', 0)}")
+            else:
+                print(f"  Error: {result.get('detail')}")
 
-        log_event(job_id, {
-            "status": status,
-            "stage": result.get("stage"),
-            "detail": detail,
-            "action": "check"
-        })
+        else:
+            # Default: HF space status check
+            result = hf_space_status(space_id="tao-shen/HuggingClaw-Cain")
 
-        print(f"  Status: {status} - {detail}")
+            status = result.get("stage", "UNKNOWN")
+            detail = result.get("detail", "")
 
-        # Check failure condition
-        if "on_failure" in job:
-            failure_config = job["on_failure"]
-            condition = failure_config.get("condition", "")
+            log_event(job_id, {
+                "status": status,
+                "stage": result.get("stage"),
+                "detail": detail,
+                "action": "check"
+            })
 
-            if check_condition(status, condition):
-                print(f"  Failure condition met: {condition}")
-                log_event(job_id, {
-                    "status": status,
-                    "action": "recovery_triggered",
-                    "condition": condition
-                })
+            print(f"  Status: {status} - {detail}")
 
-                # Execute recovery action
-                if failure_config.get("tool") == "hf_restart_space":
-                    print("  Attempting recovery via hf_restart_space...")
-                    recovery_result = hf_restart_space(space_id="tao-shen/HuggingClaw-Cain")
+            # Check failure condition
+            if "on_failure" in job:
+                failure_config = job["on_failure"]
+                condition = failure_config.get("condition", "")
 
+                if check_condition(status, condition):
+                    print(f"  Failure condition met: {condition}")
                     log_event(job_id, {
-                        "action": "recovery_attempt",
-                        "success": recovery_result.get("success", False),
-                        "detail": recovery_result.get("detail", "")
+                        "status": status,
+                        "action": "recovery_triggered",
+                        "condition": condition
                     })
 
-                    if recovery_result.get("success"):
-                        print("  Recovery initiated successfully")
-                    else:
-                        print(f"  Recovery failed: {recovery_result.get('detail')}")
+                    # Execute recovery action
+                    if failure_config.get("tool") == "hf_restart_space":
+                        print("  Attempting recovery via hf_restart_space...")
+                        recovery_result = hf_restart_space(space_id="tao-shen/HuggingClaw-Cain")
+
+                        log_event(job_id, {
+                            "action": "recovery_attempt",
+                            "success": recovery_result.get("success", False),
+                            "detail": recovery_result.get("detail", "")
+                        })
+
+                        if recovery_result.get("success"):
+                            print("  Recovery initiated successfully")
+                        else:
+                            print(f"  Recovery failed: {recovery_result.get('detail')}")
 
     except Exception as e:
         print(f"  Error: {e}")

.openclaw/cron/jobs/session-archive.json

@@ -0,0 +1,27 @@
+{
+  "id": "session-archive",
+  "schedule": "0 2 * * 0",
+  "enabled": true,
+  "description": "Weekly archival of sessions older than 7 days to the archived directory",
+  "tool": "session_archive",
+  "params": {
+    "threshold_days": 7,
+    "dry_run": false
+  },
+  "logging": {
+    "file": ".openclaw/logs/session-archive.jsonl",
+    "format": "jsonl",
+    "retention_days": 90
+  },
+  "on_failure": {
+    "notify": true,
+    "retry": {
+      "max_attempts": 3,
+      "delay_minutes": 60
+    }
+  },
+  "metadata": {
+    "created_at": "2026-03-14T00:00:00Z",
+    "version": "1.0"
+  }
+}