"""``muse coord plan-merge`` — dry-run semantic merge planning. Computes which symbols diverged between two commits using a true **three-way merge** (base + ours + theirs), classifies every conflict by a semantic taxonomy, and recommends a merge strategy — without writing anything to disk. The base commit is the Lowest Common Ancestor (LCA) of ``OURS`` and ``THEIRS``, computed automatically via :func:`~muse.core.merge_engine.find_merge_base`. You can supply ``--base`` to override it (useful for first-commit repos or disconnected branches). When no common ancestor exists and ``--base`` is absent, the analysis falls back to a two-way diff with reduced accuracy. Three-way merge semantics -------------------------- A three-way merge is the foundation of correct conflict detection: * ``base=X, ours=X, theirs=Y`` → only theirs changed → **no conflict** * ``base=X, ours=Y, theirs=X`` → only ours changed → **no conflict** * ``base=X, ours=Y, theirs=Z`` → both changed → **conflict** * ``base=None, ours=Y, theirs=Z`` → both added independently → **conflict** (if content differs) Without a base, the "only one side changed" cases above are indistinguishable from "both changed", producing false positives. Conflict taxonomy ----------------- ``symbol_edit_overlap`` Both branches modified the same symbol. Content diverged — three-way merge required. Confidence: **1.00**. ``rename_edit`` One branch renamed a symbol (same :attr:`body_hash`, new address). The other edited that symbol's body. A text merge would silently corrupt the codebase. Confidence: **0.90**. ``move_edit`` One branch moved a symbol to a different file (same :attr:`body_hash`, different file prefix in address). The other modified the original. Confidence: **0.90**. ``delete_use`` One branch deleted a symbol; the other added **new call sites** for it. Detected by comparing the forward call graphs of base and the surviving branch. Requires the code index — skipped with a warning if unavailable. Confidence: **0.95**. ``dependency_conflict`` Branch A changed symbol X; Branch B's changes include a symbol that transitively calls X. Detected via reverse call graph. Requires the code index. Confidence: **0.75**. ``no_conflict`` Symbol was changed on only one branch, or is identical on both, or was deleted on both. Safe to auto-merge. Usage:: muse coord plan-merge HEAD main muse coord plan-merge feature/billing main muse coord plan-merge feature/billing main --base a1b2c3d4 muse coord plan-merge feature/billing main --skip-call-graph muse coord plan-merge feature/billing main --format json muse coord plan-merge feature/billing main --json JSON output schema:: { "schema_version": str, "ours": str, // full commit ID "theirs": str, // full commit ID "base": str | null, // LCA commit ID, or null "base_auto_computed": bool, "call_graph_available": bool, "call_graph_skipped": bool, "warnings": [str, ...], "total_symbols": int, "conflicts": int, "clean": int, "conflicts_by_type": {"symbol_edit_overlap": int, "rename_edit": int, ...}, "items": [ { "address": str, "conflict_type": str, "ours_change": str, "theirs_change": str, "recommendation": str }, ... ], "elapsed_seconds": float } Exit codes:: 0 — success 1 — one or both refs not found, or bad arguments """ from __future__ import annotations import argparse import json import logging import pathlib import sys import time from typing import TypedDict from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.merge_engine import find_merge_base from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( Manifest, get_commit_snapshot_manifest, read_current_branch, read_commit, resolve_commit_ref, ) from muse.plugins.code._query import symbols_for_snapshot from muse.plugins.code.ast_parser import SymbolRecord from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) type SymbolMap = dict[str, SymbolRecord] # address → symbol record type BodyHashMap = dict[str, list[str]] # body_hash → list of addresses type AddressMap = dict[str, str] # old_address → new_address (renames/moves) type ConflictTypeCount = dict[str, int] # conflict_type → count type MergeItemIndex = dict[str, _MergeItem] # address → merge item # ── Error helper ────────────────────────────────────────────────────────────── def _err(message: str, as_json: bool, code: int) -> None: """Print an error to stdout (JSON) or stderr (text) and exit. Never returns. Args: message: Human-readable error description. as_json: When ``True``, emit compact JSON ``{"error": ..., "status": "error"}`` to *stdout*; otherwise emit ``❌ {message}`` to *stderr*. code: :class:`~muse.core.errors.ExitCode` value to pass to :class:`SystemExit`. """ if as_json: print(json.dumps({"error": message, "status": "error"})) else: print(f"❌ {message}", file=sys.stderr) raise SystemExit(code) # ── Data model ──────────────────────────────────────────────────────────────── class _MergeItemDict(TypedDict): """JSON-serialisable form of a :class:`_MergeItem`.""" address: str conflict_type: str ours_change: str theirs_change: str recommendation: str class _MergeItem: """One symbol's classification in the merge plan. Attributes ---------- address: Full symbol address, e.g. ``"src/billing.py::compute_total"``. conflict_type: One of the taxonomy values: ``"symbol_edit_overlap"``, ``"rename_edit"``, ``"move_edit"``, ``"delete_use"``, ``"dependency_conflict"``, ``"no_conflict"``. ours_change: Human-readable description of how *ours* changed this symbol relative to base (e.g. ``"impl_only"``, ``"renamed to X"``, ``"deleted"``). theirs_change: Same for *theirs*. recommendation: A concrete, actionable suggestion for the person integrating the merge. """ __slots__ = ("address", "conflict_type", "ours_change", "theirs_change", "recommendation") def __init__( self, address: str, conflict_type: str, ours_change: str, theirs_change: str, recommendation: str, ) -> None: self.address = address self.conflict_type = conflict_type self.ours_change = ours_change self.theirs_change = theirs_change self.recommendation = recommendation def to_dict(self) -> _MergeItemDict: return { "address": self.address, "conflict_type": self.conflict_type, "ours_change": self.ours_change, "theirs_change": self.theirs_change, "recommendation": self.recommendation, } # ── Classification helpers ──────────────────────────────────────────────────── def _classify_change(base: SymbolRecord, target: SymbolRecord) -> str: """Describe how *target* differs from *base* for a single symbol. Uses the three orthogonal hash dimensions of :class:`~muse.plugins.code.ast_parser.SymbolRecord`: * ``content_id`` — the full symbol (name + signature + body) is unchanged. * ``body_hash`` — body changed (implementation change); if name also changed it's a rename+modify. * ``signature_id`` — signature (name + args + return type) changed but body did not (signature-only / rename). Returns ------- str One of ``"unchanged"``, ``"signature_only"``, ``"impl_only"``, ``"rename+modify"``, ``"metadata_only"``, ``"full_rewrite"``. """ if base["content_id"] == target["content_id"]: return "unchanged" if base["body_hash"] == target["body_hash"] and base["signature_id"] == target["signature_id"]: # Only metadata changed (decorators, async flag, etc.). return "metadata_only" if base["body_hash"] == target["body_hash"]: # Body unchanged, signature changed → rename or type-annotation tweak. return "signature_only" if base["signature_id"] == target["signature_id"]: # Signature unchanged, body changed → pure implementation edit. return "impl_only" if base["name"] != target["name"]: # Both name and body changed simultaneously. return "rename+modify" return "full_rewrite" def _classify_conflict( addr: str, base: SymbolRecord | None, ours: SymbolRecord | None, theirs: SymbolRecord | None, ) -> _MergeItem: """Classify the merge conflict for a single symbol using three-way semantics. Three-way merge rules --------------------- ``base=X, ours=X, theirs=Y`` → only theirs changed → no_conflict ``base=X, ours=Y, theirs=X`` → only ours changed → no_conflict ``base=X, ours=Y, theirs=Z`` → both changed → symbol_edit_overlap or rename_edit ``base=None, ours=Y, theirs=Z`` → both added new → no_conflict (identical) or overlap ``base=X, ours=None, theirs=None`` → both deleted → no_conflict ``base=X, ours=None, theirs=Y`` → ours deleted only → no_conflict (delete_use handled separately) ``base=X, ours=Y, theirs=None`` → theirs deleted only → no_conflict (delete_use handled separately) Parameters ---------- addr: The full symbol address being classified. base: The symbol record at the merge base commit, or ``None`` if the symbol did not exist yet at the base. ours: The symbol record on the *ours* branch, or ``None`` if deleted/absent. theirs: The symbol record on the *theirs* branch, or ``None`` if deleted/absent. Returns ------- _MergeItem A classified merge item. ``delete_use`` and ``move_edit`` conflicts that require cross-symbol or call-graph analysis are detected in separate passes — this function never returns those types. """ # ── Both absent (should not appear in all_addrs, but be safe) ───────────── if ours is None and theirs is None: return _MergeItem(addr, "no_conflict", "deleted", "deleted", "nothing to merge") # ── Only ours has it (ours added, theirs never had it or deleted) ────────── if ours is not None and theirs is None: if base is None: return _MergeItem(addr, "no_conflict", "added", "absent", "apply ours (insert)") # base exists, theirs deleted it. our_change = _classify_change(base, ours) if our_change == "unchanged": return _MergeItem(addr, "no_conflict", "unchanged", "deleted", "apply theirs (delete)") # Ours modified AND theirs deleted → potential delete_use, handled in separate pass. return _MergeItem(addr, "no_conflict", our_change, "deleted", "review: ours modified, theirs deleted") # ── Only theirs has it ───────────────────────────────────────────────────── if theirs is not None and ours is None: if base is None: return _MergeItem(addr, "no_conflict", "absent", "added", "apply theirs (insert)") their_change = _classify_change(base, theirs) if their_change == "unchanged": return _MergeItem(addr, "no_conflict", "deleted", "unchanged", "apply ours (delete)") return _MergeItem(addr, "no_conflict", "deleted", their_change, "review: theirs modified, ours deleted") # ── Both have it ─────────────────────────────────────────────────────────── assert ours is not None and theirs is not None # Identical on both → no conflict. if ours["content_id"] == theirs["content_id"]: return _MergeItem(addr, "no_conflict", "same", "same", "auto-merge (identical)") if base is not None: our_change = _classify_change(base, ours) their_change = _classify_change(base, theirs) # Only one side changed → no conflict. if our_change == "unchanged": return _MergeItem(addr, "no_conflict", "unchanged", their_change, "apply theirs (fast-forward)") if their_change == "unchanged": return _MergeItem(addr, "no_conflict", our_change, "unchanged", "apply ours (fast-forward)") # Both changed relative to base. # Rename detection: same body hash but different name → signature_only # rename on one side is surfaced as rename_edit. if ( our_change in ("signature_only", "rename+modify") and ours["body_hash"] == base["body_hash"] and ours["name"] != base["name"] ): return _MergeItem(addr, "rename_edit", our_change, their_change, "manual: rename ours, rebase theirs onto new name") if ( their_change in ("signature_only", "rename+modify") and theirs["body_hash"] == base["body_hash"] and theirs["name"] != base["name"] ): return _MergeItem(addr, "rename_edit", our_change, their_change, "manual: rebase ours onto theirs' rename") return _MergeItem(addr, "symbol_edit_overlap", our_change, their_change, "manual: three-way merge required") # No base — both added independently. return _MergeItem(addr, "symbol_edit_overlap", "added", "added", "manual: both branches added this symbol with different content") # ── Rename and move detection ───────────────────────────────────────────────── def _build_body_hash_map(syms: SymbolMap) -> BodyHashMap: """Return a mapping of ``body_hash → [address, ...]`` for *syms*. Enables O(1) lookup of a symbol's new location after a rename or move. Trivial body hashes (empty bodies shared across many symbols) could produce false positives; the caller should validate further. """ result: BodyHashMap = {} for addr, rec in syms.items(): result.setdefault(rec["body_hash"], []).append(addr) return result def _find_renames_and_moves( base_syms: SymbolMap, branch_syms: SymbolMap, ) -> tuple[AddressMap, AddressMap]: """Detect renames and moves between *base_syms* and *branch_syms*. A **rename** is when a symbol's address disappears in *branch_syms* but another address in the same file has the same ``body_hash``. A **move** is when a symbol's address disappears and an address in a *different file* has the same ``body_hash``. Only non-trivial body hashes (symbols with non-empty bodies) are considered, to reduce false positives from boilerplate methods. Parameters ---------- base_syms: Symbols at the merge base. branch_syms: Symbols on the branch being inspected. Returns ------- renames : Manifest Mapping ``{old_address: new_address}`` for renames in this branch. moves : Manifest Mapping ``{old_address: new_address}`` for moves in this branch. """ branch_hash_map = _build_body_hash_map(branch_syms) renames: AddressMap = {} moves: AddressMap = {} for old_addr, base_rec in base_syms.items(): if old_addr in branch_syms: continue # Still present — not a rename/move candidate. body_hash = base_rec["body_hash"] # Skip trivial / empty bodies to avoid false positives. if len(body_hash) < 10: continue candidates = [ a for a in branch_hash_map.get(body_hash, []) if a not in base_syms # Must be new on this branch. ] if not candidates: continue # Prefer the candidate in the same file (rename) over one in another file (move). old_file = old_addr.split("::")[0] same_file = [a for a in candidates if a.split("::")[0] == old_file] other_file = [a for a in candidates if a.split("::")[0] != old_file] if same_file: renames[old_addr] = same_file[0] elif other_file: moves[old_addr] = other_file[0] return renames, moves # ── delete_use detection ────────────────────────────────────────────────────── def _find_delete_use_conflicts( root: pathlib.Path, base_manifest: Manifest, ours_manifest: Manifest, theirs_manifest: Manifest, base_syms: SymbolMap, ours_syms: SymbolMap, theirs_syms: SymbolMap, ) -> tuple[list[_MergeItem], bool, str | None]: """Detect ``delete_use`` conflicts using forward call graphs. A ``delete_use`` conflict occurs when: * One branch **deletes** a symbol that existed at the base, AND * The other branch adds **new call sites** for that same symbol. This is the semantic equivalent of a use-after-free: one side removes a function while the other side starts relying on it more. Algorithm --------- 1. Find all symbols deleted on *ours* that still exist on *theirs*. 2. Find all symbols deleted on *theirs* that still exist on *ours*. 3. Build forward call graphs for base, ours, and theirs. 4. For each deleted symbol, compare the call-site sets in base vs the surviving branch. New call sites (in surviving branch but not in base) = ``delete_use`` conflict. Parameters ---------- root: Repository root. base_manifest, ours_manifest, theirs_manifest: Snapshot manifests for base, ours, and theirs. base_syms, ours_syms, theirs_syms: Already-loaded symbol maps for each snapshot. Returns ------- items : list[_MergeItem] ``delete_use`` conflict items, one per deleted symbol with new callers. call_graph_available : bool ``True`` if the call graph analysis ran; ``False`` if it was skipped. warning : str | None Reason message when the call graph was unavailable. """ from muse.plugins.code._callgraph import build_forward_graph # Symbols deleted on ours (present in base, absent in ours, still in theirs). deleted_on_ours = { addr for addr in base_syms if addr not in ours_syms and addr in theirs_syms } # Symbols deleted on theirs (present in base, absent in theirs, still in ours). deleted_on_theirs = { addr for addr in base_syms if addr not in theirs_syms and addr in ours_syms } if not deleted_on_ours and not deleted_on_theirs: return [], True, None # Nothing to check. try: base_fg = build_forward_graph(root, base_manifest) ours_fg = build_forward_graph(root, ours_manifest) theirs_fg = build_forward_graph(root, theirs_manifest) except (OSError, KeyError, ValueError, AttributeError) as exc: return [], False, f"delete_use analysis skipped — call graph unavailable: {exc}" items: list[_MergeItem] = [] for addr in sorted(deleted_on_ours): bare = addr.split("::")[-1] base_callers = {a for a, callees in base_fg.items() if bare in callees} theirs_callers = {a for a, callees in theirs_fg.items() if bare in callees} new_callers = sorted(theirs_callers - base_callers) if new_callers: caller_preview = ", ".join(new_callers[:3]) if len(new_callers) > 3: caller_preview += f" (+{len(new_callers) - 3} more)" items.append(_MergeItem( addr, "delete_use", "deleted", f"new caller(s): {caller_preview}", "manual: keep the symbol or remove all new call sites on theirs", )) for addr in sorted(deleted_on_theirs): bare = addr.split("::")[-1] base_callers = {a for a, callees in base_fg.items() if bare in callees} ours_callers = {a for a, callees in ours_fg.items() if bare in callees} new_callers = sorted(ours_callers - base_callers) if new_callers: caller_preview = ", ".join(new_callers[:3]) if len(new_callers) > 3: caller_preview += f" (+{len(new_callers) - 3} more)" items.append(_MergeItem( addr, "delete_use", f"new caller(s): {caller_preview}", "deleted", "manual: keep the symbol or remove all new call sites on ours", )) return items, True, None # ── dependency_conflict detection ───────────────────────────────────────────── def _find_dependency_conflicts( root: pathlib.Path, ours_manifest: Manifest, theirs_manifest: Manifest, ours_changed: set[str], theirs_changed: set[str], ) -> tuple[list[_MergeItem], bool, str | None]: """Detect ``dependency_conflict`` via the reverse call graph. A ``dependency_conflict`` occurs when: * Ours changed symbol A, AND * Theirs changed symbol B, AND * B transitively calls A (B depends on A's exact behaviour). Even though A and B were changed on separate branches (no direct overlap), B's semantics depend on A. If both changes merge cleanly at the text level, the result may still be functionally broken. Algorithm --------- 1. Build the reverse call graph for *ours* (callee → callers). 2. For each symbol changed on *ours* (``ours_changed``), find all transitive callers in *ours* up to depth 3. 3. Intersect those callers with ``theirs_changed``. Any intersection indicates that theirs modified something that (transitively) calls what ours modified. Parameters ---------- root: Repository root. ours_manifest, theirs_manifest: Snapshot manifests. ours_changed: Set of symbol addresses changed relative to base on *ours*. theirs_changed: Set of symbol addresses changed relative to base on *theirs*. Returns ------- items : list[_MergeItem] Dependency conflict items. available : bool Whether the call graph was available. warning : str | None Reason if unavailable. """ from muse.plugins.code._callgraph import build_reverse_graph, transitive_callers if not ours_changed or not theirs_changed: return [], True, None try: reverse = build_reverse_graph(root, ours_manifest) except (OSError, KeyError, ValueError, AttributeError) as exc: return [], False, f"dependency_conflict analysis skipped — call graph unavailable: {exc}" items: list[_MergeItem] = [] seen: set[tuple[str, str]] = set() # Deduplicate (ours_addr, theirs_addr) pairs. for our_addr in sorted(ours_changed): bare = our_addr.split("::")[-1] callers_by_depth = transitive_callers(bare, reverse, max_depth=3) all_callers = {a for lvl in callers_by_depth.values() for a in lvl} for their_addr in sorted(theirs_changed): if their_addr in all_callers: key = (our_addr, their_addr) if key not in seen: seen.add(key) items.append(_MergeItem( our_addr, "dependency_conflict", f"changed (caller: {their_addr})", f"depends on {our_addr}", "review: theirs depends on ours — test integration after merge", )) return items, True, None # ── CLI registration ────────────────────────────────────────────────────────── def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``plan-merge`` subcommand on *subparsers* (under ``muse coord``).""" parser = subparsers.add_parser( "plan-merge", help="Dry-run semantic merge planning between two commits.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "ours_ref", metavar="OURS", help="Our commit/branch ref (e.g. HEAD, feature/billing, a1b2c3d4).", ) parser.add_argument( "theirs_ref", metavar="THEIRS", help="Their commit/branch ref (e.g. main, e5f6a7b8).", ) parser.add_argument( "--base", default=None, dest="base_ref", metavar="BASE_REF", help=( "Override the merge base commit/branch. When absent, the base is " "auto-computed as the Lowest Common Ancestor of OURS and THEIRS." ), ) parser.add_argument( "--skip-call-graph", action="store_true", dest="skip_call_graph", help=( "Skip delete_use and dependency_conflict analysis (requires call " "graph index). Use when the index is not built or for faster output." ), ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), 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) # ── Command implementation ──────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Perform a dry-run semantic merge plan between two commits. Analysis passes --------------- **Pass 0 — Base commit resolution** The merge base (LCA) of ``OURS`` and ``THEIRS`` is computed via :func:`~muse.core.merge_engine.find_merge_base`. When ``--base`` is supplied, that commit is used directly. When no common ancestor exists (disconnected branches or first commits), analysis proceeds without a base — conflict detection is less accurate. **Pass 1 — Per-symbol three-way classification** Every symbol address present in ours, theirs, or base is classified using :func:`_classify_conflict` with full three-way semantics. This identifies ``symbol_edit_overlap``, ``rename_edit`` (basic), and marks unilateral deletions for Pass 3. **Pass 2 — Cross-symbol rename and move detection** Uses ``body_hash`` matching to find symbols that were renamed (same file, different name) or moved (different file, same body). Items already classified as ``symbol_edit_overlap`` may be upgraded to ``rename_edit`` or ``move_edit`` when a matching rename/move is found on either branch. **Pass 3 — ``delete_use`` detection** (skipped if ``--skip-call-graph``) Builds forward call graphs for base, ours, and theirs. For each symbol deleted on one branch, checks whether the other branch gained new callers. Skipped with a warning when the call graph is unavailable. **Pass 4 — ``dependency_conflict`` detection** (skipped if ``--skip-call-graph``) Builds the reverse call graph for ours and finds theirs-changed symbols that transitively call ours-changed symbols. Confidence 0.75. Security -------- All user-supplied ref strings are passed through :func:`~muse.core.store.resolve_commit_ref`, which strips glob metacharacters before any path construction. Symbol addresses from persisted records are passed through :func:`~muse.core.validation.sanitize_display` before text output to prevent ANSI injection. Error messages are emitted to *stdout* as compact JSON when ``--format json`` is active, or to *stderr* with an ``❌`` prefix otherwise — never in the reverse order. Performance ----------- Symbol trees are loaded via :func:`~muse.plugins.code._query.symbols_for_snapshot` which uses the persistent symbol cache — three warm-cache loads for a typical repo are sub-100 ms total. Pass 1 is O(V) where V is the union of all symbol addresses. Pass 2 is O(V) with O(1) hash-map lookups. Call graph construction (passes 3 and 4) is the most expensive step; use ``--skip-call-graph`` to bypass it when speed matters more than completeness. Output is compact JSON (no ``indent``); all JSON is single-line so it can be piped directly to ``jq`` or agent parsers without special handling. The command does not modify any files or repository state. Args: args: Parsed ``argparse.Namespace`` with attributes ``ours_ref``, ``theirs_ref``, ``base_ref``, ``skip_call_graph``, and ``fmt``. Exit codes: 0 — success (conflicts present is still success). 1 — a ref was not found or an unexpected error occurred. """ t0 = time.monotonic() ours_ref: str = args.ours_ref theirs_ref: str = args.theirs_ref base_ref: str | None = args.base_ref skip_call_graph: bool = args.skip_call_graph fmt: str = args.fmt as_json = fmt == "json" root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) # ── Resolve OURS and THEIRS ─────────────────────────────────────────────── ours_commit = resolve_commit_ref(root, repo_id, branch, ours_ref) if ours_commit is None: _err(f"ref '{sanitize_display(ours_ref)}' not found", as_json, ExitCode.USER_ERROR) theirs_commit = resolve_commit_ref(root, repo_id, branch, theirs_ref) if theirs_commit is None: _err(f"ref '{sanitize_display(theirs_ref)}' not found", as_json, ExitCode.USER_ERROR) # ── Pass 0: Resolve base commit ─────────────────────────────────────────── base_commit_id: str | None = None base_auto_computed = False warnings: list[str] = [] if base_ref is not None: base_commit = resolve_commit_ref(root, repo_id, branch, base_ref) if base_commit is None: _err(f"--base ref '{sanitize_display(base_ref)}' not found", as_json, ExitCode.USER_ERROR) base_commit_id = base_commit.commit_id else: base_auto_computed = True try: base_commit_id = find_merge_base( root, ours_commit.commit_id, theirs_commit.commit_id ) except Exception as exc: # noqa: BLE001 — merge-base walk can fail on deep/corrupt history warnings.append(f"merge-base auto-detection failed: {exc}") logger.debug("find_merge_base failed: %s", exc) if base_commit_id is None: warnings.append( "no common ancestor found — analysis uses two-way diff (less accurate)" ) # ── Load manifests ──────────────────────────────────────────────────────── ours_manifest = get_commit_snapshot_manifest(root, ours_commit.commit_id) or {} theirs_manifest = get_commit_snapshot_manifest(root, theirs_commit.commit_id) or {} base_manifest: Manifest = {} if base_commit_id is not None: base_manifest = get_commit_snapshot_manifest(root, base_commit_id) or {} # ── Load symbol maps ────────────────────────────────────────────────────── ours_syms: SymbolMap = {} for _fp, tree in symbols_for_snapshot(root, ours_manifest).items(): ours_syms.update(tree) theirs_syms: SymbolMap = {} for _fp, tree in symbols_for_snapshot(root, theirs_manifest).items(): theirs_syms.update(tree) base_syms: SymbolMap = {} if base_manifest: for _fp, tree in symbols_for_snapshot(root, base_manifest).items(): base_syms.update(tree) # Collect all addresses across all three snapshots. all_addrs = sorted(set(ours_syms) | set(theirs_syms) | set(base_syms)) # ── Pass 1: Per-symbol three-way classification ─────────────────────────── items: list[_MergeItem] = [] for addr in all_addrs: item = _classify_conflict( addr, base=base_syms.get(addr), ours=ours_syms.get(addr), theirs=theirs_syms.get(addr), ) items.append(item) # Index items by address for Pass 2 upgrades. item_map: MergeItemIndex = {item.address: item for item in items} # ── Pass 2: Rename and move detection ───────────────────────────────────── if base_syms: ours_renames, ours_moves = _find_renames_and_moves(base_syms, ours_syms) theirs_renames, theirs_moves = _find_renames_and_moves(base_syms, theirs_syms) # Upgrade to rename_edit when one side renamed. # This covers two cases: # 1. Both sides still have fn_old at the same address (symbol_edit_overlap). # 2. Ours deleted fn_old (renamed it away) while theirs modified fn_old (no_conflict # from Pass 1 — delete vs modify). The rename detection reveals the true nature. for old_addr, new_addr in ours_renames.items(): if old_addr in item_map: item = item_map[old_addr] if item.conflict_type == "symbol_edit_overlap": item_map[old_addr] = _MergeItem( old_addr, "rename_edit", f"renamed to {new_addr.split('::')[-1]}", item.theirs_change, "manual: rename ours, rebase theirs onto new name", ) elif ( item.conflict_type == "no_conflict" and old_addr in theirs_syms and old_addr in base_syms and theirs_syms[old_addr]["content_id"] != base_syms[old_addr]["content_id"] ): # Ours renamed fn_old → fn_new; theirs modified fn_old → rename_edit. their_change = _classify_change(base_syms[old_addr], theirs_syms[old_addr]) item_map[old_addr] = _MergeItem( old_addr, "rename_edit", f"renamed to {new_addr.split('::')[-1]}", their_change, "manual: rename ours, rebase theirs onto new name", ) for old_addr, new_addr in theirs_renames.items(): if old_addr in item_map: item = item_map[old_addr] if item.conflict_type == "symbol_edit_overlap": item_map[old_addr] = _MergeItem( old_addr, "rename_edit", item.ours_change, f"renamed to {new_addr.split('::')[-1]}", "manual: rebase ours onto theirs' rename", ) elif ( item.conflict_type == "no_conflict" and old_addr in ours_syms and old_addr in base_syms and ours_syms[old_addr]["content_id"] != base_syms[old_addr]["content_id"] ): # Theirs renamed fn_old → fn_new; ours modified fn_old → rename_edit. our_change = _classify_change(base_syms[old_addr], ours_syms[old_addr]) item_map[old_addr] = _MergeItem( old_addr, "rename_edit", our_change, f"renamed to {new_addr.split('::')[-1]}", "manual: rebase ours onto theirs' rename", ) # Add move_edit items (new items, not upgrades). for old_addr, new_addr in ours_moves.items(): # Only a move_edit if theirs also modified the original address. if old_addr in theirs_syms and old_addr in base_syms: their_change = _classify_change(base_syms[old_addr], theirs_syms[old_addr]) if their_change != "unchanged": old_file = old_addr.split("::")[0] new_file = new_addr.split("::")[0] # Remove the per-symbol no_conflict entry and replace with move_edit. item_map[old_addr] = _MergeItem( old_addr, "move_edit", f"moved to {new_file}", their_change, f"manual: apply theirs' edit to the new location {new_addr}", ) for old_addr, new_addr in theirs_moves.items(): if old_addr in ours_syms and old_addr in base_syms: our_change = _classify_change(base_syms[old_addr], ours_syms[old_addr]) if our_change != "unchanged": new_file = new_addr.split("::")[0] item_map[old_addr] = _MergeItem( old_addr, "move_edit", our_change, f"moved to {new_file}", f"manual: apply ours' edit to the new location {new_addr}", ) # Reconstruct items list from the (possibly upgraded) map. items = [item_map[addr] for addr in all_addrs] # ── Pass 3: delete_use detection ────────────────────────────────────────── call_graph_available = False call_graph_skipped = skip_call_graph if not skip_call_graph and base_syms: du_items, cg_ok, cg_warn = _find_delete_use_conflicts( root, base_manifest, ours_manifest, theirs_manifest, base_syms, ours_syms, theirs_syms, ) call_graph_available = cg_ok if cg_warn: warnings.append(cg_warn) items.extend(du_items) # ── Pass 4: dependency_conflict detection ────────────────────────────────── if not skip_call_graph and base_syms: # Compute changed-symbol sets relative to base. ours_changed = { addr for addr in ours_syms if addr in base_syms and ours_syms[addr]["content_id"] != base_syms[addr]["content_id"] } theirs_changed = { addr for addr in theirs_syms if addr in base_syms and theirs_syms[addr]["content_id"] != base_syms[addr]["content_id"] } dc_items, dc_ok, dc_warn = _find_dependency_conflicts( root, ours_manifest, theirs_manifest, ours_changed, theirs_changed, ) if not call_graph_available and dc_ok: call_graph_available = True if dc_warn and dc_warn not in warnings: warnings.append(dc_warn) items.extend(dc_items) conflicts = [i for i in items if i.conflict_type != "no_conflict"] clean = [i for i in items if i.conflict_type == "no_conflict"] elapsed = round(time.monotonic() - t0, 4) # Build per-type breakdown (useful for agent decision logic). conflicts_by_type: ConflictTypeCount = {} for c in conflicts: conflicts_by_type[c.conflict_type] = conflicts_by_type.get(c.conflict_type, 0) + 1 # ── JSON output ─────────────────────────────────────────────────────────── if as_json: print(json.dumps({ "schema_version": __version__, "ours": ours_commit.commit_id, "theirs": theirs_commit.commit_id, "base": base_commit_id, "base_auto_computed": base_auto_computed, "call_graph_available": call_graph_available, "call_graph_skipped": call_graph_skipped, "warnings": warnings, "total_symbols": len(all_addrs), "conflicts": len(conflicts), "clean": len(clean), "conflicts_by_type": conflicts_by_type, "items": [i.to_dict() for i in conflicts], "elapsed_seconds": elapsed, })) return # ── Text output ─────────────────────────────────────────────────────────── ours_short = ours_commit.commit_id[:8] theirs_short = theirs_commit.commit_id[:8] base_short = base_commit_id[:8] if base_commit_id else "none" print( f"\nSemantic merge plan — {ours_short} ← (merging) {theirs_short}" f" [base: {base_short}]" ) print("─" * 62) if warnings: for w in warnings: print(f"\n ⚠ Note: {w}") if not conflicts: print( f"\n ✅ No conflicts detected" f" ({len(clean)} symbol(s) auto-merge safely)" ) else: _CONFLICT_ICONS = { "symbol_edit_overlap": "🔴", "rename_edit": "⚠️ ", "move_edit": "⚠️ ", "delete_use": "🔴", "dependency_conflict": "🟡", } for item in sorted(conflicts, key=lambda i: i.conflict_type): icon = _CONFLICT_ICONS.get(item.conflict_type, "⚠️ ") addr = sanitize_display(item.address) print(f"\n{icon} {item.conflict_type:<24} {addr}") print(f" ours: {sanitize_display(item.ours_change)}") print(f" theirs: {sanitize_display(item.theirs_change)}") print(f" → {sanitize_display(item.recommendation)}") by_type: ConflictTypeCount = {} for c in conflicts: by_type[c.conflict_type] = by_type.get(c.conflict_type, 0) + 1 summary = ", ".join(f"{n} {t}" for t, n in sorted(by_type.items())) print(f"\n Summary: {len(conflicts)} conflict(s) [{summary}], {len(clean)} clean") print(" Run 'muse coord reconcile' for a detailed integration strategy.") print(f"\n ({elapsed:.3f}s)")