"""TDD tests for Phase 5 — Phantom conflict guard (universal). Audit finding: all four guard locations already have equivalent logic. This file provides direct unit-level proof of each guard so regressions are caught immediately rather than through multi-layer integration failures. Issue #86 Phase 5 deliverables: PG_01: detect_conflicts() — same object ID on both sides → never conflicts PG_02: CodePlugin.merge() — l == r at file level → no conflict produced PG_03: ops_commute() — ReplaceOp convergent (same new_content_id) → True PG_04: CodePlugin.merge_ops() Step 1 — same file content → no symbol conflict PG_05: CodePlugin.merge_ops() Step 1.5 — ours_id == theirs_id excludes path PG_06: musehub merge_overlay — convergent edit → no ConflictEntry PG_07: musehub merge_weave — convergent edit → no ConflictEntry PG_08: musehub merge_replay — convergent edit → no ConflictEntry Background ---------- A phantom conflict occurs when both branches end up with the same object ID for a file but the merge engine incorrectly reports a conflict. The guard ``if ours_object_id == theirs_object_id: skip`` is logically equivalent to ``if l == r:`` (CodePlugin.merge), ``ours_manifest.get(path) != theirs_manifest.get(path)`` (detect_conflicts), and ``frm_id != to_id`` (musehub weave). All four locations must independently enforce this invariant — a single layer failing is enough to corrupt the merged result. """ from __future__ import annotations import datetime import json import pathlib import pytest from muse.core.types import blob_id, fake_id from muse.core.object_store import write_object from muse.core.paths import heads_dir, muse_dir, ref_path # --------------------------------------------------------------------------- # PG_01 — detect_conflicts unit guard # --------------------------------------------------------------------------- class TestDetectConflictsGuard: """detect_conflicts() must never flag a path where both sides agree.""" def test_PG_01_convergent_same_id_not_in_conflicts(self) -> None: """PG_01 — detect_conflicts: same object ID on both sides → not in conflict set. This is the most direct test of the core guard: ours_manifest.get(path) != theirs_manifest.get(path) When both manifests have the same object ID the condition is False and the path is excluded from the returned set. """ from muse.core.merge_engine import detect_conflicts shared_id = "sha256:" + "a" * 64 other_id = "sha256:" + "b" * 64 # Both branches changed 'shared.py' but arrived at the same content. # Only 'real.py' genuinely diverged. ours_changed = {"shared.py", "real.py"} theirs_changed = {"shared.py", "real.py"} ours_manifest = {"shared.py": shared_id, "real.py": "sha256:" + "c" * 64} theirs_manifest = {"shared.py": shared_id, "real.py": "sha256:" + "d" * 64} result = detect_conflicts(ours_changed, theirs_changed, ours_manifest, theirs_manifest) assert "shared.py" not in result, ( "PG_01: detect_conflicts must NOT flag shared.py — both sides have the same object ID" ) assert "real.py" in result, ( "PG_01: detect_conflicts must still flag real.py — genuinely divergent" ) def test_PG_01b_all_convergent_returns_empty(self) -> None: """PG_01b — when ALL changed paths converged to the same ID, result is empty.""" from muse.core.merge_engine import detect_conflicts shared_id = "sha256:" + "e" * 64 ours_changed = theirs_changed = {"a.py", "b.py"} manifest = {"a.py": shared_id, "b.py": shared_id} assert detect_conflicts(ours_changed, theirs_changed, manifest, dict(manifest)) == set(), ( "PG_01b: fully convergent change set must produce empty conflict set" ) # --------------------------------------------------------------------------- # PG_02 — CodePlugin.merge() file-level guard # --------------------------------------------------------------------------- class TestCodePluginMergeGuard: """CodePlugin.merge() l == r at line 845 must not produce conflicts.""" def test_PG_02_same_object_id_not_conflicted(self, tmp_path: pathlib.Path) -> None: """PG_02 — CodePlugin.merge(): when l == r (both sides agree), no conflict is produced. This directly exercises the ``if l == r:`` branch in plugin.merge(). Even when the file differs from base (both branches independently made the same change), the result must be clean. """ from muse.plugins.code.plugin import CodePlugin base_id = "sha256:" + "0" * 64 conv_id = "sha256:" + "1" * 64 base_snap = {"files": {"config.py": base_id}} left_snap = {"files": {"config.py": conv_id}} # left changed it right_snap = {"files": {"config.py": conv_id}} # right changed it to the same plugin = CodePlugin() result = plugin.merge(base_snap, left_snap, right_snap, repo_root=None) assert result.conflicts == [], ( "PG_02: same object ID on both sides must produce no conflicts in CodePlugin.merge()" ) assert result.merged["files"].get("config.py") == conv_id, ( "PG_02: merged snapshot must contain the convergent object ID" ) def test_PG_02b_real_divergence_still_conflicts(self, tmp_path: pathlib.Path) -> None: """PG_02b — verify the guard does NOT suppress genuine divergence.""" from muse.plugins.code.plugin import CodePlugin base_id = "sha256:" + "0" * 64 left_id = "sha256:" + "1" * 64 right_id = "sha256:" + "2" * 64 base_snap = {"files": {"config.py": base_id}} left_snap = {"files": {"config.py": left_id}} right_snap = {"files": {"config.py": right_id}} plugin = CodePlugin() result = plugin.merge(base_snap, left_snap, right_snap, repo_root=None) assert "config.py" in result.conflicts, ( "PG_02b: genuine divergence must still be detected" ) # --------------------------------------------------------------------------- # PG_03 — ops_commute convergent ReplaceOp guard # --------------------------------------------------------------------------- class TestOpsCommuteConvergentGuard: """ops_commute must return True for convergent ReplaceOps at the same address.""" def test_PG_03_replace_same_new_content_id_commutes(self) -> None: """PG_03 — ops_commute: ReplaceOp at same address with same new_content_id → True. This is the Step 1 implicit guard: when both branches produce a ReplaceOp for the same symbol to the same content, ops_commute returns True and the symbol is NOT added to conflict_addresses. """ from muse.core.op_merge import ops_commute conv_id = "sha256:" + "a" * 64 op_a: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": conv_id} op_b: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": conv_id} assert ops_commute(op_a, op_b) is True, ( "PG_03: ReplaceOp at same address with same new_content_id must commute" ) def test_PG_03b_replace_different_new_content_id_conflicts(self) -> None: """PG_03b — genuine symbol divergence: different new_content_id → False (conflict).""" from muse.core.op_merge import ops_commute op_a: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": "sha256:" + "a" * 64} op_b: dict = {"op": "replace", "address": "config.py::MAX_CONN", "new_content_id": "sha256:" + "b" * 64} assert ops_commute(op_a, op_b) is False, ( "PG_03b: different new_content_id at same address must not commute" ) def test_PG_03c_replace_different_address_commutes(self) -> None: """PG_03c — ReplaceOps at different addresses always commute (independent symbols).""" from muse.core.op_merge import ops_commute op_a: dict = {"op": "replace", "address": "config.py::ALPHA", "new_content_id": "sha256:" + "a" * 64} op_b: dict = {"op": "replace", "address": "config.py::BETA", "new_content_id": "sha256:" + "b" * 64} assert ops_commute(op_a, op_b) is True, ( "PG_03c: ReplaceOps at different addresses must always commute" ) # --------------------------------------------------------------------------- # PG_04 — CodePlugin.merge_ops() Step 1 — no symbol conflict for convergent content # --------------------------------------------------------------------------- class TestMergeOpsStep1Guard: """CodePlugin.merge_ops() Step 1 must not produce symbol conflicts for convergent content.""" def _init_repo(self, tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() repo_id = fake_id("repo") (dot_muse / "repo.json").write_text(json.dumps({ "repo_id": repo_id, "domain": "code", "default_branch": "main", "created_at": "2025-01-01T00:00:00+00:00", }), encoding="utf-8") (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "snapshots").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "objects").mkdir() return tmp_path, repo_id def _write_obj(self, root: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(root, oid, content) return oid def test_PG_04_merge_ops_convergent_file_no_symbol_conflict( self, tmp_path: pathlib.Path ) -> None: """PG_04 — merge_ops(): when ours and theirs have the same file object ID, Step 1 produces no symbol conflicts and Step 1.5 skips the path. Both branches independently changed config.py to the same content (convergent). merge_ops must not return any conflict addressing config.py. """ from muse.plugins.code.plugin import CodePlugin from muse.core.types import blob_id as make_id root, _ = self._init_repo(tmp_path) # valid Python for all three: base, convergent result base_content = b"x = 1\n" conv_content = b"x = 99\n" base_id = self._write_obj(root, base_content) conv_id = self._write_obj(root, conv_content) base_snap = {"files": {"config.py": base_id}} ours_snap = {"files": {"config.py": conv_id}} theirs_snap = {"files": {"config.py": conv_id}} # same as ours — convergent plugin = CodePlugin() result = plugin.merge_ops( base_snap, ours_snap, theirs_snap, ours_ops=[], theirs_ops=[], repo_root=root, ) conflict_files = { c.split("::")[0] if "::" in c else c for c in result.conflicts } assert "config.py" not in conflict_files, ( "PG_04: convergent file (same object ID on both sides) must not produce " "any conflict in merge_ops — either Step 1 or Step 1.5 must guard it" ) # --------------------------------------------------------------------------- # PG_05 — CodePlugin.merge_ops() Step 1.5 explicit exclusion # --------------------------------------------------------------------------- class TestMergeOpsStep15ExplicitGuard: """Step 1.5 candidate filter must exclude paths where ours_id == theirs_id.""" def test_PG_05_step15_excludes_convergent_path( self, tmp_path: pathlib.Path ) -> None: """PG_05 — merge_ops Step 1.5: when ours_snap[path] == theirs_snap[path], the path is NOT a candidate for independence merge. The guard at line 1260: ours_snap["files"].get(p) != theirs_snap["files"].get(p) excludes convergent paths from candidate_paths. We verify indirectly: if the path were incorrectly processed, it would either appear in conflicts or in the merged snapshot with unexpected content. The clean assertion is that no conflict is produced and the correct content is in the snapshot. """ import json dot_muse = muse_dir(tmp_path) dot_muse.mkdir() repo_id = fake_id("repo") (dot_muse / "repo.json").write_text(json.dumps({ "repo_id": repo_id, "domain": "code", "default_branch": "main", "created_at": "2025-01-01T00:00:00+00:00", }), encoding="utf-8") (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "snapshots").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "objects").mkdir() # Both ours and theirs changed the same file to the same content. base_content = b"value = 1\n" conv_content = b"value = 99\n" # valid Python — both sides arrived here base_id = blob_id(base_content) conv_id = blob_id(conv_content) write_object(tmp_path, base_id, base_content) write_object(tmp_path, conv_id, conv_content) base_snap = {"files": {"cfg.py": base_id}} ours_snap = {"files": {"cfg.py": conv_id}} theirs_snap = {"files": {"cfg.py": conv_id}} # convergent from muse.plugins.code.plugin import CodePlugin plugin = CodePlugin() result = plugin.merge_ops( base_snap, ours_snap, theirs_snap, ours_ops=[], theirs_ops=[], repo_root=tmp_path, ) assert result.conflicts == [], ( "PG_05: Step 1.5 convergent-file guard must prevent any conflict " "when ours_snap[path] == theirs_snap[path]" ) assert result.merged["files"].get("cfg.py") == conv_id, ( "PG_05: merged snapshot must contain the convergent content ID" ) # --------------------------------------------------------------------------- # PG_06 — musehub merge_overlay phantom guard # --------------------------------------------------------------------------- class TestMusehubOverlayGuard: """musehub merge_overlay must not create ConflictEntry for convergent edits.""" def test_PG_06_overlay_convergent_no_conflict_entry(self) -> None: """PG_06 — merge_overlay: both branches changed a file to the same ID → no conflict. Guard in merge_overlay (lines 206-213): if to_manifest.get(path) != from_manifest.get(path): conflicts.append(...) When both sides have the same object ID the condition is False. """ from musehub.services.proposal_merge_strategies import merge_overlay anc_id = "sha256:" + "0" * 64 conv_id = "sha256:" + "1" * 64 other_id = "sha256:" + "2" * 64 ancestor = {"cfg.py": anc_id, "util.py": anc_id} to_m = {"cfg.py": conv_id, "util.py": other_id} # both changed cfg.py convergently; util differs from_m = {"cfg.py": conv_id, "util.py": anc_id} # from branch only changed cfg.py result = merge_overlay(to_m, from_m, ancestor_manifest=ancestor) conflict_paths = {c.path for c in result.conflicts} assert "cfg.py" not in conflict_paths, ( "PG_06: merge_overlay must not flag cfg.py — both sides have the same object ID" ) def test_PG_06b_overlay_divergent_creates_entry(self) -> None: """PG_06b — genuine divergence still produces a ConflictEntry.""" from musehub.services.proposal_merge_strategies import merge_overlay anc_id = "sha256:" + "0" * 64 to_id = "sha256:" + "1" * 64 frm_id = "sha256:" + "2" * 64 ancestor = {"cfg.py": anc_id} to_m = {"cfg.py": to_id} from_m = {"cfg.py": frm_id} result = merge_overlay(to_m, from_m, ancestor_manifest=ancestor) conflict_paths = {c.path for c in result.conflicts} assert "cfg.py" in conflict_paths, ( "PG_06b: genuine divergence must produce a ConflictEntry in overlay" ) # --------------------------------------------------------------------------- # PG_07 — musehub merge_weave phantom guard # --------------------------------------------------------------------------- class TestMusehubWeaveGuard: """musehub merge_weave must not create ConflictEntry for convergent edits.""" def test_PG_07_weave_convergent_no_conflict_entry(self) -> None: """PG_07 — merge_weave: frm_id != to_id guard excludes convergent paths. Guard in merge_weave (lines 282-290): if frm_changed and to_changed and frm_id != to_id: conflicts.append(...) When both sides arrive at the same ID, frm_id == to_id → no conflict. """ from musehub.services.proposal_merge_strategies import merge_weave anc_id = "sha256:" + "0" * 64 conv_id = "sha256:" + "1" * 64 ancestor = {"cfg.py": anc_id} to_m = {"cfg.py": conv_id} from_m = {"cfg.py": conv_id} # convergent — same as to_m result = merge_weave(to_m, from_m, ancestor_manifest=ancestor) assert result.conflicts == [], ( "PG_07: merge_weave must not flag cfg.py — both sides converged to same ID" ) assert result.manifest.get("cfg.py") == conv_id, ( "PG_07: merged manifest must contain the convergent content" ) def test_PG_07b_weave_divergent_creates_entry(self) -> None: """PG_07b — genuine divergence still produces a ConflictEntry.""" from musehub.services.proposal_merge_strategies import merge_weave anc_id = "sha256:" + "0" * 64 to_id = "sha256:" + "1" * 64 frm_id = "sha256:" + "2" * 64 ancestor = {"cfg.py": anc_id} to_m = {"cfg.py": to_id} from_m = {"cfg.py": frm_id} result = merge_weave(to_m, from_m, ancestor_manifest=ancestor) assert any(c.path == "cfg.py" for c in result.conflicts), ( "PG_07b: genuine divergence must produce a ConflictEntry in weave" ) # --------------------------------------------------------------------------- # PG_08 — musehub merge_replay phantom guard # --------------------------------------------------------------------------- class TestMusehubReplayGuard: """musehub merge_replay must not create ConflictEntry for convergent edits.""" def test_PG_08_replay_convergent_no_conflict_entry(self) -> None: """PG_08 — merge_replay: to_manifest.get(path) != from_manifest[path] guard. Guard in merge_replay (lines 349-357): if to_manifest.get(path) != from_manifest[path]: conflicts.append(...) When both branches independently arrive at the same content, no entry is created. """ from musehub.services.proposal_merge_strategies import merge_replay anc_id = "sha256:" + "0" * 64 conv_id = "sha256:" + "1" * 64 # Both branches changed cfg.py to the same content (convergent). ancestor = {"cfg.py": anc_id} to_m = {"cfg.py": conv_id} from_m = {"cfg.py": conv_id} result = merge_replay(to_m, from_m, ancestor_manifest=ancestor) assert result.conflicts == [], ( "PG_08: merge_replay must not flag cfg.py — both sides have the same object ID" ) def test_PG_08b_replay_divergent_creates_entry(self) -> None: """PG_08b — genuine divergence still produces a ConflictEntry.""" from musehub.services.proposal_merge_strategies import merge_replay anc_id = "sha256:" + "0" * 64 to_id = "sha256:" + "1" * 64 frm_id = "sha256:" + "2" * 64 ancestor = {"cfg.py": anc_id} to_m = {"cfg.py": to_id} from_m = {"cfg.py": frm_id} result = merge_replay(to_m, from_m, ancestor_manifest=ancestor) assert any(c.path == "cfg.py" for c in result.conflicts), ( "PG_08b: genuine divergence must produce a ConflictEntry in replay" )