openvibers/openvibe.ai Commits
Show each run's usage reading state in explain and free allowance on…
Changes
@@@ -149,8 +149,8 @@ needs scripts and inline styles), like Billing's staff console. Code: `server/co| `/console/providers` | every provider (kind, status, **secret reference name and whether it resolves, never a value**, circuit, base URL without query/user info, default model, priority/timeout) and the models || `/console/routes`, `/templates`, `/workflows` (`/:key[?version=]`) | every key's newest version and the version in use; per key every version with its status and content (primary and fallbacks in order, prompts, steps, schemas) || `/console/runs` | runs newest first, filtered by status (tabs), workflow, requester (`live` = `service:live`), error code and a UTC time range; failed runs grouped by error code |--| `/console/runs/:id` | versions used, provider/model, fallback, attempts, tokens, cost, error code and detail, requester/attribution/target, grounding, citations, the request log (hashes only), and the stored input/output **bounded**: strings cut at 240 characters, lists at 20 items, 6 KB in all, credential-named keys blanked; raw debug fields never |--| `/console/quotas`, `/console/usage` | active quotas with their current windows and counters, every quota; usage by day, requester, workflow and provider/model over a day range (`quotas.usage`) |++| `/console/runs/:id` | versions used, provider/model, fallback, attempts, tokens, cost, the Billing reading with its delivery state, free share and last send error, error code and detail, requester/attribution/target, grounding, citations, the request log (hashes only), and the stored input/output **bounded**: strings cut at 240 characters, lists at 20 items, 6 KB in all, credential-named keys blanked; raw debug fields never |++| `/console/quotas`, `/console/usage` | active quotas with their current windows and counters, every quota; usage by day, requester, workflow and provider/model over a day range (`quotas.usage`); the **free allowance this period** per subject × provider × metric (free used, allowance, remaining, period end; `quotas.freeAllowance.current()`); the `platform.usage-sample@1` readings by delivery state (queued, retrying, sent, failed) and the last send errors with their run. Staff only: the public `/stats` page never shows a subject || `/console/cache` | live entries and hits by workflow || `/console/audit` | the `audit_log`, filtered by kind (configuration changes by default, runs, console, everything), action, actor, target |@@@ -216,7 +216,11 @@ provider that is disabled, missing credentials, circuit-open or unable to do thefailure; a fallback that answers is recorded on the run (`fallback_used`), in the request log(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded--reason). Speech and embeddings use placement pools of registered `stt` and `embedding` models; until++reason). `GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: per++reading its `idempotency_key`, `state` (`queued` until Billing accepts it, `sent`, or `failed` when Billing++refused it), `attempts`, `queued_at`, `sent_at`, `rejected_at`, `next_attempt_at`, `last_error` (the++outbox's last send error, also while a queued reading is retried) and `free_allowance_used` when some++was free; a sent reading the relay has pruned reads `{ state: 'sent', pruned: true }`. Speech and embeddings use placement pools of registered `stt` and `embedding` models; untila second upstream exists, an unavailable sole provider answers `503 provider.unavailable`. Mediauses separate local and paid pools so paid work still requires its explicit budget. The syntheticstub can join general text routes as a last resort outside production only (`AI_STUB_FALLBACK`).@@@ -29,7 +29,7 @@],"livePatch": "applied: Live fa22de5 runs AI_SERVICE=remote in production since 2026-09-23 ~15:34 UTC (309 runs by 18:30 UTC across 13 live.* workflows and 1 network.site_copy, 0 failed). As of 2026-09-23, not moved yet: VOD captions/whisper and streamers' own provider keys stayed in Live and most Live features were passthrough prompts built by Live (since 2026-09-27 AI has live.media.transcribe, stored provider keys and versioned templates for most live.* workflows); the deployed Network still falls back to Live's /internal/ai/site-copy; scripts/import-from-live.js ran on production at 2026-09-23 18:29 UTC (79,657 ledger rows: 79,647 live.ai_usage, 2 channel_ai_config, 8 site_settings; 11 holds: provider settings and the key stay env references, 1 streamer BYO key stays in Live, 3,021 translations are unrepresentable and stay in Live); text routes (live.<role>, default.chat, default.json) are capability pools placed by openvibe-sdk/placement; live.stt, default.embedding and media.paid are single-provider and answer 503 provider.unavailable when it fails; 0 citation rows; no /metrics","deployed": true,-- "updated": "2026-09-28",++ "updated": "2026-10-04","features": ["versioned templates, workflows and routing profiles","runs with idempotency, cancel, retry, async polling and restart recovery",@@@ -42,6 +42,7 @@"ai.run.* events through the SDK outbox","ai.preferences and ai.usage_summary user modules","server-rendered operator console for Network staff",-- "Live import"++ "Live import",++ "platform.usage-sample@1 readings to Billing with a per-subject provider free allowance; each run's reading state in explain and the free allowance on the staff console"]}@@@ -6,6 +6,7 @@* GET /api/v1/runs the caller's runs (ai.run.read)* GET /api/v1/runs/:id one run (+ citations, request log metadata)* run.usage.usage_sample_id: the run's Billing reading (server/usage-samples.js), null when none++ * run.explain.usage_readings: that reading's idempotency key, delivery state (queued/sent/failed), last error, free_allowance_used* POST /api/v1/runs/:id/cancel (ai.run.create, owner)* POST /api/v1/runs/:id/retry (ai.run.create, owner) -> a new run with retry_of* GET /api/v1/runs/:id/citations (ai.run.read)@@@ -64,7 +65,7 @@ function runsRouter({ runs, registry, auth, config, log = console }) {r.get('/api/v1/runs/:id', read, async (req, res) => {try {-- const run = await owned(req);++ const run = await runs.explained(await owned(req));res.json({ run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id) });} catch (err) { sendError(res, err, req.ov, log); }});@@@ -14,7 +14,8 @@* GET /console/runs[?status&workflow&requester&code&from&to&before], /console/runs/:id* POST /console/runs/:id/cancel (ai.workflow.manage; runs.cancel)* GET /console/quotas; POST /console/quotas (writes: ai.provider.manage; quotas.upsert)-- * GET /console/usage[?from&to&requester] (quotas.usage)++ * GET /console/usage[?from&to&requester] (quotas.usage; free allowance this period, quotas.freeAllowance.current;++ * Billing readings by delivery state, usage-samples deliveryReport)* GET /console/cache; POST /console/cache/purge (writes: ai.provider.manage; ops.purgeCache)* GET /console/audit[?kind&action&actor&target_type&target_id&before]*@@@ -38,6 +39,7 @@const crypto = require('crypto');const express = require('express');const { AiError, resolveSecret } = require('../util');++const usageSamples = require('../usage-samples');const { CAPS } = require('../auth');const { staff: staffMap } = require('openvibe-contracts');const { createSessions, sameString, random } = require('./session');@@@ -323,7 +325,7 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e}));}));r.get('/console/runs/:id', read, wrap(async (req, res) => {-- const run = await runs.get(req.params.id);++ const run = await runs.explained(await runs.get(req.params.id));if (!run) return await send(res, 404, pages.message({ title: 'Not found', text: `No run ${req.params.id}.`, back: '/console/runs', ...common(req) }));return await send(res, 200, pages.run({...common(req), r: run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id), input: q.preview(run.input), output: q.preview(run.output),@@@ -354,7 +356,10 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e}));r.get('/console/usage', read, wrap(async (req, res) => {const { filters, form, bad } = q.parseUsageFilters(req.query, clock.now());-- await send(res, bad.length ? 400 : 200, pages.usage({ ...common(req), form, bad, report: await q.usageReport(quotas, filters) }));++ await send(res, bad.length ? 400 : 200, pages.usage({++ ...common(req), form, bad, report: await q.usageReport(quotas, filters),++ free: quotas.freeAllowance ? await quotas.freeAllowance.current() : [], readings: await usageSamples.deliveryReport(db),++ }));}));// ── Cache ────────────────────────────────────────────────@@@ -88,7 +88,7 @@ function secs(s) {return v < 120 ? `${v} s` : v < 7200 ? `${Math.round(v / 60)} min` : `${Math.round(v / 3600)} h`;}const clip = (s, max = 200) => { const t = String(s == null ? '' : s); return t.length > max ? `${t.slice(0, max)}…` : t; };--const GOOD = new Set(['succeeded', 'active', 'closed', 'configured', 'ok', 'cached', 'not_required']);++const GOOD = new Set(['succeeded', 'sent', 'active', 'closed', 'configured', 'ok', 'cached', 'not_required']);const BAD = new Set(['failed', 'open', 'disabled', 'missing', 'error', 'timeout', 'archived']);const badge = (s) => html`<span class="${GOOD.has(s) ? 'ok' : BAD.has(s) ? 'bad' : 'warn'}">${s || '—'}</span>`;/** A base URL without user info, query or fragment (a key must never ride along in a URL shown here). */@@@ -346,7 +346,7 @@ function run({ staff, csrf, r, citations, requests, input, output, canCancel, no<dt>Attempts</dt><dd>${n(r.usage.attempts)}</dd><dt>Tokens</dt><dd>${n(r.usage.tokens_in)} in · ${n(r.usage.tokens_out)} out</dd><dt>Cost</dt><dd>${usd(r.usage.cost_usd)}</dd>--<dt>Billing reading</dt><dd>${r.usage.usage_sample_id ? html`<code>${r.usage.usage_sample_id}</code>` : '—'}</dd>++<dt>Billing reading</dt><dd>${r.usage.usage_sample_id ? html`<code>${r.usage.usage_sample_id}</code>` : '—'}${((r.explain && r.explain.usage_readings) || []).map((u) => html` ${badge(u.state)}${u.free_allowance_used != null ? html` <span class="muted">${n(u.free_allowance_used)} free</span>` : ''}${u.last_error ? html`<br><span class="bad">${clip(u.last_error, 300)}</span>` : ''}`)}</dd>${r.error ? html`<dt>Error</dt><dd><code class="bad">${r.error.code}</code><br>${clip(r.error.detail, 500)}</dd>` : ''}</dl></div><div class="card"><dl><dt>Requester</dt><dd><code>${r.requester.type}:${r.requester.id}</code></dd>@@@ -410,7 +410,9 @@ function usageRows(caption, first, rows) {<td>${n(r.tokens_in)}</td><td>${n(r.tokens_out)}</td><td>${n(r.tokens_cached)}</td><td>${usd(r.cost_usd)}</td></tr>`), 'no usage in this range.');}--function usage({ staff, csrf, form, bad, report }) {++const day = (ms) => (ms >= 8.64e15 ? 'never' : when(new Date(ms).toISOString()));++++function usage({ staff, csrf, form, bad, report, free = [], readings }) {const t = report.total;return layout({title: 'Usage', section: 'usage', staff, csrf, error: bad.length ? `ignored malformed filter(s): ${bad.join(', ')}` : null, body: html`<h1>Usage</h1>@@@ -424,7 +426,15 @@ function usage({ staff, csrf, form, bad, report }) {${report.truncated ? html`<p class="warn">Only the first 1,000 daily rows are summed; narrow the range.</p>` : ''}<h2>Breakdown</h2>${usageRows('By day', 'Day', report.byDay)}${usageRows('By requester', 'Requester', report.byRequester)}--${usageRows('By workflow', 'Workflow', report.byWorkflow)}${usageRows('By provider and model', 'Provider / model', report.byModel)}`,++${usageRows('By workflow', 'Workflow', report.byWorkflow)}${usageRows('By provider and model', 'Provider / model', report.byModel)}++<h2>Free allowance this period</h2>++${table('Free allowance by subject, provider and metric', ['Subject', 'Provider', 'Metric', 'Free used', 'Allowance', 'Remaining', 'Period ends'], free.map((f) => html`<tr><td><code>${f.subject}</code></td><td><code>${f.provider}</code></td>++<td><code>${f.metric}</code></td><td>${n(f.free_used)}</td><td>${n(f.free_allowance)}</td><td>${n(f.remaining)}</td><td>${day(f.period_end)}</td></tr>`), 'no free allowance used this period.')}++<h2>Billing readings</h2>++${readings ? html`<p class="muted">${readings.relay ? 'Sending to Billing.' : readings.enabled ? 'Queued only: Billing sending is not configured.' : 'Readings are off (USAGE_SAMPLES=off).'}</p>++${table('Readings by delivery state', ['Queued', 'Retrying', 'Sent', 'Failed'], [html`<tr><td>${n(readings.counts.queued)}</td><td>${n(readings.counts.retrying)}</td><td>${n(readings.counts.sent)}</td><td>${n(readings.counts.failed)}</td></tr>`])}++${table('Last send errors', ['Reading', 'Run', 'State', 'Attempts', 'Queued', 'Last error'], readings.failures.map((f) => html`<tr><td><code>${f.idempotency_key}</code></td>++<td>${f.run_id ? html`<a class="mono" href="/console/runs/${enc(f.run_id)}">${f.run_id}</a>` : '—'}</td><td>${badge(f.state)}</td><td>${n(f.attempts)}</td><td>${when(f.queued_at)}</td><td>${clip(f.last_error, 300)}</td></tr>`), 'no send error.')}` : ''}`,});}@@@ -400,6 +400,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowreturn (await db.prepare(`SELECT * FROM runs ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY id DESC LIMIT ?`).all(...args, Math.min(200, Number(limit) || 50))).map(decode);}++ /** The run with its Billing readings in explain.usage_readings (server/usage-samples.js readingsFor); unchanged when it has none. */++ async function explained(run) {++ const readings = run ? await usageSamples.readingsFor(db, run.usage.usage_sample_id) : [];++ return readings.length ? { ...run, explain: { ...(run.explain || {}), usage_readings: readings } } : run;++ }++async function requestsFor(runId) {return await db.prepare('SELECT id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash, tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, created_at FROM requests WHERE run_id = ? ORDER BY seq').all(runId);}@@@ -437,7 +443,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now/** Queue depth for /metrics and /api/ready: runs waiting for a slot, and runs holding one. */function stats() { return { queued: queue.length, running: active, inflight: inflight.size, max_concurrent: config.runs.maxConcurrent, max_queued: config.runs.maxQueued }; }-- return { create, wait, get, list, cancel, retry, citations, addCitations, requestsFor, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode };++ return { create, wait, get, list, cancel, retry, citations, addCitations, requestsFor, explained, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode };}module.exports = { createRuns, entityKey, subjectKey };@@@ -20,8 +20,9 @@ const contracts = require('openvibe-contracts');const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';const REFUSED = new Set([400, 409, 413, 422]); const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;let relay = null; let on = false; let pruneTimer = null; let lastError = null; let queued = 0;++const keyOf = (runId) => `ai:${runId}:${METRIC}`;function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {-- const key = `ai:${runId}:${METRIC}`;++ const key = keyOf(runId);const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: runId, operation: workflowKey,quantity: Math.max(0, (Number(tokensIn) || 0) + (Number(tokensOut) || 0)), unit: 'tokens', at: new Date(at).toISOString(),cost_estimate: Math.max(0, Number(cost) || 0), source: 'ai.runs' };@@@ -66,6 +67,34 @@ function init(db, { enabled = true, billingUrl = '', audience = 'openvibe.billinlog.log(`[Billing] ai usage samples → ${billingUrl}`);return relay;}++// Delivery state as the outbox records it: sent, failed (Billing refused it; never retried), else queued (last_error++// set while a send is being retried).++const stateOf = (r) => (r.sent_at != null ? 'sent' : r.rejected_at != null ? 'failed' : 'queued');++const isoOrNull = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());++const envelopeOf = (e) => (typeof e === 'string' ? JSON.parse(e) : e) || {};++/**++ * A run's readings for its explain: [{ idempotency_key, state, attempts, queued_at, sent_at, rejected_at,++ * next_attempt_at, last_error, free_allowance_used? }]. A sent row the relay has pruned reads { state: 'sent', pruned: true }.++ */++async function readingsFor(db, usageSampleId) {++ if (!usageSampleId) return [];++ const r = await db.prepare(`SELECT event_id, envelope, created_at, attempts, next_attempt_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE event_id = ?`).get(usageSampleId);++ if (!r) return [{ idempotency_key: usageSampleId, state: 'sent', pruned: true }];++ const state = stateOf(r);++ const out = { idempotency_key: r.event_id, state, attempts: Number(r.attempts), queued_at: isoOrNull(r.created_at), sent_at: isoOrNull(r.sent_at), rejected_at: isoOrNull(r.rejected_at),++ next_attempt_at: state === 'queued' ? isoOrNull(r.next_attempt_at) : null, last_error: r.last_error || null };++ const free = envelopeOf(r.envelope).free_allowance_used;++ if (free != null) out.free_allowance_used = free;++ return [out];++}++/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */++async function deliveryReport(db, { failures = 10 } = {}) {++ const c = await db.prepare(`SELECT count(*) FILTER (WHERE sent_at IS NOT NULL) AS sent, count(*) FILTER (WHERE rejected_at IS NOT NULL) AS failed,++ count(*) FILTER (WHERE sent_at IS NULL AND rejected_at IS NULL) AS queued, count(*) FILTER (WHERE sent_at IS NULL AND rejected_at IS NULL AND last_error IS NOT NULL) AS retrying FROM ${TABLE}`).get();++ const rows = await db.prepare(`SELECT event_id, envelope, attempts, created_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE last_error IS NOT NULL ORDER BY id DESC LIMIT ?`).all(failures);++ return { enabled: on, relay: Boolean(relay), counts: { queued: Number(c.queued), retrying: Number(c.retrying), sent: Number(c.sent), failed: Number(c.failed) },++ failures: rows.map((r) => ({ idempotency_key: r.event_id, run_id: envelopeOf(r.envelope).resource || null, state: stateOf(r), attempts: Number(r.attempts), queued_at: isoOrNull(r.created_at), last_error: r.last_error })) };++}async function status() { if (!relay) return { enabled: on, relay: false }; return { enabled: on, relay: true, pending: Number(await relay.pending()), rejected: Number(await relay.rejected()), queued_since_boot: queued, last_error: lastError }; }/** Graceful stop (after runs.drain(), before db.close()): one last bounded pass sends what is due, then the relay stops; unsent rows wait for the next start. */async function stop({ drainMs = 3000 } = {}) {@@@ -74,4 +103,4 @@ async function stop({ drainMs = 3000 } = {}) {if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }}function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }--module.exports = { init, record, sampleOf, status, stop, _reset, TABLE };++module.exports = { init, record, sampleOf, keyOf, readingsFor, deliveryReport, status, stop, _reset, TABLE };@@@ -168,7 +168,7 @@ t.test('the staff map: admin reads and manages workflows, only the owner manages});t.test('anonymous: console pages answer the sign-in page (401), never data; writes change nothing', async () => {-- for (const p of ['/console', '/console/runs?status=failed', '/console/providers', '/console/audit', `/console/runs/${runIds.failLive}`, '/console/nope']) {++ for (const p of ['/console', '/console/runs?status=failed', '/console/providers', '/console/usage', '/console/audit', `/console/runs/${runIds.failLive}`, '/console/nope']) {const r = await get(p);assert.strictEqual(r.status, 401, p);assert.ok(r.text.includes('Sign in with OpenVibe.Network'), p);@@@ -432,6 +432,25 @@ t.test('run detail: error code, attempts, provider, request log; input bounded aassert.ok(r.text.includes('Grounding'));});++t.test('usage: free allowance per subject this period and Billing readings by state, with the last send errors', async () => {++ await h.registry.upsertModel({ provider_key: 'okseam', model_key: 'seam-model', type: 'chat', cost: { in_per_mtok: 1, out_per_mtok: 1, free_allowance: 500, reset_period: 'month' } });++ assert.strictEqual((await h.quotas.freeAllowance.claim('live:user:4242', 'okseam', 'seam-model', { in: 120 })).tokens, 120);++ const key = `ai:${runIds.ok}:tokens`;++ assert.ok(await h.db.prepare('SELECT 1 AS x FROM usage_sample_outbox WHERE event_id = ?').get(key), 'the succeeded run queued a reading');++ await h.db.prepare("UPDATE usage_sample_outbox SET attempts = 2, last_error = 'billing.usage.record answered 503' WHERE event_id = ?").run(key);++ const r = await get('/console/usage', cookies.admin);++ assertPage(r, '/console/usage');++ const row = r.text.split('</tr>').find((x) => x.includes('<code>live:user:4242</code>'));++ assert.ok(row, 'the subject has a row');++ const cells = [...row.matchAll(/<td[^>]*>([\s\S]*?)<\/td>/g)].map((m) => m[1].replace(/<[^>]+>/g, '').trim());++ assert.deepStrictEqual(cells.slice(0, 6), ['live:user:4242', 'okseam', 'input-tokens:seam-model', '120', '500', '380']);++ assert.ok(r.text.includes('Readings by delivery state') && r.text.includes(`<code>${key}</code>`) && r.text.includes('billing.usage.record answered 503'));++ const run = await get(`/console/runs/${runIds.ok}`, cookies.admin);++ assert.ok(run.text.includes(`<code>${key}</code>`) && run.text.includes('billing.usage.record answered 503'), 'the run page shows its reading\'s state and error');++ assert.strictEqual((await get('/console/usage')).status, 401, 'signed out: refused');++ assert.ok(!(await fetch(`${h.base}/stats`).then((x) => x.text())).includes('live:user:4242'), 'the public page has no subject');++});++t.test('cancel: an admin cancels a running run (ai.workflow.manage), audited with the person as actor', async () => {assert.strictEqual((await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: ops, body: { primary: { provider: 'slowseam' }, fallbacks: [] } })).status, 201);const c = await request(h.base, 'POST', '/api/v1/runs', { tok: live, body: { workflow: 'ai.generate', input: { prompt: 'slow one' }, options: { cache: false } } });@@@ -140,6 +140,14 @@ t.test('the first run is free up to the allowance per metric, the next one is fuassert.ok(Math.abs(Number(run.cost_usd) - 2) < 1e-9);});++t.test('explain shows the free share on the first run\'s reading and leaves it out on the next', async () => {++ const [first, next] = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();++ const explain = async (id) => (await request(h.base, 'GET', `/api/v1/runs/${id}`, { tok: live })).body.run.explain.usage_readings[0];++ const a = await explain(first.id);++ assert.deepStrictEqual([a.idempotency_key, a.state, a.free_allowance_used], [`ai:${first.id}:tokens`, 'queued', 200]);++ assert.ok(!('free_allowance_used' in await explain(next.id)));++});++t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });t.run();@@@ -52,9 +52,15 @@ t.test('seeded stats: one row per provider and model with p50, p95, success rateassert.match(r.text, /last 7 UTC days \(\d{4}-\d{2}-\d{2} to \d{4}-\d{2}-\d{2}\)/);});++t.test('a subject with free allowance used this period is still not on the page', async () => {++ const free = await h.quotas.freeAllowance.claim('live:user:4242', 'seam', 'seam-model', { in: 10 });++ assert.strictEqual(free.tokens, 10);++ assert.ok((await h.quotas.freeAllowance.current()).some((r) => r.subject === 'live:user:4242'), 'the staff console has the row');++});++t.test('no caller identifiers, internal URLs or secrets in the HTML', async () => {const r = await request(h.base, 'GET', '/stats');-- for (const leak of ['byo:', 'usr_owner1', 'usr_subject1', 'own-model', 'service:live', 'live:user:4242', 'default.generate', SEAM_URL, '127.0.0.1', 'SEAM_SECRET_KEY', 'env:']) {++ for (const leak of ['byo:', 'usr_owner1', 'usr_subject1', 'own-model', 'service:live', 'live:user:4242', 'default.generate', 'ai:run_', SEAM_URL, '127.0.0.1', 'SEAM_SECRET_KEY', 'env:']) {assert.ok(!r.text.includes(leak), `the page shows ${leak}`);}});@@@ -18,12 +18,13 @@ const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sen.map((r) => ({ ...r, envelope: typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope }));function fakeBilling() {-- const b = { fail: false, got: [] };++ const b = { fail: false, refuse: false, got: [] };b.fetch = async (url, init = {}) => {if (String(url).endsWith('/oauth/token')) {return new Response(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300, scope: 'billing.usage.record' }), { status: 200, headers: { 'Content-Type': 'application/json' } });}if (String(url).endsWith('/api/v1/usage')) {++ if (b.refuse) return new Response(JSON.stringify({ error: 'invalid' }), { status: 400, headers: { 'Content-Type': 'application/json' } });if (b.fail) return new Response(JSON.stringify({ error: 'unavailable' }), { status: 503, headers: { 'Content-Type': 'application/json' } });b.got.push(JSON.parse(init.body));return new Response(JSON.stringify({ record: { id: 'use_1' } }), { status: 200, headers: { 'Content-Type': 'application/json' } });@@@ -63,6 +64,15 @@ t.test('a succeeded run queues exactly one valid reading that keeps the prompt oassert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');const got = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: live });assert.strictEqual(got.body.run.usage.usage_sample_id, `ai:${runId}:tokens`);++ // explain carries the reading: its key and delivery state (no relay here, so it waits queued).++ const [u] = got.body.run.explain.usage_readings;++ assert.strictEqual(got.body.run.explain.usage_readings.length, 1);++ assert.strictEqual(u.idempotency_key, `ai:${runId}:tokens`);++ assert.strictEqual(u.state, 'queued');++ assert.strictEqual(u.attempts, 0);++ assert.strictEqual(u.last_error, null);++ assert.ok(!('free_allowance_used' in u), 'nothing was free');++ assert.ok(Array.isArray(got.body.run.explain.candidates), 'the placement explain is kept');});t.test('a cache hit writes its own zero reading', async () => {@@@ -99,6 +109,11 @@ t.test('a failed send stays queued and replays', async () => {assert.ok(pending.length >= 1, 'rows exist');for (const r of pending) { assert.strictEqual(r.sent_at, null, 'every row is unsent'); assert.ok(r.attempts >= 1, 'the send was attempted'); }assert.strictEqual(b.got.length, 0, 'nothing accepted');++ const explainOf = async (key) => (await request(h.base, 'GET', `/api/v1/runs/${key.split(':')[1]}`, { tok: live })).body.run.explain.usage_readings[0];++ let u = await explainOf(pending[0].event_id);++ assert.strictEqual(u.state, 'queued', 'a transient failure is retried');++ assert.match(u.last_error, /billing\.usage\.record answered 503/);++ assert.ok(u.attempts >= 1 && u.next_attempt_at, JSON.stringify(u));b.fail = false;await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();await relay.flush();@@@ -107,6 +122,10 @@ t.test('a failed send stays queued and replays', async () => {const keys = b.got.map((s) => s.idempotency_key).sort();assert.deepStrictEqual(keys, sent.map((r) => r.event_id).sort());for (const s of b.got) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));++ u = await explainOf(pending[0].event_id);++ assert.strictEqual(u.state, 'sent');++ assert.strictEqual(u.last_error, null);++ assert.ok(u.sent_at);});t.test('the outbox outlives a restart', async () => {@@@ -141,6 +160,25 @@ t.test('runs on default cards (free_allowance 0) carry no free_allowance_used',for (const r of await readings()) assert.ok(!('free_allowance_used' in r.envelope), JSON.stringify(r.envelope));});++t.test('a reading Billing refuses is failed, with its error, in explain and on the console report', async () => {++ usageSamples._reset();++ b = fakeBilling();++ const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });++ b.refuse = true;++ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'refused reading', options: { cache: false } } });++ assert.strictEqual(r.status, 201, r.text);++ await relay.flush();++ const [u] = (await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: live })).body.run.explain.usage_readings;++ assert.strictEqual(u.idempotency_key, `ai:${r.body.run.id}:tokens`);++ assert.strictEqual(u.state, 'failed');++ assert.match(u.last_error, /answered 400/);++ assert.ok(u.rejected_at);++ const report = await usageSamples.deliveryReport(h.db);++ assert.ok(report.relay && report.counts.failed >= 1 && report.counts.sent >= 1, JSON.stringify(report.counts));++ const f = report.failures.find((x) => x.idempotency_key === u.idempotency_key);++ assert.deepStrictEqual([f.run_id, f.state], [r.body.run.id, 'failed']);++});++t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });t.run();
@@ -149,8 +149,8 @@ needs scripts and inline styles), like Billing's staff console. Code: `server/co
@@ -149,8 +149,8 @@ needs scripts and inline styles), like Billing's staff console. Code: `server/co
| `/console/providers` | every provider (kind, status, **secret reference name and whether it resolves, never a value**, circuit, base URL without query/user info, default model, priority/timeout) and the models |
| `/console/providers` | every provider (kind, status, **secret reference name and whether it resolves, never a value**, circuit, base URL without query/user info, default model, priority/timeout) and the models |
| `/console/routes`, `/templates`, `/workflows` (`/:key[?version=]`) | every key's newest version and the version in use; per key every version with its status and content (primary and fallbacks in order, prompts, steps, schemas) |
| `/console/routes`, `/templates`, `/workflows` (`/:key[?version=]`) | every key's newest version and the version in use; per key every version with its status and content (primary and fallbacks in order, prompts, steps, schemas) |
| `/console/runs` | runs newest first, filtered by status (tabs), workflow, requester (`live` = `service:live`), error code and a UTC time range; failed runs grouped by error code |
| `/console/runs` | runs newest first, filtered by status (tabs), workflow, requester (`live` = `service:live`), error code and a UTC time range; failed runs grouped by error code |
-| `/console/runs/:id` | versions used, provider/model, fallback, attempts, tokens, cost, error code and detail, requester/attribution/target, grounding, citations, the request log (hashes only), and the stored input/output **bounded**: strings cut at 240 characters, lists at 20 items, 6 KB in all, credential-named keys blanked; raw debug fields never |
-| `/console/quotas`, `/console/usage` | active quotas with their current windows and counters, every quota; usage by day, requester, workflow and provider/model over a day range (`quotas.usage`) |
+| `/console/runs/:id` | versions used, provider/model, fallback, attempts, tokens, cost, the Billing reading with its delivery state, free share and last send error, error code and detail, requester/attribution/target, grounding, citations, the request log (hashes only), and the stored input/output **bounded**: strings cut at 240 characters, lists at 20 items, 6 KB in all, credential-named keys blanked; raw debug fields never |
+| `/console/quotas`, `/console/usage` | active quotas with their current windows and counters, every quota; usage by day, requester, workflow and provider/model over a day range (`quotas.usage`); the **free allowance this period** per subject × provider × metric (free used, allowance, remaining, period end; `quotas.freeAllowance.current()`); the `platform.usage-sample@1` readings by delivery state (queued, retrying, sent, failed) and the last send errors with their run. Staff only: the public `/stats` page never shows a subject |
| `/console/cache` | live entries and hits by workflow |
| `/console/cache` | live entries and hits by workflow |
| `/console/audit` | the `audit_log`, filtered by kind (configuration changes by default, runs, console, everything), action, actor, target |
| `/console/audit` | the `audit_log`, filtered by kind (configuration changes by default, runs, console, everything), action, actor, target |
@@ -216,7 +216,11 @@ provider that is disabled, missing credentials, circuit-open or unable to do the
@@ -216,7 +216,11 @@ provider that is disabled, missing credentials, circuit-open or unable to do the
failure; a fallback that answers is recorded on the run (`fallback_used`), in the request log
failure; a fallback that answers is recorded on the run (`fallback_used`), in the request log
(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`
(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
-reason). Speech and embeddings use placement pools of registered `stt` and `embedding` models; until
+reason). `GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: per
+reading its `idempotency_key`, `state` (`queued` until Billing accepts it, `sent`, or `failed` when Billing
+refused it), `attempts`, `queued_at`, `sent_at`, `rejected_at`, `next_attempt_at`, `last_error` (the
+outbox's last send error, also while a queued reading is retried) and `free_allowance_used` when some
+was free; a sent reading the relay has pruned reads `{ state: 'sent', pruned: true }`. Speech and embeddings use placement pools of registered `stt` and `embedding` models; until
a second upstream exists, an unavailable sole provider answers `503 provider.unavailable`. Media
a second upstream exists, an unavailable sole provider answers `503 provider.unavailable`. Media
uses separate local and paid pools so paid work still requires its explicit budget. The synthetic
uses separate local and paid pools so paid work still requires its explicit budget. The synthetic
stub can join general text routes as a last resort outside production only (`AI_STUB_FALLBACK`).
stub can join general text routes as a last resort outside production only (`AI_STUB_FALLBACK`).
@@ -29,7 +29,7 @@
@@ -29,7 +29,7 @@
],
],
"livePatch": "applied: Live fa22de5 runs AI_SERVICE=remote in production since 2026-09-23 ~15:34 UTC (309 runs by 18:30 UTC across 13 live.* workflows and 1 network.site_copy, 0 failed). As of 2026-09-23, not moved yet: VOD captions/whisper and streamers' own provider keys stayed in Live and most Live features were passthrough prompts built by Live (since 2026-09-27 AI has live.media.transcribe, stored provider keys and versioned templates for most live.* workflows); the deployed Network still falls back to Live's /internal/ai/site-copy; scripts/import-from-live.js ran on production at 2026-09-23 18:29 UTC (79,657 ledger rows: 79,647 live.ai_usage, 2 channel_ai_config, 8 site_settings; 11 holds: provider settings and the key stay env references, 1 streamer BYO key stays in Live, 3,021 translations are unrepresentable and stay in Live); text routes (live.<role>, default.chat, default.json) are capability pools placed by openvibe-sdk/placement; live.stt, default.embedding and media.paid are single-provider and answer 503 provider.unavailable when it fails; 0 citation rows; no /metrics",
"livePatch": "applied: Live fa22de5 runs AI_SERVICE=remote in production since 2026-09-23 ~15:34 UTC (309 runs by 18:30 UTC across 13 live.* workflows and 1 network.site_copy, 0 failed). As of 2026-09-23, not moved yet: VOD captions/whisper and streamers' own provider keys stayed in Live and most Live features were passthrough prompts built by Live (since 2026-09-27 AI has live.media.transcribe, stored provider keys and versioned templates for most live.* workflows); the deployed Network still falls back to Live's /internal/ai/site-copy; scripts/import-from-live.js ran on production at 2026-09-23 18:29 UTC (79,657 ledger rows: 79,647 live.ai_usage, 2 channel_ai_config, 8 site_settings; 11 holds: provider settings and the key stay env references, 1 streamer BYO key stays in Live, 3,021 translations are unrepresentable and stay in Live); text routes (live.<role>, default.chat, default.json) are capability pools placed by openvibe-sdk/placement; live.stt, default.embedding and media.paid are single-provider and answer 503 provider.unavailable when it fails; 0 citation rows; no /metrics",
"deployed": true,
"deployed": true,
- "updated": "2026-09-28",
+ "updated": "2026-10-04",
"features": [
"features": [
"versioned templates, workflows and routing profiles",
"versioned templates, workflows and routing profiles",
"runs with idempotency, cancel, retry, async polling and restart recovery",
"runs with idempotency, cancel, retry, async polling and restart recovery",
@@ -42,6 +42,7 @@
@@ -42,6 +42,7 @@
"ai.run.* events through the SDK outbox",
"ai.run.* events through the SDK outbox",
"ai.preferences and ai.usage_summary user modules",
"ai.preferences and ai.usage_summary user modules",
"server-rendered operator console for Network staff",
"server-rendered operator console for Network staff",
- "Live import"
+ "Live import",
+ "platform.usage-sample@1 readings to Billing with a per-subject provider free allowance; each run's reading state in explain and the free allowance on the staff console"
]
]
}
}
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
* GET /api/v1/runs the caller's runs (ai.run.read)
* GET /api/v1/runs the caller's runs (ai.run.read)
* GET /api/v1/runs/:id one run (+ citations, request log metadata)
* GET /api/v1/runs/:id one run (+ citations, request log metadata)
* run.usage.usage_sample_id: the run's Billing reading (server/usage-samples.js), null when none
* run.usage.usage_sample_id: the run's Billing reading (server/usage-samples.js), null when none
+ * run.explain.usage_readings: that reading's idempotency key, delivery state (queued/sent/failed), last error, free_allowance_used
* POST /api/v1/runs/:id/cancel (ai.run.create, owner)
* POST /api/v1/runs/:id/cancel (ai.run.create, owner)
* POST /api/v1/runs/:id/retry (ai.run.create, owner) -> a new run with retry_of
* POST /api/v1/runs/:id/retry (ai.run.create, owner) -> a new run with retry_of
* GET /api/v1/runs/:id/citations (ai.run.read)
* GET /api/v1/runs/:id/citations (ai.run.read)
@@ -64,7 +65,7 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
@@ -64,7 +65,7 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
r.get('/api/v1/runs/:id', read, async (req, res) => {
r.get('/api/v1/runs/:id', read, async (req, res) => {
try {
try {
- const run = await owned(req);
+ const run = await runs.explained(await owned(req));
res.json({ run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id) });
res.json({ run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id) });
} catch (err) { sendError(res, err, req.ov, log); }
} catch (err) { sendError(res, err, req.ov, log); }
});
});
@@ -14,7 +14,8 @@
@@ -14,7 +14,8 @@
* GET /console/runs[?status&workflow&requester&code&from&to&before], /console/runs/:id
* GET /console/runs[?status&workflow&requester&code&from&to&before], /console/runs/:id
* POST /console/runs/:id/cancel (ai.workflow.manage; runs.cancel)
* POST /console/runs/:id/cancel (ai.workflow.manage; runs.cancel)
* GET /console/quotas; POST /console/quotas (writes: ai.provider.manage; quotas.upsert)
* GET /console/quotas; POST /console/quotas (writes: ai.provider.manage; quotas.upsert)
- * GET /console/usage[?from&to&requester] (quotas.usage)
+ * GET /console/usage[?from&to&requester] (quotas.usage; free allowance this period, quotas.freeAllowance.current;
+ * Billing readings by delivery state, usage-samples deliveryReport)
* GET /console/cache; POST /console/cache/purge (writes: ai.provider.manage; ops.purgeCache)
* GET /console/cache; POST /console/cache/purge (writes: ai.provider.manage; ops.purgeCache)
* GET /console/audit[?kind&action&actor&target_type&target_id&before]
* GET /console/audit[?kind&action&actor&target_type&target_id&before]
*
*
@@ -38,6 +39,7 @@
@@ -38,6 +39,7 @@
const crypto = require('crypto');
const crypto = require('crypto');
const express = require('express');
const express = require('express');
const { AiError, resolveSecret } = require('../util');
const { AiError, resolveSecret } = require('../util');
+const usageSamples = require('../usage-samples');
const { CAPS } = require('../auth');
const { CAPS } = require('../auth');
const { staff: staffMap } = require('openvibe-contracts');
const { staff: staffMap } = require('openvibe-contracts');
const { createSessions, sameString, random } = require('./session');
const { createSessions, sameString, random } = require('./session');
@@ -323,7 +325,7 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e
@@ -323,7 +325,7 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e
}));
}));
}));
}));
r.get('/console/runs/:id', read, wrap(async (req, res) => {
r.get('/console/runs/:id', read, wrap(async (req, res) => {
- const run = await runs.get(req.params.id);
+ const run = await runs.explained(await runs.get(req.params.id));
if (!run) return await send(res, 404, pages.message({ title: 'Not found', text: `No run ${req.params.id}.`, back: '/console/runs', ...common(req) }));
if (!run) return await send(res, 404, pages.message({ title: 'Not found', text: `No run ${req.params.id}.`, back: '/console/runs', ...common(req) }));
return await send(res, 200, pages.run({
return await send(res, 200, pages.run({
...common(req), r: run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id), input: q.preview(run.input), output: q.preview(run.output),
...common(req), r: run, citations: await runs.citations(run.id), requests: await runs.requestsFor(run.id), input: q.preview(run.input), output: q.preview(run.output),
@@ -354,7 +356,10 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e
@@ -354,7 +356,10 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e
}));
}));
r.get('/console/usage', read, wrap(async (req, res) => {
r.get('/console/usage', read, wrap(async (req, res) => {
const { filters, form, bad } = q.parseUsageFilters(req.query, clock.now());
const { filters, form, bad } = q.parseUsageFilters(req.query, clock.now());
- await send(res, bad.length ? 400 : 200, pages.usage({ ...common(req), form, bad, report: await q.usageReport(quotas, filters) }));
+ await send(res, bad.length ? 400 : 200, pages.usage({
+ ...common(req), form, bad, report: await q.usageReport(quotas, filters),
+ free: quotas.freeAllowance ? await quotas.freeAllowance.current() : [], readings: await usageSamples.deliveryReport(db),
+ }));
}));
}));
// ── Cache ────────────────────────────────────────────────
// ── Cache ────────────────────────────────────────────────
@@ -88,7 +88,7 @@ function secs(s) {
@@ -88,7 +88,7 @@ function secs(s) {
return v < 120 ? `${v} s` : v < 7200 ? `${Math.round(v / 60)} min` : `${Math.round(v / 3600)} h`;
return v < 120 ? `${v} s` : v < 7200 ? `${Math.round(v / 60)} min` : `${Math.round(v / 3600)} h`;
}
}
const clip = (s, max = 200) => { const t = String(s == null ? '' : s); return t.length > max ? `${t.slice(0, max)}…` : t; };
const clip = (s, max = 200) => { const t = String(s == null ? '' : s); return t.length > max ? `${t.slice(0, max)}…` : t; };
-const GOOD = new Set(['succeeded', 'active', 'closed', 'configured', 'ok', 'cached', 'not_required']);
+const GOOD = new Set(['succeeded', 'sent', 'active', 'closed', 'configured', 'ok', 'cached', 'not_required']);
const BAD = new Set(['failed', 'open', 'disabled', 'missing', 'error', 'timeout', 'archived']);
const BAD = new Set(['failed', 'open', 'disabled', 'missing', 'error', 'timeout', 'archived']);
const badge = (s) => html`<span class="${GOOD.has(s) ? 'ok' : BAD.has(s) ? 'bad' : 'warn'}">${s || '—'}</span>`;
const badge = (s) => html`<span class="${GOOD.has(s) ? 'ok' : BAD.has(s) ? 'bad' : 'warn'}">${s || '—'}</span>`;
/** A base URL without user info, query or fragment (a key must never ride along in a URL shown here). */
/** A base URL without user info, query or fragment (a key must never ride along in a URL shown here). */
@@ -346,7 +346,7 @@ function run({ staff, csrf, r, citations, requests, input, output, canCancel, no
@@ -346,7 +346,7 @@ function run({ staff, csrf, r, citations, requests, input, output, canCancel, no
<dt>Attempts</dt><dd>${n(r.usage.attempts)}</dd>
<dt>Attempts</dt><dd>${n(r.usage.attempts)}</dd>
<dt>Tokens</dt><dd>${n(r.usage.tokens_in)} in · ${n(r.usage.tokens_out)} out</dd>
<dt>Tokens</dt><dd>${n(r.usage.tokens_in)} in · ${n(r.usage.tokens_out)} out</dd>
<dt>Cost</dt><dd>${usd(r.usage.cost_usd)}</dd>
<dt>Cost</dt><dd>${usd(r.usage.cost_usd)}</dd>
-<dt>Billing reading</dt><dd>${r.usage.usage_sample_id ? html`<code>${r.usage.usage_sample_id}</code>` : '—'}</dd>
+<dt>Billing reading</dt><dd>${r.usage.usage_sample_id ? html`<code>${r.usage.usage_sample_id}</code>` : '—'}${((r.explain && r.explain.usage_readings) || []).map((u) => html` ${badge(u.state)}${u.free_allowance_used != null ? html` <span class="muted">${n(u.free_allowance_used)} free</span>` : ''}${u.last_error ? html`<br><span class="bad">${clip(u.last_error, 300)}</span>` : ''}`)}</dd>
${r.error ? html`<dt>Error</dt><dd><code class="bad">${r.error.code}</code><br>${clip(r.error.detail, 500)}</dd>` : ''}
${r.error ? html`<dt>Error</dt><dd><code class="bad">${r.error.code}</code><br>${clip(r.error.detail, 500)}</dd>` : ''}
</dl></div><div class="card"><dl>
</dl></div><div class="card"><dl>
<dt>Requester</dt><dd><code>${r.requester.type}:${r.requester.id}</code></dd>
<dt>Requester</dt><dd><code>${r.requester.type}:${r.requester.id}</code></dd>
@@ -410,7 +410,9 @@ function usageRows(caption, first, rows) {
@@ -410,7 +410,9 @@ function usageRows(caption, first, rows) {
<td>${n(r.tokens_in)}</td><td>${n(r.tokens_out)}</td><td>${n(r.tokens_cached)}</td><td>${usd(r.cost_usd)}</td></tr>`), 'no usage in this range.');
<td>${n(r.tokens_in)}</td><td>${n(r.tokens_out)}</td><td>${n(r.tokens_cached)}</td><td>${usd(r.cost_usd)}</td></tr>`), 'no usage in this range.');
}
}
-function usage({ staff, csrf, form, bad, report }) {
+const day = (ms) => (ms >= 8.64e15 ? 'never' : when(new Date(ms).toISOString()));
+
+function usage({ staff, csrf, form, bad, report, free = [], readings }) {
const t = report.total;
const t = report.total;
return layout({
return layout({
title: 'Usage', section: 'usage', staff, csrf, error: bad.length ? `ignored malformed filter(s): ${bad.join(', ')}` : null, body: html`<h1>Usage</h1>
title: 'Usage', section: 'usage', staff, csrf, error: bad.length ? `ignored malformed filter(s): ${bad.join(', ')}` : null, body: html`<h1>Usage</h1>
@@ -424,7 +426,15 @@ function usage({ staff, csrf, form, bad, report }) {
@@ -424,7 +426,15 @@ function usage({ staff, csrf, form, bad, report }) {
${report.truncated ? html`<p class="warn">Only the first 1,000 daily rows are summed; narrow the range.</p>` : ''}
${report.truncated ? html`<p class="warn">Only the first 1,000 daily rows are summed; narrow the range.</p>` : ''}
<h2>Breakdown</h2>
<h2>Breakdown</h2>
${usageRows('By day', 'Day', report.byDay)}${usageRows('By requester', 'Requester', report.byRequester)}
${usageRows('By day', 'Day', report.byDay)}${usageRows('By requester', 'Requester', report.byRequester)}
-${usageRows('By workflow', 'Workflow', report.byWorkflow)}${usageRows('By provider and model', 'Provider / model', report.byModel)}`,
+${usageRows('By workflow', 'Workflow', report.byWorkflow)}${usageRows('By provider and model', 'Provider / model', report.byModel)}
+<h2>Free allowance this period</h2>
+${table('Free allowance by subject, provider and metric', ['Subject', 'Provider', 'Metric', 'Free used', 'Allowance', 'Remaining', 'Period ends'], free.map((f) => html`<tr><td><code>${f.subject}</code></td><td><code>${f.provider}</code></td>
+<td><code>${f.metric}</code></td><td>${n(f.free_used)}</td><td>${n(f.free_allowance)}</td><td>${n(f.remaining)}</td><td>${day(f.period_end)}</td></tr>`), 'no free allowance used this period.')}
+<h2>Billing readings</h2>
+${readings ? html`<p class="muted">${readings.relay ? 'Sending to Billing.' : readings.enabled ? 'Queued only: Billing sending is not configured.' : 'Readings are off (USAGE_SAMPLES=off).'}</p>
+${table('Readings by delivery state', ['Queued', 'Retrying', 'Sent', 'Failed'], [html`<tr><td>${n(readings.counts.queued)}</td><td>${n(readings.counts.retrying)}</td><td>${n(readings.counts.sent)}</td><td>${n(readings.counts.failed)}</td></tr>`])}
+${table('Last send errors', ['Reading', 'Run', 'State', 'Attempts', 'Queued', 'Last error'], readings.failures.map((f) => html`<tr><td><code>${f.idempotency_key}</code></td>
+<td>${f.run_id ? html`<a class="mono" href="/console/runs/${enc(f.run_id)}">${f.run_id}</a>` : '—'}</td><td>${badge(f.state)}</td><td>${n(f.attempts)}</td><td>${when(f.queued_at)}</td><td>${clip(f.last_error, 300)}</td></tr>`), 'no send error.')}` : ''}`,
});
});
}
}
@@ -400,6 +400,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -400,6 +400,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
return (await db.prepare(`SELECT * FROM runs ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY id DESC LIMIT ?`).all(...args, Math.min(200, Number(limit) || 50))).map(decode);
return (await db.prepare(`SELECT * FROM runs ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY id DESC LIMIT ?`).all(...args, Math.min(200, Number(limit) || 50))).map(decode);
}
}
+ /** The run with its Billing readings in explain.usage_readings (server/usage-samples.js readingsFor); unchanged when it has none. */
+ async function explained(run) {
+ const readings = run ? await usageSamples.readingsFor(db, run.usage.usage_sample_id) : [];
+ return readings.length ? { ...run, explain: { ...(run.explain || {}), usage_readings: readings } } : run;
+ }
+
async function requestsFor(runId) {
async function requestsFor(runId) {
return await db.prepare('SELECT id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash, tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, created_at FROM requests WHERE run_id = ? ORDER BY seq').all(runId);
return await db.prepare('SELECT id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash, tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, created_at FROM requests WHERE run_id = ? ORDER BY seq').all(runId);
}
}
@@ -437,7 +443,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -437,7 +443,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
/** Queue depth for /metrics and /api/ready: runs waiting for a slot, and runs holding one. */
/** Queue depth for /metrics and /api/ready: runs waiting for a slot, and runs holding one. */
function stats() { return { queued: queue.length, running: active, inflight: inflight.size, max_concurrent: config.runs.maxConcurrent, max_queued: config.runs.maxQueued }; }
function stats() { return { queued: queue.length, running: active, inflight: inflight.size, max_concurrent: config.runs.maxConcurrent, max_queued: config.runs.maxQueued }; }
- return { create, wait, get, list, cancel, retry, citations, addCitations, requestsFor, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode };
+ return { create, wait, get, list, cancel, retry, citations, addCitations, requestsFor, explained, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode };
}
}
module.exports = { createRuns, entityKey, subjectKey };
module.exports = { createRuns, entityKey, subjectKey };
@@ -20,8 +20,9 @@ const contracts = require('openvibe-contracts');
@@ -20,8 +20,9 @@ const contracts = require('openvibe-contracts');
const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';
const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';
const REFUSED = new Set([400, 409, 413, 422]); const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
const REFUSED = new Set([400, 409, 413, 422]); const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
let relay = null; let on = false; let pruneTimer = null; let lastError = null; let queued = 0;
let relay = null; let on = false; let pruneTimer = null; let lastError = null; let queued = 0;
+const keyOf = (runId) => `ai:${runId}:${METRIC}`;
function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {
function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {
- const key = `ai:${runId}:${METRIC}`;
+ const key = keyOf(runId);
const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: runId, operation: workflowKey,
const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: runId, operation: workflowKey,
quantity: Math.max(0, (Number(tokensIn) || 0) + (Number(tokensOut) || 0)), unit: 'tokens', at: new Date(at).toISOString(),
quantity: Math.max(0, (Number(tokensIn) || 0) + (Number(tokensOut) || 0)), unit: 'tokens', at: new Date(at).toISOString(),
cost_estimate: Math.max(0, Number(cost) || 0), source: 'ai.runs' };
cost_estimate: Math.max(0, Number(cost) || 0), source: 'ai.runs' };
@@ -66,6 +67,34 @@ function init(db, { enabled = true, billingUrl = '', audience = 'openvibe.billin
@@ -66,6 +67,34 @@ function init(db, { enabled = true, billingUrl = '', audience = 'openvibe.billin
log.log(`[Billing] ai usage samples → ${billingUrl}`);
log.log(`[Billing] ai usage samples → ${billingUrl}`);
return relay;
return relay;
}
}
+// Delivery state as the outbox records it: sent, failed (Billing refused it; never retried), else queued (last_error
+// set while a send is being retried).
+const stateOf = (r) => (r.sent_at != null ? 'sent' : r.rejected_at != null ? 'failed' : 'queued');
+const isoOrNull = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
+const envelopeOf = (e) => (typeof e === 'string' ? JSON.parse(e) : e) || {};
+/**
+ * A run's readings for its explain: [{ idempotency_key, state, attempts, queued_at, sent_at, rejected_at,
+ * next_attempt_at, last_error, free_allowance_used? }]. A sent row the relay has pruned reads { state: 'sent', pruned: true }.
+ */
+async function readingsFor(db, usageSampleId) {
+ if (!usageSampleId) return [];
+ const r = await db.prepare(`SELECT event_id, envelope, created_at, attempts, next_attempt_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE event_id = ?`).get(usageSampleId);
+ if (!r) return [{ idempotency_key: usageSampleId, state: 'sent', pruned: true }];
+ const state = stateOf(r);
+ const out = { idempotency_key: r.event_id, state, attempts: Number(r.attempts), queued_at: isoOrNull(r.created_at), sent_at: isoOrNull(r.sent_at), rejected_at: isoOrNull(r.rejected_at),
+ next_attempt_at: state === 'queued' ? isoOrNull(r.next_attempt_at) : null, last_error: r.last_error || null };
+ const free = envelopeOf(r.envelope).free_allowance_used;
+ if (free != null) out.free_allowance_used = free;
+ return [out];
+}
+/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */
+async function deliveryReport(db, { failures = 10 } = {}) {
+ const c = await db.prepare(`SELECT count(*) FILTER (WHERE sent_at IS NOT NULL) AS sent, count(*) FILTER (WHERE rejected_at IS NOT NULL) AS failed,
+ count(*) FILTER (WHERE sent_at IS NULL AND rejected_at IS NULL) AS queued, count(*) FILTER (WHERE sent_at IS NULL AND rejected_at IS NULL AND last_error IS NOT NULL) AS retrying FROM ${TABLE}`).get();
+ const rows = await db.prepare(`SELECT event_id, envelope, attempts, created_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE last_error IS NOT NULL ORDER BY id DESC LIMIT ?`).all(failures);
+ return { enabled: on, relay: Boolean(relay), counts: { queued: Number(c.queued), retrying: Number(c.retrying), sent: Number(c.sent), failed: Number(c.failed) },
+ failures: rows.map((r) => ({ idempotency_key: r.event_id, run_id: envelopeOf(r.envelope).resource || null, state: stateOf(r), attempts: Number(r.attempts), queued_at: isoOrNull(r.created_at), last_error: r.last_error })) };
+}
async function status() { if (!relay) return { enabled: on, relay: false }; return { enabled: on, relay: true, pending: Number(await relay.pending()), rejected: Number(await relay.rejected()), queued_since_boot: queued, last_error: lastError }; }
async function status() { if (!relay) return { enabled: on, relay: false }; return { enabled: on, relay: true, pending: Number(await relay.pending()), rejected: Number(await relay.rejected()), queued_since_boot: queued, last_error: lastError }; }
/** Graceful stop (after runs.drain(), before db.close()): one last bounded pass sends what is due, then the relay stops; unsent rows wait for the next start. */
/** Graceful stop (after runs.drain(), before db.close()): one last bounded pass sends what is due, then the relay stops; unsent rows wait for the next start. */
async function stop({ drainMs = 3000 } = {}) {
async function stop({ drainMs = 3000 } = {}) {
@@ -74,4 +103,4 @@ async function stop({ drainMs = 3000 } = {}) {
@@ -74,4 +103,4 @@ async function stop({ drainMs = 3000 } = {}) {
if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }
if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }
}
}
function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }
function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }
-module.exports = { init, record, sampleOf, status, stop, _reset, TABLE };
+module.exports = { init, record, sampleOf, keyOf, readingsFor, deliveryReport, status, stop, _reset, TABLE };
@@ -168,7 +168,7 @@ t.test('the staff map: admin reads and manages workflows, only the owner manages
@@ -168,7 +168,7 @@ t.test('the staff map: admin reads and manages workflows, only the owner manages
});
});
t.test('anonymous: console pages answer the sign-in page (401), never data; writes change nothing', async () => {
t.test('anonymous: console pages answer the sign-in page (401), never data; writes change nothing', async () => {
- for (const p of ['/console', '/console/runs?status=failed', '/console/providers', '/console/audit', `/console/runs/${runIds.failLive}`, '/console/nope']) {
+ for (const p of ['/console', '/console/runs?status=failed', '/console/providers', '/console/usage', '/console/audit', `/console/runs/${runIds.failLive}`, '/console/nope']) {
const r = await get(p);
const r = await get(p);
assert.strictEqual(r.status, 401, p);
assert.strictEqual(r.status, 401, p);
assert.ok(r.text.includes('Sign in with OpenVibe.Network'), p);
assert.ok(r.text.includes('Sign in with OpenVibe.Network'), p);
@@ -432,6 +432,25 @@ t.test('run detail: error code, attempts, provider, request log; input bounded a
@@ -432,6 +432,25 @@ t.test('run detail: error code, attempts, provider, request log; input bounded a
assert.ok(r.text.includes('Grounding'));
assert.ok(r.text.includes('Grounding'));
});
});
+t.test('usage: free allowance per subject this period and Billing readings by state, with the last send errors', async () => {
+ await h.registry.upsertModel({ provider_key: 'okseam', model_key: 'seam-model', type: 'chat', cost: { in_per_mtok: 1, out_per_mtok: 1, free_allowance: 500, reset_period: 'month' } });
+ assert.strictEqual((await h.quotas.freeAllowance.claim('live:user:4242', 'okseam', 'seam-model', { in: 120 })).tokens, 120);
+ const key = `ai:${runIds.ok}:tokens`;
+ assert.ok(await h.db.prepare('SELECT 1 AS x FROM usage_sample_outbox WHERE event_id = ?').get(key), 'the succeeded run queued a reading');
+ await h.db.prepare("UPDATE usage_sample_outbox SET attempts = 2, last_error = 'billing.usage.record answered 503' WHERE event_id = ?").run(key);
+ const r = await get('/console/usage', cookies.admin);
+ assertPage(r, '/console/usage');
+ const row = r.text.split('</tr>').find((x) => x.includes('<code>live:user:4242</code>'));
+ assert.ok(row, 'the subject has a row');
+ const cells = [...row.matchAll(/<td[^>]*>([\s\S]*?)<\/td>/g)].map((m) => m[1].replace(/<[^>]+>/g, '').trim());
+ assert.deepStrictEqual(cells.slice(0, 6), ['live:user:4242', 'okseam', 'input-tokens:seam-model', '120', '500', '380']);
+ assert.ok(r.text.includes('Readings by delivery state') && r.text.includes(`<code>${key}</code>`) && r.text.includes('billing.usage.record answered 503'));
+ const run = await get(`/console/runs/${runIds.ok}`, cookies.admin);
+ assert.ok(run.text.includes(`<code>${key}</code>`) && run.text.includes('billing.usage.record answered 503'), 'the run page shows its reading\'s state and error');
+ assert.strictEqual((await get('/console/usage')).status, 401, 'signed out: refused');
+ assert.ok(!(await fetch(`${h.base}/stats`).then((x) => x.text())).includes('live:user:4242'), 'the public page has no subject');
+});
+
t.test('cancel: an admin cancels a running run (ai.workflow.manage), audited with the person as actor', async () => {
t.test('cancel: an admin cancels a running run (ai.workflow.manage), audited with the person as actor', async () => {
assert.strictEqual((await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: ops, body: { primary: { provider: 'slowseam' }, fallbacks: [] } })).status, 201);
assert.strictEqual((await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: ops, body: { primary: { provider: 'slowseam' }, fallbacks: [] } })).status, 201);
const c = await request(h.base, 'POST', '/api/v1/runs', { tok: live, body: { workflow: 'ai.generate', input: { prompt: 'slow one' }, options: { cache: false } } });
const c = await request(h.base, 'POST', '/api/v1/runs', { tok: live, body: { workflow: 'ai.generate', input: { prompt: 'slow one' }, options: { cache: false } } });
@@ -140,6 +140,14 @@ t.test('the first run is free up to the allowance per metric, the next one is fu
@@ -140,6 +140,14 @@ t.test('the first run is free up to the allowance per metric, the next one is fu
assert.ok(Math.abs(Number(run.cost_usd) - 2) < 1e-9);
assert.ok(Math.abs(Number(run.cost_usd) - 2) < 1e-9);
});
});
+t.test('explain shows the free share on the first run\'s reading and leaves it out on the next', async () => {
+ const [first, next] = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();
+ const explain = async (id) => (await request(h.base, 'GET', `/api/v1/runs/${id}`, { tok: live })).body.run.explain.usage_readings[0];
+ const a = await explain(first.id);
+ assert.deepStrictEqual([a.idempotency_key, a.state, a.free_allowance_used], [`ai:${first.id}:tokens`, 'queued', 200]);
+ assert.ok(!('free_allowance_used' in await explain(next.id)));
+});
+
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.run();
t.run();
@@ -52,9 +52,15 @@ t.test('seeded stats: one row per provider and model with p50, p95, success rate
@@ -52,9 +52,15 @@ t.test('seeded stats: one row per provider and model with p50, p95, success rate
assert.match(r.text, /last 7 UTC days \(\d{4}-\d{2}-\d{2} to \d{4}-\d{2}-\d{2}\)/);
assert.match(r.text, /last 7 UTC days \(\d{4}-\d{2}-\d{2} to \d{4}-\d{2}-\d{2}\)/);
});
});
+t.test('a subject with free allowance used this period is still not on the page', async () => {
+ const free = await h.quotas.freeAllowance.claim('live:user:4242', 'seam', 'seam-model', { in: 10 });
+ assert.strictEqual(free.tokens, 10);
+ assert.ok((await h.quotas.freeAllowance.current()).some((r) => r.subject === 'live:user:4242'), 'the staff console has the row');
+});
+
t.test('no caller identifiers, internal URLs or secrets in the HTML', async () => {
t.test('no caller identifiers, internal URLs or secrets in the HTML', async () => {
const r = await request(h.base, 'GET', '/stats');
const r = await request(h.base, 'GET', '/stats');
- for (const leak of ['byo:', 'usr_owner1', 'usr_subject1', 'own-model', 'service:live', 'live:user:4242', 'default.generate', SEAM_URL, '127.0.0.1', 'SEAM_SECRET_KEY', 'env:']) {
+ for (const leak of ['byo:', 'usr_owner1', 'usr_subject1', 'own-model', 'service:live', 'live:user:4242', 'default.generate', 'ai:run_', SEAM_URL, '127.0.0.1', 'SEAM_SECRET_KEY', 'env:']) {
assert.ok(!r.text.includes(leak), `the page shows ${leak}`);
assert.ok(!r.text.includes(leak), `the page shows ${leak}`);
}
}
});
});
@@ -18,12 +18,13 @@ const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sen
@@ -18,12 +18,13 @@ const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sen
.map((r) => ({ ...r, envelope: typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope }));
.map((r) => ({ ...r, envelope: typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope }));
function fakeBilling() {
function fakeBilling() {
- const b = { fail: false, got: [] };
+ const b = { fail: false, refuse: false, got: [] };
b.fetch = async (url, init = {}) => {
b.fetch = async (url, init = {}) => {
if (String(url).endsWith('/oauth/token')) {
if (String(url).endsWith('/oauth/token')) {
return new Response(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300, scope: 'billing.usage.record' }), { status: 200, headers: { 'Content-Type': 'application/json' } });
return new Response(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300, scope: 'billing.usage.record' }), { status: 200, headers: { 'Content-Type': 'application/json' } });
}
}
if (String(url).endsWith('/api/v1/usage')) {
if (String(url).endsWith('/api/v1/usage')) {
+ if (b.refuse) return new Response(JSON.stringify({ error: 'invalid' }), { status: 400, headers: { 'Content-Type': 'application/json' } });
if (b.fail) return new Response(JSON.stringify({ error: 'unavailable' }), { status: 503, headers: { 'Content-Type': 'application/json' } });
if (b.fail) return new Response(JSON.stringify({ error: 'unavailable' }), { status: 503, headers: { 'Content-Type': 'application/json' } });
b.got.push(JSON.parse(init.body));
b.got.push(JSON.parse(init.body));
return new Response(JSON.stringify({ record: { id: 'use_1' } }), { status: 200, headers: { 'Content-Type': 'application/json' } });
return new Response(JSON.stringify({ record: { id: 'use_1' } }), { status: 200, headers: { 'Content-Type': 'application/json' } });
@@ -63,6 +64,15 @@ t.test('a succeeded run queues exactly one valid reading that keeps the prompt o
@@ -63,6 +64,15 @@ t.test('a succeeded run queues exactly one valid reading that keeps the prompt o
assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');
assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');
const got = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: live });
const got = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: live });
assert.strictEqual(got.body.run.usage.usage_sample_id, `ai:${runId}:tokens`);
assert.strictEqual(got.body.run.usage.usage_sample_id, `ai:${runId}:tokens`);
+ // explain carries the reading: its key and delivery state (no relay here, so it waits queued).
+ const [u] = got.body.run.explain.usage_readings;
+ assert.strictEqual(got.body.run.explain.usage_readings.length, 1);
+ assert.strictEqual(u.idempotency_key, `ai:${runId}:tokens`);
+ assert.strictEqual(u.state, 'queued');
+ assert.strictEqual(u.attempts, 0);
+ assert.strictEqual(u.last_error, null);
+ assert.ok(!('free_allowance_used' in u), 'nothing was free');
+ assert.ok(Array.isArray(got.body.run.explain.candidates), 'the placement explain is kept');
});
});
t.test('a cache hit writes its own zero reading', async () => {
t.test('a cache hit writes its own zero reading', async () => {
@@ -99,6 +109,11 @@ t.test('a failed send stays queued and replays', async () => {
@@ -99,6 +109,11 @@ t.test('a failed send stays queued and replays', async () => {
assert.ok(pending.length >= 1, 'rows exist');
assert.ok(pending.length >= 1, 'rows exist');
for (const r of pending) { assert.strictEqual(r.sent_at, null, 'every row is unsent'); assert.ok(r.attempts >= 1, 'the send was attempted'); }
for (const r of pending) { assert.strictEqual(r.sent_at, null, 'every row is unsent'); assert.ok(r.attempts >= 1, 'the send was attempted'); }
assert.strictEqual(b.got.length, 0, 'nothing accepted');
assert.strictEqual(b.got.length, 0, 'nothing accepted');
+ const explainOf = async (key) => (await request(h.base, 'GET', `/api/v1/runs/${key.split(':')[1]}`, { tok: live })).body.run.explain.usage_readings[0];
+ let u = await explainOf(pending[0].event_id);
+ assert.strictEqual(u.state, 'queued', 'a transient failure is retried');
+ assert.match(u.last_error, /billing\.usage\.record answered 503/);
+ assert.ok(u.attempts >= 1 && u.next_attempt_at, JSON.stringify(u));
b.fail = false;
b.fail = false;
await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();
await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();
await relay.flush();
await relay.flush();
@@ -107,6 +122,10 @@ t.test('a failed send stays queued and replays', async () => {
@@ -107,6 +122,10 @@ t.test('a failed send stays queued and replays', async () => {
const keys = b.got.map((s) => s.idempotency_key).sort();
const keys = b.got.map((s) => s.idempotency_key).sort();
assert.deepStrictEqual(keys, sent.map((r) => r.event_id).sort());
assert.deepStrictEqual(keys, sent.map((r) => r.event_id).sort());
for (const s of b.got) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));
for (const s of b.got) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));
+ u = await explainOf(pending[0].event_id);
+ assert.strictEqual(u.state, 'sent');
+ assert.strictEqual(u.last_error, null);
+ assert.ok(u.sent_at);
});
});
t.test('the outbox outlives a restart', async () => {
t.test('the outbox outlives a restart', async () => {
@@ -141,6 +160,25 @@ t.test('runs on default cards (free_allowance 0) carry no free_allowance_used',
@@ -141,6 +160,25 @@ t.test('runs on default cards (free_allowance 0) carry no free_allowance_used',
for (const r of await readings()) assert.ok(!('free_allowance_used' in r.envelope), JSON.stringify(r.envelope));
for (const r of await readings()) assert.ok(!('free_allowance_used' in r.envelope), JSON.stringify(r.envelope));
});
});
+t.test('a reading Billing refuses is failed, with its error, in explain and on the console report', async () => {
+ usageSamples._reset();
+ b = fakeBilling();
+ const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });
+ b.refuse = true;
+ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'refused reading', options: { cache: false } } });
+ assert.strictEqual(r.status, 201, r.text);
+ await relay.flush();
+ const [u] = (await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: live })).body.run.explain.usage_readings;
+ assert.strictEqual(u.idempotency_key, `ai:${r.body.run.id}:tokens`);
+ assert.strictEqual(u.state, 'failed');
+ assert.match(u.last_error, /answered 400/);
+ assert.ok(u.rejected_at);
+ const report = await usageSamples.deliveryReport(h.db);
+ assert.ok(report.relay && report.counts.failed >= 1 && report.counts.sent >= 1, JSON.stringify(report.counts));
+ const f = report.failures.find((x) => x.idempotency_key === u.idempotency_key);
+ assert.deepStrictEqual([f.run_id, f.state], [r.body.run.id, 'failed']);
+});
+
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.run();
t.run();