Plan: generation-published report runs + bulk extract contracts (#1202 + #1203, epic &73)

Output tables gain generation_id UUID NOT NULL REFERENCES report_generations (same-DB FK — one service): snap_qc_universe, tanf_acf199_snapshots, medicaid_tmsis_eligibility_extracts, medicaid_cms416_reports, snap_monthly_reports. Uniqueness moves to (generation_id, <natural key>). Writes are plain INSERTs into a fresh generation — the partial-column-upsert and stale-row rerun defects vanish by construction (no cross-generation upserts exist at all).

The migration backfills one legacy published generation per existing (kind, period) and stamps existing rows — readers never need a NULL-generation fallback path.

Readers (QC pages + CSV, ACF/T-MSIS/CMS lists + CSVs, FNS get/list) resolve the published generation for (kind, period) and filter by it. Non-published generations are invisible. Each read response carries the generation’s typed provenance (from report_generations — no progress decoding on the read path). MR6 (R2): the TANF/Medicaid reader shape — the QC idiom translated to URLs without a period segment: an OPTIONAL period query param (month=YYYY-MM for ACF-199/T-MSIS, year=YYYY for CMS-416) with latest-published fallback, a typed {items, provenance} envelope per kind, and 404 when no published generation exists (the pre-existing 100-row list bound applies within the resolved generation; keyset list envelopes stay the filed follow-up).

Promotion happens inside the finalize tx: verify output reconciliation (staged row count == the kind’s expected emission count derived from processed/detail counters; FNS: the monthly row exists; CMS-416: band rows consistent with the map), then staged→published + prior published→superseded, then report_run_finalize_in(tx, done). One commit = the visibility swap.

Rerun semantics (explicit): a new enqueue for a (kind, period) with a published generation creates a NEW generation; promotion supersedes the old one. Immutability guard: promotion REFUSES (error/immutable_target) when the target period’s FNS-388 report row is submitted/accepted (or a generation is flagged locked by ops) — shipped federal filings are never silently replaced.

Cleanup: a janitor deletes superseded/abandoned generations' OUTPUT ROWS (+ their report_run_universe rows) after a retention window (run_generation_row_retention_days, default 30, domain 7..=365); report_generations rows are permanent. Retention clocks (impl, MR4 — the schema carries no superseded_at): superseded is measured from the SUCCESSOR’s published_at (exact by construction — promotion supersedes and publishes in one tx), abandoned from created_at (conservative; never visible). Batch-bounded (50 generations/pass) with a dirty-candidate filter so the daily cadence always makes progress. report_runs (the job table) keeps its 7-day reap (run_reap_days, domain 7..=90) — provenance lives on the generation, so reaping breaks nothing (the #1205 durable-record split, applied).

Universe materialization: phase 1 of every run drains the universe (keyset pages) into report_run_universe (generation_id, seq, item_id UUID, aux JSONB NULL) — checkpointed like any other phase. universe_total = the materialized count (exact; later total_in_scope disagreement is irrelevant — the snapshot is the work definition). Processing iterates the MATERIALIZED list by seq keyset: upstream insert/delete churn cannot skip or duplicate work. Rows reaped with the generation.

CMS-416’s local universe: materialized the same way from SELECT DISTINCT person_id … WHERE report_month >= $jan1 AND report_month < $jan1 + interval '1 year' AND chip_indicator = false (half-open dates, no EXTRACT) — plus a supporting index migration (report_month, chip_indicator, person_id) and an EXPLAIN pin. Downstream pinning: CMS-416 refuses to enqueue unless every month of the target year has a PUBLISHED tmsis generation; the ids land in input_generation_ids and the universe query filters generation_id = ANY(pinned). CMS-64 and WPR (staying synchronous) get the same guard: refuse unless the input generations for the period are published.

as_of: persons households:batchGet and persons:batchGet take as_of: NaiveDate (the store already supports it — batch.rs:28; the batch handler’s today-hardcoding is the gap being closed); the run pins as_of = period end for monthly kinds / snapshot date for QC. Sources WITHOUT a valid-time corpus (applications, enrollment, tanf, caps, snap-abawd) are read as-of-extraction; the generation’s extracted_at records it, the api doc states it, and the plan’s claim is narrowed accordingly: universes are reproducible; enrichments are as-of-extraction snapshots. Valid-time for those sources = filed follow-up.

Pins: build_version (CARGO_PKG_VERSION) + params_hash (canonical hash of the loaded ReportingParameterTable) captured at generation creation; claim/reclaim compares — mismatch ⇒ finalize(error, stale_pins); the operator re-enqueues (fresh generation, fresh pins). A resumed run can never mix two FPL tables or fold algorithms.

generation_id UUID NOT NULL (created with the run at enqueue).

abandon_reason TEXT + abandoned_at — token-fenced best-effort write when the worker abandons; cleared on successful reclaim. Terminal attribution is durable: attempts-cap ⇒ upstream_unavailable when the last abandon_reason is transient-class, else crashed (fence loss / worker death / no recorded reason).

Attempts semantics (exact): report_run_claim refuses rows with attempts >= max_attempts — it terminalizes them (error, code per abandon_reason) in the same statement. Claim increments attempts; a run does work at attempts 1..=max and is terminalized at the (max+1)th claim attempt, never worked. Pinned by test.

Progress = a TAGGED enum, phase-explicit (sqlx::types::Json<RunProgress>; zero serde_json::Value in src): RunProgress::V1 { kind_tag, phase }, phase ∈ Draining { source_cursor: Option<KindCursor> } | Processing { after_seq: i64, aggregates: KindAggregates } | Drained { aggregates }. Kind↔tag mismatch on decode ⇒ contract_violation. Empty universe: Draining completes with total=0 ⇒ phase goes straight to Drained ⇒ finalize (0+0=0 reconciles).

Checkpoint integrity: report_run_checkpoint(id, token, expected_seq, new_progress, Δs…) — deltas CHECKed non-negative in-fn; cursor/seq strictly monotonic vs stored progress (regression RAISE); expected_seq makes a duplicate/replayed checkpoint a detectable no-op (idempotent — an ambiguous commit is safe to re-issue); universe_total set-once at Draining completion; degraded_count ⇐ processed_count table CHECK. MR5 (R2): a checkpoint_in(tx) variant was added (D4’s chunk tx requires the checkpoint inside the same transaction; MR4 shipped only the pool variant); a DuplicateNoOp observed mid-pass HALTS the pass conservatively — the run resumes cleanly on the next claim, and the safe-re-issue semantics hold at the SQL layer; the Fns388 aggregates carry expedited_unknown so the D6 detail class survives resume.

Malformed-progress handling: the claim wrapper fetches id/token/state RAW first, decodes progress SEPARATELY; decode failure ⇒ the worker holds a valid token and finalize(error, contract_violation)`s — never abandonment-by-panic. The status endpoint reads counters from COLUMNS + `detail_counters from the generation row — it never decodes progress.

The done CHECK (state <> 'done' OR (universe_total IS NOT NULL AND processed_count + skipped_orphan_count = universe_total)) proves counter arithmetic; OUTPUT completeness is proven by the promotion-time reconciliation (D1). Together they are the correctness gate — the claim is stated exactly that way, no stronger.

Heartbeat is a supervised pulse task through fetch/enrich/fold/DB work — a 90s upstream call under a 300s lease can’t be reclaimed mid-flight; config relationship rules require claim_secs >= 3×heartbeat_secs AND claim_secs > overall_call_deadline + heartbeat_secs (validated at boot, never clamped).

Full-call deadline: the client funnels wrap bearer acquisition + send + bounded body read + decode under ONE overall_timeout (~90s) — a stalled body cannot escape the budget (today’s retry_request wraps only send()).

Typed client errors: the reporting funnels return ReportingClientError { kind: TransientExhausted{class} | OverallTimeout | Auth | NotFound | Conflict | Unprocessable | Decode | Contract } instead of anyhow strings — the ladder dispatches on it. Idempotency keys are minted ONCE per logical call outside the retry closure.

Error dispatch (exhaustive): transient-exhausted / overall-timeout ⇒ abandon (durable reason); auth ⇒ abandon (auth); decode/contract/short-batch/drift/ pins ⇒ finalize(error, contract-class); local DB statement-timeout or deadlock ⇒ abandon (retryable); pool outage ⇒ engine error (backoff loop); promotion reconciliation failure or unique violation ⇒ finalize(error, contract_violation); final CHECK refusal ⇒ ditto (a bug, loud).

Concurrency: in-chunk legs try_join!; the one knob run_upstream_concurrency (default 16, domain 1..=64) bounds any residual per-row fan-out AND the buffer_unordered sub-batching when >500 adults arise in one ACF-199 chunk. No cross-chunk pipelining (out-of-order completion breaks the monotone cursor).

runs_enabled default TRUE; when false the worker parks AND enqueue answers 503 (a 202 for work that will never run is a lie) — the per-control operator override, accountable and visible.

Enqueue lock ordering: under the advisory lock, re-check active-target FIRST (⇒ 409 with handle) then capacity (⇒ 503). Queue cap bounds queued rows; concurrent RUNNING is bounded by worker replicas (documented; per-kind global bounds = filed follow-up).

Chunk txs use SET LOCAL statement/lock timeouts. Knobs CANOPY_REPORTING__RUN_* (from_config check()-validated): tick_ms 5000 [500..=60000] · first_tick_delay_secs 60 · claim_secs 300 [60..=600] · heartbeat_secs 60 · max_attempts 5 · max_queued 10 · chunk_size 200 [50..=200, must not exceed pagination::MAX_LIMIT] · upstream_concurrency 16 [1..=64] · runs_enabled true · reap_days 7 [7..=90] · generation_row_retention_days 30 [7..=365].

Worker health (impl, MR4): a reusable non-gating /readyz worker check — canopy_api::BackgroundWorkerHealth atomics (last pass/beat, last success, error streak, serviced count) stamped by the loop + the pulse; degraded NEVER 503s readiness (the outbox-check posture — pulling a replica cannot revive its in-process worker). OTel: claims/serviced/abandons/finalizes counters, queue depth + oldest-queued-age gauges (sampled in reap_if_due), and a last-success-age observable.

abawd_household migration: nullable + historical values NULLed with a migration comment (outage-fabricated false is indistinguishable from sourced false — they cannot be trusted as verified; NULL = "unverified legacy"). Ripple: domain Option<bool>, contracts, OpenAPI, CSV renders empty (#1155 precedent), data-model doc. The new pipeline always writes Some(_).

ACF-196/CMS-64: rows_affected()-honest counts + the refuse-unpublished-inputs guard (D2). Their synchronous 201 shape is unchanged.

Overclaim narrowing (docs + this plan’s claims): CMS-416 member_months = enrolled, eligible = enrolled, T-MSIS eligibility_start ≈ report_month + hardcoded managed-care fields, ACF unavailable-state defaults are PRE-EXISTING approximations — documented in api/canopy-reporting.adoc as known limitations, filed as a follow-up, excluded from this plan’s correctness claims.