fix: recover transient coordinator dead letters

test/autonomy.test.js

@@ -10,7 +10,7 @@ const { decide } = require('../src/autonomy/policy');
 const { validatePaperIntent, createSimulator } = require('../src/autonomy/execution');
 const { calculateOutcome } = require('../src/autonomy/outcomes');
 const { createOrderIntent } = require('../src/autonomy/orderIntents');
-const { reconcileArchiveBatch, reconcileLiveBatch } = require('../workers/autonomyWorker');
+const { enqueueCoordinatorEvent, reconcileArchiveBatch, reconcileLiveBatch } = require('../workers/autonomyWorker');
 const { scheduleNext } = require('../workers/replayWorker');
 
 test('autonomy schema and leased jobs are restart-safe', () => {
@@ -143,6 +143,60 @@ test('archive reconciliation prioritizes recent usable events and is bounded', (
   assert.equal(live.scanned, 1);
 });
 
+test('archive reconciliation recovers transient dead-lettered coordinator jobs', () => {
+  const intelligence = new Database(':memory:');
+  initAutonomySchema(intelligence);
+  enqueueJob(intelligence, {
+    jobType: 'coordinator_event', lane: 'historical', priority: 10, entityType: 'event', entityId: 99,
+    idempotencyKey: 'coordinator_event:99',
+  });
+  intelligence.prepare(`
+    UPDATE autonomy_jobs
+    SET status='dead_letter', attempts=5, last_error='TypeError: fetch failed'
+    WHERE idempotency_key='coordinator_event:99'
+  `).run();
+
+  const result = enqueueCoordinatorEvent(intelligence, {
+    event_id: 99,
+    ingested_at: new Date().toISOString(),
+    content: 'content',
+    has_embedding: 1,
+  });
+  assert.equal(result.recovered, true);
+  const job = intelligence.prepare("SELECT status, lane, priority, attempts, last_error FROM autonomy_jobs WHERE idempotency_key='coordinator_event:99'").get();
+  assert.equal(job.status, 'pending');
+  assert.equal(job.lane, 'live');
+  assert.equal(job.priority, 100);
+  assert.equal(job.attempts, 0);
+  assert.match(job.last_error, /Recovered transient coordinator failure/);
+});
+
+test('archive reconciliation leaves non-transient coordinator dead letters alone', () => {
+  const intelligence = new Database(':memory:');
+  initAutonomySchema(intelligence);
+  enqueueJob(intelligence, {
+    jobType: 'coordinator_event', lane: 'historical', priority: 10, entityType: 'event', entityId: 100,
+    idempotencyKey: 'coordinator_event:100',
+  });
+  intelligence.prepare(`
+    UPDATE autonomy_jobs
+    SET status='dead_letter', attempts=5, last_error='no tradable instruments are allowlisted'
+    WHERE idempotency_key='coordinator_event:100'
+  `).run();
+
+  const result = enqueueCoordinatorEvent(intelligence, {
+    event_id: 100,
+    ingested_at: new Date().toISOString(),
+    content: 'content',
+    has_embedding: 1,
+  });
+  assert.equal(result.recovered, false);
+  const job = intelligence.prepare("SELECT status, attempts, last_error FROM autonomy_jobs WHERE idempotency_key='coordinator_event:100'").get();
+  assert.equal(job.status, 'dead_letter');
+  assert.equal(job.attempts, 5);
+  assert.equal(job.last_error, 'no tradable instruments are allowlisted');
+});
+
 test('outcome calculation uses benchmark-relative return', () => {
   const outcome = calculateOutcome(
     { information_cutoff: '2026-01-02T00:00:00Z', horizon_days: 5, direction: 'positive' },

workers/autonomyWorker.js

@@ -5,6 +5,54 @@ const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/auton
 
 function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
 
+function isTransientCoordinatorFailure(error) {
+  const value = String(error || '').toLowerCase();
+  return value.includes('fetch failed')
+    || value.includes('network')
+    || value.includes('timeout')
+    || value.includes('before response')
+    || /\b(408|429|5\d\d)\b/.test(value);
+}
+
+function enqueueCoordinatorEvent(intelligenceDb, row) {
+  const isLive = row.ingested_at && Date.now() - Date.parse(row.ingested_at) <= 48 * 60 * 60 * 1000;
+  const lane = isLive ? 'live' : 'historical';
+  const priority = isLive ? 100 : 10;
+  const idempotencyKey = `coordinator_event:${row.event_id}`;
+  const result = enqueueJob(intelligenceDb, {
+    jobType: 'coordinator_event',
+    lane,
+    priority,
+    entityType: 'event',
+    entityId: row.event_id,
+    idempotencyKey,
+  });
+  if (result.inserted) return { inserted: true, recovered: false };
+
+  const existing = intelligenceDb.prepare(`
+    SELECT id, status, last_error
+    FROM autonomy_jobs
+    WHERE idempotency_key = ? AND job_type = 'coordinator_event'
+  `).get(idempotencyKey);
+  if (!existing || existing.status !== 'dead_letter' || !isTransientCoordinatorFailure(existing.last_error)) {
+    return { inserted: false, recovered: false };
+  }
+
+  const recovered = intelligenceDb.prepare(`
+    UPDATE autonomy_jobs
+    SET status = 'pending',
+        lane = ?,
+        priority = ?,
+        attempts = 0,
+        available_at = datetime('now'),
+        leased_by = NULL,
+        lease_expires_at = NULL,
+        last_error = ?
+    WHERE id = ? AND status = 'dead_letter'
+  `).run(lane, priority, `Recovered transient coordinator failure: ${existing.last_error || 'unknown error'}`, existing.id);
+  return { inserted: false, recovered: recovered.changes > 0 };
+}
+
 function reconcileArchiveBatch(archiveDb, intelligenceDb, batchSize = 250) {
   const cursor = intelligenceDb.prepare("SELECT value FROM autonomy_cursors WHERE key = 'archive_reconcile'").get();
   const afterId = cursor ? cursor.value : 0;
@@ -24,17 +72,9 @@ function reconcileArchiveBatch(archiveDb, intelligenceDb, batchSize = 250) {
   }
   const enqueue = intelligenceDb.transaction(() => {
     for (const row of rows) {
-      const isLive = row.ingested_at && Date.now() - Date.parse(row.ingested_at) <= 48 * 60 * 60 * 1000;
       const readyForIntelligence = row.event_id && row.content && row.has_embedding;
       if (readyForIntelligence) {
-        enqueueJob(intelligenceDb, {
-          jobType: 'coordinator_event',
-          lane: isLive ? 'live' : 'historical',
-          priority: isLive ? 100 : 10,
-          entityType: 'event',
-          entityId: row.event_id,
-          idempotencyKey: `coordinator_event:${row.event_id}`,
-        });
+        enqueueCoordinatorEvent(intelligenceDb, row);
       }
     }
     intelligenceDb.prepare(`
@@ -57,12 +97,8 @@ function reconcileLiveBatch(archiveDb, intelligenceDb, batchSize = 250) {
   let queued = 0;
   for (const row of rows) {
     if (!row.event_id || !row.content || !row.has_embedding) continue;
-    const result = enqueueJob(intelligenceDb, {
-      jobType: 'coordinator_event', lane: 'live', priority: 100,
-      entityType: 'event', entityId: row.event_id,
-      idempotencyKey: `coordinator_event:${row.event_id}`,
-    });
-    if (result.inserted) queued++;
+    const result = enqueueCoordinatorEvent(intelligenceDb, row);
+    if (result.inserted || result.recovered) queued++;
   }
   return { scanned: rows.length, queued };
 }
@@ -98,4 +134,4 @@ async function runAutonomyWorker({ archivePath, intelligencePath, workerId = `au
   }
 }
 
-module.exports = { reconcileArchiveBatch, reconcileLiveBatch, runAutonomyWorker };
+module.exports = { enqueueCoordinatorEvent, reconcileArchiveBatch, reconcileLiveBatch, runAutonomyWorker };