openvibers/openvibe.ai Commits

Emit one platform.usage-sample@1 reading per accounted run to Billing (#13)

OP OpenVibers committed 4 Oct 2026 · 5abaa65b35 · 12 files, +378 −4

checks run by GitHub CI (test, test-pg) · CI ✓ 3/3 · review SHIP by DeepSeek harness · 1 fix round

https://github.com/OpenVibers/OpenVibe.AI/pull/13

Changes

ddiff --git a/.env.example b/.env.example
iindex 404bd77..5e8dbaf 100644
---- a/.env.example
++++ b/.env.example
@@@ -47,6 +47,13 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
  #AI_CONSOLE_SESSION_TTL_MIN=60
  # Audience a Network user token must carry
  #AI_SSO_AUDIENCE=openvibe.network
++# ── Billing (plan T5 lane F) ──
++# One platform.usage-sample@1 reading per accounted run, posted to billing.usage.record with AI's
++# service token, so Network must grant client `ai` billing.usage.record on openvibe.billing. Readings
++# are queued from the start and sent while OV_BILLING_INTERNAL_URL is set; USAGE_SAMPLES=off stops queueing them.
++#OV_BILLING_INTERNAL_URL=http://127.0.0.1:4600
++#USAGE_SAMPLES=on
++#USAGE_SAMPLES_INTERVAL_MS=2000
  
  # ── A person's own provider keys (server/credentials.js, WS-O task 2) ──
  # 64 hex characters (openssl rand -hex 32). Keys stored through PUT /api/v1/credentials/:subject are AES-256-GCM
ddiff --git a/docs/cutover-pr-openvibe-ai-13.md b/docs/cutover-pr-openvibe-ai-13.md
nnew file mode 100644
iindex 0000000..a1f4074
---- /dev/null
++++ b/docs/cutover-pr-openvibe-ai-13.md
@@@ -0,0 +1,108 @@
++# Cutover — AI usage readings to Billing (`migrations/0005_usage_samples.sql`, PR #13)
++
++The change: every accounted run writes one `platform.usage-sample@1` reading into a retry-safe
++outbox, and a relay posts it to Billing's `billing.usage.record` (`POST /api/v1/usage`, audience
++`openvibe.billing`). Details in `server/usage-samples.js`.
++
++## What the migration changes
++
++`migrations/0005_usage_samples.sql` is an **expand** migration, additive and transactional:
++
++- a new table `usage_sample_outbox` (one row per reading) and two partial indexes,
++  `usage_sample_outbox_due` and `usage_sample_outbox_sent`;
++- one new **nullable** column `runs.usage_sample_id text COLLATE "C"`, naming the reading a run
++  produced.
++
++No existing row is rewritten, no column gains `NOT NULL`, no data moves. The previous release
++(N-1) neither reads the table nor the column, so a rollback to it needs no data change. Both
++writers are the AI service itself: there is no import, no dual-write window and no backfill.
++
++## Order
++
++1. **Preconditions (pre-merge).** CI green on the branch head; the PR body carries the `cutover`
++   manifest; the rehearsal below recorded green. Billing does **not** have to ship first: a missing
++   route, a missing grant (`401`/`403`/`404`) or a `429`/`5xx` leaves the row queued and retried
++   with backoff across restarts; only `400`/`409`/`413`/`422` mark a reading rejected.
++
++2. **Backup (before anything touches the schema).**
++   `ov access run openvibe-ovh db-backup ai` — ovhost takes a backup of AI's declared database and
++   logs it. Keep the backup id/artifact for the deployment until the verify step is signed off.
++   By hand (owner role, same result): `pg_dump --format=custom --no-owner "$DATABASE_DIRECT_URL" >
++   ai-<utc-timestamp>.dump`, then `pg_restore --list` on it to prove it is complete. This is the
++   only backup the cutover needs; the migration itself loses nothing.
++
++3. **Rehearse on a copy.** Restore the backup into a scratch database (`pg_restore` into
++   `ai-rehearsal`, or copy `/…/data/pglite`), start the release against it as the owner
++   (`DATABASE_URL`/`DATABASE_DIRECT_URL` pointed at the copy, e.g. `USAGE_SAMPLES=on node
++   server/index.js`), and run the verify checks below against the copy. The migration is applied by
++   the service at boot (`server/db.js` `openDb()`), as the owner on the direct connection, in a
++   transaction, serialised by the SDK's advisory lock, and recorded in `ov_migrations` with a
++   checksum. Whoever runs this writes the marker `ds/deploy/rehearsals/ai-usage-samples-0005.json`
++   as `{"ok": true}` — the author never writes it.
++
++4. **Apply in production.** Merged work deploys through its pipeline (the recipe runs with every
++   guard). The migration runs on the next service start, in the same boot path as the rehearsal.
++   No manual `psql` step.
++
++5. **Verify** (see below). Sign the deploy off only when both the schema checks and one end-to-end
++   reading pass.
++
++## How the result is verified
++
++Schema, as the owner on the direct connection:
++
++```sql
++SELECT id, name, phase, applied_at FROM ov_migrations WHERE id = '0005';   -- one row, phase expand
++\d usage_sample_outbox                                                     -- table + the two indexes
++SELECT usage_sample_id FROM runs LIMIT 1;                                  -- column exists, nullable
++```
++
++Service and follow-through:
++
++- `ov access run openvibe-ovh health ai` → ready (`GET /api/ready`).
++- `ov access run openvibe-ovh journal openvibe-ai.service 200` → no migration error; at most a
++  `[Billing] usage sample send failed (will retry)` line while Billing is unavailable.
++- One accounted run end to end: with `USAGE_SAMPLES` on and the relay configured
++  (`OV_BILLING_INTERNAL_URL`, `OV_OAUTH_CLIENT_SECRET`), drive one workflow run, then
++
++  ```sql
++  SELECT event_id, attempts, sent_at, last_error FROM usage_sample_outbox ORDER BY id DESC LIMIT 5;
++  SELECT usage_sample_id FROM runs WHERE id = '<run id>';
++  ```
++
++  The reading's id is `ai:<run id>:tokens`, the same value lands in `runs.usage_sample_id`, and
++  `sent_at` is set with `last_error` null when Billing accepted it. If Billing is not deployed yet,
++  rows waiting with `attempts`/`last_error` are the expected state, not a failure — they drain when
++  it is. A person's own key, and runs that called no provider without a cache hit, produce no
++  reading.
++- The envelope carries ids only (`service`, `subject`, `resource`, `operation`, `quantity`, `unit`,
++  `cost_estimate`, `provider`, `trace_id`) — never the input, the output or a prompt. Spot-check one:
++  `SELECT envelope FROM usage_sample_outbox ORDER BY id DESC LIMIT 1;`.
++
++## Way back
++
++- **Roll back the release (the normal way).** The migration is expand and additive and N-1 ignores
++  both objects, so `ov access run openvibe-ovh rollback ai` (owner) is enough; the schema stays and
++  is harmless. Readings queued by the failed release simply wait.
++- **Stop the flow without changing code or schema:** set `USAGE_SAMPLES=off` and restart — nothing
++  is queued while it is off; rows already queued stay until they are sent.
++- **Only if the objects themselves must go** (not required for a rollback), with the owner's
++  approval and the backup in hand, after the previous release is running:
++
++  ```sql
++  DROP TABLE IF EXISTS usage_sample_outbox;
++  ALTER TABLE runs DROP COLUMN IF EXISTS usage_sample_id;
++  DELETE FROM ov_migrations WHERE id = '0005';
++  ```
++
++  The only loss is readings not yet sent; `usage_daily` and the run ledger are untouched.
++- **Never edit `0005_usage_samples.sql` in place:** `ov_migrations` records its checksum and the
++  runner refuses a changed file. Undo with new SQL (or a later migration), not by rewriting it.
++
++## Notes
++
++- Not a `migrate`/`contract` pair: nothing depends on the new objects and nothing needs the N-1
++  window, so no follow-up migration is scheduled.
++- `deploy/` needs no change: the service already reads `DATABASE_DIRECT_URL` for migrations, and
++  `.env.example` documents the `OV_BILLING_INTERNAL_URL` / `USAGE_SAMPLES` /
++  `USAGE_SAMPLES_INTERVAL_MS` names.
ddiff --git a/migrations/0005_usage_samples.sql b/migrations/0005_usage_samples.sql
nnew file mode 100644
iindex 0000000..cac43c8
---- /dev/null
++++ b/migrations/0005_usage_samples.sql
@@@ -0,0 +1,21 @@
++-- phase: expand
++-- T5 lane F: AI → Billing usage readings (platform.usage-sample@1, server/usage-samples.js). A second outbox with the
++-- same shape as event_outbox (openvibe-sdk outboxSchema). Its relay posts each row to billing.usage.record. event_id is
++-- the reading's idempotency_key (ai:<run id>:tokens). runs.usage_sample_id names the reading a run produced.
++-- Additive only: a new table and a nullable column; no existing row is rewritten.
++CREATE TABLE IF NOT EXISTS usage_sample_outbox (
++    id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    event_id        text NOT NULL UNIQUE,
++    envelope        jsonb NOT NULL,
++    traceparent     text,
++    created_at      bigint NOT NULL,
++    attempts        integer NOT NULL DEFAULT 0,
++    next_attempt_at bigint NOT NULL DEFAULT 0,
++    sent_at         bigint,
++    seq             bigint,
++    rejected_at     bigint,
++    last_error      text
++);
++CREATE INDEX IF NOT EXISTS usage_sample_outbox_due ON usage_sample_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS usage_sample_outbox_sent ON usage_sample_outbox (sent_at) WHERE sent_at IS NOT NULL;
++ALTER TABLE runs ADD COLUMN usage_sample_id text COLLATE "C";
ddiff --git a/server/api/runs.js b/server/api/runs.js
iindex c8493de..b92404d 100644
---- a/server/api/runs.js
++++ b/server/api/runs.js
@@@ -5,6 +5,7 @@
   *   POST /api/v1/runs                       create (ai.run.create); ?wait=ms waits for the result
   *   GET  /api/v1/runs                       the caller's runs (ai.run.read)
   *   GET  /api/v1/runs/:id                   one run (+ citations, request log metadata)
++ *   run.usage.usage_sample_id: the run's Billing reading (server/usage-samples.js), null when none
   *   POST /api/v1/runs/:id/cancel            (ai.run.create, owner)
   *   POST /api/v1/runs/:id/retry             (ai.run.create, owner) -> a new run with retry_of
   *   GET  /api/v1/runs/:id/citations         (ai.run.read)
ddiff --git a/server/config.js b/server/config.js
iindex 9c6f5f8..e150603 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -193,6 +193,18 @@ function load(env = process.env) {
              maxImageBytes: int(env.AI_FETCH_MAX_IMAGE_BYTES, 12 * 1024 * 1024),
              timeoutMs: int(env.AI_FETCH_TIMEOUT_MS, 120000),
          },
++
++        // ── Billing (plan T5 lane F): one platform.usage-sample@1 reading per accounted run, posted by
++        // server/usage-samples.js to billing.usage.record (POST /api/v1/usage, audience openvibe.billing).
++        // Readings are queued unless USAGE_SAMPLES=off; they are sent while OV_BILLING_INTERNAL_URL is set.
++        // No `capability:` key: AI asks for billing.usage.record (named in usage-samples.js), Billing enforces it,
++        // and openvibe-contracts-check reads a `capability: 'x'` literal as one this service enforces.
++        billing: {
++            enabled: bool(env.USAGE_SAMPLES, true),
++            billingUrl: trimUrl(env.OV_BILLING_INTERNAL_URL),
++            audience: 'openvibe.billing',
++            intervalMs: Math.max(100, int(env.USAGE_SAMPLES_INTERVAL_MS, 2000)),
++        },
      };
  }
  
ddiff --git a/server/console/views.js b/server/console/views.js
iindex f70e882..ada4efc 100644
---- a/server/console/views.js
++++ b/server/console/views.js
@@@ -346,6 +346,7 @@ function run({ staff, csrf, r, citations, requests, input, output, canCancel, no
  <dt>Attempts</dt><dd>${n(r.usage.attempts)}</dd>
  <dt>Tokens</dt><dd>${n(r.usage.tokens_in)} in · ${n(r.usage.tokens_out)} out</dd>
  <dt>Cost</dt><dd>${usd(r.usage.cost_usd)}</dd>
++<dt>Billing reading</dt><dd>${r.usage.usage_sample_id ? html`<code>${r.usage.usage_sample_id}</code>` : '—'}</dd>
  ${r.error ? html`<dt>Error</dt><dd><code class="bad">${r.error.code}</code><br>${clip(r.error.detail, 500)}</dd>` : ''}
  </dl></div><div class="card"><dl>
  <dt>Requester</dt><dd><code>${r.requester.type}:${r.requester.id}</code></dd>
ddiff --git a/server/index.js b/server/index.js
iindex 6f5b126..09fe90a 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -36,6 +36,8 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
      const engine = createEngine({ registry, pool, fetcher, quotas, config, log });
      // ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
      require('./events').init(db, { log });
++    // platform.usage-sample@1 readings to OpenVibe.Billing through their own outbox (server/usage-samples.js).
++    require('./usage-samples').init(db, { ...config.billing, clientId: config.console.clientId, networkUrl: config.networkInternalUrl, log });
      // ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).
      const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });
      // A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.
@@@ -92,6 +94,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
          }
          await runs.drain();
          await require('./events').stop();   // after the runs: their last ai.run.* rows are queued first
++        await require('./usage-samples').stop();   // the readings the runs queued get one last send; unsent rows wait for the next start
          const w = await pool.adapter('whisper');
          if (w && w.adapter.killActive) w.adapter.killActive();
          if (!givenDb) await db.close();
ddiff --git a/server/quota.js b/server/quota.js
iindex 848d674..27ac894 100644
---- a/server/quota.js
++++ b/server/quota.js
@@@ -14,6 +14,7 @@
   * request once a window's spend has reached them, the same way Live's daily USD cap worked).
   */
  const { AiError, iso } = require('./util');
++const usageSamples = require('./usage-samples');
  
  const WINDOWS = { minute: 60, hour: 3600, day: 86400 };
  
@@@ -170,13 +171,17 @@ function createQuotas(db, { clock = { now: () => Date.now() }, registry } = {})
       * Add real usage to the reserved windows, the daily usage table and — in the same transaction — the
       * provider rollup and per-route placement state. `attempts` is the run's finished attempts (the same
       * entries logged to `requests`); `writeUsage` is false when a run spent nothing (stats are still kept).
++     * `runId`/`traceId` name the run a reading is written for (server/usage-samples.js): with a runId the
++     * run's platform.usage-sample@1 reading commits with its usage.
       */
--    async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true }) {
++    async function account(ctx, reserved, { provider, model, tokensIn = 0, tokensOut = 0, tokensCached = 0, cost = 0, attempts = null, writeUsage = true, runId = null, traceId = null }) {
          const tokens = tokensIn + tokensOut;
          const day = iso(clock.now()).slice(0, 10);
          await db.tx(async () => {
              for (const r of reserved || []) await bump.run(r.scope_type, r.scope_id, r.window, r.ws, r.workflow_prefix, 0, tokens, cost);
              if (writeUsage) await insDaily.run(day, `${ctx.requesterType}:${ctx.requesterId}`, ctx.attributionKey || '', ctx.workflowKey, provider || '', model || '', tokensIn, tokensOut, tokensCached, cost);
++            // The run's platform.usage-sample@1 reading for Billing commits with its usage (server/usage-samples.js).
++            if (writeUsage && runId) await usageSamples.record(db, { runId, workflowKey: ctx.workflowKey, requester: `${ctx.requesterType}:${ctx.requesterId}`, provider, tokensIn, tokensOut, cost, at: clock.now(), traceId });
              await recordStats(day, attempts);
          });
      }
ddiff --git a/server/runs.js b/server/runs.js
iindex 16a3233..10c045c 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -21,6 +21,7 @@ const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');
  const { preferenceLines } = require('./user-modules');
  const schemas = require('./schemas');
  const events = require('./events');
++const usageSamples = require('./usage-samples');
  
  const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);
  const QUEUE_RETRY_AFTER_S = 5;
@@@ -84,7 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  origin: 'ai', workflow: r.workflow_key, workflow_version: r.workflow_version, template_version: r.template_version, route: r.route_key, route_version: r.route_version,
                  provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),
              },
--            usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts },
++            usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_id: r.usage_sample_id || null },
              explain: parseJson(r.explain, null),
              citations_count: r.citations_count,
              grounding: parseJson(r.grounding, null),
@@@ -214,6 +215,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                      if (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
                      // The reused output carries the grounding it was produced with.
                      await db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);
++                    // A cache hit is still a served run: its reading (no tokens, no cost) goes to Billing with it.
++                    await usageSamples.record(db, { runId: id, workflowKey: wf.key, requester: `${requester.type}:${requester.id}`, at: now, traceId: trace });
                      await events.runChanged(await getRow(id));
                  });
                  await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
@@@ -312,7 +315,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  return d;
              });
              if (!done.changes) return;       // cancelled while finishing
--            if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost, attempts: logged });
++            if (r.provider && reserved) await quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost, attempts: logged, runId: id, traceId: row.trace_id });
              if (r.fallbackUsed) await registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
              // Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
              if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
@@@ -335,7 +338,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              // 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.
--            if (reserved && logged.some((e) => e.status !== 'skipped')) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c, attempts: logged, writeUsage: Boolean(spent.ti || spent.tout) });
++            if (reserved && logged.some((e) => e.status !== 'skipped')) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c, attempts: logged, writeUsage: Boolean(spent.ti || spent.tout), runId: id, traceId: row.trace_id });
          } finally {
              inflight.delete(id);
          }
ddiff --git a/server/usage-samples.js b/server/usage-samples.js
nnew file mode 100644
iindex 0000000..4f014c2
---- /dev/null
++++ b/server/usage-samples.js
@@@ -0,0 +1,73 @@
++'use strict';
++/**
++ * AI → OpenVibe.Billing: one platform.usage-sample@1 reading per accounted run, posted to
++ * billing.usage.record (POST /api/v1/usage, audience openvibe.billing) through a retry-safe outbox.
++ *
++ * quota.js account() writes the reading in the SAME transaction as usage_daily. A cache hit writes its
++ * zero reading in the transaction that records the run. The relay (openvibe-sdk createPgOutbox) posts
++ * each row with AI's service token. If Billing is down, has no grant yet, or is not deployed, the row
++ * waits and is retried with backoff, across restarts. Only Billing refusing the reading itself (400, 409,
++ * 413, 422) marks a row rejected. A reading carries ids, the workflow, the requester, tokens and cost. It
++ * never carries the input, the output or a prompt. Readings are queued once init() ran, unless
++ * USAGE_SAMPLES=off. The relay runs while OV_BILLING_INTERNAL_URL and OV_OAUTH_CLIENT_SECRET are set.
++ * Runs on a person's own key, and runs that called no provider and were not a cache hit, are not
++ * accounted and have no reading.
++ */
++const { createPgOutbox } = require('openvibe-sdk/events');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++const contracts = require('openvibe-contracts');
++const TABLE = 'usage_sample_outbox'; const METRIC = 'tokens'; const SCHEMA = 'platform.usage-sample@1';
++const REFUSED = new Set([400, 409, 413, 422]); const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
++let relay = null; let on = false; let pruneTimer = null; let lastError = null; let queued = 0;
++function sampleOf({ runId, workflowKey, requester, provider = null, tokensIn = 0, tokensOut = 0, cost = 0, at, traceId = null }) {
++  const key = `ai:${runId}:${METRIC}`;
++  const s = { id: key, idempotency_key: key, service: 'ai', subject: requester, resource: runId, operation: workflowKey,
++    quantity: Math.max(0, (Number(tokensIn) || 0) + (Number(tokensOut) || 0)), unit: 'tokens', at: new Date(at).toISOString(),
++    cost_estimate: Math.max(0, Number(cost) || 0), source: 'ai.runs' };
++  if (provider && provider !== 'mixed') s.provider = provider;
++  if (traceId) s.trace_id = String(traceId);
++  return s;
++}
++/** MUST be awaited inside the transaction that accounts the run; a second call for the same run changes nothing. */
++async function record(db, fields) {
++  if (!on) return null;
++  const s = sampleOf(fields);
++  const v = contracts.validate(SCHEMA, s);
++  if (!v.valid) throw new Error(`usage sample for ${fields.runId} is not a valid ${SCHEMA}: ${JSON.stringify(v.errors)}`);
++  const ins = await db.prepare(`INSERT INTO ${TABLE} (event_id, envelope, created_at) VALUES (?, ?, ?) ON CONFLICT (event_id) DO NOTHING`).run(s.id, JSON.stringify(s), Date.now());
++  await db.prepare('UPDATE runs SET usage_sample_id = ? WHERE id = ?').run(s.id, fields.runId);
++  if (ins.changes) queued++;
++  if (relay) db.afterCommit(() => relay && relay.kick());
++  return s.id;
++}
++// createPgOutbox's `events` seam pointed at Billing: one reading per POST. isPermanent() reads err.status.
++function billingSink({ billingUrl, tokens, audience, fetchImpl, timeoutMs = 5000 }) {
++  return { prepare: (sample) => sample, async publish(sample) {
++    const token = await tokens.getToken({ audience });
++    const res = await fetchImpl(`${billingUrl}/api/v1/usage`, { method: 'POST', headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', Accept: 'application/json' }, body: JSON.stringify(sample), signal: AbortSignal.timeout(timeoutMs) });
++    if (!res.ok) { const err = new Error(`billing.usage.record answered ${res.status}`); err.status = REFUSED.has(res.status) ? 422 : 503; throw err; }   // 401/403 before the grant, 404 before Billing ships the route, 429, 5xx: retried
++    return { event_id: sample.id, seq: null };
++  } };
++}
++function init(db, { enabled = true, billingUrl = '', audience = 'openvibe.billing', capability = 'billing.usage.record', clientId = process.env.OV_OAUTH_CLIENT_ID || 'ai', clientSecret = process.env.OV_OAUTH_CLIENT_SECRET, networkUrl = process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000', fetchImpl = globalThis.fetch, intervalMs = 2000, log = console } = {}) {
++  if (relay) return relay;
++  on = Boolean(enabled);
++  if (!on || !billingUrl || !clientSecret) return null;
++  const tokens = createServiceTokenClient({ tokenUrl: `${String(networkUrl).replace(/\/+$/, '')}/oauth/token`, clientId, clientSecret, scope: { [audience]: capability }, fetch: fetchImpl });
++  relay = createPgOutbox(db, { events: billingSink({ billingUrl: String(billingUrl).replace(/\/+$/, ''), tokens, audience, fetchImpl }), table: TABLE, batchSize: 1, intervalMs,
++    onError: (err) => { const msg = err && err.message; if (msg !== lastError) log.warn(`[Billing] usage sample send failed (will retry): ${msg}`); lastError = msg; } });
++  relay.start();
++  pruneTimer = setInterval(() => { if (relay) relay.prune().catch((err) => log.warn('[Billing] usage outbox prune failed:', err && err.message)); }, PRUNE_EVERY_MS);
++  if (pruneTimer.unref) pruneTimer.unref();
++  log.log(`[Billing] ai usage samples → ${billingUrl}`);
++  return relay;
++}
++async function status() { if (!relay) return { enabled: on, relay: false }; return { enabled: on, relay: true, pending: Number(await relay.pending()), rejected: Number(await relay.rejected()), queued_since_boot: queued, last_error: lastError }; }
++/** Graceful stop (after runs.drain(), before db.close()): one last bounded pass sends what is due, then the relay stops; unsent rows wait for the next start. */
++async function stop({ drainMs = 3000 } = {}) {
++  if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null;
++  const r = relay; relay = null; on = false;
++  if (r) { let timer; await Promise.race([r.flush().catch(() => {}), new Promise((res) => { timer = setTimeout(res, drainMs); if (timer.unref) timer.unref(); })]); clearTimeout(timer); await r.stop(); }
++}
++function _reset() { if (pruneTimer) clearInterval(pruneTimer); pruneTimer = null; if (relay) relay.stop(); relay = null; on = false; queued = 0; lastError = null; }
++module.exports = { init, record, sampleOf, status, stop, _reset, TABLE };
ddiff --git a/test/quota.test.js b/test/quota.test.js
iindex cb8a867..ac360da 100644
---- a/test/quota.test.js
++++ b/test/quota.test.js
@@@ -44,6 +44,14 @@ t.test("'*' service quotas give each service its own window", async () => {
      assert.strictEqual(r.status, 201, 'another service is not limited by live\'s usage');
  });
  
++t.test('every accounted run has one Billing reading; the refused request queued none', async () => {
++    const runs = await h.db.prepare("SELECT id, usage_sample_id FROM runs WHERE status = 'succeeded'").all();
++    assert.ok(runs.length > 0, 'runs succeeded');
++    for (const r of runs) assert.strictEqual(r.usage_sample_id, `ai:${r.id}:tokens`, `no reading for ${r.id}`);
++    const n = Number((await h.db.prepare('SELECT COUNT(*) n FROM usage_sample_outbox').get()).n);
++    assert.strictEqual(n, runs.length, 'one reading per succeeded run; the 429 request queued none');
++});
++
  t.test('attribution quotas limit one attributed owner (Live per-streamer budget)', async () => {
      await request(h.base, 'POST', '/api/v1/quotas', { tok: live, body: { scope_type: 'service', scope_id: '*', window: 'minute', max_requests: 1000 } });
      const q = await request(h.base, 'POST', '/api/v1/quotas', { tok: live, body: { scope_type: 'attribution', scope_id: 'live:user:42', window: 'day', max_requests: 1 } });
ddiff --git a/test/usage-sample.test.js b/test/usage-sample.test.js
nnew file mode 100644
iindex 0000000..0878023
---- /dev/null
++++ b/test/usage-sample.test.js
@@@ -0,0 +1,132 @@
++'use strict';
++// platform.usage-sample@1 readings: one per accounted run, queued in the same transaction as the run's
++// usage and posted to billing.usage.record by a retry-safe outbox (server/usage-samples.js). A reading
++// carries ids, tokens and cost — never the prompt, the input or the output.
++const assert = require('assert');
++const contracts = require('openvibe-contracts');
++const { boot, request, token, suite, seamServer, tmpDir, ALL } = require('./helpers');
++
++const t = suite('usage-sample');
++const live = token('live', ALL);
++const usageSamples = require('../server/usage-samples');
++let h;
++let seam;
++let b;   // the Billing stub the live relay posts through
++const env = { AI_STUB_FALLBACK: 'false' };
++
++const readings = async () => (await h.db.prepare('SELECT event_id, envelope, sent_at, attempts FROM usage_sample_outbox ORDER BY id').all())
++    .map((r) => ({ ...r, envelope: typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope }));
++
++function fakeBilling() {
++    const b = { fail: false, got: [] };
++    b.fetch = async (url, init = {}) => {
++        if (String(url).endsWith('/oauth/token')) {
++            return new Response(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300, scope: 'billing.usage.record' }), { status: 200, headers: { 'Content-Type': 'application/json' } });
++        }
++        if (String(url).endsWith('/api/v1/usage')) {
++            if (b.fail) return new Response(JSON.stringify({ error: 'unavailable' }), { status: 503, headers: { 'Content-Type': 'application/json' } });
++            b.got.push(JSON.parse(init.body));
++            return new Response(JSON.stringify({ record: { id: 'use_1' } }), { status: 200, headers: { 'Content-Type': 'application/json' } });
++        }
++        return new Response('not found', { status: 404 });
++    };
++    return b;
++}
++
++t.test('boot with a counting seam provider as the only route target', async () => {
++    seam = await seamServer(() => ({ text: 'ok', usage: { input: 1000, output: 1000 } }));
++    h = await boot({ dir: tmpDir(), env });   // a dir: the database outlives h.stop()
++    await request(h.base, 'POST', '/api/v1/providers', { tok: live, body: { key: 'seam', kind: 'http', base_url: seam.url, auth_mode: 'none', capabilities: ['generate', 'chat'] } });
++    await request(h.base, 'POST', '/api/v1/models', { tok: live, body: { provider_key: 'seam', model_key: 'm1', cost: { in_per_mtok: 1000, out_per_mtok: 1000 } } });
++    const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: live, body: { primary: { provider: 'seam', model: 'm1' }, fallbacks: [] } });
++    assert.strictEqual(r.status, 201, r.text);
++});
++
++t.test('a succeeded run queues exactly one valid reading that keeps the prompt out', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'secret prompt words', options: { cache: false } } });
++    assert.strictEqual(r.status, 201, r.text);
++    const runId = r.body.run.id;
++    const rows = await readings();
++    assert.strictEqual(rows.length, 1, 'exactly one reading');
++    const s = rows[0].envelope;
++    const v = contracts.validate('platform.usage-sample@1', s);
++    assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
++    assert.strictEqual(s.idempotency_key, `ai:${runId}:tokens`);
++    assert.strictEqual(s.id, `ai:${runId}:tokens`);
++    assert.strictEqual(rows[0].event_id, s.idempotency_key);
++    assert.strictEqual(s.service, 'ai');
++    assert.strictEqual(s.operation, 'ai.generate');
++    assert.strictEqual(s.unit, 'tokens');
++    assert.strictEqual(s.quantity, 2000);
++    assert.ok(s.cost_estimate > 0, 'cost is carried');
++    assert.strictEqual(s.subject, 'service:live');
++    assert.ok(!JSON.stringify(s).includes('secret prompt'), 'no prompt');
++    const got = await request(h.base, 'GET', `/api/v1/runs/${runId}`, { tok: live });
++    assert.strictEqual(got.body.run.usage.usage_sample_id, `ai:${runId}:tokens`);
++});
++
++t.test('a cache hit writes its own zero reading', async () => {
++    const body = { prompt: 'cache me for a reading' };
++    const a = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body });
++    const b = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body });
++    assert.strictEqual(b.status, 201, b.text);
++    if (b.body.run.status !== 'cached') { assert.ok(a.body.run.id, 'this workflow may not cache'); return; }
++    const s = (await readings()).find((r) => r.event_id === `ai:${b.body.run.id}:tokens`).envelope;
++    const v = contracts.validate('platform.usage-sample@1', s);
++    assert.ok(v.valid, `platform.usage-sample@1: ${JSON.stringify(v.errors)}`);
++    assert.strictEqual(s.quantity, 0);
++    assert.strictEqual(s.cost_estimate, 0);
++});
++
++t.test('recording the same run again leaves one row, unchanged', async () => {
++    const runId = (await h.db.prepare("SELECT id FROM runs WHERE status = 'succeeded' ORDER BY id LIMIT 1").get()).id;
++    const key = `ai:${runId}:tokens`;
++    await h.db.tx(async () => { await usageSamples.record(h.db, { runId, workflowKey: 'ai.generate', requester: 'service:live', tokensIn: 1, tokensOut: 0, cost: 0, at: Date.now() }); });
++    const rows = (await readings()).filter((r) => r.event_id === key);
++    assert.strictEqual(rows.length, 1, 'one row');
++    assert.strictEqual(rows[0].envelope.quantity, 2000, 'the first reading is unchanged');
++});
++
++t.test('a failed send stays queued and replays', async () => {
++    usageSamples._reset();
++    b = fakeBilling();
++    b.fail = true;
++    const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });
++    await relay.flush();
++    // batchSize is 1 and a failed pass stops there: flush until every row has been attempted once.
++    for (let i = 0; i < 10 && (await readings()).some((r) => r.sent_at === null && r.attempts === 0); i++) await relay.flush();
++    const pending = await readings();
++    assert.ok(pending.length >= 1, 'rows exist');
++    for (const r of pending) { assert.strictEqual(r.sent_at, null, 'every row is unsent'); assert.ok(r.attempts >= 1, 'the send was attempted'); }
++    assert.strictEqual(b.got.length, 0, 'nothing accepted');
++    b.fail = false;
++    await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();
++    await relay.flush();
++    const sent = await readings();
++    for (const r of sent) assert.notStrictEqual(r.sent_at, null, 'every row is sent');
++    const keys = b.got.map((s) => s.idempotency_key).sort();
++    assert.deepStrictEqual(keys, sent.map((r) => r.event_id).sort());
++    for (const s of b.got) assert.ok(contracts.validate('platform.usage-sample@1', s).valid, JSON.stringify(s));
++});
++
++t.test('the outbox outlives a restart', async () => {
++    b.fail = true;   // the still-running relay fails every send
++    const r = await request(h.base, 'POST', '/api/v1/generate', { tok: live, body: { prompt: 'survive a restart', options: { cache: false } } });
++    assert.strictEqual(r.status, 201, r.text);
++    const key = `ai:${r.body.run.id}:tokens`;
++    await h.stop();
++    h = await boot({ dir: h.dir, env });
++    const row = await h.db.prepare('SELECT sent_at FROM usage_sample_outbox WHERE event_id = ?').get(key);
++    assert.ok(row, 'the reading is still there');
++    assert.strictEqual(row.sent_at, null, 'and unsent');
++    usageSamples._reset();
++    b = fakeBilling();
++    const relay = usageSamples.init(h.db, { billingUrl: 'http://billing.test', clientSecret: 's', networkUrl: 'http://network.test', fetchImpl: b.fetch, intervalMs: 60000 });
++    await h.db.prepare('UPDATE usage_sample_outbox SET next_attempt_at = 0').run();
++    await relay.flush();
++    assert.strictEqual(b.got.filter((s) => s.idempotency_key === key).length, 1, 'sent exactly once');
++});
++
++t.test('shutdown', async () => { usageSamples._reset(); await h.stop(); await seam.close(); });
++
++t.run();