"""MWP-1 Phase 0 (RED) — commit-graph generation authority. Sub-ticket: musehub#106 — https://staging.musehub.ai/gabriel/musehub/issues/106 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58 Reproduces the clone-after-push staleness root cause (RC-1): both generation computers fall back to ``generation 0`` when a commit's parent generation cannot be resolved — musehub_wire_push.py:926 (wire_push_unpack_mpack, step 9b) musehub_wire_push.py:1188 (_build_commit_graph_from_raw) both: ``gen = (max(parent_gens) + 1) if parent_gens else 0`` That conflates a genuine root commit (parent_ids == [], gen 0 correct) with a commit whose parents exist but were not resolved (must be parent_gen + 1). A gen-0 tip then poisons the generation-bounded range scan in ``_walk_commit_delta`` (``generation <= max_want_gen``), truncating the fetch walk and shipping a clone that is missing the newest commits. Expected Phase 0 status: - MWP1_01 RED (currently yields generation 0 instead of 3) - MWP1_02 GREEN (guard — genuine root must stay 0 through every later phase) - MWP1_03 RED (push path yields generation 0 for the new tip) The fix (Phases 1–2) makes generations authoritative by backfilling missing parent generations from ``musehub_commits`` (the source-of-truth DAG). """ from __future__ import annotations import datetime import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest from sqlalchemy import delete as _sa_delete, 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_branch_id, compute_identity_id from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubCommitGraph, ) from musehub.services.musehub_repository import create_repo from musehub.services.musehub_wire_fetch import wire_fetch_mpack from musehub.services.musehub_wire_push import ( _build_commit_graph_from_raw, _resolve_generation_with_backfill, check_commit_graph_invariant, repair_corrupt_commit_generations, wire_push_unpack_mpack, ) _OWNER = "gabriel" _IDENTITY_ID = compute_identity_id(b"gabriel") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _cid(seed: str) -> str: """Deterministic sha256: commit id from a seed string.""" return blob_id(f"mwp1-commit-{seed}".encode()) def _now() -> datetime.datetime: return datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _raw_commit( cid: str, parent: str | None, snap_id: str, *, branch: str = "main", ) -> dict: """A raw commit dict in the shape both generation computers consume.""" 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": "", } async def _seed_commit_row( session: AsyncSession, cid: str, parent_ids: list[str], snap_id: str, ) -> None: """Insert a musehub_commits row (source-of-truth DAG) without a 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_row( session: AsyncSession, cid: str, parent_ids: list[str], generation: int, snap_id: str, ) -> None: session.add(MusehubCommitGraph( commit_id=cid, parent_ids=parent_ids, generation=generation, snapshot_id=snap_id, created_at=_now(), )) await session.flush() async def _graph_gen(session: AsyncSession, cid: str) -> int | None: return (await session.execute( select(MusehubCommitGraph.generation).where(MusehubCommitGraph.commit_id == cid) )).scalar_one_or_none() def _mock_backend(mpack_bytes: bytes) -> MagicMock: backend = MagicMock() backend.get_mpack = AsyncMock(return_value=mpack_bytes) backend.put = AsyncMock(return_value=None) backend.put_mpack = AsyncMock(return_value=None) backend.quarantine_mpack = AsyncMock(return_value=None) backend.presign_get = AsyncMock(return_value="") backend.presign_mpack_get = AsyncMock(return_value="") return backend class _InMemBackend: """In-memory backend for end-to-end push→fetch tests. Pre-seed push mpacks with ``backend._mpacks[mpack_key] = mpack_bytes`` BEFORE calling ``wire_push_unpack_mpack``. The fetch path writes its assembled mpack via ``put_mpack`` and 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 # --------------------------------------------------------------------------- # MWP1_01 — unresolved parent must get parent_gen + 1, never 0 # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_01_unresolved_parent_gets_correct_generation( db_session: AsyncSession, ) -> None: """RED: a child whose parent is absent from both the mpack and the commit graph must receive ``parent_generation + 1`` — not 0. Precondition: C1<-C2<-C3 exist in musehub_commits (the authoritative DAG) but NONE of them have a musehub_commit_graph row. We then compute the graph for [C4] alone (C4's parent C3 is the external, ungraphed parent). True generations: C1=0, C2=1, C3=2, therefore C4 must be 3. Buggy behaviour: db_parent_gens is empty -> C4 gets generation 0. """ c1, c2, c3, c4 = _cid("01-c1"), _cid("01-c2"), _cid("01-c3"), _cid("01-c4") s1, s2, s3, s4 = (hash_snapshot({"f.txt": blob_id(f"01-{n}".encode())}) for n in ("s1", "s2", "s3", "s4")) await _seed_commit_row(db_session, c1, [], s1) await _seed_commit_row(db_session, c2, [c1], s2) await _seed_commit_row(db_session, c3, [c2], s3) await _build_commit_graph_from_raw(db_session, [_raw_commit(c4, c3, s4)]) await db_session.flush() gen_c4 = await _graph_gen(db_session, c4) assert gen_c4 == 3, ( f"C4 must inherit parent_generation + 1 = 3 (C1=0,C2=1,C3=2,C4=3); " f"got {gen_c4}. A gen-0 tip poisons the fetch range scan and ships a " f"stale clone." ) # --------------------------------------------------------------------------- # MWP1_02 — genuine root commit stays generation 0 (guard, GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_02_root_commit_generation_zero( db_session: AsyncSession, ) -> None: """GREEN guard: a commit with no parents is a genuine root — generation 0. Later phases must not over-correct the gen-0 fallback into breaking the legitimate root case. """ root = _cid("02-root") s_root = hash_snapshot({"f.txt": blob_id(b"02-root")}) await _build_commit_graph_from_raw(db_session, [_raw_commit(root, None, s_root)]) await db_session.flush() gen_root = await _graph_gen(db_session, root) assert gen_root == 0, f"root commit (no parents) must be generation 0; got {gen_root}" # --------------------------------------------------------------------------- # MWP1_03 — push of a child of an ungraphed remote tip (push-path site) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_03_push_child_of_ungraphed_remote_tip( db_session: AsyncSession, ) -> None: """RED: pushing C4 (child of remote tip C3) when C3 has no graph row must backfill C3's generation and assign C4 = C3 + 1. This exercises the OTHER generation site — wire_push_unpack_mpack step 9b. Precondition mirrors a real backfill gap: - musehub_commits: C1<-C2<-C3 (authoritative DAG) - musehub_commit_graph: C1=0, C2=1 only (C3 deliberately absent) - branch main -> C3 (remote tip) Push C4 (parent C3, in the 'have' set so excluded from the mpack). True result: C3 backfilled to 2, C4 = 3. Buggy result: external parent C3 not found in graph -> C4 gets generation 0. force=True isolates step-9 generation logic from the FF/ancestor check, which itself reads the commit graph we are deliberately leaving incomplete. """ repo = await create_repo( db_session, name="mwp1-03", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() c1, c2, c3, c4 = _cid("03-c1"), _cid("03-c2"), _cid("03-c3"), _cid("03-c4") s1 = hash_snapshot({"f1.txt": blob_id(b"03-1")}) s2 = hash_snapshot({"f2.txt": blob_id(b"03-2")}) s3 = hash_snapshot({"f3.txt": blob_id(b"03-3")}) # Authoritative DAG: all three commits exist in musehub_commits. await _seed_commit_row(db_session, c1, [], s1) await _seed_commit_row(db_session, c2, [c1], s2) await _seed_commit_row(db_session, c3, [c2], s3) # Graph has C1, C2 — but NOT C3 (the backfill gap). await _seed_graph_row(db_session, c1, [], 0, s1) await _seed_graph_row(db_session, c2, [c1], 1, s2) # Branch tip is C3. db_session.add(MusehubBranch( branch_id=compute_branch_id(repo.repo_id, "main"), repo_id=repo.repo_id, name="main", head_commit_id=c3, )) await db_session.commit() # Build a MUSE-binary mpack carrying ONLY C4 (its parent C3 is external). oid4 = blob_id(b"03-c4-content") s4 = hash_snapshot({"f4.txt": oid4}) mpack_bytes = build_wire_mpack({ "commits": [_raw_commit(c4, c3, s4)], "snapshots": [{ "snapshot_id": s4, "parent_snapshot_id": None, "delta_upsert": {"f4.txt": oid4}, "delta_remove": [], "directories": [], }], "blobs": [{"object_id": oid4, "content": b"03-c4-content"}], "tags": [], }) mpack_key = blob_id(mpack_bytes) backend = _mock_backend(mpack_bytes) with patch("musehub.services.musehub_wire.get_backend", return_value=backend), \ patch("musehub.services.musehub_wire_push.get_backend", return_value=backend), \ patch("musehub.storage.backends.get_backend", return_value=backend): await wire_push_unpack_mpack( db_session, repo.repo_id, mpack_key, pusher_id=_OWNER, branch="main", head_commit_id=c4, commits_count=1, blobs_count=1, force=True, ) await db_session.flush() gen_c3 = await _graph_gen(db_session, c3) gen_c4 = await _graph_gen(db_session, c4) assert gen_c3 == 2, f"C3 must be backfilled to generation 2; got {gen_c3}" assert gen_c4 == 3, ( f"C4 (child of remote tip C3) must be generation 3; got {gen_c4}. " f"A gen-0 new tip is the clone-after-push staleness bug." ) # --------------------------------------------------------------------------- # MWP1_04 — Phase 1: seam test for _build_commit_graph_from_raw # --------------------------------------------------------------------------- _WIRE_SHARED_LOGGER = "musehub.services.musehub_wire_shared" _UNRESOLVED_MARKER = "[MWP1] unresolved parent generations" @pytest.mark.asyncio async def test_mwp1_04_seam_root_vs_unresolved_build_commit_graph( db_session: AsyncSession, caplog: pytest.LogCaptureFixture, ) -> None: """GREEN: _build_commit_graph_from_raw separates 'genuine root' from 'unresolved parent' at the generation seam, then backfills via Phase 2. - Root commit (parent_ids == []) → generation 0, NO unresolved warning. - Child of a parent in musehub_commits but NOT in the graph → fires the unresolved branch, logs a warning, then backfills correctly. """ root = _cid("04-root") child = _cid("04-child") ungraphed_parent = _cid("04-ungraphed") s_root = hash_snapshot({"r.txt": blob_id(b"04-root")}) s_ungraphed = hash_snapshot({"u.txt": blob_id(b"04-ungraphed")}) s_child = hash_snapshot({"c.txt": blob_id(b"04-child")}) # Seed ungraphed_parent in musehub_commits (authoritative DAG) but NOT in # musehub_commit_graph — this is the realistic backfill-gap scenario. await _seed_commit_row(db_session, ungraphed_parent, [], s_ungraphed) await db_session.flush() with caplog.at_level(logging.WARNING, logger=_WIRE_SHARED_LOGGER): # Root case — must NOT emit unresolved warning. await _build_commit_graph_from_raw(db_session, [_raw_commit(root, None, s_root)]) await db_session.flush() root_warnings = [ r.message for r in caplog.records if _UNRESOLVED_MARKER in r.message ] assert root_warnings == [], ( f"Root commit must not trigger the unresolved-parent warning; got: {root_warnings}" ) caplog.clear() with caplog.at_level(logging.WARNING, logger=_WIRE_SHARED_LOGGER): # Unresolved case — ungraphed_parent is in musehub_commits but NOT in graph. # The seam must fire the unresolved branch, log a warning, then backfill. await _build_commit_graph_from_raw( db_session, [_raw_commit(child, ungraphed_parent, s_child)], ) await db_session.flush() unresolved_warnings = [ r.message for r in caplog.records if _UNRESOLVED_MARKER in r.message ] assert unresolved_warnings, ( "Expected an unresolved-parent warning from _build_commit_graph_from_raw " "when the parent has no graph row — none was emitted. " "The Phase 1 seam is not wired correctly at site 2." ) warning_text = unresolved_warnings[0] assert "_build_commit_graph_from_raw" in warning_text, ( f"Warning should identify the site (_build_commit_graph_from_raw); got: {warning_text!r}" ) assert ungraphed_parent in warning_text or child in warning_text, ( f"Warning should mention the commit or unresolved parent; got: {warning_text!r}" ) # --------------------------------------------------------------------------- # MWP1_05 — Phase 1: seam test for wire_push_unpack_mpack (step 9b) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_05_seam_push_path_logs_unresolved_warning( db_session: AsyncSession, caplog: pytest.LogCaptureFixture, ) -> None: """GREEN: the wire_push_unpack_mpack generation site logs the unresolved warning when a pushed commit's parent has no graph row. This is the MWP1_03 scenario restated as a spy/log assertion (Phase 1 seam for the push-path site). The generation value is still wrong (0) until Phase 2; but the seam is wired and observable. """ repo = await create_repo( db_session, name="mwp1-05", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() c1, c2, c3, c4 = _cid("05-c1"), _cid("05-c2"), _cid("05-c3"), _cid("05-c4") s1 = hash_snapshot({"f1.txt": blob_id(b"05-1")}) s2 = hash_snapshot({"f2.txt": blob_id(b"05-2")}) s3 = hash_snapshot({"f3.txt": blob_id(b"05-3")}) await _seed_commit_row(db_session, c1, [], s1) await _seed_commit_row(db_session, c2, [c1], s2) await _seed_commit_row(db_session, c3, [c2], s3) await _seed_graph_row(db_session, c1, [], 0, s1) await _seed_graph_row(db_session, c2, [c1], 1, s2) # C3 deliberately absent from musehub_commit_graph. db_session.add(MusehubBranch( branch_id=compute_branch_id(repo.repo_id, "main"), repo_id=repo.repo_id, name="main", head_commit_id=c3, )) await db_session.commit() oid4 = blob_id(b"05-c4-content") s4 = hash_snapshot({"f4.txt": oid4}) mpack_bytes = build_wire_mpack({ "commits": [_raw_commit(c4, c3, s4)], "snapshots": [{ "snapshot_id": s4, "parent_snapshot_id": None, "delta_upsert": {"f4.txt": oid4}, "delta_remove": [], "directories": [], }], "blobs": [{"object_id": oid4, "content": b"05-c4-content"}], "tags": [], }) mpack_key = blob_id(mpack_bytes) backend = _mock_backend(mpack_bytes) with caplog.at_level(logging.WARNING, logger=_WIRE_SHARED_LOGGER), \ patch("musehub.services.musehub_wire.get_backend", return_value=backend), \ patch("musehub.services.musehub_wire_push.get_backend", return_value=backend), \ patch("musehub.storage.backends.get_backend", return_value=backend): await wire_push_unpack_mpack( db_session, repo.repo_id, mpack_key, pusher_id=_OWNER, branch="main", head_commit_id=c4, commits_count=1, blobs_count=1, force=True, ) await db_session.flush() unresolved_warnings = [ r.message for r in caplog.records if _UNRESOLVED_MARKER in r.message ] assert unresolved_warnings, ( "Expected the unresolved-parent warning from wire_push_unpack_mpack " "when C4's parent C3 has no graph row — none was emitted. " "The Phase 1 seam is not wired at site 1." ) warning_text = unresolved_warnings[0] assert "wire_push_unpack_mpack" in warning_text, ( f"Warning should identify the site (wire_push_unpack_mpack); got: {warning_text!r}" ) # --------------------------------------------------------------------------- # MWP1_06/MWP1_07 — Phase 2: backfill resolves full missing-parent chain # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_07_backfill_resolves_chain_of_depth_n( db_session: AsyncSession, ) -> None: """GREEN: _build_commit_graph_from_raw now calls _resolve_generation_with_backfill, which walks musehub_commits.parent_ids to compute generations for all missing ancestors. Scenario: C1<-C2<-C3<-C4 exist in musehub_commits but NONE in the graph. Push [C5] (parent C4). True generations: C1=0 C2=1 C3=2 C4=3 C5=4. After Phase 2 the backfill must upsert C1..C4 and compute C5=4. """ c1, c2, c3, c4, c5 = (_cid(f"07-c{i}") for i in range(1, 6)) snaps = { c1: hash_snapshot({"a.txt": blob_id(b"07-1")}), c2: hash_snapshot({"b.txt": blob_id(b"07-2")}), c3: hash_snapshot({"c.txt": blob_id(b"07-3")}), c4: hash_snapshot({"d.txt": blob_id(b"07-4")}), c5: hash_snapshot({"e.txt": blob_id(b"07-5")}), } # Seed the authoritative DAG — no graph rows. for cid, parent in [(c1, None), (c2, c1), (c3, c2), (c4, c3)]: await _seed_commit_row(db_session, cid, [parent] if parent else [], snaps[cid]) await _build_commit_graph_from_raw(db_session, [_raw_commit(c5, c4, snaps[c5])]) await db_session.flush() expected = {c1: 0, c2: 1, c3: 2, c4: 3, c5: 4} for cid, exp_gen in expected.items(): actual = await _graph_gen(db_session, cid) assert actual == exp_gen, ( f"commit {cid[:16]} expected generation {exp_gen}; got {actual}. " f"Backfill must compute bottom-up from the root." ) # --------------------------------------------------------------------------- # MWP1_08 — Phase 2: missing-from-musehub_commits raises integrity error # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_08_absent_parent_raises_integrity_error( db_session: AsyncSession, ) -> None: """GREEN: pushing a commit whose parent is absent from musehub_commits raises ValueError with the [MWP1] integrity error tag. C2 claims C1 as its parent, but C1 is NOT in musehub_commits. The fast-forward push invariant requires every referenced commit to already be on the server. """ phantom = _cid("08-phantom") # not inserted into musehub_commits c2 = _cid("08-c2") s2 = hash_snapshot({"f.txt": blob_id(b"08-2")}) with pytest.raises(ValueError, match=r"\[MWP1\] integrity error"): await _resolve_generation_with_backfill( db_session, [phantom], inline_gen={}, db_gen={}, ) # Also verify via the full build path (C2 -> phantom, phantom absent). with pytest.raises(ValueError, match=r"\[MWP1\] integrity error"): await _build_commit_graph_from_raw( db_session, [_raw_commit(c2, phantom, s2)] ) # --------------------------------------------------------------------------- # MWP1_09 — Phase 2 stress: 1 000-commit chain backfills correctly and bounded # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_09_stress_1000_commit_chain( db_session: AsyncSession, ) -> None: """GREEN: backfill over a 1,000-commit missing chain completes with correct generations and stops at the first anchor (a commit already in the graph). Scenario: - C0 in musehub_commit_graph with generation=0 (anchor). - C1..C999 in musehub_commits only — no graph rows. - Call _resolve_generation_with_backfill([C999]). Expected: C999=999. The walk must stop at C0 (the anchor) without walking its parents (C0 has no parents, but the point is the graph lookup short-circuits). The chain must be fully backfilled: C1=1, C500=500, C999=999. """ n = 1000 # Build commit IDs: C0 is the root anchor; C1..C999 are ungraphed. chain = [_cid(f"09-c{i:04d}") for i in range(n)] snap = hash_snapshot({"f.txt": blob_id(b"09-root")}) # Seed musehub_commits for ALL n commits. for i, cid in enumerate(chain): parent = [chain[i - 1]] if i > 0 else [] await _seed_commit_row(db_session, cid, parent, snap) # Seed musehub_commit_graph ONLY for C0 (the anchor). await _seed_graph_row(db_session, chain[0], [], 0, snap) await db_session.flush() # Backfill from C999 upward. result = await _resolve_generation_with_backfill( db_session, [chain[n - 1]], inline_gen={}, db_gen={}, ) await db_session.flush() assert result == n - 1, ( f"_resolve_generation_with_backfill([C{n-1}]) must return {n - 1}; got {result}" ) # Spot-check a few points in the chain. for i in [1, 100, 500, 999]: actual = await _graph_gen(db_session, chain[i]) assert actual == i, ( f"C{i} must have generation {i} after backfill; got {actual}" ) # Anchor C0 must not have been modified (it was already gen=0). assert await _graph_gen(db_session, chain[0]) == 0, ( "Anchor commit C0 must remain at generation 0 after backfill." ) # --------------------------------------------------------------------------- # MWP1_10 — Phase 3: repair routine corrects pre-existing corrupt rows # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_10_repair_corrects_corrupt_rows( db_session: AsyncSession, ) -> None: """GREEN: repair_corrupt_commit_generations fixes graph rows that were written with generation=0 but have non-empty parent_ids (the RC-1 bug artefacts). Scenario mirrors a database state left behind by the pre-Phase-2 push path: - C1: graph row gen=0, parent_ids=[] (true root — must NOT be touched) - C2: graph row gen=0, parent_ids=[C1] (corrupt — parent is a true root) - C3: graph row gen=0, parent_ids=[C2] (corrupt — parent is also corrupt) - C4: graph row gen=0, parent_ids=[C3] (corrupt — deepest leaf) After repair: C1=0 (unchanged), C2=1, C3=2, C4=3. """ c1, c2, c3, c4 = (_cid(f"10-c{i}") for i in range(1, 5)) snap = hash_snapshot({"f.txt": blob_id(b"10-root")}) # Pre-seed corrupt graph rows (simulating the old buggy push output). # ALL four rows have generation=0 — C1 is a true root, C2/C3/C4 are corrupt. await _seed_graph_row(db_session, c1, [], 0, snap) # true root — correct await _seed_graph_row(db_session, c2, [c1], 0, snap) # corrupt await _seed_graph_row(db_session, c3, [c2], 0, snap) # corrupt await _seed_graph_row(db_session, c4, [c3], 0, snap) # corrupt await db_session.flush() result = await repair_corrupt_commit_generations(db_session) await db_session.flush() assert result["total_corrupt"] == 3, ( f"Repair must identify exactly 3 corrupt rows (C2,C3,C4); got {result}" ) assert result["repaired"] == 3, ( f"Repair must correct all 3 corrupt rows; got {result}" ) # Verify correct generations after repair. expected = {c1: 0, c2: 1, c3: 2, c4: 3} for cid, exp_gen in expected.items(): actual = await _graph_gen(db_session, cid) assert actual == exp_gen, ( f"After repair: commit {cid[:16]} expected gen={exp_gen}; got {actual}" ) # --------------------------------------------------------------------------- # MWP1_11 — Phase 3: re-running the repair is a no-op (idempotent) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_11_repair_is_idempotent( db_session: AsyncSession, ) -> None: """GREEN: calling repair_corrupt_commit_generations twice is a no-op on the second call — it finds zero corrupt rows and returns immediately. Uses the same scenario as MWP1_10 to confirm that after the first repair pass all previously-corrupt rows now have correct non-zero generations, so the WHERE generation=0 AND cardinality(parent_ids)>0 predicate matches nothing. """ c1, c2, c3 = (_cid(f"11-c{i}") for i in range(1, 4)) snap = hash_snapshot({"f.txt": blob_id(b"11-root")}) await _seed_graph_row(db_session, c1, [], 0, snap) # true root await _seed_graph_row(db_session, c2, [c1], 0, snap) # corrupt await _seed_graph_row(db_session, c3, [c2], 0, snap) # corrupt await db_session.flush() # First call — repairs C2 and C3. first = await repair_corrupt_commit_generations(db_session) await db_session.flush() assert first["total_corrupt"] == 2 assert first["repaired"] == 2 # Second call — must find nothing to repair. second = await repair_corrupt_commit_generations(db_session) await db_session.flush() assert second["total_corrupt"] == 0, ( f"Second repair call must find 0 corrupt rows; got {second}" ) assert second["repaired"] == 0, ( f"Second repair call must repair 0 rows; got {second}" ) # Generations must still be correct after both passes. assert await _graph_gen(db_session, c1) == 0 assert await _graph_gen(db_session, c2) == 1 assert await _graph_gen(db_session, c3) == 2 # --------------------------------------------------------------------------- # MWP1_12 — Phase 4: data-integrity invariant # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_12_commit_graph_invariant( db_session: AsyncSession, ) -> None: """GREEN: check_commit_graph_invariant detects corrupt rows (gen=0 with parents) and reports clean after repair_corrupt_commit_generations fixes them. This is the assertable invariant for CI / verify: no musehub_commit_graph row may have generation==0 AND non-empty parent_ids. """ c1, c2, c3 = (_cid(f"12-c{i}") for i in range(1, 4)) snap = hash_snapshot({"f.txt": blob_id(b"12-root")}) # Seed: C1 is a true root (gen=0, no parents — valid). # C2 and C3 are corrupt (gen=0 with parents). await _seed_graph_row(db_session, c1, [], 0, snap) # valid root await _seed_graph_row(db_session, c2, [c1], 0, snap) # corrupt await _seed_graph_row(db_session, c3, [c2], 0, snap) # corrupt await db_session.flush() # Invariant must report violations before repair. pre = await check_commit_graph_invariant(db_session) assert pre["valid"] is False, ( f"Invariant must detect corrupt rows before repair; got {pre}" ) assert pre["violations"] == 2, ( f"Invariant must count exactly 2 violations (C2, C3); got {pre}" ) # Repair, then re-check. await repair_corrupt_commit_generations(db_session) await db_session.flush() post = await check_commit_graph_invariant(db_session) assert post["valid"] is True, ( f"Invariant must pass after repair; got {post}" ) assert post["violations"] == 0, ( f"Invariant must report 0 violations after repair; got {post}" ) # --------------------------------------------------------------------------- # MWP1_13 — Phase 4: fetch-side guard fires on snapshot mismatch # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_13_fetch_side_guard_fires_on_snapshot_mismatch( db_session: AsyncSession, ) -> None: """GREEN: wire_fetch_mpack calls repair_corrupt_commit_generations when the CommitGraph's max-gen snapshot disagrees with the authoritative MusehubCommit snapshot (the BLOB-DEBUG mismatch condition). Setup: - tip in CommitGraph: generation=0, snapshot_id=OLD_SNAP (stale — RC-1 artefact) - tip in MusehubCommit: snapshot_id=NEW_SNAP (authoritative) The guard detects want_tip_snap_id (OLD_SNAP) ∉ want_snap_ids_from_commit_rows ({NEW_SNAP}) and calls repair_corrupt_commit_generations before assembling. """ repo = await create_repo( db_session, name="mwp1-13", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() tip = _cid("13-tip") old_snap = hash_snapshot({"old.txt": blob_id(b"13-old")}) new_snap = hash_snapshot({"new.txt": blob_id(b"13-new")}) # Authoritative DAG: tip has new_snap (what the client should receive). await _seed_commit_row(db_session, tip, [], new_snap) # CommitGraph: tip has gen=0 with OLD snapshot (the stale RC-1 state). await _seed_graph_row(db_session, tip, [], 0, old_snap) await db_session.flush() backend = _mock_backend(b"") with patch( "musehub.services.musehub_wire_fetch.repair_corrupt_commit_generations", new_callable=AsyncMock, ) as mock_repair, \ patch("musehub.services.musehub_wire.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): mock_repair.return_value = {"total_corrupt": 0, "repaired": 0} try: await wire_fetch_mpack( db_session, repo.repo_id, want=[tip], have=[], force_build=True, ) except Exception: pass # function may fail on missing blobs — only the guard call matters mock_repair.assert_called_once_with(db_session) # --------------------------------------------------------------------------- # MWP1_14 — Phase 5: end-to-end regression proof # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp1_14_end_to_end_push_clone_push_clone( db_session: AsyncSession, ) -> None: """GREEN: push C1→C2→C3, clone, push C4, clone again → the second clone contains all 4 commits (C1 through C4) and C4's blob. This is the acceptance gate for MWP-1. It directly reproduces the clone-after-push staleness (RC-1 bug) and proves the fix holds: 1. Push C1/C2/C3 individually. 2. Corrupt C3's CommitGraph row (delete it) — simulates the backfill gap that the RC-1 bug created: a parent that is on the server in musehub_commits but absent from musehub_commit_graph. 3. Push C4 (parent C3). - WITHOUT the fix: C3 not in graph → parent_gens empty → C4 gets gen=0. The range scan (gen ≤ max_want_gen=0) returns only C4 (and possibly C1) so the clone is missing C2 and C3. - WITH the fix: _resolve_generation_with_backfill restores C3=2, so C4 gets gen=3 and the full history is returned. 4. Clone with want=[C4], have=[] → assert all 4 commits and C4's blob. """ repo = await create_repo( db_session, name="mwp1-14", owner=_OWNER, owner_user_id=_IDENTITY_ID, visibility="public", initialize=False, ) await db_session.commit() c1, c2, c3, c4 = (_cid(f"14-c{i}") for i in range(1, 5)) oid1 = blob_id(b"14-c1-blob") oid2 = blob_id(b"14-c2-blob") oid3 = blob_id(b"14-c3-blob") oid4 = blob_id(b"14-c4-blob") s1 = hash_snapshot({"f1.txt": oid1}) s2 = hash_snapshot({"f2.txt": oid2}) s3 = hash_snapshot({"f3.txt": oid3}) s4 = hash_snapshot({"f4.txt": oid4}) backend = _InMemBackend() def _single_commit_mpack( cid: str, parent: str | None, snap_id: str, fpath: str, oid: str, data: bytes, ) -> tuple[bytes, str]: 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 return mpack_bytes, mpack_key async def _push( cid: str, parent: str | None, snap_id: str, fpath: str, oid: str, data: bytes, ) -> None: _, mpack_key = _single_commit_mpack(cid, parent, snap_id, fpath, oid, data) with patch("musehub.services.musehub_wire.get_backend", return_value=backend), \ patch("musehub.services.musehub_wire_push.get_backend", return_value=backend), \ patch("musehub.storage.backends.get_backend", return_value=backend): await wire_push_unpack_mpack( db_session, repo.repo_id, mpack_key, pusher_id=_OWNER, branch="main", head_commit_id=cid, commits_count=1, blobs_count=1, force=True, ) await db_session.flush() # Step 1: push C1, C2, C3 in sequence. await _push(c1, None, s1, "f1.txt", oid1, b"14-c1-blob") await _push(c2, c1, s2, "f2.txt", oid2, b"14-c2-blob") await _push(c3, c2, s3, "f3.txt", oid3, b"14-c3-blob") # Verify correct generations after initial pushes. assert await _graph_gen(db_session, c1) == 0 assert await _graph_gen(db_session, c2) == 1 assert await _graph_gen(db_session, c3) == 2 # Step 2: simulate the RC-1 backfill gap — delete C3 from CommitGraph. # This is the state that the pre-fix push path produced when a parent's # graph row was missing: the next push would compute gen=0 for C4. await db_session.execute( _sa_delete(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == c3) ) await db_session.flush() assert await _graph_gen(db_session, c3) is None, "C3 must be absent from graph after deletion" # Step 3: push C4 (parent C3). Phase 2 fix backfills C3=2, so C4 gets gen=3. await _push(c4, c3, s4, "f4.txt", oid4, b"14-c4-blob") gen_c3_after = await _graph_gen(db_session, c3) gen_c4 = await _graph_gen(db_session, c4) assert gen_c3_after == 2, ( f"C3 must be backfilled to generation 2 after pushing C4; got {gen_c3_after}. " f"Without the fix C3 stays absent and C4 gets gen=0." ) assert gen_c4 == 3, ( f"C4 must receive generation 3 (C3+1); got {gen_c4}. " f"A gen-0 tip causes the range scan to miss C1/C2/C3 in the second clone." ) # Step 4: clone with want=[C4], have=[] — the second clone must include all history. fetch_result: dict | None = None with patch("musehub.services.musehub_wire.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): fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[c4], have=[], force_build=True, ) assert fetch_result is not None assert fetch_result["mpack_id"] is not None, "Fetch must return a mpack_id" assembled_mpack_id = fetch_result["mpack_id"] assert assembled_mpack_id in backend._mpacks, ( "Assembled mpack must be stored in the backend" ) assembled_bytes = backend._mpacks[assembled_mpack_id] parsed = parse_wire_mpack(assembled_bytes) cloned_commit_ids = {c["commit_id"] for c in parsed.get("commits", [])} cloned_blob_oids = {b["object_id"] for b in parsed.get("blobs", [])} # Primary acceptance gate: C4 and its blob must be in the second clone. assert c4 in cloned_commit_ids, ( f"C4 (the just-pushed commit) must be in the second clone; " f"commits found: {[cid[:16] for cid in cloned_commit_ids]}" ) assert oid4 in cloned_blob_oids, ( f"C4's blob (oid4) must be in the second clone's blobs; " f"blobs found: {[o[:16] for o in cloned_blob_oids]}" ) # Full-history gate: a fresh clone must include the complete ancestor chain. # WITHOUT the fix: C4 gets gen=0, max_want_gen=0, range scan misses C2/C3 # (gen=1,2 are > max_want_gen=0), so they are absent from the clone. assert c1 in cloned_commit_ids, ( f"C1 must be in the second clone (full history); " f"WITHOUT the fix this fails because C4 gets gen=0 and the range scan " f"(gen <= 0) misses C1's ancestors. commits: {[cid[:16] for cid in cloned_commit_ids]}" ) assert c2 in cloned_commit_ids, ( f"C2 must be in the second clone (full history); got {[cid[:16] for cid in cloned_commit_ids]}" ) assert c3 in cloned_commit_ids, ( f"C3 must be in the second clone (full history); got {[cid[:16] for cid in cloned_commit_ids]}" )