gabriel / musehub public
test_mwp2_walk_fallback.py python
947 lines 36.2 KB
Raw
sha256:34035d72cef530c1ab9d6a6f53be18d803dde705fc3157617d70352a96d0747b Merge branch 'fix/wire-push-external-parent-manifest' into dev Human 8 days ago
1 """MWP-2 Phase 0 (RED) — _walk_commit_delta correctness fallback.
2
3 Sub-ticket: musehub#107 — https://staging.musehub.ai/gabriel/musehub/issues/107
4 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58
5 Predecessor: musehub#106 (MWP-1, generation authority) — closed.
6
7 Root cause (RC-2): ``_walk_commit_delta`` (musehub_wire_fetch.py:49) pins
8 ``if True:`` at line 66, making the authoritative MusehubCommit DAG walk (lines
9 159–184) dead code. The live path is a generation-bounded range scan:
10
11 generation > min_have_gen AND generation <= max_want_gen
12
13 This is correct only when every generation in ``musehub_commit_graph`` is
14 correct. When a generation is wrong or a graph row is missing, the range scan
15 either excludes real ancestors (corrupt low generation causes max_want_gen to
16 understate the range) or the BFS dead-ends (missing graph row yields empty
17 parents from ``graph_map.get(cid, ([], None, 0))``). Either way
18 ``_walk_commit_delta`` silently returns a truncated commit set, the assembled
19 mpack is missing commits and blobs, and clone exits 0 with a stale working tree.
20
21 MWP-1 (musehub#106) fixed the write side: no new commit is ever written with a
22 wrong generation, and ``repair_corrupt_commit_generations`` heals pre-existing
23 artefacts. MWP-2 fixes the read side: ``_walk_commit_delta`` must detect that
24 its BFS is incomplete and fall back to the authoritative parent-pointer walk,
25 so the *current* response is always complete regardless of graph state.
26
27 Expected Phase 0 status:
28 - MWP2_01 RED — corrupt generation excludes ancestors; truncated result
29 - MWP2_02 RED — missing graph row dead-ends BFS; missing commits in result
30 - MWP2_03 GREEN — fully consistent graph uses fast path; no fallback triggered
31
32 The fix (Phases 1–2) adds a ``graph_incomplete`` flag to the BFS and calls a
33 promoted ``_walk_commit_delta_dag`` helper when the flag is set.
34 """
35 from __future__ import annotations
36
37 import datetime
38 import logging
39 from unittest.mock import AsyncMock, MagicMock, patch
40
41 import pytest
42 from sqlalchemy import delete as _sa_delete, insert as _sa_insert, select
43 from sqlalchemy.ext.asyncio import AsyncSession
44
45 from muse.core.ids import hash_snapshot
46 from muse.core.mpack import build_wire_mpack, parse_wire_mpack
47 from muse.core.types import blob_id
48 from musehub.core.genesis import compute_identity_id
49 from musehub.db.musehub_repo_models import (
50 MusehubCommit,
51 MusehubCommitGraph,
52 )
53 from musehub.services.musehub_repository import create_repo
54 from musehub.services.musehub_wire_fetch import (
55 _walk_commit_delta,
56 _walk_commit_delta_dag,
57 wire_fetch_mpack,
58 )
59 from musehub.services.musehub_wire_push import wire_push_unpack_mpack
60
61 _OWNER = "gabriel"
62 _IDENTITY_ID = compute_identity_id(b"gabriel")
63
64
65 # ---------------------------------------------------------------------------
66 # Helpers — mirrors test_mwp1_generation_authority.py conventions
67 # ---------------------------------------------------------------------------
68
69
70 def _cid(seed: str) -> str:
71 """Deterministic sha256: commit id from a seed string."""
72 return blob_id(f"mwp2-commit-{seed}".encode())
73
74
75 def _sid(seed: str) -> str:
76 """Deterministic snapshot id."""
77 return hash_snapshot({f"{seed}.txt": blob_id(seed.encode())})
78
79
80 def _now() -> datetime.datetime:
81 return datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
82
83
84 async def _seed_commit(
85 session: AsyncSession,
86 cid: str,
87 parent_ids: list[str],
88 snap_id: str,
89 ) -> None:
90 """Insert a musehub_commits row (authoritative DAG) with no graph row."""
91 session.add(MusehubCommit(
92 commit_id=cid,
93 branch="main",
94 message=f"commit {cid[:12]}",
95 author=_OWNER,
96 timestamp=_now(),
97 parent_ids=parent_ids,
98 snapshot_id=snap_id,
99 ))
100 await session.flush()
101
102
103 async def _seed_graph(
104 session: AsyncSession,
105 cid: str,
106 parent_ids: list[str],
107 generation: int,
108 snap_id: str,
109 ) -> None:
110 """Insert a musehub_commit_graph row with an explicit (possibly corrupt) generation."""
111 session.add(MusehubCommitGraph(
112 commit_id=cid,
113 parent_ids=parent_ids,
114 generation=generation,
115 snapshot_id=snap_id,
116 created_at=_now(),
117 ))
118 await session.flush()
119
120
121 class _InMemBackend:
122 """In-memory storage backend for E2E push→fetch tests.
123
124 Pre-seed push mpacks in ``backend._mpacks[mpack_key] = mpack_bytes``
125 BEFORE calling ``wire_push_unpack_mpack``. The fetch path assembles and
126 stores its mpack via ``put_mpack``; the test reads it back directly.
127 """
128
129 def __init__(self) -> None:
130 self._mpacks: dict[str, bytes] = {}
131
132 async def put_mpack(self, mpack_id: str, data: bytes) -> None:
133 self._mpacks[mpack_id] = data
134
135 async def get_mpack(self, mpack_id: str) -> bytes | None:
136 return self._mpacks.get(mpack_id)
137
138 async def presign_mpack_get(self, mpack_id: str, ttl: int = 3600) -> str:
139 return f"inmem://{mpack_id}"
140
141 async def put(self, oid: str, data: bytes) -> str:
142 return oid
143
144 async def get(self, oid: str) -> bytes | None:
145 return None
146
147 async def presign_get(self, oid: str, ttl: int = 3600) -> str:
148 return f"inmem://{oid}"
149
150 async def quarantine_mpack(self, *args: object, **kwargs: object) -> None:
151 pass
152
153
154 def _raw_commit(
155 cid: str,
156 parent: str | None,
157 snap_id: str,
158 *,
159 branch: str = "main",
160 ) -> dict:
161 """Raw commit wire dict in the shape wire_push_unpack_mpack consumes."""
162 return {
163 "commit_id": cid,
164 "branch": branch,
165 "message": f"commit {cid[:12]}",
166 "author": _OWNER,
167 "committed_at": _now().isoformat(),
168 "parent_commit_id": parent,
169 "parent2_commit_id": None,
170 "snapshot_id": snap_id,
171 "agent_id": "",
172 "model_id": "",
173 "toolchain_id": "",
174 "sem_ver_bump": "none",
175 "breaking_changes": [],
176 "signature": "",
177 "signer_key_id": "",
178 "signer_public_key": "",
179 "prompt_hash": "",
180 }
181
182
183 # ---------------------------------------------------------------------------
184 # MWP2_01 — corrupt generation excludes ancestors from range scan (RED)
185 # ---------------------------------------------------------------------------
186
187
188 @pytest.mark.asyncio
189 async def test_mwp2_01_corrupt_generation_truncates_walk(
190 db_session: AsyncSession,
191 ) -> None:
192 """RED: a missing interior graph row dead-ends the BFS, dropping ancestors.
193
194 Chain: C1 <- C2 <- C3 <- C4 (all in musehub_commits, authoritative).
195 Graph: C1(gen=0), C2(gen=1), C4(gen=3) — C3 has NO graph row.
196 want=[C4], have=[].
197
198 BFS: C4 is in graph_map (gen=3, parents=[C3]). C3 → graph_map.get(C3)
199 returns default ([], None, 0) → BFS treats C3 as a leaf and stops.
200 C2 and C1 are never added to the reachable set.
201
202 Result without fix: {C4, C3} — C2 and C1 missing.
203 Result with fix: {C4, C3, C2, C1}.
204
205 This is the canonical post-MWP-1 RC-2 scenario: the tip has a correct high
206 generation, but an interior ancestor's graph row was never written (a push
207 that wrote musehub_commits but not the graph row, or a pre-backfill gap).
208 MWP2_02 covers the case where the *start tip itself* is absent from the graph.
209 """
210 c1 = _cid("01-c1")
211 c2 = _cid("01-c2")
212 c3 = _cid("01-c3")
213 c4 = _cid("01-c4")
214 s1, s2, s3, s4 = _sid("01-s1"), _sid("01-s2"), _sid("01-s3"), _sid("01-s4")
215
216 # Authoritative DAG — all four commits exist in musehub_commits.
217 await _seed_commit(db_session, c1, [], s1)
218 await _seed_commit(db_session, c2, [c1], s2)
219 await _seed_commit(db_session, c3, [c2], s3)
220 await _seed_commit(db_session, c4, [c3], s4)
221
222 # Graph: C1, C2 present with correct generations; C3 absent (the backfill
223 # gap that produces the BFS dead-end); C4 present with correct generation
224 # relative to the true chain (gen=3).
225 await _seed_graph(db_session, c1, [], 0, s1)
226 await _seed_graph(db_session, c2, [c1], 1, s2)
227 # C3 deliberately NOT seeded in graph.
228 await _seed_graph(db_session, c4, [c3], 3, s4)
229 await db_session.commit()
230
231 result = await _walk_commit_delta(db_session, want=[c4], have=[])
232
233 assert c4 in result, "C4 (want tip) must always be in result"
234 assert c3 in result, (
235 f"C3 must be in result — it is a direct parent of C4. "
236 f"It is missing because graph_map.get(C3) returned empty parents "
237 f"(C3 has no graph row), so the BFS dead-ended at C3. "
238 f"Got result keys: {sorted(result.keys())}"
239 )
240 assert c2 in result, (
241 f"C2 must be in result — ancestor of C4 reachable via C3. "
242 f"Got result keys: {sorted(result.keys())}"
243 )
244 assert c1 in result, (
245 f"C1 must be in result — root ancestor of C4. "
246 f"Got result keys: {sorted(result.keys())}"
247 )
248
249
250 # ---------------------------------------------------------------------------
251 # MWP2_02 — missing graph row for the start tip dead-ends BFS (RED)
252 # ---------------------------------------------------------------------------
253
254
255 @pytest.mark.asyncio
256 async def test_mwp2_02_missing_start_tip_graph_row(
257 db_session: AsyncSession,
258 ) -> None:
259 """RED: when the start tip (a want commit) has no graph row at all, the
260 BFS dead-ends immediately and returns an empty or incomplete set.
261
262 Setup:
263 musehub_commits: C1(root) <- C2 <- C3
264 musehub_commit_graph: C1(gen=0), C2(gen=1) — C3 has NO graph row.
265 want=[C3], have=[].
266
267 max_want_gen: C3 not in graph → query returns None → max_want_gen = 0.
268 Range scan: gen > -1 AND gen <= 0 → only C1 (gen=0) returned in graph_map.
269 BFS: starts from C3; C3 not in graph_map → treated as having empty parents;
270 BFS adds C3 to reachable but never follows its parents (C2, C1).
271 Result without fix: {C3} only. Missing: C2, C1.
272 Result with fix: {C3, C2, C1}.
273 """
274 c1 = _cid("02-c1")
275 c2 = _cid("02-c2")
276 c3 = _cid("02-c3")
277 s1, s2, s3 = _sid("02-s1"), _sid("02-s2"), _sid("02-s3")
278
279 # Authoritative DAG.
280 await _seed_commit(db_session, c1, [], s1)
281 await _seed_commit(db_session, c2, [c1], s2)
282 await _seed_commit(db_session, c3, [c2], s3)
283
284 # Graph: C1 and C2 present; C3 (the want tip) absent.
285 await _seed_graph(db_session, c1, [], 0, s1)
286 await _seed_graph(db_session, c2, [c1], 1, s2)
287 await db_session.commit()
288
289 result = await _walk_commit_delta(db_session, want=[c3], have=[])
290
291 assert c3 in result, "C3 (want tip) must be in result"
292 assert c2 in result, (
293 f"C2 must be in result — parent of C3 (the want tip). "
294 f"C3 had no graph row, so max_want_gen=0 and the range scan only "
295 f"returned C1. BFS from C3 found no parents in graph_map. "
296 f"Got result keys: {sorted(result.keys())}"
297 )
298 assert c1 in result, (
299 f"C1 must be in result — root ancestor reachable from C3. "
300 f"Got result keys: {sorted(result.keys())}"
301 )
302
303
304 # ---------------------------------------------------------------------------
305 # MWP2_03 — consistent graph uses fast path, no fallback invoked (GREEN)
306 # ---------------------------------------------------------------------------
307
308
309 @pytest.mark.asyncio
310 async def test_mwp2_03_consistent_graph_uses_fast_path(
311 db_session: AsyncSession,
312 ) -> None:
313 """GREEN guard: a fully consistent graph must return the complete ancestor
314 closure via the fast path, and ``_walk_commit_delta_dag`` must NOT be called.
315
316 This test pins the happy-path contract so later phases cannot accidentally
317 make the fast path degrade into always calling the fallback (which would be
318 correct but slow).
319
320 After Phase 2 the ``_walk_commit_delta_dag`` helper will exist; this test
321 patches it to detect if it is invoked. Before Phase 1 it doesn't exist, so
322 we patch ``musehub.graph.walk.walk_dag_async`` (the underlying function the
323 legacy walk calls) as a proxy — if the fast path is truly live and exclusive,
324 walk_dag_async should never be called during _walk_commit_delta.
325
326 Note: after Phase 1 promotes _walk_commit_delta_dag to a real named function,
327 this test should be updated to patch that directly. For Phase 0 we patch
328 walk_dag_async as the canary.
329
330 Setup:
331 musehub_commits: C1(root) <- C2 <- C3
332 musehub_commit_graph: C1(gen=0), C2(gen=1), C3(gen=2) — fully consistent.
333 want=[C3], have=[].
334
335 Expected: result contains C1, C2, C3 and walk_dag_async is not called.
336 """
337 c1 = _cid("03-c1")
338 c2 = _cid("03-c2")
339 c3 = _cid("03-c3")
340 s1, s2, s3 = _sid("03-s1"), _sid("03-s2"), _sid("03-s3")
341
342 await _seed_commit(db_session, c1, [], s1)
343 await _seed_commit(db_session, c2, [c1], s2)
344 await _seed_commit(db_session, c3, [c2], s3)
345
346 await _seed_graph(db_session, c1, [], 0, s1)
347 await _seed_graph(db_session, c2, [c1], 1, s2)
348 await _seed_graph(db_session, c3, [c2], 2, s3)
349 await db_session.commit()
350
351 with patch(
352 "musehub.services.musehub_wire_fetch._walk_commit_delta_dag",
353 new_callable=AsyncMock,
354 ) as mock_dag:
355 mock_dag.side_effect = AssertionError(
356 "_walk_commit_delta invoked the DAG fallback (_walk_commit_delta_dag) "
357 "on a fully consistent graph — the fast path must be used exclusively."
358 )
359
360 result = await _walk_commit_delta(db_session, want=[c3], have=[])
361
362 assert c3 in result, f"C3 must be in result; got {sorted(result.keys())}"
363 assert c2 in result, f"C2 must be in result; got {sorted(result.keys())}"
364 assert c1 in result, f"C1 must be in result; got {sorted(result.keys())}"
365
366
367 # ---------------------------------------------------------------------------
368 # MWP2_05 — _walk_commit_delta_dag returns full closure with populated fields
369 # ---------------------------------------------------------------------------
370
371
372 @pytest.mark.asyncio
373 async def test_mwp2_05_dag_helper_returns_full_closure(
374 db_session: AsyncSession,
375 ) -> None:
376 """GREEN: _walk_commit_delta_dag over a musehub_commits chain returns the
377 complete ancestor closure of want minus have, with snapshot_id and
378 parent_ids populated on every returned row.
379
380 No graph rows are seeded — the helper walks musehub_commits.parent_ids
381 directly (the authoritative DAG), so a missing/corrupt graph is irrelevant.
382
383 Chain: C1(root) <- C2 <- C3
384 Call: _walk_commit_delta_dag(session, starts=[C3], have_set=frozenset())
385 Expect: {C3, C2, C1}, each with correct snapshot_id and parent_ids.
386 """
387 c1 = _cid("05-c1")
388 c2 = _cid("05-c2")
389 c3 = _cid("05-c3")
390 s1, s2, s3 = _sid("05-s1"), _sid("05-s2"), _sid("05-s3")
391
392 await _seed_commit(db_session, c1, [], s1)
393 await _seed_commit(db_session, c2, [c1], s2)
394 await _seed_commit(db_session, c3, [c2], s3)
395 await db_session.commit()
396
397 result = await _walk_commit_delta_dag(db_session, starts=[c3], have_set=frozenset())
398
399 assert set(result.keys()) == {c1, c2, c3}, (
400 f"Must return all three commits; got {sorted(result.keys())}"
401 )
402
403 # Every row must have snapshot_id and parent_ids populated.
404 r1, r2, r3 = result[c1], result[c2], result[c3]
405
406 assert r1.snapshot_id == s1, f"C1 snapshot_id wrong: {r1.snapshot_id}"
407 assert r2.snapshot_id == s2, f"C2 snapshot_id wrong: {r2.snapshot_id}"
408 assert r3.snapshot_id == s3, f"C3 snapshot_id wrong: {r3.snapshot_id}"
409
410 assert r1.parent_ids == [], f"C1 parent_ids wrong: {r1.parent_ids}"
411 assert r2.parent_ids == [c1], f"C2 parent_ids wrong: {r2.parent_ids}"
412 assert r3.parent_ids == [c2], f"C3 parent_ids wrong: {r3.parent_ids}"
413
414
415 # ---------------------------------------------------------------------------
416 # MWP2_06 — _walk_commit_delta_dag returns commits in parents-first topo order
417 # ---------------------------------------------------------------------------
418
419
420 @pytest.mark.asyncio
421 async def test_mwp2_06_dag_helper_parents_first_order(
422 db_session: AsyncSession,
423 ) -> None:
424 """GREEN: the dict returned by _walk_commit_delta_dag is in parents-first
425 topological order — every commit appears after all of its in-set parents.
426
427 walk_dag_async BFS order is children-before-parents (C3, C2, C1 for a
428 linear chain starting at C3). After Kahn's topo-sort the result must be
429 C1, C2, C3 — each parent strictly before its child.
430
431 Chain: C1(root) <- C2 <- C3
432 Expected key order: [C1, C2, C3]
433 """
434 c1 = _cid("06-c1")
435 c2 = _cid("06-c2")
436 c3 = _cid("06-c3")
437 s1, s2, s3 = _sid("06-s1"), _sid("06-s2"), _sid("06-s3")
438
439 await _seed_commit(db_session, c1, [], s1)
440 await _seed_commit(db_session, c2, [c1], s2)
441 await _seed_commit(db_session, c3, [c2], s3)
442 await db_session.commit()
443
444 result = await _walk_commit_delta_dag(db_session, starts=[c3], have_set=frozenset())
445
446 keys = list(result.keys())
447 assert len(keys) == 3, f"Expected 3 commits, got {len(keys)}: {keys}"
448
449 idx = {cid: i for i, cid in enumerate(keys)}
450 assert idx[c1] < idx[c2], (
451 f"C1 (root) must appear before C2 in the result. "
452 f"Got order: C1={idx[c1]}, C2={idx[c2]}, C3={idx[c3]}"
453 )
454 assert idx[c2] < idx[c3], (
455 f"C2 must appear before C3 in the result. "
456 f"Got order: C1={idx[c1]}, C2={idx[c2]}, C3={idx[c3]}"
457 )
458
459
460 # ---------------------------------------------------------------------------
461 # MWP2_09 — have boundary is honored under the DAG fallback (GREEN)
462 # ---------------------------------------------------------------------------
463
464
465 @pytest.mark.asyncio
466 async def test_mwp2_09_have_boundary_honored_under_fallback(
467 db_session: AsyncSession,
468 ) -> None:
469 """GREEN: when the fallback fires, the have set is respected — commits in
470 have and their ancestors are excluded from the result, same as the fast path.
471
472 Chain: C1(root) <- C2 <- C3 <- C4
473 Graph: C1(gen=0), C2(gen=1), C4(gen=3) — C3 absent (triggers fallback).
474 want=[C4], have=[C2].
475
476 Fast-path sequence (pre-fallback):
477 max_want_gen=3, min_have_gen=1 (gen of C2).
478 Range scan gen=(1,3] → only C4(gen=3). C3 absent → graph_incomplete=True.
479 Dispatches to _walk_commit_delta_dag(session, starts=[C4], have_set={C2}).
480
481 DAG walk: BFS from C4 with exclude={C2}.
482 C4 visited → adj=[C3] → C3 visited → adj=[C2] (excluded, already in seen).
483 C2 never yielded. C1 never queued.
484 Kahn's sort: {C3, C4} in parents-first order.
485
486 Assert: {C3, C4} in result; C1 and C2 absent.
487 """
488 c1 = _cid("09-c1")
489 c2 = _cid("09-c2")
490 c3 = _cid("09-c3")
491 c4 = _cid("09-c4")
492 s1, s2, s3, s4 = _sid("09-s1"), _sid("09-s2"), _sid("09-s3"), _sid("09-s4")
493
494 # Authoritative DAG — all four commits.
495 await _seed_commit(db_session, c1, [], s1)
496 await _seed_commit(db_session, c2, [c1], s2)
497 await _seed_commit(db_session, c3, [c2], s3)
498 await _seed_commit(db_session, c4, [c3], s4)
499
500 # Graph: C1, C2, C4 present; C3 absent → triggers fallback.
501 await _seed_graph(db_session, c1, [], 0, s1)
502 await _seed_graph(db_session, c2, [c1], 1, s2)
503 await _seed_graph(db_session, c4, [c3], 3, s4)
504 await db_session.commit()
505
506 result = await _walk_commit_delta(db_session, want=[c4], have=[c2])
507
508 assert c4 in result, f"C4 (want tip) must be in result; got {sorted(result.keys())}"
509 assert c3 in result, f"C3 must be in result (it is between the tip and the have boundary)"
510 assert c2 not in result, (
511 f"C2 is in have — it must be excluded from the result; "
512 f"got {sorted(result.keys())}"
513 )
514 assert c1 not in result, (
515 f"C1 is an ancestor of have — it must be excluded from the result; "
516 f"got {sorted(result.keys())}"
517 )
518
519
520 # ---------------------------------------------------------------------------
521 # MWP2_10 — max_nodes bound is wired into walk_dag_async + warning fires (GREEN)
522 # ---------------------------------------------------------------------------
523
524
525 @pytest.mark.asyncio
526 async def test_mwp2_10_max_nodes_bound_is_wired(
527 db_session: AsyncSession,
528 caplog: pytest.LogCaptureFixture,
529 ) -> None:
530 """GREEN (Phase 3): two assertions about the max_nodes cap in _walk_commit_delta_dag.
531
532 Part A — kwarg wired: walk_dag_async is called with max_nodes=100_000.
533 Setup: a 2-commit chain with no graph rows → triggers _walk_commit_delta_dag.
534 A wrapper around walk_dag_async captures the kwargs.
535
536 Part B — warning fires: when walk_dag_async yields exactly 100_000 commits,
537 the `[MWP2] ... max_nodes cap hit` warning is emitted.
538 A mock replaces walk_dag_async to yield 100_000 fake cid strings (not in DB).
539 Kahn's sort runs over them (all in_degree=0, empty result dict).
540 The deque-based Kahn's queue keeps this O(n), not O(n²).
541 """
542 import logging as _logging
543 from unittest.mock import patch as _patch
544
545 # ---- Part A: max_nodes=100_000 kwarg reaches walk_dag_async ----
546 c1 = _cid("10a-c1")
547 c2 = _cid("10a-c2")
548 s1, s2 = _sid("10a-s1"), _sid("10a-s2")
549
550 await _seed_commit(db_session, c1, [], s1)
551 await _seed_commit(db_session, c2, [c1], s2)
552 # No graph rows — forces _walk_commit_delta_dag.
553
554 received: dict[str, object] = {}
555
556 from musehub.graph import walk as _walk_mod
557
558 _original_walk = _walk_mod.walk_dag_async
559
560 async def _capturing_walk(starts, adj_fn, *, exclude=None, max_nodes=None):
561 received["max_nodes"] = max_nodes
562 async for item in _original_walk(starts, adj_fn, exclude=exclude, max_nodes=max_nodes):
563 yield item
564
565 with _patch.object(_walk_mod, "walk_dag_async", new=_capturing_walk):
566 result_a = await _walk_commit_delta(db_session, want=[c2], have=[])
567
568 assert received.get("max_nodes") == 100_000, (
569 f"_walk_commit_delta_dag must pass max_nodes=100_000 to walk_dag_async; "
570 f"got max_nodes={received.get('max_nodes')!r}"
571 )
572 assert c1 in result_a and c2 in result_a, (
573 f"Full result still expected even with wrapping; got {sorted(result_a.keys())}"
574 )
575
576 # ---- Part B: WARNING fires when reachable count hits the cap ----
577 # Mock walk_dag_async to yield exactly 100_000 fake cids (not in DB).
578 # Kahn's sort sees all in_degree=0 (no _row_cache entries → no parent edges).
579 # With deque.popleft() this is O(n); with list.pop(0) it would be O(n²).
580 async def _capped_walk(starts, adj_fn, *, exclude=None, max_nodes=None):
581 for i in range(100_000):
582 yield f"sha256:{'a' * 60}{i:04d}"
583
584 with _patch.object(_walk_mod, "walk_dag_async", new=_capped_walk):
585 with caplog.at_level(_logging.WARNING, logger="musehub.services.musehub_wire_fetch"):
586 await _walk_commit_delta_dag(
587 db_session,
588 starts=[c2],
589 have_set=frozenset(),
590 )
591
592 assert any("max_nodes cap hit" in r.message for r in caplog.records), (
593 "WARNING must be logged when reachable count reaches max_nodes=100_000. "
594 f"Log records captured: {[r.message for r in caplog.records]}"
595 )
596
597
598 # ---------------------------------------------------------------------------
599 # MWP2_11 — 5,000-commit stress: full fallback, correct topo order, O(n) (GREEN)
600 # ---------------------------------------------------------------------------
601
602
603 @pytest.mark.asyncio
604 async def test_mwp2_11_stress_5000_commit_linear_chain(
605 db_session: AsyncSession,
606 ) -> None:
607 """GREEN (Phase 3): a 5,000-commit linear chain with no graph rows triggers
608 the DAG fallback and returns all 5,000 commits in parents-first topo order.
609
610 No graph rows are seeded → fast-path BFS detects graph_incomplete=True at
611 the very first start (want tip absent from graph_map → max_want_gen=0).
612 Fallback walks musehub_commits.parent_ids over the entire chain.
613
614 Performance note: commits are seeded with a single flush (no commit), so
615 SQLAlchemy's identity map stays warm — session.get() hits the map for each
616 adjacency call (O(1) per call) rather than issuing 5,000 DB round trips.
617 Kahn's topo-sort uses collections.deque to guarantee O(n), not O(n²).
618
619 Assertions:
620 - result contains exactly 5,000 commits
621 - every parent appears before its child in dict iteration order
622 - result is bounded well below max_nodes=100_000
623 """
624 N = 5_000
625
626 # Build linear chain C0 ← C1 ← ... ← C4999.
627 chain_cids = [_cid(f"11-c{i:04d}") for i in range(N)]
628 chain_sids = [_sid(f"11-s{i:04d}") for i in range(N)]
629
630 for i in range(N):
631 parent_ids = [] if i == 0 else [chain_cids[i - 1]]
632 db_session.add(MusehubCommit(
633 commit_id=chain_cids[i],
634 branch="main",
635 message=f"commit {chain_cids[i][:12]}",
636 author=_OWNER,
637 timestamp=_now(),
638 parent_ids=parent_ids,
639 snapshot_id=chain_sids[i],
640 ))
641 # Single flush — keeps identity map warm so session.get() is O(1) per call.
642 # No graph rows seeded → forces full DAG fallback.
643 await db_session.flush()
644
645 tip = chain_cids[-1]
646 result = await _walk_commit_delta(db_session, want=[tip], have=[])
647
648 assert len(result) == N, (
649 f"Fallback must return all {N} commits; got {len(result)}"
650 )
651
652 # Verify parents-first topological order: every parent in the result set must
653 # appear strictly before its child in dict iteration order.
654 keys = list(result.keys())
655 idx = {cid: pos for pos, cid in enumerate(keys)}
656 violations = []
657 for cid in keys:
658 row = result[cid]
659 for pid in (row.parent_ids or []):
660 if pid in idx and idx[pid] >= idx[cid]:
661 violations.append((pid[:12], cid[:12]))
662 assert not violations, (
663 f"Topo-order violation: parent appears after child in {len(violations)} "
664 f"case(s); first few: {violations[:5]}"
665 )
666
667
668 # ---------------------------------------------------------------------------
669 # Shared E2E helpers (MWP2_12 / MWP2_13)
670 # ---------------------------------------------------------------------------
671
672
673 def _e2e_backend_patch(backend: _InMemBackend) -> tuple:
674 """Return the three backend patches needed for push + fetch E2E tests."""
675 return (
676 patch("musehub.services.musehub_wire.get_backend", return_value=backend),
677 patch("musehub.services.musehub_wire_push.get_backend", return_value=backend),
678 patch("musehub.services.musehub_wire_fetch.get_backend", return_value=backend),
679 patch("musehub.storage.backends.get_backend", return_value=backend),
680 )
681
682
683 async def _e2e_push(
684 session: AsyncSession,
685 repo_id: str,
686 backend: _InMemBackend,
687 cid: str,
688 parent: str | None,
689 snap_id: str,
690 fpath: str,
691 oid: str,
692 data: bytes,
693 ) -> None:
694 """Build a single-commit mpack, store in backend, and push via wire_push_unpack_mpack."""
695 mpack_bytes = build_wire_mpack({
696 "commits": [_raw_commit(cid, parent, snap_id)],
697 "snapshots": [{
698 "snapshot_id": snap_id,
699 "parent_snapshot_id": None,
700 "delta_upsert": {fpath: oid},
701 "delta_remove": [],
702 "directories": [],
703 }],
704 "blobs": [{"object_id": oid, "content": data}],
705 "tags": [],
706 })
707 mpack_key = blob_id(mpack_bytes)
708 backend._mpacks[mpack_key] = mpack_bytes
709
710 push_patches = _e2e_backend_patch(backend)
711 with push_patches[0], push_patches[1], push_patches[2], push_patches[3]:
712 await wire_push_unpack_mpack(
713 session,
714 repo_id,
715 mpack_key,
716 pusher_id=_OWNER,
717 branch="main",
718 head_commit_id=cid,
719 commits_count=1,
720 blobs_count=1,
721 force=True,
722 )
723 await session.flush()
724
725
726 # ---------------------------------------------------------------------------
727 # MWP2_12 — E2E acceptance gate: fallback alone ships a complete clone (GREEN)
728 # ---------------------------------------------------------------------------
729
730
731 @pytest.mark.asyncio
732 async def test_mwp2_12_e2e_fallback_alone_ships_complete_clone(
733 db_session: AsyncSession,
734 ) -> None:
735 """GREEN (Phase 4 — acceptance gate): MWP-2 fallback alone delivers a
736 complete clone even when the commit graph has a missing interior row.
737
738 Setup:
739 1. Create a repo and push C1 → C2 → C3 individually.
740 CommitGraph after pushes: C1(gen=0), C2(gen=1), C3(gen=2).
741 2. Delete C2's CommitGraph row — interior node missing.
742 3. Patch out repair_corrupt_commit_generations (MWP-1 cannot self-heal).
743 4. Clone via wire_fetch_mpack(want=[C3], have=[], force_build=True).
744
745 Without MWP-2 (the bug):
746 max_want_gen = 2 (C3 in graph). Range scan: gen=(−1,2] → C1(0), C3(2).
747 BFS: C3 → C2 (C2 not in graph_map, parents=[]) → BFS dead-end.
748 Reachable = {C3, C2}. C1 ABSENT from assembled mpack.
749
750 With MWP-2 (the fix):
751 BFS detects C2 absent from graph_map → graph_incomplete=True → fallback.
752 _walk_commit_delta_dag returns {C1, C2, C3}. All three commits in mpack.
753
754 Note: MWP-1's guard does NOT fire here because C3's graph snapshot still
755 matches C3's commit snapshot — the corruption is only a missing graph row,
756 not a wrong generation. repair_corrupt_commit_generations is patched as a
757 safety rail to prove MWP-2 is self-sufficient.
758 """
759 repo = await create_repo(
760 db_session,
761 name="mwp2-12",
762 owner=_OWNER,
763 owner_user_id=_IDENTITY_ID,
764 visibility="public",
765 initialize=False,
766 )
767 await db_session.commit()
768
769 c1, c2, c3 = _cid("12-c1"), _cid("12-c2"), _cid("12-c3")
770 oid1 = blob_id(b"12-c1-blob")
771 oid2 = blob_id(b"12-c2-blob")
772 oid3 = blob_id(b"12-c3-blob")
773 s1 = hash_snapshot({"f1.txt": oid1})
774 s2 = hash_snapshot({"f2.txt": oid2})
775 s3 = hash_snapshot({"f3.txt": oid3})
776
777 backend = _InMemBackend()
778
779 # Push C1, C2, C3 — each push writes correct CommitGraph generations.
780 await _e2e_push(db_session, repo.repo_id, backend, c1, None, s1, "f1.txt", oid1, b"12-c1-blob")
781 await _e2e_push(db_session, repo.repo_id, backend, c2, c1, s2, "f2.txt", oid2, b"12-c2-blob")
782 await _e2e_push(db_session, repo.repo_id, backend, c3, c2, s3, "f3.txt", oid3, b"12-c3-blob")
783
784 # Corrupt: delete C2's CommitGraph row (missing interior node).
785 await db_session.execute(
786 _sa_delete(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == c2)
787 )
788 await db_session.flush()
789
790 # Patch out MWP-1's repair guard to prove MWP-2 handles this alone.
791 fetch_patches = _e2e_backend_patch(backend)
792 with patch(
793 "musehub.services.musehub_wire_fetch.repair_corrupt_commit_generations",
794 new_callable=AsyncMock,
795 ) as mock_repair, \
796 fetch_patches[0], fetch_patches[2], fetch_patches[3]:
797
798 mock_repair.return_value = {"total_corrupt": 0, "repaired": 0}
799
800 fetch_result = await wire_fetch_mpack(
801 db_session,
802 repo.repo_id,
803 want=[c3],
804 have=[],
805 force_build=True,
806 )
807
808 # MWP-1 guard should NOT fire: C3's graph snapshot still matches its commit
809 # snapshot — only the missing C2 row caused truncation (RC-2, not RC-1).
810 mock_repair.assert_not_called()
811
812 assert fetch_result["mpack_id"] is not None, "wire_fetch_mpack must return a mpack_id"
813 assembled_bytes = backend._mpacks[fetch_result["mpack_id"]]
814 parsed = parse_wire_mpack(assembled_bytes)
815
816 commit_ids = {c["commit_id"] for c in parsed.get("commits", [])}
817 blob_oids = {b["object_id"] for b in parsed.get("blobs", [])}
818
819 assert c1 in commit_ids, (
820 f"C1 must be in assembled mpack — WITHOUT MWP-2 the BFS dead-ends at C2 "
821 f"(C2 absent from graph_map → parents=[]) and C1 is never reachable. "
822 f"commits present: {[cid[:16] for cid in commit_ids]}"
823 )
824 assert c2 in commit_ids, (
825 f"C2 must be in assembled mpack; got {[cid[:16] for cid in commit_ids]}"
826 )
827 assert c3 in commit_ids, (
828 f"C3 (want tip) must be in assembled mpack; got {[cid[:16] for cid in commit_ids]}"
829 )
830 assert oid3 in blob_oids, (
831 f"C3's blob (oid3) must be in assembled mpack; got {[o[:16] for o in blob_oids]}"
832 )
833
834
835 # ---------------------------------------------------------------------------
836 # MWP2_13 — E2E: fallback + graph healing → fast path on second call (GREEN)
837 # ---------------------------------------------------------------------------
838
839
840 @pytest.mark.asyncio
841 async def test_mwp2_13_e2e_fast_path_after_graph_healed(
842 db_session: AsyncSession,
843 ) -> None:
844 """GREEN (Phase 4): MWP-2 handles the corrupt state; once the graph is healed
845 by any mechanism, a follow-up _walk_commit_delta uses the fast path.
846
847 Proves MWP-1 + MWP-2 compose: the fallback is the safety net for the
848 *current* response; the fast path is restored once the graph is intact.
849
850 Setup (same as MWP2_12):
851 Push C1 → C2 → C3. Delete C2's CommitGraph row. repair NOT patched.
852
853 First clone: MWP-2 fallback fires. Result: {C1, C2, C3} — complete.
854 MWP-1 guard does NOT fire (no snapshot mismatch for this corruption type).
855
856 Graph heal: INSERT C2's CommitGraph row back (gen=1) — simulating what a
857 subsequent push or an explicit repair would do once the graph gap is noticed.
858
859 Second call (_walk_commit_delta, not full wire_fetch_mpack):
860 _walk_commit_delta_dag is patched to raise AssertionError if called.
861 With the healed graph (C1=0, C2=1, C3=2 all present), the fast-path BFS
862 completes without any missing nodes → graph_incomplete=False → no fallback.
863 The patched dag helper is never invoked — no AssertionError.
864 """
865 repo = await create_repo(
866 db_session,
867 name="mwp2-13",
868 owner=_OWNER,
869 owner_user_id=_IDENTITY_ID,
870 visibility="public",
871 initialize=False,
872 )
873 await db_session.commit()
874
875 c1, c2, c3 = _cid("13-c1"), _cid("13-c2"), _cid("13-c3")
876 oid1 = blob_id(b"13-c1-blob")
877 oid2 = blob_id(b"13-c2-blob")
878 oid3 = blob_id(b"13-c3-blob")
879 s1 = hash_snapshot({"f1.txt": oid1})
880 s2 = hash_snapshot({"f2.txt": oid2})
881 s3 = hash_snapshot({"f3.txt": oid3})
882
883 backend = _InMemBackend()
884
885 # Push C1, C2, C3.
886 await _e2e_push(db_session, repo.repo_id, backend, c1, None, s1, "f1.txt", oid1, b"13-c1-blob")
887 await _e2e_push(db_session, repo.repo_id, backend, c2, c1, s2, "f2.txt", oid2, b"13-c2-blob")
888 await _e2e_push(db_session, repo.repo_id, backend, c3, c2, s3, "f3.txt", oid3, b"13-c3-blob")
889
890 # Corrupt: delete C2's CommitGraph row (same as MWP2_12).
891 await db_session.execute(
892 _sa_delete(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == c2)
893 )
894 await db_session.flush()
895
896 # First clone — MWP-2 fallback fires; repair is NOT patched (it won't run
897 # for this corruption type, but we leave it live to prove composition).
898 fetch_patches = _e2e_backend_patch(backend)
899 with fetch_patches[0], fetch_patches[2], fetch_patches[3]:
900 fetch_result = await wire_fetch_mpack(
901 db_session,
902 repo.repo_id,
903 want=[c3],
904 have=[],
905 force_build=True,
906 )
907
908 assert fetch_result["mpack_id"] is not None
909 assembled_bytes = backend._mpacks[fetch_result["mpack_id"]]
910 parsed_first = parse_wire_mpack(assembled_bytes)
911 commit_ids_first = {c["commit_id"] for c in parsed_first.get("commits", [])}
912
913 assert {c1, c2, c3} == commit_ids_first, (
914 f"First clone must contain all 3 commits; got {[cid[:16] for cid in commit_ids_first]}"
915 )
916
917 # Heal the graph: re-insert C2's CommitGraph row with the correct generation.
918 # This simulates what a follow-up push or repair routine would do once it
919 # detects the gap. The INSERT is safe because the DELETE was already flushed.
920 await db_session.execute(
921 _sa_insert(MusehubCommitGraph).values(
922 commit_id=c2,
923 parent_ids=[c1],
924 generation=1,
925 snapshot_id=s2,
926 created_at=_now(),
927 )
928 )
929 await db_session.flush()
930
931 # Second call: _walk_commit_delta must use the fast path now.
932 # Patch _walk_commit_delta_dag to raise if called — any invocation = regression.
933 with patch(
934 "musehub.services.musehub_wire_fetch._walk_commit_delta_dag",
935 new_callable=AsyncMock,
936 ) as mock_dag:
937 mock_dag.side_effect = AssertionError(
938 "MWP2_13: _walk_commit_delta invoked the DAG fallback after the graph "
939 "was healed — the fast path must be used exclusively when graph is intact."
940 )
941 result2 = await _walk_commit_delta(db_session, want=[c3], have=[])
942
943 # Fast path returned — verify completeness (now as SimpleNamespace proxies).
944 assert {c1, c2, c3} == set(result2.keys()), (
945 f"Second _walk_commit_delta must return all 3 commits via fast path; "
946 f"got {[cid[:16] for cid in result2.keys()]}"
947 )
File History 5 commits
sha256:34035d72cef530c1ab9d6a6f53be18d803dde705fc3157617d70352a96d0747b Merge branch 'fix/wire-push-external-parent-manifest' into dev Human 8 days ago
sha256:fc04e4cae9e1774d6a21b65c45daeed0e6787eb581d13aa1b03bfe9384a34226 Merge branch 'fix/two-column-scroll-layout' into dev Human 8 days ago
sha256:408916fc5973ba59c6e4eebaa80ebdcc801c0a63205651e25009d11548f79454 chore: bump version to 0.2.0.dev2 — nightly.2, matching muse Sonnet 4.6 patch 11 days ago
sha256:d035733f21ccff27735fddebfbbe0ed24565a32a22db8de5885402262671ecd2 chore: bump version to 0.2.0rc15 for musehub#113 fix release Sonnet 4.6 patch 14 days ago
sha256:0032d6cfa33bc3c8367436ad768e7dd0e339b4332153160247da8266cb5fa352 Merge branch 'task/version-tags-phase3-server' into dev Human 17 days ago