openvibers/openvibe.sdk Commits
T1 lane A step 4 — OpenVibe.SDK: add the Fabric modules… (#4)
Changes
@@@ -9,6 +9,26 @@ release may change an API and says so here.## Unreleased++- **`openvibe-sdk/usage` (new, server; plan T1):** `platform.usage-sample@1` readings in one place.++ `usageSample(fields)` builds the record — `id`, `idempotency_key`, `service`, `operation`, `quantity`, `unit` and++ every optional dimension pass through, `at` defaults to now (ISO) and `source` to `openvibe-sdk/usage`, null fields++ are stripped and no money field is invented (`cost_estimate`, `free_allowance_used`, `vibes_charged` are the++ caller's; rating is Billing's). `usageKey(service, ...parts)` joins an idempotency key with `:` (the service alone++ when there are no parts; a non-empty service is enforced). `validateUsageSample(record)` → `{ ok, errors }`++ (`openvibe-contracts` is required lazily on first use; a missing package never claims validity: `ok: false`,++ `errors: []`). `openvibe-sdk/govern` now builds its `onUsage` record with `usageSample()`: the record is unchanged.++- **`openvibe-sdk/telemetry` (new, server; plan T1):** `platform.telemetry-sample@1`, the universal operation++ observation. `telemetrySample(fields)` and `validateTelemetrySample(record)` as in usage;++ `createTelemetry({ service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000 })` →++ `{ record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop() }`: a collector++ that buffers samples and flushes them to `await sink(samples)` every `intervalMs` (an unref'd timer) and on++ `stop()` (clears the timer, flushes once — a `gracefulStop` stop step). `record` builds++ `{ service, instance, operation, at, latency_ms, ...labels }`; a value that is not a number goes to `extra[name]`++ as a scalar. A sink that rejects is logged once and its samples kept for the next flush; past `maxBuffered` the++ oldest sample is dropped, with a warning at most once a minute. The SDK holds no network code: the sink is the service's.++- **`openvibe-sdk/placement`:** `TRUST_ORDER` gains `user-owned` (between `first-party` and `partner`; the T1 trust++ classes): a `user-owned` offer is eligible under `trust: ['user-owned']` and for workloads that name no trust list.++- The devDependency `openvibe-contracts` moves to v0.86.0 (it adds a `node` principal to`identity.service-token-claims`); the copied types and the lockfile integrity hash follow.- **`openvibe-sdk/govern` emits a full usage record** (plan T5). `createGovernor({ service, provider?, resource?, region? })`; `reserve()` takes optional `operation`, `provider`, `resource`, `region`, `trace_id` and `route_epoch`, which override the defaults. `onUsage` now gets a `platform.usage-sample@1` reading: `id` (the reservation id), `idempotency_key`, `service`, `project`, `subject`, `resource`, `provider`, `region`, `operation` (default `reserve`), `quantity`, `unit`, `at`, `route_epoch`, `trace_id`, `source` (`openvibe-sdk/govern`); absent values are left out. **Changed:** `amount` is now `quantity` and `state` is gone (the schema allows neither; only new reservations are emitted). No money fields: rating is Billing's. A governor created without `service` still works and logs one warning (`log`, default `console`). Tested against openvibe-contracts 0.86.0.@@@ -181,6 +181,8 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client| `openvibe-sdk/geo` | both | `createGeoClient(client, {samples?, timeoutMs?})`: `nodes({role, region})` from Network's public node registry, `measure(nodes)` (beacon round trips, fastest first), `nearest({role, region, preferRegion})` → `{node, rtt_ms, measured}` (up/degraded nodes only; falls back to preferRegion when nothing can be measured) || `openvibe-sdk/openre` | server | `createOpenReClient(client, {baseUrl?, publicUrl?})`: OpenRe.Stream, the streaming engine. `streams` (`byExternalRef`, `create`, `get`, `update`, `rotateKey`, `destinations`, `addDestination`), `destinations` (`test`, `start`, `stop`, `logs`), `sessions` (`get`, `playback` cached 10 s, `end`, `outputs`), `workers`, `manageUrl`; `{subject}` acts for a person (X-OV-Subject) || `openvibe-sdk/vip` | server | `createVipClient({baseUrl?, tokenClient \| getToken, fetch?, timeoutMs?})`: `evaluate({subject, resource, owner, fallback, mode})`, `checkEntitlement({subject, creator, product, mode})`, `isMember(subject, creator)`; every failure is a denial. `createVipCache({vip, ttlMs, denyTtlMs, unavailableTtlMs})`: `entitlement`, `peekEntitlement`, `evaluate`, `invalidate`, `handleEvent` (`vip.membership.changed` and Billing's entitlement events drop a member's answers at once), `clear`, `bounds` |++| `openvibe-sdk/usage` | server | `usageSample(fields)` builds a `platform.usage-sample@1` record (required `id`, `idempotency_key`, `service`, `operation`, `quantity`, `unit`; `at` defaults to now ISO, `source` to `openvibe-sdk/usage`; null fields stripped; money fields never invented); `usageKey(service, ...parts)` joins an idempotency key with `:`; `validateUsageSample(record)` → `{ok, errors}` (openvibe-contracts, required lazily). `openvibe-sdk/govern` builds its `onUsage` record with it |++| `openvibe-sdk/telemetry` | server | `telemetrySample(fields)` / `validateTelemetrySample(record)` for `platform.telemetry-sample@1`; `createTelemetry({service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000})` → `{record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop()}`: buffered samples flushed to `await sink(samples)` on an unref'd interval and on `stop()` (a `gracefulStop` stop step); `record` builds `{service, instance, operation, at, latency_ms, ...labels}` (a non-number value goes to `extra[name]`); a rejecting sink is logged once and its samples kept; past `maxBuffered` the oldest sample is dropped, with a warning at most once a minute || `openvibe-sdk/service` | server | the service kit: `gracefulStop({name, server, stop, close, drainMs, deadlineMs, deadlineExitCode, beforeDrain, handles})` → `{stop, stopping}`, `within(ms, p)`; `createServiceError(name)`, `ServiceError`, `asServiceError`, `run`, `wrap`, `sendError`, `jsonBody({limit})`, `privateNoStore`, `jsonErrors()`; re-exports (loaded on first use) of openvibe-shared `createReadiness`, `skip`, `safeReason`, `createRegistry`, `instrument`, `metricsHandler`, `isLoopbackDirect`, `releaseInfo`, `createRelease` and openvibe-contracts `problem`, `sendProblem` || `openvibe-sdk/testing` | Node | `createMockPlatform()`: fake Network (incl. developer apps and projects), Events, Media, Tools jobs and the Tools platform API on an in-process `fetch` || `openvibe-sdk/browser/openvibe-sdk.mjs` | browser | one self-contained ES module: core + auth (browser), registry, modules, realtime, media, community, jobs, tools, projects |@@@ -7,6 +7,6 @@ export const {satisfiesRange, compareVersions, CONTRACTS_RANGE, DEFAULT_NETWORK, SDK_VERSION, auth,registry, identity, modules, events, realtime, media,community, search, chat, ai, jobs, tools,-- projects, vip, openre, geo,++ projects, vip, openre, geo, usage, telemetry,} = cjs;export default cjs;nnew file mode 100644@@@ -0,0 +1,7 @@++// Generated by scripts/esm.js from src/telemetry.js. Do not edit.++import cjs from '../src/telemetry.js';++++export const {++ telemetrySample, validateTelemetrySample, createTelemetry,++} = cjs;++export default cjs;nnew file mode 100644@@@ -0,0 +1,7 @@++// Generated by scripts/esm.js from src/usage.js. Do not edit.++import cjs from '../src/usage.js';++++export const {++ usageSample, usageKey, validateUsageSample,++} = cjs;++export default cjs;@@@ -25,4 +25,6 @@ module.exports = {vip: require('./src/vip'),openre: require('./src/openre'),geo: require('./src/geo'),++ usage: require('./src/usage'),++ telemetry: require('./src/telemetry'),};@@@ -94,6 +94,18 @@"import": "./esm/placement.mjs","default": "./src/placement.js"},++ "./usage": {++ "types": "./types/usage.d.ts",++ "browser": null,++ "import": "./esm/usage.mjs",++ "default": "./src/usage.js"++ },++ "./telemetry": {++ "types": "./types/telemetry.d.ts",++ "browser": null,++ "import": "./esm/telemetry.mjs",++ "default": "./src/telemetry.js"++ },"./db": {"types": "./types/db.d.ts","browser": null,@@@ -28,6 +28,8 @@ const ENTRIES = {'limits.mjs': '../src/limits.js','govern.mjs': '../src/govern.js','placement.mjs': '../src/placement.js',++ 'usage.mjs': '../src/usage.js',++ 'telemetry.mjs': '../src/telemetry.js','db.mjs': '../src/db/index.js','valkey.mjs': '../src/valkey.js','cache.mjs': '../src/cache.js',@@@ -18,6 +18,8 @@*/const crypto = require('crypto');++const { usageSample } = require('./usage');++const WINDOWS = { minute: 60e3, hour: 3600e3, day: 86400e3, month: 30 * 86400e3 };const UNITS = ['browser-second', 'ai-token', 'ai-usd', 'gpu-second', 'video-minute', 'bandwidth-byte', 'storage-byte-day', 'event','watch-check', 'bot-control-second', 'job-run', 'upload-byte', 'download-byte'];@@@ -163,10 +165,8 @@ function createGovernor({ policy = {}, valkey = null, now = () => Date.now(), re}if (onUsage && !r.replay) {// A platform.usage-sample@1 reading: pass-through only, no money fields (rating is Billing's).-- const record = { id, idempotency_key: key, service, project, subject, resource: res ?? resource, provider: prov ?? provider, region: reg ?? region,-- operation: operation || 'reserve', quantity: Number(amount), unit, at: new Date(t).toISOString(), route_epoch, trace_id, source: 'openvibe-sdk/govern' };-- for (const k of Object.keys(record)) if (record[k] == null) delete record[k];-- onUsage(record);++ onUsage(usageSample({ id, idempotency_key: key, service, project, subject, resource: res ?? resource, provider: prov ?? provider, region: reg ?? region,++ operation: operation || 'reserve', quantity: Number(amount), unit, at: new Date(t).toISOString(), route_epoch, trace_id, source: 'openvibe-sdk/govern' }));}return { ok: true, id, replay: Boolean(r.replay) };}@@@ -22,7 +22,7 @@*/const crypto = require('crypto');--const TRUST_ORDER = ['first-party', 'partner', 'community', 'external'];++const TRUST_ORDER = ['first-party', 'user-owned', 'partner', 'community', 'external'];const HEALTH_OK = new Set(['up', 'degraded']);// ── Cost ─────────────────────────────────────────────────────────────────────nnew file mode 100644@@@ -0,0 +1,140 @@++'use strict';++/**++ * openvibe-sdk/telemetry — platform.telemetry-sample@1, the universal operation observation every service++ * can emit (T1 Universal Fabric): one sample schema for routing and autoscaling, product dimensions in extra.++ * No I/O and no dependency at require time: the sink is the service's (the SDK holds no network code), and++ * openvibe-contracts is required lazily by validateTelemetrySample, the way src/service re-exports its packages.++ *++ * const { telemetrySample, validateTelemetrySample, createTelemetry } = require('openvibe-sdk/telemetry');++ * const t = createTelemetry({ service: 'media', instance: 'i-1', sink: (samples) => post('/telemetry', samples) });++ * t.record('deliver', 28, { subject, resource, status: 'ok' }); // { service, instance, operation, at, latency_ms, ...labels }++ * t.count('cache.miss'); // a convenience wrapper around record()++ * stop: [() => t.stop()], // a gracefulStop stop step: one flush++ */++++/** The fields platform.telemetry-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */++function telemetrySample(fields) {++ const f = fields || {};++ const record = {++ service: f.service,++ project: f.project,++ subject: f.subject,++ resource: f.resource,++ provider: f.provider,++ node: f.node,++ cell: f.cell,++ region: f.region,++ operation: f.operation,++ at: f.at == null ? new Date().toISOString() : f.at,++ latency_ms: f.latency_ms,++ queue_delay_ms: f.queue_delay_ms,++ ttfb_ms: f.ttfb_ms,++ throughput_per_second: f.throughput_per_second,++ bytes: f.bytes,++ status: f.status,++ cache_status: f.cache_status,++ cost_estimate: f.cost_estimate,++ route_epoch: f.route_epoch,++ trace_id: f.trace_id,++ extra: f.extra,++ };++ for (const k of Object.keys(record)) if (record[k] == null) delete record[k];++ return record;++}++++/** { ok, errors } for a record against platform.telemetry-sample@1. A missing openvibe-contracts is never a++ * claim of validity: ok is false and errors stay empty (nothing was checked). */++function validateTelemetrySample(record) {++ let contracts = null;++ try { contracts = require('openvibe-contracts'); } catch { /* not installed: nothing to check against */ }++ if (!contracts) return { ok: false, errors: [] };++ const r = contracts.validate('platform.telemetry-sample@1', record);++ return { ok: Boolean(r.valid), errors: r.errors || [] };++}++++/**++ * A collector: it buffers samples and flushes them to `await sink(samples)` every `intervalMs` (an unref'd++ * timer, so it never holds the process open) and on stop(). A sink that rejects is logged once per attempt++ * and its samples kept for the next flush; past `maxBuffered` the oldest sample is dropped, with a warning++ * at most once a minute. `stop()` clears the timer and flushes once, so it can serve as a gracefulStop stop step.++ */++function createTelemetry({ service = null, instance = null, sink = null, intervalMs = 15000, now = () => Date.now(), log = console, maxBuffered = 1000 } = {}) {++ if (typeof sink !== 'function') throw new TypeError('createTelemetry: a sink(samples) function is required');++ if (!service) log.warn('[telemetry] createTelemetry() without `service`: samples lack the service platform.telemetry-sample@1 requires');++++ let buffer = [];++ let flushing = null;++ let dropped = 0;++ let lastWarn = -Infinity;++++ function warnDropped() {++ const t = now();++ if (t - lastWarn < 60e3) return; // at most one warning per minute++ lastWarn = t;++ log.warn(`[telemetry] buffer over ${maxBuffered}: dropped the oldest sample (${dropped} dropped so far)`);++ }++++ function push(sample) {++ buffer.push(sample);++ while (buffer.length > maxBuffered) {++ buffer.shift();++ dropped += 1;++ warnDropped();++ }++ }++++ /** `{ service, instance, operation: name, at, latency_ms: value, ...labels }`; a value that is not a++ * number goes to `extra[name]` as a scalar (a product dimension, never a payload or a secret). */++ function record(name, value, labels = {}) {++ const { extra, ...rest } = labels || {};++ const base = { service, instance, operation: name, at: new Date(now()).toISOString(), ...rest };++ const sample = typeof value === 'number' ? { ...base, latency_ms: value, extra } : { ...base, extra: { [name]: value, ...(extra || {}) } };++ for (const k of Object.keys(sample)) if (sample[k] == null) delete sample[k];++ push(sample);++ return sample;++ }++++ /** A gauge's level (memory in use, queue depth) and a counter's tick are samples like any other. */++ function gauge(name, value, labels) { return record(name, value, labels); }++ function count(name, labels) { return record(name, 1, labels); }++++ /** Await the sink for everything buffered; a rejection is logged once and the samples kept for the next flush. */++ async function flush() {++ if (flushing) return flushing;++ if (!buffer.length) return;++ const batch = buffer;++ buffer = [];++ flushing = (async () => {++ try {++ await sink(batch);++ } catch (err) {++ log.error(`[telemetry] sink rejected ${batch.length} samples (${err && err.message ? err.message : err}); kept for the next flush`);++ buffer = [...batch, ...buffer];++ while (buffer.length > maxBuffered) {++ buffer.shift();++ dropped += 1;++ warnDropped();++ }++ } finally {++ flushing = null;++ }++ })();++ return flushing;++ }++++ const timer = setInterval(() => { flush(); }, intervalMs);++ if (timer.unref) timer.unref();++++ let stopped = false;++ /** Clears the timer and flushes once (a gracefulStop stop step; a second stop changes nothing). */++ async function stop() {++ if (stopped) return;++ stopped = true;++ clearInterval(timer);++ await flush();++ }++++ return { record, gauge, count, flush, stop };++}++++module.exports = { telemetrySample, validateTelemetrySample, createTelemetry };nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++/**++ * openvibe-sdk/usage — platform.usage-sample@1, one metered usage reading with retry-safe attribution++ * (T1 Universal Fabric). Pure construction and validation, no I/O and no dependency at require time:++ * `openvibe-contracts` is required lazily by validateUsageSample, the way src/service re-exports its packages.++ *++ * const { usageSample, usageKey, validateUsageSample } = require('openvibe-sdk/usage');++ * const record = usageSample({ id, idempotency_key: usageKey('media', 'delivery', 1), service: 'media',++ * operation: 'deliver', quantity: 1.5, unit: 'GiB' }); // at and source default++ * validateUsageSample(record); // { ok, errors }++ *++ * No money fields are invented: cost_estimate, free_allowance_used and vibes_charged ride along only when++ * the caller passes them (rating is Billing's). openvibe-sdk/govern builds its onUsage record with usageSample().++ */++++/** The fields platform.usage-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */++function usageSample(fields) {++ const f = fields || {};++ const record = {++ id: f.id,++ idempotency_key: f.idempotency_key,++ service: f.service,++ project: f.project,++ subject: f.subject,++ resource: f.resource,++ provider: f.provider,++ node: f.node,++ cell: f.cell,++ region: f.region,++ operation: f.operation,++ quantity: f.quantity,++ unit: f.unit,++ at: f.at == null ? new Date().toISOString() : f.at,++ cost_estimate: f.cost_estimate,++ free_allowance_used: f.free_allowance_used,++ vibes_charged: f.vibes_charged,++ route_epoch: f.route_epoch,++ trace_id: f.trace_id,++ source: f.source == null ? 'openvibe-sdk/usage' : f.source,++ };++ for (const k of Object.keys(record)) if (record[k] == null) delete record[k];++ return record;++}++++/** The stable idempotency key of a reading: the service, then each part, joined with ':'. */++function usageKey(service, ...parts) {++ if (typeof service !== 'string' || !service) throw new TypeError('usageKey: service must be a non-empty string');++ const rest = parts.filter((p) => p != null).map(String);++ return rest.length ? [service, ...rest].join(':') : service;++}++++/** { ok, errors } for a record against platform.usage-sample@1. A missing openvibe-contracts is never a++ * claim of validity: ok is false and errors stay empty (nothing was checked). */++function validateUsageSample(record) {++ let contracts = null;++ try { contracts = require('openvibe-contracts'); } catch { /* not installed: nothing to check against */ }++ if (!contracts) return { ok: false, errors: [] };++ const r = contracts.validate('platform.usage-sample@1', record);++ return { ok: Boolean(r.valid), errors: r.errors || [] };++}++++module.exports = { usageSample, usageKey, validateUsageSample };@@@ -76,7 +76,7 @@ run([for (const [re, why] of FORBIDDEN) if (re.test(src)) problems.push(`${path.relative(ROOT, file)}: ${why}`);}assert.deepEqual(problems, []);-- assert.deepEqual(blocked.sort(), ['./cache', './db', './events', './govern', './identity', './limits', './openre', './placement', './pubsub', './queue', './service', './sso', './testing', './valkey', './vip']);++ assert.deepEqual(blocked.sort(), ['./cache', './db', './events', './govern', './identity', './limits', './openre', './placement', './pubsub', './queue', './service', './sso', './telemetry', './testing', './usage', './valkey', './vip']);const rel = [...scanned].map((f) => path.relative(ROOT, f));for (const must of ['browser.js', 'src/core/client.js', 'src/auth/browser.js', 'src/realtime.js', 'src/media.js', 'src/community.js', 'src/modules.js', 'src/registry.js', 'src/jobs.js', 'src/tools.js', 'src/core/form.js', 'src/projects.js', 'browser/openvibe-sdk.mjs']) {assert.ok(rel.includes(must), `${must} is scanned`);@@@ -66,7 +66,7 @@ run([})().catch((err) => { console.error(err); process.exit(1); });`;const res = execFileSync(process.execPath, ['-e', script], { cwd: app, encoding: 'utf8' });assert.match(res, new RegExp(`consumer ok: ${subpaths.length} subpaths \\+ the browser bundle`));-- assert.equal(subpaths.length, 32);++ assert.equal(subpaths.length, 34);assert.ok(!fs.existsSync(path.join(app, 'node_modules/openvibe-contracts')), 'the optional peer is not installed');} finally {fs.rmSync(tmp, { recursive: true, force: true });@@@ -3,7 +3,7 @@const assert = require('node:assert/strict');const crypto = require('node:crypto');const { run } = require('./helpers');--const { plan, marginalCost, rendezvous, pickTwo, signPlan, verifyPlan, createPlanHolder, ownedLoadCost } = require('../src/placement');++const { plan, excluded, marginalCost, rendezvous, pickTwo, signPlan, verifyPlan, createPlanHolder, ownedLoadCost } = require('../src/placement');const T = Date.parse('2026-09-16T00:00:00Z'); // half of September goneconst period = { period_start: '2026-09-01T00:00:00Z', period_end: '2026-10-01T00:00:00Z' };@@@ -44,6 +44,15 @@ run([const r = plan(req(), [provider('mystery', 'rc_missing')], { now: T });assert.equal(r.selected, null);}],++ ['a user-owned offer is eligible under trust: [\'user-owned\'], and for workloads that name no trust list', async () => {++ const own = node('usr-node', 0.2, { trust: 'user-owned' });++ assert.equal(excluded(req({ trust: ['user-owned'] }), own), null, 'user-owned trust is accepted');++ assert.equal(excluded(req(), own), null, 'a workload that names no trust list accepts every TRUST_ORDER class');++ assert.match(excluded(req({ trust: ['partner'] }), own), /partner/, 'a narrower trust list still refuses it');++ const r = plan(req({ objective: 'cheapest', trust: ['user-owned'] }), [own, provider('ext', 'rc')], { rateCards: [card('rc', 'ext', 1e9, 0)], now: T });++ assert.equal(r.selected, 'usr-node');++ assert.match(r.candidates.find((c) => c.id === 'ext').excluded_because, /user-owned/);++ }],['hysteresis: stay unless clearly better; fail over at once when the current one drops out', async () => {const a = node('a', 0.3, { latency_ms: { start_p95: 50 } }); const b = node('b', 0.3, { latency_ms: { start_p95: 46 } });const r = plan(req({ objective: 'lowest-latency' }), [a, b], { current: 'a', now: T });nnew file mode 100644@@@ -0,0 +1,110 @@++'use strict';++/** openvibe-sdk/telemetry: platform.telemetry-sample@1 — build, validate, and a collector that buffers and flushes. */++const assert = require('node:assert/strict');++const fs = require('node:fs');++const path = require('node:path');++const { run, sleep } = require('./helpers');++const { telemetrySample, validateTelemetrySample, createTelemetry } = require('../src/telemetry');++++const FIXTURES = path.join(__dirname, '..', 'node_modules', 'openvibe-contracts', 'fixtures', 'platform.telemetry-sample');++let contracts = null;++try { contracts = require('openvibe-contracts'); } catch { /* the fixture cases skip */ }++++const quiet = { log() {}, warn() {}, error() {} };++const T = Date.parse('2026-09-30T10:00:00Z');++const AT = '2026-09-30T10:00:00.000Z';++++run([++ ['telemetrySample passes every valid fixture through unchanged, and they validate', async () => {++ if (!contracts) { console.log('telemetry fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }++ for (const f of fs.readdirSync(path.join(FIXTURES, 'valid'))) {++ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'valid', f), 'utf8'));++ assert.deepEqual(telemetrySample({ ...fixture }), fixture, f);++ assert.deepEqual(validateTelemetrySample(fixture), { ok: true, errors: [] }, f);++ }++ }],++ ['telemetrySample defaults at and strips null/undefined', async () => {++ const before = Date.now();++ const record = telemetrySample({ service: 'media', operation: 'deliver', at: null, subject: undefined, latency_ms: null });++ assert.deepEqual(record, { service: 'media', operation: 'deliver', at: record.at });++ assert.ok(Date.parse(record.at) >= before, 'at defaults to now');++ const zero = telemetrySample({ service: 'media', operation: 'deliver', at: AT, latency_ms: 0, route_epoch: 0, extra: {} });++ assert.equal(zero.latency_ms, 0); assert.equal(zero.route_epoch, 0); assert.deepEqual(zero.extra, {}, 'a 0 and {} are kept, not stripped');++ }],++ ['validateTelemetrySample refuses every invalid fixture with its errors', async () => {++ if (!contracts) { console.log('telemetry invalid fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }++ for (const f of fs.readdirSync(path.join(FIXTURES, 'invalid'))) {++ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'invalid', f), 'utf8'));++ const r = validateTelemetrySample(fixture);++ assert.equal(r.ok, false, f);++ assert.ok(r.errors.length > 0, f);++ }++ }],++ ['the collector flushes on the interval and on stop, with the sample record() builds', async () => {++ const flushed = [];++ const t = createTelemetry({ service: 'media', instance: 'i-1', sink: async (samples) => { flushed.push(...samples); },++ intervalMs: 30, now: () => T, log: quiet });++ const sample = t.record('deliver', 28, { subject: 'user:u1', resource: 'object-1', status: 'ok', trace_id: 'tr-42' });++ assert.deepEqual(sample, { service: 'media', instance: 'i-1', operation: 'deliver', at: AT, latency_ms: 28,++ subject: 'user:u1', resource: 'object-1', status: 'ok', trace_id: 'tr-42' });++ await sleep(90); // the unref'd interval fired++ assert.equal(flushed.length, 1);++ assert.deepEqual(flushed[0], sample);++ t.gauge('queue_depth', 4); // a gauge's level++ t.count('cache.miss'); // a counter's tick++ t.record('mode', 'fast'); // a non-number value goes to extra[name]++ await t.stop(); // flushes once, timer cleared++ assert.equal(flushed.length, 4);++ assert.deepEqual(flushed[1], { service: 'media', instance: 'i-1', operation: 'queue_depth', at: AT, latency_ms: 4 });++ assert.deepEqual(flushed[2], { service: 'media', instance: 'i-1', operation: 'cache.miss', at: AT, latency_ms: 1 });++ assert.deepEqual(flushed[3], { service: 'media', instance: 'i-1', operation: 'mode', at: AT, extra: { mode: 'fast' } });++ await sleep(60); // the timer is cleared: nothing more flushes++ assert.equal(flushed.length, 4);++ await t.stop(); // a second stop changes nothing++ assert.equal(flushed.length, 4);++ }],++ ['the collector bounds its buffer, drops the oldest and warns once per minute', async () => {++ const warns = [];++ const flushed = [];++ let clock = T;++ const t = createTelemetry({ service: 'media', sink: async (samples) => { flushed.push(...samples); },++ intervalMs: 3600e3, now: () => clock, log: { warn: (...a) => warns.push(a.join(' ')), error() {} }, maxBuffered: 3 });++ t.record('a', 1); t.record('b', 2); t.record('c', 3); // exactly maxBuffered: nothing dropped yet++ assert.equal(warns.length, 0);++ t.record('d', 4); t.record('e', 5); // past the bound: the oldest go, one warning++ assert.equal(warns.length, 1);++ assert.match(warns[0], /dropped/);++ clock += 61e3; // a minute later the warning is allowed again++ t.record('f', 6);++ assert.equal(warns.length, 2);++ clock += 1e3; // within the minute: counted, not repeated++ t.record('g', 7);++ assert.equal(warns.length, 2);++ await t.stop();++ assert.deepEqual(flushed.map((s) => s.operation), ['e', 'f', 'g'], 'the oldest samples were dropped');++ }],++ ['a rejecting sink is logged once per attempt and its samples kept for the next flush', async () => {++ const errors = [];++ let calls = 0;++ const t = createTelemetry({ service: 'media', sink: async () => { calls += 1; throw new Error('boom'); },++ intervalMs: 3600e3, now: () => T, log: { warn() {}, error: (...a) => errors.push(a.join(' ')) } });++ t.record('deliver', 5);++ await t.stop(); // must not reject++ assert.equal(calls, 1);++ assert.equal(errors.length, 1);++ assert.match(errors[0], /boom/);++ await t.flush(); // the buffer was kept, not lost++ assert.equal(calls, 2);++ assert.equal(errors.length, 2);++ }],++ ['createTelemetry requires a sink, and warns once without a service', async () => {++ const warns = [];++ assert.throws(() => createTelemetry({ service: 'media', log: quiet }), /sink/);++ const t = createTelemetry({ sink: () => {}, log: { warn: (...a) => warns.push(a.join(' ')), error() {} } });++ assert.equal(warns.length, 1);++ assert.match(warns[0], /service/);++ t.record('deliver', 1); // still works, like a governor without service++ assert.ok(!('service' in t.record('deliver', 2)));++ await t.stop();++ }],++]);nnew file mode 100644@@@ -0,0 +1,68 @@++'use strict';++/** openvibe-sdk/usage: platform.usage-sample@1 — build, key, validate; govern emits the same record through it. */++const assert = require('node:assert/strict');++const fs = require('node:fs');++const path = require('node:path');++const { run } = require('./helpers');++const { usageSample, usageKey, validateUsageSample } = require('../src/usage');++const { createGovernor } = require('../src/govern');++++const FIXTURES = path.join(__dirname, '..', 'node_modules', 'openvibe-contracts', 'fixtures', 'platform.usage-sample');++let contracts = null;++try { contracts = require('openvibe-contracts'); } catch { /* the fixture cases skip */ }++++const quiet = { log() {}, warn() {}, error() {} };++++run([++ ['usageSample passes every valid fixture through unchanged, and they validate', async () => {++ if (!contracts) { console.log('usage fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }++ for (const f of fs.readdirSync(path.join(FIXTURES, 'valid'))) {++ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'valid', f), 'utf8'));++ assert.deepEqual(usageSample({ ...fixture }), fixture, f);++ assert.deepEqual(validateUsageSample(fixture), { ok: true, errors: [] }, f);++ }++ }],++ ['usageSample defaults at and source, strips null/undefined and invents no money field', async () => {++ const before = Date.now();++ const record = usageSample({ id: 'use-9', idempotency_key: 'media:delivery:9', service: 'media', operation: 'deliver',++ quantity: 1, unit: 'GiB', at: null, source: undefined, project: null, region: undefined, cost_estimate: null });++ assert.deepEqual(Object.keys(record), ['id', 'idempotency_key', 'service', 'operation', 'quantity', 'unit', 'at', 'source']);++ assert.ok(Date.parse(record.at) >= before, 'at defaults to now');++ assert.equal(record.source, 'openvibe-sdk/usage');++ for (const k of ['cost_estimate', 'free_allowance_used', 'vibes_charged', 'project', 'region']) assert.ok(!(k in record), k);++ const passed = usageSample({ id: 'use-8', idempotency_key: 'k', service: 'media', operation: 'deliver', quantity: 0,++ unit: 'GiB', at: '2026-09-30T10:00:00Z', source: 'media.egress', route_epoch: 0 });++ assert.equal(passed.quantity, 0); assert.equal(passed.route_epoch, 0, 'a 0 is kept, not stripped');++ }],++ ['validateUsageSample refuses every invalid fixture with its errors', async () => {++ if (!contracts) { console.log('usage invalid fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }++ for (const f of fs.readdirSync(path.join(FIXTURES, 'invalid'))) {++ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'invalid', f), 'utf8'));++ const r = validateUsageSample(fixture);++ assert.equal(r.ok, false, f);++ assert.ok(r.errors.length > 0, f);++ }++ }],++ ['usageKey joins the service and parts with :, and enforces the service', async () => {++ assert.equal(usageKey('media'), 'media');++ assert.equal(usageKey('media', 'delivery', 1), 'media:delivery:1');++ assert.equal(usageKey('run', 'job_01JAB2C3D4E5F6G7H8J9K0MNPQ', 12), 'run:job_01JAB2C3D4E5F6G7H8J9K0MNPQ:12');++ assert.throws(() => usageKey(''), /non-empty string/);++ assert.throws(() => usageKey(null), /non-empty string/);++ }],++ ['govern: onUsage emits exactly usageSample() of the same fields, and it validates', async () => {++ const policy = { 'job-run': { user: { hour: 10 } } };++ const seen = [];++ const gov = createGovernor({ policy, service: 'openvibe.test', provider: 'local', region: 'eu-1', log: quiet,++ now: () => Date.parse('2026-09-28T12:00:30Z'), onUsage: (r) => seen.push(r) });++ const res = await gov.reserve({ subject: 'user:u1', unit: 'job-run', amount: 3, key: 'usage-mod-1',++ operation: 'render', trace_id: 'tr-7', route_epoch: 2 });++ assert.equal(res.ok, true);++ assert.equal(seen.length, 1);++ assert.deepEqual(seen[0], usageSample({ id: res.id, idempotency_key: 'usage-mod-1', service: 'openvibe.test', subject: 'user:u1',++ provider: 'local', region: 'eu-1', operation: 'render', quantity: 3, unit: 'job-run', at: '2026-09-28T12:00:30.000Z',++ route_epoch: 2, trace_id: 'tr-7', source: 'openvibe-sdk/govern' }));++ if (!contracts) { console.log('govern usage record: skipped validation (openvibe-contracts not installed: npm install)'); return; }++ assert.deepEqual(validateUsageSample(seen[0]), { ok: true, errors: [] });++ }],++]);@@@ -16,3 +16,5 @@ export * as projects from './projects';export * as vip from './vip';export * as openre from './openre';export * as geo from './geo';++export * as usage from './usage';++export * as telemetry from './telemetry';@@@ -1,5 +1,5 @@/** platform.* contract shapes are validated by openvibe-contracts; these are the fields the planner reads. */--export type Offer = { offer_id: string; kind: 'node' | 'provider'; provider?: string; region: string; regions?: string[]; trust: 'first-party' | 'partner' | 'community' | 'external';++export type Offer = { offer_id: string; kind: 'node' | 'provider'; provider?: string; region: string; regions?: string[]; trust: 'first-party' | 'user-owned' | 'partner' | 'community' | 'external';capabilities: string[]; capacity?: Record<string, any>; latency_ms?: Record<string, number>; health: { status: 'up' | 'degraded' | 'down' | 'draining' };pricing: { model: string; marginal_usd_per_unit?: number; rate_card?: string } };export type Requirements = { kind: string; mobility: 'request' | 'job' | 'session' | 'stateful-partition'; latency_class: 'realtime' | 'interactive' | 'background' | 'bulk' | 'critical';nnew file mode 100644@@@ -0,0 +1,30 @@++/** platform.telemetry-sample@1: universal operation observation, product dimensions in extra. Validated by++ * openvibe-contracts (required lazily by validateTelemetrySample, the first time it is called). */++export type TelemetrySample = { service: string; project?: string; subject?: string; resource?: string; provider?: string; node?: string;++ cell?: string; region?: string; operation: string; at: string; latency_ms?: number; queue_delay_ms?: number; ttfb_ms?: number;++ throughput_per_second?: number; bytes?: number; status?: string; cache_status?: string; cost_estimate?: number;++ route_epoch?: number; trace_id?: string; extra?: Record<string, string | number | boolean> };++export type ValidationError = { path: string; message: string };++/** `at` defaults to now (ISO); null/undefined fields are stripped. */++export declare function telemetrySample(fields: Partial<TelemetrySample>): TelemetrySample;++/** { ok, errors } against platform.telemetry-sample@1; a missing openvibe-contracts never claims validity (ok: false, errors: []). */++export declare function validateTelemetrySample(record: unknown): { ok: boolean; errors: ValidationError[] };++/** Labels are the sample's other fields (project, subject, resource, status, …); an `extra` object is merged in. */++export type TelemetryLabels = Partial<Omit<TelemetrySample, 'service' | 'operation' | 'at' | 'latency_ms'>>;++export interface TelemetryCollector {++ /** Builds `{ service, instance, operation: name, at, latency_ms: value, ...labels }` and queues it; a value that++ * is not a number goes to `extra[name]` as a scalar. Returns the sample. */++ record(name: string, value: number | string | boolean, labels?: TelemetryLabels): TelemetrySample;++ /** A gauge's level, as record(name, value, labels). */++ gauge(name: string, value: number, labels?: TelemetryLabels): TelemetrySample;++ /** A counter's tick, as record(name, 1, labels). */++ count(name: string, labels?: TelemetryLabels): TelemetrySample;++ /** Await the sink for everything buffered; a rejection is logged once and the samples kept for the next flush. */++ flush(): Promise<void>;++ /** Clears the timer and flushes once (a gracefulStop stop step; a second stop changes nothing). */++ stop(): Promise<void>;++}++/** A collector: samples buffered and flushed to `await sink(samples)` every intervalMs (unref'd) and on stop();++ * past maxBuffered the oldest sample is dropped, with a warning at most once a minute. */++export declare function createTelemetry(opts: { service?: string | null; instance?: string | null; sink: (samples: TelemetrySample[]) => Promise<void> | void;++ intervalMs?: number; now?: () => number; log?: { warn(...a: unknown[]): void; error(...a: unknown[]): void }; maxBuffered?: number }): TelemetryCollector;nnew file mode 100644@@@ -0,0 +1,13 @@++/** platform.usage-sample@1: one metered usage reading with retry-safe attribution. Validated by openvibe-contracts++ * (required lazily by validateUsageSample, the first time it is called). */++export type UsageSample = { id: string; idempotency_key: string; service: string; project?: string; subject?: string; resource?: string;++ provider?: string; node?: string; cell?: string; region?: string; operation: string; quantity: number; unit: string; at: string;++ /** Money fields are the caller's (rating is Billing's); usageSample never invents them. */++ cost_estimate?: number; free_allowance_used?: number; vibes_charged?: number; route_epoch?: number; trace_id?: string; source: string };++export type ValidationError = { path: string; message: string };++/** `at` defaults to now (ISO) and `source` to 'openvibe-sdk/usage'; null/undefined fields are stripped. */++export declare function usageSample(fields: Partial<UsageSample>): UsageSample;++/** The stable idempotency key of a reading: the service, then each part, joined with ':'; a non-empty service is required. */++export declare function usageKey(service: string, ...parts: (string | number)[]): string;++/** { ok, errors } against platform.usage-sample@1; a missing openvibe-contracts never claims validity (ok: false, errors: []). */++export declare function validateUsageSample(record: unknown): { ok: boolean; errors: ValidationError[] };
@@ -9,6 +9,26 @@ release may change an API and says so here.
@@ -9,6 +9,26 @@ release may change an API and says so here.
## Unreleased
## Unreleased
+- **`openvibe-sdk/usage` (new, server; plan T1):** `platform.usage-sample@1` readings in one place.
+ `usageSample(fields)` builds the record — `id`, `idempotency_key`, `service`, `operation`, `quantity`, `unit` and
+ every optional dimension pass through, `at` defaults to now (ISO) and `source` to `openvibe-sdk/usage`, null fields
+ are stripped and no money field is invented (`cost_estimate`, `free_allowance_used`, `vibes_charged` are the
+ caller's; rating is Billing's). `usageKey(service, ...parts)` joins an idempotency key with `:` (the service alone
+ when there are no parts; a non-empty service is enforced). `validateUsageSample(record)` → `{ ok, errors }`
+ (`openvibe-contracts` is required lazily on first use; a missing package never claims validity: `ok: false`,
+ `errors: []`). `openvibe-sdk/govern` now builds its `onUsage` record with `usageSample()`: the record is unchanged.
+- **`openvibe-sdk/telemetry` (new, server; plan T1):** `platform.telemetry-sample@1`, the universal operation
+ observation. `telemetrySample(fields)` and `validateTelemetrySample(record)` as in usage;
+ `createTelemetry({ service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000 })` →
+ `{ record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop() }`: a collector
+ that buffers samples and flushes them to `await sink(samples)` every `intervalMs` (an unref'd timer) and on
+ `stop()` (clears the timer, flushes once — a `gracefulStop` stop step). `record` builds
+ `{ service, instance, operation, at, latency_ms, ...labels }`; a value that is not a number goes to `extra[name]`
+ as a scalar. A sink that rejects is logged once and its samples kept for the next flush; past `maxBuffered` the
+ oldest sample is dropped, with a warning at most once a minute. The SDK holds no network code: the sink is the service's.
+- **`openvibe-sdk/placement`:** `TRUST_ORDER` gains `user-owned` (between `first-party` and `partner`; the T1 trust
+ classes): a `user-owned` offer is eligible under `trust: ['user-owned']` and for workloads that name no trust list.
+
- The devDependency `openvibe-contracts` moves to v0.86.0 (it adds a `node` principal to
- The devDependency `openvibe-contracts` moves to v0.86.0 (it adds a `node` principal to
`identity.service-token-claims`); the copied types and the lockfile integrity hash follow.
`identity.service-token-claims`); the copied types and the lockfile integrity hash follow.
- **`openvibe-sdk/govern` emits a full usage record** (plan T5). `createGovernor({ service, provider?, resource?, region? })`; `reserve()` takes optional `operation`, `provider`, `resource`, `region`, `trace_id` and `route_epoch`, which override the defaults. `onUsage` now gets a `platform.usage-sample@1` reading: `id` (the reservation id), `idempotency_key`, `service`, `project`, `subject`, `resource`, `provider`, `region`, `operation` (default `reserve`), `quantity`, `unit`, `at`, `route_epoch`, `trace_id`, `source` (`openvibe-sdk/govern`); absent values are left out. **Changed:** `amount` is now `quantity` and `state` is gone (the schema allows neither; only new reservations are emitted). No money fields: rating is Billing's. A governor created without `service` still works and logs one warning (`log`, default `console`). Tested against openvibe-contracts 0.86.0.
- **`openvibe-sdk/govern` emits a full usage record** (plan T5). `createGovernor({ service, provider?, resource?, region? })`; `reserve()` takes optional `operation`, `provider`, `resource`, `region`, `trace_id` and `route_epoch`, which override the defaults. `onUsage` now gets a `platform.usage-sample@1` reading: `id` (the reservation id), `idempotency_key`, `service`, `project`, `subject`, `resource`, `provider`, `region`, `operation` (default `reserve`), `quantity`, `unit`, `at`, `route_epoch`, `trace_id`, `source` (`openvibe-sdk/govern`); absent values are left out. **Changed:** `amount` is now `quantity` and `state` is gone (the schema allows neither; only new reservations are emitted). No money fields: rating is Billing's. A governor created without `service` still works and logs one warning (`log`, default `console`). Tested against openvibe-contracts 0.86.0.
@@ -181,6 +181,8 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client
@@ -181,6 +181,8 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client
| `openvibe-sdk/geo` | both | `createGeoClient(client, {samples?, timeoutMs?})`: `nodes({role, region})` from Network's public node registry, `measure(nodes)` (beacon round trips, fastest first), `nearest({role, region, preferRegion})` → `{node, rtt_ms, measured}` (up/degraded nodes only; falls back to preferRegion when nothing can be measured) |
| `openvibe-sdk/geo` | both | `createGeoClient(client, {samples?, timeoutMs?})`: `nodes({role, region})` from Network's public node registry, `measure(nodes)` (beacon round trips, fastest first), `nearest({role, region, preferRegion})` → `{node, rtt_ms, measured}` (up/degraded nodes only; falls back to preferRegion when nothing can be measured) |
| `openvibe-sdk/openre` | server | `createOpenReClient(client, {baseUrl?, publicUrl?})`: OpenRe.Stream, the streaming engine. `streams` (`byExternalRef`, `create`, `get`, `update`, `rotateKey`, `destinations`, `addDestination`), `destinations` (`test`, `start`, `stop`, `logs`), `sessions` (`get`, `playback` cached 10 s, `end`, `outputs`), `workers`, `manageUrl`; `{subject}` acts for a person (X-OV-Subject) |
| `openvibe-sdk/openre` | server | `createOpenReClient(client, {baseUrl?, publicUrl?})`: OpenRe.Stream, the streaming engine. `streams` (`byExternalRef`, `create`, `get`, `update`, `rotateKey`, `destinations`, `addDestination`), `destinations` (`test`, `start`, `stop`, `logs`), `sessions` (`get`, `playback` cached 10 s, `end`, `outputs`), `workers`, `manageUrl`; `{subject}` acts for a person (X-OV-Subject) |
| `openvibe-sdk/vip` | server | `createVipClient({baseUrl?, tokenClient \| getToken, fetch?, timeoutMs?})`: `evaluate({subject, resource, owner, fallback, mode})`, `checkEntitlement({subject, creator, product, mode})`, `isMember(subject, creator)`; every failure is a denial. `createVipCache({vip, ttlMs, denyTtlMs, unavailableTtlMs})`: `entitlement`, `peekEntitlement`, `evaluate`, `invalidate`, `handleEvent` (`vip.membership.changed` and Billing's entitlement events drop a member's answers at once), `clear`, `bounds` |
| `openvibe-sdk/vip` | server | `createVipClient({baseUrl?, tokenClient \| getToken, fetch?, timeoutMs?})`: `evaluate({subject, resource, owner, fallback, mode})`, `checkEntitlement({subject, creator, product, mode})`, `isMember(subject, creator)`; every failure is a denial. `createVipCache({vip, ttlMs, denyTtlMs, unavailableTtlMs})`: `entitlement`, `peekEntitlement`, `evaluate`, `invalidate`, `handleEvent` (`vip.membership.changed` and Billing's entitlement events drop a member's answers at once), `clear`, `bounds` |
+| `openvibe-sdk/usage` | server | `usageSample(fields)` builds a `platform.usage-sample@1` record (required `id`, `idempotency_key`, `service`, `operation`, `quantity`, `unit`; `at` defaults to now ISO, `source` to `openvibe-sdk/usage`; null fields stripped; money fields never invented); `usageKey(service, ...parts)` joins an idempotency key with `:`; `validateUsageSample(record)` → `{ok, errors}` (openvibe-contracts, required lazily). `openvibe-sdk/govern` builds its `onUsage` record with it |
+| `openvibe-sdk/telemetry` | server | `telemetrySample(fields)` / `validateTelemetrySample(record)` for `platform.telemetry-sample@1`; `createTelemetry({service, instance?, sink, intervalMs = 15000, now?, log?, maxBuffered = 1000})` → `{record(name, value, labels), gauge(name, value, labels), count(name, labels), flush(), stop()}`: buffered samples flushed to `await sink(samples)` on an unref'd interval and on `stop()` (a `gracefulStop` stop step); `record` builds `{service, instance, operation, at, latency_ms, ...labels}` (a non-number value goes to `extra[name]`); a rejecting sink is logged once and its samples kept; past `maxBuffered` the oldest sample is dropped, with a warning at most once a minute |
| `openvibe-sdk/service` | server | the service kit: `gracefulStop({name, server, stop, close, drainMs, deadlineMs, deadlineExitCode, beforeDrain, handles})` → `{stop, stopping}`, `within(ms, p)`; `createServiceError(name)`, `ServiceError`, `asServiceError`, `run`, `wrap`, `sendError`, `jsonBody({limit})`, `privateNoStore`, `jsonErrors()`; re-exports (loaded on first use) of openvibe-shared `createReadiness`, `skip`, `safeReason`, `createRegistry`, `instrument`, `metricsHandler`, `isLoopbackDirect`, `releaseInfo`, `createRelease` and openvibe-contracts `problem`, `sendProblem` |
| `openvibe-sdk/service` | server | the service kit: `gracefulStop({name, server, stop, close, drainMs, deadlineMs, deadlineExitCode, beforeDrain, handles})` → `{stop, stopping}`, `within(ms, p)`; `createServiceError(name)`, `ServiceError`, `asServiceError`, `run`, `wrap`, `sendError`, `jsonBody({limit})`, `privateNoStore`, `jsonErrors()`; re-exports (loaded on first use) of openvibe-shared `createReadiness`, `skip`, `safeReason`, `createRegistry`, `instrument`, `metricsHandler`, `isLoopbackDirect`, `releaseInfo`, `createRelease` and openvibe-contracts `problem`, `sendProblem` |
| `openvibe-sdk/testing` | Node | `createMockPlatform()`: fake Network (incl. developer apps and projects), Events, Media, Tools jobs and the Tools platform API on an in-process `fetch` |
| `openvibe-sdk/testing` | Node | `createMockPlatform()`: fake Network (incl. developer apps and projects), Events, Media, Tools jobs and the Tools platform API on an in-process `fetch` |
| `openvibe-sdk/browser/openvibe-sdk.mjs` | browser | one self-contained ES module: core + auth (browser), registry, modules, realtime, media, community, jobs, tools, projects |
| `openvibe-sdk/browser/openvibe-sdk.mjs` | browser | one self-contained ES module: core + auth (browser), registry, modules, realtime, media, community, jobs, tools, projects |
@@ -7,6 +7,6 @@ export const {
@@ -7,6 +7,6 @@ export const {
satisfiesRange, compareVersions, CONTRACTS_RANGE, DEFAULT_NETWORK, SDK_VERSION, auth,
satisfiesRange, compareVersions, CONTRACTS_RANGE, DEFAULT_NETWORK, SDK_VERSION, auth,
registry, identity, modules, events, realtime, media,
registry, identity, modules, events, realtime, media,
community, search, chat, ai, jobs, tools,
community, search, chat, ai, jobs, tools,
- projects, vip, openre, geo,
+ projects, vip, openre, geo, usage, telemetry,
} = cjs;
} = cjs;
export default cjs;
export default cjs;
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+// Generated by scripts/esm.js from src/telemetry.js. Do not edit.
+import cjs from '../src/telemetry.js';
+
+export const {
+ telemetrySample, validateTelemetrySample, createTelemetry,
+} = cjs;
+export default cjs;
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+// Generated by scripts/esm.js from src/usage.js. Do not edit.
+import cjs from '../src/usage.js';
+
+export const {
+ usageSample, usageKey, validateUsageSample,
+} = cjs;
+export default cjs;
@@ -25,4 +25,6 @@ module.exports = {
@@ -25,4 +25,6 @@ module.exports = {
vip: require('./src/vip'),
vip: require('./src/vip'),
openre: require('./src/openre'),
openre: require('./src/openre'),
geo: require('./src/geo'),
geo: require('./src/geo'),
+ usage: require('./src/usage'),
+ telemetry: require('./src/telemetry'),
};
};
@@ -94,6 +94,18 @@
@@ -94,6 +94,18 @@
"import": "./esm/placement.mjs",
"import": "./esm/placement.mjs",
"default": "./src/placement.js"
"default": "./src/placement.js"
},
},
+ "./usage": {
+ "types": "./types/usage.d.ts",
+ "browser": null,
+ "import": "./esm/usage.mjs",
+ "default": "./src/usage.js"
+ },
+ "./telemetry": {
+ "types": "./types/telemetry.d.ts",
+ "browser": null,
+ "import": "./esm/telemetry.mjs",
+ "default": "./src/telemetry.js"
+ },
"./db": {
"./db": {
"types": "./types/db.d.ts",
"types": "./types/db.d.ts",
"browser": null,
"browser": null,
@@ -28,6 +28,8 @@ const ENTRIES = {
@@ -28,6 +28,8 @@ const ENTRIES = {
'limits.mjs': '../src/limits.js',
'limits.mjs': '../src/limits.js',
'govern.mjs': '../src/govern.js',
'govern.mjs': '../src/govern.js',
'placement.mjs': '../src/placement.js',
'placement.mjs': '../src/placement.js',
+ 'usage.mjs': '../src/usage.js',
+ 'telemetry.mjs': '../src/telemetry.js',
'db.mjs': '../src/db/index.js',
'db.mjs': '../src/db/index.js',
'valkey.mjs': '../src/valkey.js',
'valkey.mjs': '../src/valkey.js',
'cache.mjs': '../src/cache.js',
'cache.mjs': '../src/cache.js',
@@ -18,6 +18,8 @@
@@ -18,6 +18,8 @@
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
+const { usageSample } = require('./usage');
+
const WINDOWS = { minute: 60e3, hour: 3600e3, day: 86400e3, month: 30 * 86400e3 };
const WINDOWS = { minute: 60e3, hour: 3600e3, day: 86400e3, month: 30 * 86400e3 };
const UNITS = ['browser-second', 'ai-token', 'ai-usd', 'gpu-second', 'video-minute', 'bandwidth-byte', 'storage-byte-day', 'event',
const UNITS = ['browser-second', 'ai-token', 'ai-usd', 'gpu-second', 'video-minute', 'bandwidth-byte', 'storage-byte-day', 'event',
'watch-check', 'bot-control-second', 'job-run', 'upload-byte', 'download-byte'];
'watch-check', 'bot-control-second', 'job-run', 'upload-byte', 'download-byte'];
@@ -163,10 +165,8 @@ function createGovernor({ policy = {}, valkey = null, now = () => Date.now(), re
@@ -163,10 +165,8 @@ function createGovernor({ policy = {}, valkey = null, now = () => Date.now(), re
}
}
if (onUsage && !r.replay) {
if (onUsage && !r.replay) {
// A platform.usage-sample@1 reading: pass-through only, no money fields (rating is Billing's).
// A platform.usage-sample@1 reading: pass-through only, no money fields (rating is Billing's).
- const record = { id, idempotency_key: key, service, project, subject, resource: res ?? resource, provider: prov ?? provider, region: reg ?? region,
- operation: operation || 'reserve', quantity: Number(amount), unit, at: new Date(t).toISOString(), route_epoch, trace_id, source: 'openvibe-sdk/govern' };
- for (const k of Object.keys(record)) if (record[k] == null) delete record[k];
- onUsage(record);
+ onUsage(usageSample({ id, idempotency_key: key, service, project, subject, resource: res ?? resource, provider: prov ?? provider, region: reg ?? region,
+ operation: operation || 'reserve', quantity: Number(amount), unit, at: new Date(t).toISOString(), route_epoch, trace_id, source: 'openvibe-sdk/govern' }));
}
}
return { ok: true, id, replay: Boolean(r.replay) };
return { ok: true, id, replay: Boolean(r.replay) };
}
}
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
-const TRUST_ORDER = ['first-party', 'partner', 'community', 'external'];
+const TRUST_ORDER = ['first-party', 'user-owned', 'partner', 'community', 'external'];
const HEALTH_OK = new Set(['up', 'degraded']);
const HEALTH_OK = new Set(['up', 'degraded']);
// ── Cost ─────────────────────────────────────────────────────────────────────
// ── Cost ─────────────────────────────────────────────────────────────────────
new file mode 100644
new file mode 100644
@@ -0,0 +1,140 @@
@@ -0,0 +1,140 @@
+'use strict';
+/**
+ * openvibe-sdk/telemetry — platform.telemetry-sample@1, the universal operation observation every service
+ * can emit (T1 Universal Fabric): one sample schema for routing and autoscaling, product dimensions in extra.
+ * No I/O and no dependency at require time: the sink is the service's (the SDK holds no network code), and
+ * openvibe-contracts is required lazily by validateTelemetrySample, the way src/service re-exports its packages.
+ *
+ * const { telemetrySample, validateTelemetrySample, createTelemetry } = require('openvibe-sdk/telemetry');
+ * const t = createTelemetry({ service: 'media', instance: 'i-1', sink: (samples) => post('/telemetry', samples) });
+ * t.record('deliver', 28, { subject, resource, status: 'ok' }); // { service, instance, operation, at, latency_ms, ...labels }
+ * t.count('cache.miss'); // a convenience wrapper around record()
+ * stop: [() => t.stop()], // a gracefulStop stop step: one flush
+ */
+
+/** The fields platform.telemetry-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */
+function telemetrySample(fields) {
+ const f = fields || {};
+ const record = {
+ service: f.service,
+ project: f.project,
+ subject: f.subject,
+ resource: f.resource,
+ provider: f.provider,
+ node: f.node,
+ cell: f.cell,
+ region: f.region,
+ operation: f.operation,
+ at: f.at == null ? new Date().toISOString() : f.at,
+ latency_ms: f.latency_ms,
+ queue_delay_ms: f.queue_delay_ms,
+ ttfb_ms: f.ttfb_ms,
+ throughput_per_second: f.throughput_per_second,
+ bytes: f.bytes,
+ status: f.status,
+ cache_status: f.cache_status,
+ cost_estimate: f.cost_estimate,
+ route_epoch: f.route_epoch,
+ trace_id: f.trace_id,
+ extra: f.extra,
+ };
+ for (const k of Object.keys(record)) if (record[k] == null) delete record[k];
+ return record;
+}
+
+/** { ok, errors } for a record against platform.telemetry-sample@1. A missing openvibe-contracts is never a
+ * claim of validity: ok is false and errors stay empty (nothing was checked). */
+function validateTelemetrySample(record) {
+ let contracts = null;
+ try { contracts = require('openvibe-contracts'); } catch { /* not installed: nothing to check against */ }
+ if (!contracts) return { ok: false, errors: [] };
+ const r = contracts.validate('platform.telemetry-sample@1', record);
+ return { ok: Boolean(r.valid), errors: r.errors || [] };
+}
+
+/**
+ * A collector: it buffers samples and flushes them to `await sink(samples)` every `intervalMs` (an unref'd
+ * timer, so it never holds the process open) and on stop(). A sink that rejects is logged once per attempt
+ * and its samples kept for the next flush; past `maxBuffered` the oldest sample is dropped, with a warning
+ * at most once a minute. `stop()` clears the timer and flushes once, so it can serve as a gracefulStop stop step.
+ */
+function createTelemetry({ service = null, instance = null, sink = null, intervalMs = 15000, now = () => Date.now(), log = console, maxBuffered = 1000 } = {}) {
+ if (typeof sink !== 'function') throw new TypeError('createTelemetry: a sink(samples) function is required');
+ if (!service) log.warn('[telemetry] createTelemetry() without `service`: samples lack the service platform.telemetry-sample@1 requires');
+
+ let buffer = [];
+ let flushing = null;
+ let dropped = 0;
+ let lastWarn = -Infinity;
+
+ function warnDropped() {
+ const t = now();
+ if (t - lastWarn < 60e3) return; // at most one warning per minute
+ lastWarn = t;
+ log.warn(`[telemetry] buffer over ${maxBuffered}: dropped the oldest sample (${dropped} dropped so far)`);
+ }
+
+ function push(sample) {
+ buffer.push(sample);
+ while (buffer.length > maxBuffered) {
+ buffer.shift();
+ dropped += 1;
+ warnDropped();
+ }
+ }
+
+ /** `{ service, instance, operation: name, at, latency_ms: value, ...labels }`; a value that is not a
+ * number goes to `extra[name]` as a scalar (a product dimension, never a payload or a secret). */
+ function record(name, value, labels = {}) {
+ const { extra, ...rest } = labels || {};
+ const base = { service, instance, operation: name, at: new Date(now()).toISOString(), ...rest };
+ const sample = typeof value === 'number' ? { ...base, latency_ms: value, extra } : { ...base, extra: { [name]: value, ...(extra || {}) } };
+ for (const k of Object.keys(sample)) if (sample[k] == null) delete sample[k];
+ push(sample);
+ return sample;
+ }
+
+ /** A gauge's level (memory in use, queue depth) and a counter's tick are samples like any other. */
+ function gauge(name, value, labels) { return record(name, value, labels); }
+ function count(name, labels) { return record(name, 1, labels); }
+
+ /** Await the sink for everything buffered; a rejection is logged once and the samples kept for the next flush. */
+ async function flush() {
+ if (flushing) return flushing;
+ if (!buffer.length) return;
+ const batch = buffer;
+ buffer = [];
+ flushing = (async () => {
+ try {
+ await sink(batch);
+ } catch (err) {
+ log.error(`[telemetry] sink rejected ${batch.length} samples (${err && err.message ? err.message : err}); kept for the next flush`);
+ buffer = [...batch, ...buffer];
+ while (buffer.length > maxBuffered) {
+ buffer.shift();
+ dropped += 1;
+ warnDropped();
+ }
+ } finally {
+ flushing = null;
+ }
+ })();
+ return flushing;
+ }
+
+ const timer = setInterval(() => { flush(); }, intervalMs);
+ if (timer.unref) timer.unref();
+
+ let stopped = false;
+ /** Clears the timer and flushes once (a gracefulStop stop step; a second stop changes nothing). */
+ async function stop() {
+ if (stopped) return;
+ stopped = true;
+ clearInterval(timer);
+ await flush();
+ }
+
+ return { record, gauge, count, flush, stop };
+}
+
+module.exports = { telemetrySample, validateTelemetrySample, createTelemetry };
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+/**
+ * openvibe-sdk/usage — platform.usage-sample@1, one metered usage reading with retry-safe attribution
+ * (T1 Universal Fabric). Pure construction and validation, no I/O and no dependency at require time:
+ * `openvibe-contracts` is required lazily by validateUsageSample, the way src/service re-exports its packages.
+ *
+ * const { usageSample, usageKey, validateUsageSample } = require('openvibe-sdk/usage');
+ * const record = usageSample({ id, idempotency_key: usageKey('media', 'delivery', 1), service: 'media',
+ * operation: 'deliver', quantity: 1.5, unit: 'GiB' }); // at and source default
+ * validateUsageSample(record); // { ok, errors }
+ *
+ * No money fields are invented: cost_estimate, free_allowance_used and vibes_charged ride along only when
+ * the caller passes them (rating is Billing's). openvibe-sdk/govern builds its onUsage record with usageSample().
+ */
+
+/** The fields platform.usage-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */
+function usageSample(fields) {
+ const f = fields || {};
+ const record = {
+ id: f.id,
+ idempotency_key: f.idempotency_key,
+ service: f.service,
+ project: f.project,
+ subject: f.subject,
+ resource: f.resource,
+ provider: f.provider,
+ node: f.node,
+ cell: f.cell,
+ region: f.region,
+ operation: f.operation,
+ quantity: f.quantity,
+ unit: f.unit,
+ at: f.at == null ? new Date().toISOString() : f.at,
+ cost_estimate: f.cost_estimate,
+ free_allowance_used: f.free_allowance_used,
+ vibes_charged: f.vibes_charged,
+ route_epoch: f.route_epoch,
+ trace_id: f.trace_id,
+ source: f.source == null ? 'openvibe-sdk/usage' : f.source,
+ };
+ for (const k of Object.keys(record)) if (record[k] == null) delete record[k];
+ return record;
+}
+
+/** The stable idempotency key of a reading: the service, then each part, joined with ':'. */
+function usageKey(service, ...parts) {
+ if (typeof service !== 'string' || !service) throw new TypeError('usageKey: service must be a non-empty string');
+ const rest = parts.filter((p) => p != null).map(String);
+ return rest.length ? [service, ...rest].join(':') : service;
+}
+
+/** { ok, errors } for a record against platform.usage-sample@1. A missing openvibe-contracts is never a
+ * claim of validity: ok is false and errors stay empty (nothing was checked). */
+function validateUsageSample(record) {
+ let contracts = null;
+ try { contracts = require('openvibe-contracts'); } catch { /* not installed: nothing to check against */ }
+ if (!contracts) return { ok: false, errors: [] };
+ const r = contracts.validate('platform.usage-sample@1', record);
+ return { ok: Boolean(r.valid), errors: r.errors || [] };
+}
+
+module.exports = { usageSample, usageKey, validateUsageSample };
@@ -76,7 +76,7 @@ run([
@@ -76,7 +76,7 @@ run([
for (const [re, why] of FORBIDDEN) if (re.test(src)) problems.push(`${path.relative(ROOT, file)}: ${why}`);
for (const [re, why] of FORBIDDEN) if (re.test(src)) problems.push(`${path.relative(ROOT, file)}: ${why}`);
}
}
assert.deepEqual(problems, []);
assert.deepEqual(problems, []);
- assert.deepEqual(blocked.sort(), ['./cache', './db', './events', './govern', './identity', './limits', './openre', './placement', './pubsub', './queue', './service', './sso', './testing', './valkey', './vip']);
+ assert.deepEqual(blocked.sort(), ['./cache', './db', './events', './govern', './identity', './limits', './openre', './placement', './pubsub', './queue', './service', './sso', './telemetry', './testing', './usage', './valkey', './vip']);
const rel = [...scanned].map((f) => path.relative(ROOT, f));
const rel = [...scanned].map((f) => path.relative(ROOT, f));
for (const must of ['browser.js', 'src/core/client.js', 'src/auth/browser.js', 'src/realtime.js', 'src/media.js', 'src/community.js', 'src/modules.js', 'src/registry.js', 'src/jobs.js', 'src/tools.js', 'src/core/form.js', 'src/projects.js', 'browser/openvibe-sdk.mjs']) {
for (const must of ['browser.js', 'src/core/client.js', 'src/auth/browser.js', 'src/realtime.js', 'src/media.js', 'src/community.js', 'src/modules.js', 'src/registry.js', 'src/jobs.js', 'src/tools.js', 'src/core/form.js', 'src/projects.js', 'browser/openvibe-sdk.mjs']) {
assert.ok(rel.includes(must), `${must} is scanned`);
assert.ok(rel.includes(must), `${must} is scanned`);
@@ -66,7 +66,7 @@ run([
@@ -66,7 +66,7 @@ run([
})().catch((err) => { console.error(err); process.exit(1); });`;
})().catch((err) => { console.error(err); process.exit(1); });`;
const res = execFileSync(process.execPath, ['-e', script], { cwd: app, encoding: 'utf8' });
const res = execFileSync(process.execPath, ['-e', script], { cwd: app, encoding: 'utf8' });
assert.match(res, new RegExp(`consumer ok: ${subpaths.length} subpaths \\+ the browser bundle`));
assert.match(res, new RegExp(`consumer ok: ${subpaths.length} subpaths \\+ the browser bundle`));
- assert.equal(subpaths.length, 32);
+ assert.equal(subpaths.length, 34);
assert.ok(!fs.existsSync(path.join(app, 'node_modules/openvibe-contracts')), 'the optional peer is not installed');
assert.ok(!fs.existsSync(path.join(app, 'node_modules/openvibe-contracts')), 'the optional peer is not installed');
} finally {
} finally {
fs.rmSync(tmp, { recursive: true, force: true });
fs.rmSync(tmp, { recursive: true, force: true });
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
const assert = require('node:assert/strict');
const assert = require('node:assert/strict');
const crypto = require('node:crypto');
const crypto = require('node:crypto');
const { run } = require('./helpers');
const { run } = require('./helpers');
-const { plan, marginalCost, rendezvous, pickTwo, signPlan, verifyPlan, createPlanHolder, ownedLoadCost } = require('../src/placement');
+const { plan, excluded, marginalCost, rendezvous, pickTwo, signPlan, verifyPlan, createPlanHolder, ownedLoadCost } = require('../src/placement');
const T = Date.parse('2026-09-16T00:00:00Z'); // half of September gone
const T = Date.parse('2026-09-16T00:00:00Z'); // half of September gone
const period = { period_start: '2026-09-01T00:00:00Z', period_end: '2026-10-01T00:00:00Z' };
const period = { period_start: '2026-09-01T00:00:00Z', period_end: '2026-10-01T00:00:00Z' };
@@ -44,6 +44,15 @@ run([
@@ -44,6 +44,15 @@ run([
const r = plan(req(), [provider('mystery', 'rc_missing')], { now: T });
const r = plan(req(), [provider('mystery', 'rc_missing')], { now: T });
assert.equal(r.selected, null);
assert.equal(r.selected, null);
}],
}],
+ ['a user-owned offer is eligible under trust: [\'user-owned\'], and for workloads that name no trust list', async () => {
+ const own = node('usr-node', 0.2, { trust: 'user-owned' });
+ assert.equal(excluded(req({ trust: ['user-owned'] }), own), null, 'user-owned trust is accepted');
+ assert.equal(excluded(req(), own), null, 'a workload that names no trust list accepts every TRUST_ORDER class');
+ assert.match(excluded(req({ trust: ['partner'] }), own), /partner/, 'a narrower trust list still refuses it');
+ const r = plan(req({ objective: 'cheapest', trust: ['user-owned'] }), [own, provider('ext', 'rc')], { rateCards: [card('rc', 'ext', 1e9, 0)], now: T });
+ assert.equal(r.selected, 'usr-node');
+ assert.match(r.candidates.find((c) => c.id === 'ext').excluded_because, /user-owned/);
+ }],
['hysteresis: stay unless clearly better; fail over at once when the current one drops out', async () => {
['hysteresis: stay unless clearly better; fail over at once when the current one drops out', async () => {
const a = node('a', 0.3, { latency_ms: { start_p95: 50 } }); const b = node('b', 0.3, { latency_ms: { start_p95: 46 } });
const a = node('a', 0.3, { latency_ms: { start_p95: 50 } }); const b = node('b', 0.3, { latency_ms: { start_p95: 46 } });
const r = plan(req({ objective: 'lowest-latency' }), [a, b], { current: 'a', now: T });
const r = plan(req({ objective: 'lowest-latency' }), [a, b], { current: 'a', now: T });
new file mode 100644
new file mode 100644
@@ -0,0 +1,110 @@
@@ -0,0 +1,110 @@
+'use strict';
+/** openvibe-sdk/telemetry: platform.telemetry-sample@1 — build, validate, and a collector that buffers and flushes. */
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const path = require('node:path');
+const { run, sleep } = require('./helpers');
+const { telemetrySample, validateTelemetrySample, createTelemetry } = require('../src/telemetry');
+
+const FIXTURES = path.join(__dirname, '..', 'node_modules', 'openvibe-contracts', 'fixtures', 'platform.telemetry-sample');
+let contracts = null;
+try { contracts = require('openvibe-contracts'); } catch { /* the fixture cases skip */ }
+
+const quiet = { log() {}, warn() {}, error() {} };
+const T = Date.parse('2026-09-30T10:00:00Z');
+const AT = '2026-09-30T10:00:00.000Z';
+
+run([
+ ['telemetrySample passes every valid fixture through unchanged, and they validate', async () => {
+ if (!contracts) { console.log('telemetry fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }
+ for (const f of fs.readdirSync(path.join(FIXTURES, 'valid'))) {
+ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'valid', f), 'utf8'));
+ assert.deepEqual(telemetrySample({ ...fixture }), fixture, f);
+ assert.deepEqual(validateTelemetrySample(fixture), { ok: true, errors: [] }, f);
+ }
+ }],
+ ['telemetrySample defaults at and strips null/undefined', async () => {
+ const before = Date.now();
+ const record = telemetrySample({ service: 'media', operation: 'deliver', at: null, subject: undefined, latency_ms: null });
+ assert.deepEqual(record, { service: 'media', operation: 'deliver', at: record.at });
+ assert.ok(Date.parse(record.at) >= before, 'at defaults to now');
+ const zero = telemetrySample({ service: 'media', operation: 'deliver', at: AT, latency_ms: 0, route_epoch: 0, extra: {} });
+ assert.equal(zero.latency_ms, 0); assert.equal(zero.route_epoch, 0); assert.deepEqual(zero.extra, {}, 'a 0 and {} are kept, not stripped');
+ }],
+ ['validateTelemetrySample refuses every invalid fixture with its errors', async () => {
+ if (!contracts) { console.log('telemetry invalid fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }
+ for (const f of fs.readdirSync(path.join(FIXTURES, 'invalid'))) {
+ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'invalid', f), 'utf8'));
+ const r = validateTelemetrySample(fixture);
+ assert.equal(r.ok, false, f);
+ assert.ok(r.errors.length > 0, f);
+ }
+ }],
+ ['the collector flushes on the interval and on stop, with the sample record() builds', async () => {
+ const flushed = [];
+ const t = createTelemetry({ service: 'media', instance: 'i-1', sink: async (samples) => { flushed.push(...samples); },
+ intervalMs: 30, now: () => T, log: quiet });
+ const sample = t.record('deliver', 28, { subject: 'user:u1', resource: 'object-1', status: 'ok', trace_id: 'tr-42' });
+ assert.deepEqual(sample, { service: 'media', instance: 'i-1', operation: 'deliver', at: AT, latency_ms: 28,
+ subject: 'user:u1', resource: 'object-1', status: 'ok', trace_id: 'tr-42' });
+ await sleep(90); // the unref'd interval fired
+ assert.equal(flushed.length, 1);
+ assert.deepEqual(flushed[0], sample);
+ t.gauge('queue_depth', 4); // a gauge's level
+ t.count('cache.miss'); // a counter's tick
+ t.record('mode', 'fast'); // a non-number value goes to extra[name]
+ await t.stop(); // flushes once, timer cleared
+ assert.equal(flushed.length, 4);
+ assert.deepEqual(flushed[1], { service: 'media', instance: 'i-1', operation: 'queue_depth', at: AT, latency_ms: 4 });
+ assert.deepEqual(flushed[2], { service: 'media', instance: 'i-1', operation: 'cache.miss', at: AT, latency_ms: 1 });
+ assert.deepEqual(flushed[3], { service: 'media', instance: 'i-1', operation: 'mode', at: AT, extra: { mode: 'fast' } });
+ await sleep(60); // the timer is cleared: nothing more flushes
+ assert.equal(flushed.length, 4);
+ await t.stop(); // a second stop changes nothing
+ assert.equal(flushed.length, 4);
+ }],
+ ['the collector bounds its buffer, drops the oldest and warns once per minute', async () => {
+ const warns = [];
+ const flushed = [];
+ let clock = T;
+ const t = createTelemetry({ service: 'media', sink: async (samples) => { flushed.push(...samples); },
+ intervalMs: 3600e3, now: () => clock, log: { warn: (...a) => warns.push(a.join(' ')), error() {} }, maxBuffered: 3 });
+ t.record('a', 1); t.record('b', 2); t.record('c', 3); // exactly maxBuffered: nothing dropped yet
+ assert.equal(warns.length, 0);
+ t.record('d', 4); t.record('e', 5); // past the bound: the oldest go, one warning
+ assert.equal(warns.length, 1);
+ assert.match(warns[0], /dropped/);
+ clock += 61e3; // a minute later the warning is allowed again
+ t.record('f', 6);
+ assert.equal(warns.length, 2);
+ clock += 1e3; // within the minute: counted, not repeated
+ t.record('g', 7);
+ assert.equal(warns.length, 2);
+ await t.stop();
+ assert.deepEqual(flushed.map((s) => s.operation), ['e', 'f', 'g'], 'the oldest samples were dropped');
+ }],
+ ['a rejecting sink is logged once per attempt and its samples kept for the next flush', async () => {
+ const errors = [];
+ let calls = 0;
+ const t = createTelemetry({ service: 'media', sink: async () => { calls += 1; throw new Error('boom'); },
+ intervalMs: 3600e3, now: () => T, log: { warn() {}, error: (...a) => errors.push(a.join(' ')) } });
+ t.record('deliver', 5);
+ await t.stop(); // must not reject
+ assert.equal(calls, 1);
+ assert.equal(errors.length, 1);
+ assert.match(errors[0], /boom/);
+ await t.flush(); // the buffer was kept, not lost
+ assert.equal(calls, 2);
+ assert.equal(errors.length, 2);
+ }],
+ ['createTelemetry requires a sink, and warns once without a service', async () => {
+ const warns = [];
+ assert.throws(() => createTelemetry({ service: 'media', log: quiet }), /sink/);
+ const t = createTelemetry({ sink: () => {}, log: { warn: (...a) => warns.push(a.join(' ')), error() {} } });
+ assert.equal(warns.length, 1);
+ assert.match(warns[0], /service/);
+ t.record('deliver', 1); // still works, like a governor without service
+ assert.ok(!('service' in t.record('deliver', 2)));
+ await t.stop();
+ }],
+]);
new file mode 100644
new file mode 100644
@@ -0,0 +1,68 @@
@@ -0,0 +1,68 @@
+'use strict';
+/** openvibe-sdk/usage: platform.usage-sample@1 — build, key, validate; govern emits the same record through it. */
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const path = require('node:path');
+const { run } = require('./helpers');
+const { usageSample, usageKey, validateUsageSample } = require('../src/usage');
+const { createGovernor } = require('../src/govern');
+
+const FIXTURES = path.join(__dirname, '..', 'node_modules', 'openvibe-contracts', 'fixtures', 'platform.usage-sample');
+let contracts = null;
+try { contracts = require('openvibe-contracts'); } catch { /* the fixture cases skip */ }
+
+const quiet = { log() {}, warn() {}, error() {} };
+
+run([
+ ['usageSample passes every valid fixture through unchanged, and they validate', async () => {
+ if (!contracts) { console.log('usage fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }
+ for (const f of fs.readdirSync(path.join(FIXTURES, 'valid'))) {
+ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'valid', f), 'utf8'));
+ assert.deepEqual(usageSample({ ...fixture }), fixture, f);
+ assert.deepEqual(validateUsageSample(fixture), { ok: true, errors: [] }, f);
+ }
+ }],
+ ['usageSample defaults at and source, strips null/undefined and invents no money field', async () => {
+ const before = Date.now();
+ const record = usageSample({ id: 'use-9', idempotency_key: 'media:delivery:9', service: 'media', operation: 'deliver',
+ quantity: 1, unit: 'GiB', at: null, source: undefined, project: null, region: undefined, cost_estimate: null });
+ assert.deepEqual(Object.keys(record), ['id', 'idempotency_key', 'service', 'operation', 'quantity', 'unit', 'at', 'source']);
+ assert.ok(Date.parse(record.at) >= before, 'at defaults to now');
+ assert.equal(record.source, 'openvibe-sdk/usage');
+ for (const k of ['cost_estimate', 'free_allowance_used', 'vibes_charged', 'project', 'region']) assert.ok(!(k in record), k);
+ const passed = usageSample({ id: 'use-8', idempotency_key: 'k', service: 'media', operation: 'deliver', quantity: 0,
+ unit: 'GiB', at: '2026-09-30T10:00:00Z', source: 'media.egress', route_epoch: 0 });
+ assert.equal(passed.quantity, 0); assert.equal(passed.route_epoch, 0, 'a 0 is kept, not stripped');
+ }],
+ ['validateUsageSample refuses every invalid fixture with its errors', async () => {
+ if (!contracts) { console.log('usage invalid fixtures: skipped (openvibe-contracts not installed: npm install)'); return; }
+ for (const f of fs.readdirSync(path.join(FIXTURES, 'invalid'))) {
+ const fixture = JSON.parse(fs.readFileSync(path.join(FIXTURES, 'invalid', f), 'utf8'));
+ const r = validateUsageSample(fixture);
+ assert.equal(r.ok, false, f);
+ assert.ok(r.errors.length > 0, f);
+ }
+ }],
+ ['usageKey joins the service and parts with :, and enforces the service', async () => {
+ assert.equal(usageKey('media'), 'media');
+ assert.equal(usageKey('media', 'delivery', 1), 'media:delivery:1');
+ assert.equal(usageKey('run', 'job_01JAB2C3D4E5F6G7H8J9K0MNPQ', 12), 'run:job_01JAB2C3D4E5F6G7H8J9K0MNPQ:12');
+ assert.throws(() => usageKey(''), /non-empty string/);
+ assert.throws(() => usageKey(null), /non-empty string/);
+ }],
+ ['govern: onUsage emits exactly usageSample() of the same fields, and it validates', async () => {
+ const policy = { 'job-run': { user: { hour: 10 } } };
+ const seen = [];
+ const gov = createGovernor({ policy, service: 'openvibe.test', provider: 'local', region: 'eu-1', log: quiet,
+ now: () => Date.parse('2026-09-28T12:00:30Z'), onUsage: (r) => seen.push(r) });
+ const res = await gov.reserve({ subject: 'user:u1', unit: 'job-run', amount: 3, key: 'usage-mod-1',
+ operation: 'render', trace_id: 'tr-7', route_epoch: 2 });
+ assert.equal(res.ok, true);
+ assert.equal(seen.length, 1);
+ assert.deepEqual(seen[0], usageSample({ id: res.id, idempotency_key: 'usage-mod-1', service: 'openvibe.test', subject: 'user:u1',
+ provider: 'local', region: 'eu-1', operation: 'render', quantity: 3, unit: 'job-run', at: '2026-09-28T12:00:30.000Z',
+ route_epoch: 2, trace_id: 'tr-7', source: 'openvibe-sdk/govern' }));
+ if (!contracts) { console.log('govern usage record: skipped validation (openvibe-contracts not installed: npm install)'); return; }
+ assert.deepEqual(validateUsageSample(seen[0]), { ok: true, errors: [] });
+ }],
+]);
@@ -16,3 +16,5 @@ export * as projects from './projects';
@@ -16,3 +16,5 @@ export * as projects from './projects';
export * as vip from './vip';
export * as vip from './vip';
export * as openre from './openre';
export * as openre from './openre';
export * as geo from './geo';
export * as geo from './geo';
+export * as usage from './usage';
+export * as telemetry from './telemetry';
@@ -1,5 +1,5 @@
@@ -1,5 +1,5 @@
/** platform.* contract shapes are validated by openvibe-contracts; these are the fields the planner reads. */
/** platform.* contract shapes are validated by openvibe-contracts; these are the fields the planner reads. */
-export type Offer = { offer_id: string; kind: 'node' | 'provider'; provider?: string; region: string; regions?: string[]; trust: 'first-party' | 'partner' | 'community' | 'external';
+export type Offer = { offer_id: string; kind: 'node' | 'provider'; provider?: string; region: string; regions?: string[]; trust: 'first-party' | 'user-owned' | 'partner' | 'community' | 'external';
capabilities: string[]; capacity?: Record<string, any>; latency_ms?: Record<string, number>; health: { status: 'up' | 'degraded' | 'down' | 'draining' };
capabilities: string[]; capacity?: Record<string, any>; latency_ms?: Record<string, number>; health: { status: 'up' | 'degraded' | 'down' | 'draining' };
pricing: { model: string; marginal_usd_per_unit?: number; rate_card?: string } };
pricing: { model: string; marginal_usd_per_unit?: number; rate_card?: string } };
export type Requirements = { kind: string; mobility: 'request' | 'job' | 'session' | 'stateful-partition'; latency_class: 'realtime' | 'interactive' | 'background' | 'bulk' | 'critical';
export type Requirements = { kind: string; mobility: 'request' | 'job' | 'session' | 'stateful-partition'; latency_class: 'realtime' | 'interactive' | 'background' | 'bulk' | 'critical';
new file mode 100644
new file mode 100644
@@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
+/** platform.telemetry-sample@1: universal operation observation, product dimensions in extra. Validated by
+ * openvibe-contracts (required lazily by validateTelemetrySample, the first time it is called). */
+export type TelemetrySample = { service: string; project?: string; subject?: string; resource?: string; provider?: string; node?: string;
+ cell?: string; region?: string; operation: string; at: string; latency_ms?: number; queue_delay_ms?: number; ttfb_ms?: number;
+ throughput_per_second?: number; bytes?: number; status?: string; cache_status?: string; cost_estimate?: number;
+ route_epoch?: number; trace_id?: string; extra?: Record<string, string | number | boolean> };
+export type ValidationError = { path: string; message: string };
+/** `at` defaults to now (ISO); null/undefined fields are stripped. */
+export declare function telemetrySample(fields: Partial<TelemetrySample>): TelemetrySample;
+/** { ok, errors } against platform.telemetry-sample@1; a missing openvibe-contracts never claims validity (ok: false, errors: []). */
+export declare function validateTelemetrySample(record: unknown): { ok: boolean; errors: ValidationError[] };
+/** Labels are the sample's other fields (project, subject, resource, status, …); an `extra` object is merged in. */
+export type TelemetryLabels = Partial<Omit<TelemetrySample, 'service' | 'operation' | 'at' | 'latency_ms'>>;
+export interface TelemetryCollector {
+ /** Builds `{ service, instance, operation: name, at, latency_ms: value, ...labels }` and queues it; a value that
+ * is not a number goes to `extra[name]` as a scalar. Returns the sample. */
+ record(name: string, value: number | string | boolean, labels?: TelemetryLabels): TelemetrySample;
+ /** A gauge's level, as record(name, value, labels). */
+ gauge(name: string, value: number, labels?: TelemetryLabels): TelemetrySample;
+ /** A counter's tick, as record(name, 1, labels). */
+ count(name: string, labels?: TelemetryLabels): TelemetrySample;
+ /** Await the sink for everything buffered; a rejection is logged once and the samples kept for the next flush. */
+ flush(): Promise<void>;
+ /** Clears the timer and flushes once (a gracefulStop stop step; a second stop changes nothing). */
+ stop(): Promise<void>;
+}
+/** A collector: samples buffered and flushed to `await sink(samples)` every intervalMs (unref'd) and on stop();
+ * past maxBuffered the oldest sample is dropped, with a warning at most once a minute. */
+export declare function createTelemetry(opts: { service?: string | null; instance?: string | null; sink: (samples: TelemetrySample[]) => Promise<void> | void;
+ intervalMs?: number; now?: () => number; log?: { warn(...a: unknown[]): void; error(...a: unknown[]): void }; maxBuffered?: number }): TelemetryCollector;
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+/** platform.usage-sample@1: one metered usage reading with retry-safe attribution. Validated by openvibe-contracts
+ * (required lazily by validateUsageSample, the first time it is called). */
+export type UsageSample = { id: string; idempotency_key: string; service: string; project?: string; subject?: string; resource?: string;
+ provider?: string; node?: string; cell?: string; region?: string; operation: string; quantity: number; unit: string; at: string;
+ /** Money fields are the caller's (rating is Billing's); usageSample never invents them. */
+ cost_estimate?: number; free_allowance_used?: number; vibes_charged?: number; route_epoch?: number; trace_id?: string; source: string };
+export type ValidationError = { path: string; message: string };
+/** `at` defaults to now (ISO) and `source` to 'openvibe-sdk/usage'; null/undefined fields are stripped. */
+export declare function usageSample(fields: Partial<UsageSample>): UsageSample;
+/** The stable idempotency key of a reading: the service, then each part, joined with ':'; a non-empty service is required. */
+export declare function usageKey(service: string, ...parts: (string | number)[]): string;
+/** { ok, errors } against platform.usage-sample@1; a missing openvibe-contracts never claims validity (ok: false, errors: []). */
+export declare function validateUsageSample(record: unknown): { ok: boolean; errors: ValidationError[] };