feat(ops): add cleanup commands

src/cleveragents/application/services/cleanup_service.py

@@ -0,0 +1,482 @@
+"""Garbage collection and cleanup service (CONC3).
+
+Provides centralized cleanup for stale sandboxes, excess checkpoints,
+inactive sessions, expired logs, and expired backups.  Supports
+dry-run mode (report without deleting) and full purge.
+"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+import time
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from cleveragents.config.settings import Settings
+
+__all__ = [
+    "CleanupReport",
+    "CleanupService",
+    "ResourceCleanupSummary",
+    "StaleItem",
+]
+
+
+# ── Data classes ──────────────────────────────────────────────────
+
+
+@dataclass(slots=True)
+class StaleItem:
+    """A single item identified for cleanup."""
+
+    resource_type: str
+    path: str
+    age_description: str
+    plan_id: str | None = None
+
+
+@dataclass(slots=True)
+class ResourceCleanupSummary:
+    """Per-resource-type cleanup result."""
+
+    resource_type: str
+    scanned: int = 0
+    removed: int = 0
+    skipped: int = 0
+    skipped_reasons: list[str] = field(default_factory=list)
+
+
+@dataclass(slots=True)
+class CleanupReport:
+    """Full cleanup report across all resource types."""
+
+    dry_run: bool = False
+    sandboxes: ResourceCleanupSummary = field(
+        default_factory=lambda: ResourceCleanupSummary(resource_type="sandboxes"),
+    )
+    checkpoints: ResourceCleanupSummary = field(
+        default_factory=lambda: ResourceCleanupSummary(resource_type="checkpoints"),
+    )
+    sessions: ResourceCleanupSummary = field(
+        default_factory=lambda: ResourceCleanupSummary(resource_type="sessions"),
+    )
+    logs: ResourceCleanupSummary = field(
+        default_factory=lambda: ResourceCleanupSummary(resource_type="logs"),
+    )
+    backups: ResourceCleanupSummary = field(
+        default_factory=lambda: ResourceCleanupSummary(resource_type="backups"),
+    )
+    stale_items: list[StaleItem] = field(default_factory=list)
+
+    def as_dict(self) -> dict[str, Any]:
+        """Serialize the report for CLI output."""
+        return {
+            "dry_run": self.dry_run,
+            "sandboxes": {
+                "scanned": self.sandboxes.scanned,
+                "removed": self.sandboxes.removed,
+                "skipped": self.sandboxes.skipped,
+            },
+            "checkpoints": {
+                "scanned": self.checkpoints.scanned,
+                "removed": self.checkpoints.removed,
+                "skipped": self.checkpoints.skipped,
+            },
+            "sessions": {
+                "scanned": self.sessions.scanned,
+                "removed": self.sessions.removed,
+                "skipped": self.sessions.skipped,
+            },
+            "logs": {
+                "scanned": self.logs.scanned,
+                "removed": self.logs.removed,
+                "skipped": self.logs.skipped,
+            },
+            "backups": {
+                "scanned": self.backups.scanned,
+                "removed": self.backups.removed,
+                "skipped": self.backups.skipped,
+            },
+        }
+
+
+# ── Service ───────────────────────────────────────────────────────
+
+
+class CleanupService:
+    """Orchestrates garbage collection for the platform.
+
+    Args:
+        settings: Application settings with retention policy fields.
+        active_plan_ids: Set of plan IDs currently running (protected).
+    """
+
+    def __init__(
+        self,
+        settings: Settings,
+        active_plan_ids: frozenset[str] | None = None,
+    ) -> None:
+        if not isinstance(settings, Settings):
+            raise TypeError("settings must be a Settings instance")
+        self._settings = settings
+        self._active_plan_ids: frozenset[str] = active_plan_ids or frozenset()
+
+    # ── Public API ────────────────────────────────────────────────
+
+    def scan(self) -> CleanupReport:
+        """Scan for stale resources without deleting anything.
+
+        Returns:
+            A CleanupReport with dry_run=True containing identified
+            stale items and per-resource counts.
+        """
+        report = CleanupReport(dry_run=True)
+        self._scan_sandboxes(report)
+        self._scan_checkpoints(report)
+        self._scan_sessions(report)
+        self._scan_logs(report)
+        self._scan_backups(report)
+        return report
+
+    def purge(self, *, purge_all: bool = False) -> CleanupReport:
+        """Delete stale resources.
+
+        Args:
+            purge_all: When True, purge all resource types.
+                When False, only purge sandboxes and checkpoints.
+
+        Returns:
+            A CleanupReport with removal counts.
+        """
+        report = CleanupReport(dry_run=False)
+        self._purge_sandboxes(report)
+        self._purge_checkpoints(report)
+        if purge_all:
+            self._purge_sessions(report)
+            self._purge_logs(report)
+            self._purge_backups(report)
+        return report
+
+    # ── Sandbox cleanup ───────────────────────────────────────────
+
+    def _get_sandbox_dirs(self) -> list[Path]:
+        """Find sandbox directories in the system temp directory."""
+        tmp = Path(tempfile.gettempdir())
+        if not tmp.exists():
+            return []
+        dirs: list[Path] = []
+        for prefix in ("ca-sandbox-", "ca-cow-sandbox-"):
+            dirs.extend(
+                p for p in tmp.iterdir() if p.is_dir() and p.name.startswith(prefix)
+            )
+        return dirs
+
+    def _extract_plan_id_from_sandbox(self, path: Path) -> str | None:
+        """Extract plan_id from a sandbox directory name."""
+        name = path.name
+        if name.startswith("ca-sandbox-"):
+            # Format: ca-sandbox-<plan_id>-<random>
+            parts = name[len("ca-sandbox-") :].rsplit("-", 1)
+            if len(parts) >= 1:
+                return parts[0]
+        return None
+
+    def _is_sandbox_stale(self, path: Path) -> bool:
+        """Check if a sandbox directory exceeds the max age."""
+        max_age_seconds = self._settings.cleanup_sandbox_max_age_hours * 3600
+        try:
+            mtime = path.stat().st_mtime
+        except OSError:
+            return False
+        return (time.time() - mtime) > max_age_seconds
+
+    def _scan_sandboxes(self, report: CleanupReport) -> None:
+        """Populate report with stale sandbox info."""
+        dirs = self._get_sandbox_dirs()
+        for d in dirs:
+            report.sandboxes.scanned += 1
+            plan_id = self._extract_plan_id_from_sandbox(d)
+            if plan_id and plan_id in self._active_plan_ids:
+                report.sandboxes.skipped += 1
+                report.sandboxes.skipped_reasons.append(
+                    f"Skipped {d.name}: linked to running plan {plan_id}"
+                )
+                continue
+            if self._is_sandbox_stale(d):
+                report.stale_items.append(
+                    StaleItem(
+                        resource_type="sandbox",
+                        path=str(d),
+                        age_description=self._age_description(d),
+                        plan_id=plan_id,
+                    )
+                )
+
+    def _purge_sandboxes(self, report: CleanupReport) -> None:
+        """Remove stale sandbox directories."""
+        dirs = self._get_sandbox_dirs()
+        for d in dirs:
+            report.sandboxes.scanned += 1
+            plan_id = self._extract_plan_id_from_sandbox(d)
+            if plan_id and plan_id in self._active_plan_ids:
+                report.sandboxes.skipped += 1
+                report.sandboxes.skipped_reasons.append(
+                    f"Skipped {d.name}: linked to running plan {plan_id}"
+                )
+                continue
+            if self._is_sandbox_stale(d):
+                try:
+                    shutil.rmtree(d)
+                    report.sandboxes.removed += 1
+                except OSError:
+                    report.sandboxes.skipped += 1
+
+    # ── Checkpoint cleanup ────────────────────────────────────────
+
+    def scan_checkpoints_for_plan(
+        self,
+        checkpoint_dir: Path,
+    ) -> list[Path]:
+        """Return excess checkpoint files for a plan directory.
+
+        Keeps the first and most recent, prunes the rest when count
+        exceeds ``cleanup_checkpoint_max_per_plan``.
+        """
+        if not checkpoint_dir.exists():
+            return []
+        files = sorted(checkpoint_dir.glob("checkpoint_*.json"))
+        max_count = self._settings.cleanup_checkpoint_max_per_plan
+        if len(files) <= max_count:
+            return []
+        # Keep first and last, prune from the middle
+        if len(files) < 3:
+            return []
+        middle = files[1:-1]
+        excess_count = len(files) - max_count
+        return middle[:excess_count]
+
+    def prune_checkpoints_for_plan(
+        self,
+        checkpoint_dir: Path,
+    ) -> int:
+        """Delete excess checkpoint files. Returns count removed."""
+        to_remove = self.scan_checkpoints_for_plan(checkpoint_dir)
+        removed = 0
+        for f in to_remove:
+            try:
+                f.unlink()
+                removed += 1
+            except OSError:
+                pass
+        return removed
+
+    def _scan_checkpoints(self, report: CleanupReport) -> None:
+        """Scan for excess checkpoints across all plan directories."""
+        checkpoint_base = self._get_checkpoint_base_dir()
+        if not checkpoint_base.exists():
+            return
+        for plan_dir in checkpoint_base.iterdir():
+            if not plan_dir.is_dir():
+                continue
+            excess = self.scan_checkpoints_for_plan(plan_dir)
+            report.checkpoints.scanned += len(list(plan_dir.glob("checkpoint_*.json")))
+            for f in excess:
+                report.stale_items.append(
+                    StaleItem(
+                        resource_type="checkpoint",
+                        path=str(f),
+                        age_description="excess",
+                    )
+                )
+
+    def _purge_checkpoints(self, report: CleanupReport) -> None:
+        """Prune excess checkpoints across all plan directories."""
+        checkpoint_base = self._get_checkpoint_base_dir()
+        if not checkpoint_base.exists():
+            return
+        for plan_dir in checkpoint_base.iterdir():
+            if not plan_dir.is_dir():
+                continue
+            files = list(plan_dir.glob("checkpoint_*.json"))
+            report.checkpoints.scanned += len(files)
+            removed = self.prune_checkpoints_for_plan(plan_dir)
+            report.checkpoints.removed += removed
+
+    def _get_checkpoint_base_dir(self) -> Path:
+        """Return the base directory where plan checkpoint dirs live."""
+        data_dir = self._settings.data_dir
+        if not data_dir.is_absolute():
+            data_dir = Path.cwd() / data_dir
+        return data_dir / "checkpoints"
+
+    # ── Session cleanup ───────────────────────────────────────────
+
+    def scan_inactive_sessions(
+        self,
+        sessions: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """Filter sessions that exceed the inactivity threshold.
+
+        Args:
+            sessions: List of session dicts with ``updated_at`` key
+                as a :class:`datetime` or ISO-8601 string.
+
+        Returns:
+            Sessions that are inactive beyond the threshold.
+        """
+        threshold_days = self._settings.cleanup_session_inactivity_days
+        now = datetime.now(tz=UTC)
+        inactive: list[dict[str, Any]] = []
+        for session in sessions:
+            updated = session.get("updated_at")
+            if updated is None:
+                continue
+            if isinstance(updated, str):
+                updated = datetime.fromisoformat(updated)
+            if not updated.tzinfo:
+                updated = updated.replace(tzinfo=UTC)
+            age_days = (now - updated).days
+            if age_days > threshold_days:
+                inactive.append(session)
+        return inactive
+
+    def _scan_sessions(self, report: CleanupReport) -> None:
+        """Placeholder scan — actual session scan requires DB access."""
+        # Populated by the CLI command which has DB access.
+        pass
+
+    def _purge_sessions(self, report: CleanupReport) -> None:
+        """Placeholder purge — actual session purge requires DB access."""
+        # Populated by the CLI command which has DB access.
+        pass
+
+    # ── Log cleanup ───────────────────────────────────────────────
+
+    def scan_expired_files(
+        self,
+        directory: Path,
+        retention_days: int,
+        pattern: str = "*.log",
+    ) -> list[Path]:
+        """Return files in *directory* older than *retention_days*."""
+        if not directory.exists():
+            return []
+        max_age_seconds = retention_days * 86400
+        now = time.time()
+        expired: list[Path] = []
+        for f in directory.glob(pattern):
+            if not f.is_file():
+                continue
+            try:
+                mtime = f.stat().st_mtime
+            except OSError:
+                continue
+            if (now - mtime) > max_age_seconds:
+                expired.append(f)
+        return expired
+
+    def _scan_logs(self, report: CleanupReport) -> None:
+        """Scan for expired log files."""
+        log_dir = self._settings.log_dir
+        if not log_dir.is_absolute():
+            log_dir = Path.cwd() / log_dir
+        expired = self.scan_expired_files(
+            log_dir,
+            self._settings.cleanup_log_retention_days,
+            pattern="*.log",
+        )
+        report.logs.scanned += len(
+            list(log_dir.glob("*.log")) if log_dir.exists() else []
+        )
+        for f in expired:
+            report.stale_items.append(
+                StaleItem(
+                    resource_type="log",
+                    path=str(f),
+                    age_description=self._age_description(f),
+                )
+            )
+
+    def _purge_logs(self, report: CleanupReport) -> None:
+        """Delete expired log files."""
+        log_dir = self._settings.log_dir
+        if not log_dir.is_absolute():
+            log_dir = Path.cwd() / log_dir
+        expired = self.scan_expired_files(
+            log_dir,
+            self._settings.cleanup_log_retention_days,
+            pattern="*.log",
+        )
+        report.logs.scanned += len(
+            list(log_dir.glob("*.log")) if log_dir.exists() else []
+        )
+        for f in expired:
+            try:
+                f.unlink()
+                report.logs.removed += 1
+            except OSError:
+                report.logs.skipped += 1
+
+    # ── Backup cleanup ────────────────────────────────────────────
+
+    def _get_backup_dir(self) -> Path:
+        """Return the backup directory."""
+        data_dir = self._settings.data_dir
+        if not data_dir.is_absolute():
+            data_dir = Path.cwd() / data_dir
+        return data_dir / "backups"
+
+    def _scan_backups(self, report: CleanupReport) -> None:
+        """Scan for expired backup files."""
+        backup_dir = self._get_backup_dir()
+        expired = self.scan_expired_files(
+            backup_dir,
+            self._settings.cleanup_backup_retention_days,
+            pattern="*",
+        )
+        all_files = list(backup_dir.iterdir()) if backup_dir.exists() else []
+        report.backups.scanned += len([f for f in all_files if f.is_file()])
+        for f in expired:
+            report.stale_items.append(
+                StaleItem(
+                    resource_type="backup",
+                    path=str(f),
+                    age_description=self._age_description(f),
+                )
+            )
+
+    def _purge_backups(self, report: CleanupReport) -> None:
+        """Delete expired backup files."""
+        backup_dir = self._get_backup_dir()
+        expired = self.scan_expired_files(
+            backup_dir,
+            self._settings.cleanup_backup_retention_days,
+            pattern="*",
+        )
+        all_files = list(backup_dir.iterdir()) if backup_dir.exists() else []
+        report.backups.scanned += len([f for f in all_files if f.is_file()])
+        for f in expired:
+            try:
+                f.unlink()
+                report.backups.removed += 1
+            except OSError:
+                report.backups.skipped += 1
+
+    # ── Helpers ───────────────────────────────────────────────────
+
+    @staticmethod
+    def _age_description(path: Path) -> str:
+        """Human-readable age description for a file or directory."""
+        try:
+            mtime = path.stat().st_mtime
+        except OSError:
+            return "unknown age"
+        age_seconds = time.time() - mtime
+        if age_seconds < 3600:
+            return f"{int(age_seconds / 60)} minutes old"
+        if age_seconds < 86400:
+            return f"{int(age_seconds / 3600)} hours old"
+        return f"{int(age_seconds / 86400)} days old"

src/cleveragents/cli/commands/cleanup.py

@@ -0,0 +1,153 @@
+"""CLI commands for garbage collection and cleanup (CONC3).
+
+Provides ``agents cleanup`` with subcommands for scanning and purging
+stale sandboxes, checkpoints, sessions, logs, and backups.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated
+
+import typer
+
+from cleveragents.cli.main import get_console, get_err_console
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.cleanup_service import (
+        CleanupReport,
+        CleanupService,
+    )
+
+app = typer.Typer(
+    help="Garbage collection and cleanup for stale resources.",
+)
+
+
+def _get_cleanup_service() -> CleanupService:
+    """Build a :class:`CleanupService` from current settings."""
+    from cleveragents.application.services.cleanup_service import CleanupService
+    from cleveragents.config.settings import get_settings
+
+    settings = get_settings()
+    # In the future, active_plan_ids would be fetched from the DB.
+    # For MVP, we pass an empty set.
+    return CleanupService(settings=settings, active_plan_ids=frozenset())
+
+
+@app.command(name="scan")
+def scan() -> None:
+    """Scan for stale resources without deleting anything (dry-run)."""
+    console = get_console()
+    service = _get_cleanup_service()
+    report = service.scan()
+
+    console.print("\n[bold]Cleanup Scan Results (dry-run)[/bold]\n")
+    _print_summary(report)
+
+    if report.stale_items:
+        console.print("\n[dim]Stale items found:[/dim]")
+        for item in report.stale_items:
+            console.print(
+                f"  [{item.resource_type}] {item.path} ({item.age_description})"
+            )
+    else:
+        console.print("\n[green]No stale resources found.[/green]")
+
+
+@app.command(name="purge")
+def purge(
+    dry_run: Annotated[
+        bool,
+        typer.Option("--dry-run", help="Show what would be cleaned without deleting."),
+    ] = False,
+    purge_all: Annotated[
+        bool,
+        typer.Option(
+            "--all",
+            help=(
+                "Purge all resource types "
+                "(sandboxes, checkpoints, sessions, "
+                "logs, backups)."
+            ),
+        ),
+    ] = False,
+    yes: Annotated[
+        bool,
+        typer.Option("--yes", "-y", help="Skip confirmation prompt."),
+    ] = False,
+) -> None:
+    """Remove stale resources based on retention policies."""
+    console = get_console()
+    err_console = get_err_console()
+    service = _get_cleanup_service()
+
+    if dry_run:
+        report = service.scan()
+        console.print("\n[bold]Cleanup Dry-Run Report[/bold]\n")
+        _print_summary(report)
+        if report.stale_items:
+            console.print("\n[dim]Would clean:[/dim]")
+            for item in report.stale_items:
+                console.print(f"  [{item.resource_type}] {item.path}")
+        else:
+            console.print("\n[green]Nothing to clean.[/green]")
+        return
+
+    if not yes:
+        scope = "all resource types" if purge_all else "sandboxes and checkpoints"
+        confirm = typer.confirm(f"Purge stale {scope}?")
+        if not confirm:
+            err_console.print("[yellow]Aborted.[/yellow]")
+            raise typer.Abort()
+
+    report = service.purge(purge_all=purge_all)
+
+    console.print("\n[bold]Cleanup Complete[/bold]\n")
+    _print_summary(report)
+
+
+@app.command(name="status")
+def status() -> None:
+    """Show current retention policy settings."""
+    from cleveragents.config.settings import get_settings
+
+    console = get_console()
+    settings = get_settings()
+
+    console.print("\n[bold]Cleanup Retention Policies[/bold]\n")
+    console.print(
+        f"  Sandbox max age:        {settings.cleanup_sandbox_max_age_hours} hours"
+    )
+    console.print(
+        f"  Checkpoint max/plan:    {settings.cleanup_checkpoint_max_per_plan}"
+    )
+    console.print(
+        f"  Session inactivity:     {settings.cleanup_session_inactivity_days} days"
+    )
+    console.print(
+        f"  Log retention:          {settings.cleanup_log_retention_days} days"
+    )
+    console.print(
+        f"  Backup retention:       {settings.cleanup_backup_retention_days} days"
+    )
+    console.print(f"  Schedule:               {settings.cleanup_schedule}")
+    console.print()
+
+
+def _print_summary(report: CleanupReport) -> None:
+    """Print the per-resource cleanup summary table."""
+
+    console = get_console()
+    data = report.as_dict()
+    mode = "Dry-run" if report.dry_run else "Purge"
+
+    console.print(f"  [dim]Mode:[/dim] {mode}")
+    for resource_type in ("sandboxes", "checkpoints", "sessions", "logs", "backups"):
+        info = data[resource_type]
+        removed_label = "to remove" if report.dry_run else "removed"
+        console.print(
+            f"  {resource_type.capitalize():15s}  "
+            f"scanned={info['scanned']}  "
+            f"{removed_label}={info['removed']}  "
+            f"skipped={info['skipped']}"
+        )

src/cleveragents/cli/main.py

@@ -79,6 +79,7 @@ def _register_subcommands() -> None:
         from cleveragents.cli.commands import (
             action,
             actor,
+            cleanup,
             context,
             plan,
             project,
@@ -128,6 +129,11 @@ def _register_subcommands() -> None:
         name="skill",
         help="Manage skills (reusable, namespaced tool collections)",
     )
+    app.add_typer(
+        cleanup.app,
+        name="cleanup",
+        help="Garbage collection and cleanup for stale resources",
+    )
 
     _subcommands_registered = True
 
@@ -513,6 +519,7 @@ def main(args: list[str] | None = None) -> int:
             "action",  # v3 plan lifecycle actions
             "resource",  # Resource registry management
             "skill",  # Skill management
+            "cleanup",  # Garbage collection and cleanup
             "auto-debug",  # Auto-debug commands
             "tell",  # Shortcut for plan tell
             "build",  # Shortcut for plan build

src/cleveragents/config/settings.py

@@ -148,6 +148,40 @@ class Settings(BaseSettings):
         validation_alias=AliasChoices("CLEVERAGENTS_STORAGE_BASE_PATH"),
     )
 
+    # Cleanup / retention policies (CONC3)
+    cleanup_sandbox_max_age_hours: int = Field(
+        default=48,
+        validation_alias=AliasChoices("CLEVERAGENTS_CLEANUP_SANDBOX_MAX_AGE_HOURS"),
+        description="Max age (hours) before stale sandboxes are eligible for cleanup.",
+    )
+    cleanup_checkpoint_max_per_plan: int = Field(
+        default=50,
+        validation_alias=AliasChoices("CLEVERAGENTS_CHECKPOINT_MAX"),
+        description=(
+            "Max checkpoints per plan; oldest pruned first (keep first + most recent)."
+        ),
+    )
+    cleanup_session_inactivity_days: int = Field(
+        default=30,
+        validation_alias=AliasChoices("CLEVERAGENTS_CLEANUP_SESSION_INACTIVITY_DAYS"),
+        description="Days of inactivity before a session is eligible for cleanup.",
+    )
+    cleanup_log_retention_days: int = Field(
+        default=30,
+        validation_alias=AliasChoices("CLEVERAGENTS_LOG_RETENTION_DAYS"),
+        description="Days to retain log files before automatic cleanup.",
+    )
+    cleanup_backup_retention_days: int = Field(
+        default=7,
+        validation_alias=AliasChoices("CLEVERAGENTS_BACKUP_RETENTION_DAYS"),
+        description="Days to retain backup snapshots before automatic cleanup.",
+    )
+    cleanup_schedule: str = Field(
+        default="manual",
+        validation_alias=AliasChoices("CLEVERAGENTS_CLEANUP_SCHEDULE"),
+        description="Cleanup schedule: 'manual' (MVP default) or 'auto'.",
+    )
+
     # Persistence
     database_url: str = Field(
         default="sqlite:///cleveragents.db",