diff --git a/scripts/resync-postgres-schema.js b/scripts/resync-postgres-schema.js new file mode 100644 index 0000000..97b830e --- /dev/null +++ b/scripts/resync-postgres-schema.js @@ -0,0 +1,120 @@ +#!/usr/bin/env node +/* + * Re-sync one logical SQLite database into an existing PostgreSQL schema. + * Unlike the initial migration, this updates conflicting primary-key rows so + * mutable runtime state such as job status, replay cursors, and broker snapshots + * can be cut over safely. + */ +const Database = require('better-sqlite3'); +const { Pool } = require('pg'); + +const sqlitePath = process.env.SQLITE_PATH || process.env.SQLITE_INTELLIGENCE_PATH || '/data/intelligence.sqlite'; +const schema = process.env.POSTGRES_SCHEMA || 'intelligence'; +const connectionString = process.env.DATABASE_URL; +if (!connectionString) throw new Error('DATABASE_URL is required'); + +const batchSize = Math.max(1, Number(process.env.POSTGRES_SYNC_BATCH_SIZE) || 500); +const sqlite = new Database(sqlitePath, { readonly: true }); +const pool = new Pool({ connectionString, max: 2 }); + +function quote(name) { return `"${String(name).replaceAll('"', '""')}"`; } +function pgType(sqliteType) { + const type = String(sqliteType || '').toUpperCase(); + if (type.includes('INT')) return 'BIGINT'; + if (type.includes('REAL') || type.includes('FLOA') || type.includes('DOUB')) return 'DOUBLE PRECISION'; + if (type.includes('BLOB')) return 'BYTEA'; + return 'TEXT'; +} +function normalizeValue(value) { + if (typeof value === 'string' && value.includes('\u0000')) return value.replace(/\u0000/g, ''); + return value ?? null; +} + +function sourceTables() { + return sqlite.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all().map((row) => row.name); +} + +function columnsFor(table) { + return sqlite.prepare(`PRAGMA table_info(${quote(table)})`).all(); +} + +async function ensureTable(client, table, columns) { + const primary = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk).map((column) => quote(column.name)); + const definitions = columns.map((column) => `${quote(column.name)} ${pgType(column.type)}${column.notnull ? ' NOT NULL' : ''}`); + if (primary.length) definitions.push(`PRIMARY KEY (${primary.join(', ')})`); + await client.query(`CREATE TABLE IF NOT EXISTS ${quote(schema)}.${quote(table)} (${definitions.join(', ')})`); +} + +async function repairIdentity(client, table, columns) { + const id = columns.find((column) => column.pk === 1 && column.name === 'id' && /INT/i.test(column.type)); + if (!id) return; + const seq = `${schema}_${table}_id_seq`; + await client.query(`CREATE SEQUENCE IF NOT EXISTS ${quote(schema)}.${quote(seq)}`); + await client.query(`ALTER TABLE ${quote(schema)}.${quote(table)} ALTER COLUMN id SET DEFAULT nextval('${quote(schema)}.${quote(seq)}')`); + await client.query(`ALTER SEQUENCE ${quote(schema)}.${quote(seq)} OWNED BY ${quote(schema)}.${quote(table)}.id`); + await client.query(`SELECT setval('${quote(schema)}.${quote(seq)}', COALESCE((SELECT MAX(id) FROM ${quote(schema)}.${quote(table)}), 0) + 1, false)`); +} + +async function syncTable(client, table) { + const columns = columnsFor(table); + if (!columns.length) return; + await ensureTable(client, table, columns); + const names = columns.map((column) => column.name); + const pk = columns.filter((column) => column.pk).sort((a, b) => a.pk - b.pk).map((column) => column.name); + const sourceRows = sqlite.prepare(`SELECT COUNT(*) AS count FROM ${quote(table)}`).get().count; + if (!sourceRows) { + await repairIdentity(client, table, columns); + console.log(`[sync] ${schema}.${table}: empty`); + return; + } + if (!pk.length) { + const targetRows = await client.query(`SELECT COUNT(*)::bigint AS count FROM ${quote(schema)}.${quote(table)}`); + if (Number(targetRows.rows[0].count) === Number(sourceRows)) { + console.log(`[sync] ${schema}.${table}: no primary key, count matches (${sourceRows})`); + return; + } + throw new Error(`${schema}.${table} has no primary key and count differs; refusing ambiguous sync`); + } + const placeholders = names.map((_, index) => `$${index + 1}`).join(', '); + const updates = names.filter((name) => !pk.includes(name)).map((name) => `${quote(name)}=EXCLUDED.${quote(name)}`).join(', '); + const conflict = pk.map(quote).join(', '); + const sql = `INSERT INTO ${quote(schema)}.${quote(table)} (${names.map(quote).join(', ')}) VALUES (${placeholders}) ON CONFLICT (${conflict}) ${updates ? `DO UPDATE SET ${updates}` : 'DO NOTHING'}`; + const key = columns.find((column) => column.pk === 1 && /INT/i.test(column.type)); + let cursor = 0; + let copied = 0; + console.log(`[sync] ${schema}.${table}: ${sourceRows} source rows`); + while (true) { + const rows = key + ? sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} WHERE ${quote(key.name)} > ? ORDER BY ${quote(key.name)} LIMIT ?`).all(cursor, batchSize) + : sqlite.prepare(`SELECT ${names.map(quote).join(', ')} FROM ${quote(table)} LIMIT ? OFFSET ?`).all(batchSize, copied); + if (!rows.length) break; + await client.query('BEGIN'); + try { + for (const row of rows) await client.query(sql, names.map((name) => normalizeValue(row[name]))); + await client.query('COMMIT'); + } catch (error) { + await client.query('ROLLBACK'); + throw error; + } + copied += rows.length; + if (key) cursor = Number(rows[rows.length - 1][key.name]); + process.stdout.write(`\r[sync] ${schema}.${table}: ${copied}/${sourceRows}`); + } + await repairIdentity(client, table, columns); + console.log(`\r[sync] ${schema}.${table}: synced ${sourceRows}`); +} + +(async () => { + const client = await pool.connect(); + try { + await client.query(`CREATE SCHEMA IF NOT EXISTS ${quote(schema)}`); + for (const table of sourceTables()) await syncTable(client, table); + } finally { + client.release(); + await pool.end(); + sqlite.close(); + } +})().catch((error) => { + console.error('[sync] fatal:', error); + process.exit(1); +}); diff --git a/src/autonomy/schema.js b/src/autonomy/schema.js index eba4b91..5c1f141 100644 --- a/src/autonomy/schema.js +++ b/src/autonomy/schema.js @@ -1,6 +1,7 @@ const AUTONOMY_SCHEMA_VERSION = 2; function initAutonomySchema(db) { + if (db.dialect === 'postgres') return; db.exec(` CREATE TABLE IF NOT EXISTS autonomy_schema ( version INTEGER PRIMARY KEY, diff --git a/src/db/runtime.js b/src/db/runtime.js new file mode 100644 index 0000000..7bb7978 --- /dev/null +++ b/src/db/runtime.js @@ -0,0 +1,144 @@ +const { Pool } = require('pg'); +const deasync = require('deasync'); +const SqliteDatabase = require('better-sqlite3'); + +const pools = new Map(); + +function isPostgresEnabled() { + return String(process.env.DURIIN_DB_BACKEND || '').toLowerCase() === 'postgres' || Boolean(process.env.DATABASE_URL && process.env.DURIIN_USE_POSTGRES === 'true'); +} + +function poolFor(schema) { + const connectionString = process.env.DATABASE_URL; + if (!connectionString) throw new Error('DATABASE_URL is required for postgres runtime'); + const key = `${connectionString}|${schema}`; + if (!pools.has(key)) { + pools.set(key, new Pool({ + connectionString, + max: Math.max(1, Number(process.env.POSTGRES_RUNTIME_POOL_SIZE) || 4), + options: `-c search_path=${schema},public`, + })); + } + return pools.get(key); +} + +function querySync(pool, sql, params = []) { + let done = false; + let result; + let error; + pool.query(sql, params).then((value) => { result = value; done = true; }).catch((err) => { error = err; done = true; }); + deasync.loopWhile(() => !done); + if (error) throw error; + return result; +} + +function normalizeParams(params) { + if (params.length === 1 && params[0] && typeof params[0] === 'object' && !Array.isArray(params[0]) && !Buffer.isBuffer(params[0])) { + return params[0]; + } + return params.flat(); +} + +function rewritePlaceholders(sql, params) { + if (params && !Array.isArray(params)) { + const values = []; + const text = sql.replace(/@([A-Za-z_][A-Za-z0-9_]*)/g, (_, name) => { + values.push(params[name]); + return `$${values.length}`; + }); + return { sql: text, params: values }; + } + let index = 0; + return { sql: sql.replace(/\?/g, () => `$${++index}`), params: params || [] }; +} + +function rewriteSql(sql, params) { + let text = String(sql).trim(); + text = text.replace(/INSERT\s+OR\s+IGNORE\s+INTO/gi, 'INSERT INTO'); + text = text.replace(/INSERT\s+OR\s+REPLACE\s+INTO\s+autonomy_outcomes\s*\(([^)]+)\)\s*VALUES\s*\(([^)]+)\)/i, + (match, columns, values) => { + const names = columns.split(',').map((item) => item.trim().replace(/"/g, '')); + const updates = names.filter((name) => name !== 'prediction_id').map((name) => `${name}=EXCLUDED.${name}`).join(', '); + return `INSERT INTO autonomy_outcomes (${columns}) VALUES (${values}) ON CONFLICT (prediction_id) DO UPDATE SET ${updates}`; + }); + text = text.replace(/AUTOINCREMENT/gi, 'GENERATED BY DEFAULT AS IDENTITY'); + text = text.replace(/INTEGER\s+PRIMARY\s+KEY\s+GENERATED BY DEFAULT AS IDENTITY/gi, 'BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY'); + text = text.replace(/INTEGER\s+PRIMARY\s+KEY\s+AUTOINCREMENT/gi, 'BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY'); + text = text.replace(/datetime\('now',\s*\?\)/gi, 'CURRENT_TIMESTAMP + (?::interval)'); + text = text.replace(/datetime\('now',\s*'\+60 seconds'\)/gi, "CURRENT_TIMESTAMP + interval '60 seconds'"); + text = text.replace(/datetime\('now',\s*'([^']+)'\)/gi, "CURRENT_TIMESTAMP + interval '$1'"); + text = text.replace(/datetime\('now'\)/gi, 'CURRENT_TIMESTAMP'); + text = text.replace(/date\('now'\)/gi, 'CURRENT_DATE'); + text = text.replace(/datetime\(COALESCE\(([^)]+)\)\)/gi, 'COALESCE($1)::timestamp'); + text = text.replace(/datetime\((p\.information_cutoff),\s*'\+'\s*\|\|\s*(p\.horizon_days)\s*\|\|\s*' days'\)/gi, "($1::timestamp + ($2 || ' days')::interval)"); + text = text.replace(/datetime\(([^)]+)\)/gi, '($1)::timestamp'); + + const rewritten = rewritePlaceholders(text, params); + let finalSql = rewritten.sql; + if (/^INSERT\s+INTO\s+autonomy_jobs\b/i.test(finalSql) && !/ON\s+CONFLICT/i.test(finalSql)) finalSql += ' ON CONFLICT DO NOTHING'; + if (/^INSERT\s+INTO\s+autonomy_order_intents\b/i.test(finalSql) && !/ON\s+CONFLICT/i.test(finalSql)) finalSql += ' ON CONFLICT DO NOTHING'; + if (/^INSERT\s+INTO\s+autonomy_schema\b/i.test(finalSql) && !/ON\s+CONFLICT/i.test(finalSql)) finalSql += ' ON CONFLICT DO NOTHING'; + if (/^INSERT\s+INTO\s+autonomy_proposals\b/i.test(finalSql) && !/RETURNING\s+id/i.test(finalSql)) finalSql += ' RETURNING id'; + return { sql: finalSql, params: rewritten.params }; +} + +class PgCompatDb { + constructor(schema) { + this.schema = schema; + this.dialect = 'postgres'; + this.pool = poolFor(schema); + } + + pragma() { return undefined; } + + prepare(sql) { + const db = this; + return { + get(...rawParams) { + const { sql: text, params } = rewriteSql(sql, normalizeParams(rawParams)); + return querySync(db.pool, text, params).rows[0]; + }, + all(...rawParams) { + const { sql: text, params } = rewriteSql(sql, normalizeParams(rawParams)); + return querySync(db.pool, text, params).rows; + }, + run(...rawParams) { + const { sql: text, params } = rewriteSql(sql, normalizeParams(rawParams)); + const result = querySync(db.pool, text, params); + return { changes: result.rowCount || 0, lastInsertRowid: result.rows?.[0]?.id ?? null }; + }, + }; + } + + exec(sql) { + const statements = String(sql).split(';').map((statement) => statement.trim()).filter(Boolean); + for (const statement of statements) { + const { sql: text, params } = rewriteSql(statement, []); + querySync(this.pool, text, params); + } + } + + transaction(fn) { + const db = this; + const run = (...args) => { + querySync(db.pool, 'BEGIN', []); + try { + const result = fn(...args); + querySync(db.pool, 'COMMIT', []); + return result; + } catch (error) { + querySync(db.pool, 'ROLLBACK', []); + throw error; + } + }; + run.immediate = run; + return run; + } +} + +function openRuntimeDb(path, { schema = 'intelligence', readonly = false } = {}) { + if (isPostgresEnabled()) return new PgCompatDb(schema); + return new SqliteDatabase(path, readonly ? { readonly: true } : undefined); +} + +module.exports = { openRuntimeDb, isPostgresEnabled, PgCompatDb, rewriteSql }; diff --git a/src/routes/admin.js b/src/routes/admin.js index 7b3394b..92cb4de 100644 --- a/src/routes/admin.js +++ b/src/routes/admin.js @@ -5,6 +5,7 @@ const { Worker } = require('node:worker_threads'); const db = require('../db'); const config = require('../config'); const Database = require('better-sqlite3'); +const { openRuntimeDb, isPostgresEnabled } = require('../db/runtime'); let idb = null; let statsSummaryCache = null; @@ -46,9 +47,9 @@ function getIntelligenceDb() { ? (path.isAbsolute(config.intelligence_db) ? config.intelligence_db : path.resolve(configDir, config.intelligence_db)) : path.resolve(configDir, 'intelligence.sqlite')); - if (!fs.existsSync(rawPath)) return null; + if (!isPostgresEnabled() && !fs.existsSync(rawPath)) return null; - idb = new Database(rawPath); + idb = isPostgresEnabled() ? openRuntimeDb(rawPath, { schema: 'intelligence' }) : new Database(rawPath); return idb; } @@ -153,9 +154,9 @@ async function adminRoutes(fastify) { if (!checkAuth(request, reply)) return; const intelligenceDb = getIntelligenceDb(); if (!intelligenceDb) return { enabled: false, reason: 'intelligence database unavailable' }; - const hasSchema = intelligenceDb.prepare( - "SELECT 1 FROM sqlite_master WHERE type='table' AND name='autonomy_jobs'" - ).get(); + const hasSchema = isPostgresEnabled() + ? intelligenceDb.prepare("SELECT 1 FROM information_schema.tables WHERE table_schema = ? AND table_name = ?").get('intelligence', 'autonomy_jobs') + : intelligenceDb.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='autonomy_jobs'").get(); if (!hasSchema) return { enabled: false, reason: 'autonomy schema is not initialized' }; const jobs = intelligenceDb.prepare(` diff --git a/src/routes/autonomy.js b/src/routes/autonomy.js index b502913..136b31e 100644 --- a/src/routes/autonomy.js +++ b/src/routes/autonomy.js @@ -1,14 +1,16 @@ const path = require('path'); -const Database = require('better-sqlite3'); +const { openRuntimeDb, isPostgresEnabled } = require('../db/runtime'); const intelligencePath = process.env.INTELLIGENCE_DB || path.resolve(process.cwd(), 'intelligence.sqlite'); -const db = new Database(intelligencePath, { readonly: true }); +const db = openRuntimeDb(intelligencePath, { schema: 'intelligence', readonly: true }); async function autonomyRoutes(fastify) { fastify.get('/health', async () => ({ ok: true, service: 'duriin-api' })); fastify.get('/autonomy/status', async () => { - const exists = db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='autonomy_jobs'").get(); + 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(); if (!exists) return { enabled: false, reason: 'autonomy schema is not initialized' }; const jobs = db.prepare(` SELECT lane, status, COUNT(*) AS count diff --git a/workers/autonomyWorker.js b/workers/autonomyWorker.js index 3f23202..d154b7f 100644 --- a/workers/autonomyWorker.js +++ b/workers/autonomyWorker.js @@ -1,5 +1,5 @@ const os = require('os'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs'); @@ -104,8 +104,8 @@ function reconcileLiveBatch(archiveDb, intelligenceDb, batchSize = 250) { } async function runAutonomyWorker({ archivePath, intelligencePath, workerId = `autonomy-${os.hostname()}-${process.pid}`, pollMs = 1000 } = {}) { - const archiveDb = new Database(archivePath, { readonly: true }); - const intelligenceDb = new Database(intelligencePath); + const archiveDb = openRuntimeDb(archivePath, { schema: 'archive', readonly: true }); + const intelligenceDb = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); intelligenceDb.pragma('journal_mode = WAL'); intelligenceDb.pragma('busy_timeout = 5000'); initAutonomySchema(intelligenceDb); diff --git a/workers/calibrationWorker.js b/workers/calibrationWorker.js index 8d5eda9..c38d370 100644 --- a/workers/calibrationWorker.js +++ b/workers/calibrationWorker.js @@ -1,5 +1,5 @@ const os = require('os'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { calibrateOutcomes, cohortKey } = require('../src/autonomy/calibration'); const { decide } = require('../src/autonomy/policy'); @@ -144,7 +144,7 @@ function refreshReplayEvaluations(db) { } async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId = `calibration-${os.hostname()}-${process.pid}` } = {}) { - const db = new Database(intelligencePath); + const db = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); db.pragma('journal_mode = WAL'); db.pragma('busy_timeout = 5000'); initAutonomySchema(db); diff --git a/workers/coordinatorWorker.js b/workers/coordinatorWorker.js index 5737a6e..11549a1 100644 --- a/workers/coordinatorWorker.js +++ b/workers/coordinatorWorker.js @@ -1,7 +1,7 @@ const os = require('os'); const fs = require('fs'); const path = require('path'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs'); const { callCoordinator } = require('../src/autonomy/llm'); @@ -31,8 +31,8 @@ function buildPrompt(event, articles) { } async function runCoordinatorWorker({ archivePath, intelligencePath, workerId = `coordinator-${os.hostname()}-${process.pid}`, pollMs = 1000 } = {}) { - const archiveDb = new Database(archivePath, { readonly: true }); - const intelligenceDb = new Database(intelligencePath); + const archiveDb = openRuntimeDb(archivePath, { schema: 'archive', readonly: true }); + const intelligenceDb = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); intelligenceDb.pragma('journal_mode = WAL'); intelligenceDb.pragma('busy_timeout = 5000'); initAutonomySchema(intelligenceDb); diff --git a/workers/executionWorker.js b/workers/executionWorker.js index 0521744..da3740b 100644 --- a/workers/executionWorker.js +++ b/workers/executionWorker.js @@ -1,5 +1,5 @@ const os = require('os'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { createOrderIntent } = require('../src/autonomy/orderIntents'); const { createAlpacaPaperClient } = require('../src/brokers/alpacaPaper'); @@ -8,7 +8,7 @@ function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } async function runExecutionWorker({ intelligencePath, pollMs = 10000, mode = 'shadow', notional = 100, workerId = `execution-${os.hostname()}-${process.pid}` } = {}) { if (!['shadow', 'paper'].includes(mode)) throw new Error(`unsupported execution mode: ${mode}`); - const db = new Database(intelligencePath); + const db = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); db.pragma('journal_mode = WAL'); db.pragma('busy_timeout = 5000'); initAutonomySchema(db); diff --git a/workers/outcomeAutonomyWorker.js b/workers/outcomeAutonomyWorker.js index 05e7716..57afec5 100644 --- a/workers/outcomeAutonomyWorker.js +++ b/workers/outcomeAutonomyWorker.js @@ -1,6 +1,6 @@ const os = require('os'); const https = require('https'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { calculateOutcome } = require('../src/autonomy/outcomes'); @@ -33,7 +33,7 @@ async function history(symbol) { } async function resolveAutonomyOutcomes({ intelligencePath, workerId = `outcome-${os.hostname()}-${process.pid}`, pollMs = 60000 } = {}) { - const db = new Database(intelligencePath); + const db = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); db.pragma('journal_mode = WAL'); db.pragma('busy_timeout = 5000'); initAutonomySchema(db); @@ -56,9 +56,18 @@ async function resolveAutonomyOutcomes({ intelligencePath, workerId = `outcome-$ continue; } db.prepare(` - INSERT OR REPLACE INTO autonomy_outcomes + INSERT INTO autonomy_outcomes (prediction_id, price_0, price_horizon, benchmark_0, benchmark_horizon, excess_return, direction_correct, error_type) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(prediction_id) DO UPDATE SET + price_0=excluded.price_0, + price_horizon=excluded.price_horizon, + benchmark_0=excluded.benchmark_0, + benchmark_horizon=excluded.benchmark_horizon, + excess_return=excluded.excess_return, + direction_correct=excluded.direction_correct, + error_type=excluded.error_type, + evaluated_at=datetime('now') `).run(prediction.id, result.price0, result.priceHorizon, result.benchmark0, result.benchmarkHorizon, result.excessReturn, result.directionCorrect, result.directionCorrect ? null : 'direction_error'); db.prepare("UPDATE autonomy_predictions SET status = 'resolved' WHERE id = ?").run(prediction.id); diff --git a/workers/replayWorker.js b/workers/replayWorker.js index 5f038d7..4cd97b1 100644 --- a/workers/replayWorker.js +++ b/workers/replayWorker.js @@ -1,7 +1,7 @@ const os = require('os'); const fs = require('fs'); const path = require('path'); -const Database = require('better-sqlite3'); +const { openRuntimeDb } = require('../src/db/runtime'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs'); const { callCoordinator } = require('../src/autonomy/llm'); @@ -94,8 +94,8 @@ function scheduleNext(db, archiveDb, run) { } 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); + const archiveDb = openRuntimeDb(archivePath, { schema: 'archive', readonly: true }); + const db = openRuntimeDb(intelligencePath, { schema: 'intelligence' }); db.pragma('journal_mode = WAL'); db.pragma('busy_timeout = 5000'); initAutonomySchema(db);