openvibers/openvibe.network Commits
Grant AI, Events and Media billing.usage.record and Bot its OpenRe capabilities; emit aggregated platform.telemetry-sample@1 (#52)
Changes
@@@ -45,8 +45,8 @@ Identity and account service for the OpenVibe network. Manages user accounts, OA## Depends on--- `openvibe-contracts` v0.80.0 (registry manifests, capability checks), `openvibe-sdk` v0.25.1 (PostgreSQL-- and Valkey adapters, events, per-actor limits), `openvibe-shared` v1.29.2, pinned by release tarball++- `openvibe-contracts` v0.80.0 (registry manifests, capability checks), `openvibe-sdk` v0.28.0 (PostgreSQL++ and Valkey adapters, events, per-actor limits, universal telemetry), `openvibe-shared` v1.29.2, pinned by release tarball- OpenVibe.Events (Network's outbox relay; the events it consumes), OpenVibe.AI (the `network.site_copy`workflow), OpenVibe.Blog (the changelog feed it proxies)- email (Resend), Discord and GitHub when their secrets are set ([Provider secrets](#provider-secrets))@@@ -560,6 +560,12 @@ Google-style account management supporting up to 5 accounts:open tabs report to `POST /release-metrics`). Network's own counters are`network_tokens_issued_total{grant_type}`, `network_token_failures_total{grant_type,error}` and`network_principal_token_failures_total{code}`.++- Universal telemetry: requests are aggregated per `route|method|status_class` into a single++ `platform.telemetry-sample@1` per key per flush (count, sum, max and p95 in `extra`, mean in++ `latency_ms`), plus the HTTP autoscaling gauges (active requests, p95, event-loop lag) once per flush;++ health, readiness, metrics, chrome and static/shared assets are skipped. Buffered by++ `openvibe-sdk/telemetry` and written into the analytics store (`server/telemetry.js`,++ `server/observability.js`); `gracefulStop` flushes it once at shutdown.- `GET /release.json` is Network's release manifest (ADR-016, `registry.release-manifest@1`, from`openvibe-shared/release`'s `release.mount`): the deployed commit, the library versions and, sinceopenvibe-contracts 0.32.0, the 1.1.0 fields. No components are declared, so every release still@@@ -42,7 +42,8 @@"service registry, public discovery and the status page with incidents","operator alerts relay target and admin panel","serves /shared/* from its pinned openvibe-shared",-- "per-actor limits (openvibe-sdk/limits)"++ "per-actor limits (openvibe-sdk/limits)",++ "universal telemetry: per-route aggregated platform.telemetry-sample@1 plus per-flush HTTP autoscaling gauges (openvibe-sdk/telemetry)"],-- "updated": "2026-09-29"++ "updated": "2026-10-04"}@@@ -22,7 +22,7 @@"jsonwebtoken": "^9.0.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0","pg": "^8.23.0","uuid": "^11.1.0",@@@ -1572,9 +1572,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.26.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",-- "integrity": "sha512-+RK89ZG4WAKIRh89yr3ieIvnyJVNSOI+dsbBqGNcr9UHsJStdxtM75zSDq8cUNpBPQ1X6ElatZrjU9S3s0aEkg==",++ "version": "0.28.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",++ "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==","license": "MIT","engines": {"node": ">=20"@@@ -1,7 +1,7 @@{"name": "openvibe-network","version": "1.0.0",-- "description": "OpenVibe.Network — Identity, SSO, Accounts, Themes, Notifications, and OpenCoins Wallet",++ "description": "OpenVibe.Network \u2014 Identity, SSO, Accounts, Themes, Notifications, and OpenCoins Wallet","main": "server/index.js","scripts": {"start": "node server/index.js",@@@ -31,7 +31,7 @@"jsonwebtoken": "^9.0.0","multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0","pg": "^8.23.0","uuid": "^11.1.0",@@@ -156,6 +156,10 @@ const DEFAULT_GRANTS = [['host', 'network.resource.report', SELF_AUDIENCE, []],// Plan T2 N4b: Bot pairs its users' machines with Network and reads or revokes only the principals it paired.['bot', 'network.node.manage', SELF_AUDIENCE, []],++ // Plan T15: Bot runs each robot's OpenRe stream for its owner (find/create/archive, key rotation), plays the live++ // session in the panel (session.read), restreams out (output.*) and sets the owner's streaming and recording++ // toggles on the stream, minting its own token with its Network client instead of a hand-set BOT_OPENRE_TOKEN.++ ...['openre.stream.read', 'openre.stream.write', 'openre.key.rotate', 'openre.session.read', 'openre.output.read', 'openre.output.write'].map(c => ['bot', c, 'openvibe.openre', []]),// Plan T2 lane B step 3: Host places and runs per project, so it reads a project's tenancy, placement and quotas.// Other services get this row only when they ship a caller of GET /internal/projects/:project_id.['host', 'network.project.read', SELF_AUDIENCE, []],@@@ -254,6 +258,13 @@ const DEFAULT_GRANTS = [['tools', 'search.document.write', 'openvibe.search', []],// Plan T5 step 7: Tools posts one usage reading per ended tool job to Billing (POST /api/v1/usage).['tools', 'billing.usage.record', 'openvibe.billing', []],++ // Plan T5 step 14: Network's own usage producers — AI (per model run), Events (per delivered event)++ // and Media (per stored object) post their platform.usage-sample@1 readings to Billing++ // (POST /api/v1/usage) with their own token. Recording usage only: none of them holds a money++ // capability (billing.ledger.admin / billing.cashout.manage).++ ['ai', 'billing.usage.record', 'openvibe.billing', []],++ ['events', 'billing.usage.record', 'openvibe.billing', []],++ ['media', 'billing.usage.record', 'openvibe.billing', []],];// Grants withdrawn by decision; applied at every boot so an old default can't come back.@@@ -335,6 +335,12 @@ config._registry = resolvedRegistry;const analytics = networkAnalytics.openAnalytics(db);app.locals.analytics = analytics;app.use(analytics.middleware());++// Universal telemetry (plan T1, server/telemetry.js): one platform.telemetry-sample@1 per request plus++// the HTTP autoscaling signals, into the same analytics store. Wired before any route runs; flushed once++// by gracefulStop's stop array below.++const telemetry = require('./telemetry');++telemetry.init({ analytics });++app.use(observability.telemetryMiddleware);// ── Extract bearer token (available as req.token for optional-auth proxies) ─app.use((req, _res, next) => {@@@ -985,6 +991,7 @@ gracefulStop({() => require('./identity/profile-events').stop(),() => require('./identity/grants-admin').stop(),() => require('./developer/confirmations').stop(),++ () => telemetry.stop(),],close: [() => within(1500, require('./developer/event-relay').stopRelay(db)),@@@ -9,9 +9,11 @@* and service-token (principal) failures at Network's capability-guarded internal routes.* Labels are fixed vocabularies; a client id, subject or token never becomes a label.*/++const path = require('path');const jwt = require('jsonwebtoken');const metrics = require('openvibe-shared/metrics');const { createReadiness, skip } = require('openvibe-shared/ready');++const telemetry = require('./telemetry');const registry = metrics.createRegistry();@@@ -50,6 +52,79 @@ function principalDenied({ req, code }) {principalFailures.inc({ code: c });}++// ── Universal telemetry (plan T1, §15.19): aggregated platform.telemetry-sample@1 + autoscaling ──++// A request is not a row: the middleware hands each finished request to server/telemetry.js, which++// aggregates per route|method|status_class and emits one http.request sample per key per flush (plus the++// HTTP autoscaling gauges once per flush). A sample carries the route template, method and status class —++// never a raw URL, client id or token. Health, readiness, metrics, chrome and static/shared assets carry++// no product signal and are skipped before anything is counted.++++// The event-loop monitor: created and enabled by startEventLoopMonitor (called from telemetry.init), and++// disabled by stopEventLoopMonitor (telemetry.stop). Nothing runs at module load.++let loopMonitor = null;++function startEventLoopMonitor() {++ if (loopMonitor) return;++ try {++ loopMonitor = require('perf_hooks').monitorEventLoopDelay({ resolution: 20 });++ loopMonitor.enable();++ } catch { loopMonitor = null; }++}++function stopEventLoopMonitor() {++ if (!loopMonitor) return;++ try { loopMonitor.disable(); } catch { /* already gone */ }++ loopMonitor = null;++}++/** Whether the monitor is running (tests assert requiring the module starts nothing). */++function eventLoopMonitorEnabled() { return loopMonitor != null; }++++/** The event-loop lag since the last flush, ms (mean includes the sampling interval: subtracted); null++ * before the monitor has a sample. The monitor resets per read, so each gauge covers one flush interval. */++function eventLoopLagMs() {++ if (!loopMonitor || !(loopMonitor.count > 0 || loopMonitor.max > 0)) return null;++ const lag = Math.max(0, loopMonitor.mean / 1e6 - 20);++ loopMonitor.reset();++ return lag;++}++++// server/telemetry.js owns the aggregation; it pulls the lag and calls these lifecycle hooks.++telemetry.registerSignals({ start: startEventLoopMonitor, stop: stopEventLoopMonitor, lag: eventLoopLagMs });++++const SKIP_EXACT = new Set(['/api/health', '/ready', '/api/ready', '/metrics']);++const SKIP_PREFIXES = ['/shared', '/api/chrome'];++// Anything express.static serves: public/ files and /data/avatars images. JSON is deliberately not here++// (routes such as /release.json and /contracts/*.json are product/observability API, not static assets).++const STATIC_EXTENSIONS = new Set([++ '.js', '.mjs', '.cjs', '.css', '.map', '.html', '.htm', '.txt', '.xml', '.webmanifest', '.wasm',++ '.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp', '.avif', '.ico', '.bmp',++ '.woff', '.woff2', '.ttf', '.otf', '.eot', '.mp3', '.mp4', '.webm', '.ogg', '.pdf',++]);++function telemetrySkipped(req) {++ const p = req.path || '';++ if (SKIP_EXACT.has(p)) return true;++ if (SKIP_PREFIXES.some((prefix) => p === prefix || p.startsWith(`${prefix}/`))) return true;++ return STATIC_EXTENSIONS.has(path.extname(p).toLowerCase());++}++++/** Express middleware: hand the finished request to the aggregator; the in-flight peak is sampled at start. */++function telemetryMiddleware(req, res, next) {++ if (telemetrySkipped(req)) return next();++ telemetry.requestStarted();++ const t0 = process.hrtime.bigint();++ let done = false;++ const finish = () => {++ if (done) return;++ done = true;++ try {++ const latencyMs = Number(process.hrtime.bigint() - t0) / 1e6;++ const httpStatus = res.writableFinished || res.finished ? (res.statusCode || 0) : 0;++ telemetry.requestFinished({ route: metrics.routeLabel(req), method: req.method, httpStatus, latencyMs });++ } catch { /* telemetry never breaks a request */ }++ };++ res.once('finish', finish);++ res.once('close', finish);++ next();++}++/*** Readiness with Network's real dependencies:* db (required) a query against the identity database@@@ -98,4 +173,4 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discorreturn createReadiness({ service: 'network', release, checks });}--module.exports = { registry, tokenEndpointMetrics, principalDenied, createNetworkReadiness, grantLabel };++module.exports = { registry, tokenEndpointMetrics, principalDenied, telemetryMiddleware, telemetrySkipped, createNetworkReadiness, grantLabel, eventLoopMonitorEnabled };nnew file mode 100644@@@ -0,0 +1,226 @@++'use strict';++/**++ * Network's platform.telemetry-sample@1 collector (openvibe-sdk/telemetry; plan T1 universal Fabric,++ * §15.19). One sample schema every service emits; autoscaling and routing consume it.++ *++ * createNetworkTelemetry({ analytics }) service 'network', with a sink that writes each flushed batch++ * into Network's ADR-021 analytics tracker as custom (non-HTTP)++ * events and flushes it, so one batch lands together.++ * init({ analytics }) the process-wide collector, wired once at boot (server/index.js).++ * record / gauge / count / flush / stop the collector's API; before init they are no-ops.++ *++ * Volume: one analytics row per request does not scale (at 50 req/s that is ~200 rows/s). HTTP requests++ * are therefore aggregated per `route|method|status_class` per flush interval into count, sum, max and++ * p95, and ONE `http.request` sample per key is emitted per flush — not one per request. The HTTP++ * autoscaling gauges (active requests, p95, event-loop lag) are emitted once per flush from a timer++ * started in init(), never per request, so the SDK buffer (maxBuffered 1000) cannot overflow.++ *++ * Field mapping (platform.telemetry-sample@1; `latency_ms` is the schema's latency field, there is no++ * generic gauge field): an aggregated request carries its mean in `latency_ms` and count/sum/max/p95 in++ * `extra`; the p95 gauge carries its value in `latency_ms` (it is a latency); the active-requests and++ * event-loop-lag gauges carry their value in `extra` — never in `latency_ms`, which the analytics sink++ * maps to `response_time_ms` and would make indistinguishable from a latency.++ *++ * server/observability.js feeds the middleware's per-request observations and registers the event-loop++ * monitor (started by init, stopped by stop) so nothing starts at module load; gracefulStop's stop array++ * flushes it once at shutdown.++ */++const { createTelemetry, telemetrySample, validateTelemetrySample } = require('openvibe-sdk/telemetry');++++const SERVICE = 'network';++const DEFAULT_INTERVAL_MS = 15000;++++// analytics/event.v1 event_type: [A-Za-z][A-Za-z0-9_.:-]{0,63}; an operation is sanitised to fit.++const EVENT_RE = /^[A-Za-z][A-Za-z0-9_.:-]{0,63}$/;++function eventName(operation) {++ const name = `telemetry.${String(operation == null ? 'operation' : operation)}`.replace(/[^A-Za-z0-9_.:-]/g, '_').slice(0, 64);++ return EVENT_RE.test(name) ? name : 'telemetry.operation';++}++++/**++ * The sink: write each sample in the batch into Network's analytics tracker (its `trackEvent`, the++ * non-HTTP form; ADR-021 reduces the personal fields), then await its flush so the batch lands together.++ * Only the fields analytics/event.v1 can hold are passed; the numbers live in `extra.http_status`.++ */++function analyticsSink(analytics) {++ if (!analytics || typeof analytics.trackEvent !== 'function') {++ throw new TypeError('analyticsSink: Network\'s analytics tracker (trackEvent) is required');++ }++ return async function sink(samples) {++ for (const s of samples) {++ const extra = s.extra || {};++ analytics.trackEvent(eventName(s.operation), {++ path: typeof s.resource === 'string' ? s.resource : (typeof extra.route === 'string' ? extra.route : undefined),++ method: typeof extra.method === 'string' ? extra.method : undefined,++ status_code: Number.isInteger(extra.http_status) ? extra.http_status : undefined,++ response_time_ms: Number.isFinite(s.latency_ms) ? s.latency_ms : undefined,++ });++ }++ if (typeof analytics.flush === 'function') await analytics.flush();++ };++}++++// ── HTTP aggregation (volume) ─────────────────────────────────────++const HTTP_METHODS = new Set(['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']);++const statusClass = (code) => (code ? `${Math.floor(code / 100)}xx` : 'aborted');++const round = (n, digits = 3) => { const f = 10 ** digits; return Math.round(n * f) / f; };++++/** p95 (nearest-rank) of an array of ms latencies; null when empty. Computed at most once per flush. */++function percentile95(values) {++ if (!values.length) return null;++ const sorted = [...values].sort((a, b) => a - b);++ return sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * 0.95))];++}++++// The event-loop monitor lives in observability.js and is injected here at its require: requiring either++// module starts nothing. telemetry.init() calls start(), telemetry.stop() calls stop(), and the gauges++// read lag(). Unregistered (unit tests, a bare collector) they are safe no-ops.++let signals = { start() {}, stop() {}, lag() { return null; } };++function registerSignals(injected) {++ signals = { ...signals, ...(injected || {}) };++}++++/**++ * Build one collector. `sink` overrides the analytics sink (tests inject one); everything else is the SDK.++ * HTTP observations are aggregated in memory and drained on flush(); the flush timer itself starts in++ * init(), so a bare collector only flushes when asked.++ */++function createNetworkTelemetry({ analytics = null, sink = null, instance = null, intervalMs = DEFAULT_INTERVAL_MS, now, log, maxBuffered } = {}) {++ const write = typeof sink === 'function' ? sink : analyticsSink(analytics);++ const sdk = createTelemetry({ service: SERVICE, instance, sink: write, intervalMs, now, log, maxBuffered });++++ let routes = new Map(); // key -> { route, method, status, count, sumMs, maxMs, latencies, codes }++ let allLatencies = []; // the flush interval's latencies, for the single p95 gauge++ let active = 0; // requests in flight right now++ let activeMax = 0; // the interval's peak, sampled at request start (item 3)++ let stopped = false;++++ /** A request began: track the in-flight peak here, not after the decrement. */++ function requestStarted() {++ active += 1;++ if (active > activeMax) activeMax = active;++ }++++ function observeRequest(info = {}) {++ const r = String(info.route == null ? 'unmatched' : info.route).slice(0, 200);++ const m = HTTP_METHODS.has(info.method) ? info.method : 'OTHER';++ const s = statusClass(info.httpStatus);++ const latency = Number.isFinite(info.latencyMs) && info.latencyMs >= 0 ? info.latencyMs : 0;++ const key = `${r}|${m}|${s}`;++ let bucket = routes.get(key);++ if (!bucket) {++ bucket = { route: r, method: m, status: s, count: 0, sumMs: 0, maxMs: 0, latencies: [], codes: new Set() };++ routes.set(key, bucket);++ }++ bucket.count += 1;++ bucket.sumMs += latency;++ if (latency > bucket.maxMs) bucket.maxMs = latency;++ bucket.latencies.push(latency);++ if (Number.isInteger(info.httpStatus) && info.httpStatus > 0) bucket.codes.add(info.httpStatus);++ allLatencies.push(latency);++ }++++ /** A request ended: free the in-flight slot, then fold it into its aggregation key. */++ function requestFinished(info) {++ if (active > 0) active -= 1;++ observeRequest(info);++ }++++ /** Push a gauge whose value(s) live in `extra`; the schema has no generic gauge field and latency_ms++ * is the sink's response_time_ms, so extra is the documented home (never latency_ms). */++ function emitExtraGauge(name, extra) {++ const sample = sdk.record(name, 0, { extra });++ delete sample.latency_ms;++ return sample;++ }++++ /** Drain this interval: one http.request sample per route|method|status_class, then the three gauges. */++ function emitSamples() {++ const buckets = [...routes.values()];++ const latencies = allLatencies;++ routes = new Map();++ allLatencies = [];++ for (const b of buckets) {++ const mean = b.count ? b.sumMs / b.count : 0;++ const p95 = percentile95(b.latencies);++ const extra = {++ method: b.method,++ status_class: b.status,++ count: b.count,++ sum_ms: round(b.sumMs),++ max_ms: round(b.maxMs),++ };++ if (p95 != null) extra.p95_ms = round(p95);++ if (b.codes.size === 1) extra.http_status = b.codes.values().next().value;++ sdk.record('http.request', round(mean), { status: b.status, resource: b.route, extra });++ }++ // The gauges, once per flush (never per request).++ emitExtraGauge('http.active_requests', { last: active, max: activeMax });++ activeMax = active; // the next interval's peak starts from what is still in flight++ const p95 = percentile95(latencies);++ if (p95 != null) sdk.record('http.latency_p95_ms', round(p95)); // latency_ms: the schema's latency field++ const lag = signals.lag();++ if (lag != null) emitExtraGauge('http.eventloop_lag_ms', { value: round(lag) });++ }++++ /** Emit the interval's samples, then hand the SDK everything buffered (a flush is also the tick). */++ function flush() {++ if (stopped) return Promise.resolve();++ emitSamples();++ return sdk.flush();++ }++++ /** Emit once, then let the SDK clear its timer and flush (idempotent). */++ async function stop() {++ if (stopped) return;++ stopped = true;++ emitSamples();++ await sdk.stop();++ }++++ return { record: sdk.record, gauge: sdk.gauge, count: sdk.count, flush, stop, requestStarted, requestFinished, observeRequest };++}++++// The process-wide collector. It exists only once the analytics tracker does (server/index.js), so an++// early request is recorded by neither; the HTTP middleware is mounted before any route runs.++let current = null;++let timer = null;++function init(opts = {}) {++ current = createNetworkTelemetry(opts);++ signals.start(); // the event-loop monitor (observability.js): created and enabled here, not at load++ if (timer) clearInterval(timer);++ const interval = Number.isFinite(opts.intervalMs) ? opts.intervalMs : DEFAULT_INTERVAL_MS;++ timer = setInterval(() => { current.flush(); }, interval);++ if (timer.unref) timer.unref();++ return current;++}++++function record(...args) { return current ? current.record(...args) : undefined; }++function gauge(...args) { return current ? current.gauge(...args) : undefined; }++function count(...args) { return current ? current.count(...args) : undefined; }++function requestStarted() { return current ? current.requestStarted() : undefined; }++function requestFinished(info) { return current ? current.requestFinished(info) : undefined; }++function flush() { return current ? current.flush() : Promise.resolve(); }++async function stop() {++ if (timer) { clearInterval(timer); timer = null; }++ if (current) await current.stop();++ signals.stop(); // disable the event-loop monitor that init enabled++}++++module.exports = {++ SERVICE,++ eventName,++ analyticsSink,++ createNetworkTelemetry,++ telemetrySample,++ validateTelemetrySample,++ registerSignals,++ init,++ record,++ gauge,++ count,++ requestStarted,++ requestFinished,++ flush,++ stop,++};@@@ -22,6 +22,10 @@ await db.prepare("UPDATE oauth_clients SET client_secret = 'live-secret' WHERE cawait db.prepare("UPDATE oauth_clients SET client_secret = 'media-secret' WHERE client_id = 'media'").run();// The `search` client is created with server/setup/service-principal.js, not seeded with the signed-in products.await db.prepare("INSERT INTO oauth_clients (client_id, client_secret, name, redirect_uris, is_first_party) VALUES ('search', 'search-secret', 'OpenVibe.Search', '[]', 1) ON CONFLICT DO NOTHING").run();++// Events is created the same way (setup/service-principal.js); it is a usage producer with its own token.++await db.prepare("INSERT INTO oauth_clients (client_id, client_secret, name, redirect_uris, is_first_party) VALUES ('events', 'events-secret', 'OpenVibe.Events', '[]', 1) ON CONFLICT DO NOTHING").run();++await db.prepare("UPDATE oauth_clients SET client_secret = 'ai-secret' WHERE client_id = 'ai'").run();++await db.prepare("UPDATE oauth_clients SET client_secret = 'bot-secret' WHERE client_id = 'bot'").run();await db.prepare("INSERT INTO users (id, username, password_hash) VALUES (7, 'payee', 'x'), (8, 'payer', 'x')").run();// A database seeded before media.analyze (Live's AI grants as the old default) is moved at the next boot.await db.prepare("UPDATE principal_grants SET namespaces = ? WHERE client_id = 'live' AND audience = 'openvibe.ai'").run(JSON.stringify(['live.*', 'network.site_copy']));@@@ -178,6 +182,18 @@ const server = http.createServer(app);assert.strictEqual(ts.body.scope, 'search.document.write', 'Tools indexes its tools in Search');const tb = await token({ client_id: 'tools', client_secret: 'tools-secret', audience: 'openvibe.billing' });assert.strictEqual(tb.body.scope, 'billing.usage.record', 'Tools posts usage readings to Billing and holds no money capability');++ // Plan T5 step 14: Network's own usage producers post platform.usage-sample@1 to Billing with their++ // own token; usage recording only, never a money capability.++ for (const [client, secret] of [['ai', 'ai-secret'], ['events', 'events-secret'], ['media', 'media-secret']]) {++ const ub = await token({ client_id: client, client_secret: secret, audience: 'openvibe.billing' });++ assert.strictEqual(ub.status, 200, `${client}: ${JSON.stringify(ub.body)}`);++ assert.strictEqual(ub.body.scope, 'billing.usage.record', `${client} records usage on Billing and holds no money capability`);++ }++ // Plan T15: Bot mints its own OpenRe token for the robots' streams, sessions, outputs and toggles; nothing beyond OpenRe's++ // stream, key, session-read and output capabilities (never openre.session.end).++ const bo = await token({ client_id: 'bot', client_secret: 'bot-secret', audience: 'openvibe.openre' });++ assert.strictEqual(bo.status, 200, JSON.stringify(bo.body));++ assert.deepStrictEqual(bo.body.scope.split(' ').sort(), ['openre.key.rotate', 'openre.output.read', 'openre.output.write', 'openre.session.read', 'openre.stream.read', 'openre.stream.write'], "Bot runs its robots' OpenRe streams");// Coupons subscribes to its Events sources with its own token (OpenVibe.Coupons scripts/subscribe.js).await db.prepare("UPDATE oauth_clients SET client_secret = 'coupons-secret' WHERE client_id = 'coupons'").run();const cps = await token({ client_id: 'coupons', client_secret: 'coupons-secret', audience: 'openvibe.events', scope: 'events.subscription.manage' });nnew file mode 100644@@@ -0,0 +1,146 @@++'use strict';++// Universal telemetry (server/telemetry.js on openvibe-sdk/telemetry, plan T1): every emitted++// platform.telemetry-sample@1 validates, HTTP requests are aggregated per route|method|status_class++// (one sample for each such group at each flush, with count, sum, max and p95 in extra), skipped paths produce nothing, the three++// autoscaling gauges are emitted once per flush, one flush posts one batch, stop() flushes once and is++// idempotent, the analytics sink maps a batch to custom events, and nothing starts at module load.++// No database: the sink is injected.++// node --test test/telemetry.test.js++const assert = require('assert');++const http = require('http');++const express = require('express');++const telemetry = require('../server/telemetry');++const { createNetworkTelemetry, analyticsSink, validateTelemetrySample } = require('../server/telemetry');++const observability = require('../server/observability');++++const quiet = { log() {}, warn() {}, error() {} };++const AT = '2026-09-30T10:00:00.000Z';++const valid = (sample, label) => {++ const r = validateTelemetrySample(sample);++ assert.ok(r.ok, `${label}: ${JSON.stringify(r.errors)} — ${JSON.stringify(sample)}`);++};++++(async () => {++ // ── Item 2: requiring the modules starts nothing (no monitor, no timer) ──++ assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'requiring observability starts no event-loop monitor');++++ // ── Every emitted sample is a well-formed platform.telemetry-sample@1 ──++ const batches = [];++ const t = createNetworkTelemetry({ sink: async (samples) => { batches.push(samples); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });++ const request = t.record('http.request', 12.5, { status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } });++ const gauge = t.gauge('demo.queue_depth', 3);++ const count = t.count('cache.miss');++ for (const [sample, label] of [[request, 'record'], [gauge, 'gauge'], [count, 'count']]) {++ valid(sample, label);++ assert.strictEqual(sample.service, 'network', `${label} carries the service`);++ assert.strictEqual(sample.at, AT, `${label} carries at`);++ }++++ // ── A flush posts exactly one well-formed batch, drains the interval's gauges, stop() is idempotent ──++ await t.flush();++ assert.strictEqual(batches.length, 1, 'one flush = one sink call');++ assert.deepStrictEqual(batches[0].slice(0, 3), [request, gauge, count], 'the buffered samples, in order');++ const active0 = batches[0].find((s) => s.operation === 'http.active_requests');++ assert.ok(active0, 'the active-requests gauge is emitted once per flush');++ assert.deepStrictEqual(active0.extra, { last: 0, max: 0 }, 'the gauge value lives in extra, not latency_ms');++ assert.ok(!('latency_ms' in active0), 'a gauge is never in the latency field');++ for (const s of batches[0]) valid(s, 'flushed batch');++ await t.flush();++ assert.strictEqual(batches.length, 2, 'a second flush emits the gauges again (once per flush)');++ assert.strictEqual(batches[1].length, 1);++ assert.strictEqual(batches[1][0].operation, 'http.active_requests');++ t.record('http.request', 7, { status: '5xx' });++ await t.stop();++ assert.strictEqual(batches.length, 3, 'stop() flushes the remaining buffer once');++ assert.strictEqual(batches[2].length, 2);++ await t.stop();++ await t.flush();++ assert.strictEqual(batches.length, 3, 'a second stop() (or a later flush) posts nothing');++++ // ── Item 1: N requests to one route aggregate into ONE http.request sample with count N ──++ const aggBatches = [];++ const a = createNetworkTelemetry({ sink: async (s) => { aggBatches.push(s); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });++ for (let i = 0; i < 5; i++) a.requestStarted(); // all in flight at once: the peak is 5++ for (let i = 0; i < 5; i++) a.requestFinished({ route: '/api/v1/things/:id', method: 'GET', httpStatus: 200, latencyMs: 10 + i });++ await a.flush();++ const httpSamples = aggBatches[0].filter((s) => s.operation === 'http.request');++ assert.strictEqual(httpSamples.length, 1, `five requests to one route = one sample: ${JSON.stringify(aggBatches[0])}`);++ const agg = httpSamples[0];++ assert.strictEqual(agg.resource, '/api/v1/things/:id', 'the route template, never the raw id');++ assert.strictEqual(agg.status, '2xx');++ assert.strictEqual(agg.latency_ms, 12, 'the mean latency is the sample’s latency_ms');++ assert.deepStrictEqual(agg.extra, { method: 'GET', status_class: '2xx', count: 5, sum_ms: 60, max_ms: 14, p95_ms: 14, http_status: 200 },++ 'count, sum, max and p95 live in extra');++ for (const s of aggBatches[0]) valid(s, 'aggregated batch');++ const activeAgg = aggBatches[0].filter((s) => s.operation === 'http.active_requests');++ assert.strictEqual(activeAgg.length, 1, 'the active gauge is emitted once per flush');++ assert.deepStrictEqual(activeAgg[0].extra, { last: 0, max: 5 }, 'item 3: the peak is sampled at request start');++ assert.strictEqual(aggBatches[0].filter((s) => s.operation === 'http.latency_p95_ms').length, 1, 'p95 is emitted once per flush');++ assert.strictEqual(aggBatches[0].find((s) => s.operation === 'http.latency_p95_ms').latency_ms, 14);++ await a.flush();++ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.request').length, 0, 'an idle interval emits no request sample');++ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.active_requests').length, 1);++ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.latency_p95_ms').length, 0, 'no requests: no p95');++ await a.stop();++++ // ── The analytics sink writes the batch as custom events and flushes once ──++ const events = [];++ const analytics = { trackEvent: (name, data) => events.push({ name, data }), flush() { analytics.flushes = (analytics.flushes || 0) + 1; return Promise.resolve(); } };++ await analyticsSink(analytics)([++ { service: 'network', operation: 'http.request', at: AT, latency_ms: 12.5, status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } },++ { service: 'network', operation: 'a b/c:d', at: AT, latency_ms: 1 },++ ]);++ assert.strictEqual(analytics.flushes, 1, 'the sink flushes the tracker once per batch');++ assert.strictEqual(events.length, 2);++ assert.match(events[0].name, /^telemetry\.http\.request$/);++ assert.deepStrictEqual(events[0].data, { path: '/api/v1/x', method: 'GET', status_code: 200, response_time_ms: 12.5 });++ assert.strictEqual(events[1].name, 'telemetry.a_b_c:d', 'spaces and slashes are sanitised; : is valid in analytics/event.v1');++++ // ── Items 1 & 2: the middleware aggregates; skipped paths are never counted; init starts the monitor ──++ const raw = [];++ telemetry.init({ sink: async (samples) => { for (const s of samples) raw.push(s); }, intervalMs: 3600e3, log: quiet });++ assert.strictEqual(observability.eventLoopMonitorEnabled(), true, 'init() creates and enables the event-loop monitor');++ await new Promise((r) => setTimeout(r, 60)); // let the monitor take a sample before the flush++ const app = express();++ app.use(observability.telemetryMiddleware);++ app.get('/api/v1/things/:id', (req, res) => res.json({ ok: true }));++ app.get('/api/health', (_req, res) => res.json({ ok: true }));++ app.get('/metrics', (_req, res) => res.type('text/plain').send('ok'));++ app.get('/api/ready', (_req, res) => res.json({ ok: true }));++ app.get('/api/chrome', (_req, res) => res.json({ ok: true }));++ app.get('/shared/navbar.js', (_req, res) => res.type('application/javascript').send('ok'));++ app.get('/logo.png', (_req, res) => res.type('image/png').send('ok'));++ app.get('/data/avatars/a.png', (_req, res) => res.type('image/png').send('ok'));++ app.get('/release.json', (_req, res) => res.json({ ok: true }));++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${server.address().port}`;++ await fetch(`${base}/api/v1/things/42`);++ await fetch(`${base}/api/v1/things/43`);++ for (const p of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/navbar.js', '/logo.png', '/data/avatars/a.png']) await fetch(base + p);++ await fetch(`${base}/release.json`); // a JSON route is product API, not a static asset: still counted++ await telemetry.flush();++ server.close();++ const sampled = raw.filter((s) => s.operation === 'http.request');++ const things = sampled.filter((s) => s.resource === '/api/v1/things/:id');++ assert.strictEqual(things.length, 1, `two requests to one route = one sample: ${JSON.stringify(sampled)}`);++ assert.strictEqual(things[0].extra.count, 2);++ assert.strictEqual(things[0].status, '2xx');++ const resources = new Set(sampled.map((s) => s.resource));++ for (const skipped of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/*', '/logo.png', '/data/avatars/a.png']) {++ assert.ok(!resources.has(skipped), `skipped path ${skipped} produced no sample: ${JSON.stringify([...resources])}`);++ }++ assert.ok(resources.has('/release.json'), '/release.json is a route, not a static asset');++ assert.strictEqual(raw.filter((s) => s.operation === 'http.active_requests').length, 1, 'gauges once per flush');++ assert.deepStrictEqual(raw.find((s) => s.operation === 'http.active_requests').extra, { last: 0, max: 1 }, 'the two requests are sequential, so the peak is one in flight');++ const lag = raw.filter((s) => s.operation === 'http.eventloop_lag_ms');++ assert.strictEqual(lag.length, 1, 'the event-loop lag gauge is emitted once per flush');++ assert.ok(Number.isFinite(lag[0].extra.value), 'lag lives in extra');++ assert.ok(!('latency_ms' in lag[0]), 'lag is never in the latency field');++ for (const s of raw) valid(s, 'middleware batch');++++ await telemetry.stop();++ assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'stop() disables the event-loop monitor');++++ console.log('telemetry: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -45,8 +45,8 @@ Identity and account service for the OpenVibe network. Manages user accounts, OA
@@ -45,8 +45,8 @@ Identity and account service for the OpenVibe network. Manages user accounts, OA
## Depends on
## Depends on
-- `openvibe-contracts` v0.80.0 (registry manifests, capability checks), `openvibe-sdk` v0.25.1 (PostgreSQL
- and Valkey adapters, events, per-actor limits), `openvibe-shared` v1.29.2, pinned by release tarball
+- `openvibe-contracts` v0.80.0 (registry manifests, capability checks), `openvibe-sdk` v0.28.0 (PostgreSQL
+ and Valkey adapters, events, per-actor limits, universal telemetry), `openvibe-shared` v1.29.2, pinned by release tarball
- OpenVibe.Events (Network's outbox relay; the events it consumes), OpenVibe.AI (the `network.site_copy`
- OpenVibe.Events (Network's outbox relay; the events it consumes), OpenVibe.AI (the `network.site_copy`
workflow), OpenVibe.Blog (the changelog feed it proxies)
workflow), OpenVibe.Blog (the changelog feed it proxies)
- email (Resend), Discord and GitHub when their secrets are set ([Provider secrets](#provider-secrets))
- email (Resend), Discord and GitHub when their secrets are set ([Provider secrets](#provider-secrets))
@@ -560,6 +560,12 @@ Google-style account management supporting up to 5 accounts:
@@ -560,6 +560,12 @@ Google-style account management supporting up to 5 accounts:
open tabs report to `POST /release-metrics`). Network's own counters are
open tabs report to `POST /release-metrics`). Network's own counters are
`network_tokens_issued_total{grant_type}`, `network_token_failures_total{grant_type,error}` and
`network_tokens_issued_total{grant_type}`, `network_token_failures_total{grant_type,error}` and
`network_principal_token_failures_total{code}`.
`network_principal_token_failures_total{code}`.
+- Universal telemetry: requests are aggregated per `route|method|status_class` into a single
+ `platform.telemetry-sample@1` per key per flush (count, sum, max and p95 in `extra`, mean in
+ `latency_ms`), plus the HTTP autoscaling gauges (active requests, p95, event-loop lag) once per flush;
+ health, readiness, metrics, chrome and static/shared assets are skipped. Buffered by
+ `openvibe-sdk/telemetry` and written into the analytics store (`server/telemetry.js`,
+ `server/observability.js`); `gracefulStop` flushes it once at shutdown.
- `GET /release.json` is Network's release manifest (ADR-016, `registry.release-manifest@1`, from
- `GET /release.json` is Network's release manifest (ADR-016, `registry.release-manifest@1`, from
`openvibe-shared/release`'s `release.mount`): the deployed commit, the library versions and, since
`openvibe-shared/release`'s `release.mount`): the deployed commit, the library versions and, since
openvibe-contracts 0.32.0, the 1.1.0 fields. No components are declared, so every release still
openvibe-contracts 0.32.0, the 1.1.0 fields. No components are declared, so every release still
@@ -42,7 +42,8 @@
@@ -42,7 +42,8 @@
"service registry, public discovery and the status page with incidents",
"service registry, public discovery and the status page with incidents",
"operator alerts relay target and admin panel",
"operator alerts relay target and admin panel",
"serves /shared/* from its pinned openvibe-shared",
"serves /shared/* from its pinned openvibe-shared",
- "per-actor limits (openvibe-sdk/limits)"
+ "per-actor limits (openvibe-sdk/limits)",
+ "universal telemetry: per-route aggregated platform.telemetry-sample@1 plus per-flush HTTP autoscaling gauges (openvibe-sdk/telemetry)"
],
],
- "updated": "2026-09-29"
+ "updated": "2026-10-04"
}
}
@@ -22,7 +22,7 @@
@@ -22,7 +22,7 @@
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"uuid": "^11.1.0",
"uuid": "^11.1.0",
@@ -1572,9 +1572,9 @@
@@ -1572,9 +1572,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.26.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
- "integrity": "sha512-+RK89ZG4WAKIRh89yr3ieIvnyJVNSOI+dsbBqGNcr9UHsJStdxtM75zSDq8cUNpBPQ1X6ElatZrjU9S3s0aEkg==",
+ "version": "0.28.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
+ "integrity": "sha512-jjLisFDWBN+xDVsixa5BJRr3355kRijmvOh1/9rz4s1yVb3wZWCbpTLNcOvXm57ksnRoqK57393W38G0Ey3PFw==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -1,7 +1,7 @@
@@ -1,7 +1,7 @@
{
{
"name": "openvibe-network",
"name": "openvibe-network",
"version": "1.0.0",
"version": "1.0.0",
- "description": "OpenVibe.Network — Identity, SSO, Accounts, Themes, Notifications, and OpenCoins Wallet",
+ "description": "OpenVibe.Network \u2014 Identity, SSO, Accounts, Themes, Notifications, and OpenCoins Wallet",
"main": "server/index.js",
"main": "server/index.js",
"scripts": {
"scripts": {
"start": "node server/index.js",
"start": "node server/index.js",
@@ -31,7 +31,7 @@
@@ -31,7 +31,7 @@
"jsonwebtoken": "^9.0.0",
"jsonwebtoken": "^9.0.0",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.94.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.28.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.8.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
"uuid": "^11.1.0",
"uuid": "^11.1.0",
@@ -156,6 +156,10 @@ const DEFAULT_GRANTS = [
@@ -156,6 +156,10 @@ const DEFAULT_GRANTS = [
['host', 'network.resource.report', SELF_AUDIENCE, []],
['host', 'network.resource.report', SELF_AUDIENCE, []],
// Plan T2 N4b: Bot pairs its users' machines with Network and reads or revokes only the principals it paired.
// Plan T2 N4b: Bot pairs its users' machines with Network and reads or revokes only the principals it paired.
['bot', 'network.node.manage', SELF_AUDIENCE, []],
['bot', 'network.node.manage', SELF_AUDIENCE, []],
+ // Plan T15: Bot runs each robot's OpenRe stream for its owner (find/create/archive, key rotation), plays the live
+ // session in the panel (session.read), restreams out (output.*) and sets the owner's streaming and recording
+ // toggles on the stream, minting its own token with its Network client instead of a hand-set BOT_OPENRE_TOKEN.
+ ...['openre.stream.read', 'openre.stream.write', 'openre.key.rotate', 'openre.session.read', 'openre.output.read', 'openre.output.write'].map(c => ['bot', c, 'openvibe.openre', []]),
// Plan T2 lane B step 3: Host places and runs per project, so it reads a project's tenancy, placement and quotas.
// Plan T2 lane B step 3: Host places and runs per project, so it reads a project's tenancy, placement and quotas.
// Other services get this row only when they ship a caller of GET /internal/projects/:project_id.
// Other services get this row only when they ship a caller of GET /internal/projects/:project_id.
['host', 'network.project.read', SELF_AUDIENCE, []],
['host', 'network.project.read', SELF_AUDIENCE, []],
@@ -254,6 +258,13 @@ const DEFAULT_GRANTS = [
@@ -254,6 +258,13 @@ const DEFAULT_GRANTS = [
['tools', 'search.document.write', 'openvibe.search', []],
['tools', 'search.document.write', 'openvibe.search', []],
// Plan T5 step 7: Tools posts one usage reading per ended tool job to Billing (POST /api/v1/usage).
// Plan T5 step 7: Tools posts one usage reading per ended tool job to Billing (POST /api/v1/usage).
['tools', 'billing.usage.record', 'openvibe.billing', []],
['tools', 'billing.usage.record', 'openvibe.billing', []],
+ // Plan T5 step 14: Network's own usage producers — AI (per model run), Events (per delivered event)
+ // and Media (per stored object) post their platform.usage-sample@1 readings to Billing
+ // (POST /api/v1/usage) with their own token. Recording usage only: none of them holds a money
+ // capability (billing.ledger.admin / billing.cashout.manage).
+ ['ai', 'billing.usage.record', 'openvibe.billing', []],
+ ['events', 'billing.usage.record', 'openvibe.billing', []],
+ ['media', 'billing.usage.record', 'openvibe.billing', []],
];
];
// Grants withdrawn by decision; applied at every boot so an old default can't come back.
// Grants withdrawn by decision; applied at every boot so an old default can't come back.
@@ -335,6 +335,12 @@ config._registry = resolvedRegistry;
@@ -335,6 +335,12 @@ config._registry = resolvedRegistry;
const analytics = networkAnalytics.openAnalytics(db);
const analytics = networkAnalytics.openAnalytics(db);
app.locals.analytics = analytics;
app.locals.analytics = analytics;
app.use(analytics.middleware());
app.use(analytics.middleware());
+// Universal telemetry (plan T1, server/telemetry.js): one platform.telemetry-sample@1 per request plus
+// the HTTP autoscaling signals, into the same analytics store. Wired before any route runs; flushed once
+// by gracefulStop's stop array below.
+const telemetry = require('./telemetry');
+telemetry.init({ analytics });
+app.use(observability.telemetryMiddleware);
// ── Extract bearer token (available as req.token for optional-auth proxies) ─
// ── Extract bearer token (available as req.token for optional-auth proxies) ─
app.use((req, _res, next) => {
app.use((req, _res, next) => {
@@ -985,6 +991,7 @@ gracefulStop({
@@ -985,6 +991,7 @@ gracefulStop({
() => require('./identity/profile-events').stop(),
() => require('./identity/profile-events').stop(),
() => require('./identity/grants-admin').stop(),
() => require('./identity/grants-admin').stop(),
() => require('./developer/confirmations').stop(),
() => require('./developer/confirmations').stop(),
+ () => telemetry.stop(),
],
],
close: [
close: [
() => within(1500, require('./developer/event-relay').stopRelay(db)),
() => within(1500, require('./developer/event-relay').stopRelay(db)),
@@ -9,9 +9,11 @@
@@ -9,9 +9,11 @@
* and service-token (principal) failures at Network's capability-guarded internal routes.
* and service-token (principal) failures at Network's capability-guarded internal routes.
* Labels are fixed vocabularies; a client id, subject or token never becomes a label.
* Labels are fixed vocabularies; a client id, subject or token never becomes a label.
*/
*/
+const path = require('path');
const jwt = require('jsonwebtoken');
const jwt = require('jsonwebtoken');
const metrics = require('openvibe-shared/metrics');
const metrics = require('openvibe-shared/metrics');
const { createReadiness, skip } = require('openvibe-shared/ready');
const { createReadiness, skip } = require('openvibe-shared/ready');
+const telemetry = require('./telemetry');
const registry = metrics.createRegistry();
const registry = metrics.createRegistry();
@@ -50,6 +52,79 @@ function principalDenied({ req, code }) {
@@ -50,6 +52,79 @@ function principalDenied({ req, code }) {
principalFailures.inc({ code: c });
principalFailures.inc({ code: c });
}
}
+// ── Universal telemetry (plan T1, §15.19): aggregated platform.telemetry-sample@1 + autoscaling ──
+// A request is not a row: the middleware hands each finished request to server/telemetry.js, which
+// aggregates per route|method|status_class and emits one http.request sample per key per flush (plus the
+// HTTP autoscaling gauges once per flush). A sample carries the route template, method and status class —
+// never a raw URL, client id or token. Health, readiness, metrics, chrome and static/shared assets carry
+// no product signal and are skipped before anything is counted.
+
+// The event-loop monitor: created and enabled by startEventLoopMonitor (called from telemetry.init), and
+// disabled by stopEventLoopMonitor (telemetry.stop). Nothing runs at module load.
+let loopMonitor = null;
+function startEventLoopMonitor() {
+ if (loopMonitor) return;
+ try {
+ loopMonitor = require('perf_hooks').monitorEventLoopDelay({ resolution: 20 });
+ loopMonitor.enable();
+ } catch { loopMonitor = null; }
+}
+function stopEventLoopMonitor() {
+ if (!loopMonitor) return;
+ try { loopMonitor.disable(); } catch { /* already gone */ }
+ loopMonitor = null;
+}
+/** Whether the monitor is running (tests assert requiring the module starts nothing). */
+function eventLoopMonitorEnabled() { return loopMonitor != null; }
+
+/** The event-loop lag since the last flush, ms (mean includes the sampling interval: subtracted); null
+ * before the monitor has a sample. The monitor resets per read, so each gauge covers one flush interval. */
+function eventLoopLagMs() {
+ if (!loopMonitor || !(loopMonitor.count > 0 || loopMonitor.max > 0)) return null;
+ const lag = Math.max(0, loopMonitor.mean / 1e6 - 20);
+ loopMonitor.reset();
+ return lag;
+}
+
+// server/telemetry.js owns the aggregation; it pulls the lag and calls these lifecycle hooks.
+telemetry.registerSignals({ start: startEventLoopMonitor, stop: stopEventLoopMonitor, lag: eventLoopLagMs });
+
+const SKIP_EXACT = new Set(['/api/health', '/ready', '/api/ready', '/metrics']);
+const SKIP_PREFIXES = ['/shared', '/api/chrome'];
+// Anything express.static serves: public/ files and /data/avatars images. JSON is deliberately not here
+// (routes such as /release.json and /contracts/*.json are product/observability API, not static assets).
+const STATIC_EXTENSIONS = new Set([
+ '.js', '.mjs', '.cjs', '.css', '.map', '.html', '.htm', '.txt', '.xml', '.webmanifest', '.wasm',
+ '.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp', '.avif', '.ico', '.bmp',
+ '.woff', '.woff2', '.ttf', '.otf', '.eot', '.mp3', '.mp4', '.webm', '.ogg', '.pdf',
+]);
+function telemetrySkipped(req) {
+ const p = req.path || '';
+ if (SKIP_EXACT.has(p)) return true;
+ if (SKIP_PREFIXES.some((prefix) => p === prefix || p.startsWith(`${prefix}/`))) return true;
+ return STATIC_EXTENSIONS.has(path.extname(p).toLowerCase());
+}
+
+/** Express middleware: hand the finished request to the aggregator; the in-flight peak is sampled at start. */
+function telemetryMiddleware(req, res, next) {
+ if (telemetrySkipped(req)) return next();
+ telemetry.requestStarted();
+ const t0 = process.hrtime.bigint();
+ let done = false;
+ const finish = () => {
+ if (done) return;
+ done = true;
+ try {
+ const latencyMs = Number(process.hrtime.bigint() - t0) / 1e6;
+ const httpStatus = res.writableFinished || res.finished ? (res.statusCode || 0) : 0;
+ telemetry.requestFinished({ route: metrics.routeLabel(req), method: req.method, httpStatus, latencyMs });
+ } catch { /* telemetry never breaks a request */ }
+ };
+ res.once('finish', finish);
+ res.once('close', finish);
+ next();
+}
+
/**
/**
* Readiness with Network's real dependencies:
* Readiness with Network's real dependencies:
* db (required) a query against the identity database
* db (required) a query against the identity database
@@ -98,4 +173,4 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discor
@@ -98,4 +173,4 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discor
return createReadiness({ service: 'network', release, checks });
return createReadiness({ service: 'network', release, checks });
}
}
-module.exports = { registry, tokenEndpointMetrics, principalDenied, createNetworkReadiness, grantLabel };
+module.exports = { registry, tokenEndpointMetrics, principalDenied, telemetryMiddleware, telemetrySkipped, createNetworkReadiness, grantLabel, eventLoopMonitorEnabled };
new file mode 100644
new file mode 100644
@@ -0,0 +1,226 @@
@@ -0,0 +1,226 @@
+'use strict';
+/**
+ * Network's platform.telemetry-sample@1 collector (openvibe-sdk/telemetry; plan T1 universal Fabric,
+ * §15.19). One sample schema every service emits; autoscaling and routing consume it.
+ *
+ * createNetworkTelemetry({ analytics }) service 'network', with a sink that writes each flushed batch
+ * into Network's ADR-021 analytics tracker as custom (non-HTTP)
+ * events and flushes it, so one batch lands together.
+ * init({ analytics }) the process-wide collector, wired once at boot (server/index.js).
+ * record / gauge / count / flush / stop the collector's API; before init they are no-ops.
+ *
+ * Volume: one analytics row per request does not scale (at 50 req/s that is ~200 rows/s). HTTP requests
+ * are therefore aggregated per `route|method|status_class` per flush interval into count, sum, max and
+ * p95, and ONE `http.request` sample per key is emitted per flush — not one per request. The HTTP
+ * autoscaling gauges (active requests, p95, event-loop lag) are emitted once per flush from a timer
+ * started in init(), never per request, so the SDK buffer (maxBuffered 1000) cannot overflow.
+ *
+ * Field mapping (platform.telemetry-sample@1; `latency_ms` is the schema's latency field, there is no
+ * generic gauge field): an aggregated request carries its mean in `latency_ms` and count/sum/max/p95 in
+ * `extra`; the p95 gauge carries its value in `latency_ms` (it is a latency); the active-requests and
+ * event-loop-lag gauges carry their value in `extra` — never in `latency_ms`, which the analytics sink
+ * maps to `response_time_ms` and would make indistinguishable from a latency.
+ *
+ * server/observability.js feeds the middleware's per-request observations and registers the event-loop
+ * monitor (started by init, stopped by stop) so nothing starts at module load; gracefulStop's stop array
+ * flushes it once at shutdown.
+ */
+const { createTelemetry, telemetrySample, validateTelemetrySample } = require('openvibe-sdk/telemetry');
+
+const SERVICE = 'network';
+const DEFAULT_INTERVAL_MS = 15000;
+
+// analytics/event.v1 event_type: [A-Za-z][A-Za-z0-9_.:-]{0,63}; an operation is sanitised to fit.
+const EVENT_RE = /^[A-Za-z][A-Za-z0-9_.:-]{0,63}$/;
+function eventName(operation) {
+ const name = `telemetry.${String(operation == null ? 'operation' : operation)}`.replace(/[^A-Za-z0-9_.:-]/g, '_').slice(0, 64);
+ return EVENT_RE.test(name) ? name : 'telemetry.operation';
+}
+
+/**
+ * The sink: write each sample in the batch into Network's analytics tracker (its `trackEvent`, the
+ * non-HTTP form; ADR-021 reduces the personal fields), then await its flush so the batch lands together.
+ * Only the fields analytics/event.v1 can hold are passed; the numbers live in `extra.http_status`.
+ */
+function analyticsSink(analytics) {
+ if (!analytics || typeof analytics.trackEvent !== 'function') {
+ throw new TypeError('analyticsSink: Network\'s analytics tracker (trackEvent) is required');
+ }
+ return async function sink(samples) {
+ for (const s of samples) {
+ const extra = s.extra || {};
+ analytics.trackEvent(eventName(s.operation), {
+ path: typeof s.resource === 'string' ? s.resource : (typeof extra.route === 'string' ? extra.route : undefined),
+ method: typeof extra.method === 'string' ? extra.method : undefined,
+ status_code: Number.isInteger(extra.http_status) ? extra.http_status : undefined,
+ response_time_ms: Number.isFinite(s.latency_ms) ? s.latency_ms : undefined,
+ });
+ }
+ if (typeof analytics.flush === 'function') await analytics.flush();
+ };
+}
+
+// ── HTTP aggregation (volume) ─────────────────────────────────────
+const HTTP_METHODS = new Set(['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']);
+const statusClass = (code) => (code ? `${Math.floor(code / 100)}xx` : 'aborted');
+const round = (n, digits = 3) => { const f = 10 ** digits; return Math.round(n * f) / f; };
+
+/** p95 (nearest-rank) of an array of ms latencies; null when empty. Computed at most once per flush. */
+function percentile95(values) {
+ if (!values.length) return null;
+ const sorted = [...values].sort((a, b) => a - b);
+ return sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * 0.95))];
+}
+
+// The event-loop monitor lives in observability.js and is injected here at its require: requiring either
+// module starts nothing. telemetry.init() calls start(), telemetry.stop() calls stop(), and the gauges
+// read lag(). Unregistered (unit tests, a bare collector) they are safe no-ops.
+let signals = { start() {}, stop() {}, lag() { return null; } };
+function registerSignals(injected) {
+ signals = { ...signals, ...(injected || {}) };
+}
+
+/**
+ * Build one collector. `sink` overrides the analytics sink (tests inject one); everything else is the SDK.
+ * HTTP observations are aggregated in memory and drained on flush(); the flush timer itself starts in
+ * init(), so a bare collector only flushes when asked.
+ */
+function createNetworkTelemetry({ analytics = null, sink = null, instance = null, intervalMs = DEFAULT_INTERVAL_MS, now, log, maxBuffered } = {}) {
+ const write = typeof sink === 'function' ? sink : analyticsSink(analytics);
+ const sdk = createTelemetry({ service: SERVICE, instance, sink: write, intervalMs, now, log, maxBuffered });
+
+ let routes = new Map(); // key -> { route, method, status, count, sumMs, maxMs, latencies, codes }
+ let allLatencies = []; // the flush interval's latencies, for the single p95 gauge
+ let active = 0; // requests in flight right now
+ let activeMax = 0; // the interval's peak, sampled at request start (item 3)
+ let stopped = false;
+
+ /** A request began: track the in-flight peak here, not after the decrement. */
+ function requestStarted() {
+ active += 1;
+ if (active > activeMax) activeMax = active;
+ }
+
+ function observeRequest(info = {}) {
+ const r = String(info.route == null ? 'unmatched' : info.route).slice(0, 200);
+ const m = HTTP_METHODS.has(info.method) ? info.method : 'OTHER';
+ const s = statusClass(info.httpStatus);
+ const latency = Number.isFinite(info.latencyMs) && info.latencyMs >= 0 ? info.latencyMs : 0;
+ const key = `${r}|${m}|${s}`;
+ let bucket = routes.get(key);
+ if (!bucket) {
+ bucket = { route: r, method: m, status: s, count: 0, sumMs: 0, maxMs: 0, latencies: [], codes: new Set() };
+ routes.set(key, bucket);
+ }
+ bucket.count += 1;
+ bucket.sumMs += latency;
+ if (latency > bucket.maxMs) bucket.maxMs = latency;
+ bucket.latencies.push(latency);
+ if (Number.isInteger(info.httpStatus) && info.httpStatus > 0) bucket.codes.add(info.httpStatus);
+ allLatencies.push(latency);
+ }
+
+ /** A request ended: free the in-flight slot, then fold it into its aggregation key. */
+ function requestFinished(info) {
+ if (active > 0) active -= 1;
+ observeRequest(info);
+ }
+
+ /** Push a gauge whose value(s) live in `extra`; the schema has no generic gauge field and latency_ms
+ * is the sink's response_time_ms, so extra is the documented home (never latency_ms). */
+ function emitExtraGauge(name, extra) {
+ const sample = sdk.record(name, 0, { extra });
+ delete sample.latency_ms;
+ return sample;
+ }
+
+ /** Drain this interval: one http.request sample per route|method|status_class, then the three gauges. */
+ function emitSamples() {
+ const buckets = [...routes.values()];
+ const latencies = allLatencies;
+ routes = new Map();
+ allLatencies = [];
+ for (const b of buckets) {
+ const mean = b.count ? b.sumMs / b.count : 0;
+ const p95 = percentile95(b.latencies);
+ const extra = {
+ method: b.method,
+ status_class: b.status,
+ count: b.count,
+ sum_ms: round(b.sumMs),
+ max_ms: round(b.maxMs),
+ };
+ if (p95 != null) extra.p95_ms = round(p95);
+ if (b.codes.size === 1) extra.http_status = b.codes.values().next().value;
+ sdk.record('http.request', round(mean), { status: b.status, resource: b.route, extra });
+ }
+ // The gauges, once per flush (never per request).
+ emitExtraGauge('http.active_requests', { last: active, max: activeMax });
+ activeMax = active; // the next interval's peak starts from what is still in flight
+ const p95 = percentile95(latencies);
+ if (p95 != null) sdk.record('http.latency_p95_ms', round(p95)); // latency_ms: the schema's latency field
+ const lag = signals.lag();
+ if (lag != null) emitExtraGauge('http.eventloop_lag_ms', { value: round(lag) });
+ }
+
+ /** Emit the interval's samples, then hand the SDK everything buffered (a flush is also the tick). */
+ function flush() {
+ if (stopped) return Promise.resolve();
+ emitSamples();
+ return sdk.flush();
+ }
+
+ /** Emit once, then let the SDK clear its timer and flush (idempotent). */
+ async function stop() {
+ if (stopped) return;
+ stopped = true;
+ emitSamples();
+ await sdk.stop();
+ }
+
+ return { record: sdk.record, gauge: sdk.gauge, count: sdk.count, flush, stop, requestStarted, requestFinished, observeRequest };
+}
+
+// The process-wide collector. It exists only once the analytics tracker does (server/index.js), so an
+// early request is recorded by neither; the HTTP middleware is mounted before any route runs.
+let current = null;
+let timer = null;
+function init(opts = {}) {
+ current = createNetworkTelemetry(opts);
+ signals.start(); // the event-loop monitor (observability.js): created and enabled here, not at load
+ if (timer) clearInterval(timer);
+ const interval = Number.isFinite(opts.intervalMs) ? opts.intervalMs : DEFAULT_INTERVAL_MS;
+ timer = setInterval(() => { current.flush(); }, interval);
+ if (timer.unref) timer.unref();
+ return current;
+}
+
+function record(...args) { return current ? current.record(...args) : undefined; }
+function gauge(...args) { return current ? current.gauge(...args) : undefined; }
+function count(...args) { return current ? current.count(...args) : undefined; }
+function requestStarted() { return current ? current.requestStarted() : undefined; }
+function requestFinished(info) { return current ? current.requestFinished(info) : undefined; }
+function flush() { return current ? current.flush() : Promise.resolve(); }
+async function stop() {
+ if (timer) { clearInterval(timer); timer = null; }
+ if (current) await current.stop();
+ signals.stop(); // disable the event-loop monitor that init enabled
+}
+
+module.exports = {
+ SERVICE,
+ eventName,
+ analyticsSink,
+ createNetworkTelemetry,
+ telemetrySample,
+ validateTelemetrySample,
+ registerSignals,
+ init,
+ record,
+ gauge,
+ count,
+ requestStarted,
+ requestFinished,
+ flush,
+ stop,
+};
@@ -22,6 +22,10 @@ await db.prepare("UPDATE oauth_clients SET client_secret = 'live-secret' WHERE c
@@ -22,6 +22,10 @@ await db.prepare("UPDATE oauth_clients SET client_secret = 'live-secret' WHERE c
await db.prepare("UPDATE oauth_clients SET client_secret = 'media-secret' WHERE client_id = 'media'").run();
await db.prepare("UPDATE oauth_clients SET client_secret = 'media-secret' WHERE client_id = 'media'").run();
// The `search` client is created with server/setup/service-principal.js, not seeded with the signed-in products.
// The `search` client is created with server/setup/service-principal.js, not seeded with the signed-in products.
await db.prepare("INSERT INTO oauth_clients (client_id, client_secret, name, redirect_uris, is_first_party) VALUES ('search', 'search-secret', 'OpenVibe.Search', '[]', 1) ON CONFLICT DO NOTHING").run();
await db.prepare("INSERT INTO oauth_clients (client_id, client_secret, name, redirect_uris, is_first_party) VALUES ('search', 'search-secret', 'OpenVibe.Search', '[]', 1) ON CONFLICT DO NOTHING").run();
+// Events is created the same way (setup/service-principal.js); it is a usage producer with its own token.
+await db.prepare("INSERT INTO oauth_clients (client_id, client_secret, name, redirect_uris, is_first_party) VALUES ('events', 'events-secret', 'OpenVibe.Events', '[]', 1) ON CONFLICT DO NOTHING").run();
+await db.prepare("UPDATE oauth_clients SET client_secret = 'ai-secret' WHERE client_id = 'ai'").run();
+await db.prepare("UPDATE oauth_clients SET client_secret = 'bot-secret' WHERE client_id = 'bot'").run();
await db.prepare("INSERT INTO users (id, username, password_hash) VALUES (7, 'payee', 'x'), (8, 'payer', 'x')").run();
await db.prepare("INSERT INTO users (id, username, password_hash) VALUES (7, 'payee', 'x'), (8, 'payer', 'x')").run();
// A database seeded before media.analyze (Live's AI grants as the old default) is moved at the next boot.
// A database seeded before media.analyze (Live's AI grants as the old default) is moved at the next boot.
await db.prepare("UPDATE principal_grants SET namespaces = ? WHERE client_id = 'live' AND audience = 'openvibe.ai'").run(JSON.stringify(['live.*', 'network.site_copy']));
await db.prepare("UPDATE principal_grants SET namespaces = ? WHERE client_id = 'live' AND audience = 'openvibe.ai'").run(JSON.stringify(['live.*', 'network.site_copy']));
@@ -178,6 +182,18 @@ const server = http.createServer(app);
@@ -178,6 +182,18 @@ const server = http.createServer(app);
assert.strictEqual(ts.body.scope, 'search.document.write', 'Tools indexes its tools in Search');
assert.strictEqual(ts.body.scope, 'search.document.write', 'Tools indexes its tools in Search');
const tb = await token({ client_id: 'tools', client_secret: 'tools-secret', audience: 'openvibe.billing' });
const tb = await token({ client_id: 'tools', client_secret: 'tools-secret', audience: 'openvibe.billing' });
assert.strictEqual(tb.body.scope, 'billing.usage.record', 'Tools posts usage readings to Billing and holds no money capability');
assert.strictEqual(tb.body.scope, 'billing.usage.record', 'Tools posts usage readings to Billing and holds no money capability');
+ // Plan T5 step 14: Network's own usage producers post platform.usage-sample@1 to Billing with their
+ // own token; usage recording only, never a money capability.
+ for (const [client, secret] of [['ai', 'ai-secret'], ['events', 'events-secret'], ['media', 'media-secret']]) {
+ const ub = await token({ client_id: client, client_secret: secret, audience: 'openvibe.billing' });
+ assert.strictEqual(ub.status, 200, `${client}: ${JSON.stringify(ub.body)}`);
+ assert.strictEqual(ub.body.scope, 'billing.usage.record', `${client} records usage on Billing and holds no money capability`);
+ }
+ // Plan T15: Bot mints its own OpenRe token for the robots' streams, sessions, outputs and toggles; nothing beyond OpenRe's
+ // stream, key, session-read and output capabilities (never openre.session.end).
+ const bo = await token({ client_id: 'bot', client_secret: 'bot-secret', audience: 'openvibe.openre' });
+ assert.strictEqual(bo.status, 200, JSON.stringify(bo.body));
+ assert.deepStrictEqual(bo.body.scope.split(' ').sort(), ['openre.key.rotate', 'openre.output.read', 'openre.output.write', 'openre.session.read', 'openre.stream.read', 'openre.stream.write'], "Bot runs its robots' OpenRe streams");
// Coupons subscribes to its Events sources with its own token (OpenVibe.Coupons scripts/subscribe.js).
// Coupons subscribes to its Events sources with its own token (OpenVibe.Coupons scripts/subscribe.js).
await db.prepare("UPDATE oauth_clients SET client_secret = 'coupons-secret' WHERE client_id = 'coupons'").run();
await db.prepare("UPDATE oauth_clients SET client_secret = 'coupons-secret' WHERE client_id = 'coupons'").run();
const cps = await token({ client_id: 'coupons', client_secret: 'coupons-secret', audience: 'openvibe.events', scope: 'events.subscription.manage' });
const cps = await token({ client_id: 'coupons', client_secret: 'coupons-secret', audience: 'openvibe.events', scope: 'events.subscription.manage' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,146 @@
@@ -0,0 +1,146 @@
+'use strict';
+// Universal telemetry (server/telemetry.js on openvibe-sdk/telemetry, plan T1): every emitted
+// platform.telemetry-sample@1 validates, HTTP requests are aggregated per route|method|status_class
+// (one sample for each such group at each flush, with count, sum, max and p95 in extra), skipped paths produce nothing, the three
+// autoscaling gauges are emitted once per flush, one flush posts one batch, stop() flushes once and is
+// idempotent, the analytics sink maps a batch to custom events, and nothing starts at module load.
+// No database: the sink is injected.
+// node --test test/telemetry.test.js
+const assert = require('assert');
+const http = require('http');
+const express = require('express');
+const telemetry = require('../server/telemetry');
+const { createNetworkTelemetry, analyticsSink, validateTelemetrySample } = require('../server/telemetry');
+const observability = require('../server/observability');
+
+const quiet = { log() {}, warn() {}, error() {} };
+const AT = '2026-09-30T10:00:00.000Z';
+const valid = (sample, label) => {
+ const r = validateTelemetrySample(sample);
+ assert.ok(r.ok, `${label}: ${JSON.stringify(r.errors)} — ${JSON.stringify(sample)}`);
+};
+
+(async () => {
+ // ── Item 2: requiring the modules starts nothing (no monitor, no timer) ──
+ assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'requiring observability starts no event-loop monitor');
+
+ // ── Every emitted sample is a well-formed platform.telemetry-sample@1 ──
+ const batches = [];
+ const t = createNetworkTelemetry({ sink: async (samples) => { batches.push(samples); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });
+ const request = t.record('http.request', 12.5, { status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } });
+ const gauge = t.gauge('demo.queue_depth', 3);
+ const count = t.count('cache.miss');
+ for (const [sample, label] of [[request, 'record'], [gauge, 'gauge'], [count, 'count']]) {
+ valid(sample, label);
+ assert.strictEqual(sample.service, 'network', `${label} carries the service`);
+ assert.strictEqual(sample.at, AT, `${label} carries at`);
+ }
+
+ // ── A flush posts exactly one well-formed batch, drains the interval's gauges, stop() is idempotent ──
+ await t.flush();
+ assert.strictEqual(batches.length, 1, 'one flush = one sink call');
+ assert.deepStrictEqual(batches[0].slice(0, 3), [request, gauge, count], 'the buffered samples, in order');
+ const active0 = batches[0].find((s) => s.operation === 'http.active_requests');
+ assert.ok(active0, 'the active-requests gauge is emitted once per flush');
+ assert.deepStrictEqual(active0.extra, { last: 0, max: 0 }, 'the gauge value lives in extra, not latency_ms');
+ assert.ok(!('latency_ms' in active0), 'a gauge is never in the latency field');
+ for (const s of batches[0]) valid(s, 'flushed batch');
+ await t.flush();
+ assert.strictEqual(batches.length, 2, 'a second flush emits the gauges again (once per flush)');
+ assert.strictEqual(batches[1].length, 1);
+ assert.strictEqual(batches[1][0].operation, 'http.active_requests');
+ t.record('http.request', 7, { status: '5xx' });
+ await t.stop();
+ assert.strictEqual(batches.length, 3, 'stop() flushes the remaining buffer once');
+ assert.strictEqual(batches[2].length, 2);
+ await t.stop();
+ await t.flush();
+ assert.strictEqual(batches.length, 3, 'a second stop() (or a later flush) posts nothing');
+
+ // ── Item 1: N requests to one route aggregate into ONE http.request sample with count N ──
+ const aggBatches = [];
+ const a = createNetworkTelemetry({ sink: async (s) => { aggBatches.push(s); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });
+ for (let i = 0; i < 5; i++) a.requestStarted(); // all in flight at once: the peak is 5
+ for (let i = 0; i < 5; i++) a.requestFinished({ route: '/api/v1/things/:id', method: 'GET', httpStatus: 200, latencyMs: 10 + i });
+ await a.flush();
+ const httpSamples = aggBatches[0].filter((s) => s.operation === 'http.request');
+ assert.strictEqual(httpSamples.length, 1, `five requests to one route = one sample: ${JSON.stringify(aggBatches[0])}`);
+ const agg = httpSamples[0];
+ assert.strictEqual(agg.resource, '/api/v1/things/:id', 'the route template, never the raw id');
+ assert.strictEqual(agg.status, '2xx');
+ assert.strictEqual(agg.latency_ms, 12, 'the mean latency is the sample’s latency_ms');
+ assert.deepStrictEqual(agg.extra, { method: 'GET', status_class: '2xx', count: 5, sum_ms: 60, max_ms: 14, p95_ms: 14, http_status: 200 },
+ 'count, sum, max and p95 live in extra');
+ for (const s of aggBatches[0]) valid(s, 'aggregated batch');
+ const activeAgg = aggBatches[0].filter((s) => s.operation === 'http.active_requests');
+ assert.strictEqual(activeAgg.length, 1, 'the active gauge is emitted once per flush');
+ assert.deepStrictEqual(activeAgg[0].extra, { last: 0, max: 5 }, 'item 3: the peak is sampled at request start');
+ assert.strictEqual(aggBatches[0].filter((s) => s.operation === 'http.latency_p95_ms').length, 1, 'p95 is emitted once per flush');
+ assert.strictEqual(aggBatches[0].find((s) => s.operation === 'http.latency_p95_ms').latency_ms, 14);
+ await a.flush();
+ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.request').length, 0, 'an idle interval emits no request sample');
+ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.active_requests').length, 1);
+ assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.latency_p95_ms').length, 0, 'no requests: no p95');
+ await a.stop();
+
+ // ── The analytics sink writes the batch as custom events and flushes once ──
+ const events = [];
+ const analytics = { trackEvent: (name, data) => events.push({ name, data }), flush() { analytics.flushes = (analytics.flushes || 0) + 1; return Promise.resolve(); } };
+ await analyticsSink(analytics)([
+ { service: 'network', operation: 'http.request', at: AT, latency_ms: 12.5, status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } },
+ { service: 'network', operation: 'a b/c:d', at: AT, latency_ms: 1 },
+ ]);
+ assert.strictEqual(analytics.flushes, 1, 'the sink flushes the tracker once per batch');
+ assert.strictEqual(events.length, 2);
+ assert.match(events[0].name, /^telemetry\.http\.request$/);
+ assert.deepStrictEqual(events[0].data, { path: '/api/v1/x', method: 'GET', status_code: 200, response_time_ms: 12.5 });
+ assert.strictEqual(events[1].name, 'telemetry.a_b_c:d', 'spaces and slashes are sanitised; : is valid in analytics/event.v1');
+
+ // ── Items 1 & 2: the middleware aggregates; skipped paths are never counted; init starts the monitor ──
+ const raw = [];
+ telemetry.init({ sink: async (samples) => { for (const s of samples) raw.push(s); }, intervalMs: 3600e3, log: quiet });
+ assert.strictEqual(observability.eventLoopMonitorEnabled(), true, 'init() creates and enables the event-loop monitor');
+ await new Promise((r) => setTimeout(r, 60)); // let the monitor take a sample before the flush
+ const app = express();
+ app.use(observability.telemetryMiddleware);
+ app.get('/api/v1/things/:id', (req, res) => res.json({ ok: true }));
+ app.get('/api/health', (_req, res) => res.json({ ok: true }));
+ app.get('/metrics', (_req, res) => res.type('text/plain').send('ok'));
+ app.get('/api/ready', (_req, res) => res.json({ ok: true }));
+ app.get('/api/chrome', (_req, res) => res.json({ ok: true }));
+ app.get('/shared/navbar.js', (_req, res) => res.type('application/javascript').send('ok'));
+ app.get('/logo.png', (_req, res) => res.type('image/png').send('ok'));
+ app.get('/data/avatars/a.png', (_req, res) => res.type('image/png').send('ok'));
+ app.get('/release.json', (_req, res) => res.json({ ok: true }));
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${server.address().port}`;
+ await fetch(`${base}/api/v1/things/42`);
+ await fetch(`${base}/api/v1/things/43`);
+ for (const p of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/navbar.js', '/logo.png', '/data/avatars/a.png']) await fetch(base + p);
+ await fetch(`${base}/release.json`); // a JSON route is product API, not a static asset: still counted
+ await telemetry.flush();
+ server.close();
+ const sampled = raw.filter((s) => s.operation === 'http.request');
+ const things = sampled.filter((s) => s.resource === '/api/v1/things/:id');
+ assert.strictEqual(things.length, 1, `two requests to one route = one sample: ${JSON.stringify(sampled)}`);
+ assert.strictEqual(things[0].extra.count, 2);
+ assert.strictEqual(things[0].status, '2xx');
+ const resources = new Set(sampled.map((s) => s.resource));
+ for (const skipped of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/*', '/logo.png', '/data/avatars/a.png']) {
+ assert.ok(!resources.has(skipped), `skipped path ${skipped} produced no sample: ${JSON.stringify([...resources])}`);
+ }
+ assert.ok(resources.has('/release.json'), '/release.json is a route, not a static asset');
+ assert.strictEqual(raw.filter((s) => s.operation === 'http.active_requests').length, 1, 'gauges once per flush');
+ assert.deepStrictEqual(raw.find((s) => s.operation === 'http.active_requests').extra, { last: 0, max: 1 }, 'the two requests are sequential, so the peak is one in flight');
+ const lag = raw.filter((s) => s.operation === 'http.eventloop_lag_ms');
+ assert.strictEqual(lag.length, 1, 'the event-loop lag gauge is emitted once per flush');
+ assert.ok(Number.isFinite(lag[0].extra.value), 'lag lives in extra');
+ assert.ok(!('latency_ms' in lag[0]), 'lag is never in the latency field');
+ for (const s of raw) valid(s, 'middleware batch');
+
+ await telemetry.stop();
+ assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'stop() disables the event-loop monitor');
+
+ console.log('telemetry: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });