Merge pull request #27 from computersciencefreshmen/codex/publisher-mapping-recovery

Keep validated candidates recoverable until exact mappings exist

Author: computersciencefreshmen

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) {

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 }

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
 }