"""``muse merge`` — three-way merge a branch into the current branch. Algorithm --------- 1. Find the merge base (LCA of HEAD and the target branch). 2. Delegate conflict detection and manifest reconciliation to the domain plugin. 3. If clean → apply merged manifest, write new commit, advance HEAD. 4. If conflicts → write conflict markers to the working tree, write ``.muse/MERGE_STATE.json``, exit non-zero. Usage:: muse merge — three-way merge into current branch muse merge --dry-run — simulate without writing anything muse merge --strategy=ours — auto-resolve conflicts keeping ours muse merge --abort — cancel in-progress merge JSON output (``--format json`` or ``--json``):: { "status": "merged|fast_forward|conflict|up_to_date", "commit_id": " | null", "branch": "", "current_branch": "", "base_commit_id": " | null", "conflicts": ["", ...], "files_changed": {"added": N, "modified": N, "deleted": N}, "semver_impact": "MAJOR|MINOR|PATCH|", "strategy": "ours|theirs|recursive|null", "dry_run": false, "duration_ms": 12.4, "exit_code": 0 } JSON error schema (usage/internal errors):: { "status": "error", "error": "", "exit_code": } When ``--json`` is active all errors go to stdout as JSON — no prose on stderr. Agents should parse stdout and check ``exit_code``. Exit codes:: 0 — success (merged, fast-forward, up-to-date, dry-run) 1 — conflict detected (use ``muse conflicts`` / ``muse checkout --ours/--theirs``) 1 — invalid arguments (missing branch, bad format) 3 — internal error (unreadable commit or snapshot) """ import argparse import datetime import json import logging import os import pathlib import sys import time from typing import TypedDict from muse.core.types import short_id from muse.core.paths import merge_state_path as _merge_state_path from muse.core.errors import ExitCode from muse.core.merge_engine import ( STRATEGY_MAP, apply_merge, detect_conflicts, diff_snapshots, find_merge_base, read_merge_state, run_merge, write_merge_state, ) from muse.core.repo import require_repo from muse.cli.config import get_config_value from muse.core.validation import sanitize_provenance from muse.core.terminal import use_color from muse.core.harmony import auto_apply as harmony_auto_apply from muse.core.ids import hash_commit, hash_snapshot from muse.core.snapshot import directories_from_manifest from muse.core.types import JsonObject, Manifest from muse.core.refs import ( RefConflictError, get_head_commit_id, read_current_branch, resolve_any_ref, write_branch_ref, ) from muse.core.commits import ( CommitRecord, read_commit, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, get_head_snapshot_manifest, read_snapshot, write_snapshot, ) from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.reflog import append_reflog from muse.core.validation import sanitize_display, validate_branch_name from muse.cli.config import get_protected_branches, is_branch_protected from muse.core.workdir import apply_manifest from muse.core.object_store import restore_object from muse.core.merge_debug import merge_debug_log, merge_debug_manifest_summary from muse.core.timing import start_timer from muse.plugins.code.stage import read_stage, write_stage from muse.cli.guard import require_clean_workdir from muse.cli.commands.checkout import _apply_autoshelf from muse.cli.commands.shelf import _shelf_push_programmatic from muse.domain import AddressedMergePlugin, ConflictDict, DomainOp, MergeResult, SnapshotManifest from muse.plugins.registry import read_domain, resolve_plugin logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Wire-format TypedDicts # --------------------------------------------------------------------------- class _SymbolConflictDict(TypedDict): path: str symbol: str ours: str theirs: str type _FilesChangedMap = dict[str, int] class _MergeResultDict(TypedDict): """Merge execution summary — mirrored in ProposalResponse.merge_result.""" status: str commit_id: str | None strategy: str | None on_conflict: str | None history: str | None conflicts: list[str] files_changed: _FilesChangedMap semver_impact: str class _MergeJsonBase(EnvelopeJson): """Common JSON envelope fields for all ``muse merge --json`` outcome paths. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. """ status: str # "merged" | "fast_forward" | "conflict" | "up_to_date" commit_id: str | None branch: str current_branch: str base_commit_id: str | None conflicts: list[str] files_changed: _FilesChangedMap semver_impact: str strategy: str | None on_conflict: str | None history: str | None dry_run: bool merge_result: _MergeResultDict class _MergeJson(_MergeJsonBase, total=False): """Extended merge JSON — adds optional conflict-detail fields.""" conflict_records: list[ConflictDict] symbol_conflicts: list[_SymbolConflictDict] class _MergeErrorJson(EnvelopeJson): """Error payload for ``muse merge --json`` on usage/internal errors.""" status: str # "error" error: str def _emit_error(json_out: bool, msg: str, code: "ExitCode", elapsed: float) -> None: """Print an error and raise SystemExit. Never returns. In ``--json`` mode the error goes to stdout as a JSON payload so agents always get parseable output. In text mode it goes to stderr. """ if json_out: print(json.dumps(_MergeErrorJson( **make_envelope(elapsed, exit_code=int(code)), status="error", error=msg, ))) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(code) # --------------------------------------------------------------------------- # Colour helpers — respect NO_COLOR and TERM=dumb # --------------------------------------------------------------------------- _RESET = "\033[0m" _BOLD = "\033[1m" _DIM = "\033[2m" _GREEN = "\033[32m" _RED = "\033[31m" _YELLOW = "\033[33m" _CYAN = "\033[36m" def _c(text: str, *codes: str) -> str: """Wrap *text* in ANSI escape *codes* only when writing to a colour-capable TTY.""" if not use_color(): return text return "".join(codes) + text + _RESET def _diff_stats( old: Manifest, new: Manifest, ) -> tuple[int, int, int]: """Return (added, modified, deleted) file counts between two manifests.""" added = sum(1 for k in new if k not in old) deleted = sum(1 for k in old if k not in new) modified = sum(1 for k in new if k in old and old[k] != new[k]) return added, modified, deleted def _print_file_stats( added: int, modified: int, deleted: int, ) -> None: """Emit the 'N files changed (A added, M modified, D deleted)' summary line.""" total = added + modified + deleted if total == 0: return files_word = "file" if total == 1 else "files" parts: list[str] = [] if added: parts.append(_c(f"{added} added", _GREEN)) if modified: parts.append(f"{modified} modified") if deleted: parts.append(_c(f"{deleted} deleted", _RED)) detail = ", ".join(parts) print(f" {_c(str(total), _BOLD)} {files_word} changed ({detail})") def _semver_from_op_log(op_log: list[DomainOp]) -> str: """Infer a proposed semver bump from a structured merge operation log. This is Muse-unique: because Muse tracks named symbols across time (not just line changes), it can tell you whether a merge would introduce a breaking API change (MAJOR), a new feature (MINOR), or a fix (PATCH). Heuristic (conservative — errs toward higher impact): - Any ``delete`` or ``rename`` on a public symbol → MAJOR - Any ``add`` → MINOR - Any ``modify`` → PATCH - Empty log → "" (cannot determine) Args: op_log: List of operation dicts from ``MergeResult.op_log``. Each dict has at least ``"op"`` (add/modify/delete/rename) and ``"symbol"`` keys. Public symbols lack a ``_`` prefix. Returns: "MAJOR", "MINOR", "PATCH", or "" when the log is empty. """ if not op_log: return "" impact = "PATCH" for op in op_log: op_kind = str(op.get("op", "")).lower() symbol = str(op.get("symbol", "")) is_public = symbol and not symbol.startswith("_") if op_kind in ("delete", "rename") and is_public: return "MAJOR" if op_kind == "add" and is_public and impact != "MAJOR": impact = "MINOR" return impact def _run_octopus_merge( root: pathlib.Path, branches: list[str], *, message: str | None, dry_run: bool, json_out: bool, elapsed: "() -> float", ) -> None: """Implement ``muse merge ... `` (muse#105). One commit, N+1 parents (current branch tip + every named branch's tip), written only when none of the N+1 sides make divergent changes to the same path relative to their shared merge base. Refuses cleanly — naming the conflicting pair and paths — the moment two sides disagree, exactly git's own octopus behavior: no partial commit, no invented resolution, fall back to sequential two-way merges instead. JSON fields ----------- status ``"merged"`` or ``"conflict"``. commit_id New merge commit ID, or ``null`` on conflict/dry-run. branches The N branch names passed on the command line. current_branch Branch being merged into. base_commit_id Shared merge base across all N+1 tips, or ``null``. conflict ``{"branch_a", "branch_b", "paths"}`` when refused. files_changed ``{"added", "modified", "deleted"}`` counts. dry_run ``true`` when ``--dry-run`` was passed. """ from muse.core.octopus_merge import ( OctopusConflictError, OctopusNoCommonAncestorError, compute_octopus_merge, ) require_clean_workdir(root, "merge", force=False, json_out=json_out) try: current_branch = read_current_branch(root) except ValueError: _emit_error( json_out, "Octopus merge requires an active branch — detached HEAD has no " "branch to record as a parent. Run 'muse checkout ' first.", ExitCode.USER_ERROR, elapsed, ) return current_commit_id = get_head_commit_id(root, current_branch) if current_commit_id is None: _emit_error(json_out, f"Branch '{current_branch}' has no commits.", ExitCode.USER_ERROR, elapsed) return target_commit_ids: list[str] = [] display_names: dict[str, str] = {current_commit_id: current_branch} for b in branches: cid = resolve_any_ref(root, b) if cid is None: _emit_error(json_out, f"Branch not found: '{sanitize_display(b)}'", ExitCode.USER_ERROR, elapsed) return target_commit_ids.append(cid) display_names[cid] = b try: result = compute_octopus_merge( root, current_commit_id, target_commit_ids, display_names=display_names, ) except OctopusConflictError as exc: if json_out: print(json.dumps({ **make_envelope(elapsed, exit_code=ExitCode.USER_ERROR), "status": "conflict", "commit_id": None, "branches": branches, "current_branch": current_branch, "base_commit_id": None, "conflict": { "branch_a": exc.branch_a, "branch_b": exc.branch_b, "paths": sorted(exc.paths), }, "dry_run": dry_run, })) else: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) except OctopusNoCommonAncestorError as exc: _emit_error(json_out, str(exc), ExitCode.USER_ERROR, elapsed) return current_manifest = get_head_snapshot_manifest(root, current_branch) or {} added = len([p for p in result.manifest if p not in current_manifest]) deleted = len([p for p in current_manifest if p not in result.manifest]) modified = len([ p for p in result.manifest if p in current_manifest and current_manifest[p] != result.manifest[p] ]) if dry_run: if json_out: print(json.dumps({ **make_envelope(elapsed), "status": "merged", "commit_id": None, "branches": branches, "current_branch": current_branch, "base_commit_id": result.base_commit_id, "conflict": None, "files_changed": {"added": added, "modified": modified, "deleted": deleted}, "dry_run": True, })) else: print( f"[dry-run] Would octopus-merge {len(branches)} branches " f"({', '.join(sanitize_display(b) for b in branches)}) into " f"'{sanitize_display(current_branch)}': " f"{added} added, {modified} modified, {deleted} deleted." ) return merged_dirs = directories_from_manifest(result.manifest) snapshot_id = hash_snapshot(result.manifest, merged_dirs) merge_message = message or ( f"Octopus merge branches {', '.join(branches)} into {current_branch}" ) committed_at = datetime.datetime.now(tz=datetime.timezone.utc) commit_id = hash_commit( parent_ids=result.parent_commit_ids, snapshot_id=snapshot_id, message=merge_message, committed_at_iso=committed_at.isoformat(), author="user", ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=result.manifest, directories=merged_dirs)) write_commit(root, CommitRecord( commit_id=commit_id, branch=current_branch, snapshot_id=snapshot_id, message=merge_message, committed_at=committed_at, parent_commit_id=result.parent_commit_ids[0], parent2_commit_id=result.parent_commit_ids[1] if len(result.parent_commit_ids) > 1 else None, parent_commit_ids=result.parent_commit_ids, author="user", )) apply_manifest(root, current_manifest, result.manifest, target_directories=merged_dirs) try: write_branch_ref(root, current_branch, commit_id, expected_id=current_commit_id) except RefConflictError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) append_reflog( root, current_branch, old_id=current_commit_id, new_id=commit_id, author="user", operation=f"octopus merge: {', '.join(sanitize_display(b) for b in branches)} into {sanitize_display(current_branch)}", ) if json_out: print(json.dumps({ **make_envelope(elapsed), "status": "merged", "commit_id": commit_id, "branches": branches, "current_branch": current_branch, "base_commit_id": result.base_commit_id, "conflict": None, "files_changed": {"added": added, "modified": modified, "deleted": deleted}, "dry_run": False, })) else: print( f"Octopus merge: {len(branches)} branches " f"({', '.join(sanitize_display(b) for b in branches)}) merged into " f"'{sanitize_display(current_branch)}' — {commit_id}" ) def _run_abort(json_out: bool) -> None: """Abort an in-progress merge and restore the pre-merge working tree.""" root = require_repo() merge_state = read_merge_state(root) if merge_state is None: msg = "No merge in progress." if json_out: print(json.dumps({"error": "no_merge_in_progress", "message": msg})) else: print(f"❌ {msg}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) ours_commit_id = merge_state.ours_commit or "" if not ours_commit_id: print( "❌ MERGE_STATE has no ours_commit — cannot restore working tree.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) ours_commit = read_commit(root, ours_commit_id) if ours_commit is None: print( f"❌ Cannot restore: pre-merge commit '{ours_commit_id}' not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) snapshot_id: str | None = ours_commit.snapshot_id or None if snapshot_id: snap = read_snapshot(root, snapshot_id) if snap is not None: theirs_manifest: Manifest = {} if merge_state.theirs_commit: theirs_commit_rec = read_commit(root, merge_state.theirs_commit) if theirs_commit_rec: theirs_snap_rec = read_snapshot(root, theirs_commit_rec.snapshot_id) if theirs_snap_rec: theirs_manifest = dict(theirs_snap_rec.manifest) apply_manifest(root, {**snap.manifest, **theirs_manifest}, snap.manifest, target_directories=directories_from_manifest(snap.manifest)) # Clear the stage — abort means "undo everything back to the pre-merge state", # including any files staged during conflict resolution. write_stage(root, {}) _merge_state_path(root).unlink(missing_ok=True) if json_out: print(json.dumps({ "status": "aborted", "restored_to": ours_commit_id, })) else: print(f"Merge aborted. Working tree restored to {ours_commit_id}.") def _build_explain_trace( requested_strategy: str | None, on_conflict: str | None, history: str | None, diff_unit: str, resolution: str, base_commit_id: str | None, base_manifest: Manifest, ours_manifest: Manifest, theirs_manifest: Manifest, result_manifest: Manifest, pre_harmony_conflicts: list[str], harmony_decisions: JsonObject, harmony_was_run: bool, ) -> JsonObject: """Build the explain trace dict from merge inputs and outcomes.""" from collections import Counter # Normalise symbol-level conflict addresses (e.g. "file.py::Sym") to file paths. pre_harmony_file_conflicts: set[str] = { p.split("::")[0] if "::" in p else p for p in pre_harmony_conflicts } all_paths = sorted( set(base_manifest) | set(ours_manifest) | set(theirs_manifest) | set(result_manifest) ) per_path = [] for path in all_paths: base_id = base_manifest.get(path) ours_id = ours_manifest.get(path) theirs_id = theirs_manifest.get(path) ours_changed = ours_id != base_id theirs_changed = theirs_id != base_id harmony_info = harmony_decisions.get(path, {}) harmony_result_str = harmony_info.get("result", "not_applicable") harmony_pattern_id = harmony_info.get("pattern_id") harmony_checked = path in pre_harmony_file_conflicts and harmony_was_run if harmony_info.get("result") == "auto_resolved": decision = "harmony_auto_resolved" reason = "harmony auto-resolved via saved pattern" elif path in pre_harmony_file_conflicts: decision = "conflict" reason = "both sides modified; content diverged" if harmony_was_run: harmony_result_str = harmony_info.get("result", "no_pattern") elif not ours_changed and not theirs_changed: decision = "no_change" reason = "identical on both sides" elif ours_changed and not theirs_changed: decision = "take_ours_only" reason = "only ours changed" elif not ours_changed and theirs_changed: decision = "take_theirs_only" reason = "only theirs changed" elif ours_id == theirs_id: decision = "convergent" reason = "both sides independently arrived at same content" elif resolution == "prefer_theirs": decision = "overlay_auto_resolved" reason = "overlay strategy: theirs wins" elif resolution == "prefer_ours": decision = "take_ours_only" reason = "prefer-ours strategy: ours wins" else: decision = "independence_merged" reason = "symbol-level independence confirmed; text merge applied" per_path.append({ "path": path, "ours_changed": ours_changed, "theirs_changed": theirs_changed, "ours_id": ours_id, "theirs_id": theirs_id, "base_id": base_id, "decision": decision, "reason": reason, "harmony_checked": harmony_checked, "harmony_result": harmony_result_str if harmony_checked else "not_applicable", "harmony_pattern_id": harmony_pattern_id, "attributes_rule": None, "independence_merge_attempted": decision == "independence_merged", }) counts = Counter(e["decision"] for e in per_path) return { "strategy_routing": { "requested_strategy": requested_strategy, "on_conflict": on_conflict, "history": history, "resolved_diff_unit": diff_unit, "resolved_resolution": resolution, "case": 4, }, "merge_base_commit_id": base_commit_id, "per_path": per_path, "summary": { "total_paths": len(per_path), "clean_no_change": counts.get("no_change", 0), "take_ours_only": counts.get("take_ours_only", 0), "take_theirs_only": counts.get("take_theirs_only", 0), "convergent": counts.get("convergent", 0), "harmony_auto_resolved": counts.get("harmony_auto_resolved", 0), "attributes_auto_resolved": counts.get("attributes_auto_resolved", 0), "independence_merged": counts.get("independence_merged", 0), "conflicts": counts.get("conflict", 0), }, } def _print_explain_trace(trace: JsonObject) -> None: """Print the explain trace in human-readable format to stdout.""" sr = trace["strategy_routing"] diff_unit = sr.get("resolved_diff_unit", "") resolution = sr.get("resolved_resolution", "") strategy_name = sr.get("requested_strategy") or "recursive" base_cid = trace.get("merge_base_commit_id") or "(none)" print(f"\n Merge base: {base_cid}") print(f" Strategy: {strategy_name} ({diff_unit} + {resolution}) → Case {sr.get('case', '?')}") print() _DECISION_LABEL = { "no_change": "no-change", "take_ours_only": "take-ours", "take_theirs_only": "take-theirs", "convergent": "convergent", "harmony_auto_resolved": "harmony", "attributes_auto_resolved": "attributes", "independence_merged": "independence", "conflict": "CONFLICT", "overlay_auto_resolved": "overlay", } for entry in trace["per_path"]: path = sanitize_display(entry["path"]) label = _DECISION_LABEL.get(entry["decision"], entry["decision"]) reason = entry.get("reason", "") pattern_id = entry.get("harmony_pattern_id") extra = "" if pattern_id: extra = f" (pattern {pattern_id})" print(f" {path:<30} {label:<14} {reason}{extra}") s = trace["summary"] parts = [] if s.get("conflicts"): parts.append(f"{s['conflicts']} conflict{'s' if s['conflicts'] != 1 else ''}") if s.get("harmony_auto_resolved"): parts.append(f"{s['harmony_auto_resolved']} harmony auto-resolved") if s.get("convergent"): parts.append(f"{s['convergent']} convergent") if s.get("clean_no_change"): parts.append(f"{s['clean_no_change']} no-change") print() print(" " + " · ".join(parts) if parts else " (no changes)") def _run_merge( root: pathlib.Path, branch: str, no_ff: bool, message: str | None, harmony_autoupdate: bool, force: bool, dry_run: bool, strategy: str | None, json_out: bool, on_conflict: str | None = None, history: str | None = None, explain: bool = False, ) -> None: """Execute the merge — called by :func:`run` after autoshelf setup.""" elapsed = start_timer() # Dry-run never touches the working tree; skip the clean-workdir check. if not dry_run: require_clean_workdir(root, "merge", force=force, json_out=json_out) current_branch = read_current_branch(root) domain = read_domain(root) plugin = resolve_plugin(root) protected = get_protected_branches(root) if is_branch_protected(current_branch, protected): _emit_error( json_out, f"Branch '{sanitize_display(current_branch)}' is protected — merge directly into it is not allowed. " "Merge into an integration branch (e.g. dev) and promote via the standard flow.", ExitCode.USER_ERROR, elapsed, ) if branch == current_branch: _emit_error(json_out, "Cannot merge a branch into itself.", ExitCode.USER_ERROR, elapsed) ours_commit_id = get_head_commit_id(root, current_branch) theirs_commit_id = resolve_any_ref(root, branch) if theirs_commit_id is None: _emit_error(json_out, f"Branch '{sanitize_display(branch)}' has no commits.", ExitCode.USER_ERROR, elapsed) if ours_commit_id is None: _emit_error(json_out, "Current branch has no commits.", ExitCode.USER_ERROR, elapsed) base_commit_id = find_merge_base(root, ours_commit_id, theirs_commit_id) # Capture the user-visible strategy name before any normalization. # Cases 1 and 2 emit JSON before the STRATEGY_MAP routing block, so # reported_strategy must be set here. reported_strategy = strategy # ------------------------------------------------------------------- # Case 1: already up to date # ------------------------------------------------------------------- if base_commit_id == theirs_commit_id: if json_out: print(json.dumps(_MergeJsonBase( **make_envelope(elapsed), status="up_to_date", commit_id=ours_commit_id, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=[], files_changed={"added": 0, "modified": 0, "deleted": 0}, semver_impact="", strategy=reported_strategy, on_conflict=on_conflict, history=history, dry_run=dry_run, merge_result=_MergeResultDict( status="up_to_date", commit_id=ours_commit_id, strategy=reported_strategy, on_conflict=on_conflict, history=history, conflicts=[], files_changed={"added": 0, "modified": 0, "deleted": 0}, semver_impact="", ), ))) else: print("Already up to date.") return # ------------------------------------------------------------------- # Case 2: fast-forward # ------------------------------------------------------------------- if base_commit_id == ours_commit_id and not no_ff: theirs_commit = read_commit(root, theirs_commit_id) if theirs_commit is None: print( f"❌ Cannot read commit {theirs_commit_id} for branch " f"'{sanitize_display(branch)}' — aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) ff_snap = read_snapshot(root, theirs_commit.snapshot_id) if ff_snap is None: print( f"❌ Cannot read snapshot {theirs_commit.snapshot_id} for branch " f"'{sanitize_display(branch)}' — aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) ff_manifest: Manifest = ff_snap.manifest ours_commit_rec = read_commit(root, ours_commit_id) ours_ff_manifest: Manifest = {} if ours_commit_rec: ours_snap_rec = read_snapshot(root, ours_commit_rec.snapshot_id) if ours_snap_rec: ours_ff_manifest = ours_snap_rec.manifest added, modified, deleted = _diff_stats(ours_ff_manifest, ff_manifest) if not dry_run: apply_manifest(root, ours_ff_manifest, ff_manifest, target_directories=directories_from_manifest(ff_manifest)) try: validate_branch_name(current_branch) except ValueError as exc: print( f"❌ Current branch name is invalid: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) try: write_branch_ref(root, current_branch, theirs_commit_id, expected_id=ours_commit_id) except RefConflictError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) append_reflog( root, current_branch, old_id=ours_commit_id, new_id=theirs_commit_id, author="user", operation=( f"merge: fast-forward {sanitize_display(branch)} " f"→ {sanitize_display(current_branch)}" ), ) if json_out: _ff_commit_id = theirs_commit_id if not dry_run else None _ff_files = {"added": added, "modified": modified, "deleted": deleted} print(json.dumps(_MergeJsonBase( **make_envelope(elapsed), status="fast_forward", commit_id=_ff_commit_id, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=[], files_changed=_ff_files, semver_impact="", strategy=reported_strategy, on_conflict=on_conflict, history=history, dry_run=dry_run, merge_result=_MergeResultDict( status="fast_forward", commit_id=_ff_commit_id, strategy=reported_strategy, on_conflict=on_conflict, history=history, conflicts=[], files_changed=_ff_files, semver_impact="", ), ))) else: if dry_run: print(f"{_c('[dry-run]', _CYAN)} Nothing will be written.\n") print( f"{_c('Updating', _DIM)} " f"{_c(short_id(ours_commit_id), _YELLOW)}.." f"{_c(short_id(theirs_commit_id), _YELLOW)}" ) label = "Would fast-forward" if dry_run else "Fast-forward" print( _c(label, _BOLD) + f" {sanitize_display(branch)} → {sanitize_display(current_branch)}" ) _print_file_stats(added, modified, deleted) return # Read both branch manifests — needed by Cases 3 and 4. _ours_manifest = get_head_snapshot_manifest(root, current_branch) if _ours_manifest is None: print( f"❌ Cannot read snapshot for branch '{sanitize_display(current_branch)}' " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) ours_manifest: Manifest = _ours_manifest _theirs_manifest = get_head_snapshot_manifest(root, branch) if _theirs_manifest is None: print( f"❌ Cannot read snapshot for branch '{sanitize_display(branch)}' " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) theirs_manifest: Manifest = _theirs_manifest # Use STRATEGY_MAP as the single source of truth for routing. # Look up the engine; default to recursive if no strategy specified. _engine = STRATEGY_MAP.get(strategy or "recursive") # Determine effective resolution: --on-conflict overrides engine default. if on_conflict == "ours": _resolution = "prefer_ours" elif on_conflict == "theirs": _resolution = "prefer_theirs" else: _resolution = _engine.resolution if _engine else "escalate" # diff_unit determines execution path. snapshot/replay_* fall through to the # Case 4 three-way path until Phase 6 adds separate execution paths. _diff_unit = _engine.diff_unit if _engine else "three_way" # Route prefer_ours/prefer_theirs to Case 3 auto-resolve. # For the case 3 path, map resolution back to a strategy string. if _resolution == "prefer_ours": strategy = "ours" elif _resolution == "prefer_theirs": strategy = "theirs" else: strategy = None # three-way escalate → Case 4 # ------------------------------------------------------------------- # Case 3: auto-resolve — ours / theirs (driven by STRATEGY_MAP resolution) # ------------------------------------------------------------------- if strategy in ("ours", "theirs") and not dry_run: base_manifest_strategy: Manifest = {} if base_commit_id: base_commit_rec = read_commit(root, base_commit_id) if base_commit_rec is None: print( f"❌ Cannot read merge base commit {base_commit_id} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_snap_rec = read_snapshot(root, base_commit_rec.snapshot_id) if base_snap_rec is None: print( f"❌ Cannot read snapshot for merge base {base_commit_id} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_manifest_strategy = base_snap_rec.manifest from muse.core.merge_engine import MergeEngine _case3_engine = MergeEngine( diff_unit=_diff_unit, resolution="prefer_ours" if strategy == "ours" else "prefer_theirs", ) _case3_result = run_merge( base_manifest_strategy, ours_manifest, theirs_manifest, _case3_engine, plugin=plugin, repo_root=root, domain=domain, ) chosen_manifest = dict(_case3_result.merged["files"]) chosen_dirs = directories_from_manifest(chosen_manifest) apply_manifest(root, {**ours_manifest, **theirs_manifest}, chosen_manifest, target_directories=chosen_dirs) committed_at = datetime.datetime.now(datetime.timezone.utc) merge_author = sanitize_provenance(get_config_value("user.handle", root) or "") safe_branch = sanitize_display(branch) safe_current = sanitize_display(current_branch) chosen_dirs = directories_from_manifest(chosen_manifest) strategy_snap_id = hash_snapshot(chosen_manifest, chosen_dirs) # --history squash/rebase: single-parent commit, no merge parent — # Case 3 (ours/theirs/overlay auto-resolve) previously always wrote # a two-parent commit regardless of --history, unlike Case 4's # three-way path which already honoured it (muse#29). case3_squash = history in ("squash", "rebase") if case3_squash: if history == "rebase": merge_msg = message or f"Rebase branch '{safe_branch}' onto {safe_current}" else: merge_msg = message or f"Squash merge branch '{safe_branch}' into {safe_current}" strategy_commit_id = hash_commit( parent_ids=[ours_commit_id], snapshot_id=strategy_snap_id, message=merge_msg, committed_at_iso=committed_at.isoformat(), author=merge_author, ) write_snapshot(root, SnapshotRecord(snapshot_id=strategy_snap_id, manifest=chosen_manifest, directories=chosen_dirs)) write_commit(root, CommitRecord( commit_id=strategy_commit_id, branch=current_branch, snapshot_id=strategy_snap_id, message=merge_msg, committed_at=committed_at, parent_commit_id=ours_commit_id, parent2_commit_id=None, author=merge_author, )) else: merge_msg = message or f"Merge branch '{safe_branch}' (--strategy={strategy})" strategy_commit_id = hash_commit( parent_ids=[ours_commit_id, theirs_commit_id], snapshot_id=strategy_snap_id, message=merge_msg, committed_at_iso=committed_at.isoformat(), author=merge_author, ) write_snapshot(root, SnapshotRecord(snapshot_id=strategy_snap_id, manifest=chosen_manifest, directories=chosen_dirs)) write_commit(root, CommitRecord( commit_id=strategy_commit_id, branch=current_branch, snapshot_id=strategy_snap_id, message=merge_msg, committed_at=committed_at, parent_commit_id=ours_commit_id, parent2_commit_id=theirs_commit_id, author=merge_author, )) try: validate_branch_name(current_branch) except ValueError as exc: print( f"❌ Current branch name is invalid: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) try: write_branch_ref(root, current_branch, strategy_commit_id, expected_id=ours_commit_id) except RefConflictError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) append_reflog( root, current_branch, old_id=ours_commit_id, new_id=strategy_commit_id, author="user", operation=( f"merge (--strategy={strategy}): " f"{sanitize_display(branch)} → {sanitize_display(current_branch)}" ), ) strategy_added, strategy_modified, strategy_deleted = _diff_stats( ours_manifest, chosen_manifest ) if json_out: _s3_files = {"added": strategy_added, "modified": strategy_modified, "deleted": strategy_deleted} print(json.dumps(_MergeJsonBase( **make_envelope(elapsed), status="merged", commit_id=strategy_commit_id, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=[], files_changed=_s3_files, semver_impact="", strategy=reported_strategy, on_conflict=on_conflict, history=history, dry_run=False, merge_result=_MergeResultDict( status="merged", commit_id=strategy_commit_id, strategy=reported_strategy, on_conflict=on_conflict, history=history, conflicts=[], files_changed=_s3_files, semver_impact="", ), ))) else: print( f"✅ Merged '{sanitize_display(branch)}' into " f"'{sanitize_display(current_branch)}' " f"using strategy '{strategy}' → {strategy_commit_id}" ) _print_file_stats(strategy_added, strategy_modified, strategy_deleted) return # ------------------------------------------------------------------- # Case 4: three-way merge # ------------------------------------------------------------------- base_manifest: Manifest = {} if base_commit_id: base_commit = read_commit(root, base_commit_id) if base_commit is None: print( f"❌ Cannot read merge base commit {base_commit_id} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_snap = read_snapshot(root, base_commit.snapshot_id) if base_snap is None: print( f"❌ Cannot read snapshot for merge base {base_commit_id} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_manifest = base_snap.manifest merge_debug_log("merge.case4.enter", { "caller": "muse_merge", "current_branch": current_branch, "source_branch": branch, "base_commit_id": base_commit_id, "ours_commit_id": ours_commit_id, "theirs_commit_id": theirs_commit_id, "base_manifest": merge_debug_manifest_summary(base_manifest), "ours_manifest": merge_debug_manifest_summary(ours_manifest), "theirs_manifest": merge_debug_manifest_summary(theirs_manifest), "strategy": strategy, }) from muse.core.merge_engine import MergeEngine _case4_engine = MergeEngine(diff_unit=_diff_unit, resolution=_resolution) result = run_merge( base_manifest, ours_manifest, theirs_manifest, _case4_engine, plugin=plugin, repo_root=root, domain=domain, ) structured = isinstance(plugin, AddressedMergePlugin) # Capture pre-harmony conflicts and prepare harmony decisions dict for explain. pre_harmony_conflicts: list[str] = list(result.conflicts) harmony_decisions: JsonObject = {} merge_debug_log("merge.case4.result", { "caller": "muse_merge", "is_clean": result.is_clean, "conflicts": result.conflicts, "applied_strategies": result.applied_strategies, "merged_file_count": len(result.merged["files"]), }) if result.applied_strategies and not json_out: for p, strat in sorted(result.applied_strategies.items()): safe_p = sanitize_display(p) safe_strat = sanitize_display(strat) if strat == "dimension-merge": dim_detail = result.dimension_reports.get(p, {}) dim_summary = ", ".join( f"{sanitize_display(d)}={sanitize_display(str(v))}" for d, v in sorted(dim_detail.items()) ) print(f" ✔ dimension-merge: {safe_p} ({dim_summary})") elif strat != "manual": print(f" ✔ [{safe_strat}] {safe_p}") if not result.is_clean: harmony_resolved: Manifest = {} remaining_conflicts = result.conflicts if harmony_autoupdate and not dry_run: harmony_resolved, remaining_conflicts = harmony_auto_apply( root, result.conflicts, ours_manifest, theirs_manifest, domain, plugin, explain_decisions=harmony_decisions if explain else None, ) if not json_out: for p in sorted(harmony_resolved): print(f" ✔ [harmony] auto-resolved: {sanitize_display(p)}") # prefer_ours/prefer_theirs auto-resolves all conflicts — skip conflict reporting. if remaining_conflicts and _resolution in ("prefer_ours", "prefer_theirs"): remaining_conflicts = [] if remaining_conflicts: if not dry_run: # apply_manifest BEFORE write_merge_state — safe crash ordering. # A crash after apply but before write leaves a dirty tree with no # MERGE_STATE.json (recoverable: re-run merge). A crash after # write_merge_state but before apply leaves MERGE_STATE claiming a # conflict while the workdir is still clean — the user cannot resolve. conflict_file_paths: set[str] = { p.split("::")[0] for p in remaining_conflicts } partial_merged: Manifest = dict(ours_manifest) for path, oid in result.merged["files"].items(): if path not in conflict_file_paths: partial_merged[path] = oid apply_manifest( root, ours_manifest, partial_merged, target_directories=directories_from_manifest(partial_merged), ) # Write Cohen Transform conflict markers into each conflicting # file so the user can see both sides and resolve manually. # Binary files are left at the ours version (same policy as checkout). from muse.core.cohen_transform import three_way_merge_lines from muse.core.object_store import read_object from muse.core.io import write_text_atomic def _is_binary(b: bytes) -> bool: return b"\0" in b for cpath in conflict_file_paths: ours_oid = ours_manifest.get(cpath) theirs_oid = theirs_manifest.get(cpath) base_oid = base_manifest.get(cpath) if not ours_oid or not theirs_oid: continue ours_bytes = read_object(root, ours_oid) or b"" theirs_bytes = read_object(root, theirs_oid) or b"" base_bytes = read_object(root, base_oid) if base_oid else b"" if _is_binary(ours_bytes) or _is_binary(theirs_bytes) or _is_binary(base_bytes): continue base_lines = base_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) ours_lines = ours_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) theirs_lines = theirs_bytes.decode("utf-8", errors="replace").splitlines(keepends=True) merged_lines, _ = three_way_merge_lines( base_lines, ours_lines, theirs_lines, label_ours=current_branch, label_base="base", label_theirs=branch, ) dest = root / cpath dest.parent.mkdir(parents=True, exist_ok=True) write_text_atomic(dest, "".join(merged_lines)) write_merge_state( root, base_commit=base_commit_id or "", ours_commit=ours_commit_id, theirs_commit=theirs_commit_id, conflict_paths=remaining_conflicts, other_branch=branch, history=history, ) symbol_conflicts: list[_SymbolConflictDict] = [] conflict_records_out: list[ConflictDict] = [] if result.conflict_records: for rec in result.conflict_records: conflict_records_out.append(rec.to_dict()) if structured and rec.addresses: symbol_conflicts.append({ "path": sanitize_display(rec.path), "symbol": sanitize_display(",".join(rec.addresses)), "ours": sanitize_display(rec.ours_summary), "theirs": sanitize_display(rec.theirs_summary), }) if json_out: _conflict_list = sorted(remaining_conflicts) out = _MergeJson( **make_envelope(elapsed, exit_code=int(ExitCode.USER_ERROR)), status="conflict", commit_id=None, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=_conflict_list, files_changed={"added": 0, "modified": 0, "deleted": 0}, semver_impact="", strategy=strategy, on_conflict=on_conflict, history=history, dry_run=dry_run, merge_result=_MergeResultDict( status="conflict", commit_id=None, strategy=strategy, on_conflict=on_conflict, history=history, conflicts=_conflict_list, files_changed={"added": 0, "modified": 0, "deleted": 0}, semver_impact="", ), ) out["conflict_records"] = conflict_records_out out["symbol_conflicts"] = symbol_conflicts if explain: out["explain"] = _build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=result.merged["files"], pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=(harmony_autoupdate and not dry_run), ) print(json.dumps(out)) else: if dry_run: print(f"{_c('[dry-run]', _CYAN)} Nothing will be written.\n") print( f"❌ {'Would have merge' if dry_run else 'Merge'} " f"conflict in {len(remaining_conflicts)} file(s):", file=sys.stderr, ) for p in sorted(remaining_conflicts): print(f" CONFLICT (both modified): {sanitize_display(p)}", file=sys.stderr) if symbol_conflicts: print( f"\n Symbol-level conflicts ({len(symbol_conflicts)}):", file=sys.stderr, ) for sc in symbol_conflicts[:10]: print(f" {sc['path']}::{sc['symbol']}", file=sys.stderr) if len(symbol_conflicts) > 10: print( f" … and {len(symbol_conflicts) - 10} more", file=sys.stderr, ) if not dry_run: print( '\nFix conflicts and run "muse commit" to complete the merge.', file=sys.stderr, ) if explain: _print_explain_trace(_build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=result.merged["files"], pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=(harmony_autoupdate and not dry_run), )) raise SystemExit(ExitCode.USER_ERROR) if not dry_run: merged_files = dict(result.merged["files"]) # harmony_resolved is keyed by the *original conflict path*, which # for the code domain is a symbol address ("calc.py::add"), not a # plain file path. merged_files is keyed by plain file paths, same # as every manifest in this function. Updating with the raw # symbol-addressed keys silently inserted a bogus extra key while # leaving the real file-path entry untouched at its pre-harmony # value (muse#93) -- the merge reported "auto-resolved" and bumped # applied_count, but the committed content stayed "ours". Multiple # resolved symbols in the same file both restore_object() the # whole file to its own outcome_blob in auto_apply already # (working-tree writes happen sequentially there); re-deriving # file keys here just needs to match whichever write landed last, # not invent new per-symbol merge logic. merged_files.update({p.split("::")[0]: oid for p, oid in harmony_resolved.items()}) result = MergeResult( merged=SnapshotManifest(files=merged_files, domain=domain, directories=directories_from_manifest(merged_files)), conflicts=[], applied_strategies=result.applied_strategies, dimension_reports=result.dimension_reports, op_log=result.op_log, conflict_records=result.conflict_records, ) merged_manifest = result.merged["files"] added, modified, deleted = _diff_stats(ours_manifest, merged_manifest) if dry_run: semver_impact = _semver_from_op_log(result.op_log if structured else []) if json_out: _dr_files = {"added": added, "modified": modified, "deleted": deleted} _dr_envelope = _MergeJsonBase( **make_envelope(elapsed), status="merged", commit_id=None, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=[], files_changed=_dr_files, semver_impact=semver_impact, strategy=reported_strategy, on_conflict=on_conflict, history=history, dry_run=True, merge_result=_MergeResultDict( status="merged", commit_id=None, strategy=reported_strategy, on_conflict=on_conflict, history=history, conflicts=[], files_changed=_dr_files, semver_impact=semver_impact, ), ) if explain: _dr_envelope["explain"] = _build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=result.merged["files"], pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=False, ) print(json.dumps(_dr_envelope)) else: print(f"{_c('[dry-run]', _CYAN)} Nothing will be written.\n") print(f"{_c('Would merge', _BOLD)} by the three-way strategy.") print(f" {sanitize_display(branch)} → {sanitize_display(current_branch)}") _print_file_stats(added, modified, deleted) if semver_impact: print(f" Proposed semver bump: {_c(semver_impact, _YELLOW)}") if explain: _print_explain_trace(_build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=result.merged["files"], pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=False, )) return if not merged_manifest and (ours_manifest or theirs_manifest): print( "❌ Internal error: merge produced an empty manifest despite non-empty " "inputs — aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) merged_dirs = directories_from_manifest(merged_manifest) apply_manifest(root, {**ours_manifest, **theirs_manifest}, merged_manifest, target_directories=merged_dirs) snapshot_id = hash_snapshot(merged_manifest, merged_dirs) committed_at = datetime.datetime.now(datetime.timezone.utc) merge_author = sanitize_provenance(get_config_value("user.handle", root) or "") safe_branch = sanitize_display(branch) safe_current = sanitize_display(current_branch) # --history squash/rebase: single-parent commit, no merge parent # (rebase full commit-by-commit replay is Phase 4 scope) squash = history in ("squash", "rebase") if squash: if history == "rebase": merge_message = message or f"Rebase branch '{safe_branch}' onto {safe_current}" else: merge_message = message or f"Squash merge branch '{safe_branch}' into {safe_current}" commit_id = hash_commit( parent_ids=[ours_commit_id], snapshot_id=snapshot_id, message=merge_message, committed_at_iso=committed_at.isoformat(), author=merge_author, ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=merged_manifest, directories=merged_dirs)) write_commit(root, CommitRecord( commit_id=commit_id, branch=current_branch, snapshot_id=snapshot_id, message=merge_message, committed_at=committed_at, parent_commit_id=ours_commit_id, parent2_commit_id=None, author=merge_author, )) else: merge_message = message or f"Merge branch '{safe_branch}' into {safe_current}" commit_id = hash_commit( parent_ids=[ours_commit_id, theirs_commit_id], snapshot_id=snapshot_id, message=merge_message, committed_at_iso=committed_at.isoformat(), author=merge_author, ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=merged_manifest, directories=merged_dirs)) write_commit(root, CommitRecord( commit_id=commit_id, branch=current_branch, snapshot_id=snapshot_id, message=merge_message, committed_at=committed_at, parent_commit_id=ours_commit_id, parent2_commit_id=theirs_commit_id, author=merge_author, )) try: validate_branch_name(current_branch) except ValueError as exc: print( f"❌ Current branch name is invalid: {sanitize_display(str(exc))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) try: write_branch_ref(root, current_branch, commit_id, expected_id=ours_commit_id) except RefConflictError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) append_reflog( root, current_branch, old_id=ours_commit_id, new_id=commit_id, author="user", operation=f"merge: {sanitize_display(branch)} into {sanitize_display(current_branch)}", ) if json_out: _live_files = {"added": added, "modified": modified, "deleted": deleted} _live_semver = _semver_from_op_log(result.op_log if structured else []) _live_envelope = _MergeJsonBase( **make_envelope(elapsed), status="merged", commit_id=commit_id, branch=branch, current_branch=current_branch, base_commit_id=base_commit_id, conflicts=[], files_changed=_live_files, semver_impact=_live_semver, strategy=strategy, on_conflict=on_conflict, history=history, dry_run=False, merge_result=_MergeResultDict( status="merged", commit_id=commit_id, strategy=strategy, on_conflict=on_conflict, history=history, conflicts=[], files_changed=_live_files, semver_impact=_live_semver, ), ) if explain: _live_envelope["explain"] = _build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=merged_manifest, pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=(harmony_autoupdate and not dry_run), ) print(json.dumps(_live_envelope)) else: print(f"{_c('Merge', _BOLD)} made by the three-way strategy.") print( f" {sanitize_display(branch)} → {sanitize_display(current_branch)}" f" {_c(short_id(commit_id), _YELLOW)}" ) _print_file_stats(added, modified, deleted) if explain: _print_explain_trace(_build_explain_trace( requested_strategy=reported_strategy, on_conflict=on_conflict, history=history, diff_unit=_diff_unit, resolution=_resolution, base_commit_id=base_commit_id, base_manifest=base_manifest, ours_manifest=ours_manifest, theirs_manifest=theirs_manifest, result_manifest=merged_manifest, pre_harmony_conflicts=pre_harmony_conflicts, harmony_decisions=harmony_decisions, harmony_was_run=(harmony_autoupdate and not dry_run), )) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse merge`` subcommand and all its flags.""" parser = subparsers.add_parser( "merge", help="Three-way merge a branch into the current branch.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "branch", nargs="*", default=[], help=( "Branch to merge into the current branch. " "Pass 2 or more for an octopus (N-way) merge — one commit with " "all branches as parents, when their changes don't overlap " "(muse#105). Refuses with a clear error naming the conflicting " "pair if any two branches touch the same path — same as git's " "own octopus strategy, use sequential merges instead." ), ) parser.add_argument( "--no-ff", action="store_true", help="Always create a merge commit, even for fast-forward.", ) parser.add_argument( "-m", "--message", default=None, help="Override the merge commit message.", ) parser.add_argument( "--harmony-autoupdate", action="store_true", default=True, dest="harmony_autoupdate", help="Automatically apply cached harmony resolutions (default: on).", ) parser.add_argument( "--no-harmony-autoupdate", action="store_false", dest="harmony_autoupdate", help="Disable harmony auto-update.", ) parser.add_argument( "--force", "-f", action="store_true", help="Proceed even with uncommitted changes (data-loss risk).", ) parser.add_argument( "--dry-run", "-n", action="store_true", dest="dry_run", help=( "Simulate the merge without writing anything. " "Reports fast-forward, clean merge, or conflicts — including " "symbol-level conflict detail for structured-merge plugins." ), ) parser.add_argument( "--strategy", "-s", choices=["recursive", "overlay", "snapshot", "replay", "ours", "theirs"], default=None, dest="strategy", help=( "Merge strategy (diff_unit). " "'recursive' — three-way delta merge (default); " "'overlay' — apply theirs on top of ours (snapshot, no base); " "'snapshot' — snapshot-level comparison, surface conflicts; " "'replay' — apply ours' delta onto theirs' state; " "'ours' / 'theirs' — convenience aliases for recursive + --on-conflict." ), ) parser.add_argument( "--on-conflict", choices=["escalate", "ours", "theirs"], default=None, dest="on_conflict", help=( "Resolution policy when a conflict is detected. " "'escalate' — surface it for human/Harmony resolution (default); " "'ours' — auto-resolve all conflicts keeping our version; " "'theirs' — auto-resolve all conflicts keeping their version." ), ) parser.add_argument( "--history", choices=["merge", "squash", "rebase"], default=None, dest="history", help=( "History mode. " "'merge' — two-parent merge commit (default); " "'squash' — flatten incoming branch into one commit, no merge parent; " "'rebase' — replay incoming commits linearly on top of ours." ), ) parser.add_argument( "--abort", action="store_true", help="Abort an in-progress merge and restore the working tree.", ) parser.add_argument( "--autoshelf", dest="autoshelf", action="store_true", help=( "Automatically shelf uncommitted changes before merging and pop " "them back onto the working tree after the merge completes. " "Equivalent to running ``muse shelf save`` before merge and " "``muse shelf pop`` after. " "Mutually exclusive with --force." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON.", ) parser.add_argument( "--explain", action="store_true", default=False, dest="explain", help=( "Print a per-path decision trace alongside the merge output. " "With --json, embeds an 'explain' key in the JSON envelope. " "With --dry-run, shows what decisions would be made without writing anything. " "Does not change merge behavior — only adds observation." ), ) parser.set_defaults(func=run, autoshelf=False) def run(args: argparse.Namespace) -> None: """Three-way merge a branch into the current branch. Performs a three-way merge (or fast-forward when possible) and emits a structured result. Pass ``--dry-run`` to simulate without writing. Pass ``--abort`` to cancel an in-progress merge and restore the working tree to the pre-merge state. Agent quickstart ---------------- :: muse merge feat/billing --format json muse merge feat/billing --dry-run --format json muse merge feat/billing --strategy ours --format json muse merge --abort --format json JSON fields ----------- status ``"merged"``, ``"fast_forward"``, ``"conflict"``, or ``"up_to_date"``. commit_id New merge commit ID, or ``null`` on conflict or dry-run. branch Source branch being merged in. current_branch Target branch being merged into. base_commit_id Common ancestor commit ID, or ``null``. conflicts List of paths with unresolved conflicts. files_changed ``{"added": N, "modified": N, "deleted": N}`` file counts. semver_impact Proposed semver bump (``"MAJOR"``, ``"MINOR"``, ``"PATCH"``, or ``""``). strategy Merge strategy used, or ``null``. dry_run ``true`` when ``--dry-run`` was passed. Exit codes ---------- 0 Success — merged, fast-forwarded, or already up to date. 1 Conflict detected, or invalid arguments. 3 Internal error — missing snapshot or commit. """ elapsed = start_timer() abort: bool = getattr(args, "abort", False) json_out: bool = args.json_out if abort: _run_abort(json_out) return branches: list[str] = list(args.branch) if not branches: _emit_error( json_out, "Usage: muse merge [options]. To cancel an in-progress merge: muse merge --abort", ExitCode.USER_ERROR, elapsed, ) if len(branches) >= 2: _run_octopus_merge( root=require_repo(), branches=branches, message=args.message, dry_run=getattr(args, "dry_run", False), json_out=json_out, elapsed=elapsed, ) return branch: str = branches[0] no_ff: bool = args.no_ff message: str | None = args.message harmony_autoupdate: bool = args.harmony_autoupdate force: bool = args.force dry_run: bool = getattr(args, "dry_run", False) strategy: str | None = getattr(args, "strategy", None) on_conflict: str | None = getattr(args, "on_conflict", None) history: str | None = getattr(args, "history", None) autoshelf: bool = getattr(args, "autoshelf", False) explain: bool = getattr(args, "explain", False) if autoshelf and force: _emit_error(json_out, "--autoshelf and --force are mutually exclusive.", ExitCode.USER_ERROR, elapsed) root = require_repo() # Autoshelf: save working-tree changes before the merge so require_clean_workdir # passes, then pop them back in the finally block regardless of outcome. autoshelf_entry = None if autoshelf and not dry_run: current_branch_for_merge = read_current_branch(root) autoshelf_entry = _shelf_push_programmatic( root, name=f"_auto/{sanitize_display(current_branch_for_merge)}", intent_type="interrupt", intent=f"autoshelf before merge of {sanitize_display(branch)}", created_by="muse", ) if autoshelf_entry is not None and not json_out: print("autoshelf: saved working directory state", file=sys.stderr) try: _run_merge( root=root, branch=branch, no_ff=no_ff, message=message, harmony_autoupdate=harmony_autoupdate, force=force, dry_run=dry_run, strategy=strategy, json_out=json_out, on_conflict=on_conflict, history=history, explain=explain, ) finally: if autoshelf_entry is not None: _apply_autoshelf(root, json_out)