"""muse plumbing snapshot-diff — diff two snapshot manifests. Compares two Muse snapshots and categorises every path change into one of three buckets: added, modified, or deleted. Accepts snapshot IDs directly or commit IDs / branch names (in which case the commit's snapshot is resolved first). Single-pair mode (default) -------------------------- Output (JSON, default):: { "snapshot_a": "", "snapshot_b": "", "added": [{"path": "src/foo.py", "object_id": ""}], "modified": [{"path": "src/bar.py", "object_id_a": "", "object_id_b": ""}], "deleted": [{"path": "src/old.py", "object_id": ""}], "total_changes": 3 } Text output (``--format text``):: A src/foo.py M src/bar.py D src/old.py Text output with ``--raw`` (includes object IDs):: A src/foo.py M src/bar.py D src/old.py Batch mode (``--stdin``) ------------------------ Reads ref pairs from stdin (one `` `` per line) and processes each pair. Analogous to ``git diff-tree --stdin``. JSON mode emits one JSON object per pair (newline-delimited):: {"snapshot_a": "...", "snapshot_b": "...", "added": [...], ...} {"snapshot_a": "...", "snapshot_b": "...", "added": [...], ...} Text mode emits the text diff for each pair separated by a blank line. Invalid or unresolvable pairs emit an error object/line and continue. Plumbing contract ----------------- - Exit 0: diff computed (even when zero changes). - Exit 1: snapshot or commit ID cannot be resolved; bad ``--format`` value. - Exit 3: I/O error reading snapshot records. - Batch mode (``--stdin``) always exits 0; individual errors are reported inline. """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import ( get_head_commit_id, read_commit, read_current_branch, read_snapshot, ) from muse.core.validation import sanitize_display, validate_object_id logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("json", "text") # Column widths for --raw text output alignment. _OID_WIDTH = 64 class _AddedEntry(TypedDict): path: str object_id: str class _ModifiedEntry(TypedDict): path: str object_id_a: str object_id_b: str class _DeletedEntry(TypedDict): path: str object_id: str class _DiffResult(TypedDict): snapshot_a: str snapshot_b: str added: list[_AddedEntry] modified: list[_ModifiedEntry] deleted: list[_DeletedEntry] total_changes: int def _resolve_to_snapshot_id(root: pathlib.Path, ref: str) -> str | None: """Return the snapshot ID for *ref*. *ref* may be a 64-char snapshot ID, a 64-char commit ID, a branch name, or ``HEAD``. Returns ``None`` when the ref cannot be resolved. """ if ref.upper() == "HEAD": branch = read_current_branch(root) commit_id = get_head_commit_id(root, branch) if commit_id is None: return None commit = read_commit(root, commit_id) return commit.snapshot_id if commit else None commit_id = get_head_commit_id(root, ref) if commit_id is not None: commit = read_commit(root, commit_id) return commit.snapshot_id if commit else None try: validate_object_id(ref) except ValueError: return None snap = read_snapshot(root, ref) if snap is not None: return snap.snapshot_id commit = read_commit(root, ref) if commit is not None: return commit.snapshot_id return None def _compute_diff(root: pathlib.Path, ref_a: str, ref_b: str) -> "_DiffResult | dict[str, str]": """Compute the diff between *ref_a* and *ref_b*. Returns a ``_DiffResult`` on success or a ``{"error": "..."}`` dict when either ref cannot be resolved or a snapshot cannot be read. """ snap_id_a = _resolve_to_snapshot_id(root, ref_a) if snap_id_a is None: return {"error": f"Cannot resolve ref: {ref_a!r}"} snap_id_b = _resolve_to_snapshot_id(root, ref_b) if snap_id_b is None: return {"error": f"Cannot resolve ref: {ref_b!r}"} try: snap_a = read_snapshot(root, snap_id_a) snap_b = read_snapshot(root, snap_id_b) except Exception as exc: return {"error": str(exc)} if snap_a is None: return {"error": f"Snapshot not found: {snap_id_a}"} if snap_b is None: return {"error": f"Snapshot not found: {snap_id_b}"} manifest_a = snap_a.manifest manifest_b = snap_b.manifest keys_a = set(manifest_a) keys_b = set(manifest_b) added: list[_AddedEntry] = sorted( [{"path": p, "object_id": manifest_b[p]} for p in (keys_b - keys_a)], key=lambda e: e["path"], ) deleted: list[_DeletedEntry] = sorted( [{"path": p, "object_id": manifest_a[p]} for p in (keys_a - keys_b)], key=lambda e: e["path"], ) modified: list[_ModifiedEntry] = sorted( [ {"path": p, "object_id_a": manifest_a[p], "object_id_b": manifest_b[p]} for p in (keys_a & keys_b) if manifest_a[p] != manifest_b[p] ], key=lambda e: e["path"], ) return _DiffResult( snapshot_a=snap_id_a, snapshot_b=snap_id_b, added=added, modified=modified, deleted=deleted, total_changes=len(added) + len(modified) + len(deleted), ) def _emit_text(result: _DiffResult, raw: bool, stat: bool) -> None: """Print text-format diff output for *result*.""" if raw: for entry in result["added"]: oid = entry["object_id"] print(f"A {oid} {sanitize_display(entry['path'])}") for entry in result["modified"]: oid_a = entry["object_id_a"] oid_b = entry["object_id_b"] print(f"M {oid_a} {oid_b} {sanitize_display(entry['path'])}") for entry in result["deleted"]: oid = entry["object_id"] print(f"D {oid} {sanitize_display(entry['path'])}") else: for entry in result["added"]: print(f"A {sanitize_display(entry['path'])}") for entry in result["modified"]: print(f"M {sanitize_display(entry['path'])}") for entry in result["deleted"]: print(f"D {sanitize_display(entry['path'])}") if stat: na = len(result["added"]) nm = len(result["modified"]) nd = len(result["deleted"]) print(f"\n{na} added, {nm} modified, {nd} deleted") def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the snapshot-diff subcommand.""" parser = subparsers.add_parser( "snapshot-diff", help="Diff two snapshot manifests: added, modified, deleted paths.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "ref_a", nargs="?", default=None, help=( "First snapshot ID, commit ID, branch name, or HEAD. " "Required in single-pair mode; omit when using --stdin." ), ) parser.add_argument( "ref_b", nargs="?", default=None, help=( "Second snapshot ID, commit ID, branch name, or HEAD. " "Required in single-pair mode; omit when using --stdin." ), ) parser.add_argument( "--stdin", action="store_true", dest="from_stdin", help=( "Batch mode: read ' ' pairs from stdin (one per line) " "and process each. JSON mode emits one object per line; " "text mode separates pairs with a blank line." ), ) parser.add_argument( "--raw", action="store_true", dest="raw", help=( "Include object IDs in text-format output. " "Has no effect on JSON output (which always includes OIDs). " "Format: 'A ', 'M ', 'D '." ), ) parser.add_argument( "--format", "-f", dest="fmt", default="json", metavar="FORMAT", help="Output format: json or text. (default: json)", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.add_argument( "--stat", "-s", action="store_true", help="Append a summary line: N added, M modified, D deleted (text mode only).", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Diff two snapshots and report added, modified, and deleted paths.""" fmt: str = args.fmt ref_a: str | None = args.ref_a ref_b: str | None = args.ref_b from_stdin: bool = args.from_stdin raw: bool = args.raw stat: bool = args.stat if fmt not in _FORMAT_CHOICES: print( json.dumps({"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # ── Batch (--stdin) mode ────────────────────────────────────────────────── if from_stdin: first = True for raw_line in sys.stdin: line = raw_line.strip() if not line or line.startswith("#"): continue parts = line.split() if len(parts) < 2: if fmt == "json": print(json.dumps({"error": f"Expected ' ', got: {line!r}"})) else: print(f"error: expected ' ', got: {sanitize_display(line)}") if not first: print() first = False continue pair_a, pair_b = parts[0], parts[1] result = _compute_diff(root, pair_a, pair_b) if "error" in result: if fmt == "json": print(json.dumps(result)) else: if not first: print() print(f"error: {sanitize_display(result['error'])}") # type: ignore[index] elif fmt == "json": print(json.dumps(result)) else: if not first: print() _emit_text(result, raw=raw, stat=stat) # type: ignore[arg-type] first = False return # ── Single-pair mode ────────────────────────────────────────────────────── if ref_a is None or ref_b is None: print( json.dumps({"error": "ref_a and ref_b are required in single-pair mode " "(or use --stdin for batch processing)"}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) result = _compute_diff(root, ref_a, ref_b) if "error" in result: print(json.dumps(result), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if fmt == "text": _emit_text(result, raw=raw, stat=stat) # type: ignore[arg-type] return print(json.dumps(result))