Add a per-service PostgreSQL telemetry sink, schema and retention

ddiff --git a/CHANGELOG.md b/CHANGELOG.md
iindex 684da5c..3b55b47 100644
---- a/CHANGELOG.md
++++ b/CHANGELOG.md
@@@ -3,6 +3,21 @@
  All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
  release may change an API and says so here.
  
++## 0.45.0 (2026-10-13)
++
++The telemetry sink every service stores into (plan T1, step 10). Step 5 left `createTelemetry({ sink })` and
++`createHttpTelemetry({ sink })` without a sink for any service but Network; `openvibe-sdk/telemetry` now ships one,
++a per-service PostgreSQL table (the decision and its reason are in `docs/service.md`, "Where the samples go"):
++
++- `telemetrySchema(table = 'telemetry_samples')`: the DDL for a service migration, `platform.telemetry-sample@1`'s
++  21 columns in schema order (no `instance`), `extra` jsonb, indexes on `(at)` and `(service, operation, at)`.
++- `createPgTelemetrySink({ db, table?, log? })`: the sink. One multi-row parameterized INSERT per flush; a malformed
++  sample (no service, operation or parseable `at`, or a field of the wrong type) is dropped with one warning and the
++  rest stored; a rejected `db.query` propagates, so the collector keeps the batch for its next flush.
++- `pruneTelemetry({ db, table?, olderThanDays = 30 })`: the retention delete.
++- A table name that does not match `/^[a-z_][a-z0-9_]{0,62}$/` throws, so no SQL goes in through it.
++- **Tests:** `test/telemetry-sink.test.js` on PGlite.
++
  ## 0.44.0 (2026-10-11)
  
  The SQLite kits are gone (plan T1, the deletion ledger): every OpenVibe service has run on PostgreSQL since
ddiff --git a/README.md b/README.md
iindex d12f2e4..8327f90 100644
---- a/README.md
++++ b/README.md
@@@ -187,7 +187,7 @@ Until OpenVibe.Tools serves the gateway facade (Tools S6), point the jobs client
  | `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). `createUsageReporter({ db, service, source, table, billingUrl, tokenClient, audience, requireProject, ... })` owns one service's readings: `key(...parts)`/`sample(fields)`, and `record(t, reading)` queues the reading idempotency-keyed in the outbox inside the caller's transaction, relayed to Billing's `billing.usage.record` with backoff (a retry is never dropped; only Billing refusing the reading itself — any 4xx but 401/403/404/408/425/429, which stay pending — marks a row rejected; a 401 drops the cached token). The reporter bills whatever reading it is given: never record first-party or sandbox traffic (pass `requireProject: true` to refuse a reading without a `prj_…` project). `openvibe-sdk/govern` builds its `onUsage` record with it |
  | `openvibe-sdk/commerce` | server | `createCommerceClient(config, {caps, fetchImpl?, tokenClients?, timeoutMs?})`: the Billing client Tips and VIP share. `config` = `{billing: {url, audience, timeoutMs = 10000}, network: {internalUrl}, oauth: {clientId, clientSecret}}`; `caps` maps `intent`, `transfer`, `subscription`, `entitlement` (and optional `rates`, default `entitlement`) to capability strings, one lazily created `serviceAuth` token client per distinct string (`scope` = the string) unless `tokenClients` supplies one, keyed by logical key or capability. Methods: `createIntent` (`kind` purchase with `bits`, or subscription with `creator`/`autoRenew`), `createTransfer`, `refundTransfer`, `subscribeWithCredit`, `cancelSubscription`, `getSubscription`, `listSubscriptions`, `entitlement`, `rates`, `baseUrl`. `Idempotency-Key` is sent verbatim; a 401 invalidates the token and retries once with the same key; problem+json becomes `CommerceError` (`status`, `code`, `body`, `retryable`). `intentKey(prefix, parts)` joins the parts that are set; `receipt(kind, fields)` → `{kind, at, ...fields}` (null stripped, never posted). `openvibe-contracts` is required lazily |
--| `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/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; the per-service PostgreSQL sink: `telemetrySchema(table = 'telemetry_samples')` (migration DDL), `createPgTelemetrySink({db, table?, log?})` (one multi-row INSERT per flush, malformed samples dropped with a warning), `pruneTelemetry({db, table?, olderThanDays = 30})` |
  | `openvibe-sdk/cost` | server | Build and validate `platform.rate-card@1` and `platform.cost-snapshot@1` records with `rateCard()`, `costSnapshot()`, `validateRateCard()`, and `validateCostSnapshot()`; `estimate()` calculates a charge from a rate card |
  | `openvibe-sdk/runtime` | server | Build and validate `platform.runtime-offer@1` records with `runtimeOffer()` and `validateRuntimeOffer()`; `runtimeClass()`, `RUNTIME_CLASSES`, and `validateRuntimeClass()` handle known runtime classes |
  | `openvibe-sdk/storage` | server | `storageOffer()` and `validateStorageOffer()` for `platform.storage-offer@1` |
ddiff --git a/STATUS.json b/STATUS.json
iindex 7465939..3988d77 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -3,7 +3,7 @@
    "stage": "alpha",
    "code": true,
    "wave": "W2",
--  "version": "0.44.0",
++  "version": "0.45.0",
    "plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
    "deployed": false,
    "deployment": "none: a package released as git tags; services and apps pin a tag",
@@@ -28,6 +28,7 @@
      "events: the PostgreSQL outbox (lease-claimed, several relays per table) and inbox (handler in its transaction)",
      "usage: createUsageReporter (openvibe-sdk/usage), the shared platform.usage-sample@1 reporter — validated readings queued idempotency-keyed in an outbox and relayed to Billing with retry, never dropped, refused readings kept",
      "the HTTP telemetry kit (openvibe-sdk/service): the platform.telemetry-sample@1 middleware aggregated per route|method|status_class per flush, with active-requests, p95 and event-loop-lag gauges, lifted from Network's server/telemetry.js and server/observability.js",
++    "the telemetry sink (openvibe-sdk/telemetry): telemetrySchema() DDL, createPgTelemetrySink() and pruneTelemetry(), a per-service PostgreSQL table every service stores platform.telemetry-sample@1 into",
      "resources (openvibe-sdk/resources): the ADR-048 resource index fan-out and the idempotent control client"
    ],
    "updated": "2026-10-11"
ddiff --git a/browser/openvibe-sdk.mjs b/browser/openvibe-sdk.mjs
iindex 0fb6185..1e7ee36 100644
---- a/browser/openvibe-sdk.mjs
++++ b/browser/openvibe-sdk.mjs
@@@ -1,4 +1,4 @@
--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.44.0, MIT). Do not edit.
++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.45.0, MIT). Do not edit.
  // One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
  // (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
  /* eslint-disable */
@@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);
  const semver = __load(6);
  
  /** The package version (test/version.test.js keeps it equal to package.json). */
--const SDK_VERSION = '0.44.0';
++const SDK_VERSION = '0.45.0';
  
  module.exports = {
      createClient,
ddiff --git a/docs/service.md b/docs/service.md
iindex 003011f..80922e2 100644
---- a/docs/service.md
++++ b/docs/service.md
@@@ -174,7 +174,7 @@ that matched no route is `unmatched`, and the per-flush key map and latency samp
  const svc = require('openvibe-sdk/service');
  const collector = svc.createHttpTelemetry({
      service: 'blog',                                     // the schema's service; there is no `instance` field
--    sink: (samples) => post('/internal/telemetry', samples),   // one batch per flush
++    sink: createPgTelemetrySink({ db, log }),             // one batch per flush (see below)
      intervalMs: 15000,                                   // also the p95 window
      skipPrefixes: ['/shared'],                           // default ['/shared']; Network passes '/shared', '/api/chrome'
      routeLabel: (req) => metrics.routeLabel(req),        // optional; default reads req.route.path under req.baseUrl
@@@ -198,6 +198,36 @@ and the counts in `extra`; the p95 gauge carries its value in `latency_ms` (it i
  and event-loop-lag gauges carry theirs in `extra`, never in `latency_ms` (which a consumer reads as a response
  time). `registerSignals({ start, stop, lag })` injects a service's own monitor in place of the default.
  
++### Where the samples go
++
++**Decision (T1 step 10): each service stores its samples in its own PostgreSQL table**, not in a central ingest
++(a Network `/internal/telemetry` route behind service-token auth). Why:
++
++- Every service already has an `openvibe-sdk/db` handle and a migration directory; a sink that is an INSERT adds no
++  network hop, no token, no new route and no new failure mode. A central route would make Network a dependency of
++  every service's telemetry, and an outage there would turn into buffered (then dropped) samples everywhere.
++- It is the pattern the SDK already ships for the outbox and the usage relay (`outboxSchema()` in a migration, a
++  helper on the handle), so a service adds one migration file and one argument.
++- A row costs the service's own database, which the placement and cost work (T5/T13) can already see per service;
++  a consumer that needs the fleet view reads the tables (or a later relay copies them) without changing any emitter.
++
++```js
++const { telemetrySchema, createPgTelemetrySink, pruneTelemetry } = require('openvibe-sdk/telemetry');
++// migrations/NNN_telemetry.sql: the text of telemetrySchema() (21 columns in platform.telemetry-sample@1 order,
++// no `instance`, `extra` jsonb, indexes on (at) and (service, operation, at)); the default table is telemetry_samples.
++const collector = svc.createHttpTelemetry({ service: 'blog', sink: createPgTelemetrySink({ db, log }) });
++svc.gracefulStop({ name: 'Blog', server, stop: [() => collector.stop()] });   // the last flush is one INSERT
++await pruneTelemetry({ db });   // a daily job: DELETE … WHERE at < now() - 30 days (olderThanDays to change it)
++```
++
++The sink does one multi-row parameterized INSERT per flush (a second only past 3120 rows, PostgreSQL's 65535
++parameter limit). A malformed sample (no `service`, `operation` or parseable `at`, a text field that is not a string,
++a number that is not finite, a non-integer `bytes`/`route_epoch`, an `extra` that is not an object) is dropped with
++one warning per flush and the rest stored, so one bad sample never blocks a batch the collector would otherwise
++retry forever. A missing field is NULL and a missing `extra` is `{}`. A rejected `db.query` propagates: the collector
++logs it and keeps the batch for the next flush. The table name must match `/^[a-z_][a-z0-9_]{0,62}$/` or the helpers
++throw, so no SQL goes in through it.
++
  **Network and Community** delete `server/telemetry.js` and the telemetry block of `server/observability.js`
  (`telemetrySkipped`/`telemetryMiddleware`/the loop monitor) and register the kit's middleware with
  `skipPrefixes: ['/shared', '/api/chrome']`; their `analyticsSink` becomes the `sink` passed to `init`.
ddiff --git a/esm/telemetry.mjs b/esm/telemetry.mjs
iindex 2954be9..03d764c 100644
---- a/esm/telemetry.mjs
++++ b/esm/telemetry.mjs
@@@ -2,6 +2,6 @@
  import cjs from '../src/telemetry.js';
  
  export const {
--    telemetrySample, validateTelemetrySample, createTelemetry,
++    telemetrySample, validateTelemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry,
  } = cjs;
  export default cjs;
ddiff --git a/package-lock.json b/package-lock.json
iindex 2dae190..d1eb624 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -1,12 +1,12 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.44.0",
++  "version": "0.45.0",
    "lockfileVersion": 3,
    "requires": true,
    "packages": {
      "": {
        "name": "openvibe-sdk",
--      "version": "0.44.0",
++      "version": "0.45.0",
        "license": "MIT",
        "devDependencies": {
          "@electric-sql/pglite": "^0.5.8",
ddiff --git a/package.json b/package.json
iindex 32399f1..8e0046e 100644
---- a/package.json
++++ b/package.json
@@@ -1,6 +1,6 @@
  {
    "name": "openvibe-sdk",
--  "version": "0.44.0",
++  "version": "0.45.0",
    "description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
    "main": "index.js",
    "types": "types/index.d.ts",
ddiff --git a/src/core/index.js b/src/core/index.js
iindex ad018c9..1f8fdff 100644
---- a/src/core/index.js
++++ b/src/core/index.js
@@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
  const semver = require('./semver');
  
  /** The package version (test/version.test.js keeps it equal to package.json). */
--const SDK_VERSION = '0.44.0';
++const SDK_VERSION = '0.45.0';
  
  module.exports = {
      createClient,
ddiff --git a/src/telemetry.js b/src/telemetry.js
iindex 7dce6e4..112b123 100644
---- a/src/telemetry.js
++++ b/src/telemetry.js
@@@ -10,6 +10,9 @@
   *   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
++ *
++ * Where samples land: each service's own PostgreSQL table (telemetrySchema() in a migration, createPgTelemetrySink()
++ * as the sink, pruneTelemetry() for retention), not a central ingest; docs/service.md says why.
   */
  
  /** The fields platform.telemetry-sample@1 requires or allows, in the schema's order; null/undefined are stripped. */
@@@ -137,4 +140,100 @@ function createTelemetry({ service = null, instance = null, sink = null, interva
      return { record, gauge, count, flush, stop };
  }
  
--module.exports = { telemetrySample, validateTelemetrySample, createTelemetry };
++const TABLE_RE = /^[a-z_][a-z0-9_]{0,62}$/;
++
++function checkTable(table, who) {
++    if (typeof table !== 'string' || !TABLE_RE.test(table)) throw new TypeError(`${who}: bad telemetry table name (lowercase letters, digits and _, at most 63)`);
++    return table;
++}
++
++/** The table's columns in platform.telemetry-sample@1 order (no `instance`: the schema has none), then `extra`. */
++const COLUMNS = ['at', 'service', 'project', 'subject', 'resource', 'provider', 'node', 'cell', 'region', 'operation',
++    'latency_ms', 'queue_delay_ms', 'ttfb_ms', 'throughput_per_second', 'bytes', 'status', 'cache_status',
++    'cost_estimate', 'route_epoch', 'trace_id', 'extra'];
++const TEXT = new Set(['project', 'subject', 'resource', 'provider', 'node', 'cell', 'region', 'status', 'cache_status', 'trace_id']);
++const NUMBER = new Set(['latency_ms', 'queue_delay_ms', 'ttfb_ms', 'throughput_per_second', 'cost_estimate']);
++const INTEGER = new Set(['bytes', 'route_epoch']);
++// PostgreSQL binds at most 65535 parameters per statement: past this many rows a flush takes a second INSERT.
++const MAX_ROWS = Math.floor(65535 / COLUMNS.length);
++
++/** The telemetry table's DDL, for a service migration (and tests): 21 columns, `extra` jsonb, and two indexes. */
++function telemetrySchema(table = 'telemetry_samples') {
++    checkTable(table, 'telemetrySchema');
++    return `CREATE TABLE IF NOT EXISTS ${table} (
++    id                    bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++    at                    timestamptz NOT NULL,
++    service               text NOT NULL,
++    project               text,
++    subject               text,
++    resource              text,
++    provider              text,
++    node                  text,
++    cell                  text,
++    region                text,
++    operation             text NOT NULL,
++    latency_ms            double precision,
++    queue_delay_ms        double precision,
++    ttfb_ms               double precision,
++    throughput_per_second double precision,
++    bytes                 bigint,
++    status                text,
++    cache_status          text,
++    cost_estimate         double precision,
++    route_epoch           bigint,
++    trace_id              text,
++    extra                 jsonb NOT NULL DEFAULT '{}'::jsonb
++);
++CREATE INDEX IF NOT EXISTS ${table}_at_idx ON ${table} (at);
++CREATE INDEX IF NOT EXISTS ${table}_service_idx ON ${table} (service, operation, at);`;
++}
++
++/** Whether a row would insert: one bad field must not fail (and, kept by the collector, block) the whole batch. */
++function storable(s) {
++    if (!s || typeof s !== 'object' || Array.isArray(s)) return false;
++    if (typeof s.service !== 'string' || !s.service || typeof s.operation !== 'string' || !s.operation) return false;
++    if (typeof s.at !== 'string' || Number.isNaN(Date.parse(s.at))) return false;
++    for (const c of TEXT) if (s[c] != null && typeof s[c] !== 'string') return false;
++    for (const c of NUMBER) if (s[c] != null && !Number.isFinite(s[c])) return false;
++    for (const c of INTEGER) if (s[c] != null && !Number.isSafeInteger(s[c])) return false;
++    if (s.extra != null && (typeof s.extra !== 'object' || Array.isArray(s.extra))) return false;
++    return true;
++}
++
++/**
++ * A createTelemetry() / createHttpTelemetry() sink that stores each flush in `table` with one multi-row
++ * parameterized INSERT (a second one only past MAX_ROWS rows). A malformed sample (no service, operation or
++ * parseable `at`, or a field of the wrong type) is dropped with one warning per flush and the rest are stored;
++ * a missing field is NULL, `extra` is JSON ('{}' when absent). A rejected db.query propagates, so the collector
++ * logs it and keeps the batch for its next flush.
++ */
++function createPgTelemetrySink({ db, table = 'telemetry_samples', log = console } = {}) {
++    if (!db || typeof db.query !== 'function') throw new TypeError('createPgTelemetrySink: an openvibe-sdk/db handle is required');
++    checkTable(table, 'createPgTelemetrySink');
++    return async function sink(samples) {
++        const all = Array.isArray(samples) ? samples : [];
++        const rows = all.filter(storable);
++        if (rows.length < all.length) log.warn(`[telemetry] ${table}: dropped ${all.length - rows.length} malformed sample(s), stored ${rows.length}`);
++        for (let start = 0; start < rows.length; start += MAX_ROWS) {
++            const chunk = rows.slice(start, start + MAX_ROWS);
++            const values = [];
++            const tuples = chunk.map((s) => `(${COLUMNS.map((c) => {
++                values.push(c === 'extra' ? JSON.stringify(s.extra || {}) : (s[c] == null ? null : s[c]));
++                return `$${values.length}`;
++            }).join(', ')})`);
++            await db.query(`INSERT INTO ${table} (${COLUMNS.join(', ')}) VALUES ${tuples.join(', ')}`, values);
++        }
++    };
++}
++
++/** Retention: deletes the rows older than `olderThanDays` (default 30) and resolves with how many went. A
++ *  service runs it from a daily job or timer; the `at` index keeps it cheap. */
++async function pruneTelemetry({ db, table = 'telemetry_samples', olderThanDays = 30 } = {}) {
++    if (!db || typeof db.query !== 'function') throw new TypeError('pruneTelemetry: an openvibe-sdk/db handle is required');
++    checkTable(table, 'pruneTelemetry');
++    if (!(Number.isFinite(olderThanDays) && olderThanDays > 0)) throw new TypeError('pruneTelemetry: olderThanDays must be a positive number');
++    const r = await db.query(`DELETE FROM ${table} WHERE at < now() - make_interval(secs => $1)`, [olderThanDays * 86400]);
++    return r.rowCount;
++}
++
++module.exports = { telemetrySample, validateTelemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry };
ddiff --git a/test/telemetry-sink.test.js b/test/telemetry-sink.test.js
nnew file mode 100644
iindex 0000000..255ba6b
---- /dev/null
++++ b/test/telemetry-sink.test.js
@@@ -0,0 +1,141 @@
++'use strict';
++/**
++ * openvibe-sdk/telemetry's PostgreSQL sink: telemetrySchema() DDL, createPgTelemetrySink() (one multi-row
++ * INSERT per flush, malformed samples dropped with a warning, a db rejection propagated) and pruneTelemetry().
++ * PGlite is the database.
++ */
++const assert = require('node:assert/strict');
++const { run } = require('./helpers');
++const { createDb } = require('../src/db');
++const { telemetrySample, createTelemetry, telemetrySchema, createPgTelemetrySink, pruneTelemetry } = require('../src/telemetry');
++
++const COLUMNS = ['id', 'at', 'service', 'project', 'subject', 'resource', 'provider', 'node', 'cell', 'region', 'operation',
++    'latency_ms', 'queue_delay_ms', 'ttfb_ms', 'throughput_per_second', 'bytes', 'status', 'cache_status',
++    'cost_estimate', 'route_epoch', 'trace_id', 'extra'];
++
++function logger() {
++    const warns = [];
++    return { warns, warn: (m) => warns.push(String(m)), error() {}, log() {} };
++}
++
++async function open(table) {
++    const db = createDb({ pglite: true });
++    await db.query(telemetrySchema(table));
++    return db;
++}
++
++const full = (i) => telemetrySample({
++    service: 'blog', project: 'prj_1', subject: 'usr_1', resource: '/api/posts/:id', provider: 'ovh', node: 'n1',
++    cell: 'c1', region: 'eu-west', operation: 'http.request', at: `2026-10-13T10:00:0${i}.000Z`, latency_ms: 12.5 + i,
++    queue_delay_ms: 1, ttfb_ms: 3, throughput_per_second: 40, bytes: 1024 + i, status: '2xx', cache_status: 'miss',
++    cost_estimate: 0.001, route_epoch: 7, trace_id: `trace-${i}`, extra: { count: 5 + i, method: 'GET' },
++});
++
++run([
++    ['telemetrySchema() runs on PGlite (twice: idempotent) and has the columns in schema order, plus two indexes', async () => {
++        const db = await open();
++        try {
++            await db.query(telemetrySchema());
++            const cols = await db.many(`SELECT column_name FROM information_schema.columns WHERE table_name = 'telemetry_samples' ORDER BY ordinal_position`);
++            assert.deepEqual(cols.map((c) => c.column_name), COLUMNS);
++            const idx = await db.many(`SELECT indexname FROM pg_indexes WHERE tablename = 'telemetry_samples' ORDER BY indexname`);
++            assert.deepEqual(idx.map((r) => r.indexname), ['telemetry_samples_at_idx', 'telemetry_samples_pkey', 'telemetry_samples_service_idx']);
++        } finally { await db.close(); }
++    }],
++
++    ['the sink inserts N rows with one INSERT, every field in its column and extra JSON-encoded', async () => {
++        const db = await open();
++        const queries = [];
++        const spy = { query: (text, values) => { queries.push({ text, values }); return db.query(text, values); } };
++        try {
++            const log = logger();
++            await createPgTelemetrySink({ db: spy, log })([full(0), full(1), full(2)]);
++            assert.equal(queries.length, 1);
++            assert.match(queries[0].text, /^INSERT INTO telemetry_samples \(at, service, project, .*, trace_id, extra\) VALUES /);
++            assert.equal(queries[0].values.length, 3 * 21);
++            assert.equal(queries[0].values[20], JSON.stringify({ count: 5, method: 'GET' }));
++            assert.deepEqual(log.warns, []);
++            const rows = await db.many('SELECT * FROM telemetry_samples ORDER BY id');
++            assert.equal(rows.length, 3);
++            rows.forEach((r, i) => {
++                const { id, ...stored } = r;
++                assert.ok(id);
++                assert.deepEqual(stored, { ...full(i) });
++            });
++        } finally { await db.close(); }
++    }],
++
++    ['missing fields are NULL and a missing extra is {}', async () => {
++        const db = await open();
++        try {
++            await createPgTelemetrySink({ db, log: logger() })([{ service: 'blog', operation: 'cache.miss', at: '2026-10-13T10:00:00.000Z' }]);
++            const r = await db.one('SELECT * FROM telemetry_samples');
++            for (const c of COLUMNS) {
++                if (['id', 'at', 'service', 'operation', 'extra'].includes(c)) continue;
++                assert.equal(r[c], null, c);
++            }
++            assert.deepEqual(r.extra, {});
++        } finally { await db.close(); }
++    }],
++
++    ['a malformed sample is dropped with one warning and the rest of the batch is stored', async () => {
++        const db = await open();
++        try {
++            const log = logger();
++            const bad = [null, { operation: 'x', at: '2026-10-13T10:00:00Z' }, { ...full(1), at: 'yesterday' },
++                { ...full(2), latency_ms: '12' }, { ...full(3), bytes: 1.5 }, { ...full(4), extra: [1] }, { ...full(5), status: 200 }];
++            await createPgTelemetrySink({ db, log })([full(0), ...bad, full(6)]);
++            assert.equal(log.warns.length, 1);
++            assert.match(log.warns[0], /dropped 7 malformed sample\(s\), stored 2/);
++            const rows = await db.many('SELECT trace_id FROM telemetry_samples ORDER BY id');
++            assert.deepEqual(rows.map((r) => r.trace_id), ['trace-0', 'trace-6']);
++            // A batch with nothing storable inserts nothing (and does not query).
++            await createPgTelemetrySink({ db: { query: () => { throw new Error('queried'); } }, log })([null]);
++            await createPgTelemetrySink({ db: { query: () => { throw new Error('queried'); } }, log })([]);
++        } finally { await db.close(); }
++    }],
++
++    ['a bad table name throws, so no SQL goes in through it', async () => {
++        const db = { query: async () => ({ rows: [], rowCount: 0 }) };
++        for (const t of ['bad-name', 'x; DROP TABLE users', 'Upper', '1abc', '', 'a'.repeat(64), 'public.t', null]) {
++            assert.throws(() => telemetrySchema(t), /bad telemetry table name/, String(t));
++            assert.throws(() => createPgTelemetrySink({ db, table: t }), /bad telemetry table name/, String(t));
++            await assert.rejects(pruneTelemetry({ db, table: t }), /bad telemetry table name/, String(t));
++        }
++        assert.match(telemetrySchema('blog_telemetry'), /CREATE TABLE IF NOT EXISTS blog_telemetry \(/);
++        assert.throws(() => createPgTelemetrySink({}), /db handle/);
++    }],
++
++    ['a rejected db.query propagates, and createTelemetry keeps the batch for its next flush', async () => {
++        const failing = { query: async () => { throw new Error('connection refused'); } };
++        await assert.rejects(createPgTelemetrySink({ db: failing, log: logger() })([full(0)]), /connection refused/);
++
++        const db = await open('svc_telemetry');
++        try {
++            let down = true;
++            const flaky = { query: (text, values) => (down ? Promise.reject(new Error('down')) : db.query(text, values)) };
++            const log = logger();
++            const t = createTelemetry({ service: 'blog', sink: createPgTelemetrySink({ db: flaky, table: 'svc_telemetry', log }), log, intervalMs: 3600e3 });
++            t.record('http.request', 10, { status: '2xx', extra: { count: 1 } });
++            await t.flush();
++            assert.equal(await db.value('SELECT count(*)::int FROM svc_telemetry'), 0);
++            down = false;
++            await t.stop();
++            const r = await db.one('SELECT service, operation, latency_ms, status, extra FROM svc_telemetry');
++            assert.deepEqual(r, { service: 'blog', operation: 'http.request', latency_ms: 10, status: '2xx', extra: { count: 1 } });
++        } finally { await db.close(); }
++    }],
++
++    ['pruneTelemetry deletes rows older than 30 days by default and keeps the rest', async () => {
++        const db = await open();
++        try {
++            const day = 86400e3;
++            const at = (daysAgo) => new Date(Date.now() - daysAgo * day).toISOString();
++            await createPgTelemetrySink({ db, log: logger() })([40, 31, 29, 1].map((d) => ({ service: 'blog', operation: `d${d}`, at: at(d) })));
++            assert.equal(await pruneTelemetry({ db }), 2);
++            assert.deepEqual((await db.many('SELECT operation FROM telemetry_samples ORDER BY id')).map((r) => r.operation), ['d29', 'd1']);
++            assert.equal(await pruneTelemetry({ db, olderThanDays: 7 }), 1);
++            await assert.rejects(pruneTelemetry({ db, olderThanDays: 0 }), /olderThanDays/);
++        } finally { await db.close(); }
++    }],
++]);
ddiff --git a/types/telemetry.d.ts b/types/telemetry.d.ts
iindex 0bfe439..b7d3545 100644
---- a/types/telemetry.d.ts
++++ b/types/telemetry.d.ts
@@@ -28,3 +28,15 @@ export interface TelemetryCollector {
   *  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;
++/** Anything with openvibe-sdk/db's query(text, values) (a handle or a transaction's). */
++export type TelemetryDb = { query(text: string, values?: unknown[]): Promise<{ rows: unknown[]; rowCount: number }> };
++/** The telemetry table's DDL for a service migration: platform.telemetry-sample@1's 21 columns in order (no
++ *  `instance`), `extra` jsonb, indexes on (at) and (service, operation, at). The table name must match
++ *  /^[a-z_][a-z0-9_]{0,62}$/ or it throws. Default `telemetry_samples`. */
++export declare function telemetrySchema(table?: string): string;
++/** A sink for createTelemetry() / createHttpTelemetry(): one multi-row parameterized INSERT per flush. A malformed
++ *  sample is dropped with one warning and the rest stored; missing fields are NULL, `extra` is JSON; a rejected
++ *  db.query propagates (the collector keeps the batch). */
++export declare function createPgTelemetrySink(opts: { db: TelemetryDb; table?: string; log?: { warn(...a: unknown[]): void } }): (samples: TelemetrySample[]) => Promise<void>;
++/** Retention: deletes rows whose `at` is older than olderThanDays (default 30); resolves with the count. */
++export declare function pruneTelemetry(opts: { db: TelemetryDb; table?: string; olderThanDays?: number }): Promise<number>;