diff.py
python
| 1 | """muse diff — compare working tree against HEAD, or compare two commits.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import logging |
| 7 | import pathlib |
| 8 | |
| 9 | import typer |
| 10 | |
| 11 | from muse.core.errors import ExitCode |
| 12 | from muse.core.repo import require_repo |
| 13 | from muse.core.store import get_commit_snapshot_manifest, get_head_snapshot_manifest, resolve_commit_ref |
| 14 | from muse.core.validation import sanitize_display |
| 15 | from muse.domain import DomainOp, SnapshotManifest |
| 16 | from muse.plugins.registry import read_domain, resolve_plugin |
| 17 | |
| 18 | logger = logging.getLogger(__name__) |
| 19 | |
| 20 | app = typer.Typer() |
| 21 | |
| 22 | |
| 23 | def _read_branch(root: pathlib.Path) -> str: |
| 24 | head_ref = (root / ".muse" / "HEAD").read_text().strip() |
| 25 | return head_ref.removeprefix("refs/heads/").strip() |
| 26 | |
| 27 | |
| 28 | def _read_repo_id(root: pathlib.Path) -> str: |
| 29 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 30 | |
| 31 | |
| 32 | def _print_structured_delta(ops: list[DomainOp]) -> int: |
| 33 | """Print a structured delta op-by-op. Returns the number of ops printed. |
| 34 | |
| 35 | Each branch checks ``op["op"]`` directly so mypy can narrow the |
| 36 | TypedDict union to the specific subtype before accessing its fields. |
| 37 | """ |
| 38 | for op in ops: |
| 39 | if op["op"] == "insert": |
| 40 | typer.echo(f"A {op['address']}") |
| 41 | elif op["op"] == "delete": |
| 42 | typer.echo(f"D {op['address']}") |
| 43 | elif op["op"] == "replace": |
| 44 | typer.echo(f"M {op['address']}") |
| 45 | elif op["op"] == "move": |
| 46 | typer.echo( |
| 47 | f"R {op['address']} ({op['from_position']} → {op['to_position']})" |
| 48 | ) |
| 49 | elif op["op"] == "patch": |
| 50 | typer.echo(f"M {op['address']}") |
| 51 | if op["child_summary"]: |
| 52 | typer.echo(f" └─ {op['child_summary']}") |
| 53 | return len(ops) |
| 54 | |
| 55 | |
| 56 | @app.callback(invoke_without_command=True) |
| 57 | def diff( |
| 58 | ctx: typer.Context, |
| 59 | commit_a: str | None = typer.Argument(None, help="Base commit ID (default: HEAD)."), |
| 60 | commit_b: str | None = typer.Argument(None, help="Target commit ID (default: working tree)."), |
| 61 | stat: bool = typer.Option(False, "--stat", help="Show summary statistics only."), |
| 62 | ) -> None: |
| 63 | """Compare working tree against HEAD, or compare two commits.""" |
| 64 | root = require_repo() |
| 65 | repo_id = _read_repo_id(root) |
| 66 | branch = _read_branch(root) |
| 67 | domain = read_domain(root) |
| 68 | plugin = resolve_plugin(root) |
| 69 | |
| 70 | def _resolve_manifest(ref: str) -> dict[str, str]: |
| 71 | """Resolve a ref (branch, short SHA, full SHA) to its snapshot manifest.""" |
| 72 | resolved = resolve_commit_ref(root, repo_id, branch, ref) |
| 73 | if resolved is None: |
| 74 | typer.echo(f"⚠️ Commit '{sanitize_display(ref)}' not found.") |
| 75 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 76 | return get_commit_snapshot_manifest(root, resolved.commit_id) or {} |
| 77 | |
| 78 | if commit_a is None: |
| 79 | base_snap = SnapshotManifest( |
| 80 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 81 | domain=domain, |
| 82 | ) |
| 83 | target_snap = plugin.snapshot(root) |
| 84 | elif commit_b is None: |
| 85 | # Single ref provided: diff HEAD vs that ref's snapshot. |
| 86 | base_snap = SnapshotManifest( |
| 87 | files=get_head_snapshot_manifest(root, repo_id, branch) or {}, |
| 88 | domain=domain, |
| 89 | ) |
| 90 | target_snap = SnapshotManifest( |
| 91 | files=_resolve_manifest(commit_a), |
| 92 | domain=domain, |
| 93 | ) |
| 94 | else: |
| 95 | base_snap = SnapshotManifest( |
| 96 | files=_resolve_manifest(commit_a), |
| 97 | domain=domain, |
| 98 | ) |
| 99 | target_snap = SnapshotManifest( |
| 100 | files=_resolve_manifest(commit_b), |
| 101 | domain=domain, |
| 102 | ) |
| 103 | |
| 104 | delta = plugin.diff(base_snap, target_snap, repo_root=root) |
| 105 | |
| 106 | if stat: |
| 107 | typer.echo(delta["summary"] if delta["ops"] else "No differences.") |
| 108 | return |
| 109 | |
| 110 | changed = _print_structured_delta(delta["ops"]) |
| 111 | |
| 112 | if changed == 0: |
| 113 | typer.echo("No differences.") |
| 114 | else: |
| 115 | typer.echo(f"\n{delta['summary']}") |