Implement the smallest T7 PR on current origin/main: an optional NATS… (#6)

ddiff --git a/.env.example b/.env.example
iindex ae4b632..f1719e1 100644
---- a/.env.example
++++ b/.env.example
@@@ -30,6 +30,12 @@ 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.
++#NATS_URL=
++#NATS_SUBJECT_PREFIX=ov.events.
++#EVENTS_CARRIERS=pg-v1,valkey-v1,nats-v1
++#EVENTS_CARRIERS_DISABLED=
  # Events (and their deliveries) older than this are pruned; resuming from before it yields a gap.
  EVENTS_RETENTION_DAYS=30
  # Accepted event ids are remembered this long, so late re-publishes stay duplicates.
ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex a720b4f..b51891f 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -11,7 +11,8 @@ jobs:
        contracts-service: events
        # ADR-035: PostgreSQL 18 + PgBouncer (transaction mode) + Valkey 9 containers (openvibe-sdk
        # scripts/test-services.sh). npm test runs every file on PGlite; test:pg runs them all through PgBouncer.
--      test: 'eval "$(node_modules/openvibe-sdk/scripts/test-services.sh up)" && npm test && npm run test:pg'
++      # A NATS Core container (scripts/test-nats.sh) backs the optional nats-v1 TOPIC carrier's tests.
++      test: 'eval "$(node_modules/openvibe-sdk/scripts/test-services.sh up)" && eval "$(bash scripts/test-nats.sh up)" && npm test && npm run test:pg'
    # Secret scan (gitleaks) and dependency audit, shared by every OpenVibers repo (Track Q).
    security:
      uses: OpenVibers/OpenVibe.Shared/.github/workflows/security.yml@423fe519b5fe9594cc81b670e5ba5cb9f21f6176
ddiff --git a/README.md b/README.md
iindex ee50498..84a1963 100644
---- a/README.md
++++ b/README.md
@@@ -118,11 +118,11 @@ 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`):
++Carriers (`server/fabric/carriers`; `EVENTS_CARRIERS`, default `pg-v1,valkey-v1,nats-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` (STREAM) is not built yet: STREAM is carried by `pg-v1`.
++- `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`.
  
  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 4baeffb..1b6352f 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), 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) 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",
      "at_most_once delivery semantics (a failed attempt is final)"
    ]
  }
ddiff --git a/scripts/nats-test.conf b/scripts/nats-test.conf
nnew file mode 100644
iindex 0000000..bd4b915
---- /dev/null
++++ b/scripts/nats-test.conf
@@@ -0,0 +1,11 @@
++# The test broker for test/fabric-nats.test.js (scripts/test-nats.sh). Not production. A connection without
++# credentials is the unrestricted user `open`; `nopub` may not publish and `nosub` may not subscribe under ov.events.
++# (the permissions-violation cases).
++authorization {
++  users = [
++    { user: open, password: open }
++    { user: nopub, password: nopub, permissions: { publish: { deny: ["ov.events.>"] } } }
++    { user: nosub, password: nosub, permissions: { subscribe: { deny: ["ov.events.>"] } } }
++  ]
++}
++no_auth_user: open
ddiff --git a/scripts/test-nats.sh b/scripts/test-nats.sh
nnew file mode 100755
iindex 0000000..87886c5
---- /dev/null
++++ b/scripts/test-nats.sh
@@@ -0,0 +1,26 @@
++#!/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.
++#
++#   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.
++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)
++case "${1:-up}" in
++up)
++  # Reused only while it runs this config 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
++  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
++  if [ -z "$ready" ]; then echo "$NAME not listening on 127.0.0.1:$PORT after 10s" >&2; exit 1; fi
++  echo "export OV_TEST_NATS_URL=nats://127.0.0.1:$PORT"
++  ;;
++down) docker rm -f $NAME >/dev/null 2>&1 || true; echo removed ;;
++*) echo "usage: $0 up|down" >&2; exit 2 ;;
++esac
ddiff --git a/server/config.js b/server/config.js
iindex c5c6a7d..2e5e54f 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -69,14 +69,18 @@ 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 is 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 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.
          carriers: {
--            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1']),
++            enabled: list(env.EVENTS_CARRIERS, ['pg-v1', 'valkey-v1', 'nats-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.' },
          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 5353bfb..ae4f4c9 100644
---- a/server/fabric/carriers/index.js
++++ b/server/fabric/carriers/index.js
@@@ -2,11 +2,12 @@
  /**
   * 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 not built yet (it needs the host
-- * unit), so STREAM is left to pg-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.
   *
   * Config (server/config.js):
-- *   EVENTS_CARRIERS           ids to instantiate (default pg-v1,valkey-v1). An unknown id refuses to start.
++ *   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_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
@@@ -14,8 +15,9 @@
   */
  const { createPgCarrier } = require('./pg');
  const { createValkeyCarrier, newInstanceId } = require('./valkey');
++const { createNatsCarrier } = require('./nats');
  
--const KNOWN = ['pg-v1', 'valkey-v1'];
++const KNOWN = ['pg-v1', 'valkey-v1', 'nats-v1'];
  
  function createCarriers({ config, clock = { now: () => Date.now() }, log = console, valkey = null } = {}) {
      const enabled = new Set(config.carriers.enabled);
@@@ -31,6 +33,10 @@ 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.
++    if (config.nats && config.nats.url) {
++        adapters.push(createNatsCarrier({ url: config.nats.url, subjectPrefix: config.nats.subjectPrefix, clock, log, instanceId: newInstanceId() }));
++    }
      for (const a of adapters) if (!a.disabledReason && off(a.id)) a.disabledReason = 'disabled by configuration';
  
      const get = (id) => adapters.find((a) => a.id === id) || null;
ddiff --git a/server/fabric/carriers/nats.js b/server/fabric/carriers/nats.js
nnew file mode 100644
iindex 0000000..3f0468a
---- /dev/null
++++ b/server/fabric/carriers/nats.js
@@@ -0,0 +1,118 @@
++'use strict';
++/**
++ * nats-v1 — a NATS Core TOPIC carrier (ADR-042 decision 5), only when NATS_URL is set. Optional: without it, or with
++ * EVENTS_CARRIERS_DISABLED=nats-v1, nothing connects and TOPIC stays on valkey-v1 or pg-v1.
++ *
++ * TOPIC: after a publish commits, signal(rows) PUBLISHes the event ids (never the envelopes) on the subject
++ * `<NATS_SUBJECT_PREFIX>fabric.topic`, tagged with this process's id. Every process subscribes and, for ids from
++ * OTHER processes, loads the committed rows from PostgreSQL (handlers.getEvent) and re-emits them to its own SSE
++ * clients in arrival order, deduped by event id with a small LRU (the gateway's per-connection seq check is the second
++ * guard). The broker is never trusted with content: an id that is not committed is ignored, so a message from anyone
++ * else allowed on the subject can neither inject an event nor advance a connection's seq. Visibility and topic
++ * matching are applied by the receiving gateway exactly as for local events. Core NATS is at most once and keeps
++ * nothing: a lost message reaches a remote browser on its next resume (Last-Event-ID replays from PostgreSQL, the
++ * record), a duplicate is dropped.
++ *
++ * Health: connected (the socket is up and the handshake done), no permissions violation from the broker (a refused
++ * PUB or SUB keeps the connection open but drops the message; the handshake probes the subject) and the breaker on
++ * 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.
++ */
++const { createNatsCore } = require('../nats-core');
++const { createSignals } = require('../signals');
++
++const TOPIC_SUBJECT = 'fabric.topic';
++const SEEN_MAX = 5000;
++const IDS_MAX = 1000;     // ids per message taken from the broker (a publish batch is smaller)
++
++function createNatsCarrier({ url, subjectPrefix = 'ov.events.', clock = { now: () => Date.now() }, log = console, instanceId, deniedRetryMs } = {}) {
++    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`);
++    const signals = createSignals({ now: () => clock.now() });
++    const subject = `${subjectPrefix}${TOPIC_SUBJECT}`;
++    const nats = createNatsCore({ url, name: `openvibe-events ${instanceId}`, log, probe: [subject], deniedRetryMs });
++    const seen = new Map();   // event id -> first-seen ms
++    let unsubscribe = null;
++    let remote = Promise.resolve();   // remote messages are handled one after another, in arrival order
++
++    function markSeen(id) {
++        seen.set(id, clock.now());
++        if (seen.size > SEEN_MAX) {
++            const cut = clock.now() - 300000;
++            for (const [k, t] of seen) { if (t < cut) seen.delete(k); if (seen.size <= SEEN_MAX) break; }
++            while (seen.size > SEEN_MAX) seen.delete(seen.keys().next().value);
++        }
++    }
++
++    async function signalTopic(rows) {
++        const t0 = clock.now();
++        try {
++            nats.publish(subject, { from: instanceId, ids: rows.map((r) => r.id) });
++            signals.record('topic', true, clock.now() - t0);
++        } catch (err) {
++            signals.record('topic', false);
++            log.warn(`[nats-v1] topic publish: ${err.message}`);
++        }
++    }
++
++    return {
++        id: 'nats-v1',
++        classes: new Set(['TOPIC']),
++        client: nats,
++        signals,
++
++        async start(handlers = {}) {
++            const fanOut = async (ids) => {
++                const rows = [];
++                for (const id of ids) {
++                    if (seen.has(id)) continue;
++                    const row = await handlers.getEvent(id);
++                    if (!row) continue;   // not committed (or pruned): nothing to re-emit
++                    markSeen(id);
++                    rows.push(row);
++                }
++                rows.sort((x, y) => x.seq - y.seq);
++                if (rows.length) handlers.onRemote(rows);
++            };
++            unsubscribe = nats.subscribe(subject, (msg) => {
++                if (!msg || msg.from === instanceId || !Array.isArray(msg.ids) || !handlers.onRemote || !handlers.getEvent) return;
++                const ids = msg.ids.slice(0, IDS_MAX).filter((id) => typeof id === 'string' && id.length <= 64);
++                if (!ids.length) return;
++                remote = remote.then(() => fanOut(ids)).catch((err) => log.warn(`[nats-v1] remote fan-out: ${err.message}`));
++            });
++            // Not fatal: the client keeps reconnecting, and nats-v1 is ineligible (not connected) until it is up.
++            await nats.connect().catch((err) => log.warn(`[nats-v1] not connected yet: ${err.message}`));
++        },
++
++        async stop() {
++            if (unsubscribe) { await unsubscribe().catch(() => {}); unsubscribe = null; }
++            await nats.close();
++            await remote;
++        },
++
++        signal(rows, kind) { return kind === 'TOPIC' ? signalTopic(rows) : undefined; },
++        healthy: () => nats.connected() && !nats.denied() && signals.healthy('topic'),
++        unhealthyReason: () => (!nats.connected() ? 'not connected (nats-v1)' : nats.denied() ? 'permission denied (nats-v1)'
++            : !signals.healthy('topic') ? 'breaker open (nats-v1)' : null),
++
++        offer() {
++            const topicMs = signals.ewma('topic') || 1;
++            return {
++                offer_id: 'nats-v1',
++                kind: 'provider',
++                provider: 'nats',
++                adapter: 'nats-v1',
++                region: 'cell',
++                trust: 'first-party',
++                capabilities: ['events:gateway'],
++                latency_ms: { publish_p95: topicMs, ack_p95: topicMs },
++                health: { status: 'up', checked_at: new Date(clock.now()).toISOString() },
++                pricing: { model: 'per-operation', unit: 'signal-operation', rate_card: 'rc-nats-v1' },
++                updated_at: new Date(clock.now()).toISOString(),
++            };
++        },
++    };
++}
++
++module.exports = { createNatsCarrier, TOPIC_SUBJECT };
ddiff --git a/server/fabric/nats-core.js b/server/fabric/nats-core.js
nnew file mode 100644
iindex 0000000..1a68f39
---- /dev/null
++++ b/server/fabric/nats-core.js
@@@ -0,0 +1,196 @@
++'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).
++ *
++ *   createNatsCore({ url, name, log, reconnectMs, maxReconnectMs, connectTimeoutMs, probe, deniedRetryMs })
++ *     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.
++ *     connected()             true while the handshake is done and the socket open
++ *     denied()                the broker's last permissions violation (a refused PUB or SUB, after which NATS keeps
++ *                             the connection open and drops the message), else null. The handshake SUBs and PUBs `{}`
++ *                             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
++ *     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 CRLF = '\r\n';
++const MAX_LINE = 64 * 1024;
++const MAX_BUFFERED = 8 * 1024 * 1024;   // unsent bytes before publish fails (and the breaker counts it)
++
++function parseUrl(raw) {
++    let u;
++    try { u = new URL(raw); } catch { throw new Error('NATS_URL is not a URL (nats://[user:pass@]host:port)'); }
++    if (u.protocol !== 'nats:') throw new Error('NATS_URL must use nats:// (TLS is not supported by nats-v1)');
++    const user = decodeURIComponent(u.username || '');
++    const pass = decodeURIComponent(u.password || '');
++    return {
++        host: u.hostname.replace(/^\[|\]$/g, ''),
++        port: Number(u.port) || 4222,
++        auth: user && pass ? { user, pass } : user ? { auth_token: user } : {},
++    };
++}
++
++function createNatsCore({ url, name = 'openvibe-events', log = console, reconnectMs = 250, maxReconnectMs = 5000, connectTimeoutMs = 2000, probe = [], deniedRetryMs = 30000 } = {}) {
++    const target = parseUrl(url);
++    const where = `${target.host}:${target.port}`;
++    const subs = new Map();   // sid -> { subject, fn }
++    let nextSid = 1;
++    let sock = null;
++    let ready = false;
++    let stopped = false;
++    let timer = null;
++    let delay = reconnectMs;
++    let maxPayload = 1024 * 1024;
++    let first = null;         // { resolve, reject } of the pending connect()
++    let denied = null;        // the last permissions violation on the current connection
++    let deniedTimer = null;
++
++    function settle(err) {
++        if (!first) return;
++        const f = first; first = null;
++        if (err) f.reject(err); else f.resolve();
++    }
++
++    function scheduleReconnect() {
++        if (stopped || timer) return;
++        timer = setTimeout(() => { timer = null; open(); }, delay);
++        timer.unref?.();
++        delay = Math.min(delay * 2, maxReconnectMs);
++    }
++
++    // A permissions change needs a new connection to be seen (and the handshake probe to pass): re-open later.
++    function retryDenied(s) {
++        if (deniedTimer) return;
++        deniedTimer = setTimeout(() => { deniedTimer = null; if (sock === s) s.destroy(); }, deniedRetryMs);
++        deniedTimer.unref?.();
++    }
++
++    function open() {
++        if (stopped) return;
++        let buf = Buffer.alloc(0);
++        let handshaken = false;
++        let violation = null;     // a permissions violation seen during this connection's handshake
++        const s = net.connect({ host: target.host, port: target.port });
++        sock = s;
++        s.setNoDelay(true);
++        const deadline = setTimeout(() => { if (!handshaken) s.destroy(new Error(`nats ${where}: connect timed out`)); }, connectTimeoutMs);
++        deadline.unref?.();
++
++        s.on('data', (chunk) => {
++            buf = buf.length ? Buffer.concat([buf, chunk]) : chunk;
++            for (;;) {
++                const eol = buf.indexOf(CRLF);
++                if (eol < 0) { if (buf.length > MAX_LINE) s.destroy(new Error(`nats ${where}: protocol line too long`)); return; }
++                const line = buf.subarray(0, eol).toString('utf8');
++                const op = line.split(' ', 1)[0].toUpperCase();
++                if (op === 'MSG') {
++                    // MSG <subject> <sid> [reply-to] <#bytes>
++                    const parts = line.split(' ');
++                    const bytes = Number(parts[parts.length - 1]);
++                    if (buf.length < eol + 2 + bytes + 2) return;   // wait for the payload
++                    const payload = buf.subarray(eol + 2, eol + 2 + bytes);
++                    buf = buf.subarray(eol + 2 + bytes + 2);
++                    const sub = subs.get(Number(parts[2]));
++                    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}`); }
++                    }
++                    continue;
++                }
++                buf = buf.subarray(eol + 2);
++                if (op === 'PING') s.write(`PONG${CRLF}`);
++                else if (op === 'INFO') {
++                    let info = {};
++                    try { info = JSON.parse(line.slice(5)); } catch { /* keep the defaults */ }
++                    if (info.max_payload) maxPayload = info.max_payload;
++                    if (info.tls_required) { s.destroy(new Error(`nats ${where}: the server requires TLS (not supported by nats-v1)`)); return; }
++                    if (!handshaken) {
++                        const connect = { verbose: false, pedantic: false, lang: 'node', version: '0', name, protocol: 1, headers: false, no_echo: true, ...target.auth };
++                        let out = `CONNECT ${JSON.stringify(connect)}${CRLF}`;
++                        for (const [sid, sub] of subs) out += `SUB ${sub.subject} ${sid}${CRLF}`;
++                        for (const subject of probe) out += `PUB ${subject} 2${CRLF}{}${CRLF}`;
++                        s.write(`${out}PING${CRLF}`);
++                    }
++                } else if (op === 'PONG') {
++                    if (!handshaken) {
++                        handshaken = true; ready = true; delay = reconnectMs;
++                        clearTimeout(deadline);
++                        denied = violation;
++                        if (denied) retryDenied(s);
++                        settle(null);
++                    }
++                } else if (op === '-ERR') {
++                    const why = line.slice(5).trim();
++                    log.warn(`[nats] ${where}: ${why}`);
++                    // A refused PUB or SUB leaves the connection open (the message is dropped): record it, so the
++                    // carrier is unhealthy and placement falls back. Any other -ERR before the handshake is fatal.
++                    if (/permissions violation/i.test(why)) {
++                        if (!handshaken) violation = why;
++                        else if (sock === s) { denied = why; retryDenied(s); }
++                    } else if (!handshaken) s.destroy(new Error(`nats ${where}: ${why}`));
++                }
++                // +OK and anything else: nothing to do
++            }
++        });
++        s.on('error', (err) => { if (!handshaken) settle(err); else log.warn(`[nats] ${where}: ${err.message}`); });
++        s.on('close', () => {
++            clearTimeout(deadline);
++            if (sock === s) { sock = null; ready = false; denied = null; if (deniedTimer) { clearTimeout(deniedTimer); deniedTimer = null; } }
++            if (!handshaken) settle(new Error(`nats ${where}: connection closed before the handshake`));
++            scheduleReconnect();
++        });
++    }
++
++    return {
++        connect() {
++            if (ready) return Promise.resolve();
++            stopped = false;
++            return new Promise((resolve, reject) => {
++                first = { resolve, reject };
++                if (!sock && !timer) open();
++            });
++        },
++        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}`);
++            };
++        },
++        async close() {
++            stopped = true;
++            if (timer) { clearTimeout(timer); timer = null; }
++            if (deniedTimer) { clearTimeout(deniedTimer); deniedTimer = null; }
++            ready = false;
++            settle(new Error('nats: closed'));
++            const s = sock;
++            sock = null;
++            if (s && !s.destroyed) await new Promise((r) => { s.once('close', r); s.end(); setTimeout(() => s.destroy(), 200).unref?.(); });
++        },
++        get maxPayload() { return maxPayload; },
++    };
++}
++
++module.exports = { createNatsCore, parseUrl };
ddiff --git a/server/fabric/planner.js b/server/fabric/planner.js
iindex 51a89c8..1cd5564 100644
---- a/server/fabric/planner.js
++++ b/server/fabric/planner.js
@@@ -65,10 +65,10 @@ function createPlanner({ registry, db = null, clock = { now: () => Date.now() },
              offers.push(o);
          }
          const r = plan(requirements(klass, key, policy), offers, { rateCards: RATE_CARDS, now, current: current ? current.carrier : null });
--        // The offer's health already excludes an unhealthy adapter; name the breaker in the reason.
++        // The offer's health already excludes an unhealthy adapter; name the breaker (or the lost connection) in the reason.
          for (const c of r.candidates) {
              const a = registry.get(c.id);
--            if (a && !a.disabledReason && !a.healthy() && !c.eligible) c.excluded_because = `breaker open (${a.id})`;
++            if (a && !a.disabledReason && !a.healthy() && !c.eligible) c.excluded_because = (a.unhealthyReason && a.unhealthyReason()) || `breaker open (${a.id})`;
          }
          for (const e of excluded) if (!r.candidates.some((c) => c.id === e.id)) r.candidates.push(e);
          if (!r.selected) {
ddiff --git a/server/fabric/rate-cards.json b/server/fabric/rate-cards.json
iindex 2c17a1f..15515a0 100644
---- a/server/fabric/rate-cards.json
++++ b/server/fabric/rate-cards.json
@@@ -26,5 +26,19 @@
      "effective_from": "2026-09-01",
      "source": "https://openvibe.network/pricing/events",
      "verified_at": "2026-09-01"
++  },
++  {
++    "id": "rc-nats-v1",
++    "provider": "nats",
++    "adapter": "nats-v1",
++    "metric": "signal-operation",
++    "unit_size": 1,
++    "unit_price_usd": 0.0000001,
++    "free_allowance": 1000000000,
++    "reset_period": "month",
++    "region": "cell",
++    "effective_from": "2026-10-02",
++    "source": "https://openvibe.network/pricing/events",
++    "verified_at": "2026-10-02"
    }
  ]
ddiff --git a/server/index.js b/server/index.js
iindex fc27413..af30f2a 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -59,6 +59,7 @@ async function start({
              claim: (eventId, subscriptionId) => store.claimDelivery(eventId, subscriptionId, worker.lease()),
              deliver: (delivery) => worker.deliver(delivery),
              onRemote: (rows) => realtime.publish(rows),
++            getEvent: (id) => store.getEvent(id),   // nats-v1 re-emits only rows committed in PostgreSQL
          });
      } catch (err) {
          log.warn(`[events] carriers did not start: ${err.message}`);
ddiff --git a/test/fabric-nats.test.js b/test/fabric-nats.test.js
nnew file mode 100644
iindex 0000000..43995dd
---- /dev/null
++++ b/test/fabric-nats.test.js
@@@ -0,0 +1,342 @@
++'use strict';
++// nats-v1 (ADR-042 decision 5, optional): a NATS Core TOPIC carrier for cross-process SSE fan-out. The registry builds
++// it only with NATS_URL, refuses a bad URL or subject prefix, and honours the disable switch; the planner prefers it for
++// TOPIC while it is connected and explains why it is excluded when it is not. Against a real broker: two instances
++// see one remote public event once, visibility holds on the remote gateway, a duplicate signal is dropped, a lost one
++// is replayed from PostgreSQL on resume, a message naming an uncommitted event (or carrying a forged row) is ignored, a
++// broker outage falls back to pg-v1 and recovers, a disabled nats-v1 never connects, and a broker that refuses its
++// PUB or SUB (a permissions violation) drops it to pg-v1 with a reason. The broker parts need OV_TEST_NATS_URL (scripts/test-nats.sh up); 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, userToken, envelope, sse, 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 { createNatsCarrier } = require('../server/fabric/carriers/nats');
++const { createNatsCore } = require('../server/fabric/nats-core');
++
++const t = suite('fabric-nats');
++const NATS = process.env.OV_TEST_NATS_URL || '';
++const admin = serviceToken('ops', ['events.delivery.admin']);
++const live = serviceToken('live', ['events.event.publish']);
++const reader = serviceToken('media', ['events.event.read']);
++const topicPolicy = { class: 'realtime_ephemeral', durability: 'none', delivery_semantics: 'at_least_once' };
++const silent = { log() {}, warn() {}, error() {} };
++
++// A valkey-shaped stub (as in fabric.test.js): the registry constructs valkey-v1 from it, nothing connects.
++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 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 count = (client, id) => client.events().filter((m) => m.event.event_id === id).length;
++// envelope() defaults to internal visibility: a public event says so.
++const pub = (over = {}) => envelope('live', { event_type: 'live.nt.x', visibility: 'public', ...over });
++const has = (id) => (c) => c.events().some((m) => m.event.event_id === id);
++
++// A TCP proxy in front of the broker: down() drops every connection and stops listening (a broker outage as one
++// instance sees it), up() listens again on the same port.
++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-v1 only with NATS_URL; a bad URL or prefix refuses to start; the rate card and offer validate', async () => {
++    const without = createCarriers({ config: load({ NODE_ENV: 'test' }), log: silent });
++    assert.strictEqual(without.get('nats-v1'), null, 'no NATS_URL: no nats-v1');
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1,kafka-v1' }) }), /unknown carrier adapter "kafka-v1"/);
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'http://127.0.0.1:4222' }), log: silent }), /nats:\/\//);
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', NATS_SUBJECT_PREFIX: 'a b.' }), log: silent }), /NATS_SUBJECT_PREFIX/);
++
++    const reg = createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1' }), log: silent });
++    const n = reg.get('nats-v1');
++    assert.deepStrictEqual([...n.classes], ['TOPIC'], 'TOPIC only: JetStream (STREAM) is not built');
++    assert.ok(!n.disabledReason);
++    assert.ok(reg.forClass('QUEUE').every((a) => a.id !== 'nats-v1') && reg.forClass('STREAM').every((a) => a.id !== 'nats-v1'));
++    const card = RATE_CARDS.find((c) => c.id === 'rc-nats-v1');
++    assert.ok(card && validate('platform.rate-card@1', card).valid, JSON.stringify(card));
++    const offer = n.offer();
++    assert.ok(validate('platform.resource-offer@1', offer).valid, JSON.stringify(offer));
++    assert.strictEqual(offer.pricing.rate_card, 'rc-nats-v1');
++});
++
++t('planner: TOPIC prefers a connected nats-v1; not connected or disabled, it is excluded 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, so valkey-v1 carries TOPIC and the reason names the connection.
++    let ex = planner.explain('TOPIC', 'live.x', topicPolicy);
++    assert.strictEqual(ex.carrier, 'valkey-v1');
++    assert.strictEqual(ex.result.candidates.find((c) => c.id === 'nats-v1').excluded_because, 'not connected (nats-v1)');
++
++    reg.get('nats-v1').client.connected = () => true;   // as if the handshake were done
++    ex = planner.explain('TOPIC', 'live.x', topicPolicy);
++    assert.strictEqual(ex.carrier, 'nats-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', topicPolicy).carrier, 'valkey-v1', 'QUEUE never goes to NATS Core');
++    assert.strictEqual(planner.explain('STREAM', 'k', topicPolicy).carrier, 'pg-v1', 'STREAM stays on pg-v1');
++
++    // The breaker on consecutive publish errors also drops it.
++    const n = reg.get('nats-v1');
++    for (let i = 0; i < 5; i++) n.signals.record('topic', false);
++    assert.strictEqual(planner.explain('TOPIC', 'live.x', topicPolicy).result.candidates.find((c) => c.id === 'nats-v1').excluded_because, 'breaker open (nats-v1)');
++
++    // EVENTS_CARRIERS_DISABLED=nats-v1 (the rollback): excluded at once; with no valkey, pg-v1 carries TOPIC.
++    const off = createCarriers({ config: load({ NODE_ENV: 'test', NATS_URL: 'nats://127.0.0.1:1', EVENTS_CARRIERS_DISABLED: 'nats-v1' }), log: silent });
++    off.get('nats-v1').client.connected = () => true;
++    const ex2 = createPlanner({ registry: off }).explain('TOPIC', 'live.x', topicPolicy);
++    assert.strictEqual(ex2.carrier, 'pg-v1');
++    assert.strictEqual(ex2.result.candidates.find((c) => c.id === 'nats-v1').excluded_because, 'disabled by configuration');
++});
++
++// A NATS-shaped server speaking just enough of the protocol: it refuses every PUB while `denyPub` is set (the
++// permissions violation NATS answers on an open connection) and records the payloads it accepted.
++async function fakeNats() {
++    const state = { denyPub: false, accepted: [], conns: 0 };
++    const server = net.createServer((c) => {
++        state.conns++;
++        c.write(`INFO {"server_id":"fake","max_payload":1048576}\r\n`);
++        let buf = '';
++        let pub = null;   // the subject of a PUB whose payload line comes next
++        c.on('error', () => {});
++        c.on('data', (d) => {
++            buf += d.toString('utf8');
++            let eol;
++            while ((eol = buf.indexOf('\r\n')) >= 0) {
++                const line = buf.slice(0, eol);
++                buf = buf.slice(eol + 2);
++                if (pub !== null) {
++                    if (state.denyPub) c.write(`-ERR 'Permissions Violation for Publish to "${pub}"'\r\n`);
++                    else state.accepted.push(line);
++                    pub = null;
++                } else if (line.startsWith('PUB ')) pub = line.split(' ')[1];
++                else if (line === 'PING') c.write('PONG\r\n');
++            }
++        });
++    });
++    await new Promise((r) => server.listen(0, '127.0.0.1', r));
++    return { state, url: `nats://127.0.0.1:${server.address().port}`, close: () => new Promise((r) => { server.close(() => r()); server.unref(); }) };
++}
++
++t('nats-v1: a PUB refused after the handshake marks it unhealthy, TOPIC falls back to pg-v1 until the permission returns', async () => {
++    const fake = await fakeNats();
++    const carrier = createNatsCarrier({ url: fake.url, log: silent, instanceId: 'test', deniedRetryMs: 1000 });
++    try {
++        await carrier.start({});
++        assert.ok(carrier.healthy(), 'the handshake probe was accepted');
++        assert.ok(fake.state.accepted.includes('{}'), 'the handshake probes the subject');
++        const reg = createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1' }), log: silent });
++        reg.adapters.push(carrier);   // as if registered: the planner sees this nats-v1
++        const planner = createPlanner({ registry: reg });
++        assert.strictEqual(planner.explain('TOPIC', 'live.x', topicPolicy).carrier, 'nats-v1');
++
++        fake.state.denyPub = true;
++        await carrier.signal([{ id: 'evt_lost' }], 'TOPIC');
++        await waitFor(() => !carrier.healthy(), 2000);
++        assert.strictEqual(carrier.client.connected(), true, 'NATS keeps the connection open after a permissions violation');
++        assert.strictEqual(carrier.unhealthyReason(), 'permission denied (nats-v1)');
++        const ex = planner.explain('TOPIC', 'live.x', topicPolicy);
++        assert.strictEqual(ex.carrier, 'pg-v1', JSON.stringify(ex.result));
++        assert.strictEqual(ex.result.candidates.find((c) => c.id === 'nats-v1').excluded_because, 'permission denied (nats-v1)');
++
++        // Re-opened while still refused: the handshake probe keeps it denied (no real signal is spent finding out).
++        await waitFor(() => fake.state.conns >= 2, 5000);
++        await waitFor(() => carrier.client.connected(), 5000);
++        assert.strictEqual(carrier.unhealthyReason(), 'permission denied (nats-v1)');
++        fake.state.denyPub = false;
++        await waitFor(() => carrier.healthy(), 3000);
++        assert.strictEqual(planner.explain('TOPIC', 'live.x', topicPolicy).carrier, 'nats-v1', 'chosen again once permitted');
++    } finally {
++        await carrier.stop();
++        await fake.close();
++    }
++});
++
++// ── Against a NATS broker (OV_TEST_NATS_URL) ────────────────────────────────
++
++t('nats-v1: two instances, remote fan-out once, visibility, duplicate and lost signals, outage and disable fall back to pg-v1', async () => {
++    if (!NATS) return;   // the skip line is printed once, below
++    const shared = await testDb();
++    const clock = manualClock();
++    const env = { NATS_URL: NATS, NATS_SUBJECT_PREFIX: `ov.events.test.${process.pid}.${Date.now()}.`, EVENTS_CARRIERS: 'pg-v1,nats-v1' };
++    const proxy = natsProxy(NATS);
++    await proxy.up();
++    const a = await boot({ db: shared.db, clock, env, worker: 'off' });
++    const b = await boot({ db: shared.db, clock, env, worker: 'off' });
++    const c = await boot({ db: shared.db, clock, env: { ...env, NATS_URL: proxy.url }, worker: 'off' });
++    const d = await boot({ db: shared.db, clock, env: { ...env, EVENTS_CARRIERS_DISABLED: 'nats-v1' }, worker: 'off' });
++    const clients = [];
++    const open = async (inst, headers = {}) => { const x = await sse(inst.base, '/realtime/stream?topics=live.nt.*', { headers }); clients.push(x); return x; };
++    try {
++        await waitFor(() => [a, b, c].every((i) => i.carriers.get('nats-v1').healthy()));
++        const put = await request(a.base, 'PUT', '/api/v1/delivery-policies/live.nt.*', { token: admin, body: topicPolicy });
++        assert.strictEqual(put.status, 200, put.text);
++        for (const i of [b, c, d]) await i.policies.invalidate();
++
++        const U1 = ids.newId('user');
++        const anon = await open(b);
++        const u1 = await open(b, { Authorization: `Bearer ${userToken({ subjectId: U1 })}` });
++        const u2 = await open(b, { Authorization: `Bearer ${userToken({ subjectId: ids.newId('user') })}` });
++        const svc = await open(b, { Authorization: `Bearer ${reader}` });
++        // A sentinel published on `a` over nats-v1: once `anon` on `b` has it, everything before it has arrived too.
++        const sentinel = async () => { const s = pub({ event_type: 'live.nt.sentinel' }); await publish(a.base, s); await anon.waitFor(has(s.event_id), 5000); return s; };
++
++        // Explain on `a`: nats-v1 is selected for TOPIC, pg-v1 eligible underneath.
++        const ex = await explain(a.base, 'live.nt.x');
++        assert.strictEqual(ex.carrier_class, 'TOPIC');
++        assert.strictEqual(ex.placement.selected, 'nats-v1', JSON.stringify(ex.placement));
++        assert.ok(cand(ex, 'pg-v1').eligible);
++
++        // One public event published on `a` reaches the browser on `b` once, and is already committed when it does.
++        let capturedRows = null;
++        const origPublish = a.realtime.publish;
++        a.realtime.publish = (rows) => { capturedRows = rows; return origPublish(rows); };
++        const P = pub();
++        await publish(a.base, P);
++        await anon.waitFor(has(P.event_id), 5000);
++        assert.strictEqual((await request(b.base, 'GET', `/api/v1/events/${P.event_id}`, { token: reader })).status, 200, 'signalled after the commit');
++        a.realtime.publish = origPublish;
++
++        // Visibility on the remote gateway: a subject event only to its person, an internal one only to a service.
++        const S = envelope('live', { event_type: 'live.nt.x', visibility: 'subject', subject: { type: 'user', id: U1 }, actor: { type: 'user', id: U1 } });
++        const I = envelope('live', { event_type: 'live.nt.x', visibility: 'internal' });
++        await publish(a.base, S);
++        await publish(a.base, I);
++        await sentinel();
++        await u1.waitFor(has(S.event_id), 2000);
++        await svc.waitFor(has(I.event_id), 2000);
++        assert.strictEqual(count(u2, S.event_id), 0, 'another person never sees a subject event');
++        assert.strictEqual(count(anon, S.event_id) + count(anon, I.event_id) + count(u1, I.event_id), 0, 'internal never reaches a browser');
++
++        // Duplicate signals (the same rows re-published twice): still exactly once on `b`.
++        await a.carriers.signal('nats-v1', capturedRows, 'TOPIC');
++        await a.carriers.signal('nats-v1', capturedRows, 'TOPIC');
++        await sentinel();
++        assert.strictEqual(count(anon, P.event_id), 1, 'a duplicate signal is dropped');
++
++        // A lost signal: the live stream on `b` misses it, a resume (Last-Event-ID) replays it from PostgreSQL.
++        const before = anon.messages[anon.messages.length - 1].id;
++        const client = a.carriers.get('nats-v1').client;
++        const realPub = client.publish;
++        client.publish = () => { client.publish = realPub; };   // the next message vanishes
++        const L = pub();
++        await publish(a.base, L);
++        await sentinel();
++        assert.strictEqual(count(anon, L.event_id), 0, 'the lost signal never arrived live');
++        const resumed = await open(b, { 'Last-Event-ID': before });
++        await resumed.waitFor(has(L.event_id), 3000);
++
++        // Anyone else allowed on the subject: an id that was never committed is ignored, and a forged row (the old
++        // shape, with a seq far ahead) is never streamed nor advances the connection, so the sentinel still arrives.
++        const forger = createNatsCore({ url: NATS, log: silent });
++        await forger.connect();
++        const subject = `${env.NATS_SUBJECT_PREFIX}fabric.topic`;
++        const F = pub();
++        forger.publish(subject, { from: 'forged', ids: [F.event_id] });
++        forger.publish(subject, { from: 'forged', rows: [{ id: F.event_id, seq: 1e12, event_type: 'live.nt.x', visibility: 'public', envelope: JSON.stringify(F) }] });
++        await forger.close();
++        await sentinel();
++        assert.strictEqual(count(anon, F.event_id), 0, 'an uncommitted event is never streamed');
++
++        // Broker outage as `c` sees it: nats-v1 drops out with a reason, pg-v1 is selected, the publish still commits
++        // and reaches `c`'s own browsers; `b` does not get it live. Back up, `c` reconnects and nats-v1 is chosen again.
++        const local = await open(c);
++        await proxy.down();
++        await waitFor(() => !c.carriers.get('nats-v1').healthy(), 3000);
++        const exDown = await explain(c.base, 'live.nt.x');
++        assert.strictEqual(exDown.placement.selected, 'pg-v1', JSON.stringify(exDown.placement));
++        assert.strictEqual(cand(exDown, 'nats-v1').excluded_because, 'not connected (nats-v1)');
++        const O = pub();
++        await publish(c.base, O);
++        await local.waitFor(has(O.event_id), 2000);
++        await sentinel();
++        assert.strictEqual(count(anon, O.event_id), 0, 'nothing crossed the broker while it was down');
++        await proxy.up();
++        await waitFor(() => c.carriers.get('nats-v1').healthy(), 8000);
++        assert.strictEqual((await explain(c.base, 'live.nt.x')).placement.selected, 'nats-v1', 'chosen again once reconnected');
++        const R = pub();
++        await publish(c.base, R);
++        await anon.waitFor(has(R.event_id), 5000);
++
++        // Disabled (the rollback): `d` never connects, explains why, and its TOPIC events stay on pg-v1.
++        assert.strictEqual(d.carriers.get('nats-v1').client.connected(), false, 'a disabled nats-v1 never connects');
++        const exOff = await explain(d.base, 'live.nt.x');
++        assert.strictEqual(exOff.placement.selected, 'pg-v1');
++        assert.strictEqual(cand(exOff, 'nats-v1').excluded_because, 'disabled by configuration');
++        const X = pub();
++        await publish(d.base, X);
++        await sentinel();
++        assert.strictEqual(count(anon, X.event_id), 0, 'a disabled carrier is never signalled');
++
++        // Permissions: a broker user that may not publish (or subscribe) on the subject keeps its connection, but
++        // nats-v1 is excluded with a reason and TOPIC goes to pg-v1; the publish still commits and reaches local clients.
++        const withUser = (u) => NATS.replace('nats://', `nats://${u}:${u}@`);
++        for (const user of ['nopub', 'nosub']) {
++            const e = await boot({ db: shared.db, clock, env: { ...env, NATS_URL: withUser(user) }, worker: 'off' });
++            try {
++                const n = e.carriers.get('nats-v1');
++                await waitFor(() => n.client.connected() && !n.healthy(), 3000);
++                await e.policies.invalidate();
++                const exDenied = await explain(e.base, 'live.nt.x');
++                assert.strictEqual(exDenied.placement.selected, 'pg-v1', `${user}: ${JSON.stringify(exDenied.placement)}`);
++                assert.strictEqual(cand(exDenied, 'nats-v1').excluded_because, 'permission denied (nats-v1)', user);
++                const mine = await open(e);
++                const D = pub();
++                await publish(e.base, D);
++                await mine.waitFor(has(D.event_id), 2000);
++            } finally {
++                await e.stop();
++            }
++        }
++    } finally {
++        for (const x of clients) x.close();
++        for (const i of [a, b, c, d]) await i.stop();
++        await proxy.down().catch(() => {});
++        await shared.close();
++    }
++});
++
++if (!NATS) console.log('fabric nats: skipped (OV_TEST_NATS_URL not set: eval "$(scripts/test-nats.sh up)")');
++
++t.run();
ddiff --git a/test/fabric.test.js b/test/fabric.test.js
iindex ddb5ae3..499fc1e 100644
---- a/test/fabric.test.js
++++ b/test/fabric.test.js
@@@ -77,7 +77,7 @@ t('registry: every class -> one carrier class; unknown id refused; disabled valk
          assert.strictEqual(carrierClass(keyed('subject', c)), 'STREAM');
      }
      // An unknown adapter id in config refuses to start.
--    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1,nats-v1' }) }), /unknown carrier adapter "nats-v1"/);
++    assert.throws(() => createCarriers({ config: load({ NODE_ENV: 'test', EVENTS_CARRIERS: 'pg-v1,kafka-v1' }) }), /unknown carrier adapter "kafka-v1"/);
  
      // The rate cards are valid platform.rate-card@1 (the model is exercised with a small per-op figure).
      for (const card of RATE_CARDS) {
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 5513c7d..b0510a8 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -220,6 +220,7 @@ const REVIEWED = {
      'server/worker.js appPost': [1, 'developer-app deliveries: the guarded poster (server/egress.js), covered above'],
      'server/egress.js http.request': [1, 'the guarded poster\'s https.request (default requestImpl): its own lookup pins the checked address'],
      'server/egress.js requestImpl': [1, 'the guarded poster\'s request, after parseAppEndpoint and with the guarded lookup'],
++    'server/fabric/nats-core.js net.connect': [1, 'the nats-v1 broker: NATS_URL (operator config, nats:// on loopback or the private network); never a request input'],
  };
  /** JS source with comments blanked (strings, template literals and regex literals kept). */
  function stripComments(src) {