openvibers/openvibe.ai Commits

Record receipt timing and recover historical measured latency (#440)

OP OpenVibers committed 3 days ago · 1fe043af80 · 9 files, +74 −22

* Record run timing and backfill latency from retained run timestamps

* Merge main before shipping receipt timing

Changes

ddiff --git a/docs/measured-metrics.md b/docs/measured-metrics.md
iindex a501b4c..4435946 100644
---- a/docs/measured-metrics.md
++++ b/docs/measured-metrics.md
@@@ -16,6 +16,12 @@ answering or final attempted offer. It is run health, not individual fallback at
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.
++Older receipts omitted timing because the workflow result did not forward provider
++timings. For those receipts, total latency is the recorded run finish minus start;
++missing or invalid timestamps stay null. New receipts record that same run-boundary
++duration and, for streaming, the time of the first content/reasoning/tool-argument
++delta. No historical first-token time is invented. Job algorithm version 2 recomputes
++retained daily and hourly cells once, without manual production writes.
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.
ddiff --git a/migrations/0071_metrics_timing_version.sql b/migrations/0071_metrics_timing_version.sql
nnew file mode 100644
iindex 0000000..d30b62a
---- /dev/null
++++ b/migrations/0071_metrics_timing_version.sql
@@@ -0,0 +1,4 @@
++-- phase: expand
++-- Versioned jobs backfill retained run timestamps once when receipt timing improves.
++ALTER TABLE exchange_metrics_job ADD COLUMN IF NOT EXISTS algorithm_version integer NOT NULL DEFAULT 1;
++ALTER TABLE exchange_observations_job ADD COLUMN IF NOT EXISTS algorithm_version integer NOT NULL DEFAULT 1;
ddiff --git a/server/exchange/metrics-observations.js b/server/exchange/metrics-observations.js
iindex 4df15c4..2c5b7c6 100644
---- a/server/exchange/metrics-observations.js
++++ b/server/exchange/metrics-observations.js
@@@ -1,6 +1,7 @@
'use strict';
const { publicOffers, quantile, DAY, MIN_SAMPLES } = require('./metrics-rollup');
const { modelMatches } = require('./observed');
++const { latencyOf } = require('./timing');
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);
@@@ -17,7 +18,7 @@ function healthCells(rows, offers) {
cell.runs++;
if (row.status === 'succeeded') {
cell.successes++;
-- const latency = numeric(row.latency_ms);
++ const latency = latencyOf(row);
if (latency != null) cell.latency.push(latency);
}
cells.set(key, cell);
@@@ -32,16 +33,16 @@ function createObservations({ db, registry, catalog, clock = { now: () => Date.n
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));
++ const state = await db.prepare('SELECT * FROM exchange_observations_job WHERE id = 1 AND (next_due <= ?::timestamptz OR algorithm_version < 2) 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 from = state.completed_hour && state.algorithm_version >= 2 ? 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
++ r.status, r.started_at, r.finished_at, 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);
@@@ -62,7 +63,7 @@ function createObservations({ db, registry, catalog, clock = { now: () => Date.n
prices += out.changes;
}
}
-- await db.prepare('UPDATE exchange_observations_job SET next_due = ?::timestamptz, completed_hour = ?::timestamptz WHERE id = 1')
++ await db.prepare('UPDATE exchange_observations_job SET algorithm_version = 2, next_due = ?::timestamptz, completed_hour = ?::timestamptz WHERE id = 1')
.run(iso(end + HOUR + 5 * 60000), iso(end - HOUR));
return { hours: cells.length, prices };
});
ddiff --git a/server/exchange/metrics-rollup.js b/server/exchange/metrics-rollup.js
iindex 68b211e..764ca97 100644
---- a/server/exchange/metrics-rollup.js
++++ b/server/exchange/metrics-rollup.js
@@@ -3,6 +3,7 @@
const { canonicalOf, slugOf } = require('./catalog');
const { modelMatches } = require('./observed');
const { upstreamOf } = require('../providers/upstream');
++const { latencyOf } = require('./timing');
const DAY = 86400000;
const MIN_SAMPLES = 20;
const TASK_KINDS = ['chat', 'code', 'design', 'search', 'vision', 'speech', 'embedding', 'image', 'video', 'other'];
@@@ -60,7 +61,7 @@ function aggregate(rows, offers) {
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);
++ const latency = latencyOf(row), 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));
@@@ -86,7 +87,7 @@ function aggregate(rows, offers) {
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->>'kind' AS kind, c.body->>'workflow' AS workflow, r.status, r.started_at, r.finished_at,
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
@@@ -96,7 +97,7 @@ function createMetricsRollup({ db, registry, clock = { now: () => Date.now() },
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));
++ const state = await db.prepare('SELECT * FROM exchange_metrics_job WHERE id = 1 AND (next_due <= ?::timestamptz OR algorithm_version < 2) 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' };
@@@ -105,7 +106,7 @@ function createMetricsRollup({ db, registry, clock = { now: () => Date.now() },
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);
++ const from = through == null || state.algorithm_version < 2 ? retainedFrom : Math.max(retainedFrom, through - DAY);
let days = 0, cells = 0;
for (let start = from; start < today; start += DAY) {
const day = dayOf(start);
@@@ -117,7 +118,7 @@ function createMetricsRollup({ db, registry, clock = { now: () => Date.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')
++ await db.prepare('UPDATE exchange_metrics_job SET algorithm_version = 2, 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) };
});
ddiff --git a/server/exchange/timing.js b/server/exchange/timing.js
nnew file mode 100644
iindex 0000000..4be15e7
---- /dev/null
++++ b/server/exchange/timing.js
@@@ -0,0 +1,12 @@
++'use strict';
++function durationMs(started, finished) {
++ const time = (v) => typeof v === 'number' ? v : v == null ? NaN : Date.parse(v);
++ const start = time(started), end = time(finished);
++ return Number.isFinite(start) && Number.isFinite(end) && end >= start ? end - start : null;
++}
++function latencyOf(row) {
++ const value = row.latency_ms;
++ if (value != null && value !== '' && Number.isFinite(Number(value)) && Number(value) >= 0) return Number(value);
++ return durationMs(row.started_at, row.finished_at);
++}
++module.exports = { durationMs, latencyOf };
ddiff --git a/server/runs.js b/server/runs.js
iindex 6caaff3..da0fe51 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -826,7 +826,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
explain, usage, price: prices, privacy: privacy.view(label),
payer: { paid_by: paidBy, detail: (payer && payer.detail) || null, fee_rate: feeRate, fee_usd: money ? money.fee_usd : 0 },
preset: opts.preset || null,
-- latencyMs, ttftMs,
++ latencyMs: latencyMs ?? require('./exchange/timing').durationMs(row.started_at, row.finished_at), ttftMs,
provider: answerProvider, model: answerModel,
createdAt: row.finished_at || new Date(clock.now()).toISOString(),
});
@@@ -916,7 +916,8 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
async function execute(id, { controller, reserved, scope, ctx, input, hold = null }) {
const row = await getRow(id);
if (!row || row.status !== 'queued') { inflight.delete(id); return; }
-- await db.prepare("UPDATE runs SET status = 'running', started_at = ? WHERE id = ? AND status = 'queued'").run(iso(clock.now()), id);
++ const startedAt = clock.now();
++ await db.prepare("UPDATE runs SET status = 'running', started_at = ? WHERE id = ? AND status = 'queued'").run(iso(startedAt), id);
const wf = await registry.getWorkflow(row.workflow_key, row.workflow_version);
const opts = parseJson(row.options, {});
let seq = 0;
@@@ -936,6 +937,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
};
let credit = null;
let billingDelta = null;
++ let firstDeltaAt = null;
let onRoute = null;
try {
// AI Exchange F: hold prepaid Balance before the first managed provider call (AI_CREDITS, off by
@@@ -950,7 +952,13 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
ctx.creditsRequired = !credit && !ctx.byok && !ctx.serviceBudgeted;
}
billingDelta = typeof ctx.onDelta === 'function'
-- ? (credit ? (d) => { credit.observe(d); ctx.onDelta(d); } : ctx.onDelta)
++ ? (d) => {
++ const delta = d && d.delta;
++ if (firstDeltaAt === null && (delta?.content || delta?.reasoning_content || delta?.thinking
++ || delta?.tool_calls?.some((c) => c.function?.arguments))) firstDeltaAt = clock.now();
++ if (credit) credit.observe(d);
++ ctx.onDelta(d);
++ }
: null;
// Balance holds first (a 402 must land before the provider), then the router's own disclosure hook.
onRoute = (credit || typeof ctx.onRoute === 'function')
@@@ -1068,12 +1076,15 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
catch (err) { log.warn(`[runs] ${id}: the abuse counters were not written: ${err && err.message}`); }
}
// The record of the run (L40): the tokens with the cache apart, the money and who paid.
-- const published = await publishUsage({ row: await getRow(id), outcome: 'succeeded', usage: r.usage, provider: r.provider, model: r.model, attempts, latencyMs: r.latencyMs, ttftMs: r.ttftMs });
++ const finishedRun = await getRow(id);
++ const timing = { latencyMs: require('./exchange/timing').durationMs(finishedRun.started_at, finishedRun.finished_at),
++ ttftMs: firstDeltaAt === null ? null : Math.max(0, firstDeltaAt - startedAt) };
++ const published = await publishUsage({ row: finishedRun, outcome: 'succeeded', usage: r.usage, provider: r.provider, model: r.model, attempts, ...timing });
// The receipt of the answer (server/exchange/receipts.js): the offer, the price, the tokens, the cache,
// the latency, why the router chose it, the fallbacks it would have used, the payer and the privacy label.
await writeReceipt({
-- row: await getRow(id), explain: r.explain, usage: r.usage, provider: r.provider, model: r.model,
-- latencyMs: r.latencyMs, ttftMs: r.ttftMs, money: published && published.money, payer: published && published.payer,
++ row: finishedRun, explain: r.explain, usage: r.usage, provider: r.provider, model: r.model,
++ ...timing, money: published && published.money, payer: published && published.payer,
});
// What the run spent goes on the budget of its run token (Codes C3): the same tokens and cost the
// quota accounting counts, so the next request of a run over its budget is refused.
ddiff --git a/test/exchange-presets-receipts.test.js b/test/exchange-presets-receipts.test.js
iindex 9468747..db592bc 100644
---- a/test/exchange-presets-receipts.test.js
++++ b/test/exchange-presets-receipts.test.js
@@@ -213,6 +213,7 @@ t.test('every answer carries a receipt: the headers, the id, and GET /v1/receipt
assert.strictEqual(compact.offer, 'crowd/crowd-m');
assert.deepStrictEqual([compact.price.in, compact.price.out], [0.2, 0.6]);
assert.deepStrictEqual([compact.tokens.in, compact.tokens.out, compact.tokens.cached], [120, 30, 40]);
++ assert.strictEqual(typeof compact.latency_ms, 'number', 'ordinary runs record latency rather than null');
assert.ok(compact.latency_ms >= 0);
assert.strictEqual(compact.privacy, 'community', 'a community machine is labelled as one');
assert.ok(Array.isArray(compact.fallbacks) && compact.fallbacks.length, 'the fallbacks it would have used are named');
@@@ -282,6 +283,10 @@ t.test('a streamed answer names its receipt in the headers and carries it in the
const last = chunks[chunks.length - 1];
assert.ok(last.receipt, 'the last chunk carries the receipt');
assert.strictEqual(last.receipt.id, id);
++ const full = await request(h.base, 'GET', `/v1/receipts/${id}`, { tok: service });
++ assert.strictEqual(typeof full.body.receipt.latency_ms, 'number');
++ assert.strictEqual(typeof full.body.receipt.ttft_ms, 'number', 'the first content delta is timed at the run boundary');
++ assert.ok(full.body.receipt.ttft_ms <= full.body.receipt.latency_ms);
});
t.test('a preset is refused when nobody has it, and the receipt route needs a token', async () => {
ddiff --git a/test/measured-metrics.test.js b/test/measured-metrics.test.js
iindex 01c5e76..e6d448e 100644
---- a/test/measured-metrics.test.js
++++ b/test/measured-metrics.test.js
@@@ -8,6 +8,14 @@ const models = [{ provider_key: 'openai', model_key: 'gpt-test', metadata: { id:
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('historical receipts use recorded run duration and never invent first-token time', () => {
++ const historical = { ...row, latency_ms: null, ttft_ms: null, started_at: '2026-11-09T12:00:00Z', finished_at: '2026-11-09T12:00:02Z' };
++ const cell = aggregate(Array(20).fill(historical), offers)[0];
++ assert.equal(cell.metrics.latency_p50_ms, 2000);
++ assert.equal(cell.metrics.ttft_p50_ms, null);
++ assert.equal(cell.metrics.tokens_per_second_p50, null);
++ assert.equal(aggregate(Array(20).fill({ ...historical, finished_at: null }), offers)[0].metrics.latency_p50_ms, null);
++});
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' }));
@@@ -46,11 +54,11 @@ t.test('durable due job backfills and survives restarts without losing older day
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', ?, ?)`)
++ await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, synthetic, created_at, started_at, finished_at)
++ VALUES (?, 'ai.complete', 1, 'succeeded', 'service', 'private-site', '{}', 'hash', ?, ?, '2026-11-09T12:00:00Z', '2026-11-09T12:00:01Z')`)
.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,
++ offer: { provider: 'openai', model: 'gpt-test' }, workflow: 'ai.complete', latency_ms: null, ttft_ms: 200, usage: { tokens_out: 80 }, cost_usd: 0.01,
}), at);
}
const first = await createMetricsRollup(options).tick();
@@@ -72,6 +80,8 @@ t.test('durable due job backfills and survives restarts without losing older day
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('UPDATE exchange_metrics_job SET algorithm_version = 1 WHERE id = 1').run();
++ assert.equal((await createMetricsRollup(options).tick()).cells, 1, 'a new algorithm backfills retained days even before the next due time');
await db.prepare('DELETE FROM ai_receipts').run();
now += 5 * 86400000;
await createMetricsRollup(options).tick();
ddiff --git a/test/metrics-reads.test.js b/test/metrics-reads.test.js
iindex 9d9dc92..72b6e16 100644
---- a/test/metrics-reads.test.js
++++ b/test/metrics-reads.test.js
@@@ -56,16 +56,18 @@ t.test('hour job records receipts and daily catalogue prices once across restart
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', ?)`)
++ await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, created_at, started_at, finished_at)
++ VALUES (?, 'ai.complete', 1, 'succeeded', 'service', 'private', '{}', 'hash', ?, '2026-11-10T11:00:00Z', '2026-11-10T11:00:01Z')`)
.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);
++ JSON.stringify({ offer: { provider: row.provider, model: row.model }, latency_ms: null }), 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');
++ await db.prepare('UPDATE exchange_observations_job SET algorithm_version = 1 WHERE id = 1').run();
++ assert.deepEqual(await createObservations({ db, registry, catalog, clock }).tick(), { hours: 1, prices: 0 });
const prices = await get('/prices/openai%2Fgpt-test');
assert.equal(prices.body.series[0].price_out, 0.5);
const health = await get('/providers/openai');