feat: new ops console for the admin surface
The old admin was five separate html pages, so every navigation was a full reload and the operational picture was scattered across all of them. Worth saying: the api was never the problem, every endpoint answers in under 200ms. It felt slow because of the architecture, not the backend. This is a single page console. React and htm from a cdn, no bundler and no babel-in-the-browser, because a runtime transpiler on every load is exactly the slowness we are trying to get rid of. Hash routing, polling that keeps the last good payload on screen instead of flashing a spinner, and stale responses are dropped so a slow request cannot overwrite a newer one. /admin/api/ops/overview answers the whole dashboard in one call rather than making the browser fan out and stitch. It carries the things that actually matter and were not visible anywhere before: when live evidence matures, why a cohort does or does not clear the trade gate, which stage of the pipeline has gone quiet, and what is sitting in dead letters. Controls, all of which change production and all of which ask twice: - requeue dead letters, which only ever moves dead_letter back to pending - execution mode and a kill switch, now read from autonomy_settings on every poll instead of only from AUTONOMY_EXECUTION_MODE, so halting no longer needs a redeploy first - run the reaction analysis and read its output The d3 graph is framed rather than ported. It works, and rewriting it would risk something valuable for nothing the operator can see. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WnNxwxfXSbeNtjvtz5gayb
This commit is contained in:
ImBenji
@@ -0,0 +1,254 @@
|
||||
// Control plane for the operations dashboard.
|
||||
//
|
||||
// Kept apart from admin.js on purpose: everything in here either changes what the
|
||||
// autonomy stack does or is read by an operator while something is on fire, so it
|
||||
// wants to stay small enough to audit in one sitting.
|
||||
const { execFile } = require('child_process');
|
||||
const path = require('path');
|
||||
const { getExecutionControls, setExecutionMode, setKillSwitch, EXECUTION_MODES } = require('../autonomy/settings');
|
||||
|
||||
// Which of the pipeline stages we consider "recent enough to be alive". These are
|
||||
// generous, they exist to catch a stall not to police a few seconds of jitter.
|
||||
const STALE_AFTER_MINUTES = { ingest: 90, prediction: 180, outcome: 24 * 60 };
|
||||
|
||||
const ANALYSES = {
|
||||
reaction: {
|
||||
label: 'Reaction conditioning',
|
||||
script: 'scripts/analyze-reaction-conditioning.js',
|
||||
detail: 'Does the initial market reaction predict anything. Read only, a few minutes.',
|
||||
},
|
||||
};
|
||||
|
||||
function minutesSince(value) {
|
||||
if (!value) return null;
|
||||
const stamp = String(value).includes('T') ? String(value) : `${String(value).replace(' ', 'T')}Z`;
|
||||
const then = Date.parse(stamp);
|
||||
if (!Number.isFinite(then)) return null;
|
||||
return Math.max(0, (Date.now() - then) / 60000);
|
||||
}
|
||||
|
||||
function one(db, sql, params = []) {
|
||||
try {
|
||||
return db.prepare(sql).get(...params) || {};
|
||||
} catch (error) {
|
||||
console.error('[ops] query failed:', sql.trim().slice(0, 80), error.message);
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
function many(db, sql, params = []) {
|
||||
try {
|
||||
return db.prepare(sql).all(...params) || [];
|
||||
} catch (error) {
|
||||
console.error('[ops] query failed:', sql.trim().slice(0, 80), error.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
// One request for the whole dashboard. The old admin made the browser fire a
|
||||
// handful of sequential calls and stitch them together, which is most of why it
|
||||
// felt sluggish even though every individual endpoint was fast.
|
||||
function buildOverview(intel) {
|
||||
const predictions = one(intel, `
|
||||
SELECT COUNT(*) AS total,
|
||||
SUM(CASE WHEN origin='live' AND status='open' THEN 1 ELSE 0 END) AS live_open,
|
||||
SUM(CASE WHEN origin='live' AND status='resolved' THEN 1 ELSE 0 END) AS live_resolved,
|
||||
SUM(CASE WHEN origin IN ('historical','replay') AND status='resolved' THEN 1 ELSE 0 END) AS offline_resolved,
|
||||
SUM(CASE WHEN status='unresolvable' THEN 1 ELSE 0 END) AS unresolvable,
|
||||
MAX(created_at) AS latest
|
||||
FROM autonomy_predictions
|
||||
`);
|
||||
|
||||
const byOrigin = many(intel, `
|
||||
SELECT p.origin, COUNT(*) AS total,
|
||||
SUM(o.direction_correct) AS correct,
|
||||
AVG(o.excess_return) AS mean_excess
|
||||
FROM autonomy_predictions p JOIN autonomy_outcomes o ON o.prediction_id = p.id
|
||||
GROUP BY p.origin
|
||||
`);
|
||||
|
||||
// When live evidence actually arrives. This is the number that decides whether
|
||||
// anything can ever qualify, and nothing in the old UI showed it.
|
||||
const maturity = many(intel, `
|
||||
SELECT horizon_days, COUNT(*) AS n,
|
||||
MIN(date(information_cutoff, '+' || horizon_days || ' days')) AS first_matures
|
||||
FROM autonomy_predictions
|
||||
WHERE origin='live' AND status='open'
|
||||
GROUP BY horizon_days ORDER BY horizon_days
|
||||
`);
|
||||
|
||||
const decisions = many(intel, "SELECT action, COUNT(*) AS n FROM autonomy_decisions GROUP BY action");
|
||||
const intents = one(intel, 'SELECT COUNT(*) AS n FROM autonomy_order_intents');
|
||||
|
||||
const jobs = many(intel, `
|
||||
SELECT job_type, lane, status, COUNT(*) AS n, MAX(created_at) AS newest
|
||||
FROM autonomy_jobs GROUP BY job_type, lane, status
|
||||
`);
|
||||
const deadLetters = many(intel, `
|
||||
SELECT job_type, lane, COUNT(*) AS n, substr(MAX(last_error), 1, 160) AS sample_error
|
||||
FROM autonomy_jobs WHERE status='dead_letter' GROUP BY job_type, lane
|
||||
`);
|
||||
|
||||
let cohorts = [];
|
||||
try {
|
||||
cohorts = many(intel, `
|
||||
SELECT cohort_key, source, sample_size, distinct_instruments, top_instrument_share,
|
||||
directional_probability, expected_excess_return
|
||||
FROM autonomy_calibration_snapshots
|
||||
WHERE cohort_key LIKE 'v2|%'
|
||||
GROUP BY cohort_key, source
|
||||
HAVING MAX(created_at) = created_at
|
||||
ORDER BY sample_size DESC LIMIT 40
|
||||
`);
|
||||
} catch (error) {
|
||||
console.error('[ops] cohort snapshot read failed:', error.message);
|
||||
}
|
||||
|
||||
const outcomes = one(intel, 'SELECT COUNT(*) AS n, MAX(evaluated_at) AS latest FROM autonomy_outcomes');
|
||||
|
||||
return {
|
||||
generatedAt: new Date().toISOString(),
|
||||
predictions,
|
||||
byOrigin,
|
||||
maturity,
|
||||
decisions,
|
||||
orderIntents: intents.n || 0,
|
||||
outcomes,
|
||||
jobs,
|
||||
deadLetters,
|
||||
cohorts,
|
||||
freshness: {
|
||||
predictionMinutes: minutesSince(predictions.latest),
|
||||
outcomeMinutes: minutesSince(outcomes.latest),
|
||||
thresholds: STALE_AFTER_MINUTES,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function opsRoutes(fastify, options) {
|
||||
const { checkAuth, getIntelligenceDb, getArchiveDb } = options;
|
||||
|
||||
const withIntel = (reply) => {
|
||||
const intel = getIntelligenceDb();
|
||||
if (!intel) {
|
||||
reply.code(503).send({ error: 'intelligence database unavailable' });
|
||||
return null;
|
||||
}
|
||||
return intel;
|
||||
};
|
||||
|
||||
fastify.get('/admin/api/ops/overview', async (request, reply) => {
|
||||
if (!checkAuth(request, reply)) return;
|
||||
const intel = withIntel(reply);
|
||||
if (!intel) return;
|
||||
|
||||
const payload = buildOverview(intel);
|
||||
try {
|
||||
payload.controls = getExecutionControls(intel);
|
||||
} catch (error) {
|
||||
console.error('[ops] could not read execution controls:', error.message);
|
||||
payload.controls = null;
|
||||
}
|
||||
|
||||
// Archive counts are the one genuinely expensive thing here, so they are
|
||||
// cheap approximations rather than COUNT(*) over 2.2M rows on every poll.
|
||||
try {
|
||||
const archive = getArchiveDb ? getArchiveDb() : null;
|
||||
if (archive) {
|
||||
payload.archive = one(archive, `
|
||||
SELECT MAX(id) AS max_id, MAX(ingested_at) AS latest_ingest FROM articles
|
||||
`);
|
||||
payload.freshness.ingestMinutes = minutesSince(payload.archive.latest_ingest);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[ops] archive probe failed:', error.message);
|
||||
payload.archive = null;
|
||||
}
|
||||
return payload;
|
||||
});
|
||||
|
||||
fastify.get('/admin/api/ops/settings', async (request, reply) => {
|
||||
if (!checkAuth(request, reply)) return;
|
||||
const intel = withIntel(reply);
|
||||
if (!intel) return;
|
||||
return { controls: getExecutionControls(intel), modes: EXECUTION_MODES, analyses: ANALYSES };
|
||||
});
|
||||
|
||||
fastify.post('/admin/api/ops/settings', async (request, reply) => {
|
||||
if (!checkAuth(request, reply)) return;
|
||||
const intel = withIntel(reply);
|
||||
if (!intel) return;
|
||||
const { mode, killSwitch } = request.body || {};
|
||||
try {
|
||||
if (mode !== undefined) setExecutionMode(intel, String(mode), 'admin-ui');
|
||||
if (killSwitch !== undefined) setKillSwitch(intel, Boolean(killSwitch), 'admin-ui');
|
||||
} catch (error) {
|
||||
console.error('[ops] settings write rejected:', error.message);
|
||||
reply.code(400).send({ error: error.message });
|
||||
return;
|
||||
}
|
||||
const controls = getExecutionControls(intel);
|
||||
console.log(`[ops] execution controls now mode=${controls.mode} kill=${controls.killSwitch}`);
|
||||
return { controls };
|
||||
});
|
||||
|
||||
// Requeue is deliberately narrow: it only ever moves dead_letter back to pending,
|
||||
// it never deletes and never edits payloads, so the worst case is repeated work.
|
||||
fastify.post('/admin/api/ops/dead-letters/requeue', async (request, reply) => {
|
||||
if (!checkAuth(request, reply)) return;
|
||||
const intel = withIntel(reply);
|
||||
if (!intel) return;
|
||||
const { jobType, lane } = request.body || {};
|
||||
|
||||
const filters = ["status = 'dead_letter'"];
|
||||
const params = [];
|
||||
if (jobType) { filters.push('job_type = ?'); params.push(String(jobType)); }
|
||||
if (lane) { filters.push('lane = ?'); params.push(String(lane)); }
|
||||
const where = filters.join(' AND ');
|
||||
|
||||
try {
|
||||
const before = one(intel, `SELECT COUNT(*) AS n FROM autonomy_jobs WHERE ${where}`, params).n || 0;
|
||||
if (!before) return { requeued: 0, remaining: 0 };
|
||||
const result = intel.prepare(`
|
||||
UPDATE autonomy_jobs
|
||||
SET status='pending', attempts=0, available_at=datetime('now'),
|
||||
leased_by=NULL, lease_expires_at=NULL,
|
||||
last_error='requeued from the ops dashboard'
|
||||
WHERE ${where}
|
||||
`).run(...params);
|
||||
const remaining = one(intel, `SELECT COUNT(*) AS n FROM autonomy_jobs WHERE ${where}`, params).n || 0;
|
||||
console.log(`[ops] requeued ${result.changes} dead letters (jobType=${jobType || 'any'} lane=${lane || 'any'})`);
|
||||
return { requeued: result.changes, before, remaining };
|
||||
} catch (error) {
|
||||
console.error('[ops] requeue failed:', error.message, error.stack);
|
||||
reply.code(500).send({ error: error.message });
|
||||
}
|
||||
});
|
||||
|
||||
fastify.post('/admin/api/ops/analysis/:name', async (request, reply) => {
|
||||
if (!checkAuth(request, reply)) return;
|
||||
const spec = ANALYSES[request.params.name];
|
||||
if (!spec) {
|
||||
reply.code(404).send({ error: `unknown analysis: ${request.params.name}` });
|
||||
return;
|
||||
}
|
||||
const script = path.resolve(__dirname, '..', '..', spec.script);
|
||||
return new Promise((resolve) => {
|
||||
execFile('node', [script], { timeout: 15 * 60 * 1000, maxBuffer: 8 * 1024 * 1024 },
|
||||
(error, stdout, stderr) => {
|
||||
if (error) console.error(`[ops] analysis ${request.params.name} failed:`, error.message);
|
||||
resolve({
|
||||
analysis: request.params.name,
|
||||
label: spec.label,
|
||||
ok: !error,
|
||||
output: String(stdout || '').slice(-20000),
|
||||
error: error ? String(stderr || error.message).slice(-4000) : null,
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
module.exports = opsRoutes;
|
||||
module.exports.buildOverview = buildOverview;
|
||||
module.exports.ANALYSES = ANALYSES;
|
||||
Reference in New Issue
Block a user