openvibers/openvibe.actor Commits
Actor: meter its own model calls (#465)
Changes
@@@ -1,6 +1,16 @@# Changelog++## Unreleased — every AI call measured++++- **AI meter** (`openvibe-sdk/ai-meter`, SDK 0.164.0): every model call Actor makes straight to a provider is recorded++ and sent to OpenVibe.AI's `POST /api/v1/usage/calls` (a call through OpenVibe.AI, with a person's own key or the++ router, is recorded by AI itself, so it is never counted twice): provider,++ model, task and call id, TTFT, tokens per second, latency, tokens, tool calls, the outcome, a failure's class, and++ the task's wall time on its last step. Actor's service token is narrowed to `ai.usage.report`. The meter is off in++ a drill and in tests, and it never fails or blocks a model call.++- Pins: openvibe-contracts v0.421.0, openvibe-sdk v0.164.0.++## Unreleased — make your own agent (blueprints)- **Blueprints** (`agents.blueprint@1`, Contracts 0.419.0): a person's agent as one document of ten parts.@@@ -265,7 +265,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).<!-- versions:start -->--- openvibe-contracts: v0.419.1--- openvibe-sdk: v0.162.0++- openvibe-contracts: v0.421.0++- openvibe-sdk: v0.164.0- openvibe-shared: v4.233.0<!-- versions:end -->@@@ -7,12 +7,12 @@"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9","port": 4950,"domain": "openvibe.actor",-- "contracts": "openvibe-contracts v0.419.1",-- "sdk": "openvibe-sdk v0.162.0",++ "contracts": "openvibe-contracts v0.421.0",++ "sdk": "openvibe-sdk v0.164.0","shared": "openvibe-shared v4.233.0","packages": [-- "openvibe-contracts v0.419.1",-- "openvibe-sdk v0.162.0",++ "openvibe-contracts v0.421.0",++ "openvibe-sdk v0.164.0","openvibe-shared v4.233.0"],"summary": "Public at https://openvibe.actor (sudo ovhost deploy actor; unit openvibe-actor on 127.0.0.1:4950). A task in plain words is classified, routed by openvibe-sdk/placement to the agent system that can do it (the OpenVibe runtime acting through OpenVibe.Search and OpenVibe.Tools, OpenAI's web agent, an open model on OpenVibe's server for private mode), run with its steps streamed live, checked by a model of another family, and delivered as platform.task@1 with its cost to a fraction of a cent and the explanation; a failed agent or check hands it to the next agent. Free tier per person with an operator ceiling.",@@@ -15,8 +15,8 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0","pg": "^8.23.0","sharp": "^0.34.5"@@@ -1712,9 +1712,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.419.1",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",-- "integrity": "sha512-iSJ63vTpvOOHBo4eU+MGC91SFsUUroolrnO+4loG31KP0XEbcn4Dpt+9S4Pqo/Y5DrQIA6X1Hckom6Z6/vKTcw==",++ "version": "0.421.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",++ "integrity": "sha512-iRSbGGqTS6Lr4XYrAvN5yhoCTkPkxrU/l3dufJCsc5BhbGU9F4ZSNt0FrgpqDseNxRM/8zx8Zu3t/w7vryAfBw==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -1728,9 +1728,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.162.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",-- "integrity": "sha512-LQC+6X21smbLzoYIpk0K7f1fjewi9KRk7x/qNoX/89cGg5cHGNLW9gd/VcQWZ1O8KuzDG4IOwWctSZ5CStMttQ==",++ "version": "0.164.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0",++ "integrity": "sha512-SjTWeFTaMhQfGVDLmIM0HlzbVbNWkqeRCub+1FnGBPC6rDDbZKlsCPYW55l36EEVh/Hxph2LWSVQjsaZqyz2rQ==","license": "MIT","bin": {"openvibe-download": "bin/openvibe-download.js"@@@ -28,8 +28,8 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0","pg": "^8.23.0","sharp": "^0.34.5"@@@ -103,9 +103,12 @@ async function createApp(opts = {}) {const sso = config.selfHosted.on ? createLocalSignIn({ config, now: s.now }) : createSso({ config, keys, fetchImpl, now: s.now, log });const stream = createTaskStream({ log });const providerFetch = opts.providerFetch || globalThis.fetch;++ // One AI meter for the process (ai-metrics.md): Actor's own model calls are sent to AI's usage route; it is off++ // in a drill and in tests, and without a Network client secret. It never fails or blocks a model call.++ const aiMeter = opts.aiMeter !== undefined ? opts.aiMeter : require('./runtime/ai-meter').createAiMeter({ config, fetchImpl: providerFetch, log });// Actor v2: one model layer (tiers), one tool registry (every module in runtime/tools) and one loop. The tools// are validated strictly in tests, leniently in production; the model lists are read once at boot.-- const models = opts.models || createModels({ config, rates: catalog.RATES, fetchImpl: providerFetch, log });++ const models = opts.models || createModels({ config, rates: catalog.RATES, fetchImpl: providerFetch, log, meter: aiMeter });const registry = opts.registry || createRegistry({ log, strict: config.nodeEnv === 'test', fetchImpl: providerFetch });await models.warm().catch((err) => log.warn('[Actor] model listing failed:', err && err.message));// Live per-tool/per-model numbers (stage V2-10), one aggregate over actor_spans cached for 5 minutes: the brain's@@@ -168,7 +171,7 @@ async function createApp(opts = {}) {const live = opts.live || createLive({ config, fetchImpl });// An agent answers the mail sent to its own address (track HO6): the consumer below takes the event, the engine// reports each run's end back here, and the approvals decide() asks it to send an approved draft.-- const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers, live, triggers, skillRun, media, tips };++ const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, aiMeter, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers, live, triggers, skillRun, media, tips };const fabricMail = require('./mail/client').createMailClient({ config, fetchImpl: opts.mailFetch || fetchImpl, tokens: opts.mailTokens || null });ctx.fabric = require('./fabric/service').createFabric({ s, enabled: !!config.fabric?.enabled, readMail: async (args) => {const out = await fabricMail.getMessage(args);@@@ -21,7 +21,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {return gracefulStop({name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.chatAsks.stop(), () => { if (ctx.vibeDelivery) ctx.vibeDelivery.stop(); }, () => ctx.keys.client.stop(), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.chatAsks.stop(), () => { if (ctx.vibeDelivery) ctx.vibeDelivery.stop(); }, () => ctx.keys.client.stop(), () => (ctx.aiMeter ? ctx.aiMeter.close() : null), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],});}nnew file mode 100644@@@ -0,0 +1,32 @@++'use strict';++++/**++ * The process's one meter client (ai-metrics.md): every model call Actor makes straight to a provider is recorded with++ * openvibe-sdk/ai-meter and sent to AI's POST /api/v1/usage/calls. A call through OpenVibe.AI (a person's own key, the++ * router) is AI's own to record, so it is never reported here.++ *++ * AI is also the only place that prices a call Actor did not route, so this client sends tokens, timing and outcome++ * only (never a price): AI applies its rate cards and publishes ai.call.recorded@1.++ *++ * Off in a drill and in tests, where the process makes no outbound calls, in a self-hosted Actor (no Network identity++ * to mint the service token), and without a Network client secret: a missing meter never stops a model call.++ */++++const { serviceAuth } = require('openvibe-contracts');++const { createMeterClient } = require('openvibe-sdk/ai-meter');++++function createAiMeter({ config, fetchImpl = globalThis.fetch, log = console } = {}) {++ const ai = config && config.ai;++ if (!ai || !ai.url || !ai.clientSecret) return null;++ if (config.drill || config.nodeEnv === 'test') return null;++ if (config.selfHosted && config.selfHosted.on) return null;++ const tokens = serviceAuth.createTokenClient({++ tokenUrl: (config.networkInternalUrl || config.networkUrl) + '/oauth/token',++ clientId: ai.clientId, clientSecret: ai.clientSecret,++ // Its own token, narrowed to reporting: a grant that is missing breaks metering only, never a model call.++ audience: ai.audience || 'openvibe.ai', scope: 'ai.usage.report', fetchImpl,++ });++ return createMeterClient({ aiUrl: ai.url, tokenClient: tokens, fetch: fetchImpl, log });++}++++module.exports = { createAiMeter };@@@ -311,7 +311,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.}/** One model turn. Checks the clock and the budget first; meters its cost after. */-- async function callModel(state, ctx, { tools, forceNoTools = false, stepKind = 'work' } = {}) {++ async function callModel(state, ctx, { tools, forceNoTools = false, stepKind = 'work', taskEnd = false } = {}) {if (state.signal && state.signal.aborted) throw Object.assign(new Error('cancelled'), { code: 'actor.task.cancelled' });if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;const defs = forceNoTools ? null : shortlistTools(tools, state);@@@ -332,6 +332,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date....(onDelta ? { onDelta } : {}),// A routine's run may wait (plan BA): the class it asks AI's router for, when the caller named one....(ctx.latency ? { latency: ctx.latency } : {}),++ // The last step of a task carries its whole wall time on the record (AI's task_ms).++ ...(taskEnd ? { taskEnd: true } : {}),});} catch (err) {await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });@@@ -476,7 +478,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.if (ctx.budgetLeft && ctx.budgetLeft() > 0) {state.messages.push({ role: 'user', content: `Stop working now (${why}). Answer from the evidence you already have, citing [E#] for every live fact; remove anything the evidence does not support. If the evidence cannot answer the task, say exactly what is missing.` });try {-- const r = await callModel(state, ctx, { tools: null, forceNoTools: true });++ const r = await callModel(state, ctx, { tools: null, forceNoTools: true, taskEnd: state.depth === 0 });if (r && r.text && r.text.trim()) return r.text.trim();} catch (err) {if (state.signal && state.signal.aborted) throw err;@@@ -619,7 +621,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.const tier = await tierFor(state, ctx);if (tier === null) { await answered('the budget cannot pay for another model call'); break; }state.tier = tier;-- const r = await callModel(state, ctx, { tools: state.tools, stepKind: 'work' });++ const r = await callModel(state, ctx, { tools: state.tools, stepKind: 'work', taskEnd: state.depth === 0 && state.selfReflections >= reflections });if (!r) { await answered('the task reached its budget'); break; }state.messages.push(r.message);if (r.text && r.text.trim()) { state.lastText = r.text.trim(); if (state.brain) state.brain.lastLine = clip(r.text.trim(), 160); }@@@ -37,6 +37,10 @@ const { createClient: createSdkClient } = require('openvibe-sdk/core');const { createAiClient } = require('openvibe-sdk/ai');const { createServiceTokenClient } = require('openvibe-sdk/auth');const { chatClient, priceUsage, readStream, ProviderError } = require('../agents/providers');++++// openvibe-sdk/ai-meter (0.164.0+) is required when the process has a meter, so a build without it still boots.++let meterLib = null;++const meterLibOf = () => (meterLib || (meterLib = require('openvibe-sdk/ai-meter')));const aiCatalogLib = require('./ai-catalog');class ModelError extends Error {@@@ -85,7 +89,7 @@ function parsePin(value) {const trimSlash = (s) => String(s || '').replace(/\/+$/, '');const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));--function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() } = {}) {++function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now(), meter = null } = {}) {const cfg = config.models || { tiers: {}, providerOrder: [], listTimeoutMs: 5000 };const p = config.providers || {};const ai = config.ai || {};@@@ -284,6 +288,110 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =const choiceStore = new AsyncLocalStorage();let aiTokens = null;++ // ── The AI meter (ai-metrics.md) ──────────────────────────────────────────────────────────────++ // The task a model call belongs to rides in an async store set by the task's engine: the id, the task's wall++ // clock and the call number. A call outside a task (a title, a page) is not metered; a call inside one always++ // is, and nothing here may delay or fail it. The client itself lives in server/runtime/ai-meter.js.++ const meterStore = new AsyncLocalStorage();++++ /** Run the rest of the work as one task's model calls (the engine wraps a task in this). */++ function withMetering(meta, fn) {++ return meterStore.run({ ...(meta || {}) }, fn);++ }++++ function currentMetering() {++ return meterStore.getStore() || null;++ }++++ /** Start one call's record; null (never a throw) when there is no meter, no task or a bad shape. */++ function startMetered({ provider, model, place, streamed }) {++ if (!meter) return null;++ const m = currentMetering();++ if (!m || !m.taskId) return null;++ try {++ m.seq = (Number(m.seq) || 0) + 1;++ return meterLibOf().startCall({++ system: 'actor',++ turn_id: `${m.taskId}:${m.seq}`,++ provider: String(provider || 'unknown').slice(0, 120),++ model: String(model || 'unknown').slice(0, 200),++ operation: 'chat',++ task: m.task || 'actor.task',++ task_id: m.taskId,++ harness: 'openvibe',++ place: place === 'machine' ? 'machine' : 'provider',++ streamed: streamed === true,++ now,++ });++ } catch (err) {++ if (log && log.warn) log.warn('[Actor] AI meter call not started:', (err && err.message) || err);++ return null;++ }++ }++++ /** The caller's onDelta, with the first piece of the answer marked on the call (TTFT). */++ function watchFirstToken(call, onDelta) {++ if (!call || typeof onDelta !== 'function') return onDelta;++ let first = false;++ return (text) => {++ if (!first) {++ first = true;++ try { call.firstToken(); } catch { /* a meter never drops a token */ }++ }++ return onDelta(text);++ };++ }++++ /** The usage Actor keeps ({ input total, cached, output }) as the meter's uncached-input shape. */++ function meterUsage(usage) {++ const u = usage || {};++ const input = Math.max(0, Number(u.input) || 0);++ const cached = Math.max(0, Number(u.cached) || 0);++ return {++ input: Math.max(0, input - cached),++ output: Math.max(0, Number(u.output) || 0),++ reasoning: Math.max(0, Number(u.reasoning) || 0),++ cache_read: cached,++ cache_write: Math.max(0, Number(u.cache_write) || 0),++ };++ }++++ /** End and record a succeeded call: its tools, its tokens, and on the task's last step its whole wall time. */++ function endMetered(call, { usage, toolCalls, taskEnd = false } = {}) {++ if (!call) return;++ try {++ const calls = Array.isArray(toolCalls) ? toolCalls.length : 0;++ for (let i = 0; i < calls; i++) call.tool();++ const m = currentMetering();++ const taskMs = taskEnd && calls === 0 && m && m.startedAt ? Math.max(0, now() - m.startedAt) : null;++ const record = call.end({++ usage: meterUsage(usage),++ outcome: 'succeeded',++ fallbacks: 0,++ ...(taskMs != null ? { task_ms: taskMs } : {}),++ });++ meter.record(record);++ } catch (err) {++ if (log && log.warn) log.warn('[Actor] AI meter record not written:', (err && err.message) || err);++ }++ }++++ /** End and record a failed or cancelled call, with its class and never a message. */++ function failMetered(call, err, signal) {++ if (!call) return;++ try {++ const cancelled = Boolean((signal && signal.aborted) || (err && err.code === 'actor.task.cancelled'));++ const record = call.end({++ usage: null,++ outcome: cancelled ? 'cancelled' : 'failed',++ error_class: String((err && err.code) || 'actor.model.failed').slice(0, 60),++ });++ meter.record(record);++ } catch (e) {++ if (log && log.warn) log.warn('[Actor] AI meter record not written:', (e && e.message) || e);++ }++ }++/** Actor's service token for OpenVibe.AI (audience openvibe.ai). Never a provider key. */function aiTokenClient() {if (aiTokens) return aiTokens;@@@ -629,7 +737,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =}/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */-- async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null, latency = null, subject = null, policy = null } = {}) {++ async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null, latency = null, subject = null, policy = null, taskEnd = false } = {}) {// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it. That path has// no latency class: the classes live on AI's router.const choice = currentChoice();@@@ -637,15 +745,36 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =// that tier. A caller may also name it directly.const policyOf = policy || (choice && choice.policy) || null;if (choice && choice.provider) {-- const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta, policy: policyOf });++ // OpenVibe.AI records every call it serves (ai.call.recorded@1) itself: Actor reporting it too would count++ // it twice. Only the calls Actor makes straight to a provider, below, are Actor's to report.++ const call = null;++ let r;++ try {++ r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta: watchFirstToken(call, onDelta), policy: policyOf });++ } catch (err) {++ failMetered(call, err, signal);++ throw err;++ }++ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER, latency_class: null };}// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.if (routerOn() || (choice && choice.model && !choice.provider)) {const args = { tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf };-- const r = routerOn() && ai.exchangeRouter && !args.target && !args.latency-- ? await chatViaExchange(args)-- : await chatViaRouter(args);++ // OpenVibe.AI records every call it serves (ai.call.recorded@1) itself: Actor reporting it too would count++ // it twice. Only the calls Actor makes straight to a provider, below, are Actor's to report.++ const call = null;++ args.onDelta = watchFirstToken(call, onDelta);++ let r;++ try {++ r = routerOn() && ai.exchangeRouter && !args.target && !args.latency++ ? await chatViaExchange(args)++ : await chatViaRouter(args);++ } catch (err) {++ failMetered(call, err, signal);++ throw err;++ }++ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };}await warm();@@@ -662,7 +791,15 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =// The check is a short structured call, so its thinking is off; the loop keeps it on.const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });-- const r = await client({ messages, tools, json, maxTokens, signal, onDelta });++ const call = startMetered({ provider: d.provider, model: d.model, place: 'provider', streamed: Boolean(onDelta) });++ let r;++ try {++ r = await client({ messages, tools, json, maxTokens, signal, onDelta: watchFirstToken(call, onDelta) });++ } catch (err) {++ failMetered(call, err, signal);++ throw err;++ }++ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };}@@@ -710,6 +847,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =return {warm, chat, resolve, tierList, modelList, resolveAll, withChoice, currentChoice, allowedKeys, aiCatalog, latencyClass,++ withMetering, currentMetering,tierConfig: cfg, PREFERRED, PREFERRED_CHECKED, TIERS,get warmed() { return warmed; },listErrors: () => ({ ...listErrors }),@@@ -586,8 +586,13 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);if (wakePayload) wakes.delete(id);let out = null;++ // The AI meter (ai-metrics.md): every model call of this task carries its id and wall clock, wherever++ // the call is made — the run, its plan, its check, its revision. A product without a model layer or++ // a process without a meter simply runs the call.++ const aiMetering = { taskId: id, task: 'actor.task', startedAt: now(), seq: 0 };++ const metered = (fn) => (typeof models.withMetering === 'function' ? models.withMetering(aiMetering, fn) : fn());// The files attached to this message in the chat (plan L55 C2): text as data; images described first.-- const attachments = await inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', err && err.message); return []; });++ const attachments = await metered(() => inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', (err && err.message) || err); return []; }));// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.// Every model call of the run — the loop's, the plan's, the check's — reads it, and no other run's does.let choice = await modelPrefs.choiceFor(s, row.requester).catch(() => null);@@@ -616,7 +621,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =// tick starts no turn by itself, so the next run is told which item changed.const context = row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [];if (!resume && !fork) context.push(...await canvases.takeNoticesForTask(s, row));-- out = await withChoice(models, choice, async () => loop.run({++ out = await metered(() => withChoice(models, choice, async () => loop.run({task: row.task, requester: fromChat ? null : row.requester, taskId: id, mode: row.mode, signal,// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.context,@@@ -636,7 +641,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =latency: await latencyClass(id),// A product conversation turn (one-engine slice): the tools this run may call.allowedTools: unifiedCode && !turnAllowed ? ['codes_task'] : turnAllowed,-- }));++ })));// The run parked (plan A2): the wait is open, so the task is not finished. It stays open// ('running' is the closest state there is today), its wall clock waits with it, and its// budget clock stops because no model runs while it waits. The desk is closed with the run.@@@ -677,11 +682,11 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =}try {await setState(id, 'verifying', { agentForEvent: agentId });-- const verify = async (current) => withChoice(models, choice, () => checkClaims({++ const verify = async (current) => metered(() => withChoice(models, choice, () => checkClaims({task: row.task, answer: current.text, evidence: current.evidence, models,writerFamily: current.family || null, signal, now: () => now(),trace: current.state && current.state.ctx && current.state.ctx.trace,-- }));++ })));let verdict = await verify(out);await meter(verdict.cost_usd);const showCheck = (v) => emit(id, { kind: 'output', step: 'check', agent: agentId, text: `${v.ok ? 'Checked' : 'Check failed'}${v.by ? ` by ${v.by}${v.cross ? '' : ' (same family)'}` : ''}: ${v.reason || ''}`.slice(0, 2000), is_error: !v.ok });@@@ -707,9 +712,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =let revised = null;if (!escalated) {escalated = true;-- revised = await loop.escalate(out, verdict.unsupported);++ revised = await metered(() => loop.escalate(out, verdict.unsupported));}-- if (!revised || revised === out || !revised.text || revised.text === out.text) revised = await loop.revise(out, verdict.unsupported);++ if (!revised || revised === out || !revised.text || revised.text === out.text) revised = await metered(() => loop.revise(out, verdict.unsupported));if (!revised || revised === out || !revised.text || revised.text === out.text) break;out = revised;verdict = await verify(out);@@@ -724,7 +729,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =for (const c of verdict.contested.slice(0, 2)) {if (budgetLeft() <= 0) break;try {-- const j = await withChoice(models, choice, () => adjudicate({ claim: c, evidence: out.evidence, task: row.task, models, writerFamily: out.family || null, signal, now: () => now(), trace: out.state && out.state.ctx && out.state.ctx.trace }));++ const j = await metered(() => withChoice(models, choice, () => adjudicate({ claim: c, evidence: out.evidence, task: row.task, models, writerFamily: out.family || null, signal, now: () => now(), trace: out.state && out.state.ctx && out.state.ctx.trace })));if (j) { await meter(j.cost_usd); if (!j.failed) contested.push({ claim: j.claim, sides: j.sides, pick: j.pick, reason: j.reason, by: j.by }); }} catch (err) {if (signal && signal.aborted) throw err;@@@ -740,14 +745,14 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =if (Array.isArray(verdict.claims) && verdict.claims.length) {const outward = !!(out.state && out.state.brain && out.state.brain.understanding && out.state.brain.understanding.outward);try {-- review = await withChoice(models, choice, () => confidence.review({++ review = await metered(() => withChoice(models, choice, () => confidence.review({claims: verdict.claims, evidence: out.evidence, task: row.task, mode: row.mode, models,outward, signal, trace: out.state && out.state.ctx && out.state.ctx.trace,budgetLeft, writerFamily: out.family || null, now: () => now(),search: async (query) => (registry? registry.run('search_web', { query }, { config, log, signal, requester: row.requester, now }): { text: 'no search is configured', isError: true }),-- }));++ })));await meter(review.cost_usd);if (review.challenged) await emit(id, { kind: 'output', step: 'check', agent: agentId, text: `Challenged ${review.challenged} high-stakes claim${review.challenged === 1 ? '' : 's'} with a counter-evidence search; ${review.unsure} could not be confirmed.`.slice(0, 2000) });} catch (err) {nnew file mode 100644@@@ -0,0 +1,159 @@++'use strict';++/**++ * The AI meter for Actor's own model calls (ai-metrics.md): every call through the tier layer carries the task id,++ * its provider and model, TTFT, tokens per second and its tokens; a failure carries its class; the one process client++ * posts the records to AI's /api/v1/usage/calls with Actor's service token.++ */++const assert = require('assert');++const http = require('node:http');++const configLib = require('../server/config');++const { createModels } = require('../server/runtime/models');++const { createAiMeter } = require('../server/runtime/ai-meter');++const { RATES } = require('../server/agents/catalog');++++const results = [];++const ok = async (name, fn) => { try { await fn(); console.log(` ✓ ${name}`); } catch (e) { results.push(e); console.log(` ✗ ${name}\n ${(e.stack || String(e)).split('\n').slice(0, 7).join('\n ')}`); } };++++const quiet = { log() {}, warn() {} };++const json = (status, obj) => ({ ok: status >= 200 && status < 300, status, headers: { get: () => 'application/json' }, json: async () => obj, text: async () => JSON.stringify(obj) });++++const env = {++ NODE_ENV: 'test',++ ACTOR_DEEPSEEK_API_KEY: 'ds', ACTOR_DEEPSEEK_BASE_URL: 'https://deepseek.test',++ ACTOR_MODEL_SMART: 'deepseek:deepseek-v4-flash',++};++++function provider({ usage, answer = 'the answer', toolCalls = null, status = 200, error = null, onFetched = null } = {}) {++ const seen = [];++ const fetchImpl = async (url, opts = {}) => {++ const u = String(url);++ if (u.endsWith('/models')) return json(200, { data: [{ id: 'deepseek-v4-flash' }] });++ seen.push({ url: u, body: opts.body ? JSON.parse(opts.body) : null });++ if (onFetched) onFetched();++ if (status !== 200) return json(status, error || { error: { message: 'no' } });++ const message = { role: 'assistant', content: answer, ...(toolCalls ? { tool_calls: toolCalls } : {}) };++ return json(200, { model: 'deepseek-v4-flash', choices: [{ message, finish_reason: 'stop' }], usage });++ };++ return { fetchImpl, seen };++}++++const tickingMeter = () => ({ records: [], record(rec) { this.records.push(rec); return true; } });++++(async () => {++ await ok('a provider call records TTFT, TPS, tokens, the provider and the task id', async () => {++ let t = 10000;++ const usage = { prompt_tokens: 1200, completion_tokens: 300, prompt_tokens_details: { cached_tokens: 1000 } };++ const stub = provider({ usage, onFetched: () => { t += 800; } });++ const meter = tickingMeter();++ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter, now: () => t });++ await models.warm();++ const meta = { taskId: 'tsk_meter', task: 'actor.task', startedAt: 10000, seq: 0 };++ const r = await models.withMetering(meta, () => models.chat({++ tier: 'smart', messages: [{ role: 'user', content: 'x' }], onDelta: () => { t += 3000; }, taskEnd: true,++ }));++ assert.strictEqual(r.provider, 'deepseek');++ assert.strictEqual(r.text, 'the answer');++ assert.strictEqual(meter.records.length, 1);++ const rec = meter.records[0];++ assert.strictEqual(rec.system, 'actor');++ assert.strictEqual(rec.task_id, 'tsk_meter');++ assert.strictEqual(rec.turn_id, 'tsk_meter:1');++ assert.strictEqual(rec.provider, 'deepseek');++ assert.strictEqual(rec.model, 'deepseek-v4-flash');++ assert.strictEqual(rec.operation, 'chat');++ assert.strictEqual(rec.task, 'actor.task');++ assert.strictEqual(rec.harness, 'openvibe');++ assert.strictEqual(rec.place, 'provider');++ assert.strictEqual(rec.streamed, true);++ assert.strictEqual(rec.outcome, 'succeeded');++ assert.deepStrictEqual([rec.input, rec.output, rec.cache_read], [200, 300, 1000]);++ assert.deepStrictEqual([rec.ttft_ms, rec.latency_ms, rec.tps], [800, 3800, 100]);++ assert.strictEqual(rec.task_ms, 3800, 'the task\'s wall time on its last step');++ });++++ await ok('tool calls are counted, and a step with tools is not the task\'s last', async () => {++ let t = 5000;++ const calls = [{ id: 'c1', type: 'function', function: { name: 'lookup', arguments: '{}' } }, { id: 'c2', type: 'function', function: { name: 'search_web', arguments: '{}' } }];++ const stub = provider({ usage: { prompt_tokens: 100, completion_tokens: 10 }, toolCalls: calls, onFetched: () => { t += 100; } });++ const meter = tickingMeter();++ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter, now: () => t });++ await models.warm();++ const meta = { taskId: 'tsk_tools', startedAt: 5000, seq: 0 };++ const r = await models.withMetering(meta, () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }], taskEnd: true }));++ assert.strictEqual(r.toolCalls.length, 2);++ assert.strictEqual(meter.records[0].tool_calls, 2);++ assert.strictEqual(meter.records[0].task_ms, undefined);++ });++++ await ok('a failed call records outcome failed and its class; metering never hides the error', async () => {++ const stub = provider({ status: 429, error: { error: { message: 'slow down' } } });++ const meter = tickingMeter();++ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter });++ await models.warm();++ await assert.rejects(++ () => models.withMetering({ taskId: 'tsk_fail', startedAt: 1, seq: 0 }, () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }] })),++ (err) => err && err.code === 'actor.provider.rate_limited',++ );++ assert.strictEqual(meter.records.length, 1);++ assert.strictEqual(meter.records[0].outcome, 'failed');++ assert.strictEqual(meter.records[0].error_class, 'actor.provider.rate_limited');++ assert.strictEqual(meter.records[0].task_id, 'tsk_fail');++ assert.strictEqual(meter.records[0].turn_id, 'tsk_fail:1');++ });++++ await ok('a call that goes through OpenVibe.AI is AI\'s to record, never Actor\'s (no double count)', async () => {++ const meter = tickingMeter();++ const fetchImpl = async () => { throw new Error('AI is down'); };++ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl, log: quiet, meter });++ const meta = { taskId: 'tsk_via_ai', task: 'actor.task', startedAt: 0, seq: 0 };++ await assert.rejects(models.withMetering(meta, () => models.withChoice({ provider: 'openai', model: 'gpt-5-mini' },++ () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }] }))));++ assert.strictEqual(meter.records.length, 0, 'AI publishes its own ai.call.recorded@1 for this call');++ });++++ await ok('the meter client posts the records to a stub AI with Actor\'s service token', async () => {++ const seen = [];++ const tokenRequests = [];++ const server = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => {++ if (req.url === '/oauth/token') {++ tokenRequests.push(body);++ res.writeHead(200, { 'content-type': 'application/json' });++ return res.end(JSON.stringify({ access_token: 'tok_actor', token_type: 'Bearer', expires_in: 300, scope: 'ai.usage.report' }));++ }++ seen.push({ url: req.url, auth: req.headers.authorization, body: JSON.parse(body) });++ res.writeHead(202, { 'content-type': 'application/json' });++ res.end('{}');++ });++ });++ await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));++ const url = `http://127.0.0.1:${server.address().port}`;++ let meter = null;++ try {++ const config = configLib.load({++ NODE_ENV: 'production', OV_OAUTH_CLIENT_SECRET: 'secret', OV_NETWORK_INTERNAL_URL: url,++ OV_AI_URL: url, ACTOR_AI_ROUTER: '0',++ });++ meter = createAiMeter({ config, log: quiet });++ assert.ok(meter, 'a hosted Actor with a secret has a meter');++ meter.record({ system: 'actor', turn_id: 'tsk_post:1', provider: 'deepseek', model: 'deepseek-v4-flash', outcome: 'succeeded', at: new Date().toISOString(), input: 1, output: 1 });++ await meter.flush();++ assert.strictEqual(seen.length, 1);++ assert.strictEqual(seen[0].url, '/api/v1/usage/calls');++ assert.strictEqual(seen[0].auth, 'Bearer tok_actor');++ assert.strictEqual(seen[0].body.calls.length, 1);++ assert.strictEqual(seen[0].body.calls[0].turn_id, 'tsk_post:1');++ assert.strictEqual(new URLSearchParams(tokenRequests[0]).get('scope'), 'ai.usage.report');++ assert.strictEqual(createAiMeter({ config: configLib.load({ ...env, ACTOR_DRILL: '1' }), log: quiet }), null, 'off in a drill');++ assert.strictEqual(createAiMeter({ config: configLib.load(env), log: quiet }), null, 'off in tests');++ } finally {++ if (meter) await meter.close();++ await new Promise((resolve) => server.close(resolve));++ }++ });++++ if (results.length) { console.log(`\n${results.length} failed`); process.exit(1); }++ console.log('\nall passed');++})();
@@ -1,6 +1,16 @@
@@ -1,6 +1,16 @@
# Changelog
# Changelog
+## Unreleased — every AI call measured
+
+- **AI meter** (`openvibe-sdk/ai-meter`, SDK 0.164.0): every model call Actor makes straight to a provider is recorded
+ and sent to OpenVibe.AI's `POST /api/v1/usage/calls` (a call through OpenVibe.AI, with a person's own key or the
+ router, is recorded by AI itself, so it is never counted twice): provider,
+ model, task and call id, TTFT, tokens per second, latency, tokens, tool calls, the outcome, a failure's class, and
+ the task's wall time on its last step. Actor's service token is narrowed to `ai.usage.report`. The meter is off in
+ a drill and in tests, and it never fails or blocks a model call.
+- Pins: openvibe-contracts v0.421.0, openvibe-sdk v0.164.0.
+
## Unreleased — make your own agent (blueprints)
## Unreleased — make your own agent (blueprints)
- **Blueprints** (`agents.blueprint@1`, Contracts 0.419.0): a person's agent as one document of ten parts.
- **Blueprints** (`agents.blueprint@1`, Contracts 0.419.0): a person's agent as one document of ten parts.
@@ -265,7 +265,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
@@ -265,7 +265,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.419.1
-- openvibe-sdk: v0.162.0
+- openvibe-contracts: v0.421.0
+- openvibe-sdk: v0.164.0
- openvibe-shared: v4.233.0
- openvibe-shared: v4.233.0
<!-- versions:end -->
<!-- versions:end -->
@@ -7,12 +7,12 @@
@@ -7,12 +7,12 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": 4950,
"port": 4950,
"domain": "openvibe.actor",
"domain": "openvibe.actor",
- "contracts": "openvibe-contracts v0.419.1",
- "sdk": "openvibe-sdk v0.162.0",
+ "contracts": "openvibe-contracts v0.421.0",
+ "sdk": "openvibe-sdk v0.164.0",
"shared": "openvibe-shared v4.233.0",
"shared": "openvibe-shared v4.233.0",
"packages": [
"packages": [
- "openvibe-contracts v0.419.1",
- "openvibe-sdk v0.162.0",
+ "openvibe-contracts v0.421.0",
+ "openvibe-sdk v0.164.0",
"openvibe-shared v4.233.0"
"openvibe-shared v4.233.0"
],
],
"summary": "Public at https://openvibe.actor (sudo ovhost deploy actor; unit openvibe-actor on 127.0.0.1:4950). A task in plain words is classified, routed by openvibe-sdk/placement to the agent system that can do it (the OpenVibe runtime acting through OpenVibe.Search and OpenVibe.Tools, OpenAI's web agent, an open model on OpenVibe's server for private mode), run with its steps streamed live, checked by a model of another family, and delivered as platform.task@1 with its cost to a fraction of a cent and the explanation; a failed agent or check hands it to the next agent. Free tier per person with an operator ceiling.",
"summary": "Public at https://openvibe.actor (sudo ovhost deploy actor; unit openvibe-actor on 127.0.0.1:4950). A task in plain words is classified, routed by openvibe-sdk/placement to the agent system that can do it (the OpenVibe runtime acting through OpenVibe.Search and OpenVibe.Tools, OpenAI's web agent, an open model on OpenVibe's server for private mode), run with its steps streamed live, checked by a model of another family, and delivered as platform.task@1 with its cost to a fraction of a cent and the explanation; a failed agent or check hands it to the next agent. Free tier per person with an operator ceiling.",
@@ -15,8 +15,8 @@
@@ -15,8 +15,8 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.34.5"
"sharp": "^0.34.5"
@@ -1712,9 +1712,9 @@
@@ -1712,9 +1712,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.419.1",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",
- "integrity": "sha512-iSJ63vTpvOOHBo4eU+MGC91SFsUUroolrnO+4loG31KP0XEbcn4Dpt+9S4Pqo/Y5DrQIA6X1Hckom6Z6/vKTcw==",
+ "version": "0.421.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",
+ "integrity": "sha512-iRSbGGqTS6Lr4XYrAvN5yhoCTkPkxrU/l3dufJCsc5BhbGU9F4ZSNt0FrgpqDseNxRM/8zx8Zu3t/w7vryAfBw==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -1728,9 +1728,9 @@
@@ -1728,9 +1728,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.162.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",
- "integrity": "sha512-LQC+6X21smbLzoYIpk0K7f1fjewi9KRk7x/qNoX/89cGg5cHGNLW9gd/VcQWZ1O8KuzDG4IOwWctSZ5CStMttQ==",
+ "version": "0.164.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0",
+ "integrity": "sha512-SjTWeFTaMhQfGVDLmIM0HlzbVbNWkqeRCub+1FnGBPC6rDDbZKlsCPYW55l36EEVh/Hxph2LWSVQjsaZqyz2rQ==",
"license": "MIT",
"license": "MIT",
"bin": {
"bin": {
"openvibe-download": "bin/openvibe-download.js"
"openvibe-download": "bin/openvibe-download.js"
@@ -28,8 +28,8 @@
@@ -28,8 +28,8 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.1",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.162.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.421.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.164.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.233.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"sharp": "^0.34.5"
"sharp": "^0.34.5"
@@ -103,9 +103,12 @@ async function createApp(opts = {}) {
@@ -103,9 +103,12 @@ async function createApp(opts = {}) {
const sso = config.selfHosted.on ? createLocalSignIn({ config, now: s.now }) : createSso({ config, keys, fetchImpl, now: s.now, log });
const sso = config.selfHosted.on ? createLocalSignIn({ config, now: s.now }) : createSso({ config, keys, fetchImpl, now: s.now, log });
const stream = createTaskStream({ log });
const stream = createTaskStream({ log });
const providerFetch = opts.providerFetch || globalThis.fetch;
const providerFetch = opts.providerFetch || globalThis.fetch;
+ // One AI meter for the process (ai-metrics.md): Actor's own model calls are sent to AI's usage route; it is off
+ // in a drill and in tests, and without a Network client secret. It never fails or blocks a model call.
+ const aiMeter = opts.aiMeter !== undefined ? opts.aiMeter : require('./runtime/ai-meter').createAiMeter({ config, fetchImpl: providerFetch, log });
// Actor v2: one model layer (tiers), one tool registry (every module in runtime/tools) and one loop. The tools
// Actor v2: one model layer (tiers), one tool registry (every module in runtime/tools) and one loop. The tools
// are validated strictly in tests, leniently in production; the model lists are read once at boot.
// are validated strictly in tests, leniently in production; the model lists are read once at boot.
- const models = opts.models || createModels({ config, rates: catalog.RATES, fetchImpl: providerFetch, log });
+ const models = opts.models || createModels({ config, rates: catalog.RATES, fetchImpl: providerFetch, log, meter: aiMeter });
const registry = opts.registry || createRegistry({ log, strict: config.nodeEnv === 'test', fetchImpl: providerFetch });
const registry = opts.registry || createRegistry({ log, strict: config.nodeEnv === 'test', fetchImpl: providerFetch });
await models.warm().catch((err) => log.warn('[Actor] model listing failed:', err && err.message));
await models.warm().catch((err) => log.warn('[Actor] model listing failed:', err && err.message));
// Live per-tool/per-model numbers (stage V2-10), one aggregate over actor_spans cached for 5 minutes: the brain's
// Live per-tool/per-model numbers (stage V2-10), one aggregate over actor_spans cached for 5 minutes: the brain's
@@ -168,7 +171,7 @@ async function createApp(opts = {}) {
@@ -168,7 +171,7 @@ async function createApp(opts = {}) {
const live = opts.live || createLive({ config, fetchImpl });
const live = opts.live || createLive({ config, fetchImpl });
// An agent answers the mail sent to its own address (track HO6): the consumer below takes the event, the engine
// An agent answers the mail sent to its own address (track HO6): the consumer below takes the event, the engine
// reports each run's end back here, and the approvals decide() asks it to send an approved draft.
// reports each run's end back here, and the approvals decide() asks it to send an approved draft.
- const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers, live, triggers, skillRun, media, tips };
+ const ctx = { config, s, keys, aiCatalogConfig, sso, stream, adapters, engine, webhooks, principal, projects, log, chatAsks, vibeDelivery, models, aiMeter, registry, loop, notifier, stats, checkpoints, connectors, waits, sharing, computers, live, triggers, skillRun, media, tips };
const fabricMail = require('./mail/client').createMailClient({ config, fetchImpl: opts.mailFetch || fetchImpl, tokens: opts.mailTokens || null });
const fabricMail = require('./mail/client').createMailClient({ config, fetchImpl: opts.mailFetch || fetchImpl, tokens: opts.mailTokens || null });
ctx.fabric = require('./fabric/service').createFabric({ s, enabled: !!config.fabric?.enabled, readMail: async (args) => {
ctx.fabric = require('./fabric/service').createFabric({ s, enabled: !!config.fabric?.enabled, readMail: async (args) => {
const out = await fabricMail.getMessage(args);
const out = await fabricMail.getMessage(args);
@@ -21,7 +21,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;
@@ -21,7 +21,7 @@ const PRUNE_EVERY_MS = 6 * 60 * 60_000;
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
return gracefulStop({
name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'Actor', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.chatAsks.stop(), () => { if (ctx.vibeDelivery) ctx.vibeDelivery.stop(); }, () => ctx.keys.client.stop(), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
+ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.engine.stop(), () => ctx.webhooks.stop(), () => ctx.chatAsks.stop(), () => { if (ctx.vibeDelivery) ctx.vibeDelivery.stop(); }, () => ctx.keys.client.stop(), () => (ctx.aiMeter ? ctx.aiMeter.close() : null), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
});
});
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,32 @@
@@ -0,0 +1,32 @@
+'use strict';
+
+/**
+ * The process's one meter client (ai-metrics.md): every model call Actor makes straight to a provider is recorded with
+ * openvibe-sdk/ai-meter and sent to AI's POST /api/v1/usage/calls. A call through OpenVibe.AI (a person's own key, the
+ * router) is AI's own to record, so it is never reported here.
+ *
+ * AI is also the only place that prices a call Actor did not route, so this client sends tokens, timing and outcome
+ * only (never a price): AI applies its rate cards and publishes ai.call.recorded@1.
+ *
+ * Off in a drill and in tests, where the process makes no outbound calls, in a self-hosted Actor (no Network identity
+ * to mint the service token), and without a Network client secret: a missing meter never stops a model call.
+ */
+
+const { serviceAuth } = require('openvibe-contracts');
+const { createMeterClient } = require('openvibe-sdk/ai-meter');
+
+function createAiMeter({ config, fetchImpl = globalThis.fetch, log = console } = {}) {
+ const ai = config && config.ai;
+ if (!ai || !ai.url || !ai.clientSecret) return null;
+ if (config.drill || config.nodeEnv === 'test') return null;
+ if (config.selfHosted && config.selfHosted.on) return null;
+ const tokens = serviceAuth.createTokenClient({
+ tokenUrl: (config.networkInternalUrl || config.networkUrl) + '/oauth/token',
+ clientId: ai.clientId, clientSecret: ai.clientSecret,
+ // Its own token, narrowed to reporting: a grant that is missing breaks metering only, never a model call.
+ audience: ai.audience || 'openvibe.ai', scope: 'ai.usage.report', fetchImpl,
+ });
+ return createMeterClient({ aiUrl: ai.url, tokenClient: tokens, fetch: fetchImpl, log });
+}
+
+module.exports = { createAiMeter };
@@ -311,7 +311,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -311,7 +311,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
}
}
/** One model turn. Checks the clock and the budget first; meters its cost after. */
/** One model turn. Checks the clock and the budget first; meters its cost after. */
- async function callModel(state, ctx, { tools, forceNoTools = false, stepKind = 'work' } = {}) {
+ async function callModel(state, ctx, { tools, forceNoTools = false, stepKind = 'work', taskEnd = false } = {}) {
if (state.signal && state.signal.aborted) throw Object.assign(new Error('cancelled'), { code: 'actor.task.cancelled' });
if (state.signal && state.signal.aborted) throw Object.assign(new Error('cancelled'), { code: 'actor.task.cancelled' });
if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;
if (ctx.budgetLeft && ctx.budgetLeft() <= 0) return null;
const defs = forceNoTools ? null : shortlistTools(tools, state);
const defs = forceNoTools ? null : shortlistTools(tools, state);
@@ -332,6 +332,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -332,6 +332,8 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
...(onDelta ? { onDelta } : {}),
...(onDelta ? { onDelta } : {}),
// A routine's run may wait (plan BA): the class it asks AI's router for, when the caller named one.
// A routine's run may wait (plan BA): the class it asks AI's router for, when the caller named one.
...(ctx.latency ? { latency: ctx.latency } : {}),
...(ctx.latency ? { latency: ctx.latency } : {}),
+ // The last step of a task carries its whole wall time on the record (AI's task_ms).
+ ...(taskEnd ? { taskEnd: true } : {}),
});
});
} catch (err) {
} catch (err) {
await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });
await spanModel(ctx.trace, { tier: state.tier, family: state.family, startedAt, ms: now() - startedAt, error: err, step: state.calls + 1 });
@@ -476,7 +478,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -476,7 +478,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
if (ctx.budgetLeft && ctx.budgetLeft() > 0) {
if (ctx.budgetLeft && ctx.budgetLeft() > 0) {
state.messages.push({ role: 'user', content: `Stop working now (${why}). Answer from the evidence you already have, citing [E#] for every live fact; remove anything the evidence does not support. If the evidence cannot answer the task, say exactly what is missing.` });
state.messages.push({ role: 'user', content: `Stop working now (${why}). Answer from the evidence you already have, citing [E#] for every live fact; remove anything the evidence does not support. If the evidence cannot answer the task, say exactly what is missing.` });
try {
try {
- const r = await callModel(state, ctx, { tools: null, forceNoTools: true });
+ const r = await callModel(state, ctx, { tools: null, forceNoTools: true, taskEnd: state.depth === 0 });
if (r && r.text && r.text.trim()) return r.text.trim();
if (r && r.text && r.text.trim()) return r.text.trim();
} catch (err) {
} catch (err) {
if (state.signal && state.signal.aborted) throw err;
if (state.signal && state.signal.aborted) throw err;
@@ -619,7 +621,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -619,7 +621,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
const tier = await tierFor(state, ctx);
const tier = await tierFor(state, ctx);
if (tier === null) { await answered('the budget cannot pay for another model call'); break; }
if (tier === null) { await answered('the budget cannot pay for another model call'); break; }
state.tier = tier;
state.tier = tier;
- const r = await callModel(state, ctx, { tools: state.tools, stepKind: 'work' });
+ const r = await callModel(state, ctx, { tools: state.tools, stepKind: 'work', taskEnd: state.depth === 0 && state.selfReflections >= reflections });
if (!r) { await answered('the task reached its budget'); break; }
if (!r) { await answered('the task reached its budget'); break; }
state.messages.push(r.message);
state.messages.push(r.message);
if (r.text && r.text.trim()) { state.lastText = r.text.trim(); if (state.brain) state.brain.lastLine = clip(r.text.trim(), 160); }
if (r.text && r.text.trim()) { state.lastText = r.text.trim(); if (state.brain) state.brain.lastLine = clip(r.text.trim(), 160); }
@@ -37,6 +37,10 @@ const { createClient: createSdkClient } = require('openvibe-sdk/core');
@@ -37,6 +37,10 @@ const { createClient: createSdkClient } = require('openvibe-sdk/core');
const { createAiClient } = require('openvibe-sdk/ai');
const { createAiClient } = require('openvibe-sdk/ai');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { createServiceTokenClient } = require('openvibe-sdk/auth');
const { chatClient, priceUsage, readStream, ProviderError } = require('../agents/providers');
const { chatClient, priceUsage, readStream, ProviderError } = require('../agents/providers');
+
+// openvibe-sdk/ai-meter (0.164.0+) is required when the process has a meter, so a build without it still boots.
+let meterLib = null;
+const meterLibOf = () => (meterLib || (meterLib = require('openvibe-sdk/ai-meter')));
const aiCatalogLib = require('./ai-catalog');
const aiCatalogLib = require('./ai-catalog');
class ModelError extends Error {
class ModelError extends Error {
@@ -85,7 +89,7 @@ function parsePin(value) {
@@ -85,7 +89,7 @@ function parsePin(value) {
const trimSlash = (s) => String(s || '').replace(/\/+$/, '');
const trimSlash = (s) => String(s || '').replace(/\/+$/, '');
const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));
const clip = (s, n) => (String(s || '').length > n ? `${String(s).slice(0, n - 1)}…` : String(s || ''));
-function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now() } = {}) {
+function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log = console, now = () => Date.now(), meter = null } = {}) {
const cfg = config.models || { tiers: {}, providerOrder: [], listTimeoutMs: 5000 };
const cfg = config.models || { tiers: {}, providerOrder: [], listTimeoutMs: 5000 };
const p = config.providers || {};
const p = config.providers || {};
const ai = config.ai || {};
const ai = config.ai || {};
@@ -284,6 +288,110 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -284,6 +288,110 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
const choiceStore = new AsyncLocalStorage();
const choiceStore = new AsyncLocalStorage();
let aiTokens = null;
let aiTokens = null;
+ // ── The AI meter (ai-metrics.md) ──────────────────────────────────────────────────────────────
+ // The task a model call belongs to rides in an async store set by the task's engine: the id, the task's wall
+ // clock and the call number. A call outside a task (a title, a page) is not metered; a call inside one always
+ // is, and nothing here may delay or fail it. The client itself lives in server/runtime/ai-meter.js.
+ const meterStore = new AsyncLocalStorage();
+
+ /** Run the rest of the work as one task's model calls (the engine wraps a task in this). */
+ function withMetering(meta, fn) {
+ return meterStore.run({ ...(meta || {}) }, fn);
+ }
+
+ function currentMetering() {
+ return meterStore.getStore() || null;
+ }
+
+ /** Start one call's record; null (never a throw) when there is no meter, no task or a bad shape. */
+ function startMetered({ provider, model, place, streamed }) {
+ if (!meter) return null;
+ const m = currentMetering();
+ if (!m || !m.taskId) return null;
+ try {
+ m.seq = (Number(m.seq) || 0) + 1;
+ return meterLibOf().startCall({
+ system: 'actor',
+ turn_id: `${m.taskId}:${m.seq}`,
+ provider: String(provider || 'unknown').slice(0, 120),
+ model: String(model || 'unknown').slice(0, 200),
+ operation: 'chat',
+ task: m.task || 'actor.task',
+ task_id: m.taskId,
+ harness: 'openvibe',
+ place: place === 'machine' ? 'machine' : 'provider',
+ streamed: streamed === true,
+ now,
+ });
+ } catch (err) {
+ if (log && log.warn) log.warn('[Actor] AI meter call not started:', (err && err.message) || err);
+ return null;
+ }
+ }
+
+ /** The caller's onDelta, with the first piece of the answer marked on the call (TTFT). */
+ function watchFirstToken(call, onDelta) {
+ if (!call || typeof onDelta !== 'function') return onDelta;
+ let first = false;
+ return (text) => {
+ if (!first) {
+ first = true;
+ try { call.firstToken(); } catch { /* a meter never drops a token */ }
+ }
+ return onDelta(text);
+ };
+ }
+
+ /** The usage Actor keeps ({ input total, cached, output }) as the meter's uncached-input shape. */
+ function meterUsage(usage) {
+ const u = usage || {};
+ const input = Math.max(0, Number(u.input) || 0);
+ const cached = Math.max(0, Number(u.cached) || 0);
+ return {
+ input: Math.max(0, input - cached),
+ output: Math.max(0, Number(u.output) || 0),
+ reasoning: Math.max(0, Number(u.reasoning) || 0),
+ cache_read: cached,
+ cache_write: Math.max(0, Number(u.cache_write) || 0),
+ };
+ }
+
+ /** End and record a succeeded call: its tools, its tokens, and on the task's last step its whole wall time. */
+ function endMetered(call, { usage, toolCalls, taskEnd = false } = {}) {
+ if (!call) return;
+ try {
+ const calls = Array.isArray(toolCalls) ? toolCalls.length : 0;
+ for (let i = 0; i < calls; i++) call.tool();
+ const m = currentMetering();
+ const taskMs = taskEnd && calls === 0 && m && m.startedAt ? Math.max(0, now() - m.startedAt) : null;
+ const record = call.end({
+ usage: meterUsage(usage),
+ outcome: 'succeeded',
+ fallbacks: 0,
+ ...(taskMs != null ? { task_ms: taskMs } : {}),
+ });
+ meter.record(record);
+ } catch (err) {
+ if (log && log.warn) log.warn('[Actor] AI meter record not written:', (err && err.message) || err);
+ }
+ }
+
+ /** End and record a failed or cancelled call, with its class and never a message. */
+ function failMetered(call, err, signal) {
+ if (!call) return;
+ try {
+ const cancelled = Boolean((signal && signal.aborted) || (err && err.code === 'actor.task.cancelled'));
+ const record = call.end({
+ usage: null,
+ outcome: cancelled ? 'cancelled' : 'failed',
+ error_class: String((err && err.code) || 'actor.model.failed').slice(0, 60),
+ });
+ meter.record(record);
+ } catch (e) {
+ if (log && log.warn) log.warn('[Actor] AI meter record not written:', (e && e.message) || e);
+ }
+ }
+
/** Actor's service token for OpenVibe.AI (audience openvibe.ai). Never a provider key. */
/** Actor's service token for OpenVibe.AI (audience openvibe.ai). Never a provider key. */
function aiTokenClient() {
function aiTokenClient() {
if (aiTokens) return aiTokens;
if (aiTokens) return aiTokens;
@@ -629,7 +737,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -629,7 +737,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
}
}
/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */
/** One model call. The caller checks the budget first and meters cost_usd after; this layer only prices it. */
- async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null, latency = null, subject = null, policy = null } = {}) {
+ async function chat({ tier = 'smart', messages = null, tools = null, json = false, maxTokens = 4000, signal, avoidFamily = null, provider = null, model = null, onDelta = null, latency = null, subject = null, policy = null, taskEnd = false } = {}) {
// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it. That path has
// The person's own key (plan L55 C3): every model call of this run goes to OpenVibe.AI with it. That path has
// no latency class: the classes live on AI's router.
// no latency class: the classes live on AI's router.
const choice = currentChoice();
const choice = currentChoice();
@@ -637,15 +745,36 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -637,15 +745,36 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
// that tier. A caller may also name it directly.
// that tier. A caller may also name it directly.
const policyOf = policy || (choice && choice.policy) || null;
const policyOf = policy || (choice && choice.policy) || null;
if (choice && choice.provider) {
if (choice && choice.provider) {
- const r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta, policy: policyOf });
+ // OpenVibe.AI records every call it serves (ai.call.recorded@1) itself: Actor reporting it too would count
+ // it twice. Only the calls Actor makes straight to a provider, below, are Actor's to report.
+ const call = null;
+ let r;
+ try {
+ r = await chatViaAi({ choice, messages, tools, json, maxTokens, signal, onDelta: watchFirstToken(call, onDelta), policy: policyOf });
+ } catch (err) {
+ failMetered(call, err, signal);
+ throw err;
+ }
+ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER, latency_class: null };
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER, latency_class: null };
}
}
// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.
// The router (the AI layer): the tier names a policy on AI and the person's allowance pays. No provider key.
if (routerOn() || (choice && choice.model && !choice.provider)) {
if (routerOn() || (choice && choice.model && !choice.provider)) {
const args = { tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf };
const args = { tier, messages, tools, json, maxTokens, signal, onDelta, subject: subject || (choice && choice.subject), target: choice && choice.target, latency, policy: policyOf };
- const r = routerOn() && ai.exchangeRouter && !args.target && !args.latency
- ? await chatViaExchange(args)
- : await chatViaRouter(args);
+ // OpenVibe.AI records every call it serves (ai.call.recorded@1) itself: Actor reporting it too would count
+ // it twice. Only the calls Actor makes straight to a provider, below, are Actor's to report.
+ const call = null;
+ args.onDelta = watchFirstToken(call, onDelta);
+ let r;
+ try {
+ r = routerOn() && ai.exchangeRouter && !args.target && !args.latency
+ ? await chatViaExchange(args)
+ : await chatViaRouter(args);
+ } catch (err) {
+ failMetered(call, err, signal);
+ throw err;
+ }
+ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
return { ...r, tier, provider: AI_PROVIDER, family: AI_PROVIDER };
}
}
await warm();
await warm();
@@ -662,7 +791,15 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -662,7 +791,15 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
// The check is a short structured call, so its thinking is off; the loop keeps it on.
// The check is a short structured call, so its thinking is off; the loop keeps it on.
const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;
const extraBody = d.provider === 'deepseek' && tier === 'check' ? { thinking: { type: 'disabled' } } : null;
const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });
const client = chatClient({ baseUrl: endpoints[d.provider].baseUrl, apiKey: endpoints[d.provider].apiKey, model: d.model, rates: d.rates, fetchImpl, timeoutMs: 180_000, extraBody });
- const r = await client({ messages, tools, json, maxTokens, signal, onDelta });
+ const call = startMetered({ provider: d.provider, model: d.model, place: 'provider', streamed: Boolean(onDelta) });
+ let r;
+ try {
+ r = await client({ messages, tools, json, maxTokens, signal, onDelta: watchFirstToken(call, onDelta) });
+ } catch (err) {
+ failMetered(call, err, signal);
+ throw err;
+ }
+ endMetered(call, { usage: r.usage, toolCalls: r.toolCalls, taskEnd });
return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };
return { ...r, cost_usd: r.cost_usd || priceUsage(d.rates, r.usage || {}), tier, provider: d.provider, family: d.family };
}
}
@@ -710,6 +847,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
@@ -710,6 +847,7 @@ function createModels({ config, rates = {}, fetchImpl = globalThis.fetch, log =
return {
return {
warm, chat, resolve, tierList, modelList, resolveAll, withChoice, currentChoice, allowedKeys, aiCatalog, latencyClass,
warm, chat, resolve, tierList, modelList, resolveAll, withChoice, currentChoice, allowedKeys, aiCatalog, latencyClass,
+ withMetering, currentMetering,
tierConfig: cfg, PREFERRED, PREFERRED_CHECKED, TIERS,
tierConfig: cfg, PREFERRED, PREFERRED_CHECKED, TIERS,
get warmed() { return warmed; },
get warmed() { return warmed; },
listErrors: () => ({ ...listErrors }),
listErrors: () => ({ ...listErrors }),
@@ -586,8 +586,13 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -586,8 +586,13 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);
const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);
if (wakePayload) wakes.delete(id);
if (wakePayload) wakes.delete(id);
let out = null;
let out = null;
+ // The AI meter (ai-metrics.md): every model call of this task carries its id and wall clock, wherever
+ // the call is made — the run, its plan, its check, its revision. A product without a model layer or
+ // a process without a meter simply runs the call.
+ const aiMetering = { taskId: id, task: 'actor.task', startedAt: now(), seq: 0 };
+ const metered = (fn) => (typeof models.withMetering === 'function' ? models.withMetering(aiMetering, fn) : fn());
// The files attached to this message in the chat (plan L55 C2): text as data; images described first.
// The files attached to this message in the chat (plan L55 C2): text as data; images described first.
- const attachments = await inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', err && err.message); return []; });
+ const attachments = await metered(() => inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', (err && err.message) || err); return []; }));
// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.
// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.
// Every model call of the run — the loop's, the plan's, the check's — reads it, and no other run's does.
// Every model call of the run — the loop's, the plan's, the check's — reads it, and no other run's does.
let choice = await modelPrefs.choiceFor(s, row.requester).catch(() => null);
let choice = await modelPrefs.choiceFor(s, row.requester).catch(() => null);
@@ -616,7 +621,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -616,7 +621,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
// tick starts no turn by itself, so the next run is told which item changed.
// tick starts no turn by itself, so the next run is told which item changed.
const context = row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [];
const context = row.parent_id && row.fork_of_step == null && !resume ? await store.followContext(s, row) : [];
if (!resume && !fork) context.push(...await canvases.takeNoticesForTask(s, row));
if (!resume && !fork) context.push(...await canvases.takeNoticesForTask(s, row));
- out = await withChoice(models, choice, async () => loop.run({
+ out = await metered(() => withChoice(models, choice, async () => loop.run({
task: row.task, requester: fromChat ? null : row.requester, taskId: id, mode: row.mode, signal,
task: row.task, requester: fromChat ? null : row.requester, taskId: id, mode: row.mode, signal,
// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.
// A fork carries its parent's state in the checkpoint, not as a follow-up conversation.
context,
context,
@@ -636,7 +641,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -636,7 +641,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
latency: await latencyClass(id),
latency: await latencyClass(id),
// A product conversation turn (one-engine slice): the tools this run may call.
// A product conversation turn (one-engine slice): the tools this run may call.
allowedTools: unifiedCode && !turnAllowed ? ['codes_task'] : turnAllowed,
allowedTools: unifiedCode && !turnAllowed ? ['codes_task'] : turnAllowed,
- }));
+ })));
// The run parked (plan A2): the wait is open, so the task is not finished. It stays open
// The run parked (plan A2): the wait is open, so the task is not finished. It stays open
// ('running' is the closest state there is today), its wall clock waits with it, and its
// ('running' is the closest state there is today), its wall clock waits with it, and its
// budget clock stops because no model runs while it waits. The desk is closed with the run.
// budget clock stops because no model runs while it waits. The desk is closed with the run.
@@ -677,11 +682,11 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -677,11 +682,11 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
}
}
try {
try {
await setState(id, 'verifying', { agentForEvent: agentId });
await setState(id, 'verifying', { agentForEvent: agentId });
- const verify = async (current) => withChoice(models, choice, () => checkClaims({
+ const verify = async (current) => metered(() => withChoice(models, choice, () => checkClaims({
task: row.task, answer: current.text, evidence: current.evidence, models,
task: row.task, answer: current.text, evidence: current.evidence, models,
writerFamily: current.family || null, signal, now: () => now(),
writerFamily: current.family || null, signal, now: () => now(),
trace: current.state && current.state.ctx && current.state.ctx.trace,
trace: current.state && current.state.ctx && current.state.ctx.trace,
- }));
+ })));
let verdict = await verify(out);
let verdict = await verify(out);
await meter(verdict.cost_usd);
await meter(verdict.cost_usd);
const showCheck = (v) => emit(id, { kind: 'output', step: 'check', agent: agentId, text: `${v.ok ? 'Checked' : 'Check failed'}${v.by ? ` by ${v.by}${v.cross ? '' : ' (same family)'}` : ''}: ${v.reason || ''}`.slice(0, 2000), is_error: !v.ok });
const showCheck = (v) => emit(id, { kind: 'output', step: 'check', agent: agentId, text: `${v.ok ? 'Checked' : 'Check failed'}${v.by ? ` by ${v.by}${v.cross ? '' : ' (same family)'}` : ''}: ${v.reason || ''}`.slice(0, 2000), is_error: !v.ok });
@@ -707,9 +712,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -707,9 +712,9 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
let revised = null;
let revised = null;
if (!escalated) {
if (!escalated) {
escalated = true;
escalated = true;
- revised = await loop.escalate(out, verdict.unsupported);
+ revised = await metered(() => loop.escalate(out, verdict.unsupported));
}
}
- if (!revised || revised === out || !revised.text || revised.text === out.text) revised = await loop.revise(out, verdict.unsupported);
+ if (!revised || revised === out || !revised.text || revised.text === out.text) revised = await metered(() => loop.revise(out, verdict.unsupported));
if (!revised || revised === out || !revised.text || revised.text === out.text) break;
if (!revised || revised === out || !revised.text || revised.text === out.text) break;
out = revised;
out = revised;
verdict = await verify(out);
verdict = await verify(out);
@@ -724,7 +729,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -724,7 +729,7 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
for (const c of verdict.contested.slice(0, 2)) {
for (const c of verdict.contested.slice(0, 2)) {
if (budgetLeft() <= 0) break;
if (budgetLeft() <= 0) break;
try {
try {
- const j = await withChoice(models, choice, () => adjudicate({ claim: c, evidence: out.evidence, task: row.task, models, writerFamily: out.family || null, signal, now: () => now(), trace: out.state && out.state.ctx && out.state.ctx.trace }));
+ const j = await metered(() => withChoice(models, choice, () => adjudicate({ claim: c, evidence: out.evidence, task: row.task, models, writerFamily: out.family || null, signal, now: () => now(), trace: out.state && out.state.ctx && out.state.ctx.trace })));
if (j) { await meter(j.cost_usd); if (!j.failed) contested.push({ claim: j.claim, sides: j.sides, pick: j.pick, reason: j.reason, by: j.by }); }
if (j) { await meter(j.cost_usd); if (!j.failed) contested.push({ claim: j.claim, sides: j.sides, pick: j.pick, reason: j.reason, by: j.by }); }
} catch (err) {
} catch (err) {
if (signal && signal.aborted) throw err;
if (signal && signal.aborted) throw err;
@@ -740,14 +745,14 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -740,14 +745,14 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
if (Array.isArray(verdict.claims) && verdict.claims.length) {
if (Array.isArray(verdict.claims) && verdict.claims.length) {
const outward = !!(out.state && out.state.brain && out.state.brain.understanding && out.state.brain.understanding.outward);
const outward = !!(out.state && out.state.brain && out.state.brain.understanding && out.state.brain.understanding.outward);
try {
try {
- review = await withChoice(models, choice, () => confidence.review({
+ review = await metered(() => withChoice(models, choice, () => confidence.review({
claims: verdict.claims, evidence: out.evidence, task: row.task, mode: row.mode, models,
claims: verdict.claims, evidence: out.evidence, task: row.task, mode: row.mode, models,
outward, signal, trace: out.state && out.state.ctx && out.state.ctx.trace,
outward, signal, trace: out.state && out.state.ctx && out.state.ctx.trace,
budgetLeft, writerFamily: out.family || null, now: () => now(),
budgetLeft, writerFamily: out.family || null, now: () => now(),
search: async (query) => (registry
search: async (query) => (registry
? registry.run('search_web', { query }, { config, log, signal, requester: row.requester, now })
? registry.run('search_web', { query }, { config, log, signal, requester: row.requester, now })
: { text: 'no search is configured', isError: true }),
: { text: 'no search is configured', isError: true }),
- }));
+ })));
await meter(review.cost_usd);
await meter(review.cost_usd);
if (review.challenged) await emit(id, { kind: 'output', step: 'check', agent: agentId, text: `Challenged ${review.challenged} high-stakes claim${review.challenged === 1 ? '' : 's'} with a counter-evidence search; ${review.unsure} could not be confirmed.`.slice(0, 2000) });
if (review.challenged) await emit(id, { kind: 'output', step: 'check', agent: agentId, text: `Challenged ${review.challenged} high-stakes claim${review.challenged === 1 ? '' : 's'} with a counter-evidence search; ${review.unsure} could not be confirmed.`.slice(0, 2000) });
} catch (err) {
} catch (err) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,159 @@
@@ -0,0 +1,159 @@
+'use strict';
+/**
+ * The AI meter for Actor's own model calls (ai-metrics.md): every call through the tier layer carries the task id,
+ * its provider and model, TTFT, tokens per second and its tokens; a failure carries its class; the one process client
+ * posts the records to AI's /api/v1/usage/calls with Actor's service token.
+ */
+const assert = require('assert');
+const http = require('node:http');
+const configLib = require('../server/config');
+const { createModels } = require('../server/runtime/models');
+const { createAiMeter } = require('../server/runtime/ai-meter');
+const { RATES } = require('../server/agents/catalog');
+
+const results = [];
+const ok = async (name, fn) => { try { await fn(); console.log(` ✓ ${name}`); } catch (e) { results.push(e); console.log(` ✗ ${name}\n ${(e.stack || String(e)).split('\n').slice(0, 7).join('\n ')}`); } };
+
+const quiet = { log() {}, warn() {} };
+const json = (status, obj) => ({ ok: status >= 200 && status < 300, status, headers: { get: () => 'application/json' }, json: async () => obj, text: async () => JSON.stringify(obj) });
+
+const env = {
+ NODE_ENV: 'test',
+ ACTOR_DEEPSEEK_API_KEY: 'ds', ACTOR_DEEPSEEK_BASE_URL: 'https://deepseek.test',
+ ACTOR_MODEL_SMART: 'deepseek:deepseek-v4-flash',
+};
+
+function provider({ usage, answer = 'the answer', toolCalls = null, status = 200, error = null, onFetched = null } = {}) {
+ const seen = [];
+ const fetchImpl = async (url, opts = {}) => {
+ const u = String(url);
+ if (u.endsWith('/models')) return json(200, { data: [{ id: 'deepseek-v4-flash' }] });
+ seen.push({ url: u, body: opts.body ? JSON.parse(opts.body) : null });
+ if (onFetched) onFetched();
+ if (status !== 200) return json(status, error || { error: { message: 'no' } });
+ const message = { role: 'assistant', content: answer, ...(toolCalls ? { tool_calls: toolCalls } : {}) };
+ return json(200, { model: 'deepseek-v4-flash', choices: [{ message, finish_reason: 'stop' }], usage });
+ };
+ return { fetchImpl, seen };
+}
+
+const tickingMeter = () => ({ records: [], record(rec) { this.records.push(rec); return true; } });
+
+(async () => {
+ await ok('a provider call records TTFT, TPS, tokens, the provider and the task id', async () => {
+ let t = 10000;
+ const usage = { prompt_tokens: 1200, completion_tokens: 300, prompt_tokens_details: { cached_tokens: 1000 } };
+ const stub = provider({ usage, onFetched: () => { t += 800; } });
+ const meter = tickingMeter();
+ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter, now: () => t });
+ await models.warm();
+ const meta = { taskId: 'tsk_meter', task: 'actor.task', startedAt: 10000, seq: 0 };
+ const r = await models.withMetering(meta, () => models.chat({
+ tier: 'smart', messages: [{ role: 'user', content: 'x' }], onDelta: () => { t += 3000; }, taskEnd: true,
+ }));
+ assert.strictEqual(r.provider, 'deepseek');
+ assert.strictEqual(r.text, 'the answer');
+ assert.strictEqual(meter.records.length, 1);
+ const rec = meter.records[0];
+ assert.strictEqual(rec.system, 'actor');
+ assert.strictEqual(rec.task_id, 'tsk_meter');
+ assert.strictEqual(rec.turn_id, 'tsk_meter:1');
+ assert.strictEqual(rec.provider, 'deepseek');
+ assert.strictEqual(rec.model, 'deepseek-v4-flash');
+ assert.strictEqual(rec.operation, 'chat');
+ assert.strictEqual(rec.task, 'actor.task');
+ assert.strictEqual(rec.harness, 'openvibe');
+ assert.strictEqual(rec.place, 'provider');
+ assert.strictEqual(rec.streamed, true);
+ assert.strictEqual(rec.outcome, 'succeeded');
+ assert.deepStrictEqual([rec.input, rec.output, rec.cache_read], [200, 300, 1000]);
+ assert.deepStrictEqual([rec.ttft_ms, rec.latency_ms, rec.tps], [800, 3800, 100]);
+ assert.strictEqual(rec.task_ms, 3800, 'the task\'s wall time on its last step');
+ });
+
+ await ok('tool calls are counted, and a step with tools is not the task\'s last', async () => {
+ let t = 5000;
+ const calls = [{ id: 'c1', type: 'function', function: { name: 'lookup', arguments: '{}' } }, { id: 'c2', type: 'function', function: { name: 'search_web', arguments: '{}' } }];
+ const stub = provider({ usage: { prompt_tokens: 100, completion_tokens: 10 }, toolCalls: calls, onFetched: () => { t += 100; } });
+ const meter = tickingMeter();
+ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter, now: () => t });
+ await models.warm();
+ const meta = { taskId: 'tsk_tools', startedAt: 5000, seq: 0 };
+ const r = await models.withMetering(meta, () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }], taskEnd: true }));
+ assert.strictEqual(r.toolCalls.length, 2);
+ assert.strictEqual(meter.records[0].tool_calls, 2);
+ assert.strictEqual(meter.records[0].task_ms, undefined);
+ });
+
+ await ok('a failed call records outcome failed and its class; metering never hides the error', async () => {
+ const stub = provider({ status: 429, error: { error: { message: 'slow down' } } });
+ const meter = tickingMeter();
+ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl: stub.fetchImpl, log: quiet, meter });
+ await models.warm();
+ await assert.rejects(
+ () => models.withMetering({ taskId: 'tsk_fail', startedAt: 1, seq: 0 }, () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }] })),
+ (err) => err && err.code === 'actor.provider.rate_limited',
+ );
+ assert.strictEqual(meter.records.length, 1);
+ assert.strictEqual(meter.records[0].outcome, 'failed');
+ assert.strictEqual(meter.records[0].error_class, 'actor.provider.rate_limited');
+ assert.strictEqual(meter.records[0].task_id, 'tsk_fail');
+ assert.strictEqual(meter.records[0].turn_id, 'tsk_fail:1');
+ });
+
+ await ok('a call that goes through OpenVibe.AI is AI\'s to record, never Actor\'s (no double count)', async () => {
+ const meter = tickingMeter();
+ const fetchImpl = async () => { throw new Error('AI is down'); };
+ const models = createModels({ config: configLib.load(env), rates: RATES, fetchImpl, log: quiet, meter });
+ const meta = { taskId: 'tsk_via_ai', task: 'actor.task', startedAt: 0, seq: 0 };
+ await assert.rejects(models.withMetering(meta, () => models.withChoice({ provider: 'openai', model: 'gpt-5-mini' },
+ () => models.chat({ tier: 'smart', messages: [{ role: 'user', content: 'x' }] }))));
+ assert.strictEqual(meter.records.length, 0, 'AI publishes its own ai.call.recorded@1 for this call');
+ });
+
+ await ok('the meter client posts the records to a stub AI with Actor\'s service token', async () => {
+ const seen = [];
+ const tokenRequests = [];
+ const server = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => {
+ if (req.url === '/oauth/token') {
+ tokenRequests.push(body);
+ res.writeHead(200, { 'content-type': 'application/json' });
+ return res.end(JSON.stringify({ access_token: 'tok_actor', token_type: 'Bearer', expires_in: 300, scope: 'ai.usage.report' }));
+ }
+ seen.push({ url: req.url, auth: req.headers.authorization, body: JSON.parse(body) });
+ res.writeHead(202, { 'content-type': 'application/json' });
+ res.end('{}');
+ });
+ });
+ await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
+ const url = `http://127.0.0.1:${server.address().port}`;
+ let meter = null;
+ try {
+ const config = configLib.load({
+ NODE_ENV: 'production', OV_OAUTH_CLIENT_SECRET: 'secret', OV_NETWORK_INTERNAL_URL: url,
+ OV_AI_URL: url, ACTOR_AI_ROUTER: '0',
+ });
+ meter = createAiMeter({ config, log: quiet });
+ assert.ok(meter, 'a hosted Actor with a secret has a meter');
+ meter.record({ system: 'actor', turn_id: 'tsk_post:1', provider: 'deepseek', model: 'deepseek-v4-flash', outcome: 'succeeded', at: new Date().toISOString(), input: 1, output: 1 });
+ await meter.flush();
+ assert.strictEqual(seen.length, 1);
+ assert.strictEqual(seen[0].url, '/api/v1/usage/calls');
+ assert.strictEqual(seen[0].auth, 'Bearer tok_actor');
+ assert.strictEqual(seen[0].body.calls.length, 1);
+ assert.strictEqual(seen[0].body.calls[0].turn_id, 'tsk_post:1');
+ assert.strictEqual(new URLSearchParams(tokenRequests[0]).get('scope'), 'ai.usage.report');
+ assert.strictEqual(createAiMeter({ config: configLib.load({ ...env, ACTOR_DRILL: '1' }), log: quiet }), null, 'off in a drill');
+ assert.strictEqual(createAiMeter({ config: configLib.load(env), log: quiet }), null, 'off in tests');
+ } finally {
+ if (meter) await meter.close();
+ await new Promise((resolve) => server.close(resolve));
+ }
+ });
+
+ if (results.length) { console.log(`\n${results.length} failed`); process.exit(1); }
+ console.log('\nall passed');
+})();