From d291e54686c3ba34f30e5080093bad97301bb2a7 Mon Sep 17 00:00:00 2001 From: Henrythefoodie <13022037121@163.com> Date: Sat, 29 Aug 2026 15:14:26 +0800 Subject: [PATCH] fix(ingestion): harden evidence replay gates --- .../0016_ingestion_candidate_replay_index.sql | 6 + workers/ingestion/src/minimax.ts | 6 + workers/ingestion/src/pipeline.ts | 9 +- workers/ingestion/src/provenance.ts | 2 + .../entity-persistence-resilience.test.ts | 110 ++++++++++++++++++ workers/ingestion/tests/minimax.test.ts | 41 +++++++ 6 files changed, 173 insertions(+), 1 deletion(-) create mode 100644 infra/d1/pipeline/migrations/0016_ingestion_candidate_replay_index.sql diff --git a/infra/d1/pipeline/migrations/0016_ingestion_candidate_replay_index.sql b/infra/d1/pipeline/migrations/0016_ingestion_candidate_replay_index.sql new file mode 100644 index 0000000..ea9cb47 --- /dev/null +++ b/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; diff --git a/workers/ingestion/src/minimax.ts b/workers/ingestion/src/minimax.ts index d3b909f..8c19820 100644 --- a/workers/ingestion/src/minimax.ts +++ b/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}`) diff --git a/workers/ingestion/src/pipeline.ts b/workers/ingestion/src/pipeline.ts index b34e423..0c8e940 100644 --- a/workers/ingestion/src/pipeline.ts +++ b/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, diff --git a/workers/ingestion/src/provenance.ts b/workers/ingestion/src/provenance.ts index 2c75997..1509ac7 100644 --- a/workers/ingestion/src/provenance.ts +++ b/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 { return sha256Hex(stableJson({ version: MINIMAX_PROMPT_SPEC_VERSION, + evidenceGateVersion: EVIDENCE_GATE_VERSION, model, promptFingerprint, schemaVersion: manifest.extraction.schemaVersion, diff --git a/workers/ingestion/tests/entity-persistence-resilience.test.ts b/workers/ingestion/tests/entity-persistence-resilience.test.ts index 4316fdb..61cd57f 100644 --- a/workers/ingestion/tests/entity-persistence-resilience.test.ts +++ b/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() + } +}) diff --git a/workers/ingestion/tests/minimax.test.ts b/workers/ingestion/tests/minimax.test.ts index 55543a1..eef0ef6 100644 --- a/workers/ingestion/tests/minimax.test.ts +++ b/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 = []