"""Phase 8 TDD tests — Harmony↔rerere and Shelf↔Stash bridge. NOTE: git subprocess calls in this file are INTENTIONAL — they create real git repositories used as import/export targets. The bridge command itself converts between Muse and git formats. The Muse codebase otherwise never uses git. Tiers ----- Tier 1 Harmony ↔ rerere unit tests Tier 2 Shelf ↔ Stash unit tests Tier 3 Argparse flag presence tests """ from __future__ import annotations import os import pathlib import subprocess import pytest from tests.cli_test_helper import CliRunner runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _invoke(*args: str, cwd: pathlib.Path | None = None) -> "CliRunner": """Invoke the muse CLI from *cwd* (or CWD if None).""" return runner.invoke(None, list(args), cwd=cwd) def _make_muse_repo(path: pathlib.Path) -> pathlib.Path: """Initialise a Muse repository at *path*.""" path.mkdir(parents=True, exist_ok=True) result = _invoke("init", cwd=path) assert result.exit_code == 0, f"muse init failed: {result.stderr}" return path def _make_git_repo(path: pathlib.Path) -> pathlib.Path: """Create a minimal git repo at *path* (single initial commit).""" path.mkdir(parents=True, exist_ok=True) subprocess.run(["git", "init", str(path)], check=True, capture_output=True) subprocess.run( ["git", "-C", str(path), "config", "user.email", "test@example.com"], check=True, capture_output=True, ) subprocess.run( ["git", "-C", str(path), "config", "user.name", "Test User"], check=True, capture_output=True, ) # Initial commit so stash has something to work from. (path / "README.md").write_text("initial") subprocess.run(["git", "-C", str(path), "add", "."], check=True, capture_output=True) subprocess.run( ["git", "-C", str(path), "commit", "-m", "initial"], check=True, capture_output=True, env={**os.environ, "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_AUTHOR_NAME": "Test User", "GIT_COMMITTER_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "Test User"}, ) return path def _make_rr_cache_entry( git_dir: pathlib.Path, name: str, preimage: bytes, postimage: bytes, *, only_preimage: bool = False, ) -> pathlib.Path: """Create a fake rerere cache entry under .git/rr-cache//.""" entry_dir = git_dir / ".git" / "rr-cache" / name entry_dir.mkdir(parents=True, exist_ok=True) (entry_dir / "preimage").write_bytes(preimage) if not only_preimage: (entry_dir / "postimage").write_bytes(postimage) return entry_dir # --------------------------------------------------------------------------- # Tier 1 — Harmony ↔ rerere # --------------------------------------------------------------------------- class TestImportRerereToHarmony: def test_import_rerere_empty_rr_cache_returns_zero( self, tmp_path: pathlib.Path ) -> None: """Empty rr-cache dir → 0 patterns imported, no error.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") # Create an empty rr-cache dir. (git_root / ".git" / "rr-cache").mkdir(parents=True, exist_ok=True) count = import_rerere_to_harmony(muse_root, git_root) assert count == 0 def test_import_rerere_missing_rr_cache_raises( self, tmp_path: pathlib.Path ) -> None: """If rr-cache does not exist at all, FileNotFoundError is raised.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") # rr-cache intentionally absent. with pytest.raises(FileNotFoundError): import_rerere_to_harmony(muse_root, git_root) def test_import_rerere_imports_one_entry( self, tmp_path: pathlib.Path ) -> None: """Single complete rr-cache entry → 1 pattern + 1 resolution created.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony from muse.core.harmony import list_patterns, list_resolutions muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") _make_rr_cache_entry( git_root, name="abc123conflict", preimage=b"<<<<<<< HEAD\nours\n=======\ntheirs\n>>>>>>> feat\n", postimage=b"resolved\n", ) count = import_rerere_to_harmony(muse_root, git_root) assert count == 1 patterns = list_patterns(muse_root) assert len(patterns) == 1 resolutions = list_resolutions(muse_root, patterns[0].pattern_id) assert len(resolutions) == 1 assert resolutions[0].confidence == pytest.approx(0.7) def test_import_rerere_dry_run_writes_nothing( self, tmp_path: pathlib.Path ) -> None: """dry_run=True returns the count but writes no patterns.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony from muse.core.harmony import list_patterns muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") _make_rr_cache_entry( git_root, name="dryrun_entry", preimage=b"conflict content", postimage=b"resolved content", ) count = import_rerere_to_harmony(muse_root, git_root, dry_run=True) assert count == 1 assert list_patterns(muse_root) == [] def test_import_rerere_missing_postimage_skipped( self, tmp_path: pathlib.Path ) -> None: """Entry with only preimage (no postimage) is skipped silently.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") _make_rr_cache_entry( git_root, name="partial_entry", preimage=b"conflict\n", postimage=b"", only_preimage=True, ) count = import_rerere_to_harmony(muse_root, git_root) assert count == 0 def test_import_rerere_custom_confidence( self, tmp_path: pathlib.Path ) -> None: """Custom confidence score is stored on the resolution.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony from muse.core.harmony import list_patterns, list_resolutions muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") _make_rr_cache_entry( git_root, name="highconf_entry", preimage=b"conflict data", postimage=b"resolved data", ) import_rerere_to_harmony(muse_root, git_root, confidence=0.9) patterns = list_patterns(muse_root) resolutions = list_resolutions(muse_root, patterns[0].pattern_id) assert resolutions[0].confidence == pytest.approx(0.9) def test_import_rerere_idempotent( self, tmp_path: pathlib.Path ) -> None: """Importing the same rr-cache entry twice does not create duplicates.""" from muse.core.bridge.harmony_shelf import import_rerere_to_harmony from muse.core.harmony import list_patterns, list_resolutions muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") _make_rr_cache_entry( git_root, name="idem_entry", preimage=b"conflict bytes", postimage=b"resolved bytes", ) import_rerere_to_harmony(muse_root, git_root) import_rerere_to_harmony(muse_root, git_root) patterns = list_patterns(muse_root) assert len(patterns) == 1 resolutions = list_resolutions(muse_root, patterns[0].pattern_id) assert len(resolutions) == 1 class TestExportHarmonyToRerere: def test_export_harmony_no_patterns_returns_zero( self, tmp_path: pathlib.Path ) -> None: """Empty harmony store → 0 exported.""" from muse.core.bridge.harmony_shelf import export_harmony_to_rerere muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") count = export_harmony_to_rerere(muse_root, git_root) assert count == 0 def test_export_harmony_low_confidence_skipped( self, tmp_path: pathlib.Path ) -> None: """Pattern with confidence < 0.8 and not human_verified → not exported.""" from muse.core.bridge.harmony_shelf import export_harmony_to_rerere from muse.core.harmony import ( AgentProvenance, ConflictPattern, ConflictType, Resolution, ResolutionStrategy, blob_fingerprint, compute_pattern_id, compute_resolution_id, record_pattern, save_resolution, ) from muse.core.object_store import write_object from muse.core.types import blob_id, fake_id import datetime muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") preimage_bytes = b"conflict content low conf" postimage_bytes = b"resolved low conf" ours_id = blob_id(preimage_bytes) write_object(muse_root, ours_id, preimage_bytes) outcome_blob = blob_id(postimage_bytes) write_object(muse_root, outcome_blob, postimage_bytes) theirs_id = fake_id("theirs-low") blob_fp = blob_fingerprint(ours_id, theirs_id) pattern_id = compute_pattern_id("src/foo.py", blob_fp, blob_fp) now = datetime.datetime.now(datetime.timezone.utc) pattern = ConflictPattern( pattern_id=pattern_id, path="src/foo.py", domain="code", conflict_type=ConflictType.CONTENT, blob_fingerprint=blob_fp, semantic_fingerprint=blob_fp, ours_id=ours_id, theirs_id=theirs_id, description={}, recorded_at=now, recorded_by="test", ) record_pattern(muse_root, pattern) prov = AgentProvenance.agent("test", "test-model") res_id = compute_resolution_id(pattern_id, outcome_blob, ResolutionStrategy.MANUAL, prov, now) resolution = Resolution( resolution_id=res_id, pattern_id=pattern_id, strategy=ResolutionStrategy.MANUAL, policy_id=None, outcome_blob=outcome_blob, resolved_by=prov, human_verified=False, confidence=0.5, # low — should be skipped rationale="low confidence", resolved_at=now, ) save_resolution(muse_root, resolution) count = export_harmony_to_rerere(muse_root, git_root) assert count == 0 def test_export_harmony_high_confidence_exported( self, tmp_path: pathlib.Path ) -> None: """Pattern with confidence >= 0.8 → written to rr-cache.""" from muse.core.bridge.harmony_shelf import export_harmony_to_rerere from muse.core.harmony import ( AgentProvenance, ConflictPattern, ConflictType, Resolution, ResolutionStrategy, blob_fingerprint, compute_pattern_id, compute_resolution_id, record_pattern, save_resolution, ) from muse.core.object_store import write_object from muse.core.types import blob_id, fake_id import datetime muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") preimage_bytes = b"conflict content high conf" postimage_bytes = b"resolved high conf" ours_id = blob_id(preimage_bytes) write_object(muse_root, ours_id, preimage_bytes) outcome_blob = blob_id(postimage_bytes) write_object(muse_root, outcome_blob, postimage_bytes) theirs_id = fake_id("theirs-high") blob_fp = blob_fingerprint(ours_id, theirs_id) pattern_id = compute_pattern_id("src/bar.py", blob_fp, blob_fp) now = datetime.datetime.now(datetime.timezone.utc) pattern = ConflictPattern( pattern_id=pattern_id, path="src/bar.py", domain="code", conflict_type=ConflictType.CONTENT, blob_fingerprint=blob_fp, semantic_fingerprint=blob_fp, ours_id=ours_id, theirs_id=theirs_id, description={}, recorded_at=now, recorded_by="test", ) record_pattern(muse_root, pattern) prov = AgentProvenance.agent("test", "test-model") res_id = compute_resolution_id(pattern_id, outcome_blob, ResolutionStrategy.MANUAL, prov, now) resolution = Resolution( resolution_id=res_id, pattern_id=pattern_id, strategy=ResolutionStrategy.MANUAL, policy_id=None, outcome_blob=outcome_blob, resolved_by=prov, human_verified=False, confidence=0.9, # high — should be exported rationale="high confidence", resolved_at=now, ) save_resolution(muse_root, resolution) count = export_harmony_to_rerere(muse_root, git_root) assert count == 1 rr_cache = git_root / ".git" / "rr-cache" entries = list(rr_cache.iterdir()) assert len(entries) == 1 assert (entries[0] / "preimage").read_bytes() == preimage_bytes assert (entries[0] / "postimage").read_bytes() == postimage_bytes def test_export_harmony_human_verified_exported_regardless_of_confidence( self, tmp_path: pathlib.Path ) -> None: """human_verified=True bypasses the confidence threshold.""" from muse.core.bridge.harmony_shelf import export_harmony_to_rerere from muse.core.harmony import ( AgentProvenance, ConflictPattern, ConflictType, Resolution, ResolutionStrategy, blob_fingerprint, compute_pattern_id, compute_resolution_id, record_pattern, save_resolution, ) from muse.core.object_store import write_object from muse.core.types import blob_id, fake_id import datetime muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") preimage_bytes = b"human verified pre" postimage_bytes = b"human verified post" ours_id = blob_id(preimage_bytes) write_object(muse_root, ours_id, preimage_bytes) outcome_blob = blob_id(postimage_bytes) write_object(muse_root, outcome_blob, postimage_bytes) theirs_id = fake_id("theirs-human") blob_fp = blob_fingerprint(ours_id, theirs_id) pattern_id = compute_pattern_id("src/baz.py", blob_fp, blob_fp) now = datetime.datetime.now(datetime.timezone.utc) pattern = ConflictPattern( pattern_id=pattern_id, path="src/baz.py", domain="code", conflict_type=ConflictType.CONTENT, blob_fingerprint=blob_fp, semantic_fingerprint=blob_fp, ours_id=ours_id, theirs_id=theirs_id, description={}, recorded_at=now, recorded_by="test", ) record_pattern(muse_root, pattern) prov = AgentProvenance.human() res_id = compute_resolution_id(pattern_id, outcome_blob, ResolutionStrategy.MANUAL, prov, now) resolution = Resolution( resolution_id=res_id, pattern_id=pattern_id, strategy=ResolutionStrategy.MANUAL, policy_id=None, outcome_blob=outcome_blob, resolved_by=prov, human_verified=True, confidence=0.3, # low confidence but human_verified → should export rationale="human verified", resolved_at=now, ) save_resolution(muse_root, resolution) count = export_harmony_to_rerere(muse_root, git_root) assert count == 1 def test_export_harmony_dry_run_writes_nothing( self, tmp_path: pathlib.Path ) -> None: """dry_run=True returns count but creates no rr-cache entries.""" from muse.core.bridge.harmony_shelf import export_harmony_to_rerere from muse.core.harmony import ( AgentProvenance, ConflictPattern, ConflictType, Resolution, ResolutionStrategy, blob_fingerprint, compute_pattern_id, compute_resolution_id, record_pattern, save_resolution, ) from muse.core.object_store import write_object from muse.core.types import blob_id, fake_id import datetime muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") preimage_bytes = b"dry run pre" postimage_bytes = b"dry run post" ours_id = blob_id(preimage_bytes) write_object(muse_root, ours_id, preimage_bytes) outcome_blob = blob_id(postimage_bytes) write_object(muse_root, outcome_blob, postimage_bytes) theirs_id = fake_id("theirs-dry") blob_fp = blob_fingerprint(ours_id, theirs_id) pattern_id = compute_pattern_id("src/dry.py", blob_fp, blob_fp) now = datetime.datetime.now(datetime.timezone.utc) pattern = ConflictPattern( pattern_id=pattern_id, path="src/dry.py", domain="code", conflict_type=ConflictType.CONTENT, blob_fingerprint=blob_fp, semantic_fingerprint=blob_fp, ours_id=ours_id, theirs_id=theirs_id, description={}, recorded_at=now, recorded_by="test", ) record_pattern(muse_root, pattern) prov = AgentProvenance.agent("test", "test-model") res_id = compute_resolution_id(pattern_id, outcome_blob, ResolutionStrategy.MANUAL, prov, now) resolution = Resolution( resolution_id=res_id, pattern_id=pattern_id, strategy=ResolutionStrategy.MANUAL, policy_id=None, outcome_blob=outcome_blob, resolved_by=prov, human_verified=False, confidence=0.95, rationale="high conf dry run", resolved_at=now, ) save_resolution(muse_root, resolution) count = export_harmony_to_rerere(muse_root, git_root, dry_run=True) assert count == 1 rr_cache = git_root / ".git" / "rr-cache" assert not rr_cache.exists() or len(list(rr_cache.iterdir())) == 0 # --------------------------------------------------------------------------- # Tier 2 — Shelf ↔ Stash # --------------------------------------------------------------------------- class TestImportStashesToShelf: def test_import_stashes_no_stashes_returns_zero( self, tmp_path: pathlib.Path ) -> None: """No git stashes → 0 shelf entries created.""" from muse.core.bridge.harmony_shelf import import_stashes_to_shelf from muse.core.shelf import list_shelf_entries muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") count = import_stashes_to_shelf(muse_root, git_root) assert count == 0 assert list_shelf_entries(muse_root) == [] def test_import_one_stash_creates_shelf_entry( self, tmp_path: pathlib.Path ) -> None: """One git stash → one Muse shelf entry with intent_type='handoff'.""" from muse.core.bridge.harmony_shelf import import_stashes_to_shelf from muse.core.shelf import list_shelf_entries muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") # Create a stash in the git repo. (git_root / "work.py").write_text("in progress work") subprocess.run(["git", "-C", str(git_root), "add", "."], check=True, capture_output=True) subprocess.run( ["git", "-C", str(git_root), "stash", "push", "-m", "WIP: some feature"], check=True, capture_output=True, ) count = import_stashes_to_shelf(muse_root, git_root) assert count == 1 entries = list_shelf_entries(muse_root) assert len(entries) == 1 entry = entries[0] assert entry.get("intent_type") == "handoff" assert "WIP: some feature" in str(entry.get("intent", "")) def test_import_stash_dry_run_writes_nothing( self, tmp_path: pathlib.Path ) -> None: """dry_run=True returns count but creates no shelf entries.""" from muse.core.bridge.harmony_shelf import import_stashes_to_shelf from muse.core.shelf import list_shelf_entries muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") (git_root / "wip.py").write_text("work in progress") subprocess.run(["git", "-C", str(git_root), "add", "."], check=True, capture_output=True) subprocess.run( ["git", "-C", str(git_root), "stash", "push", "-m", "dry run stash"], check=True, capture_output=True, ) count = import_stashes_to_shelf(muse_root, git_root, dry_run=True) assert count == 1 assert list_shelf_entries(muse_root) == [] def test_import_stashes_idempotent( self, tmp_path: pathlib.Path ) -> None: """Calling import twice does not duplicate shelf entries.""" from muse.core.bridge.harmony_shelf import import_stashes_to_shelf from muse.core.shelf import list_shelf_entries muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") (git_root / "feature.py").write_text("feature code") subprocess.run(["git", "-C", str(git_root), "add", "."], check=True, capture_output=True) subprocess.run( ["git", "-C", str(git_root), "stash", "push", "-m", "feature stash"], check=True, capture_output=True, ) first = import_stashes_to_shelf(muse_root, git_root) second = import_stashes_to_shelf(muse_root, git_root) assert first == 1 assert second == 0 assert len(list_shelf_entries(muse_root)) == 1 class TestExportShelvesToStash: def test_export_shelves_no_entries_returns_zero( self, tmp_path: pathlib.Path ) -> None: """No Muse shelf entries → 0 stashes created.""" from muse.core.bridge.harmony_shelf import export_shelves_to_stash muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") count = export_shelves_to_stash(muse_root, git_root) assert count == 0 # Verify no stashes were created. result = subprocess.run( ["git", "-C", str(git_root), "stash", "list"], capture_output=True, text=True, check=True, ) assert result.stdout.strip() == "" def test_export_one_shelf_creates_git_stash( self, tmp_path: pathlib.Path ) -> None: """One Muse shelf entry → one git stash.""" from muse.core.bridge.harmony_shelf import export_shelves_to_stash from muse.core.object_store import write_object from muse.core.types import blob_id, content_hash, now_utc_iso from muse.core.shelf import write_shelf_entry muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") # Write a blob into Muse object store. blob_bytes = b"shelf file content" obj_id = blob_id(blob_bytes) write_object(muse_root, obj_id, blob_bytes) snapshot = {"feature.py": obj_id} snapshot_id = content_hash(snapshot) now_iso = now_utc_iso() entry_without_id = { "name": "test-shelf-entry", "snapshot": snapshot, "deleted": [], "snapshot_id": snapshot_id, "parent_commit": "", "branch": "dev", "created_at": now_iso, "created_by": "test", "intent_type": "handoff", "intent": "test shelf export", "resumable": True, "tags": [], "expires_at": None, "domain_state": {}, } shelf_id = content_hash(entry_without_id) entry = dict(entry_without_id) entry["id"] = shelf_id write_shelf_entry(muse_root, entry) count = export_shelves_to_stash(muse_root, git_root) assert count == 1 result = subprocess.run( ["git", "-C", str(git_root), "stash", "list"], capture_output=True, text=True, check=True, ) assert "muse-shelf" in result.stdout def test_export_shelves_dry_run_writes_nothing( self, tmp_path: pathlib.Path ) -> None: """dry_run=True returns count but creates no git stashes.""" from muse.core.bridge.harmony_shelf import export_shelves_to_stash from muse.core.object_store import write_object from muse.core.types import blob_id, content_hash, now_utc_iso from muse.core.shelf import write_shelf_entry muse_root = _make_muse_repo(tmp_path / "muse") git_root = _make_git_repo(tmp_path / "git") blob_bytes = b"dry run blob" obj_id = blob_id(blob_bytes) write_object(muse_root, obj_id, blob_bytes) snapshot = {"dry.py": obj_id} snapshot_id = content_hash(snapshot) now_iso = now_utc_iso() entry_without_id = { "name": "dry-shelf", "snapshot": snapshot, "deleted": [], "snapshot_id": snapshot_id, "parent_commit": "", "branch": "dev", "created_at": now_iso, "created_by": "test", "intent_type": "checkpoint", "intent": "dry run test", "resumable": False, "tags": [], "expires_at": None, "domain_state": {}, } shelf_id = content_hash(entry_without_id) entry = dict(entry_without_id) entry["id"] = shelf_id write_shelf_entry(muse_root, entry) count = export_shelves_to_stash(muse_root, git_root, dry_run=True) assert count == 1 result = subprocess.run( ["git", "-C", str(git_root), "stash", "list"], capture_output=True, text=True, check=True, ) assert result.stdout.strip() == "" # --------------------------------------------------------------------------- # Tier 3 — Argparse flag presence # --------------------------------------------------------------------------- class TestArgparseFlags: def test_git_import_has_import_rerere_flag(self) -> None: result = _invoke("bridge", "git-import", "--help") assert "--import-rerere" in result.output def test_git_import_has_rerere_confidence_flag(self) -> None: result = _invoke("bridge", "git-import", "--help") assert "--rerere-confidence" in result.output def test_git_import_has_import_stashes_flag(self) -> None: result = _invoke("bridge", "git-import", "--help") assert "--import-stashes" in result.output def test_git_export_has_export_rerere_flag(self) -> None: result = _invoke("bridge", "git-export", "--help") assert "--export-rerere" in result.output def test_git_export_has_export_shelves_flag(self) -> None: result = _invoke("bridge", "git-export", "--help") assert "--export-shelves" in result.output