"""TDD tests for Phase 4 — Harmony integration per history mode. Issue #86 Phase 4 deliverables: HA_01: --history merge conflict → resolve → commit → reset → re-merge → Harmony auto-resolves (no conflict) — baseline HA_02: --history squash conflict → resolve → commit → reset → re-merge → Harmony auto-resolves (no conflict) HA_03: --history rebase conflict → resolve → commit → reset → re-merge → Harmony auto-resolves (no conflict) (rebase is squash-equivalent in Phase 3; per-commit Harmony in replay loop is deferred to Phase 6) HA_04: MERGE_STATE.theirs_commit is set for squash merges (guard: ensures the condition at commit.py:676 fires and Harmony actually records) HA_05: Harmony learns from all three history modes and the pattern is present in the store after commit Background ---------- Harmony keying is content-based: blob_fingerprint(ours_object_id, theirs_object_id) uses the *file content hashes* from the ours/theirs snapshots, NOT commit IDs. MERGE_STATE always carries theirs_commit_id regardless of --history mode, so the commit.py guard (merge_state.ours_commit and merge_state.theirs_commit) fires correctly for all three modes. No code change was required — this phase adds coverage to guarantee the behaviour and catch regressions. """ from __future__ import annotations import datetime import json import pathlib import pytest from tests.cli_test_helper import CliRunner from muse.core.types import blob_id, fake_id from muse.core.object_store import write_object, read_object from muse.core.paths import heads_dir, muse_dir, ref_path from muse.core.merge_engine import read_merge_state runner = CliRunner() cli = None # --------------------------------------------------------------------------- # Helpers (same pattern as Phase 2/3) # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> dict: return {"MUSE_REPO_ROOT": str(root)} def _init_repo(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(root: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(root, oid, content) return oid def _make_commit( root: pathlib.Path, repo_id: str, branch: str = "main", message: str = "test", manifest: dict | None = None, parent_id: str | None = None, ) -> str: from muse.core.commits import CommitRecord, write_commit from muse.core.snapshots import SnapshotRecord, write_snapshot from muse.core.ids import hash_snapshot, hash_commit ref_file = ref_path(root, branch) if parent_id is None: parent_id = ref_file.read_text().strip() if ref_file.exists() else None m = manifest or {} snap_id = hash_snapshot(m) committed_at = datetime.datetime.now(datetime.timezone.utc) commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [], snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m)) write_commit(root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=message, committed_at=committed_at, parent_commit_id=parent_id, )) ref_file.parent.mkdir(parents=True, exist_ok=True) ref_file.write_text(commit_id, encoding="utf-8") return commit_id def _checkout(root: pathlib.Path, branch: str, manifest: dict) -> None: """Set HEAD to branch and write manifest file contents to disk.""" (muse_dir(root) / "HEAD").write_text(f"ref: refs/heads/{branch}", encoding="utf-8") for path, oid in manifest.items(): content = read_object(root, oid) if content is not None: dest = root / path dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(content) def _setup_conflict_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str, str, str, str]: """Create a repo where main and feat diverge on config.py. Returns (root, repo_id, ours_commit_id, cfg_ours, cfg_theirs). Both sides use valid Python (single assignment) so the CodePlugin detects a genuine symbol-level conflict without triggering the independence-merge path. """ root, repo_id = _init_repo(tmp_path) cfg_base = _write_obj(root, b"config = 1\n") base_id = _make_commit(root, repo_id, "main", "base", {"config.py": cfg_base}) cfg_ours = _write_obj(root, b"config = 2\n") cfg_theirs = _write_obj(root, b"config = 3\n") (heads_dir(root) / "feat").write_text(base_id, encoding="utf-8") _make_commit(root, repo_id, "feat", "feat changes config", {"config.py": cfg_theirs}, parent_id=base_id) ours_commit_id = _make_commit(root, repo_id, "main", "main changes config", {"config.py": cfg_ours}, parent_id=base_id) _checkout(root, "main", {"config.py": cfg_ours}) return root, repo_id, ours_commit_id, cfg_ours, cfg_theirs def _run_conflict_cycle( root: pathlib.Path, history: str, resolved_content: bytes, ) -> None: """Run a full conflict → resolve → commit cycle for the given history mode. After this function returns: - Harmony has recorded the resolution for the conflict pattern. - MERGE_STATE.json is cleared. - main branch tip is the merge commit. """ # 1. Merge → conflict r = runner.invoke(cli, ["merge", "feat", "--history", history, "--json"], env=_env(root), catch_exceptions=False) data = json.loads(r.output.strip().splitlines()[-1]) assert len(data.get("conflicts", [])) > 0, ( f"Expected a conflict for --history {history}, got {data.get('conflicts')}" ) # 2. Write resolved content to disk resolved_id = _write_obj(root, resolved_content) (root / "config.py").write_bytes(resolved_content) # 3. Mark resolved (stages automatically) r2 = runner.invoke(cli, ["resolve", "config.py", "--json"], env=_env(root), catch_exceptions=False) assert json.loads(r2.output).get("exit_code") == 0, ( f"muse resolve failed: {r2.output}" ) # 4. Commit → Harmony records the pattern r3 = runner.invoke(cli, ["commit", "-m", f"merge: {history} with resolution", "--json"], env=_env(root), catch_exceptions=False) d3 = json.loads(r3.output.strip().splitlines()[-1]) assert d3.get("exit_code") == 0, f"muse commit failed: {r3.output}" # --------------------------------------------------------------------------- # Group 1 — Harmony learns from all three history modes (HA_01–HA_03) # --------------------------------------------------------------------------- class TestHarmonyLearnsPerHistoryMode: """After a conflict is resolved and committed, re-running the same merge must auto-resolve via Harmony — regardless of history mode.""" def _run_full_harmony_test(self, tmp_path: pathlib.Path, history: str) -> None: root, repo_id, ours_commit_id, cfg_ours, cfg_theirs = _setup_conflict_repo(tmp_path) # Teach Harmony: conflict → resolve → commit resolved_content = b"config = 99\n" _run_conflict_cycle(root, history, resolved_content) # Check Harmony learned the pattern from muse.core.harmony.engine import list_patterns patterns = list_patterns(root) assert len(patterns) >= 1, ( f"HA: Harmony must have at least 1 pattern after --history {history} commit" ) # Reset main back to ours_commit_id and restore disk content ref_path(root, "main").write_text(ours_commit_id, encoding="utf-8") ours_content = read_object(root, cfg_ours) assert ours_content is not None (root / "config.py").write_bytes(ours_content) # Re-run the same merge — Harmony must auto-resolve (no conflict) r_final = runner.invoke(cli, ["merge", "feat", "--history", history, "--json"], env=_env(root), catch_exceptions=False) data = json.loads(r_final.output.strip().splitlines()[-1]) assert data.get("conflicts", []) == [], ( f"HA: Harmony must auto-resolve conflict on re-merge with --history {history}; " f"got conflicts={data.get('conflicts')}" ) assert data.get("exit_code") == 0, ( f"HA: re-merge must succeed with --history {history}" ) def test_HA_01_merge_harmony_auto_resolves(self, tmp_path: pathlib.Path) -> None: """HA_01 — --history merge: conflict → resolve → commit → re-merge → Harmony auto-resolves.""" self._run_full_harmony_test(tmp_path, "merge") def test_HA_02_squash_harmony_auto_resolves(self, tmp_path: pathlib.Path) -> None: """HA_02 — --history squash: conflict → resolve → commit → re-merge → Harmony auto-resolves. Squash commits have no parent2_commit_id. Harmony still learns because MERGE_STATE.theirs_commit is set from the merge operation (before the commit), not from the commit record itself. """ self._run_full_harmony_test(tmp_path, "squash") def test_HA_03_rebase_harmony_auto_resolves(self, tmp_path: pathlib.Path) -> None: """HA_03 — --history rebase: conflict → resolve → commit → re-merge → Harmony auto-resolves. In Phase 3, --history rebase produces a single-parent commit (same as squash). Full commit-by-commit replay with per-commit Harmony recording is deferred to Phase 6. This test ensures the squash-equivalent path works correctly. """ self._run_full_harmony_test(tmp_path, "rebase") # --------------------------------------------------------------------------- # Group 2 — Structural guarantees (HA_04–HA_05) # --------------------------------------------------------------------------- class TestHarmonyStructuralGuarantees: """Lower-level checks on the data invariants that make Harmony learning work.""" def test_HA_04_merge_state_theirs_commit_set_for_squash( self, tmp_path: pathlib.Path ) -> None: """HA_04 — MERGE_STATE.theirs_commit is non-None after a squash conflict. This is the guard that makes commit.py:676 fire and Harmony actually record. If theirs_commit were None, the condition would short-circuit and Harmony would silently skip learning. """ root, *_ = _setup_conflict_repo(tmp_path) runner.invoke(cli, ["merge", "feat", "--history", "squash", "--json"], env=_env(root), catch_exceptions=False) ms = read_merge_state(root) assert ms is not None, "HA_04: MERGE_STATE must be written after a squash conflict" assert ms.theirs_commit is not None, ( "HA_04: MERGE_STATE.theirs_commit must be set for squash merges so " "commit.py Harmony recording fires" ) assert ms.ours_commit is not None, ( "HA_04: MERGE_STATE.ours_commit must be set" ) def test_HA_05_harmony_pattern_in_store_after_squash_commit( self, tmp_path: pathlib.Path ) -> None: """HA_05 — After a squash merge commit, a Harmony pattern exists in the store.""" root, *_ = _setup_conflict_repo(tmp_path) _run_conflict_cycle(root, "squash", b"config = 99\n") from muse.core.harmony.engine import list_patterns patterns = list_patterns(root) assert len(patterns) >= 1, ( "HA_05: Harmony must have at least 1 pattern after a squash merge commit" ) p = patterns[0] assert p.path is not None, "HA_05: pattern must have a path" # Pattern should reference the file involved in the conflict assert "config.py" in p.path, ( f"HA_05: pattern path must reference config.py, got {p.path!r}" ) def test_HA_06_harmony_does_not_fire_for_clean_squash_merge( self, tmp_path: pathlib.Path ) -> None: """HA_06 — A clean squash merge (no conflicts) produces no Harmony patterns. Harmony only learns when conflicts are surfaced and resolved. Clean merges that never write MERGE_STATE must not accidentally teach Harmony anything. """ root, repo_id = _init_repo(tmp_path) a_id = _write_obj(root, b"x = 1\n") b_id = _write_obj(root, b"y = 1\n") base_id = _make_commit(root, repo_id, "main", "base", {"a.py": a_id, "b.py": b_id}) b_v2 = _write_obj(root, b"y = 2\n") (heads_dir(root) / "feat").write_text(base_id, encoding="utf-8") _make_commit(root, repo_id, "feat", "feat changes b", {"a.py": a_id, "b.py": b_v2}, parent_id=base_id) _checkout(root, "main", {"a.py": a_id, "b.py": b_id}) r = runner.invoke(cli, ["merge", "feat", "--history", "squash", "--json"], env=_env(root), catch_exceptions=False) data = json.loads(r.output.strip().splitlines()[-1]) assert data.get("exit_code") == 0, "HA_06: clean squash merge must succeed" assert data.get("conflicts", []) == [], "HA_06: clean squash merge must have no conflicts" from muse.core.harmony.engine import list_patterns patterns = list_patterns(root) assert len(patterns) == 0, ( f"HA_06: clean squash merge must not create Harmony patterns, got {len(patterns)}" )