MWP-3 — Job-enqueue idempotency drops fresh payloads on back-to-back pushes (fixes RC-3)

Sub-ticket of the MWP MVP master tracker: 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. (Formal cross-issue links are not available yet; this issue and #58 cross-reference by URL.)

Background

muse clone has no dedicated endpoint — it is a full POST /fetch/mpack with have=[], served from the prebuilt-mpack cache. After every push the server enqueues two clone-correctness-critical background jobs:

• mpack.index — reads the just-pushed mpack from object storage and writes byte-range offsets into musehub_mpack_index so the fetch path can serve blobs via byte-range GETs. One job per pushed mpack (each mpack is distinct work).
• fetch.mpack.prebuild — rebuilds one combined mpack covering all current branch tips and upserts a musehub_fetch_mpack_cache row per tip, so the next clone is a sub-second cache HIT.

A separate worker process polls the queue and runs these jobs. Reality check — the worker today runs only these two job types; every other job type is a no-op (musehub/worker.py:107-108: if job_type not in ("mpack.index", "fetch.mpack.prebuild"): pass). So RC-3's live blast radius is exactly these two types — both of which gate clone correctness.

The bug (RC-3)

Both enqueue paths dedup pending jobs on the coarse key (repo_id, job_type, status="pending"):

• enqueue_job (musehub/services/musehub_jobs.py:72-84) — single-job path: SELECT job_id WHERE repo_id=? AND job_type=? AND status='pending' LIMIT 1; if a row exists it returns None and inserts nothing.
• enqueue_push_intel (musehub/services/musehub_jobs.py:158-189) — batch path: one SELECT job_type … WHERE job_type IN (…) AND status='pending', then add_all([... for job_type, payload in jobs if job_type not in existing]).

This coarse dedup assumes any pending job of a type covers all future work of that type. That assumption is false for both live job types, in two distinct ways:

Face 1 — mpack.index: push B's mpack is never indexed

mpack.index's payload is push-specific: {"mpack_key": <this push's mpack>, "head": …, "branch": …} (built at musehub_jobs.py:128 as mpack_payload). Each push produces a different mpack_key. If push A's mpack.index is still pending when push B lands, push B is deduped — and push B's mpack_key is never indexed at all. The pending job carries push A's mpack_key; when it runs it indexes A's mpack, not B's. B's blobs have no byte-range index until yet another push happens to be processed while no index job is pending. Coarse dedup is simply wrong for mpack.index: distinct mpacks are distinct, non-substitutable work.

Face 2 — fetch.mpack.prebuild: stale captured tip set

fetch.mpack.prebuild's payload captures the branch tips at enqueue time: prebuild_payload = {"tip_commit_ids": [row[0] for row in branch_tips_q]} (musehub_jobs.py:131-136). The handler process_fetch_mpack_prebuild_job (musehub/services/musehub_wire_fetch.py:1034) trusts that captured list verbatim (musehub_wire_fetch.py:1067: tip_commit_ids = [str(t) for t in (payload.get("tip_commit_ids") or [])]).

When push A's prebuild is still pending and push B advances a branch, push B is deduped. Push A's pending job still carries push A's tip set — missing push B's new tip. When the single job finally runs, it builds a combined mpack for the old tip set, and the clone cache for push B's new tip is never written (or is written pointing at an mpack that predates B). Result: clone of the new tip is a cache MISS forever (until another push coincidentally rebuilds), or worse, a stale HIT.

Net effect: rapid / back-to-back pushes ⇒ permanently stale clone cache and unindexed mpacks. This is RC-3.

Why the two faces need different fixes

The two job types fail for structurally different reasons, so they take different remedies — this is the core design insight of MWP-3:

After MWP-3, the invariant is: a pending job is only ever skipped when an existing pending job will do the exact same effective work.

Relationship to RC-4 (out of scope — MWP-4)

RC-4 is a separate bug: fetch.mpack.prebuild and mpack.index share a created_at and claim_next_job (musehub_jobs.py:206-231) orders by created_at with no tiebreaker, so prebuild can run before index. MWP-3 does not fix ordering — it fixes which jobs exist and what payload they carry. The two are independent and compose. Do not change _FOUNDATION_TYPES staggering or claim_next_job ordering in this ticket. See "Out of scope".

Goal

• No fresh work is ever dropped on back-to-back pushes. Every pushed mpack is indexed; every branch-tip advance results in a prebuilt clone cache covering the latest tip.
• fetch.mpack.prebuild is self-coalescing: the handler re-reads current branch tips at run time, so a single pending prebuild always reflects the latest state no matter how many pushes were deduped into it.
• mpack.index dedup is keyed on mpack_key: two distinct mpacks always produce two index jobs; an identical re-pushed mpack is still deduped (idempotent).
• A non-skipped regression test proves a two-push (A then B) sequence drained by the worker leaves both mpacks indexed and the clone cache pointing at B's tip.
• Steady-state single-push latency and behavior are unchanged (no duplicate-job explosion, no extra round-trips beyond what the per-key check costs).

Design

Fix 1 — fetch.mpack.prebuild re-reads branch tips at run time

File: musehub/services/musehub_wire_fetch.py::process_fetch_mpack_prebuild_job

Today the handler reads the tip set from the payload (musehub_wire_fetch.py:1067). Change it to query the current branch tips authoritatively, mirroring the query already used in enqueue_push_intel (musehub_jobs.py:131-135):

from musehub.db.musehub_repo_models import MusehubBranch

# Authoritative: the tips that exist RIGHT NOW, not whatever was captured at
# enqueue time. This makes the job self-coalescing — any number of pushes that
# were deduped into one pending job are covered, because we read live state here.
branch_tips_q = await session.execute(
    select(MusehubBranch.head_commit_id)
    .where(MusehubBranch.repo_id == repo_id)
    .where(MusehubBranch.head_commit_id.isnot(None))
)
tip_commit_ids: list[str] = [str(row[0]) for row in branch_tips_q]

# payload["tip_commit_ids"] is retained ONLY for observability (logged below);
# it is NOT authoritative. A stale captured list must never shrink the build.
_payload_tips = [str(t) for t in (payload.get("tip_commit_ids") or [])]
if set(_payload_tips) != set(tip_commit_ids):
    logger.info(
        "[fetch.mpack.prebuild] job=%s payload tips=%d differ from live tips=%d "
        "(coalesced pushes) — using live tips",
        job_id[:16], len(_payload_tips), len(tip_commit_ids),
    )

The rest of the handler (cache-freshness check, combined build, per-tip upsert) is unchanged — it already builds one combined mpack covering all tip_commit_ids and upserts a cache row per tip so the multi-tip clone cache-hit check (len(mpack_ids) == 1 across all want tips) fires. Reading live tips simply guarantees that set is current.

Consequence: the coarse dedup on (repo_id, "fetch.mpack.prebuild", pending) in enqueue_push_intel is now correct and stays as-is. One pending prebuild covers all branches at the moment it runs.

The "running" race is already safe: if push A's prebuild is mid-flight (status="running", not "pending") when push B lands, push B's coarse check finds no pending prebuild and enqueues a fresh one, which re-reads live tips again. No tip can be lost.

Why we keep writing payload["tip_commit_ids"]: it is cheap, aids debugging, and tests/test_fetch_mpack_prebuild.py::test_fmc_18_… asserts its presence. The handler simply no longer trusts it as authoritative.

Fix 2 — mpack.index dedup keyed on mpack_key

File: musehub/services/musehub_jobs.py::enqueue_push_intel

Pull mpack.index out of the coarse job_type IN (...) dedup and give it a per-mpack_key existence check. The cleanest shape: partition the job list into coarse-deduped types and the per-key type, run the existing batch SELECT for the coarse set, and a targeted SELECT for mpack.index keyed on the JSONB mpack_key.

# mpack.index is per-mpack work: dedup on the exact mpack_key, never on job_type
# alone. Two distinct mpacks must both be indexed; an identical re-pushed mpack
# (same mpack_key) is still deduped.
_PER_KEY_TYPE = "mpack.index"

coarse_jobs = [(jt, p) for jt, p in jobs if jt != _PER_KEY_TYPE]
index_jobs  = [(jt, p) for jt, p in jobs if jt == _PER_KEY_TYPE]

# --- coarse dedup (unchanged) ---
coarse_types = [jt for jt, _ in coarse_jobs]
existing_coarse = set((await session.execute(
    select(MusehubBackgroundJob.job_type).where(
        MusehubBackgroundJob.repo_id == repo_id,
        MusehubBackgroundJob.job_type.in_(coarse_types),
        MusehubBackgroundJob.status == "pending",
    )
)).scalars().all())

# --- per-key dedup for mpack.index ---
index_to_add: list[tuple[str, JSONObject]] = []
for jt, payload in index_jobs:
    mk = str(payload.get("mpack_key", ""))
    if not mk:
        continue  # nothing to index; skip (matches process_mpack_index_job guard)
    already = (await session.execute(
        select(MusehubBackgroundJob.job_id).where(
            MusehubBackgroundJob.repo_id == repo_id,
            MusehubBackgroundJob.job_type == _PER_KEY_TYPE,
            MusehubBackgroundJob.status == "pending",
            MusehubBackgroundJob.payload["mpack_key"].astext == mk,
        ).limit(1)
    )).scalar_one_or_none()
    if already is None:
        index_to_add.append((jt, payload))

The add_all then unions: [coarse jobs not in existing_coarse] + index_to_add, preserving the existing _FOUNDATION_TYPES / subtype created_at staggering verbatim (RC-4 territory — do not touch).

JSONB note: MusehubBackgroundJob.payload is a Postgres JSONB column (musehub/db/musehub_jobs_models.py:…). payload["mpack_key"].astext == mk compiles to payload ->> 'mpack_key' = :mk. The test DB is Postgres, so this is exercised directly. (If a SQLite unit path is ever added, this comparison must be guarded — but the current suite runs on Postgres via the db_session fixture.)

Optional generalization — enqueue_job(dedup_payload_key=…)

enqueue_job (musehub/services/musehub_jobs.py:57) is the generic single-job path (currently used only by enqueue_profile_snapshot). It is not on the mpack.index enqueue path today, so fixing it is not required for RC-3. To make the per-key pattern reusable and prevent a future regression if a caller routes a per-key job through enqueue_job, add an optional parameter:

async def enqueue_job(
    session, repo_id, job_type, payload, *, dedup_payload_key: str | None = None,
) -> str | None:
    """... When dedup_payload_key is given, idempotency is keyed on
    (repo_id, job_type, payload[dedup_payload_key], pending) instead of the
    coarse (repo_id, job_type, pending) key."""

This is a small, well-tested addition (deliverable MWP3_08b). Keep it scoped: do not reroute mpack.index through enqueue_job — the batch path stays the single enqueue site for push jobs.

Phases (load-bearing; each green before the next)

Phase 0 — Reproduce (RED) ✅

• [x] MWP3_01 Unit: two enqueue_push_intel calls for the same repo with distinct mpack_keys (simulating push A then push B, A's index still pending) → assert two mpack.index pending rows exist. RED today (coarse dedup drops the second). Asserts Face 1.
• [x] MWP3_02 Unit: create branch main→tipA, call enqueue_push_intel; then advance main→tipB (or add branch dev→tipB) and call again. The single pending fetch.mpack.prebuild row's payload["tip_commit_ids"] still contains only {tipA} — and (the real defect) running process_fetch_mpack_prebuild_job against that pending row builds a cache that omits tipB. RED today (handler trusts the stale payload). Asserts Face 2.
• [x] MWP3_03 Guard (must stay GREEN through all phases): a single enqueue_push_intel call enqueues exactly one mpack.index and exactly one fetch.mpack.prebuild — no duplicate explosion.

Phase 0 exit: MWP3_01 and MWP3_02 RED for the documented reasons; MWP3_03 GREEN. No production code changed. Committed sha256:78ce8754bfa2b… on task/mwp-3-job-idempotency.

Phase 1 — fetch.mpack.prebuild self-coalescing (runtime re-read) ✅

• [x] MWP3_04 Modify process_fetch_mpack_prebuild_job to read branch tips from MusehubBranch at run time (query mirrors enqueue_push_intel:131-135). Unit: insert a prebuild job with an empty payload["tip_commit_ids"] but two live MusehubBranch rows (tipA, tipB) + mock wire_fetch_mpack; assert the handler requests a build for {tipA, tipB}.
• [x] MWP3_05 Unit: payload carries only {tipA} but DB has {tipA, tipB} → handler builds {tipA, tipB} (live tips win over stale payload). Makes MWP3_02 GREEN.
• [x] MWP3_06 Integration: enqueue prebuild for push A (payload {tipA}), advance branch to add tipB, then run the single pending handler against a real (or in-mem-backed) build → musehub_fetch_mpack_cache has rows for both tipA and tipB pointing at one mpack_id.

Phase 1 exit: prebuild is authoritative on live state; MWP3_02 GREEN; FMC_18 still GREEN (payload still written for observability). Committed sha256:b63ee1a008d0af… on task/mwp-3-job-idempotency.

Phase 2 — mpack.index per-mpack_key dedup ✅

• [x] MWP3_07 Modify enqueue_push_intel to dedup mpack.index on (repo_id, "mpack.index", payload->>'mpack_key', pending). Unit: two calls with distinct mpack_keys → two mpack.index pending. Makes MWP3_01 GREEN.
• [x] MWP3_08 Unit: two calls with the same mpack_key (idempotent re-push) → exactly one mpack.index pending (per-key dedup still collapses true duplicates).
• [x] MWP3_08b (optional generalization) Add dedup_payload_key param to enqueue_job; unit: two enqueue_job calls, same type, distinct payload[key] → two rows; same payload[key] → one row. Default (no param) = unchanged coarse behavior.
• [x] MWP3_09 Regression: across the same two distinct-mpack_key calls, the coarse types (fetch.mpack.prebuild, intel.*, gc, …) remain deduped to one pending each — only mpack.index multiplied. Confirms the partition didn't loosen coarse dedup.

Phase 2 exit: MWP3_01 GREEN; distinct mpacks always indexed; coarse dedup intact; MWP3_03 still GREEN. Committed sha256:58981a2606ddc4f96980b255b0f771d2a0f9f6f4786cebb0edb4e6640d3beea5 on task/mwp-3-job-idempotency.

Phase 3 — End-to-end back-to-back drain (the acceptance gate) ✅

• [x] MWP3_10 E2E: push A (C1) then push B (C2 on the same branch) through wire_push_unpack_mpack with an in-memory backend (mirror tests/test_mwp2_walk_fallback.py::_InMemBackend / _e2e_push). Between the two pushes do not drain the worker (so A's jobs are still pending when B lands). Then drain all pending jobs (claim → dispatch via the same handlers the worker calls). Assert: (a) musehub_mpack_index has rows for both A's and B's mpack_key; (b) musehub_fetch_mpack_cache for the branch tip points at an mpack that contains C2. This is the acceptance gate.
• [x] MWP3_11 Stress: N=10 rapid pushes on one branch with the worker draining opportunistically (interleave a partial drain mid-sequence) → after the final drain, every distinct pushed mpack_key is indexed and the clone cache serves the final tip. No permanently-stale cache row remains.

Phase 3 exit: the originally-reported "rapid pushes ⇒ stale clone" symptom is impossible; both faces are closed end-to-end. All 14 tests GREEN.

Phase 4 — Lock & document ✅

• [x] MWP3_12 Confirm the full MWP-3 file is non-skipped and GREEN; run the adjacent suites that touch these symbols and confirm no regression: tests/test_enqueue_batch.py, tests/test_fetch_mpack_prebuild.py, tests/test_mpack_index_always_enqueued.py, tests/test_process_mpack_index_job.py. All 36 tests across 5 suites GREEN. Three tests in test_fetch_mpack_prebuild.py (FMC_07, FMC_07b, FMC_08) required a one-line fix each: they relied on the handler trusting the job payload's tip list, which the Phase 1 fix superseded. Fix: add MusehubBranch rows matching each test's tip IDs so the handler's live-state query returns the expected set. The semantics of each test are unchanged; only the setup is brought into alignment with the current handler contract.
• [x] Update this issue body marking every phase ✅ and every acceptance criterion [x], then close with a comment and update muse#58.

Acceptance criteria

• [x] Two back-to-back pushes with distinct mpacks both get indexed — no mpack.index job is dropped by dedup (MWP3_01, MWP3_07, MWP3_10).
• [x] fetch.mpack.prebuild reflects the current branch tips whenever it runs, regardless of how many pushes were deduped into one pending job (MWP3_04, MWP3_05, MWP3_06).
• [x] An identical re-pushed mpack (same mpack_key) is still deduped to one mpack.index (MWP3_08).
• [x] Coarse dedup for all non-mpack.index types is unchanged — a single push enqueues exactly one of each (MWP3_03, MWP3_09).
• [x] A worker drain of a no-intermediate-drain A→B push sequence leaves both mpacks indexed and the clone cache pointing at B's tip (MWP3_10). Codified as a non-skipped regression test.
• [x] RC-4 ordering and _FOUNDATION_TYPES staggering are untouched; existing job-enqueue suites stay green (MWP3_12).

Testing tiers

Out of scope

• RC-4 — prebuild/index claim ordering. created_at tiebreaker / _FOUNDATION_TYPES stagger / claim_next_job ORDER BY → MWP-4. Do not modify ordering here.
• Intel/profile staleness — and do NOT try to re-enable the intel jobs. intel.*, push.file_last_commits, profile.snapshot also use coarse dedup, but (a) the worker currently no-ops them (worker.py:107-108) and (b) they are recomputed on the next push and are not clone-blocking. The ~17 intel job types were deliberately disabled in the worker because they failed or ran too long and were actively preventing debugging of the mpack.index / fetch.mpack.prebuild pipeline. Getting them operational is a separate, deferred effort — stay out of that rabbit hole. This ticket touches only the two mpack jobs. Making intel re-read HEAD at run time is a future hardening, not RC-3.
• gc coalescing. gc is naturally idempotent on current state; coarse dedup is already correct for it.
• Re-enabling the broad disabled wire suite → MWP-6.

Anchors

• musehub/services/musehub_jobs.py::enqueue_push_intel (batch enqueue; add per-mpack_key dedup for mpack.index, lines 156-189; keep _FOUNDATION_TYPES stagger at 170-189 intact)
• musehub/services/musehub_jobs.py::enqueue_job (generic path; optional dedup_payload_key param — MWP3_08b)
• musehub/services/musehub_jobs.py::claim_next_job (worker claim — read only, do not change ordering: RC-4)
• musehub/services/musehub_wire_fetch.py::process_fetch_mpack_prebuild_job (re-read MusehubBranch tips at run time; replace trust of payload["tip_commit_ids"] at line 1067)
• musehub/services/musehub_wire_push.py::process_mpack_index_job (consumer of mpack.index; reads payload["mpack_key"] at line 1557 — confirms why per-key dedup is required)
• musehub/services/musehub_intel_providers.py::job_types_for_push (source of the job-type list; mpack.index + fetch.mpack.prebuild at line 2716)
• musehub/db/musehub_jobs_models.py::MusehubBackgroundJob (JSONB payload column enabling payload->>'mpack_key' dedup)
• musehub/db/musehub_repo_models.py::MusehubBranch (head_commit_id — authoritative tip source for the prebuild handler)
• musehub/worker.py::_process_one (dispatch; the two live job types at lines 107-131 — mirror this in the MWP3_10 E2E drain)
• New test file: musehub/tests/test_mwp3_job_idempotency.py (test IDs MWP3_01–MWP3_12)
• Adjacent regression suites: tests/test_enqueue_batch.py, tests/test_fetch_mpack_prebuild.py (FMC_18), tests/test_mpack_index_always_enqueued.py, tests/test_process_mpack_index_job.py