Add the nats-js-v1 JetStream STREAM carrier over the PostgreSQL record

ddiff --git a/.env.example b/.env.example
iindex 6c7c0dc..e052b00 100644
---- a/.env.example
++++ b/.env.example
@@@ -30,11 +30,15 @@ DATABASE_URL=
  DATABASE_DIRECT_URL=
  VALKEY_URL=
  VALKEY_PREFIX=
--# Optional NATS Core (nats-v1: cross-process SSE fan-out, the TOPIC carrier). nats://[user:pass@]host:4222 on loopback or
--# the private network; unset, nothing connects and pg-v1/valkey-v1 carry TOPIC. EVENTS_CARRIERS_DISABLED=nats-v1 rolls back.
++# Optional NATS (deploy/nats/nats-server.conf): nats-v1, cross-process SSE fan-out (the TOPIC carrier), and nats-js-v1,
++# JetStream (the STREAM carrier, needs nats-server -js). nats://[user:pass@]host:4222 on loopback or the private network;
++# unset, nothing connects and pg-v1/valkey-v1 carry TOPIC, pg-v1 STREAM. EVENTS_CARRIERS_DISABLED=nats-v1 or nats-js-v1
++# rolls one back; NATS_JETSTREAM=off leaves nats-js-v1 out.
  #NATS_URL=
  #NATS_SUBJECT_PREFIX=ov.events.
--#EVENTS_CARRIERS=pg-v1,valkey-v1,nats-v1
++#NATS_JETSTREAM=
++#NATS_STREAM=OV_EVENTS_STREAM
++#EVENTS_CARRIERS=pg-v1,valkey-v1,nats-v1,nats-js-v1
  #EVENTS_CARRIERS_DISABLED=
  # Events (and their deliveries) older than this are pruned; resuming from before it yields a gap.
  EVENTS_RETENTION_DAYS=30
ddiff --git a/README.md b/README.md
iindex e17f404..2a36dfd 100644
---- a/README.md
++++ b/README.md
@@@ -118,11 +118,12 @@ Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, the
  
  How an event type is carried is its delivery policy (`events.delivery-policy@1`, [ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 1). The carrier class — TOPIC (fan-out, no ack), QUEUE (competing consumers, ack) or STREAM (retained, ordered per key) — is a derived, internal label, never in the envelope. PostgreSQL stays the record: every event and every delivery is committed before any carrier sees it, and the worker's lease is the only thing that makes a send happen. A publish commits, then the chosen carrier is signalled; a lost or duplicated signal changes latency, never the outcome.
  
--Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1,nats-v1`):
++Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1,nats-v1,nats-js-v1`):
  
  - `pg-v1` — the poll. Always present, every class; the worker's 500 ms tick and its in-process `kick()` carry delivery. It is the floor every delivery falls back to.
  - `valkey-v1` — push, only when `VALKEY_URL` is set. **QUEUE:** after the commit, each new delivery is XADDed to a Valkey stream (`openvibe-sdk/queue`, consumer group `workers`); each Events process claims just that one row (`store.claimDelivery`, the same key-head and inflight-cap rules as the poll) and hands it to the worker's normal send — the item is acked either way, so a lost item only falls to the poll. **TOPIC:** the stored envelopes are PUBLISHed on a Valkey channel tagged with this process's id, and every process re-emits envelopes from other processes to its own SSE clients (deduped by event id), so an event published on one process reaches a browser connected to another. Health is an EWMA plus a breaker; while a breaker is open the planner excludes the carrier with a reason and `pg-v1` delivers.
--- `nats-v1` — optional NATS Core, TOPIC only, only when `NATS_URL` is set (`nats://[user:pass@]host:4222`, loopback or the private network; no TLS; subjects under `NATS_SUBJECT_PREFIX`, default `ov.events.`). After the commit the event ids (never the envelopes) are published on `<prefix>fabric.topic`; every process loads those from other processes from PostgreSQL and re-emits the committed rows to its own SSE clients (an id that is not committed is ignored; deduped by event id; visibility applied by the receiving gateway). Core NATS keeps nothing: a lost message reaches a remote browser on its next resume (`Last-Event-ID` replays from PostgreSQL). While it is not connected, the broker refuses its publish or subscribe (a permissions violation; the handshake probes the subject, a refused connection is re-opened every 30 s) or its breaker is open, the planner excludes it with that reason and `valkey-v1` or `pg-v1` carries TOPIC; `EVENTS_CARRIERS_DISABLED=nats-v1` stops it connecting at all. During a rolling deploy, old processes drop `{ids}` messages and new ones drop old `{rows}` messages, so cross-process TOPIC fan-out falls back to resume-from-PostgreSQL until every process runs the new version. JetStream (STREAM) is not built yet: STREAM is carried by `pg-v1`. Tests: `eval "$(scripts/test-nats.sh up)"` sets `OV_TEST_NATS_URL` for `test/fabric-nats.test.js`.
++- `nats-v1` — optional NATS Core, TOPIC only, only when `NATS_URL` is set (`nats://[user:pass@]host:4222`, loopback or the private network; no TLS; subjects under `NATS_SUBJECT_PREFIX`, default `ov.events.`). After the commit the event ids (never the envelopes) are published on `<prefix>fabric.topic`; every process loads those from other processes from PostgreSQL and re-emits the committed rows to its own SSE clients (an id that is not committed is ignored; deduped by event id; visibility applied by the receiving gateway). Core NATS keeps nothing: a lost message reaches a remote browser on its next resume (`Last-Event-ID` replays from PostgreSQL). While it is not connected, the broker refuses its publish or subscribe (a permissions violation; the handshake probes the subject, a refused connection is re-opened every 30 s) or its breaker is open, the planner excludes it with that reason and `valkey-v1` or `pg-v1` carries TOPIC; `EVENTS_CARRIERS_DISABLED=nats-v1` stops it connecting at all. During a rolling deploy, old processes drop `{ids}` messages and new ones drop old `{rows}` messages, so cross-process TOPIC fan-out falls back to resume-from-PostgreSQL until every process runs the new version. Tests: `eval "$(scripts/test-nats.sh up)"` (a broker with JetStream) sets `OV_TEST_NATS_URL` for `test/fabric-nats.test.js` and `test/fabric-jetstream.test.js`.
++- `nats-js-v1` — optional JetStream, STREAM only, when `NATS_URL` is set and `NATS_JETSTREAM` is not `off` (nats-server 2.10+ with `-js`; `deploy/nats/nats-server.conf`). Spoken as JSON over the same dependency-free client (`server/fabric/nats-core.js`, request/reply), no client library. After the commit each new STREAM delivery's `{ event_id, subscription_id }` (never content) is published, with the stream's ack awaited, on `<prefix>fabric.stream.<key>` of the stream `NATS_STREAM` (default `OV_EVENTS_STREAM`, file storage, kept 24 h; `<key>` is a subject-safe hash of the ordering key). Every process running the delivery worker has its own durable push consumer (new messages only, explicit ack, one unacked message at a time) and, in stream order, claims exactly that row (`store.claimDelivery`: the key-head lease, `max_inflight` and due rules of the poll), sends it through the worker, then acks — whatever the claim answered. A message whose claim yields (another process or the poll has it, not its key's head yet, done, or never committed) leaves the row to the poll; a process that dies before its ack gets the message again after `ack_wait` and the claim, finding it delivered, sends nothing. Order per key is the store's, never the broker's. While it is not connected, the broker refuses it, the stream cannot be ensured (re-checked after every reconnect) or its breaker is open, the planner excludes it with that reason and `pg-v1` carries STREAM. The planner asks `events:durable` + `events:ordered` for STREAM, so only `nats-js-v1` and `pg-v1` can carry it; `rc-nats-js-v1` is cheaper than `rc-pg-v1`, so JetStream is preferred while healthy. Rollback: `EVENTS_CARRIERS_DISABLED=nats-js-v1` (or `NATS_JETSTREAM=off`) — it never connects and new STREAM deliveries go to `pg-v1`; rows already placed on it are delivered by the poll either way.
  
  The planner (`server/fabric/planner.js`) calls `openvibe-sdk/placement` `plan()` per (carrier class, ordering key) with one `resource-offer@1` per adapter and the rate cards in `server/fabric/rate-cards.json`, caching decisions with hysteresis and recording a change in `delivery_placements`. A key's backlog stays on one carrier until it drains (a placement is pinned per (subscription, key), not per time window). An adapter named in `EVENTS_CARRIERS_DISABLED` is ineligible at once with reason `disabled by configuration` — the ADR's rollback; an unknown id in `EVENTS_CARRIERS` refuses to start.
  
ddiff --git a/STATUS.json b/STATUS.json
iindex 1b6352f..30e516f 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -31,7 +31,7 @@
      "hourly project usage rollups (events.usage.recorded)",
      "redaction tombstones",
      "per-actor limits (openvibe-sdk/limits; shared on Valkey)",
--    "carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out) and an optional nats-v1 NATS Core TOPIC carrier (NATS_URL), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
++    "carrier fabric (ADR-042): the pg-v1 poll beside a valkey-v1 push carrier for QUEUE (streams) and TOPIC (SSE fan-out) an optional nats-v1 NATS Core TOPIC carrier (NATS_URL) and an optional nats-js-v1 JetStream STREAM carrier (NATS_URL, NATS_JETSTREAM; deploy/nats), with a placement planner, per-key pinning, delivery_placements history and a GET /api/v1/placement explain API; EVENTS_CARRIERS[_DISABLED] switches it",
      "at_most_once delivery semantics (a failed attempt is final)"
    ]
  }
ddiff --git a/deploy/nats/nats-server.conf b/deploy/nats/nats-server.conf
nnew file mode 100644
iindex 0000000..0c96db5
---- /dev/null
++++ b/deploy/nats/nats-server.conf
@@@ -0,0 +1,40 @@
++# The cell's NATS for OpenVibe.Events (ADR-042 decisions 5-6): the nats-v1 TOPIC carrier (NATS Core) and the
++# nats-js-v1 STREAM carrier (JetStream). Run as `nats-server -js -c /etc/nats/nats-server.conf` (the unit is
++# OpenVibe.Host's openvibe-nats.service). Never the record: PostgreSQL is; losing this store loses latency, not events.
++#
++# Loopback only (put the private-network address here instead for a multi-host cell, and firewall 4222 to the cell's
++# hosts). No monitoring port, no cluster, no leafnodes.
++host: 127.0.0.1
++port: 4222
++max_payload: 1MB
++
++jetstream {
++  store_dir: "/var/lib/nats"
++  max_memory_store: 256MB
++  max_file_store: 4GB
++}
++
++# One user for Events. The password comes from the environment of the unit (NATS_EVENTS_PASSWORD, e.g. from an
++# EnvironmentFile readable only by the nats user), never from this file; Events gets it in NATS_URL
++# (nats://events:<password>@127.0.0.1:4222). The user may use its own subjects (NATS_SUBJECT_PREFIX, default
++# ov.events.: the topic subject, the stream's subjects, its consumers' deliver subjects and its request inboxes) and
++# exactly the JetStream API and ack subjects of its one stream (NATS_STREAM, default OV_EVENTS_STREAM).
++authorization {
++  users = [
++    {
++      user: events
++      password: $NATS_EVENTS_PASSWORD
++      permissions: {
++        publish: [
++          "ov.events.>"
++          "$JS.API.STREAM.INFO.OV_EVENTS_STREAM"
++          "$JS.API.STREAM.CREATE.OV_EVENTS_STREAM"
++          "$JS.API.CONSUMER.CREATE.OV_EVENTS_STREAM.>"
++          "$JS.API.CONSUMER.DELETE.OV_EVENTS_STREAM.>"
++          "$JS.ACK.OV_EVENTS_STREAM.>"
++        ]
++        subscribe: ["ov.events.>"]
++      }
++    }
++  ]
++}
ddiff --git a/scripts/test-nats.sh b/scripts/test-nats.sh
iindex 87886c5..e05f93c 100755
---- a/scripts/test-nats.sh
++++ b/scripts/test-nats.sh
@@@ -1,20 +1,22 @@
  #!/usr/bin/env bash
--# A NATS Core broker in a container for test/fabric-nats.test.js (the optional nats-v1 TOPIC carrier). Not production:
--# loopback only, no JetStream; scripts/nats-test.conf adds two users without publish or subscribe permission.
++# A NATS broker with JetStream (-js) in a container for test/fabric-nats.test.js (the optional nats-v1 TOPIC carrier)
++# and test/fabric-jetstream.test.js (the nats-js-v1 STREAM carrier). Not production: loopback only, the JetStream store
++# inside the container; scripts/nats-test.conf adds two users without publish or subscribe permission.
  #
  #   scripts/test-nats.sh up      start (or reuse) and print the env to export (OV_TEST_NATS_URL)
  #   scripts/test-nats.sh down    remove the container
  #
--# Without OV_TEST_NATS_URL the broker-backed cases print "fabric nats: skipped (…)" and the rest still runs.
++# Without OV_TEST_NATS_URL the broker-backed cases print "fabric nats: skipped (…)" (and "fabric jetstream: skipped
++# (…)") and the rest still runs.
  set -euo pipefail
  NAME=events-test-nats; PORT=${OV_TEST_NATS_PORT:-54222}; IMAGE=nats:2.11-alpine
--CONF="$(cd "$(dirname "$0")" && pwd)/nats-test.conf"; SUM=$(sha256sum "$CONF" | cut -c1-12)
++CONF="$(cd "$(dirname "$0")" && pwd)/nats-test.conf"; SUM=js-$(sha256sum "$CONF" | cut -c1-12)
  case "${1:-up}" in
  up)
--  # Reused only while it runs this config on this port (the labels carry the checksum and the port).
++  # Reused only while it runs this config, with -js, on this port (the labels carry the checksum and the port).
    if [ "$(docker inspect -f '{{.State.Running}} {{index .Config.Labels "ov.conf"}} {{index .Config.Labels "ov.port"}}' $NAME 2>/dev/null)" != "true $SUM $PORT" ]; then
      docker rm -f $NAME >/dev/null 2>&1 || true
--    docker run -d --name $NAME --label ov.conf=$SUM --label ov.port=$PORT -p 127.0.0.1:$PORT:4222 -v "$CONF:/etc/nats/test.conf:ro" $IMAGE -c /etc/nats/test.conf >/dev/null
++    docker run -d --name $NAME --label ov.conf=$SUM --label ov.port=$PORT -p 127.0.0.1:$PORT:4222 -v "$CONF:/etc/nats/test.conf:ro" $IMAGE -js -c /etc/nats/test.conf >/dev/null
    fi
    ready=""
    for i in $(seq 1 40); do (exec 3<>/dev/tcp/127.0.0.1/$PORT) 2>/dev/null && { ready=1; break; }; sleep 0.25; done
ddiff --git a/server/config.js b/server/config.js
iindex da1a002..9efa412 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -92,18 +92,24 @@ function load(env = process.env) {
          // PostgreSQL (ADR-035): DATABASE_URL serves (PgBouncer), DATABASE_DIRECT_URL migrates (owner role).
          db: { url: env.DATABASE_URL || '', directUrl: env.DATABASE_DIRECT_URL || '' },
          // Carriers (ADR-042 decisions 1-2, 5; server/fabric/carriers): the adapters to instantiate and the ones the
--        // operator switch removes. pg-v1 is always present; valkey-v1 needs VALKEY_URL and nats-v1 NATS_URL, each
--        // silently absent without it. A disabled adapter is ineligible at once (reason `disabled by configuration`):
--        // the ADR's rollback.
++        // operator switch removes. pg-v1 is always present; valkey-v1 needs VALKEY_URL, nats-v1 NATS_URL and nats-js-v1
++        // NATS_URL with JetStream on, each silently absent without it. A disabled adapter is ineligible at once (reason
++        // `disabled by configuration`): the ADR's rollback.
          carriers: {
--            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1', 'nats-v1']),
++            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1']),
              disabled: list(env.EVENTS_CARRIERS_DISABLED, []),
          },
          // Valkey (ADR-035): per-actor limit counters shared across processes; without it they count in this process.
          valkey: { url: env.VALKEY_URL || '', prefix: env.VALKEY_PREFIX || 'ov:events:' },
--        // NATS Core (ADR-042 decision 5, optional): the nats-v1 TOPIC carrier, cross-process SSE fan-out. Loopback or
--        // the private network only; nats:// with an optional user:pass or token. Never the record.
--        nats: { url: env.NATS_URL || '', subjectPrefix: env.NATS_SUBJECT_PREFIX || 'ov.events.' },
++        // NATS (ADR-042 decisions 5-6, optional): the nats-v1 TOPIC carrier (cross-process SSE fan-out) and, unless
++        // NATS_JETSTREAM=off, the nats-js-v1 STREAM carrier on the JetStream stream NATS_STREAM. Loopback or the private
++        // network only; nats:// with an optional user:pass or token. Never the record.
++        nats: {
++            url: env.NATS_URL || '',
++            subjectPrefix: env.NATS_SUBJECT_PREFIX || 'ov.events.',
++            jetstream: env.NATS_JETSTREAM !== 'off',
++            stream: env.NATS_STREAM || 'OV_EVENTS_STREAM',
++        },
          retentionDays: int(env.EVENTS_RETENTION_DAYS, 30),
          receiptRetentionDays: int(env.EVENTS_RECEIPT_RETENTION_DAYS, 90),
          pruneIntervalMs: int(env.EVENTS_PRUNE_INTERVAL_MS, 60 * 60 * 1000),
ddiff --git a/server/fabric/carriers/index.js b/server/fabric/carriers/index.js
iindex ae4f4c9..911581f 100644
---- a/server/fabric/carriers/index.js
++++ b/server/fabric/carriers/index.js
@@@ -3,11 +3,13 @@
   * The carrier registry (ADR-042 decisions 1-2, 5): one adapter per carrier, each with an id, the classes it can
   * carry, an offer, health, a signal() and a start/stop. pg-v1 is always present and carries every class; valkey-v1
   * is registered only when VALKEY_URL is set and carries QUEUE and TOPIC; nats-v1 is registered only when NATS_URL is
-- * set and carries TOPIC over NATS Core. JetStream is not built yet (it needs the host unit), so STREAM is left to pg-v1.
++ * set and carries TOPIC over NATS Core; nats-js-v1 is registered when NATS_URL is set and NATS_JETSTREAM is not `off`
++ * and carries STREAM over JetStream. Without them STREAM is pg-v1's.
   *
   * Config (server/config.js):
-- *   EVENTS_CARRIERS           ids to instantiate (default pg-v1,valkey-v1,nats-v1). An unknown id refuses to start.
-- *   NATS_URL                  nats://[user:pass@]host:port of the cell's NATS (nats-v1); NATS_SUBJECT_PREFIX its subjects.
++ *   EVENTS_CARRIERS           ids to instantiate (default pg-v1,valkey-v1,nats-v1,nats-js-v1). An unknown id refuses to start.
++ *   NATS_URL                  nats://[user:pass@]host:port of the cell's NATS (nats-v1, nats-js-v1); NATS_SUBJECT_PREFIX
++ *                             its subjects; NATS_JETSTREAM=off leaves nats-js-v1 out; NATS_STREAM its stream name.
   *   EVENTS_CARRIERS_DISABLED  ids excluded at once with reason `disabled by configuration` (the ADR's rollback).
   *
   * The planner asks offers(class) for one resource-offer@1 per eligible adapter and forClass(class) for the
@@@ -16,8 +18,9 @@
  const { createPgCarrier } = require('./pg');
  const { createValkeyCarrier, newInstanceId } = require('./valkey');
  const { createNatsCarrier } = require('./nats');
++const { createJetStreamCarrier } = require('./jetstream');
  
--const KNOWN = ['pg-v1', 'valkey-v1', 'nats-v1'];
++const KNOWN = ['pg-v1', 'valkey-v1', 'nats-v1', 'nats-js-v1'];
  
  function createCarriers({ config, clock = { now: () => Date.now() }, log = console, valkey = null } = {}) {
      const enabled = new Set(config.carriers.enabled);
@@@ -33,9 +36,12 @@ function createCarriers({ config, clock = { now: () => Date.now() }, log = conso
          v.disabledReason = off('valkey-v1') ? 'disabled by configuration' : null;
          adapters.push(v);
      }
--    // nats-v1 connects in start(), which skips a disabled adapter: switched off, it never opens a socket.
++    // nats-v1 and nats-js-v1 connect in start(), which skips a disabled adapter: switched off, they never open a socket.
      if (config.nats && config.nats.url) {
          adapters.push(createNatsCarrier({ url: config.nats.url, subjectPrefix: config.nats.subjectPrefix, clock, log, instanceId: newInstanceId() }));
++        if (config.nats.jetstream) {
++            adapters.push(createJetStreamCarrier({ url: config.nats.url, subjectPrefix: config.nats.subjectPrefix, stream: config.nats.stream, clock, log, instanceId: newInstanceId() }));
++        }
      }
      for (const a of adapters) if (!a.disabledReason && off(a.id)) a.disabledReason = 'disabled by configuration';
  
@@@ -60,23 +66,25 @@ function createCarriers({ config, clock = { now: () => Date.now() }, log = conso
          forClass,
          eligible,
  
--        /** Signal one carrier with rows (kind is 'QUEUE' or 'TOPIC'); fail open — never throws to the caller. */
++        /** Signal one carrier with rows (kind is 'QUEUE', 'TOPIC' or 'STREAM'); fail open — never throws to the caller. */
          async signal(carrierId, rows, kind) {
              const a = get(carrierId);
              if (!a || a.disabledReason || !rows || !rows.length) return;
              try { await a.signal(rows, kind); } catch (err) { log.warn(`[carriers] ${carrierId} signal: ${err.message}`); }
          },
  
--        /** After a publish commits: QUEUE deliveries to their stream, TOPIC events to their channel. The worker's
--         *  kick() beside this is what makes a local process fast; a remote process is woken by the carrier. */
++        /** After a publish commits: QUEUE and STREAM deliveries to their carrier, TOPIC events to their channel. The
++         *  worker's kick() beside this is what makes a local process fast; a remote process is woken by the carrier. */
          async signalPublish(events = [], deliveries = []) {
--            const byQueue = new Map();
--            for (const d of deliveries) {
--                if (d.carrier_class !== 'QUEUE' || !d.carrier) continue;
--                if (!byQueue.has(d.carrier)) byQueue.set(d.carrier, []);
--                byQueue.get(d.carrier).push(d);
++            for (const kind of ['QUEUE', 'STREAM']) {
++                const byCarrier = new Map();
++                for (const d of deliveries) {
++                    if (d.carrier_class !== kind || !d.carrier) continue;
++                    if (!byCarrier.has(d.carrier)) byCarrier.set(d.carrier, []);
++                    byCarrier.get(d.carrier).push(d);
++                }
++                for (const [cid, rows] of byCarrier) await this.signal(cid, rows, kind);
              }
--            for (const [cid, rows] of byQueue) await this.signal(cid, rows, 'QUEUE');
  
              const byTopic = new Map();
              for (const e of events) {
ddiff --git a/server/fabric/carriers/jetstream.js b/server/fabric/carriers/jetstream.js
nnew file mode 100644
iindex 0000000..9c4829e
---- /dev/null
++++ b/server/fabric/carriers/jetstream.js
@@@ -0,0 +1,196 @@
++'use strict';
++/**
++ * nats-js-v1 — the JetStream STREAM carrier (ADR-042 decisions 1, 2, 6: low-latency, durable, cell-local), only when
++ * NATS_URL is set and NATS_JETSTREAM is not `off`. Without it, or with EVENTS_CARRIERS_DISABLED=nats-js-v1, nothing
++ * connects and STREAM stays on pg-v1.
++ *
++ * PostgreSQL stays the record. After a publish commits, signal(rows) publishes one JetStream message per new STREAM
++ * delivery — only `{ event_id, subscription_id }`, never content — on `<prefix>fabric.stream.<key>` of the stream
++ * NATS_STREAM (default OV_EVENTS_STREAM), and waits for the stream's ack. `<key>` is a subject-safe hash of the
++ * ordering key (a key may hold dots, spaces or wildcards), so one key is one subject.
++ *
++ * Every process that runs the delivery worker has its own durable push consumer on that stream (deliver_policy new,
++ * ack explicit, max_ack_pending 1) and handles its messages one at a time in stream order: claim exactly that row
++ * (handlers.claim -> store.claimDelivery: the same key-head lease, max_inflight and due rules as the poll), send it
++ * through the worker (handlers.deliver), then ack. The message is acked whatever the claim answered: a null claim
++ * means another process or the poll has it, it is not its key's head yet, it is done, or it was never committed (a
++ * forged message names nothing) — the row stays pending and the 500 ms poll finds it when it is due. A process that
++ * dies before its ack gets the message again after ack_wait (same durable on restart): the claim finds the row
++ * delivered or leased, so the redelivery sends nothing. Per-key order is the store's (a key's head first, one lease
++ * per key); the stream order only makes the head usually arrive first. A durable left by a crashed process is removed
++ * by the broker after inactive_threshold; a clean stop deletes it.
++ *
++ * The JetStream API ($JS.API.*) is spoken as JSON over ../nats-core.js request/reply, without a client library: Events
++ * needs only stream info/create, consumer create/delete, publish-with-ack and ack. Replies come back under
++ * `<prefix>_inbox.`, so the broker user needs only `<prefix>>` plus that stream's API and ack subjects
++ * (deploy/nats/nats-server.conf). Needs nats-server 2.10+ started with -js.
++ *
++ * Health: connected, no permissions violation, the stream ensured on this connection (re-checked after every
++ * reconnect, retried while it fails) and the breaker on consecutive publish errors (../signals.js). While any fails the
++ * planner drops nats-js-v1 with the reason from unhealthyReason() and pg-v1 carries STREAM; nothing is lost because
++ * nothing depends on the message.
++ */
++const crypto = require('crypto');
++const { createNatsCore } = require('../nats-core');
++const { createSignals } = require('../signals');
++
++const STREAM = 'OV_EVENTS_STREAM';
++const SUBJECT_ROOT = 'fabric.stream';
++const NS = 1e6;   // JetStream durations are nanoseconds
++const ID_MAX = 64;
++
++/** The subject token of an ordering key: unambiguous, fixed length, always subject-safe. */
++function keyToken(key) {
++    return key == null || key === '' ? '_' : crypto.createHash('sha256').update(String(key)).digest('base64url').slice(0, 22);
++}
++
++function createJetStreamCarrier({ url, subjectPrefix = 'ov.events.', stream = STREAM, clock = { now: () => Date.now() }, log = console, instanceId,
++    consumerName = null, ackWaitMs = 60000, maxAgeMs = 24 * 3600 * 1000, requestTimeoutMs = 2000, publishTimeoutMs = 1000, retryMs = 5000, deniedRetryMs, inactiveMs = 3600 * 1000 } = {}) {
++    if (!/^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*\.$/.test(subjectPrefix)) throw new Error(`NATS_SUBJECT_PREFIX "${subjectPrefix}" must be dot-separated tokens ending in a dot`);
++    if (!/^[A-Za-z0-9_-]{1,64}$/.test(stream)) throw new Error(`NATS_STREAM "${stream}" must be 1-64 letters, digits, _ or -`);
++    const signals = createSignals({ now: () => clock.now() });
++    const durable = (consumerName || `events_${String(instanceId || crypto.randomBytes(6).toString('hex'))}`).replace(/[^A-Za-z0-9_-]/g, '_').slice(0, 64);
++    const subjects = `${subjectPrefix}${SUBJECT_ROOT}.>`;
++    const deliverSubject = `${subjectPrefix}fabric.jsdeliver.${durable}`;
++    const ackPrefix = `$JS.ACK.${stream}.${durable}.`;
++    let handlers = null;      // set by start() when this process consumes
++    let streamReady = false;  // the stream (and our consumer, when consuming) exist on the current connection
++    let lastError = null;
++    let ensuring = null;
++    let retryTimer = null;
++    let retryDelay = retryMs;        // doubles up to a minute while the stream cannot be ensured (no -js, refused)
++    let chain = Promise.resolve();   // consumer messages, one after another
++    let unsubscribe = null;
++    let stopped = false;
++
++    const nats = createNatsCore({
++        url, name: `openvibe-events js ${instanceId || durable}`, log, probe: [deliverSubject], deniedRetryMs,
++        inboxPrefix: `${subjectPrefix}_inbox.${crypto.randomBytes(8).toString('hex')}.`,
++        onConnect: () => { streamReady = false; ensure(); },
++    });
++
++    async function api(subject, body = {}) {
++        const r = await nats.request(`$JS.API.${subject}`, body, { timeoutMs: requestTimeoutMs });
++        if (r && r.error) throw Object.assign(new Error(`${subject}: ${r.error.description || 'error'} (${r.error.err_code || r.error.code})`), { jsCode: r.error.err_code, status: r.error.code });
++        return r;
++    }
++
++    // Stream, then (when consuming) our durable push consumer. Idempotent: an existing stream is used as it is.
++    function ensure() {
++        if (ensuring || stopped) return ensuring;
++        ensuring = (async () => {
++            try {
++                try { await api(`STREAM.INFO.${stream}`); }
++                catch (err) {
++                    if (err.jsCode !== 10059 && err.status !== 404) throw err;
++                    await api(`STREAM.CREATE.${stream}`, {
++                        name: stream, subjects: [subjects], retention: 'limits', storage: 'file', discard: 'old',
++                        max_age: maxAgeMs * NS, max_msg_size: 1024, duplicate_window: 120000 * NS, num_replicas: 1,
++                    }).catch((e) => { if (e.jsCode !== 10058) throw e; });   // another process created it meanwhile
++                }
++                if (handlers) {
++                    await api(`CONSUMER.CREATE.${stream}.${durable}`, {
++                        stream_name: stream,
++                        config: {
++                            durable_name: durable, deliver_subject: deliverSubject, filter_subject: subjects,
++                            deliver_policy: 'new', ack_policy: 'explicit', ack_wait: ackWaitMs * NS, max_ack_pending: 1,
++                            max_deliver: 5, replay_policy: 'instant', inactive_threshold: inactiveMs * NS,
++                        },
++                    });
++                }
++                streamReady = true; lastError = null; retryDelay = retryMs;
++            } catch (err) {
++                streamReady = false; lastError = err.message;
++                log.warn(`[nats-js-v1] stream not ready: ${err.message}`);
++                if (!retryTimer && !stopped) {
++                    retryTimer = setTimeout(() => { retryTimer = null; if (nats.connected()) ensure(); }, retryDelay);
++                    retryTimer.unref?.();
++                    retryDelay = Math.min(retryDelay * 2, 60000);
++                }
++            } finally { ensuring = null; }
++        })();
++        return ensuring;
++    }
++
++    async function handle(msg, reply) {
++        // Only JetStream's own deliveries of this consumer carry an ack subject; anything else on the subject is ignored.
++        if (!reply || !reply.startsWith(ackPrefix)) return;
++        try {
++            const ok = msg && typeof msg.event_id === 'string' && typeof msg.subscription_id === 'string'
++                && msg.event_id.length <= ID_MAX && msg.subscription_id.length <= ID_MAX;
++            const row = ok ? await handlers.claim(msg.event_id, msg.subscription_id) : null;
++            if (row) await handlers.deliver(row);
++        } catch (err) {
++            log.warn(`[nats-js-v1] delivery ${msg && msg.event_id}: ${err.message}`);   // acked anyway: the poll retries
++        }
++        try { nats.publishRaw(reply, ''); } catch { /* not acked: redelivered after ack_wait, and the claim dedupes */ }
++    }
++
++    async function signalStream(rows) {
++        if (!streamReady) return;   // not ready: the poll carries these (the planner already prefers pg-v1)
++        const t0 = clock.now();
++        try {
++            const acks = await Promise.all(rows.map((r) => nats.request(`${subjectPrefix}${SUBJECT_ROOT}.${keyToken(r.ordering_key)}`,
++                { event_id: r.event_id, subscription_id: r.subscription_id }, { timeoutMs: publishTimeoutMs })));
++            const bad = acks.find((a) => !a || a.error || a.stream !== stream);
++            if (bad) throw new Error(`stream publish refused: ${(bad && bad.error && bad.error.description) || JSON.stringify(bad)}`);
++            signals.record('stream', true, clock.now() - t0);
++        } catch (err) {
++            signals.record('stream', false);
++            log.warn(`[nats-js-v1] stream publish: ${err.message}`);
++        }
++    }
++
++    return {
++        id: 'nats-js-v1',
++        classes: new Set(['STREAM']),
++        client: nats,
++        signals,
++        durable,
++
++        async start(h = {}) {
++            stopped = false;
++            // The consumer is a delivery path: only where the delivery worker runs (EVENTS_WORKER), as valkey-v1's QUEUE.
++            if (h.consumeQueue !== false && h.claim && h.deliver) {
++                handlers = h;
++                unsubscribe = nats.subscribe(deliverSubject, (msg, meta) => { chain = chain.then(() => handle(msg, meta.reply)); });
++            }
++            // Not fatal: the client keeps reconnecting, and nats-js-v1 is ineligible (not connected) until it is up.
++            await nats.connect().catch((err) => log.warn(`[nats-js-v1] not connected yet: ${err.message}`));
++            if (ensuring) await ensuring;
++        },
++
++        async stop() {
++            stopped = true;
++            if (retryTimer) { clearTimeout(retryTimer); retryTimer = null; }
++            if (unsubscribe) { await unsubscribe().catch(() => {}); unsubscribe = null; }
++            await chain;
++            if (handlers && nats.connected()) await api(`CONSUMER.DELETE.${stream}.${durable}`).catch(() => {});
++            await nats.close();
++        },
++
++        signal(rows, kind) { return kind === 'STREAM' ? signalStream(rows) : undefined; },
++        healthy: () => nats.connected() && !nats.denied() && streamReady && signals.healthy('stream'),
++        unhealthyReason: () => (!nats.connected() ? 'not connected (nats-js-v1)' : nats.denied() ? 'permission denied (nats-js-v1)'
++            : !streamReady ? `stream not ready (nats-js-v1)${lastError ? `: ${lastError}` : ''}` : !signals.healthy('stream') ? 'breaker open (nats-js-v1)' : null),
++
++        offer() {
++            const ms = signals.ewma('stream') || 1;
++            return {
++                offer_id: 'nats-js-v1',
++                kind: 'provider',
++                provider: 'nats',
++                adapter: 'nats-js-v1',
++                region: 'cell',
++                trust: 'first-party',
++                capabilities: ['events:gateway', 'events:durable', 'events:ordered'],
++                latency_ms: { publish_p95: ms, ack_p95: ms },
++                health: { status: 'up', checked_at: new Date(clock.now()).toISOString() },
++                pricing: { model: 'per-operation', unit: 'stream-operation', rate_card: 'rc-nats-js-v1' },
++                updated_at: new Date(clock.now()).toISOString(),
++            };
++        },
++    };
++}
++
++module.exports = { createJetStreamCarrier, STREAM, SUBJECT_ROOT, keyToken };
ddiff --git a/server/fabric/carriers/nats.js b/server/fabric/carriers/nats.js
iindex 3f0468a..374f6db 100644
---- a/server/fabric/carriers/nats.js
++++ b/server/fabric/carriers/nats.js
@@@ -18,7 +18,7 @@
   * consecutive publish errors (../signals.js). While any fails the planner drops nats-v1 with the reason from
   * unhealthyReason(); the client reconnects in the background and the adapter is eligible again once it is back.
   *
-- * JetStream (STREAM) is not built here: STREAM stays on pg-v1.
++ * JetStream (STREAM) is ./jetstream.js (nats-js-v1), on the same broker and client.
   */
  const { createNatsCore } = require('../nats-core');
  const { createSignals } = require('../signals');
ddiff --git a/server/fabric/nats-core.js b/server/fabric/nats-core.js
iindex 1a68f39..97dd647 100644
---- a/server/fabric/nats-core.js
++++ b/server/fabric/nats-core.js
@@@ -1,11 +1,12 @@
  'use strict';
  /**
   * A minimal NATS Core client (the text protocol: INFO, CONNECT, PING/PONG, PUB, SUB, UNSUB, MSG, +OK, -ERR) for the
-- * nats-v1 TOPIC carrier. Core NATS only: no JetStream, no headers, no request/reply, no TLS (a server that requires
-- * TLS is refused). It is kept here, without a dependency, because Events needs exactly fire-and-forget publish and a
-- * subscription; the broker is on loopback or the private network (deploy/nats).
++ * nats-v1 TOPIC carrier and the nats-js-v1 STREAM carrier. No headers and no TLS (a server that requires TLS is
++ * refused). JetStream is reached through its JSON API over plain request/reply (carriers/jetstream.js), so it needs no
++ * client library either. It is kept here, without a dependency, because Events needs exactly publish, a subscription
++ * and request/reply; the broker is on loopback or the private network (deploy/nats).
   *
-- *   createNatsCore({ url, name, log, reconnectMs, maxReconnectMs, connectTimeoutMs, probe, deniedRetryMs })
++ *   createNatsCore({ url, name, log, reconnectMs, maxReconnectMs, connectTimeoutMs, probe, deniedRetryMs, inboxPrefix, onConnect })
   *     connect()               resolves after the first PONG; rejects on refusal, auth error or timeout. A dropped
   *                             connection reconnects with backoff (from reconnectMs up to maxReconnectMs) and
   *                             re-subscribes; connect() rejecting does not stop the retries.
@@@ -15,13 +16,21 @@
   *                             to each `probe` subject before its PING, so a refusal shows up before the first real
   *                             publish. A denied connection is re-opened after deniedRetryMs: a handshake without a
   *                             violation clears it.
-- *     publish(subject, obj)   JSON-encodes and writes one PUB; throws when not connected or over max_payload
-- *     subscribe(subject, fn)  fn(obj) per MSG (a non-JSON payload is dropped); returns an unsubscribe function
++ *     publish(subject, obj, reply)       JSON-encodes and writes one PUB (with an optional reply subject); throws
++ *                                        when not connected or over max_payload
++ *     publishRaw(subject, data, reply)   the same with a string or Buffer as is (a JetStream ack is an empty body)
++ *     request(subject, obj, { timeoutMs })  publish with a reply subject under `inboxPrefix` (default `_INBOX.<id>.`)
++ *                             and resolve with the first JSON reply; rejects on timeout (a non-JSON reply is
++ *                             dropped, so it times out too) or close()
++ *     subscribe(subject, fn)  fn(obj, { subject, reply }) per MSG (a non-JSON payload is dropped); returns an
++ *                             unsubscribe function
++ *     onConnect()             (option) called after every handshake, the first and each reconnect
   *     close()                 stops reconnecting and closes the socket
   *
   * The URL is nats://[user:pass@|token@]host[:4222]. It is never logged (it can carry a credential).
   */
  const net = require('net');
++const crypto = require('crypto');
  
  const CRLF = '\r\n';
  const MAX_LINE = 64 * 1024;
@@@ -40,7 +49,8 @@ function parseUrl(raw) {
      };
  }
  
--function createNatsCore({ url, name = 'openvibe-events', log = console, reconnectMs = 250, maxReconnectMs = 5000, connectTimeoutMs = 2000, probe = [], deniedRetryMs = 30000 } = {}) {
++function createNatsCore({ url, name = 'openvibe-events', log = console, reconnectMs = 250, maxReconnectMs = 5000, connectTimeoutMs = 2000, probe = [], deniedRetryMs = 30000,
++    inboxPrefix = `_INBOX.${crypto.randomBytes(8).toString('hex')}.`, onConnect = null } = {}) {
      const target = parseUrl(url);
      const where = `${target.host}:${target.port}`;
      const subs = new Map();   // sid -> { subject, fn }
@@@ -54,6 +64,9 @@ function createNatsCore({ url, name = 'openvibe-events', log = console, reconnec
      let first = null;         // { resolve, reject } of the pending connect()
      let denied = null;        // the last permissions violation on the current connection
      let deniedTimer = null;
++    const pending = new Map();   // request token -> { resolve, reject, timer }
++    let nextToken = 1;
++    let inboxSid = 0;            // the one wildcard subscription replies arrive on (made at the first request)
  
      function settle(err) {
          if (!first) return;
@@@ -104,7 +117,8 @@ function createNatsCore({ url, name = 'openvibe-events', log = console, reconnec
                      if (sub) {
                          let obj;
                          try { obj = JSON.parse(payload.toString('utf8')); } catch { continue; }
--                        try { sub.fn(obj); } catch (err) { log.warn(`[nats] handler ${sub.subject}: ${err.message}`); }
++                        const meta = { subject: parts[1], reply: parts.length === 5 ? parts[3] : null };
++                        try { sub.fn(obj, meta); } catch (err) { log.warn(`[nats] handler ${sub.subject}: ${err.message}`); }
                      }
                      continue;
                  }
@@@ -129,6 +143,7 @@ function createNatsCore({ url, name = 'openvibe-events', log = console, reconnec
                          denied = violation;
                          if (denied) retryDenied(s);
                          settle(null);
++                        if (onConnect) { try { onConnect(); } catch (err) { log.warn(`[nats] onConnect: ${err.message}`); } }
                      }
                  } else if (op === '-ERR') {
                      const why = line.slice(5).trim();
@@@ -152,6 +167,35 @@ function createNatsCore({ url, name = 'openvibe-events', log = console, reconnec
          });
      }
  
++    function publishRaw(subject, body, reply) {
++        if (!ready || !sock) throw new Error(`nats ${where}: not connected`);
++        if (sock.writableLength > MAX_BUFFERED) throw new Error(`nats ${where}: ${sock.writableLength} bytes unsent (the broker is not reading)`);
++        const data = Buffer.isBuffer(body) ? body : Buffer.from(String(body ?? ''), 'utf8');
++        if (data.length > maxPayload) throw Object.assign(new Error(`nats: payload ${data.length} bytes over max_payload ${maxPayload}`), { code: 'max_payload' });
++        sock.write(Buffer.concat([Buffer.from(`PUB ${subject}${reply ? ` ${reply}` : ''} ${data.length}${CRLF}`), data, Buffer.from(CRLF)]));
++    }
++
++    function subscribe(subject, fn) {
++        const sid = nextSid++;
++        subs.set(sid, { subject, fn });
++        if (ready && sock) sock.write(`SUB ${subject} ${sid}${CRLF}`);
++        return {
++            sid,
++            async unsubscribe() {
++                subs.delete(sid);
++                if (ready && sock) sock.write(`UNSUB ${sid}${CRLF}`);
++            },
++        };
++    }
++
++    function onReply(obj, { subject }) {
++        const p = pending.get(subject.slice(inboxPrefix.length));
++        if (!p) return;   // late (timed out) or not ours
++        pending.delete(subject.slice(inboxPrefix.length));
++        clearTimeout(p.timer);
++        p.resolve(obj);
++    }
++
      return {
          connect() {
              if (ready) return Promise.resolve();
@@@ -163,24 +207,23 @@ function createNatsCore({ url, name = 'openvibe-events', log = console, reconnec
          },
          connected: () => ready,
          denied: () => (ready ? denied : null),
--        publish(subject, obj) {
--            if (!ready || !sock) throw new Error(`nats ${where}: not connected`);
--            if (sock.writableLength > MAX_BUFFERED) throw new Error(`nats ${where}: ${sock.writableLength} bytes unsent (the broker is not reading)`);
--            const data = Buffer.from(JSON.stringify(obj), 'utf8');
--            if (data.length > maxPayload) throw Object.assign(new Error(`nats: payload ${data.length} bytes over max_payload ${maxPayload}`), { code: 'max_payload' });
--            sock.write(Buffer.concat([Buffer.from(`PUB ${subject} ${data.length}${CRLF}`), data, Buffer.from(CRLF)]));
--        },
--        subscribe(subject, fn) {
--            const sid = nextSid++;
--            subs.set(sid, { subject, fn });
--            if (ready && sock) sock.write(`SUB ${subject} ${sid}${CRLF}`);
--            return async () => {
--                subs.delete(sid);
--                if (ready && sock) sock.write(`UNSUB ${sid}${CRLF}`);
--            };
++        publish(subject, obj, reply) { publishRaw(subject, JSON.stringify(obj), reply); },
++        publishRaw,
++        request(subject, obj, { timeoutMs = 2000 } = {}) {
++            if (!inboxSid) inboxSid = subscribe(`${inboxPrefix}*`, onReply).sid;
++            const token = String(nextToken++);
++            return new Promise((resolve, reject) => {
++                const timer = setTimeout(() => { pending.delete(token); reject(new Error(`nats ${where}: request ${subject} timed out`)); }, timeoutMs);
++                timer.unref?.();
++                pending.set(token, { resolve, reject, timer });
++                try { publishRaw(subject, JSON.stringify(obj), `${inboxPrefix}${token}`); }
++                catch (err) { clearTimeout(timer); pending.delete(token); reject(err); }
++            });
          },
++        subscribe(subject, fn) { return subscribe(subject, fn).unsubscribe; },
          async close() {
              stopped = true;
++            for (const [token, p] of pending) { clearTimeout(p.timer); pending.delete(token); p.reject(new Error('nats: closed')); }
              if (timer) { clearTimeout(timer); timer = null; }
              if (deniedTimer) { clearTimeout(deniedTimer); deniedTimer = null; }
              ready = false;
ddiff --git a/server/fabric/planner.js b/server/fabric/planner.js
iindex 1cd5564..e9f4407 100644
---- a/server/fabric/planner.js
++++ b/server/fabric/planner.js
@@@ -29,6 +29,10 @@ const LATENCY_CLASS = {
      domain: 'interactive', critical: 'critical', background: 'background', bulk: 'bulk', webhook: 'interactive',
  };
  
++// STREAM (durability required + an ordering key) may only go to a carrier that keeps its messages and an order per key;
++// QUEUE and TOPIC carriers only signal over the PostgreSQL record, so they ask the gateway capability alone.
++const CLASS_CAPABILITIES = { STREAM: ['events:gateway', 'events:durable', 'events:ordered'] };
++
  function requirements(klass, key, policy) {
      const lat = (policy && policy.latency) || {};
      const objective = OBJECTIVES[(policy && policy.routing && policy.routing.objective) || 'balanced'] || 'balanced';
@@@ -38,7 +42,7 @@ function requirements(klass, key, policy) {
          latency_class: LATENCY_CLASS[klass] || 'interactive',
          objective,
          partition_key: key || undefined,
--        capabilities: ['events:gateway'],
++        capabilities: CLASS_CAPABILITIES[klass] || ['events:gateway'],
          latency_op: 'ack_p95',
          ...(Number.isFinite(lat.maximum_p95_ms) ? { max_latency_ms: lat.maximum_p95_ms } : {}),
          units: 1,
ddiff --git a/server/fabric/rate-cards.json b/server/fabric/rate-cards.json
iindex 15515a0..4bcdadd 100644
---- a/server/fabric/rate-cards.json
++++ b/server/fabric/rate-cards.json
@@@ -40,5 +40,19 @@
      "effective_from": "2026-10-02",
      "source": "https://openvibe.network/pricing/events",
      "verified_at": "2026-10-02"
++  },
++  {
++    "id": "rc-nats-js-v1",
++    "provider": "nats",
++    "adapter": "nats-js-v1",
++    "metric": "stream-operation",
++    "unit_size": 1,
++    "unit_price_usd": 0.0000002,
++    "free_allowance": 1000000000,
++    "reset_period": "month",
++    "region": "cell",
++    "effective_from": "2026-10-04",
++    "source": "https://openvibe.network/pricing/events",
++    "verified_at": "2026-10-04"
    }
  ]
ddiff --git a/server/store.js b/server/store.js
iindex 926c6d7..99d6a7c 100644
---- a/server/store.js
++++ b/server/store.js
@@@ -258,7 +258,7 @@ function createStore(db, { clock = { now: () => Date.now() }, maxHops = 8, usage
                      if (pin && pin.carrier) carrier = pin.carrier;
                  }
                  await q.insertDelivery.run(env.event_id, s.id, seq, PRIORITY_RANK[env.priority], now, key, carrier, now, now);
--                if (carrier) deliveries.push({ event_id: env.event_id, subscription_id: s.id, carrier, carrier_class: klass });
++                if (carrier) deliveries.push({ event_id: env.event_id, subscription_id: s.id, carrier, carrier_class: klass, ordering_key: key });
              }
              await revocationHook(row, now);
              results.push({ event_id: env.event_id, seq, duplicate: false, cursor: cursor.encode(seq, epoch) });
ddiff --git a/test/fabric-jetstream.test.js b/test/fabric-jetstream.test.js
nnew file mode 100644
iindex 0000000..009a7e0
---- /dev/null
++++ b/test/fabric-jetstream.test.js
@@@ -0,0 +1,312 @@
++'use strict';
++// nats-js-v1 (ADR-042 decisions 1, 2, 6, optional): the JetStream STREAM carrier. The registry builds it only with
++// NATS_URL and NATS_JETSTREAM not off, and honours the disable switch; the planner asks events:durable +
++// events:ordered for STREAM, prefers nats-js-v1 while it is healthy and explains why it is excluded when it is not.
++// Against a real broker: (a) a durability-required + ordering-key policy is STREAM and the explain API selects
++// nats-js-v1; (b) a broker outage drops it to pg-v1 with the reason in excluded_because and nothing is lost (the poll
++// delivers what the carrier signalled and what it did not), and it comes back; (c) two consuming workers and one due
++// delivery send it exactly once; (d) per-key order holds across two keys and across a crash before the ack (the
++// redelivery claims nothing); (e) a forged (event_id, subscription_id) that was never committed claims nothing;
++// (f) EVENTS_CARRIERS_DISABLED=nats-js-v1 never connects and leaves STREAM on pg-v1. The broker parts need
++// OV_TEST_NATS_URL (scripts/test-nats.sh up, which starts the broker with -js); without it they print one
++// `skipped (…)` line.
++const assert = require('assert');
++const net = require('net');
++const { validate, ids } = require('openvibe-contracts');
++const { boot, request, serviceToken, envelope, subscriber, suite, sleep, manualClock } = require('./helpers');
++const { testDb } = require('./db');
++const { load } = require('../server/config');
++const { createCarriers } = require('../server/fabric/carriers');
++const { createPlanner, RATE_CARDS } = require('../server/fabric/planner');
++const { createJetStreamCarrier, keyToken } = require('../server/fabric/carriers/jetstream');
++const { createNatsCore } = require('../server/fabric/nats-core');
++
++const t = suite('fabric-jetstream');
++const NATS = process.env.OV_TEST_NATS_URL || '';
++const admin = serviceToken('ops', ['events.delivery.admin']);
++const live = serviceToken('live', ['events.event.publish']);
++const streamPolicy = { class: 'domain', durability: 'required', delivery_semantics: 'at_least_once', ordering: { scope: 'key', key: 'subject' } };
++const silent = { log() {}, warn() {}, error() {} };
++
++const fakeValkey = {
++    client: { ovQueuePromote() {}, defineCommand() {}, on() {} },
++    key: (...p) => `k:${p.join(':')}`,
++    duplicate() { return { on() {}, subscribe() {}, unsubscribe: async () => {}, disconnect() {} }; },
++};
++
++async function waitFor(pred, ms = 8000) {
++    const until = Date.now() + ms;
++    for (;;) { if (await pred()) return; if (Date.now() > until) throw new Error('waitFor timed out'); await sleep(25); }
++}
++async function publish(base, env) {
++    const r = await request(base, 'POST', '/api/v1/events', { token: live, body: env });
++    assert.strictEqual(r.status, 201, r.text);
++    return r.body;
++}
++async function putPolicy(base, pattern, body) {
++    const r = await request(base, 'PUT', `/api/v1/delivery-policies/${pattern}`, { token: admin, body });
++    assert.strictEqual(r.status, 200, r.text);
++}
++async function subscribe(base, slug, topic, url) {
++    const r = await request(base, 'POST', '/api/v1/subscriptions', { token: serviceToken(slug, ['events.subscription.manage']), body: { topic_pattern: topic, endpoint: url } });
++    assert.strictEqual(r.status, 201, r.text);
++    return r.body;
++}
++async function explain(base, eventType) {
++    const r = await request(base, 'GET', `/api/v1/placement?event_type=${eventType}`, { token: admin });
++    assert.strictEqual(r.status, 200, r.text);
++    assert.ok(validate('platform.placement-result@1', r.body.placement).valid, JSON.stringify(r.body.placement));
++    return r.body;
++}
++const cand = (ex, id) => ex.placement.candidates.find((c) => c.id === id);
++const keyed = (k, over = {}) => envelope('live', { event_type: 'live.js.x', subject: { type: 'stream', id: k }, ...over });
++const sent = (stub, id) => stub.calls.filter((c) => c.body.event.event_id === id).length;
++const carrierOf = (db, eventId) => db.prepare('SELECT carrier, status FROM deliveries WHERE event_id = ?').get(eventId);
++
++// A TCP proxy in front of the broker: down() drops every connection and stops listening, up() listens again.
++function natsProxy(target) {
++    const u = new URL(target);
++    const socks = new Set();
++    let server = null;
++    let port = 0;
++    return {
++        get url() { return `nats://127.0.0.1:${port}`; },
++        up() {
++            server = net.createServer((c) => {
++                const b = net.connect(Number(u.port) || 4222, u.hostname);
++                socks.add(c); socks.add(b);
++                const end = () => { c.destroy(); b.destroy(); socks.delete(c); socks.delete(b); };
++                c.on('error', end); b.on('error', end); c.on('close', end); b.on('close', end);
++                c.pipe(b); b.pipe(c);
++            });
++            return new Promise((r) => server.listen(port, '127.0.0.1', () => { port = server.address().port; r(); }));
++        },
++        down() {
++            for (const s of socks) s.destroy();
++            socks.clear();
++            return new Promise((r) => server.close(() => r()));
++        },
++    };
++}
++
++// ── Registry, configuration gate, rate card, planner (no broker) ───────────
++
++t('registry: nats-js-v1 only with NATS_URL and JetStream on; STREAM only; the rate card is cheaper than pg-v1 and validates', async () => {
++    assert.strictEqual(createCarriers({ config: load({ NODE_ENV: 'test' }), log: silent }).get('nats-js-v1'), null, 'no NATS_URL: no nats-js-v1');
++    assert.strictEqual(createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', NATS_JETSTREAM: 'off' }), log: silent }).get('nats-js-v1'), null, 'NATS_JETSTREAM=off');
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', NATS_STREAM: 'a.b' }), log: silent }), /NATS_STREAM/);
++
++    const cfg = load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1' });
++    assert.deepStrictEqual([cfg.nats.jetstream, cfg.nats.stream], [true, 'OV_EVENTS_STREAM']);
++    assert.ok(cfg.carriers.enabled.includes('nats-js-v1'), 'on by default once NATS_URL is set');
++    const reg = createCarriers({ config: cfg, log: silent });
++    const js = reg.get('nats-js-v1');
++    assert.deepStrictEqual([...js.classes], ['STREAM']);
++    assert.ok(reg.forClass('QUEUE').every((a) => a.id !== 'nats-js-v1') && reg.forClass('TOPIC').every((a) => a.id !== 'nats-js-v1'));
++    const card = RATE_CARDS.find((c) => c.id === 'rc-nats-js-v1');
++    assert.ok(card && validate('platform.rate-card@1', card).valid, JSON.stringify(card));
++    assert.ok(card.unit_price_usd < RATE_CARDS.find((c) => c.id === 'rc-pg-v1').unit_price_usd, 'preferred over pg-v1 while healthy');
++    const offer = js.offer();
++    assert.ok(validate('platform.resource-offer@1', offer).valid, JSON.stringify(offer));
++    assert.deepStrictEqual(offer.capabilities, ['events:gateway', 'events:durable', 'events:ordered']);
++    // The ordering key becomes one subject token: fixed length, subject-safe, distinct per key.
++    assert.match(keyToken('stream:a.b *>'), /^[A-Za-z0-9_-]{22}$/);
++    assert.notStrictEqual(keyToken('a'), keyToken('b'));
++    assert.strictEqual(keyToken(null), '_');
++});
++
++t('planner: STREAM needs durable + ordered; prefers a healthy nats-js-v1; (f) disabled or unhealthy, pg-v1 with a reason', async () => {
++    const reg = createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1' }), valkey: fakeValkey, log: silent });
++    const planner = createPlanner({ registry: reg });
++    // Never started: not connected.
++    let ex = planner.explain('STREAM', 'stream:K', streamPolicy);
++    assert.strictEqual(ex.carrier, 'pg-v1');
++    assert.strictEqual(ex.result.candidates.find((c) => c.id === 'nats-js-v1').excluded_because, 'not connected (nats-js-v1)');
++
++    const js = reg.get('nats-js-v1');
++    js.healthy = () => true;   // as if connected with the stream ensured
++    ex = planner.explain('STREAM', 'stream:K', streamPolicy);
++    assert.strictEqual(ex.carrier, 'nats-js-v1', JSON.stringify(ex.result));
++    assert.ok(ex.result.candidates.some((c) => c.id === 'pg-v1' && c.eligible), 'pg-v1 stays a candidate underneath');
++    assert.strictEqual(planner.explain('QUEUE', 'k', streamPolicy).carrier, 'valkey-v1', 'QUEUE never goes to JetStream');
++    assert.notStrictEqual(planner.explain('TOPIC', 'k', streamPolicy).carrier, 'nats-js-v1', 'TOPIC never goes to JetStream');
++
++    // valkey-v1 and nats-v1 only signal: never selected for STREAM even when they are the only push carriers.
++    const reg2 = createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', NATS_JETSTREAM: 'off' }), valkey: fakeValkey, log: silent });
++    assert.deepStrictEqual(reg2.forClass('STREAM').map((a) => a.id), ['pg-v1']);
++
++    // (f) EVENTS_CARRIERS_DISABLED=nats-js-v1 (the rollback): excluded at once, STREAM on pg-v1.
++    const off = createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', EVENTS_CARRIERS_DISABLED: 'nats-js-v1' }), log: silent });
++    off.get('nats-js-v1').healthy = () => true;
++    const ex2 = createPlanner({ registry: off }).explain('STREAM', 'stream:K', streamPolicy);
++    assert.strictEqual(ex2.carrier, 'pg-v1');
++    assert.strictEqual(ex2.result.candidates.find((c) => c.id === 'nats-js-v1').excluded_because, 'disabled by configuration');
++
++    // signalPublish groups STREAM deliveries by their carrier.
++    const got = [];
++    js.signal = async (rows, kind) => { got.push([kind, rows.map((r) => r.event_id)]); };
++    await reg.signalPublish([], [
++        { event_id: 'e1', subscription_id: 's', carrier: 'nats-js-v1', carrier_class: 'STREAM', ordering_key: 'k' },
++        { event_id: 'e2', subscription_id: 's', carrier: 'pg-v1', carrier_class: 'STREAM', ordering_key: 'k' },
++        { event_id: 'e3', subscription_id: 's', carrier: 'nats-js-v1', carrier_class: 'STREAM', ordering_key: 'j' },
++    ]);
++    assert.deepStrictEqual(got, [['STREAM', ['e1', 'e3']]]);
++});
++
++// ── Against a NATS broker with JetStream (OV_TEST_NATS_URL) ────────────────
++
++t('nats-js-v1: explain, outage to pg-v1 and back, exactly once, per-key order across a crash, forged ids, disable', async () => {
++    if (!NATS) return;   // the skip line is printed once, below
++    const run = `${process.pid}_${Date.now()}`;
++    const prefix = `ov.events.test.js${run}.`;
++    const stream = `OV_EVENTS_TEST_${run}`;
++    const env = { NATS_URL: NATS, NATS_SUBJECT_PREFIX: prefix, NATS_STREAM: stream, EVENTS_CARRIERS: 'pg-v1,nats-js-v1' };
++    const shared = await testDb();
++    const clock = manualClock();
++    const proxy = natsProxy(NATS);
++    await proxy.up();
++    // a publishes (its worker is manual: no poll, no consumer); b and c are the consuming workers' stores.
++    const a = await boot({ db: shared.db, clock, env });
++    const b = await boot({ db: shared.db, clock, env: { EVENTS_CARRIERS: 'pg-v1' } });
++    const c = await boot({ db: shared.db, clock, env: { EVENTS_CARRIERS: 'pg-v1' } });
++    const p = await boot({ db: shared.db, clock, env: { ...env, NATS_URL: proxy.url } });
++    const d = await boot({ db: shared.db, clock, env: { ...env, EVENTS_CARRIERS_DISABLED: 'nats-js-v1' } });
++    const stub = await subscriber(() => 204);
++    const raw = createNatsCore({ url: NATS, log: silent, inboxPrefix: `${prefix}_inbox.raw.` });
++    const consumers = [];
++    // A consuming carrier for instance `h`: claims through h's store with h's worker lease and sends with h's worker.
++    const consumer = async (h, opts = {}, wrap = {}) => {
++        const claims = [];
++        const k = createJetStreamCarrier({ url: NATS, subjectPrefix: prefix, stream, log: silent, instanceId: `${opts.consumerName || ids.newId('event').slice(-8)}`, ...opts });
++        await k.start({
++            claim: async (e, s) => { const row = await h.store.claimDelivery(e, s, h.worker.lease()); claims.push([e, Boolean(row)]); return row; },
++            deliver: async (row) => { await h.worker.deliver(row); if (wrap.afterDeliver) await wrap.afterDeliver(row, k); },
++        });
++        consumers.push(k);
++        k.claims = claims;
++        return k;
++    };
++    try {
++        await raw.connect();
++        await waitFor(() => a.carriers.get('nats-js-v1').healthy() && p.carriers.get('nats-js-v1').healthy());
++        const sub = await subscribe(a.base, 'fabjs', 'live.js.*', stub.url);
++        await putPolicy(a.base, 'live.js.*', streamPolicy);
++        for (const i of [b, c, p, d]) await i.policies.invalidate();
++
++        // (a) STREAM, and the explain API selects nats-js-v1; the delivery is written on it.
++        const ex = await explain(a.base, 'live.js.x');
++        assert.strictEqual(ex.carrier_class, 'STREAM');
++        assert.strictEqual(ex.placement.selected, 'nats-js-v1', JSON.stringify(ex.placement));
++        assert.ok(cand(ex, 'pg-v1').eligible);
++
++        // (c) Two consuming workers, one due delivery: both get the message, exactly one claims and sends it.
++        const cb = await consumer(b);
++        const cc = await consumer(c);
++        const E1 = keyed('K1');
++        await publish(a.base, E1);
++        assert.strictEqual((await carrierOf(shared.db, E1.event_id)).carrier, 'nats-js-v1');
++        await waitFor(() => sent(stub, E1.event_id) === 1);
++        await waitFor(() => [cb, cc].every((k) => k.claims.some(([e]) => e === E1.event_id)));
++        assert.strictEqual([cb, cc].flatMap((k) => k.claims).filter(([e, won]) => e === E1.event_id && won).length, 1, 'one claim won');
++        await sleep(200);
++        assert.strictEqual(sent(stub, E1.event_id), 1, 'sent exactly once');
++        assert.strictEqual((await carrierOf(shared.db, E1.event_id)).status, 'delivered');
++
++        // (d) Per-key order across two keys, interleaved: each key's sends in publish order, each once. The poll runs
++        // underneath as in production: a message whose claim yields (not its key's head yet, or the subscription row
++        // locked by the other consumer's claim on PostgreSQL) is acked and the poll sends that row; the store keeps the order.
++        const order = { K1: [], K2: [] };
++        const poll = setInterval(() => { c.worker.dispatch().catch(() => {}); }, 500);
++        try {
++            for (let i = 0; i < 4; i++) for (const k of ['K1', 'K2']) { const e = keyed(k); order[k].push(e.event_id); await publish(a.base, e); }
++            await waitFor(() => [...order.K1, ...order.K2].every((id) => sent(stub, id) === 1));
++        } finally { clearInterval(poll); }
++        const all = [...order.K1, ...order.K2];
++        assert.ok([cb, cc].flatMap((k) => k.claims).some(([e, won]) => won && all.includes(e)), 'the carrier claimed and sent');
++        for (const k of ['K1', 'K2']) {
++            const seen = stub.calls.map((x) => x.body.event.event_id).filter((id) => order[k].includes(id));
++            assert.deepStrictEqual(seen, order[k], `${k}: sent in publish order`);
++        }
++
++        // (e) A forged message on the stream naming an uncommitted event (and one naming a committed event for a
++        // subscription that has no delivery of it) claims nothing; a real event behind them is still delivered.
++        const forged = ids.newId('event');
++        const before = stub.calls.length;
++        const subj = `${prefix}fabric.stream.${keyToken('stream:K1')}`;
++        assert.strictEqual((await raw.request(subj, { event_id: forged, subscription_id: sub.id })).stream, stream);
++        await raw.request(subj, { event_id: E1.event_id, subscription_id: 'sub_forged0000000000' });
++        await raw.request(subj, { nonsense: true });
++        const E2 = keyed('K1');
++        await publish(a.base, E2);
++        await waitFor(() => sent(stub, E2.event_id) === 1);
++        assert.strictEqual(stub.calls.length, before + 1, 'the forged messages sent nothing');
++        assert.ok(cb.claims.some(([e, won]) => e === forged && !won), 'the forged id was looked up and claimed nothing');
++        assert.strictEqual(await shared.db.prepare('SELECT 1 FROM deliveries WHERE event_id = ?').get(forged), undefined);
++        await cc.stop();
++
++        // (d, crash) A worker that sends and dies before its ack: the restarted worker (same durable) gets the message
++        // again after ack_wait, the claim finds it delivered and sends nothing; the key's next events follow in order.
++        await cb.stop();
++        let crashed = false;
++        const durable = `crash_${run}`;
++        await consumer(b, { consumerName: durable, ackWaitMs: 1000 }, {
++            afterDeliver: async (row, k) => { if (!crashed) { crashed = true; await k.client.close(); } },
++        });
++        const C1 = keyed('K3');
++        await publish(a.base, C1);
++        await waitFor(() => crashed && sent(stub, C1.event_id) === 1);
++        const C2 = keyed('K3');
++        const C3 = keyed('K3');
++        await publish(a.base, C2);
++        await publish(a.base, C3);
++        const again = await consumer(b, { consumerName: durable, ackWaitMs: 1000 });
++        await waitFor(() => sent(stub, C3.event_id) === 1, 10000);
++        assert.ok(again.claims.some(([e, won]) => e === C1.event_id && !won), 'the unacked message was redelivered and claimed nothing');
++        assert.strictEqual(sent(stub, C1.event_id), 1, 'the redelivery was deduped by the claim');
++        const k3 = stub.calls.map((x) => x.body.event.event_id).filter((id) => [C1, C2, C3].some((e) => e.event_id === id));
++        assert.deepStrictEqual(k3, [C1.event_id, C2.event_id, C3.event_id], 'K3 in order across the crash');
++        await again.stop();
++
++        // (b) Broker down for `p`: excluded with the reason, STREAM falls to pg-v1, nothing is lost; back up, preferred again.
++        const pj = p.carriers.get('nats-js-v1');
++        await proxy.down();
++        await waitFor(() => !pj.healthy());
++        const exDown = await explain(p.base, 'live.js.x');
++        assert.strictEqual(exDown.placement.selected, 'pg-v1');
++        assert.strictEqual(cand(exDown, 'nats-js-v1').excluded_because, 'not connected (nats-js-v1)');
++        const L1 = keyed('K4');
++        await publish(p.base, L1);
++        assert.strictEqual((await carrierOf(shared.db, L1.event_id)).carrier, 'pg-v1');
++        // A delivery signalled on nats-js-v1 that no worker consumed (no consumer runs now) is the poll's too.
++        const L2 = keyed('K5');
++        await publish(a.base, L2);
++        assert.strictEqual((await carrierOf(shared.db, L2.event_id)).carrier, 'nats-js-v1');
++        await p.worker.drain();
++        assert.strictEqual(sent(stub, L1.event_id) + sent(stub, L2.event_id), 2, 'the poll delivered both: nothing lost');
++        await proxy.up();
++        await waitFor(() => pj.healthy(), 10000);
++        assert.strictEqual((await explain(p.base, 'live.js.x')).placement.selected, 'nats-js-v1', 'preferred again once back');
++
++        // (f) Disabled: never connects, STREAM stays on pg-v1.
++        const dj = d.carriers.get('nats-js-v1');
++        assert.strictEqual(dj.client.connected(), false, 'a disabled nats-js-v1 never connects');
++        const exOff = await explain(d.base, 'live.js.x');
++        assert.strictEqual(exOff.placement.selected, 'pg-v1');
++        assert.strictEqual(cand(exOff, 'nats-js-v1').excluded_because, 'disabled by configuration');
++        const F = keyed('K6');
++        await publish(d.base, F);
++        assert.strictEqual((await carrierOf(shared.db, F.event_id)).carrier, 'pg-v1');
++    } finally {
++        for (const k of consumers) await k.stop().catch(() => {});
++        for (const i of [a, b, c, p, d]) await i.stop();
++        await raw.request(`$JS.API.STREAM.DELETE.${stream}`, {}).catch(() => {});
++        await raw.close();
++        await proxy.down().catch(() => {});
++        await stub.close();
++        await shared.close();
++    }
++});
++
++if (!NATS) console.log('fabric jetstream: skipped (OV_TEST_NATS_URL not set: eval "$(scripts/test-nats.sh up)")');
++
++t.run();