openvibers/openvibe.ai Commits
AI phase 0 of the provider router (plan T6): every finished attempt rolls up into provider_stats_daily (requests, ok, errors, tokens, cost, a fixed-bucket latency histogram with p50/p95) and moves the route's placement_state EWMA (latency and error rate, alpha 0.2), in the same transaction as usage_daily, so a rollup failure rolls the accounting back; a failed run now records the provider error even when nothing was spent. statsFor() reads the rollup only (the price/latency page's source). Migration 0002 (expand). test/router-stats.test.js; both suites green (verified).
Changes
nnew file mode 100644@@@ -0,0 +1,34 @@++-- phase: expand++-- T6 provider router, phase 0 (brief-t6-ai.out.md §4 "Prices/latency page", decisions-t4-t6.md "T6"):++-- the daily per-provider/model rollup the public price/latency page will read (never the requests table),++-- and the persisted placement state the router will rebalance on. Both are written in the same transaction++-- as usage_daily, from server/quota.js account(); nothing reads them yet.++++CREATE TABLE provider_stats_daily (++ day date NOT NULL, -- UTC day++ provider text COLLATE "C" NOT NULL,++ model text COLLATE "C" NOT NULL DEFAULT '',++ requests bigint NOT NULL DEFAULT 0,++ ok bigint NOT NULL DEFAULT 0,++ errors bigint NOT NULL DEFAULT 0,++ latency_p50_ms integer,++ latency_p95_ms integer,++ cost_usd_total numeric NOT NULL DEFAULT 0,++ tokens_in bigint NOT NULL DEFAULT 0,++ tokens_out bigint NOT NULL DEFAULT 0,++ -- A fixed-bucket latency histogram (server/quota.js LATENCY_BUCKETS_MS): the streaming estimate the++ -- p50/p95 columns are read from. Length = length(bounds)+1, the last bucket the overflow (>= top++ -- bound). Keeping it per row means the percentiles never scan the requests table.++ latency_hist bigint[] NOT NULL DEFAULT '{}',++ PRIMARY KEY (day, provider, model)++);++++CREATE TABLE placement_state (++ route_key text COLLATE "C" PRIMARY KEY,++ current_provider text COLLATE "C", -- the last finished candidate on the route++ current_model text COLLATE "C",++ ewma_latency_ms double precision NOT NULL DEFAULT 0, -- alpha 0.2 per finished attempt (quota.js)++ ewma_error_rate double precision NOT NULL DEFAULT 0, -- alpha 0.2, errors / attempts++ quality double precision, -- no quality signal yet: always null++ updated_at timestamptz NOT NULL DEFAULT now()++);@@@ -9,6 +9,7 @@* 4 templates (versioned) 9 cache_entries* 5 workflows (versioned) 10 quotas + usage_counters* 11 audit_log++ * 12 provider_stats_daily + placement_state (T6 provider router)** Templates, workflows and routes are append-only per key: an edit inserts version n+1 and the* previous version stays readable, so every run can say exactly which versions produced it.@@@ -17,6 +17,35 @@ const { AiError, iso } = require('./util');const WINDOWS = { minute: 60, hour: 3600, day: 86400 };++/**++ * T6 provider router, phase 0 (nothing reads these yet). Every finished provider attempt is rolled up++ * into provider_stats_daily and per-route placement_state in the SAME transaction as usage_daily, so++ * the rollup can never drift from the accounting and a failed rollup write rolls the accounting back.++ *++ * Latency p50/p95 come from a fixed-bucket histogram (LATENCY_BUCKETS_MS = the inclusive upper bound of++ * each bucket; the last array slot is the overflow), never from a scan of the requests table. The estimate++ * is the bucket's upper bound — the smallest bucket whose cumulative count reaches the quantile.++ */++const EWMA_ALPHA = 0.2; // documented: 20 % of the new sample, 80 % of the history++const LATENCY_BUCKETS_MS = [50, 100, 200, 400, 800, 1600, 3200, 6400, 12800, 25600];++++function latencyBucket(ms) {++ const i = LATENCY_BUCKETS_MS.findIndex((b) => ms <= b);++ return i === -1 ? LATENCY_BUCKETS_MS.length : i; // -1 → the overflow bucket++}++function estimatePercentile(hist, q) {++ let total = 0;++ for (const n of hist) total += n;++ if (!total) return null;++ const target = Math.ceil(q * total);++ let cum = 0;++ for (let i = 0; i < hist.length; i++) {++ cum += hist[i];++ if (cum >= target) return i < LATENCY_BUCKETS_MS.length ? LATENCY_BUCKETS_MS[i] : LATENCY_BUCKETS_MS[LATENCY_BUCKETS_MS.length - 1];++ }++ return LATENCY_BUCKETS_MS[LATENCY_BUCKETS_MS.length - 1];++}++function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {}) {const nowSec = () => Math.floor(clock.now() / 1000);const windowStart = (w, t = nowSec()) => Math.floor(t / WINDOWS[w]) * WINDOWS[w];@@@ -101,15 +130,71 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})ON CONFLICT(day, requester, attribution, workflow_key, provider_key, model_key) DO UPDATE SET requests = usage_daily.requests + 1,tokens_in = usage_daily.tokens_in + excluded.tokens_in, tokens_out = usage_daily.tokens_out + excluded.tokens_out, tokens_cached = usage_daily.tokens_cached + excluded.tokens_cached, cost_usd = usage_daily.cost_usd + excluded.cost_usd`);-- /** Add real usage to the reserved windows and the daily usage table. */-- async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0 }) {++ // ── Provider rollup + placement state (T6 phase 0; nothing reads them yet) ──++ const ensStats = db.prepare('INSERT INTO provider_stats_daily (day, provider, model) VALUES (?, ?, ?) ON CONFLICT (day, provider, model) DO NOTHING');++ const getStats = db.prepare('SELECT latency_hist FROM provider_stats_daily WHERE day = ? AND provider = ? AND model = ? FOR UPDATE');++ const bumpStats = db.prepare(`UPDATE provider_stats_daily SET requests = requests + 1, ok = ok + ?, errors = errors + ?, cost_usd_total = cost_usd_total + ?,++ tokens_in = tokens_in + ?, tokens_out = tokens_out + ?, latency_hist = ?::bigint[], latency_p50_ms = ?, latency_p95_ms = ?++ WHERE day = ? AND provider = ? AND model = ?`);++ const putPlacement = db.prepare(`INSERT INTO placement_state (route_key, current_provider, current_model, ewma_latency_ms, ewma_error_rate, quality, updated_at)++ VALUES (?, ?, ?, ?, ?, NULL, ?)++ ON CONFLICT (route_key) DO UPDATE SET current_provider = excluded.current_provider, current_model = excluded.current_model,++ ewma_latency_ms = excluded.ewma_latency_ms * ? + placement_state.ewma_latency_ms * ?, ewma_error_rate = excluded.ewma_error_rate * ? + placement_state.ewma_error_rate * ?, updated_at = excluded.updated_at`);++++ /**++ * One provider_stats_daily upsert and one placement_state EWMA update per finished attempt++ * (ok | error | timeout | cancelled; skips never touched a provider). The first sample for a key++ * initialises its EWMA; later samples move it by EWMA_ALPHA. quality stays null (no signal yet).++ */++ async function recordStats(day, attempts) {++ for (const a of attempts || []) {++ if (!a.provider_key || a.status === 'skipped') continue;++ const model = a.model_key || '';++ const ok = a.status === 'ok' ? 1 : 0;++ const errors = ok ? 0 : 1;++ const latency = Number.isFinite(a.latency_ms) ? Math.max(0, Math.round(a.latency_ms)) : null;++ await ensStats.run(day, a.provider_key, model);++ const cur = await getStats.get(day, a.provider_key, model);++ const hist = Array.from({ length: LATENCY_BUCKETS_MS.length + 1 }, (_, i) => Number((cur && cur.latency_hist && cur.latency_hist[i]) || 0));++ if (latency !== null) hist[latencyBucket(latency)] += 1;++ await bumpStats.run(ok, errors, Number(a.cost_usd) || 0, Number(a.tokens_in) || 0, Number(a.tokens_out) || 0, hist,++ estimatePercentile(hist, 0.5), estimatePercentile(hist, 0.95), day, a.provider_key, model);++ if (a.route_key && latency !== null) {++ const t = iso(clock.now());++ await putPlacement.run(a.route_key, a.provider_key, a.model_key || null, latency, errors, t, EWMA_ALPHA, 1 - EWMA_ALPHA, EWMA_ALPHA, 1 - EWMA_ALPHA);++ }++ }++ }++++ /**++ * Add real usage to the reserved windows, the daily usage table and — in the same transaction — the++ * provider rollup and per-route placement state. `attempts` is the run's finished attempts (the same++ * entries logged to `requests`); `writeUsage` is false when a run spent nothing (stats are still kept).++ */++ async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true }) {const tokens = tokensIn + tokensOut;++ const day = iso(clock.now()).slice(0, 10);await db.tx(async () => {for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);-- await insDaily.run(iso(clock.now()).slice(0, 10), `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);++ if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);++ await recordStats(day, attempts);});}++ /**++ * Per provider/model totals over the last `days` UTC days, from provider_stats_daily only — never the++ * requests table. latency_p50_ms/latency_p95_ms are the worst daily estimate in the window (a day's++ * histogram is not summable in SQL without a custom aggregate); the router and the price/latency page++ * read this. No route yet (T6 phase 0).++ */++ async function statsFor({ days = 7 } = {}) {++ const from = new Date(clock.now() - Math.max(1, days) * 86400000).toISOString().slice(0, 10);++ return await db.prepare(`SELECT provider, model, SUM(requests)::bigint AS requests, SUM(ok)::bigint AS ok, SUM(errors)::bigint AS errors,++ SUM(cost_usd_total)::double precision AS cost_usd_total, SUM(tokens_in)::bigint AS tokens_in, SUM(tokens_out)::bigint AS tokens_out,++ MAX(latency_p50_ms) AS latency_p50_ms, MAX(latency_p95_ms) AS latency_p95_ms++ FROM provider_stats_daily WHERE day >= ? GROUP BY provider, model ORDER BY provider, model`).all(from);++ }++// ── Admin ──async function list() { return await db.prepare('SELECT * FROM quotas ORDER BY scope_type, scope_id, "window"').all(); }async function upsert(q, { actor = 'system', origin = 'admin', trace = null } = {}) {@@@ -169,7 +254,7 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})return { max_cost_usd: Math.min(...rows.map((r) => r.max_cost_usd)), spent, window: 'day' };}-- return { reserve, account, list, upsert, usage, counters, paidBudget, WINDOWS };++ return { reserve, account, statsFor, list, upsert, usage, counters, paidBudget, WINDOWS };}module.exports = { createQuotas, WINDOWS };@@@ -274,8 +274,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst opts = parseJson(row.options, {});let seq = 0;let attempts = 0;++ const logged = []; // every attempt entry, for the provider rollup in quotas.account()const logRequest = async (e) => {if (e.status !== 'skipped') attempts++;++ logged.push(e);await insReq.run({run_id: id, seq: seq++, operation: e.operation, provider_key: e.provider_key, model_key: e.model_key || null, route_key: e.route_key || null, route_version: e.route_version || null,status: e.status, skip_reason: e.skip_reason || null, fallback: e.fallback || 0, prompt_hash: e.prompt_hash || null, input_hash: row.input_hash, output_hash: e.output_hash || null,@@@ -308,7 +310,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowreturn d;});if (!done.changes) return; // cancelled while finishing-- if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });++ if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost, attempts: logged });if (r.fallbackUsed) await registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {@@@ -327,7 +329,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (f.changes) await events.runChanged(await getRow(id));});const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);-- if (reserved && spent && (spent.ti || spent.tout)) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });++ // Account the spent tokens/cost as before, but always roll up the finished attempts (the failed++ // ones included) into provider_stats_daily/placement_state: a run that answered nothing still++ // tells the router a provider was unhealthy.++ if (reserved && logged.some((e) => e.status !== 'skipped')) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c, attempts: logged, writeUsage: Boolean(spent.ti || spent.tout) });} finally {inflight.delete(id);}nnew file mode 100644@@@ -0,0 +1,121 @@++'use strict';++// T6 provider router, phase 0 (data layer): every finished attempt rolls into provider_stats_daily++// (counts, tokens, cost, a fixed-bucket p50/p95) and per-route placement_state (EWMA latency/error rate,++// alpha 0.2) in the SAME transaction as usage_daily. A failing rollup write rolls the accounting back++// with it; statsFor reads provider_stats_daily only, never the requests table.++//++// Requires migrations/0002_provider_router_stats.sql. Until it is applied this file prints a skip line.++const assert = require('assert');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++++const t = suite('router-stats');++const tok = token('live', ALL);++let h;++let seam;++++async function provider(body) {++ const r = await request(h.base, 'POST', '/api/v1/providers', { tok, body: { kind: 'http', auth_mode: 'none', capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000, ...body } });++ assert.strictEqual(r.status, 201, r.text);++}++async function route(primary, fallbacks) {++ const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok, body: { primary: { provider: primary }, fallbacks: fallbacks.map((p) => ({ provider: p })), timeout_ms: 5000 } });++ assert.strictEqual(r.status, 201, r.text);++}++const generate = (prompt) => request(h.base, 'POST', '/api/v1/generate', { tok, body: { prompt, options: { cache: false } } });++const statsRow = (p, m = '') => h.db.prepare('SELECT * FROM provider_stats_daily WHERE provider = ? AND model = ?').get(p, m);++const placementRow = (k) => h.db.prepare('SELECT * FROM placement_state WHERE route_key = ?').get(k);++const usageRows = async () => (await h.db.prepare("SELECT COUNT(*)::bigint AS n FROM usage_daily WHERE requester = 'service:live'").get()).n;++++t.test('boot with a working seam provider, a dead one, and the route pinned', async () => {++ seam = await seamServer(() => ({ text: 'a real answer', model: 'seam-model', usage: { input: 12, output: 4 } }));++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '5' } });++ const table = await h.db.prepare("SELECT to_regclass('provider_stats_daily') AS t").get();++ if (!table.t) {++ process.stdout.write('router-stats: skipped (migrations/0002_provider_router_stats.sql is not applied)\n');++ process.exit(0);++ }++ await provider({ key: 'seam', base_url: seam.url });++ await provider({ key: 'dead', base_url: 'http://127.0.0.1:1/x' });++ await route('seam', []);++});++++t.test('a successful run rolls up one ok attempt: counts, tokens, cost, a percentile estimate', async () => {++ const r = await generate('hello');++ assert.strictEqual(r.status, 201, r.text);++ assert.strictEqual(r.body.run.status, 'succeeded');++ assert.strictEqual(r.body.run.provenance.provider, 'seam');++ const row = await statsRow('seam', 'seam-model');++ assert.ok(row, 'a provider_stats_daily row for the answered model');++ assert.strictEqual(row.requests, 1);++ assert.strictEqual(row.ok, 1);++ assert.strictEqual(row.errors, 0);++ assert.strictEqual(row.tokens_in, 12);++ assert.strictEqual(row.tokens_out, 4);++ assert.ok(Number(row.cost_usd_total) > 0, 'the attempt cost accumulated');++ assert.ok(row.latency_p50_ms >= 0 && row.latency_p95_ms >= row.latency_p50_ms, 'p50/p95 from the histogram');++ assert.strictEqual(row.latency_hist.map(Number).reduce((a, b) => a + b, 0), 1, 'the histogram counted this one attempt');++ const p = await placementRow('default.chat');++ assert.ok(p, 'a placement_state row for the route');++ assert.strictEqual(p.current_provider, 'seam');++ assert.strictEqual(p.current_model, 'seam-model');++ assert.strictEqual(p.ewma_error_rate, 0);++ assert.ok(p.ewma_latency_ms > 0, 'the first sample initialises the EWMA');++ assert.strictEqual(p.quality, null);++});++++t.test('a failed run counts an error and moves the route EWMA by alpha 0.2', async () => {++ const before = await placementRow('default.chat');++ await route('dead', []);++ const r = await generate('nobody home');++ assert.strictEqual(r.body.run.status, 'failed');++ assert.strictEqual(r.body.run.error.code, 'provider.unavailable');++ const row = await statsRow('dead', '');++ assert.ok(row, 'the dead provider has a rollup row');++ assert.strictEqual(row.requests, 1);++ assert.strictEqual(row.ok, 0);++ assert.strictEqual(row.errors, 1);++ const attempt = await h.db.prepare("SELECT latency_ms FROM requests WHERE provider_key = 'dead' AND status = 'error' ORDER BY id DESC LIMIT 1").get();++ assert.ok(attempt && attempt.latency_ms != null, 'the failed attempt recorded a latency');++ const p = await placementRow('default.chat');++ assert.strictEqual(p.current_provider, 'dead');++ assert.ok(Math.abs(p.ewma_error_rate - 0.2) < 1e-9, `first error after a clean sample: got ${p.ewma_error_rate}`);++ assert.ok(Math.abs(p.ewma_latency_ms - (0.2 * attempt.latency_ms + 0.8 * before.ewma_latency_ms)) < 1e-6, 'latency EWMA');++ assert.ok(Math.abs(p.ewma_latency_ms - before.ewma_latency_ms) > 1e-9, 'the latency EWMA moved');++});++++t.test('a failing rollup write rolls the accounting back with it', async () => {++ await route('seam', []);++ const seamBefore = await statsRow('seam', 'seam-model');++ const usageBefore = await usageRows();++ // Mirror test/events.test.js: interpose on the write the rollup goes through and make it reject, then++ // assert the accounting in the same transaction rolled back with it. (No DDL: the pg runtime role may++ // not create a trigger.)++ const realQuery = h.db.query.bind(h.db);++ h.db.query = async (text, values) => {++ if (typeof text === 'string' && text.includes('provider_stats_daily')) throw new Error('rollup write failed');++ return await realQuery(text, values);++ };++ try {++ const r = await generate('rollback please');++ assert.strictEqual(r.body.run.status, 'succeeded', 'the provider still answered');++ } finally {++ h.db.query = realQuery;++ }++ const seamAfter = await statsRow('seam', 'seam-model');++ assert.strictEqual(await usageRows(), usageBefore, 'usage_daily rolled back with the failed rollup');++ assert.strictEqual(seamAfter.requests, seamBefore.requests, 'the rollup row did not move');++});++++t.test('statsFor totals provider_stats_daily only, never the requests table', async () => {++ await h.db.query('DELETE FROM requests');++ assert.strictEqual((await h.db.prepare('SELECT COUNT(*)::bigint AS n FROM requests').get()).n, 0, 'requests is empty');++ const rows = await h.quotas.statsFor({ days: 7 });++ const seamRow = rows.find((r) => r.provider === 'seam' && r.model === 'seam-model');++ const deadRow = rows.find((r) => r.provider === 'dead');++ assert.ok(seamRow && seamRow.requests >= 1 && seamRow.ok >= 1 && seamRow.errors === 0, 'seam totals from the rollup');++ assert.ok(deadRow && deadRow.errors >= 1, 'dead totals from the rollup');++ assert.ok(Number(seamRow.cost_usd_total) > 0, 'cost totalled');++});++++t.test('shutdown', async () => { await h.stop(); await seam.close(); });++t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,34 @@
@@ -0,0 +1,34 @@
+-- phase: expand
+-- T6 provider router, phase 0 (brief-t6-ai.out.md §4 "Prices/latency page", decisions-t4-t6.md "T6"):
+-- the daily per-provider/model rollup the public price/latency page will read (never the requests table),
+-- and the persisted placement state the router will rebalance on. Both are written in the same transaction
+-- as usage_daily, from server/quota.js account(); nothing reads them yet.
+
+CREATE TABLE provider_stats_daily (
+ day date NOT NULL, -- UTC day
+ provider text COLLATE "C" NOT NULL,
+ model text COLLATE "C" NOT NULL DEFAULT '',
+ requests bigint NOT NULL DEFAULT 0,
+ ok bigint NOT NULL DEFAULT 0,
+ errors bigint NOT NULL DEFAULT 0,
+ latency_p50_ms integer,
+ latency_p95_ms integer,
+ cost_usd_total numeric NOT NULL DEFAULT 0,
+ tokens_in bigint NOT NULL DEFAULT 0,
+ tokens_out bigint NOT NULL DEFAULT 0,
+ -- A fixed-bucket latency histogram (server/quota.js LATENCY_BUCKETS_MS): the streaming estimate the
+ -- p50/p95 columns are read from. Length = length(bounds)+1, the last bucket the overflow (>= top
+ -- bound). Keeping it per row means the percentiles never scan the requests table.
+ latency_hist bigint[] NOT NULL DEFAULT '{}',
+ PRIMARY KEY (day, provider, model)
+);
+
+CREATE TABLE placement_state (
+ route_key text COLLATE "C" PRIMARY KEY,
+ current_provider text COLLATE "C", -- the last finished candidate on the route
+ current_model text COLLATE "C",
+ ewma_latency_ms double precision NOT NULL DEFAULT 0, -- alpha 0.2 per finished attempt (quota.js)
+ ewma_error_rate double precision NOT NULL DEFAULT 0, -- alpha 0.2, errors / attempts
+ quality double precision, -- no quality signal yet: always null
+ updated_at timestamptz NOT NULL DEFAULT now()
+);
@@ -9,6 +9,7 @@
@@ -9,6 +9,7 @@
* 4 templates (versioned) 9 cache_entries
* 4 templates (versioned) 9 cache_entries
* 5 workflows (versioned) 10 quotas + usage_counters
* 5 workflows (versioned) 10 quotas + usage_counters
* 11 audit_log
* 11 audit_log
+ * 12 provider_stats_daily + placement_state (T6 provider router)
*
*
* Templates, workflows and routes are append-only per key: an edit inserts version n+1 and the
* Templates, workflows and routes are append-only per key: an edit inserts version n+1 and the
* previous version stays readable, so every run can say exactly which versions produced it.
* previous version stays readable, so every run can say exactly which versions produced it.
@@ -17,6 +17,35 @@ const { AiError, iso } = require('./util');
@@ -17,6 +17,35 @@ const { AiError, iso } = require('./util');
const WINDOWS = { minute: 60, hour: 3600, day: 86400 };
const WINDOWS = { minute: 60, hour: 3600, day: 86400 };
+/**
+ * T6 provider router, phase 0 (nothing reads these yet). Every finished provider attempt is rolled up
+ * into provider_stats_daily and per-route placement_state in the SAME transaction as usage_daily, so
+ * the rollup can never drift from the accounting and a failed rollup write rolls the accounting back.
+ *
+ * Latency p50/p95 come from a fixed-bucket histogram (LATENCY_BUCKETS_MS = the inclusive upper bound of
+ * each bucket; the last array slot is the overflow), never from a scan of the requests table. The estimate
+ * is the bucket's upper bound — the smallest bucket whose cumulative count reaches the quantile.
+ */
+const EWMA_ALPHA = 0.2; // documented: 20 % of the new sample, 80 % of the history
+const LATENCY_BUCKETS_MS = [50, 100, 200, 400, 800, 1600, 3200, 6400, 12800, 25600];
+
+function latencyBucket(ms) {
+ const i = LATENCY_BUCKETS_MS.findIndex((b) => ms <= b);
+ return i === -1 ? LATENCY_BUCKETS_MS.length : i; // -1 → the overflow bucket
+}
+function estimatePercentile(hist, q) {
+ let total = 0;
+ for (const n of hist) total += n;
+ if (!total) return null;
+ const target = Math.ceil(q * total);
+ let cum = 0;
+ for (let i = 0; i < hist.length; i++) {
+ cum += hist[i];
+ if (cum >= target) return i < LATENCY_BUCKETS_MS.length ? LATENCY_BUCKETS_MS[i] : LATENCY_BUCKETS_MS[LATENCY_BUCKETS_MS.length - 1];
+ }
+ return LATENCY_BUCKETS_MS[LATENCY_BUCKETS_MS.length - 1];
+}
+
function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {}) {
function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {}) {
const nowSec = () => Math.floor(clock.now() / 1000);
const nowSec = () => Math.floor(clock.now() / 1000);
const windowStart = (w, t = nowSec()) => Math.floor(t / WINDOWS[w]) * WINDOWS[w];
const windowStart = (w, t = nowSec()) => Math.floor(t / WINDOWS[w]) * WINDOWS[w];
@@ -101,15 +130,71 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})
@@ -101,15 +130,71 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})
ON CONFLICT(day, requester, attribution, workflow_key, provider_key, model_key) DO UPDATE SET requests = usage_daily.requests + 1,
ON CONFLICT(day, requester, attribution, workflow_key, provider_key, model_key) DO UPDATE SET requests = usage_daily.requests + 1,
tokens_in = usage_daily.tokens_in + excluded.tokens_in, tokens_out = usage_daily.tokens_out + excluded.tokens_out, tokens_cached = usage_daily.tokens_cached + excluded.tokens_cached, cost_usd = usage_daily.cost_usd + excluded.cost_usd`);
tokens_in = usage_daily.tokens_in + excluded.tokens_in, tokens_out = usage_daily.tokens_out + excluded.tokens_out, tokens_cached = usage_daily.tokens_cached + excluded.tokens_cached, cost_usd = usage_daily.cost_usd + excluded.cost_usd`);
- /** Add real usage to the reserved windows and the daily usage table. */
- async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0 }) {
+ // ── Provider rollup + placement state (T6 phase 0; nothing reads them yet) ──
+ const ensStats = db.prepare('INSERT INTO provider_stats_daily (day, provider, model) VALUES (?, ?, ?) ON CONFLICT (day, provider, model) DO NOTHING');
+ const getStats = db.prepare('SELECT latency_hist FROM provider_stats_daily WHERE day = ? AND provider = ? AND model = ? FOR UPDATE');
+ const bumpStats = db.prepare(`UPDATE provider_stats_daily SET requests = requests + 1, ok = ok + ?, errors = errors + ?, cost_usd_total = cost_usd_total + ?,
+ tokens_in = tokens_in + ?, tokens_out = tokens_out + ?, latency_hist = ?::bigint[], latency_p50_ms = ?, latency_p95_ms = ?
+ WHERE day = ? AND provider = ? AND model = ?`);
+ const putPlacement = db.prepare(`INSERT INTO placement_state (route_key, current_provider, current_model, ewma_latency_ms, ewma_error_rate, quality, updated_at)
+ VALUES (?, ?, ?, ?, ?, NULL, ?)
+ ON CONFLICT (route_key) DO UPDATE SET current_provider = excluded.current_provider, current_model = excluded.current_model,
+ ewma_latency_ms = excluded.ewma_latency_ms * ? + placement_state.ewma_latency_ms * ?, ewma_error_rate = excluded.ewma_error_rate * ? + placement_state.ewma_error_rate * ?, updated_at = excluded.updated_at`);
+
+ /**
+ * One provider_stats_daily upsert and one placement_state EWMA update per finished attempt
+ * (ok | error | timeout | cancelled; skips never touched a provider). The first sample for a key
+ * initialises its EWMA; later samples move it by EWMA_ALPHA. quality stays null (no signal yet).
+ */
+ async function recordStats(day, attempts) {
+ for (const a of attempts || []) {
+ if (!a.provider_key || a.status === 'skipped') continue;
+ const model = a.model_key || '';
+ const ok = a.status === 'ok' ? 1 : 0;
+ const errors = ok ? 0 : 1;
+ const latency = Number.isFinite(a.latency_ms) ? Math.max(0, Math.round(a.latency_ms)) : null;
+ await ensStats.run(day, a.provider_key, model);
+ const cur = await getStats.get(day, a.provider_key, model);
+ const hist = Array.from({ length: LATENCY_BUCKETS_MS.length + 1 }, (_, i) => Number((cur && cur.latency_hist && cur.latency_hist[i]) || 0));
+ if (latency !== null) hist[latencyBucket(latency)] += 1;
+ await bumpStats.run(ok, errors, Number(a.cost_usd) || 0, Number(a.tokens_in) || 0, Number(a.tokens_out) || 0, hist,
+ estimatePercentile(hist, 0.5), estimatePercentile(hist, 0.95), day, a.provider_key, model);
+ if (a.route_key && latency !== null) {
+ const t = iso(clock.now());
+ await putPlacement.run(a.route_key, a.provider_key, a.model_key || null, latency, errors, t, EWMA_ALPHA, 1 - EWMA_ALPHA, EWMA_ALPHA, 1 - EWMA_ALPHA);
+ }
+ }
+ }
+
+ /**
+ * Add real usage to the reserved windows, the daily usage table and — in the same transaction — the
+ * provider rollup and per-route placement state. `attempts` is the run's finished attempts (the same
+ * entries logged to `requests`); `writeUsage` is false when a run spent nothing (stats are still kept).
+ */
+ async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true }) {
const tokens = tokensIn + tokensOut;
const tokens = tokensIn + tokensOut;
+ const day = iso(clock.now()).slice(0, 10);
await db.tx(async () => {
await db.tx(async () => {
for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);
for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);
- await insDaily.run(iso(clock.now()).slice(0, 10), `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);
+ if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);
+ await recordStats(day, attempts);
});
});
}
}
+ /**
+ * Per provider/model totals over the last `days` UTC days, from provider_stats_daily only — never the
+ * requests table. latency_p50_ms/latency_p95_ms are the worst daily estimate in the window (a day's
+ * histogram is not summable in SQL without a custom aggregate); the router and the price/latency page
+ * read this. No route yet (T6 phase 0).
+ */
+ async function statsFor({ days = 7 } = {}) {
+ const from = new Date(clock.now() - Math.max(1, days) * 86400000).toISOString().slice(0, 10);
+ return await db.prepare(`SELECT provider, model, SUM(requests)::bigint AS requests, SUM(ok)::bigint AS ok, SUM(errors)::bigint AS errors,
+ SUM(cost_usd_total)::double precision AS cost_usd_total, SUM(tokens_in)::bigint AS tokens_in, SUM(tokens_out)::bigint AS tokens_out,
+ MAX(latency_p50_ms) AS latency_p50_ms, MAX(latency_p95_ms) AS latency_p95_ms
+ FROM provider_stats_daily WHERE day >= ? GROUP BY provider, model ORDER BY provider, model`).all(from);
+ }
+
// ── Admin ──
// ── Admin ──
async function list() { return await db.prepare('SELECT * FROM quotas ORDER BY scope_type, scope_id, "window"').all(); }
async function list() { return await db.prepare('SELECT * FROM quotas ORDER BY scope_type, scope_id, "window"').all(); }
async function upsert(q, { actor = 'system', origin = 'admin', trace = null } = {}) {
async function upsert(q, { actor = 'system', origin = 'admin', trace = null } = {}) {
@@ -169,7 +254,7 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})
@@ -169,7 +254,7 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})
return { max_cost_usd: Math.min(...rows.map((r) => r.max_cost_usd)), spent, window: 'day' };
return { max_cost_usd: Math.min(...rows.map((r) => r.max_cost_usd)), spent, window: 'day' };
}
}
- return { reserve, account, list, upsert, usage, counters, paidBudget, WINDOWS };
+ return { reserve, account, statsFor, list, upsert, usage, counters, paidBudget, WINDOWS };
}
}
module.exports = { createQuotas, WINDOWS };
module.exports = { createQuotas, WINDOWS };
@@ -274,8 +274,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -274,8 +274,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const opts = parseJson(row.options, {});
const opts = parseJson(row.options, {});
let seq = 0;
let seq = 0;
let attempts = 0;
let attempts = 0;
+ const logged = []; // every attempt entry, for the provider rollup in quotas.account()
const logRequest = async (e) => {
const logRequest = async (e) => {
if (e.status !== 'skipped') attempts++;
if (e.status !== 'skipped') attempts++;
+ logged.push(e);
await insReq.run({
await insReq.run({
run_id: id, seq: seq++, operation: e.operation, provider_key: e.provider_key, model_key: e.model_key || null, route_key: e.route_key || null, route_version: e.route_version || null,
run_id: id, seq: seq++, operation: e.operation, provider_key: e.provider_key, model_key: e.model_key || null, route_key: e.route_key || null, route_version: e.route_version || null,
status: e.status, skip_reason: e.skip_reason || null, fallback: e.fallback || 0, prompt_hash: e.prompt_hash || null, input_hash: row.input_hash, output_hash: e.output_hash || null,
status: e.status, skip_reason: e.skip_reason || null, fallback: e.fallback || 0, prompt_hash: e.prompt_hash || null, input_hash: row.input_hash, output_hash: e.output_hash || null,
@@ -308,7 +310,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -308,7 +310,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
return d;
return d;
});
});
if (!done.changes) return; // cancelled while finishing
if (!done.changes) return; // cancelled while finishing
- if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
+ if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost, attempts: logged });
if (r.fallbackUsed) await registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
if (r.fallbackUsed) await registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
@@ -327,7 +329,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -327,7 +329,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (f.changes) await events.runChanged(await getRow(id));
if (f.changes) await events.runChanged(await getRow(id));
});
});
const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
- if (reserved && spent && (spent.ti || spent.tout)) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });
+ // Account the spent tokens/cost as before, but always roll up the finished attempts (the failed
+ // ones included) into provider_stats_daily/placement_state: a run that answered nothing still
+ // tells the router a provider was unhealthy.
+ if (reserved && logged.some((e) => e.status !== 'skipped')) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c, attempts: logged, writeUsage: Boolean(spent.ti || spent.tout) });
} finally {
} finally {
inflight.delete(id);
inflight.delete(id);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,121 @@
@@ -0,0 +1,121 @@
+'use strict';
+// T6 provider router, phase 0 (data layer): every finished attempt rolls into provider_stats_daily
+// (counts, tokens, cost, a fixed-bucket p50/p95) and per-route placement_state (EWMA latency/error rate,
+// alpha 0.2) in the SAME transaction as usage_daily. A failing rollup write rolls the accounting back
+// with it; statsFor reads provider_stats_daily only, never the requests table.
+//
+// Requires migrations/0002_provider_router_stats.sql. Until it is applied this file prints a skip line.
+const assert = require('assert');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+
+const t = suite('router-stats');
+const tok = token('live', ALL);
+let h;
+let seam;
+
+async function provider(body) {
+ const r = await request(h.base, 'POST', '/api/v1/providers', { tok, body: { kind: 'http', auth_mode: 'none', capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000, ...body } });
+ assert.strictEqual(r.status, 201, r.text);
+}
+async function route(primary, fallbacks) {
+ const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok, body: { primary: { provider: primary }, fallbacks: fallbacks.map((p) => ({ provider: p })), timeout_ms: 5000 } });
+ assert.strictEqual(r.status, 201, r.text);
+}
+const generate = (prompt) => request(h.base, 'POST', '/api/v1/generate', { tok, body: { prompt, options: { cache: false } } });
+const statsRow = (p, m = '') => h.db.prepare('SELECT * FROM provider_stats_daily WHERE provider = ? AND model = ?').get(p, m);
+const placementRow = (k) => h.db.prepare('SELECT * FROM placement_state WHERE route_key = ?').get(k);
+const usageRows = async () => (await h.db.prepare("SELECT COUNT(*)::bigint AS n FROM usage_daily WHERE requester = 'service:live'").get()).n;
+
+t.test('boot with a working seam provider, a dead one, and the route pinned', async () => {
+ seam = await seamServer(() => ({ text: 'a real answer', model: 'seam-model', usage: { input: 12, output: 4 } }));
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '5' } });
+ const table = await h.db.prepare("SELECT to_regclass('provider_stats_daily') AS t").get();
+ if (!table.t) {
+ process.stdout.write('router-stats: skipped (migrations/0002_provider_router_stats.sql is not applied)\n');
+ process.exit(0);
+ }
+ await provider({ key: 'seam', base_url: seam.url });
+ await provider({ key: 'dead', base_url: 'http://127.0.0.1:1/x' });
+ await route('seam', []);
+});
+
+t.test('a successful run rolls up one ok attempt: counts, tokens, cost, a percentile estimate', async () => {
+ const r = await generate('hello');
+ assert.strictEqual(r.status, 201, r.text);
+ assert.strictEqual(r.body.run.status, 'succeeded');
+ assert.strictEqual(r.body.run.provenance.provider, 'seam');
+ const row = await statsRow('seam', 'seam-model');
+ assert.ok(row, 'a provider_stats_daily row for the answered model');
+ assert.strictEqual(row.requests, 1);
+ assert.strictEqual(row.ok, 1);
+ assert.strictEqual(row.errors, 0);
+ assert.strictEqual(row.tokens_in, 12);
+ assert.strictEqual(row.tokens_out, 4);
+ assert.ok(Number(row.cost_usd_total) > 0, 'the attempt cost accumulated');
+ assert.ok(row.latency_p50_ms >= 0 && row.latency_p95_ms >= row.latency_p50_ms, 'p50/p95 from the histogram');
+ assert.strictEqual(row.latency_hist.map(Number).reduce((a, b) => a + b, 0), 1, 'the histogram counted this one attempt');
+ const p = await placementRow('default.chat');
+ assert.ok(p, 'a placement_state row for the route');
+ assert.strictEqual(p.current_provider, 'seam');
+ assert.strictEqual(p.current_model, 'seam-model');
+ assert.strictEqual(p.ewma_error_rate, 0);
+ assert.ok(p.ewma_latency_ms > 0, 'the first sample initialises the EWMA');
+ assert.strictEqual(p.quality, null);
+});
+
+t.test('a failed run counts an error and moves the route EWMA by alpha 0.2', async () => {
+ const before = await placementRow('default.chat');
+ await route('dead', []);
+ const r = await generate('nobody home');
+ assert.strictEqual(r.body.run.status, 'failed');
+ assert.strictEqual(r.body.run.error.code, 'provider.unavailable');
+ const row = await statsRow('dead', '');
+ assert.ok(row, 'the dead provider has a rollup row');
+ assert.strictEqual(row.requests, 1);
+ assert.strictEqual(row.ok, 0);
+ assert.strictEqual(row.errors, 1);
+ const attempt = await h.db.prepare("SELECT latency_ms FROM requests WHERE provider_key = 'dead' AND status = 'error' ORDER BY id DESC LIMIT 1").get();
+ assert.ok(attempt && attempt.latency_ms != null, 'the failed attempt recorded a latency');
+ const p = await placementRow('default.chat');
+ assert.strictEqual(p.current_provider, 'dead');
+ assert.ok(Math.abs(p.ewma_error_rate - 0.2) < 1e-9, `first error after a clean sample: got ${p.ewma_error_rate}`);
+ assert.ok(Math.abs(p.ewma_latency_ms - (0.2 * attempt.latency_ms + 0.8 * before.ewma_latency_ms)) < 1e-6, 'latency EWMA');
+ assert.ok(Math.abs(p.ewma_latency_ms - before.ewma_latency_ms) > 1e-9, 'the latency EWMA moved');
+});
+
+t.test('a failing rollup write rolls the accounting back with it', async () => {
+ await route('seam', []);
+ const seamBefore = await statsRow('seam', 'seam-model');
+ const usageBefore = await usageRows();
+ // Mirror test/events.test.js: interpose on the write the rollup goes through and make it reject, then
+ // assert the accounting in the same transaction rolled back with it. (No DDL: the pg runtime role may
+ // not create a trigger.)
+ const realQuery = h.db.query.bind(h.db);
+ h.db.query = async (text, values) => {
+ if (typeof text === 'string' && text.includes('provider_stats_daily')) throw new Error('rollup write failed');
+ return await realQuery(text, values);
+ };
+ try {
+ const r = await generate('rollback please');
+ assert.strictEqual(r.body.run.status, 'succeeded', 'the provider still answered');
+ } finally {
+ h.db.query = realQuery;
+ }
+ const seamAfter = await statsRow('seam', 'seam-model');
+ assert.strictEqual(await usageRows(), usageBefore, 'usage_daily rolled back with the failed rollup');
+ assert.strictEqual(seamAfter.requests, seamBefore.requests, 'the rollup row did not move');
+});
+
+t.test('statsFor totals provider_stats_daily only, never the requests table', async () => {
+ await h.db.query('DELETE FROM requests');
+ assert.strictEqual((await h.db.prepare('SELECT COUNT(*)::bigint AS n FROM requests').get()).n, 0, 'requests is empty');
+ const rows = await h.quotas.statsFor({ days: 7 });
+ const seamRow = rows.find((r) => r.provider === 'seam' && r.model === 'seam-model');
+ const deadRow = rows.find((r) => r.provider === 'dead');
+ assert.ok(seamRow && seamRow.requests >= 1 && seamRow.ok >= 1 && seamRow.errors === 0, 'seam totals from the rollup');
+ assert.ok(deadRow && deadRow.errors >= 1, 'dead totals from the rollup');
+ assert.ok(Number(seamRow.cost_usd_total) > 0, 'cost totalled');
+});
+
+t.test('shutdown', async () => { await h.stop(); await seam.close(); });
+t.run();