openvibers/openvibe.ai Commits

Exchange catalog: uptime from real attempts, 20 samples before a percentage (#353)

OP OpenVibers committed 6 days ago · 7a82900042 · 3 files, +179 −38

A run marked failed after its provider answered (output.empty) was counted as a
failure and its ok attempt was thrown away: production's 8 DeepSeek Flash answers
were all runs failed by validation, so Flash read 0 succeeded / 8 failed = 0.0% up.

Fold one sample per real provider attempt instead: requests.status ok is a success,
error/timeout a failure, skipped/cancelled not a sample. A failed run with no attempt
row (refusal before routing, quota, validation) counts against no offer, and a failed
attempt counts against the provider and model its request row names. Cached answers
stay apart, following cached_from.

Uptime and speed are null below 20 real attempts in the window; requests_7d still
reports the count. The Exchange already omits a null uptime.

Changes

ddiff --git a/server/exchange/observed.js b/server/exchange/observed.js
iindex 07dd5aa..d30ece1 100644
---- a/server/exchange/observed.js
++++ b/server/exchange/observed.js
@@@ -1,20 +1,28 @@
'use strict';
/**
-- * What the Exchange catalog measures: uptime, p50/p95 and request counts from the runs that actually
-- * happened, never from a rollup keyed by a different model id.
++ * What the Exchange catalog measures: uptime, p50/p95 and request counts from the provider attempts
++ * that actually happened, never from a rollup keyed by a different model id.
*
* A registry model row names the model ("gpt-5-nano"); a run records the id its upstream resolved
* ("gpt-5-nano-2025-08-07"). A run belongs to the longest registry key it extends by a revision —
* a date, a version or a tag. `gpt-5` therefore never collects `gpt-5-mini-2025-08-07`; the mini row
* does, by exact name. Cached answers carry no provider of their own, so they follow the run they
-- * reused (`cached_from`). A failed run answered nothing, so the provider and model it last tried are
-- * read from its request rows.
++ * reused (`cached_from`).
++ *
++ * Uptime is per real provider attempt, not per run status. A failed run counts against an offer only
++ * when an attempt row names that provider and model; a run refused before routing, over quota or by
++ * validation has no such row and counts against nothing. An attempt that answered (status ok) is a
++ * success even when the run afterwards failed validation (`output.empty` / `output.invalid`): the
++ * offer did answer. A skipped or cancelled row was never an answer and is not a sample.
*
* The latency of an answer is its time to the first token when the receipt recorded one (a streamed
-- * answer), else the total provider latency of the attempt that answered. No samples means null, not 0.
++ * answer), else the total provider latency of the attempt that answered. Uptime and speed need enough
++ * samples: 20 real attempts in the window. Below that both are null while `requests_7d` still says how
++ * many attempts there were. No samples means null, not 0.
* Nothing here is per person, and neither a prompt nor an answer is read.
*/
const DAY_MS = 86400000;
++const MIN_SAMPLES = 20;
const dayOf = (ms) => new Date(ms).toISOString().slice(0, 10);
// What an upstream appends to a registry name: -2025-08-07, -v3-0324, -4-5-20251001, :free, @rev.
@@@ -52,40 +60,50 @@ function percentile(values, p) {
}
function createObserved({ db, clock = { now: () => Date.now() } }) {
-- // Counted per day, status and effective offer. A cached run's provider is the run it reused; a
-- // failed run's is its last attempt (only a provider call can fail an offer — a refused request
-- // never reached one).
-- const runRows = db.prepare(`SELECT to_char(r.created_at::timestamptz AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day, r.status,
-- COALESCE(r.provider_key, src.provider_key, a.provider_key) AS provider_key,
-- COALESCE(r.model_key, src.model_key, a.model_key) AS model_key,
++ // One sample per real provider attempt: ok is a success, error or timeout is a failure. Skipped and
++ // cancelled rows are not samples (no provider was tried, or the caller left). The run's own provider
++ // and model are not used here: a fallback run can name several offers, and a failed run keeps none.
++ const attemptRows = db.prepare(`SELECT to_char(r.created_at::timestamptz AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day,
++ CASE q.status WHEN 'ok' THEN 'succeeded' ELSE 'failed' END AS status,
++ q.provider_key AS provider_key, q.model_key AS model_key,
++ COUNT(*)::bigint AS n
++ FROM requests q
++ JOIN runs r ON r.id = q.run_id
++ WHERE q.status IN ('ok', 'error', 'timeout')
++ AND r.status IN ('succeeded', 'failed')
++ AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz
++ GROUP BY 1, 2, 3, 4`);
++ // A cached answer called no provider; it follows the run it reused (`cached_from`).
++ const cachedRows = db.prepare(`SELECT to_char(r.created_at::timestamptz AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day,
++ 'cached' AS status,
++ COALESCE(r.provider_key, src.provider_key) AS provider_key,
++ COALESCE(r.model_key, src.model_key) AS model_key,
COUNT(*)::bigint AS n
FROM runs r
LEFT JOIN runs src ON src.id = r.cached_from
-- LEFT JOIN LATERAL (
-- SELECT q.provider_key, q.model_key FROM requests q
-- WHERE q.run_id = r.id AND q.status <> 'skipped'
-- ORDER BY q.seq DESC LIMIT 1
-- ) a ON r.status = 'failed'
-- WHERE r.status IN ('succeeded', 'failed', 'cached')
++ WHERE r.status = 'cached'
AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz
GROUP BY 1, 2, 3, 4`);
-- const latencyRows = db.prepare(`SELECT COALESCE(r.provider_key, q.provider_key) AS provider_key,
-- COALESCE(r.model_key, q.model_key) AS model_key,
++ const latencyRows = db.prepare(`SELECT q.provider_key AS provider_key,
++ q.model_key AS model_key,
q.latency_ms, rc.body->>'ttft_ms' AS ttft_ms
-- FROM runs r
-- JOIN requests q ON q.run_id = r.id AND q.status = 'ok'
++ FROM requests q
++ JOIN runs r ON r.id = q.run_id
LEFT JOIN ai_receipts rc ON rc.run_id = r.id
-- WHERE r.status = 'succeeded' AND q.latency_ms IS NOT NULL
++ WHERE q.status = 'ok' AND q.latency_ms IS NOT NULL
++ AND r.status IN ('succeeded', 'failed')
AND r.created_at::timestamptz >= ?::timestamptz AND r.created_at::timestamptz <= ?::timestamptz`);
-- /** The real runs and answering attempts of the window, and the UTC day that is "today". */
++ /** The real attempts and cached answers of the window, and the UTC day that is "today". */
async function read({ days = 7 } = {}) {
const now = clock.now();
const span = Math.max(1, Number(days) || 7);
const from = new Date(now - span * DAY_MS).toISOString();
const to = new Date(now).toISOString();
-- const [runs, latencies] = await Promise.all([runRows.all(from, to), latencyRows.all(from, to)]);
-- return { from, to, today: dayOf(now), runs, latencies };
++ const [attempts, cached, latencies] = await Promise.all([
++ attemptRows.all(from, to), cachedRows.all(from, to), latencyRows.all(from, to),
++ ]);
++ return { from, to, today: dayOf(now), runs: [...attempts, ...cached], latencies };
}
return { read };
@@@ -143,8 +161,11 @@ function fold(observed, drafts) {
}
for (const [d, s] of stats) {
const real = s.succeeded + s.failed;
-- d.offer.speed = { p50_ms: s.p50, p95_ms: s.p95 };
-- d.offer.uptime = real > 0 ? s.succeeded / real : null;
++ // A measured number needs enough samples to mean something: below 20 real attempts the catalog
++ // says "not measured yet" (null) but still reports how many attempts there were.
++ const enough = real >= MIN_SAMPLES;
++ d.offer.speed = { p50_ms: enough ? s.p50 : null, p95_ms: enough ? s.p95 : null };
++ d.offer.uptime = enough ? s.succeeded / real : null;
d.offer.requests_7d = real;
d.offer.requests_today = s.succeeded_today + s.failed_today;
d.offer.cached_7d = s.cached;
@@@ -153,4 +174,4 @@ function fold(observed, drafts) {
return stats;
}
--module.exports = { createObserved, fold, modelMatches, percentile };
++module.exports = { createObserved, fold, modelMatches, percentile, MIN_SAMPLES };
ddiff --git a/test/exchange-observed.test.js b/test/exchange-observed.test.js
nnew file mode 100644
iindex 0000000..6f795c3
---- /dev/null
++++ b/test/exchange-observed.test.js
@@@ -0,0 +1,120 @@
++'use strict';
++// The catalog's uptime is per real provider attempt, not per run status: a failed run counts against an
++// offer only when an attempt row names it, and an ok attempt is a success even when the run afterwards
++// failed validation (output.empty). Uptime and speed need 20 samples; below that they are null.
++const assert = require('assert');
++const { boot, request, suite } = require('./helpers');
++
++const t = suite('exchange-observed');
++let h;
++let nowMs = Date.now();
++let seq = 0;
++// The catalog read is memoised for 30 s per kind; advancing this clock past that makes each test read
++// what it just inserted.
++const clock = { now: () => nowMs };
++const at = (n) => new Date(nowMs - n * 86400000).toISOString();
++const uid = (p) => `${p}_${Date.now()}_${seq++}`;
++
++async function insertRun({ id = uid('run'), status = 'succeeded', provider = null, model = null, created = at(0) } = {}) {
++ await h.db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash,
++ provider_key, model_key, created_at, started_at, finished_at)
++ VALUES (?, 'ai.complete', 1, ?, 'service', 'test', '{}', 'input-hash', ?, ?, ?, ?, ?)`)
++ .run(id, status, provider, model, created, created, created);
++ return id;
++}
++async function insertRequest({ runId, provider = 'deepseek', model = 'deepseek-flash', status = 'ok', latency = 100, created = at(0), seqNo = 0 }) {
++ await h.db.prepare(`INSERT INTO requests (run_id, seq, operation, provider_key, model_key, status, latency_ms, created_at)
++ VALUES (?, ?, 'chat', ?, ?, ?, ?, ?)`).run(runId, seqNo, provider, model, status, latency, created);
++}
++async function reset() {
++ await h.db.prepare('DELETE FROM requests').run();
++ await h.db.prepare('DELETE FROM runs').run();
++}
++async function catalog() {
++ nowMs += 31000;
++ const r = await request(h.base, 'GET', '/api/v1/exchange/catalog');
++ assert.strictEqual(r.status, 200, r.text);
++ return r.body;
++}
++function flash(body) {
++ const m = body.models.find((x) => x.slug === 'deepseek/deepseek-flash');
++ assert.ok(m, 'the Flash model is an offer');
++ const offer = m.offers.find((o) => o.provider === 'deepseek');
++ assert.ok(offer, 'DeepSeek offers Flash');
++ return offer;
++}
++
++t.test('boot with DeepSeek configured', async () => {
++ h = await boot({ clock, env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER: 'openai', AI_API_KEY: 'test-openai-key', DEEPSEEK_KEY: 'test-deepseek-key', DEEPSEEK_API_KEY: 'test-deepseek-key' } });
++ await h.registry.upsertProvider({
++ key: 'deepseek', display_name: 'DeepSeek', kind: 'openai', auth_mode: 'bearer', secret_ref: 'env:DEEPSEEK_KEY',
++ base_url: 'https://api.deepseek.test', capabilities: ['chat', 'json', 'tools'], provider_access: 'OWN_DEPLOYMENT',
++ });
++ await h.registry.upsertModel({ provider_key: 'deepseek', model_key: 'deepseek-flash', type: 'chat', context_window: 131072,
++ cost: { in_per_mtok: 0.3, out_per_mtok: 1.2 } });
++});
++
++t.test('a failed run that reached no provider counts against no offer', async () => {
++ await reset();
++ // The production shape: refusals before routing keep no provider and no request row.
++ for (let i = 0; i < 50; i++) await insertRun({ status: 'failed' });
++ // A route that refused the offer logs only a skipped row: no provider was tried.
++ const skipped = await insertRun({ status: 'failed' });
++ await insertRequest({ runId: skipped, status: 'skipped' });
++
++ const body = await catalog();
++ assert.strictEqual(flash(body).requests_7d, 0);
++ assert.strictEqual(flash(body).uptime, null);
++ const total = body.models.reduce((n, m) => n + m.offers.reduce((a, o) => a + o.requests_7d, 0), 0);
++ assert.strictEqual(total, 0, 'no offer counts a run that never reached a provider');
++});
++
++t.test('a failed run counts against the offer its failed attempt names', async () => {
++ await reset();
++ const failed = await insertRun({ status: 'failed' });
++ await insertRequest({ runId: failed, status: 'error', latency: 900 });
++ for (let i = 0; i < 19; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id });
++ }
++
++ const offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 20);
++ assert.strictEqual(offer.uptime, 0.95, '19 answered attempts and 1 failed attempt');
++});
++
++t.test('an ok attempt is a success even when its run failed validation', async () => {
++ await reset();
++ // Every run failed with output.empty, but every provider attempt answered ok (production's 8 Flash runs).
++ for (let i = 0; i < 20; i++) {
++ const id = await insertRun({ status: 'failed' });
++ await insertRequest({ runId: id });
++ }
++
++ const offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 20);
++ assert.strictEqual(offer.uptime, 1, '20 answered attempts, 0 failed attempts');
++});
++
++t.test('8 successes and 0 real failures are 100% only once 20 samples exist', async () => {
++ await reset();
++ for (let i = 0; i < 8; i++) {
++ const id = await insertRun({ status: 'failed' });
++ await insertRequest({ runId: id });
++ }
++ let offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 8);
++ assert.strictEqual(offer.uptime, null, '8 answers is below the 20-sample floor');
++ assert.deepStrictEqual(offer.speed, { p50_ms: null, p95_ms: null });
++
++ for (let i = 0; i < 12; i++) {
++ const id = await insertRun();
++ await insertRequest({ runId: id });
++ }
++ offer = flash(await catalog());
++ assert.strictEqual(offer.requests_7d, 20);
++ assert.strictEqual(offer.uptime, 1);
++ assert.deepStrictEqual(offer.speed, { p50_ms: 100, p95_ms: 100 });
++});
++
++t.run();
ddiff --git a/test/exchange-read.test.js b/test/exchange-read.test.js
iindex 37dff31..d17d438 100644
---- a/test/exchange-read.test.js
++++ b/test/exchange-read.test.js
@@@ -62,11 +62,11 @@ t.test('catalog groups the same model across providers, with prices, speed, upti
await h.registry.upsertModel({ provider_key: 'deepinfra', model_key: 'deepseek-chat', type: 'chat', context_window: 131072,
cost: { in_per_mtok: 0.20, out_per_mtok: 0.30 } });
// Real runs name the id the upstream resolved ("deepseek-chat-2026-01-01"); the registry row names the
-- // model ("deepseek-chat"). 8 answers, 2 failures attributed through their last attempt, 3 cache hits
-- // whose provider is the run they reused. One answer recorded a TTFT (streamed); the rest have only the
-- // attempt's total latency.
++ // model ("deepseek-chat"). 18 answered attempts, 2 failed attempts, 3 cache hits whose provider is the
++ // run they reused. One answer recorded a TTFT (streamed); the rest have only the attempt's total
++ // latency. 20 real attempts is the floor at which uptime and speed are shown.
const answered = [];
-- for (let i = 0; i < 8; i++) {
++ for (let i = 0; i < 18; i++) {
const id = await insertRun({ provider: 'deepseek', model: 'deepseek-chat-2026-01-01' });
answered.push(id);
await insertRequest({ runId: id, provider: 'deepseek', model: 'deepseek-chat-2026-01-01', latency: (i + 1) * 100 });
@@@ -89,18 +89,18 @@ t.test('catalog groups the same model across providers, with prices, speed, upti
const [cheap, dear] = m.offers;
assert.strictEqual(cheap.provider, 'deepinfra', 'sorted by price');
assert.deepStrictEqual(cheap.price, { in: 0.2, out: 0.3, cached: 0.02, per: '1M tokens', currency: 'USD' });
-- // The deepseek offer: p50/p95 over [100,200,300,400,500,600,700,800] with 400 replaced by the recorded TTFT 50.
-- assert.strictEqual(dear.uptime, 0.8, '8 succeeded / (8 + 2 failed), from runs');
-- assert.strictEqual(dear.requests_7d, 10, 'real runs only; the 3 cached answers are counted apart');
++ // The deepseek offer: p50/p95 over [100..1800] with 400 replaced by the recorded TTFT 50.
++ assert.strictEqual(dear.uptime, 0.9, '18 answered / (18 + 2 failed), from attempts');
++ assert.strictEqual(dear.requests_7d, 20, 'real attempts only; the 3 cached answers are counted apart');
assert.strictEqual(dear.cached_7d, 3);
-- assert.deepStrictEqual(dear.speed, { p50_ms: 400, p95_ms: 765 }, 'TTFT where recorded, else total');
-- assert.strictEqual(dear.requests_today, 10);
++ assert.deepStrictEqual(dear.speed, { p50_ms: 950, p95_ms: 1715 }, 'TTFT where recorded, else total');
++ assert.strictEqual(dear.requests_today, 20);
assert.strictEqual(cheap.uptime, null, 'no data is unknown, not 0');
assert.strictEqual(cheap.requests_7d, 0);
assert.deepStrictEqual(cheap.speed, { p50_ms: null, p95_ms: null });
assert.deepStrictEqual(dear.data_policy, { trains: false, retention_days: 0, region: 'cn', anonymized: null });
assert.strictEqual(cheap.data_policy && cheap.data_policy.trains, false, 'DeepInfra carries its published policy (seeded by AI #307): no training');
-- assert.strictEqual(m.requests_7d, 10);
++ assert.strictEqual(m.requests_7d, 20);
assert.strictEqual(m.cached_7d, 3);
});