From 2c023c8962bcd6c9a2514c934d306f44dfd4d504 Mon Sep 17 00:00:00 2001 From: ImBenji Date: Sat, 8 Aug 2026 23:32:41 +0100 Subject: [PATCH] fix: let replay recover from dead letter jobs --- src/autonomy/llm.js | 43 ++++++++++++++++++++---------- test/autonomy.test.js | 35 +++++++++++++++++++++++++ workers/replayWorker.js | 58 ++++++++++++++++++++++++++++------------- 3 files changed, 104 insertions(+), 32 deletions(-) diff --git a/src/autonomy/llm.js b/src/autonomy/llm.js index 7796e4e..ebe81b6 100644 --- a/src/autonomy/llm.js +++ b/src/autonomy/llm.js @@ -11,20 +11,35 @@ function extractJson(text) { async function callCoordinator(config, prompt) { const apiKey = String(config?.openRouter?.apiKey || '').trim(); if (!apiKey) throw new Error('OpenRouter API key is not configured'); - const response = await fetch('https://openrouter.ai/api/v1/chat/completions', { - method: 'POST', - headers: { Authorization: `Bearer ${apiKey}`, 'Content-Type': 'application/json' }, - body: JSON.stringify({ - model: config.openRouter.llmModel, - temperature: 0, - response_format: { type: 'json_object' }, - messages: [ - { role: 'system', content: 'You are a coordinator. Extract only evidence-backed categorical hypotheses. Never output probabilities, expected returns, confidence scores, position sizes, or trade actions.' }, - { role: 'user', content: prompt }, - ], - }), - }); - if (!response.ok) throw new Error(`coordinator request failed with ${response.status}`); + const timeoutMs = Math.max(1000, Number(config?.openRouter?.timeoutMs || process.env.OPEN_ROUTER_TIMEOUT_MS) || 60000); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), timeoutMs); + let response; + try { + response = await fetch('https://openrouter.ai/api/v1/chat/completions', { + method: 'POST', + signal: controller.signal, + headers: { Authorization: `Bearer ${apiKey}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + model: config.openRouter.llmModel, + temperature: 0, + response_format: { type: 'json_object' }, + messages: [ + { role: 'system', content: 'You are a coordinator. Extract only evidence-backed categorical hypotheses. Never output probabilities, expected returns, confidence scores, position sizes, or trade actions.' }, + { role: 'user', content: prompt }, + ], + }), + }); + } catch (error) { + const cause = error?.cause?.code || error?.code || error?.name || 'network_error'; + throw new Error(`coordinator request failed before response (${cause})`); + } finally { + clearTimeout(timeout); + } + if (!response.ok) { + const body = await response.text().catch(() => ''); + throw new Error(`coordinator request failed with ${response.status}: ${body.slice(0, 300)}`); + } const body = await response.json(); return extractJson(body?.choices?.[0]?.message?.content); } diff --git a/test/autonomy.test.js b/test/autonomy.test.js index 190b099..31b7979 100644 --- a/test/autonomy.test.js +++ b/test/autonomy.test.js @@ -11,6 +11,7 @@ const { validatePaperIntent, createSimulator } = require('../src/autonomy/execut const { calculateOutcome } = require('../src/autonomy/outcomes'); const { createOrderIntent } = require('../src/autonomy/orderIntents'); const { reconcileArchiveBatch, reconcileLiveBatch } = require('../workers/autonomyWorker'); +const { scheduleNext } = require('../workers/replayWorker'); test('autonomy schema and leased jobs are restart-safe', () => { const db = new Database(':memory:'); @@ -73,6 +74,40 @@ test('replay evidence cannot look beyond its information cutoff and remains non- assert.equal(executable.length, 0); }); +test('replay scheduler skips terminal replay jobs instead of pinning the cursor', () => { + const archive = new Database(':memory:'); + archive.exec(` + CREATE TABLE articles ( + id INTEGER PRIMARY KEY, + title TEXT, + description TEXT, + content TEXT, + pub_date_effective TEXT, + is_index_page INTEGER + ) + `); + archive.prepare("INSERT INTO articles VALUES (1, 'bad', '', 'content', '2020-01-01T00:00:00Z', 0)").run(); + archive.prepare("INSERT INTO articles VALUES (2, 'next', '', 'content', '2020-01-02T00:00:00Z', 0)").run(); + + const intelligence = new Database(':memory:'); + initAutonomySchema(intelligence); + const runId = intelligence.prepare(` + INSERT INTO autonomy_replay_runs (watermark_at, strategy_version, prompt_version, coordinator_model) + VALUES ('2020-01-03T00:00:00Z', 'test', 'test', 'test') + `).run().lastInsertRowid; + enqueueJob(intelligence, { + jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: 1, + idempotencyKey: `replay:${runId}:article:1`, + }); + intelligence.prepare("UPDATE autonomy_jobs SET status='dead_letter', attempts=5, last_error='fetch failed'").run(); + + const run = intelligence.prepare('SELECT * FROM autonomy_replay_runs WHERE id=?').get(runId); + const next = scheduleNext(intelligence, archive, run); + assert.equal(next.id, 2); + assert.equal(intelligence.prepare('SELECT cursor_article_id FROM autonomy_replay_runs WHERE id=?').get(runId).cursor_article_id, 1); + assert.equal(intelligence.prepare("SELECT COUNT(*) count FROM autonomy_jobs WHERE status='pending' AND entity_id='2'").get().count, 1); +}); + test('calibration and policy abstain on insufficient evidence', () => { const calibration = calibrateOutcomes([ { excess_return: 0.02, direction_correct: 1 }, diff --git a/workers/replayWorker.js b/workers/replayWorker.js index 8288f49..5f038d7 100644 --- a/workers/replayWorker.js +++ b/workers/replayWorker.js @@ -23,7 +23,7 @@ function articleTimeColumns(archiveDb) { const columns = new Set(archiveDb.prepare('PRAGMA table_info(articles)').all().map((row) => row.name)); const candidates = ['pub_date_effective', 'pub_date', 'ingested_at'].filter((name) => columns.has(name)); if (!candidates.length) throw new Error('archive articles need publication or ingestion timestamps for replay'); - return { columns, effective: `COALESCE(${candidates.join(', ')})` }; + return { columns, effective: candidates.length === 1 ? candidates[0] : `COALESCE(${candidates.join(', ')})` }; } function replayPrompt(article) { @@ -52,23 +52,45 @@ function scheduleNext(db, archiveDb, run) { const { columns, effective } = articleTimeColumns(archiveDb); const content = columns.has('content') ? "content IS NOT NULL AND content != ''" : '1=1'; const indexFilter = columns.has('is_index_page') ? 'AND (is_index_page = 0 OR is_index_page IS NULL)' : ''; - const cursorFilter = run.cursor_effective_at - ? `AND (datetime(${effective}) > datetime(?) OR (datetime(${effective}) = datetime(?) AND id > ?))` - : ''; - const params = [run.watermark_at]; - if (run.cursor_effective_at) params.push(run.cursor_effective_at, run.cursor_effective_at, run.cursor_article_id); - const article = archiveDb.prepare(` - SELECT id, title, description, content, ${effective} AS effective_at - FROM articles - WHERE ${content} ${indexFilter} AND datetime(${effective}) <= datetime(?) ${cursorFilter} - ORDER BY datetime(${effective}) ASC, id ASC LIMIT 1 - `).get(...params); - if (!article) return null; - enqueueJob(db, { - jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: article.id, - idempotencyKey: `replay:${run.id}:article:${article.id}`, - }); - return article; + let cursorEffectiveAt = run.cursor_effective_at; + let cursorArticleId = run.cursor_article_id; + + for (let skipped = 0; skipped < 100; skipped += 1) { + const cursorFilter = cursorEffectiveAt + ? `AND (datetime(${effective}) > datetime(?) OR (datetime(${effective}) = datetime(?) AND id > ?))` + : ''; + const params = [run.watermark_at]; + if (cursorEffectiveAt) params.push(cursorEffectiveAt, cursorEffectiveAt, cursorArticleId); + const article = archiveDb.prepare(` + SELECT id, title, description, content, ${effective} AS effective_at + FROM articles + WHERE ${content} ${indexFilter} AND datetime(${effective}) <= datetime(?) ${cursorFilter} + ORDER BY datetime(${effective}) ASC, id ASC LIMIT 1 + `).get(...params); + if (!article) return null; + + const idempotencyKey = `replay:${run.id}:article:${article.id}`; + const existing = db.prepare('SELECT status, last_error FROM autonomy_jobs WHERE idempotency_key = ?').get(idempotencyKey); + if (existing && ['complete', 'dead_letter'].includes(existing.status)) { + db.prepare(` + UPDATE autonomy_replay_runs + SET cursor_article_id=?, cursor_effective_at=?, last_error=?, updated_at=datetime('now') + WHERE id=? + `).run(article.id, article.effective_at, existing.status === 'dead_letter' + ? `Skipped dead-letter replay job for article ${article.id}: ${existing.last_error || 'unknown error'}` + : run.last_error, run.id); + cursorEffectiveAt = article.effective_at; + cursorArticleId = article.id; + continue; + } + + enqueueJob(db, { + jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: article.id, + idempotencyKey, + }); + return article; + } + throw new Error('replay scheduler skipped too many terminal jobs in one pass'); } async function runReplayWorker({ archivePath, intelligencePath, workerId = `replay-${os.hostname()}-${process.pid}`, pollMs = 15000 } = {}) {