Story-2982: the results consumer — verify, then make the artifact terminal truth

RFD-AGT-014 decoupled the brain from the invoker: the broker (Story-2981) writes the pending row and mints one create-only presigned key, the brain PUTs its agent_result_artifact.v1 there, and S3 rings the bell (Chore-2929's results queue). This story is the moment the artifact becomes truth. The consumer (src/workers/agent_results_consumer/) verifies what landed and — in ONE transaction — CASes the run terminal, persists the verified effects, appends one billing.vendor_cost_event per usage receipt, and inserts the outbox event. The sweeper (Story-2980) deliberately never stamps delivered: digest/size/schema verification and the outcome discriminator are this consumer's contract (D15).

Verify-before-parse: recompute, never trust the label

The sha256 in S3 object metadata is written by the same party as the body — it proves nothing about substitution (adversary A3). The consumer RECOMPUTES the digest of the bytes it actually read and compares against the declared pair. The red-first proof: a tampered body with valid metadata (the original's digest) is REFUSED — tests/unit/test_agent_results_consumer.py::test_tampered_body_with_valid_metadata_is_refused.

The validator is revived, not forked: src/shared/context/production_blake.py::resolve_blake_payload already did pointer-parse → firm-prefix assert → fetch → verify size + sha256 → parse (live call site at document_registration_consumer/processor.py:368). The verify core is now shared, and the new resolve_agent_run_artifact applies it to the broker's terminal-key grammar (agent-runs/{firm_id}/{yyyymm}/{run_id}/{seat}/result.json): prefix assert BEFORE any read → GetObject → recompute + compare → size limit → strict schema parse. An unknown top-level schema version is refused, never best-effort parsed (D15) — proven red-first by test_unknown_schema_version_is_refused_not_parsed (unit) and test_unknown_schema_version_is_refused_never_parsed (integration, real CHECKs).

Precedence (D5) — six rows, all integration-tested against real CHECKs + the absorbing trigger

CurrentIncomingOutcomeProof
awaitingsuccesspersisted — CAS to delivered+succeeded test_awaiting_success_persists_the_four_writes
awaitingfailure envelopepersisted_failure — delivered+failed, honest error from the brain's envelope; receipts still land test_awaiting_failure_envelope_persists_failed_honest
deliveredsame digestduplicate_ignored — the (run_id, sha256) receipt row is the dedup test_delivered_same_digest_is_duplicate_ignored
delivereddifferent digestrejected_after_success — logged loud + metric; run row untouched test_delivered_different_digest_is_rejected_after_success
sweptsuccessanomaly_swept_success — NEVER applied: no effects, no receipts, no status change (succeeded is absorbing; D7's generous clock makes a late arrival genuinely anomalous) test_swept_success_is_anomaly_never_applied
sweptfailureswept_failure_restamped — terminal failure stands; the envelope's receipts are real spend and land idempotently test_swept_failure_is_idempotent_restamp

One transaction, four writes

staff-bracket read of the pending row (0219 staff SELECT policy) → key match BEFORE any read
→ per-artifact span AGENT_RESULT_DELIVERED (CONSUMER) PARENTED from the row's stored trace context (D8)
→ verify (revived resolve core) → strict parse → D21 provenance
→ firm-scope transaction (SB-48: the database refuses cross-tenant, not our control flow):
    1. SELECT ... FOR UPDATE, re-check delivery_status under the lock
    2. delivery receipt INSERT ... ON CONFLICT (run_id, sha256) DO NOTHING  ← redelivery dedup
    3. CAS awaiting_artifact → delivered, with the 2959 status product
       (succeeded | failed; failed ⟺ error_message non-empty — the 0214 CHECK makes a lie unrepresentable)
    4. persist effects → agents.agent_run_artifact_effect (0219), D19-monotonic per projection target
    5. one billing.vendor_cost_event per receipt (D6 / ADR-177 — dimensions-as-columns, frozen-FX INR)
    6. insert_domain_event agent_run.artifact_delivered (R7 — never put_events)
→ AFTER commit: structured outcome log (D13) + bounded delivery_outcomes metric

D6 — N generations → N ledger rows, idempotent under redelivery

Every receipt in a delivered artifact lands exactly one billing.vendor_cost_event, with the provider generation id as the idempotency key and the clerk's openrouter:<run_id>:seq:N fallback when the provider reported none (D6's corrected receipt identity). provider_actual rides the integer micro-USD end-to-end; the estimate lane prices from the brain-stamped estimated_cost_usd via Decimal — money never takes a float detour (ADR-173 §4). The frozen-FX INR columns ALREADY exist (ADR-177, pramaan-functions#773) — this story adds none. Proof: test_receipts_fall_back_to_artifact_local_sequence — 3 generations, 1 with a provider id → exactly 3 rows, 2 on the seq fallback, still 3 after redelivery.

D19 — cross-run ordering is monotonic (AC-40)

D5 governs precedence WITHIN a run; nothing governed across runs. Two reads of the same document now resolve by run start: an effect applies only if its run is not older than what is already applied for that projection target. The guard is one indexed comparison against agent_run_artifact_effect (which denormalizes document_id + run_started_at exactly for this). A stale arrival is recorded as stale_effect_skipped, never applied — the fresher findings remain the record. Proof: test_stale_run_effect_is_anomaly_never_applied.

D21 — the 412 detector, consumer half (AC-37)

Create-only is first-writer-wins, and the key is empty for the whole run. A second holder of the presigned URL who writes first plants terminal truth and 412s the honest brain out. The brain's own detector is its PUT refusing with HTTP 412; the consumer's detector is provenance: the broker stamps the trace context into the invoke envelope and onto the pending row (Story-2981), and the URL never carries it — so a writer that cannot echo it in the artifact envelope is not this run's brain. Provenance failure ⇒ loud anomaly + QUARANTINE: honest terminal failure (artifact_quarantined_untrusted_writer), the object kept as evidence, effects and receipts NEVER applied. Proof: test_quarantine_when_the_writer_cannot_echo_the_trace.

D8 — one trace across the artifact hop

The per-artifact span functions.agent_result.delivered (registered CONSUMER, Chore-2935 amendment) parents from the pending row's stored traceparent/tracestate. The oracle test (tests/unit/test_agent_results_consumer_trace_oracle.py) asserts consumer.trace_id == producer.trace_id across the S3→SQS hop, plus the CONSUMER kind and the required-attribute contract (firm_id, run_id, agent.key, outcome — never baggage).

D13 — no silent skip

Every seam reports a structured agent_results_consumer.outcome line and feeds the bounded pramaan.agent_result.delivery_outcomes counter: unknown run_id · expected_key mismatch · digest/size failure · unknown version · duplicate · rejected_after_success · swept anomaly · quarantine · stale effect. The anomalous outcomes log at WARNING and their metric samples are pager signals.

Triggers: S3→SQS, the sweeper's event, and the startup backfill

TriggerShapeNote
SQS pramaan-agent-results-stagingS3 ObjectCreated notification Bound by Terraform output, never by name prefix — pramaan-agents-runs-* is a DIFFERENT epic; a notification from any other bucket is refused, never fetched.
EventBridge agent_run.artifact_observedThe sweeper's self-heal outbox event The event carries ids only (content ban) — the key is read back from run truth. Environment mismatch refused (the envelope contract).
Cold startBounded staff-bracket backfill scan The observed event fires ONCE ever per run; a consumer that was down would never learn. One bounded scan per cold start settles awaiting runs whose artifact IS in the bucket (AGENT_RESULTS_BACKFILL_BATCH_SIZE, default 25). Proof: test_startup_backfill_settles_present_and_skips_absent.

R5 — refused artifacts still terminate honestly

A refused artifact (tamper, unknown version, limit breach, quarantine) cannot be fixed by redelivery — the key is create-only. Leaving the row awaiting_artifact would make the sweeper self-heal it back to the consumer forever, so every refusal stamps the honest terminal: delivered+failed with a typed error_class, a lawyer-readable error_message, and the recomputed digest of the refused bytes as evidence. No non-terminal forever.

Schema: migration 0219

agents.agent_run_artifact_effect — append-only verified effects: one row per effect, FK-chained to the run, RLS firm-isolated, UPDATE/DELETE revoked from the app role. This is the durable truth the per-seat projections (rail cutovers T2c+) read from, so the consumer never reaches into a seat's own tables. The migration also creates the staff SELECT bracket on agents.agent_run the backfill needs — guarded on pg_policies because the sweeper's migration owns the same policy name; whichever lands second no-ops. Roundtrip-proven (downgrade → upgrade → alembic check clean) on the isolated test database.

Deploy path (R4 — non-vacuous canary)

.github/workflows/deploy-agent-results-consumer.yml mirrors the sweeper's ownership boundary: pramaan-infra owns the shell, IAM (role pramaan-agent-results-consumer-staging), the ESM, and the EventBridge rule; this workflow publishes code only — no aws iam, no queue/ESM/notification mutation anywhere (mechanically gated by tests/unit/test_agent_results_consumer_deploy_workflow.py, with planted-writer red proofs). The candidate canary exercises the REAL capabilities — the staff-scoped run-truth query and a HeadObject against the runs bucket (a deliberately-absent key must read as absent; AccessDenied fails the deploy) — before the live alias moves, and the post-promotion gate requires an ENABLED event-source mapping from EXACTLY pramaan-agent-results-staging and no foreign queue.

Verification