"""MWP-3 Phase 0 (RED) — Job-enqueue idempotency drops fresh payloads on back-to-back pushes. Sub-ticket: musehub#108 — https://staging.musehub.ai/gabriel/musehub/issues/108 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58 Predecessors: musehub#106 (MWP-1, generation authority) — closed. musehub#107 (MWP-2, walk fallback) — closed. Root cause (RC-3): ``enqueue_push_intel`` deduplicates pending jobs on the coarse key ``(repo_id, job_type, status='pending')``. This assumption — that any pending job of a given type covers all future work of that type — is false for two of the three live job types: Face 1 — ``mpack.index``: each pushed mpack has a distinct ``mpack_key`` (built as ``mpack_payload`` at musehub_jobs.py:128). If push A's index job is still pending when push B lands, B's job is silently dropped by the coarse dedup. Push B's mpack is never indexed. Face 2 — ``fetch.mpack.prebuild``: the payload captures branch tips *at enqueue time* (musehub_jobs.py:131-136). The handler ``process_fetch_mpack_prebuild_job`` trusts that stale list verbatim (musehub_wire_fetch.py:1067). If push B advances a branch while A's prebuild is still pending, the single deduped job carries push A's stale tip set, and push B's new tip is never prebuilt. Worker scope reminder: only ``mpack.index`` and ``fetch.mpack.prebuild`` are live today (musehub/worker.py:107-108: ``if job_type not in (...): pass``). RC-3's entire blast radius is these two types — both gate clone correctness. Expected Phase 0 status: MWP3_01 RED — two enqueue_push_intel calls with distinct mpack_keys → coarse dedup drops the second; only one mpack.index exists. MWP3_02 RED — prebuild handler runs with stale payload; live branch tip tipB is absent from the wire_fetch_mpack want= list. MWP3_03 GREEN — single enqueue_push_intel call → exactly one mpack.index and one fetch.mpack.prebuild pending (no duplicate explosion). No production code is changed in Phase 0. """ from __future__ import annotations import hashlib from datetime import datetime, timezone from unittest.mock import AsyncMock, patch import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.ids import hash_snapshot from muse.core.mpack import build_wire_mpack from muse.core.types import blob_id from musehub.core.genesis import compute_job_id from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.db.musehub_repo_models import MusehubFetchMPackCache, MusehubMPackIndex from musehub.services.musehub_jobs import complete_job, enqueue_push_intel from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job from musehub.services.musehub_wire_push import process_mpack_index_job, wire_push_unpack_mpack from tests.factories import create_branch, create_repo # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fake_commit_id(seed: str) -> str: return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() def _fake_mpack_key(seed: str) -> str: return "sha256:" + hashlib.sha256(f"mpack-{seed}".encode()).hexdigest() def _now() -> datetime: return datetime.now(tz=timezone.utc) async def _insert_prebuild_job( session: AsyncSession, repo_id: str, tip_commit_ids: list[str], ) -> str: """Insert a fetch.mpack.prebuild job directly with a given (possibly stale) tip set. Used in MWP3_02 to simulate a job whose payload was captured at push A time, before push B advanced a branch. """ now = _now() job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat()) session.add(MusehubBackgroundJob( job_id=job_id, repo_id=repo_id, job_type="fetch.mpack.prebuild", payload={"tip_commit_ids": tip_commit_ids}, status="pending", created_at=now, attempt=0, )) await session.commit() return job_id # --------------------------------------------------------------------------- # MWP3_01 — Face 1: coarse dedup silently drops push B's mpack.index (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_01_distinct_mpack_keys_both_enqueued(db_session: AsyncSession) -> None: """MWP3_01 RED: two back-to-back pushes with distinct mpack_keys must each produce a pending mpack.index job. Simulation: push A lands and enqueues mpack.index with mpack_key=mpack_a. The job is committed (pending). Push B lands before the worker drains — enqueue_push_intel is called again with mpack_key=mpack_b. Current (buggy) behavior: the coarse SELECT finds mpack.index already pending for this repo_id and skips the insert. Only mpack_a is indexed; mpack_b is never indexed, leaving push B's blobs unreachable. After fix (Phase 2): enqueue_push_intel deduplicates mpack.index on (repo_id, mpack.index, payload->>'mpack_key', pending), so distinct mpack_keys always produce two rows. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-01-tip-a") tip_b = _fake_commit_id("mwp3-01-tip-b") mpack_a = _fake_mpack_key("mwp3-01-a") mpack_b = _fake_mpack_key("mwp3-01-b") # Push A: create branch main→tipA, enqueue push intel for push A. await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) await db_session.commit() # Push B arrives before the worker drains push A's jobs. # Its mpack_key is distinct — this is separate, non-substitutable index work. await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) await db_session.commit() # Both mpack_keys must have a pending mpack.index job. pending_index_rows = (await db_session.execute( select(MusehubBackgroundJob.job_id, MusehubBackgroundJob.payload).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).all() mpack_keys_in_db = [row[1].get("mpack_key") for row in pending_index_rows] assert len(pending_index_rows) == 2, ( f"Expected 2 pending mpack.index jobs (one per distinct mpack_key), " f"got {len(pending_index_rows)}. " f"mpack_keys found in DB: {mpack_keys_in_db}. " f"Root cause: coarse (repo_id, job_type, pending) dedup in " f"enqueue_push_intel (musehub_jobs.py:158-164) silently dropped push " f"B's mpack.index. Fix: dedup mpack.index on payload->>'mpack_key' " f"instead of job_type alone." ) assert mpack_a in mpack_keys_in_db, ( f"Push A mpack_key={mpack_a!r} missing from DB. Got: {mpack_keys_in_db}" ) assert mpack_b in mpack_keys_in_db, ( f"Push B mpack_key={mpack_b!r} was dropped by coarse dedup. " f"Got: {mpack_keys_in_db}" ) # --------------------------------------------------------------------------- # MWP3_02 — Face 2: prebuild handler trusts stale payload tip set (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_02_prebuild_handler_uses_stale_payload_tips(db_session: AsyncSession) -> None: """MWP3_02 RED: process_fetch_mpack_prebuild_job trusts payload["tip_commit_ids"] verbatim, missing branch tips that advanced after the job was enqueued. Scenario: - Branch 'main' points to tipA at enqueue time → payload captures {tipA}. - Push B adds branch 'dev' pointing to tipB; push B is deduped (no new prebuild job). - The single pending prebuild job has payload={"tip_commit_ids": [tipA]}. - MusehubBranch now has both tipA (main) and tipB (dev). - Handler runs → reads only from payload → calls wire_fetch_mpack with want=[tipA]; tipB is silently omitted from the prebuild. After fix (Phase 1): the handler queries MusehubBranch.head_commit_id at run time, getting {tipA, tipB}, and passes both to wire_fetch_mpack. The pending job becomes self-coalescing: it always reflects live state. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-02-tip-a") tip_b = _fake_commit_id("mwp3-02-tip-b") # Live DB state at handler run time: two branches, both tips present. # This reflects the situation after push B has committed its branch advance. await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) # The pending prebuild job carries a STALE payload — only tipA was known at # push A's enqueue time. Push B was deduped; tipB is absent from the payload. job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-02-combined-mpack").hexdigest() with patch( "musehub.services.musehub_wire_fetch.wire_fetch_mpack", new_callable=AsyncMock, return_value={ "mpack_id": fake_mpack_id, "mpack_url": "https://r2.example/combined", "commit_count": 2, "blob_count": 4, }, ) as mock_build: await process_fetch_mpack_prebuild_job(db_session, job_id) await db_session.commit() assert mock_build.call_count == 1, ( f"Expected wire_fetch_mpack to be called exactly once; " f"got call_count={mock_build.call_count}" ) # wire_fetch_mpack is called as: # wire_fetch_mpack(session, repo_id, want=tips_to_build, have=[], force_build=True) actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) assert tip_b in actual_want, ( f"wire_fetch_mpack was called with want={actual_want!r}, " f"but tipB={tip_b!r} is absent. " f"Root cause: process_fetch_mpack_prebuild_job reads tip_commit_ids " f"from payload at musehub_wire_fetch.py:1067, which was stale at " f"enqueue time. Fix: query MusehubBranch.head_commit_id from the DB " f"at run time so the handler is self-coalescing." ) assert tip_a in actual_want, ( f"tipA={tip_a!r} must also be in the want list; got {actual_want!r}" ) # --------------------------------------------------------------------------- # MWP3_03 — Guard: single push enqueues exactly one of each (GREEN) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_03_single_push_enqueues_exactly_one_each(db_session: AsyncSession) -> None: """MWP3_03 GREEN: a single enqueue_push_intel call enqueues exactly one mpack.index job and exactly one fetch.mpack.prebuild job. This guard must remain GREEN through all MWP-3 phases. The per-key dedup fix for mpack.index (Phase 2) must not cause duplicate jobs for the normal single-push case — only distinct mpack_keys should produce multiple rows. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-03-tip-a") mpack_a = _fake_mpack_key("mwp3-03-a") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) await db_session.commit() count_index = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() count_prebuild = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.status == "pending", ) )).scalar() assert count_index == 1, ( f"Expected exactly 1 pending mpack.index job for a single push; " f"got {count_index}. The per-key dedup must not produce duplicates " f"when only one mpack_key is enqueued." ) assert count_prebuild == 1, ( f"Expected exactly 1 pending fetch.mpack.prebuild job for a single push; " f"got {count_prebuild}." ) # --------------------------------------------------------------------------- # MWP3_04 — Phase 1: handler reads live tips even when payload is empty (GREEN after fix) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_04_handler_reads_live_tips_when_payload_empty(db_session: AsyncSession) -> None: """MWP3_04: process_fetch_mpack_prebuild_job must query MusehubBranch for live tips at run time, even when payload["tip_commit_ids"] is empty. Scenario: a prebuild job was somehow enqueued with an empty tip list (e.g. a race where the branch was advanced after enqueue_push_intel ran its branch query). Two live branches exist in the DB pointing to tipA and tipB. Current (buggy) behavior: the handler returns early at line 1069 because payload["tip_commit_ids"] is []. wire_fetch_mpack is never called. After fix: the handler queries MusehubBranch, gets {tipA, tipB}, and calls wire_fetch_mpack with want=[tipA, tipB]. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-04-tip-a") tip_b = _fake_commit_id("mwp3-04-tip-b") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) # Deliberately empty payload — branch tips not captured. job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[]) fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-04-mpack").hexdigest() with patch( "musehub.services.musehub_wire_fetch.wire_fetch_mpack", new_callable=AsyncMock, return_value={ "mpack_id": fake_mpack_id, "mpack_url": "https://r2.example/combined", "commit_count": 2, "blob_count": 4, }, ) as mock_build: result = await process_fetch_mpack_prebuild_job(db_session, job_id) await db_session.commit() assert mock_build.call_count == 1, ( f"wire_fetch_mpack must be called once — handler must query live branches " f"instead of returning early on empty payload. call_count={mock_build.call_count}. " f"result={result}" ) actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) assert tip_a in actual_want, f"tipA must be in want; got {actual_want!r}" assert tip_b in actual_want, f"tipB must be in want; got {actual_want!r}" assert result["tips_requested"] == 2, ( f"tips_requested must reflect the live tip count (2); got {result['tips_requested']}" ) # --------------------------------------------------------------------------- # MWP3_05 — Phase 1: stale payload tips lose to live DB tips (GREEN after fix) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_05_live_tips_supersede_stale_payload(db_session: AsyncSession) -> None: """MWP3_05: when payload["tip_commit_ids"] carries only {tipA} but MusehubBranch has {tipA, tipB} live, the handler must build for {tipA, tipB}. This directly exercises the self-coalescing property: a single pending prebuild job whose payload was captured before push B advanced a branch must still cover push B's tip when it finally runs. This is the same scenario as MWP3_02 (which became RED in Phase 0) — Phase 1 makes both MWP3_05 and MWP3_02 GREEN by the same code change. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-05-tip-a") tip_b = _fake_commit_id("mwp3-05-tip-b") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) # Stale payload: only tipA. tipB is live in MusehubBranch but absent from payload. job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-05-mpack").hexdigest() with patch( "musehub.services.musehub_wire_fetch.wire_fetch_mpack", new_callable=AsyncMock, return_value={ "mpack_id": fake_mpack_id, "mpack_url": "https://r2.example/combined", "commit_count": 2, "blob_count": 4, }, ) as mock_build: await process_fetch_mpack_prebuild_job(db_session, job_id) await db_session.commit() actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) assert tip_b in actual_want, ( f"tipB must be in want — live DB tip must supersede stale payload. " f"want={actual_want!r}" ) assert tip_a in actual_want, ( f"tipA must also be in want. want={actual_want!r}" ) # --------------------------------------------------------------------------- # MWP3_06 — Phase 1 integration: both tips get cache rows (GREEN after fix) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_06_cache_rows_written_for_all_live_tips(db_session: AsyncSession) -> None: """MWP3_06 Integration: after the Phase 1 fix, running the single pending prebuild job writes musehub_fetch_mpack_cache rows for ALL live branch tips, even those absent from the stale payload. Simulates the real scenario: push A enqueues prebuild with payload={tipA}; push B adds branch dev→tipB (deduped — no new prebuild job); the worker drains by running the single pending job. Both tipA and tipB must end up with a cache row pointing at the same combined mpack_id. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-06-tip-a") tip_b = _fake_commit_id("mwp3-06-tip-b") combined_mpack_id = _fake_mpack_key("mwp3-06-combined") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) # Stale payload — only tipA captured at push A's enqueue time. job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) with patch( "musehub.services.musehub_wire_fetch.wire_fetch_mpack", new_callable=AsyncMock, return_value={ "mpack_id": combined_mpack_id, "mpack_url": "https://r2.example/combined", "commit_count": 4, "blob_count": 8, }, ): result = await process_fetch_mpack_prebuild_job(db_session, job_id) await db_session.commit() # Both tips must have a cache row pointing at the same combined mpack_id. cache_rows = (await db_session.execute( select( MusehubFetchMPackCache.tip_commit_id, MusehubFetchMPackCache.mpack_id, ).where(MusehubFetchMPackCache.repo_id == repo_id) )).all() cached = {row[0]: row[1] for row in cache_rows} assert tip_a in cached, ( f"tipA={tip_a[:20]!r} must have a cache row; got keys: {[k[:20] for k in cached]}" ) assert tip_b in cached, ( f"tipB={tip_b[:20]!r} must have a cache row — it was absent from the " f"stale payload but is a live branch tip. Got keys: {[k[:20] for k in cached]}" ) assert cached[tip_a] == combined_mpack_id, ( f"tipA cache must point at combined_mpack_id. Got: {cached.get(tip_a)}" ) assert cached[tip_b] == combined_mpack_id, ( f"tipB cache must point at combined_mpack_id. Got: {cached.get(tip_b)}" ) assert result["tips_requested"] == 2, ( f"tips_requested must be 2 (live tip count); got {result['tips_requested']}" ) # =========================================================================== # Phase 2 — mpack.index per-mpack_key dedup # =========================================================================== # --------------------------------------------------------------------------- # MWP3_07 — distinct mpack_keys must each produce a pending mpack.index (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_07_distinct_mpack_keys_both_produce_index_jobs( db_session: AsyncSession, ) -> None: """MWP3_07 RED → GREEN after Phase 2 fix: two enqueue_push_intel calls with distinct mpack_keys must both produce a pending mpack.index job. This is the same core assertion as MWP3_01 but written as a Phase 2 deliverable — the canonical test that makes MWP3_01 GREEN and validates the per-key dedup path. Current (buggy) behavior: the coarse SELECT finds mpack.index already pending after push A; push B's job is silently dropped. Only one row. After fix: enqueue_push_intel deduplicates mpack.index on payload->>'mpack_key', so distinct keys always produce distinct rows. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-07-tip-a") tip_b = _fake_commit_id("mwp3-07-tip-b") mpack_a = _fake_mpack_key("mwp3-07-a") mpack_b = _fake_mpack_key("mwp3-07-b") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) await db_session.commit() # Push B while push A's jobs are still pending. await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) await db_session.commit() rows = (await db_session.execute( select(MusehubBackgroundJob.payload).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalars().all() keys_in_db = [r.get("mpack_key") for r in rows] assert len(rows) == 2, ( f"Expected 2 pending mpack.index rows (one per mpack_key), got {len(rows)}. " f"mpack_keys in DB: {keys_in_db}" ) assert mpack_a in keys_in_db, f"Push A key {mpack_a!r} missing: {keys_in_db}" assert mpack_b in keys_in_db, f"Push B key {mpack_b!r} missing: {keys_in_db}" # --------------------------------------------------------------------------- # MWP3_08 — same mpack_key is still deduped to one row (GREEN guard) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_08_same_mpack_key_deduped_to_one_row( db_session: AsyncSession, ) -> None: """MWP3_08 GREEN (guard): two enqueue_push_intel calls with the SAME mpack_key must still produce exactly one pending mpack.index job. Per-key dedup must not cause a duplicate-job explosion for idempotent re-pushes (same mpack content re-uploaded). This guard must remain GREEN through Phase 2 and all subsequent phases. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-08-tip-a") tip_b = _fake_commit_id("mwp3-08-tip-b") mpack_same = _fake_mpack_key("mwp3-08-same") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_same) await db_session.commit() await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_same) await db_session.commit() count = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert count == 1, ( f"Expected exactly 1 pending mpack.index for an identical re-push " f"(same mpack_key); got {count}. Per-key dedup must collapse true duplicates." ) # --------------------------------------------------------------------------- # MWP3_08b — enqueue_job dedup_payload_key param (distinct → 2 rows, RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_08b_distinct_payload_key_produces_two_rows( db_session: AsyncSession, ) -> None: """MWP3_08b RED → GREEN: enqueue_job with dedup_payload_key='mpack_key' and distinct payload values must produce two pending rows. Without the parameter, enqueue_job uses coarse (repo_id, job_type, pending) dedup and silently drops the second call. With dedup_payload_key, it keys on payload->>'mpack_key', allowing distinct mpacks to each have a job. """ from musehub.services.musehub_jobs import enqueue_job repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) mpack_a = _fake_mpack_key("mwp3-08b-a") mpack_b = _fake_mpack_key("mwp3-08b-b") await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": mpack_a, "head": "sha256:aaa"}, dedup_payload_key="mpack_key", ) await db_session.commit() await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": mpack_b, "head": "sha256:bbb"}, dedup_payload_key="mpack_key", ) await db_session.commit() rows = (await db_session.execute( select(MusehubBackgroundJob.payload).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalars().all() keys_in_db = [r.get("mpack_key") for r in rows] assert len(rows) == 2, ( f"Expected 2 rows with dedup_payload_key and distinct mpack_keys; " f"got {len(rows)}. keys in DB: {keys_in_db}" ) assert mpack_a in keys_in_db and mpack_b in keys_in_db, ( f"Both keys must be present. Got: {keys_in_db}" ) @pytest.mark.asyncio async def test_mwp3_08b_same_payload_key_deduped_to_one_row( db_session: AsyncSession, ) -> None: """MWP3_08b guard GREEN: enqueue_job with dedup_payload_key and identical payload values deduplicates to one row — per-key dedup collapses true duplicates. """ from musehub.services.musehub_jobs import enqueue_job repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) mpack_same = _fake_mpack_key("mwp3-08b-same") await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": mpack_same, "head": "sha256:aaa"}, dedup_payload_key="mpack_key", ) await db_session.commit() await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": mpack_same, "head": "sha256:bbb"}, dedup_payload_key="mpack_key", ) await db_session.commit() count = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert count == 1, ( f"Same mpack_key must dedup to 1 row with dedup_payload_key; got {count}" ) @pytest.mark.asyncio async def test_mwp3_08b_default_coarse_behavior_unchanged( db_session: AsyncSession, ) -> None: """MWP3_08b guard GREEN: enqueue_job without dedup_payload_key retains existing coarse (repo_id, job_type, pending) dedup — second call returns None, one row. """ from musehub.services.musehub_jobs import enqueue_job repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) id1 = await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": _fake_mpack_key("mwp3-08b-coarse-a"), "head": "sha256:aaa"}, ) await db_session.commit() id2 = await enqueue_job( db_session, repo_id, "mpack.index", {"mpack_key": _fake_mpack_key("mwp3-08b-coarse-b"), "head": "sha256:bbb"}, ) await db_session.commit() assert id1 is not None, "First call must return a job_id" assert id2 is None, "Second call without dedup_payload_key must be deduped (return None)" count = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert count == 1, f"Coarse dedup must yield 1 row; got {count}" # --------------------------------------------------------------------------- # MWP3_09 — coarse types stay deduped; only mpack.index multiplied (GREEN guard) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_09_coarse_types_remain_deduplicated( db_session: AsyncSession, ) -> None: """MWP3_09 GREEN (guard): after two enqueue_push_intel calls with distinct mpack_keys, all coarse job types (fetch.mpack.prebuild, intel.*, gc, …) must each have exactly one pending row — the fix must not loosen coarse dedup. Only mpack.index is allowed to multiply (per distinct mpack_key). """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) tip_a = _fake_commit_id("mwp3-09-tip-a") tip_b = _fake_commit_id("mwp3-09-tip-b") mpack_a = _fake_mpack_key("mwp3-09-a") mpack_b = _fake_mpack_key("mwp3-09-b") await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) await db_session.commit() await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) await db_session.commit() # Query all pending rows and group by job_type. all_rows = (await db_session.execute( select(MusehubBackgroundJob.job_type).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.status == "pending", ) )).scalars().all() from collections import Counter counts = Counter(all_rows) # mpack.index should have 2 (one per distinct mpack_key). # Every other type should have exactly 1. duplicated_coarse = { jt: cnt for jt, cnt in counts.items() if jt != "mpack.index" and cnt > 1 } assert not duplicated_coarse, ( f"Coarse job types must remain deduplicated (≤1 each) after two " f"enqueue_push_intel calls. Found duplicates: {duplicated_coarse}" ) # --------------------------------------------------------------------------- # Phase 3 — E2E helpers # --------------------------------------------------------------------------- _E2E_OWNER = "gabriel" def _e2e_raw_commit( cid: str, parent: str | None, snap_id: str, branch: str = "main", ) -> dict: return { "commit_id": cid, "branch": branch, "message": f"commit {cid[:12]}", "author": _E2E_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 _e2e_push( session: AsyncSession, repo_id: str, cid: str, parent: str | None, snap_id: str, fpath: str, oid: str, data: bytes, branch: str = "main", ) -> str: """Build a single-commit mpack, store it in the conftest MemoryBackend, push it. The conftest autouse fixture patches musehub.storage.backends.get_backend to a MemoryBackend instance shared across all modules. Storing in _mpacks[key] here is equivalent to pre-seeding MinIO — wire_push_unpack_mpack and process_mpack_index_job both call get_backend() and see the same bytes. Returns the mpack_key so callers can assert on MPackIndex after draining. """ import musehub.storage.backends as _b_mod backend = _b_mod.get_backend() mpack_bytes = build_wire_mpack({ "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)], "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 await wire_push_unpack_mpack( session, repo_id, mpack_key, pusher_id=_E2E_OWNER, branch=branch, head_commit_id=cid, commits_count=1, blobs_count=1, force=True, ) await session.commit() return mpack_key async def _drain_index_jobs(session: AsyncSession, repo_id: str, limit: int | None = None) -> list[str]: """Drain pending mpack.index jobs in created_at order, return job_ids drained.""" q = ( select(MusehubBackgroundJob) .where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) .order_by(MusehubBackgroundJob.created_at) ) if limit is not None: q = q.limit(limit) jobs = (await session.execute(q)).scalars().all() drained = [] for job in jobs: await process_mpack_index_job(session, job.job_id) await complete_job(session, job.job_id) await session.commit() drained.append(job.job_id) return drained async def _drain_prebuild_job(session: AsyncSession, repo_id: str) -> None: """Drain the single pending fetch.mpack.prebuild job.""" row = (await session.execute( select(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.status == "pending", ).limit(1) )).scalar_one_or_none() assert row is not None, "Expected a pending fetch.mpack.prebuild job" await process_fetch_mpack_prebuild_job(session, row.job_id) await session.commit() # --------------------------------------------------------------------------- # MWP3_10 — E2E acceptance gate: two pushes + full drain → both indexed + cache # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_10_e2e_two_pushes_full_drain(db_session: AsyncSession) -> None: """MWP3_10 GREEN (Phase 3 — E2E acceptance gate): two back-to-back pushes, no drain between them, then full drain → both mpacks indexed and clone cache updated to push B's tip. This is the canonical RC-3 scenario — two pushes before the worker wakes up. Before the MWP-3 fix (Phase 0 state): - Push A enqueues mpack.index(key_a). - Push B: coarse dedup finds mpack.index pending → drops key_b. Only key_a ever gets indexed. clone cache never advances to c2. After the MWP-3 fix (Phases 1+2): - Push A enqueues mpack.index(key_a) + fetch.mpack.prebuild. - Push B enqueues mpack.index(key_b) (per-key dedup allows it). The prebuild is deduped to one row but re-reads live branch tips (Phase 1). - Drain order: index jobs first (populates MusehubMPackIndex so wire_fetch_mpack does not raise FetchNotIndexedError), then prebuild. - After drain: MPackIndex has rows for key_a AND key_b; clone cache has c2. """ repo = await create_repo(db_session, owner=_E2E_OWNER) repo_id = str(repo.repo_id) data_a = b"mwp3-10-blob-a" data_b = b"mwp3-10-blob-b" oid_a = blob_id(data_a) oid_b = blob_id(data_b) snap_a = hash_snapshot({"file_a.txt": oid_a}) snap_b = hash_snapshot({"file_b.txt": oid_b}) c1 = _fake_commit_id("mwp3-10-c1") c2 = _fake_commit_id("mwp3-10-c2") # Push A — creates branch main → c1; enqueues mpack.index(key_a) + prebuild. mpack_key_a = await _e2e_push(db_session, repo_id, c1, None, snap_a, "file_a.txt", oid_a, data_a) # Push B — advances main → c2; enqueues mpack.index(key_b); prebuild deduped. mpack_key_b = await _e2e_push(db_session, repo_id, c2, c1, snap_b, "file_b.txt", oid_b, data_b) # Both mpack.index keys must be pending — Phase 2 fix: per-key dedup. pending_index = (await db_session.execute( select(MusehubBackgroundJob.payload).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalars().all() pending_keys = {pl.get("mpack_key") for pl in pending_index} assert mpack_key_a in pending_keys and mpack_key_b in pending_keys, ( f"Expected both mpack_keys pending before drain. " f"Found: {pending_keys}" ) assert len(pending_index) == 2, ( f"Expected 2 pending mpack.index jobs, got {len(pending_index)}" ) # Drain: index jobs first (wire_fetch_mpack raises FetchNotIndexedError # if index is absent), then prebuild. await _drain_index_jobs(db_session, repo_id) await _drain_prebuild_job(db_session, repo_id) # (a) Both mpack keys must appear in MusehubMPackIndex. count_a = (await db_session.execute( select(func.count()).select_from(MusehubMPackIndex).where( MusehubMPackIndex.mpack_id == mpack_key_a, MusehubMPackIndex.entity_type == "object", ) )).scalar() count_b = (await db_session.execute( select(func.count()).select_from(MusehubMPackIndex).where( MusehubMPackIndex.mpack_id == mpack_key_b, MusehubMPackIndex.entity_type == "object", ) )).scalar() assert count_a > 0, ( f"MusehubMPackIndex has no rows for mpack_key_a={mpack_key_a[:20]}. " f"process_mpack_index_job for push A never ran or wrote no rows." ) assert count_b > 0, ( f"MusehubMPackIndex has no rows for mpack_key_b={mpack_key_b[:20]}. " f"process_mpack_index_job for push B was never enqueued (RC-3 Face 1 not fixed)." ) # (b) Clone cache must point to c2 (push B's tip). # process_fetch_mpack_prebuild_job reads live branch tip (c2 — Phase 1 fix). cache_row = (await db_session.execute( select(MusehubFetchMPackCache).where( MusehubFetchMPackCache.repo_id == repo_id, MusehubFetchMPackCache.tip_commit_id == c2, ) )).scalar_one_or_none() assert cache_row is not None, ( f"MusehubFetchMPackCache has no row for tip_commit_id=c2={c2[:20]}. " f"The prebuild job used a stale tip (RC-3 Face 2 not fixed) or " f"wire_fetch_mpack raised FetchNotIndexedError because push B's mpack " f"was never indexed (RC-3 Face 1 not fixed)." ) # --------------------------------------------------------------------------- # MWP3_11 — Stress: N=10 pushes, partial drain mid-sequence, full final drain # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp3_11_stress_partial_drain_n10(db_session: AsyncSession) -> None: """MWP3_11 GREEN (Phase 3 — stress): 10 rapid pushes with a partial drain mid-sequence. All 10 mpacks must be indexed and clone cache must point to the final tip after the full drain. Sequence: 1. Push commits 1-5 without draining → 5 pending mpack.index, 1 prebuild. 2. Partial drain: drain only the first 3 mpack.index jobs (chronological order). Commits 4 and 5 remain unindexed (2 pending index jobs). 3. Push commits 6-10 without draining → 5 new mpack.index jobs enqueued (dedup allows them because each has a distinct mpack_key). Prebuild stays at 1 pending row (still deduped from push 1). 4. Full drain: drain all 7 remaining mpack.index jobs (2 + 5), then drain the single pending prebuild job. Assertions: (a) MusehubMPackIndex has rows for all 10 mpack_keys. (b) MusehubFetchMPackCache has row for commit_10's tip (live branch tip at prebuild run time = Phase 1 fix; all 10 index jobs drained before prebuild = Phase 2 fix). """ repo = await create_repo(db_session, owner=_E2E_OWNER) repo_id = str(repo.repo_id) n = 10 cids: list[str] = [_fake_commit_id(f"mwp3-11-c{i}") for i in range(1, n + 1)] datas: list[bytes] = [f"mwp3-11-blob-{i}".encode() for i in range(1, n + 1)] oids: list[str] = [blob_id(d) for d in datas] snaps: list[str] = [hash_snapshot({f"f{i}.txt": oids[i - 1]}) for i in range(1, n + 1)] mpack_keys: list[str] = [] # Step 1 — push commits 1-5 without draining. for i in range(5): parent = cids[i - 1] if i > 0 else None key = await _e2e_push( db_session, repo_id, cids[i], parent, snaps[i], f"f{i + 1}.txt", oids[i], datas[i], ) mpack_keys.append(key) pending_after_5 = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert pending_after_5 == 5, ( f"Expected 5 pending mpack.index jobs after 5 pushes, got {pending_after_5}" ) # Step 2 — partial drain: drain only the first 3 index jobs. await _drain_index_jobs(db_session, repo_id, limit=3) pending_after_partial = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert pending_after_partial == 2, ( f"Expected 2 pending mpack.index jobs after draining 3 of 5, " f"got {pending_after_partial}" ) # Step 3 — push commits 6-10 without draining. for i in range(5, n): key = await _e2e_push( db_session, repo_id, cids[i], cids[i - 1], snaps[i], f"f{i + 1}.txt", oids[i], datas[i], ) mpack_keys.append(key) # 2 remaining from commits 4+5 + 5 new from commits 6-10 = 7 pending. pending_after_10 = (await db_session.execute( select(func.count()).select_from(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "mpack.index", MusehubBackgroundJob.status == "pending", ) )).scalar() assert pending_after_10 == 7, ( f"Expected 7 pending mpack.index jobs after pushes 6-10 " f"(2 leftover + 5 new = 7), got {pending_after_10}" ) # Step 4 — full drain: all remaining index jobs, then prebuild. await _drain_index_jobs(db_session, repo_id) await _drain_prebuild_job(db_session, repo_id) # (a) All 10 mpack_keys must appear in MusehubMPackIndex. for i, mk in enumerate(mpack_keys, start=1): cnt = (await db_session.execute( select(func.count()).select_from(MusehubMPackIndex).where( MusehubMPackIndex.mpack_id == mk, MusehubMPackIndex.entity_type == "object", ) )).scalar() assert cnt > 0, ( f"MusehubMPackIndex has no rows for commit {i} (mpack_key={mk[:20]}). " f"process_mpack_index_job for this push was either never enqueued " f"(RC-3 Face 1) or silently skipped during partial drain." ) # (b) Clone cache must point to commit 10 (live branch tip at prebuild run time). tip_10 = cids[n - 1] cache_row = (await db_session.execute( select(MusehubFetchMPackCache).where( MusehubFetchMPackCache.repo_id == repo_id, MusehubFetchMPackCache.tip_commit_id == tip_10, ) )).scalar_one_or_none() assert cache_row is not None, ( f"MusehubFetchMPackCache has no row for tip_commit_id=commit_10={tip_10[:20]}. " f"The prebuild ran with a stale tip (RC-3 Face 2 not fixed) or " f"wire_fetch_mpack raised FetchNotIndexedError because some push mpack " f"was not indexed (RC-3 Face 1 not fixed)." )