fix(ingestion): harden evidence replay gates

infra/d1/pipeline/migrations/0016_ingestion_candidate_replay_index.sql

@@ -0,0 +1,6 @@
+-- Speeds extractor-fingerprint replay checks for a saved source snapshot.
+-- candidate_id keeps the index covering for the provenance join.
+CREATE INDEX IF NOT EXISTS idx_ingestion_candidates_source_snapshot
+  ON ingestion_candidates(source_id, snapshot_id, candidate_id);
+
+PRAGMA optimize;

workers/ingestion/src/minimax.ts

@@ -295,6 +295,12 @@ function validateEnvelope(
       issues.push(`${label}: invalid ${field.type} value for ${fact.fieldPath}`)
       continue
     }
+    if (field.critical && (field.type === 'object' || field.type === 'string-array')) {
+      issues.push(
+        `${label}: critical composite field requires leaf-level evidence: ${fact.fieldPath}`,
+      )
+      continue
+    }
     const quote = normalizeEvidenceText(fact.evidence.quote)
     if (quote.length < 2 || quote.length > 1_000 || !normalizedSource.includes(quote)) {
       issues.push(`${label}: evidence is not grounded for ${fact.fieldPath}`)

workers/ingestion/src/pipeline.ts

@@ -587,7 +587,7 @@ export async function processIngestionJob(
     // An extractor or prompt upgrade must receive the official body again. Sending
     // validators here could yield 304 and strand the saved snapshot on an old
     // extraction fingerprint forever.
-    if (state.rawSha256 === null || previousCandidateExtractionCurrent) {
+    if (previousSnapshotId !== null && previousCandidateExtractionCurrent) {
       if (state.etag) headers.set('If-None-Match', state.etag)
       if (state.lastModified) headers.set('If-Modified-Since', state.lastModified)
     }
@@ -609,6 +609,13 @@ export async function processIngestionJob(
 
     if (response.status === 304) {
       clearTimeout(timeout)
+      if (previousSnapshotId === null || !previousCandidateExtractionCurrent) {
+        throw new IngestionError(
+          'Official source returned 304 without a current saved extraction',
+          'unexpected_304',
+          true,
+        )
+      }
       await recordNoChange(environment, {
         job,
         sourceId: manifest.id,

workers/ingestion/src/provenance.ts

@@ -8,6 +8,7 @@ import type {
 } from './types'
 
 export const MINIMAX_PROMPT_SPEC_VERSION = 'studyinchina-minimax-dual-v3'
+export const EVIDENCE_GATE_VERSION = 'studyinchina-evidence-gate-v2'
 export const RULE_EXTRACTOR_VERSION = 'studyinchina-rules-v1'
 
 export const MINIMAX_SYSTEM_INSTRUCTIONS = [
@@ -56,6 +57,7 @@ export async function miniMaxExtractorFingerprint(
 ): Promise<string> {
   return sha256Hex(stableJson({
     version: MINIMAX_PROMPT_SPEC_VERSION,
+    evidenceGateVersion: EVIDENCE_GATE_VERSION,
     model,
     promptFingerprint,
     schemaVersion: manifest.extraction.schemaVersion,

workers/ingestion/tests/entity-persistence-resilience.test.ts

@@ -4,6 +4,7 @@ import { resolve } from 'node:path'
 import { DatabaseSync } from 'node:sqlite'
 import test from 'node:test'
 import { IngestionError } from '../src/errors'
+import { sha256Hex } from '../src/hash'
 import { processIngestionJob } from '../src/pipeline'
 import { recordJobFailure } from '../src/repository'
 import type {
@@ -395,3 +396,112 @@ test('ordinary rules-only pages retain the existing completed-result path', asyn
     database.close()
   }
 })
+
+test('304 without a saved snapshot is retryable and never records no-change', async () => {
+  const database = databaseWithEntitySchema()
+  try {
+    const manifest = sourceManifest()
+    const job = testJob(manifest.id)
+    seedSourceAndJob(database, manifest, job)
+    database.prepare(
+      `UPDATE ingestion_sources
+          SET etag = '"orphan-etag"', last_modified = 'Mon, 20 Jul 2026 00:00:00 GMT'
+        WHERE source_id = ?`,
+    ).run(manifest.id)
+    const environment = environmentFor(database)
+    let officialFetches = 0
+    const fetcher: Fetcher = async (_input, init) => {
+      officialFetches += 1
+      const headers = new Headers(init?.headers)
+      assert.equal(headers.has('if-none-match'), false)
+      assert.equal(headers.has('if-modified-since'), false)
+      return new Response(null, { status: 304 })
+    }
+
+    await assert.rejects(
+      processIngestionJob(environment, job, fetcher, checkedAt),
+      (error: unknown) => error instanceof IngestionError
+        && error.code === 'unexpected_304'
+        && error.retryable,
+    )
+
+    assert.equal(officialFetches, 1)
+    assert.deepEqual(
+      plainRow(database.prepare(
+        `SELECT status, outcome, completed_at FROM ingestion_jobs WHERE job_id = ?`,
+      ).get(job.jobId)),
+      { status: 'running', outcome: null, completed_at: null },
+    )
+  } finally {
+    database.close()
+  }
+})
+
+test('304 with a stale extractor fingerprint is retryable and never records no-change', async () => {
+  const database = databaseWithEntitySchema()
+  try {
+    const manifest = sourceManifest()
+    const job = testJob(manifest.id)
+    seedSourceAndJob(database, manifest, job)
+    const rawSha256 = 'a'.repeat(64)
+    const canonicalSha256 = 'b'.repeat(64)
+    const snapshotId = await sha256Hex(`${manifest.id}:${rawSha256}`)
+    database.prepare(
+      `UPDATE ingestion_sources
+          SET raw_sha256 = ?, canonical_sha256 = ?,
+              etag = '"stale-etag"', last_modified = 'Mon, 20 Jul 2026 00:00:00 GMT'
+        WHERE source_id = ?`,
+    ).run(rawSha256, canonicalSha256, manifest.id)
+    database.prepare(
+      `INSERT INTO ingestion_snapshots
+         (snapshot_id, source_id, r2_key, raw_sha256, canonical_sha256, content_type,
+          byte_length, final_url, fetched_at, etag, last_modified)
+       VALUES (?, ?, ?, ?, ?, 'text/html', 1, ?, ?, '"stale-etag"',
+               'Mon, 20 Jul 2026 00:00:00 GMT')`,
+    ).run(
+      snapshotId,
+      manifest.id,
+      `snapshots/${manifest.id}/${rawSha256}.html`,
+      rawSha256,
+      canonicalSha256,
+      manifest.officialUrl,
+      checkedAt.toISOString(),
+    )
+    database.prepare(
+      `INSERT INTO ingestion_candidates
+         (candidate_id, source_id, snapshot_id, extractor, gate_status,
+          candidate_status, facts_json, issues_json, created_at)
+       VALUES ('stale-candidate', ?, ?, 'rules', 'rule-pass',
+               'extracted', '[]', '[]', ?)`,
+    ).run(manifest.id, snapshotId, checkedAt.toISOString())
+    database.prepare(
+      `INSERT INTO ingestion_candidate_provenance
+         (candidate_id, schema_version, extractor_fingerprint, field_evidence_json,
+          contains_critical, created_at)
+       VALUES ('stale-candidate', ?, ?, '[]', 0, ?)`,
+    ).run(manifest.extraction.schemaVersion, 'c'.repeat(64), checkedAt.toISOString())
+    const environment = environmentFor(database)
+    const fetcher: Fetcher = async (_input, init) => {
+      const headers = new Headers(init?.headers)
+      assert.equal(headers.has('if-none-match'), false)
+      assert.equal(headers.has('if-modified-since'), false)
+      return new Response(null, { status: 304 })
+    }
+
+    await assert.rejects(
+      processIngestionJob(environment, job, fetcher, checkedAt),
+      (error: unknown) => error instanceof IngestionError
+        && error.code === 'unexpected_304'
+        && error.retryable,
+    )
+
+    assert.deepEqual(
+      plainRow(database.prepare(
+        `SELECT status, outcome, completed_at FROM ingestion_jobs WHERE job_id = ?`,
+      ).get(job.jobId)),
+      { status: 'running', outcome: null, completed_at: null },
+    )
+  } finally {
+    database.close()
+  }
+})

workers/ingestion/tests/minimax.test.ts

@@ -52,6 +52,47 @@ test('dual extraction passes only when values agree and evidence is grounded', (
   )
 })
 
+test('critical composite fields remain quarantined without leaf-level evidence', () => {
+  const manifest = sourceManifest()
+  manifest.extraction.fields = [
+    { path: 'fees', type: 'object', required: true, critical: true },
+    { path: 'availableLanguages', type: 'string-array', required: true, critical: true },
+  ]
+  const compositeSource = 'Tuition information. Languages available.'
+  const compositeExtraction: ExtractionEnvelope = {
+    schemaVersion: manifest.extraction.schemaVersion,
+    sourceId: manifest.id,
+    facts: [
+      {
+        fieldPath: 'fees',
+        value: { tuition: 30_000, currency: 'CNY' },
+        evidence: { quote: 'Tuition information.' },
+      },
+      {
+        fieldPath: 'availableLanguages',
+        value: ['Chinese', 'English'],
+        evidence: { quote: 'Languages available.' },
+      },
+    ],
+  }
+
+  const result = gateDualExtractions(
+    compositeExtraction,
+    structuredClone(compositeExtraction),
+    manifest,
+    compositeSource,
+  )
+
+  assert.equal(result.status, 'quarantined')
+  assert.equal(result.facts.length, 0)
+  assert.ok(result.issues.some((issue) => issue.includes(
+    'critical composite field requires leaf-level evidence: fees',
+  )))
+  assert.ok(result.issues.some((issue) => issue.includes(
+    'critical composite field requires leaf-level evidence: availableLanguages',
+  )))
+})
+
 test('MiniMax adapter performs two independent passes through a configurable endpoint', async () => {
   const passes: string[] = []
   const redirectModes: Array<RequestRedirect | undefined> = []