openvibers/openvibe.ai Commits
Emit one rateable usage reading per attempt per token kind: resource is the rate-card metric, unit its leading word, free allowance per kind (#20)
Changes
nnew file mode 100644@@@ -0,0 +1,6 @@++-- phase: expand++-- B22: a run's Billing readings are one platform.usage-sample@1 per finished provider attempt per token metric++-- (server/usage-samples.js), keyed ai:<run id>:<attempt>:<kind>. runs.usage_sample_ids links that whole set in++-- order; runs.usage_sample_id (0005) named the single aggregate reading and is no longer written by AI. Additive++-- only: a new nullable column, no existing row rewritten, and the old column left in place.++ALTER TABLE runs ADD COLUMN usage_sample_ids text[];@@@ -5,8 +5,8 @@* POST /api/v1/runs create (ai.run.create); ?wait=ms waits for the result* 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 (on every run response): that reading's idempotency key, delivery state (queued/sent/failed), last error, free_allowance_used++ * run.usage.usage_sample_ids: the run's Billing readings (one per attempt per token metric), null when none++ * run.explain.usage_readings (on every run response): each 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)@@@ -334,6 +334,7 @@ const ref = (x) => (x ? (x.service ? `${x.service}:${x.type}:${x.id}` : `${x.typfunction run({ staff, csrf, r, citations, requests, input, output, canCancel, notice, error }) {const g = r.grounding;++ const sampleIds = r.usage.usage_sample_ids || [];const open = r.status === 'queued' || r.status === 'running';return layout({title: `Run ${r.id}`, section: 'runs', staff, csrf, notice, error, body: html`<h1>Run <span class="mono">${r.id}</span> ${badge(r.status)}</h1>@@@ -346,7 +347,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>` : '—'}${((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>++<dt>Billing readings</dt><dd>${sampleIds.length ? html`${sampleIds.map((id) => html`<code>${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>@@@ -16,7 +16,10 @@* a key was lost and what the console reads. The numbers are the same either way.** const free = createFreeAllowance(db, { cardsFor: pool.rateCardsFor });-- * const { tokens, usd } = await free.claim(subject, provider, model, { in, cached, out }); // inside db.tx++ * const { tokens, usd, byKind } = await free.claim(subject, provider, model, { in, cached, out }); // inside db.tx++ *++ * `byKind[kind]` is { tokens, usd, cost }: that kind's free tokens, their list price, and the list price of its++ * whole quantity — account() turns each into one reading with its own free_allowance_used and cost_estimate.*/const rates = require('./providers/rate-cards');@@@ -63,11 +66,16 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },/*** Claim this attempt's free tokens: usage { in (excluding cached), cached, out } on `provider`/`model`.-- * Returns { tokens, usd }: the free tokens and their list price on the same cards. A provider with no-- * cards (never billed) or a card with free_allowance 0 claims nothing and writes nothing.++ * Returns { tokens, usd, byKind }: the free tokens and their list price in total, and, per kind++ * (`in` | `cached` | `out`), { tokens, usd, cost } — the free tokens, their list price, and the list++ * price of the whole requested quantity. account() writes one platform.usage-sample@1 per kind from++ * this, so each reading's free_allowance_used and cost_estimate are its own (server/usage-samples.js).++ * A provider with no cards (never billed) claims nothing: byKind is all zeros. A card with++ * free_allowance 0 claims nothing but still reports its list cost (the reading is fully priced).*/async function claim(subject, provider, model, usage = {}, now = clock.now()) {-- const out = { tokens: 0, usd: 0 };++ const byKind = Object.fromEntries(KINDS.map((kind) => [kind, { tokens: 0, usd: 0, cost: 0 }]));++ const out = { tokens: 0, usd: 0, byKind };if (!subject || !provider) return out;const cards = await cardsFor(provider, model || null);if (!cards) return out;@@@ -77,21 +85,21 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },const c = cards[kind];const cap = Math.floor(Number(c && c.free_allowance) || 0);const q = Math.max(0, Math.floor(Number(usage[kind]) || 0));++ const k = byKind[kind];++ if (c) k.cost = (q / c.unit_size) * c.unit_price_usd; // the whole quantity's list price, free or notif (cap <= 0 || q <= 0) continue;const p = periodFor(c, now);if (store) {-- const free = await claimOnStore(subject, provider, c, p, q, cap, at, now);-- out.tokens += free;-- out.usd += (free / c.unit_size) * c.unit_price_usd;-- continue;++ k.tokens = await claimOnStore(subject, provider, c, p, q, cap, at, now);++ } else {++ await open.run(subject, provider, c.metric, p.start, p.end, at);++ const row = await lock.get(subject, provider, c.metric, p.start);++ const free = Math.min(q, Math.max(0, cap - Number(row.free_used)));++ if (free) { await take.run(free, at, subject, provider, c.metric, p.start); k.tokens = free; }}-- await open.run(subject, provider, c.metric, p.start, p.end, at);-- const row = await lock.get(subject, provider, c.metric, p.start);-- const free = Math.min(q, Math.max(0, cap - Number(row.free_used)));-- if (!free) continue;-- await take.run(free, at, subject, provider, c.metric, p.start);-- out.tokens += free;-- out.usd += (free / c.unit_size) * c.unit_price_usd;++ k.usd = (k.tokens / c.unit_size) * c.unit_price_usd;++ out.tokens += k.tokens;++ out.usd += k.usd;}});return out;@@@ -170,14 +170,16 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll/*** Add real usage to the reserved windows, the daily usage table and — in the same transaction — the-- * provider rollup and per-route placement state. `attempts` is the run's finished attempts (the same-- * entries logged to `requests`); `writeUsage` is false when a run spent nothing (stats are still kept).-- * `runId`/`traceId` name the run a reading is written for (server/usage-samples.js): with a runId the-- * run's platform.usage-sample@1 reading commits with its usage. Before it, the run's tokens claim the subject's-- * provider free allowance (server/free-allowance.js, §2.1.8): the free share is the reading's free_allowance_used-- * and its list price is left out of the reading's cost_estimate. Pricing only: the counters above are unchanged.-- * `hold` is the run's tier hold (server/govern.js): once the accounting committed, it is settled with the-- * run's tokens and its priced cost (the free share left out), so a fully free run spends no ai-usd.++ * provider rollup, per-route placement state and the run's Billing readings. `attempts` is the run's++ * finished attempts (the same entries logged to `requests`); `writeUsage` is false when a run spent++ * nothing (stats are still kept). `runId`/`traceId` name the run a reading is written for++ * (server/usage-samples.js): with a runId, one platform.usage-sample@1 per attempt per token kind++ * (in excluding cached, cached, out) commits with its usage, under the attempt's provider/model metric.++ * Each attempt's tokens claim the subject's provider free allowance (server/free-allowance.js, §2.1.8) per++ * kind: the free share is that reading's free_allowance_used and its list price is left out of the++ * reading's cost_estimate. Pricing only: the counters above are unchanged. `hold` is the run's tier hold++ * (server/govern.js): once the accounting committed, it is settled with the run's tokens and its priced++ * cost (the free share left out), so a fully free run spends no ai-usd.*/async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true, runId = null, traceId = null, hold = null }) {const tokens = tokensIn + tokensOut;@@@ -186,11 +188,26 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAllawait db.tx(async () => {for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);-- // The run's platform.usage-sample@1 reading for Billing commits with its usage (server/usage-samples.js).++ // One reading per finished attempt per token kind, for Billing (server/usage-samples.js).if (writeUsage && runId) {-- const free = freeAllowance ? await freeAllowance.claim(subjectOf(ctx), provider, model, { in: Math.max(0, tokensIn - tokensCached), cached: tokensCached, out: tokensOut }, clock.now()) : { tokens: 0, usd: 0 };-- freeUsd = free.usd;-- await usageSamples.record(db, { runId, workflowKey: ctx.workflowKey, requester: `${ctx.requesterType}:${ctx.requesterId}`, provider, tokensIn, tokensOut, cost: Math.max(0, cost - free.usd), freeAllowanceUsed: free.tokens, at: clock.now(), traceId });++ const at = clock.now();++ const readings = [];++ const list = attempts || [];++ for (let attempt = 0; attempt < list.length; attempt++) {++ const a = list[attempt];++ if (!a || !a.provider_key || a.status === 'skipped') continue; // a skipped attempt never touched a provider++ const attemptModel = a.model_key || null;++ const qty = { in: Math.max(0, (Number(a.tokens_in) || 0) - (Number(a.tokens_cached) || 0)), cached: Math.max(0, Number(a.tokens_cached) || 0), out: Math.max(0, Number(a.tokens_out) || 0) };++ const free = freeAllowance ? await freeAllowance.claim(subjectOf(ctx), a.provider_key, attemptModel, qty, at) : null;++ const byKind = (free && free.byKind) || {};++ if (free) freeUsd += Number(free.usd) || 0;++ for (const kind of usageSamples.KINDS) {++ const f = byKind[kind] || { tokens: 0, usd: 0, cost: 0 };++ readings.push({ attempt, kind, provider: a.provider_key, model: attemptModel, quantity: qty[kind],++ cost: Math.max(0, (Number(f.cost) || 0) - (Number(f.usd) || 0)), freeAllowanceUsed: Number(f.tokens) || 0 });++ }++ }++ await usageSamples.record(db, { runId, workflowKey: ctx.workflowKey, requester: `${ctx.requesterType}:${ctx.requesterId}`, at, traceId, readings });}await recordStats(day, attempts);});@@@ -85,7 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { noworigin: 'ai', workflow: r.workflow_key, workflow_version: r.workflow_version, template_version: r.template_version, route: r.route_key, route_version: r.route_version,provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),},-- usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_id: r.usage_sample_id || null },++ usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_ids: r.usage_sample_ids || (r.usage_sample_id ? [r.usage_sample_id] : null) },explain: parseJson(r.explain, null),citations_count: r.citations_count,grounding: parseJson(r.grounding, null),@@@ -215,8 +215,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');// The reused output carries the grounding it was produced with.await db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);-- // A cache hit is still a served run: its reading (no tokens, no cost) goes to Billing with it.-- await usageSamples.record(db, { runId: id, workflowKey: wf.key, requester: `${requester.type}:${requester.id}`, at: now, traceId: trace });++ // A cache hit is still a served run: its zero reading (no attempt, no tokens, no cost) goes to Billing with it.++ await usageSamples.record(db, { runId: id, workflowKey: wf.key, requester: `${requester.type}:${requester.id}`, at: now, traceId: trace,++ readings: [{ attempt: 0, kind: 'in', model: hit.model_key || null, quantity: 0, cost: 0, freeAllowanceUsed: 0 }] });await events.runChanged(await getRow(id));});await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });@@@ -410,7 +411,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now/** 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) : [];++ const ids = run && run.usage ? run.usage.usage_sample_ids : null;++ const readings = ids && ids.length ? await usageSamples.readingsFor(db, ids) : [];return readings.length ? { ...run, explain: { ...(run.explain || {}), usage_readings: readings } } : run;}@@@ -1,59 +1,81 @@'use strict';/**-- * AI → OpenVibe.Billing: one platform.usage-sample@1 reading per accounted run, posted to-- * billing.usage.record (POST /api/v1/usage, audience openvibe.billing) through a retry-safe outbox.++ * AI → OpenVibe.Billing: one platform.usage-sample@1 reading per finished provider attempt per token metric,++ * posted to billing.usage.record (POST /api/v1/usage, audience openvibe.billing) through a retry-safe outbox.*-- * quota.js account() writes the reading in the SAME transaction as usage_daily. A cache hit writes its-- * zero reading in the transaction that records the run. The relay (openvibe-sdk createPgOutbox) posts-- * each row with AI's service token. If Billing is down, has no grant yet, or is not deployed, the row-- * waits and is retried with backoff, across restarts. Only Billing refusing the reading itself (400, 409,-- * 413, 422) marks a row rejected. A reading carries ids, the workflow, the requester, tokens and cost. It-- * never carries the input, the output or a prompt. Readings are queued once init() ran, unless-- * USAGE_SAMPLES=off. The relay runs while OV_BILLING_INTERNAL_URL and OV_OAUTH_CLIENT_SECRET are set.-- * Runs on a person's own key, and runs that called no provider and were not a cache hit, are not-- * accounted and have no reading. free_allowance_used (§2.1.8) is the share the provider free allowance covered; the-- * cost_estimate the caller passes already leaves that share out. Each reading comes from the onUsage record of an-- * openvibe-sdk/govern meter (server/govern.js createMeter) reserving the run's tokens under the reading's-- * idempotency key, so a key read twice at once yields one reading.++ * quota.js account() writes a run's readings in the SAME transaction as usage_daily: for every attempt that++ * reached a provider (`provider_key`, status != 'skipped'), one reading for each token kind — `in` (input++ * excluding the cached part), `cached` and `out`. A reading's `resource` is the rate-card metric for that++ * kind and model (server/providers/rate-cards.js metricFor, e.g. `input-tokens:gpt-4o-mini`), its `unit` the++ * metric's leading word, and its idempotency key `ai:<run id>:<attempt>:<kind>`. That is what Billing's++ * rate() looks up (`rate_cards WHERE provider = ? AND metric = ?`) and its unitFits() accepts, so the++ * reading is rateable. A cache hit writes its zero reading in the transaction that records the run.++ *++ * The relay (openvibe-sdk createPgOutbox) posts each row with AI's service token. If Billing is down, has no++ * grant yet, or is not deployed, the row waits and is retried with backoff, across restarts. Only Billing++ * refusing the reading itself (400, 409, 413, 422) marks a row rejected. A reading carries ids, the workflow,++ * the requester, tokens and cost. It never carries the input, the output or a prompt. Readings are queued once++ * init() ran, unless USAGE_SAMPLES=off. The relay runs while OV_BILLING_INTERNAL_URL and OV_OAUTH_CLIENT_SECRET++ * are set. Runs on a person's own key, and runs that called no provider and were not a cache hit, are not++ * accounted and have no reading. free_allowance_used (§2.1.8) is the share the provider free allowance covered++ * of that reading's own quantity; its cost_estimate is that quantity's list price less the free share. Each++ * reading comes from the onUsage record of an openvibe-sdk/govern meter (server/govern.js createMeter)++ * reserving its tokens under its idempotency key, so a key read twice at once yields one reading.*/const { createPgOutbox } = require('openvibe-sdk/events');const { createServiceTokenClient } = require('openvibe-sdk/auth');const contracts = require('openvibe-contracts');const { createMeter } = require('./govern');--const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';++const rates = require('./providers/rate-cards');++const TABLE = 'usage_sample_outbox'; const SCHEMA = 'platform.usage-sample@1';++// Billing's unitFits() accepts the card metric's leading word ("GiB" fits gib-delivered; a trailing plural s++// is dropped from the unit, so the full `<kind>-tokens` would not). The metric is `<KINDS[kind]>-tokens[:model]`.++const KINDS = ['in', 'cached', 'out'];++const UNIT = { in: 'input', cached: 'cached-input', out: 'output' };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}`;++const keyOf = (runId, attempt, kind) => `ai:${runId}:${attempt}:${kind}`;const meter = createMeter();--function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {-- 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(),++/** One reading: an attempt's tokens of one kind, under its rate-card metric and the metric's leading-word unit. */++function sampleOf({ runId, attempt, kind, workflowKey, requester, provider = null, model = null, quantity = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {++ const key = keyOf(runId, attempt, kind);++ const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: rates.metricFor(kind, model || null), operation: workflowKey,++ quantity: Math.max(0, Number(quantity) || 0), unit: UNIT[kind], at: new Date(at).toISOString(),cost_estimate: Math.max(0, Number(cost) || 0), source: 'ai.runs' };if (provider && provider !== 'mixed') s.provider = provider;-- // §2.1.8: the tokens the provider free allowance covered (server/free-allowance.js); absent when nothing was free.++ // §2.1.8: the tokens the provider free allowance covered of this reading's quantity (server/free-allowance.js); absent at 0.const free = Math.min(s.quantity, Math.max(0, Math.floor(Number(freeAllowanceUsed) || 0)));if (free > 0) s.free_allowance_used = free;if (traceId) s.trace_id = String(traceId);return s;}--/** The reading: govern's onUsage record (id, subject, quantity) on the run's own fields. */++/** The reading: govern's onUsage record (id, subject, quantity) on the attempt's own fields. */function sampleFromUsage(e, fields) {return { ...sampleOf(fields), id: e.idempotency_key, idempotency_key: e.idempotency_key, subject: e.subject, quantity: e.amount };}--/** MUST be awaited inside the transaction that accounts the run; a second call for the same run changes nothing. */--async function record(db, fields) {-- if (!on) return null;-- const e = await meter({ subject: fields.requester, quantity: Math.max(0, (Number(fields.tokensIn) || 0) + (Number(fields.tokensOut) || 0)), key: keyOf(fields.runId) });-- if (!e) return null; // this run's reading is being written by another call right now-- const s = sampleFromUsage(e, fields);-- const v = contracts.validate(SCHEMA, s);-- if (!v.valid) throw new Error(`usage sample for ${fields.runId} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);-- const ins = await db.prepare(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES (?, ?, ?) ON CONFLICT (event_id) DO NOTHING`).run(s.id, JSON.stringify(s), Date.now());-- await db.prepare('UPDATE runs SET usage_sample_id = ? WHERE id = ?').run(s.id, fields.runId);-- if (ins.changes) queued++;-- if (relay) db.afterCommit(() => relay && relay.kick());-- return s.id;++/**++ * MUST be awaited inside the transaction that accounts the run. `readings` are the run's per-attempt,++ * per-kind descriptors (quota.account builds them from the free allowance's byKind): { attempt, kind,++ * provider, model, quantity, cost, freeAllowanceUsed }. Writes them all, links their keys on the run, and++ * returns the keys. A second call for the same keys changes nothing (ON CONFLICT DO NOTHING).++ */++async function record(db, { runId, workflowKey, requester, at, traceId = null, readings = [] }) {++ if (!on) return [];++ const ids = [];++ for (const r of readings) {++ const fields = { runId, attempt: r.attempt, kind: r.kind, workflowKey, requester, provider: r.provider, model: r.model, quantity: r.quantity, cost: r.cost, freeAllowanceUsed: r.freeAllowanceUsed, at, traceId };++ const key = keyOf(runId, r.attempt, r.kind);++ const e = await meter({ subject: requester, quantity: Math.max(0, Number(r.quantity) || 0), key });++ const s = e ? sampleFromUsage(e, fields) : sampleOf(fields);++ const v = contracts.validate(SCHEMA, s);++ if (!v.valid) throw new Error(`usage sample ${key} for ${runId} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);++ const ins = await db.prepare(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES (?, ?, ?) ON CONFLICT (event_id) DO NOTHING`).run(s.id, JSON.stringify(s), Date.now());++ if (ins.changes) queued++;++ ids.push(key);++ }++ if (ids.length) await db.prepare('UPDATE runs SET usage_sample_ids = ?::text[] WHERE id = ?').run(ids, runId);++ if (ids.length && relay) db.afterCommit(() => relay && relay.kick());++ return ids;}// createPgOutbox's `events` seam pointed at Billing: one reading per POST. isPermanent() reads err.status.function billingSink({ billingUrl, tokens, audience, fetchImpl, timeoutMs = 5000 }) {@@@ -83,27 +105,35 @@ const stateOf = (r) => (r.sent_at != null ? 'sent' : r.rejected_at != null ? 'faconst 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 }.++ * A run's readings for its explain, in the order of `usageSampleIds` (the keys linked on the run):++ * [{ 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];++async function readingsFor(db, usageSampleIds) {++ const ids = (Array.isArray(usageSampleIds) ? usageSampleIds : usageSampleIds ? [usageSampleIds] : []).filter(Boolean);++ if (!ids.length) return [];++ const rows = await db.prepare(`SELECT event_id, envelope, created_at, attempts, next_attempt_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE event_id IN (${ids.map(() => '?').join(', ')})`).all(...ids);++ const byId = new Map(rows.map((r) => [r.event_id, r]));++ return ids.map((id) => {++ const r = byId.get(id);++ if (!r) return { idempotency_key: id, 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;++ });}++/** The run a reading key names: `ai:<run id>:<attempt>:<kind>` (the run id is the second segment). */++const runIdOf = (key) => { const p = String(key || '').split(':'); return p[0] === 'ai' && p[1] ? p[1] : null; };/** 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 })) };++ failures: rows.map((r) => ({ idempotency_key: r.event_id, run_id: runIdOf(r.event_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. */@@@ -113,4 +143,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, sampleFromUsage, keyOf, readingsFor, deliveryReport, status, stop, _reset, TABLE };++module.exports = { init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, status, stop, _reset, TABLE, KINDS, UNIT };@@@ -435,7 +435,7 @@ t.test('run detail: error code, attempts, provider, request log; input bounded at.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`;++ const key = `ai:${runIds.ok}:0:in`;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);@@@ -23,6 +23,8 @@ const cardsWith = (terms) => async (provider, model) => rates.buildCards({ proviconst usedRows = async () => (await tdb.db.prepare('SELECT subject, provider, metric, period_start, period_end, free_used FROM free_allowance_usage ORDER BY subject, metric, period_start').all()).map((r) => ({ ...r, period_start: Number(r.period_start), period_end: Number(r.period_end), free_used: Number(r.free_used) }));const clear = async () => { await tdb.db.prepare('DELETE FROM free_allowance_usage').run(); };++// claim() also returns byKind (per metric, for the readings); the aggregate is what these totals assert.++const agg = (c) => ({ tokens: c.tokens, usd: c.usd });t.test('a migrated database', async () => { tdb = await testDb(); });@@@ -32,7 +34,7 @@ t.test('the subject is the attribution, else the actor, else the requester; noneassert.strictEqual(subjectOf({ requesterType: 'service', requesterId: 'live' }), 'service:live');assert.strictEqual(subjectOf({}), null);const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({ free_allowance: 100, reset_period: 'day' }) });-- assert.deepStrictEqual(await fa.claim(null, 'p', 'm', { in: 50 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });++ assert.deepStrictEqual(agg(await fa.claim(null, 'p', 'm', { in: 50 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });assert.deepStrictEqual(await usedRows(), []);});@@@ -48,7 +50,8 @@ t.test('free up to the allowance per metric, then fully priced; the real periodc = await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000);assert.strictEqual(c.tokens, 10 + 40, 'input exhausted; cached has its own 100; output the 40 left');c = await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000);-- assert.deepStrictEqual(c, { tokens: 0, usd: 0 }, 'fully priced once exhausted');++ assert.deepStrictEqual(agg(c), { tokens: 0, usd: 0 }, 'fully priced once exhausted');++ assert.ok(Math.abs(c.byKind.in.cost - 60 / 1e6) < 1e-15 && Math.abs(c.byKind.out.cost - 120 / 1e6) < 1e-15 && c.byKind.cached.cost === 0, 'byKind carries each metric\'s list cost even when nothing is free');const rows = await usedRows();assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);@@@ -79,9 +82,9 @@ t.test('the counter resets at the period boundary: day and month; none never rest.test('free_allowance 0 (the default) claims nothing and writes nothing', async () => {await clear();const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({}) });-- assert.deepStrictEqual(await fa.claim('s1', 'p', 'm', { in: 100, cached: 10, out: 100 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });++ assert.deepStrictEqual(agg(await fa.claim('s1', 'p', 'm', { in: 100, cached: 10, out: 100 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });const unbilled = createFreeAllowance(tdb.db, { cardsFor: async () => null });-- assert.deepStrictEqual(await unbilled.claim('s1', 'p', 'm', { in: 100 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });++ assert.deepStrictEqual(agg(await unbilled.claim('s1', 'p', 'm', { in: 100 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });assert.deepStrictEqual(await usedRows(), []);});@@@ -125,13 +128,17 @@ t.test('boot with a seam provider whose model has a free allowance of 100 tokenst.test('the first run is free up to the allowance per metric, the next one is fully priced', async () => {assert.strictEqual((await gen()).status, 201);assert.strictEqual((await gen()).status, 201);-- const [a, b] = await readings();-- for (const s of [a, b]) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));-- assert.strictEqual(a.quantity, 2000);-- assert.strictEqual(a.free_allowance_used, 200, '100 input + 100 output tokens free');-- assert.ok(Math.abs(a.cost_estimate - 1.8) < 1e-9, `the free share is not priced: ${a.cost_estimate}`);-- assert.ok(!('free_allowance_used' in b), 'nothing free: the field is absent');-- assert.ok(Math.abs(b.cost_estimate - 2) < 1e-9, `fully priced: ${b.cost_estimate}`);++ const all = await readings();++ for (const s of all) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));++ const runs = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();++ const byKind = (s) => Object.fromEntries(all.filter((x) => x.idempotency_key.startsWith(`ai:${s}:0:`)).map((x) => [x.idempotency_key.split(':')[3], x]));++ const a = byKind(runs[0].id); const b = byKind(runs[1].id);++ assert.deepStrictEqual([a.in.quantity, a.cached.quantity, a.out.quantity], [1000, 0, 1000], 'one reading per kind');++ assert.deepStrictEqual([a.in.free_allowance_used, a.out.free_allowance_used], [100, 100], '100 input + 100 output tokens free, each on its own reading');++ assert.ok(!('free_allowance_used' in a.cached), 'the cached metric was fully free of charge');++ assert.ok(Math.abs(a.in.cost_estimate - 0.9) < 1e-9 && Math.abs(a.out.cost_estimate - 0.9) < 1e-9, `the free share is not priced: ${a.in.cost_estimate}, ${a.out.cost_estimate}`);++ for (const s of [b.in, b.cached, b.out]) assert.ok(!('free_allowance_used' in s), 'nothing free: the field is absent');++ assert.ok(Math.abs(b.in.cost_estimate - 1) < 1e-9 && Math.abs(b.out.cost_estimate - 1) < 1e-9, `fully priced: ${b.in.cost_estimate}, ${b.out.cost_estimate}`);const rows = await h.quotas.freeAllowance.current();assert.deepStrictEqual(rows.map((r) => [r.subject, r.provider, r.metric, r.free_used, r.free_allowance, r.remaining]),[['service:live', 'seam', 'input-tokens:m1', 100, 100, 0], ['service:live', 'seam', 'output-tokens:m1', 100, 100, 0]]);@@@ -140,12 +147,13 @@ 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 () => {++t.test('explain shows the free share on the first run\'s readings 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 explain = async (id) => Object.fromEntries((await request(h.base, 'GET', `/api/v1/runs/${id}`, { tok: live })).body.run.explain.usage_readings.map((u) => [u.idempotency_key.split(':')[3], u]));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)));++ assert.deepStrictEqual([a.in.idempotency_key, a.in.state, a.in.free_allowance_used], [`ai:${first.id}:0:in`, 'queued', 100]);++ assert.strictEqual(a.out.free_allowance_used, 100);++ assert.ok(!('free_allowance_used' in (await explain(next.id)).in));});t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });@@@ -97,7 +97,8 @@ t.test('on the store: free up to the allowance per metric, then fully priced; thassert.ok(Math.abs(c.usd - (60 * 1 + 30 * 2) / 1e6) < 1e-12, `usd ${c.usd}`);assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, out: 30 }, now + 1000)).tokens, 40 + 30);assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000)).tokens, 10 + 40);-- assert.deepStrictEqual(await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000), { tokens: 0, usd: 0 });++ const exhausted = await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000);++ assert.deepStrictEqual([exhausted.tokens, exhausted.usd], [0, 0], 'fully priced once exhausted');const rows = await usedRows();assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);@@@ -182,16 +183,18 @@ t.test('boot with AI_GOVERN_TIERS=1 and a seam model with a free allowance of 10assert.strictEqual(r.status, 201, r.text);});--t.test('the readings carry step 2\'s numbers and come one per accounted run', async () => {++t.test('the readings carry step 2\'s numbers and come one per attempt per metric', async () => {assert.strictEqual((await gen()).status, 201);assert.strictEqual((await gen()).status, 201);const runs = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();-- const [a, b] = await readings();-- assert.deepStrictEqual([a, b].map((s) => s.idempotency_key), runs.map((r) => `ai:${r.id}:tokens`), 'one reading per run, the same key as before');-- assert.strictEqual(a.free_allowance_used, 200);-- assert.ok(Math.abs(a.cost_estimate - 1.8) < 1e-9, `${a.cost_estimate}`);-- assert.ok(!('free_allowance_used' in b));-- assert.ok(Math.abs(b.cost_estimate - 2) < 1e-9, `${b.cost_estimate}`);++ const all = await readings();++ const byKind = (id) => Object.fromEntries(all.filter((x) => x.idempotency_key.startsWith(`ai:${id}:0:`)).map((x) => [x.idempotency_key.split(':')[3], x]));++ const a = byKind(runs[0].id); const b = byKind(runs[1].id);++ assert.deepStrictEqual(Object.keys(a).sort(), ['cached', 'in', 'out'], 'one reading per metric');++ assert.deepStrictEqual([a.in.free_allowance_used, a.out.free_allowance_used], [100, 100]);++ assert.ok(Math.abs(a.in.cost_estimate - 0.9) < 1e-9 && Math.abs(a.out.cost_estimate - 0.9) < 1e-9, `${a.in.cost_estimate}, ${a.out.cost_estimate}`);++ assert.ok(!('free_allowance_used' in b.in) && !('free_allowance_used' in b.out));++ assert.ok(Math.abs(b.in.cost_estimate - 1) < 1e-9 && Math.abs(b.out.cost_estimate - 1) < 1e-9, `${b.in.cost_estimate}, ${b.out.cost_estimate}`);const rows = await h.quotas.freeAllowance.current();assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used, r.remaining]), [['input-tokens:m1', 100, 0], ['output-tokens:m1', 100, 0]]);});@@@ -206,7 +209,7 @@ t.test('a run whose hold no longer fits is refused before the provider; reserve(assert.deepStrictEqual([no.status, no.body.code], [429, 'govern.exceeded'], no.text);assert.strictEqual(seam.calls, calls, 'no provider call');assert.strictEqual(await counted(), before + 1, 'the refused run is not counted');-- assert.strictEqual((await readings()).length, 3);++ assert.strictEqual((await readings()).length, 9, 'three runs, one reading per attempt per metric');});t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });@@@ -44,12 +44,17 @@ t.test("'*' service quotas give each service its own window", async () => {assert.strictEqual(r.status, 201, 'another service is not limited by live\'s usage');});--t.test('every accounted run has one Billing reading; the refused request queued none', async () => {-- const runs = await h.db.prepare("SELECT id, usage_sample_id FROM runs WHERE status = 'succeeded'").all();++t.test('every accounted run has one Billing reading per attempt per token kind; the refused request queued none', async () => {++ const runs = await h.db.prepare("SELECT id, usage_sample_ids FROM runs WHERE status = 'succeeded'").all();assert.ok(runs.length > 0, 'runs succeeded');-- for (const r of runs) assert.strictEqual(r.usage_sample_id, `ai:${r.id}:tokens`, `no reading for ${r.id}`);++ let expected = 0;++ for (const r of runs) {++ const ids = r.usage_sample_ids || [];++ assert.deepStrictEqual(ids, ['in', 'cached', 'out'].map((k) => `ai:${r.id}:0:${k}`), `no readings for ${r.id}`);++ expected += ids.length;++ }const n = Number((await h.db.prepare('SELECT COUNT(*) n FROM usage_sample_outbox').get()).n);-- assert.strictEqual(n, runs.length, 'one reading per succeeded run; the 429 request queued none');++ assert.strictEqual(n, expected, 'one reading per attempt per kind; the 429 request queued none');});t.test('attribution quotas limit one attributed owner (Live per-streamer budget)', async () => {@@@ -1,9 +1,10 @@'use strict';--// platform.usage-sample@1 readings: one per accounted run, queued in the same transaction as the run's--// usage and posted to billing.usage.record by a retry-safe outbox (server/usage-samples.js). A reading--// carries ids, tokens and cost — never the prompt, the input or the output.++// platform.usage-sample@1 readings: one per finished provider attempt per token metric, queued in the same++// transaction as the run's usage and posted to billing.usage.record by a retry-safe outbox (server/usage-samples.js).++// A reading carries ids, tokens and cost — never the prompt, the input or the output.const assert = require('assert');const contracts = require('openvibe-contracts');++const rates = require('../server/providers/rate-cards');const { boot, request, token, suite, seamServer, tmpDir, ALL } = require('./helpers');const t = suite('usage-sample');@@@ -17,6 +18,10 @@ const env = { AI_STUB_FALLBACK: 'false' };const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sent_at, attempts FROM usage_sample_outbox ORDER BY id').all()).map((r) => ({ ...r, envelope: typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope }));++// Billing's ops/rating.js unitFits(): the unit (case/spacing/plural-s aside) is the card metric or its leading word.++const unitKey = (s) => String(s || '').trim().toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').replace(/s$/, '');++const unitFits = (metric, unit) => { const u = unitKey(unit); const m = unitKey(metric); return !!u && (u === m || m.startsWith(`${u}-`)); };++function fakeBilling() {const b = { fail: false, refuse: false, got: [] };b.fetch = async (url, init = {}) => {@@@ -43,35 +48,42 @@ t.test('boot with a counting seam provider as the only route target', async () =assert.strictEqual(r.status, 201, r.text);});--t.test('a succeeded run queues exactly one valid reading that keeps the prompt out', async () => {++t.test('a succeeded run queues one valid reading per attempt per token kind, each under its rate-card metric', async () => {const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'secret prompt words', options: { cache: false } } });assert.strictEqual(r.status, 201, r.text);const runId = r.body.run.id;const rows = await readings();-- assert.strictEqual(rows.length, 1, 'exactly one reading');-- const s = rows[0].envelope;-- const v = contracts.validate('platform.usage-sample@1', s);-- assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);-- assert.strictEqual(s.idempotency_key, `ai:${runId}:tokens`);-- assert.strictEqual(s.id, `ai:${runId}:tokens`);-- assert.strictEqual(rows[0].event_id, s.idempotency_key);-- assert.strictEqual(s.service, 'ai');-- assert.strictEqual(s.operation, 'ai.generate');-- assert.strictEqual(s.unit, 'tokens');-- assert.strictEqual(s.quantity, 2000);-- assert.ok(s.cost_estimate > 0, 'cost is carried');-- assert.strictEqual(s.subject, 'service:live');-- assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');++ assert.deepStrictEqual(rows.map((x) => x.envelope.idempotency_key), ['in', 'cached', 'out'].map((k) => `ai:${runId}:0:${k}`), 'one per kind, in order');++ const qty = { in: 1000, cached: 0, out: 1000 };++ for (const x of rows) {++ const s = x.envelope;++ const kind = s.idempotency_key.split(':')[3];++ const v = contracts.validate('platform.usage-sample@1', s);++ assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);++ assert.strictEqual(x.event_id, s.idempotency_key);++ assert.strictEqual(s.id, s.idempotency_key);++ assert.strictEqual(s.service, 'ai');++ assert.strictEqual(s.operation, 'ai.generate');++ assert.strictEqual(s.provider, 'seam');++ assert.strictEqual(s.resource, rates.metricFor(kind, 'm1'), 'resource is the rate-card metric');++ assert.ok(unitFits(s.resource, s.unit), `unit ${s.unit} fits metric ${s.resource}`);++ assert.strictEqual(s.unit, usageSamples.UNIT[kind]);++ assert.strictEqual(s.quantity, qty[kind], `${kind} quantity`);++ assert.strictEqual(s.subject, 'service:live');++ if (kind === 'in' || kind === 'out') assert.ok(s.cost_estimate > 0, 'cost is carried'); else assert.strictEqual(s.cost_estimate, 0);++ assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');++ }++ // A Billing card with that provider+metric picks the reading, and its unit fits it.++ const card = { id: 'seam:m1/input-tokens', provider: 'seam', metric: rates.metricFor('in', 'm1'), unit_size: 1e6, unit_price_usd: 1000 };++ const inp = rows.find((x) => x.envelope.unit === 'input').envelope;++ assert.deepStrictEqual([card.provider === inp.provider, card.metric === inp.resource, unitFits(card.metric, inp.unit)], [true, true, true], 'rateable');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.deepStrictEqual(got.body.run.usage.usage_sample_ids, rows.map((x) => x.envelope.idempotency_key));++ // explain carries the readings: their keys and delivery state (no relay here, so they wait queued).++ const us = got.body.run.explain.usage_readings;++ assert.deepStrictEqual(us.map((u) => u.idempotency_key), rows.map((x) => x.envelope.idempotency_key));++ assert.ok(us.every((u) => u.state === 'queued' && u.attempts === 0 && u.last_error === null));++ assert.ok(us.every((u) => !('free_allowance_used' in u)), 'nothing was free');assert.ok(Array.isArray(got.body.run.explain.candidates), 'the placement explain is kept');});@@@ -81,20 +93,27 @@ t.test('a cache hit writes its own zero reading', async () => {const b = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body });assert.strictEqual(b.status, 201, b.text);if (b.body.run.status !== 'cached') { assert.ok(a.body.run.id, 'this workflow may not cache'); return; }-- const s = (await readings()).find((r) => r.event_id === `ai:${b.body.run.id}:tokens`).envelope;++ const key = `ai:${b.body.run.id}:0:in`;++ assert.deepStrictEqual(b.body.run.usage.usage_sample_ids, [key]);++ const s = (await readings()).find((r) => r.event_id === key).envelope;const v = contracts.validate('platform.usage-sample@1', s);assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);assert.strictEqual(s.quantity, 0);assert.strictEqual(s.cost_estimate, 0);++ assert.ok(unitFits(s.resource, s.unit));});--t.test('recording the same run again leaves one row, unchanged', async () => {++t.test('recording the same run again leaves the rows, unchanged', async () => {const runId = (await h.db.prepare("SELECT id FROM runs WHERE status = 'succeeded' ORDER BY id LIMIT 1").get()).id;-- const key = `ai:${runId}:tokens`;-- await h.db.tx(async () => { await usageSamples.record(h.db, { runId, workflowKey: 'ai.generate', requester: 'service:live', tokensIn: 1, tokensOut: 0, cost: 0, at: Date.now() }); });-- const rows = (await readings()).filter((r) => r.event_id === key);-- assert.strictEqual(rows.length, 1, 'one row');-- assert.strictEqual(rows[0].envelope.quantity, 2000, 'the first reading is unchanged');++ const before = (await readings()).filter((r) => r.event_id.startsWith(`ai:${runId}:`));++ await h.db.tx(async () => {++ await usageSamples.record(h.db, { runId, workflowKey: 'ai.generate', requester: 'service:live', at: Date.now(),++ readings: before.map((r) => ({ attempt: 0, kind: r.envelope.idempotency_key.split(':')[3], model: 'm1', quantity: r.envelope.quantity, cost: 0, freeAllowanceUsed: 0 })) });++ });++ const after = (await readings()).filter((r) => r.event_id.startsWith(`ai:${runId}:`));++ assert.strictEqual(after.length, before.length, 'no new rows');++ assert.strictEqual(after[0].envelope.quantity, before[0].envelope.quantity, 'the first reading is unchanged');++ assert.strictEqual(after[0].envelope.quantity, 1000);});t.test('a failed send stays queued and replays', async () => {@@@ -104,12 +123,12 @@ t.test('a failed send stays queued and replays', async () => {const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });await relay.flush();// batchSize is 1 and a failed pass stops there: flush until every row has been attempted once.-- for (let i = 0; i < 10 && (await readings()).some((r) => r.sent_at === null && r.attempts === 0); i++) await relay.flush();++ for (let i = 0; i < 20 && (await readings()).some((r) => r.sent_at === null && r.attempts === 0); i++) await relay.flush();const pending = await readings();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];++ const explainOf = async (key) => (await request(h.base, 'GET', `/api/v1/runs/${key.split(':')[1]}`, { tok: live })).body.run.explain.usage_readings.find((u) => u.idempotency_key === key);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/);@@@ -117,6 +136,7 @@ t.test('a failed send stays queued and replays', async () => {b.fail = false;await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();await relay.flush();++ for (let i = 0; i < 20 && (await readings()).some((r) => r.sent_at === null); i++) await relay.flush();const sent = await readings();for (const r of sent) assert.notStrictEqual(r.sent_at, null, 'every row is sent');const keys = b.got.map((s) => s.idempotency_key).sort();@@@ -132,7 +152,7 @@ t.test('the outbox outlives a restart', async () => {b.fail = true; // the still-running relay fails every sendconst r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'survive a restart', options: { cache: false } } });assert.strictEqual(r.status, 201, r.text);-- const key = `ai:${r.body.run.id}:tokens`;++ const key = `ai:${r.body.run.id}:0:in`;await h.stop();h = await boot({ dir: h.dir, env });const row = await h.db.prepare('SELECT sent_at FROM usage_sample_outbox WHERE event_id = ?').get(key);@@@ -147,11 +167,14 @@ t.test('the outbox outlives a restart', async () => {});t.test('free_allowance_used: the free share when some was free, never more than the quantity, absent at 0', () => {-- const base = { runId: 'run_x', workflowKey: 'ai.generate', requester: 'service:live', provider: 'seam', tokensIn: 300, tokensOut: 100, cost: 0.1, at: Date.UTC(2026, 9, 4) };++ const base = { runId: 'run_x', attempt: 2, kind: 'out', workflowKey: 'ai.generate', requester: 'service:live', provider: 'seam', model: 'm1', quantity: 300, cost: 0.1, at: Date.UTC(2026, 9, 4) };const some = usageSamples.sampleOf({ ...base, freeAllowanceUsed: 150 });assert.strictEqual(some.free_allowance_used, 150);++ assert.strictEqual(some.idempotency_key, 'ai:run_x:2:out');++ assert.strictEqual(some.resource, rates.metricFor('out', 'm1'));++ assert.ok(unitFits(some.resource, some.unit));assert.ok(contracts.validate('platform.usage-sample@1', some).valid);-- assert.strictEqual(usageSamples.sampleOf({ ...base, freeAllowanceUsed: 9999 }).free_allowance_used, 400);++ assert.strictEqual(usageSamples.sampleOf({ ...base, freeAllowanceUsed: 9999 }).free_allowance_used, 300);assert.ok(!('free_allowance_used' in usageSamples.sampleOf({ ...base, freeAllowanceUsed: 0 })));assert.ok(!('free_allowance_used' in usageSamples.sampleOf(base)));});@@@ -169,7 +192,7 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on tassert.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.idempotency_key, `ai:${r.body.run.id}:0:in`);assert.strictEqual(u.state, 'failed');assert.match(u.last_error, /answered 400/);assert.ok(u.rejected_at);
new file mode 100644
new file mode 100644
@@ -0,0 +1,6 @@
@@ -0,0 +1,6 @@
+-- phase: expand
+-- B22: a run's Billing readings are one platform.usage-sample@1 per finished provider attempt per token metric
+-- (server/usage-samples.js), keyed ai:<run id>:<attempt>:<kind>. runs.usage_sample_ids links that whole set in
+-- order; runs.usage_sample_id (0005) named the single aggregate reading and is no longer written by AI. Additive
+-- only: a new nullable column, no existing row rewritten, and the old column left in place.
+ALTER TABLE runs ADD COLUMN usage_sample_ids text[];
@@ -5,8 +5,8 @@
@@ -5,8 +5,8 @@
* POST /api/v1/runs create (ai.run.create); ?wait=ms waits for the result
* POST /api/v1/runs create (ai.run.create); ?wait=ms waits for the result
* 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.explain.usage_readings (on every run response): that reading's idempotency key, delivery state (queued/sent/failed), last error, free_allowance_used
+ * run.usage.usage_sample_ids: the run's Billing readings (one per attempt per token metric), null when none
+ * run.explain.usage_readings (on every run response): each 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)
@@ -334,6 +334,7 @@ const ref = (x) => (x ? (x.service ? `${x.service}:${x.type}:${x.id}` : `${x.typ
@@ -334,6 +334,7 @@ const ref = (x) => (x ? (x.service ? `${x.service}:${x.type}:${x.id}` : `${x.typ
function run({ staff, csrf, r, citations, requests, input, output, canCancel, notice, error }) {
function run({ staff, csrf, r, citations, requests, input, output, canCancel, notice, error }) {
const g = r.grounding;
const g = r.grounding;
+ const sampleIds = r.usage.usage_sample_ids || [];
const open = r.status === 'queued' || r.status === 'running';
const open = r.status === 'queued' || r.status === 'running';
return layout({
return layout({
title: `Run ${r.id}`, section: 'runs', staff, csrf, notice, error, body: html`<h1>Run <span class="mono">${r.id}</span> ${badge(r.status)}</h1>
title: `Run ${r.id}`, section: 'runs', staff, csrf, notice, error, body: html`<h1>Run <span class="mono">${r.id}</span> ${badge(r.status)}</h1>
@@ -346,7 +347,7 @@ function run({ staff, csrf, r, citations, requests, input, output, canCancel, no
@@ -346,7 +347,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>` : '—'}${((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>
+<dt>Billing readings</dt><dd>${sampleIds.length ? html`${sampleIds.map((id) => html`<code>${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>
@@ -16,7 +16,10 @@
@@ -16,7 +16,10 @@
* a key was lost and what the console reads. The numbers are the same either way.
* a key was lost and what the console reads. The numbers are the same either way.
*
*
* const free = createFreeAllowance(db, { cardsFor: pool.rateCardsFor });
* const free = createFreeAllowance(db, { cardsFor: pool.rateCardsFor });
- * const { tokens, usd } = await free.claim(subject, provider, model, { in, cached, out }); // inside db.tx
+ * const { tokens, usd, byKind } = await free.claim(subject, provider, model, { in, cached, out }); // inside db.tx
+ *
+ * `byKind[kind]` is { tokens, usd, cost }: that kind's free tokens, their list price, and the list price of its
+ * whole quantity — account() turns each into one reading with its own free_allowance_used and cost_estimate.
*/
*/
const rates = require('./providers/rate-cards');
const rates = require('./providers/rate-cards');
@@ -63,11 +66,16 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },
@@ -63,11 +66,16 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },
/**
/**
* Claim this attempt's free tokens: usage { in (excluding cached), cached, out } on `provider`/`model`.
* Claim this attempt's free tokens: usage { in (excluding cached), cached, out } on `provider`/`model`.
- * Returns { tokens, usd }: the free tokens and their list price on the same cards. A provider with no
- * cards (never billed) or a card with free_allowance 0 claims nothing and writes nothing.
+ * Returns { tokens, usd, byKind }: the free tokens and their list price in total, and, per kind
+ * (`in` | `cached` | `out`), { tokens, usd, cost } — the free tokens, their list price, and the list
+ * price of the whole requested quantity. account() writes one platform.usage-sample@1 per kind from
+ * this, so each reading's free_allowance_used and cost_estimate are its own (server/usage-samples.js).
+ * A provider with no cards (never billed) claims nothing: byKind is all zeros. A card with
+ * free_allowance 0 claims nothing but still reports its list cost (the reading is fully priced).
*/
*/
async function claim(subject, provider, model, usage = {}, now = clock.now()) {
async function claim(subject, provider, model, usage = {}, now = clock.now()) {
- const out = { tokens: 0, usd: 0 };
+ const byKind = Object.fromEntries(KINDS.map((kind) => [kind, { tokens: 0, usd: 0, cost: 0 }]));
+ const out = { tokens: 0, usd: 0, byKind };
if (!subject || !provider) return out;
if (!subject || !provider) return out;
const cards = await cardsFor(provider, model || null);
const cards = await cardsFor(provider, model || null);
if (!cards) return out;
if (!cards) return out;
@@ -77,21 +85,21 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },
@@ -77,21 +85,21 @@ function createFreeAllowance(db, { cardsFor, clock = { now: () => Date.now() },
const c = cards[kind];
const c = cards[kind];
const cap = Math.floor(Number(c && c.free_allowance) || 0);
const cap = Math.floor(Number(c && c.free_allowance) || 0);
const q = Math.max(0, Math.floor(Number(usage[kind]) || 0));
const q = Math.max(0, Math.floor(Number(usage[kind]) || 0));
+ const k = byKind[kind];
+ if (c) k.cost = (q / c.unit_size) * c.unit_price_usd; // the whole quantity's list price, free or not
if (cap <= 0 || q <= 0) continue;
if (cap <= 0 || q <= 0) continue;
const p = periodFor(c, now);
const p = periodFor(c, now);
if (store) {
if (store) {
- const free = await claimOnStore(subject, provider, c, p, q, cap, at, now);
- out.tokens += free;
- out.usd += (free / c.unit_size) * c.unit_price_usd;
- continue;
+ k.tokens = await claimOnStore(subject, provider, c, p, q, cap, at, now);
+ } else {
+ await open.run(subject, provider, c.metric, p.start, p.end, at);
+ const row = await lock.get(subject, provider, c.metric, p.start);
+ const free = Math.min(q, Math.max(0, cap - Number(row.free_used)));
+ if (free) { await take.run(free, at, subject, provider, c.metric, p.start); k.tokens = free; }
}
}
- await open.run(subject, provider, c.metric, p.start, p.end, at);
- const row = await lock.get(subject, provider, c.metric, p.start);
- const free = Math.min(q, Math.max(0, cap - Number(row.free_used)));
- if (!free) continue;
- await take.run(free, at, subject, provider, c.metric, p.start);
- out.tokens += free;
- out.usd += (free / c.unit_size) * c.unit_price_usd;
+ k.usd = (k.tokens / c.unit_size) * c.unit_price_usd;
+ out.tokens += k.tokens;
+ out.usd += k.usd;
}
}
});
});
return out;
return out;
@@ -170,14 +170,16 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
@@ -170,14 +170,16 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
/**
/**
* Add real usage to the reserved windows, the daily usage table and — in the same transaction — the
* Add real usage to the reserved windows, the daily usage table and — in the same transaction — the
- * provider rollup and per-route placement state. `attempts` is the run's finished attempts (the same
- * entries logged to `requests`); `writeUsage` is false when a run spent nothing (stats are still kept).
- * `runId`/`traceId` name the run a reading is written for (server/usage-samples.js): with a runId the
- * run's platform.usage-sample@1 reading commits with its usage. Before it, the run's tokens claim the subject's
- * provider free allowance (server/free-allowance.js, §2.1.8): the free share is the reading's free_allowance_used
- * and its list price is left out of the reading's cost_estimate. Pricing only: the counters above are unchanged.
- * `hold` is the run's tier hold (server/govern.js): once the accounting committed, it is settled with the
- * run's tokens and its priced cost (the free share left out), so a fully free run spends no ai-usd.
+ * provider rollup, per-route placement state and the run's Billing readings. `attempts` is the run's
+ * finished attempts (the same entries logged to `requests`); `writeUsage` is false when a run spent
+ * nothing (stats are still kept). `runId`/`traceId` name the run a reading is written for
+ * (server/usage-samples.js): with a runId, one platform.usage-sample@1 per attempt per token kind
+ * (in excluding cached, cached, out) commits with its usage, under the attempt's provider/model metric.
+ * Each attempt's tokens claim the subject's provider free allowance (server/free-allowance.js, §2.1.8) per
+ * kind: the free share is that reading's free_allowance_used and its list price is left out of the
+ * reading's cost_estimate. Pricing only: the counters above are unchanged. `hold` is the run's tier hold
+ * (server/govern.js): once the accounting committed, it is settled with the run's tokens and its priced
+ * cost (the free share left out), so a fully free run spends no ai-usd.
*/
*/
async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true, runId = null, traceId = null, hold = null }) {
async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true, runId = null, traceId = null, hold = null }) {
const tokens = tokensIn + tokensOut;
const tokens = tokensIn + tokensOut;
@@ -186,11 +188,26 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
@@ -186,11 +188,26 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry, freeAll
await db.tx(async () => {
await db.tx(async () => {
for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);
for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);
if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);
if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);
- // The run's platform.usage-sample@1 reading for Billing commits with its usage (server/usage-samples.js).
+ // One reading per finished attempt per token kind, for Billing (server/usage-samples.js).
if (writeUsage && runId) {
if (writeUsage && runId) {
- const free = freeAllowance ? await freeAllowance.claim(subjectOf(ctx), provider, model, { in: Math.max(0, tokensIn - tokensCached), cached: tokensCached, out: tokensOut }, clock.now()) : { tokens: 0, usd: 0 };
- freeUsd = free.usd;
- await usageSamples.record(db, { runId, workflowKey: ctx.workflowKey, requester: `${ctx.requesterType}:${ctx.requesterId}`, provider, tokensIn, tokensOut, cost: Math.max(0, cost - free.usd), freeAllowanceUsed: free.tokens, at: clock.now(), traceId });
+ const at = clock.now();
+ const readings = [];
+ const list = attempts || [];
+ for (let attempt = 0; attempt < list.length; attempt++) {
+ const a = list[attempt];
+ if (!a || !a.provider_key || a.status === 'skipped') continue; // a skipped attempt never touched a provider
+ const attemptModel = a.model_key || null;
+ const qty = { in: Math.max(0, (Number(a.tokens_in) || 0) - (Number(a.tokens_cached) || 0)), cached: Math.max(0, Number(a.tokens_cached) || 0), out: Math.max(0, Number(a.tokens_out) || 0) };
+ const free = freeAllowance ? await freeAllowance.claim(subjectOf(ctx), a.provider_key, attemptModel, qty, at) : null;
+ const byKind = (free && free.byKind) || {};
+ if (free) freeUsd += Number(free.usd) || 0;
+ for (const kind of usageSamples.KINDS) {
+ const f = byKind[kind] || { tokens: 0, usd: 0, cost: 0 };
+ readings.push({ attempt, kind, provider: a.provider_key, model: attemptModel, quantity: qty[kind],
+ cost: Math.max(0, (Number(f.cost) || 0) - (Number(f.usd) || 0)), freeAllowanceUsed: Number(f.tokens) || 0 });
+ }
+ }
+ await usageSamples.record(db, { runId, workflowKey: ctx.workflowKey, requester: `${ctx.requesterType}:${ctx.requesterId}`, at, traceId, readings });
}
}
await recordStats(day, attempts);
await recordStats(day, attempts);
});
});
@@ -85,7 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -85,7 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
origin: 'ai', workflow: r.workflow_key, workflow_version: r.workflow_version, template_version: r.template_version, route: r.route_key, route_version: r.route_version,
origin: 'ai', workflow: r.workflow_key, workflow_version: r.workflow_version, template_version: r.template_version, route: r.route_key, route_version: r.route_version,
provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),
provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),
},
},
- usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_id: r.usage_sample_id || null },
+ usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_ids: r.usage_sample_ids || (r.usage_sample_id ? [r.usage_sample_id] : null) },
explain: parseJson(r.explain, null),
explain: parseJson(r.explain, null),
citations_count: r.citations_count,
citations_count: r.citations_count,
grounding: parseJson(r.grounding, null),
grounding: parseJson(r.grounding, null),
@@ -215,8 +215,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -215,8 +215,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
if (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
// The reused output carries the grounding it was produced with.
// The reused output carries the grounding it was produced with.
await db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);
await db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);
- // A cache hit is still a served run: its reading (no tokens, no cost) goes to Billing with it.
- await usageSamples.record(db, { runId: id, workflowKey: wf.key, requester: `${requester.type}:${requester.id}`, at: now, traceId: trace });
+ // A cache hit is still a served run: its zero reading (no attempt, no tokens, no cost) goes to Billing with it.
+ await usageSamples.record(db, { runId: id, workflowKey: wf.key, requester: `${requester.type}:${requester.id}`, at: now, traceId: trace,
+ readings: [{ attempt: 0, kind: 'in', model: hit.model_key || null, quantity: 0, cost: 0, freeAllowanceUsed: 0 }] });
await events.runChanged(await getRow(id));
await events.runChanged(await getRow(id));
});
});
await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
@@ -410,7 +411,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -410,7 +411,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
/** The run with its Billing readings in explain.usage_readings (server/usage-samples.js readingsFor); unchanged when it has none. */
/** The run with its Billing readings in explain.usage_readings (server/usage-samples.js readingsFor); unchanged when it has none. */
async function explained(run) {
async function explained(run) {
- const readings = run ? await usageSamples.readingsFor(db, run.usage.usage_sample_id) : [];
+ const ids = run && run.usage ? run.usage.usage_sample_ids : null;
+ const readings = ids && ids.length ? await usageSamples.readingsFor(db, ids) : [];
return readings.length ? { ...run, explain: { ...(run.explain || {}), usage_readings: readings } } : run;
return readings.length ? { ...run, explain: { ...(run.explain || {}), usage_readings: readings } } : run;
}
}
@@ -1,59 +1,81 @@
@@ -1,59 +1,81 @@
'use strict';
'use strict';
/**
/**
- * AI → OpenVibe.Billing: one platform.usage-sample@1 reading per accounted run, posted to
- * billing.usage.record (POST /api/v1/usage, audience openvibe.billing) through a retry-safe outbox.
+ * AI → OpenVibe.Billing: one platform.usage-sample@1 reading per finished provider attempt per token metric,
+ * posted to billing.usage.record (POST /api/v1/usage, audience openvibe.billing) through a retry-safe outbox.
*
*
- * quota.js account() writes the reading in the SAME transaction as usage_daily. A cache hit writes its
- * zero reading in the transaction that records the run. The relay (openvibe-sdk createPgOutbox) posts
- * each row with AI's service token. If Billing is down, has no grant yet, or is not deployed, the row
- * waits and is retried with backoff, across restarts. Only Billing refusing the reading itself (400, 409,
- * 413, 422) marks a row rejected. A reading carries ids, the workflow, the requester, tokens and cost. It
- * never carries the input, the output or a prompt. Readings are queued once init() ran, unless
- * USAGE_SAMPLES=off. The relay runs while OV_BILLING_INTERNAL_URL and OV_OAUTH_CLIENT_SECRET are set.
- * Runs on a person's own key, and runs that called no provider and were not a cache hit, are not
- * accounted and have no reading. free_allowance_used (§2.1.8) is the share the provider free allowance covered; the
- * cost_estimate the caller passes already leaves that share out. Each reading comes from the onUsage record of an
- * openvibe-sdk/govern meter (server/govern.js createMeter) reserving the run's tokens under the reading's
- * idempotency key, so a key read twice at once yields one reading.
+ * quota.js account() writes a run's readings in the SAME transaction as usage_daily: for every attempt that
+ * reached a provider (`provider_key`, status != 'skipped'), one reading for each token kind — `in` (input
+ * excluding the cached part), `cached` and `out`. A reading's `resource` is the rate-card metric for that
+ * kind and model (server/providers/rate-cards.js metricFor, e.g. `input-tokens:gpt-4o-mini`), its `unit` the
+ * metric's leading word, and its idempotency key `ai:<run id>:<attempt>:<kind>`. That is what Billing's
+ * rate() looks up (`rate_cards WHERE provider = ? AND metric = ?`) and its unitFits() accepts, so the
+ * reading is rateable. A cache hit writes its zero reading in the transaction that records the run.
+ *
+ * The relay (openvibe-sdk createPgOutbox) posts each row with AI's service token. If Billing is down, has no
+ * grant yet, or is not deployed, the row waits and is retried with backoff, across restarts. Only Billing
+ * refusing the reading itself (400, 409, 413, 422) marks a row rejected. A reading carries ids, the workflow,
+ * the requester, tokens and cost. It never carries the input, the output or a prompt. Readings are queued once
+ * init() ran, unless USAGE_SAMPLES=off. The relay runs while OV_BILLING_INTERNAL_URL and OV_OAUTH_CLIENT_SECRET
+ * are set. Runs on a person's own key, and runs that called no provider and were not a cache hit, are not
+ * accounted and have no reading. free_allowance_used (§2.1.8) is the share the provider free allowance covered
+ * of that reading's own quantity; its cost_estimate is that quantity's list price less the free share. Each
+ * reading comes from the onUsage record of an openvibe-sdk/govern meter (server/govern.js createMeter)
+ * reserving its tokens under its idempotency key, so a key read twice at once yields one reading.
*/
*/
const { createPgOutbox } = require('openvibe-sdk/events');
const { createPgOutbox } = require('openvibe-sdk/events');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
const { createMeter } = require('./govern');
const { createMeter } = require('./govern');
-const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';
+const rates = require('./providers/rate-cards');
+const TABLE = 'usage_sample_outbox'; const SCHEMA = 'platform.usage-sample@1';
+// Billing's unitFits() accepts the card metric's leading word ("GiB" fits gib-delivered; a trailing plural s
+// is dropped from the unit, so the full `<kind>-tokens` would not). The metric is `<KINDS[kind]>-tokens[:model]`.
+const KINDS = ['in', 'cached', 'out'];
+const UNIT = { in: 'input', cached: 'cached-input', out: 'output' };
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}`;
+const keyOf = (runId, attempt, kind) => `ai:${runId}:${attempt}:${kind}`;
const meter = createMeter();
const meter = createMeter();
-function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {
- 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(),
+/** One reading: an attempt's tokens of one kind, under its rate-card metric and the metric's leading-word unit. */
+function sampleOf({ runId, attempt, kind, workflowKey, requester, provider = null, model = null, quantity = 0, cost = 0, freeAllowanceUsed = 0, at, traceId = null }) {
+ const key = keyOf(runId, attempt, kind);
+ const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: rates.metricFor(kind, model || null), operation: workflowKey,
+ quantity: Math.max(0, Number(quantity) || 0), unit: UNIT[kind], 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' };
if (provider && provider !== 'mixed') s.provider = provider;
if (provider && provider !== 'mixed') s.provider = provider;
- // §2.1.8: the tokens the provider free allowance covered (server/free-allowance.js); absent when nothing was free.
+ // §2.1.8: the tokens the provider free allowance covered of this reading's quantity (server/free-allowance.js); absent at 0.
const free = Math.min(s.quantity, Math.max(0, Math.floor(Number(freeAllowanceUsed) || 0)));
const free = Math.min(s.quantity, Math.max(0, Math.floor(Number(freeAllowanceUsed) || 0)));
if (free > 0) s.free_allowance_used = free;
if (free > 0) s.free_allowance_used = free;
if (traceId) s.trace_id = String(traceId);
if (traceId) s.trace_id = String(traceId);
return s;
return s;
}
}
-/** The reading: govern's onUsage record (id, subject, quantity) on the run's own fields. */
+/** The reading: govern's onUsage record (id, subject, quantity) on the attempt's own fields. */
function sampleFromUsage(e, fields) {
function sampleFromUsage(e, fields) {
return { ...sampleOf(fields), id: e.idempotency_key, idempotency_key: e.idempotency_key, subject: e.subject, quantity: e.amount };
return { ...sampleOf(fields), id: e.idempotency_key, idempotency_key: e.idempotency_key, subject: e.subject, quantity: e.amount };
}
}
-/** MUST be awaited inside the transaction that accounts the run; a second call for the same run changes nothing. */
-async function record(db, fields) {
- if (!on) return null;
- const e = await meter({ subject: fields.requester, quantity: Math.max(0, (Number(fields.tokensIn) || 0) + (Number(fields.tokensOut) || 0)), key: keyOf(fields.runId) });
- if (!e) return null; // this run's reading is being written by another call right now
- const s = sampleFromUsage(e, fields);
- const v = contracts.validate(SCHEMA, s);
- if (!v.valid) throw new Error(`usage sample for ${fields.runId} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);
- const ins = await db.prepare(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES (?, ?, ?) ON CONFLICT (event_id) DO NOTHING`).run(s.id, JSON.stringify(s), Date.now());
- await db.prepare('UPDATE runs SET usage_sample_id = ? WHERE id = ?').run(s.id, fields.runId);
- if (ins.changes) queued++;
- if (relay) db.afterCommit(() => relay && relay.kick());
- return s.id;
+/**
+ * MUST be awaited inside the transaction that accounts the run. `readings` are the run's per-attempt,
+ * per-kind descriptors (quota.account builds them from the free allowance's byKind): { attempt, kind,
+ * provider, model, quantity, cost, freeAllowanceUsed }. Writes them all, links their keys on the run, and
+ * returns the keys. A second call for the same keys changes nothing (ON CONFLICT DO NOTHING).
+ */
+async function record(db, { runId, workflowKey, requester, at, traceId = null, readings = [] }) {
+ if (!on) return [];
+ const ids = [];
+ for (const r of readings) {
+ const fields = { runId, attempt: r.attempt, kind: r.kind, workflowKey, requester, provider: r.provider, model: r.model, quantity: r.quantity, cost: r.cost, freeAllowanceUsed: r.freeAllowanceUsed, at, traceId };
+ const key = keyOf(runId, r.attempt, r.kind);
+ const e = await meter({ subject: requester, quantity: Math.max(0, Number(r.quantity) || 0), key });
+ const s = e ? sampleFromUsage(e, fields) : sampleOf(fields);
+ const v = contracts.validate(SCHEMA, s);
+ if (!v.valid) throw new Error(`usage sample ${key} for ${runId} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);
+ const ins = await db.prepare(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES (?, ?, ?) ON CONFLICT (event_id) DO NOTHING`).run(s.id, JSON.stringify(s), Date.now());
+ if (ins.changes) queued++;
+ ids.push(key);
+ }
+ if (ids.length) await db.prepare('UPDATE runs SET usage_sample_ids = ?::text[] WHERE id = ?').run(ids, runId);
+ if (ids.length && relay) db.afterCommit(() => relay && relay.kick());
+ return ids;
}
}
// createPgOutbox's `events` seam pointed at Billing: one reading per POST. isPermanent() reads err.status.
// createPgOutbox's `events` seam pointed at Billing: one reading per POST. isPermanent() reads err.status.
function billingSink({ billingUrl, tokens, audience, fetchImpl, timeoutMs = 5000 }) {
function billingSink({ billingUrl, tokens, audience, fetchImpl, timeoutMs = 5000 }) {
@@ -83,27 +105,35 @@ const stateOf = (r) => (r.sent_at != null ? 'sent' : r.rejected_at != null ? 'fa
@@ -83,27 +105,35 @@ const stateOf = (r) => (r.sent_at != null ? 'sent' : r.rejected_at != null ? 'fa
const isoOrNull = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const isoOrNull = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const envelopeOf = (e) => (typeof e === 'string' ? JSON.parse(e) : e) || {};
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 }.
+ * A run's readings for its explain, in the order of `usageSampleIds` (the keys linked on the run):
+ * [{ 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];
+async function readingsFor(db, usageSampleIds) {
+ const ids = (Array.isArray(usageSampleIds) ? usageSampleIds : usageSampleIds ? [usageSampleIds] : []).filter(Boolean);
+ if (!ids.length) return [];
+ const rows = await db.prepare(`SELECT event_id, envelope, created_at, attempts, next_attempt_at, sent_at, rejected_at, last_error FROM ${TABLE} WHERE event_id IN (${ids.map(() => '?').join(', ')})`).all(...ids);
+ const byId = new Map(rows.map((r) => [r.event_id, r]));
+ return ids.map((id) => {
+ const r = byId.get(id);
+ if (!r) return { idempotency_key: id, 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;
+ });
}
}
+/** The run a reading key names: `ai:<run id>:<attempt>:<kind>` (the run id is the second segment). */
+const runIdOf = (key) => { const p = String(key || '').split(':'); return p[0] === 'ai' && p[1] ? p[1] : null; };
/** Staff console: readings by delivery state and the latest rows carrying a send error (never a public page: keys name runs). */
/** 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 } = {}) {
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,
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();
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);
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) },
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 })) };
+ failures: rows.map((r) => ({ idempotency_key: r.event_id, run_id: runIdOf(r.event_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. */
@@ -113,4 +143,4 @@ async function stop({ drainMs = 3000 } = {}) {
@@ -113,4 +143,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, sampleFromUsage, keyOf, readingsFor, deliveryReport, status, stop, _reset, TABLE };
+module.exports = { init, record, sampleOf, sampleFromUsage, keyOf, runIdOf, readingsFor, deliveryReport, status, stop, _reset, TABLE, KINDS, UNIT };
@@ -435,7 +435,7 @@ t.test('run detail: error code, attempts, provider, request log; input bounded a
@@ -435,7 +435,7 @@ t.test('run detail: error code, attempts, provider, request log; input bounded a
t.test('usage: free allowance per subject this period and Billing readings by state, with the last send errors', async () => {
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' } });
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);
assert.strictEqual((await h.quotas.freeAllowance.claim('live:user:4242', 'okseam', 'seam-model', { in: 120 })).tokens, 120);
- const key = `ai:${runIds.ok}:tokens`;
+ const key = `ai:${runIds.ok}:0:in`;
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');
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);
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);
const r = await get('/console/usage', cookies.admin);
@@ -23,6 +23,8 @@ const cardsWith = (terms) => async (provider, model) => rates.buildCards({ provi
@@ -23,6 +23,8 @@ const cardsWith = (terms) => async (provider, model) => rates.buildCards({ provi
const usedRows = async () => (await tdb.db.prepare('SELECT subject, provider, metric, period_start, period_end, free_used FROM free_allowance_usage ORDER BY subject, metric, period_start').all())
const usedRows = async () => (await tdb.db.prepare('SELECT subject, provider, metric, period_start, period_end, free_used FROM free_allowance_usage ORDER BY subject, metric, period_start').all())
.map((r) => ({ ...r, period_start: Number(r.period_start), period_end: Number(r.period_end), free_used: Number(r.free_used) }));
.map((r) => ({ ...r, period_start: Number(r.period_start), period_end: Number(r.period_end), free_used: Number(r.free_used) }));
const clear = async () => { await tdb.db.prepare('DELETE FROM free_allowance_usage').run(); };
const clear = async () => { await tdb.db.prepare('DELETE FROM free_allowance_usage').run(); };
+// claim() also returns byKind (per metric, for the readings); the aggregate is what these totals assert.
+const agg = (c) => ({ tokens: c.tokens, usd: c.usd });
t.test('a migrated database', async () => { tdb = await testDb(); });
t.test('a migrated database', async () => { tdb = await testDb(); });
@@ -32,7 +34,7 @@ t.test('the subject is the attribution, else the actor, else the requester; none
@@ -32,7 +34,7 @@ t.test('the subject is the attribution, else the actor, else the requester; none
assert.strictEqual(subjectOf({ requesterType: 'service', requesterId: 'live' }), 'service:live');
assert.strictEqual(subjectOf({ requesterType: 'service', requesterId: 'live' }), 'service:live');
assert.strictEqual(subjectOf({}), null);
assert.strictEqual(subjectOf({}), null);
const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({ free_allowance: 100, reset_period: 'day' }) });
const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({ free_allowance: 100, reset_period: 'day' }) });
- assert.deepStrictEqual(await fa.claim(null, 'p', 'm', { in: 50 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });
+ assert.deepStrictEqual(agg(await fa.claim(null, 'p', 'm', { in: 50 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });
assert.deepStrictEqual(await usedRows(), []);
assert.deepStrictEqual(await usedRows(), []);
});
});
@@ -48,7 +50,8 @@ t.test('free up to the allowance per metric, then fully priced; the real period
@@ -48,7 +50,8 @@ t.test('free up to the allowance per metric, then fully priced; the real period
c = await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000);
c = await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000);
assert.strictEqual(c.tokens, 10 + 40, 'input exhausted; cached has its own 100; output the 40 left');
assert.strictEqual(c.tokens, 10 + 40, 'input exhausted; cached has its own 100; output the 40 left');
c = await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000);
c = await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000);
- assert.deepStrictEqual(c, { tokens: 0, usd: 0 }, 'fully priced once exhausted');
+ assert.deepStrictEqual(agg(c), { tokens: 0, usd: 0 }, 'fully priced once exhausted');
+ assert.ok(Math.abs(c.byKind.in.cost - 60 / 1e6) < 1e-15 && Math.abs(c.byKind.out.cost - 120 / 1e6) < 1e-15 && c.byKind.cached.cost === 0, 'byKind carries each metric\'s list cost even when nothing is free');
const rows = await usedRows();
const rows = await usedRows();
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);
for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);
for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);
@@ -79,9 +82,9 @@ t.test('the counter resets at the period boundary: day and month; none never res
@@ -79,9 +82,9 @@ t.test('the counter resets at the period boundary: day and month; none never res
t.test('free_allowance 0 (the default) claims nothing and writes nothing', async () => {
t.test('free_allowance 0 (the default) claims nothing and writes nothing', async () => {
await clear();
await clear();
const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({}) });
const fa = createFreeAllowance(tdb.db, { cardsFor: cardsWith({}) });
- assert.deepStrictEqual(await fa.claim('s1', 'p', 'm', { in: 100, cached: 10, out: 100 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });
+ assert.deepStrictEqual(agg(await fa.claim('s1', 'p', 'm', { in: 100, cached: 10, out: 100 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });
const unbilled = createFreeAllowance(tdb.db, { cardsFor: async () => null });
const unbilled = createFreeAllowance(tdb.db, { cardsFor: async () => null });
- assert.deepStrictEqual(await unbilled.claim('s1', 'p', 'm', { in: 100 }, at(2026, 10, 4)), { tokens: 0, usd: 0 });
+ assert.deepStrictEqual(agg(await unbilled.claim('s1', 'p', 'm', { in: 100 }, at(2026, 10, 4))), { tokens: 0, usd: 0 });
assert.deepStrictEqual(await usedRows(), []);
assert.deepStrictEqual(await usedRows(), []);
});
});
@@ -125,13 +128,17 @@ t.test('boot with a seam provider whose model has a free allowance of 100 tokens
@@ -125,13 +128,17 @@ t.test('boot with a seam provider whose model has a free allowance of 100 tokens
t.test('the first run is free up to the allowance per metric, the next one is fully priced', async () => {
t.test('the first run is free up to the allowance per metric, the next one is fully priced', async () => {
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
- const [a, b] = await readings();
- for (const s of [a, b]) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));
- assert.strictEqual(a.quantity, 2000);
- assert.strictEqual(a.free_allowance_used, 200, '100 input + 100 output tokens free');
- assert.ok(Math.abs(a.cost_estimate - 1.8) < 1e-9, `the free share is not priced: ${a.cost_estimate}`);
- assert.ok(!('free_allowance_used' in b), 'nothing free: the field is absent');
- assert.ok(Math.abs(b.cost_estimate - 2) < 1e-9, `fully priced: ${b.cost_estimate}`);
+ const all = await readings();
+ for (const s of all) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));
+ const runs = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();
+ const byKind = (s) => Object.fromEntries(all.filter((x) => x.idempotency_key.startsWith(`ai:${s}:0:`)).map((x) => [x.idempotency_key.split(':')[3], x]));
+ const a = byKind(runs[0].id); const b = byKind(runs[1].id);
+ assert.deepStrictEqual([a.in.quantity, a.cached.quantity, a.out.quantity], [1000, 0, 1000], 'one reading per kind');
+ assert.deepStrictEqual([a.in.free_allowance_used, a.out.free_allowance_used], [100, 100], '100 input + 100 output tokens free, each on its own reading');
+ assert.ok(!('free_allowance_used' in a.cached), 'the cached metric was fully free of charge');
+ assert.ok(Math.abs(a.in.cost_estimate - 0.9) < 1e-9 && Math.abs(a.out.cost_estimate - 0.9) < 1e-9, `the free share is not priced: ${a.in.cost_estimate}, ${a.out.cost_estimate}`);
+ for (const s of [b.in, b.cached, b.out]) assert.ok(!('free_allowance_used' in s), 'nothing free: the field is absent');
+ assert.ok(Math.abs(b.in.cost_estimate - 1) < 1e-9 && Math.abs(b.out.cost_estimate - 1) < 1e-9, `fully priced: ${b.in.cost_estimate}, ${b.out.cost_estimate}`);
const rows = await h.quotas.freeAllowance.current();
const rows = await h.quotas.freeAllowance.current();
assert.deepStrictEqual(rows.map((r) => [r.subject, r.provider, r.metric, r.free_used, r.free_allowance, r.remaining]),
assert.deepStrictEqual(rows.map((r) => [r.subject, r.provider, r.metric, r.free_used, r.free_allowance, r.remaining]),
[['service:live', 'seam', 'input-tokens:m1', 100, 100, 0], ['service:live', 'seam', 'output-tokens:m1', 100, 100, 0]]);
[['service:live', 'seam', 'input-tokens:m1', 100, 100, 0], ['service:live', 'seam', 'output-tokens:m1', 100, 100, 0]]);
@@ -140,12 +147,13 @@ t.test('the first run is free up to the allowance per metric, the next one is fu
@@ -140,12 +147,13 @@ 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 () => {
+t.test('explain shows the free share on the first run\'s readings 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 [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 explain = async (id) => Object.fromEntries((await request(h.base, 'GET', `/api/v1/runs/${id}`, { tok: live })).body.run.explain.usage_readings.map((u) => [u.idempotency_key.split(':')[3], u]));
const a = await explain(first.id);
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)));
+ assert.deepStrictEqual([a.in.idempotency_key, a.in.state, a.in.free_allowance_used], [`ai:${first.id}:0:in`, 'queued', 100]);
+ assert.strictEqual(a.out.free_allowance_used, 100);
+ assert.ok(!('free_allowance_used' in (await explain(next.id)).in));
});
});
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
@@ -97,7 +97,8 @@ t.test('on the store: free up to the allowance per metric, then fully priced; th
@@ -97,7 +97,8 @@ t.test('on the store: free up to the allowance per metric, then fully priced; th
assert.ok(Math.abs(c.usd - (60 * 1 + 30 * 2) / 1e6) < 1e-12, `usd ${c.usd}`);
assert.ok(Math.abs(c.usd - (60 * 1 + 30 * 2) / 1e6) < 1e-12, `usd ${c.usd}`);
assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, out: 30 }, now + 1000)).tokens, 40 + 30);
assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, out: 30 }, now + 1000)).tokens, 40 + 30);
assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000)).tokens, 10 + 40);
assert.strictEqual((await fa.claim('s1', 'p', 'm', { in: 60, cached: 10, out: 100 }, now + 2000)).tokens, 10 + 40);
- assert.deepStrictEqual(await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000), { tokens: 0, usd: 0 });
+ const exhausted = await fa.claim('s1', 'p', 'm', { in: 60, out: 60 }, now + 3000);
+ assert.deepStrictEqual([exhausted.tokens, exhausted.usd], [0, 0], 'fully priced once exhausted');
const rows = await usedRows();
const rows = await usedRows();
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used]), [['cached-input-tokens:m', 10], ['input-tokens:m', 100], ['output-tokens:m', 100]]);
for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);
for (const r of rows) assert.deepStrictEqual([r.period_start, r.period_end], [Date.UTC(2026, 9, 4), Date.UTC(2026, 9, 5)]);
@@ -182,16 +183,18 @@ t.test('boot with AI_GOVERN_TIERS=1 and a seam model with a free allowance of 10
@@ -182,16 +183,18 @@ t.test('boot with AI_GOVERN_TIERS=1 and a seam model with a free allowance of 10
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);
});
});
-t.test('the readings carry step 2\'s numbers and come one per accounted run', async () => {
+t.test('the readings carry step 2\'s numbers and come one per attempt per metric', async () => {
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
assert.strictEqual((await gen()).status, 201);
const runs = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();
const runs = await h.db.prepare('SELECT id FROM runs ORDER BY created_at, id').all();
- const [a, b] = await readings();
- assert.deepStrictEqual([a, b].map((s) => s.idempotency_key), runs.map((r) => `ai:${r.id}:tokens`), 'one reading per run, the same key as before');
- assert.strictEqual(a.free_allowance_used, 200);
- assert.ok(Math.abs(a.cost_estimate - 1.8) < 1e-9, `${a.cost_estimate}`);
- assert.ok(!('free_allowance_used' in b));
- assert.ok(Math.abs(b.cost_estimate - 2) < 1e-9, `${b.cost_estimate}`);
+ const all = await readings();
+ const byKind = (id) => Object.fromEntries(all.filter((x) => x.idempotency_key.startsWith(`ai:${id}:0:`)).map((x) => [x.idempotency_key.split(':')[3], x]));
+ const a = byKind(runs[0].id); const b = byKind(runs[1].id);
+ assert.deepStrictEqual(Object.keys(a).sort(), ['cached', 'in', 'out'], 'one reading per metric');
+ assert.deepStrictEqual([a.in.free_allowance_used, a.out.free_allowance_used], [100, 100]);
+ assert.ok(Math.abs(a.in.cost_estimate - 0.9) < 1e-9 && Math.abs(a.out.cost_estimate - 0.9) < 1e-9, `${a.in.cost_estimate}, ${a.out.cost_estimate}`);
+ assert.ok(!('free_allowance_used' in b.in) && !('free_allowance_used' in b.out));
+ assert.ok(Math.abs(b.in.cost_estimate - 1) < 1e-9 && Math.abs(b.out.cost_estimate - 1) < 1e-9, `${b.in.cost_estimate}, ${b.out.cost_estimate}`);
const rows = await h.quotas.freeAllowance.current();
const rows = await h.quotas.freeAllowance.current();
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used, r.remaining]), [['input-tokens:m1', 100, 0], ['output-tokens:m1', 100, 0]]);
assert.deepStrictEqual(rows.map((r) => [r.metric, r.free_used, r.remaining]), [['input-tokens:m1', 100, 0], ['output-tokens:m1', 100, 0]]);
});
});
@@ -206,7 +209,7 @@ t.test('a run whose hold no longer fits is refused before the provider; reserve(
@@ -206,7 +209,7 @@ t.test('a run whose hold no longer fits is refused before the provider; reserve(
assert.deepStrictEqual([no.status, no.body.code], [429, 'govern.exceeded'], no.text);
assert.deepStrictEqual([no.status, no.body.code], [429, 'govern.exceeded'], no.text);
assert.strictEqual(seam.calls, calls, 'no provider call');
assert.strictEqual(seam.calls, calls, 'no provider call');
assert.strictEqual(await counted(), before + 1, 'the refused run is not counted');
assert.strictEqual(await counted(), before + 1, 'the refused run is not counted');
- assert.strictEqual((await readings()).length, 3);
+ assert.strictEqual((await readings()).length, 9, 'three runs, one reading per attempt per metric');
});
});
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
@@ -44,12 +44,17 @@ t.test("'*' service quotas give each service its own window", async () => {
@@ -44,12 +44,17 @@ t.test("'*' service quotas give each service its own window", async () => {
assert.strictEqual(r.status, 201, 'another service is not limited by live\'s usage');
assert.strictEqual(r.status, 201, 'another service is not limited by live\'s usage');
});
});
-t.test('every accounted run has one Billing reading; the refused request queued none', async () => {
- const runs = await h.db.prepare("SELECT id, usage_sample_id FROM runs WHERE status = 'succeeded'").all();
+t.test('every accounted run has one Billing reading per attempt per token kind; the refused request queued none', async () => {
+ const runs = await h.db.prepare("SELECT id, usage_sample_ids FROM runs WHERE status = 'succeeded'").all();
assert.ok(runs.length > 0, 'runs succeeded');
assert.ok(runs.length > 0, 'runs succeeded');
- for (const r of runs) assert.strictEqual(r.usage_sample_id, `ai:${r.id}:tokens`, `no reading for ${r.id}`);
+ let expected = 0;
+ for (const r of runs) {
+ const ids = r.usage_sample_ids || [];
+ assert.deepStrictEqual(ids, ['in', 'cached', 'out'].map((k) => `ai:${r.id}:0:${k}`), `no readings for ${r.id}`);
+ expected += ids.length;
+ }
const n = Number((await h.db.prepare('SELECT COUNT(*) n FROM usage_sample_outbox').get()).n);
const n = Number((await h.db.prepare('SELECT COUNT(*) n FROM usage_sample_outbox').get()).n);
- assert.strictEqual(n, runs.length, 'one reading per succeeded run; the 429 request queued none');
+ assert.strictEqual(n, expected, 'one reading per attempt per kind; the 429 request queued none');
});
});
t.test('attribution quotas limit one attributed owner (Live per-streamer budget)', async () => {
t.test('attribution quotas limit one attributed owner (Live per-streamer budget)', async () => {
@@ -1,9 +1,10 @@
@@ -1,9 +1,10 @@
'use strict';
'use strict';
-// platform.usage-sample@1 readings: one per accounted run, queued in the same transaction as the run's
-// usage and posted to billing.usage.record by a retry-safe outbox (server/usage-samples.js). A reading
-// carries ids, tokens and cost — never the prompt, the input or the output.
+// platform.usage-sample@1 readings: one per finished provider attempt per token metric, queued in the same
+// transaction as the run's usage and posted to billing.usage.record by a retry-safe outbox (server/usage-samples.js).
+// A reading carries ids, tokens and cost — never the prompt, the input or the output.
const assert = require('assert');
const assert = require('assert');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
+const rates = require('../server/providers/rate-cards');
const { boot, request, token, suite, seamServer, tmpDir, ALL } = require('./helpers');
const { boot, request, token, suite, seamServer, tmpDir, ALL } = require('./helpers');
const t = suite('usage-sample');
const t = suite('usage-sample');
@@ -17,6 +18,10 @@ const env = { AI_STUB_FALLBACK: 'false' };
@@ -17,6 +18,10 @@ const env = { AI_STUB_FALLBACK: 'false' };
const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sent_at, attempts FROM usage_sample_outbox ORDER BY id').all())
const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sent_at, attempts FROM usage_sample_outbox ORDER BY id').all())
.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 }));
+// Billing's ops/rating.js unitFits(): the unit (case/spacing/plural-s aside) is the card metric or its leading word.
+const unitKey = (s) => String(s || '').trim().toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').replace(/s$/, '');
+const unitFits = (metric, unit) => { const u = unitKey(unit); const m = unitKey(metric); return !!u && (u === m || m.startsWith(`${u}-`)); };
+
function fakeBilling() {
function fakeBilling() {
const b = { fail: false, refuse: false, got: [] };
const b = { fail: false, refuse: false, got: [] };
b.fetch = async (url, init = {}) => {
b.fetch = async (url, init = {}) => {
@@ -43,35 +48,42 @@ t.test('boot with a counting seam provider as the only route target', async () =
@@ -43,35 +48,42 @@ t.test('boot with a counting seam provider as the only route target', async () =
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);
});
});
-t.test('a succeeded run queues exactly one valid reading that keeps the prompt out', async () => {
+t.test('a succeeded run queues one valid reading per attempt per token kind, each under its rate-card metric', async () => {
const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'secret prompt words', options: { cache: false } } });
const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'secret prompt words', options: { cache: false } } });
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);
const runId = r.body.run.id;
const runId = r.body.run.id;
const rows = await readings();
const rows = await readings();
- assert.strictEqual(rows.length, 1, 'exactly one reading');
- const s = rows[0].envelope;
- const v = contracts.validate('platform.usage-sample@1', s);
- assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
- assert.strictEqual(s.idempotency_key, `ai:${runId}:tokens`);
- assert.strictEqual(s.id, `ai:${runId}:tokens`);
- assert.strictEqual(rows[0].event_id, s.idempotency_key);
- assert.strictEqual(s.service, 'ai');
- assert.strictEqual(s.operation, 'ai.generate');
- assert.strictEqual(s.unit, 'tokens');
- assert.strictEqual(s.quantity, 2000);
- assert.ok(s.cost_estimate > 0, 'cost is carried');
- assert.strictEqual(s.subject, 'service:live');
- assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');
+ assert.deepStrictEqual(rows.map((x) => x.envelope.idempotency_key), ['in', 'cached', 'out'].map((k) => `ai:${runId}:0:${k}`), 'one per kind, in order');
+ const qty = { in: 1000, cached: 0, out: 1000 };
+ for (const x of rows) {
+ const s = x.envelope;
+ const kind = s.idempotency_key.split(':')[3];
+ const v = contracts.validate('platform.usage-sample@1', s);
+ assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
+ assert.strictEqual(x.event_id, s.idempotency_key);
+ assert.strictEqual(s.id, s.idempotency_key);
+ assert.strictEqual(s.service, 'ai');
+ assert.strictEqual(s.operation, 'ai.generate');
+ assert.strictEqual(s.provider, 'seam');
+ assert.strictEqual(s.resource, rates.metricFor(kind, 'm1'), 'resource is the rate-card metric');
+ assert.ok(unitFits(s.resource, s.unit), `unit ${s.unit} fits metric ${s.resource}`);
+ assert.strictEqual(s.unit, usageSamples.UNIT[kind]);
+ assert.strictEqual(s.quantity, qty[kind], `${kind} quantity`);
+ assert.strictEqual(s.subject, 'service:live');
+ if (kind === 'in' || kind === 'out') assert.ok(s.cost_estimate > 0, 'cost is carried'); else assert.strictEqual(s.cost_estimate, 0);
+ assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');
+ }
+ // A Billing card with that provider+metric picks the reading, and its unit fits it.
+ const card = { id: 'seam:m1/input-tokens', provider: 'seam', metric: rates.metricFor('in', 'm1'), unit_size: 1e6, unit_price_usd: 1000 };
+ const inp = rows.find((x) => x.envelope.unit === 'input').envelope;
+ assert.deepStrictEqual([card.provider === inp.provider, card.metric === inp.resource, unitFits(card.metric, inp.unit)], [true, true, true], 'rateable');
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`);
- // 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.deepStrictEqual(got.body.run.usage.usage_sample_ids, rows.map((x) => x.envelope.idempotency_key));
+ // explain carries the readings: their keys and delivery state (no relay here, so they wait queued).
+ const us = got.body.run.explain.usage_readings;
+ assert.deepStrictEqual(us.map((u) => u.idempotency_key), rows.map((x) => x.envelope.idempotency_key));
+ assert.ok(us.every((u) => u.state === 'queued' && u.attempts === 0 && u.last_error === null));
+ assert.ok(us.every((u) => !('free_allowance_used' in u)), 'nothing was free');
assert.ok(Array.isArray(got.body.run.explain.candidates), 'the placement explain is kept');
assert.ok(Array.isArray(got.body.run.explain.candidates), 'the placement explain is kept');
});
});
@@ -81,20 +93,27 @@ t.test('a cache hit writes its own zero reading', async () => {
@@ -81,20 +93,27 @@ t.test('a cache hit writes its own zero reading', async () => {
const b = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body });
const b = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body });
assert.strictEqual(b.status, 201, b.text);
assert.strictEqual(b.status, 201, b.text);
if (b.body.run.status !== 'cached') { assert.ok(a.body.run.id, 'this workflow may not cache'); return; }
if (b.body.run.status !== 'cached') { assert.ok(a.body.run.id, 'this workflow may not cache'); return; }
- const s = (await readings()).find((r) => r.event_id === `ai:${b.body.run.id}:tokens`).envelope;
+ const key = `ai:${b.body.run.id}:0:in`;
+ assert.deepStrictEqual(b.body.run.usage.usage_sample_ids, [key]);
+ const s = (await readings()).find((r) => r.event_id === key).envelope;
const v = contracts.validate('platform.usage-sample@1', s);
const v = contracts.validate('platform.usage-sample@1', s);
assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
assert.strictEqual(s.quantity, 0);
assert.strictEqual(s.quantity, 0);
assert.strictEqual(s.cost_estimate, 0);
assert.strictEqual(s.cost_estimate, 0);
+ assert.ok(unitFits(s.resource, s.unit));
});
});
-t.test('recording the same run again leaves one row, unchanged', async () => {
+t.test('recording the same run again leaves the rows, unchanged', async () => {
const runId = (await h.db.prepare("SELECT id FROM runs WHERE status = 'succeeded' ORDER BY id LIMIT 1").get()).id;
const runId = (await h.db.prepare("SELECT id FROM runs WHERE status = 'succeeded' ORDER BY id LIMIT 1").get()).id;
- const key = `ai:${runId}:tokens`;
- await h.db.tx(async () => { await usageSamples.record(h.db, { runId, workflowKey: 'ai.generate', requester: 'service:live', tokensIn: 1, tokensOut: 0, cost: 0, at: Date.now() }); });
- const rows = (await readings()).filter((r) => r.event_id === key);
- assert.strictEqual(rows.length, 1, 'one row');
- assert.strictEqual(rows[0].envelope.quantity, 2000, 'the first reading is unchanged');
+ const before = (await readings()).filter((r) => r.event_id.startsWith(`ai:${runId}:`));
+ await h.db.tx(async () => {
+ await usageSamples.record(h.db, { runId, workflowKey: 'ai.generate', requester: 'service:live', at: Date.now(),
+ readings: before.map((r) => ({ attempt: 0, kind: r.envelope.idempotency_key.split(':')[3], model: 'm1', quantity: r.envelope.quantity, cost: 0, freeAllowanceUsed: 0 })) });
+ });
+ const after = (await readings()).filter((r) => r.event_id.startsWith(`ai:${runId}:`));
+ assert.strictEqual(after.length, before.length, 'no new rows');
+ assert.strictEqual(after[0].envelope.quantity, before[0].envelope.quantity, 'the first reading is unchanged');
+ assert.strictEqual(after[0].envelope.quantity, 1000);
});
});
t.test('a failed send stays queued and replays', async () => {
t.test('a failed send stays queued and replays', async () => {
@@ -104,12 +123,12 @@ t.test('a failed send stays queued and replays', async () => {
@@ -104,12 +123,12 @@ t.test('a failed send stays queued and replays', async () => {
const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });
const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });
await relay.flush();
await relay.flush();
// batchSize is 1 and a failed pass stops there: flush until every row has been attempted once.
// batchSize is 1 and a failed pass stops there: flush until every row has been attempted once.
- for (let i = 0; i < 10 && (await readings()).some((r) => r.sent_at === null && r.attempts === 0); i++) await relay.flush();
+ for (let i = 0; i < 20 && (await readings()).some((r) => r.sent_at === null && r.attempts === 0); i++) await relay.flush();
const pending = await readings();
const pending = await readings();
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];
+ const explainOf = async (key) => (await request(h.base, 'GET', `/api/v1/runs/${key.split(':')[1]}`, { tok: live })).body.run.explain.usage_readings.find((u) => u.idempotency_key === key);
let u = await explainOf(pending[0].event_id);
let u = await explainOf(pending[0].event_id);
assert.strictEqual(u.state, 'queued', 'a transient failure is retried');
assert.strictEqual(u.state, 'queued', 'a transient failure is retried');
assert.match(u.last_error, /billing\.usage\.record answered 503/);
assert.match(u.last_error, /billing\.usage\.record answered 503/);
@@ -117,6 +136,7 @@ t.test('a failed send stays queued and replays', async () => {
@@ -117,6 +136,7 @@ t.test('a failed send stays queued and replays', async () => {
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();
+ for (let i = 0; i < 20 && (await readings()).some((r) => r.sent_at === null); i++) await relay.flush();
const sent = await readings();
const sent = await readings();
for (const r of sent) assert.notStrictEqual(r.sent_at, null, 'every row is sent');
for (const r of sent) assert.notStrictEqual(r.sent_at, null, 'every row is sent');
const keys = b.got.map((s) => s.idempotency_key).sort();
const keys = b.got.map((s) => s.idempotency_key).sort();
@@ -132,7 +152,7 @@ t.test('the outbox outlives a restart', async () => {
@@ -132,7 +152,7 @@ t.test('the outbox outlives a restart', async () => {
b.fail = true; // the still-running relay fails every send
b.fail = true; // the still-running relay fails every send
const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'survive a restart', options: { cache: false } } });
const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'survive a restart', options: { cache: false } } });
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);
- const key = `ai:${r.body.run.id}:tokens`;
+ const key = `ai:${r.body.run.id}:0:in`;
await h.stop();
await h.stop();
h = await boot({ dir: h.dir, env });
h = await boot({ dir: h.dir, env });
const row = await h.db.prepare('SELECT sent_at FROM usage_sample_outbox WHERE event_id = ?').get(key);
const row = await h.db.prepare('SELECT sent_at FROM usage_sample_outbox WHERE event_id = ?').get(key);
@@ -147,11 +167,14 @@ t.test('the outbox outlives a restart', async () => {
@@ -147,11 +167,14 @@ t.test('the outbox outlives a restart', async () => {
});
});
t.test('free_allowance_used: the free share when some was free, never more than the quantity, absent at 0', () => {
t.test('free_allowance_used: the free share when some was free, never more than the quantity, absent at 0', () => {
- const base = { runId: 'run_x', workflowKey: 'ai.generate', requester: 'service:live', provider: 'seam', tokensIn: 300, tokensOut: 100, cost: 0.1, at: Date.UTC(2026, 9, 4) };
+ const base = { runId: 'run_x', attempt: 2, kind: 'out', workflowKey: 'ai.generate', requester: 'service:live', provider: 'seam', model: 'm1', quantity: 300, cost: 0.1, at: Date.UTC(2026, 9, 4) };
const some = usageSamples.sampleOf({ ...base, freeAllowanceUsed: 150 });
const some = usageSamples.sampleOf({ ...base, freeAllowanceUsed: 150 });
assert.strictEqual(some.free_allowance_used, 150);
assert.strictEqual(some.free_allowance_used, 150);
+ assert.strictEqual(some.idempotency_key, 'ai:run_x:2:out');
+ assert.strictEqual(some.resource, rates.metricFor('out', 'm1'));
+ assert.ok(unitFits(some.resource, some.unit));
assert.ok(contracts.validate('platform.usage-sample@1', some).valid);
assert.ok(contracts.validate('platform.usage-sample@1', some).valid);
- assert.strictEqual(usageSamples.sampleOf({ ...base, freeAllowanceUsed: 9999 }).free_allowance_used, 400);
+ assert.strictEqual(usageSamples.sampleOf({ ...base, freeAllowanceUsed: 9999 }).free_allowance_used, 300);
assert.ok(!('free_allowance_used' in usageSamples.sampleOf({ ...base, freeAllowanceUsed: 0 })));
assert.ok(!('free_allowance_used' in usageSamples.sampleOf({ ...base, freeAllowanceUsed: 0 })));
assert.ok(!('free_allowance_used' in usageSamples.sampleOf(base)));
assert.ok(!('free_allowance_used' in usageSamples.sampleOf(base)));
});
});
@@ -169,7 +192,7 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on t
@@ -169,7 +192,7 @@ t.test('a reading Billing refuses is failed, with its error, in explain and on t
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);
await relay.flush();
await relay.flush();
const [u] = (await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: live })).body.run.explain.usage_readings;
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.idempotency_key, `ai:${r.body.run.id}:0:in`);
assert.strictEqual(u.state, 'failed');
assert.strictEqual(u.state, 'failed');
assert.match(u.last_error, /answered 400/);
assert.match(u.last_error, /answered 400/);
assert.ok(u.rejected_at);
assert.ok(u.rejected_at);