"""MWP-2 Phase 0 (RED) — _walk_commit_delta correctness fallback. Sub-ticket: musehub#107 — https://staging.musehub.ai/gabriel/musehub/issues/107 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58 Predecessor: musehub#106 (MWP-1, generation authority) — closed. Root cause (RC-2): ``_walk_commit_delta`` (musehub_wire_fetch.py:49) pins ``if True:`` at line 66, making the authoritative MusehubCommit DAG walk (lines 159–184) dead code. The live path is a generation-bounded range scan: generation > min_have_gen AND generation <= max_want_gen This is correct only when every generation in ``musehub_commit_graph`` is correct. When a generation is wrong or a graph row is missing, the range scan either excludes real ancestors (corrupt low generation causes max_want_gen to understate the range) or the BFS dead-ends (missing graph row yields empty parents from ``graph_map.get(cid, ([], None, 0))``). Either way ``_walk_commit_delta`` silently returns a truncated commit set, the assembled mpack is missing commits and blobs, and clone exits 0 with a stale working tree. MWP-1 (musehub#106) fixed the write side: no new commit is ever written with a wrong generation, and ``repair_corrupt_commit_generations`` heals pre-existing artefacts. MWP-2 fixes the read side: ``_walk_commit_delta`` must detect that its BFS is incomplete and fall back to the authoritative parent-pointer walk, so the *current* response is always complete regardless of graph state. Expected Phase 0 status: - MWP2_01 RED — corrupt generation excludes ancestors; truncated result - MWP2_02 RED — missing graph row dead-ends BFS; missing commits in result - MWP2_03 GREEN — fully consistent graph uses fast path; no fallback triggered The fix (Phases 1–2) adds a ``graph_incomplete`` flag to the BFS and calls a promoted ``_walk_commit_delta_dag`` helper when the flag is set. """ from __future__ import annotations import datetime import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest from sqlalchemy import delete as _sa_delete, insert as _sa_insert, select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.ids import hash_snapshot from muse.core.mpack import build_wire_mpack, parse_wire_mpack from muse.core.types import blob_id from musehub.core.genesis import compute_identity_id from musehub.db.musehub_repo_models import ( MusehubCommit, MusehubCommitGraph, ) from musehub.services.musehub_repository import create_repo from musehub.services.musehub_wire_fetch import ( _walk_commit_delta, _walk_commit_delta_dag, wire_fetch_mpack, ) from musehub.services.musehub_wire_push import wire_push_unpack_mpack _OWNER = "gabriel" _IDENTITY_ID = compute_identity_id(b"gabriel") # --------------------------------------------------------------------------- # Helpers — mirrors test_mwp1_generation_authority.py conventions # --------------------------------------------------------------------------- def _cid(seed: str) -> str: """Deterministic sha256: commit id from a seed string.""" return blob_id(f"mwp2-commit-{seed}".encode()) def _sid(seed: str) -> str: """Deterministic snapshot id.""" return hash_snapshot({f"{seed}.txt": blob_id(seed.encode())}) def _now() -> datetime.datetime: return datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) async def _seed_commit( session: AsyncSession, cid: str, parent_ids: list[str], snap_id: str, ) -> None: """Insert a musehub_commits row (authoritative DAG) with no graph row.""" session.add(MusehubCommit( commit_id=cid, branch="main", message=f"commit {cid[:12]}", author=_OWNER, timestamp=_now(), parent_ids=parent_ids, snapshot_id=snap_id, )) await session.flush() async def _seed_graph( session: AsyncSession, cid: str, parent_ids: list[str], generation: int, snap_id: str, ) -> None: """Insert a musehub_commit_graph row with an explicit (possibly corrupt) generation.""" session.add(MusehubCommitGraph( commit_id=cid, parent_ids=parent_ids, generation=generation, snapshot_id=snap_id, created_at=_now(), )) await session.flush() class _InMemBackend: """In-memory storage backend for E2E push→fetch tests. Pre-seed push mpacks in ``backend._mpacks[mpack_key] = mpack_bytes`` BEFORE calling ``wire_push_unpack_mpack``. The fetch path assembles and stores its mpack via ``put_mpack``; the test reads it back directly. """ def __init__(self) -> None: self._mpacks: dict[str, bytes] = {} async def put_mpack(self, mpack_id: str, data: bytes) -> None: self._mpacks[mpack_id] = data async def get_mpack(self, mpack_id: str) -> bytes | None: return self._mpacks.get(mpack_id) async def presign_mpack_get(self, mpack_id: str, ttl: int = 3600) -> str: return f"inmem://{mpack_id}" async def put(self, oid: str, data: bytes) -> str: return oid async def get(self, oid: str) -> bytes | None: return None async def presign_get(self, oid: str, ttl: int = 3600) -> str: return f"inmem://{oid}" async def quarantine_mpack(self, *args: object, **kwargs: object) -> None: pass def _raw_commit( cid: str, parent: str | None, snap_id: str, *, branch: str = "main", ) -> dict: """Raw commit wire dict in the shape wire_push_unpack_mpack consumes.""" return { "commit_id": cid, "branch": branch, "message": f"commit {cid[:12]}", "author": _OWNER, "committed_at": _now().isoformat(), "parent_commit_id": parent, "parent2_commit_id": None, "snapshot_id": snap_id, "agent_id": "", "model_id": "", "toolchain_id": "", "sem_ver_bump": "none", "breaking_changes": [], "signature": "", "signer_key_id": "", "signer_public_key": "", "prompt_hash": "", } # --------------------------------------------------------------------------- # MWP2_01 — corrupt generation excludes ancestors from range scan (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_01_corrupt_generation_truncates_walk( db_session: AsyncSession, ) -> None: """RED: a missing interior graph row dead-ends the BFS, dropping ancestors. Chain: C1 <- C2 <- C3 <- C4 (all in musehub_commits, authoritative). Graph: C1(gen=0), C2(gen=1), C4(gen=3) — C3 has NO graph row. want=[C4], have=[]. BFS: C4 is in graph_map (gen=3, parents=[C3]). C3 → graph_map.get(C3) returns default ([], None, 0) → BFS treats C3 as a leaf and stops. C2 and C1 are never added to the reachable set. Result without fix: {C4, C3} — C2 and C1 missing. Result with fix: {C4, C3, C2, C1}. This is the canonical post-MWP-1 RC-2 scenario: the tip has a correct high generation, but an interior ancestor's graph row was never written (a push that wrote musehub_commits but not the graph row, or a pre-backfill gap). MWP2_02 covers the case where the *start tip itself* is absent from the graph. """ c1 = _cid("01-c1") c2 = _cid("01-c2") c3 = _cid("01-c3") c4 = _cid("01-c4") s1, s2, s3, s4 = _sid("01-s1"), _sid("01-s2"), _sid("01-s3"), _sid("01-s4") # Authoritative DAG — all four commits exist in musehub_commits. await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) await _seed_commit(db_session, c4, [c3], s4) # Graph: C1, C2 present with correct generations; C3 absent (the backfill # gap that produces the BFS dead-end); C4 present with correct generation # relative to the true chain (gen=3). await _seed_graph(db_session, c1, [], 0, s1) await _seed_graph(db_session, c2, [c1], 1, s2) # C3 deliberately NOT seeded in graph. await _seed_graph(db_session, c4, [c3], 3, s4) await db_session.commit() result = await _walk_commit_delta(db_session, want=[c4], have=[]) assert c4 in result, "C4 (want tip) must always be in result" assert c3 in result, ( f"C3 must be in result — it is a direct parent of C4. " f"It is missing because graph_map.get(C3) returned empty parents " f"(C3 has no graph row), so the BFS dead-ended at C3. " f"Got result keys: {sorted(result.keys())}" ) assert c2 in result, ( f"C2 must be in result — ancestor of C4 reachable via C3. " f"Got result keys: {sorted(result.keys())}" ) assert c1 in result, ( f"C1 must be in result — root ancestor of C4. " f"Got result keys: {sorted(result.keys())}" ) # --------------------------------------------------------------------------- # MWP2_02 — missing graph row for the start tip dead-ends BFS (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_02_missing_start_tip_graph_row( db_session: AsyncSession, ) -> None: """RED: when the start tip (a want commit) has no graph row at all, the BFS dead-ends immediately and returns an empty or incomplete set. Setup: musehub_commits: C1(root) <- C2 <- C3 musehub_commit_graph: C1(gen=0), C2(gen=1) — C3 has NO graph row. want=[C3], have=[]. max_want_gen: C3 not in graph → query returns None → max_want_gen = 0. Range scan: gen > -1 AND gen <= 0 → only C1 (gen=0) returned in graph_map. BFS: starts from C3; C3 not in graph_map → treated as having empty parents; BFS adds C3 to reachable but never follows its parents (C2, C1). Result without fix: {C3} only. Missing: C2, C1. Result with fix: {C3, C2, C1}. """ c1 = _cid("02-c1") c2 = _cid("02-c2") c3 = _cid("02-c3") s1, s2, s3 = _sid("02-s1"), _sid("02-s2"), _sid("02-s3") # Authoritative DAG. await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) # Graph: C1 and C2 present; C3 (the want tip) absent. await _seed_graph(db_session, c1, [], 0, s1) await _seed_graph(db_session, c2, [c1], 1, s2) await db_session.commit() result = await _walk_commit_delta(db_session, want=[c3], have=[]) assert c3 in result, "C3 (want tip) must be in result" assert c2 in result, ( f"C2 must be in result — parent of C3 (the want tip). " f"C3 had no graph row, so max_want_gen=0 and the range scan only " f"returned C1. BFS from C3 found no parents in graph_map. " f"Got result keys: {sorted(result.keys())}" ) assert c1 in result, ( f"C1 must be in result — root ancestor reachable from C3. " f"Got result keys: {sorted(result.keys())}" ) # --------------------------------------------------------------------------- # MWP2_03 — consistent graph uses fast path, no fallback invoked (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_03_consistent_graph_uses_fast_path( db_session: AsyncSession, ) -> None: """GREEN guard: a fully consistent graph must return the complete ancestor closure via the fast path, and ``_walk_commit_delta_dag`` must NOT be called. This test pins the happy-path contract so later phases cannot accidentally make the fast path degrade into always calling the fallback (which would be correct but slow). After Phase 2 the ``_walk_commit_delta_dag`` helper will exist; this test patches it to detect if it is invoked. Before Phase 1 it doesn't exist, so we patch ``musehub.graph.walk.walk_dag_async`` (the underlying function the legacy walk calls) as a proxy — if the fast path is truly live and exclusive, walk_dag_async should never be called during _walk_commit_delta. Note: after Phase 1 promotes _walk_commit_delta_dag to a real named function, this test should be updated to patch that directly. For Phase 0 we patch walk_dag_async as the canary. Setup: musehub_commits: C1(root) <- C2 <- C3 musehub_commit_graph: C1(gen=0), C2(gen=1), C3(gen=2) — fully consistent. want=[C3], have=[]. Expected: result contains C1, C2, C3 and walk_dag_async is not called. """ c1 = _cid("03-c1") c2 = _cid("03-c2") c3 = _cid("03-c3") s1, s2, s3 = _sid("03-s1"), _sid("03-s2"), _sid("03-s3") await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) await _seed_graph(db_session, c1, [], 0, s1) await _seed_graph(db_session, c2, [c1], 1, s2) await _seed_graph(db_session, c3, [c2], 2, s3) await db_session.commit() with patch( "musehub.services.musehub_wire_fetch._walk_commit_delta_dag", new_callable=AsyncMock, ) as mock_dag: mock_dag.side_effect = AssertionError( "_walk_commit_delta invoked the DAG fallback (_walk_commit_delta_dag) " "on a fully consistent graph — the fast path must be used exclusively." ) result = await _walk_commit_delta(db_session, want=[c3], have=[]) assert c3 in result, f"C3 must be in result; got {sorted(result.keys())}" assert c2 in result, f"C2 must be in result; got {sorted(result.keys())}" assert c1 in result, f"C1 must be in result; got {sorted(result.keys())}" # --------------------------------------------------------------------------- # MWP2_05 — _walk_commit_delta_dag returns full closure with populated fields # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_05_dag_helper_returns_full_closure( db_session: AsyncSession, ) -> None: """GREEN: _walk_commit_delta_dag over a musehub_commits chain returns the complete ancestor closure of want minus have, with snapshot_id and parent_ids populated on every returned row. No graph rows are seeded — the helper walks musehub_commits.parent_ids directly (the authoritative DAG), so a missing/corrupt graph is irrelevant. Chain: C1(root) <- C2 <- C3 Call: _walk_commit_delta_dag(session, starts=[C3], have_set=frozenset()) Expect: {C3, C2, C1}, each with correct snapshot_id and parent_ids. """ c1 = _cid("05-c1") c2 = _cid("05-c2") c3 = _cid("05-c3") s1, s2, s3 = _sid("05-s1"), _sid("05-s2"), _sid("05-s3") await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) await db_session.commit() result = await _walk_commit_delta_dag(db_session, starts=[c3], have_set=frozenset()) assert set(result.keys()) == {c1, c2, c3}, ( f"Must return all three commits; got {sorted(result.keys())}" ) # Every row must have snapshot_id and parent_ids populated. r1, r2, r3 = result[c1], result[c2], result[c3] assert r1.snapshot_id == s1, f"C1 snapshot_id wrong: {r1.snapshot_id}" assert r2.snapshot_id == s2, f"C2 snapshot_id wrong: {r2.snapshot_id}" assert r3.snapshot_id == s3, f"C3 snapshot_id wrong: {r3.snapshot_id}" assert r1.parent_ids == [], f"C1 parent_ids wrong: {r1.parent_ids}" assert r2.parent_ids == [c1], f"C2 parent_ids wrong: {r2.parent_ids}" assert r3.parent_ids == [c2], f"C3 parent_ids wrong: {r3.parent_ids}" # --------------------------------------------------------------------------- # MWP2_06 — _walk_commit_delta_dag returns commits in parents-first topo order # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_06_dag_helper_parents_first_order( db_session: AsyncSession, ) -> None: """GREEN: the dict returned by _walk_commit_delta_dag is in parents-first topological order — every commit appears after all of its in-set parents. walk_dag_async BFS order is children-before-parents (C3, C2, C1 for a linear chain starting at C3). After Kahn's topo-sort the result must be C1, C2, C3 — each parent strictly before its child. Chain: C1(root) <- C2 <- C3 Expected key order: [C1, C2, C3] """ c1 = _cid("06-c1") c2 = _cid("06-c2") c3 = _cid("06-c3") s1, s2, s3 = _sid("06-s1"), _sid("06-s2"), _sid("06-s3") await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) await db_session.commit() result = await _walk_commit_delta_dag(db_session, starts=[c3], have_set=frozenset()) keys = list(result.keys()) assert len(keys) == 3, f"Expected 3 commits, got {len(keys)}: {keys}" idx = {cid: i for i, cid in enumerate(keys)} assert idx[c1] < idx[c2], ( f"C1 (root) must appear before C2 in the result. " f"Got order: C1={idx[c1]}, C2={idx[c2]}, C3={idx[c3]}" ) assert idx[c2] < idx[c3], ( f"C2 must appear before C3 in the result. " f"Got order: C1={idx[c1]}, C2={idx[c2]}, C3={idx[c3]}" ) # --------------------------------------------------------------------------- # MWP2_09 — have boundary is honored under the DAG fallback (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_09_have_boundary_honored_under_fallback( db_session: AsyncSession, ) -> None: """GREEN: when the fallback fires, the have set is respected — commits in have and their ancestors are excluded from the result, same as the fast path. Chain: C1(root) <- C2 <- C3 <- C4 Graph: C1(gen=0), C2(gen=1), C4(gen=3) — C3 absent (triggers fallback). want=[C4], have=[C2]. Fast-path sequence (pre-fallback): max_want_gen=3, min_have_gen=1 (gen of C2). Range scan gen=(1,3] → only C4(gen=3). C3 absent → graph_incomplete=True. Dispatches to _walk_commit_delta_dag(session, starts=[C4], have_set={C2}). DAG walk: BFS from C4 with exclude={C2}. C4 visited → adj=[C3] → C3 visited → adj=[C2] (excluded, already in seen). C2 never yielded. C1 never queued. Kahn's sort: {C3, C4} in parents-first order. Assert: {C3, C4} in result; C1 and C2 absent. """ c1 = _cid("09-c1") c2 = _cid("09-c2") c3 = _cid("09-c3") c4 = _cid("09-c4") s1, s2, s3, s4 = _sid("09-s1"), _sid("09-s2"), _sid("09-s3"), _sid("09-s4") # Authoritative DAG — all four commits. await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) await _seed_commit(db_session, c3, [c2], s3) await _seed_commit(db_session, c4, [c3], s4) # Graph: C1, C2, C4 present; C3 absent → triggers fallback. await _seed_graph(db_session, c1, [], 0, s1) await _seed_graph(db_session, c2, [c1], 1, s2) await _seed_graph(db_session, c4, [c3], 3, s4) await db_session.commit() result = await _walk_commit_delta(db_session, want=[c4], have=[c2]) assert c4 in result, f"C4 (want tip) must be in result; got {sorted(result.keys())}" assert c3 in result, f"C3 must be in result (it is between the tip and the have boundary)" assert c2 not in result, ( f"C2 is in have — it must be excluded from the result; " f"got {sorted(result.keys())}" ) assert c1 not in result, ( f"C1 is an ancestor of have — it must be excluded from the result; " f"got {sorted(result.keys())}" ) # --------------------------------------------------------------------------- # MWP2_10 — max_nodes bound is wired into walk_dag_async + warning fires (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_10_max_nodes_bound_is_wired( db_session: AsyncSession, caplog: pytest.LogCaptureFixture, ) -> None: """GREEN (Phase 3): two assertions about the max_nodes cap in _walk_commit_delta_dag. Part A — kwarg wired: walk_dag_async is called with max_nodes=100_000. Setup: a 2-commit chain with no graph rows → triggers _walk_commit_delta_dag. A wrapper around walk_dag_async captures the kwargs. Part B — warning fires: when walk_dag_async yields exactly 100_000 commits, the `[MWP2] ... max_nodes cap hit` warning is emitted. A mock replaces walk_dag_async to yield 100_000 fake cid strings (not in DB). Kahn's sort runs over them (all in_degree=0, empty result dict). The deque-based Kahn's queue keeps this O(n), not O(n²). """ import logging as _logging from unittest.mock import patch as _patch # ---- Part A: max_nodes=100_000 kwarg reaches walk_dag_async ---- c1 = _cid("10a-c1") c2 = _cid("10a-c2") s1, s2 = _sid("10a-s1"), _sid("10a-s2") await _seed_commit(db_session, c1, [], s1) await _seed_commit(db_session, c2, [c1], s2) # No graph rows — forces _walk_commit_delta_dag. received: dict[str, object] = {} from musehub.graph import walk as _walk_mod _original_walk = _walk_mod.walk_dag_async async def _capturing_walk(starts, adj_fn, *, exclude=None, max_nodes=None): received["max_nodes"] = max_nodes async for item in _original_walk(starts, adj_fn, exclude=exclude, max_nodes=max_nodes): yield item with _patch.object(_walk_mod, "walk_dag_async", new=_capturing_walk): result_a = await _walk_commit_delta(db_session, want=[c2], have=[]) assert received.get("max_nodes") == 100_000, ( f"_walk_commit_delta_dag must pass max_nodes=100_000 to walk_dag_async; " f"got max_nodes={received.get('max_nodes')!r}" ) assert c1 in result_a and c2 in result_a, ( f"Full result still expected even with wrapping; got {sorted(result_a.keys())}" ) # ---- Part B: WARNING fires when reachable count hits the cap ---- # Mock walk_dag_async to yield exactly 100_000 fake cids (not in DB). # Kahn's sort sees all in_degree=0 (no _row_cache entries → no parent edges). # With deque.popleft() this is O(n); with list.pop(0) it would be O(n²). async def _capped_walk(starts, adj_fn, *, exclude=None, max_nodes=None): for i in range(100_000): yield f"sha256:{'a' * 60}{i:04d}" with _patch.object(_walk_mod, "walk_dag_async", new=_capped_walk): with caplog.at_level(_logging.WARNING, logger="musehub.services.musehub_wire_fetch"): await _walk_commit_delta_dag( db_session, starts=[c2], have_set=frozenset(), ) assert any("max_nodes cap hit" in r.message for r in caplog.records), ( "WARNING must be logged when reachable count reaches max_nodes=100_000. " f"Log records captured: {[r.message for r in caplog.records]}" ) # --------------------------------------------------------------------------- # MWP2_11 — 5,000-commit stress: full fallback, correct topo order, O(n) (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_11_stress_5000_commit_linear_chain( db_session: AsyncSession, ) -> None: """GREEN (Phase 3): a 5,000-commit linear chain with no graph rows triggers the DAG fallback and returns all 5,000 commits in parents-first topo order. No graph rows are seeded → fast-path BFS detects graph_incomplete=True at the very first start (want tip absent from graph_map → max_want_gen=0). Fallback walks musehub_commits.parent_ids over the entire chain. Performance note: commits are seeded with a single flush (no commit), so SQLAlchemy's identity map stays warm — session.get() hits the map for each adjacency call (O(1) per call) rather than issuing 5,000 DB round trips. Kahn's topo-sort uses collections.deque to guarantee O(n), not O(n²). Assertions: - result contains exactly 5,000 commits - every parent appears before its child in dict iteration order - result is bounded well below max_nodes=100_000 """ N = 5_000 # Build linear chain C0 ← C1 ← ... ← C4999. chain_cids = [_cid(f"11-c{i:04d}") for i in range(N)] chain_sids = [_sid(f"11-s{i:04d}") for i in range(N)] for i in range(N): parent_ids = [] if i == 0 else [chain_cids[i - 1]] db_session.add(MusehubCommit( commit_id=chain_cids[i], branch="main", message=f"commit {chain_cids[i][:12]}", author=_OWNER, timestamp=_now(), parent_ids=parent_ids, snapshot_id=chain_sids[i], )) # Single flush — keeps identity map warm so session.get() is O(1) per call. # No graph rows seeded → forces full DAG fallback. await db_session.flush() tip = chain_cids[-1] result = await _walk_commit_delta(db_session, want=[tip], have=[]) assert len(result) == N, ( f"Fallback must return all {N} commits; got {len(result)}" ) # Verify parents-first topological order: every parent in the result set must # appear strictly before its child in dict iteration order. keys = list(result.keys()) idx = {cid: pos for pos, cid in enumerate(keys)} violations = [] for cid in keys: row = result[cid] for pid in (row.parent_ids or []): if pid in idx and idx[pid] >= idx[cid]: violations.append((pid[:12], cid[:12])) assert not violations, ( f"Topo-order violation: parent appears after child in {len(violations)} " f"case(s); first few: {violations[:5]}" ) # --------------------------------------------------------------------------- # Shared E2E helpers (MWP2_12 / MWP2_13) # --------------------------------------------------------------------------- def _e2e_backend_patch(backend: _InMemBackend) -> tuple: """Return the three backend patches needed for push + fetch E2E tests.""" return ( patch("musehub.services.musehub_wire.get_backend", return_value=backend), patch("musehub.services.musehub_wire_push.get_backend", return_value=backend), patch("musehub.services.musehub_wire_fetch.get_backend", return_value=backend), patch("musehub.storage.backends.get_backend", return_value=backend), ) async def _e2e_push( session: AsyncSession, repo_id: str, backend: _InMemBackend, cid: str, parent: str | None, snap_id: str, fpath: str, oid: str, data: bytes, ) -> None: """Build a single-commit mpack, store in backend, and push via wire_push_unpack_mpack.""" mpack_bytes = build_wire_mpack({ "commits": [_raw_commit(cid, parent, snap_id)], "snapshots": [{ "snapshot_id": snap_id, "parent_snapshot_id": None, "delta_upsert": {fpath: oid}, "delta_remove": [], "directories": [], }], "blobs": [{"object_id": oid, "content": data}], "tags": [], }) mpack_key = blob_id(mpack_bytes) backend._mpacks[mpack_key] = mpack_bytes push_patches = _e2e_backend_patch(backend) with push_patches[0], push_patches[1], push_patches[2], push_patches[3]: await wire_push_unpack_mpack( session, repo_id, mpack_key, pusher_id=_OWNER, branch="main", head_commit_id=cid, commits_count=1, blobs_count=1, force=True, ) await session.flush() # --------------------------------------------------------------------------- # MWP2_12 — E2E acceptance gate: fallback alone ships a complete clone (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_12_e2e_fallback_alone_ships_complete_clone( db_session: AsyncSession, ) -> None: """GREEN (Phase 4 — acceptance gate): MWP-2 fallback alone delivers a complete clone even when the commit graph has a missing interior row. Setup: 1. Create a repo and push C1 → C2 → C3 individually. CommitGraph after pushes: C1(gen=0), C2(gen=1), C3(gen=2). 2. Delete C2's CommitGraph row — interior node missing. 3. Patch out repair_corrupt_commit_generations (MWP-1 cannot self-heal). 4. Clone via wire_fetch_mpack(want=[C3], have=[], force_build=True). Without MWP-2 (the bug): max_want_gen = 2 (C3 in graph). Range scan: gen=(−1,2] → C1(0), C3(2). BFS: C3 → C2 (C2 not in graph_map, parents=[]) → BFS dead-end. Reachable = {C3, C2}. C1 ABSENT from assembled mpack. With MWP-2 (the fix): BFS detects C2 absent from graph_map → graph_incomplete=True → fallback. _walk_commit_delta_dag returns {C1, C2, C3}. All three commits in mpack. Note: MWP-1's guard does NOT fire here because C3's graph snapshot still matches C3's commit snapshot — the corruption is only a missing graph row, not a wrong generation. repair_corrupt_commit_generations is patched as a safety rail to prove MWP-2 is self-sufficient. """ repo = await create_repo( db_session, name="mwp2-12", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() c1, c2, c3 = _cid("12-c1"), _cid("12-c2"), _cid("12-c3") oid1 = blob_id(b"12-c1-blob") oid2 = blob_id(b"12-c2-blob") oid3 = blob_id(b"12-c3-blob") s1 = hash_snapshot({"f1.txt": oid1}) s2 = hash_snapshot({"f2.txt": oid2}) s3 = hash_snapshot({"f3.txt": oid3}) backend = _InMemBackend() # Push C1, C2, C3 — each push writes correct CommitGraph generations. await _e2e_push(db_session, repo.repo_id, backend, c1, None, s1, "f1.txt", oid1, b"12-c1-blob") await _e2e_push(db_session, repo.repo_id, backend, c2, c1, s2, "f2.txt", oid2, b"12-c2-blob") await _e2e_push(db_session, repo.repo_id, backend, c3, c2, s3, "f3.txt", oid3, b"12-c3-blob") # Corrupt: delete C2's CommitGraph row (missing interior node). await db_session.execute( _sa_delete(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == c2) ) await db_session.flush() # Patch out MWP-1's repair guard to prove MWP-2 handles this alone. fetch_patches = _e2e_backend_patch(backend) with patch( "musehub.services.musehub_wire_fetch.repair_corrupt_commit_generations", new_callable=AsyncMock, ) as mock_repair, \ fetch_patches[0], fetch_patches[2], fetch_patches[3]: mock_repair.return_value = {"total_corrupt": 0, "repaired": 0} fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[c3], have=[], force_build=True, ) # MWP-1 guard should NOT fire: C3's graph snapshot still matches its commit # snapshot — only the missing C2 row caused truncation (RC-2, not RC-1). mock_repair.assert_not_called() assert fetch_result["mpack_id"] is not None, "wire_fetch_mpack must return a mpack_id" assembled_bytes = backend._mpacks[fetch_result["mpack_id"]] parsed = parse_wire_mpack(assembled_bytes) commit_ids = {c["commit_id"] for c in parsed.get("commits", [])} blob_oids = {b["object_id"] for b in parsed.get("blobs", [])} assert c1 in commit_ids, ( f"C1 must be in assembled mpack — WITHOUT MWP-2 the BFS dead-ends at C2 " f"(C2 absent from graph_map → parents=[]) and C1 is never reachable. " f"commits present: {[cid[:16] for cid in commit_ids]}" ) assert c2 in commit_ids, ( f"C2 must be in assembled mpack; got {[cid[:16] for cid in commit_ids]}" ) assert c3 in commit_ids, ( f"C3 (want tip) must be in assembled mpack; got {[cid[:16] for cid in commit_ids]}" ) assert oid3 in blob_oids, ( f"C3's blob (oid3) must be in assembled mpack; got {[o[:16] for o in blob_oids]}" ) # --------------------------------------------------------------------------- # MWP2_13 — E2E: fallback + graph healing → fast path on second call (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp2_13_e2e_fast_path_after_graph_healed( db_session: AsyncSession, ) -> None: """GREEN (Phase 4): MWP-2 handles the corrupt state; once the graph is healed by any mechanism, a follow-up _walk_commit_delta uses the fast path. Proves MWP-1 + MWP-2 compose: the fallback is the safety net for the *current* response; the fast path is restored once the graph is intact. Setup (same as MWP2_12): Push C1 → C2 → C3. Delete C2's CommitGraph row. repair NOT patched. First clone: MWP-2 fallback fires. Result: {C1, C2, C3} — complete. MWP-1 guard does NOT fire (no snapshot mismatch for this corruption type). Graph heal: INSERT C2's CommitGraph row back (gen=1) — simulating what a subsequent push or an explicit repair would do once the graph gap is noticed. Second call (_walk_commit_delta, not full wire_fetch_mpack): _walk_commit_delta_dag is patched to raise AssertionError if called. With the healed graph (C1=0, C2=1, C3=2 all present), the fast-path BFS completes without any missing nodes → graph_incomplete=False → no fallback. The patched dag helper is never invoked — no AssertionError. """ repo = await create_repo( db_session, name="mwp2-13", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() c1, c2, c3 = _cid("13-c1"), _cid("13-c2"), _cid("13-c3") oid1 = blob_id(b"13-c1-blob") oid2 = blob_id(b"13-c2-blob") oid3 = blob_id(b"13-c3-blob") s1 = hash_snapshot({"f1.txt": oid1}) s2 = hash_snapshot({"f2.txt": oid2}) s3 = hash_snapshot({"f3.txt": oid3}) backend = _InMemBackend() # Push C1, C2, C3. await _e2e_push(db_session, repo.repo_id, backend, c1, None, s1, "f1.txt", oid1, b"13-c1-blob") await _e2e_push(db_session, repo.repo_id, backend, c2, c1, s2, "f2.txt", oid2, b"13-c2-blob") await _e2e_push(db_session, repo.repo_id, backend, c3, c2, s3, "f3.txt", oid3, b"13-c3-blob") # Corrupt: delete C2's CommitGraph row (same as MWP2_12). await db_session.execute( _sa_delete(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == c2) ) await db_session.flush() # First clone — MWP-2 fallback fires; repair is NOT patched (it won't run # for this corruption type, but we leave it live to prove composition). fetch_patches = _e2e_backend_patch(backend) with fetch_patches[0], fetch_patches[2], fetch_patches[3]: fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[c3], have=[], force_build=True, ) assert fetch_result["mpack_id"] is not None assembled_bytes = backend._mpacks[fetch_result["mpack_id"]] parsed_first = parse_wire_mpack(assembled_bytes) commit_ids_first = {c["commit_id"] for c in parsed_first.get("commits", [])} assert {c1, c2, c3} == commit_ids_first, ( f"First clone must contain all 3 commits; got {[cid[:16] for cid in commit_ids_first]}" ) # Heal the graph: re-insert C2's CommitGraph row with the correct generation. # This simulates what a follow-up push or repair routine would do once it # detects the gap. The INSERT is safe because the DELETE was already flushed. await db_session.execute( _sa_insert(MusehubCommitGraph).values( commit_id=c2, parent_ids=[c1], generation=1, snapshot_id=s2, created_at=_now(), ) ) await db_session.flush() # Second call: _walk_commit_delta must use the fast path now. # Patch _walk_commit_delta_dag to raise if called — any invocation = regression. with patch( "musehub.services.musehub_wire_fetch._walk_commit_delta_dag", new_callable=AsyncMock, ) as mock_dag: mock_dag.side_effect = AssertionError( "MWP2_13: _walk_commit_delta invoked the DAG fallback after the graph " "was healed — the fast path must be used exclusively when graph is intact." ) result2 = await _walk_commit_delta(db_session, want=[c3], have=[]) # Fast path returned — verify completeness (now as SimpleNamespace proxies). assert {c1, c2, c3} == set(result2.keys()), ( f"Second _walk_commit_delta must return all 3 commits via fast path; " f"got {[cid[:16] for cid in result2.keys()]}" )