gabriel / muse public
test_core_merge_engine.py python
613 lines 25.0 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Tests for muse.core.merge_engine — three-way merge logic.
2
3 Extended to cover the structured (operation-level) merge path via
4 :func:`~muse.core.op_transform.merge_structured` and the
5 :class:`~muse.domain.StructuredMergePlugin` integration.
6 """
7
8 from collections.abc import Mapping
9 import datetime
10 import json
11 import pathlib
12 import unittest.mock
13
14 import pytest
15
16 from muse.core._types import blob_id, long_id
17 from muse.core.merge_engine import (
18 MergeState,
19 apply_merge,
20 clear_merge_state,
21 detect_conflicts,
22 diff_snapshots,
23 find_merge_base,
24 read_merge_state,
25 write_merge_state,
26 )
27 from muse.core.op_transform import MergeOpsResult, merge_op_lists, merge_structured
28 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
29 from muse.core.store import CommitRecord, write_commit
30 from muse.domain import (
31 DeleteOp,
32 DomainOp,
33 InsertOp,
34 ReplaceOp,
35 SnapshotManifest,
36 StructuredDelta,
37 StructuredMergePlugin,
38 )
39 from muse.core.attributes import AttributeRule
40 from muse.plugins.code.plugin import CodePlugin
41 from muse.plugins.midi.plugin import MidiPlugin
42
43 _OID_AAA = blob_id(b"aaa")
44 _OID_OLD = blob_id(b"old")
45 _OID_NEW = blob_id(b"new")
46 _OID_BASE = blob_id(b"base")
47 _OID_OURS = blob_id(b"ours")
48 _OID_THEIRS = blob_id(b"theirs")
49 _OID_K = blob_id(b"k")
50 _OID_FIXED = blob_id(b"fixed")
51
52
53 @pytest.fixture
54 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
55 muse_dir = tmp_path / ".muse"
56 (muse_dir / "commits").mkdir(parents=True)
57 (muse_dir / "refs" / "heads").mkdir(parents=True)
58 return tmp_path
59
60
61 def _commit(root: pathlib.Path, cid: str, parent: str | None = None, parent2: str | None = None) -> str:
62 """Write a commit with a valid content-hash commit_id. Returns the actual commit_id."""
63 snap_id = compute_snapshot_id({})
64 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
65 parent_ids = [p for p in [parent, parent2] if p is not None]
66 commit_id = compute_commit_id(
67 repo_id="r",
68 parent_ids=parent_ids,
69 snapshot_id=snap_id,
70 message=cid,
71 committed_at_iso=committed_at.isoformat(),
72 )
73 write_commit(root, CommitRecord(
74 commit_id=commit_id,
75 repo_id="r",
76 created_on_branch="main",
77 snapshot_id=snap_id,
78 message=cid,
79 committed_at=committed_at,
80 parent_commit_id=parent,
81 parent2_commit_id=parent2,
82 ))
83 return commit_id
84
85
86 class TestDiffSnapshots:
87 def test_no_change(self) -> None:
88 m = {"a.mid": "h1", "b.mid": "h2"}
89 assert diff_snapshots(m, m) == set()
90
91 def test_added(self) -> None:
92 assert diff_snapshots({}, {"a.mid": "h1"}) == {"a.mid"}
93
94 def test_removed(self) -> None:
95 assert diff_snapshots({"a.mid": "h1"}, {}) == {"a.mid"}
96
97 def test_modified(self) -> None:
98 assert diff_snapshots({"a.mid": "old"}, {"a.mid": "new"}) == {"a.mid"}
99
100
101 class TestDetectConflicts:
102 def test_no_conflict_disjoint(self) -> None:
103 ours_m = {"a.mid": "h_a"}
104 theirs_m = {"b.mid": "h_b"}
105 assert detect_conflicts({"a.mid"}, {"b.mid"}, ours_m, theirs_m) == set()
106
107 def test_conflict_divergent_content(self) -> None:
108 ours_m = {"a.mid": "h_a", "b.mid": "h_b_ours"}
109 theirs_m = {"b.mid": "h_b_theirs", "c.mid": "h_c"}
110 assert detect_conflicts({"a.mid", "b.mid"}, {"b.mid", "c.mid"}, ours_m, theirs_m) == {"b.mid"}
111
112 def test_both_empty(self) -> None:
113 assert detect_conflicts(set(), set(), {}, {}) == set()
114
115 def test_convergent_both_delete(self) -> None:
116 """Both branches deleted the same file — convergent, NOT a conflict."""
117 ours_m: Manifest = {} # a.py deleted
118 theirs_m: Manifest = {} # a.py deleted
119 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == set()
120
121 def test_convergent_same_add(self) -> None:
122 """Both branches independently added the same file with identical content."""
123 ours_m = {"new.py": "hash_n"}
124 theirs_m = {"new.py": "hash_n"}
125 assert detect_conflicts({"new.py"}, {"new.py"}, ours_m, theirs_m) == set()
126
127 def test_delete_vs_modify_is_conflict(self) -> None:
128 """One side deleted, other modified — genuinely divergent."""
129 ours_m: Manifest = {} # deleted a.py
130 theirs_m = {"a.py": "hash_new"} # modified a.py
131 assert detect_conflicts({"a.py"}, {"a.py"}, ours_m, theirs_m) == {"a.py"}
132
133
134 class TestApplyMerge:
135 def test_clean_merge(self) -> None:
136 base = {"a.mid": "h0", "b.mid": "h0"}
137 ours = {"a.mid": "h_ours", "b.mid": "h0"}
138 theirs = {"a.mid": "h0", "b.mid": "h_theirs"}
139 ours_changed = {"a.mid"}
140 theirs_changed = {"b.mid"}
141 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, set())
142 assert result == {"a.mid": "h_ours", "b.mid": "h_theirs"}
143
144 def test_conflict_paths_excluded(self) -> None:
145 base = {"a.mid": "h0"}
146 ours = {"a.mid": "h_ours"}
147 theirs = {"a.mid": "h_theirs"}
148 ours_changed = theirs_changed = {"a.mid"}
149 result = apply_merge(base, ours, theirs, ours_changed, theirs_changed, {"a.mid"})
150 assert result == {"a.mid": "h0"} # Falls back to base
151
152 def test_ours_deletion_applied(self) -> None:
153 base = {"a.mid": "h0", "b.mid": "h0"}
154 ours = {"b.mid": "h0"} # a.mid deleted on ours
155 theirs = {"a.mid": "h0", "b.mid": "h0"}
156 result = apply_merge(base, ours, theirs, {"a.mid"}, set(), set())
157 assert "a.mid" not in result
158
159
160 class TestMergeStateIO:
161 def test_write_and_read(self, repo: pathlib.Path) -> None:
162 base_id = long_id("b" * 64)
163 ours_id = long_id("1" * 64)
164 theirs_id = long_id("2" * 64)
165 write_merge_state(
166 repo,
167 base_commit=base_id,
168 ours_commit=ours_id,
169 theirs_commit=theirs_id,
170 conflict_paths=["a.mid", "b.mid"],
171 other_branch="feature/x",
172 )
173 state = read_merge_state(repo)
174 assert state is not None
175 assert state.base_commit == base_id
176 assert state.conflict_paths == ["a.mid", "b.mid"]
177 assert state.other_branch == "feature/x"
178
179 def test_read_no_state(self, repo: pathlib.Path) -> None:
180 assert read_merge_state(repo) is None
181
182 def test_clear(self, repo: pathlib.Path) -> None:
183 write_merge_state(repo, base_commit=long_id("b" * 64), ours_commit=long_id("c" * 64), theirs_commit=long_id("d" * 64), conflict_paths=[])
184 clear_merge_state(repo)
185 assert read_merge_state(repo) is None
186
187
188 class TestFindMergeBase:
189 def test_direct_parent(self, repo: pathlib.Path) -> None:
190 root_id = _commit(repo, "root")
191 a_id = _commit(repo, "a", parent=root_id)
192 b_id = _commit(repo, "b", parent=root_id)
193 base = find_merge_base(repo, a_id, b_id)
194 assert base == root_id
195
196 def test_same_commit(self, repo: pathlib.Path) -> None:
197 _commit(repo, "root")
198 base = find_merge_base(repo, "root", "root")
199 assert base == "root"
200
201 def test_linear_history(self, repo: pathlib.Path) -> None:
202 a_id = _commit(repo, "a")
203 b_id = _commit(repo, "b", parent=a_id)
204 c_id = _commit(repo, "c", parent=b_id)
205 base = find_merge_base(repo, c_id, b_id)
206 assert base == b_id
207
208 def test_no_common_ancestor(self, repo: pathlib.Path) -> None:
209 x_id = _commit(repo, "x")
210 y_id = _commit(repo, "y")
211 assert find_merge_base(repo, x_id, y_id) is None
212
213 def test_bidirectional_terminates_early(self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
214 """Bidirectional BFS reads O(distance_to_LCA) commits, not O(total_history)."""
215 import muse.core.store as store_mod
216
217 # 100-commit chain: root → c0 → ... → c97 → head_a
218 # ↘ feat (branches from c97)
219 # LCA = c97, one hop from each tip
220 root = _commit(repo, "root")
221 tip = root
222 for i in range(97):
223 tip = _commit(repo, f"c{i}", parent=tip)
224 lca = tip
225 head_a = _commit(repo, "head_a", parent=lca)
226 feat = _commit(repo, "feat", parent=lca)
227
228 call_count = 0
229 original = store_mod.read_commit
230
231 def counting_read(rr: pathlib.Path, cid: str):
232 nonlocal call_count
233 call_count += 1
234 return original(rr, cid)
235
236 monkeypatch.setattr(store_mod, "read_commit", counting_read)
237
238 base = find_merge_base(repo, head_a, feat)
239 assert base == lca
240 # Bidirectional BFS finds LCA in ~2 reads (one per tip).
241 # Old two-phase BFS read all 99 A ancestors before touching B.
242 assert call_count <= 10
243
244 def test_deep_chain_diamond(self, repo: pathlib.Path) -> None:
245 """LCA correct for a long chain with diverging feature branches."""
246 root = _commit(repo, "root")
247 tip = root
248 for i in range(50):
249 tip = _commit(repo, f"m{i}", parent=tip)
250 lca = tip
251 branch_a = _commit(repo, "a0", parent=lca)
252 branch_b = _commit(repo, "b0", parent=lca)
253 for i in range(1, 5):
254 branch_a = _commit(repo, f"a{i}", parent=branch_a)
255 branch_b = _commit(repo, f"b{i}", parent=branch_b)
256 base = find_merge_base(repo, branch_a, branch_b)
257 assert base == lca
258
259
260 # ===========================================================================
261 # Structured merge engine integration tests
262 # ===========================================================================
263
264
265 def _ins(addr: str, pos: int | None, cid: str) -> InsertOp:
266 return InsertOp(op="insert", address=addr, position=pos, content_id=cid, content_summary=cid)
267
268
269 def _del(addr: str, pos: int | None, cid: str) -> DeleteOp:
270 return DeleteOp(op="delete", address=addr, position=pos, content_id=cid, content_summary=cid)
271
272
273 def _rep(addr: str, old: str, new: str) -> ReplaceOp:
274 return ReplaceOp(
275 op="replace",
276 address=addr,
277 position=None,
278 old_content_id=old,
279 new_content_id=new,
280 old_summary="old",
281 new_summary="new",
282 )
283
284
285 def _delta(ops: list[DomainOp]) -> StructuredDelta:
286 return StructuredDelta(domain="midi", ops=ops, summary="test")
287
288
289 class TestMergeStructuredIntegration:
290 """Verify merge_structured delegates correctly to merge_op_lists."""
291
292 def test_clean_non_overlapping_file_ops(self) -> None:
293 ours = _delta([_ins("a.mid", pos=0, cid="a-hash")])
294 theirs = _delta([_ins("b.mid", pos=0, cid="b-hash")])
295 result = merge_structured(_delta([]), ours, theirs)
296 assert result.is_clean is True
297 assert len(result.merged_ops) == 2
298
299 def test_conflicting_same_address_replaces_detected(self) -> None:
300 ours = _delta([_rep("shared.mid", "old", "v-ours")])
301 theirs = _delta([_rep("shared.mid", "old", "v-theirs")])
302 result = merge_structured(_delta([]), ours, theirs)
303 assert result.is_clean is False
304 assert len(result.conflict_ops) == 1
305
306 def test_base_ops_kept_by_both_sides_preserved(self) -> None:
307 shared = _ins("base.mid", pos=0, cid="base-cid")
308 result = merge_structured(
309 _delta([shared]),
310 _delta([shared]),
311 _delta([shared]),
312 )
313 assert result.is_clean is True
314 assert any(_op_key_tuple(op) == _op_key_tuple(shared) for op in result.merged_ops)
315
316 def test_position_adjustment_in_structured_merge(self) -> None:
317 """Non-conflicting note inserts get position-adjusted in structured merge."""
318 ours = _delta([_ins("lead.mid", pos=3, cid="note-A")])
319 theirs = _delta([_ins("lead.mid", pos=7, cid="note-B")])
320 result = merge_structured(_delta([]), ours, theirs)
321 assert result.is_clean is True
322 pos_by_cid = {
323 op["content_id"]: op["position"]
324 for op in result.merged_ops
325 if op["op"] == "insert"
326 }
327 # note-A(3): no theirs ≤ 3 → stays 3
328 assert pos_by_cid["note-A"] == 3
329 # note-B(7): ours A(3) ≤ 7 → 7+1 = 8
330 assert pos_by_cid["note-B"] == 8
331
332
333 def _op_key_tuple(op: DomainOp) -> tuple[str, ...]:
334 """Re-implementation of _op_key for test assertions."""
335 if op["op"] == "insert":
336 return ("insert", op["address"], str(op["position"]), op["content_id"])
337 if op["op"] == "delete":
338 return ("delete", op["address"], str(op["position"]), op["content_id"])
339 if op["op"] == "replace":
340 return ("replace", op["address"], str(op["position"]), op["old_content_id"], op["new_content_id"])
341 return (op["op"], op["address"])
342
343
344 class TestStructuredMergePluginProtocol:
345 """Verify MidiPlugin satisfies the StructuredMergePlugin protocol."""
346
347 def test_midi_plugin_isinstance_structured_merge_plugin(self) -> None:
348 plugin = MidiPlugin()
349 assert isinstance(plugin, StructuredMergePlugin)
350
351 def test_merge_ops_non_conflicting_files_is_clean(self) -> None:
352 plugin = MidiPlugin()
353 base = SnapshotManifest(files={}, domain="midi")
354 ours_snap = SnapshotManifest(files={"a.mid": "hash-a"}, domain="midi")
355 theirs_snap = SnapshotManifest(files={"b.mid": "hash-b"}, domain="midi")
356 ours_ops: list[DomainOp] = [_ins("a.mid", pos=None, cid="hash-a")]
357 theirs_ops: list[DomainOp] = [_ins("b.mid", pos=None, cid="hash-b")]
358
359 result = plugin.merge_ops(
360 base, ours_snap, theirs_snap, ours_ops, theirs_ops
361 )
362 assert result.is_clean is True
363 assert "a.mid" in result.merged["files"]
364 assert "b.mid" in result.merged["files"]
365
366 def test_merge_ops_conflicting_same_file_replace_not_clean(self) -> None:
367 plugin = MidiPlugin()
368 base = SnapshotManifest(files={"f.mid": "base-hash"}, domain="midi")
369 ours_snap = SnapshotManifest(files={"f.mid": "ours-hash"}, domain="midi")
370 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-hash"}, domain="midi")
371 ours_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "ours-hash")]
372 theirs_ops: list[DomainOp] = [_rep("f.mid", "base-hash", "theirs-hash")]
373
374 result = plugin.merge_ops(
375 base, ours_snap, theirs_snap, ours_ops, theirs_ops
376 )
377 assert not result.is_clean
378 assert "f.mid" in result.conflicts
379
380 def test_merge_ops_ours_strategy_resolves_conflict(self) -> None:
381 plugin = MidiPlugin()
382 base = SnapshotManifest(files={"f.mid": "base"}, domain="midi")
383 ours_snap = SnapshotManifest(files={"f.mid": "ours-v"}, domain="midi")
384 theirs_snap = SnapshotManifest(files={"f.mid": "theirs-v"}, domain="midi")
385 ours_ops: list[DomainOp] = [_rep("f.mid", "base", "ours-v")]
386 theirs_ops: list[DomainOp] = [_rep("f.mid", "base", "theirs-v")]
387
388 result = plugin.merge_ops(
389 base,
390 ours_snap,
391 theirs_snap,
392 ours_ops,
393 theirs_ops,
394 )
395 # Without .museattributes the conflict stands — verify conflict is reported.
396 assert not result.is_clean
397
398 def test_merge_ops_delete_on_only_one_side_is_clean(self) -> None:
399 plugin = MidiPlugin()
400 base = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
401 ours_snap = SnapshotManifest(files={"keep.mid": "k"}, domain="midi")
402 theirs_snap = SnapshotManifest(files={"keep.mid": "k", "remove.mid": "r"}, domain="midi")
403 ours_ops: list[DomainOp] = [_del("remove.mid", pos=None, cid="r")]
404 theirs_ops: list[DomainOp] = []
405
406 result = plugin.merge_ops(
407 base, ours_snap, theirs_snap, ours_ops, theirs_ops
408 )
409 assert result.is_clean is True
410 assert "keep.mid" in result.merged["files"]
411 assert "remove.mid" not in result.merged["files"]
412
413 def test_merge_ops_empty_changes_returns_base(self) -> None:
414 plugin = MidiPlugin()
415 base = SnapshotManifest(files={"f.mid": "h"}, domain="midi")
416 result = plugin.merge_ops(base, base, base, [], [])
417 assert result.is_clean is True
418 assert result.merged["files"] == {"f.mid": "h"}
419
420
421 # ---------------------------------------------------------------------------
422 # Bug: "manual" attribute strategy on l==r paths causes false conflicts
423 #
424 # The `manual` strategy fires "even when the engine would auto-resolve" a
425 # DIVERGENT change (one side changed, the engine would take it automatically).
426 # It must NOT fire when both sides agree (l == r) — whether nothing changed
427 # (b == l == r) or both made the same convergent edit (b != l == r).
428 #
429 # Regression for: muse merge task/core-cat → 73 false conflicts in muse/core/**
430 # caused by [[rules]] path="muse/core/**" strategy="manual" in .museattributes.
431 # ---------------------------------------------------------------------------
432
433 _DUMMY_ROOT = pathlib.Path("/nonexistent-repo-for-testing")
434
435
436 def _code_plugin() -> CodePlugin:
437 return CodePlugin()
438
439
440 def _snap(files: Mapping[str, str]) -> SnapshotManifest:
441 return SnapshotManifest(files=files, domain="code")
442
443
444 def _manual_attrs():
445 return [AttributeRule(path_pattern="core/**", dimension="*", strategy="manual", priority=100)]
446
447
448 def _merge_with_manual(
449 base: SnapshotManifest,
450 ours: SnapshotManifest,
451 theirs: SnapshotManifest,
452 ) -> MergeResult:
453 plugin = _code_plugin()
454 with unittest.mock.patch(
455 "muse.plugins.code.plugin.load_attributes",
456 return_value=_manual_attrs(),
457 ):
458 return plugin.merge(base, ours, theirs, repo_root=_DUMMY_ROOT)
459
460
461 class TestManualStrategyUnchangedFiles:
462 """manual fires for single-branch changes but NEVER for l == r paths."""
463
464 def test_unchanged_no_conflict(self) -> None:
465 """b == l == r: neither branch touched the file → no conflict."""
466 base = _snap({"core/store.py": _OID_AAA})
467 result = _merge_with_manual(base, base, base)
468 assert result.is_clean
469 assert "core/store.py" not in result.conflicts
470
471 def test_convergent_same_change_no_conflict(self) -> None:
472 """b != l == r: both independently made the same edit → convergent, no conflict."""
473 base = _snap({"core/store.py": _OID_OLD})
474 same = _snap({"core/store.py": _OID_NEW})
475 result = _merge_with_manual(base, same, same)
476 assert result.is_clean
477 assert "core/store.py" not in result.conflicts
478
479 def test_only_ours_changed_manual_forces_conflict(self) -> None:
480 """b == theirs != ours: one side changed → manual forces human review."""
481 base = _snap({"core/store.py": _OID_OLD})
482 ours = _snap({"core/store.py": _OID_NEW})
483 theirs = _snap({"core/store.py": _OID_OLD})
484 result = _merge_with_manual(base, ours, theirs)
485 assert "core/store.py" in result.conflicts
486
487 def test_only_theirs_changed_manual_forces_conflict(self) -> None:
488 """b == ours != theirs: one side changed → manual forces human review."""
489 base = _snap({"core/store.py": _OID_OLD})
490 ours = _snap({"core/store.py": _OID_OLD})
491 theirs = _snap({"core/store.py": _OID_NEW})
492 result = _merge_with_manual(base, ours, theirs)
493 assert "core/store.py" in result.conflicts
494
495 def test_divergent_changes_conflict(self) -> None:
496 """Both changed differently → conflict regardless."""
497 base = _snap({"core/store.py": _OID_BASE})
498 ours = _snap({"core/store.py": _OID_OURS})
499 theirs = _snap({"core/store.py": _OID_THEIRS})
500 result = _merge_with_manual(base, ours, theirs)
501 assert "core/store.py" in result.conflicts
502
503 def test_73_unchanged_plus_one_real_conflict(self) -> None:
504 """Regression: 73 unchanged core files + 1 real conflict → only 1 conflict."""
505 unchanged = {f"core/file_{i}.py": blob_id(f"file_{i}".encode()) for i in range(73)}
506 base_files = {**unchanged, "core/store.py": _OID_BASE}
507 ours_files = {**unchanged, "core/store.py": _OID_OURS}
508 theirs_files = {**unchanged, "core/store.py": _OID_THEIRS}
509 result = _merge_with_manual(
510 _snap(base_files), _snap(ours_files), _snap(theirs_files)
511 )
512 false_conflicts = [p for p in result.conflicts if p != "core/store.py"]
513 assert false_conflicts == [], f"{len(false_conflicts)} false conflicts: {false_conflicts[:5]}"
514 assert "core/store.py" in result.conflicts
515
516
517 # ---------------------------------------------------------------------------
518 # Bug: one-sided changes must NEVER produce false conflicts
519 #
520 # When only one branch changes a file (b == ours or b == theirs), the merge
521 # must take the changed side cleanly — no conflict, no manual review needed.
522 # This covers both the file-level merge() path and the operation-level
523 # merge_ops() path used by the code plugin (StructuredMergePlugin).
524 # ---------------------------------------------------------------------------
525
526
527 class TestOneSidedChangeNeverConflicts:
528 """One side changes a file, other side doesn't → always clean."""
529
530 def test_only_theirs_changed_no_conflict(self) -> None:
531 base = _snap({"pyproject.toml": _OID_OLD, "describe.py": _OID_OLD})
532 ours = _snap({"pyproject.toml": _OID_OLD, "describe.py": _OID_OLD})
533 theirs = _snap({"pyproject.toml": _OID_NEW, "describe.py": _OID_FIXED})
534 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
535 assert result.is_clean
536 assert result.conflicts == []
537 assert result.merged["files"]["pyproject.toml"] == _OID_NEW
538 assert result.merged["files"]["describe.py"] == _OID_FIXED
539
540 def test_only_ours_changed_no_conflict(self) -> None:
541 base = _snap({"a.py": _OID_OLD})
542 ours = _snap({"a.py": _OID_OURS})
543 theirs = _snap({"a.py": _OID_OLD})
544 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
545 assert result.is_clean
546 assert result.conflicts == []
547 assert result.merged["files"]["a.py"] == _OID_OURS
548
549 def test_theirs_deleted_file_ours_untouched(self) -> None:
550 base = _snap({"gone.py": _OID_OLD, "keep.py": _OID_K})
551 ours = _snap({"gone.py": _OID_OLD, "keep.py": _OID_K})
552 theirs = _snap({"keep.py": _OID_K})
553 result = _code_plugin().merge(base, ours, theirs, repo_root=None)
554 assert result.is_clean
555 assert "gone.py" not in result.merged["files"]
556
557 def test_merge_ops_one_sided_no_conflict(self) -> None:
558 """merge_ops() must not flag one-sided changes as conflicts."""
559 plugin = _code_plugin()
560 base = _snap({"pyproject.toml": _OID_OLD})
561 ours = _snap({"pyproject.toml": _OID_OLD})
562 theirs = _snap({"pyproject.toml": _OID_NEW})
563 ours_delta: StructuredDelta = {"ops": [], "summary": "", "domain": "code"}
564 theirs_delta: StructuredDelta = {
565 "ops": [{"op": "patch", "address": "pyproject.toml", "child_ops": [],
566 "file_change": "modified", "content_summary": ""}],
567 "summary": "1 change",
568 "domain": "code",
569 }
570 result = plugin.merge_ops(base, ours, theirs, ours_delta["ops"], theirs_delta["ops"])
571 assert result.conflicts == [], f"False conflicts: {result.conflicts}"
572
573
574 # ---------------------------------------------------------------------------
575 # Bug: merge commit with two parents must pass write_commit hash verification
576 #
577 # compute_commit_id and _verify_commit_id must be symmetric for merge commits
578 # (two parents). If they disagree, write_commit raises ValueError and the
579 # merge cannot be completed — data is permanently stuck.
580 # ---------------------------------------------------------------------------
581
582
583 class TestMergeCommitHashVerification:
584 """write_commit must accept merge commits (two parents) without raising."""
585
586 def test_two_parent_commit_passes_verification(self, repo: pathlib.Path) -> None:
587 parent1 = _commit(repo, "ours")
588 parent2 = _commit(repo, "theirs")
589 # A merge commit with both parents must write cleanly.
590 _commit(repo, "merge", parent=parent1, parent2=parent2)
591
592 def test_merge_commit_id_is_deterministic(self, repo: pathlib.Path) -> None:
593 """Same inputs → same commit_id regardless of parent order in the list."""
594 snap_id = compute_snapshot_id({})
595 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
596 p1 = _commit(repo, "p1")
597 p2 = _commit(repo, "p2")
598 id_ab = compute_commit_id(repo_id="r", parent_ids=[p1, p2], snapshot_id=snap_id, message="merge", committed_at_iso=committed_at.isoformat())
599 id_ba = compute_commit_id(repo_id="r", parent_ids=[p2, p1], snapshot_id=snap_id, message="merge", committed_at_iso=committed_at.isoformat())
600 assert id_ab == id_ba, "Merge commit ID must be order-independent"
601
602 def test_verify_sees_same_id_as_compute(self, repo: pathlib.Path) -> None:
603 """_verify_commit_id (called inside write_commit) must agree with compute_commit_id."""
604 from muse.core.store import read_commit
605 parent1 = _commit(repo, "ours")
606 parent2 = _commit(repo, "theirs")
607 merge_id = _commit(repo, "merge", parent=parent1, parent2=parent2)
608 # If write_commit succeeded, read_commit must return the record intact.
609 rec = read_commit(repo, merge_id)
610 assert rec is not None
611 assert rec.commit_id == merge_id
612 assert rec.parent_commit_id == parent1
613 assert rec.parent2_commit_id == parent2
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 140 days ago