openvibers/openvibe.ai Commits

AI: meter every call with the 1.1.0 fields and accept usage reports (#517)

OP OpenVibers committed 10 h ago · cf8c33b10e · 16 files, +620 −23

* AI: meter every call with the 1.1.0 fields and accept reports

* AI: measure Anthropic streams and parent advisor tasks

* usage/calls: the capability proposal for ai.usage.report, and a dedupe key that names the calling service

The proposals test wants every capability AI enforces in docs/capabilities-proposal and the manifest. The dedupe key
now carries the token's subject, so one product's run ids can never mark another product's report a duplicate.

Changes

ddiff --git a/README.md b/README.md
iindex 4992ab0..fcd1a4d 100644
---- a/README.md
++++ b/README.md
@@@ -65,14 +65,16 @@ owns publication truth: Wiki, Blog, News, Live, … decide what to publish.
  - OpenVibe.Media (recordings `media.analyze` reads, over allow-listed https URLs or signed URLs)
  - provider APIs configured by the operator (OpenAI-compatible, OpenAI Responses, Anthropic, the HTTP seam), whisper.cpp
    and the local model server `openvibe-llm.service` on this host
--- `openvibe-contracts` v0.252.0, `openvibe-sdk` v0.82.0 (service tokens, events outbox, redact),
++- `openvibe-contracts` v0.421.0, `openvibe-sdk` v0.82.0 (service tokens, events outbox, redact),
    `openvibe-shared` v4.109.2, pinned by release tarball
  
  ## Capabilities
  
  Implemented here (the service manifest's `capabilities`, audience `openvibe.ai`; routes under
  [API](#api)): `ai.run.create`, `ai.run.read`, `ai.workflow.manage`, `ai.provider.manage`,
--`ai.usage.read`, `ai.credential.manage` (`/api/v1/credentials/:subject[/:provider]`),
++`ai.usage.read`, `ai.usage.report` (`POST /api/v1/usage/calls`: another product reports the AI calls it made
++without AI, `ai.call-report@1`, and AI prices, dedupes and publishes each as `ai.call.recorded@1`),
++`ai.credential.manage` (`/api/v1/credentials/:subject[/:provider]`),
  `ai.quota.attribution.manage` (per-streamer quotas), `ai.memory.read` and `ai.memory.write` (Memory M2a,
  `/api/v1/memory/*`), `ai.me.read` and `ai.me.manage` ("Your AI", `/api/v1/me/*`: a person's own keys and
  routing preferences; only the person's own token or AI-site session) and `ai.app.run` (developer apps, ADR-014; public,
@@@ -145,6 +147,7 @@ idempotent replay):
  | `POST /api/v1/{templates,workflows,routes}/:key/versions`, `…/versions/:v/status` | `ai.workflow.manage` |
  | `POST/PATCH /api/v1/providers…`, `…/disable\|enable\|reset`, `POST/PATCH /api/v1/models…`, `POST /api/v1/quotas`, `DELETE /api/v1/cache` | `ai.provider.manage` |
  | `GET /api/v1/status\|usage\|quotas\|requests\|audit\|cache\|providers\|models` | `ai.usage.read` |
++| `POST /api/v1/usage/calls` — up to 100 calls a product made outside AI (`ai.call-report@1`, 202 `{ accepted, duplicates, rejected }`) | `ai.usage.report` |
  | `GET /api/v1/usage/series?days=N` (N 1–90) — aggregate demand per UTC day and model: requests, successes and the four token totals, never a caller id or text (`AI_USAGE_SERIES=on`) | `ai.usage.read` |
  | `GET /api/health`, `GET /api/ready`, `GET /release.json` | public |
  | `GET /metrics` | direct loopback callers only (a request carrying X-Forwarded-For gets 404) |
@@@ -1102,7 +1105,7 @@ ai.openvibe.network notice from OpenVibe.Sites.
  Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
  
  <!-- versions:start -->
--- openvibe-contracts: v0.416.0
++- openvibe-contracts: v0.421.0
  - openvibe-sdk: v0.152.0
  - openvibe-shared: v4.231.0
  <!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex ea20e86..4e02a44 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
    "runtime": "node 22, express, postgresql",
    "port": 4700,
    "domain": "ai.openvibe.services (the old ai.openvibe.network address was retired on 2026-10-10); the service runs on openvibe-ovh (/opt/openvibe.ai, unit openvibe-ai) on 127.0.0.1:4700 behind nginx; OpenVibe.Live and OpenVibe.Network call it on loopback",
--  "contracts": "openvibe-contracts v0.416.0 (ai.config.manage and the common.config-* contracts carry D11; billing.credit-hold@1 takes a nullable project_id for personal holds)",
++  "contracts": "openvibe-contracts v0.421.0 (ai.call.recorded@1 1.1.0, ai.call-report@1 and ai.usage.report; ai.config.manage and the common.config-* contracts carry D11)",
    "sdk": "openvibe-sdk v0.152.0",
    "shared": "openvibe-shared v4.231.0",
    "capabilities": [
@@@ -16,6 +16,7 @@
      "ai.workflow.manage",
      "ai.provider.manage",
      "ai.usage.read",
++    "ai.usage.report",
      "ai.credential.manage",
      "ai.quota.attribution.manage",
      "ai.actions.read",
ddiff --git a/docs/capabilities-proposal/ai.usage.report.json b/docs/capabilities-proposal/ai.usage.report.json
nnew file mode 100644
iindex 0000000..4554636
---- /dev/null
++++ b/docs/capabilities-proposal/ai.usage.report.json
@@@ -0,0 +1,21 @@
++{
++  "id": "ai.usage.report",
++  "version": "1.0",
++  "owner": "ai",
++  "status": "active",
++  "visibility": "internal",
++  "description": "Report AI calls a product made without OpenVibe.AI (ai.call-report@1). AI prices each call, records it once by system and run or turn id, and publishes it as ai.call.recorded@1.",
++  "permissions": [
++    "ai:write"
++  ],
++  "resourceConstraints": [
++    "none"
++  ],
++  "quotaClass": "ai-write",
++  "events": [
++    "ai.call.recorded"
++  ],
++  "implementedBy": [
++    "POST /api/v1/usage/calls"
++  ]
++}
ddiff --git a/docs/capabilities-proposal/service-manifest.ai.json b/docs/capabilities-proposal/service-manifest.ai.json
iindex 39847cf..10a5ae9 100644
---- a/docs/capabilities-proposal/service-manifest.ai.json
++++ b/docs/capabilities-proposal/service-manifest.ai.json
@@@ -29,6 +29,7 @@
      "ai.run.create",
      "ai.run.read",
      "ai.usage.read",
++    "ai.usage.report",
      "ai.workflow.manage",
      "exchange.search.icons"
    ],
ddiff --git a/migrations/0084_reported_ai_calls.sql b/migrations/0084_reported_ai_calls.sql
nnew file mode 100644
iindex 0000000..a705c77
---- /dev/null
++++ b/migrations/0084_reported_ai_calls.sql
@@@ -0,0 +1,15 @@
++-- phase: expand
++-- The idempotency key of ai.call-report@1 (Contracts 0.421.0, ai-metrics.md): a product reports an AI call it
++-- made without OpenVibe.AI, and AI records it once. SYSTEM + RUN_ID (or TURN_ID, for a harness turn) is the key:
++-- a report delivered twice inserts nothing the second time and is counted as a duplicate. The rows are small and
++-- are kept for 30 days (POST /api/v1/usage/calls prunes older ones), long enough for a retrying sender.
++CREATE TABLE IF NOT EXISTS ai_usage_reported_calls (
++    system text COLLATE "C" NOT NULL,
++    call_key text COLLATE "C" NOT NULL,
++    run_id text COLLATE "C",
++    turn_id text COLLATE "C",
++    received_at text COLLATE "C" NOT NULL,
++    PRIMARY KEY (system, call_key),
++    CHECK (run_id IS NOT NULL OR turn_id IS NOT NULL)
++);
++CREATE INDEX IF NOT EXISTS ai_usage_reported_calls_received ON ai_usage_reported_calls(received_at);
ddiff --git a/package-lock.json b/package-lock.json
iindex 12d3eef..b0edb0f 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -14,7 +14,7 @@
          "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.416.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.152.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.231.0",
          "pg": "^8.23.0",
@@@ -1635,9 +1635,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.416.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.416.0",
--      "integrity": "sha512-gFZGSXY7GM2SHdyOmqZ0699yJhF9ZsF/Rb0WHo3asI+S4LTqthqIQPOCoYczlF/dcB5p13KBjLCMdAvEqmNLoQ==",
++      "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",
ddiff --git a/package.json b/package.json
iindex 5c1f08b..bd34363 100644
---- a/package.json
++++ b/package.json
@@@ -30,7 +30,7 @@
      "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.416.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.152.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.231.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.231.0",
ddiff --git a/server/api/usage-calls.js b/server/api/usage-calls.js
nnew file mode 100644
iindex 0000000..c0f8f9e
---- /dev/null
++++ b/server/api/usage-calls.js
@@@ -0,0 +1,142 @@
++'use strict';
++/**
++ * The one front door for AI calls made outside OpenVibe.AI (ai-metrics.md, Contracts 0.421.0).
++ *
++ *   POST /api/v1/usage/calls        ai.usage.report, up to 100 calls in ai.call-report@1
++ *
++ * Actor's own model calls, a Codes harness turn, a local model on a person's machine through Bot and every
++ * other product send their calls here instead of writing their own ledger. AI stays the one owner of
++ * ai.call.recorded@1: it validates the report, keeps the subject only when the token is a service acting for
++ * that person, dedupes by system + run_id or turn_id, prices a call that brought no charged amount from its
++ * own rate cards (charged first, then list; 'unknown' with zeros when there is no card) and publishes each
++ * accepted call through server/events.js. The dedupe table keeps 30 days.
++ */
++const express = require('express');
++const { validate } = require('openvibe-contracts');
++const { AiError, sendError, iso } = require('../util');
++const usageRecord = require('../usage-record');
++const events = require('../events');
++const { CAPS } = require('../auth');
++
++const KEEP_MS = 30 * 24 * 60 * 60 * 1000;
++
++function usageCallsRouter({ db, pool, config, auth, clock = { now: () => Date.now() }, log = console }) {
++    const r = express.Router();
++    const report = auth.requireCap(CAPS.usageReport);
++
++    /** The subject the caller may name: only the person named by a service's own on_behalf_of claim. */
++    function allowedSubject(req, subject) {
++        const principal = req.principal || {};
++        if (!subject || !principal.onBehalfOf) return null;
++        return String(subject) === String(principal.onBehalfOf) ? String(subject) : null;
++    }
++
++    /**
++     * A call's money: what it brought wins; otherwise AI's rate cards price it (the same pricesOf/moneyOf a run
++     * uses). No card means 'unknown' and zeros, never an invented price.
++     */
++    async function moneyFor(call) {
++        const brought = {
++            list_cost_usd: call.list_cost_usd == null ? 0 : call.list_cost_usd,
++            charged_usd: call.charged_usd == null ? 0 : call.charged_usd,
++            fee_usd: call.fee_usd == null ? 0 : call.fee_usd,
++            cache_saved_usd: call.cache_saved_usd == null ? 0 : call.cache_saved_usd,
++            price_source: call.price_source || '',
++            price_date: call.price_date || null,
++        };
++        if (call.charged_usd != null) return brought;
++        const cards = pool && call.provider ? await pool.rateCardsFor(call.provider, call.model || null) : null;
++        if (!cards) return { ...brought, list_cost_usd: 0, charged_usd: 0, fee_usd: 0, cache_saved_usd: 0, price_source: 'unknown', price_date: null };
++        const prices = usageRecord.pricesOf(cards);
++        const money = usageRecord.moneyOf({
++            prices,
++            tokens: { input: call.input || 0, output: call.output || 0, cache_read: call.cache_read || 0, cache_write: call.cache_write || 0 },
++            paidBy: call.paid_by || 'openvibe_key',
++            feeRate: (config.usage && config.usage.feeRate) || 0,
++        });
++        return { ...money, price_source: 'catalog', price_date: prices.date };
++    }
++
++    /** One accepted report -> the record AI publishes, with every field ai.call.recorded@1 1.1.0 carries. */
++    async function recordFor(req, call) {
++        const money = await moneyFor(call);
++        return usageRecord.recordOf({
++            system: call.system,
++            run_id: call.run_id || null, turn_id: call.turn_id || null,
++            subject: allowedSubject(req, call.subject),
++            attribution: call.attribution || '', surface: call.surface || '',
++            provider: call.provider || null, model: call.model || null, effort: call.effort || null,
++            paid_by: call.paid_by || 'openvibe_key', paid_by_detail: call.paid_by_detail || null,
++            input: call.input || 0, output: call.output || 0, reasoning: call.reasoning || 0,
++            cache_read: call.cache_read || 0, cache_write: call.cache_write || 0, cache_write_ttl: call.cache_write_ttl || null,
++            images: call.images || 0, audio_seconds: call.audio_seconds || 0, video_seconds: call.video_seconds || 0,
++            list_cost_usd: money.list_cost_usd, charged_usd: money.charged_usd, fee_usd: money.fee_usd, cache_saved_usd: money.cache_saved_usd,
++            price_source: money.price_source, price_date: money.price_date,
++            latency_ms: call.latency_ms == null ? null : call.latency_ms,
++            ttft_ms: call.ttft_ms == null ? null : call.ttft_ms,
++            attempts: call.attempts == null ? 0 : call.attempts, outcome: call.outcome, at: call.at,
++            operation: call.operation || null, streamed: call.streamed == null ? null : call.streamed,
++            tps: call.tps == null ? null : call.tps, queue_ms: call.queue_ms == null ? null : call.queue_ms,
++            task: call.task || null, task_id: call.task_id || call.run_id || call.turn_id || null,
++            task_ms: call.task_ms == null ? null : call.task_ms, harness: call.harness || null, place: call.place || null,
++            error_class: call.error_class || null, fallbacks: call.fallbacks == null ? null : call.fallbacks,
++            cache_hit: call.cache_hit == null ? (call.outcome === 'cached') : call.cache_hit,
++            tool_calls: call.tool_calls == null ? null : call.tool_calls, region: call.region || null,
++        });
++    }
++
++    r.post('/api/v1/usage/calls', report, async (req, res) => {
++        try {
++            const body = req.body || {};
++            // Validate the whole report first. An error inside one call is that call's rejection (with its
++            // index); an error outside the calls array (too many, a missing array, an unknown field) refuses
++            // the report itself.
++            const whole = validate('ai.call-report@1', body);
++            if (!whole.valid) {
++                const outside = whole.errors.find(e => !/^\/calls\/\d+/.test(String(e.path || '')));
++                if (outside) throw new AiError(422, 'input.invalid', `body does not match ai.call-report@1: ${outside.path} ${outside.message}`, { errors: whole.errors });
++            }
++            const calls = body.calls;
++            if (!Array.isArray(calls)) throw new AiError(422, 'input.invalid', 'body does not match ai.call-report@1: calls must be an array');
++
++            // The dedupe rows are small and only needed for a sender's retries: keep them 30 days.
++            await db.prepare('DELETE FROM ai_usage_reported_calls WHERE received_at < ?')
++                .run(iso(clock.now() - KEEP_MS)).catch((err) => log.warn(`[usage] reported calls prune failed: ${err && err.message}`));
++
++            let accepted = 0;
++            let duplicates = 0;
++            const rejected = [];
++            for (const [index, call] of calls.entries()) {
++                const one = validate('ai.call-report@1', { calls: [call] });
++                if (!one.valid) {
++                    rejected.push({ index, error: one.errors.map(e => `${e.path} ${e.message}`).join('; ') });
++                    continue;
++                }
++                if (!call.run_id && !call.turn_id) {
++                    rejected.push({ index, error: 'run_id or turn_id is required' });
++                    continue;
++                }
++                // The key names the calling service too, so one product's ids can never block another's report.
++                const caller = String((req.principal && req.principal.sub) || 'unknown');
++                const key = `${caller}|${call.run_id ? `run:${call.run_id}` : `turn:${call.turn_id}`}`;
++                const rec = await recordFor(req, call);
++                let duplicate = false;
++                await db.tx(async () => {
++                    const inserted = await db.prepare(`INSERT INTO ai_usage_reported_calls (system, call_key, run_id, turn_id, received_at)
++                        VALUES (?, ?, ?, ?, ?) ON CONFLICT (system, call_key) DO NOTHING`)
++                        .run(String(call.system), key, call.run_id || null, call.turn_id || null, iso(clock.now()));
++                    if (!Number(inserted.changes)) { duplicate = true; return; }
++                    await events.callRecorded(rec);
++                });
++                if (duplicate) duplicates++;
++                else accepted++;
++            }
++            // 202: the calls are recorded and each event is queued; the sender may forget them.
++            res.status(202).json({ accepted, duplicates, rejected });
++        } catch (err) { sendError(res, err, req.ov, log); }
++    });
++
++    return r;
++}
++
++module.exports = { usageCallsRouter };
ddiff --git a/server/app.js b/server/app.js
iindex 45980f8..ce55e10 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -262,6 +262,9 @@ function createApp({ choices = null, searchFetch = null, config, db, registry, p
      // Operator actions shared by the admin API and the operator console (one implementation, one audit row).
      const ops = createOps({ db, registry, pool, quotas, cache, runs });
      app.use(runsRouter({ choices, runs, registry, auth, config, log }));
++    // The one front door for AI calls made outside AI (ai.call-report@1, ai.usage.report): AI prices, dedupes
++    // and publishes each as ai.call.recorded@1, so every product's calls land in the same meter.
++    app.use(require('./api/usage-calls').usageCallsRouter({ db, pool, config, auth, clock, log }));
      // Deep dives (workflow design.deep_dive): the Exchange feature's API and the Design integration's.
      app.use(require('./api/deep-dives').deepDivesRouter({ config, runs, auth, deepDive, store: deepDive && deepDive.store, capture: deepDive && deepDive.capture, mediaStore, log }));
      app.use(require('./api/orientation').createOrientation({ choices, runs, registry, auth, db, config, log }));
ddiff --git a/server/auth.js b/server/auth.js
iindex 5859ac6..500c5d3 100644
---- a/server/auth.js
++++ b/server/auth.js
@@@ -31,6 +31,9 @@ const CAPS = Object.freeze({
      workflowManage: 'ai.workflow.manage',
      providerManage: 'ai.provider.manage',
      usageRead: 'ai.usage.read',
++    // A product reports the AI calls it made without AI (Contracts 0.421.0, ai.call-report@1): AI prices,
++    // dedupes and publishes each as ai.call.recorded@1, so one meter covers every AI call on the network.
++    usageReport: 'ai.usage.report',
      credentialManage: 'ai.credential.manage',
      quotaAttributionManage: 'ai.quota.attribution.manage',
      appRun: 'ai.app.run',
ddiff --git a/server/events.js b/server/events.js
iindex 851e058..05761f5 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -125,6 +125,14 @@ function callPayload(rec) {
          price_source: rec.price_source, price_date: rec.price_date || null,
          latency_ms: rec.latency_ms, ttft_ms: rec.ttft_ms, attempts: rec.attempts, outcome: rec.outcome,
          run_id: rec.run_id || null, turn_id: rec.turn_id || null, at: rec.at,
++        // ai.call.recorded@1 1.1.0 (ai-metrics.md): what, how fast, for which task, where and what went wrong.
++        operation: rec.operation || null, streamed: rec.streamed == null ? null : Boolean(rec.streamed),
++        tps: rec.tps == null ? null : Number(rec.tps), queue_ms: rec.queue_ms == null ? null : Number(rec.queue_ms),
++        task: rec.task || null, task_id: rec.task_id || null, task_ms: rec.task_ms == null ? null : Number(rec.task_ms),
++        harness: rec.harness || null, place: rec.place || null, error_class: rec.error_class || null,
++        fallbacks: rec.fallbacks == null ? null : Number(rec.fallbacks),
++        cache_hit: rec.cache_hit == null ? null : Boolean(rec.cache_hit),
++        tool_calls: rec.tool_calls == null ? null : Number(rec.tool_calls), region: rec.region || null,
      };
  }
  
ddiff --git a/server/runs.js b/server/runs.js
iindex e03078c..baab2c5 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -234,7 +234,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
       * principal: { sub, subject: {type, id} }
       * Returns { run, created, promise? }.
       */
--    async function create(body, principal, { trace = null, requestId = null, retryOf = null, onDelta = null, onRoute = null, abuse: requestAbuse = null, discount = null, exact = null, costCheck = null, communityOfferId = null, callerWaitMs = 0, advisorInternal = false } = {}) {
++    async function create(body, principal, { trace = null, requestId = null, retryOf = null, onDelta = null, onRoute = null, abuse: requestAbuse = null, discount = null, exact = null, costCheck = null, communityOfferId = null, callerWaitMs = 0, advisorInternal = false, taskParent = null } = {}) {
          if (!body || typeof body !== 'object') throw new AiError(400, 'input.invalid', 'JSON body required');
          const namedTask = choices && body.task ? await choices.task(String(body.task)) : null;
          if (body.task && !namedTask) throw new AiError(404, 'task.not_found', 'unknown task');
@@@ -506,6 +506,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
                  ...(projectProvider ? { project_provider: projectProvider } : {}), ...(pinned ? { pinned_offer: pinned } : {}),
                  ...(communityOptIn ? { community: true } : {}), ...(prefer ? { prefer } : {}),
                  ...(contentPolicy !== 'standard' ? { policy: contentPolicy } : {}),
++                ...(taskParent ? { task_parent: taskParent } : {}),
                  ...(preset ? { preset: { name: String(preset.name).trim(), scope: preset.scope || 'owner', policy: presetPolicy } } : {}) }), created_at: now,
          };
          base.input = stableStringify(kept.value);
@@@ -540,7 +541,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
                      await events.runChanged(await getRow(id));
                  });
                  // A cache hit is still a run: its record says so (no tokens, no money).
--                await publishUsage({ row: await getRow(id), outcome: 'cached' });
++                await publishUsage({ row: await getRow(id), outcome: 'cached', streamed: false, workflow: wf });
                  // A reused answer carries a receipt too: the offer that produced it, and the fact that no
                  // provider was called for this one.
                  const cachedRow = await getRow(id);
@@@ -768,6 +769,38 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
          };
      }
  
++    /**
++     * Where the model ran, for the meter's `place` (ai.call.recorded@1 1.1.0): a person's own machine, a
++     * signed-in plan, a community machine or endpoint, OpenVibe's own host (the local model), else a provider.
++     */
++    async function placeOf({ payer = null, provider = null, explain = null } = {}) {
++        if (payer && payer.paid_by === 'own_machine') return 'machine';
++        if (payer && payer.paid_by === 'plan') return 'plan';
++        if (/^community[:-]/.test(String(provider || '')) || (explain && explain.community)) return 'community';
++        if (provider) {
++            const p = await registry.getProvider(provider).catch(() => null);
++            if (p && p.metadata && p.metadata.local) return 'host';
++            return 'provider';
++        }
++        return null;
++    }
++
++    /** The attempts that failed before the one that answered (or, when none answered, all the failed ones). */
++    function fallbacksOf(requests) {
++        const answered = requests.findIndex(r => r.status === 'ok');
++        const before = answered >= 0 ? requests.slice(0, answered) : requests;
++        return before.filter(r => r.status === 'error' || r.status === 'timeout').length;
++    }
++
++    /** The parent run when an advice call asked for this one (run_advisor_uses), else the run itself. */
++    async function taskIdOf(row) {
++        try {
++            const parent = await db.prepare('SELECT run_id FROM run_advisor_uses WHERE advice_run_id = ? ORDER BY created_at DESC LIMIT 1').get(row.id);
++            if (parent && parent.run_id) return parent.run_id;
++        } catch { /* an older database without advisor uses: the run is its own task */ }
++        return row.id;
++    }
++
      /**
       * The record of a run (plan L40): the tokens (the cache read and the cache write counted apart), the money and
       * who paid. It is published as one ai.call.recorded@1 event; OpenVibe.Report keeps the ledger. The record never
@@@ -776,9 +809,10 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
       * It answers the money it computed, so the receipt of the same run (server/exchange/receipts.js) carries the
       * same numbers and never a second, different price.
       */
--    async function publishUsage({ row, outcome, usage = {}, provider = null, model = null, attempts = 0, latencyMs = null, ttftMs = null }) {
++    async function publishUsage({ row, outcome, usage = {}, provider = null, model = null, attempts = 0, latencyMs = null, ttftMs = null, streamed = false, firstDeltaAt = null, workflow = null }) {
          try {
              const opts = parseJson(row.options, {});
++            const explain = parseJson(row.explain, {}) || {};
              const payer = await payerOf({ row, providerKey: provider, opts });
              const tokens = {
                  input: Math.max(0, (usage.input || 0) - (usage.cached || 0) - (usage.cache_write || 0)),
@@@ -788,9 +822,19 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
              const cards = provider ? await pool.rateCardsFor(provider, model) : null;
              const prices = usageRecord.pricesOf(cards);
              const money = usageRecord.moneyOf({ prices, tokens, paidBy: payer.paid_by, feeRate: (config.usage && config.usage.feeRate) || 0 });
++            // The attempt log is where the operation, the failed attempts and the error class come from. A run
++            // served from the cache has no attempts; its workflow still names the operation.
++            const requests = await db.prepare('SELECT operation, status, error_class, error FROM requests WHERE run_id = ? ORDER BY seq').all(row.id).catch(() => []);
++            const answered = requests.find(r => r.status === 'ok');
++            const lastAttempt = [...requests].reverse().find(r => r.status !== 'skipped');
++            const operation = usageRecord.operationOf({ workflowKey: row.workflow_key, workflow, operation: (answered || lastAttempt || {}).operation || null });
++            const failed = outcome === 'failed' ? [...requests].reverse().find(r => r.status === 'error' || r.status === 'timeout') : null;
++            const errorClass = failed ? usageRecord.errorClassOf({ failureClass: failed.error_class, code: row.error_code, message: failed.error }) : null;
++            const finishedAt = row.finished_at || new Date(clock.now()).toISOString();
++            const taskId = opts.task_parent || await taskIdOf(row);
              const rec = usageRecord.recordOf({
                  system: 'ai', run_id: row.id, subject: subjectOf(row), attribution: attributionOf(row), surface: row.workflow_key || '',
--                provider: payer.provider || provider || null, model: model || null, effort: parseJson(row.explain, {})?.effort_applied || opts.effort || null,
++                provider: payer.provider || provider || null, model: model || null, effort: explain.effort_applied || opts.effort || null,
                  paid_by: payer.paid_by, paid_by_detail: payer.detail,
                  input: tokens.input, output: tokens.output, reasoning: tokens.reasoning, cache_read: tokens.cache_read,
                  cache_write: tokens.cache_write, cache_write_ttl: usage.cache_write_ttl || null,
@@@ -798,7 +842,18 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
                  list_cost_usd: money.list_cost_usd, charged_usd: money.charged_usd, fee_usd: money.fee_usd, cache_saved_usd: money.cache_saved_usd,
                  price_source: prices.source, price_date: prices.date,
                  latency_ms: latencyMs, ttft_ms: ttftMs, attempts, outcome,
--                at: row.finished_at || new Date(clock.now()).toISOString(),
++                // ai.call.recorded@1 1.1.0 (ai-metrics.md): what ran, how fast, for which task, where, and what failed.
++                operation, streamed: outcome === 'cached' ? false : Boolean(streamed),
++                tps: usageRecord.tpsOf({ output: tokens.output, streamed, firstDeltaAt, finishedAt }),
++                queue_ms: outcome === 'cached' ? null : require('./exchange/timing').durationMs(row.created_at, row.started_at),
++                task: opts.task_choice && opts.task_choice.task ? opts.task_choice.task.name : (row.workflow_key || null),
++                task_id: taskId,
++                harness: 'openvibe',
++                place: await placeOf({ payer, provider: payer.provider || provider, explain }),
++                error_class: errorClass,
++                fallbacks: fallbacksOf(requests),
++                cache_hit: outcome === 'cached',
++                at: finishedAt,
              });
              // The event (ai.call.recorded) is queued in its own transaction, like every other event here.
              await db.tx(async () => { await events.callRecorded(rec); });
@@@ -986,6 +1041,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
          let credit = null;
          let billingDelta = null;
          let firstDeltaAt = null;
++        let streamed = false;
          let onRoute = null;
          try {
              // AI Exchange F: hold prepaid Balance before the first managed provider call (AI_CREDITS, off by
@@@ -1001,9 +1057,13 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
              }
              billingDelta = typeof ctx.onDelta === 'function'
                  ? (d) => {
--                    const delta = d && d.delta;
--                    if (firstDeltaAt === null && (delta?.content || delta?.reasoning_content || delta?.thinking
--                        || delta?.tool_calls?.some((c) => c.function?.arguments))) firstDeltaAt = clock.now();
++                    streamed = true;
++                    // OpenAI-shaped deltas carry `delta`; the Messages API's own stream carries `data.delta`.
++                    const delta = d && (d.delta || (d.data && d.data.delta));
++                    const kind = d && (d.event || d.type || (d.data && d.data.type)) || '';
++                    if (firstDeltaAt === null && (delta?.content || delta?.text || delta?.reasoning_content || delta?.thinking
++                        || delta?.tool_calls?.some((c) => c.function?.arguments)
++                        || /content_block_delta|text_delta|input_json_delta/.test(String(kind)))) firstDeltaAt = clock.now();
                      if (credit) credit.observe(d);
                      ctx.onDelta(d);
                  }
@@@ -1127,7 +1187,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
              const finishedRun = await getRow(id);
              const timing = { latencyMs: require('./exchange/timing').durationMs(finishedRun.started_at, finishedRun.finished_at),
                  ttftMs: firstDeltaAt === null ? null : Math.max(0, firstDeltaAt - startedAt) };
--            const published = await publishUsage({ row: finishedRun, outcome: 'succeeded', usage: r.usage, provider: r.provider, model: r.model, attempts, ...timing });
++            const published = await publishUsage({ row: finishedRun, outcome: 'succeeded', usage: r.usage, provider: r.provider, model: r.model, attempts, streamed, firstDeltaAt, workflow: wf, ...timing });
              // The receipt of the answer (server/exchange/receipts.js): the offer, the price, the tokens, the cache,
              // the latency, why the router chose it, the fallbacks it would have used, the payer and the privacy label.
              await writeReceipt({
@@@ -1193,7 +1253,8 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
              // A run that broke still spent: its record says what it cost and who paid (L40).
              await publishUsage({ row: await getRow(id), outcome: cancelled ? 'cancelled' : 'failed',
                  usage: { input: Number(spent.ti) || 0, output: Number(spent.tout) || 0, cached: Number(spent.tc) || 0 },
--                provider: logged.length ? logged[logged.length - 1].provider_key : null, model: null, attempts });
++                provider: logged.length ? logged[logged.length - 1].provider_key : null, model: null, attempts,
++                streamed, firstDeltaAt, workflow: wf });
              // Account the spent tokens/cost as before, but always roll up the finished attempts (the failed
              // ones included) into provider_stats_daily/placement_state: a run that answered nothing still
              // tells the router a provider was unhealthy.
@@@ -1334,7 +1395,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
              ...(row.on_behalf_of ? { on_behalf_of: parseJson(row.on_behalf_of, null) } : {}),
              ...(row.attribution ? { attribution: parseJson(row.attribution, null) } : {}),
              options: { cache: false },
--        }, principal, { trace, advisorInternal: true });
++        }, principal, { trace, advisorInternal: true, taskParent: id });
          const done = await wait(created, config.runs.maxWaitMs);
          if (!done || ['failed', 'cancelled'].includes(done.status)) throw new AiError(502, 'advisor.failed', (done && done.error && done.error.detail) || 'the advisor run did not answer');
          const cut = advisor.words(done.output && done.output.text, maxWords);
ddiff --git a/server/usage-record.js b/server/usage-record.js
iindex 9c908c0..5414ef8 100644
---- a/server/usage-record.js
++++ b/server/usage-record.js
@@@ -17,6 +17,104 @@ const iso = (v) => (v instanceof Date ? v : new Date(v)).toISOString();
  /** Six decimals, the way money is written here. */
  const money = (v) => Math.round((Number(v) || 0) * 1e6) / 1e6;
  
++// ── The shape of a call (ai.call.recorded@1 1.1.0) ─────────────────────────────────────
++//
++// The new fields are all optional. The operation is one of the contract's names: the run's own operation
++// (generate, summarize, chat, embed, image, speak, transcribe) is mapped onto it, never invented.
++const OPERATIONS = Object.freeze(['chat', 'completion', 'embedding', 'image', 'image_edit', 'speech', 'transcription',
++    'translation', 'rerank', 'moderation', 'decision', 'realtime', 'batch', 'tool', 'other']);
++const PLACES = Object.freeze(['provider', 'machine', 'community', 'plan', 'host']);
++const OP_ALIASES = Object.freeze({
++    chat: 'chat', complete: 'chat', completion: 'completion', messages: 'chat', passthrough: 'chat',
++    generate: 'chat', summarize: 'chat', classify: 'chat', extract: 'chat', enrich: 'chat',
++    embed: 'embedding', embedding: 'embedding',
++    transcribe: 'transcription', transcription: 'transcription', translate: 'translation', translation: 'translation',
++    image: 'image', image_edit: 'image_edit', picture: 'image',
++    speak: 'speech', speech: 'speech', rerank: 'rerank', moderation: 'moderation',
++    decide: 'decision', decision: 'decision', realtime: 'realtime', batch: 'batch', tool: 'tool', other: 'other',
++});
++
++/**
++ * The contract's operation for a run: the workflow key names the special ones (decide.*, moderation.*,
++ * live.translate, a rerank), then the operation the engine/provider called, then the workflow's first step.
++ */
++function operationOf({ workflowKey = null, workflow = null, operation = null } = {}) {
++    const key = String(workflowKey || '');
++    if (/^decide\./.test(key)) return 'decision';
++    if (/moderation/.test(key)) return 'moderation';
++    if (/translat/.test(key)) return 'translation';
++    if (/rerank/.test(key)) return 'rerank';
++    if (operation) {
++        const mapped = OP_ALIASES[String(operation).toLowerCase()];
++        if (mapped) return mapped;
++        return 'other';
++    }
++    const steps = workflow && Array.isArray(workflow.steps) ? workflow.steps : [];
++    const step = steps.find(s => s && ['llm', 'passthrough', 'complete', 'messages', 'transcribe', 'embed', 'image', 'speak', 'picture'].includes(s.kind)) || steps[0];
++    if (!step) {
++        if (/image/.test(key)) return 'image';
++        if (/embed/.test(key)) return 'embedding';
++        if (/transcri|\.stt\b/.test(key)) return 'transcription';
++        if (/speak|speech|tts/.test(key)) return 'speech';
++        return null;
++    }
++    if (step.kind === 'transcribe') return 'transcription';
++    if (step.kind === 'embed') return 'embedding';
++    if (step.kind === 'image' || step.kind === 'picture') return 'image';
++    if (step.kind === 'speak') return 'speech';
++    if (step.operation) return OP_ALIASES[String(step.operation).toLowerCase()] || 'other';
++    return 'chat';
++}
++
++/**
++ * The contract's error class, from a provider attempt's class and its words. The provider router's five
++ * classes (balance, auth, rate_limit, transient, request) are finer than the meter needs; the meter keeps
++ * the shape the ledger and Report group by (rate_limited, timeout, context_length, refused, auth, server,
++ * network, invalid).
++ */
++function errorClassOf({ failureClass = null, code = null, message = null, status = null } = {}) {
++    const text = `${code || ''} ${message || ''}`.toLowerCase();
++    const cls = failureClass ? String(failureClass) : null;
++    if (cls === 'rate_limit') return 'rate_limited';
++    if (cls === 'auth' || cls === 'balance') return 'auth';
++    if (cls === 'transient') {
++        if (code === 'provider.timeout' || /timeout|timed out|no response within/.test(text)) return 'timeout';
++        if (code === 'provider.unreachable' || code === 'provider.stream_broken'
++            || /fetch failed|econn|etimedout|eai_again|socket|connection|network|unreachable/.test(text)) return 'network';
++        return 'server';
++    }
++    if (cls === 'request') {
++        if (/context (length|window)|too long|maximum context|token limit/.test(text)) return 'context_length';
++        if (/refus|floor_refused|policy/.test(text)) return 'refused';
++        return 'invalid';
++    }
++    if (/rate[ _-]?limit/.test(text)) return 'rate_limited';
++    if (/timeout|timed out/.test(text)) return 'timeout';
++    if (/context (length|window)|too long|maximum context/.test(text)) return 'context_length';
++    if (/refus|floor_refused/.test(text)) return 'refused';
++    if (/unauthorized|forbidden|permission|auth/.test(text) || status === 401 || status === 403) return 'auth';
++    if (/network|unreachable|fetch failed|econn/.test(text)) return 'network';
++    if (/invalid|bad request|schema|400|422/.test(text) || (status != null && status >= 400 && status < 500)) return 'invalid';
++    if (/server|unavailable|resting/.test(text) || (status != null && status >= 500)) return 'server';
++    return null;
++}
++
++/**
++ * Output tokens per second after the first delta: the decode speed the Exchange's Measured panels show.
++ * Null when the call did not stream, or when there are too few tokens (under 8) or too little time
++ * (under 200 ms) for the number to mean anything.
++ */
++function tpsOf({ output = 0, streamed = false, firstDeltaAt = null, finishedAt = null } = {}) {
++    if (!streamed || firstDeltaAt == null || finishedAt == null) return null;
++    const tokens = Number(output) || 0;
++    const start = Number(firstDeltaAt);
++    const end = typeof finishedAt === 'number' ? finishedAt : Date.parse(finishedAt);
++    if (!Number.isFinite(start) || !Number.isFinite(end)) return null;
++    const ms = end - start;
++    if (tokens < 8 || !(ms >= 200)) return null;
++    return Math.round((tokens / (ms / 1000)) * 100) / 100;
++}
++
  // ── The money of one record ────────────────────────────────────────────────────────────
  
  /**
@@@ -82,8 +180,13 @@ function recordOf({ system = 'ai', run_id: runId = null, turn_id: turnId = null,
      images = 0, audio_seconds: audioSeconds = 0, video_seconds: videoSeconds = 0,
      list_cost_usd: listCost = 0, charged_usd: charged = 0, fee_usd: fee = 0, cache_saved_usd: cacheSaved = 0,
      price_source: priceSource = '', price_date: priceDate = null, latency_ms: latencyMs = null, ttft_ms: ttftMs = null,
--    attempts = 1, outcome = 'succeeded', at = new Date() }) {
++    attempts = 1, outcome = 'succeeded', at = new Date(),
++    operation = null, streamed = null, tps = null, queue_ms: queueMs = null, task = null, task_id: taskId = null,
++    task_ms: taskMs = null, harness = null, place = null, error_class: errorClass = null, fallbacks = null,
++    cache_hit: cacheHit = null, tool_calls: toolCalls = null, region = null }) {
      if (!PAID_BY.includes(paidBy)) throw new Error(`paid_by must be one of ${PAID_BY.join(', ')}`);
++    const bounded = (v, max) => (v == null ? null : String(v).slice(0, max));
++    const int = (v) => (v == null ? null : Math.max(0, Math.round(Number(v) || 0)));
      return {
          system: String(system || 'ai'),
          run_id: runId || null,
@@@ -116,7 +219,21 @@ function recordOf({ system = 'ai', run_id: runId = null, turn_id: turnId = null,
          attempts: Math.max(0, num(attempts)),
          outcome: String(outcome || 'succeeded'),
          at: iso(at),
++        operation: operation && OPERATIONS.includes(String(operation)) ? String(operation) : null,
++        streamed: streamed == null ? null : Boolean(streamed),
++        tps: tps == null || !Number.isFinite(Number(tps)) ? null : Math.round(Math.max(0, Number(tps)) * 100) / 100,
++        queue_ms: int(queueMs),
++        task: bounded(task, 120),
++        task_id: bounded(taskId, 120),
++        task_ms: int(taskMs),
++        harness: bounded(harness, 60),
++        place: place && PLACES.includes(String(place)) ? String(place) : null,
++        error_class: bounded(errorClass, 60),
++        fallbacks: fallbacks == null ? null : Math.min(50, int(fallbacks)),
++        cache_hit: cacheHit == null ? null : Boolean(cacheHit),
++        tool_calls: int(toolCalls),
++        region: bounded(region, 40),
      };
  }
  
--module.exports = { PAID_BY, money, pricesOf, moneyOf, recordOf, iso };
++module.exports = { PAID_BY, OPERATIONS, PLACES, money, pricesOf, moneyOf, operationOf, errorClassOf, tpsOf, recordOf, iso };
ddiff --git a/test/ai-usage.test.js b/test/ai-usage.test.js
iindex ce31bcf..52dafa1 100644
---- a/test/ai-usage.test.js
++++ b/test/ai-usage.test.js
@@@ -103,4 +103,52 @@ t.test('the record of a run carries the tokens and the money, and no text at all
      assert.throws(() => record.recordOf({ paid_by: 'someone_else' }), /paid_by/, 'only the known payers');
  });
  
++t.test('the 1.1.0 fields: operation, speed, task, place and the failure class', () => {
++    const rec = record.recordOf({
++        system: 'ai', run_id: 'run_01JAB2C3D4E5F6G7H8J9K0MNPQ', subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ',
++        attribution: 'site:design', surface: 'ai.generate', provider: 'anthropic', model: 'claude-sonnet-4-5',
++        paid_by: 'openvibe_key', input: 200, output: 80, list_cost_usd: 0.0021, charged_usd: 0.0021,
++        latency_ms: 1200, ttft_ms: 300, attempts: 3, outcome: 'succeeded', at: '2026-10-31T18:00:00.000Z',
++        operation: 'chat', streamed: true, tps: 21.5, queue_ms: 12, task: 'chat.reply', task_id: 'run_PARENT',
++        harness: 'openvibe', place: 'provider', error_class: null, fallbacks: 2, cache_hit: false,
++    });
++    const payload = events.callPayload(rec);
++    assert.strictEqual(payload.operation, 'chat', 'what kind of call it was');
++    assert.strictEqual(payload.streamed, true, 'whether it streamed');
++    assert.strictEqual(payload.tps, 21.5, 'decode tokens per second');
++    assert.strictEqual(payload.queue_ms, 12, 'time waited before the call');
++    assert.strictEqual(payload.task, 'chat.reply');
++    assert.strictEqual(payload.task_id, 'run_PARENT', 'the task the call is one step of');
++    assert.strictEqual(payload.place, 'provider', 'where the model ran');
++    assert.strictEqual(payload.fallbacks, 2, 'the offers that failed before the answer');
++    assert.strictEqual(payload.cache_hit, false, 'not an exact-cache answer');
++    assert.strictEqual(payload.harness, 'openvibe');
++
++    // The operation comes from the run's own operation or workflow, never invented.
++    assert.strictEqual(record.operationOf({ workflowKey: 'ai.generate', operation: 'generate' }), 'chat');
++    assert.strictEqual(record.operationOf({ workflowKey: 'decide.fallback', operation: 'classify' }), 'decision');
++    assert.strictEqual(record.operationOf({ workflowKey: 'ai.embed', operation: 'embed' }), 'embedding');
++    assert.strictEqual(record.operationOf({ workflowKey: 'ai.speak', operation: 'speak' }), 'speech');
++    assert.strictEqual(record.operationOf({ workflowKey: 'live.media.transcribe', operation: 'transcribe' }), 'transcription');
++    assert.strictEqual(record.operationOf({ workflowKey: 'live.translate', operation: 'generate' }), 'translation');
++    assert.strictEqual(record.operationOf({ workflowKey: 'moderation.classify', operation: 'classify' }), 'moderation');
++    assert.strictEqual(record.operationOf({ workflowKey: 'ai.image', operation: 'image' }), 'image');
++
++    // The provider's five failure classes become the meter's names.
++    assert.strictEqual(record.errorClassOf({ failureClass: 'rate_limit' }), 'rate_limited');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'auth' }), 'auth');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'transient', code: 'provider.timeout' }), 'timeout');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'transient', message: 'fetch failed: ECONNRESET' }), 'network');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'transient', status: 503 }), 'server');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'request', message: 'maximum context length exceeded' }), 'context_length');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'request', message: 'the policy floor refused it' }), 'refused');
++    assert.strictEqual(record.errorClassOf({ failureClass: 'request', message: 'invalid schema' }), 'invalid');
++
++    // TPS is null when there is too little to measure: under 8 output tokens or under 200 ms.
++    assert.strictEqual(record.tpsOf({ output: 20, streamed: true, firstDeltaAt: 1000, finishedAt: 2000 }), 20);
++    assert.strictEqual(record.tpsOf({ output: 7, streamed: true, firstDeltaAt: 1000, finishedAt: 2000 }), null);
++    assert.strictEqual(record.tpsOf({ output: 20, streamed: true, firstDeltaAt: 1000, finishedAt: 1199 }), null);
++    assert.strictEqual(record.tpsOf({ output: 20, streamed: false, firstDeltaAt: 1000, finishedAt: 2000 }), null);
++});
++
  (async () => { await t.run(); })();
ddiff --git a/test/events.test.js b/test/events.test.js
iindex 29a61a0..b82d7bf 100644
---- a/test/events.test.js
++++ b/test/events.test.js
@@@ -7,11 +7,12 @@ const contracts = require('openvibe-contracts');
  process.env.EVENTS_URL = 'http://127.0.0.1:9';           // unreachable: rows stay in the outbox
  process.env.OV_OAUTH_CLIENT_SECRET = 'ai-secret-for-tests';
  process.env.OV_NETWORK_INTERNAL_URL = 'http://127.0.0.1:9';
--const { boot, request, token, suite, tmpDir, ALL } = require('./helpers');
++const { boot, request, token, suite, tmpDir, seamServer, ALL } = require('./helpers');
  
  const t = suite('events');
  const live = token('live', ALL);
  let h;
++let seam;
  const queued = async () => (await h.db.prepare('SELECT envelope FROM event_outbox ORDER BY id').all()).map((r) => (typeof r.envelope === 'string' ? (typeof r.envelope === 'string' ? (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope) : r.envelope) : r.envelope));
  const valid = (env) => { const r = contracts.validate(`${env.event_type}@1`, env.payload); assert.ok(r.valid, `${env.event_type}: ${JSON.stringify(r.errors)}`); };
  // The ai.call.recorded schema rides with the contracts release that is not out yet: until it is, the shape is checked here.
@@@ -129,6 +130,56 @@ t.test('a refused own key queues ai.credential.failed, never the key (L36 Y8)',
      assert.ok(!JSON.stringify(ev).includes('sk-'), 'no key');
  });
  
++t.test('a streamed run says so, and how fast, on its ai.call.recorded', async () => {
++    const res = await fetch(`${h.base}/v1/chat/completions`, {
++        method: 'POST', headers: { Authorization: `Bearer ${live}`, 'Content-Type': 'application/json' },
++        body: JSON.stringify({ model: 'openvibe', stream: true, messages: [{ role: 'user', content: 'stream this answer' }] }),
++    });
++    assert.strictEqual(res.status, 200);
++    await res.text();
++    const usage = (await queued()).filter((e) => e.event_type === 'ai.call.recorded').at(-1);
++    validUsage(usage);
++    assert.strictEqual(usage.payload.streamed, true, 'the deltas went to the caller');
++    assert.strictEqual(usage.payload.operation, 'chat', 'the Chat Completions operation');
++    assert.strictEqual(usage.payload.task, 'ai.complete', 'the workflow it served');
++    assert.strictEqual(usage.payload.task_id, usage.payload.run_id, 'a run is its own task');
++    assert.strictEqual(usage.payload.cache_hit, false);
++    assert.ok(usage.payload.tps === null || usage.payload.tps > 0, 'TPS is measured or null, never negative');
++});
++
++t.test('a failed run carries the failed attempt class and the fallbacks before it', async () => {
++    seam = await seamServer(() => ({ status: 429, body: { error: { message: 'too many requests: rate limit', type: 'rate_limit_error' } } }));
++    const p = await request(h.base, 'POST', '/api/v1/providers', { tok: live, body: {
++        key: 'bad', kind: 'http', auth_mode: 'none', base_url: seam.url, default_model: 'bad-1',
++        capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000 } });
++    assert.strictEqual(p.status, 201, p.text);
++    const m = await request(h.base, 'POST', '/api/v1/models', { tok: live, body: {
++        provider_key: 'bad', model_key: 'bad-1', type: 'chat', cost: { in_per_mtok: 1, out_per_mtok: 1 } } });
++    assert.strictEqual(m.status, 201, m.text);
++    const routed = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: live, body: {
++        primary: { provider: 'bad', model: 'bad-1' }, fallbacks: [], timeout_ms: 5000 } });
++    assert.strictEqual(routed.status, 201, routed.text);
++    await request(h.base, 'POST', '/api/v1/providers/stub/disable', { tok: live });
++    try {
++        const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: {
++            workflow: 'ai.generate', input: { prompt: 'this will be rate limited' }, options: { cache: false } } });
++        assert.strictEqual(r.status, 201, r.text);
++        assert.strictEqual(r.body.run.status, 'failed', r.text);
++        const usage = (await queued()).find((e) => e.event_type === 'ai.call.recorded' && e.payload.run_id === r.body.run.id);
++        assert.ok(usage, 'the failed run has its usage record');
++        validUsage(usage);
++        assert.strictEqual(usage.payload.outcome, 'failed');
++        assert.strictEqual(usage.payload.error_class, 'rate_limited', '429 from the provider is rate_limited');
++        assert.ok(usage.payload.fallbacks >= 1, 'the failed offers before the answer are counted');
++        assert.strictEqual(usage.payload.operation, 'chat');
++        assert.strictEqual(usage.payload.task, 'ai.generate');
++        assert.strictEqual(usage.payload.task_id, r.body.run.id);
++    } finally {
++        await request(h.base, 'POST', '/api/v1/providers/stub/enable', { tok: live });
++        await seam.close();
++    }
++});
++
  t.test('close() stops the ai.run.* relay (unsent rows stay in the outbox)', async () => {
      const events = require('../server/events');
      const pending = (await events.status()).pending;
ddiff --git a/test/usage-calls.test.js b/test/usage-calls.test.js
nnew file mode 100644
iindex 0000000..7acd51c
---- /dev/null
++++ b/test/usage-calls.test.js
@@@ -0,0 +1,123 @@
++'use strict';
++/**
++ * POST /api/v1/usage/calls (Contracts 0.421.0, ai-metrics.md): a product reports the AI calls it made without
++ * OpenVibe.AI. AI validates ai.call-report@1, keeps a subject only for the person its token acts for, dedupes by
++ * system + run_id/turn_id, prices an unpriced call from its own rate cards, publishes ai.call.recorded@1 and
++ * refuses a token without ai.usage.report.
++ */
++const assert = require('assert');
++
++process.env.EVENTS_URL = 'http://127.0.0.1:9';           // unreachable: the published events stay in the outbox
++process.env.OV_OAUTH_CLIENT_SECRET = 'ai-secret-for-tests';
++process.env.OV_NETWORK_INTERNAL_URL = 'http://127.0.0.1:9';
++
++const { boot, request, token, suite, ALL } = require('./helpers');
++
++const t = suite('usage-calls');
++const USR = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ';
++const actor = token('actor', ['ai.usage.report'], { extra: { on_behalf_of: USR } });
++const plain = token('codes', ['ai.usage.report']);
++const noReport = token('codes', ALL);
++const admin = token('live', ALL);
++let h;
++
++const calls = async () => (await h.db.prepare("SELECT envelope FROM event_outbox WHERE envelope->>'event_type' = 'ai.call.recorded' ORDER BY id").all())
++    .map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope).payload);
++const at = () => new Date().toISOString();
++
++t.test('boot with events on and a priced provider/model', async () => {
++    h = await boot({ env: { AI_STUB_FALLBACK: 'true' } });
++    const p = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: {
++        key: 'priced', kind: 'http', auth_mode: 'none', base_url: 'http://127.0.0.1:1', default_model: 'm1',
++        capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000 } });
++    assert.strictEqual(p.status, 201, p.text);
++    const m = await request(h.base, 'POST', '/api/v1/models', { tok: admin, body: {
++        provider_key: 'priced', model_key: 'm1', type: 'chat',
++        cost: { in_per_mtok: 1, out_per_mtok: 2, cached_per_mtok: 0.1, source: 'urn:openvibe:ai:test', verified_at: '2026-01-01' } } });
++    assert.strictEqual(m.status, 201, m.text);
++});
++
++t.test('a valid report is accepted, keeps the subject its token acts for, and is priced from the rate card', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: actor, body: { calls: [{
++        system: 'actor', provider: 'priced', model: 'm1', outcome: 'succeeded', at: at(), run_id: 'run_REPORT0000000000000000001',
++        subject: USR, attribution: 'site:actor', surface: 'actor.turn', paid_by: 'openvibe_key',
++        input: 1000000, output: 1000000, operation: 'chat', streamed: true, tps: 22.5, queue_ms: 7,
++        task: 'actor.turn', task_id: 'actor_task_1', fallbacks: 1, place: 'provider',
++    }] } });
++    assert.strictEqual(r.status, 202, r.text);
++    assert.deepStrictEqual(r.body, { accepted: 1, duplicates: 0, rejected: [] });
++    const ev = (await calls()).find((p) => p.run_id === 'run_REPORT0000000000000000001');
++    assert.ok(ev, 'the accepted call was published as ai.call.recorded@1');
++    assert.strictEqual(ev.subject, USR, 'a service acting for the person keeps their subject');
++    assert.strictEqual(ev.input, 1000000);
++    assert.strictEqual(ev.list_cost_usd, 3, '1M in at $1/M and 1M out at $2/M');
++    assert.strictEqual(ev.charged_usd, 3, 'the OpenVibe key pays the list cost');
++    assert.strictEqual(ev.price_source, 'catalog', 'AI priced it from its own rate card');
++    assert.strictEqual(ev.price_date, '2026-01-01');
++    assert.strictEqual(ev.operation, 'chat');
++    assert.strictEqual(ev.streamed, true);
++    assert.strictEqual(ev.tps, 22.5);
++    assert.strictEqual(ev.queue_ms, 7);
++    assert.strictEqual(ev.task, 'actor.turn');
++    assert.strictEqual(ev.task_id, 'actor_task_1');
++    assert.strictEqual(ev.fallbacks, 1);
++    assert.strictEqual(ev.place, 'provider');
++    assert.strictEqual(ev.cache_hit, false);
++});
++
++t.test('the same report twice is recorded once and counted as a duplicate', async () => {
++    const body = { calls: [{
++        system: 'actor', provider: 'priced', model: 'm1', outcome: 'succeeded', at: at(), run_id: 'run_REPORT0000000000000000002',
++        input: 10, output: 10, paid_by: 'openvibe_key',
++    }] };
++    const first = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: actor, body });
++    assert.deepStrictEqual(first.body, { accepted: 1, duplicates: 0, rejected: [] }, first.text);
++    const second = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: actor, body });
++    assert.strictEqual(second.status, 202, second.text);
++    assert.deepStrictEqual(second.body, { accepted: 0, duplicates: 1, rejected: [] });
++    assert.strictEqual((await calls()).filter((p) => p.run_id === 'run_REPORT0000000000000000002').length, 1, 'one event, not two');
++    // Another service's report with the same ids is its own call: one product can never block another's.
++    const other = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: plain, body });
++    assert.deepStrictEqual(other.body, { accepted: 1, duplicates: 0, rejected: [] }, other.text);
++});
++
++t.test('a token that does not act for the person keeps no subject; no rate card means unknown with zeros', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: plain, body: { calls: [{
++        system: 'codes', provider: 'no-such-provider', model: 'no-such-model', outcome: 'succeeded', at: at(),
++        turn_id: 'turn_REPORT0000000000000000003', subject: USR, input: 100, output: 100,
++    }] } });
++    assert.strictEqual(r.status, 202, r.text);
++    assert.strictEqual(r.body.accepted, 1, r.text);
++    const ev = (await calls()).find((p) => p.turn_id === 'turn_REPORT0000000000000000003');
++    assert.ok(ev, 'the turn was published');
++    assert.strictEqual(ev.subject, null, 'a service that does not act for the person may not name them');
++    assert.strictEqual(ev.price_source, 'unknown');
++    assert.deepStrictEqual([ev.list_cost_usd, ev.charged_usd, ev.fee_usd, ev.cache_saved_usd], [0, 0, 0, 0]);
++});
++
++t.test('an invalid call is rejected with its index; the valid ones are accepted', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: actor, body: { calls: [
++        { system: 'actor', provider: 'priced', model: 'm1', outcome: 'succeeded', at: at(), run_id: 'run_REPORT0000000000000000004' },
++        { system: 'actor', provider: 'priced', model: 'm1', outcome: 'made_up', at: at(), run_id: 'run_REPORT0000000000000000005' },
++    ] } });
++    assert.strictEqual(r.status, 202, r.text);
++    assert.strictEqual(r.body.accepted, 1);
++    assert.strictEqual(r.body.duplicates, 0);
++    assert.strictEqual(r.body.rejected.length, 1);
++    assert.strictEqual(r.body.rejected[0].index, 1, 'the rejected item names its index');
++    assert.ok(/outcome|enum/i.test(r.body.rejected[0].error), r.body.rejected[0].error);
++    assert.ok((await calls()).some((p) => p.run_id === 'run_REPORT0000000000000000004'));
++    assert.ok(!(await calls()).some((p) => p.run_id === 'run_REPORT0000000000000000005'));
++});
++
++t.test('a token without ai.usage.report is refused', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/usage/calls', { tok: noReport, body: { calls: [{
++        system: 'actor', provider: 'priced', model: 'm1', outcome: 'succeeded', at: at(), run_id: 'run_REPORT0000000000000000006',
++    }] } });
++    assert.strictEqual(r.status, 403, r.text);
++    assert.strictEqual(r.body.code, 'capability.denied');
++});
++
++t.test('close', async () => { await h.stop(); });
++
++t.run();