"""muse rerere — reuse recorded resolutions for merge conflicts. Records how you resolve a merge conflict and replays that resolution automatically the next time the same conflict appears. The conflict is identified by a content fingerprint (SHA-256 of the sorted ours/theirs object ID pair) so the same resolution applies regardless of surrounding context changes. Subcommands ----------- ``muse rerere`` Apply cached resolutions to current conflicts. ``muse rerere record`` Record preimages for current conflicts (no replay). ``muse rerere status`` List cached resolutions and match against current conflicts. ``muse rerere forget`` Remove the cached resolution for specific paths or fingerprints. ``muse rerere clear`` Remove all cached resolutions. ``muse rerere gc`` Garbage-collect preimage-only entries older than N days. Automatic integration --------------------- ``muse merge`` calls rerere automatically: - **Before writing conflict markers**: tries to auto-apply any cached resolutions, resolving matching conflicts without user intervention. - **Records preimages** for any remaining conflicts so that the user's resolution can be saved when they run ``muse commit``. ``muse commit`` (during a conflicted merge) calls rerere to save the user's chosen resolution for replay on future identical conflicts. Security model -------------- All user-controlled values (paths, fingerprints, domain names) are validated and passed through ``sanitize_display()`` before terminal output to prevent ANSI injection. Fingerprints are validated as 64-char hex strings. Conflict paths are validated not to escape the repository root. Error messages go to **stderr**; **stdout** carries only data. Agent UX -------- Pass ``--json`` to any subcommand for machine-readable output. JSON schemas are defined by TypedDicts below — all fields are always present. JSON schemas ~~~~~~~~~~~~ ``muse rerere --json``:: {"dry_run": bool, "auto_resolved": [str], "remaining": [str]} ``muse rerere record --json``:: {"recorded": [str], "skipped": [str]} ``muse rerere status --json``:: { "total": int, "records": [ { "fingerprint": str, // 64-char hex "path": str, "domain": str, "has_resolution": bool, "resolution_id": str | null, "recorded_at": str, // ISO 8601 "matches_current_conflict": bool } ] } ``muse rerere forget --json``:: {"forgotten": [str], "not_found": [str]} ``muse rerere clear --json`` / ``muse rerere gc --json``:: {"removed": int} Exit codes ---------- - 0 — success - 1 — invalid arguments or missing MERGE_STATE - 2 — not inside a Muse repository """ from __future__ import annotations import argparse import json import logging import pathlib import sys from typing import TYPE_CHECKING, TypedDict from muse.core.errors import ExitCode from muse.core.merge_engine import read_merge_state from muse.core.repo import require_repo from muse.core.rerere import ( RerereRecord, apply_cached, clear_all, compute_fingerprint, forget_record, gc_stale, list_records, record_preimage, rr_cache_dir, ) from muse.core.store import read_commit, read_snapshot from muse.core.validation import clamp_int, sanitize_display from muse.plugins.registry import read_domain, resolve_plugin from muse.core._types import Manifest if TYPE_CHECKING: pass logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("text", "json") # --------------------------------------------------------------------------- # JSON TypedDicts — stable machine-readable output schemas # --------------------------------------------------------------------------- class _RerereApplyJson(TypedDict): """JSON output from ``muse rerere`` (default apply action).""" dry_run: bool auto_resolved: list[str] remaining: list[str] class _RerereRecordJson(TypedDict): """JSON output from ``muse rerere record``.""" recorded: list[str] skipped: list[str] class _RerereStatusRecordJson(TypedDict): """One record in ``muse rerere status --json`` output.""" fingerprint: str path: str domain: str has_resolution: bool resolution_id: str | None recorded_at: str matches_current_conflict: bool class _RerereStatusJson(TypedDict): """JSON output from ``muse rerere status``.""" total: int records: list[_RerereStatusRecordJson] class _RerereForgetJson(TypedDict): """JSON output from ``muse rerere forget``.""" forgotten: list[str] not_found: list[str] class _RerereScalarJson(TypedDict): """JSON output from ``muse rerere clear`` and ``muse rerere gc``.""" removed: int # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _load_conflict_manifests( root: pathlib.Path, ) -> tuple[dict[str, str], dict[str, str]] | None: """Return ``(ours_manifest, theirs_manifest)`` from the active MERGE_STATE. Returns ``None`` when MERGE_STATE is absent or incomplete. """ state = read_merge_state(root) if state is None or not state.ours_commit or not state.theirs_commit: return None def _manifest(commit_id: str) -> Manifest: commit = read_commit(root, commit_id) if commit is None: return {} snap = read_snapshot(root, commit.snapshot_id) return snap.manifest if snap else {} return _manifest(state.ours_commit), _manifest(state.theirs_commit) def _fmt_record(rec: RerereRecord) -> str: """Format one :class:`RerereRecord` for human-readable terminal output. Applies ``sanitize_display()`` to the path to prevent ANSI injection. """ status = "✅ resolved" if rec.has_resolution else "⏳ preimage only" ts = rec.recorded_at.strftime("%Y-%m-%d %H:%M") return f"{rec.fingerprint[:12]} {status} {ts} {sanitize_display(rec.path)}" def _check_format(fmt: str) -> None: """Exit with USER_ERROR if *fmt* is not a known output format.""" if fmt not in _FORMAT_CHOICES: print( f"❌ Unknown format {fmt!r}. Valid choices: {', '.join(_FORMAT_CHOICES)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the rerere subcommand.""" parser = subparsers.add_parser( "rerere", help="Reuse recorded resolutions for merge conflicts.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--dry-run", "-n", action="store_true", help="Show what would be auto-resolved without writing files.", ) parser.add_argument( "--format", "-f", default="text", dest="fmt", help="Output format: text or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") # -- clear ---------------------------------------------------------------- clear_p = subs.add_parser( "clear", help="Remove all cached rerere resolutions.", description=( "Delete every entry in ``.muse/rr-cache/``. This is irreversible —\n" "all recorded resolutions will be lost. A confirmation prompt is\n" "shown unless ``--yes`` is passed.\n\n" "Agent quickstart\n" "----------------\n" " muse rerere clear --yes --json\n\n" "JSON output schema\n" "------------------\n" ' {"removed": }\n\n' "Exit codes\n" "----------\n" " 0 — clear completed (removed may be 0 if cache was empty)\n" " 1 — invalid --format value\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) clear_p.add_argument("--yes", "-y", action="store_true", help="Skip confirmation prompt.") clear_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") clear_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") clear_p.set_defaults(func=run_clear) # -- forget --------------------------------------------------------------- forget_p = subs.add_parser( "forget", help="Remove the cached rerere resolution for specific paths or fingerprints.", description=( "Remove rr-cache entries so the next identical conflict is resolved\n" "manually instead of auto-applied.\n\n" "Two modes\n" "---------\n" "Path mode muse rerere forget src/foo.py\n" " Requires an active merge state to compute fingerprints.\n\n" "Fingerprint muse rerere forget --fingerprint \n" " Works without an active merge. Fingerprints come from\n" " ``muse rerere status --json``.\n\n" "Agent quickstart\n" "----------------\n" " muse rerere forget --fingerprint --json\n\n" "JSON output schema\n" "------------------\n" ' {"forgotten": ["", ...],\n' ' "not_found": ["", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — forget completed (some items may be in not_found)\n" " 1 — no args given; invalid format; no merge state in path mode\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) forget_p.add_argument( "paths", nargs="*", help="Workspace-relative paths whose rerere resolution should be removed " "(requires an active merge state to compute fingerprints).", ) forget_p.add_argument( "--fingerprint", dest="fingerprints", action="append", metavar="HEX", help="Forget by 64-char hex fingerprint directly — does not require an " "active merge state. May be repeated.", ) forget_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") forget_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") forget_p.set_defaults(func=run_forget) # -- gc ------------------------------------------------------------------- gc_p = subs.add_parser( "gc", help="Remove preimage-only rerere entries older than N days.", description=( "Garbage-collect ``.muse/rr-cache/`` by removing preimage-only\n" "entries older than ``--age`` days. Entries that have a saved\n" "resolution are always kept regardless of age.\n\n" "Agent quickstart\n" "----------------\n" " muse rerere gc --json\n" " muse rerere gc --age 30 --json\n\n" "JSON output schema\n" "------------------\n" ' {"removed": }\n\n' "Exit codes\n" "----------\n" " 0 — gc completed (removed may be 0)\n" " 1 — invalid --age value or --format value\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) gc_p.add_argument( "--age", type=int, default=60, metavar="DAYS", help="Remove preimage-only entries older than DAYS days (default: 60).", ) gc_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") gc_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") gc_p.set_defaults(func=run_gc) # -- record --------------------------------------------------------------- record_p = subs.add_parser( "record", help="Record preimages for current conflicts without applying cached resolutions.", description=( "Write a preimage entry to ``.muse/rr-cache/`` for every\n" "conflicting path in MERGE_STATE. The preimage is the fingerprint\n" "of the two conflicting blob IDs; the resolution is saved later\n" "when the user commits. Idempotent: re-recording a known conflict\n" "is a no-op.\n\n" "Agent quickstart\n" "----------------\n" " muse rerere record --json\n\n" "JSON output schema\n" "------------------\n" ' {"recorded": ["", ...], "skipped": ["", ...]}\n\n' "Exit codes\n" "----------\n" " 0 — preimages recorded (or nothing to record)\n" " 1 — MERGE_STATE incomplete\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) record_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") record_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") record_p.set_defaults(func=run_record) # -- status --------------------------------------------------------------- status_p = subs.add_parser( "status", help="Show cached rerere resolutions and their match against current conflicts.", description=( "List every entry in ``.muse/rr-cache/``, showing whether a\n" "resolution has been saved. When a merge is in progress, entries\n" "that match the current conflict set are marked with ◀ current.\n\n" "Agent quickstart\n" "----------------\n" " muse rerere status --json\n\n" "JSON output schema\n" "------------------\n" ' {"total": , "records": [\n' ' {"fingerprint": "", "path": "",\n' ' "domain": "", "has_resolution": ,\n' ' "resolution_id": "|null",\n' ' "recorded_at": "",\n' ' "matches_current_conflict": }, ...]}\n\n' "Exit codes\n" "----------\n" " 0 — status listed (may be empty)\n" " 1 — invalid --format value\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) status_p.add_argument("--format", "-f", default="text", dest="fmt", help="Output format: text or json.") status_p.add_argument("--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.") status_p.set_defaults(func=run_status) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Default action: apply cached resolutions # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Apply cached conflict resolutions to the current working tree. Reads ``.muse/MERGE_STATE.json`` to discover which paths are in conflict, looks up each path's fingerprint in ``.muse/rr-cache/``, and restores any cached resolution blobs to the working tree. Paths with no cached resolution are left unchanged and reported for manual attention. Run after ``muse merge`` exits with conflicts, or let ``muse merge`` invoke this automatically via the ``--rerere-autoupdate`` flag. Each conflicting path is validated not to escape the repository root before any file write occurs. """ dry_run: bool = args.dry_run fmt: str = args.fmt _check_format(fmt) root = require_repo() state = read_merge_state(root) if state is None or not state.conflict_paths: if fmt == "json": print(json.dumps(_RerereApplyJson(dry_run=dry_run, auto_resolved=[], remaining=[]), indent=2)) else: print("No merge in progress — nothing for rerere to do.") return manifests = _load_conflict_manifests(root) if manifests is None: print("❌ MERGE_STATE.json is incomplete — cannot load conflict manifests.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) ours_manifest, theirs_manifest = manifests plugin = resolve_plugin(root) auto_resolved: list[str] = [] remaining: list[str] = [] for path in sorted(state.conflict_paths): # Path traversal guard — protect against a tampered MERGE_STATE.json. try: (root / path).relative_to(root) except ValueError: logger.warning("⚠️ rerere: path traversal attempt — skipping %r", path) remaining.append(path) continue ours_id = ours_manifest.get(path, "") theirs_id = theirs_manifest.get(path, "") if not ours_id or not theirs_id: remaining.append(path) continue fp = compute_fingerprint(path, ours_id, theirs_id, plugin, root) if not (root / ".muse" / "rr-cache" / fp / "resolution").exists(): remaining.append(path) continue if dry_run: auto_resolved.append(path) continue dest = root / path if apply_cached(root, fp, dest): auto_resolved.append(path) else: remaining.append(path) if fmt == "json": print( json.dumps( _RerereApplyJson( dry_run=dry_run, auto_resolved=auto_resolved, remaining=remaining, ), indent=2, ) ) return prefix = "[dry-run] would resolve" if dry_run else "✅ rerere auto-resolved" for p in auto_resolved: print(f" {prefix}: {sanitize_display(p)}") for p in remaining: print(f" ⏳ needs manual resolution: {sanitize_display(p)}") if not auto_resolved and not remaining: print("No conflicting paths found.") # --------------------------------------------------------------------------- # record — write preimages without replaying # --------------------------------------------------------------------------- def run_record(args: argparse.Namespace) -> None: """Record preimages for current conflicts without applying cached resolutions. Writes a preimage entry to ``.muse/rr-cache/`` for every conflicting path in MERGE_STATE. The preimage is the fingerprint of the two conflicting blob IDs; the resolution is saved later when the user commits. Idempotent: re-recording an already-known conflict is a no-op. Paths where one side deleted the file (missing ours_id or theirs_id) are placed in ``skipped`` rather than causing an error. JSON schema (``--json``):: {"recorded": ["", ...], "skipped": ["", ...]} Exit codes: 0 — preimages recorded (or no merge in progress) 1 — MERGE_STATE is incomplete (missing ours/theirs commit) 2 — not inside a Muse repository """ fmt: str = args.fmt _check_format(fmt) root = require_repo() state = read_merge_state(root) if state is None or not state.conflict_paths: if fmt == "json": print(json.dumps(_RerereRecordJson(recorded=[], skipped=[]))) else: print("No merge in progress — nothing to record.") return manifests = _load_conflict_manifests(root) if manifests is None: print("❌ MERGE_STATE.json is incomplete — cannot load conflict manifests.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) ours_manifest, theirs_manifest = manifests domain = read_domain(root) plugin = resolve_plugin(root) recorded: list[str] = [] skipped: list[str] = [] for path in sorted(state.conflict_paths): ours_id = ours_manifest.get(path, "") theirs_id = theirs_manifest.get(path, "") if not ours_id or not theirs_id: skipped.append(path) continue record_preimage(root, path, ours_id, theirs_id, domain, plugin) recorded.append(path) if fmt == "json": print(json.dumps(_RerereRecordJson( recorded=[sanitize_display(p) for p in recorded], skipped=[sanitize_display(p) for p in skipped], ))) return for p in recorded: print(f" 📝 preimage recorded: {sanitize_display(p)}") for p in skipped: print(f" ⚠️ skipped (one side deleted): {sanitize_display(p)}") # --------------------------------------------------------------------------- # status — show cached resolutions vs current conflicts # --------------------------------------------------------------------------- def run_status(args: argparse.Namespace) -> None: """Show cached rerere resolutions and their match against current conflicts. Lists every entry in ``.muse/rr-cache/``, indicating whether a resolution has been saved. When a merge is in progress, entries whose fingerprint matches the current conflict set are marked with ``◀ current``. JSON schema:: { "total": , "records": [ { "fingerprint": "", "path": "", "domain": "", "has_resolution": , "resolution_id": " | null", "recorded_at": "", "matches_current_conflict": }, ... ] } Exit codes: 0 — status listed (may be empty) 1 — invalid --format value 2 — not inside a Muse repository """ fmt: str = args.fmt _check_format(fmt) root = require_repo() records = list_records(root) state = read_merge_state(root) # Build the set of fingerprints that match current conflicts. current_fps: set[str] = set() if state and state.conflict_paths: manifests = _load_conflict_manifests(root) if manifests is not None: ours_manifest, theirs_manifest = manifests plugin = resolve_plugin(root) for path in state.conflict_paths: ours_id = ours_manifest.get(path, "") theirs_id = theirs_manifest.get(path, "") if ours_id and theirs_id: fp = compute_fingerprint(path, ours_id, theirs_id, plugin, root) current_fps.add(fp) if fmt == "json": status_records: list[_RerereStatusRecordJson] = [ _RerereStatusRecordJson( fingerprint=sanitize_display(r.fingerprint), path=sanitize_display(r.path), domain=sanitize_display(r.domain), has_resolution=r.has_resolution, resolution_id=sanitize_display(r.resolution_id) if r.resolution_id else None, recorded_at=r.recorded_at.isoformat(), matches_current_conflict=r.fingerprint in current_fps, ) for r in records ] print(json.dumps(_RerereStatusJson(total=len(records), records=status_records))) return if not records: print("No rerere records found.") return print(f"{'fingerprint':14} {'status':16} {'recorded':16} path") print("-" * 72) for rec in records: marker = " ◀ current" if rec.fingerprint in current_fps else "" print(_fmt_record(rec) + marker) # --------------------------------------------------------------------------- # forget — remove cached resolution for specific paths or fingerprints # --------------------------------------------------------------------------- def run_forget(args: argparse.Namespace) -> None: """Remove the cached rerere resolution for specific paths or fingerprints. Two modes: **Path mode** (requires an active merge): computes the conflict fingerprint from each ``PATH`` against the current MERGE_STATE and removes its rr-cache entry. **Fingerprint mode** (``--fingerprint HEX``): removes the rr-cache entry for the given 64-char hex fingerprint directly. Does not require an active merge. Useful for agents that have stored fingerprints from a previous ``muse rerere status --json`` call. Run this when a recorded resolution is wrong and you want to force manual resolution on the next identical conflict. JSON schema:: { "forgotten": ["", ...], "not_found": ["", ...] } Exit codes: 0 — forget completed (some items may appear in not_found) 1 — no args given; invalid format; no active merge in path mode 2 — not inside a Muse repository """ paths: list[str] = args.paths fingerprints: list[str] | None = args.fingerprints fmt: str = args.fmt _check_format(fmt) if not paths and not fingerprints: print( "❌ Provide at least one PATH or --fingerprint HEX.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() forgotten: list[str] = [] not_found: list[str] = [] # -- Fingerprint mode (no active merge required) -------------------------- if fingerprints: for fp in fingerprints: if forget_record(root, fp): forgotten.append(fp) else: not_found.append(fp) # -- Path mode (requires active merge state) ------------------------------ if paths: state = read_merge_state(root) if state is None or not state.conflict_paths: print( "❌ No merge in progress — cannot determine conflict fingerprints from paths. " "Use --fingerprint instead.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) manifests = _load_conflict_manifests(root) if manifests is None: print("❌ MERGE_STATE.json is incomplete.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) ours_manifest, theirs_manifest = manifests plugin = resolve_plugin(root) for path in paths: ours_id = ours_manifest.get(path, "") theirs_id = theirs_manifest.get(path, "") if not ours_id or not theirs_id: not_found.append(path) continue fp = compute_fingerprint(path, ours_id, theirs_id, plugin, root) if forget_record(root, fp): forgotten.append(path) else: not_found.append(path) if fmt == "json": print(json.dumps(_RerereForgetJson( forgotten=[sanitize_display(x) for x in forgotten], not_found=[sanitize_display(x) for x in not_found], ))) return for item in forgotten: print(f" 🗑 forgot: {sanitize_display(item)}") for item in not_found: print(f" ⚠️ no record found: {sanitize_display(item)}") # --------------------------------------------------------------------------- # clear — remove all cached resolutions # --------------------------------------------------------------------------- def run_clear(args: argparse.Namespace) -> None: """Remove all cached rerere resolutions. Deletes the entire ``.muse/rr-cache/`` directory contents. This is irreversible — all recorded resolutions will be lost. Pass ``--yes`` to suppress the interactive confirmation prompt (required for non-interactive agents). JSON schema:: {"removed": } Exit codes: 0 — clear completed (removed may be 0 if cache was empty) 1 — invalid --format value 2 — not inside a Muse repository """ yes: bool = args.yes fmt: str = args.fmt _check_format(fmt) root = require_repo() if not yes: cache = rr_cache_dir(root) # Mirror clear_all's logic: skip symlinks so the count matches what # will actually be removed. count = ( sum(1 for e in cache.iterdir() if not e.is_symlink() and e.is_dir()) if cache.exists() else 0 ) if count == 0: if fmt == "json": print(json.dumps(_RerereScalarJson(removed=0))) else: print("rr-cache is already empty.") return confirmed = input( f"This will permanently delete {count} rerere record(s). Continue? [y/N]: " ).strip().lower() in ("y", "yes") if not confirmed: print("Aborted.") return removed = clear_all(root) if fmt == "json": print(json.dumps(_RerereScalarJson(removed=removed))) return print(f"✅ Cleared {removed} rerere record(s).") # --------------------------------------------------------------------------- # gc — garbage-collect stale preimage-only entries # --------------------------------------------------------------------------- def run_gc(args: argparse.Namespace) -> None: """Remove preimage-only rerere entries older than ``--age`` days. Keeps all entries that have a resolution saved (regardless of age). Removes entries where the user never committed a resolution — these are conflicts that were abandoned or resolved in another way. The default age threshold is 60 days. Pass ``--age DAYS`` to customise. Valid range: 1–36 500 days. JSON schema:: {"removed": } Exit codes: 0 — gc completed (removed may be 0) 1 — invalid --age value or --format value 2 — not inside a Muse repository """ fmt: str = args.fmt try: age_days: int = clamp_int(args.age, 1, 36_500, "age") except ValueError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) _check_format(fmt) root = require_repo() removed = gc_stale(root, age_days=age_days) if fmt == "json": print(json.dumps(_RerereScalarJson(removed=removed))) return if removed: print(f"✅ gc: removed {removed} stale preimage-only entry(s) older than {age_days} day(s).") else: print(f"gc: nothing to remove (threshold: {age_days} day(s)).")