ezra-environment/tools/session_backup.py

#!/usr/bin/env python3
"""
Session and state backup automation for Ezra.
Backs up critical files: sessions, memory, config, state.db.

Epic: EZRA-SELF-001 / Phase 4 - Session Management
Author: Ezra (self-improvement)
"""

import json
import os
import shutil
import tarfile
import time
from datetime import datetime
from pathlib import Path


class SessionBackup:
    """Automated backup of Ezra's state and sessions."""

    def __init__(
        self,
        home_dir: str = None,
        backup_dir: str = None,
        max_backups: int = 10,
    ):
        self.home_dir = Path(home_dir or "/root/wizards/ezra/home")
        self.backup_dir = Path(backup_dir or "/root/wizards/ezra/backups")
        self.max_backups = max_backups
        self.backup_dir.mkdir(parents=True, exist_ok=True)

    # Files/patterns to back up
    CRITICAL_FILES = [
        "config.yaml",
        "memories/MEMORY.md",
        "memories/USER.md",
        "state.db",
        "channel_directory.json",
        "gateway_state.json",
        "cron/jobs.json",
    ]

    CRITICAL_DIRS = [
        "sessions",
    ]

    def create_backup(self, label: str = None) -> dict:
        """Create a compressed backup of critical state."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        label = label or "auto"
        filename = f"ezra-backup-{timestamp}-{label}.tar.gz"
        filepath = self.backup_dir / filename

        files_included = []
        files_missing = []
        total_size = 0

        with tarfile.open(filepath, "w:gz") as tar:
            # Individual critical files
            for rel_path in self.CRITICAL_FILES:
                full_path = self.home_dir / rel_path
                if full_path.exists():
                    tar.add(full_path, arcname=rel_path)
                    size = full_path.stat().st_size
                    files_included.append({"path": rel_path, "size": size})
                    total_size += size
                else:
                    files_missing.append(rel_path)

            # Session files (only metadata, not full JSONL)
            sessions_dir = self.home_dir / "sessions"
            if sessions_dir.exists():
                # Include session index
                sessions_json = sessions_dir / "sessions.json"
                if sessions_json.exists():
                    tar.add(sessions_json, arcname="sessions/sessions.json")
                    files_included.append({"path": "sessions/sessions.json", "size": sessions_json.stat().st_size})

                # Include session metadata (small files)
                for f in sessions_dir.glob("session_*.json"):
                    if f.stat().st_size < 100_000:  # Skip huge session files
                        tar.add(f, arcname=f"sessions/{f.name}")
                        files_included.append({"path": f"sessions/{f.name}", "size": f.stat().st_size})
                        total_size += f.stat().st_size

        backup_size = filepath.stat().st_size

        result = {
            "filename": filename,
            "path": str(filepath),
            "backup_size": backup_size,
            "backup_size_human": self._human_size(backup_size),
            "source_size": total_size,
            "files_included": len(files_included),
            "files_missing": files_missing,
            "timestamp": timestamp,
        }

        # Rotate old backups
        self._rotate_backups()

        return result

    def _rotate_backups(self):
        """Remove old backups beyond max_backups."""
        backups = sorted(
            self.backup_dir.glob("ezra-backup-*.tar.gz"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
        for old in backups[self.max_backups:]:
            old.unlink()

    def list_backups(self) -> list[dict]:
        """List existing backups."""
        backups = []
        for f in sorted(self.backup_dir.glob("ezra-backup-*.tar.gz"), reverse=True):
            stat = f.stat()
            backups.append({
                "filename": f.name,
                "size": self._human_size(stat.st_size),
                "created": datetime.fromtimestamp(stat.st_mtime).isoformat(),
                "age_hours": round((time.time() - stat.st_mtime) / 3600, 1),
            })
        return backups

    def restore_backup(self, filename: str, dry_run: bool = True) -> dict:
        """Restore from a backup. Use dry_run=True to preview."""
        filepath = self.backup_dir / filename
        if not filepath.exists():
            return {"error": f"Backup not found: {filename}"}

        with tarfile.open(filepath, "r:gz") as tar:
            members = tar.getmembers()

            if dry_run:
                return {
                    "mode": "dry_run",
                    "filename": filename,
                    "files": [m.name for m in members],
                    "total_files": len(members),
                }

            # Actual restore
            tar.extractall(path=str(self.home_dir))
            return {
                "mode": "restored",
                "filename": filename,
                "files_restored": len(members),
            }

    def check_freshness(self) -> dict:
        """Check if backups are fresh enough."""
        backups = self.list_backups()
        if not backups:
            return {"fresh": False, "reason": "No backups exist", "latest": None}

        latest = backups[0]
        age = latest["age_hours"]
        return {
            "fresh": age < 24,
            "latest": latest["filename"],
            "age_hours": age,
            "total_backups": len(backups),
        }

    @staticmethod
    def _human_size(size: int) -> str:
        for unit in ["B", "KB", "MB", "GB"]:
            if size < 1024:
                return f"{size:.1f}{unit}"
            size /= 1024
        return f"{size:.1f}TB"


if __name__ == "__main__":
    backup = SessionBackup()

    # Create a backup
    result = backup.create_backup("manual")
    print(f"Created: {result['filename']} ({result['backup_size_human']})")
    print(f"Files: {result['files_included']} included, {len(result['files_missing'])} missing")
    if result["files_missing"]:
        print(f"Missing: {', '.join(result['files_missing'])}")

    # List backups
    print("\nExisting backups:")
    for b in backup.list_backups():
        print(f"  {b['filename']} - {b['size']} ({b['age_hours']}h ago)")