# 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:

| Job type | Why coarse dedup fails | Correct fix |
|----------|------------------------|-------------|
| `fetch.mpack.prebuild` | Captured tip set goes stale | **Re-read current branch tips at run time** → the single pending job always builds the latest set. Coarse dedup becomes *correct*. |
| `mpack.index` | Each mpack is distinct, non-substitutable work | **Dedup per `mpack_key`** → distinct mpacks always enqueue; only an identical re-pushed mpack is deduped. |

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`):

```python
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`.

```python
# 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:

```python
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_key`s (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_key`s → 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

| Tier | Coverage |
|------|----------|
| Unit | per-key index dedup; coarse-dedup preservation; handler live-tip read; `dedup_payload_key` |
| Integration | DB-backed two-call enqueue; single pending prebuild builds live tips |
| End-to-end | push A → push B (no drain) → worker drain → both indexed + cache at B's tip |
| Stress | N rapid pushes; every mpack indexed; final clone serves latest tip |
| Data integrity | no dropped index work; no stale/missing clone cache row under rapid pushes |

## 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`