"""TDD — Phase 4: push-path prune gate and transport BFS consolidation. Phase 4 of issue #6 (generic DAG walker). Two atomic changes: Pack — prune gate ----------------- ``collect_object_ids`` and ``walk_commits`` replace ``exclude=have_set`` with ``prune=lambda cid: cid in have_set`` in their ``iter_ancestors`` calls. Effect: the walk stops at the server boundary without expanding have-commit ancestors. Both ``exclude`` and ``prune`` are semantically equivalent for reachability, but ``prune`` makes the early-termination intent explicit and consistent with the ``walk_dag`` API used everywhere else. Transport — bundle-overlay adjacency closure -------------------------------------------- ``_is_ancestor`` in ``transport.py`` replaces its inline BFS (``seen: set``, ``queue: list``, ``while queue``) with ``walk_dag``. The adjacency function is a closure over ``bundle_by_id`` and ``remote_root`` — it reads from the in-memory bundle first and falls back to the on-disk store, which is the same two-source lookup the old loop did. Coverage -------- P4-1 Structural — ``collect_object_ids`` source contains ``prune=`` P4-2 Structural — ``walk_commits`` source contains ``prune=`` P4-3 Behavioural — walk stops at have boundary; objects from ancestors of the have commit are NOT included in the result P4-4 Structural — ``_is_ancestor`` uses ``walk_dag``; no inline BFS queue P4-5 Behavioural — ``_is_ancestor`` returns True when candidate is in bundle P4-6 Behavioural — ``_is_ancestor`` returns True when candidate is in store P4-7 Behavioural — ``_is_ancestor`` returns False for unreachable commit """ from __future__ import annotations import datetime import inspect import json import pathlib import pytest from muse._version import __version__ from muse.core.object_store import write_object from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from muse.core.types import blob_id from muse.core.paths import muse_dir # --------------------------------------------------------------------------- # Repo fixture helpers # --------------------------------------------------------------------------- def _repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: dot_muse = muse_dir(tmp_path) for d in ("commits", "snapshots", "objects", "refs/heads", "remotes"): (dot_muse / d).mkdir(parents=True, exist_ok=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) return tmp_path 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, manifest: dict[str, str], parent_id: str | None = None, ) -> CommitRecord: snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) cid = compute_commit_id( parent_ids=[parent_id] if parent_id else [], snapshot_id=snap_id, message="test", committed_at_iso=ts.isoformat(), ) rec = CommitRecord( repo_id="test-repo", commit_id=cid, branch="main", snapshot_id=snap_id, message="test", committed_at=ts, parent_commit_id=parent_id, ) write_commit(root, rec) return rec # --------------------------------------------------------------------------- # P4-1 Structural — collect_object_ids uses prune= # --------------------------------------------------------------------------- def test_p4_1_collect_object_ids_uses_prune() -> None: """collect_object_ids must use prune= in its iter_ancestors call. ``prune=lambda cid: cid in have_set`` makes early-termination intent explicit. A plain ``exclude=have_set`` still works but diverges from the canonical pattern used throughout graph.py. """ from muse.core import pack as pack_mod src = inspect.getsource(pack_mod.collect_object_ids) assert "prune=" in src, ( "collect_object_ids must pass prune= to iter_ancestors (or walk_dag). " "Replace exclude=have_set with prune=lambda cid: cid in have_set." ) # --------------------------------------------------------------------------- # P4-2 Structural — walk_commits uses prune= # --------------------------------------------------------------------------- def test_p4_2_walk_commits_uses_prune() -> None: """walk_commits must use prune= in its iter_ancestors call.""" from muse.core import pack as pack_mod src = inspect.getsource(pack_mod.walk_commits) assert "prune=" in src, ( "walk_commits must pass prune= to iter_ancestors (or walk_dag). " "Replace exclude=have_set with prune=lambda cid: cid in have_set." ) # --------------------------------------------------------------------------- # P4-3 Behavioural — walk stops at have boundary # --------------------------------------------------------------------------- def test_p4_3_collect_object_ids_stops_at_have_boundary( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """Objects from commits older than the have boundary must not be returned. Chain: C1 (old) → C2 (have/server boundary) → C3 (new, local HEAD) Each commit changes the same file. C2 is the have commit the server already has. collect_object_ids(root, [C3], have=[C2]) must return only oid_c3 — the object from C3. It must NOT include oid_c1 (C1 is an ancestor of the have boundary and must be pruned) and must NOT include oid_c2 (C2 is the have commit itself). """ from muse.core.pack import collect_object_ids root = _repo(tmp_path, monkeypatch) oid_c1 = _write_obj(root, b"version-1") oid_c2 = _write_obj(root, b"version-2") oid_c3 = _write_obj(root, b"version-3") c1 = _make_commit(root, {"src/main.py": oid_c1}) c2 = _make_commit(root, {"src/main.py": oid_c2}, parent_id=c1.commit_id) c3 = _make_commit(root, {"src/main.py": oid_c3}, parent_id=c2.commit_id) result = collect_object_ids(root, [c3.commit_id], have=[c2.commit_id]) assert oid_c3 in result, "New object (C3's file version) must be in result" assert oid_c2 not in result, ( "Have-commit object must not be in result — server already has it" ) assert oid_c1 not in result, ( "Ancestor-of-have object must not be in result — pruning must stop " "the walk before reaching C1. If oid_c1 appears, the prune gate is " "not firing correctly at the have boundary." ) # --------------------------------------------------------------------------- # P4-4 Structural — _is_ancestor uses walk_dag, not inline BFS # --------------------------------------------------------------------------- def test_p4_4_is_ancestor_uses_walk_dag() -> None: """_is_ancestor must use walk_dag instead of an inline BFS queue. The inline pattern ``seen: set[str] = set(); queue: list[str] = [...]`` followed by a ``while queue`` loop must not appear in ``_is_ancestor``. Replace with ``walk_dag`` using a bundle-overlay adjacency closure. """ from muse.core import transport as transport_mod src = inspect.getsource(transport_mod._is_ancestor) # type: ignore[attr-defined] assert "walk_dag" in src, ( "_is_ancestor must use walk_dag for its BFS traversal. " "Replace the inline while-queue loop with walk_dag + adjacency closure." ) assert "while queue" not in src, ( "_is_ancestor still has an inline while-queue BFS. " "Replace with walk_dag." ) # --------------------------------------------------------------------------- # P4-5 Behavioural — _is_ancestor: candidate found in bundle # --------------------------------------------------------------------------- def test_p4_5_is_ancestor_finds_candidate_in_bundle( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """_is_ancestor returns True when the candidate is in the bundle. Simulates the case where the fast-forward check needs to confirm the remote tip (candidate) is reachable from the new push HEAD (from_commit), and the remote tip is included in the push bundle. """ from muse.core.transport import _is_ancestor # type: ignore[attr-defined] root = _repo(tmp_path, monkeypatch) oid = _write_obj(root, b"data") c1 = _make_commit(root, {"f.py": oid}) c2 = _make_commit(root, {"f.py": oid}, parent_id=c1.commit_id) c3 = _make_commit(root, {"f.py": oid}, parent_id=c2.commit_id) # Bundle contains c3 and c2; c1 is on disk (remote store). bundle_by_id = { c3.commit_id: c3.to_dict(), c2.commit_id: c2.to_dict(), } # c1 is reachable from c3 via the bundle path. assert _is_ancestor(c1.commit_id, c3.commit_id, bundle_by_id, root) is True, ( "_is_ancestor must return True when candidate is reachable through " "commits that exist in the bundle." ) # --------------------------------------------------------------------------- # P4-6 Behavioural — _is_ancestor: candidate found in store (not bundle) # --------------------------------------------------------------------------- def test_p4_6_is_ancestor_finds_candidate_in_store( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """_is_ancestor returns True when candidate is in store, not bundle. Simulates the typical push scenario: the new commits are in the bundle, the existing remote commits (including the candidate remote tip) are only on disk in the remote store. """ from muse.core.transport import _is_ancestor # type: ignore[attr-defined] root = _repo(tmp_path, monkeypatch) oid = _write_obj(root, b"data2") c1 = _make_commit(root, {"g.py": oid}) c2 = _make_commit(root, {"g.py": oid}, parent_id=c1.commit_id) # Bundle only contains c2 (the new push tip); c1 is on disk. bundle_by_id = {c2.commit_id: c2.to_dict()} # c1 is reachable from c2 via store fallback. assert _is_ancestor(c1.commit_id, c2.commit_id, bundle_by_id, root) is True, ( "_is_ancestor must fall back to the store to find ancestors not in the bundle." ) # --------------------------------------------------------------------------- # P4-7 Behavioural — _is_ancestor: returns False for unreachable commit # --------------------------------------------------------------------------- def test_p4_7_is_ancestor_returns_false_for_unreachable( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """_is_ancestor returns False when the candidate is not reachable. Two independent chains share no common ancestor. The candidate from chain B must not be found when walking chain A. """ from muse.core.transport import _is_ancestor # type: ignore[attr-defined] root = _repo(tmp_path, monkeypatch) oid = _write_obj(root, b"data3") # Chain A a1 = _make_commit(root, {"a.py": oid}) a2 = _make_commit(root, {"a.py": oid}, parent_id=a1.commit_id) # Chain B (independent) oid2 = _write_obj(root, b"data4") b1 = _make_commit(root, {"b.py": oid2}) bundle_by_id = {a2.commit_id: a2.to_dict()} assert _is_ancestor(b1.commit_id, a2.commit_id, bundle_by_id, root) is False, ( "_is_ancestor must return False when the candidate is on an independent " "chain unreachable from from_commit." )