"""`hermes checkpoints` CLI subcommand. Gives users direct visibility and control over the filesystem checkpoint store at ``~/.hermes/checkpoints/``. Actions: hermes checkpoints # same as `status` hermes checkpoints status # total size, project count, breakdown hermes checkpoints list # per-project checkpoint counts + workdir hermes checkpoints prune [opts] # force a sweep (ignores the 24h marker) hermes checkpoints clear [-f] # nuke the entire base (asks first) hermes checkpoints clear-legacy # delete just the legacy-* archives Examples:: hermes checkpoints hermes checkpoints prune --retention-days 3 --max-size-mb 200 hermes checkpoints clear -f None of these require the agent to be running. Safe to call any time. """ from __future__ import annotations import argparse import time from datetime import datetime from pathlib import Path from typing import Any, Dict def _fmt_bytes(n: int) -> str: units = ("B", "KB", "MB", "GB", "TB") size = float(n or 0) for unit in units: if size < 1024 or unit == units[-1]: if unit == "B": return f"{int(size)} {unit}" return f"{size:.1f} {unit}" size /= 1024 return f"{size:.1f} TB" def _fmt_ts(ts: Any) -> str: try: return datetime.fromtimestamp(float(ts)).strftime("%Y-%m-%d %H:%M") except (TypeError, ValueError): return "—" def _fmt_age(ts: Any) -> str: try: age = time.time() - float(ts) except (TypeError, ValueError): return "—" if age < 0: return "now" if age < 60: return f"{int(age)}s ago" if age < 3600: return f"{int(age / 60)}m ago" if age < 86400: return f"{int(age / 3600)}h ago" return f"{int(age / 86400)}d ago" def cmd_status(args: argparse.Namespace) -> int: from tools.checkpoint_manager import store_status info = store_status() base = info["base"] print(f"Checkpoint base: {base}") print(f"Total size: {_fmt_bytes(info['total_size_bytes'])}") print(f" store/ {_fmt_bytes(info['store_size_bytes'])}") print(f" legacy-* {_fmt_bytes(info['legacy_size_bytes'])}") print(f"Projects: {info['project_count']}") projects = sorted( info["projects"], key=lambda p: (p.get("last_touch") or 0), reverse=True, ) if projects: print() print(f" {'WORKDIR':<60} {'COMMITS':>7} {'LAST TOUCH':>12} STATE") for p in projects[: args.limit if hasattr(args, "limit") and args.limit else 20]: wd = p.get("workdir") or "(unknown)" if len(wd) > 60: wd = "…" + wd[-59:] exists = p.get("exists") state = "live" if exists else "orphan" commits = p.get("commits", 0) last = _fmt_age(p.get("last_touch")) print(f" {wd:<60} {commits:>7} {last:>12} {state}") legacy = info.get("legacy_archives", []) if legacy: print() print(f"Legacy archives ({len(legacy)}):") for arch in sorted(legacy, key=lambda a: a.get("mtime", 0), reverse=True): print(f" {arch['name']:<40} {_fmt_bytes(arch['size_bytes']):>10}") print() print("Clear with: hermes checkpoints clear-legacy") return 0 def cmd_list(args: argparse.Namespace) -> int: # `list` is just a terser status — already covered. return cmd_status(args) def cmd_prune(args: argparse.Namespace) -> int: from tools.checkpoint_manager import prune_checkpoints retention_days = args.retention_days max_size_mb = args.max_size_mb print("Pruning checkpoint store…") print(f" retention_days: {retention_days}") print(f" delete_orphans: {not args.keep_orphans}") print(f" max_total_size_mb: {max_size_mb}") print() result = prune_checkpoints( retention_days=retention_days, delete_orphans=not args.keep_orphans, max_total_size_mb=max_size_mb, ) print(f"Scanned: {result['scanned']}") print(f"Deleted orphan: {result['deleted_orphan']}") print(f"Deleted stale: {result['deleted_stale']}") print(f"Errors: {result['errors']}") print(f"Bytes reclaimed: {_fmt_bytes(result['bytes_freed'])}") return 0 def _confirm(prompt: str) -> bool: try: resp = input(f"{prompt} [y/N]: ").strip().lower() except (EOFError, KeyboardInterrupt): print() return False return resp in ("y", "yes") def cmd_clear(args: argparse.Namespace) -> int: from tools.checkpoint_manager import CHECKPOINT_BASE, clear_all, store_status info = store_status() if info["total_size_bytes"] == 0 and not Path(CHECKPOINT_BASE).exists(): print("Nothing to clear — checkpoint base does not exist.") return 0 print(f"This will delete the ENTIRE checkpoint base at {info['base']}") print(f" size: {_fmt_bytes(info['total_size_bytes'])}") print(f" projects: {info['project_count']}") print(f" legacy dirs: {len(info.get('legacy_archives', []))}") print() print("All /rollback history for every working directory will be lost.") if not args.force and not _confirm("Proceed?"): print("Aborted.") return 1 result = clear_all() if result["deleted"]: print(f"Cleared. Reclaimed {_fmt_bytes(result['bytes_freed'])}.") return 0 print("Could not clear checkpoint base (see logs).") return 2 def cmd_clear_legacy(args: argparse.Namespace) -> int: from tools.checkpoint_manager import clear_legacy, store_status info = store_status() legacy = info.get("legacy_archives", []) if not legacy: print("No legacy archives to clear.") return 0 total = sum(a.get("size_bytes", 0) for a in legacy) print(f"Found {len(legacy)} legacy archive(s), total {_fmt_bytes(total)}:") for arch in legacy: print(f" {arch['name']:<40} {_fmt_bytes(arch['size_bytes']):>10}") print() print("Legacy archives hold pre-v2 per-project shadow repos, moved aside") print("during the single-store migration. Delete when you're confident") print("you don't need the old /rollback history.") if not args.force and not _confirm("Delete all legacy archives?"): print("Aborted.") return 1 result = clear_legacy() print(f"Deleted {result['deleted']} archive(s), reclaimed {_fmt_bytes(result['bytes_freed'])}.") return 0 def register_cli(parser: argparse.ArgumentParser) -> None: """Wire subcommands onto the ``hermes checkpoints`` parser.""" parser.set_defaults(func=cmd_status) # bare `hermes checkpoints` → status subs = parser.add_subparsers(dest="checkpoints_command", metavar="COMMAND") p_status = subs.add_parser( "status", help="Show total size, project count, and per-project breakdown", ) p_status.add_argument("--limit", type=int, default=20, help="Max projects to list (default 20)") p_status.set_defaults(func=cmd_status) p_list = subs.add_parser( "list", help="Alias for 'status'", ) p_list.add_argument("--limit", type=int, default=20) p_list.set_defaults(func=cmd_list) p_prune = subs.add_parser( "prune", help="Delete orphan/stale checkpoints and GC the store", ) p_prune.add_argument("--retention-days", type=int, default=7, help="Drop projects whose last_touch is older than N days (default 7)") p_prune.add_argument("--max-size-mb", type=int, default=500, help="After orphan/stale prune, drop oldest commits " "per project until total size <= this (default 500)") p_prune.add_argument("--keep-orphans", action="store_true", help="Skip deleting projects whose workdir no longer exists") p_prune.set_defaults(func=cmd_prune) p_clear = subs.add_parser( "clear", help="Delete the entire checkpoint base (all /rollback history)", ) p_clear.add_argument("-f", "--force", action="store_true", help="Skip confirmation prompt") p_clear.set_defaults(func=cmd_clear) p_legacy = subs.add_parser( "clear-legacy", help="Delete only the legacy-/ archives from v1 migration", ) p_legacy.add_argument("-f", "--force", action="store_true", help="Skip confirmation prompt") p_legacy.set_defaults(func=cmd_clear_legacy)