openvibers/openvibe.ai Commits
Read measured comparisons, model series, price history and provider health (#439)
Changes
@@@ -23,5 +23,30 @@ 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.++Public reads under `/api/v1/exchange/metrics` are cached sixty seconds and allow++cross-origin GETs without credentials. Model slugs in path segments are URI-encoded.++++| Read | Parameters | Data |++| --- | --- | --- |++| `/models/:model` | `days=30|90`, optional `kind` | Daily series per provider and task kind |++| `/compare` | `models=a,b` (up to six), optional `kind`, `measures`, `day` | Exact daily metrics, grouped by model/provider/kind |++| `/leaderboard` | `kind`, `measure`, optional `day` | Ordered model/provider rows for that completed UTC day |++| `/prices/:model` | `days=30|90` | Daily catalogue prices per offer, source and capture time |++| `/providers/:provider` | none | Seven days of hourly run success rate and successful median latency |++| `/status` | none | Published date coverage and job completion |++++Comparisons and leaderboards default to the latest published day, shared by every++row. They never average daily medians or mix dates without saying so. These are++observational workloads, not controlled benchmark runs. The response names the++measure units, sample minimum, methodology, providers, task kinds and dates. An++empty series or cell is unavailable evidence, never a zero score.++++The second due job records hourly provider health and today's listed prices. It++backfills health from seven days of receipts, refreshes two prior hours and keeps++only seven days. Prices are immutable daily observations and accumulate from the++deployment date; no previous price is invented. A listed price is not a measured++number and needs no run samples. Adult models are absent from these public reads.++++For Fable: chart inputs are `series[].day`, provider/task kind, `runs`, the selected++measure and `samples`; price charts use `price_in/out/cached` in USD per million++tokens; health charts use UTC `hour`, `success_rate`, `latency_p50_ms` and samples.nnew file mode 100644@@@ -0,0 +1,31 @@++-- phase: expand++-- Public listed prices, and provider-hour aggregates of at least twenty run receipts.++CREATE TABLE IF NOT EXISTS exchange_prices_daily (++ day date NOT NULL,++ model text NOT NULL,++ provider text NOT NULL,++ offer text NOT NULL,++ price_in double precision,++ price_out double precision,++ price_cached double precision,++ source text,++ captured_at timestamptz NOT NULL,++ PRIMARY KEY (day, model, provider, offer)++);++CREATE INDEX IF NOT EXISTS exchange_prices_model_day ON exchange_prices_daily (model, day);++CREATE TABLE IF NOT EXISTS exchange_provider_hours (++ hour timestamptz NOT NULL,++ provider text NOT NULL,++ runs integer NOT NULL CHECK (runs >= 20),++ success_rate double precision NOT NULL CHECK (success_rate BETWEEN 0 AND 1),++ latency_p50_ms double precision,++ latency_samples integer CHECK (latency_samples >= 20),++ computed_at timestamptz NOT NULL,++ PRIMARY KEY (hour, provider)++);++CREATE TABLE IF NOT EXISTS exchange_observations_job (++ id integer PRIMARY KEY CHECK (id = 1),++ next_due timestamptz NOT NULL,++ completed_hour timestamptz++);++INSERT INTO exchange_observations_job (id, next_due) VALUES (1, '1970-01-01') ON CONFLICT DO NOTHING;@@@ -1,8 +1,35 @@'use strict';const express = require('express');--const { sendError } = require('../util');++const { sendError, AiError } = require('../util');function measuredMetricsRouter({ db, clock = { now: () => Date.now() }, log = console }) {const r = express.Router();++ const reads = require('../exchange/metrics-read').createMetricsRead({ db, clock });++ const cache = new Map();++ r.use('/api/v1/exchange/metrics', (req, res, next) => {++ if (!['GET', 'OPTIONS'].includes(req.method)) return next();++ res.set('Access-Control-Allow-Origin', '*');++ if (req.method === 'OPTIONS') return res.set({ 'Access-Control-Allow-Methods': 'GET, OPTIONS',++ 'Access-Control-Allow-Headers': 'Accept, Content-Type, Authorization', 'Access-Control-Max-Age': '600' }).sendStatus(204);++ next();++ });++ const read = (fn) => async (req, res) => {++ try {++ const key = req.originalUrl;++ if (key.length > 2400) throw new AiError(422, 'metrics.invalid', 'query is too long');++ let value = cache.get(key);++ if (!value || value.until <= clock.now()) {++ value = { until: clock.now() + 60000, body: await fn(req) };++ cache.set(key, value);++ if (cache.size > 256) cache.delete(cache.keys().next().value);++ }++ res.set('Cache-Control', 'public, max-age=60').json(value.body);++ } catch (err) { res.set('Cache-Control', 'no-store'); sendError(res, err, req.ov, log); }++ };++ r.get('/api/v1/exchange/metrics/models/:model', read((req) => reads.series(req.params.model, req.query)));++ r.get('/api/v1/exchange/metrics/compare', read((req) => reads.compare(req.query)));++ r.get('/api/v1/exchange/metrics/leaderboard', read((req) => reads.leaderboard(req.query)));++ r.get('/api/v1/exchange/metrics/prices/:model', read((req) => reads.prices(req.params.model, req.query)));++ r.get('/api/v1/exchange/metrics/providers/:provider', read((req) => reads.health(req.params.provider)));let hit = null;r.get('/api/v1/exchange/metrics/status', async (req, res) => {try {nnew file mode 100644@@@ -0,0 +1,82 @@++'use strict';++const { publicOffers, quantile, DAY, MIN_SAMPLES } = require('./metrics-rollup');++const { modelMatches } = require('./observed');++const HOUR = 3600000;++const iso = (ms) => new Date(ms).toISOString();++const numeric = (value) => value == null || !Number.isFinite(Number(value)) || Number(value) < 0 ? null : Number(value);++++function healthCells(rows, offers) {++ const matches = new Map(), cells = new Map();++ for (const row of rows) {++ const identity = `${row.provider}\0${row.model}`;++ if (!matches.has(identity)) matches.set(identity, offers.find((o) => o.key === row.provider && modelMatches(o.keyModel, row.model)) || null);++ const offer = matches.get(identity);++ if (!offer) continue;++ const key = `${row.hour}\0${offer.provider}`;++ const cell = cells.get(key) || { hour: row.hour, provider: offer.provider, runs: 0, successes: 0, latency: [] };++ cell.runs++;++ if (row.status === 'succeeded') {++ cell.successes++;++ const latency = numeric(row.latency_ms);++ if (latency != null) cell.latency.push(latency);++ }++ cells.set(key, cell);++ }++ return [...cells.values()].filter((c) => c.runs >= MIN_SAMPLES).map((c) => ({ hour: c.hour, provider: c.provider, runs: c.runs,++ success_rate: c.successes / c.runs, latency_p50_ms: quantile(c.latency, 0.5), latency_samples: c.latency.length >= MIN_SAMPLES ? c.latency.length : null }));++}++++function createObservations({ db, registry, catalog, clock = { now: () => Date.now() }, log = console, enabled = true }) {++ let timer, pending, stopped = true;++ async function tick() {++ if (!enabled) return { skipped: 'disabled' };++ const now = clock.now(), end = Math.floor(now / HOUR) * HOUR;++ return db.tx(async () => {++ const state = await db.prepare('SELECT * FROM exchange_observations_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());++ const oldest = end - 7 * DAY;++ const from = state.completed_hour ? Math.max(oldest, Date.parse(state.completed_hour) - HOUR) : oldest;++ const rows = await db.prepare(`SELECT to_char(c.created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:00:00"Z"') AS hour,++ c.body->'offer'->>'provider' AS provider, c.body->'offer'->>'model' AS model,++ r.status, c.body->>'latency_ms' AS latency_ms 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.cached_from IS NULL AND r.status IN ('succeeded', 'failed')`).all(iso(from), iso(end));++ const cells = healthCells(rows, offers);++ await db.prepare('DELETE FROM exchange_provider_hours WHERE hour < ?::timestamptz OR hour >= ?::timestamptz').run(iso(oldest), iso(from));++ for (const c of cells) await db.prepare(`INSERT INTO exchange_provider_hours++ (hour, provider, runs, success_rate, latency_p50_ms, latency_samples, computed_at) VALUES (?::timestamptz, ?, ?, ?, ?, ?, ?::timestamptz)`)++ .run(c.hour, c.provider, c.runs, c.success_rate, c.latency_p50_ms, c.latency_samples, iso(now));++ const day = iso(now).slice(0, 10);++ const snapshot = await catalog.models();++ let prices = 0;++ for (const model of snapshot.models || []) {++ if (model.uncensored) continue;++ for (const offer of model.offers || []) {++ const out = await db.prepare(`INSERT INTO exchange_prices_daily++ (day, model, provider, offer, price_in, price_out, price_cached, source, captured_at)++ VALUES (?::date, ?, ?, ?, ?, ?, ?, ?, ?::timestamptz) ON CONFLICT DO NOTHING`)++ .run(day, model.slug, offer.provider, offer.id || offer.model, numeric(offer.price?.in), numeric(offer.price?.out), numeric(offer.price?.cached), offer.price_source || null, iso(now));++ prices += out.changes;++ }++ }++ await db.prepare('UPDATE exchange_observations_job SET next_due = ?::timestamptz, completed_hour = ?::timestamptz WHERE id = 1')++ .run(iso(end + HOUR + 5 * 60000), iso(end - HOUR));++ return { hours: cells.length, prices };++ });++ }++ function start() {++ if (!enabled || !stopped) return;++ stopped = false;++ const run = () => { pending = tick().catch((err) => log.warn(`[ai] observations failed: ${err.message}`)).finally(() => {++ pending = null;++ if (!stopped) { timer = setTimeout(run, 60000); timer.unref?.(); }++ }); };++ timer = setTimeout(run, 5000); timer.unref?.();++ }++ async function stop() { stopped = true; clearTimeout(timer); if (pending) await pending; }++ return { tick, start, stop };++}++module.exports = { createObservations, healthCells };nnew file mode 100644@@@ -0,0 +1,75 @@++'use strict';++const { AiError } = require('../util');++const { TASK_KINDS, DAY } = require('./metrics-rollup');++const MEASURES = ['success_rate', 'latency_p50_ms', 'latency_p95_ms', 'ttft_p50_ms', 'tokens_per_second_p50', 'cost_per_run_usd', 'cost_per_1k_output_tokens_usd'];++const HIGHER = new Set(['success_rate', 'tokens_per_second_p50']);++const date = (v) => v instanceof Date ? v.toISOString().slice(0, 10) : String(v).slice(0, 10);++const fail = (detail) => { throw new AiError(422, 'metrics.invalid', detail); };++function modelId(v) { if (typeof v !== 'string' || !/^[a-z0-9][a-z0-9._/-]{0,199}$/.test(v)) fail('model must be a catalogue model slug'); return v; }++function daysOf(v) { if (v == null) return 30; if (!['30', '90'].includes(String(v))) fail('days must be 30 or 90'); return Number(v); }++function kindOf(v) { if (v == null || v === '') return null; if (!TASK_KINDS.includes(v)) fail('unknown task kind'); return v; }++function measuresOf(v) {++ if (v == null) return MEASURES;++ if (typeof v !== 'string') fail('measures must be comma-separated names');++ const values = [...new Set(v.split(','))];++ if (!values.length || values.some((m) => !MEASURES.includes(m))) fail('unknown measure');++ return values;++}++function wire(row, measures = MEASURES) {++ const metrics = typeof row.metrics === 'string' ? JSON.parse(row.metrics) : row.metrics;++ return { day: date(row.day), model: row.model, provider: row.provider, task_kind: row.task_kind, runs: Number(row.runs),++ ...Object.fromEntries(measures.map((m) => [m, metrics[m] ?? null])), samples: metrics.samples, computed_at: row.computed_at };++}++function createMetricsRead({ db, clock = { now: () => Date.now() } }) {++ const base = () => ({ schema: 'exchange.metrics@1', minimum_samples: 20, source: 'OpenVibe.AI run receipts',++ methodology: 'Complete UTC days; real uncached runs; successful total latency; decode throughput excludes first-token time. Workloads vary. Sparse measures are null.', measures: MEASURES, task_kinds: TASK_KINDS });++ async function dayOf(value) {++ if (value == null || value === '') {++ const row = await db.prepare('SELECT MAX(day) AS day FROM exchange_metrics_daily WHERE runs >= 20').get();++ return row.day ? date(row.day) : new Date(clock.now() - DAY).toISOString().slice(0, 10);++ }++ if (typeof value !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(value) || !Number.isFinite(Date.parse(value)) || date(new Date(value)) !== value++ || value >= new Date(clock.now()).toISOString().slice(0, 10)) fail('day must be a completed UTC date');++ return value;++ }++ async function series(model, query = {}) {++ modelId(model); const days = daysOf(query.days), kind = kindOf(query.kind);++ const to = new Date(clock.now()).toISOString().slice(0, 10), from = new Date(Date.parse(to) - days * DAY).toISOString().slice(0, 10);++ const rows = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE model = ? AND runs >= 20 AND day >= ?::date AND day < ?::date++ AND (?::text IS NULL OR task_kind = ?) ORDER BY day, provider, task_kind`).all(model, from, to, kind, kind);++ return { ...base(), model, days, from, to, kind, series: rows.map((r) => wire(r)), missing: 'No published cell means fewer than twenty retained observations or no observations.' };++ }++ async function compare(query = {}) {++ if (typeof query.models !== 'string') fail('models must be one to six comma-separated catalogue slugs');++ const models = [...new Set(query.models.split(','))];++ if (models.length < 1 || models.length > 6) fail('choose at most six models');++ models.forEach(modelId);++ const measures = measuresOf(query.measures), kind = kindOf(query.kind), day = await dayOf(query.day);++ const rows = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE model = ANY(?::text[]) AND day = ?::date AND runs >= 20++ AND (?::text IS NULL OR task_kind = ?) ORDER BY model, provider, task_kind`).all(models, day, kind, kind);++ return { ...base(), day, kind, measures, models: models.map((model) => ({ model, cells: rows.filter((r) => r.model === model).map((r) => wire(r, measures)) })) };++ }++ async function leaderboard(query = {}) {++ const kind = kindOf(query.kind) || 'chat', measure = query.measure || 'latency_p50_ms';++ if (!MEASURES.includes(measure)) fail('unknown measure');++ const day = await dayOf(query.day);++ const records = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE day = ?::date AND task_kind = ? AND runs >= 20`).all(day, kind);++ const rows = records.map((r) => wire(r)).filter((r) => r[measure] != null);++ rows.sort((a, b) => (HIGHER.has(measure) ? -1 : 1) * (a[measure] - b[measure]) || a.model.localeCompare(b.model) || a.provider.localeCompare(b.provider));++ return { ...base(), day, kind, measure, direction: HIGHER.has(measure) ? 'higher' : 'lower', rows: rows.slice(0, 100) };++ }++ async function prices(model, query = {}) {++ modelId(model); const days = daysOf(query.days);++ const from = new Date(clock.now() - (days - 1) * DAY).toISOString().slice(0, 10);++ const rows = await db.prepare('SELECT * FROM exchange_prices_daily WHERE model = ? AND day >= ?::date ORDER BY day, provider, offer').all(model, from);++ return { schema: 'exchange.prices@1', model, days, source: 'Daily catalogue-listed prices; not measured run costs', currency: 'USD', per: '1M tokens', series: rows.map((r) => ({ ...r, day: date(r.day) })) };++ }++ async function health(provider) {++ if (typeof provider !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,79}$/.test(provider)) fail('invalid provider');++ const to = new Date(Math.floor(clock.now() / 3600000) * 3600000).toISOString(), from = new Date(Date.parse(to) - 7 * DAY).toISOString();++ const rows = await db.prepare('SELECT * FROM exchange_provider_hours WHERE provider = ? AND hour >= ?::timestamptz AND hour < ?::timestamptz AND runs >= 20 ORDER BY hour').all(provider, from, to);++ return { ...base(), provider, days: 7, interval: 'hour', from, to, series: rows };++ }++ return { series, compare, leaderboard, prices, health };++}++module.exports = { createMetricsRead, MEASURES, modelId, daysOf, kindOf };@@@ -35,8 +35,10 @@ function publicOffers(providers, models) {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 })) }];++ if (!upstream) return [];++ const canonical = canonicalOf(m, { key: upstream.key });++ if (m.metadata?.hidden || m.metadata?.uncensored || /uncensor|abliterat/i.test(`${m.model_key} ${canonical} ${m.metadata?.variant || ''}`)) return [];++ return [{ key: provider.key, provider: upstream.key, keyModel: m.model_key, model: slugOf(canonical) }];}).sort((a, b) => b.keyModel.length - a.keyModel.length);}@@@ -128,6 +128,9 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {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 || '') });++ const observations = require('./exchange/metrics-observations').createObservations({ db, registry, clock, log,++ catalog: require('./exchange/catalog').createCatalog({ db, registry, pool, config, env, clock }),++ 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 });@@@ -273,6 +276,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {// 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();++ if (listen) observations.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@@@ -322,6 +326,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {if (memorySharing) memorySharing.stop();priceBook.stop();await measuredMetrics.stop();++ await observations.stop();gpuMarket.stop();refusal.stop();batches.stop();@@@ -588,6 +588,26 @@ 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]) },});++const measuredCell = { type: 'object', properties: {++ day: str({ format: 'date' }), model: str(), provider: str(), task_kind: str(), runs: integer({ minimum: 20 }),++ ...Object.fromEntries(require('./exchange/metrics-read').MEASURES.map((m) => [m, nullable('number')])),++ samples: { type: 'object', additionalProperties: nullable('integer', { minimum: 20 }) }, computed_at: str({ format: 'date-time' }),++} };++for (const [routePath, id, summary, parameters, properties] of [++ ['/exchange/metrics/models/{model}', 'exchangeModelMetrics', 'Measured model series over 30 or 90 completed UTC days',++ [path('model', str()), query('days', integer({ enum: [30, 90] })), query('kind', str())], { series: arrayOf(measuredCell) }],++ ['/exchange/metrics/compare', 'exchangeCompareMetrics', 'Compare up to six models on chosen measures for one completed UTC day',++ [query('models', str()), query('measures', str()), query('kind', str()), query('day', str({ format: 'date' }))],++ { models: arrayOf({ type: 'object', properties: { model: str(), cells: arrayOf(measuredCell) } }) }],++ ['/exchange/metrics/leaderboard', 'exchangeMetricsLeaderboard', 'A task kind leaderboard for one measure and completed UTC day',++ [query('kind', str()), query('measure', str({ enum: require('./exchange/metrics-read').MEASURES })), query('day', str({ format: 'date' }))], { rows: arrayOf(measuredCell) }],++ ['/exchange/metrics/prices/{model}', 'exchangeModelPriceHistory', 'Daily listed catalogue prices per model and provider',++ [path('model', str()), query('days', integer({ enum: [30, 90] }))], { series: arrayOf({ type: 'object' }) }],++ ['/exchange/metrics/providers/{provider}', 'exchangeProviderHealth', 'Seven days of hourly provider run success and median latency',++ [path('provider', str())], { series: arrayOf({ type: 'object', properties: { hour: str({ format: 'date-time' }), provider: str(), runs: integer({ minimum: 20 }), success_rate: { type: 'number' }, latency_p50_ms: nullable('number'), latency_samples: nullable('integer', { minimum: 20 }) } }) }],++]) route('get', routePath, { operationId: id, summary, tags: ['exchange'], security: [], parameters,++ description: 'Public, cached sixty seconds, CORS enabled. Measured cells and each timing measure require twenty observations. Missing is not zero; prices are listed prices, not measurements. No person, prompt or site identifiers.',++ responses: { 200: json(summary, { type: 'object', properties }), ...errors([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.',@@@ -33,6 +33,8 @@ t.test('private and synthetic providers cannot form published cells', () => {{ ...providers[0], kind: 'stub' },];for (const provider of privateProviders) assert.deepStrictEqual(publicOffers([provider], models), []);++ assert.deepStrictEqual(publicOffers(providers, [{ ...models[0], metadata: { variant: 'abliterated' } }]), []);++ assert.deepStrictEqual(publicOffers(providers, [{ ...models[0], metadata: { id: 'acme/uncensored' } }]), []);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 () => {nnew file mode 100644@@@ -0,0 +1,77 @@++'use strict';++const assert = require('assert');++const { suite } = require('./helpers');++const { aggregate } = require('../server/exchange/metrics-rollup');++const { healthCells, createObservations } = require('../server/exchange/metrics-observations');++const t = suite('metrics-reads');++const now = Date.parse('2026-11-10T12:30:00Z');++const clock = { now: () => now };++const offers = [{ key: 'openai', provider: 'openai', keyModel: 'gpt-test', model: 'openai/gpt-test' }];++const row = { provider: 'openai', model: 'gpt-test', workflow: 'ai.complete', status: 'succeeded', latency_ms: 1000, ttft_ms: 200, tokens_out: 80, cost_usd: 0.01, hour: '2026-11-10T11:00:00Z' };++let testdb, server, base;++async function get(path, opts = {}) { const res = await fetch(base + '/api/v1/exchange/metrics' + path, opts); const text = await res.text(); return { status: res.status, headers: res.headers, body: text ? JSON.parse(text) : null }; }++t.test('boot with published daily evidence for two models', async () => {++ testdb = await require('./db').testDb();++ const db = testdb.db;++ const metrics = aggregate(Array(20).fill(row), offers)[0].metrics;++ for (const [model, day, latency] of [['openai/gpt-test', '2026-11-09', 1000], ['openai/gpt-other', '2026-11-09', 2000], ['openai/gpt-test', '2026-09-10', 3000]]) {++ await db.prepare(`INSERT INTO exchange_metrics_daily (day, model, provider, task_kind, runs, metrics, computed_at)++ VALUES (?::date, ?, 'openai', 'chat', 20, ?::jsonb, ?::timestamptz)`).run(day, model, JSON.stringify({ ...metrics, latency_p50_ms: latency }), new Date(now).toISOString());++ }++ const app = require('express')();++ app.use(require('../server/api/measured-metrics').measuredMetricsRouter({ db, clock, log: { error() {} } }));++ server = app.listen(0, '127.0.0.1');++ await new Promise((r) => server.once('listening', r));++ base = `http://127.0.0.1:${server.address().port}`;++});++t.test('30/90-day series, six-model bound, selected measures, same-day ranking and CORS', async () => {++ assert.equal((await get('/models/openai%2Fgpt-test?days=30')).body.series.length, 1);++ assert.equal((await get('/models/openai%2Fgpt-test?days=90')).body.series.length, 2);++ const comparison = await get('/compare?models=openai/gpt-test,openai/gpt-other&kind=chat&measures=latency_p50_ms');++ assert.equal(comparison.status, 200);++ assert.equal(comparison.headers.get('access-control-allow-origin'), '*');++ assert.equal(comparison.headers.get('cache-control'), 'public, max-age=60');++ assert.equal(comparison.body.day, '2026-11-09');++ assert.deepEqual(comparison.body.models.map((m) => m.cells[0].latency_p50_ms), [1000, 2000]);++ assert(!Object.hasOwn(comparison.body.models[0].cells[0], 'cost_per_run_usd'));++ assert.equal((await get('/leaderboard?kind=chat&measure=latency_p50_ms')).body.rows[0].model, 'openai/gpt-test');++ const preflight = await fetch(base + '/api/v1/exchange/metrics/compare', { method: 'OPTIONS', headers: { Origin: 'https://example.test' } });++ assert.equal(preflight.status, 204);++ assert.equal(preflight.headers.get('access-control-allow-origin'), '*');++ for (const path of ['/compare?models=a,b,c,d,e,f,g', '/compare?models=a&measures=made_up', '/models/a?days=365', '/leaderboard?kind=private-site', '/compare?models=a&day=2026-02-31']) {++ const invalid = await get(path);++ assert.equal(invalid.status, 422, path);++ assert.equal(invalid.headers.get('cache-control'), 'no-store');++ }++});++t.test('hourly health withholds sparse cohorts and sparse latency separately', () => {++ assert.deepEqual(healthCells(Array(19).fill(row), offers), []);++ const health = healthCells([...Array(19).fill(row), { ...row, status: 'failed' }], offers)[0];++ assert.equal(health.runs, 20);++ assert.equal(health.success_rate, 0.95);++ assert.equal(health.latency_p50_ms, null);++ assert.equal(health.latency_samples, null);++});++t.test('hour job records receipts and daily catalogue prices once across restarts', async () => {++ const db = testdb.db;++ for (let i = 0; i < 20; i++) {++ const id = `health_${i}`;++ await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, created_at)++ VALUES (?, 'ai.complete', 1, 'succeeded', 'service', 'private', '{}', 'hash', ?)`)++ .run(id, row.hour);++ await db.prepare('INSERT INTO ai_receipts (id, run_id, body, created_at) VALUES (?, ?, ?::jsonb, ?::timestamptz)').run(id, id,++ JSON.stringify({ offer: { provider: row.provider, model: row.model }, latency_ms: row.latency_ms }), row.hour);++ }++ const registry = { listProviders: async () => [{ key: 'openai', kind: 'openai' }], listModels: async () => [{ provider_key: 'openai', model_key: 'gpt-test', metadata: { id: 'openai/gpt-test' } }] };++ const catalog = { models: async () => ({ models: [{ slug: 'openai/gpt-test', offers: [{ provider: 'openai', model: 'gpt-test', price: { in: 0.1, out: 0.5 }, price_source: 'https://example.test/prices' }] }] }) };++ assert.deepEqual(await createObservations({ db, registry, catalog, clock }).tick(), { hours: 1, prices: 1 });++ assert.equal((await createObservations({ db, registry, catalog, clock }).tick()).skipped, 'not_due');++ const prices = await get('/prices/openai%2Fgpt-test');++ assert.equal(prices.body.series[0].price_out, 0.5);++ const health = await get('/providers/openai');++ assert.equal(health.body.series.length, 1);++ assert.equal(health.body.series[0].latency_p50_ms, 1000);++ assert(!JSON.stringify(health.body).includes('private'));++});++t.test('close', async () => { if (server) await new Promise((r) => server.close(r)); if (testdb) await testdb.close(); });++t.run();
@@ -23,5 +23,30 @@ Backfill covers complete dates still inside receipt retention (normally 29 days)
@@ -23,5 +23,30 @@ Backfill covers complete dates still inside receipt retention (normally 29 days)
Longer history accumulates going forward; it is never fabricated from expired runs.
Longer history accumulates going forward; it is never fabricated from expired runs.
Models use the catalogue's canonical identity and longest matching registered revision.
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.
+Public reads under `/api/v1/exchange/metrics` are cached sixty seconds and allow
+cross-origin GETs without credentials. Model slugs in path segments are URI-encoded.
+
+| Read | Parameters | Data |
+| --- | --- | --- |
+| `/models/:model` | `days=30|90`, optional `kind` | Daily series per provider and task kind |
+| `/compare` | `models=a,b` (up to six), optional `kind`, `measures`, `day` | Exact daily metrics, grouped by model/provider/kind |
+| `/leaderboard` | `kind`, `measure`, optional `day` | Ordered model/provider rows for that completed UTC day |
+| `/prices/:model` | `days=30|90` | Daily catalogue prices per offer, source and capture time |
+| `/providers/:provider` | none | Seven days of hourly run success rate and successful median latency |
+| `/status` | none | Published date coverage and job completion |
+
+Comparisons and leaderboards default to the latest published day, shared by every
+row. They never average daily medians or mix dates without saying so. These are
+observational workloads, not controlled benchmark runs. The response names the
+measure units, sample minimum, methodology, providers, task kinds and dates. An
+empty series or cell is unavailable evidence, never a zero score.
+
+The second due job records hourly provider health and today's listed prices. It
+backfills health from seven days of receipts, refreshes two prior hours and keeps
+only seven days. Prices are immutable daily observations and accumulate from the
+deployment date; no previous price is invented. A listed price is not a measured
+number and needs no run samples. Adult models are absent from these public reads.
+
+For Fable: chart inputs are `series[].day`, provider/task kind, `runs`, the selected
+measure and `samples`; price charts use `price_in/out/cached` in USD per million
+tokens; health charts use UTC `hour`, `success_rate`, `latency_p50_ms` and samples.
new file mode 100644
new file mode 100644
@@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
+-- phase: expand
+-- Public listed prices, and provider-hour aggregates of at least twenty run receipts.
+CREATE TABLE IF NOT EXISTS exchange_prices_daily (
+ day date NOT NULL,
+ model text NOT NULL,
+ provider text NOT NULL,
+ offer text NOT NULL,
+ price_in double precision,
+ price_out double precision,
+ price_cached double precision,
+ source text,
+ captured_at timestamptz NOT NULL,
+ PRIMARY KEY (day, model, provider, offer)
+);
+CREATE INDEX IF NOT EXISTS exchange_prices_model_day ON exchange_prices_daily (model, day);
+CREATE TABLE IF NOT EXISTS exchange_provider_hours (
+ hour timestamptz NOT NULL,
+ provider text NOT NULL,
+ runs integer NOT NULL CHECK (runs >= 20),
+ success_rate double precision NOT NULL CHECK (success_rate BETWEEN 0 AND 1),
+ latency_p50_ms double precision,
+ latency_samples integer CHECK (latency_samples >= 20),
+ computed_at timestamptz NOT NULL,
+ PRIMARY KEY (hour, provider)
+);
+CREATE TABLE IF NOT EXISTS exchange_observations_job (
+ id integer PRIMARY KEY CHECK (id = 1),
+ next_due timestamptz NOT NULL,
+ completed_hour timestamptz
+);
+INSERT INTO exchange_observations_job (id, next_due) VALUES (1, '1970-01-01') ON CONFLICT DO NOTHING;
@@ -1,8 +1,35 @@
@@ -1,8 +1,35 @@
'use strict';
'use strict';
const express = require('express');
const express = require('express');
-const { sendError } = require('../util');
+const { sendError, AiError } = require('../util');
function measuredMetricsRouter({ db, clock = { now: () => Date.now() }, log = console }) {
function measuredMetricsRouter({ db, clock = { now: () => Date.now() }, log = console }) {
const r = express.Router();
const r = express.Router();
+ const reads = require('../exchange/metrics-read').createMetricsRead({ db, clock });
+ const cache = new Map();
+ r.use('/api/v1/exchange/metrics', (req, res, next) => {
+ if (!['GET', 'OPTIONS'].includes(req.method)) return next();
+ res.set('Access-Control-Allow-Origin', '*');
+ if (req.method === 'OPTIONS') return res.set({ 'Access-Control-Allow-Methods': 'GET, OPTIONS',
+ 'Access-Control-Allow-Headers': 'Accept, Content-Type, Authorization', 'Access-Control-Max-Age': '600' }).sendStatus(204);
+ next();
+ });
+ const read = (fn) => async (req, res) => {
+ try {
+ const key = req.originalUrl;
+ if (key.length > 2400) throw new AiError(422, 'metrics.invalid', 'query is too long');
+ let value = cache.get(key);
+ if (!value || value.until <= clock.now()) {
+ value = { until: clock.now() + 60000, body: await fn(req) };
+ cache.set(key, value);
+ if (cache.size > 256) cache.delete(cache.keys().next().value);
+ }
+ res.set('Cache-Control', 'public, max-age=60').json(value.body);
+ } catch (err) { res.set('Cache-Control', 'no-store'); sendError(res, err, req.ov, log); }
+ };
+ r.get('/api/v1/exchange/metrics/models/:model', read((req) => reads.series(req.params.model, req.query)));
+ r.get('/api/v1/exchange/metrics/compare', read((req) => reads.compare(req.query)));
+ r.get('/api/v1/exchange/metrics/leaderboard', read((req) => reads.leaderboard(req.query)));
+ r.get('/api/v1/exchange/metrics/prices/:model', read((req) => reads.prices(req.params.model, req.query)));
+ r.get('/api/v1/exchange/metrics/providers/:provider', read((req) => reads.health(req.params.provider)));
let hit = null;
let hit = null;
r.get('/api/v1/exchange/metrics/status', async (req, res) => {
r.get('/api/v1/exchange/metrics/status', async (req, res) => {
try {
try {
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+'use strict';
+const { publicOffers, quantile, DAY, MIN_SAMPLES } = require('./metrics-rollup');
+const { modelMatches } = require('./observed');
+const HOUR = 3600000;
+const iso = (ms) => new Date(ms).toISOString();
+const numeric = (value) => value == null || !Number.isFinite(Number(value)) || Number(value) < 0 ? null : Number(value);
+
+function healthCells(rows, offers) {
+ const matches = new Map(), cells = new Map();
+ for (const row of rows) {
+ const identity = `${row.provider}\0${row.model}`;
+ if (!matches.has(identity)) matches.set(identity, offers.find((o) => o.key === row.provider && modelMatches(o.keyModel, row.model)) || null);
+ const offer = matches.get(identity);
+ if (!offer) continue;
+ const key = `${row.hour}\0${offer.provider}`;
+ const cell = cells.get(key) || { hour: row.hour, provider: offer.provider, runs: 0, successes: 0, latency: [] };
+ cell.runs++;
+ if (row.status === 'succeeded') {
+ cell.successes++;
+ const latency = numeric(row.latency_ms);
+ if (latency != null) cell.latency.push(latency);
+ }
+ cells.set(key, cell);
+ }
+ return [...cells.values()].filter((c) => c.runs >= MIN_SAMPLES).map((c) => ({ hour: c.hour, provider: c.provider, runs: c.runs,
+ success_rate: c.successes / c.runs, latency_p50_ms: quantile(c.latency, 0.5), latency_samples: c.latency.length >= MIN_SAMPLES ? c.latency.length : null }));
+}
+
+function createObservations({ db, registry, catalog, clock = { now: () => Date.now() }, log = console, enabled = true }) {
+ let timer, pending, stopped = true;
+ async function tick() {
+ if (!enabled) return { skipped: 'disabled' };
+ const now = clock.now(), end = Math.floor(now / HOUR) * HOUR;
+ return db.tx(async () => {
+ const state = await db.prepare('SELECT * FROM exchange_observations_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());
+ const oldest = end - 7 * DAY;
+ const from = state.completed_hour ? Math.max(oldest, Date.parse(state.completed_hour) - HOUR) : oldest;
+ const rows = await db.prepare(`SELECT to_char(c.created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:00:00"Z"') AS hour,
+ c.body->'offer'->>'provider' AS provider, c.body->'offer'->>'model' AS model,
+ r.status, c.body->>'latency_ms' AS latency_ms 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.cached_from IS NULL AND r.status IN ('succeeded', 'failed')`).all(iso(from), iso(end));
+ const cells = healthCells(rows, offers);
+ await db.prepare('DELETE FROM exchange_provider_hours WHERE hour < ?::timestamptz OR hour >= ?::timestamptz').run(iso(oldest), iso(from));
+ for (const c of cells) await db.prepare(`INSERT INTO exchange_provider_hours
+ (hour, provider, runs, success_rate, latency_p50_ms, latency_samples, computed_at) VALUES (?::timestamptz, ?, ?, ?, ?, ?, ?::timestamptz)`)
+ .run(c.hour, c.provider, c.runs, c.success_rate, c.latency_p50_ms, c.latency_samples, iso(now));
+ const day = iso(now).slice(0, 10);
+ const snapshot = await catalog.models();
+ let prices = 0;
+ for (const model of snapshot.models || []) {
+ if (model.uncensored) continue;
+ for (const offer of model.offers || []) {
+ const out = await db.prepare(`INSERT INTO exchange_prices_daily
+ (day, model, provider, offer, price_in, price_out, price_cached, source, captured_at)
+ VALUES (?::date, ?, ?, ?, ?, ?, ?, ?, ?::timestamptz) ON CONFLICT DO NOTHING`)
+ .run(day, model.slug, offer.provider, offer.id || offer.model, numeric(offer.price?.in), numeric(offer.price?.out), numeric(offer.price?.cached), offer.price_source || null, iso(now));
+ prices += out.changes;
+ }
+ }
+ await db.prepare('UPDATE exchange_observations_job SET next_due = ?::timestamptz, completed_hour = ?::timestamptz WHERE id = 1')
+ .run(iso(end + HOUR + 5 * 60000), iso(end - HOUR));
+ return { hours: cells.length, prices };
+ });
+ }
+ function start() {
+ if (!enabled || !stopped) return;
+ stopped = false;
+ const run = () => { pending = tick().catch((err) => log.warn(`[ai] observations failed: ${err.message}`)).finally(() => {
+ pending = null;
+ if (!stopped) { timer = setTimeout(run, 60000); timer.unref?.(); }
+ }); };
+ timer = setTimeout(run, 5000); timer.unref?.();
+ }
+ async function stop() { stopped = true; clearTimeout(timer); if (pending) await pending; }
+ return { tick, start, stop };
+}
+module.exports = { createObservations, healthCells };
new file mode 100644
new file mode 100644
@@ -0,0 +1,75 @@
@@ -0,0 +1,75 @@
+'use strict';
+const { AiError } = require('../util');
+const { TASK_KINDS, DAY } = require('./metrics-rollup');
+const MEASURES = ['success_rate', 'latency_p50_ms', 'latency_p95_ms', 'ttft_p50_ms', 'tokens_per_second_p50', 'cost_per_run_usd', 'cost_per_1k_output_tokens_usd'];
+const HIGHER = new Set(['success_rate', 'tokens_per_second_p50']);
+const date = (v) => v instanceof Date ? v.toISOString().slice(0, 10) : String(v).slice(0, 10);
+const fail = (detail) => { throw new AiError(422, 'metrics.invalid', detail); };
+function modelId(v) { if (typeof v !== 'string' || !/^[a-z0-9][a-z0-9._/-]{0,199}$/.test(v)) fail('model must be a catalogue model slug'); return v; }
+function daysOf(v) { if (v == null) return 30; if (!['30', '90'].includes(String(v))) fail('days must be 30 or 90'); return Number(v); }
+function kindOf(v) { if (v == null || v === '') return null; if (!TASK_KINDS.includes(v)) fail('unknown task kind'); return v; }
+function measuresOf(v) {
+ if (v == null) return MEASURES;
+ if (typeof v !== 'string') fail('measures must be comma-separated names');
+ const values = [...new Set(v.split(','))];
+ if (!values.length || values.some((m) => !MEASURES.includes(m))) fail('unknown measure');
+ return values;
+}
+function wire(row, measures = MEASURES) {
+ const metrics = typeof row.metrics === 'string' ? JSON.parse(row.metrics) : row.metrics;
+ return { day: date(row.day), model: row.model, provider: row.provider, task_kind: row.task_kind, runs: Number(row.runs),
+ ...Object.fromEntries(measures.map((m) => [m, metrics[m] ?? null])), samples: metrics.samples, computed_at: row.computed_at };
+}
+function createMetricsRead({ db, clock = { now: () => Date.now() } }) {
+ const base = () => ({ schema: 'exchange.metrics@1', minimum_samples: 20, source: 'OpenVibe.AI run receipts',
+ methodology: 'Complete UTC days; real uncached runs; successful total latency; decode throughput excludes first-token time. Workloads vary. Sparse measures are null.', measures: MEASURES, task_kinds: TASK_KINDS });
+ async function dayOf(value) {
+ if (value == null || value === '') {
+ const row = await db.prepare('SELECT MAX(day) AS day FROM exchange_metrics_daily WHERE runs >= 20').get();
+ return row.day ? date(row.day) : new Date(clock.now() - DAY).toISOString().slice(0, 10);
+ }
+ if (typeof value !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(value) || !Number.isFinite(Date.parse(value)) || date(new Date(value)) !== value
+ || value >= new Date(clock.now()).toISOString().slice(0, 10)) fail('day must be a completed UTC date');
+ return value;
+ }
+ async function series(model, query = {}) {
+ modelId(model); const days = daysOf(query.days), kind = kindOf(query.kind);
+ const to = new Date(clock.now()).toISOString().slice(0, 10), from = new Date(Date.parse(to) - days * DAY).toISOString().slice(0, 10);
+ const rows = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE model = ? AND runs >= 20 AND day >= ?::date AND day < ?::date
+ AND (?::text IS NULL OR task_kind = ?) ORDER BY day, provider, task_kind`).all(model, from, to, kind, kind);
+ return { ...base(), model, days, from, to, kind, series: rows.map((r) => wire(r)), missing: 'No published cell means fewer than twenty retained observations or no observations.' };
+ }
+ async function compare(query = {}) {
+ if (typeof query.models !== 'string') fail('models must be one to six comma-separated catalogue slugs');
+ const models = [...new Set(query.models.split(','))];
+ if (models.length < 1 || models.length > 6) fail('choose at most six models');
+ models.forEach(modelId);
+ const measures = measuresOf(query.measures), kind = kindOf(query.kind), day = await dayOf(query.day);
+ const rows = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE model = ANY(?::text[]) AND day = ?::date AND runs >= 20
+ AND (?::text IS NULL OR task_kind = ?) ORDER BY model, provider, task_kind`).all(models, day, kind, kind);
+ return { ...base(), day, kind, measures, models: models.map((model) => ({ model, cells: rows.filter((r) => r.model === model).map((r) => wire(r, measures)) })) };
+ }
+ async function leaderboard(query = {}) {
+ const kind = kindOf(query.kind) || 'chat', measure = query.measure || 'latency_p50_ms';
+ if (!MEASURES.includes(measure)) fail('unknown measure');
+ const day = await dayOf(query.day);
+ const records = await db.prepare(`SELECT * FROM exchange_metrics_daily WHERE day = ?::date AND task_kind = ? AND runs >= 20`).all(day, kind);
+ const rows = records.map((r) => wire(r)).filter((r) => r[measure] != null);
+ rows.sort((a, b) => (HIGHER.has(measure) ? -1 : 1) * (a[measure] - b[measure]) || a.model.localeCompare(b.model) || a.provider.localeCompare(b.provider));
+ return { ...base(), day, kind, measure, direction: HIGHER.has(measure) ? 'higher' : 'lower', rows: rows.slice(0, 100) };
+ }
+ async function prices(model, query = {}) {
+ modelId(model); const days = daysOf(query.days);
+ const from = new Date(clock.now() - (days - 1) * DAY).toISOString().slice(0, 10);
+ const rows = await db.prepare('SELECT * FROM exchange_prices_daily WHERE model = ? AND day >= ?::date ORDER BY day, provider, offer').all(model, from);
+ return { schema: 'exchange.prices@1', model, days, source: 'Daily catalogue-listed prices; not measured run costs', currency: 'USD', per: '1M tokens', series: rows.map((r) => ({ ...r, day: date(r.day) })) };
+ }
+ async function health(provider) {
+ if (typeof provider !== 'string' || !/^[a-z0-9][a-z0-9._-]{0,79}$/.test(provider)) fail('invalid provider');
+ const to = new Date(Math.floor(clock.now() / 3600000) * 3600000).toISOString(), from = new Date(Date.parse(to) - 7 * DAY).toISOString();
+ const rows = await db.prepare('SELECT * FROM exchange_provider_hours WHERE provider = ? AND hour >= ?::timestamptz AND hour < ?::timestamptz AND runs >= 20 ORDER BY hour').all(provider, from, to);
+ return { ...base(), provider, days: 7, interval: 'hour', from, to, series: rows };
+ }
+ return { series, compare, leaderboard, prices, health };
+}
+module.exports = { createMetricsRead, MEASURES, modelId, daysOf, kindOf };
@@ -35,8 +35,10 @@ function publicOffers(providers, models) {
@@ -35,8 +35,10 @@ function publicOffers(providers, models) {
return models.flatMap((m) => {
return models.flatMap((m) => {
const provider = byKey.get(m.provider_key);
const provider = byKey.get(m.provider_key);
const upstream = provider && upstreamOf(provider);
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 })) }];
+ if (!upstream) return [];
+ const canonical = canonicalOf(m, { key: upstream.key });
+ if (m.metadata?.hidden || m.metadata?.uncensored || /uncensor|abliterat/i.test(`${m.model_key} ${canonical} ${m.metadata?.variant || ''}`)) return [];
+ return [{ key: provider.key, provider: upstream.key, keyModel: m.model_key, model: slugOf(canonical) }];
}).sort((a, b) => b.keyModel.length - a.keyModel.length);
}).sort((a, b) => b.keyModel.length - a.keyModel.length);
}
}
@@ -128,6 +128,9 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
@@ -128,6 +128,9 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
const receipts = require('./exchange/receipts').createReceipts({ db, clock, log });
const receipts = require('./exchange/receipts').createReceipts({ db, clock, log });
const measuredMetrics = require('./exchange/metrics-rollup').createMetricsRollup({ db, registry, 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 || '') });
retentionDays: config.runs.retentionDays, enabled: config.nodeEnv !== 'test' && !/^(1|true)$/i.test(env.AI_DRILL || '') });
+ const observations = require('./exchange/metrics-observations').createObservations({ db, registry, clock, log,
+ catalog: require('./exchange/catalog').createCatalog({ db, registry, pool, config, env, clock }),
+ 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
// 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.
// 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 });
const share = require('./exchange/share').createShare({ db, clock, log, k: config.trafficShare.k });
@@ -273,6 +276,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
@@ -273,6 +276,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
// The price book reads the providers' listings once a week; it never starts in test mode or under a drill.
// 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 && config.priceBook && config.priceBook.enabled) priceBook.start();
if (listen) measuredMetrics.start();
if (listen) measuredMetrics.start();
+ if (listen) observations.start();
// The public GPU price reader: public pages only, due-based, off in tests and under a drill.
// 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();
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
// The refusal job asks the offers the same questions once a week; it never starts in test mode (the
@@ -322,6 +326,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
@@ -322,6 +326,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
if (memorySharing) memorySharing.stop();
if (memorySharing) memorySharing.stop();
priceBook.stop();
priceBook.stop();
await measuredMetrics.stop();
await measuredMetrics.stop();
+ await observations.stop();
gpuMarket.stop();
gpuMarket.stop();
refusal.stop();
refusal.stop();
batches.stop();
batches.stop();
@@ -588,6 +588,26 @@ route('put', '/agent-role-defaults/{role}', {
@@ -588,6 +588,26 @@ route('put', '/agent-role-defaults/{role}', {
requestBody: body({ type: 'object', properties: { model: str(), choice_order: arrayOf(str()), effort: effortFields.effort } }),
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]) },
responses: { 200: json('Saved default', { type: 'object' }), ...errors([401, 403, 422]) },
});
});
+const measuredCell = { type: 'object', properties: {
+ day: str({ format: 'date' }), model: str(), provider: str(), task_kind: str(), runs: integer({ minimum: 20 }),
+ ...Object.fromEntries(require('./exchange/metrics-read').MEASURES.map((m) => [m, nullable('number')])),
+ samples: { type: 'object', additionalProperties: nullable('integer', { minimum: 20 }) }, computed_at: str({ format: 'date-time' }),
+} };
+for (const [routePath, id, summary, parameters, properties] of [
+ ['/exchange/metrics/models/{model}', 'exchangeModelMetrics', 'Measured model series over 30 or 90 completed UTC days',
+ [path('model', str()), query('days', integer({ enum: [30, 90] })), query('kind', str())], { series: arrayOf(measuredCell) }],
+ ['/exchange/metrics/compare', 'exchangeCompareMetrics', 'Compare up to six models on chosen measures for one completed UTC day',
+ [query('models', str()), query('measures', str()), query('kind', str()), query('day', str({ format: 'date' }))],
+ { models: arrayOf({ type: 'object', properties: { model: str(), cells: arrayOf(measuredCell) } }) }],
+ ['/exchange/metrics/leaderboard', 'exchangeMetricsLeaderboard', 'A task kind leaderboard for one measure and completed UTC day',
+ [query('kind', str()), query('measure', str({ enum: require('./exchange/metrics-read').MEASURES })), query('day', str({ format: 'date' }))], { rows: arrayOf(measuredCell) }],
+ ['/exchange/metrics/prices/{model}', 'exchangeModelPriceHistory', 'Daily listed catalogue prices per model and provider',
+ [path('model', str()), query('days', integer({ enum: [30, 90] }))], { series: arrayOf({ type: 'object' }) }],
+ ['/exchange/metrics/providers/{provider}', 'exchangeProviderHealth', 'Seven days of hourly provider run success and median latency',
+ [path('provider', str())], { series: arrayOf({ type: 'object', properties: { hour: str({ format: 'date-time' }), provider: str(), runs: integer({ minimum: 20 }), success_rate: { type: 'number' }, latency_p50_ms: nullable('number'), latency_samples: nullable('integer', { minimum: 20 }) } }) }],
+]) route('get', routePath, { operationId: id, summary, tags: ['exchange'], security: [], parameters,
+ description: 'Public, cached sixty seconds, CORS enabled. Measured cells and each timing measure require twenty observations. Missing is not zero; prices are listed prices, not measurements. No person, prompt or site identifiers.',
+ responses: { 200: json(summary, { type: 'object', properties }), ...errors([422]) } });
route('get', '/exchange/metrics/status', {
route('get', '/exchange/metrics/status', {
operationId: 'exchangeMetricsStatus', summary: 'Daily measured metrics publication status', tags: ['exchange'], security: [],
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.',
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.',
@@ -33,6 +33,8 @@ t.test('private and synthetic providers cannot form published cells', () => {
@@ -33,6 +33,8 @@ t.test('private and synthetic providers cannot form published cells', () => {
{ ...providers[0], kind: 'stub' },
{ ...providers[0], kind: 'stub' },
];
];
for (const provider of privateProviders) assert.deepStrictEqual(publicOffers([provider], models), []);
for (const provider of privateProviders) assert.deepStrictEqual(publicOffers([provider], models), []);
+ assert.deepStrictEqual(publicOffers(providers, [{ ...models[0], metadata: { variant: 'abliterated' } }]), []);
+ assert.deepStrictEqual(publicOffers(providers, [{ ...models[0], metadata: { id: 'acme/uncensored' } }]), []);
assert.equal(aggregate([{ ...row, provider: 'byo:private' }, ...Array(19).fill(row)], offers).length, 0);
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 () => {
t.test('durable due job backfills and survives restarts without losing older days', async () => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,77 @@
@@ -0,0 +1,77 @@
+'use strict';
+const assert = require('assert');
+const { suite } = require('./helpers');
+const { aggregate } = require('../server/exchange/metrics-rollup');
+const { healthCells, createObservations } = require('../server/exchange/metrics-observations');
+const t = suite('metrics-reads');
+const now = Date.parse('2026-11-10T12:30:00Z');
+const clock = { now: () => now };
+const offers = [{ key: 'openai', provider: 'openai', keyModel: 'gpt-test', model: 'openai/gpt-test' }];
+const row = { provider: 'openai', model: 'gpt-test', workflow: 'ai.complete', status: 'succeeded', latency_ms: 1000, ttft_ms: 200, tokens_out: 80, cost_usd: 0.01, hour: '2026-11-10T11:00:00Z' };
+let testdb, server, base;
+async function get(path, opts = {}) { const res = await fetch(base + '/api/v1/exchange/metrics' + path, opts); const text = await res.text(); return { status: res.status, headers: res.headers, body: text ? JSON.parse(text) : null }; }
+t.test('boot with published daily evidence for two models', async () => {
+ testdb = await require('./db').testDb();
+ const db = testdb.db;
+ const metrics = aggregate(Array(20).fill(row), offers)[0].metrics;
+ for (const [model, day, latency] of [['openai/gpt-test', '2026-11-09', 1000], ['openai/gpt-other', '2026-11-09', 2000], ['openai/gpt-test', '2026-09-10', 3000]]) {
+ await db.prepare(`INSERT INTO exchange_metrics_daily (day, model, provider, task_kind, runs, metrics, computed_at)
+ VALUES (?::date, ?, 'openai', 'chat', 20, ?::jsonb, ?::timestamptz)`).run(day, model, JSON.stringify({ ...metrics, latency_p50_ms: latency }), new Date(now).toISOString());
+ }
+ const app = require('express')();
+ app.use(require('../server/api/measured-metrics').measuredMetricsRouter({ db, clock, log: { error() {} } }));
+ server = app.listen(0, '127.0.0.1');
+ await new Promise((r) => server.once('listening', r));
+ base = `http://127.0.0.1:${server.address().port}`;
+});
+t.test('30/90-day series, six-model bound, selected measures, same-day ranking and CORS', async () => {
+ assert.equal((await get('/models/openai%2Fgpt-test?days=30')).body.series.length, 1);
+ assert.equal((await get('/models/openai%2Fgpt-test?days=90')).body.series.length, 2);
+ const comparison = await get('/compare?models=openai/gpt-test,openai/gpt-other&kind=chat&measures=latency_p50_ms');
+ assert.equal(comparison.status, 200);
+ assert.equal(comparison.headers.get('access-control-allow-origin'), '*');
+ assert.equal(comparison.headers.get('cache-control'), 'public, max-age=60');
+ assert.equal(comparison.body.day, '2026-11-09');
+ assert.deepEqual(comparison.body.models.map((m) => m.cells[0].latency_p50_ms), [1000, 2000]);
+ assert(!Object.hasOwn(comparison.body.models[0].cells[0], 'cost_per_run_usd'));
+ assert.equal((await get('/leaderboard?kind=chat&measure=latency_p50_ms')).body.rows[0].model, 'openai/gpt-test');
+ const preflight = await fetch(base + '/api/v1/exchange/metrics/compare', { method: 'OPTIONS', headers: { Origin: 'https://example.test' } });
+ assert.equal(preflight.status, 204);
+ assert.equal(preflight.headers.get('access-control-allow-origin'), '*');
+ for (const path of ['/compare?models=a,b,c,d,e,f,g', '/compare?models=a&measures=made_up', '/models/a?days=365', '/leaderboard?kind=private-site', '/compare?models=a&day=2026-02-31']) {
+ const invalid = await get(path);
+ assert.equal(invalid.status, 422, path);
+ assert.equal(invalid.headers.get('cache-control'), 'no-store');
+ }
+});
+t.test('hourly health withholds sparse cohorts and sparse latency separately', () => {
+ assert.deepEqual(healthCells(Array(19).fill(row), offers), []);
+ const health = healthCells([...Array(19).fill(row), { ...row, status: 'failed' }], offers)[0];
+ assert.equal(health.runs, 20);
+ assert.equal(health.success_rate, 0.95);
+ assert.equal(health.latency_p50_ms, null);
+ assert.equal(health.latency_samples, null);
+});
+t.test('hour job records receipts and daily catalogue prices once across restarts', async () => {
+ const db = testdb.db;
+ for (let i = 0; i < 20; i++) {
+ const id = `health_${i}`;
+ await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, created_at)
+ VALUES (?, 'ai.complete', 1, 'succeeded', 'service', 'private', '{}', 'hash', ?)`)
+ .run(id, row.hour);
+ await db.prepare('INSERT INTO ai_receipts (id, run_id, body, created_at) VALUES (?, ?, ?::jsonb, ?::timestamptz)').run(id, id,
+ JSON.stringify({ offer: { provider: row.provider, model: row.model }, latency_ms: row.latency_ms }), row.hour);
+ }
+ const registry = { listProviders: async () => [{ key: 'openai', kind: 'openai' }], listModels: async () => [{ provider_key: 'openai', model_key: 'gpt-test', metadata: { id: 'openai/gpt-test' } }] };
+ const catalog = { models: async () => ({ models: [{ slug: 'openai/gpt-test', offers: [{ provider: 'openai', model: 'gpt-test', price: { in: 0.1, out: 0.5 }, price_source: 'https://example.test/prices' }] }] }) };
+ assert.deepEqual(await createObservations({ db, registry, catalog, clock }).tick(), { hours: 1, prices: 1 });
+ assert.equal((await createObservations({ db, registry, catalog, clock }).tick()).skipped, 'not_due');
+ const prices = await get('/prices/openai%2Fgpt-test');
+ assert.equal(prices.body.series[0].price_out, 0.5);
+ const health = await get('/providers/openai');
+ assert.equal(health.body.series.length, 1);
+ assert.equal(health.body.series[0].latency_p50_ms, 1000);
+ assert(!JSON.stringify(health.body).includes('private'));
+});
+t.test('close', async () => { if (server) await new Promise((r) => server.close(r)); if (testdb) await testdb.close(); });
+t.run();