"""``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 } 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) """ from __future__ import annotations import argparse import datetime import json import logging import os import pathlib import sys from muse.core.errors import ExitCode from muse.core.merge_engine import ( apply_merge, detect_conflicts, diff_snapshots, find_merge_base, read_merge_state, write_merge_state, ) from muse.core.merge_hooks import MergeHookResult, get_hook_registry from muse.core.repo import read_repo_id, require_repo from muse.core.rerere import auto_apply as rerere_auto_apply from muse.core.snapshot import compute_commit_id, compute_snapshot_id, directories_from_manifest from muse.core.store import ( CommitRecord, Manifest, SnapshotRecord, get_head_commit_id, get_head_snapshot_manifest, read_commit, read_current_branch, read_snapshot, write_branch_ref, write_commit, write_snapshot, ) from muse.core.reflog import append_reflog from muse.core.validation import sanitize_display, validate_branch_name from muse.core.workdir import apply_manifest from muse.cli.guard import require_clean_workdir from muse.domain import DomainOp, MergeResult, SnapshotManifest, StructuredMergePlugin from muse.plugins.registry import read_domain, resolve_plugin logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 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 _use_color() -> bool: """Return True only when colour output is appropriate. Respects ``NO_COLOR`` (https://no-color.org/), ``TERM=dumb``, and the ``sys.stdout.isatty()`` check so raw ANSI never leaks into pipes or logs. """ if os.environ.get("NO_COLOR") or os.environ.get("TERM") == "dumb": return False return sys.stdout.isatty() 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 _hook_result_to_dict(result: MergeHookResult) -> dict[str, object]: """Render a :class:`MergeHookResult` as a JSON-friendly dict. Used by the ``--consolidate`` flag to embed the hook outcome under the ``consolidation_hook`` key in ``muse merge --format json`` output. Args: result: The :class:`MergeHookResult` to serialise. Returns: A dict with keys ``hook_name``, ``fired``, ``out_of_band``, ``message``, ``error`` — JSON-safe primitive values only. """ return { "hook_name": result.hook_name, "fired": result.fired, "out_of_band": result.out_of_band, "message": result.message, "error": result.error, } def _maybe_run_consolidation_hook( *, consolidate: bool, domain: str, root: pathlib.Path, ours_commit_id: str, theirs_commit_id: str, branch: str, fmt: str, ) -> list[dict[str, object]]: """Fire the registered pre-merge hook for *domain* when --consolidate is set. The hook is non-blocking by contract — see :mod:`muse.core.merge_hooks`. This helper returns immediately after dispatch and never lets a hook exception escape into the merge command. The result is rendered to stdout in text-format runs and returned to the caller as a JSON-friendly list so the caller can embed it in its own JSON output structure. Args: consolidate: ``True`` when the user passed ``--consolidate``. If ``False``, this helper is a no-op and returns ``[]``. domain: Repo domain string from :func:`muse.plugins.registry.read_domain`. Today the hook only fires for ``"knowtation"``; other domains receive no hook even when ``--consolidate`` is requested. root: Repository root directory. ours_commit_id: HEAD of the current branch before merge. theirs_commit_id: HEAD of the branch being merged in. branch: Name of the branch being merged in. fmt: ``"text"`` or ``"json"``. Controls whether the helper prints the hook message to stdout (text mode) or returns silently for JSON-mode embedding. Returns: A list of JSON-shaped hook result dicts (length 0 when no hook fired, otherwise length 1). """ if not consolidate: return [] if domain != "knowtation": return [] try: results = get_hook_registry().run_pre_merge( root, domain, ours_commit_id, theirs_commit_id, branch, consolidate=True, ) except Exception as exc: # noqa: BLE001 — must not block merge logger.warning( "merge: consolidation hook dispatch raised %s: %s", type(exc).__name__, exc, ) return [] rendered = [_hook_result_to_dict(r) for r in results] if fmt == "text": for entry in rendered: mark = "✔" if entry["fired"] and not entry["error"] else ( "✗" if entry["error"] else "·" ) name = sanitize_display(str(entry["hook_name"])) msg = sanitize_display(str(entry["message"])) print(f" {mark} [{name}] {msg}") if entry["error"]: print( f" consolidation hook error: " f"{sanitize_display(str(entry['error']))}", file=sys.stderr, ) return rendered 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_abort(fmt: str) -> 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 fmt == "json": 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[:12]}' 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: apply_manifest(root, snap.manifest) merge_state_path = root / ".muse" / "MERGE_STATE.json" merge_state_path.unlink(missing_ok=True) if fmt == "json": print(json.dumps({ "status": "aborted", "restored_to": ours_commit_id, })) else: print(f"Merge aborted. Working tree restored to {ours_commit_id[:12]}.") 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=None, help="Branch to merge into the current branch.", ) 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( "--rerere-autoupdate", action="store_true", default=True, dest="rerere_autoupdate", help="Automatically apply cached rerere resolutions (default: on).", ) parser.add_argument( "--no-rerere-autoupdate", action="store_false", dest="rerere_autoupdate", help="Disable rerere auto-update.", ) parser.add_argument( "--force", action="store_true", help="Proceed even with uncommitted changes (data-loss risk).", ) parser.add_argument( "--dry-run", 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=["ours", "theirs", "recursive"], default=None, dest="strategy", help=( "Merge strategy. " "'ours' accepts all conflicts by keeping our version; " "'theirs' accepts all conflicts by keeping their version; " "'recursive' (default) performs a three-way merge and stops on conflict." ), ) parser.add_argument( "--abort", action="store_true", help="Abort an in-progress merge and restore the working tree.", ) parser.add_argument( "--consolidate", action="store_true", dest="consolidate", help=( "After the merge completes, fire the registered pre-merge hook " "for the active domain (knowtation only today). The hook runs " "out-of-band — the merge NEVER blocks on its result." ), ) parser.add_argument( "--format", "-f", default="text", dest="fmt", help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Three-way merge a branch into the current branch. All error messages are written to **stderr**; stdout is reserved for structured output only. Agents should pass ``--format json`` to receive a stable, machine-readable result with a consistent schema across all outcomes:: { "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 } Pass ``--dry-run`` to simulate the merge without writing anything. Pass ``--abort`` to cancel an in-progress merge and restore the working tree to the pre-merge state recorded in MERGE_STATE.json. Exit codes:: 0 — success 1 — conflict detected, invalid arguments 3 — internal error (missing snapshot or commit) """ abort: bool = getattr(args, "abort", False) fmt: str = args.fmt if abort: _run_abort(fmt) return branch: str | None = args.branch if branch is None: print("❌ Usage: muse merge [options]", file=sys.stderr) print(" To cancel an in-progress merge: muse merge --abort", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) no_ff: bool = args.no_ff message: str | None = args.message rerere_autoupdate: bool = args.rerere_autoupdate force: bool = args.force dry_run: bool = getattr(args, "dry_run", False) strategy: str | None = getattr(args, "strategy", None) consolidate: bool = getattr(args, "consolidate", False) if fmt not in ("text", "json"): print( f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # Dry-run never touches the working tree; skip the clean-workdir check. if not dry_run: require_clean_workdir(root, "merge", force=force) repo_id = read_repo_id(root) current_branch = read_current_branch(root) domain = read_domain(root) plugin = resolve_plugin(root) if branch == current_branch: print("❌ Cannot merge a branch into itself.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) ours_commit_id = get_head_commit_id(root, current_branch) theirs_commit_id = get_head_commit_id(root, branch) if theirs_commit_id is None: print( f"❌ Branch '{sanitize_display(branch)}' has no commits.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if ours_commit_id is None: print("❌ Current branch has no commits.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) base_commit_id = find_merge_base(root, ours_commit_id, theirs_commit_id) # ----------------------------------------------------------------------- # Case 1: already up to date # ----------------------------------------------------------------------- if base_commit_id == theirs_commit_id: hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": payload: dict[str, object] = { "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": strategy, "dry_run": dry_run, } if hook_records: payload["consolidation_hook"] = hook_records[0] print(json.dumps(payload)) 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[:8]} 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[:8]} 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 # Compute diff stats for display only; ours snapshot failure is non-fatal here. 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, 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) write_branch_ref(root, current_branch, theirs_commit_id) 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)}" ), ) ff_hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": ff_payload: dict[str, object] = { "status": "fast_forward", "commit_id": theirs_commit_id if not dry_run else None, "branch": branch, "current_branch": current_branch, "base_commit_id": base_commit_id, "conflicts": [], "files_changed": {"added": added, "modified": modified, "deleted": deleted}, "semver_impact": "", "strategy": strategy, "dry_run": dry_run, } if ff_hook_records: ff_payload["consolidation_hook"] = ff_hook_records[0] print(json.dumps(ff_payload)) else: if dry_run: print(_c("[dry-run]", _CYAN) + " Nothing will be written.\n") print( f"{_c('Updating', _DIM)} " f"{_c(ours_commit_id[:8], _YELLOW)}.." f"{_c(theirs_commit_id[:8], _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 now — needed by both the strategy shortcuts # (Case 3) and the full three-way merge (Case 4). # Hard fail on None: silently treating an unreadable snapshot as {} causes # apply_merge({},{},{}) → {} → apply_manifest({}) which deletes all files. _ours_manifest = get_head_snapshot_manifest(root, repo_id, 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, repo_id, 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 # ----------------------------------------------------------------------- # Case 3: strategy shortcuts — ours / theirs # ----------------------------------------------------------------------- # --strategy=ours / --strategy=theirs perform a full three-way merge and # then resolve every conflict automatically by taking the chosen side. # Non-conflicting changes from BOTH sides are always applied — only the # conflicting files are decided by the strategy flag. if strategy in ("ours", "theirs") and not dry_run: # Read the merge-base manifest so we can compute per-file change-sets. 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[:8]} " "— 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[:8]} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_manifest_strategy = base_snap_rec.manifest ours_changed_s = diff_snapshots(base_manifest_strategy, ours_manifest) theirs_changed_s = diff_snapshots(base_manifest_strategy, theirs_manifest) conflict_paths_s = detect_conflicts( ours_changed_s, theirs_changed_s, ours_manifest, theirs_manifest ) # Build the merged manifest: non-conflicting changes from both sides. chosen_manifest = apply_merge( base_manifest_strategy, ours_manifest, theirs_manifest, ours_changed_s, theirs_changed_s, conflict_paths_s, ) # Resolve all conflicts by taking the chosen side. side_manifest = ours_manifest if strategy == "ours" else theirs_manifest for path in conflict_paths_s: if path in side_manifest: chosen_manifest[path] = side_manifest[path] else: chosen_manifest.pop(path, None) apply_manifest(root, chosen_manifest) committed_at = datetime.datetime.now(datetime.timezone.utc) # Sanitize branch names before embedding in commit message (stored on disk). safe_branch = sanitize_display(branch) merge_msg = message or f"Merge branch '{safe_branch}' (--strategy={strategy})" chosen_dirs = directories_from_manifest(chosen_manifest) strategy_snap_id = compute_snapshot_id(chosen_manifest, chosen_dirs) strategy_commit_id = compute_commit_id( parent_ids=[ours_commit_id, theirs_commit_id], snapshot_id=strategy_snap_id, message=merge_msg, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=strategy_snap_id, manifest=chosen_manifest, directories=chosen_dirs)) write_commit(root, CommitRecord( commit_id=strategy_commit_id, repo_id=repo_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, )) 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) write_branch_ref(root, current_branch, strategy_commit_id) 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 ) strategy_hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": strategy_payload: dict[str, object] = { "status": "merged", "commit_id": strategy_commit_id, "branch": branch, "current_branch": current_branch, "base_commit_id": base_commit_id, "conflicts": [], "files_changed": { "added": strategy_added, "modified": strategy_modified, "deleted": strategy_deleted, }, "semver_impact": "", "strategy": strategy, "dry_run": False, } if strategy_hook_records: strategy_payload["consolidation_hook"] = strategy_hook_records[0] print(json.dumps(strategy_payload)) else: print( f"✅ Merged '{sanitize_display(branch)}' into " f"'{sanitize_display(current_branch)}' " f"using strategy '{strategy}' → {strategy_commit_id[:8]}" ) _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[:8]} " "— 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[:8]} " "— aborting to prevent data loss.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) base_manifest = base_snap.manifest base_snap_obj = SnapshotManifest(files=base_manifest, domain=domain, directories=directories_from_manifest(base_manifest)) ours_snap_obj = SnapshotManifest(files=ours_manifest, domain=domain, directories=directories_from_manifest(ours_manifest)) theirs_snap_obj = SnapshotManifest(files=theirs_manifest, domain=domain, directories=directories_from_manifest(theirs_manifest)) # Prefer operation-level merge when the plugin supports it. # Produces finer-grained conflict detection (sub-file / note level). structured = isinstance(plugin, StructuredMergePlugin) if structured and isinstance(plugin, StructuredMergePlugin): ours_delta = plugin.diff(base_snap_obj, ours_snap_obj, repo_root=root) theirs_delta = plugin.diff(base_snap_obj, theirs_snap_obj, repo_root=root) result = plugin.merge_ops( base_snap_obj, ours_snap_obj, theirs_snap_obj, ours_delta["ops"], theirs_delta["ops"], repo_root=root, ) logger.debug( "merge: used operation-level merge (%s); %d conflict(s)", type(plugin).__name__, len(result.conflicts), ) else: result = plugin.merge(base_snap_obj, ours_snap_obj, theirs_snap_obj, repo_root=root) # Report any .museattributes auto-resolutions. if result.applied_strategies: 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: rerere_resolved: Manifest = {} remaining_conflicts = result.conflicts # Dry-run: skip rerere (it writes preimage files) but still report conflicts. if rerere_autoupdate and not dry_run: rerere_resolved, remaining_conflicts = rerere_auto_apply( root, result.conflicts, ours_manifest, theirs_manifest, domain, plugin, ) for p in sorted(rerere_resolved): print(f" ✔ [rerere] auto-resolved: {sanitize_display(p)}") if remaining_conflicts: if not dry_run: 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, ) # Apply the non-conflicting portion of the merge to the working # tree NOW — before exiting — so that theirs-only additions (new # files) and ours-only changes are already visible when the agent # runs `muse checkout --ours/--theirs` to resolve conflicts. 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, partial_merged) # Build symbol-level conflict detail for structured plugins. symbol_conflicts: list[dict[str, str]] = [] if structured and result.conflict_records: for rec in result.conflict_records: 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), }) conflict_hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": conflict_payload: dict[str, object] = { "status": "conflict", "commit_id": None, "branch": branch, "current_branch": current_branch, "base_commit_id": base_commit_id, "conflicts": sorted(remaining_conflicts), "symbol_conflicts": symbol_conflicts, "files_changed": {"added": 0, "modified": 0, "deleted": 0}, "semver_impact": "", "strategy": strategy, "dry_run": dry_run, } if conflict_hook_records: conflict_payload["consolidation_hook"] = conflict_hook_records[0] print(json.dumps(conflict_payload)) else: if dry_run: print(_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, ) raise SystemExit(ExitCode.USER_ERROR) if not dry_run: # All conflicts resolved by rerere — rebuild result and fall through # to the clean-merge path so a merge commit is created normally. merged_files = dict(result.merged["files"]) merged_files.update(rerere_resolved) 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) # Dry-run: report what would happen and exit without writing. if dry_run: semver_impact = _semver_from_op_log(result.op_log if structured else []) dry_hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": dry_payload: dict[str, object] = { "status": "merged", "commit_id": None, "branch": branch, "current_branch": current_branch, "base_commit_id": base_commit_id, "conflicts": [], "files_changed": {"added": added, "modified": modified, "deleted": deleted}, "semver_impact": semver_impact, "strategy": strategy, "dry_run": True, } if dry_hook_records: dry_payload["consolidation_hook"] = dry_hook_records[0] print(json.dumps(dry_payload)) else: print(_c("[dry-run]", _CYAN) + " Nothing will be written.\n") print(_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)}") return # Last-resort guard: refuse to apply an empty manifest when the inputs were # non-empty. compute_snapshot_id({}) == SHA-256(b"") — if we ever reach # this point with an empty merged_manifest, something went catastrophically # wrong upstream and we must not delete all tracked files. 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) apply_manifest(root, merged_manifest) merged_dirs = directories_from_manifest(merged_manifest) snapshot_id = compute_snapshot_id(merged_manifest, merged_dirs) committed_at = datetime.datetime.now(datetime.timezone.utc) # Sanitize branch names before embedding in the commit message (stored on disk). safe_branch = sanitize_display(branch) safe_current = sanitize_display(current_branch) merge_message = message or f"Merge branch '{safe_branch}' into {safe_current}" commit_id = compute_commit_id( parent_ids=[ours_commit_id, theirs_commit_id], snapshot_id=snapshot_id, message=merge_message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=merged_manifest, directories=merged_dirs)) write_commit(root, CommitRecord( commit_id=commit_id, repo_id=repo_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, )) 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) write_branch_ref(root, current_branch, commit_id) 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)}", ) final_hook_records = _maybe_run_consolidation_hook( consolidate=consolidate, domain=domain, root=root, ours_commit_id=ours_commit_id, theirs_commit_id=theirs_commit_id, branch=branch, fmt=fmt, ) if fmt == "json": final_payload: dict[str, object] = { "status": "merged", "commit_id": commit_id, "branch": branch, "current_branch": current_branch, "base_commit_id": base_commit_id, "conflicts": [], "files_changed": {"added": added, "modified": modified, "deleted": deleted}, "semver_impact": _semver_from_op_log(result.op_log if structured else []), "strategy": strategy, "dry_run": False, } if final_hook_records: final_payload["consolidation_hook"] = final_hook_records[0] print(json.dumps(final_payload)) else: print(_c("Merge", _BOLD) + " made by the three-way strategy.") print( f" {sanitize_display(branch)} → {sanitize_display(current_branch)}" f" {_c(commit_id[:8], _YELLOW)}" ) _print_file_stats(added, modified, deleted)