From 7ff69ad99dd97c26f3ac59594301df9a06469bbc Mon Sep 17 00:00:00 2001 From: ImBenji Date: Mon, 17 Aug 2026 14:10:29 +0100 Subject: [PATCH] fix: use async postgres for autonomy http routes --- src/db/pgAsync.js | 24 ++++++++++++++++ src/routes/admin.js | 63 ++++++++++++++++++++++++++++++++++++++++++ src/routes/autonomy.js | 14 ++++++++++ 3 files changed, 101 insertions(+) create mode 100644 src/db/pgAsync.js diff --git a/src/db/pgAsync.js b/src/db/pgAsync.js new file mode 100644 index 0000000..5e18d59 --- /dev/null +++ b/src/db/pgAsync.js @@ -0,0 +1,24 @@ +const { Pool } = require('pg'); + +const pools = new Map(); + +function postgresUrl() { return process.env.DURIIN_POSTGRES_URL || process.env.DATABASE_URL; } + +function poolFor(schema = 'intelligence') { + const connectionString = postgresUrl(); + if (!connectionString) throw new Error('DURIIN_POSTGRES_URL is required'); + const key = `${connectionString}|${schema}`; + if (!pools.has(key)) { + pools.set(key, new Pool({ + connectionString, + max: Math.max(1, Number(process.env.POSTGRES_HTTP_POOL_SIZE) || 4), + options: `-c search_path=${schema},public`, + })); + } + return pools.get(key); +} + +async function all(schema, sql, params = []) { return (await poolFor(schema).query(sql, params)).rows; } +async function get(schema, sql, params = []) { return (await poolFor(schema).query(sql, params)).rows[0]; } + +module.exports = { poolFor, all, get }; diff --git a/src/routes/admin.js b/src/routes/admin.js index 92cb4de..1b29e18 100644 --- a/src/routes/admin.js +++ b/src/routes/admin.js @@ -6,6 +6,7 @@ const db = require('../db'); const config = require('../config'); const Database = require('better-sqlite3'); const { openRuntimeDb, isPostgresEnabled } = require('../db/runtime'); +const pg = require('../db/pgAsync'); let idb = null; let statsSummaryCache = null; @@ -152,6 +153,68 @@ async function adminRoutes(fastify) { // public status endpoint. fastify.get('/admin/api/autonomy/overview', async (request, reply) => { if (!checkAuth(request, reply)) return; + if (isPostgresEnabled()) { + const hasSchema = await pg.get('intelligence', "SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2", ['intelligence', 'autonomy_jobs']); + if (!hasSchema) return { enabled: false, reason: 'autonomy schema is not initialized' }; + const [jobs, predictionCounts, decisionCounts, proposalCounts, outcomeSummary, instruments, latestRows, latestOrders, account, calibration, replay] = await Promise.all([ + pg.all('intelligence', 'SELECT lane, status, COUNT(*) AS count FROM autonomy_jobs GROUP BY lane, status ORDER BY lane, status'), + pg.all('intelligence', 'SELECT status, COUNT(*) AS count FROM autonomy_predictions GROUP BY status'), + pg.all('intelligence', 'SELECT action, COUNT(*) AS count FROM autonomy_decisions GROUP BY action'), + pg.all('intelligence', 'SELECT status, COUNT(*) AS count FROM autonomy_proposals GROUP BY status'), + pg.get('intelligence', "SELECT COUNT(*) AS total, SUM(direction_correct) AS correct, AVG(excess_return) AS average_excess_return FROM autonomy_outcomes o JOIN autonomy_predictions p ON p.id = o.prediction_id WHERE p.origin = 'live'"), + pg.get('intelligence', 'SELECT COUNT(*) AS count FROM autonomy_instruments WHERE active=1 AND tradable=1'), + pg.all('intelligence', ` + SELECT p.id, p.instrument, p.direction, p.event_type, p.causal_channel, + p.horizon_days, p.information_cutoff, p.evidence_article_ids, + p.invalidation_condition, p.learning_eligible, p.status, p.created_at, + d.action, d.calibrated_probability, d.expected_excess_return, d.rationale, + o.excess_return, o.direction_correct + FROM autonomy_predictions p + LEFT JOIN autonomy_decisions d ON d.id = ( + SELECT MAX(d2.id) FROM autonomy_decisions d2 WHERE d2.prediction_id = p.id + ) + LEFT JOIN autonomy_outcomes o ON o.prediction_id = p.id + ORDER BY p.id DESC LIMIT 12 + `), + pg.all('intelligence', ` + SELECT oi.id, oi.client_order_id, oi.instrument, oi.side, oi.notional, + oi.status, oi.broker_order_id, oi.attempts, oi.last_error, + oi.created_at, oi.updated_at, d.action + FROM autonomy_order_intents oi + JOIN autonomy_decisions d ON d.id = oi.decision_id + ORDER BY oi.id DESC LIMIT 12 + `), + pg.get('intelligence', 'SELECT broker, equity, cash, buying_power, captured_at FROM autonomy_account_snapshots ORDER BY id DESC LIMIT 1'), + pg.all('intelligence', 'SELECT cohort_key, sample_size, effective_sample_size, directional_probability, expected_excess_return, lower_return, upper_return, created_at FROM autonomy_calibration_snapshots ORDER BY id DESC LIMIT 8'), + pg.get('intelligence', ` + SELECT r.id, r.status, r.watermark_at, r.cursor_article_id, r.cursor_effective_at, + r.processed_articles, r.updated_at, + SUM(CASE WHEN p.status = 'resolved' THEN 1 ELSE 0 END) AS resolved_predictions, + COUNT(p.id) AS predictions, + SUM(o.direction_correct) AS correct_predictions, + (SELECT COUNT(*) FROM autonomy_replay_evaluations e WHERE e.replay_run_id = r.id) AS evaluations + FROM autonomy_replay_runs r + LEFT JOIN autonomy_predictions p ON p.replay_run_id = r.id + LEFT JOIN autonomy_outcomes o ON o.prediction_id = p.id + GROUP BY r.id ORDER BY r.id DESC LIMIT 1 + `), + ]); + const latestPredictions = latestRows.map((row) => { + let evidenceCount = 0; + try { evidenceCount = JSON.parse(row.evidence_article_ids || '[]').length; } catch (_) {} + const { evidence_article_ids: ignored, ...safeRow } = row; + return { ...safeRow, evidence_count: evidenceCount }; + }); + return { + enabled: true, + mode: process.env.AUTONOMY_EXECUTION_MODE || 'shadow', + broker: { name: 'Alpaca Paper', configured: Boolean(process.env.ALPACA_PAPER_KEY_ID && process.env.ALPACA_PAPER_SECRET_KEY) }, + jobs, predictionCounts, decisionCounts, proposalCounts, outcomes: outcomeSummary, + allowlistedInstruments: instruments.count, latestPredictions, latestOrders, + account: account || null, calibration, replay: replay || null, + generatedAt: new Date().toISOString(), + }; + } const intelligenceDb = getIntelligenceDb(); if (!intelligenceDb) return { enabled: false, reason: 'intelligence database unavailable' }; const hasSchema = isPostgresEnabled() diff --git a/src/routes/autonomy.js b/src/routes/autonomy.js index 136b31e..c5ec142 100644 --- a/src/routes/autonomy.js +++ b/src/routes/autonomy.js @@ -1,5 +1,6 @@ const path = require('path'); const { openRuntimeDb, isPostgresEnabled } = require('../db/runtime'); +const pg = require('../db/pgAsync'); const intelligencePath = process.env.INTELLIGENCE_DB || path.resolve(process.cwd(), 'intelligence.sqlite'); const db = openRuntimeDb(intelligencePath, { schema: 'intelligence', readonly: true }); @@ -8,6 +9,19 @@ async function autonomyRoutes(fastify) { fastify.get('/health', async () => ({ ok: true, service: 'duriin-api' })); fastify.get('/autonomy/status', async () => { + if (isPostgresEnabled()) { + const exists = await pg.get('intelligence', "SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2", ['intelligence', 'autonomy_jobs']); + if (!exists) return { enabled: false, reason: 'autonomy schema is not initialized' }; + const [jobs, predictions, decisions, outcomes, legacy, instruments] = await Promise.all([ + pg.all('intelligence', 'SELECT lane, status, COUNT(*) AS count FROM autonomy_jobs GROUP BY lane, status ORDER BY lane, status'), + pg.all('intelligence', 'SELECT status, COUNT(*) AS count FROM autonomy_predictions GROUP BY status ORDER BY status'), + pg.all('intelligence', 'SELECT action, COUNT(*) AS count FROM autonomy_decisions GROUP BY action ORDER BY action'), + pg.get('intelligence', 'SELECT COUNT(*) AS total, SUM(direction_correct) AS correct, AVG(excess_return) AS average_excess_return FROM autonomy_outcomes'), + pg.get('intelligence', 'SELECT COUNT(*) AS count FROM autonomy_legacy_records'), + pg.get('intelligence', 'SELECT COUNT(*) AS count FROM autonomy_instruments WHERE active=1 AND tradable=1'), + ]); + return { enabled: true, jobs, predictions, decisions, outcomes, legacyRecords: legacy.count, allowlistedInstruments: instruments.count }; + } const exists = isPostgresEnabled() ? db.prepare("SELECT 1 FROM information_schema.tables WHERE table_schema = ? AND table_name = ?").get('intelligence', 'autonomy_jobs') : db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='autonomy_jobs'").get();