"""Addressed-merge plugin — naming correctness and op-type structural tests. The code domain uses address-keyed (Map CRDT) merge semantics, not classical OT. Two operations on different symbol addresses commute automatically; same-address conflicts surface to Harmony. This is a Map CRDT, not an Operational Transformation system. Coverage matrix --------------- A Protocol naming A1 AddressedMergePlugin exists and is importable from muse.domain A2 StructuredMergePlugin does NOT exist in muse.domain (no backward-compat alias) A3 CodePlugin satisfies AddressedMergePlugin via isinstance A4 MidiPlugin satisfies AddressedMergePlugin via isinstance A5 IdentityPlugin satisfies AddressedMergePlugin via isinstance B AddressedInsertOp — no position field B1 AddressedInsertOp is importable from muse.domain B2 AddressedInsertOp instance has keys: op, address, content_id, content_summary B3 AddressedInsertOp instance has NO position key B4 AddressedInsertOp.__required_keys__ does not include position B5 Literal["insert"] is the op value C AddressedDeleteOp — no position field C1 AddressedDeleteOp is importable from muse.domain C2 AddressedDeleteOp instance has keys: op, address, content_id, content_summary C3 AddressedDeleteOp instance has NO position key C4 AddressedDeleteOp.__required_keys__ does not include position C5 Literal["delete"] is the op value D Sequence ops retain position — InsertOp / DeleteOp unchanged D1 InsertOp still has position in its required keys D2 DeleteOp still has position in its required keys D3 InsertOp and AddressedInsertOp are distinct types E Code plugin emits AddressedInsertOp / AddressedDeleteOp (no position in output) E1 CodePlugin.diff() file-level added op has no position key E2 CodePlugin.diff() file-level removed op has no position key E3 muse read --json structured delta ops contain no position key for code commits """ from __future__ import annotations import json import pathlib import pytest # --------------------------------------------------------------------------- # A — Protocol naming # --------------------------------------------------------------------------- class TestProtocolNaming: def test_A1_addressed_merge_plugin_importable(self) -> None: """A1: AddressedMergePlugin exists in muse.domain.""" from muse.domain import AddressedMergePlugin # noqa: F401 def test_A2_structured_merge_plugin_does_not_exist(self) -> None: """A2: StructuredMergePlugin is not exported — no backward-compat alias.""" import muse.domain as domain_module assert not hasattr(domain_module, "StructuredMergePlugin"), ( "StructuredMergePlugin must not exist — use AddressedMergePlugin" ) def test_A3_code_plugin_satisfies_addressed_merge_plugin(self) -> None: """A3: CodePlugin satisfies AddressedMergePlugin at runtime.""" from muse.domain import AddressedMergePlugin from muse.plugins.code.plugin import plugin as code_plugin assert isinstance(code_plugin, AddressedMergePlugin), ( "CodePlugin must satisfy AddressedMergePlugin" ) def test_A4_midi_plugin_satisfies_addressed_merge_plugin(self) -> None: """A4: MidiPlugin satisfies AddressedMergePlugin at runtime.""" from muse.domain import AddressedMergePlugin from muse.plugins.midi.plugin import plugin as midi_plugin assert isinstance(midi_plugin, AddressedMergePlugin), ( "MidiPlugin must satisfy AddressedMergePlugin" ) def test_A5_identity_plugin_satisfies_addressed_merge_plugin(self) -> None: """A5: IdentityPlugin satisfies AddressedMergePlugin at runtime.""" from muse.domain import AddressedMergePlugin from muse.plugins.identity.plugin import IdentityPlugin assert isinstance(IdentityPlugin(), AddressedMergePlugin), ( "IdentityPlugin must satisfy AddressedMergePlugin" ) # --------------------------------------------------------------------------- # B — AddressedInsertOp # --------------------------------------------------------------------------- class TestAddressedInsertOp: def test_B1_importable(self) -> None: """B1: AddressedInsertOp is importable from muse.domain.""" from muse.domain import AddressedInsertOp # noqa: F401 def test_B2_required_keys(self) -> None: """B2: AddressedInsertOp has exactly the expected keys.""" from muse.domain import AddressedInsertOp op = AddressedInsertOp( op="insert", address="src/billing.py::compute_total", content_id="sha256:abc123", content_summary="added compute_total", ) assert set(op.keys()) == {"op", "address", "content_id", "content_summary"} def test_B3_no_position_key_in_instance(self) -> None: """B3: an AddressedInsertOp instance has no 'position' key.""" from muse.domain import AddressedInsertOp op = AddressedInsertOp( op="insert", address="src/billing.py::compute_total", content_id="sha256:abc123", content_summary="added compute_total", ) assert "position" not in op, ( "AddressedInsertOp must not have a position key — " "code symbols are name-addressed, not position-indexed" ) def test_B4_no_position_in_required_keys(self) -> None: """B4: 'position' is absent from AddressedInsertOp.__required_keys__.""" from muse.domain import AddressedInsertOp all_keys = AddressedInsertOp.__required_keys__ | AddressedInsertOp.__optional_keys__ assert "position" not in all_keys, ( "position must not be declared on AddressedInsertOp at all" ) def test_B5_op_literal(self) -> None: """B5: op field value is 'insert'.""" from muse.domain import AddressedInsertOp op = AddressedInsertOp( op="insert", address="src/billing.py::compute_total", content_id="sha256:abc123", content_summary="added compute_total", ) assert op["op"] == "insert" # --------------------------------------------------------------------------- # C — AddressedDeleteOp # --------------------------------------------------------------------------- class TestAddressedDeleteOp: def test_C1_importable(self) -> None: """C1: AddressedDeleteOp is importable from muse.domain.""" from muse.domain import AddressedDeleteOp # noqa: F401 def test_C2_required_keys(self) -> None: """C2: AddressedDeleteOp has exactly the expected keys.""" from muse.domain import AddressedDeleteOp op = AddressedDeleteOp( op="delete", address="src/billing.py::old_fn", content_id="sha256:dead", content_summary="removed old_fn", ) assert set(op.keys()) == {"op", "address", "content_id", "content_summary"} def test_C3_no_position_key_in_instance(self) -> None: """C3: an AddressedDeleteOp instance has no 'position' key.""" from muse.domain import AddressedDeleteOp op = AddressedDeleteOp( op="delete", address="src/billing.py::old_fn", content_id="sha256:dead", content_summary="removed old_fn", ) assert "position" not in op, ( "AddressedDeleteOp must not have a position key" ) def test_C4_no_position_in_required_keys(self) -> None: """C4: 'position' is absent from AddressedDeleteOp type declaration.""" from muse.domain import AddressedDeleteOp all_keys = AddressedDeleteOp.__required_keys__ | AddressedDeleteOp.__optional_keys__ assert "position" not in all_keys def test_C5_op_literal(self) -> None: """C5: op field value is 'delete'.""" from muse.domain import AddressedDeleteOp op = AddressedDeleteOp( op="delete", address="src/billing.py::old_fn", content_id="sha256:dead", content_summary="removed old_fn", ) assert op["op"] == "delete" # --------------------------------------------------------------------------- # D — Sequence ops (InsertOp / DeleteOp) retain position — unchanged # --------------------------------------------------------------------------- class TestSequenceOpsRetainPosition: def test_D1_insert_op_has_position(self) -> None: """D1: InsertOp (for ordered sequences) still has position.""" from muse.domain import InsertOp all_keys = InsertOp.__required_keys__ | InsertOp.__optional_keys__ assert "position" in all_keys, ( "InsertOp is for ordered-sequence domains (MIDI) and must retain position" ) def test_D2_delete_op_has_position(self) -> None: """D2: DeleteOp (for ordered sequences) still has position.""" from muse.domain import DeleteOp all_keys = DeleteOp.__required_keys__ | DeleteOp.__optional_keys__ assert "position" in all_keys, ( "DeleteOp is for ordered-sequence domains (MIDI) and must retain position" ) def test_D3_addressed_and_sequence_insert_are_distinct(self) -> None: """D3: AddressedInsertOp and InsertOp are distinct types.""" from muse.domain import AddressedInsertOp, InsertOp assert AddressedInsertOp is not InsertOp, ( "AddressedInsertOp and InsertOp must be distinct TypedDicts" ) # --------------------------------------------------------------------------- # E — Code plugin emits AddressedInsertOp / AddressedDeleteOp # --------------------------------------------------------------------------- class TestCodePluginAddressedOps: @pytest.fixture() def two_file_repo(self, tmp_path: pathlib.Path) -> tuple[pathlib.Path, dict, dict]: """Two minimal snapshots: base has file_a, target adds file_b, removes file_a.""" from muse.domain import SnapshotManifest base = SnapshotManifest( files={"file_a.py": "sha256:" + "a" * 64}, domain="code", directories=[], ) target = SnapshotManifest( files={"file_b.py": "sha256:" + "b" * 64}, domain="code", directories=[], ) return tmp_path, base, target def test_E1_added_file_op_has_no_position( self, two_file_repo: tuple[pathlib.Path, dict, dict] ) -> None: """E1: code plugin file-level insert op has no 'position' key.""" from muse.plugins.code.plugin import plugin as code_plugin _, base, target = two_file_repo delta = code_plugin.diff(base, target) insert_ops = [o for o in delta["ops"] if o.get("op") == "insert"] assert insert_ops, "expected at least one insert op" for op in insert_ops: assert "position" not in op, ( f"code insert op for '{op.get('address')}' must not have position" ) def test_E2_removed_file_op_has_no_position( self, two_file_repo: tuple[pathlib.Path, dict, dict] ) -> None: """E2: code plugin file-level delete op has no 'position' key.""" from muse.plugins.code.plugin import plugin as code_plugin _, base, target = two_file_repo delta = code_plugin.diff(base, target) delete_ops = [o for o in delta["ops"] if o.get("op") == "delete"] assert delete_ops, "expected at least one delete op" for op in delete_ops: assert "position" not in op, ( f"code delete op for '{op.get('address')}' must not have position" ) def test_E3_muse_read_json_has_no_position_in_code_ops( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """E3: muse read --json for a code commit has no position key in any op.""" import subprocess import sys monkeypatch.chdir(tmp_path) env = {"MUSE_REPO_ROOT": str(tmp_path)} from tests.cli_test_helper import CliRunner runner = CliRunner() def run(*args: str) -> str: result = runner.invoke(None, list(args), env=env) assert result.exit_code == 0, f"{args} failed:\n{result.output}" return result.output.strip() run("init", "--domain", "code") (tmp_path / "hello.py").write_text("def greet(): pass\n") run("code", "add", ".") run("commit", "-m", "initial") (tmp_path / "world.py").write_text("def world(): pass\n") run("code", "add", ".") run("commit", "-m", "add world") output = run("read", "--json") data = json.loads(output) def collect_ops(ops: list[dict]) -> list[dict]: result = [] for op in ops: result.append(op) for child in op.get("child_ops", []): result.append(child) return result all_ops = collect_ops(data.get("structured_delta", {}).get("ops", [])) for op in all_ops: assert "position" not in op, ( f"code-domain op '{op.get('op')} @ {op.get('address')}' " f"must not have position — found: {op}" )