"""``muse gc`` — garbage-collect unreachable objects. Content-addressed storage accumulates blobs that no live commit can reach. These orphaned objects are safe to delete. ``muse gc`` walks the full commit graph from every live branch and tag, marks every referenced object as reachable, then removes the rest. A grace period (``--grace-period``, default 30 s) prevents deleting objects written by a concurrent ``muse commit`` that has not yet created its commit record. Usage:: muse gc # remove unreachable objects (30 s grace) muse gc --full # also prune orphaned commits + snapshots muse gc --clear-symbol-cache # delete the symbol cache (forces re-parse) muse gc --dry-run # show what would be removed, touch nothing muse gc --verbose # print each removed object ID muse gc --grace-period 0 # no grace — useful in tests / CI muse gc --json # machine-readable output Exit codes:: 0 — success (even if nothing was collected) 1 — bad arguments or internal error """ from __future__ import annotations import argparse import json import logging import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.gc import _DEFAULT_GRACE_PERIOD_SECONDS, run_gc from muse.core.repo import require_repo from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Typed JSON schema # --------------------------------------------------------------------------- class _GcJson(TypedDict): """Machine-readable output of ``muse gc --json``.""" collected_count: int collected_bytes: int reachable_count: int commits_reachable: int commits_collected: int commits_collected_bytes: int snapshots_reachable: int snapshots_collected: int snapshots_collected_bytes: int elapsed_seconds: float grace_period_seconds: int dry_run: bool full: bool collected_ids: list[str] symbol_cache_cleared: bool symbol_cache_bytes: int # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fmt_bytes(n: int) -> str: """Human-readable byte count (B / KiB / MiB).""" if n < 1024: return f"{n} B" if n < 1024 * 1024: return f"{n / 1024:.1f} KiB" return f"{n / (1024 * 1024):.1f} MiB" # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``gc`` subcommand.""" parser = subparsers.add_parser( "gc", help="Remove unreachable objects from the Muse object store.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--dry-run", "-n", action="store_true", help="Show what would be removed without deleting anything.", ) parser.add_argument( "--verbose", "-v", action="store_true", help="Print each object ID that is (or would be) removed.", ) parser.add_argument( "--grace-period", type=int, default=_DEFAULT_GRACE_PERIOD_SECONDS, dest="grace_period", metavar="SECONDS", help=( "Skip objects written within the last SECONDS seconds even if " "unreachable (protects concurrent commits). Default: " f"{_DEFAULT_GRACE_PERIOD_SECONDS}." ), ) 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.add_argument( "--full", action="store_true", help=( "Also prune orphaned commit records (.muse/commits/) and snapshot " "manifests (.muse/snapshots/), mirroring git prune. Reachability " "is computed from live branch refs, not all files in the store." ), ) parser.add_argument( "--clear-symbol-cache", action="store_true", dest="clear_symbol_cache", help=( "Delete the symbol cache (.muse/symbol_cache.msgpack). " "Forces a full re-parse on the next ``muse code`` invocation. " "Use this when the cache is stale — e.g. after installing a new " "language grammar (tree-sitter-markdown, etc.)." ), ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command implementation # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Remove unreachable objects from the Muse object store. Muse stores every tracked file as a content-addressed blob. Blobs that are no longer referenced by any commit, snapshot, branch, or tag are *garbage*. This command identifies and removes them, reclaiming disk space. The reachability walk always completes before any deletion occurs. Use ``--dry-run`` to preview the impact. Use ``--grace-period 0`` in fully-controlled environments (tests, CI) where concurrent writes are impossible. Agents should pass ``--json`` to receive a machine-readable ``_GcJson`` result with counts for blobs, commits, and snapshots. Examples:: muse gc # safe cleanup with 30 s grace period muse gc --full # also prune orphaned commits + snapshots muse gc --dry-run # preview only muse gc --verbose # show every object ID muse gc --grace-period 0 # no grace (tests / CI) muse gc --json # machine-readable """ dry_run: bool = args.dry_run verbose: bool = args.verbose fmt: str = args.fmt grace_period: int = args.grace_period full: bool = args.full clear_symbol_cache: bool = args.clear_symbol_cache if grace_period < 0: print( f"❌ --grace-period must be ≥ 0 (got {grace_period}).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) repo_root = require_repo() # --- Symbol cache clear -------------------------------------------------- symbol_cache_cleared = False symbol_cache_bytes = 0 if clear_symbol_cache: cache_path = repo_root / ".muse" / "symbol_cache.msgpack" if cache_path.is_file(): symbol_cache_bytes = cache_path.stat().st_size if not dry_run: cache_path.unlink() symbol_cache_cleared = True result = run_gc( repo_root, dry_run=dry_run, grace_period_seconds=grace_period, full=full, ) if fmt == "json": payload = _GcJson( collected_count=result.collected_count, collected_bytes=result.collected_bytes, reachable_count=result.reachable_count, commits_reachable=result.commits_reachable, commits_collected=result.commits_collected, commits_collected_bytes=result.commits_collected_bytes, snapshots_reachable=result.snapshots_reachable, snapshots_collected=result.snapshots_collected, snapshots_collected_bytes=result.snapshots_collected_bytes, elapsed_seconds=result.elapsed_seconds, grace_period_seconds=result.grace_period_seconds, dry_run=result.dry_run, full=result.full, collected_ids=sorted(result.collected_ids), symbol_cache_cleared=symbol_cache_cleared, symbol_cache_bytes=symbol_cache_bytes, ) print(json.dumps(payload)) return prefix = "[dry-run] " if dry_run else "" action = "Would remove" if dry_run else "Removed" if verbose and result.collected_ids: print(f"{prefix}Unreachable objects:") for oid in sorted(result.collected_ids): print(f" {sanitize_display(oid)}") print( f"{prefix}{action} {result.collected_count} object(s) " f"({_fmt_bytes(result.collected_bytes)}) " f"in {result.elapsed_seconds:.3f}s " f"[{result.reachable_count} reachable]" ) if full: print( f"{prefix}{action} {result.commits_collected} commit(s) " f"({_fmt_bytes(result.commits_collected_bytes)}) " f"[{result.commits_reachable} reachable]" ) print( f"{prefix}{action} {result.snapshots_collected} snapshot(s) " f"({_fmt_bytes(result.snapshots_collected_bytes)}) " f"[{result.snapshots_reachable} reachable]" ) if symbol_cache_cleared: cache_action = "Would clear" if dry_run else "Cleared" print(f"{prefix}{cache_action} symbol cache ({_fmt_bytes(symbol_cache_bytes)})")