diff --git a/test/autonomy.test.js b/test/autonomy.test.js index 31b7979..ba04966 100644 --- a/test/autonomy.test.js +++ b/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' }, diff --git a/workers/autonomyWorker.js b/workers/autonomyWorker.js index e4a0b0a..3f23202 100644 --- a/workers/autonomyWorker.js +++ b/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 };