feat: add isolated historical replay calibration

workers/calibrationWorker.js

@@ -11,7 +11,7 @@ function refreshCalibration(db, version = `cal-${Date.now()}`) {
     SELECT p.direction, p.event_type, p.horizon_days, o.*
     FROM autonomy_predictions p
     JOIN autonomy_outcomes o ON o.prediction_id = p.id
-    WHERE p.status = 'resolved' AND p.learning_eligible = 1
+    WHERE p.status = 'resolved' AND p.learning_eligible = 1 AND p.origin = 'live'
   `).all().reduce((map, row) => {
     const key = cohortKey({ direction: row.direction, eventType: row.event_type, horizonDays: row.horizon_days });
     if (!map.has(key)) map.set(key, []);
@@ -21,8 +21,8 @@ function refreshCalibration(db, version = `cal-${Date.now()}`) {
   const insert = db.prepare(`
     INSERT INTO autonomy_calibration_snapshots
       (cohort_key, sample_size, effective_sample_size, directional_probability,
-       expected_excess_return, lower_return, upper_return, parent_cohort_key, version)
-    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+       expected_excess_return, lower_return, upper_return, parent_cohort_key, version, source, replay_run_id)
+    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', NULL)
   `);
   const tx = db.transaction(() => {
     for (const [key, rows] of groups) {
@@ -40,7 +40,7 @@ function createDecisions(db, strategyVersion = 'autonomy-1') {
   const predictions = db.prepare(`
     SELECT p.* FROM autonomy_predictions p
     LEFT JOIN autonomy_decisions d ON d.prediction_id = p.id
-    WHERE d.prediction_id IS NULL AND p.status = 'open'
+    WHERE d.prediction_id IS NULL AND p.status = 'open' AND p.origin = 'live'
   `).all();
   const latest = db.prepare(`
     SELECT * FROM autonomy_calibration_snapshots
@@ -68,6 +68,44 @@ function createDecisions(db, strategyVersion = 'autonomy-1') {
   return created;
 }
 
+// Replay evaluations are walk-forward: each historical prediction is scored
+// against calibration data that had matured strictly before its cutoff.  They
+// are stored in their own ledger, never in autonomy_decisions.
+function refreshReplayEvaluations(db) {
+  const predictions = db.prepare(`
+    SELECT p.*, o.excess_return, o.direction_correct
+    FROM autonomy_predictions p JOIN autonomy_outcomes o ON o.prediction_id = p.id
+    LEFT JOIN autonomy_replay_evaluations e ON e.prediction_id = p.id
+    WHERE p.origin = 'replay' AND p.status = 'resolved' AND e.prediction_id IS NULL
+    ORDER BY datetime(p.information_cutoff), p.id LIMIT 200
+  `).all();
+  const prior = db.prepare(`
+    SELECT p.direction, p.event_type, p.horizon_days, o.*
+    FROM autonomy_predictions p JOIN autonomy_outcomes o ON o.prediction_id = p.id
+    WHERE p.origin = 'replay' AND p.status = 'resolved'
+      AND datetime(p.information_cutoff, '+' || p.horizon_days || ' days') < datetime(?)
+  `);
+  const insert = db.prepare(`
+    INSERT INTO autonomy_replay_evaluations
+      (prediction_id, replay_run_id, snapshot_cutoff, sample_size, action, calibrated_probability, expected_excess_return, rationale)
+    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+  `);
+  const tx = db.transaction(() => {
+    for (const prediction of predictions) {
+      const key = cohortKey({ direction: prediction.direction, eventType: prediction.event_type, horizonDays: prediction.horizon_days });
+      const rows = prior.all(prediction.information_cutoff).filter((row) =>
+        cohortKey({ direction: row.direction, eventType: row.event_type, horizonDays: row.horizon_days }) === key);
+      const calibration = rows.length ? calibrateOutcomes(rows) : null;
+      const decision = calibration ? decide({ ...calibration, direction: prediction.direction }, { minSampleSize: 30 })
+        : { action: 'ABSTAIN', rationale: 'walk-forward calibration unavailable' };
+      insert.run(prediction.id, prediction.replay_run_id, prediction.information_cutoff, rows.length, decision.action,
+        calibration?.directionalProbability || null, calibration?.expectedExcessReturn || null, decision.rationale);
+    }
+  });
+  tx();
+  return predictions.length;
+}
+
 async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId = `calibration-${os.hostname()}-${process.pid}` } = {}) {
   const db = new Database(intelligencePath);
   db.pragma('journal_mode = WAL');
@@ -78,7 +116,8 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId
       const state = db.prepare('SELECT COUNT(*) AS count, COALESCE(MAX(prediction_id), 0) AS max_id FROM autonomy_outcomes').get();
       const groups = refreshCalibration(db, `cal-${state.count}-${state.max_id}`);
       const decisions = createDecisions(db);
-      if (groups || decisions) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions}`);
+      const replayEvaluations = refreshReplayEvaluations(db);
+      if (groups || decisions || replayEvaluations) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions} replay_evaluations=${replayEvaluations}`);
     } catch (error) {
       console.error(`[${workerId}] calibration error:`, error.message);
     }
@@ -86,4 +125,4 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId
   }
 }
 
-module.exports = { refreshCalibration, createDecisions, runCalibrationWorker };
+module.exports = { refreshCalibration, createDecisions, refreshReplayEvaluations, runCalibrationWorker };

workers/executionWorker.js

@@ -51,8 +51,9 @@ async function runExecutionWorker({ intelligencePath, pollMs = 10000, mode = 'sh
     }
     const decisions = db.prepare(`
       SELECT d.id FROM autonomy_decisions d
+      JOIN autonomy_predictions p ON p.id = d.prediction_id
       LEFT JOIN autonomy_order_intents oi ON oi.decision_id = d.id
-      WHERE oi.id IS NULL AND d.action IN ('BUY', 'SELL')
+      WHERE oi.id IS NULL AND d.action IN ('BUY', 'SELL') AND p.origin = 'live'
       ORDER BY d.created_at ASC LIMIT 25
     `).all();
     for (const decision of decisions) {

workers/outcomeAutonomyWorker.js

@@ -20,7 +20,9 @@ function httpGet(url) {
 }
 
 async function history(symbol) {
-  const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}?range=10y&interval=1d`;
+  // GDELT backfills predate the normal rolling quote window.  Use an explicit
+  // point-in-time range so replay outcomes do not silently become unresolvable.
+  const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}?period1=946684800&period2=${Math.floor(Date.now() / 1000)}&interval=1d`;
   const body = JSON.parse(await httpGet(url));
   const result = body?.chart?.result?.[0];
   if (!result) return [];

workers/replay-entrypoint.js

@@ -0,0 +1,8 @@
+const path = require('path');
+const { runReplayWorker } = require('./replayWorker');
+
+runReplayWorker({
+  archivePath: process.env.DURIIN_DB || path.resolve('/data/archive.sqlite'),
+  intelligencePath: process.env.INTELLIGENCE_DB || path.resolve('/data/intelligence.sqlite'),
+  pollMs: Number(process.env.AUTONOMY_REPLAY_POLL_MS) || 15000,
+}).catch((error) => { console.error('[replay] fatal:', error); process.exit(1); });

workers/replayWorker.js

@@ -0,0 +1,113 @@
+const os = require('os');
+const fs = require('fs');
+const path = require('path');
+const Database = require('better-sqlite3');
+const { initAutonomySchema } = require('../src/autonomy/schema');
+const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs');
+const { callCoordinator } = require('../src/autonomy/llm');
+const { acceptProposal, recordRejectedProposal } = require('../src/autonomy/coordinator');
+
+function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
+
+function loadConfig() {
+  const configPath = path.resolve(process.env.DURIIN_CONFIG || path.join(__dirname, '..', 'config.json'));
+  const raw = JSON.parse(fs.readFileSync(configPath, 'utf8'));
+  require('dotenv').config({ path: path.resolve(path.dirname(configPath), '.env') });
+  raw.openRouter = { ...(raw.openRouter || {}) };
+  if (process.env.OPEN_ROUTER_API_KEY) raw.openRouter.apiKey = process.env.OPEN_ROUTER_API_KEY;
+  if (process.env.OPEN_ROUTER_LLM_MODEL) raw.openRouter.llmModel = process.env.OPEN_ROUTER_LLM_MODEL;
+  return raw;
+}
+
+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(', ')})` };
+}
+
+function replayPrompt(article) {
+  return `Historical evidence cutoff: ${article.effective_at}\n\n` +
+    `[Evidence 1] article_id=${article.id}\nTitle: ${article.title || ''}\n${String(article.content || article.description || '').slice(0, 6000)}\n\n` +
+    `Return JSON only in this shape:\n${JSON.stringify({ predictions: [{
+      instrument: 'NVDA', direction: 'positive|negative', event_type: 'stable_enum',
+      causal_channel: 'short description', horizon_days: 10, evidence_article_ids: [123],
+      invalidation_condition: 'condition',
+    }] }, null, 2)}\n\n` +
+    'Use only this dated evidence. Return an empty predictions array when there is no clear, tradable hypothesis. Never include probabilities, returns, confidence, position sizes, or actions.';
+}
+
+function activeRun(db, config) {
+  let run = db.prepare("SELECT * FROM autonomy_replay_runs WHERE status = 'running' ORDER BY id DESC LIMIT 1").get();
+  if (run) return run;
+  const watermarkDays = Math.max(1, Number(process.env.AUTONOMY_REPLAY_WATERMARK_DAYS) || 7);
+  const result = db.prepare(`
+    INSERT INTO autonomy_replay_runs (watermark_at, strategy_version, prompt_version, coordinator_model)
+    VALUES (datetime('now', ?), 'autonomy-1', 'replay-coordinator-1', ?)
+  `).run(`-${watermarkDays} days`, config.openRouter.llmModel || 'unknown');
+  return db.prepare('SELECT * FROM autonomy_replay_runs WHERE id = ?').get(result.lastInsertRowid);
+}
+
+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;
+}
+
+async function runReplayWorker({ archivePath, intelligencePath, workerId = `replay-${os.hostname()}-${process.pid}`, pollMs = 15000 } = {}) {
+  const archiveDb = new Database(archivePath, { readonly: true });
+  const db = new Database(intelligencePath);
+  db.pragma('journal_mode = WAL');
+  db.pragma('busy_timeout = 5000');
+  initAutonomySchema(db);
+  const config = loadConfig();
+  const dailyBudget = Math.max(1, Number(process.env.AUTONOMY_REPLAY_DAILY_BUDGET) || 100);
+  while (true) {
+    try {
+      const completedToday = db.prepare("SELECT COUNT(*) AS count FROM autonomy_jobs WHERE job_type='replay_article' AND status='complete' AND date(completed_at) = date('now')").get().count;
+      if (completedToday >= dailyBudget) { await sleep(Math.max(pollMs, 60000)); continue; }
+      const run = activeRun(db, config);
+      scheduleNext(db, archiveDb, run);
+      const job = leaseNextJob(db, workerId, 300, ['replay_article']);
+      if (!job) { await sleep(pollMs); continue; }
+      try {
+        const { effective } = articleTimeColumns(archiveDb);
+        const article = archiveDb.prepare(`SELECT id, title, description, content, ${effective} AS effective_at FROM articles WHERE id=?`).get(job.entity_id);
+        if (!article || !article.effective_at) throw new Error(`replay article ${job.entity_id} is unavailable`);
+        const raw = await callCoordinator(config, replayPrompt(article));
+        try {
+          acceptProposal(db, archiveDb, raw, {
+            informationCutoff: article.effective_at, model: config.openRouter.llmModel || 'unknown',
+            promptVersion: 'replay-coordinator-1', strategyVersion: 'autonomy-1', learningEligible: false,
+            origin: 'replay', replayRunId: run.id,
+          });
+        } catch (validationError) {
+          recordRejectedProposal(db, raw, { informationCutoff: article.effective_at, model: config.openRouter.llmModel || 'unknown', promptVersion: 'replay-coordinator-1' }, validationError.message);
+        }
+        db.prepare(`UPDATE autonomy_replay_runs SET cursor_article_id=?, cursor_effective_at=?, processed_articles=processed_articles+1, updated_at=datetime('now') WHERE id=?`)
+          .run(article.id, article.effective_at, run.id);
+        completeJob(db, job.id, workerId);
+      } catch (error) { failJob(db, job.id, workerId, error); }
+    } catch (error) { console.error(`[${workerId}] replay error:`, error.message); }
+    await sleep(pollMs);
+  }
+}
+
+module.exports = { articleTimeColumns, replayPrompt, scheduleNext, runReplayWorker };