openvibers/openvibe.ai Commits

Preserve daily measured model metrics from run receipts (#436)

OP OpenVibers committed 3 days ago · 8384f79ba0 · 8 files, +299 −0

* Preserve daily measured model metrics from run receipts

* Document the measured metrics status read in OpenAPI

* Merge remote-tracking branch 'origin/main' into feat/measured-metrics-rollup-1110

Changes

ddiff --git a/docs/measured-metrics.md b/docs/measured-metrics.md
nnew file mode 100644
iindex 0000000..5431539
---- /dev/null
++++ b/docs/measured-metrics.md
@@@ -0,0 +1,27 @@
++# Measured Exchange metrics
++
++`exchange_metrics_daily` preserves complete UTC days from AI run receipts. The due row
++in `exchange_metrics_job` survives restarts; a transaction locks it before a worker
++backfills retained receipts and refreshes the last two completed dates. The worker
++wakes each minute, first five seconds after boot, and normally runs once a day after
++00:05 UTC. It is disabled in tests and drills, and stops before the database closes.
++
++Each row is a model, public provider, fixed task kind and day with at least twenty
++real completed runs. Synthetic, cached, private/BYOK and hidden providers are excluded.
++Uncensored offers are excluded from public analytics. No person, site, prompt, answer,
++private template/workflow name or receipt identifier is stored in the rollup.
++
++Success rate counts succeeded / (succeeded + failed) receipts attributed to the
++answering or final attempted offer. It is run health, not individual fallback attempt
++health. Latency percentiles use successful receipts' total latency. First token uses
++recorded streamed TTFT; decode throughput is output tokens / (latency - TTFT) seconds.
++Each timing metric needs twenty valid observations of its own. Unknown is null.
++Mean cost per run includes failures with recorded cost; cost per thousand output
++tokens includes input cost. Samples, units and dates travel with every future read.
++
++Backfill covers complete dates still inside receipt retention (normally 29 days).
++Longer history accumulates going forward; it is never fabricated from expired runs.
++Models use the catalogue's canonical identity and longest matching registered revision.
++
++For Fable: part 1 adds storage only. The next read surface will expose the daily
++series, sample counts and null states for charts without needing raw receipts.
ddiff --git a/migrations/0068_measured_metrics.sql b/migrations/0068_measured_metrics.sql
nnew file mode 100644
iindex 0000000..3de9f9d
---- /dev/null
++++ b/migrations/0068_measured_metrics.sql
@@@ -0,0 +1,21 @@
++-- phase: expand
++-- Durable, public-safe statistics from completed run receipts. No caller, prompt,
++-- answer, template, site or run id is copied into this table. Sparse cells stay private.
++CREATE TABLE IF NOT EXISTS exchange_metrics_daily (
++ day date NOT NULL,
++ model text NOT NULL,
++ provider text NOT NULL,
++ task_kind text NOT NULL,
++ runs integer NOT NULL CHECK (runs >= 20),
++ metrics jsonb NOT NULL,
++ computed_at timestamptz NOT NULL,
++ PRIMARY KEY (day, model, provider, task_kind)
++);
++CREATE INDEX IF NOT EXISTS exchange_metrics_model_day ON exchange_metrics_daily (model, day);
++CREATE TABLE IF NOT EXISTS exchange_metrics_job (
++ id integer PRIMARY KEY CHECK (id = 1),
++ next_due timestamptz NOT NULL,
++ completed_through date,
++ completed_at timestamptz
++);
++INSERT INTO exchange_metrics_job (id, next_due) VALUES (1, '1970-01-01') ON CONFLICT DO NOTHING;
ddiff --git a/server/api/exchange.js b/server/api/exchange.js
iindex acd7ab1..54c9a24 100644
---- a/server/api/exchange.js
++++ b/server/api/exchange.js
@@@ -59,6 +59,7 @@ function exchangeRouter({ db, registry, pool, quotas = null, env = process.env,
const market = require('../exchange/market').createMarket({ db, registry, catalog: all, config, clock, fidelity: fidelitySource });
const usageByWindow = new Map();
const r = express.Router();
++ r.use(require('./measured-metrics').measuredMetricsRouter({ db, clock, log }));
const publicCache = (seconds) => (req, res, next) => { res.set('Cache-Control', `public, max-age=${seconds}`); next(); };
ddiff --git a/server/api/measured-metrics.js b/server/api/measured-metrics.js
nnew file mode 100644
iindex 0000000..2e3631b
---- /dev/null
++++ b/server/api/measured-metrics.js
@@@ -0,0 +1,19 @@
++'use strict';
++const express = require('express');
++const { sendError } = require('../util');
++function measuredMetricsRouter({ db, clock = { now: () => Date.now() }, log = console }) {
++ const r = express.Router();
++ let hit = null;
++ r.get('/api/v1/exchange/metrics/status', async (req, res) => {
++ try {
++ if (!hit || hit.until <= clock.now()) {
++ const state = await db.prepare('SELECT completed_through, completed_at FROM exchange_metrics_job WHERE id = 1').get();
++ const cells = await db.prepare('SELECT COUNT(*)::integer AS published_cells, MIN(day) AS first_day, MAX(day) AS last_day FROM exchange_metrics_daily WHERE runs >= 20').get();
++ hit = { until: clock.now() + 60000, body: { schema: 'exchange.metrics@1', minimum_samples: 20, ...state, ...cells } };
++ }
++ res.set('Cache-Control', 'public, max-age=60').json(hit.body);
++ } catch (err) { sendError(res, err, req.ov, log); }
++ });
++ return r;
++}
++module.exports = { measuredMetricsRouter };
ddiff --git a/server/exchange/metrics-rollup.js b/server/exchange/metrics-rollup.js
nnew file mode 100644
iindex 0000000..9fc3f46
---- /dev/null
++++ b/server/exchange/metrics-rollup.js
@@@ -0,0 +1,139 @@
++'use strict';
++
++const { canonicalOf, slugOf } = require('./catalog');
++const { modelMatches } = require('./observed');
++const { upstreamOf } = require('../providers/upstream');
++const DAY = 86400000;
++const MIN_SAMPLES = 20;
++const TASK_KINDS = ['chat', 'code', 'design', 'search', 'vision', 'speech', 'embedding', 'image', 'video', 'other'];
++const iso = (ms) => new Date(ms).toISOString();
++const dayOf = (ms) => iso(ms).slice(0, 10);
++const number = (value) => value == null || value === '' || !Number.isFinite(Number(value)) || Number(value) < 0 ? null : Number(value);
++
++// Only fixed public categories leave the run store; never a private workflow or template name.
++function taskKind(kind, workflow) {
++ if (TASK_KINDS.includes(kind)) return kind;
++ const name = String(kind || workflow || '').toLowerCase();
++ for (const [pattern, task] of [[/^(?:codes?|agent)\./, 'code'], [/^design\./, 'design'], [/^(?:search|research)\./, 'search'],
++ [/^(?:vision|image\.analy)/, 'vision'], [/^(?:speech|audio|ai\.transcrib)/, 'speech'], [/^(?:image|ai\.image)/, 'image'],
++ [/^(?:video|ai\.video)/, 'video'], [/^(?:embed|ai\.embed)/, 'embedding'], [/^(?:chat|text|ai\.(?:chat|complete|generate|summarize|classify|extract|enrich))/, 'chat']]) {
++ if (pattern.test(name)) return task;
++ }
++ return 'other';
++}
++
++function quantile(values, p) {
++ if (values.length < MIN_SAMPLES) return null;
++ const sorted = [...values].sort((a, b) => a - b);
++ const i = (sorted.length - 1) * p;
++ return sorted[Math.floor(i)] + (sorted[Math.ceil(i)] - sorted[Math.floor(i)]) * (i % 1);
++}
++
++function publicOffers(providers, models) {
++ const byKey = new Map(providers.filter((p) => p.kind !== 'stub' && !p.metadata?.synthetic && !p.metadata?.hidden
++ && !['CUSTOMER_BYOK', 'RESTRICTED'].includes(p.provider_access) && !/^(?:byo:|community:)/.test(p.key)).map((p) => [p.key, p]));
++ return models.flatMap((m) => {
++ const provider = byKey.get(m.provider_key);
++ const upstream = provider && upstreamOf(provider);
++ if (!upstream || m.metadata?.hidden || m.metadata?.uncensored || /uncensored|abliterat/i.test(m.model_key)) return [];
++ return [{ key: provider.key, provider: upstream.key, keyModel: m.model_key, model: slugOf(canonicalOf(m, { key: upstream.key })) }];
++ }).sort((a, b) => b.keyModel.length - a.keyModel.length);
++}
++
++function aggregate(rows, offers) {
++ const groups = new Map();
++ const matches = new Map();
++ for (const row of rows) {
++ const lookup = `${row.provider}\0${row.model}`;
++ if (!matches.has(lookup)) matches.set(lookup, offers.find((o) => o.key === row.provider && modelMatches(o.keyModel, row.model)) || null);
++ const offer = matches.get(lookup);
++ if (!offer) continue;
++ const task = taskKind(row.kind, row.workflow);
++ const key = `${offer.model}\0${offer.provider}\0${task}`;
++ let cell = groups.get(key);
++ if (!cell) {
++ cell = { model: offer.model, provider: offer.provider, task_kind: task, runs: 0, successes: 0, latency: [], ttft: [], tps: [], cost: [], tokens: [] };
++ groups.set(key, cell);
++ }
++ cell.runs++;
++ const success = row.status === 'succeeded';
++ if (success) cell.successes++;
++ const latency = number(row.latency_ms), ttft = number(row.ttft_ms), out = number(row.tokens_out), cost = number(row.cost_usd);
++ if (success && latency != null) cell.latency.push(latency);
++ if (success && ttft != null && (latency == null || ttft <= latency)) cell.ttft.push(ttft);
++ if (success && latency > ttft && ttft != null && out > 0) cell.tps.push(out * 1000 / (latency - ttft));
++ if (cost != null) cell.cost.push(cost);
++ if (cost != null && out != null) cell.tokens.push({ cost, out });
++ }
++ return [...groups.values()].filter((c) => c.runs >= MIN_SAMPLES).map((c) => {
++ const totalCost = c.cost.reduce((s, v) => s + v, 0);
++ const tokens = c.tokens.reduce((s, v) => s + v.out, 0);
++ return { model: c.model, provider: c.provider, task_kind: c.task_kind, runs: c.runs, metrics: {
++ success_rate: c.successes / c.runs,
++ latency_p50_ms: quantile(c.latency, 0.5), latency_p95_ms: quantile(c.latency, 0.95),
++ ttft_p50_ms: quantile(c.ttft, 0.5), tokens_per_second_p50: quantile(c.tps, 0.5),
++ cost_per_run_usd: c.cost.length >= MIN_SAMPLES ? totalCost / c.cost.length : null,
++ cost_per_1k_output_tokens_usd: c.tokens.length >= MIN_SAMPLES && tokens > 0 ? c.tokens.reduce((s, v) => s + v.cost, 0) * 1000 / tokens : null,
++ samples: { latency: c.latency.length >= MIN_SAMPLES ? c.latency.length : null,
++ ttft: c.ttft.length >= MIN_SAMPLES ? c.ttft.length : null, throughput: c.tps.length >= MIN_SAMPLES ? c.tps.length : null,
++ cost: c.cost.length >= MIN_SAMPLES ? c.cost.length : null },
++ } };
++ });
++}
++
++function createMetricsRollup({ db, registry, clock = { now: () => Date.now() }, log = console, retentionDays = 30, enabled = true }) {
++ let timer = null, pending = null, stopped = true;
++ const receipts = db.prepare(`SELECT c.body->'offer'->>'provider' AS provider, c.body->'offer'->>'model' AS model,
++ c.body->>'kind' AS kind, c.body->>'workflow' AS workflow, r.status,
++ c.body->>'latency_ms' AS latency_ms, c.body->>'ttft_ms' AS ttft_ms,
++ c.body->'usage'->>'tokens_out' AS tokens_out, c.body->>'cost_usd' AS cost_usd
++ FROM ai_receipts c JOIN runs r ON r.id = c.run_id
++ WHERE c.created_at >= ?::timestamptz AND c.created_at < ?::timestamptz
++ AND r.synthetic = 0 AND r.status IN ('succeeded', 'failed') AND r.cached_from IS NULL`);
++ async function tick() {
++ if (!enabled) return { skipped: 'disabled' };
++ const now = clock.now(), today = Date.parse(`${dayOf(now)}T00:00:00Z`);
++ return db.tx(async () => {
++ const state = await db.prepare('SELECT * FROM exchange_metrics_job WHERE id = 1 AND next_due <= ?::timestamptz FOR UPDATE SKIP LOCKED').get(iso(now));
++ if (!state) return { skipped: 'not_due' };
++ const providers = await registry.listProviders();
++ if (providers.some((p) => p.metadata?.drill || p.metadata?.fail)) return { skipped: 'drill' };
++ const offers = publicOffers(providers, await registry.listModels());
++ // Resume after downtime; never overwrite durable dates whose receipts were already pruned.
++ const retainedFrom = today - Math.max(1, Math.min(90, retentionDays - 1)) * DAY;
++ const savedDay = state.completed_through instanceof Date ? dayOf(state.completed_through.getTime()) : String(state.completed_through || '').slice(0, 10);
++ const through = savedDay ? Date.parse(savedDay + 'T00:00:00Z') : null;
++ const from = through == null ? retainedFrom : Math.max(retainedFrom, through - DAY);
++ let days = 0, cells = 0;
++ for (let start = from; start < today; start += DAY) {
++ const day = dayOf(start);
++ const values = aggregate(await receipts.all(iso(start), iso(start + DAY)), offers);
++ await db.prepare('DELETE FROM exchange_metrics_daily WHERE day = ?::date').run(day);
++ for (const cell of values) {
++ await db.prepare('INSERT INTO exchange_metrics_daily (day, model, provider, task_kind, runs, metrics, computed_at) VALUES (?::date, ?, ?, ?, ?, ?::jsonb, ?::timestamptz)')
++ .run(day, cell.model, cell.provider, cell.task_kind, cell.runs, JSON.stringify(cell.metrics), iso(now));
++ }
++ days++; cells += values.length;
++ }
++ await db.prepare('UPDATE exchange_metrics_job SET next_due = ?::timestamptz, completed_through = ?::date, completed_at = ?::timestamptz WHERE id = 1')
++ .run(iso(today + DAY + 5 * 60000), dayOf(today - DAY), iso(now));
++ return { days, cells, completed_through: dayOf(today - DAY) };
++ });
++ }
++ function start() {
++ if (!enabled || !stopped) return;
++ stopped = false;
++ const schedule = (delay) => { timer = setTimeout(run, delay); timer.unref?.(); };
++ const run = () => {
++ pending = tick().catch((err) => log.warn(`[ai] metrics rollup failed: ${err.message}`)).finally(() => {
++ pending = null;
++ if (!stopped) schedule(60000);
++ });
++ };
++ schedule(5000);
++ }
++ async function stop() { stopped = true; clearTimeout(timer); if (pending) await pending; }
++ return { tick, start, stop };
++}
++
++module.exports = { createMetricsRollup, aggregate, publicOffers, taskKind, quantile, MIN_SAMPLES, TASK_KINDS, DAY };
ddiff --git a/server/index.js b/server/index.js
iindex 892be4a..80ba138 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -126,6 +126,8 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
// The Exchange's routing presets and the receipts of the answered requests (server/exchange/).
const presets = require('./exchange/presets').createPresets({ db, clock, log });
const receipts = require('./exchange/receipts').createReceipts({ db, clock, log });
++ const measuredMetrics = require('./exchange/metrics-rollup').createMetricsRollup({ db, registry, clock, log,
++ retentionDays: config.runs.retentionDays, enabled: config.nodeEnv !== 'test' && !/^(1|true)$/i.test(env.AI_DRILL || '') });
// The weekly real-traffic share (stage B8): the public read stores the week; the housekeeping pass
// publishes it even when nobody reads it, before the receipts it came from are pruned.
const share = require('./exchange/share').createShare({ db, clock, log, k: config.trafficShare.k });
@@@ -270,6 +272,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
if (listen && config.nodeEnv !== 'test' && memorySharing) memorySharing.start();
// The price book reads the providers' listings once a week; it never starts in test mode or under a drill.
if (listen && config.priceBook && config.priceBook.enabled) priceBook.start();
++ if (listen) measuredMetrics.start();
// The public GPU price reader: public pages only, due-based, off in tests and under a drill.
if (listen && config.gpuMarket && config.gpuMarket.enabled) gpuMarket.start();
// The refusal job asks the offers the same questions once a week; it never starts in test mode (the
@@@ -318,6 +321,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
memoryJobs.stop();
if (memorySharing) memorySharing.stop();
priceBook.stop();
++ await measuredMetrics.stop();
gpuMarket.stop();
refusal.stop();
batches.stop();
ddiff --git a/server/openapi.js b/server/openapi.js
iindex 7107e0b..f520780 100644
---- a/server/openapi.js
++++ b/server/openapi.js
@@@ -588,6 +588,15 @@ route('put', '/agent-role-defaults/{role}', {
requestBody: body({ type: 'object', properties: { model: str(), choice_order: arrayOf(str()), effort: effortFields.effort } }),
responses: { 200: json('Saved default', { type: 'object' }), ...errors([401, 403, 422]) },
});
++route('get', '/exchange/metrics/status', {
++ operationId: 'exchangeMetricsStatus', summary: 'Daily measured metrics publication status', tags: ['exchange'], security: [],
++ description: 'Public, cached sixty seconds. Complete UTC days from retained run receipts; published cells and each timing measure require twenty samples. No caller or prompt data.',
++ responses: { 200: json('Rollup status', { type: 'object', properties: {
++ schema: { const: 'exchange.metrics@1' }, minimum_samples: { const: 20 }, published_cells: integer(),
++ completed_through: nullable('string', { format: 'date' }), completed_at: nullable('string', { format: 'date-time' }),
++ first_day: nullable('string', { format: 'date' }), last_day: nullable('string', { format: 'date' }),
++ } }) },
++});
route('get', '/exchange/catalog', {
operationId: 'exchangeCatalog', summary: 'Every model and every offer that serves it', tags: ['exchange'], security: [],
description: 'Public. The live offers: measured status (up/degraded/down, with a plain reason and since when), price in, out and cached per million tokens, measured p50/p95, uptime, requests, and the provider data policy. Speed and uptime are read from the last seven days of real runs (time to the first token where a receipt recorded one, else the answering attempt\'s total); cached answers are counted apart and excluded from speed, uptime and requests. Uptime is succeeded / (succeeded + failed); null when no data, never 0. A prepaid offer has no price (free: true); free_tier says whether the free allowance and the anonymous Try it may use an offer at all (a named AI_FREE_PROVIDERS source under the output price ceiling). Models carry a kind (chat, embedding, stt, image); chat lists must use kind "chat" and never mix embeddings or speech-to-text.',
ddiff --git a/test/measured-metrics.test.js b/test/measured-metrics.test.js
nnew file mode 100644
iindex 0000000..c1d67c8
---- /dev/null
++++ b/test/measured-metrics.test.js
@@@ -0,0 +1,79 @@
++'use strict';
++const assert = require('assert');
++const { suite } = require('./helpers');
++const { aggregate, createMetricsRollup, publicOffers } = require('../server/exchange/metrics-rollup');
++const t = suite('measured-metrics');
++const providers = [{ key: 'openai', kind: 'openai', status: 'active', provider_access: 'OWN_DEPLOYMENT' }];
++const models = [{ provider_key: 'openai', model_key: 'gpt-test', metadata: { id: 'openai/gpt-test' } }];
++const offers = [{ key: 'openai', provider: 'openai', keyModel: 'gpt-test', model: 'openai/gpt-test' }];
++const row = { provider: 'openai', model: 'gpt-test-2026-11-01', workflow: 'ai.complete', status: 'succeeded', latency_ms: 1000, ttft_ms: 200, tokens_out: 80, cost_usd: 0.01 };
++let testdb;
++t.test('each measure requires twenty valid observations; no caller data is copied', () => {
++ assert.deepStrictEqual(aggregate(Array(19).fill(row), offers), []);
++ const rows = Array.from({ length: 20 }, (_, i) => ({ ...row, latency_ms: 1000 + i * 100, owner_id: 'private', prompt: 'private' }));
++ const cell = aggregate(rows, offers)[0];
++ assert.equal(cell.model, 'openai/gpt-test');
++ assert.equal(cell.metrics.latency_p50_ms, 1950);
++ assert.equal(cell.metrics.latency_p95_ms, 2805);
++ assert.equal(cell.metrics.cost_per_1k_output_tokens_usd.toFixed(3), '0.125');
++ assert(!JSON.stringify(cell).includes('private'));
++ rows[0].ttft_ms = null;
++ rows.push({ ...row, status: 'failed', latency_ms: 99999 });
++ const metrics = aggregate(rows, offers)[0].metrics;
++ assert.equal(metrics.success_rate, 20 / 21);
++ assert.equal(metrics.latency_p50_ms, 1950);
++ assert.equal(metrics.ttft_p50_ms, null);
++ assert.equal(metrics.tokens_per_second_p50, null);
++ assert.equal(metrics.samples.ttft, null);
++});
++t.test('private and synthetic providers cannot form published cells', () => {
++ const privateProviders = [
++ { ...providers[0], metadata: { hidden: true } },
++ { ...providers[0], provider_access: 'CUSTOMER_BYOK' },
++ { ...providers[0], kind: 'stub' },
++ ];
++ for (const provider of privateProviders) assert.deepStrictEqual(publicOffers([provider], models), []);
++ assert.equal(aggregate([{ ...row, provider: 'byo:private' }, ...Array(19).fill(row)], offers).length, 0);
++});
++t.test('durable due job backfills and survives restarts without losing older days', async () => {
++ testdb = await require('./db').testDb();
++ const db = testdb.db;
++ let now = Date.parse('2026-11-10T01:00:00Z');
++ const registry = { listProviders: async () => providers, listModels: async () => models };
++ const options = { db, registry, clock: { now: () => now }, retentionDays: 3 };
++ const at = '2026-11-09T12:00:00Z';
++ for (let i = 0; i < 21; i++) {
++ const id = `metric_run_${i}`;
++ await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, synthetic, created_at)
++ VALUES (?, 'ai.complete', 1, 'succeeded', 'service', 'private-site', '{}', 'hash', ?, ?)`)
++ .run(id, i === 20 ? 1 : 0, at);
++ await db.prepare('INSERT INTO ai_receipts (id, run_id, body, created_at) VALUES (?, ?, ?::jsonb, ?::timestamptz)').run(`rcpt_${i}`, id, JSON.stringify({
++ offer: { provider: 'openai', model: 'gpt-test' }, workflow: 'ai.complete', latency_ms: 1000, ttft_ms: 200, usage: { tokens_out: 80 }, cost_usd: 0.01,
++ }), at);
++ }
++ const first = await createMetricsRollup(options).tick();
++ assert.equal(first.cells, 1);
++ const cells = await db.prepare('SELECT * FROM exchange_metrics_daily').all();
++ assert.equal(cells[0].runs, 20);
++ assert.equal(cells[0].metrics.tokens_per_second_p50, 100);
++ const app = require('express')();
++ app.use(require('../server/api/measured-metrics').measuredMetricsRouter({ db }));
++ const server = app.listen(0, '127.0.0.1');
++ await new Promise((resolve) => server.once('listening', resolve));
++ try {
++ const res = await fetch(`http://127.0.0.1:${server.address().port}/api/v1/exchange/metrics/status`);
++ assert.equal(res.status, 200);
++ assert.equal(res.headers.get('cache-control'), 'public, max-age=60');
++ const body = await res.json();
++ assert.equal(body.published_cells, 1);
++ assert.equal(body.minimum_samples, 20);
++ assert(!JSON.stringify(body).includes('private'));
++ } finally { await new Promise((resolve) => server.close(resolve)); }
++ assert.equal((await createMetricsRollup(options).tick()).skipped, 'not_due');
++ await db.prepare('DELETE FROM ai_receipts').run();
++ now += 5 * 86400000;
++ await createMetricsRollup(options).tick();
++ assert.equal((await db.prepare('SELECT * FROM exchange_metrics_daily').all()).length, 1, 'expired receipts never erase published history');
++});
++t.test('close', async () => { if (testdb) await testdb.close(); });
++t.run();