rerere.py
python
| 1 | """muse rerere — reuse recorded resolutions for musical merge conflicts. |
| 2 | |
| 3 | Commands |
| 4 | -------- |
| 5 | ``muse rerere`` |
| 6 | Attempt to auto-apply any cached resolution for conflicts currently |
| 7 | listed in ``.muse/MERGE_STATE.json``. Prints the number resolved. |
| 8 | |
| 9 | ``muse rerere list`` |
| 10 | Show all conflict fingerprints currently in the rr-cache. Entries |
| 11 | marked with ``[R]`` have a postimage (resolution recorded); entries |
| 12 | marked with ``[C]`` are conflict-only (awaiting resolution). |
| 13 | |
| 14 | ``muse rerere forget <hash>`` |
| 15 | Remove a single cached conflict/resolution from the rr-cache. |
| 16 | |
| 17 | ``muse rerere clear`` |
| 18 | Purge the entire rr-cache. Use when cached resolutions are stale or |
| 19 | incorrect. |
| 20 | """ |
| 21 | from __future__ import annotations |
| 22 | |
| 23 | import logging |
| 24 | import pathlib |
| 25 | |
| 26 | import typer |
| 27 | |
| 28 | from maestro.muse_cli._repo import require_repo |
| 29 | from maestro.muse_cli.errors import ExitCode |
| 30 | from maestro.services.muse_rerere import ( |
| 31 | ConflictDict, |
| 32 | apply_rerere, |
| 33 | clear_rerere, |
| 34 | forget_rerere, |
| 35 | list_rerere, |
| 36 | ) |
| 37 | |
| 38 | logger = logging.getLogger(__name__) |
| 39 | |
| 40 | app = typer.Typer( |
| 41 | name="rerere", |
| 42 | help="Reuse recorded resolutions for musical merge conflicts.", |
| 43 | no_args_is_help=False, |
| 44 | ) |
| 45 | |
| 46 | |
| 47 | # --------------------------------------------------------------------------- |
| 48 | # Helpers |
| 49 | # --------------------------------------------------------------------------- |
| 50 | |
| 51 | |
| 52 | def _load_current_conflicts(root: pathlib.Path) -> list[ConflictDict]: |
| 53 | """Read conflict list from .muse/MERGE_STATE.json, if present.""" |
| 54 | import json |
| 55 | |
| 56 | merge_state_path = root / ".muse" / "MERGE_STATE.json" |
| 57 | if not merge_state_path.exists(): |
| 58 | return [] |
| 59 | try: |
| 60 | data = json.loads(merge_state_path.read_text(encoding="utf-8")) |
| 61 | raw = data.get("conflict_paths", []) |
| 62 | # conflict_paths is a list of file-path strings in the merge engine |
| 63 | # (file-level conflicts), not MergeConflict dicts. Wrap each path |
| 64 | # in a minimal dict so rerere can fingerprint it. |
| 65 | return [ConflictDict(region_id=p, type="file", description=f"conflict in {p}") for p in raw] |
| 66 | except Exception as exc: # noqa: BLE001 |
| 67 | logger.warning("⚠️ muse rerere: could not read MERGE_STATE.json: %s", exc) |
| 68 | return [] |
| 69 | |
| 70 | |
| 71 | # --------------------------------------------------------------------------- |
| 72 | # Default command — apply cached resolution |
| 73 | # --------------------------------------------------------------------------- |
| 74 | |
| 75 | |
| 76 | @app.callback(invoke_without_command=True) |
| 77 | def rerere_apply(ctx: typer.Context) -> None: |
| 78 | """Auto-apply any cached resolution for current merge conflicts.""" |
| 79 | if ctx.invoked_subcommand is not None: |
| 80 | return |
| 81 | |
| 82 | root = require_repo() |
| 83 | conflicts = _load_current_conflicts(root) |
| 84 | |
| 85 | if not conflicts: |
| 86 | typer.echo("✅ No active merge conflicts found (no MERGE_STATE.json or empty conflict list).") |
| 87 | return |
| 88 | |
| 89 | applied, _resolution = apply_rerere(root, conflicts) |
| 90 | if applied: |
| 91 | typer.echo(f"✅ Resolved {applied} conflict(s) using rerere.") |
| 92 | else: |
| 93 | typer.echo("⚠️ No cached resolution found for current conflicts.") |
| 94 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 95 | |
| 96 | |
| 97 | # --------------------------------------------------------------------------- |
| 98 | # list subcommand |
| 99 | # --------------------------------------------------------------------------- |
| 100 | |
| 101 | |
| 102 | @app.command("list") |
| 103 | def rerere_list() -> None: |
| 104 | """Show all conflict fingerprints in the rr-cache.""" |
| 105 | root = require_repo() |
| 106 | hashes = list_rerere(root) |
| 107 | |
| 108 | if not hashes: |
| 109 | typer.echo("rr-cache is empty.") |
| 110 | return |
| 111 | |
| 112 | typer.echo(f"rr-cache ({len(hashes)} entr{'y' if len(hashes) == 1 else 'ies'}):") |
| 113 | cache_root = root / ".muse" / "rr-cache" |
| 114 | for h in hashes: |
| 115 | postimage = cache_root / h / "postimage" |
| 116 | tag = "[R]" if postimage.exists() else "[C]" |
| 117 | typer.echo(f" {tag} {h}") |
| 118 | |
| 119 | |
| 120 | # --------------------------------------------------------------------------- |
| 121 | # forget subcommand |
| 122 | # --------------------------------------------------------------------------- |
| 123 | |
| 124 | |
| 125 | @app.command("forget") |
| 126 | def rerere_forget( |
| 127 | conflict_hash: str = typer.Argument(..., help="SHA-256 fingerprint hash to remove."), |
| 128 | ) -> None: |
| 129 | """Remove a single cached conflict/resolution from the rr-cache.""" |
| 130 | root = require_repo() |
| 131 | removed = forget_rerere(root, conflict_hash) |
| 132 | if removed: |
| 133 | typer.echo(f"✅ Forgot rerere entry {conflict_hash[:12]}…") |
| 134 | else: |
| 135 | typer.echo(f"⚠️ Hash {conflict_hash[:12]}… not found in rr-cache.") |
| 136 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 137 | |
| 138 | |
| 139 | # --------------------------------------------------------------------------- |
| 140 | # clear subcommand |
| 141 | # --------------------------------------------------------------------------- |
| 142 | |
| 143 | |
| 144 | @app.command("clear") |
| 145 | def rerere_clear() -> None: |
| 146 | """Purge the entire rr-cache.""" |
| 147 | root = require_repo() |
| 148 | count = clear_rerere(root) |
| 149 | typer.echo(f"✅ Cleared {count} rr-cache entr{'y' if count == 1 else 'ies'}.") |