"""``muse prune [--expire ]`` — surgical removal of unreachable objects. Removes objects from the local store that are not reachable from any live ref (branch tip, tag, shelf entry, or snapshot on disk) without the full overhead of ``muse gc``. ``muse prune`` is the scalpel; ``muse gc`` is the full surgery. Use prune after an aborted commit, a reset, or an abandoned experiment leaves orphaned blob objects in the store. Reachability ------------ An object is **reachable** if it appears as a value in any snapshot manifest stored in ``.muse/objects/sha256/``. This is a conservative definition: even orphaned snapshots (not referenced by any live commit) keep their objects alive. This mirrors the conservative walk in ``muse gc`` and prevents data loss when refs are being rewritten. Shelf entries in ``.muse/shelf.json`` are also walked — objects written during a ``muse shelf save`` push are always reachable until explicitly dropped. Safety guarantees ----------------- - ``--dry-run`` (default for agent use) never deletes anything; always preview first. - ``--expire `` keeps objects newer than the threshold — a safety net for objects that are mid-flight (written but not yet referenced by a snapshot). - Refuses to prune if a merge is in progress (conflict-resolution objects must not be deleted mid-merge). - Only deletes files under ``.muse/objects/``; never touches the working tree, commits, snapshots, or any other repo state. JSON schema (``--json``) ------------------------ Dry-run:: { "pruned": 42, "bytes_freed": 18432, "dry_run": true, "reachable_count": 100, "candidates": [{"object_id": "sha256:...", "size": 1024}], "duration_ms": 1.234, "exit_code": 0 } Live:: { "pruned": 42, "bytes_freed": 18432, "dry_run": false, "reachable_count": 100, "duration_ms": 1.234, "exit_code": 0 } Exit codes:: 0 — success (including zero pruned) 1 — user error (merge in progress) 2 — not a Muse repository 3 — I/O error Examples:: muse prune --dry-run # preview without deleting muse prune --dry-run --json # machine-readable preview muse prune # prune unreachable objects muse prune --expire 3600 # only prune objects older than 1 hour muse prune --json # prune + JSON result """ import argparse import json as _json import logging import os import pathlib import sys import time from typing import TypedDict from muse.core.types import long_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.gc import _collect_reachable_objects from muse.core.object_store import iter_stored_objects, objects_dir from muse.core.repo import require_repo from muse.core.timing import start_timer logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Wire-format TypedDicts # --------------------------------------------------------------------------- class _PruneCandidateEntry(TypedDict): object_id: str size: int class _PruneResultBase(EnvelopeJson): """Common fields for both dry-run and live ``muse prune --json`` output.""" pruned: int bytes_freed: int dry_run: bool reachable_count: int class _PruneResultJson(_PruneResultBase, total=False): """Full prune result envelope; ``candidates`` present only in dry-run.""" candidates: list[_PruneCandidateEntry] # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _collect_all_reachable_ids(root: pathlib.Path) -> set[str]: """Collect all object IDs reachable from any snapshot or shelf entry. Delegates to the GC module's conservative walk: every snapshot on disk (not just those referenced by live commits) is included to prevent accidental data loss during ref rewrites. Args: root: Absolute repo root. Returns: Set of ``sha256:``-prefixed object IDs reachable from any snapshot or shelf entry. """ return _collect_reachable_objects(root) def _find_prune_candidates( root: pathlib.Path, reachable: set[str], expire_before: float | None, ) -> list[dict]: """Find unreachable objects that are candidates for pruning. Args: root: Absolute repo root. reachable: Set of reachable object IDs (will not be deleted). expire_before: Only include objects whose mtime is older than this Unix timestamp. ``None`` means include all unreachable objects regardless of age. Returns: List of ``{"object_id": str, "size": int, "path": str}`` dicts for each candidate object, sorted by object_id. """ candidates: list[_PruneCandidateEntry] = [] for oid, obj_file in iter_stored_objects(root): if oid in reachable: continue try: st = obj_file.stat() if expire_before is not None and st.st_mtime >= expire_before: # Object is too recent — skip it. continue candidates.append({ "object_id": oid, "size": st.st_size, "path": str(obj_file), }) except OSError: pass candidates.sort(key=lambda c: c["object_id"]) return candidates # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``muse prune`` subcommand.""" parser = subparsers.add_parser( "prune", help="Remove unreachable objects from the local store.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "-n", "--dry-run", action="store_true", dest="dry_run", help=( "List what would be pruned without deleting anything. " "Recommended for agent use; always preview before pruning." ), ) parser.add_argument( "--expire", type=float, default=None, metavar="SECONDS", help=( "Only prune objects older than SECONDS ago. " "Safety net for in-flight objects that are mid-write. " "E.g. --expire 3600 keeps anything written in the last hour." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON on stdout.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Prune unreachable objects from the local store. Exit codes:: 0 — success (including zero pruned) 1 — merge in progress; prune refused 2 — not inside a Muse repo 3 — I/O error """ elapsed = start_timer() dry_run: bool = args.dry_run expire_secs: float | None = args.expire json_out: bool = args.json_out root = require_repo() # ── Refuse if a merge is in progress ───────────────────────────────────── try: from muse.core.merge_engine import read_merge_state if read_merge_state(root) is not None: print( "❌ Merge in progress — prune refused. " "Resolve or abort the merge first.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) except ImportError: pass # merge_engine optional dep # ── Compute expiry threshold ────────────────────────────────────────────── expire_before: float | None = None if expire_secs is not None: expire_before = time.time() - expire_secs # ── Collect reachable IDs ───────────────────────────────────────────────── reachable = _collect_all_reachable_ids(root) # ── Find candidates ─────────────────────────────────────────────────────── candidates = _find_prune_candidates(root, reachable, expire_before) # ── Dry-run: report without deleting ───────────────────────────────────── if dry_run: if json_out: print(_json.dumps(_PruneResultJson( **make_envelope(elapsed), pruned=len(candidates), bytes_freed=sum(c["size"] for c in candidates), dry_run=True, reachable_count=len(reachable), candidates=[ {"object_id": c["object_id"], "size": c["size"]} for c in candidates ], ))) else: for c in candidates: print(f"[dry-run] would prune: {c['object_id']}") print( f"[dry-run] {len(candidates)} object(s) " f"({sum(c['size'] for c in candidates)} bytes) would be pruned." ) return # ── Delete candidates ───────────────────────────────────────────────────── pruned = 0 bytes_freed = 0 for candidate in candidates: try: os.unlink(candidate["path"]) pruned += 1 bytes_freed += candidate["size"] except FileNotFoundError: # Concurrent prune — already gone, that's fine. pass except OSError as exc: logger.warning("⚠️ Could not prune %s: %s", candidate["object_id"], exc) # ── Output ──────────────────────────────────────────────────────────────── if json_out: print(_json.dumps(_PruneResultJson( **make_envelope(elapsed), pruned=pruned, bytes_freed=bytes_freed, dry_run=False, reachable_count=len(reachable), ))) else: print(f"Pruned {pruned} object(s), freed {bytes_freed} bytes.")