openvibers/openvibe.bot Commits
Bot: meter AI on people's own machines (#433)
Changes
@@@ -108,6 +108,13 @@ OV_OAUTH_CLIENT_SECRET=#BOT_BILLING_INTERVAL_MS=2000#BOT_BILLING_TIMEOUT_MS=5000++# ── OpenVibe.AI: the models on people's machines (ai-metrics.md) ─++# Every `ai` command's answer is one ai.call-report@1 to AI's POST <url>/api/v1/usage/calls, so the local models on++# a person's machine are in the same meter as every other AI call. Bot mints its own Network service token (audience++# openvibe.ai, capability ai.usage.report) from OV_OAUTH_CLIENT_ID/SECRET and OV_NETWORK_INTERNAL_URL. Unset (or no++# client secret): metering is off and an ai command's answer is unchanged.++#OV_AI_INTERNAL_URL=http://127.0.0.1:4700++# ── Embeddable panel (plan T15 R9) ──────────────────────────# Pages that may frame a robot's read-only panel (CSP frame-ancestors), comma or space separated. Each entry is a# bare https origin (no path, query or wildcard); http://localhost:<port> is accepted outside production only.@@@ -245,7 +245,7 @@ Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras BotOpenVibe.Sites no longer generates `deploy/nginx/openvibe.bot.conf` (sites.json marks the vhost as owned by OpenVibe.Bot), and Bot serves nothing from a Sites checkout: the legal pages, `robots.txt`, `sitemap.xml`, `manifest.webmanifest` and the 404 page all come from the app. A Sites deploy leaves an installed vhost it no longer generates in place, so the current Sites vhost keeps serving openvibe.bot until Bot's first deploy installs this file over it; later Sites deploys never overwrite Bot's `/device` and `/control` routes.<!-- versions:start -->--- openvibe-contracts: v0.419.0--- openvibe-sdk: v0.161.0++- openvibe-contracts: v0.421.0++- openvibe-sdk: v0.164.0- openvibe-shared: v4.233.0<!-- versions:end -->@@@ -5,12 +5,12 @@"runtime": "node 22, express 4, ws; PostgreSQL 18 via openvibe-sdk/db (pg, or embedded PGlite in tests); Valkey optional; no build step","port": 4630,"domain": "openvibe.bot (www -> apex; /device and /control WebSockets; /install)",-- "contracts": "openvibe-contracts v0.419.0",-- "sdk": "openvibe-sdk v0.161.0",++ "contracts": "openvibe-contracts v0.421.0",++ "sdk": "openvibe-sdk v0.164.0","shared": "openvibe-shared v4.233.0","packages": [-- "openvibe-contracts v0.419.0",-- "openvibe-sdk v0.161.0",++ "openvibe-contracts v0.421.0",++ "openvibe-sdk v0.164.0","openvibe-shared v4.233.0"],"capabilities": [@@@ -13,8 +13,8 @@"dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.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","qrcode-generator": "^2.0.4",@@@ -1201,9 +1201,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.419.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",-- "integrity": "sha512-VxrTdeHchT8gvl2wWC9ASaId92en7ZKfJ8F/0hKzCAD0f2Y00rPrqgBWF9BNPoLGARoyQvsMbhBEhSkhxGNnjA==",++ "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",@@@ -1217,9 +1217,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.161.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.0",-- "integrity": "sha512-qReF7K9ct7LCLDJKuguwUQ30nc0kmrp+jlRZcFm4adpEvKL8r5G6i6Osbs0ROAMLmXKwnbI4GiZAgAyQ7oN7Ig==",++ "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"@@@ -25,8 +25,8 @@"dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.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","qrcode-generator": "^2.0.4",nnew file mode 100644@@@ -0,0 +1,48 @@++'use strict';++/**++ * Bot's meter for AI that runs on people's own machines (ai-metrics.md, platform L40/L41). An `ai` command's answer++ * is one call that did not go through OpenVibe.AI: OpenVibe Link's ai plugin relayed it to the machine's own Ollama,++ * LM Studio or llama.cpp server. Bot sends it to AI's POST /api/v1/usage/calls as ai.call-report@1 (through++ * openvibe-sdk/ai-meter), which prices it and publishes one ai.call.recorded@1.++ *++ * createAiMeter({ config, fetchImpl, log }) → { enabled, startCall, record, flush, close, stats }++ *++ * Bot mints its own Network service token (client `bot`, audience openvibe.ai, capability ai.usage.report) from its++ * OAuth client, like the OpenRestream and Billing clients. OV_AI_INTERNAL_URL unset, or no client secret to mint a++ * token with, keeps metering off: startCall returns null, record does nothing and the command path is unchanged.++ */++const { serviceAuth } = require('openvibe-contracts');++const sdkMeter = require('openvibe-sdk/ai-meter');++++const AI_AUDIENCE = 'openvibe.ai';++const AI_SCOPE = 'ai.usage.report';++++function createAiMeter({ config, fetchImpl = globalThis.fetch, log = console } = {}) {++ const aiUrl = config && config.ai && config.ai.internalUrl;++ const { clientId, clientSecret } = (config && config.oauth) || {};++ const internalUrl = config && config.network && config.network.internalUrl;++ const tokenClient = aiUrl && clientSecret && internalUrl++ ? serviceAuth.createTokenClient({ tokenUrl: `${internalUrl}/oauth/token`, clientId, clientSecret, audience: AI_AUDIENCE, scope: AI_SCOPE, fetchImpl })++ : null;++ if (!aiUrl || !tokenClient) {++ return {++ enabled: false,++ startCall: () => null,++ record: () => false,++ flush: async () => {},++ close: async () => {},++ stats: () => ({ sent: 0, dropped: 0, rejected: 0, retried: 0, queued: 0 }),++ };++ }++ const client = sdkMeter.createMeterClient({ aiUrl, tokenClient, fetch: fetchImpl, log });++ return {++ enabled: true,++ startCall: sdkMeter.startCall,++ record: client.record,++ flush: client.flush,++ close: client.close,++ stats: client.stats,++ };++}++++module.exports = { createAiMeter, AI_AUDIENCE, AI_SCOPE };@@@ -29,6 +29,7 @@ const { createOfferPublisher } = require('./jobs/offers');const { createOpenRe } = require('./openre/client');const { createBotOutbox } = require('./events/outbox');const { createUsageRelay } = require('./jobs/metering');++const { createAiMeter } = require('./ai-meter');const { createDomain } = require('./domain');const accountDataLib = require('./account-data');const { createNetworkSender } = require('openvibe-sdk/account-data');@@@ -70,7 +71,9 @@ function createApp(opts = {}) {const networkSend = opts.networkSend || (config.oauth.clientSecret? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }): null);-- const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now, networkSend });++ // The models on people's machines, into OpenVibe.AI's meter (server/ai-meter.js): off without OV_AI_INTERNAL_URL.++ const meter = opts.meter !== undefined ? opts.meter : createAiMeter({ config, fetchImpl, log });++ const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now, networkSend, meter });const nodes = opts.nodes || createNodePrincipals(config, { fetchImpl });const identity = opts.identity || createIdentity(config, { fetchImpl });const openre = opts.openre !== undefined ? opts.openre : createOpenRe(config, { fetchImpl });@@@ -274,7 +277,7 @@ function createApp(opts = {}) {return res.status(500).type('text/plain').send('Something went wrong\n');});-- Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, hub, jobs, offers, sim, onvif, userAuth, metrics, submissions, routines, webhooks, claims, setupKeys });++ Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, meter, hub, jobs, offers, sim, onvif, userAuth, metrics, submissions, routines, webhooks, claims, setupKeys });return app;}@@@ -267,6 +267,11 @@ function loadConfig(env = process.env) {intervalMs: Math.max(100, int(env.BOT_BILLING_INTERVAL_MS, 2000)),timeoutMs: Math.max(100, int(env.BOT_BILLING_TIMEOUT_MS, 5000)),},++ // OpenVibe.AI's meter for the models that run on people's own machines (server/ai-meter.js): each `ai`++ // command's answer goes to AI's POST /api/v1/usage/calls as ai.call-report@1. Bot mints its own Network++ // service token for audience openvibe.ai (ai.usage.report) from its OAuth client; OV_AI_INTERNAL_URL++ // unset, or no client secret, keeps metering off (the command path is unchanged).++ ai: { internalUrl: trim(env.OV_AI_INTERNAL_URL || '') },// The one-line installer the owner copies next to the pairing code. scriptUrl is what the command// prints; sourceUrl is where GET /install sends the client (302). Bot keeps no copy of the script:// it redirects to OpenVibe Link's canonical one, and only this config (never a query parameter)@@@ -34,7 +34,7 @@ async function main() {const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });const app = createApp({ config, db, valkey, registry });-- const { domain, keys, outbox, usage, hub, offers, sim, onvif, routines, webhooks, claims, setupKeys } = app.locals;++ const { domain, keys, outbox, usage, meter, hub, offers, sim, onvif, routines, webhooks, claims, setupKeys } = app.locals;keys.start().catch(() => { /* logged by the JWKS client */ });const timers = [];@@@ -59,7 +59,7 @@ async function main() {const socketActivated = process.env.LISTEN_FDS === '1' && Number(process.env.LISTEN_PID) === process.pid;const server = app.listen(...(socketActivated ? [{ fd: 3 }] : [config.port, config.host]), () => {console.log(`[Bot] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: per process'})`);-- console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; usage relay ${usage.enabled ? `→ ${config.billing.url}` : 'off (readings wait)'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);++ console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; usage relay ${usage.enabled ? `→ ${config.billing.url}` : 'off (readings wait)'}; ai meter ${meter.enabled ? `→ ${config.ai.internalUrl}` : 'off'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);});server.keepAliveTimeout = 65_000;// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status@@@ -105,6 +105,7 @@ async function main() {() => hub.close().catch(() => {}),() => outbox.stop(),() => usage.stop(),++ () => meter.close().catch(() => {}),() => { if (subscriptions) subscriptions.stop(); },],close: [@@@ -109,6 +109,21 @@ function computerSummary(kind, value) {return { max_width: value.max_width || null, ...where };}++/**++ * What metering keeps of an `ai` command before its answer: the requested model (without the router's community/++ * prefix), whether this was a community request, and the caller's task when it named one. The server that runs it++ * (Ollama, LM Studio, llama.cpp) is only known from the answer, so it is filled in when the ack arrives.++ */++function aiMeterMeta(value) {++ const wanted = typeof value.model === 'string' ? value.model : '';++ const community = wanted.startsWith('community/');++ return {++ model: (community ? wanted.slice('community/'.length) : wanted) || 'unknown',++ community,++ ...(typeof value.task === 'string' && value.task ? { task: value.task.slice(0, 200) } : {}),++ };++}++function parseCookies(header) {const out = {};for (const part of String(header || '').split(';')) {@@@ -147,7 +162,7 @@ function simView(sim) {/** How long a restart waits for every client's close in reply to its 1012 before it drops the rest (11-12). */const CLOSE_HANDSHAKE_MS = 600;--function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now(), networkSend = null }) {++function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now(), networkSend = null, meter = null }) {let domain = null;const tickets = createControlTickets({ now }); // the embedded panel's Drive (server/control-tickets.js)const deviceConns = new Map(); // device_id → conn@@@ -715,6 +730,42 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datfor (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);}++ /**++ * One ai command's answer is one call that did not go through OpenVibe.AI: a local model on the person's machine.++ * It goes to AI's meter after the answer is in hand (never awaited, never fatal), with the plugin's usage, model++ * and server. A nack is the call failing with its fault as the error class.++ */++ function meterAiCommand(p, result, msg) {++ if (!meter || !meter.enabled || !p.ai) return;++ // OpenVibe.AI sends most ai commands (a person's own machine as their model, the community offers): those are++ // AI's runs, and AI records each one itself. Only a call nobody else records (a person's own, an agent's) is Bot's.++ const caller = p.conn && p.conn.principal;++ if (caller && caller.kind === 'service' && caller.sub === 'svc:ai') return;++ try {++ const data = result === 'ack' && msg.data && typeof msg.data === 'object' && !Array.isArray(msg.data) ? msg.data : {};++ const usage = data.usage && typeof data.usage === 'object' ? data.usage : null;++ const server = typeof data.server === 'string' && data.server.trim() ? data.server.trim().slice(0, 120) : null;++ // Ollama's native usage already names its server; an ack that carries neither is a local model with none.++ const provider = server || (usage && usage.eval_count != null ? 'ollama' : 'local');++ const started = p.at;++ let first = true;++ const call = meter.startCall({++ system: 'node', run_id: p.opId, provider, model: String(data.model || p.ai.model || 'unknown').slice(0, 200),++ operation: 'chat', place: p.ai.community ? 'community' : 'machine',++ ...(p.ai.task ? { task: p.ai.task } : {}),++ // startCall's clock begins when the command left Bot, so latency is the machine's whole answer.++ now: () => { if (first) { first = false; return started; } return now(); },++ });++ meter.record(call.end({++ usage,++ outcome: result === 'ack' ? 'succeeded' : 'failed',++ ...(result === 'nack' && msg.fault_code ? { error_class: String(msg.fault_code) } : {}),++ }));++ } catch (e) {++ log.warn(`[Bot] ai meter: ${e.message}`);++ }++ }++async function onAck(conn, msg) {const id = String(msg.id || '');if (id.startsWith('job_')) return jobs.frames.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's@@@ -735,6 +786,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => DatcacheResult(p.key, { result, reason: msg.fault_code || null, ...(p.kind === 'headlight' && data ? { data } : {}) });await domain.audit.record({ robotId: p.robotId, deviceId: p.deviceId, subject: p.subject, operatorKind: p.operatorKind, role: p.role, kind: p.kind, value: p.value, result, reason: msg.fault_code || null, latencyMs: latency });const message = (p.computer || p.kind === 'headlight') && result === 'nack' && typeof msg.message === 'string' ? msg.message.replace(/[\u0000-\u001f\u007f]/g, ' ').slice(0, 300) : null;++ meterAiCommand(p, result, msg);sendFrame(p.conn, 'command_result', { id: p.opId, result, reason: message || msg.fault_code || null, latency_ms: latency, ...(data ? { data } : {}) });}@@@ -1094,7 +1146,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datconst deadlineMs = now() + windowMs;const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind, value, deadline_ms: deadlineMs, operator: { subject, role: 'owner' } };if (!send(device.ws, frame)) return refuse('bot.device_offline', 'the machine is offline', summary);-- const p = { id: deviceCmdId, opId: id, key, deviceId, conn, robotId: deviceId, subject, operatorKind: conn.principal.kind, role: 'owner', kind, value: summary, at: now(), timer: null, computer: true };++ const ai = kind === 'ai' ? aiMeterMeta(value) : null;++ const p = { id: deviceCmdId, opId: id, key, deviceId, conn, robotId: deviceId, subject, operatorKind: conn.principal.kind, role: 'owner', kind, value: summary, at: now(), timer: null, computer: true, ...(ai ? { ai } : {}) };pendingCmds.set(deviceCmdId, p);inflight.set(key, deviceCmdId);p.timer = setTimeout(() => expireCommand(deviceCmdId), windowMs + 1500);nnew file mode 100644@@@ -0,0 +1,160 @@++'use strict';++// AI on people's own machines is metered (ai-metrics.md): an `ai` command's ack is one ai.call-report@1 record for++// OpenVibe.AI, with the plugin's usage (Ollama's counts and decode time through openvibe-sdk/ai-meter's normalize),++// model server, model and the whole answer's latency; a nack is a failed call with its fault as the error class; a++// Bot without OV_AI_INTERNAL_URL sends nothing at all.++const assert = require('assert');++const http = require('http');++const { boot, check, done } = require('./helpers/app');++++/** OpenVibe.AI's meter stand-in: records every POST /api/v1/usage/calls and answers 202. */++async function startAi() {++ const requests = [];++ const server = http.createServer((req, res) => {++ const chunks = [];++ req.on('data', (d) => chunks.push(d));++ req.on('end', () => {++ let body = null;++ try { body = JSON.parse(Buffer.concat(chunks).toString('utf8')); } catch { /* not json */ }++ requests.push({ url: req.url, method: req.method, auth: req.headers.authorization, body });++ res.writeHead(202, { 'content-type': 'application/json' });++ res.end('{}');++ });++ });++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ return { url: `http://127.0.0.1:${server.address().port}`, requests, close: () => new Promise((r) => { server.closeAllConnections(); server.close(r); }) };++}++++(async () => {++ const ai = await startAi();++ let t = null;++ let seq = 0;++ const frame = (type, fields) => ({ v: 1, seq: ++seq, ts: Date.now(), type, ...fields });++ const poll = async (fn, ms = 8000) => { const end = Date.now() + ms; while (Date.now() < end) { const v = await fn(); if (v) return v; await t.wait(30); } return null; };++++ /**++ * A Network-paired machine whose OpenVibe Link ai plugin answers `ai` commands with `answer(m, c)`; every other++ * kind is refused. Returns its device id, and `setAnswer` swaps the plugin's reply between checks.++ */++ async function machine(owner) {++ const principal = t.network.addNode({ owner: owner.subject, ref: `worker:${owner.subject}`, name: 'ai-pc' });++ const c = await t.ws('/device', { headers: { Authorization: `Bearer ${t.network.signNode(principal)}` } });++ const deviceId = (await c.waitFor((m) => m.type === 'hello')).device_id;++ const config = await c.waitFor((m) => m.type === 'config');++ const box = { answer: null };++ c.send(frame('status', { capabilities: { ai: { ai: { models: ['llama3.1:8b'], ready: true, max_tokens: 2048 } } }, drivers: [], faults: [] }));++ c.send(frame('telemetry', { sensors: {} }));++ c.ws.on('message', (raw) => {++ let m; try { m = JSON.parse(raw.toString()); } catch { return; }++ if (m.type !== 'command') return;++ if (m.kind === 'ai' && box.answer) box.answer(m, c, frame);++ else c.send(frame('nack', { id: m.id, fault_code: 'unsupported', message: 'not this plugin' }));++ });++ const beat = setInterval(() => { try { c.send(frame('heartbeat', { t: Date.now() })); } catch { /* closed */ } }, Math.max(100, Math.floor(config.heartbeat_ms / 2)));++ c.ws.on('close', () => clearInterval(beat));++ await poll(() => t.hub.computerOf(deviceId) !== null);++ return { c, deviceId, setAnswer: (fn) => { box.answer = fn; } };++ }++++ try {++ t = await boot({ env: { OV_AI_INTERNAL_URL: ai.url } });++ assert.strictEqual(t.app.locals.meter.enabled, true, 'the meter is on with OV_AI_INTERNAL_URL');++ const alex = t.network.newUser('alex');++ const gpu = await machine(alex);++ const command = (id, value) => t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, { user: alex, body: { id, kind: 'ai', value } });++ const ask = (id) => command(id, { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'What is a mutex?' }] });++++ await check('an ack with Ollama usage records one local call with tokens and tokens per second', async () => {++ gpu.setAnswer((m, c, send) => c.send(send('ack', {++ id: m.id,++ data: {++ model: 'llama3.1:8b', server: 'ollama', ms: 9600, finish_reason: 'stop',++ message: { role: 'assistant', content: 'A lock one thread holds at a time.' },++ usage: { prompt_eval_count: 300, eval_count: 120, eval_duration: 9.6e9 },++ },++ })));++ const r = await ask('cmd_ollama_1');++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(r.json.result, 'ack');++ await t.app.locals.meter.flush();++ const sent = ai.requests.at(-1);++ assert.strictEqual(sent.url, '/api/v1/usage/calls');++ assert.match(sent.auth, /^Bearer /, 'a minted service token');++ const token = t.network.tokenCalls.find((x) => x.audience === 'openvibe.ai');++ assert.deepStrictEqual([token && token.scope, token && token.client_id], ['ai.usage.report', 'bot'], 'Bot mints audience openvibe.ai for ai.usage.report');++ const call = sent.body.calls[0];++ assert.deepStrictEqual([call.system, call.run_id, call.operation, call.place, call.outcome],++ ['node', 'cmd_ollama_1', 'chat', 'machine', 'succeeded']);++ assert.deepStrictEqual([call.provider, call.model], ['ollama', 'llama3.1:8b']);++ assert.deepStrictEqual([call.input, call.output, call.reasoning, call.cache_read], [300, 120, 0, 0]);++ assert.strictEqual(call.tps, 12.5, '120 tokens over Ollama’s 9.6 s decode');++ assert.ok(call.latency_ms >= 0 && Number.isFinite(call.ttft_ms) === false, 'the whole answer’s latency, no streamed first token');++ assert.strictEqual(call.streamed, false);++ assert.ok(call.at && !Number.isNaN(Date.parse(call.at)), 'an ISO time');++ });++++ await check('a community/ model records the bare model and place community', async () => {++ gpu.setAnswer((m, c, send) => c.send(send('ack', {++ id: m.id, data: { model: 'llama3.1:8b', server: 'lmstudio', message: { role: 'assistant', content: 'hi' }, usage: { prompt_tokens: 10, completion_tokens: 20 } },++ })));++ const r = await command('cmd_community_1', { model: 'community/llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] });++ assert.strictEqual(r.json.result, 'ack', r.text);++ await t.app.locals.meter.flush();++ const call = ai.requests.at(-1).body.calls[0];++ assert.deepStrictEqual([call.model, call.provider, call.place, call.input, call.output], ['llama3.1:8b', 'lmstudio', 'community', 10, 20]);++ });++++ await check('a nack records a failed call with its fault as the error class', async () => {++ gpu.setAnswer((m, c, send) => c.send(send('nack', { id: m.id, fault_code: 'not_ready', message: 'the model is answering another request on this machine: try again in a moment' })));++ const r = await ask('cmd_nack_1');++ assert.strictEqual(r.json.result, 'nack');++ assert.match(r.json.reason, /another request/);++ await t.app.locals.meter.flush();++ const call = ai.requests.at(-1).body.calls[0];++ assert.deepStrictEqual([call.run_id, call.outcome, call.error_class, call.place], ['cmd_nack_1', 'failed', 'not_ready', 'machine']);++ assert.deepStrictEqual([call.input, call.output], [0, 0], 'no usage: zero tokens, never null');++ });++++ await check('a command OpenVibe.AI sends is AI\'s run: AI records it, Bot never reports it again', async () => {++ const on = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/ai-models`, { user: alex, body: { on: true } });++ assert.strictEqual(on.status, 200, on.text);++ gpu.setAnswer((m, c, send) => c.send(send('ack', {++ id: m.id, data: { model: 'llama3.1:8b', server: 'ollama', message: { role: 'assistant', content: 'hi' }, usage: { prompt_eval_count: 3, eval_count: 2, eval_duration: 1e9 } },++ })));++ const before = ai.requests.length;++ const asAi = { token: t.network.signService({ sub: 'svc:ai', cap: ['bot.machine.ai'] }), headers: { 'X-OV-Subject': alex.subject } };++ const r = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, { ...asAi, body: { id: 'cmd_from_ai_1', kind: 'ai', value: { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] } } });++ assert.strictEqual(r.json.result, 'ack', r.text);++ await t.app.locals.meter.flush();++ assert.strictEqual(ai.requests.length, before, 'no report: AI publishes its own ai.call.recorded@1 for this run');++ });++++ gpu.c.close();++ await t.close();++ t = null;++++ await check('without OV_AI_INTERNAL_URL the meter is off and an ai answer sends nothing', async () => {++ const before = ai.requests.length;++ t = await boot();++ assert.strictEqual(t.app.locals.meter.enabled, false, 'no AI URL, no meter');++ const alex = t.network.newUser('alex');++ const gpu = await machine(alex);++ gpu.setAnswer((m, c, send) => c.send(send('ack', {++ id: m.id, data: { model: 'llama3.1:8b', server: 'ollama', message: { role: 'assistant', content: 'hi' }, usage: { prompt_eval_count: 3, eval_count: 2, eval_duration: 1e9 } },++ })));++ const r = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, {++ user: alex, body: { id: 'cmd_off_1', kind: 'ai', value: { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] } },++ });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(r.json.result, 'ack');++ await t.wait(150);++ assert.strictEqual(ai.requests.length, before, 'nothing reached AI');++ assert.strictEqual(t.app.locals.meter.stats().queued, 0);++ gpu.c.close();++ });++ } finally {++ if (t) await t.close().catch(() => {});++ await ai.close();++ done();++ }++})().catch((e) => { console.error(e); process.exit(1); });
@@ -108,6 +108,13 @@ OV_OAUTH_CLIENT_SECRET=
@@ -108,6 +108,13 @@ OV_OAUTH_CLIENT_SECRET=
#BOT_BILLING_INTERVAL_MS=2000
#BOT_BILLING_INTERVAL_MS=2000
#BOT_BILLING_TIMEOUT_MS=5000
#BOT_BILLING_TIMEOUT_MS=5000
+# ── OpenVibe.AI: the models on people's machines (ai-metrics.md) ─
+# Every `ai` command's answer is one ai.call-report@1 to AI's POST <url>/api/v1/usage/calls, so the local models on
+# a person's machine are in the same meter as every other AI call. Bot mints its own Network service token (audience
+# openvibe.ai, capability ai.usage.report) from OV_OAUTH_CLIENT_ID/SECRET and OV_NETWORK_INTERNAL_URL. Unset (or no
+# client secret): metering is off and an ai command's answer is unchanged.
+#OV_AI_INTERNAL_URL=http://127.0.0.1:4700
+
# ── Embeddable panel (plan T15 R9) ──────────────────────────
# ── Embeddable panel (plan T15 R9) ──────────────────────────
# Pages that may frame a robot's read-only panel (CSP frame-ancestors), comma or space separated. Each entry is a
# Pages that may frame a robot's read-only panel (CSP frame-ancestors), comma or space separated. Each entry is a
# bare https origin (no path, query or wildcard); http://localhost:<port> is accepted outside production only.
# bare https origin (no path, query or wildcard); http://localhost:<port> is accepted outside production only.
@@ -245,7 +245,7 @@ Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras Bot
@@ -245,7 +245,7 @@ Set `BOT_ONVIF_CAMERAS` (JSON, unset by default) to the server-side cameras Bot
OpenVibe.Sites no longer generates `deploy/nginx/openvibe.bot.conf` (sites.json marks the vhost as owned by OpenVibe.Bot), and Bot serves nothing from a Sites checkout: the legal pages, `robots.txt`, `sitemap.xml`, `manifest.webmanifest` and the 404 page all come from the app. A Sites deploy leaves an installed vhost it no longer generates in place, so the current Sites vhost keeps serving openvibe.bot until Bot's first deploy installs this file over it; later Sites deploys never overwrite Bot's `/device` and `/control` routes.
OpenVibe.Sites no longer generates `deploy/nginx/openvibe.bot.conf` (sites.json marks the vhost as owned by OpenVibe.Bot), and Bot serves nothing from a Sites checkout: the legal pages, `robots.txt`, `sitemap.xml`, `manifest.webmanifest` and the 404 page all come from the app. A Sites deploy leaves an installed vhost it no longer generates in place, so the current Sites vhost keeps serving openvibe.bot until Bot's first deploy installs this file over it; later Sites deploys never overwrite Bot's `/device` and `/control` routes.
<!-- versions:start -->
<!-- versions:start -->
-- openvibe-contracts: v0.419.0
-- openvibe-sdk: v0.161.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 -->
@@ -5,12 +5,12 @@
@@ -5,12 +5,12 @@
"runtime": "node 22, express 4, ws; PostgreSQL 18 via openvibe-sdk/db (pg, or embedded PGlite in tests); Valkey optional; no build step",
"runtime": "node 22, express 4, ws; PostgreSQL 18 via openvibe-sdk/db (pg, or embedded PGlite in tests); Valkey optional; no build step",
"port": 4630,
"port": 4630,
"domain": "openvibe.bot (www -> apex; /device and /control WebSockets; /install)",
"domain": "openvibe.bot (www -> apex; /device and /control WebSockets; /install)",
- "contracts": "openvibe-contracts v0.419.0",
- "sdk": "openvibe-sdk v0.161.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.0",
- "openvibe-sdk v0.161.0",
+ "openvibe-contracts v0.421.0",
+ "openvibe-sdk v0.164.0",
"openvibe-shared v4.233.0"
"openvibe-shared v4.233.0"
],
],
"capabilities": [
"capabilities": [
@@ -13,8 +13,8 @@
@@ -13,8 +13,8 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.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",
"qrcode-generator": "^2.0.4",
"qrcode-generator": "^2.0.4",
@@ -1201,9 +1201,9 @@
@@ -1201,9 +1201,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.419.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",
- "integrity": "sha512-VxrTdeHchT8gvl2wWC9ASaId92en7ZKfJ8F/0hKzCAD0f2Y00rPrqgBWF9BNPoLGARoyQvsMbhBEhSkhxGNnjA==",
+ "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",
@@ -1217,9 +1217,9 @@
@@ -1217,9 +1217,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.161.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.0",
- "integrity": "sha512-qReF7K9ct7LCLDJKuguwUQ30nc0kmrp+jlRZcFm4adpEvKL8r5G6i6Osbs0ROAMLmXKwnbI4GiZAgAyQ7oN7Ig==",
+ "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"
@@ -25,8 +25,8 @@
@@ -25,8 +25,8 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.419.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.161.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",
"qrcode-generator": "^2.0.4",
"qrcode-generator": "^2.0.4",
new file mode 100644
new file mode 100644
@@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
+'use strict';
+/**
+ * Bot's meter for AI that runs on people's own machines (ai-metrics.md, platform L40/L41). An `ai` command's answer
+ * is one call that did not go through OpenVibe.AI: OpenVibe Link's ai plugin relayed it to the machine's own Ollama,
+ * LM Studio or llama.cpp server. Bot sends it to AI's POST /api/v1/usage/calls as ai.call-report@1 (through
+ * openvibe-sdk/ai-meter), which prices it and publishes one ai.call.recorded@1.
+ *
+ * createAiMeter({ config, fetchImpl, log }) → { enabled, startCall, record, flush, close, stats }
+ *
+ * Bot mints its own Network service token (client `bot`, audience openvibe.ai, capability ai.usage.report) from its
+ * OAuth client, like the OpenRestream and Billing clients. OV_AI_INTERNAL_URL unset, or no client secret to mint a
+ * token with, keeps metering off: startCall returns null, record does nothing and the command path is unchanged.
+ */
+const { serviceAuth } = require('openvibe-contracts');
+const sdkMeter = require('openvibe-sdk/ai-meter');
+
+const AI_AUDIENCE = 'openvibe.ai';
+const AI_SCOPE = 'ai.usage.report';
+
+function createAiMeter({ config, fetchImpl = globalThis.fetch, log = console } = {}) {
+ const aiUrl = config && config.ai && config.ai.internalUrl;
+ const { clientId, clientSecret } = (config && config.oauth) || {};
+ const internalUrl = config && config.network && config.network.internalUrl;
+ const tokenClient = aiUrl && clientSecret && internalUrl
+ ? serviceAuth.createTokenClient({ tokenUrl: `${internalUrl}/oauth/token`, clientId, clientSecret, audience: AI_AUDIENCE, scope: AI_SCOPE, fetchImpl })
+ : null;
+ if (!aiUrl || !tokenClient) {
+ return {
+ enabled: false,
+ startCall: () => null,
+ record: () => false,
+ flush: async () => {},
+ close: async () => {},
+ stats: () => ({ sent: 0, dropped: 0, rejected: 0, retried: 0, queued: 0 }),
+ };
+ }
+ const client = sdkMeter.createMeterClient({ aiUrl, tokenClient, fetch: fetchImpl, log });
+ return {
+ enabled: true,
+ startCall: sdkMeter.startCall,
+ record: client.record,
+ flush: client.flush,
+ close: client.close,
+ stats: client.stats,
+ };
+}
+
+module.exports = { createAiMeter, AI_AUDIENCE, AI_SCOPE };
@@ -29,6 +29,7 @@ const { createOfferPublisher } = require('./jobs/offers');
@@ -29,6 +29,7 @@ const { createOfferPublisher } = require('./jobs/offers');
const { createOpenRe } = require('./openre/client');
const { createOpenRe } = require('./openre/client');
const { createBotOutbox } = require('./events/outbox');
const { createBotOutbox } = require('./events/outbox');
const { createUsageRelay } = require('./jobs/metering');
const { createUsageRelay } = require('./jobs/metering');
+const { createAiMeter } = require('./ai-meter');
const { createDomain } = require('./domain');
const { createDomain } = require('./domain');
const accountDataLib = require('./account-data');
const accountDataLib = require('./account-data');
const { createNetworkSender } = require('openvibe-sdk/account-data');
const { createNetworkSender } = require('openvibe-sdk/account-data');
@@ -70,7 +71,9 @@ function createApp(opts = {}) {
@@ -70,7 +71,9 @@ function createApp(opts = {}) {
const networkSend = opts.networkSend || (config.oauth.clientSecret
const networkSend = opts.networkSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
? createNetworkSender({ networkInternalUrl: config.network.internalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: null);
: null);
- const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now, networkSend });
+ // The models on people's machines, into OpenVibe.AI's meter (server/ai-meter.js): off without OV_AI_INTERNAL_URL.
+ const meter = opts.meter !== undefined ? opts.meter : createAiMeter({ config, fetchImpl, log });
+ const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now, networkSend, meter });
const nodes = opts.nodes || createNodePrincipals(config, { fetchImpl });
const nodes = opts.nodes || createNodePrincipals(config, { fetchImpl });
const identity = opts.identity || createIdentity(config, { fetchImpl });
const identity = opts.identity || createIdentity(config, { fetchImpl });
const openre = opts.openre !== undefined ? opts.openre : createOpenRe(config, { fetchImpl });
const openre = opts.openre !== undefined ? opts.openre : createOpenRe(config, { fetchImpl });
@@ -274,7 +277,7 @@ function createApp(opts = {}) {
@@ -274,7 +277,7 @@ function createApp(opts = {}) {
return res.status(500).type('text/plain').send('Something went wrong\n');
return res.status(500).type('text/plain').send('Something went wrong\n');
});
});
- Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, hub, jobs, offers, sim, onvif, userAuth, metrics, submissions, routines, webhooks, claims, setupKeys });
+ Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, meter, hub, jobs, offers, sim, onvif, userAuth, metrics, submissions, routines, webhooks, claims, setupKeys });
return app;
return app;
}
}
@@ -267,6 +267,11 @@ function loadConfig(env = process.env) {
@@ -267,6 +267,11 @@ function loadConfig(env = process.env) {
intervalMs: Math.max(100, int(env.BOT_BILLING_INTERVAL_MS, 2000)),
intervalMs: Math.max(100, int(env.BOT_BILLING_INTERVAL_MS, 2000)),
timeoutMs: Math.max(100, int(env.BOT_BILLING_TIMEOUT_MS, 5000)),
timeoutMs: Math.max(100, int(env.BOT_BILLING_TIMEOUT_MS, 5000)),
},
},
+ // OpenVibe.AI's meter for the models that run on people's own machines (server/ai-meter.js): each `ai`
+ // command's answer goes to AI's POST /api/v1/usage/calls as ai.call-report@1. Bot mints its own Network
+ // service token for audience openvibe.ai (ai.usage.report) from its OAuth client; OV_AI_INTERNAL_URL
+ // unset, or no client secret, keeps metering off (the command path is unchanged).
+ ai: { internalUrl: trim(env.OV_AI_INTERNAL_URL || '') },
// The one-line installer the owner copies next to the pairing code. scriptUrl is what the command
// The one-line installer the owner copies next to the pairing code. scriptUrl is what the command
// prints; sourceUrl is where GET /install sends the client (302). Bot keeps no copy of the script:
// prints; sourceUrl is where GET /install sends the client (302). Bot keeps no copy of the script:
// it redirects to OpenVibe Link's canonical one, and only this config (never a query parameter)
// it redirects to OpenVibe Link's canonical one, and only this config (never a query parameter)
@@ -34,7 +34,7 @@ async function main() {
@@ -34,7 +34,7 @@ async function main() {
const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });
const valkey = createValkey({ url: config.valkey.url, prefix: config.valkey.prefix });
const app = createApp({ config, db, valkey, registry });
const app = createApp({ config, db, valkey, registry });
- const { domain, keys, outbox, usage, hub, offers, sim, onvif, routines, webhooks, claims, setupKeys } = app.locals;
+ const { domain, keys, outbox, usage, meter, hub, offers, sim, onvif, routines, webhooks, claims, setupKeys } = app.locals;
keys.start().catch(() => { /* logged by the JWKS client */ });
keys.start().catch(() => { /* logged by the JWKS client */ });
const timers = [];
const timers = [];
@@ -59,7 +59,7 @@ async function main() {
@@ -59,7 +59,7 @@ async function main() {
const socketActivated = process.env.LISTEN_FDS === '1' && Number(process.env.LISTEN_PID) === process.pid;
const socketActivated = process.env.LISTEN_FDS === '1' && Number(process.env.LISTEN_PID) === process.pid;
const server = app.listen(...(socketActivated ? [{ fd: 3 }] : [config.port, config.host]), () => {
const server = app.listen(...(socketActivated ? [{ fd: 3 }] : [config.port, config.host]), () => {
console.log(`[Bot] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: per process'})`);
console.log(`[Bot] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (store ${db.store}; valkey ${valkey ? 'on' : 'off: per process'})`);
- console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; usage relay ${usage.enabled ? `→ ${config.billing.url}` : 'off (readings wait)'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);
+ console.log(`[Bot] profiles from server/profiles; events relay ${outbox.enabled ? `→ ${config.events.url}` : 'off (outbox accumulates)'}; usage relay ${usage.enabled ? `→ ${config.billing.url}` : 'off (readings wait)'}; ai meter ${meter.enabled ? `→ ${config.ai.internalUrl}` : 'off'}; heartbeat ${config.device.heartbeatMs} ms, offline after ${config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs} ms`);
});
});
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
@@ -105,6 +105,7 @@ async function main() {
@@ -105,6 +105,7 @@ async function main() {
() => hub.close().catch(() => {}),
() => hub.close().catch(() => {}),
() => outbox.stop(),
() => outbox.stop(),
() => usage.stop(),
() => usage.stop(),
+ () => meter.close().catch(() => {}),
() => { if (subscriptions) subscriptions.stop(); },
() => { if (subscriptions) subscriptions.stop(); },
],
],
close: [
close: [
@@ -109,6 +109,21 @@ function computerSummary(kind, value) {
@@ -109,6 +109,21 @@ function computerSummary(kind, value) {
return { max_width: value.max_width || null, ...where };
return { max_width: value.max_width || null, ...where };
}
}
+/**
+ * What metering keeps of an `ai` command before its answer: the requested model (without the router's community/
+ * prefix), whether this was a community request, and the caller's task when it named one. The server that runs it
+ * (Ollama, LM Studio, llama.cpp) is only known from the answer, so it is filled in when the ack arrives.
+ */
+function aiMeterMeta(value) {
+ const wanted = typeof value.model === 'string' ? value.model : '';
+ const community = wanted.startsWith('community/');
+ return {
+ model: (community ? wanted.slice('community/'.length) : wanted) || 'unknown',
+ community,
+ ...(typeof value.task === 'string' && value.task ? { task: value.task.slice(0, 200) } : {}),
+ };
+}
+
function parseCookies(header) {
function parseCookies(header) {
const out = {};
const out = {};
for (const part of String(header || '').split(';')) {
for (const part of String(header || '').split(';')) {
@@ -147,7 +162,7 @@ function simView(sim) {
@@ -147,7 +162,7 @@ function simView(sim) {
/** How long a restart waits for every client's close in reply to its 1012 before it drops the rest (11-12). */
/** How long a restart waits for every client's close in reply to its 1012 before it drops the rest (11-12). */
const CLOSE_HANDSHAKE_MS = 600;
const CLOSE_HANDSHAKE_MS = 600;
-function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now(), networkSend = null }) {
+function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now(), networkSend = null, meter = null }) {
let domain = null;
let domain = null;
const tickets = createControlTickets({ now }); // the embedded panel's Drive (server/control-tickets.js)
const tickets = createControlTickets({ now }); // the embedded panel's Drive (server/control-tickets.js)
const deviceConns = new Map(); // device_id → conn
const deviceConns = new Map(); // device_id → conn
@@ -715,6 +730,42 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -715,6 +730,42 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);
for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);
}
}
+ /**
+ * One ai command's answer is one call that did not go through OpenVibe.AI: a local model on the person's machine.
+ * It goes to AI's meter after the answer is in hand (never awaited, never fatal), with the plugin's usage, model
+ * and server. A nack is the call failing with its fault as the error class.
+ */
+ function meterAiCommand(p, result, msg) {
+ if (!meter || !meter.enabled || !p.ai) return;
+ // OpenVibe.AI sends most ai commands (a person's own machine as their model, the community offers): those are
+ // AI's runs, and AI records each one itself. Only a call nobody else records (a person's own, an agent's) is Bot's.
+ const caller = p.conn && p.conn.principal;
+ if (caller && caller.kind === 'service' && caller.sub === 'svc:ai') return;
+ try {
+ const data = result === 'ack' && msg.data && typeof msg.data === 'object' && !Array.isArray(msg.data) ? msg.data : {};
+ const usage = data.usage && typeof data.usage === 'object' ? data.usage : null;
+ const server = typeof data.server === 'string' && data.server.trim() ? data.server.trim().slice(0, 120) : null;
+ // Ollama's native usage already names its server; an ack that carries neither is a local model with none.
+ const provider = server || (usage && usage.eval_count != null ? 'ollama' : 'local');
+ const started = p.at;
+ let first = true;
+ const call = meter.startCall({
+ system: 'node', run_id: p.opId, provider, model: String(data.model || p.ai.model || 'unknown').slice(0, 200),
+ operation: 'chat', place: p.ai.community ? 'community' : 'machine',
+ ...(p.ai.task ? { task: p.ai.task } : {}),
+ // startCall's clock begins when the command left Bot, so latency is the machine's whole answer.
+ now: () => { if (first) { first = false; return started; } return now(); },
+ });
+ meter.record(call.end({
+ usage,
+ outcome: result === 'ack' ? 'succeeded' : 'failed',
+ ...(result === 'nack' && msg.fault_code ? { error_class: String(msg.fault_code) } : {}),
+ }));
+ } catch (e) {
+ log.warn(`[Bot] ai meter: ${e.message}`);
+ }
+ }
+
async function onAck(conn, msg) {
async function onAck(conn, msg) {
const id = String(msg.id || '');
const id = String(msg.id || '');
if (id.startsWith('job_')) return jobs.frames.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's
if (id.startsWith('job_')) return jobs.frames.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's
@@ -735,6 +786,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -735,6 +786,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
cacheResult(p.key, { result, reason: msg.fault_code || null, ...(p.kind === 'headlight' && data ? { data } : {}) });
cacheResult(p.key, { result, reason: msg.fault_code || null, ...(p.kind === 'headlight' && data ? { data } : {}) });
await domain.audit.record({ robotId: p.robotId, deviceId: p.deviceId, subject: p.subject, operatorKind: p.operatorKind, role: p.role, kind: p.kind, value: p.value, result, reason: msg.fault_code || null, latencyMs: latency });
await domain.audit.record({ robotId: p.robotId, deviceId: p.deviceId, subject: p.subject, operatorKind: p.operatorKind, role: p.role, kind: p.kind, value: p.value, result, reason: msg.fault_code || null, latencyMs: latency });
const message = (p.computer || p.kind === 'headlight') && result === 'nack' && typeof msg.message === 'string' ? msg.message.replace(/[\u0000-\u001f\u007f]/g, ' ').slice(0, 300) : null;
const message = (p.computer || p.kind === 'headlight') && result === 'nack' && typeof msg.message === 'string' ? msg.message.replace(/[\u0000-\u001f\u007f]/g, ' ').slice(0, 300) : null;
+ meterAiCommand(p, result, msg);
sendFrame(p.conn, 'command_result', { id: p.opId, result, reason: message || msg.fault_code || null, latency_ms: latency, ...(data ? { data } : {}) });
sendFrame(p.conn, 'command_result', { id: p.opId, result, reason: message || msg.fault_code || null, latency_ms: latency, ...(data ? { data } : {}) });
}
}
@@ -1094,7 +1146,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -1094,7 +1146,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
const deadlineMs = now() + windowMs;
const deadlineMs = now() + windowMs;
const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind, value, deadline_ms: deadlineMs, operator: { subject, role: 'owner' } };
const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind, value, deadline_ms: deadlineMs, operator: { subject, role: 'owner' } };
if (!send(device.ws, frame)) return refuse('bot.device_offline', 'the machine is offline', summary);
if (!send(device.ws, frame)) return refuse('bot.device_offline', 'the machine is offline', summary);
- const p = { id: deviceCmdId, opId: id, key, deviceId, conn, robotId: deviceId, subject, operatorKind: conn.principal.kind, role: 'owner', kind, value: summary, at: now(), timer: null, computer: true };
+ const ai = kind === 'ai' ? aiMeterMeta(value) : null;
+ const p = { id: deviceCmdId, opId: id, key, deviceId, conn, robotId: deviceId, subject, operatorKind: conn.principal.kind, role: 'owner', kind, value: summary, at: now(), timer: null, computer: true, ...(ai ? { ai } : {}) };
pendingCmds.set(deviceCmdId, p);
pendingCmds.set(deviceCmdId, p);
inflight.set(key, deviceCmdId);
inflight.set(key, deviceCmdId);
p.timer = setTimeout(() => expireCommand(deviceCmdId), windowMs + 1500);
p.timer = setTimeout(() => expireCommand(deviceCmdId), windowMs + 1500);
new file mode 100644
new file mode 100644
@@ -0,0 +1,160 @@
@@ -0,0 +1,160 @@
+'use strict';
+// AI on people's own machines is metered (ai-metrics.md): an `ai` command's ack is one ai.call-report@1 record for
+// OpenVibe.AI, with the plugin's usage (Ollama's counts and decode time through openvibe-sdk/ai-meter's normalize),
+// model server, model and the whole answer's latency; a nack is a failed call with its fault as the error class; a
+// Bot without OV_AI_INTERNAL_URL sends nothing at all.
+const assert = require('assert');
+const http = require('http');
+const { boot, check, done } = require('./helpers/app');
+
+/** OpenVibe.AI's meter stand-in: records every POST /api/v1/usage/calls and answers 202. */
+async function startAi() {
+ const requests = [];
+ const server = http.createServer((req, res) => {
+ const chunks = [];
+ req.on('data', (d) => chunks.push(d));
+ req.on('end', () => {
+ let body = null;
+ try { body = JSON.parse(Buffer.concat(chunks).toString('utf8')); } catch { /* not json */ }
+ requests.push({ url: req.url, method: req.method, auth: req.headers.authorization, body });
+ res.writeHead(202, { 'content-type': 'application/json' });
+ res.end('{}');
+ });
+ });
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ return { url: `http://127.0.0.1:${server.address().port}`, requests, close: () => new Promise((r) => { server.closeAllConnections(); server.close(r); }) };
+}
+
+(async () => {
+ const ai = await startAi();
+ let t = null;
+ let seq = 0;
+ const frame = (type, fields) => ({ v: 1, seq: ++seq, ts: Date.now(), type, ...fields });
+ const poll = async (fn, ms = 8000) => { const end = Date.now() + ms; while (Date.now() < end) { const v = await fn(); if (v) return v; await t.wait(30); } return null; };
+
+ /**
+ * A Network-paired machine whose OpenVibe Link ai plugin answers `ai` commands with `answer(m, c)`; every other
+ * kind is refused. Returns its device id, and `setAnswer` swaps the plugin's reply between checks.
+ */
+ async function machine(owner) {
+ const principal = t.network.addNode({ owner: owner.subject, ref: `worker:${owner.subject}`, name: 'ai-pc' });
+ const c = await t.ws('/device', { headers: { Authorization: `Bearer ${t.network.signNode(principal)}` } });
+ const deviceId = (await c.waitFor((m) => m.type === 'hello')).device_id;
+ const config = await c.waitFor((m) => m.type === 'config');
+ const box = { answer: null };
+ c.send(frame('status', { capabilities: { ai: { ai: { models: ['llama3.1:8b'], ready: true, max_tokens: 2048 } } }, drivers: [], faults: [] }));
+ c.send(frame('telemetry', { sensors: {} }));
+ c.ws.on('message', (raw) => {
+ let m; try { m = JSON.parse(raw.toString()); } catch { return; }
+ if (m.type !== 'command') return;
+ if (m.kind === 'ai' && box.answer) box.answer(m, c, frame);
+ else c.send(frame('nack', { id: m.id, fault_code: 'unsupported', message: 'not this plugin' }));
+ });
+ const beat = setInterval(() => { try { c.send(frame('heartbeat', { t: Date.now() })); } catch { /* closed */ } }, Math.max(100, Math.floor(config.heartbeat_ms / 2)));
+ c.ws.on('close', () => clearInterval(beat));
+ await poll(() => t.hub.computerOf(deviceId) !== null);
+ return { c, deviceId, setAnswer: (fn) => { box.answer = fn; } };
+ }
+
+ try {
+ t = await boot({ env: { OV_AI_INTERNAL_URL: ai.url } });
+ assert.strictEqual(t.app.locals.meter.enabled, true, 'the meter is on with OV_AI_INTERNAL_URL');
+ const alex = t.network.newUser('alex');
+ const gpu = await machine(alex);
+ const command = (id, value) => t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, { user: alex, body: { id, kind: 'ai', value } });
+ const ask = (id) => command(id, { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'What is a mutex?' }] });
+
+ await check('an ack with Ollama usage records one local call with tokens and tokens per second', async () => {
+ gpu.setAnswer((m, c, send) => c.send(send('ack', {
+ id: m.id,
+ data: {
+ model: 'llama3.1:8b', server: 'ollama', ms: 9600, finish_reason: 'stop',
+ message: { role: 'assistant', content: 'A lock one thread holds at a time.' },
+ usage: { prompt_eval_count: 300, eval_count: 120, eval_duration: 9.6e9 },
+ },
+ })));
+ const r = await ask('cmd_ollama_1');
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(r.json.result, 'ack');
+ await t.app.locals.meter.flush();
+ const sent = ai.requests.at(-1);
+ assert.strictEqual(sent.url, '/api/v1/usage/calls');
+ assert.match(sent.auth, /^Bearer /, 'a minted service token');
+ const token = t.network.tokenCalls.find((x) => x.audience === 'openvibe.ai');
+ assert.deepStrictEqual([token && token.scope, token && token.client_id], ['ai.usage.report', 'bot'], 'Bot mints audience openvibe.ai for ai.usage.report');
+ const call = sent.body.calls[0];
+ assert.deepStrictEqual([call.system, call.run_id, call.operation, call.place, call.outcome],
+ ['node', 'cmd_ollama_1', 'chat', 'machine', 'succeeded']);
+ assert.deepStrictEqual([call.provider, call.model], ['ollama', 'llama3.1:8b']);
+ assert.deepStrictEqual([call.input, call.output, call.reasoning, call.cache_read], [300, 120, 0, 0]);
+ assert.strictEqual(call.tps, 12.5, '120 tokens over Ollama’s 9.6 s decode');
+ assert.ok(call.latency_ms >= 0 && Number.isFinite(call.ttft_ms) === false, 'the whole answer’s latency, no streamed first token');
+ assert.strictEqual(call.streamed, false);
+ assert.ok(call.at && !Number.isNaN(Date.parse(call.at)), 'an ISO time');
+ });
+
+ await check('a community/ model records the bare model and place community', async () => {
+ gpu.setAnswer((m, c, send) => c.send(send('ack', {
+ id: m.id, data: { model: 'llama3.1:8b', server: 'lmstudio', message: { role: 'assistant', content: 'hi' }, usage: { prompt_tokens: 10, completion_tokens: 20 } },
+ })));
+ const r = await command('cmd_community_1', { model: 'community/llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] });
+ assert.strictEqual(r.json.result, 'ack', r.text);
+ await t.app.locals.meter.flush();
+ const call = ai.requests.at(-1).body.calls[0];
+ assert.deepStrictEqual([call.model, call.provider, call.place, call.input, call.output], ['llama3.1:8b', 'lmstudio', 'community', 10, 20]);
+ });
+
+ await check('a nack records a failed call with its fault as the error class', async () => {
+ gpu.setAnswer((m, c, send) => c.send(send('nack', { id: m.id, fault_code: 'not_ready', message: 'the model is answering another request on this machine: try again in a moment' })));
+ const r = await ask('cmd_nack_1');
+ assert.strictEqual(r.json.result, 'nack');
+ assert.match(r.json.reason, /another request/);
+ await t.app.locals.meter.flush();
+ const call = ai.requests.at(-1).body.calls[0];
+ assert.deepStrictEqual([call.run_id, call.outcome, call.error_class, call.place], ['cmd_nack_1', 'failed', 'not_ready', 'machine']);
+ assert.deepStrictEqual([call.input, call.output], [0, 0], 'no usage: zero tokens, never null');
+ });
+
+ await check('a command OpenVibe.AI sends is AI\'s run: AI records it, Bot never reports it again', async () => {
+ const on = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/ai-models`, { user: alex, body: { on: true } });
+ assert.strictEqual(on.status, 200, on.text);
+ gpu.setAnswer((m, c, send) => c.send(send('ack', {
+ id: m.id, data: { model: 'llama3.1:8b', server: 'ollama', message: { role: 'assistant', content: 'hi' }, usage: { prompt_eval_count: 3, eval_count: 2, eval_duration: 1e9 } },
+ })));
+ const before = ai.requests.length;
+ const asAi = { token: t.network.signService({ sub: 'svc:ai', cap: ['bot.machine.ai'] }), headers: { 'X-OV-Subject': alex.subject } };
+ const r = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, { ...asAi, body: { id: 'cmd_from_ai_1', kind: 'ai', value: { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] } } });
+ assert.strictEqual(r.json.result, 'ack', r.text);
+ await t.app.locals.meter.flush();
+ assert.strictEqual(ai.requests.length, before, 'no report: AI publishes its own ai.call.recorded@1 for this run');
+ });
+
+ gpu.c.close();
+ await t.close();
+ t = null;
+
+ await check('without OV_AI_INTERNAL_URL the meter is off and an ai answer sends nothing', async () => {
+ const before = ai.requests.length;
+ t = await boot();
+ assert.strictEqual(t.app.locals.meter.enabled, false, 'no AI URL, no meter');
+ const alex = t.network.newUser('alex');
+ const gpu = await machine(alex);
+ gpu.setAnswer((m, c, send) => c.send(send('ack', {
+ id: m.id, data: { model: 'llama3.1:8b', server: 'ollama', message: { role: 'assistant', content: 'hi' }, usage: { prompt_eval_count: 3, eval_count: 2, eval_duration: 1e9 } },
+ })));
+ const r = await t.call('POST', `/api/v1/machines/${gpu.deviceId}/commands`, {
+ user: alex, body: { id: 'cmd_off_1', kind: 'ai', value: { model: 'llama3.1:8b', messages: [{ role: 'user', content: 'hi' }] } },
+ });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(r.json.result, 'ack');
+ await t.wait(150);
+ assert.strictEqual(ai.requests.length, before, 'nothing reached AI');
+ assert.strictEqual(t.app.locals.meter.stats().queued, 0);
+ gpu.c.close();
+ });
+ } finally {
+ if (t) await t.close().catch(() => {});
+ await ai.close();
+ done();
+ }
+})().catch((e) => { console.error(e); process.exit(1); });