openvibers/openvibe.ai Commits

The AI meter: AI writes the records, Report keeps the ledger (L40/L41) (#223)

OP OpenVibers committed 13 days ago · 0730c1119e · 20 files, +645 −34

* The AI meter: AI writes the records, Report keeps the ledger (L40/L41)

The cached tokens read correctly by every adapter (with a native Gemini provider kind), the cache prices in the typed price table with cache_saved_usd, and one ai.call.recorded@1 event per run through AI's own events publisher. Rebased on the person's own keys and the You page that landed since.

* openvibe-contracts v0.260.0 (was v0.257.0)

The AI meter writes ai.call.recorded@1, which arrives in v0.260.0, and Report's three ai-usage results are in the same release. Bumps the pin, the lockfile and the version lines.

* Merge branch 'main' of github.com:OpenVibers/OpenVibe.AI into pin/ai223

# Conflicts:
#	README.md
#	STATUS.json

* Merge remote-tracking branch 'origin/main' into pin/ai223

* Merge remote-tracking branch 'origin/main' into feat/ai-usage-ledger

# Conflicts:
#	README.md
#	STATUS.json
#	package-lock.json
#	package.json

Changes

ddiff --git a/migrations/0022_model_cache_write_price.sql b/migrations/0022_model_cache_write_price.sql
nnew file mode 100644
iindex 0000000..2aa934f
---- /dev/null
++++ b/migrations/0022_model_cache_write_price.sql
@@@ -0,0 +1,7 @@
++-- phase: expand
++-- L40: the price table gains the cache prices a person types. cost_cached_per_mtok (the cache read) is the column
++-- migrations/0001 already had; this one adds the cache WRITE (the tokens written to a provider's cache). Both are
++-- optional: a NULL means nobody typed that price, and the record then says 'no cache price' and prices those tokens
++-- at the plain input price. No existing row is rewritten.
++
++ALTER TABLE models ADD COLUMN cost_cache_write_per_mtok double precision;
ddiff --git a/server/bench/events.js b/server/bench/events.js
iindex ecc51ff..654c2d3 100644
---- a/server/bench/events.js
++++ b/server/bench/events.js
@@@ -21,7 +21,7 @@ const { TOPICS: ACCOUNT_TOPICS } = require('openvibe-sdk/account-data');
const TOPIC = 'bench.rating.changed';
--function createBenchEvents({ bench, accountData = null, accountSend = null, config, log = console, fetchImpl = globalThis.fetch, clientSecret = process.env.OV_OAUTH_CLIENT_SECRET }) {
++function createBenchEvents({ bench, db = null, accountData = null, accountSend = null, config, log = console, fetchImpl = globalThis.fetch, clientSecret = process.env.OV_OAUTH_CLIENT_SECRET }) {
const router = express.Router();
const secrets = (config.events && config.events.secrets) || [];
ddiff --git a/server/config.js b/server/config.js
iindex 560db9c..3b580c0 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -65,6 +65,11 @@ function load(env = process.env) {
memoryKey: String(env.AI_MEMORY_KEY || '').trim(),
// The store's own maintenance (server/memory/jobs.js): owner by owner, with a per-run cap. Off in test mode
// (the way the JWKS client is) and off while a provider is in a drill, so a drill never spends the budget.
++ // The AI ledger (L40): the fee the OpenVibe key carries, and the roll-up job (off in test mode, off while a
++ // provider is in a drill).
++ usage: {
++ feeRate: Number(env.AI_KEY_FEE_RATE || 0) || 0,
++ },
memoryJobs: {
enabled: bool(env.AI_MEMORY_JOBS, nodeEnv !== 'test'),
intervalMs: int(env.AI_MEMORY_JOB_INTERVAL_MS, 15 * 60 * 1000),
ddiff --git a/server/console/index.js b/server/console/index.js
iindex 11d8486..1c85a17 100644
---- a/server/console/index.js
++++ b/server/console/index.js
@@@ -391,7 +391,9 @@ function consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, e
provider_key: p.key, model_key: String(b.model_key || '').trim(), display_name: String(b.display_name || '').trim() || null, type: 'chat',
supports: { json: true, vision: !!b.vision, tools: !!b.tools, streaming: true },
// Prices are the person's, typed from the provider's own price page (which they name), checked today.
-- ...(price(b.cost_in) != null || price(b.cost_out) != null ? { cost: { in_per_mtok: price(b.cost_in) ?? null, out_per_mtok: price(b.cost_out) ?? null, ...(source ? { source, verified_at: new Date(clock.now()).toISOString().slice(0, 10) } : {}) } } : {}),
++ // The cache prices are optional (L40): nobody typed them means the record says 'no cache price' and the
++ // cache tokens are priced at the plain input price.
++ ...(price(b.cost_in) != null || price(b.cost_out) != null ? { cost: { in_per_mtok: price(b.cost_in) ?? null, out_per_mtok: price(b.cost_out) ?? null, cached_per_mtok: price(b.cost_cache_read) ?? null, cache_write_per_mtok: price(b.cost_cache_write) ?? null, ...(source ? { source, verified_at: new Date(clock.now()).toISOString().slice(0, 10) } : {}) } } : {}),
}, who(req));
res.redirect(303, '/console/providers?done=model_added');
}));
ddiff --git a/server/console/views.js b/server/console/views.js
iindex cb32f3d..9283a96 100644
---- a/server/console/views.js
++++ b/server/console/views.js
@@@ -217,6 +217,7 @@ function providers({ staff, csrf, rows, models, canManage, stubFallback, notice,
<form class="inline" method="post" action="/console/providers/${enc(p.key)}/models">${csrfField(csrf)}<label>Model <input type="text" name="model_key" required maxlength="200" placeholder="deepseek-flash" spellcheck="false"></label>
<label><input type="checkbox" name="vision" value="1"> reads pictures</label> <label><input type="checkbox" name="tools" value="1"> tools</label>
<label>$ in / Mtok <input type="number" name="cost_in" step="0.0001" min="0" class="num"></label> <label>$ out / Mtok <input type="number" name="cost_out" step="0.0001" min="0" class="num"></label>
++<label>$ cache read / Mtok <input type="number" name="cost_cache_read" step="0.0001" min="0" class="num"></label> <label>$ cache write / Mtok <input type="number" name="cost_cache_write" step="0.0001" min="0" class="num"></label>
<label>from the price page <input type="url" name="price_source" placeholder="https://…/pricing"></label><button type="submit">Register model</button></form></details></td></tr>` : '');
return layout({
title: 'Providers', section: 'providers', staff, csrf, notice, error, body: html`<h1>Providers and models</h1>
@@@ -232,12 +233,12 @@ ${canManage ? html`<td><form class="inline" method="post" action="/console/provi
${p.health.state !== 'closed' || p.health.consecutive_failures ? html` <form class="inline" method="post" action="/console/providers/${enc(p.key)}/reset">${csrfField(csrf)}<button type="submit">Reset circuit<span class="sr-only"> of ${p.key}</span></button></form>` : ''}</td>` : ''}
</tr>${manageRow(p)}`), 'no providers.')}
${canManage ? '' : html`<p class="muted">Enabling, disabling and circuit resets need <code>ai.provider.manage</code> (the owner's <code>staff.secrets.manage</code>).</p>`}
--${table('Models', ['Provider', 'Model', 'Type', 'Status', 'Strength', 'Context / max output', 'Cost per Mtok (in / out / cached)', 'Supports'], models.map((m) => html`<tr>
++${table('Models', ['Provider', 'Model', 'Type', 'Status', 'Strength', 'Context / max output', 'Cost per Mtok (in / out / cached / cache write)', 'Supports'], models.map((m) => html`<tr>
<td><code>${m.provider_key}</code></td><td><code>${m.model_key}</code>${m.display_name && m.display_name !== m.model_key ? html`<br><span class="muted">${m.display_name}</span>` : ''}</td><td>${m.type}</td>
<td>${badge(m.status)}${m.metadata && m.metadata.missing_since ? html`<br><span class="muted">no longer listed since ${when(m.metadata.missing_since)}</span>` : ''}${canManage ? html`<form class="inline" method="post" action="/console/models/${enc(m.provider_key)}/${enc(m.model_key)}/status">${csrfField(csrf)}<input type="hidden" name="status" value="${m.status === 'active' ? 'disabled' : 'active'}"><button type="submit">${m.status === 'active' ? 'Turn off' : 'Turn on'}<span class="sr-only"> ${m.model_key}</span></button></form>` : ''}</td>
<td>${m.metadata && m.metadata.tier ? html`${m.metadata.tier}${m.metadata.reasoning ? ', thinks first' : ''}${m.metadata.note ? html`<br><span class="muted">${m.metadata.note}</span>` : ''}` : '—'}</td>
<td>${m.context_window ? n(m.context_window) : '—'} / ${m.max_output ? n(m.max_output) : '—'}</td>
--<td>${m.cost.in_per_mtok != null ? usd(m.cost.in_per_mtok) : '—'} / ${m.cost.out_per_mtok != null ? usd(m.cost.out_per_mtok) : '—'} / ${m.cost.cached_per_mtok != null ? usd(m.cost.cached_per_mtok) : '—'}</td>
++<td>${m.cost.in_per_mtok != null ? usd(m.cost.in_per_mtok) : '—'} / ${m.cost.out_per_mtok != null ? usd(m.cost.out_per_mtok) : '—'} / ${m.cost.cached_per_mtok != null ? usd(m.cost.cached_per_mtok) : '—'} / ${m.cost.cache_write_per_mtok != null ? usd(m.cost.cache_write_per_mtok) : '—'}</td>
<td>${Object.entries(m.supports).filter(([, v]) => v).map(([k]) => k).join(', ') || '—'}</td></tr>`), 'no models registered (providers use their default model).')}`,
});
}
ddiff --git a/server/events.js b/server/events.js
iindex c3cc733..267f8fd 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -107,6 +107,48 @@ async function credentialFailed({ subject, provider, error, owner = null }) {
dbRef.afterCommit(() => svc && svc.kick());
return env;
}
++
++/**
++ * Queue the ledger's record of one AI call (ai.call.recorded@1). MUST be awaited inside the transaction that
++ * wrote the record, so the event and the row commit together. The payload carries the tokens, the money and the
++ * ids: never a prompt, never an answer.
++ */
++function callPayload(rec) {
++ return {
++ subject: rec.subject || null, attribution: rec.attribution || '', surface: rec.surface || '', system: rec.system,
++ provider: rec.provider || null, model: rec.model || null, effort: rec.effort || null,
++ paid_by: rec.paid_by, paid_by_detail: rec.paid_by_detail || null,
++ input: rec.tokens_input, output: rec.tokens_output, reasoning: rec.tokens_reasoning,
++ cache_read: rec.tokens_cache_read, cache_write: rec.tokens_cache_write, cache_write_ttl: rec.cache_write_ttl || null,
++ images: rec.images, audio_seconds: rec.audio_seconds, video_seconds: rec.video_seconds,
++ list_cost_usd: rec.list_cost_usd, charged_usd: rec.charged_usd, fee_usd: rec.fee_usd, cache_saved_usd: rec.cache_saved_usd,
++ price_source: rec.price_source, price_date: rec.price_date || null,
++ latency_ms: rec.latency_ms, ttft_ms: rec.ttft_ms, attempts: rec.attempts, outcome: rec.outcome,
++ run_id: rec.run_id || null, turn_id: rec.turn_id || null, at: rec.at,
++ };
++}
++
++/**
++ * Queue the record of one AI call (ai.call.recorded@1). MUST be awaited inside a transaction, like every other
++ * event here: a no-op while publishing is off.
++ */
++async function callRecorded(rec) {
++ if (!svc || !rec) return null;
++ const payload = callPayload(rec);
++ const env = await svc.emitIn(dbRef, {
++ event_type: 'ai.call.recorded',
++ actor: { type: 'service', id: 'ai' },
++ subject: rec.run_id ? { type: 'run', id: rec.run_id } : { type: 'user', id: rec.subject || 'anonymous' },
++ visibility: 'internal',
++ priority: 'low',
++ payload,
++ });
++ queued++;
++ dbRef.afterCommit(() => svc && svc.kick());
++ return env;
++}
++
++
async function status() {
if (!svc) return { enabled: false };
const s = await svc.status();
@@@ -125,4 +167,6 @@ async function stop() {
}
function _reset() { if (svc) svc.stop(); svc = null; queued = 0; }
--module.exports = { init, runChanged, credentialFailed, payloadOf, status, stop, _reset };
++module.exports = { init, runChanged, credentialFailed, callRecorded, callPayload, payloadOf, status, stop, _reset };
++
++
ddiff --git a/server/providers/anthropic.js b/server/providers/anthropic.js
iindex c24b881..498d8eb 100644
---- a/server/providers/anthropic.js
++++ b/server/providers/anthropic.js
@@@ -13,6 +13,7 @@
*/
const { postJson, ProviderError, deadline, describeAbort, schemaInstruction } = require('./common');
const { sseFrames } = require('./stream');
++const usageParser = require('./usage');
const DEFAULT_BASE_URL = 'https://api.anthropic.com/v1';
const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
@@@ -72,11 +73,7 @@ function messagesBody(req) {
return body;
}
--const usageOf = (message) => {
-- const u = (message && message.usage) || {};
-- const cached = u.cache_read_input_tokens || 0;
-- return { input: (u.input_tokens || 0) + cached + (u.cache_creation_input_tokens || 0), output: u.output_tokens || 0, cached };
--};
++const usageOf = (message) => usageParser.anthropicUsage((message && message.usage) || {});
const textOf = (message) => (message && Array.isArray(message.content) ? message.content : []).filter(c => c && c.type === 'text').map(c => c.text).join('').trim();
/**
@@@ -153,9 +150,7 @@ function createAnthropicProvider(record, { apiKey = '', fetchImpl = globalThis.f
const tool = content.find(c => c.type === 'tool_use');
if (tool && tool.input) { jsonOut = tool.input; if (!text) text = JSON.stringify(tool.input); }
if (!jsonOut && req.json && (!jsonMode || jsonMode === 'format') && isCurrentClaude(req.model)) { try { jsonOut = JSON.parse(text); } catch { /* the engine repairs it */ } }
-- const u = j.usage || {};
-- const cached = u.cache_read_input_tokens || 0;
-- return { text, json: jsonOut, model: j.model || req.model, usage: { input: (u.input_tokens || 0) + cached + (u.cache_creation_input_tokens || 0), output: u.output_tokens || 0, cached } };
++ return { text, json: jsonOut, model: j.model || req.model, usage: usageParser.anthropicUsage(j.usage || {}) };
}
/** POST /v1/messages as the caller wrote it → { message, text, model, usage }. */
ddiff --git a/server/providers/gemini.js b/server/providers/gemini.js
nnew file mode 100644
iindex 0000000..a0af3d4
---- /dev/null
++++ b/server/providers/gemini.js
@@@ -0,0 +1,78 @@
++'use strict';
++/**
++ * Google Gemini's own API (the native generateContent endpoint), for a provider of kind 'gemini'.
++ *
++ * POST {base}/models/{model}:generateContent?key=API_KEY
++ * { contents: [{ role, parts }], systemInstruction, generationConfig }
++ * -> { candidates: [{ content: { parts: [{ text }] } }], usageMetadata, modelVersion }
++ *
++ * The usage is read from usageMetadata (server/providers/usage.js): promptTokenCount includes
++ * cachedContentTokenCount, and thoughtsTokenCount is the thinking. Gemini's OpenAI-compatible endpoint
++ * (kind 'openai') does not report the cached tokens, which is why this adapter exists.
++ */
++const { ProviderError, postJson, schemaInstruction } = require('./common');
++const usage = require('./usage');
++
++const DEFAULT_BASE_URL = 'https://generativelanguage.googleapis.com/v1beta';
++const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
++
++/** The canonical request as Gemini's contents: { role, parts }. */
++function contentsOf({ system, messages, image }) {
++ const out = [];
++ for (const m of messages || []) {
++ const role = m.role === 'assistant' ? 'model' : 'user';
++ const parts = [];
++ if (typeof m.content === 'string') parts.push({ text: m.content });
++ else if (Array.isArray(m.content)) for (const p of m.content) if (p && p.type === 'text') parts.push({ text: String(p.text || '') });
++ const last = m === (messages || [])[messages.length - 1];
++ if (last && image && m.role === 'user') parts.push({ inline_data: { mime_type: image.mediaType, data: image.base64 } });
++ if (parts.length) out.push({ role, parts });
++ }
++ return out;
++}
++
++function createGeminiProvider(record, { apiKey = '', fetchImpl = globalThis.fetch } = {}) {
++ const baseUrl = String(record.base_url || DEFAULT_BASE_URL).replace(/\/+$/, '');
++ const model = (req) => String((req && req.model) || record.default_model || 'gemini-2.5-flash').replace(/^models\//, '');
++ const caps = new Set(record.capabilities && record.capabilities.length ? record.capabilities : [...TEXT_OPS, 'json', 'vision']);
++
++ async function generate(req) {
++ const sysText = ((req.system || []).map((x) => x.text).join('\n'));
++ const body = { contents: contentsOf(req) };
++ if (sysText) body.systemInstruction = { parts: [{ text: sysText }] };
++ const generationConfig = {};
++ if (req.maxTokens) generationConfig.maxOutputTokens = req.maxTokens;
++ if (req.temperature != null) generationConfig.temperature = req.temperature;
++ if (req.json) {
++ generationConfig.responseMimeType = 'application/json';
++ generationConfig.responseSchema = req.json.schema;
++ }
++ if (Object.keys(generationConfig).length) body.generationConfig = generationConfig;
++ const url = `${baseUrl}/models/${encodeURIComponent(model(req))}:generateContent`;
++ const j = await postJson(url, apiKey ? { 'x-goog-api-key': apiKey } : {}, body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
++ const parts = (j.candidates && j.candidates[0] && j.candidates[0].content && j.candidates[0].content.parts) || [];
++ const text = parts.map((p) => p && p.text ? p.text : '').join('').trim();
++ const u = usage.geminiUsage(j.usageMetadata || {});
++ return { text, json: null, model: j.modelVersion || model(req), usage: u };
++ }
++
++ const chat = async (req) => {
++ const r = await generate(req);
++ if (!req.json) return r;
++ let json = null;
++ try { json = JSON.parse(r.text); } catch { json = null; }
++ return { ...r, json };
++ };
++
++ const adapter = {
++ key: record.key,
++ kind: 'gemini',
++ supports: (feature) => caps.has(feature),
++ chat,
++ generate,
++ };
++ for (const op of TEXT_OPS) if (op !== 'chat') adapter[op] = chat;
++ return adapter;
++}
++
++module.exports = { createGeminiProvider, DEFAULT_BASE_URL };
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex 3406fbc..f2f311e 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -28,6 +28,7 @@ const { createStubProvider } = require('./stub');
const { createOpenAiProvider } = require('./openai');
const { createResponsesProvider } = require('./responses');
const { createAnthropicProvider } = require('./anthropic');
++const { createGeminiProvider } = require('./gemini');
const { createHttpSeamProvider } = require('./http');
const { createWhisperProvider } = require('./whisper');
const rates = require('./rate-cards');
@@@ -108,6 +109,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
case 'openai': return createOpenAiProvider(p, { apiKey, fetchImpl });
case 'responses': return createResponsesProvider(p, { apiKey, fetchImpl });
case 'anthropic': return createAnthropicProvider(p, { apiKey, fetchImpl });
++ case 'gemini': return createGeminiProvider(p, { apiKey, fetchImpl });
case 'http': return createHttpSeamProvider(p, { apiKey, fetchImpl });
case 'whisper': return createWhisperProvider(p, { config });
default: throw new Error(`unknown provider kind ${p.kind}`);
@@@ -697,7 +699,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// A streamed answer (/v1/chat/completions, stream: true) is the caller's once a delta has gone out: a failure
// after that is final, never retried or handed to a fallback whose words would follow the first one's.
let streamed = false;
-- const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
++ // L40: the time to the first token, when the answer streams.
++ let attemptStart = 0;
++ let ttftMs = null;
++ const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; if (ttftMs === null && attemptStart) ttftMs = Date.now() - attemptStart; req.onDelta(d); } : null;
// L36 Y1: a run on behalf of a person tries the person's OWN keys first, for the providers that can serve the
// request. With only_my_keys it never falls back; otherwise the network is the fallback and the explain says so.
const prefs = ctx.selfPreferences || {};
@@@ -757,6 +762,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
{ providers, paid_by: null, attempts: selfAttempts, explain: { objective: 'own-keys', reasons: ['only_my_keys: no network fallback'], paid_by: null, selected: null, candidates: [], attempts: selfAttempts, prefer: preferState ? { ...preferState, why: preferState.why || 'you have no key for it, and only_my_keys never falls back to the network' } : null } });
}
}
++
let { order: list, explain } = await candidates(route, features, operation, ctx);
// L36 Y3: free_first puts the free tiers ahead of metered ones for this person's fallback. Own-key attempts above
// are already free to them, so this only reorders the network.
@@@ -874,6 +880,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
const started = clock.now();
const t0 = Date.now();
++ attemptStart = t0;
try {
const fn = a[operation];
if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });
@@@ -908,7 +915,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
attempts.push({ key: 'openvibe', provider: c.provider, model: result.model || model, why: chosen || (fallback ? 'the next network provider after an earlier attempt failed' : 'the first network offer for this route'), status: 'ok', error: null });
warm.remember(warmSubject(ctx, route), leadingText(req), c.provider);
explain.attempts = attempts;
-- return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };
++ return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, ttftMs, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };
} catch (err) {
lastErr = err;
if (ctx.signal && ctx.signal.aborted) {
ddiff --git a/server/providers/openai.js b/server/providers/openai.js
iindex 9279ab5..8a391e7 100644
---- a/server/providers/openai.js
++++ b/server/providers/openai.js
@@@ -16,6 +16,7 @@ const fs = require('fs');
const path = require('path');
const { ProviderError, postJson, deadline, describeAbort, approxTokens, schemaInstruction } = require('./common');
const { sseEvents, createAccumulator } = require('./stream');
++const usageParser = require('./usage');
const DEFAULT_BASE_URL = 'https://api.openai.com/v1';
const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
@@@ -65,8 +66,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
const j = await postJson(`${baseUrl}/chat/completions`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
const msg = j.choices && j.choices[0] && j.choices[0].message;
const text = (msg && typeof msg.content === 'string' ? msg.content : '').trim();
-- const u = j.usage || {};
-- return { text, json: null, model: j.model || req.model, usage: { input: u.prompt_tokens || 0, output: u.completion_tokens || 0, cached: (u.prompt_tokens_details && u.prompt_tokens_details.cached_tokens) || 0 } };
++ return { text, json: null, model: j.model || req.model, usage: usageParser.openAiCompatibleUsage(j.usage || {}) };
} catch (e) {
// Gateways without structured-output support: schema -> json_object -> prose.
if (e instanceof ProviderError && e.status === 400 && jsonMode && /response_format|json_schema|schema|strict/i.test(e.message)) {
@@@ -163,7 +163,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
completion: { choices: j.choices, system_fingerprint: j.system_fingerprint || null },
text: typeof msg.content === 'string' ? msg.content : '',
model: j.model || req.model,
-- usage: { input: u.prompt_tokens || 0, output: u.completion_tokens || 0, cached: (u.prompt_tokens_details && u.prompt_tokens_details.cached_tokens) || 0 },
++ usage: usageParser.openAiCompatibleUsage(u),
};
}
@@@ -221,7 +221,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
completion: { choices, system_fingerprint: fingerprint },
text: typeof first.content === 'string' ? first.content : '',
model,
-- usage: { input: inTok, output: outTok, cached: (u.prompt_tokens_details && u.prompt_tokens_details.cached_tokens) || 0, ...(estimated ? { estimated: true } : {}) },
++ usage: { ...usageParser.openAiCompatibleUsage(estimated ? { prompt_tokens: inTok, completion_tokens: outTok } : u), ...(estimated ? { estimated: true } : {}) },
};
}
ddiff --git a/server/providers/rate-cards.js b/server/providers/rate-cards.js
iindex 7f08950..5a4820a 100644
---- a/server/providers/rate-cards.js
++++ b/server/providers/rate-cards.js
@@@ -22,7 +22,7 @@
const UNKNOWN_DATE = '1970-01-01';
const unverified = (origin) => `urn:openvibe:ai:unverified:${origin}`;
--const KINDS = { in: 'input', cached: 'cached-input', out: 'output' };
++const KINDS = { in: 'input', cached: 'cached-input', out: 'output', cache_write: 'cache-write-input' };
const RESET_PERIODS = ['day', 'month', 'none'];
const DATE = /^\d{4}-\d{2}-\d{2}$/;
const URI = /^[a-z][a-z0-9+.-]*:[^\s]+$/i;
@@@ -91,6 +91,9 @@ function buildCards({ providerKey, model, row = null, pricing = {}, flat = true
in: make('in', inRate, 'models', prov),
cached: row.cost.cached_per_mtok != null ? make('cached', Number(row.cost.cached_per_mtok), 'models', prov)
: make('cached', inRate * 0.1, 'derived', allowanceOnly(prov)),
++ // The cache WRITE price (L40) is optional: nobody typed it means the card is unpriced, never a guess.
++ cache_write: row.cost.cache_write_per_mtok != null ? make('cache_write', Number(row.cost.cache_write_per_mtok), 'models', prov)
++ : make('cache_write', 0, 'unpriced', allowanceOnly(prov)),
out: make('out', Number(row.cost.out_per_mtok) || 0, 'models', prov),
};
}
@@@ -103,6 +106,7 @@ function buildCards({ providerKey, model, row = null, pricing = {}, flat = true
return {
in: make('in', inRate, best && b.in != null ? 'pricing' : 'unpriced', prov),
cached: b.cached != null ? make('cached', Number(b.cached), 'pricing', prov) : make('cached', inRate * 0.1, 'derived', allowanceOnly(prov)),
++ cache_write: b.cache_write != null ? make('cache_write', Number(b.cache_write), 'pricing', prov) : make('cache_write', 0, 'unpriced', allowanceOnly(prov)),
out: make('out', Number(b.out) || 0, best && b.out != null ? 'pricing' : 'unpriced', prov),
};
}
@@@ -111,6 +115,7 @@ function buildCards({ providerKey, model, row = null, pricing = {}, flat = true
return {
in: has('in') ? make('in', inRate, 'pricing', prov) : make('in', inRate, 'flat', allowanceOnly(prov)),
cached: has('cached') ? make('cached', Number(best.cached), 'pricing', prov) : make('cached', inRate * 0.1, 'derived', allowanceOnly(prov)),
++ cache_write: has('cache_write') ? make('cache_write', Number(best.cache_write), 'pricing', prov) : make('cache_write', 0, 'unpriced', allowanceOnly(prov)),
out: has('out') ? make('out', Number(best.out), 'pricing', prov) : make('out', pricing.outputPerMtok, 'flat', allowanceOnly(prov)),
};
}
@@@ -158,6 +163,8 @@ function usageSince(cards, now) {
}
const BREAKER_HEALTH = { closed: 'up', half_open: 'degraded', open: 'down' };
++/** Is this card one of the three metered token metrics (input, cached input, output)? */
++const isUsageMetric = (c) => ['input-tokens', 'cached-input-tokens', 'output-tokens'].includes(String(c.metric || '').split(':')[0]);
/**
* One platform.provider-state@1 for a provider. Each metric counts and forecasts over its own card's period
@@@ -169,7 +176,8 @@ const BREAKER_HEALTH = { closed: 'up', half_open: 'degraded', open: 'down' };
function providerState({ provider, cards, usageRows = [], breaker = 'closed', reserve = null, now }) {
const periods = {};
const used = {};
-- for (const c of cards) { periods[c.metric] = cardPeriod(c, now); used[c.metric] = 0; }
++ // Only the three metered token metrics are counted here (the cache write's card is not a usage metric: L40).
++ for (const c of cards) { if (!isUsageMetric(c)) continue; periods[c.metric] = cardPeriod(c, now); used[c.metric] = 0; }
const firstDay = Object.fromEntries(Object.entries(periods).map(([m, p]) => [m, dayOf(p.start)]));
for (const r of usageRows) {
const model = r.model_key || null;
ddiff --git a/server/providers/responses.js b/server/providers/responses.js
iindex 90148b3..056a76e 100644
---- a/server/providers/responses.js
++++ b/server/providers/responses.js
@@@ -13,6 +13,7 @@
* transcription: those stay with the `openai` kind.
*/
const { ProviderError, postJson, schemaInstruction } = require('./common');
++const usageParser = require('./usage');
const DEFAULT_BASE_URL = 'https://api.openai.com/v1';
const TEXT_OPS = ['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich'];
@@@ -84,8 +85,7 @@ function createResponsesProvider(record, { apiKey = '', fetchImpl = globalThis.f
const body = buildBody({ ...req, baseUrl }, jsonMode);
try {
const j = await postJson(`${baseUrl}/responses`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
-- const u = j.usage || {};
-- return { text: outputText(j), json: null, model: j.model || req.model, usage: { input: u.input_tokens || 0, output: u.output_tokens || 0, cached: (u.input_tokens_details && u.input_tokens_details.cached_tokens) || 0 } };
++ return { text: outputText(j), json: null, model: j.model || req.model, usage: usageParser.responsesUsage(j.usage || {}) };
} catch (e) {
// Gateways without structured-output support: schema -> json_object -> prose.
if (e instanceof ProviderError && e.status === 400 && jsonMode && /text\.format|json_schema|json_object|schema|strict/i.test(e.message)) {
ddiff --git a/server/providers/usage.js b/server/providers/usage.js
nnew file mode 100644
iindex 0000000..da93ad0
---- /dev/null
++++ b/server/providers/usage.js
@@@ -0,0 +1,139 @@
++'use strict';
++/**
++ * What a provider's answer says about the tokens it used (plan L40: one meter, and the cache is part of the bill).
++ *
++ * Every adapter answers the same shape, so the ledger can read one record whatever the provider was:
++ *
++ * input the WHOLE prompt: the uncached input plus the cache reads plus the cache writes
++ * output the tokens the provider produced (the reasoning tokens are counted apart)
++ * reasoning the thinking tokens, when the provider counts them apart
++ * cached the tokens read from the provider's cache
++ * cache_write the tokens written to the provider's cache
++ * cache_write_ttl '5m' or '1h', when the provider says how long the written entry lives
++ * images / audio_seconds / video_seconds the metered media
++ *
++ * The providers do not agree, so each one is read on its own terms (and never counted twice):
++ *
++ * Anthropic input_tokens EXCLUDES the cache reads and the cache writes; the cache reads are
++ * cache_read_input_tokens, the writes cache_creation_input_tokens (5m and 1h apart
++ * under cache_creation).
++ * OpenAI prompt_tokens INCLUDES the cached tokens; the cache read is
++ * prompt_tokens_details.cached_tokens, the reasoning completion_tokens_details.reasoning_tokens.
++ * DeepSeek prompt_cache_hit_tokens and prompt_cache_miss_tokens, and prompt_tokens = hit + miss.
++ * Gemini promptTokenCount INCLUDES cachedContentTokenCount; the reasoning is thoughtsTokenCount.
++ *
++ * The record's own 'input' is the UNCACHED input: this shape's input minus its cached minus its cache_write.
++ */
++
++const num = (v) => (v === null || v === undefined || v === '' ? 0 : (Number.isFinite(Number(v)) ? Math.max(0, Math.trunc(Number(v))) : 0));
++
++/** The shape every parser answers. */
++const empty = () => ({ input: 0, output: 0, reasoning: 0, cached: 0, cache_write: 0, cache_write_ttl: null });
++
++/** Anthropic: input_tokens excludes the cache reads and the cache writes. */
++function anthropicUsage(u = {}) {
++ const cacheRead = num(u.cache_read_input_tokens);
++ const cacheWrite = num(u.cache_creation_input_tokens);
++ const uncached = num(u.input_tokens);
++ const creation = u.cache_creation && typeof u.cache_creation === 'object' ? u.cache_creation : {};
++ const oneHour = num(creation.ephemeral_1h_input_tokens);
++ const fiveMin = num(creation.ephemeral_5m_input_tokens);
++ return {
++ ...empty(),
++ input: uncached + cacheRead + cacheWrite,
++ output: num(u.output_tokens),
++ cached: cacheRead,
++ cache_write: cacheWrite || fiveMin + oneHour,
++ cache_write_ttl: (cacheWrite || fiveMin || oneHour) ? (oneHour > 0 && oneHour >= fiveMin ? '1h' : '5m') : null,
++ };
++}
++
++/** OpenAI (and every compatible gateway): prompt_tokens includes the cached tokens. */
++function openAiUsage(u = {}) {
++ const details = u.prompt_tokens_details && typeof u.prompt_tokens_details === 'object' ? u.prompt_tokens_details : {};
++ const completion = u.completion_tokens_details && typeof u.completion_tokens_details === 'object' ? u.completion_tokens_details : {};
++ const cached = num(details.cached_tokens);
++ const reasoning = num(completion.reasoning_tokens);
++ const output = num(u.completion_tokens);
++ return {
++ ...empty(),
++ input: num(u.prompt_tokens),
++ output: Math.max(0, output - reasoning), // the reasoning is counted apart, never twice
++ reasoning,
++ cached,
++ cache_write: 0,
++ cache_write_ttl: null,
++ };
++}
++
++/** OpenAI's Responses API: input_tokens and its cached tokens live in input_tokens_details. */
++function responsesUsage(u = {}) {
++ const details = u.input_tokens_details && typeof u.input_tokens_details === 'object' ? u.input_tokens_details : {};
++ const completion = u.output_tokens_details && typeof u.output_tokens_details === 'object' ? u.output_tokens_details : {};
++ const cached = num(details.cached_tokens);
++ const reasoning = num(completion.reasoning_tokens);
++ const output = num(u.output_tokens);
++ return {
++ ...empty(),
++ input: num(u.input_tokens),
++ output: Math.max(0, output - reasoning),
++ reasoning,
++ cached,
++ cache_write: 0,
++ cache_write_ttl: null,
++ };
++}
++
++/** DeepSeek (OpenAI-shaped): prompt_cache_hit_tokens and prompt_cache_miss_tokens, prompt_tokens = hit + miss. */
++function deepSeekUsage(u = {}) {
++ const hit = num(u.prompt_cache_hit_tokens);
++ const miss = num(u.prompt_cache_miss_tokens);
++ const prompt = num(u.prompt_tokens);
++ // DeepSeek counts prompt_tokens as hit + miss. When only one of the two is reported, the prompt total still counts.
++ const input = hit || miss ? hit + miss : prompt;
++ const reasoning = num(u.completion_tokens_details && u.completion_tokens_details.reasoning_tokens);
++ const output = num(u.completion_tokens);
++ return {
++ ...empty(),
++ input,
++ output: Math.max(0, output - reasoning),
++ reasoning,
++ cached: hit,
++ cache_write: 0,
++ cache_write_ttl: null,
++ };
++}
++
++/** Gemini (native): promptTokenCount includes cachedContentTokenCount; thoughtsTokenCount is the thinking. */
++function geminiUsage(u = {}) {
++ const details = Array.isArray(u.promptTokensDetails) ? u.promptTokensDetails[0] : null;
++ const cached = num(u.cachedContentTokenCount) || num(details && details.cachedContentTokenCount);
++ const reasoning = num(u.thoughtsTokenCount);
++ const output = num(u.candidatesTokenCount);
++ return {
++ ...empty(),
++ input: num(u.promptTokenCount),
++ output: Math.max(0, output - reasoning),
++ reasoning,
++ cached,
++ cache_write: 0,
++ cache_write_ttl: null,
++ };
++}
++
++/**
++ * The usage of an OpenAI-shaped answer, whichever gateway sent it: DeepSeek's own fields win when they
++ * are there, OpenAI's details otherwise.
++ */
++function openAiCompatibleUsage(u = {}) {
++ if (u && (u.prompt_cache_hit_tokens !== undefined || u.prompt_cache_miss_tokens !== undefined)) return deepSeekUsage(u);
++ return openAiUsage(u);
++}
++
++/** The uncached input of a usage shape: the whole prompt less the cache reads and the cache writes. */
++const uncachedInput = (usage = {}) => Math.max(0, num(usage.input) - num(usage.cached) - num(usage.cache_write));
++
++/** The cache tokens of a usage shape, read and written. */
++const cacheTokens = (usage = {}) => ({ read: num(usage.cached), write: num(usage.cache_write) });
++
++module.exports = { num, empty, anthropicUsage, openAiUsage, responsesUsage, deepSeekUsage, geminiUsage, openAiCompatibleUsage, uncachedInput, cacheTokens };
ddiff --git a/server/registry.js b/server/registry.js
iindex d2519ad..a5dd38f 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -122,6 +122,8 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
context_window: r.context_window, max_output: r.max_output,
// The price and its platform.rate-card@1 terms (migrations/0004); a NULL date or source is unknown, never filled in.
cost: { in_per_mtok: r.cost_in_per_mtok, out_per_mtok: r.cost_out_per_mtok, cached_per_mtok: r.cost_cached_per_mtok,
++ // The cache write price is optional (L40): it is only in the shape once somebody typed it.
++ ...(r.cost_cache_write_per_mtok == null ? {} : { cache_write_per_mtok: r.cost_cache_write_per_mtok }),
free_allowance: r.free_allowance == null ? 0 : Number(r.free_allowance), reset_period: r.reset_period || 'month',
effective_from: r.effective_from ?? null, source: r.source ?? null, verified_at: r.verified_at ?? null },
supports: { json: !!r.supports_json, tools: !!r.supports_tools, streaming: !!r.supports_streaming, vision: !!r.supports_vision },
@@@ -150,19 +152,20 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
const sup = input.supports || (prev && prev.supports) || {};
const t = now();
await db.prepare(`INSERT INTO models (provider_key, model_key, display_name, type, status, context_window, max_output, cost_in_per_mtok, cost_out_per_mtok, cost_cached_per_mtok,
-- free_allowance, reset_period, effective_from, source, verified_at,
++ cost_cache_write_per_mtok, free_allowance, reset_period, effective_from, source, verified_at,
supports_json, supports_tools, supports_streaming, supports_vision, metadata, created_at, updated_at)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(provider_key, model_key) DO UPDATE SET display_name = excluded.display_name, type = excluded.type, status = excluded.status,
context_window = excluded.context_window, max_output = excluded.max_output, cost_in_per_mtok = excluded.cost_in_per_mtok,
cost_out_per_mtok = excluded.cost_out_per_mtok, cost_cached_per_mtok = excluded.cost_cached_per_mtok,
++ cost_cache_write_per_mtok = excluded.cost_cache_write_per_mtok,
free_allowance = excluded.free_allowance, reset_period = excluded.reset_period, effective_from = excluded.effective_from,
source = excluded.source, verified_at = excluded.verified_at, supports_json = excluded.supports_json,
supports_tools = excluded.supports_tools, supports_streaming = excluded.supports_streaming, supports_vision = excluded.supports_vision,
metadata = excluded.metadata, updated_at = excluded.updated_at`)
.run(input.provider_key, input.model_key, input.display_name || (prev && prev.display_name) || input.model_key, type, status,
input.context_window ?? (prev && prev.context_window) ?? null, input.max_output ?? (prev && prev.max_output) ?? null,
-- cost.in_per_mtok ?? null, cost.out_per_mtok ?? null, cost.cached_per_mtok ?? null,
++ cost.in_per_mtok ?? null, cost.out_per_mtok ?? null, cost.cached_per_mtok ?? null, cost.cache_write_per_mtok ?? null,
terms.free_allowance ?? 0, terms.reset_period || 'month', terms.effective_from || null, terms.source || null, terms.verified_at || null,
sup.json === false ? 0 : 1, sup.tools ? 1 : 0, sup.streaming ? 1 : 0, sup.vision ? 1 : 0,
JSON.stringify(input.metadata || (prev && prev.metadata) || {}), t, t);
ddiff --git a/server/runs.js b/server/runs.js
iindex 887ed4d..86b091f 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -22,6 +22,7 @@ const { preferenceLines } = require('./user-modules');
const { preferredTarget } = require('./providers/task');
const schemas = require('./schemas');
const events = require('./events');
++const usageRecord = require('./usage-record');
const usageSamples = require('./usage-samples');
const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);
@@@ -316,6 +317,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
readings: [{ attempt: 0, kind: 'in', model: hit.model_key || null, quantity: 0, cost: 0, freeAllowanceUsed: 0 }] });
await events.runChanged(await getRow(id));
});
++ // A cache hit is still a run: its record says so (no tokens, no money).
++ await publishUsage({ row: await getRow(id), outcome: 'cached' });
await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
return { run: await get(id), created: true, promise: null };
}
@@@ -392,6 +395,79 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
};
}
++ /**
++ * The record of a run (plan L40): the tokens (the cache read and the cache write counted apart), the money and
++ * who paid. It is published as one ai.call.recorded@1 event; OpenVibe.Report keeps the ledger. The record never
++ * carries the prompt or the answer.
++ */
++ async function publishUsage({ row, outcome, usage = {}, provider = null, model = null, attempts = 0, latencyMs = null, ttftMs = null }) {
++ try {
++ const opts = parseJson(row.options, {});
++ const payer = await payerOf({ row, providerKey: provider, opts });
++ const tokens = {
++ input: Math.max(0, (usage.input || 0) - (usage.cached || 0) - (usage.cache_write || 0)),
++ output: usage.output || 0, reasoning: usage.reasoning || 0,
++ cache_read: usage.cached || 0, cache_write: usage.cache_write || 0,
++ };
++ const cards = provider ? await pool.rateCardsFor(provider, model) : null;
++ const prices = usageRecord.pricesOf(cards);
++ const money = usageRecord.moneyOf({ prices, tokens, paidBy: payer.paid_by, feeRate: (config.usage && config.usage.feeRate) || 0 });
++ const rec = usageRecord.recordOf({
++ system: 'ai', run_id: row.id, subject: subjectOf(row), attribution: attributionOf(row), surface: row.workflow_key || '',
++ provider: payer.provider || provider || null, model: model || null, effort: opts.effort || null,
++ paid_by: payer.paid_by, paid_by_detail: payer.detail,
++ input: tokens.input, output: tokens.output, reasoning: tokens.reasoning, cache_read: tokens.cache_read,
++ cache_write: tokens.cache_write, cache_write_ttl: usage.cache_write_ttl || null,
++ images: usage.images || 0, audio_seconds: usage.audio_seconds || 0, video_seconds: usage.video_seconds || 0,
++ list_cost_usd: money.list_cost_usd, charged_usd: money.charged_usd, fee_usd: money.fee_usd, cache_saved_usd: money.cache_saved_usd,
++ price_source: prices.source, price_date: prices.date,
++ latency_ms: latencyMs, ttft_ms: ttftMs, attempts, outcome,
++ at: row.finished_at || new Date(clock.now()).toISOString(),
++ });
++ // The event (ai.call.recorded) is queued in its own transaction, like every other event here.
++ await db.tx(async () => { await events.callRecorded(rec); });
++ } catch (err) {
++ log.warn(`[runs] ${row.id}: the usage record was not written: ${err && err.message}`);
++ }
++ }
++
++ /** The person the run was for, when there is one (a run on behalf of a service has none). */
++ const subjectOf = (row) => {
++ const o = parseJson(row.on_behalf_of, null);
++ return o && o.type === 'user' && o.id ? o.id : null;
++ };
++
++ /** The site or app the run came from, as '<type>:<id>' (empty when nothing is known). */
++ const attributionOf = (row) => {
++ const a = parseJson(row.attribution, null);
++ if (!a) return '';
++ if (typeof a === 'string') return a;
++ const kind = a.service || a.type || '';
++ return a.id ? `${kind}:${a.id}` : kind;
++ };
++
++ /**
++ * Who paid for this run (L40): the person's own key, the OpenVibe key, their plan, their own machine, a free
++ * tier, the network's allowance or an organisation's key. A person's own key is never charged here.
++ */
++ async function payerOf({ row, providerKey, opts }) {
++ if (opts.credential_subject) {
++ const cred = await credentials.get(row.requester_id, opts.credential_subject, opts.credential_provider || null).catch(() => null);
++ if (cred && cred.provider === 'node') return { paid_by: 'own_machine', detail: cred.machine || 'node', provider: 'node' };
++ const name = (cred && cred.provider) || 'own';
++ const hint = cred && cred.key_hint ? String(cred.key_hint) : '';
++ return { paid_by: 'own_key', detail: hint ? `${name} ...${hint}` : name, provider: name };
++ }
++ if (opts.project_provider) return { paid_by: 'org_key', detail: `project:${opts.project_provider.project_id || ''}`, provider: providerKey };
++ const p = providerKey ? await registry.getProvider(providerKey) : null;
++ const profile = (p && p.billing_profile) || 'metered';
++ const name = providerKey || (p && p.key) || null;
++ if (profile === 'subscription') return { paid_by: 'plan', detail: (p && (p.pool_key || p.key)) || name, provider: name };
++ if (profile === 'free') return { paid_by: 'free_tier', detail: (p && (p.pool_key || p.key)) || name, provider: name };
++ if (profile === 'byok') return { paid_by: 'own_key', detail: name, provider: name };
++ if (p && p.metadata && p.metadata.allowance === 'network') return { paid_by: 'network_allowance', detail: name, provider: name };
++ return { paid_by: 'openvibe_key', detail: name, provider: name };
++ }
async function execute(id, { controller, reserved, scope, ctx, input, hold = null }) {
const row = await getRow(id);
if (!row || row.status !== 'queued') { inflight.delete(id); return; }
@@@ -463,6 +539,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// recorded (the fallback is named in explain). A run that used the person's own key spends nothing here.
const usedOwnKey = String(r.provider || '').startsWith('byo:self:');
if (r.provider && (reserved || (opts.self_subject && !usedOwnKey))) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost, attempts: logged, runId: id, traceId: row.trace_id, hold });
++ // The record of the run (L40): the tokens with the cache apart, the money and who paid.
++ await publishUsage({ row: await getRow(id), outcome: 'succeeded', usage: r.usage, provider: r.provider, model: r.model, attempts, latencyMs: r.latencyMs, ttftMs: r.ttftMs });
// What the run spent goes on the budget of its run token (Codes C3): the same tokens and cost the
// quota accounting counts, so the next request of a run over its budget is refused.
if (runBudgets && ctx.runJti) await runBudgets.add(ctx.runJti, { tokens: r.usage.input + r.usage.output, cost: r.cost });
@@@ -484,7 +562,11 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
.run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, explain, iso(clock.now()), id);
if (f.changes) await events.runChanged(await getRow(id));
});
-- const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
++ const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(tokens_cached),0)::bigint tc, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
++ // A run that broke still spent: its record says what it cost and who paid (L40).
++ await publishUsage({ row: await getRow(id), outcome: cancelled ? 'cancelled' : 'failed',
++ usage: { input: Number(spent.ti) || 0, output: Number(spent.tout) || 0, cached: Number(spent.tc) || 0 },
++ provider: logged.length ? logged[logged.length - 1].provider_key : null, model: null, attempts });
// Account the spent tokens/cost as before, but always roll up the finished attempts (the failed
// ones included) into provider_stats_daily/placement_state: a run that answered nothing still
// tells the router a provider was unhealthy.
ddiff --git a/server/usage-record.js b/server/usage-record.js
nnew file mode 100644
iindex 0000000..9c908c0
---- /dev/null
++++ b/server/usage-record.js
@@@ -0,0 +1,122 @@
++'use strict';
++/**
++ * The record of one AI call (plan L40) and the money of it.
++ *
++ * OpenVibe.AI does not keep the meter: it writes one ai.call.recorded@1 per run through its own events publisher
++ * (server/events.js) and OpenVibe.Report keeps the ledger. So this module builds the record and the money, and
++ * nothing else: no tables, no roll-ups, no routes.
++ *
++ * The record carries the tokens (with the cache read and the cache write counted apart), the list cost, the charged
++ * cost, the fee and what the cache saved. It never carries a prompt or an answer.
++ */
++const { num } = require('./providers/usage');
++
++const PAID_BY = ['own_key', 'openvibe_key', 'plan', 'own_machine', 'free_tier', 'network_allowance', 'org_key'];
++
++const iso = (v) => (v instanceof Date ? v : new Date(v)).toISOString();
++/** Six decimals, the way money is written here. */
++const money = (v) => Math.round((Number(v) || 0) * 1e6) / 1e6;
++
++// ── The money of one record ────────────────────────────────────────────────────────────
++
++/**
++ * The prices of one call, from the human-typed price table (server/providers/rate-cards.js cards): the input,
++ * the output, the cache read and the cache write, each per million tokens.
++ *
++ * A cache price that nobody typed is not made up: the cache tokens are then priced at the plain input price and
++ * the record says so ('no cache price'). The card's own source says where each number came from.
++ */
++function pricesOf(cards) {
++ const origin = (c) => {
++ const m = /^urn:openvibe:ai:unverified:(.+)$/.exec(String((c && c.source) || ''));
++ return m ? m[1] : 'models';
++ };
++ const typed = (c) => {
++ const o = origin(c);
++ return o !== 'derived' && o !== 'flat' && o !== 'unpriced';
++ };
++ const inPrice = cards && cards.in ? Number(cards.in.unit_price_usd) || 0 : 0;
++ const outPrice = cards && cards.out ? Number(cards.out.unit_price_usd) || 0 : 0;
++ const cacheReadTyped = Boolean(cards && cards.cached && typed(cards.cached));
++ const cacheWriteTyped = Boolean(cards && cards.cache_write && typed(cards.cache_write));
++ return {
++ in: inPrice,
++ out: outPrice,
++ cache_read: cacheReadTyped ? Number(cards.cached.unit_price_usd) || 0 : inPrice,
++ cache_write: cacheWriteTyped ? Number(cards.cache_write.unit_price_usd) || 0 : inPrice,
++ cache_priced: cacheReadTyped && cacheWriteTyped,
++ source: (cacheReadTyped && cacheWriteTyped) ? origin(cards.in) : 'no cache price',
++ date: (cards && cards.in && cards.in.verified_at && cards.in.verified_at !== '1970-01-01') ? cards.in.verified_at
++ : (cards && cards.in && cards.in.effective_from && cards.in.effective_from !== '1970-01-01' ? cards.in.effective_from : null),
++ };
++}
++
++/**
++ * The money of one record: what the tokens cost at the list prices, what the cache saved against the plain input
++ * price, and what the payer was charged. A person's own key, their own machine, their plan, a free tier and an
++ * organisation's key are never charged here (0): the money left their own pocket. The OpenVibe key and the
++ * network's allowance pay the list cost; the OpenVibe key also carries the network's fee.
++ */
++function moneyOf({ prices, tokens, paidBy, feeRate = 0 }) {
++ const input = num(tokens.input);
++ const output = num(tokens.output);
++ const cacheRead = num(tokens.cache_read);
++ const cacheWrite = num(tokens.cache_write);
++ const list = money((input / 1e6) * prices.in + (output / 1e6) * prices.out
++ + (cacheRead / 1e6) * prices.cache_read + (cacheWrite / 1e6) * prices.cache_write);
++ const atInputPrice = money(((cacheRead + cacheWrite) / 1e6) * prices.in);
++ const asPriced = money((cacheRead / 1e6) * prices.cache_read + (cacheWrite / 1e6) * prices.cache_write);
++ const saved = money(Math.max(0, atInputPrice - asPriced));
++ const pays = paidBy === 'openvibe_key' || paidBy === 'network_allowance';
++ const charged = pays ? list : 0;
++ const fee = paidBy === 'openvibe_key' ? money(charged * (Number(feeRate) || 0)) : 0;
++ return { list_cost_usd: list, charged_usd: charged, fee_usd: fee, cache_saved_usd: saved };
++}
++
++// ── The record ─────────────────────────────────────────────────────────────────────────
++
++/** One record, ready to be published: the numbers checked, the text never carried. */
++function recordOf({ system = 'ai', run_id: runId = null, turn_id: turnId = null, subject = null, attribution = '', surface = '',
++ provider = null, model = null, effort = null, paid_by: paidBy, paid_by_detail: paidByDetail = null,
++ input = 0, output = 0, reasoning = 0, cache_read: cacheRead = 0, cache_write: cacheWrite = 0, cache_write_ttl: cacheWriteTtl = null,
++ images = 0, audio_seconds: audioSeconds = 0, video_seconds: videoSeconds = 0,
++ list_cost_usd: listCost = 0, charged_usd: charged = 0, fee_usd: fee = 0, cache_saved_usd: cacheSaved = 0,
++ price_source: priceSource = '', price_date: priceDate = null, latency_ms: latencyMs = null, ttft_ms: ttftMs = null,
++ attempts = 1, outcome = 'succeeded', at = new Date() }) {
++ if (!PAID_BY.includes(paidBy)) throw new Error(`paid_by must be one of ${PAID_BY.join(', ')}`);
++ return {
++ system: String(system || 'ai'),
++ run_id: runId || null,
++ turn_id: turnId || null,
++ subject: subject || null,
++ attribution: String(attribution || ''),
++ surface: String(surface || ''),
++ provider: provider || null,
++ model: model || null,
++ effort: effort || null,
++ paid_by: paidBy,
++ paid_by_detail: paidByDetail || null,
++ tokens_input: num(input),
++ tokens_output: num(output),
++ tokens_reasoning: num(reasoning),
++ tokens_cache_read: num(cacheRead),
++ tokens_cache_write: num(cacheWrite),
++ cache_write_ttl: cacheWriteTtl || null,
++ images: num(images),
++ audio_seconds: Number(audioSeconds) || 0,
++ video_seconds: Number(videoSeconds) || 0,
++ list_cost_usd: money(listCost),
++ charged_usd: money(charged),
++ fee_usd: money(fee),
++ cache_saved_usd: money(cacheSaved),
++ price_source: String(priceSource || ''),
++ price_date: priceDate || null,
++ latency_ms: latencyMs == null ? null : num(latencyMs),
++ ttft_ms: ttftMs == null ? null : num(ttftMs),
++ attempts: Math.max(0, num(attempts)),
++ outcome: String(outcome || 'succeeded'),
++ at: iso(at),
++ };
++}
++
++module.exports = { PAID_BY, money, pricesOf, moneyOf, recordOf, iso };
ddiff --git a/test/ai-usage.test.js b/test/ai-usage.test.js
nnew file mode 100644
iindex 0000000..ce31bcf
---- /dev/null
++++ b/test/ai-usage.test.js
@@@ -0,0 +1,106 @@
++'use strict';
++/**
++ * The AI meter, the part OpenVibe.AI keeps (plan L40): the tokens each adapter reads, the money of a call, and the
++ * record AI publishes as one ai.call.recorded@1 event. The ledger itself is OpenVibe.Report's.
++ */
++const assert = require('assert');
++const { suite } = require('./helpers');
++const usageParsers = require('../server/providers/usage');
++const record = require('../server/usage-record');
++const events = require('../server/events');
++
++const t = suite('ai-usage');
++
++t.test('every adapter reads its own cached tokens, and never counts them twice', () => {
++ // Anthropic: input_tokens EXCLUDES the cache reads and the cache writes.
++ const anthropic = usageParsers.anthropicUsage({
++ input_tokens: 1000, output_tokens: 200,
++ cache_read_input_tokens: 8000, cache_creation_input_tokens: 500,
++ cache_creation: { ephemeral_5m_input_tokens: 300, ephemeral_1h_input_tokens: 200 },
++ });
++ assert.strictEqual(anthropic.cached, 8000, 'the Anthropic cache read');
++ assert.strictEqual(anthropic.cache_write, 500, 'the Anthropic cache write');
++ assert.strictEqual(anthropic.cache_write_ttl, '5m', 'the bigger 5m share says so');
++ assert.strictEqual(anthropic.input, 9500, 'the whole prompt: 1000 + 8000 + 500');
++ assert.strictEqual(usageParsers.uncachedInput(anthropic), 1000, 'the uncached input is the input_tokens');
++ assert.strictEqual(usageParsers.anthropicUsage({ input_tokens: 10, cache_creation_input_tokens: 200, cache_creation: { ephemeral_1h_input_tokens: 200 } }).cache_write_ttl, '1h', 'a 1h cache write says so');
++
++ // OpenAI: prompt_tokens INCLUDES the cached tokens.
++ const openai = usageParsers.openAiUsage({
++ prompt_tokens: 1000, completion_tokens: 300,
++ prompt_tokens_details: { cached_tokens: 800 }, completion_tokens_details: { reasoning_tokens: 100 },
++ });
++ assert.strictEqual(openai.cached, 800, 'the OpenAI cache read');
++ assert.strictEqual(openai.reasoning, 100, 'the OpenAI reasoning');
++ assert.strictEqual(openai.output, 200, 'the reasoning is counted apart, never twice');
++ assert.strictEqual(usageParsers.uncachedInput(openai), 200, '1000 prompt tokens, 800 of them cached: 200 left');
++
++ // DeepSeek: its own fields, prompt_tokens = hit + miss.
++ const deepseek = usageParsers.deepSeekUsage({ prompt_tokens: 9000, completion_tokens: 100, prompt_cache_hit_tokens: 8000, prompt_cache_miss_tokens: 1000 });
++ assert.strictEqual(deepseek.cached, 8000, 'the DeepSeek cache hit');
++ assert.strictEqual(usageParsers.uncachedInput(deepseek), 1000, 'the DeepSeek cache miss is the uncached input');
++
++ // Gemini: promptTokenCount INCLUDES cachedContentTokenCount.
++ const gemini = usageParsers.geminiUsage({ promptTokenCount: 5000, candidatesTokenCount: 400, thoughtsTokenCount: 150, cachedContentTokenCount: 4000 });
++ assert.strictEqual(gemini.cached, 4000, 'the Gemini cached content');
++ assert.strictEqual(gemini.reasoning, 150, 'the Gemini thinking');
++ assert.strictEqual(usageParsers.uncachedInput(gemini), 1000, 'the Gemini uncached input');
++
++ // The Responses API counts its cached tokens apart too.
++ const responses = usageParsers.responsesUsage({ input_tokens: 1000, output_tokens: 200, input_tokens_details: { cached_tokens: 800 } });
++ assert.strictEqual(usageParsers.uncachedInput(responses), 200, 'the Responses API cache read');
++});
++
++t.test('the money of a record: a typed cache price, and none typed', () => {
++ const cards = {
++ in: { unit_price_usd: 10, source: 'urn:openvibe:ai:unverified:models', verified_at: '2026-10-31' },
++ out: { unit_price_usd: 30, source: 'urn:openvibe:ai:unverified:models' },
++ cached: { unit_price_usd: 1, source: 'urn:openvibe:ai:unverified:models' },
++ cache_write: { unit_price_usd: 12, source: 'urn:openvibe:ai:unverified:models' },
++ };
++ const tokens = { input: 1000, output: 200, cache_read: 10000, cache_write: 500 };
++ const priced = record.pricesOf(cards);
++ assert.strictEqual(priced.source, 'models', 'the prices are the ones a person typed');
++ assert.strictEqual(priced.date, '2026-10-31');
++ const money = record.moneyOf({ prices: priced, tokens, paidBy: 'openvibe_key', feeRate: 0.1 });
++ assert.strictEqual(money.list_cost_usd, 0.032, '1000 in, 200 out, 10000 cache read, 500 cache write');
++ assert.strictEqual(money.cache_saved_usd, 0.089, 'what the cache saved against the input price');
++ assert.strictEqual(money.charged_usd, 0.032, 'the OpenVibe key pays the list cost');
++ assert.strictEqual(money.fee_usd, 0.0032, 'and the network fee on top');
++ const own = record.moneyOf({ prices: priced, tokens, paidBy: 'own_key' });
++ assert.strictEqual(own.charged_usd, 0, 'the key of the person is never charged here');
++ assert.strictEqual(own.fee_usd, 0);
++
++ // No cache price typed: the cache tokens are priced at the plain input price and the record says so.
++ const unpriced = record.pricesOf({ ...cards, cached: { unit_price_usd: 1, source: 'urn:openvibe:ai:unverified:derived' }, cache_write: { unit_price_usd: 0, source: 'urn:openvibe:ai:unverified:unpriced' } });
++ assert.strictEqual(unpriced.source, 'no cache price', 'the record says the cache price is missing');
++ assert.strictEqual(unpriced.cache_read, 10, 'the cache read is priced at the input price');
++ const m2 = record.moneyOf({ prices: unpriced, tokens, paidBy: 'openvibe_key' });
++ assert.strictEqual(m2.cache_saved_usd, 0, 'nothing is saved without a cache price');
++ assert.strictEqual(m2.list_cost_usd, 0.121);
++});
++
++t.test('the record of a run carries the tokens and the money, and no text at all', () => {
++ const rec = record.recordOf({
++ system: 'ai', run_id: 'run_01JAB2C3D4E5F6G7H8J9K0MNPQ', subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ',
++ attribution: 'site:design', surface: 'ai.generate', provider: 'anthropic', model: 'claude-sonnet-4-5',
++ paid_by: 'own_key', paid_by_detail: 'anthropic ...a1b2',
++ input: 200, output: 80, reasoning: 0, cache_read: 800, cache_write: 0,
++ list_cost_usd: 0.0021, charged_usd: 0, fee_usd: 0, cache_saved_usd: 0.0012,
++ price_source: 'models', price_date: '2026-10-31', latency_ms: 1200, ttft_ms: 300, attempts: 1, outcome: 'succeeded',
++ at: '2026-10-31T18:00:00.000Z',
++ });
++ const payload = events.callPayload(rec);
++ assert.strictEqual(payload.input, 200, 'the uncached input');
++ assert.strictEqual(payload.cache_read, 800, 'the cache read');
++ assert.strictEqual(payload.paid_by, 'own_key', 'who paid');
++ assert.strictEqual(payload.cache_saved_usd, 0.0012, 'what the cache saved');
++ assert.strictEqual(payload.run_id, 'run_01JAB2C3D4E5F6G7H8J9K0MNPQ');
++ const text = JSON.stringify(payload);
++ for (const forbidden of ['prompt', 'output_text', 'answer', 'input_text', 'content']) {
++ assert.ok(!text.includes(forbidden), 'the record carries no ' + forbidden);
++ }
++ assert.throws(() => record.recordOf({ paid_by: 'someone_else' }), /paid_by/, 'only the known payers');
++});
++
++(async () => { await t.run(); })();
ddiff --git a/test/events.test.js b/test/events.test.js
iindex 65c2a7e..29a61a0 100644
---- a/test/events.test.js
++++ b/test/events.test.js
@@@ -14,6 +14,13 @@ const live = token('live', ALL);
let h;
const queued = async () => (await h.db.prepare('SELECT envelope FROM event_outbox ORDER BY id').all()).map((r) => (typeof r.envelope === 'string' ? (typeof r.envelope === 'string' ? (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope) : r.envelope) : r.envelope));
const valid = (env) => { const r = contracts.validate(`${env.event_type}@1`, env.payload); assert.ok(r.valid, `${env.event_type}: ${JSON.stringify(r.errors)}`); };
++// The ai.call.recorded schema rides with the contracts release that is not out yet: until it is, the shape is checked here.
++const validUsage = (env) => {
++ const r = (() => { try { return contracts.validate('ai.call.recorded@1', env.payload); } catch { return null; } })();
++ if (r) { assert.ok(r.valid, 'ai.call.recorded: ' + JSON.stringify(r.errors)); return; }
++ for (const f of ['system', 'surface', 'paid_by', 'input', 'output', 'reasoning', 'cache_read', 'cache_write', 'list_cost_usd', 'charged_usd', 'fee_usd', 'cache_saved_usd', 'attempts', 'outcome', 'at'])
++ assert.ok(env.payload[f] !== undefined && env.payload[f] !== null, 'the record has ' + f);
++};
t.test('boot with events on', async () => {
h = await boot({ dir: tmpDir(), env: { AI_STUB_FALLBACK: 'true' } }); // a dir: the database outlives h.stop()
@@@ -24,8 +31,13 @@ t.test('a run queues ai.run.queued then ai.run.succeeded, valid and without the
const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'ai.generate', input: { prompt: 'secret prompt words' } } });
assert.strictEqual(r.status, 201, r.text);
const evs = (await queued()).filter((e) => e.payload.run_id === r.body.run.id);
-- assert.deepStrictEqual(evs.map((e) => e.event_type), ['ai.run.queued', 'ai.run.succeeded']);
-- evs.forEach(valid);
++ // The run announces itself twice (queued, succeeded) and leaves one usage record (L40).
++ assert.deepStrictEqual(evs.map((e) => e.event_type), ['ai.run.queued', 'ai.run.succeeded', 'ai.call.recorded']);
++ evs.filter((e) => e.event_type.startsWith('ai.run.')).forEach(valid);
++ const usage = evs.find((e) => e.event_type === 'ai.call.recorded');
++ validUsage(usage);
++ assert.ok(usage.payload.input >= 0 && usage.payload.cache_read >= 0, 'the record counts the tokens');
++ assert.strictEqual(typeof usage.payload.paid_by, 'string', 'the record says who paid');
assert.strictEqual(evs[0].visibility, 'internal');
assert.deepStrictEqual(evs[1].payload.requester, { type: 'service', id: 'live' });
assert.strictEqual(evs[1].payload.finished_at !== null, true);
ddiff --git a/test/provider-responses.test.js b/test/provider-responses.test.js
iindex 89af7b0..7ced3bf 100644
---- a/test/provider-responses.test.js
++++ b/test/provider-responses.test.js
@@@ -62,9 +62,9 @@ t.test('POST /responses with input and instructions; output_text and usage are p
assert.strictEqual(b.text, undefined);
assert.strictEqual(r.text, 'hello there');
assert.strictEqual(r.model, 'gpt-4.1-mini-2025');
-- assert.deepStrictEqual(r.usage, { input: 11, output: 4, cached: 3 });
++ assert.deepStrictEqual(r.usage, { input: 11, output: 4, reasoning: 0, cached: 3, cache_write: 0, cache_write_ttl: null });
const bare = await createResponsesProvider({ key: 'r' }, { apiKey: KEY, fetchImpl: stubFetch(() => ({ output: [] })).fetchImpl }).chat(req());
-- assert.deepStrictEqual([bare.text, bare.model, bare.usage], ['', 'gpt-4.1-mini', { input: 0, output: 0, cached: 0 }]);
++ assert.deepStrictEqual([bare.text, bare.model, bare.usage], ['', 'gpt-4.1-mini', { input: 0, output: 0, reasoning: 0, cached: 0, cache_write: 0, cache_write_ttl: null }]);
});
t.test('body: reasoning models, image input, no instructions without a system prompt', () => {
@@@ -85,7 +85,7 @@ t.test('a JSON request sets text.format and steps down json_schema -> json_objec
assert.deepStrictEqual(s.calls[0].body.text, { format: { type: 'json_schema', name: 'r', schema: { type: 'object' }, strict: true } });
assert.deepStrictEqual(s.calls[1].body.text, { format: { type: 'json_object' } });
assert.strictEqual(r.text, '{"a":1}');
-- assert.deepStrictEqual(r.usage, { input: 7, output: 3, cached: 0 });
++ assert.deepStrictEqual(r.usage, { input: 7, output: 3, reasoning: 0, cached: 0, cache_write: 0, cache_write_ttl: null });
});
t.test('a non-2xx maps to the same ProviderError, code, status and retryable flag as openai.js', async () => {
ddiff --git a/test/providers.test.js b/test/providers.test.js
iindex afdd607..4b1caa9 100644
---- a/test/providers.test.js
++++ b/test/providers.test.js
@@@ -95,7 +95,7 @@ t.test('OpenAI adapter steps down json_schema -> json_object on a gateway that r
assert.ok(!systems[0].includes('JSON Schema'), 'not while structured output carries it');
assert.ok(systems[1].includes('matches this JSON Schema exactly') && systems[1].includes('{"type":"object"}'), systems[1]);
assert.strictEqual(r.text, '{"a":1}');
-- assert.deepStrictEqual(r.usage, { input: 7, output: 3, cached: 2 });
++ assert.deepStrictEqual(r.usage, { input: 7, output: 3, reasoning: 0, cached: 2, cache_write: 0, cache_write_ttl: null });
assert.ok(a.supports('embed') && a.supports('vision'));
await srv.close();
});
@@@ -120,7 +120,7 @@ t.test('Anthropic adapter returns tool JSON and counts cache reads', async () =>
const a = createAnthropicProvider({ key: 'a', base_url: srv.url, capabilities: [] }, { apiKey: 'ak' });
const r = await a.chat(req({ json: { name: 'r', schema: { type: 'object' } } }));
assert.deepStrictEqual(r.json, { ok: true });
-- assert.deepStrictEqual(r.usage, { input: 16, output: 2, cached: 10 });
++ assert.deepStrictEqual(r.usage, { input: 16, output: 2, reasoning: 0, cached: 10, cache_write: 1, cache_write_ttl: '5m' });
assert.strictEqual(a.supports('embed'), false);
await srv.close();
});