diff --git a/workers/publisher/src/index.ts b/workers/publisher/src/index.ts index 9333247..49416ec 100644 --- a/workers/publisher/src/index.ts +++ b/workers/publisher/src/index.ts @@ -44,6 +44,24 @@ async function scheduleValidatedCandidates( ON promotion.candidate_id = candidate.candidate_id WHERE candidate.candidate_status = 'validated' AND candidate.gate_status IN ('rule-pass', 'dual-pass') + AND json_array_length(candidate.facts_json) > 0 + AND NOT EXISTS ( + SELECT 1 + FROM json_each(candidate.facts_json) fact + WHERE NOT EXISTS ( + SELECT 1 + FROM promotion_field_mappings mapping + JOIN records target ON target.id = mapping.subject_record_id + JOIN field_definitions definition + ON definition.record_kind = target.kind + AND definition.field_path = mapping.canonical_field_path + WHERE mapping.source_id = candidate.source_id + AND mapping.enabled = 1 + AND mapping.candidate_field_path = CASE + WHEN fact.type = 'object' THEN json_extract(fact.value, '$.fieldPath') + END + ) + ) AND ( promotion.candidate_id IS NULL OR ( @@ -94,6 +112,8 @@ async function handleQueue( if (result.status === 'busy') { message.retry({ delaySeconds: 60 }) } else { + // A deferred dependency cannot be repaired by queue retries. The + // scheduler will rediscover it once every exact mapping is available. message.ack() } } catch (error) { diff --git a/workers/publisher/src/promoter.ts b/workers/publisher/src/promoter.ts index ada98b3..8c433d0 100644 --- a/workers/publisher/src/promoter.ts +++ b/workers/publisher/src/promoter.ts @@ -138,6 +138,13 @@ class UnsafeCandidateError extends Error { } } +class DeferredCandidateError extends Error { + constructor(readonly code: 'field_mapping_missing', message: string) { + super(message) + this.name = 'DeferredCandidateError' + } +} + function unsafe(code: string, issue: string | string[]): never { throw new UnsafeCandidateError(code, Array.isArray(issue) ? issue : [issue]) } @@ -503,7 +510,12 @@ async function buildPlan( const facts: PlannedFact[] = [] for (const fact of candidate.facts) { const mapping = byField.get(fact.fieldPath) - if (!mapping) unsafe('field_mapping_missing', `No exact promotion mapping for ${fact.fieldPath}`) + if (!mapping) { + throw new DeferredCandidateError( + 'field_mapping_missing', + `No exact promotion mapping for ${fact.fieldPath}`, + ) + } if (['quarantined', 'archived', 'rejected'].includes(mapping.workflow_status)) { unsafe('target_record_blocked', `Target record ${mapping.subject_record_id} is ${mapping.workflow_status}`) } @@ -1063,6 +1075,11 @@ export async function promoteCandidate( const candidate = await validateCandidate(row) plan = await buildPlan(database, candidate) } catch (error) { + if (error instanceof DeferredCandidateError) { + // Mapping is an operator-owned dependency, not an evidence failure. + // Keep the validated candidate untouched so a later poll can retry it. + return { candidateId, status: 'deferred', reasonCode: error.code } + } if (!(error instanceof UnsafeCandidateError)) throw error await isolateCandidate(database, candidateId, error.code, error.issues, now) return { candidateId, status: 'quarantined', reasonCode: error.code } diff --git a/workers/publisher/src/types.ts b/workers/publisher/src/types.ts index 4297d05..2921775 100644 --- a/workers/publisher/src/types.ts +++ b/workers/publisher/src/types.ts @@ -65,7 +65,7 @@ export interface PublisherEnv { export type PromotionResult = { candidateId: string - status: 'applied' | 'already-applied' | 'quarantined' | 'busy' + status: 'applied' | 'already-applied' | 'quarantined' | 'busy' | 'deferred' publicationJobId?: string reasonCode?: string } diff --git a/workers/publisher/tests/promoter.sqlite.test.ts b/workers/publisher/tests/promoter.sqlite.test.ts index 81d5748..798b947 100644 --- a/workers/publisher/tests/promoter.sqlite.test.ts +++ b/workers/publisher/tests/promoter.sqlite.test.ts @@ -13,11 +13,15 @@ import type { ExtractionFact, SourceManifestV1, } from '../../ingestion/src/types' +import { handleQueue, scheduleValidatedCandidates } from '../src/index' import { promoteCandidate } from '../src/promoter' import type { D1Database, D1PreparedStatement, D1Result, + PromotionFailure, + PromotionJob, + PublisherEnv, } from '../src/types' type SqliteValue = string | number | bigint | Uint8Array | null @@ -121,6 +125,12 @@ type SeedOptions = { canonicalFieldPath?: string canonicalValueType?: 'url' | 'integer' | 'decimal_minor' | 'date' | 'json' valueTransform?: 'identity' | 'major_to_minor_2' + additionalFields?: Array<{ + fieldPath: string + value: string + canonicalFieldPath: string + withMapping: boolean + }> } async function seedCandidate(database: DatabaseSync, options: SeedOptions) { @@ -133,6 +143,7 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { const fieldType = options.fieldType ?? 'string' const canonicalFieldPath = options.canonicalFieldPath ?? 'admissions_url' const canonicalValueType = options.canonicalValueType ?? 'url' + const additionalFields = options.additionalFields ?? [] const field = { path: fieldPath, type: fieldType, @@ -154,7 +165,12 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { extraction: { mode: options.extractor === 'rules' ? 'rules-only' : 'minimax', schemaVersion: `schema-${options.candidateId}`, - fields: [field], + fields: [field, ...additionalFields.map((item) => ({ + path: item.fieldPath, + type: 'string' as const, + required: true, + critical: options.critical, + }))], rules: options.extractor === 'rules' ? [{ kind: 'regex', fieldPath, pattern: 'Apply: (https://\\S+)' }] : undefined, @@ -163,21 +179,30 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { const primaryEvidence = { quote: `Apply at ${value}`, locator: 'text:primary' } const secondaryEvidence = { quote: `Official application: ${value}`, locator: 'text:secondary' } const fact: ExtractionFact = { fieldPath, value, evidence: primaryEvidence } + const facts: ExtractionFact[] = [fact, ...additionalFields.map((item) => ({ + fieldPath: item.fieldPath, + value: item.value, + evidence: { quote: `Apply at ${item.value}`, locator: 'text:primary' }, + }))] const primary: ExtractionEnvelope = { sourceId, schemaVersion: manifest.extraction.schemaVersion, - facts: [fact], + facts, } const secondary: ExtractionEnvelope = { sourceId, schemaVersion: manifest.extraction.schemaVersion, - facts: [{ fieldPath, value, evidence: secondaryEvidence }], + facts: [{ fieldPath, value, evidence: secondaryEvidence }, ...additionalFields.map((item) => ({ + fieldPath: item.fieldPath, + value: item.value, + evidence: { quote: `Official application: ${item.value}`, locator: 'text:secondary' }, + }))], } const provenance = options.extractor === 'rules' - ? await ruleCandidateProvenance(manifest, [fact], options.critical) + ? await ruleCandidateProvenance(manifest, facts, options.critical) : await miniMaxCandidateProvenance( manifest, - [fact], + facts, primary, secondary, 'MiniMax-M2.7', @@ -194,6 +219,13 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { record_kind, field_path, value_type, risk_class, required_for_publish, max_age_days ) VALUES ('organization', ?, ?, ?, 1, 30)`, ).run(canonicalFieldPath, canonicalValueType, options.critical ? 'critical' : 'low') + for (const item of additionalFields) { + database.prepare( + `INSERT INTO field_definitions ( + record_kind, field_path, value_type, risk_class, required_for_publish, max_age_days + ) VALUES ('organization', ?, 'url', ?, 1, 30)`, + ).run(item.canonicalFieldPath, options.critical ? 'critical' : 'low') + } database.prepare( `INSERT INTO source_documents ( id, public_id, canonical_url, source_kind, authority_level, @@ -236,6 +268,14 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { ).run(sourceId, fieldPath, options.valueTransform) } } + for (const item of additionalFields) { + if (!item.withMapping) continue + database.prepare( + `INSERT INTO promotion_field_mappings ( + source_id, candidate_field_path, subject_record_id, canonical_field_path + ) VALUES (?, ?, ?, ?)`, + ).run(sourceId, item.fieldPath, recordId, item.canonicalFieldPath) + } database.prepare( `INSERT INTO ingestion_candidates ( candidate_id, source_id, snapshot_id, extractor, gate_status, @@ -247,7 +287,7 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { snapshotId, options.extractor, options.gateStatus, - JSON.stringify([fact]), + JSON.stringify(facts), ) database.prepare( `INSERT INTO ingestion_candidate_provenance ( @@ -267,7 +307,7 @@ async function seedCandidate(database: DatabaseSync, options: SeedOptions) { JSON.stringify(provenance.fieldEvidence), provenance.containsCritical ? 1 : 0, ) - return { candidateId: options.candidateId, recordId, value } + return { candidateId: options.candidateId, sourceId, recordId, value } } function fixture() { @@ -276,6 +316,18 @@ function fixture() { return { sqlite, database: new SqliteD1(sqlite) } } +function environmentFixture(database: D1Database, limit = '20') { + const jobs: PromotionJob[] = [] + const failures: PromotionFailure[] = [] + const environment: PublisherEnv = { + PIPELINE_DB: database, + PROMOTION_QUEUE: { async send(job) { jobs.push(job) } }, + PUBLISHER_DLQ: { async send(failure) { failures.push(failure) } }, + SCHEDULE_BATCH_LIMIT: limit, + } + return { environment, jobs, failures } +} + test('validated dual candidate atomically creates canonical data and a publication outbox job', async () => { const { sqlite, database } = fixture() try { @@ -411,7 +463,7 @@ test('quarantined candidates and critical rule-only candidates never publish', a } }) -test('an unknown field mapping is isolated without guessing a canonical target', async () => { +test('an unknown field mapping is deferred without guessing a target or isolating evidence', async () => { const { sqlite, database } = fixture() try { const seeded = await seedCandidate(sqlite, { @@ -428,14 +480,247 @@ test('an unknown field mapping is isolated without guessing a canonical target', new Date('2026-07-20T01:00:00.000Z'), ) - assert.equal(result.status, 'quarantined') + assert.equal(result.status, 'deferred') assert.equal(result.reasonCode, 'field_mapping_missing') assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM claims`).get()?.count, 0) assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM canonical_fields`).get()?.count, 0) assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM publication_jobs`).get()?.count, 0) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM candidate_promotions`).get()?.count, 0) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = ?`, + ).get(seeded.candidateId)?.candidate_status, 'validated') + const repeated = await promoteCandidate(database, seeded.candidateId) + assert.equal(repeated.status, 'deferred') + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + } finally { + sqlite.close() + } +}) + +test('a deferred candidate promotes exactly once after its exact mapping is added', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-mapping-added', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: false, + }) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'deferred') + sqlite.prepare( + `INSERT INTO promotion_field_mappings ( + source_id, candidate_field_path, subject_record_id, canonical_field_path + ) VALUES (?, 'applicationUrl', ?, 'admissions_url')`, + ).run(seeded.sourceId, seeded.recordId) + + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'applied') + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'already-applied') + for (const table of ['claims', 'canonical_fields', 'publication_jobs', 'outbox_events', 'record_versions']) { + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM ${table}`).get()?.count, 1, table) + } + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = ?`, + ).get(seeded.candidateId)?.candidate_status, 'applied') + } finally { + sqlite.close() + } +}) + +test('partial mappings defer the entire candidate and do not consume scheduler capacity', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-partial-mapping', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: true, + additionalFields: [{ + fieldPath: 'contactUrl', value: 'https://apply.example.edu.cn/contact', + canonicalFieldPath: 'contact_url', withMapping: false, + }], + }) + const { environment, jobs } = environmentFixture(database) + const controller = { scheduledTime: Date.parse('2026-07-20T01:00:00.000Z'), cron: '37 * * * *' } + await scheduleValidatedCandidates(controller, environment) + assert.equal(jobs.length, 0) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'deferred') + for (const table of ['claims', 'canonical_fields', 'publication_jobs', 'candidate_promotions', 'promotion_isolations']) { + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM ${table}`).get()?.count, 0, table) + } + sqlite.prepare( + `INSERT INTO promotion_field_mappings ( + source_id, candidate_field_path, subject_record_id, canonical_field_path + ) VALUES (?, 'contactUrl', ?, 'contact_url')`, + ).run(seeded.sourceId, seeded.recordId) + await scheduleValidatedCandidates(controller, environment) + assert.deepEqual(jobs.map((job) => job.candidateId), [seeded.candidateId]) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'applied') + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM claims`).get()?.count, 2) + } finally { + sqlite.close() + } +}) + +test('disabled exact mappings stay deferred until re-enabled', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-disabled-mapping', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: true, + }) + sqlite.prepare(`UPDATE promotion_field_mappings SET enabled = 0 WHERE source_id = ?`).run(seeded.sourceId) + const { environment, jobs } = environmentFixture(database) + const controller = { scheduledTime: Date.parse('2026-07-20T01:00:00.000Z'), cron: '37 * * * *' } + await scheduleValidatedCandidates(controller, environment) + assert.equal(jobs.length, 0) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'deferred') + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM candidate_promotions`).get()?.count, 0) + sqlite.prepare(`UPDATE promotion_field_mappings SET enabled = 1 WHERE source_id = ?`).run(seeded.sourceId) + await scheduleValidatedCandidates(controller, environment) + assert.deepEqual(jobs.map((job) => job.candidateId), [seeded.candidateId]) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'applied') + } finally { + sqlite.close() + } +}) + +test('scheduler excludes unmapped candidates before LIMIT so older gaps cannot starve ready facts', async () => { + const { sqlite, database } = fixture() + try { + const pending = await seedCandidate(sqlite, { + candidateId: 'candidate-a-unmapped', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: false, + canonicalFieldPath: 'unmapped_admissions_url', + }) + const ready = await seedCandidate(sqlite, { + candidateId: 'candidate-b-ready', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: true, + canonicalFieldPath: 'ready_admissions_url', + }) + const { environment, jobs } = environmentFixture(database, '1') + await scheduleValidatedCandidates({ + scheduledTime: Date.parse('2026-07-20T01:00:00.000Z'), cron: '37 * * * *', + }, environment) + assert.deepEqual(jobs.map((job) => job.candidateId), [ready.candidateId]) + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = ?`, + ).get(pending.candidateId)?.candidate_status, 'validated') + } finally { + sqlite.close() + } +}) + +test('malformed fact array entries cannot abort scheduling a healthy candidate', async () => { + const { sqlite, database } = fixture() + try { + const ready = await seedCandidate(sqlite, { + candidateId: 'candidate-z-healthy', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: true, + }) + sqlite.prepare( + `INSERT INTO ingestion_candidates ( + candidate_id, source_id, snapshot_id, extractor, gate_status, + candidate_status, facts_json, issues_json, created_at, validated_at + ) SELECT 'candidate-a-malformed', source_id, snapshot_id, extractor, gate_status, + 'validated', ?, '[]', '2026-07-19T00:00:00.000Z', '2026-07-19T00:00:00.000Z' + FROM ingestion_candidates WHERE candidate_id = ?`, + ).run(JSON.stringify(['bad', null, 42, [], {}]), ready.candidateId) + const { environment, jobs } = environmentFixture(database, '1') + await scheduleValidatedCandidates({ + scheduledTime: Date.parse('2026-07-20T01:00:00.000Z'), cron: '37 * * * *', + }, environment) + assert.deepEqual(jobs.map((job) => job.candidateId), [ready.candidateId]) + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = 'candidate-a-malformed'`, + ).get()?.candidate_status, 'validated') + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM candidate_promotions`).get()?.count, 0) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + } finally { + sqlite.close() + } +}) + +test('a mapping without its field definition stays deferred and resumes when the definition returns', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-definition-missing', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: true, + }) + sqlite.prepare( + `DELETE FROM field_definitions WHERE record_kind = 'organization' AND field_path = 'admissions_url'`, + ).run() + assert.equal(sqlite.prepare( + `SELECT count(*) AS count FROM promotion_field_mappings WHERE source_id = ? AND enabled = 1`, + ).get(seeded.sourceId)?.count, 1) + const { environment, jobs } = environmentFixture(database) + const controller = { scheduledTime: Date.parse('2026-07-20T01:00:00.000Z'), cron: '37 * * * *' } + await scheduleValidatedCandidates(controller, environment) + assert.equal(jobs.length, 0) + const deferred = await promoteCandidate(database, seeded.candidateId) + assert.equal(deferred.status, 'deferred') + assert.equal(deferred.reasonCode, 'field_mapping_missing') + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = ?`, + ).get(seeded.candidateId)?.candidate_status, 'validated') + for (const table of ['claims', 'canonical_fields', 'candidate_promotions', 'promotion_isolations']) { + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM ${table}`).get()?.count, 0, table) + } + sqlite.prepare( + `INSERT INTO field_definitions ( + record_kind, field_path, value_type, risk_class, required_for_publish, max_age_days + ) VALUES ('organization', 'admissions_url', 'url', 'low', 1, 30)`, + ).run() + await scheduleValidatedCandidates(controller, environment) + assert.deepEqual(jobs.map((job) => job.candidateId), [seeded.candidateId]) + assert.equal((await promoteCandidate(database, seeded.candidateId)).status, 'applied') + } finally { + sqlite.close() + } +}) + +test('queue acknowledges deferred dependencies without retrying or sending them to DLQ', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-queue-deferred', + extractor: 'minimax-dual', gateStatus: 'dual-pass', critical: false, withMapping: false, + }) + const { environment, failures } = environmentFixture(database) + let acknowledgements = 0 + let retries = 0 + await handleQueue({ messages: [{ + id: 'message-deferred', attempts: 4, + body: { version: 1, candidateId: seeded.candidateId, requestedAt: '2026-07-20T01:00:00.000Z' }, + ack() { acknowledgements += 1 }, + retry() { retries += 1 }, + }] }, environment) + assert.equal(acknowledgements, 1) + assert.equal(retries, 0) + assert.deepEqual(failures, []) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM publication_jobs`).get()?.count, 0) + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM promotion_isolations`).get()?.count, 0) + assert.equal(sqlite.prepare( + `SELECT candidate_status FROM ingestion_candidates WHERE candidate_id = ?`, + ).get(seeded.candidateId)?.candidate_status, 'validated') + } finally { + sqlite.close() + } +}) + +test('a critical rule-only manifest stays quarantined even when mappings are missing', async () => { + const { sqlite, database } = fixture() + try { + const seeded = await seedCandidate(sqlite, { + candidateId: 'candidate-unsafe-unmapped', + extractor: 'rules', gateStatus: 'rule-pass', critical: true, withMapping: false, + }) + const result = await promoteCandidate(database, seeded.candidateId) + assert.equal(result.status, 'quarantined') + assert.equal(result.reasonCode, 'source_manifest_invalid') + assert.equal(sqlite.prepare(`SELECT count(*) AS count FROM publication_jobs`).get()?.count, 0) assert.equal(sqlite.prepare( `SELECT reason_code FROM promotion_isolations WHERE candidate_id = ?`, - ).get(seeded.candidateId)?.reason_code, 'field_mapping_missing') + ).get(seeded.candidateId)?.reason_code, 'source_manifest_invalid') } finally { sqlite.close() }