"""``muse resolve`` — mark conflict paths as resolved after manual editing. After manually editing a conflicted file (or accepting one side with ``muse checkout --ours / --theirs``), use ``muse resolve`` to remove it from the conflict list so Muse knows the path is clean. Usage:: muse resolve — mark all conflicts in a file as resolved muse resolve :: — mark one symbol-level conflict as resolved muse resolve --all — mark every remaining conflict as resolved muse resolve --json — machine-readable output When a path still has active conflict markers in the working tree, Muse records it as resolved anyway (you are responsible for editing first). The file is staged automatically — just commit when all conflicts are clear:: muse resolve src/billing.py muse commit -m "merge: resolve billing conflicts" JSON output (``--json``):: { "resolved": ["src/billing.py::Invoice.charge", "src/billing.py::Invoice.refund"], "remaining": 1, "ready_to_commit": false, "duration_ms": 0.4, "exit_code": 0 } Exit codes:: 0 — success (resolved, nothing to resolve, or --all with 0 remaining) 1 — no merge in progress, or target not found """ from __future__ import annotations import argparse import json import pathlib from typing import Literal, TypedDict from muse.core.errors import ExitCode from muse.core.merge_engine import read_merge_state, resolve_path, resolve_symbol from muse.core.repo import require_repo from muse.core.timing import start_timer from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.validation import sanitize_display class _ResolveJson(EnvelopeJson): resolved: list[str] remaining: int ready_to_commit: bool def _stage_files(root: pathlib.Path, rel_paths: list[str]) -> None: """Stage resolved files into the code index (no-op for non-code domains).""" from muse.plugins.registry import read_domain if read_domain(root) != "code": return from muse.core.object_store import write_object from muse.core.types import blob_id as _blob_id from muse.plugins.code.stage import make_entry, read_stage, write_stage from muse.cli.commands.code_stage import _head_manifest head = _head_manifest(root) stage = dict(read_stage(root)) changed = False for rel_path in rel_paths: full_path = root / rel_path if not full_path.exists(): continue content = full_path.read_bytes() oid = _blob_id(content) write_object(root, oid, content) mode: Literal["A", "M", "D"] = "M" if rel_path in head else "A" stage[rel_path] = make_entry(oid, mode) changed = True if changed: write_stage(root, stage) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: parser = subparsers.add_parser( "resolve", help="Mark conflict paths as resolved after manual editing.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "target", nargs="?", metavar="PATH[::SYMBOL]", help=( "Path to mark resolved: plain 'file.py' resolves all conflicts in that " "file; 'file.py::Symbol' resolves exactly one symbol-level conflict." ), ) parser.add_argument( "--all", "-a", action="store_true", dest="resolve_all", help="Resolve every remaining conflict at once.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON instead of human text.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Mark conflict paths as resolved after manual editing.""" import sys elapsed = start_timer() target: str | None = args.target resolve_all: bool = args.resolve_all json_out: bool = args.json_out if not target and not resolve_all: print( "❌ Specify a path (e.g. 'muse resolve src/billing.py') " "or use --all to resolve every conflict.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if target and resolve_all: print("❌ Cannot combine a specific path with --all.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() merge_state = read_merge_state(root) if merge_state is None: if json_out: print(json.dumps(_ResolveJson( **make_envelope(elapsed, exit_code=ExitCode.USER_ERROR), resolved=[], remaining=0, ready_to_commit=False, ))) else: print("❌ No merge in progress — nothing to resolve.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) resolved: list[str] = [] if resolve_all: # Drain the entire conflict list, file by file. paths_to_clear = list(merge_state.conflict_paths) staged_files: list[str] = [] for entry in paths_to_clear: file_part = entry.split("::")[0] if "::" in entry else entry cleared = resolve_path(root, file_part) for c in cleared: if c not in resolved: resolved.append(c) if cleared and file_part not in staged_files: staged_files.append(file_part) # Re-read state after each write so we don't re-process. merge_state = read_merge_state(root) or merge_state _stage_files(root, staged_files) elif "::" in (target or ""): # Symbol-level resolve. found = resolve_symbol(root, target or "") if found: resolved.append(target or "") file_part = (target or "").split("::")[0] _stage_files(root, [file_part]) else: display = sanitize_display(target or "") if json_out: print(json.dumps(_ResolveJson( **make_envelope(elapsed), resolved=[], remaining=len(merge_state.conflict_paths), ready_to_commit=len(merge_state.conflict_paths) == 0, ))) else: print(f"ℹ️ '{display}' is not in the conflict list — already resolved or never conflicted.") return else: # File-level resolve. cleared = resolve_path(root, target or "") if not cleared: display = sanitize_display(target or "") if json_out: print(json.dumps(_ResolveJson( **make_envelope(elapsed), resolved=[], remaining=len(merge_state.conflict_paths), ready_to_commit=len(merge_state.conflict_paths) == 0, ))) else: print(f"ℹ️ '{display}' is not in the conflict list — already resolved or never conflicted.") return resolved.extend(cleared) _stage_files(root, [target or ""]) # Read final state for remaining count. final_state = read_merge_state(root) remaining = len(final_state.conflict_paths) if final_state is not None else 0 ready = remaining == 0 if json_out: print(json.dumps(_ResolveJson( **make_envelope(elapsed), resolved=resolved, remaining=remaining, ready_to_commit=ready, ))) return # Human text output. if not resolved: print("ℹ️ No conflicts were cleared.") return for entry in resolved: print(f"✅ Resolved: {sanitize_display(entry)}") if ready: print("\n✅ All conflicts resolved — files staged automatically.") print(" muse commit -m 'merge: ...'") else: print(f"\n {remaining} conflict(s) remaining — run 'muse conflicts' to see them.")