"""``muse coord gc`` — collect stale coordination records. The coordination directory (``.muse/coordination/``) accumulates records over time: expired reservations, corresponding release tombstones, heartbeat files, and old intents. ``muse coord gc`` removes records that are no longer needed so the coordination layer stays lean and fast. What gets collected ------------------- * **Expired reservations** — TTL exhausted (after heartbeat extension), past the grace period. * **Released reservations** — release tombstone present and older than the grace period. * **Corresponding release tombstones** — removed with their reservation. * **Corresponding heartbeat files** — removed with their reservation. * **Orphaned releases and heartbeats** — the reservation file is gone but the tombstone or heartbeat still exists (e.g. from a prior partial GC). * **Old intents** (opt-in via ``--include-intents``) — older than ``--max-intent-age SECONDS`` (default: 7 days). What is never collected ----------------------- * Active reservations (not released, effective expiry in the future). * Records within the grace period. Usage:: muse coord gc # dry-run to preview muse coord gc --execute # actually delete files muse coord gc --execute --grace-period 60 # shorter grace period muse coord gc --execute --include-intents # also purge old intents muse coord gc --execute --verbose # per-file detail muse coord gc --format json # machine-readable output muse coord gc --json # shorthand JSON output schema:: { "dry_run": bool, "grace_period_seconds": int, "include_intents": bool, "max_intent_age_seconds": int, "reservations_removed": int, "reservations_removed_bytes": int, "releases_removed": int, "releases_removed_bytes": int, "heartbeats_removed": int, "heartbeats_removed_bytes": int, "intents_removed": int, "intents_removed_bytes": int, "total_removed": int, "total_removed_bytes": int, "removed_ids": [str, ...], "elapsed_seconds": float } Exit codes:: 0 — success (zero removed is still success) 1 — bad arguments (--grace-period < 0 or --max-intent-age <= 0) Flags: ``--execute`` Actually delete files. Without this flag the command runs in dry-run mode and only reports what would be removed. Always preview first. ``--grace-period SECONDS`` Records expired or released within the last N seconds are protected (default: 300 = 5 min). Raise this on high-churn swarms. ``--include-intents`` Also purge intents older than ``--max-intent-age``. Intents are permanent audit records by default. ``--max-intent-age SECONDS`` Age threshold for intent cleanup (default: 604 800 = 7 days). ``--verbose`` / ``-v`` Print the UUID of every removed reservation. ``--json`` / ``--format json`` Emit result as compact JSON on stdout. """ from __future__ import annotations import argparse import json import sys from muse.core.coordination import run_coord_gc from muse.core.errors import ExitCode from muse.core.repo import require_repo # ── Helpers ─────────────────────────────────────────────────────────────────── def _fmt_bytes(n: int) -> str: """Return a human-readable byte count using binary units. Scales through B → KiB → MiB → GiB → TiB, stopping at the first unit where the value is below 1024. Examples -------- ``0`` → ``"0 B"`` ``1023`` → ``"1023 B"`` ``1024`` → ``"1.0 KiB"`` ``1_048_576`` → ``"1.0 MiB"`` ``1_073_741_824`` → ``"1.0 GiB"`` ``1_099_511_627_776`` → ``"1.0 TiB"`` """ if n < 1024: return f"{n} B" for unit in ("KiB", "MiB", "GiB", "TiB"): n /= 1024.0 if n < 1024: return f"{n:.1f} {unit}" return f"{n:.1f} TiB" # ── CLI registration ────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``gc`` subcommand on *subparsers* (under ``muse coord``). Wires all flags with their defaults, choices, and help text so that ``--help`` output is accurate. Sets ``func`` to :func:`run`. """ parser = subparsers.add_parser( "gc", help="Collect stale coordination records from .muse/coordination/.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--execute", action="store_true", dest="execute", help=( "Actually delete files. Without this flag the command runs in " "dry-run mode and only reports what would be removed." ), ) parser.add_argument( "--grace-period", type=int, default=300, dest="grace_period_seconds", metavar="SECONDS", help=( "Records expired or released within the last N seconds are skipped " "(default: 300 = 5 min). Increase this on busy swarms to avoid " "racing with agents that are still reading coordination state." ), ) parser.add_argument( "--include-intents", action="store_true", dest="include_intents", help=( "Also purge intents older than --max-intent-age. Intents are " "permanent audit records by default — only enable if you are " "confident you no longer need them for post-mortem analysis." ), ) parser.add_argument( "--max-intent-age", type=int, default=604800, dest="max_intent_age_seconds", metavar="SECONDS", help=( "Age threshold for intent cleanup when --include-intents is set " "(default: 604800 = 7 days)." ), ) parser.add_argument( "--verbose", "-v", action="store_true", help="Print the ID of every removed reservation.", ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) # ── Command implementation ──────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Remove stale coordination records from ``.muse/coordination/``. Execution order --------------- 1. **Validate inputs** — ``--grace-period`` must be >= 0; ``--max-intent-age`` must be > 0. Any failure exits :attr:`~muse.core.errors.ExitCode.USER_ERROR` (1) with a message to *stderr* before any file I/O. 2. **Run GC** — delegates to :func:`~muse.core.coordination.run_coord_gc`, which loads all coordination state in a single directory scan per subdirectory and deletes in O(n) passes. 3. **Emit output** — compact JSON to *stdout* when ``--format json``, or human-readable text otherwise. Dry-run mode (default) ---------------------- Without ``--execute`` the command reports what *would* be removed but deletes nothing. This is the safe default — always preview before executing. Grace period ------------ Records expired or released within the last *grace_period_seconds* are skipped to protect against races with agents that are still reading coordination state. The default 300 s (5 min) is appropriate for most swarms. On high-churn swarms with sub-minute TTLs, lower this carefully. Intent cleanup (opt-in) ----------------------- Intents are permanent audit records by default — they are never collected without ``--include-intents``. When enabled, intents older than *max_intent_age_seconds* are removed. Security -------- GC only scans ``.muse/coordination/`` and its known subdirectories. No user-supplied string is used to construct file paths, eliminating any directory traversal risk. Performance ----------- All coordination state is loaded with a single directory scan per subdirectory. Deletion is O(n) in the number of collectable records. The grace-period filter is applied in memory before any deletion. Args: args: Parsed ``argparse.Namespace`` with attributes ``execute``, ``grace_period_seconds``, ``include_intents``, ``max_intent_age_seconds``, ``verbose``, and ``fmt``. Exit codes: 0 — success (zero removed is also success). 1 — bad arguments. """ execute: bool = args.execute grace_period_seconds: int = args.grace_period_seconds include_intents: bool = args.include_intents max_intent_age_seconds: int = args.max_intent_age_seconds verbose: bool = args.verbose fmt: str = args.fmt as_json = fmt == "json" # ── Input validation (before any file I/O) ──────────────────────────────── if grace_period_seconds < 0: msg = f"--grace-period must be >= 0, got {grace_period_seconds}" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if max_intent_age_seconds <= 0: msg = f"--max-intent-age must be > 0, got {max_intent_age_seconds}" if as_json: print(json.dumps({"error": msg, "status": "bad_args"})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() result = run_coord_gc( root, dry_run=not execute, grace_period_seconds=grace_period_seconds, include_intents=include_intents, max_intent_age_seconds=max_intent_age_seconds, ) if as_json: print(json.dumps({ "dry_run": result.dry_run, "grace_period_seconds": result.grace_period_seconds, "include_intents": result.include_intents, "max_intent_age_seconds": max_intent_age_seconds, "reservations_removed": result.reservations_removed, "reservations_removed_bytes": result.reservations_removed_bytes, "releases_removed": result.releases_removed, "releases_removed_bytes": result.releases_removed_bytes, "heartbeats_removed": result.heartbeats_removed, "heartbeats_removed_bytes": result.heartbeats_removed_bytes, "intents_removed": result.intents_removed, "intents_removed_bytes": result.intents_removed_bytes, "total_removed": result.total_removed, "total_removed_bytes": result.total_removed_bytes, "removed_ids": result.removed_ids, "elapsed_seconds": result.elapsed_seconds, })) return # ── Text output ─────────────────────────────────────────────────────────── mode = "DRY RUN — no files deleted" if result.dry_run else "GC complete" print(f"\nCoordination GC — {mode}") print("─" * 62) if result.total_removed == 0: print("\n Nothing to collect.") else: lines = [ ("Reservations", result.reservations_removed, result.reservations_removed_bytes), ("Releases", result.releases_removed, result.releases_removed_bytes), ("Heartbeats", result.heartbeats_removed, result.heartbeats_removed_bytes), ] if include_intents: lines.append(("Intents", result.intents_removed, result.intents_removed_bytes)) for label, count, nbytes in lines: if count: verb = "would remove" if result.dry_run else "removed" print(f" {label:<14} {verb} {count:>4} ({_fmt_bytes(nbytes)})") print( f"\n Total: {result.total_removed} record(s)" f" ({_fmt_bytes(result.total_removed_bytes)})" ) if verbose and result.removed_ids: print("\n Removed reservation IDs:") for rid in result.removed_ids: print(f" {rid}") print( f"\n grace-period={grace_period_seconds}s" f" ({result.elapsed_seconds:.3f}s)" )