chore(distributed): stop telling an operator to run a NATS cluster

Every carrier had already moved and no process opened a bus connection, but
the surface an operator reads still described a deployment with a broker in
it: a compose service, a 220-line credential-generation script, two CI steps
pulling a container nothing started, two flag tables offering --nats-url, an
architecture diagram with a NATS box wired to the workers, a join-command
generator in the Nodes page that emitted --nats-url for agent workers, and a
test suite that stood a NATS server up for specs that no longer used it.

That is the one way this programme could still fail invisibly. Every test
passes, every binary works, and every production deployment goes on running
and paying for infrastructure that carries nothing.

Nothing in this repository starts a NATS server any more. The compose file is
four services, the docs say to shut the broker down and what to keep, and the
e2e suite runs on one PostgreSQL container.

The three LOCALAI_NATS_*_TIMEOUT env vars are KEPT, and are now documented
twice as being kept. They were never broker settings: each names a control-RPC
budget the frontend applies to a worker, still read and still enforced. They
carry the prefix only because they arrived with the bus, and renaming them
would break every existing deployment for cosmetics.

The agent worker's join command was the last surface still emitting the flag,
two tasks after the agent worker stopped dialling. The Playwright spec that
covered it asserted the opposite of what is now true, so it is inverted rather
than deleted, and it reads the rendered command string rather than the
component's variables: the variables are what the fix removes, so a spec
reading them would have stopped compiling instead of failing, and a compile
error is not evidence about what an operator is shown.

nats_jwt_test.go and its helpers are deleted. They pinned a real server
ENFORCING the minted permissions. The CONTENT of those allow lists is still
pinned, untouched, by pkg/natsauth's own suites, including the spec that
refuses to let the agent lists go empty, since an empty allow list in NATS
means unrestricted. The enforcement half is retired rather than moved:
enforcement is a property of a connection, and nothing opens one.

The suite's own NATS container goes with them, which the brief left for the
next task. Removing the pre-pull while BeforeSuite still ran the image would
have defeated the step rather than cleaned it up, and this change removes the
last reader of TestInfra.NC. agent_native_executor_test.go and
mcp_ci_job_test.go are moved onto infra.Bus() instead of deleted: they were
the last two specs building a bridge and a dispatcher on a client nobody uses,
which is exactly the drift TestInfra.Bus's own comment warns about.

cluster.Options.NatsURL is now fed a deliberately dead address rather than a
live container's. Frontends and agent workers still receive LOCALAI_NATS_URL,
because that is the coverage for the promise that an existing command line
still starts; sourcing it from a running server would have let a regression
that actually dialled it pass. The control in cluster_control_test.go keeps
its assertion and loses its explanation, which claimed the deployment had a
bus and no longer could.

One latent spec race surfaced and is fixed: the background-run spec waited for
a COUNT of events and then read a snapshot for the terminal status, which is
the last event of a run and therefore always arrives after the count is met.
Its immediate twin had already been fixed this way. Nothing in production
changed.

pkg/natsauth keeps its files. It is reachable from production only through the
natsauth.Config parameter thread, and that thread is the next task's.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

.github/workflows/tests-e2e-distributed.yml

@@ -56,11 +56,12 @@ jobs:
       - name: Pre-pull test images
         # Pulling here rather than inside the suite keeps container-start timing
         # out of the spec timeouts and makes a registry outage read as a
-        # setup failure instead of a test failure. These two are the only images
-        # the suite needs once the testcontainers reaper is disabled below.
+        # setup failure instead of a test failure. This is the only image the
+        # suite needs once the testcontainers reaper is disabled below: the
+        # suite stands up no message broker, because nothing under test dials
+        # one.
         run: |
           docker pull postgres:16-alpine
-          docker pull nats:2-alpine
       - name: Distributed E2E
         # TESTCONTAINERS_RYUK_DISABLED keeps the pre-pull above meaningful. The
         # reaper exists to clean up leaked containers on a long-lived host, but
@@ -91,9 +92,9 @@ jobs:
     #
     # Separate job from tests-e2e-distributed so the fast in-process suite is
     # not held behind a Go build of local-ai. Serial on purpose: each Ginkgo
-    # process would get its own PostgreSQL and NATS container and each spec
-    # spawns two or three local-ai children, so --procs on an unmeasured runner
-    # is a change to make with numbers, not by default.
+    # process would get its own PostgreSQL container and each spec spawns two or
+    # three local-ai children, so --procs on an unmeasured runner is a change to
+    # make with numbers, not by default.
     #
     # The two timeouts bound different things and are not alternatives. Ginkgo's
     # --timeout=20m bounds the SUITE only; this job timeout must additionally
@@ -160,7 +161,6 @@ jobs:
         # setup failure rather than a test failure.
         run: |
           docker pull postgres:16-alpine
-          docker pull nats:2-alpine
       - name: Cluster E2E
         env:
           # No LOCALAI_E2E_BINARY and no separate build step: make test-e2e-cluster

README.md

@@ -209,7 +209,7 @@ For older news and full release notes, see [GitHub Releases](https://github.com/
 - [Object Detection](https://localai.io/features/object-detection/)
 - [Reranker API](https://localai.io/features/reranker/)
 - [P2P Inferencing](https://localai.io/features/distribute/)
-- [Distributed Mode](https://localai.io/features/distributed-mode/) — Horizontal scaling with PostgreSQL + NATS
+- [Distributed Mode](https://localai.io/features/distributed-mode/): horizontal scaling on PostgreSQL, with no message broker to run
 - [Model Context Protocol (MCP)](https://localai.io/docs/features/mcp/)
 - [Built-in Agents](https://localai.io/features/agents/) — Autonomous AI agents with tool use, RAG, skills, SSE streaming, and [Agent Hub](https://agenthub.localai.io)
 - [Backend Gallery](https://localai.io/backends/) — Install/remove backends on the fly via OCI images

core/http/react-ui/e2e/nodes-roster.spec.js

@@ -65,9 +65,17 @@ test.describe('Nodes roster panels', () => {
 
 test.describe('Nodes join command', () => {
   // The panel emits BOTH the backend and the agent join command from one
-  // component, so the bus flag has to differ per tab rather than be deleted.
-  // Backend workers connect to no NATS server; agent workers still do.
-  test('omits the NATS flag for a backend worker and keeps it for an agent worker', async ({ page }) => {
+  // component. Neither worker kind dials a message bus any more: each holds one
+  // outward tunnel to --register-to and takes every verb on it. A join command
+  // carrying --nats-url would tell an operator to stand up, secure and pay for a
+  // broker that nothing in the deployment connects to, which is the one way this
+  // migration can still cost money after the code stopped using it.
+  //
+  // Asserted on the RENDERED command text rather than on the component's
+  // variables, because the variables are what the fix deletes: a spec reading
+  // them would stop compiling instead of failing, and a compile error is not
+  // evidence about what an operator is shown.
+  test('emits no bus flag for either worker kind', async ({ page }) => {
     await mockCluster(page, [])
     await page.goto('/app/nodes')
 
@@ -76,14 +84,21 @@ test.describe('Nodes join command', () => {
     await expect(backendCli).toContainText('local-ai worker', { timeout: 15_000 })
     await expect(backendCli).not.toContainText('--nats-url')
     const backendDocker = page.locator('.p2p-cmd pre').nth(1)
+    await expect(backendDocker).toContainText('LOCALAI_REGISTER_TO')
     await expect(backendDocker).not.toContainText('LOCALAI_NATS_URL')
 
+    // The agent tab is the one that regressed: it was the last surface still
+    // emitting the flag, and it kept emitting it for two tasks after the agent
+    // worker stopped dialling.
     await page.getByRole('radio', { name: /^Agent$/ }).click()
     const agentCli = page.locator('.p2p-cmd pre').first()
     await expect(agentCli).toContainText('local-ai agent-worker', { timeout: 15_000 })
-    await expect(agentCli).toContainText('--nats-url')
+    await expect(agentCli).toContainText('--register-to',
+    )
+    await expect(agentCli).not.toContainText('--nats-url')
     const agentDocker = page.locator('.p2p-cmd pre').nth(1)
-    await expect(agentDocker).toContainText('LOCALAI_NATS_URL')
+    await expect(agentDocker).toContainText('LOCALAI_REGISTER_TO')
+    await expect(agentDocker).not.toContainText('LOCALAI_NATS_URL')
   })
 
   test('does not advertise flags the CLI does not have', async ({ page }) => {
@@ -102,8 +117,12 @@ test.describe('Nodes join command', () => {
     // command carrying them fails at kong before LocalAI does anything.
     await expect(card).not.toContainText('--distributed-nats')
     await expect(card).not.toContainText('--distributed-db')
-    // And the worker step no longer tells an operator to point a backend
-    // worker at a bus it does not dial.
+    // Neither step tells an operator to point anything at a bus. The FRONTEND
+    // command is asserted first and by itself: it is the one that used to carry
+    // --nats-url as a required flag, so an operator following this card would
+    // have stood a broker up before starting LocalAI at all.
+    await expect(card.locator('.p2p-cmd pre').nth(0)).toContainText('--auth-database-url')
+    await expect(card.locator('.p2p-cmd pre').nth(0)).not.toContainText('--nats-url')
     await expect(card.locator('.p2p-cmd pre').nth(1)).not.toContainText('--nats-url')
   })
 })

core/http/react-ui/src/pages/Nodes.jsx

@@ -41,13 +41,6 @@ function WorkerHintCard({ addToast, activeTab, hasWorkers }) {
   const { selected, setSelected, option, dev, setDev } = useImageSelector('cpu')
   const isAgent = activeTab === 'agent'
   const workerCmd = isAgent ? 'agent-worker' : 'worker'
-  // Only the agent worker still uses the bus. A backend worker reaches this
-  // frontend over one outbound tunnel and connects to no NATS server, so
-  // emitting --nats-url on its join command would tell an operator to stand up
-  // infrastructure the command does not use. Both commands come from this one
-  // panel, which is why the flag is conditional rather than deleted.
-  const natsFlag = isAgent ? '  --nats-url "nats://nats:4222" \\\n' : ''
-  const natsEnv = isAgent ? '  -e LOCALAI_NATS_URL="nats://nats:4222" \\\n' : ''
   const flags = dockerFlags(option)
   const flagsStr = flags ? `${flags} \\\n  ` : ''
 
@@ -74,14 +67,14 @@ function WorkerHintCard({ addToast, activeTab, hasWorkers }) {
         <div>
           <p className="form-label">CLI</p>
           <CommandBlock
-            command={`local-ai ${workerCmd} \\\n  --register-to "${frontendUrl}" \\\n${natsFlag}  --registration-token "$LOCALAI_REGISTRATION_TOKEN"`}
+            command={`local-ai ${workerCmd} \\\n  --register-to "${frontendUrl}" \\\n  --registration-token "$LOCALAI_REGISTRATION_TOKEN"`}
             addToast={addToast}
           />
         </div>
         <div>
           <p className="form-label">Docker</p>
           <CommandBlock
-            command={`docker run --net host ${flagsStr}\\\n  -e LOCALAI_REGISTER_TO="${frontendUrl}" \\\n${natsEnv}  -e LOCALAI_REGISTRATION_TOKEN="$TOKEN" \\\n  ${dockerImage(option, dev)} ${workerCmd}`}
+            command={`docker run --net host ${flagsStr}\\\n  -e LOCALAI_REGISTER_TO="${frontendUrl}" \\\n  -e LOCALAI_REGISTRATION_TOKEN="$TOKEN" \\\n  ${dockerImage(option, dev)} ${workerCmd}`}
             addToast={addToast}
           />
         </div>
@@ -247,7 +240,7 @@ export default function Nodes() {
               <div className="flex-1">
                 <p className="fw-medium mb-xs">Start LocalAI with distributed mode</p>
                 <CommandBlock
-                  command={`local-ai run --distributed \\\n  --auth-database-url "postgres://user:pass@host/db" \\\n  --nats-url "nats://host:4222"`}
+                  command={`local-ai run --distributed \\\n  --auth-database-url "postgres://user:pass@host/db"`}
                   addToast={addToast}
                 />
               </div>

core/http/react-ui/src/utils/format.js

@@ -16,7 +16,8 @@ export function percentColor(pct) {
 // milliseconds, regardless of its encoding. The agent SSE bridge emits the
 // json_message timestamp in three different shapes depending on deploy mode:
 // an RFC3339 string (standalone agent pool), Unix milliseconds (local
-// dispatcher), or Unix nanoseconds (older NATS path). A numeric value is
+// dispatcher), or Unix nanoseconds (releases before the tunnel migration,
+// which a rolling upgrade still has in flight). A numeric value is
 // classified by magnitude (s / ms / us / ns) so any of them yields a sane
 // epoch. Falls back to Date.now() for null/empty/unparseable input.
 export function normalizeTimestampMs(ts) {

docker-compose.distributed.yaml

@@ -1,7 +1,11 @@
 # Docker Compose for LocalAI Distributed Mode
 #
-# Starts a full distributed stack: PostgreSQL, NATS, a LocalAI frontend,
-# and one llama-cpp backend node.
+# Starts a full distributed stack: PostgreSQL, a LocalAI frontend, one
+# llama-cpp backend node and one agent worker.
+#
+# There is no message broker in this file and none is needed. PostgreSQL carries
+# every cross-replica broadcast, and each worker dials one outbound tunnel to the
+# frontend and takes every verb on it.
 #
 # Model files are transferred from the frontend to backend nodes via HTTP
 # — no shared volumes needed between frontend and backends.
@@ -28,13 +32,6 @@ services:
       timeout: 3s
       retries: 10
 
-  nats:
-    image: nats:2-alpine
-    ports:
-      - "4222:4222"   # Client connections
-      - "8222:8222"   # HTTP monitoring (optional, useful for debugging)
-    command: ["--js", "-m", "8222"]  # Enable JetStream + monitoring
-
   # --- LocalAI Frontend ---
   # Stateless API server that routes requests to backend nodes.
   # Add more replicas behind a load balancer for HA.
@@ -52,7 +49,6 @@ services:
     environment:
       # Distributed mode
       LOCALAI_DISTRIBUTED: "true"
-      LOCALAI_NATS_URL: "nats://nats:4222"
       LOCALAI_AGENT_POOL_EMBEDDING_MODEL: "granite-embedding-107m-multilingual"
       LOCALAI_AGENT_POOL_VECTOR_ENGINE: "postgres"
       LOCALAI_AGENT_POOL_DATABASE_URL: "postgresql://localai:localai@postgres:5432/localai?sslmode=disable"
@@ -68,7 +64,7 @@ services:
       # Force pure-Go DNS resolver. The default cgo resolver follows the
       # container's nsswitch.conf and ends up forwarding to host
       # systemd-resolved (127.0.0.53), which isn't reachable from inside
-      # the container — failing every postgres/nats hostname lookup at
+      # the container, failing every postgres hostname lookup at
       # boot. The pure-Go path reads /etc/resolv.conf directly and uses
       # Docker's embedded DNS at 127.0.0.11.
       GODEBUG: "netdns=go"
@@ -83,13 +79,12 @@ services:
     depends_on:
       postgres:
         condition: service_healthy
-      nats:
-        condition: service_started
 
   # --- Worker Node ---
   # A generic worker that self-registers with the frontend.
   # The same LocalAI image is used — no separate image needed.
-  # The SmartRouter dynamically tells workers which backend to install via NATS.
+  # The SmartRouter tells a worker which backend to install over that worker's
+  # own tunnel.
   #
   # Model files are transferred from the frontend via HTTP file staging.
   # The worker has its own independent models volume.
@@ -115,10 +110,9 @@ services:
     # tunnel session, so `unhealthy` here means the frontend genuinely cannot
     # reach this worker.
     #
-    # No LOCALAI_NATS_URL and no dependency on the nats service: a backend
-    # worker connects to no bus. Everything the frontend asks of it travels the
-    # tunnel this container dials out to localai:8080. The frontend and the
-    # agent worker below still need NATS.
+    # This worker connects to nothing but the frontend. Everything the frontend
+    # asks of it travels the tunnel this container dials out to localai:8080, so
+    # the only service it depends on is localai itself.
     environment:
       LOCALAI_SERVE_ADDR: "0.0.0.0:50051"
       DEBUG: "true"
@@ -186,15 +180,15 @@ services:
   # LOCALAI_NODE_NAME really must differ: the registry upserts by name, so two
   # workers sharing one steal each other's row and each other's tunnel credential.
   #
-  # Workers are generic — no backend type needed. The SmartRouter
-  # will dynamically install the required backend via NATS when
-  # a model request arrives.
+  # Workers are generic: no backend type needed. The SmartRouter installs the
+  # required backend over the worker's tunnel when a model request arrives.
 
   # --- Agent Worker ---
   # Dedicated process for agent chat execution.
-  # Receives chat jobs from NATS, runs cogito LLM calls via the LocalAI API,
-  # and publishes results back via NATS for SSE delivery.
-  # No database access needed — config and skills are sent in the NATS payload.
+  # The frontend claims a queued run from PostgreSQL and drives it as a
+  # streaming control RPC over this container's own outbound tunnel; progress,
+  # agent events and the terminal result all come back on that same response.
+  # No database access needed: config and skills are sent in the request.
 
   agent-worker-1:
     # image: localai/localai:latest-cpu
@@ -212,11 +206,10 @@ services:
       - |
         apt-get update -qq && apt-get install -y -qq docker.io >/dev/null 2>&1
         exec /entrypoint.sh agent-worker
-    # The agent worker is NATS-only — no HTTP server to probe. The image's
-    # healthcheck detects that mode and reports healthy rather than probing a
-    # port that will never bind, so no override is needed here.
+    # The agent worker binds its control server on loopback only and publishes
+    # no port. The image's healthcheck detects that mode and reports healthy
+    # rather than probing a port that will never bind, so no override is needed.
     environment:
-      LOCALAI_NATS_URL: "nats://nats:4222"
       LOCALAI_REGISTER_TO: "http://localai:8080"
       LOCALAI_NODE_NAME: "agent-worker-1"
       LOCALAI_REGISTRATION_TOKEN: "changeme"  # Must match frontend token
@@ -226,8 +219,6 @@ services:
     depends_on:
       localai:
         condition: service_started
-      nats:
-        condition: service_started
 
 volumes:
   postgres_data:

docs/content/features/distributed-mode.md

@@ -13,7 +13,7 @@ Distributed mode requires authentication enabled with a **PostgreSQL** database
 
 ## Architecture Overview
 
-![Distributed mode architecture: a load balancer fronts stateless SmartRouter frontends backed by a shared NATS/PostgreSQL/S3 plane, with generic workers running per-model gRPC backends](/images/diagrams/distributed-mode-arch.png)
+![Distributed mode architecture: a load balancer fronts stateless SmartRouter frontends backed by a shared PostgreSQL/S3 plane, with generic workers reached over the tunnel each one dials out and running per-model gRPC backends](/images/diagrams/distributed-mode-arch.png)
 
 **Frontends** are stateless LocalAI instances that receive API requests and route them to worker nodes via the **SmartRouter**. All frontends share state through PostgreSQL, which also carries every cross-replica event they broadcast.
 
@@ -38,10 +38,10 @@ Each model gets its own gRPC backend process, so a single worker can serve multi
 ## Prerequisites
 
 - **PostgreSQL** (with pgvector extension recommended for RAG) - used for node registry, job store, auth, and shared state
-  - Each frontend replica holds **one extra PostgreSQL session** beyond its connection pool, pinned for the life of the process, and creates a `bus_messages` table. Both belong to the broadcast carrier that is replacing NATS for cross-replica fan-out; it already carries the four `state.*.delta` families (see [Cross-replica in-memory state](#cross-replica-in-memory-state)). Size `max_connections` for one additional session per frontend replica.
+  - Each frontend replica holds **one extra PostgreSQL session** beyond its connection pool, pinned for the life of the process, and creates a `bus_messages` table. Both belong to the broadcast carrier that carries every cross-replica fan-out, including the four `state.*.delta` families (see [Cross-replica in-memory state](#cross-replica-in-memory-state)). Size `max_connections` for one additional session per frontend replica.
   - That session reports an `application_name` of `localai_pgbus_<id>`, so `SELECT count(*) FROM pg_stat_activity WHERE application_name LIKE 'localai_pgbus_%'` counts the replicas currently listening. If the carrier loses its session it redials and re-registers on its own; a broadcast published while it was down is not replayed, which is why nothing that must survive a gap is carried by a broadcast alone.
   - `bus_messages` holds only broadcasts too large for a PostgreSQL notification, and every replica retires rows older than ten minutes. The table is a spill buffer, not a log: it is not a place to read past events from.
-- **No message bus.** Nothing in a distributed deployment connects to NATS any more. Everything a frontend broadcasts travels on the PostgreSQL the deployment already runs; every verb a frontend addresses to a worker, including an agent cancel, is an HTTP route on that worker's own tunnel. `LOCALAI_NATS_URL` is accepted and ignored everywhere - on the frontend, on `local-ai worker` and on `local-ai agent-worker` - so an existing command line still starts.
+- **No message broker. Do not deploy one.** Everything a frontend broadcasts travels on the PostgreSQL the deployment already runs; every verb a frontend addresses to a worker, including an agent cancel, is an HTTP route on that worker's own tunnel. If you are upgrading from a release that ran one, see [Migrating off the message broker](#migrating-off-the-message-broker): your existing command lines keep working and the broker can be shut down.
 - All services must be on the same network (or reachable via configured URLs)
 
 ## Quick Start with Docker Compose
@@ -52,7 +52,7 @@ The easiest way to try distributed mode locally is with the provided Docker Comp
 docker compose -f docker-compose.distributed.yaml up
 ```
 
-This starts PostgreSQL, a LocalAI frontend, and one worker node. The compose file still stands a NATS container up; nothing connects to it and you may delete that service. When you send an inference request, the SmartRouter automatically installs the needed backend on the worker and loads the model. See the file for details on adding GPU support, shared volumes, and additional workers.
+This starts PostgreSQL, a LocalAI frontend, one worker node and one agent worker. Those four services are the whole deployment: there is no broker in the file and none to add. When you send an inference request, the SmartRouter automatically installs the needed backend on the worker and loads the model. See the file for details on adding GPU support, shared volumes, and additional workers.
 
 {{% notice tip %}}
 Use `docker-compose.distributed.yaml` for quick local testing. For production, deploy PostgreSQL as a managed service and run frontends/workers on separate hosts. There is no message bus to deploy.
@@ -66,11 +66,10 @@ The frontend is a standard LocalAI instance with distributed mode enabled. These
 |------|---------|---------|-------------|
 | `--distributed` | `LOCALAI_DISTRIBUTED` | `false` | Enable distributed mode |
 | `--instance-id` | `LOCALAI_INSTANCE_ID` | auto UUID | Unique instance ID for this frontend |
-| `--nats-url` | `LOCALAI_NATS_URL` | *(ignored)* | **Accepted and ignored.** A frontend opens no message-bus connection. Kept so an existing command line still starts. |
 | `--distributed-advertise-addr` | `LOCALAI_DISTRIBUTED_ADVERTISE_ADDR` | *(derived)* | `host:port` the **other frontend replicas** dial to reach this one. See [Replica peer links](#replica-peer-links). |
 | `--registration-token` | `LOCALAI_REGISTRATION_TOKEN` | *(empty)* | Token that workers must provide to register |
 | `--registration-require-auth` | `LOCALAI_REGISTRATION_REQUIRE_AUTH` | `false` | Fail startup when distributed mode is enabled but the registration token is empty (node endpoints and worker file-transfer would otherwise be unauthenticated) |
-| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | **Umbrella switch.** Implies `--registration-require-auth`, which is what guards registration, the worker control planes and file transfer. It also implies the inert `--nats-require-auth`. Set this in production instead of the granular flags. |
+| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | **Umbrella switch.** Implies `--registration-require-auth`, which is what guards registration, the worker control planes and file transfer. Set this in production instead of the granular flags. |
 | `--auto-approve-nodes` | `LOCALAI_AUTO_APPROVE_NODES` | `false` | Auto-approve new worker nodes (skip admin approval) |
 | `--distributed-shared-models` | `LOCALAI_DISTRIBUTED_SHARED_MODELS` | `false` | Assert that every node mounts the **same** models directory at the **same** path (a shared volume). When `true`, the router skips file staging entirely and workers load models directly from the shared path instead of re-downloading them. See [Shared models directory](#shared-models-directory). |
 | `--distributed-disk-headroom-check` | `LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK` | `true` | Reject worker nodes that lack free space to store the model, at scheduling time rather than partway through staging. When `false`, node selection ignores free disk; the check still runs and warns when it would have rejected every node. Also toggleable at runtime via the `distributed_disk_headroom_check` setting. See [Disk headroom](#disk-headroom). |
@@ -83,6 +82,8 @@ The frontend is a standard LocalAI instance with distributed mode enabled. These
 | `--worker-reconnect-grace` | `LOCALAI_WORKER_RECONNECT_GRACE` | `90s` | How long a worker whose tunnel was lost is treated as **reconnecting** rather than **gone**. Only after this window may the scheduler stop placing work on that worker, clean up its rows and release its models. Set it below the worker's own reconnect worst case and you will condemn workers that are re-homing exactly as designed. Measured on the database clock, so every replica agrees. See [A lost tunnel is a departure, not an absence](#a-lost-tunnel-is-a-departure-not-an-absence). |
 | `--expose-node-header` | `LOCALAI_EXPOSE_NODE_HEADER` | `false` | When enabled, inference responses carry an `X-LocalAI-Node` header with the ID of the worker node that served the request. Coverage spans the OpenAI-compatible endpoints (chat completions, completions, embeddings, audio transcriptions, audio speech / TTS, image generations, image inpainting), the Jina rerank endpoint (`/v1/rerank`), the VAD endpoints (`/v1/vad`, `/vad`), and the Anthropic Messages (`/v1/messages`) and Ollama (`/api/chat`, `/api/generate`, `/api/embed`) shims. Useful for debugging, observability and load-balancer attribution. Off by default: the node ID reveals internal cluster topology and should not be exposed on a public endpoint. Best-effort: under heavy concurrency for the same model across multiple replicas, the header may reflect a recent routing decision rather than this exact request's. Acceptable for observability and debugging. |
 
+The three `LOCALAI_NATS_*_TIMEOUT` names above are **control-RPC budgets, not broker settings**, and are still read and enforced. They carry that prefix only because they were introduced alongside the message bus that distributed mode used to require; renaming them would break every existing deployment for cosmetics. Do not delete them when you [shut the broker down](#migrating-off-the-message-broker).
+
 ### Replica peer links
 
 Frontend replicas record themselves in an `instances` table and open direct links to each other, so that a request arriving at one replica can be served by state another replica holds. Each replica publishes one address for this, and every other replica dials it: it is the address **peers** use, which is not necessarily the address the process binds. A replica behind a Kubernetes Service, a load balancer or a NAT binds one and is reached at another.
@@ -125,7 +126,7 @@ The peer link is served at `/api/cluster/peer` and authenticates with `LOCALAI_R
 
 Several features keep state in a frontend's process memory and surface it over the API: fine-tune jobs, quantization jobs, agent tasks and Open Responses metadata. A round-robin load balancer sends a follow-up request to any replica, so each of those maps is kept current on every replica by a broadcast.
 
-**Those four families travel on PostgreSQL, not on NATS.** Each mutation is a `NOTIFY` on the database the deployment already runs, and each replica holds one `LISTEN` session for it. There is nothing to configure: the carrier uses the same database URL as `--auth-database-url` / `LOCALAI_AUTH_DATABASE_URL`.
+**Those four families travel on PostgreSQL.** Each mutation is a `NOTIFY` on the database the deployment already runs, and each replica holds one `LISTEN` session for it. There is nothing to configure: the carrier uses the same database URL as `--auth-database-url` / `LOCALAI_AUTH_DATABASE_URL`.
 
 | Map | Subject |
 |-----|---------|
@@ -141,7 +142,7 @@ for the life of a request: which gallery operations are in flight and how far
 along they are, which admin operations have been admitted, which model files are
 being staged onto a worker, and which replica already holds the KV/prefix cache
 for a prompt. Each of those is kept current on every replica by a broadcast, and
-**every one of them is on PostgreSQL**. Nothing in this table uses NATS.
+**every one of them is on PostgreSQL**.
 
 | Family | Subject | What a peer does with it |
 |--------|---------|--------------------------|
@@ -478,7 +479,7 @@ Registering against an upgraded frontend **clears** a node's `address` and `http
 
 A worker on this release opens **no inbound listener on a routable interface**. Its backend gRPC processes and its HTTP file-transfer server all bind loopback, and the frontend reaches both through the tunnel. Concretely:
 
-- **No inbound firewall rule, published port, Service or Ingress is needed for a worker.** A serve-backend worker needs outbound access to the frontend URL (`LOCALAI_REGISTER_TO`), and nothing else - not even to NATS. An agent worker binds only loopback too, and needs outbound access to `LOCALAI_REGISTER_TO` and nothing else either: registration, heartbeats, its tunnel and every verb the frontend addresses to it all go there.
+- **No inbound firewall rule, published port, Service or Ingress is needed for a worker.** A serve-backend worker needs outbound access to the frontend URL (`LOCALAI_REGISTER_TO`), and nothing else. An agent worker binds only loopback too, and needs outbound access to `LOCALAI_REGISTER_TO` and nothing else either: registration, heartbeats, its tunnel and every verb the frontend addresses to it all go there.
 - **`LOCALAI_ADVERTISE_ADDR` and `LOCALAI_ADVERTISE_HTTP_ADDR` are gone.** There is nothing to advertise. Both are ignored if still set; remove them.
 - **`LOCALAI_ADDR` and `LOCALAI_SERVE_ADDR` are read for their port only.** The port is the base of the backend port range, and `port-1` is the HTTP file-transfer port. The host half names an interface nothing binds.
 - The node's `address` and `http_address` fields in `GET /api/nodes` are empty, and are cleared for nodes that reported them before the upgrade.
@@ -596,22 +597,26 @@ The chat UI renders this state inline and retries automatically once the model r
 A frontend replica that dies mid-load does not wedge the model: the job row carries a heartbeat and another replica reclaims a job whose heartbeat has stopped. The heartbeat is time-based, not byte-based, because a checkpoint load legitimately transfers zero bytes for many minutes.
 {{% /notice %}}
 
-### NATS credentials (inert)
+### Migrating off the message broker
 
-**No LocalAI component connects to NATS.** The frontend's cross-replica fan-out is on PostgreSQL, a serve-backend worker takes every verb on its own tunnel, and an agent worker now does too, including the cancel that was the last family on a bus.
+Earlier releases of distributed mode required a NATS cluster alongside PostgreSQL. **They no longer do. Shut the broker down.** Nothing in LocalAI opens a connection to one: the frontend's cross-replica fan-out is on PostgreSQL, queued work is a claim on a PostgreSQL table, a serve-backend worker takes every verb on its own tunnel, and an agent worker does too, including the cancel that was the last family on a bus.
 
-Every `LOCALAI_NATS_*` setting is therefore accepted and inert, so an existing command line, unit file or Helm values file starts unchanged:
+There is no migration step and no cutover window. Stop the broker, delete its service from your compose file, chart or unit files, and delete the credentials you generated for it. A deployment that keeps running one is paying for infrastructure that carries nothing.
+
+**Your existing command lines still start.** Every `LOCALAI_NATS_*` setting below is parsed and then ignored, so an unedited command line, unit file or Helm values file needs no change on the day you upgrade. Remove them at your convenience.
 
 | Flag | Env Var | Status |
 |------|---------|--------|
-| `--nats-url` | `LOCALAI_NATS_URL` | Accepted and ignored on the frontend, `local-ai worker` and `local-ai agent-worker`. |
+| `--nats-url` | `LOCALAI_NATS_URL` | Accepted and ignored on the frontend, `local-ai worker` and `local-ai agent-worker`. The value is never dialled, so it may point at a broker that is already gone. |
 | `--nats-account-seed` | `LOCALAI_NATS_ACCOUNT_SEED` | The frontend still mints a per-node user JWT at registration (`nats_jwt` in the register response). Nothing consumes it. |
 | `--nats-service-jwt` / `--nats-service-seed` | `LOCALAI_NATS_SERVICE_JWT` / `LOCALAI_NATS_SERVICE_SEED` | Accepted, unused: the frontend opens no bus connection to present them on. |
 | `--nats-worker-jwt-ttl` | `LOCALAI_NATS_WORKER_JWT_TTL` | Lifetime of the minted-but-unused worker JWTs. |
-| `--nats-require-auth` | `LOCALAI_NATS_REQUIRE_AUTH` | On an agent worker this still makes registration **wait through admin approval** rather than starting against a pending node. It no longer gates any bus connection. |
+| `--nats-require-auth` | `LOCALAI_NATS_REQUIRE_AUTH` | On an agent worker this still makes registration **wait through admin approval** rather than starting against a pending node. It gates no connection. |
 | `--nats-tls-ca` / `--nats-tls-cert` / `--nats-tls-key` | `LOCALAI_NATS_TLS_*` | Accepted, unused. |
 
-You may stop running a NATS server, and remove these settings at your convenience.
+{{% notice warning %}}
+`LOCALAI_NATS_BACKEND_INSTALL_TIMEOUT`, `LOCALAI_NATS_BACKEND_UPGRADE_TIMEOUT` and `LOCALAI_NATS_MODEL_LOAD_TIMEOUT` are **not** in the table above and must **not** be removed. Despite their names they were never broker settings: each one is a control-RPC budget the frontend applies to a worker, and each is still read and still enforced. They are documented with the other frontend flags in [Frontend Configuration](#frontend-configuration). The names are kept because renaming them would break every existing deployment for cosmetics.
+{{% /notice %}}
 
 {{% notice note %}}
 `LOCALAI_AUTH` (HTTP users/sessions) is unrelated. HTTP registration still uses `LOCALAI_REGISTRATION_TOKEN`, and every worker control plane sits behind that same bearer check.
@@ -655,7 +660,7 @@ absolute snapshot path and skip transfer. Otherwise, the controller stages the
 complete snapshot tree to each worker before loading the backend. With an object
 store configured the controller uploads to the bucket and commands the worker to
 fetch over its tunnel; without one it pushes the files to the worker's HTTP file
-transfer server directly. Neither path uses NATS.
+transfer server directly.
 
 {{% notice warning %}}
 Every controller and worker must have enough disk space for its own snapshot
@@ -713,7 +718,7 @@ local-ai worker \
   --registration-token changeme
 ```
 
-There is no `--nats-url` here. A serve-backend worker connects to no message bus: it dials one outbound tunnel to `--register-to` and serves every request the frontend makes of it over that. The flag is still accepted and ignored, so an existing command line keeps working.
+There is no broker flag here. A serve-backend worker dials one outbound tunnel to `--register-to` and serves every request the frontend makes of it over that. A `--nats-url` left over from an older command line is still accepted and ignored; see [Migrating off the message broker](#migrating-off-the-message-broker).
 
 | Flag | Env Var | Default | Description |
 |------|---------|---------|-------------|
@@ -728,7 +733,6 @@ There is no `--nats-url` here. A serve-backend worker connects to no message bus
 | `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | Umbrella switch implying `--registration-require-auth` |
 | `--heartbeat-interval` | `LOCALAI_HEARTBEAT_INTERVAL` | `10s` | Interval between heartbeat pings |
 | `--worker-tunnel` | `LOCALAI_WORKER_TUNNEL` | `true` | Hold one outbound multiplexed tunnel to the frontend and serve its requests over it, so this worker needs no inbound port (see [Worker tunnels](#worker-tunnels)). Setting it to `false` is a **fatal startup error**, not a degraded mode: the frontend has no path that dials a worker's advertised address, so a worker without its tunnel is a worker nothing can reach. To run without tunnels, run the pre-tunnel release on both the worker and the frontend. |
-| `--nats-url` | `LOCALAI_NATS_URL` | *(ignored)* | **Accepted and ignored.** A serve-backend worker opens no NATS connection. Kept so an existing worker command line still starts. |
 | `--backends-path` | `LOCALAI_BACKENDS_PATH` | `./backends` | Path to backend binaries |
 | `--models-path` | `LOCALAI_MODELS_PATH` | `./models` | Path to model files |
 | `--vram-budget` | `LOCALAI_VRAM_BUDGET` | *(empty)* | Cap the VRAM this node advertises for model placement, as a percentage (e.g. `80%`) or an absolute amount (e.g. `12GB`). Empty uses all detected VRAM. See [Per-node VRAM budget](#per-node-vram-budget). |
@@ -1208,7 +1212,7 @@ engine_args:
 
 The ds4 backend (DeepSeek V4 Flash) supports **layer-parallel** distributed inference: a single model that is too large for one machine is split by transformer layer across several machines. Each machine must have the GGUF present locally, but loads **only its own slice** of the layers. This lets you run a model whose weights exceed any single host's memory.
 
-This is **not** routed through the SmartRouter: it is a model-internal split, configured manually (Phase 1). It is unrelated to the NATS/PostgreSQL distributed mode described above.
+This is **not** routed through the SmartRouter: it is a model-internal split, configured manually (Phase 1). It is unrelated to the PostgreSQL-backed distributed mode described above.
 
 ### Topology
 
@@ -1506,8 +1510,8 @@ Notes:
 - Check that `--registration-token` matches on both frontend and worker
 - Ensure auth is enabled on the frontend (`LOCALAI_AUTH=true`)
 
-**NATS connection errors:**
-- Nothing in LocalAI connects to NATS any more, on any component. If a release you are running still logs one, it predates the tunnel migration; on this release, look at the failing component's tunnel and its `--register-to` instead.
+**Message-broker connection errors:**
+- Nothing in LocalAI connects to a broker any more, on any component. A release that logs such an error predates the tunnel migration; on this release, look at the failing component's tunnel and its `--register-to` instead. See [Migrating off the message broker](#migrating-off-the-message-broker).
 
 **PostgreSQL connection errors:**
 - Verify the connection URL format: `postgresql://user:password@host:5432/dbname?sslmode=disable`
@@ -1562,7 +1566,7 @@ Notes:
 - The HTTP file transfer server runs on the base port - 1 (default: 50050)
 - All of those bind loopback, so a firewall cannot be the cause. What can is another service on the same host already holding a port in the range: move the worker's range with `LOCALAI_ADDR` (see [Worker Port Configuration](#worker-port-configuration)) or bound it with `LOCALAI_GRPC_MAX_PORT`
 - Verify the backend gallery configuration is correct
-- The worker needs OUTBOUND network access to the gallery and to `LOCALAI_REGISTER_TO`. It needs no inbound access at all, and no access to NATS
+- The worker needs OUTBOUND network access to the gallery and to `LOCALAI_REGISTER_TO`. It needs no inbound access at all
 
 ## Roadmap: Routing and Caching Enhancements
 

docs/content/features/distributed_inferencing.md

@@ -9,7 +9,7 @@ aliases = ["/features/distribution/"]
 ![Federated vs worker mode: federated routes a whole request to one node; worker shards one model across nodes](/images/diagrams/federated-vs-worker.png)
 
 {{% notice tip %}}
-Looking for production-grade horizontal scaling with PostgreSQL and NATS? See [Distributed Mode]({{% relref "features/distributed-mode" %}}).
+Looking for production-grade horizontal scaling backed by PostgreSQL alone? See [Distributed Mode]({{% relref "features/distributed-mode" %}}).
 {{% /notice %}}
 
 ## Choosing a distributed mode
@@ -19,7 +19,7 @@ LocalAI can spread inference across multiple machines in three ways. Pick the on
 | Mode | Best for | Guide |
 |------|----------|-------|
 | **P2P / Federated inference** | Ad-hoc clusters, community sharing, quick experimentation. Nodes discover each other via a shared libp2p token, with no central server. | This page |
-| **Distributed Mode (PostgreSQL + NATS)** | Production deployments, Kubernetes, and managed infrastructure. Stateless frontends behind a load balancer, workers self-register, and state lives in PostgreSQL. | [Distributed Mode]({{% relref "features/distributed-mode" %}}) |
+| **Distributed Mode (PostgreSQL)** | Production deployments, Kubernetes, and managed infrastructure. Stateless frontends behind a load balancer, workers self-register over an outbound tunnel, and state lives in PostgreSQL. No message broker is needed. | [Distributed Mode]({{% relref "features/distributed-mode" %}}) |
 | **MLX distributed** | Apple Silicon clusters running MLX models over the MLX distributed runtime. | [MLX Distributed]({{% relref "features/mlx-distributed" %}}) |
 
 For the low-level protocol and endpoints used by P2P workers, see the [P2P API reference]({{% relref "reference/p2p-api" %}}).

docs/content/features/text-generation.md

@@ -338,8 +338,9 @@ replicas, so retrieval, `previous_response_id` chaining and cancellation work
 regardless of which replica the load balancer picks:
 
 - `GET /v1/responses/{id}` returns the response from any replica.
-- `POST /v1/responses/{id}/cancel` is delegated over NATS to the replica that is
-  actually generating, so generation really stops. If that replica is gone, the
+- `POST /v1/responses/{id}/cancel` is delegated over the PostgreSQL broadcast
+  carrier to the replica that is actually generating, so generation really
+  stops. If that replica is gone, the
   response is reported as `cancelled` without blocking.
 - **Streaming resume (`?stream=true`) is served only by the replica that created
   the response.** The event buffer lives in that process's memory and is not

docs/static/images/diagrams/distributed-mode-arch.html

@@ -54,7 +54,7 @@
     </header>
     <div class="stage"><svg viewBox="0 0 1480 560" id="svg"></svg></div>
     <footer>
-      <div class="note">Stateless frontends, a shared <b>NATS/Postgres</b> plane, and generic workers running per-model backends.</div>
+      <div class="note">Stateless frontends, a shared <b>PostgreSQL</b> plane, and generic workers reached over the tunnel each one dials out.</div>
       <div class="url">localai.io<span>/features/distributed-mode</span></div>
     </footer>
   </div>
@@ -99,15 +99,17 @@ fY.forEach((y,i)=>{
 
 // ---------- STATE PLANE ----------
 txt(560,30,"SHARED STATE PLANE",{w:700,sz:13,ls:".2em",fill:SOFT});
-const SPX=560, SPW=300, SPY=46, SPH=470;
+const SPX=560, SPW=300, SPY=46, SPH=326;
 shadowRect(SPX,SPY,SPW,SPH,PAPER,INK,4);
 svg.appendChild(el("rect",{x:SPX,y:SPY,width:SPW,height:58,fill:RUST}));
 svg.appendChild(el("line",{x1:SPX,y1:SPY+58,x2:SPX+SPW,y2:SPY+58,stroke:INK,"stroke-width":4}));
 txt(SPX+22,SPY+38,"Control plane",{f:"Bricolage Grotesque",w:800,sz:26,fill:PAPER});
 // chips
+// No broker chip. Every cross-replica broadcast and every queued job is a row
+// or a NOTIFY on the database below, and no worker connects to this plane at
+// all: a worker is reached over the tunnel it dials out to a frontend.
 const chips=[
-  {n:"PostgreSQL", s:"shared config & state"},
-  {n:"NATS",       s:"jobs · messaging bus"},
+  {n:"PostgreSQL", s:"state · registry · broadcast · jobs"},
   {n:"S3 (optional)", s:"model & artifact store"},
 ];
 const CHX=SPX+24, CHW=SPW-48, CHH=104; let cy=SPY+82;
@@ -151,14 +153,15 @@ fY.forEach((y)=>{
   arrow(FX+FW, y+FH/2, SPX, ty, INK);
 });
 
-// NATS messaging bus -> workers (dashed). Workers coordinate via NATS;
-// PostgreSQL is the frontends' shared state, not something workers connect to.
-const natsY = SPY+82+CHH+18 + CHH/2; // NATS chip center y
-wY.forEach((y)=> arrow(SPX+SPW, natsY, WX, y+WH/2, RUSTD, "2 8"));
-// label the NATS bus arrows
-const labW=140, labH=26, labX=(SPX+SPW+WX)/2-labW/2, labY=natsY-46;
+// frontends -> workers over the tunnel (dashed), routed in the clear band BELOW
+// the state plane. The route is the point: control verbs go frontend to worker
+// and never touch the plane, and the worker opened the connection outward, so
+// nothing dials into a worker and no port is published on one.
+const tunY = 430;
+arrow(FX+FW, tunY, WX, tunY, RUSTD, "2 8");
+const labW=250, labH=26, labX=(FX+FW+WX)/2-labW/2, labY=tunY-40;
 svg.appendChild(el("rect",{x:labX,y:labY,width:labW,height:labH,fill:PAPER,stroke:RUSTD,"stroke-width":2}));
-txt(labX+labW/2,labY+18,"backend.install",{f:"Bricolage Grotesque",w:700,sz:14,a:"middle",fill:RUSTD});
+txt(labX+labW/2,labY+18,"backend.install \u00b7 worker-dialled tunnel",{f:"Bricolage Grotesque",w:700,sz:14,a:"middle",fill:RUSTD});
 
 // ---- annotated arrow: frontend -> worker : LoadModel (gRPC) ----
 arrow(FX+FW, fY[2]+FH-24, WX, wY[2]+WH-30, COLD, "4 7");

pkg/natsauth/permissions_coverage_test.go

@@ -1,8 +1,6 @@
 package natsauth_test
 
 import (
-	"os"
-	"regexp"
 	"strings"
 
 	"github.com/mudler/LocalAI/core/services/messaging"
@@ -157,32 +155,10 @@ var _ = Describe("WorkerPermissions subject coverage", func() {
 	})
 })
 
-var allowPubRe = regexp.MustCompile(`--allow-pub "([^"]*)"`)
-
-var _ = Describe("Documented NATS service-user permissions", func() {
-	// scripts/nats-auth-setup.sh ships the recommended service (frontend) JWT
-	// permissions. They must cover every subject the frontend actually publishes,
-	// or prefix-cache sync (and friends) break once LOCALAI_NATS_REQUIRE_AUTH is on.
-	const scriptPath = "../../scripts/nats-auth-setup.sh"
-
-	// Representative subjects the frontend publishes on the control plane.
-	// prefixcache.* is emitted by prefixcache.Sync in core/application/distributed.go.
-	frontendPublishes := []string{
-		messaging.SubjectPrefixCacheObserve,
-		messaging.SubjectPrefixCacheInvalidate,
-		messaging.SubjectGalleryProgress("op-1"),
-	}
-
-	It("cover every subject the frontend publishes", func() {
-		raw, err := os.ReadFile(scriptPath)
-		Expect(err).ToNot(HaveOccurred(), "cannot read %s", scriptPath)
-		m := allowPubRe.FindStringSubmatch(string(raw))
-		Expect(m).To(HaveLen(2), "no --allow-pub list found in %s", scriptPath)
-		allow := strings.Split(m[1], ",")
-
-		for _, subject := range frontendPublishes {
-			Expect(anyAllows(allow, subject)).To(BeTrue(),
-				"service-user --allow-pub %v does not cover %s (frontend publishes it)", allow, subject)
-		}
-	})
-})
+// The "Documented NATS service-user permissions" suite that stood here read
+// scripts/nats-auth-setup.sh and required its --allow-pub list to cover every
+// subject the frontend publishes. Both the script and the frontend's bus
+// connection are gone: cross-replica fan-out is a PostgreSQL NOTIFY, which has
+// no allow-list to fall out of sync with. The property was retired with the
+// artifact it guarded, not moved, and the surviving suites above still pin the
+// MINTED credentials, which are Task 17's to remove.

scripts/nats-auth-setup.sh

@@ -1,220 +0,0 @@
-#!/usr/bin/env bash
-# Generate NATS JWT authentication material and server configuration
-# for LocalAI distributed mode.
-#
-# Requires: nsc (https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro/nsc)
-#
-# Outputs:
-#   ./nats-keys/localai-nats.env
-#   ./nats-keys/localai-frontend.creds
-#   ./nats-keys/nats-auth.conf
-#   ./nats-keys/nats-server.conf
-#
-# Environment overrides:
-#   NATS_OPERATOR_NAME
-#   NATS_ACCOUNT_NAME
-#   NATS_SERVICE_USER
-#   NATS_KEYS_DIR
-
-#
-# LocalAI workers receive their own JWT and user seed when registering
-# with the frontend.
-
-set -euo pipefail
-
-# Ensure newly created secret files are private by default.
-umask 077
-
-if ! command -v nsc >/dev/null 2>&1; then
-	echo "nsc is required. Install from https://github.com/nats-io/nsc/releases" >&2
-	exit 1
-fi
-
-OPERATOR="${NATS_OPERATOR_NAME:-localai-operator}"
-ACCOUNT="${NATS_ACCOUNT_NAME:-localai}"
-SYSTEM_ACCOUNT="${NATS_SYSTEM_ACCOUNT_NAME:-SYS}"
-SERVICE_USER="${NATS_SERVICE_USER:-localai-frontend}"
-OUTPUT_DIR="${NATS_KEYS_DIR:-./nats-keys}"
-
-CREDS_FILE="$OUTPUT_DIR/${SERVICE_USER}.creds"
-ENV_FILE="$OUTPUT_DIR/localai-nats.env"
-AUTH_CONFIG_FILE="$OUTPUT_DIR/nats-auth.conf"
-SERVER_CONFIG_FILE="$OUTPUT_DIR/nats-server.conf"
-
-mkdir -p "$OUTPUT_DIR"
-
-echo "Configuring NATS operator: $OPERATOR"
-
-# Create the operator if it does not exist, otherwise select it.
-if nsc select operator "$OPERATOR" >/dev/null 2>&1; then
-	echo "[ OK ] using existing operator '$OPERATOR'"
-else
-	nsc add operator \
-		-n "$OPERATOR" \
-		--generate-signing-key
-
-	nsc select operator "$OPERATOR" >/dev/null
-fi
-
-# Create and assign the NATS system account.
-if nsc describe account \
-	-n "$SYSTEM_ACCOUNT" >/dev/null 2>&1; then
-	echo "[ OK ] using existing system account '$SYSTEM_ACCOUNT'"
-else
-	nsc add account -n "$SYSTEM_ACCOUNT"
-fi
-
-nsc edit operator \
-	--system-account "$SYSTEM_ACCOUNT"
-
-# Create the LocalAI account if it does not exist.
-if nsc describe account -n "$ACCOUNT" >/dev/null 2>&1; then
-	echo "[ OK ] using existing account '$ACCOUNT'"
-else
-	nsc add account -n "$ACCOUNT"
-fi
-
-nsc select account "$ACCOUNT" >/dev/null
-
-# Create the frontend service user if it does not exist.
-if nsc describe user \
-	-n "$SERVICE_USER" \
-	--account "$ACCOUNT" >/dev/null 2>&1; then
-	echo "[ OK ] using existing user '$SERVICE_USER'"
-else
-	nsc add user \
-		-n "$SERVICE_USER" \
-		--account "$ACCOUNT"
-fi
-
-# Frontend control-plane permissions.
-nsc edit user \
-	-n "$SERVICE_USER" \
-	--account "$ACCOUNT" \
-	--allow-pub "nodes.>,gallery.>,agent.>,staging.>,state.>,jobs.>,mcp.>,cache.>,prefixcache.>,finetune.>" \
-	--allow-sub "nodes.>,gallery.>,agent.>,staging.>,state.>,jobs.>,mcp.>,cache.>,prefixcache.>,_INBOX.>"
-
-# Generate a credentials file containing the frontend user JWT and seed.
-rm -f "$CREDS_FILE"
-
-nsc generate creds \
-	-a "$ACCOUNT" \
-	-n "$SERVICE_USER" \
-	-o "$CREDS_FILE"
-
-# Extract the frontend JWT from the credentials file.
-SERVICE_JWT="$(
-	awk '
-		/BEGIN NATS USER JWT/ {
-			capture = 1
-			next
-		}
-		/END NATS USER JWT/ {
-			capture = 0
-		}
-		capture
-	' "$CREDS_FILE" | tr -d '\r\n'
-)"
-
-# Extract the frontend user seed from the credentials file.
-SERVICE_SEED="$(
-	awk '
-		/BEGIN USER NKEY SEED/ {
-			capture = 1
-			next
-		}
-		/END USER NKEY SEED/ {
-			capture = 0
-		}
-		capture
-	' "$CREDS_FILE" | tr -d '\r\n'
-)"
-
-# Retrieve the seed belonging to this exact account rather than taking
-# the first account key found in the keystore.
-ACCOUNT_SEED="$(
-	nsc list keys \
-		--account "$ACCOUNT" \
-		--accounts \
-		--show-seeds |
-	awk -F '|' -v expected="$ACCOUNT" '
-		function trim(value) {
-			gsub(/^[[:space:]]+|[[:space:]]+$/, "", value)
-			return value
-		}
-
-		NF >= 3 {
-			entity = trim($2)
-			seed = trim($3)
-
-			if (entity == expected && seed ~ /^SA[A-Z0-9]+$/) {
-				print seed
-				exit
-			}
-		}
-	'
-)"
-
-# Validate all extracted values before writing output files.
-if [[ ! "$ACCOUNT_SEED" =~ ^SA[A-Z0-9]+$ ]]; then
-	echo "Unable to extract the account seed for '$ACCOUNT'." >&2
-	exit 1
-fi
-
-if [[ ! "$SERVICE_JWT" =~ ^eyJ ]]; then
-	echo "Unable to extract the service JWT from '$CREDS_FILE'." >&2
-	exit 1
-fi
-
-if [[ ! "$SERVICE_SEED" =~ ^SU[A-Z0-9]+$ ]]; then
-	echo "Unable to extract the service seed from '$CREDS_FILE'." >&2
-	exit 1
-fi
-
-# Generate the trusted operator and memory resolver configuration.
-# This contains public operator/account JWT claims, not the private seeds.
-nsc generate config \
-	--mem-resolver \
-	--config-file "$AUTH_CONFIG_FILE" \
-	--force
-
-# Generate the primary NATS server configuration.
-# The include path matches the Docker Compose mounts shown below.
-cat >"$SERVER_CONFIG_FILE" <<'NATS_CONFIG'
-server_name: localai-nats
-port: 4222
-http: 8222
-
-jetstream {
-	store_dir: /data/jetstream
-}
-
-include nats-auth.conf
-NATS_CONFIG
-
-# Generate the environment file consumed by the LocalAI frontend.
-cat >"$ENV_FILE" <<EOF
-LOCALAI_NATS_ACCOUNT_SEED=$ACCOUNT_SEED
-LOCALAI_NATS_SERVICE_JWT=$SERVICE_JWT
-LOCALAI_NATS_SERVICE_SEED=$SERVICE_SEED
-EOF
-
-# The environment and credentials files contain private seeds.
-chmod 600 "$CREDS_FILE" "$ENV_FILE"
-
-# These contain server configuration and public JWT claims.
-chmod 644 "$AUTH_CONFIG_FILE" "$SERVER_CONFIG_FILE"
-
-echo
-echo "=== LocalAI NATS JWT setup complete ==="
-echo
-echo "LocalAI environment: $ENV_FILE"
-echo "Service credentials: $CREDS_FILE"
-echo "NATS server config:   $SERVER_CONFIG_FILE"
-echo "NATS auth config:     $AUTH_CONFIG_FILE"
-echo
-echo "Keep '$ENV_FILE' and '$CREDS_FILE' secret."
-echo "Do not commit them to source control."
-echo
-echo "=== LocalAI NATS environment ==="
-cat "$ENV_FILE"

tests/e2e/distributed/agent_native_executor_test.go

@@ -227,10 +227,17 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 		infra *TestInfra
 		db    *gorm.DB
 		store *agents.AgentStore
+		bus   messaging.Broadcaster
 	)
 
 	BeforeEach(func() {
 		infra = SetupInfra("localai_agent_native_test")
+		// The PostgreSQL LISTEN/NOTIFY carrier, which is the one a deployment
+		// runs and the one the SSE routes read. Handing the bridge anything
+		// else would leave these specs green while an operator's agent stream
+		// stayed empty: publishing onto a carrier nobody is listening on
+		// succeeds and reports nothing.
+		bus = infra.Bus()
 
 		var err error
 		db, err = gorm.Open(pgdriver.Open(infra.PGURL), &gorm.Config{
@@ -315,8 +322,8 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 	})
 
 	Context("WorkerExecutor", func() {
-		It("should dispatch chat via NATS and receive response", func() {
-			bridge := agents.NewEventBridge(infra.NC, nil, "test-instance", nil)
+		It("should serve a dispatched chat verb and publish the response on the carrier", func() {
+			bridge := agents.NewEventBridge(bus, nil, "test-instance", nil)
 
 			configs := &mockConfigProvider{configs: map[string]*agents.AgentConfig{
 				"test-agent": {
@@ -330,7 +337,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
 
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("test-agent", "user1"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("test-agent", "user1"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -344,8 +351,8 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// The executor is handed one run by the frontend replica that
 			// claimed it, over the control verb, and publishes everything it
 			// produces onto the writer that replica is reading. Here that
-			// writer is the bus itself, so the subscription above still sees
-			// the same events on the same subject.
+			// writer is the broadcast carrier itself, so the subscription above
+			// sees the same events on the same subject.
 			executor := agents.NewWorkerExecutor(bridge, configs, "http://localhost:8080", "test-key")
 
 			reply, err := executor.Execute(infra.Ctx, mustAgentJSON(agents.AgentChatEvent{
@@ -354,7 +361,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 				Message:   "Hello",
 				MessageID: "msg-test-001",
 				Role:      "user",
-			}), infra.NC)
+			}), bus)
 			Expect(err).ToNot(HaveOccurred(), "an error here is this worker failing to SERVE the verb, not the run failing")
 			Expect(string(reply)).To(ContainSubstring("status"))
 
@@ -382,7 +389,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// anything leaving it. The cancel of a run held by an agent WORKER
 			// travels over that worker's tunnel and is driven end to end, on a
 			// real one, in agent_distributed_test.go.
-			bridge := agents.NewEventBridge(infra.NC, nil, "cancel-test", nil)
+			bridge := agents.NewEventBridge(bus, nil, "cancel-test", nil)
 
 			var cancelled atomic.Bool
 			bridge.RegisterCancel("test-msg-id", func() {
@@ -395,7 +402,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 		})
 
 		It("should execute agent chat from enriched payload without ConfigProvider", func() {
-			bridge := agents.NewEventBridge(infra.NC, nil, "enriched-test", nil)
+			bridge := agents.NewEventBridge(bus, nil, "enriched-test", nil)
 
 			// Executor with NO ConfigProvider (simulating DB-free worker)
 			executor := agents.NewWorkerExecutor(bridge, nil, "http://localhost:8080", "test-key")
@@ -403,7 +410,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// Subscribe to events to verify processing
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("enriched-agent", "user1"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("enriched-agent", "user1"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -414,8 +421,6 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			Expect(err).ToNot(HaveOccurred())
 			defer sub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			// An enriched AgentChatEvent with embedded Config, exactly the body
 			// the claim row carries.
 			evt := agents.AgentChatEvent{
@@ -430,7 +435,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 					SystemPrompt: "Be helpful.",
 				},
 			}
-			_, err = executor.Execute(infra.Ctx, mustAgentJSON(evt), infra.NC)
+			_, err = executor.Execute(infra.Ctx, mustAgentJSON(evt), bus)
 			Expect(err).ToNot(HaveOccurred())
 
 			// The executor should process this even without a ConfigProvider.
@@ -571,8 +576,8 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 	})
 
 	Context("Full Distributed Chat Flow", func() {
-		It("should dispatch chat via NATS, execute, and publish response via EventBridge", func() {
-			bridge := agents.NewEventBridge(infra.NC, nil, "flow-test", nil)
+		It("should execute a dispatched chat and publish the response via EventBridge", func() {
+			bridge := agents.NewEventBridge(bus, nil, "flow-test", nil)
 
 			// Store agent config in PostgreSQL
 			cfg := agents.AgentConfig{
@@ -592,7 +597,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// Subscribe to events to capture the full flow
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("flow-agent", "user1"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("flow-agent", "user1"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -614,7 +619,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 				Message:   "Hello flow test",
 				MessageID: "msg-flow-001",
 				Role:      "user",
-			}), infra.NC)
+			}), bus)
 			Expect(err).ToNot(HaveOccurred())
 
 			Eventually(func() int {
@@ -671,8 +676,8 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			_, _ = agents.ExecuteBackgroundRun(infra.Ctx, "http://localhost:8080", "key", cfg, agents.Callbacks{})
 		})
 
-		It("should dispatch background run via NATS with system role", func() {
-			bridge := agents.NewEventBridge(infra.NC, nil, "bg-test", nil)
+		It("should execute a dispatched background run with the system role", func() {
+			bridge := agents.NewEventBridge(bus, nil, "bg-test", nil)
 
 			cfg := agents.AgentConfig{
 				Name:          "bg-agent",
@@ -688,7 +693,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// Subscribe to capture events
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("bg-agent", "system"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("bg-agent", "system"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -709,7 +714,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 				MessageID: "bg-1",
 				Role:      "system",
 			}
-			_, err = executor.Execute(infra.Ctx, mustAgentJSON(evt), infra.NC)
+			_, err = executor.Execute(infra.Ctx, mustAgentJSON(evt), bus)
 			Expect(err).ToNot(HaveOccurred())
 
 			// Should receive at least a processing status
@@ -986,18 +991,18 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 		})
 	})
 
-	Context("Full Distributed Agent Execution via NATS with Mock LLM Server", func() {
-		It("should dispatch chat via NATS, execute on worker with mock LLM, and publish response via EventBridge", func() {
+	Context("Full Distributed Agent Execution with Mock LLM Server", func() {
+		It("should execute a dispatched chat on the worker with a mock LLM and publish the response via EventBridge", func() {
 			// Start mock LLM HTTP server
 			llmURL, llmShutdown := startAgentMockLLMServer("I found the answer using my skills.")
 			defer llmShutdown()
 
-			bridge := agents.NewEventBridge(infra.NC, nil, "full-e2e-test", nil)
+			bridge := agents.NewEventBridge(bus, nil, "full-e2e-test", nil)
 
 			// Subscribe to agent events
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("e2e-agent", "user1"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("e2e-agent", "user1"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -1011,8 +1016,6 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			// Point the executor at our mock LLM server
 			executor := agents.NewWorkerExecutor(bridge, nil, llmURL, "test-key")
 
-			FlushNATS(infra.NC)
-
 			// The enriched event with skills, as the claim row carries it
 			evt := agents.AgentChatEvent{
 				AgentName: "e2e-agent",
@@ -1031,7 +1034,7 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 					{Name: "search", Description: "Search the web", Content: "Full search skill content here"},
 				},
 			}
-			reply, err := executor.Execute(infra.Ctx, mustAgentJSON(evt), infra.NC)
+			reply, err := executor.Execute(infra.Ctx, mustAgentJSON(evt), bus)
 			Expect(err).ToNot(HaveOccurred())
 			// A run that reached a conclusion is this worker's own answer, and
 			// it travels on the response body rather than only as a publish.
@@ -1071,17 +1074,17 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 			Expect(agentMessage).To(BeTrue(), "should receive agent response message via EventBridge")
 		})
 
-		It("should execute background agent run via NATS dispatcher with mock LLM", func() {
+		It("should execute a dispatched background agent run with a mock LLM", func() {
 			// Start mock LLM HTTP server
 			llmURL, llmShutdown := startAgentMockLLMServer("All systems operational. No issues detected.")
 			defer llmShutdown()
 
-			bridge := agents.NewEventBridge(infra.NC, nil, "bg-e2e-test", nil)
+			bridge := agents.NewEventBridge(bus, nil, "bg-e2e-test", nil)
 
 			// Subscribe to agent events
 			var receivedEvents []agents.AgentEvent
 			var eventMu sync.Mutex
-			sub, err := infra.NC.Subscribe(messaging.SubjectAgentEvents("bg-e2e-agent", "system"), func(data []byte) {
+			sub, err := bus.Subscribe(messaging.SubjectAgentEvents("bg-e2e-agent", "system"), func(data []byte) {
 				var evt agents.AgentEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -1094,8 +1097,6 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 
 			executor := agents.NewWorkerExecutor(bridge, nil, llmURL, "test-key")
 
-			FlushNATS(infra.NC)
-
 			// A system-role run (background/autonomous)
 			evt := agents.AgentChatEvent{
 				AgentName: "bg-e2e-agent",
@@ -1111,36 +1112,41 @@ var _ = Describe("Native Agent Executor", Label("Distributed", "AgentNative"), f
 					InnerMonologueTemplate: "Your goal is: {{.Goal}}. What should you do?",
 				},
 			}
-			reply, err := executor.Execute(infra.Ctx, mustAgentJSON(evt), infra.NC)
+			reply, err := executor.Execute(infra.Ctx, mustAgentJSON(evt), bus)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(string(reply)).To(ContainSubstring(`"status":"completed"`))
 
-			// Wait for agent response
-			Eventually(func() int {
+			// Waited for by CONTENT and not by count, for the same reason as
+			// its sibling above: the terminal status is the LAST event of the
+			// run, so "at least two events have arrived" is satisfied strictly
+			// before it, and the snapshot read afterwards cannot contain it.
+			// A count is a proxy for arrival that is wrong by construction
+			// whenever the run publishes more than the events being counted.
+			seen := func() (agentMessage, completed bool) {
 				eventMu.Lock()
 				defer eventMu.Unlock()
-				return len(receivedEvents)
-			}, "15s").Should(BeNumerically(">=", 2))
-
-			eventMu.Lock()
-			defer eventMu.Unlock()
-
-			var hasAgentMessage, hasCompleted bool
-			for _, evt := range receivedEvents {
-				if evt.EventType == "json_message" && evt.Sender == "agent" {
-					hasAgentMessage = true
-					Expect(evt.Content).To(ContainSubstring("systems operational"))
-				}
-				if evt.EventType == "json_message_status" && evt.Metadata != "" {
-					var meta map[string]string
-					json.Unmarshal([]byte(evt.Metadata), &meta)
-					if meta["status"] == "completed" {
-						hasCompleted = true
+				for _, evt := range receivedEvents {
+					if evt.EventType == "json_message" && evt.Sender == "agent" {
+						agentMessage = true
+						Expect(evt.Content).To(ContainSubstring("systems operational"))
+					}
+					if evt.EventType == "json_message_status" && evt.Metadata != "" {
+						var meta map[string]string
+						// A metadata blob this spec cannot read is not a
+						// completed status, and saying so beats failing the
+						// whole run on one malformed event.
+						if err := json.Unmarshal([]byte(evt.Metadata), &meta); err == nil && meta["status"] == "completed" {
+							completed = true
+						}
 					}
 				}
+				return agentMessage, completed
 			}
-			Expect(hasAgentMessage).To(BeTrue(), "background agent should produce a response via EventBridge")
-			Expect(hasCompleted).To(BeTrue(), "background agent should complete")
+
+			Eventually(func() bool { _, completed := seen(); return completed }, "15s").
+				Should(BeTrue(), "background agent should complete")
+			agentMessage, _ := seen()
+			Expect(agentMessage).To(BeTrue(), "background agent should produce a response via EventBridge")
 		})
 	})
 

tests/e2e/distributed/cluster_baseline_test.go

@@ -288,10 +288,16 @@ func startClusterOnFreshDB(frontends, workers int, customise ...func(*cluster.Op
 		Binary:      binary,
 		MockBackend: mockBackend,
 		PGDSN:       infra.PGURL,
-		NatsURL:     infra.NatsURL,
-		LogDir:      logDir,
-		Frontends:   frontends,
-		Workers:     workers,
+		// Deliberately a dead address. Frontends and agent workers are still
+		// handed LOCALAI_NATS_URL so this suite keeps covering the promise that
+		// an operator's existing command line starts unchanged after the broker
+		// is shut down. Pointing it at a running server would make a regression
+		// that dialled it invisible; pointing it at nothing makes such a
+		// regression a startup failure in every cluster spec.
+		NatsURL:   staleBusURL,
+		LogDir:    logDir,
+		Frontends: frontends,
+		Workers:   workers,
 	}
 	for _, apply := range customise {
 		apply(&options)

tests/e2e/distributed/cluster_control_test.go

@@ -413,8 +413,11 @@ var _ = Describe("Control plane over the worker tunnel", Label("Distributed"), L
 			Expect(entry).ToNot(HavePrefix("LOCALAI_NATS_URL="),
 				"the worker was handed a bus URL, so this spec is not about a worker that has none")
 		}
-		// And the deployment it joined DOES have a bus, so "no NATS anywhere"
-		// is not what makes this pass.
+		// And the harness DOES still hand a LOCALAI_NATS_URL to the other
+		// processes in this cluster, so the worker's lack of one is a property
+		// of the worker and not of a harness that stopped setting the variable
+		// at all. There is no broker behind that URL any more, which is the
+		// point: nothing dials it, so nothing notices.
 		Expect(c.NatsURL()).ToNot(BeEmpty())
 
 		probe := newRosterProbe(c, client, 0)

tests/e2e/distributed/foundation_test.go

@@ -30,7 +30,6 @@ var _ = Describe("Phase 0: Foundation", Label("Distributed"), func() {
 		It("should reject --distributed without PostgreSQL configured", func() {
 			appCfg := config.NewApplicationConfig(
 				config.EnableDistributed,
-				config.WithNatsURL(infra.NatsURL),
 				// No auth/PostgreSQL configured
 			)
 			Expect(appCfg.Distributed.Enabled).To(BeTrue())
@@ -38,26 +37,30 @@ var _ = Describe("Phase 0: Foundation", Label("Distributed"), func() {
 			Expect(appCfg.Auth.Enabled).To(BeFalse())
 		})
 
-		It("should reject --distributed without NATS configured", func() {
+		It("leaves the inert bus URL empty when nothing sets it", func() {
 			appCfg := config.NewApplicationConfig(
 				config.EnableDistributed,
 				config.WithAuthEnabled(true),
 				config.WithAuthDatabaseURL(infra.PGURL),
-				// No NATS URL
 			)
 			Expect(appCfg.Distributed.NatsURL).To(BeEmpty())
 		})
 
 		It("should accept valid distributed configuration", func() {
+			// staleBusURL points at nothing on purpose. It is the shape of an
+			// operator's existing command line after the broker was shut down,
+			// and the promise this spec holds is that such a command line still
+			// STARTS: the value is parsed, stored and never dialled. Pointing
+			// it at a live server would let a regression that dialled it pass.
 			appCfg := config.NewApplicationConfig(
 				config.EnableDistributed,
 				config.WithAuthEnabled(true),
 				config.WithAuthDatabaseURL(infra.PGURL),
-				config.WithNatsURL(infra.NatsURL),
+				config.WithNatsURL(staleBusURL),
 			)
 			Expect(appCfg.Distributed.Enabled).To(BeTrue())
 			Expect(appCfg.Auth.Enabled).To(BeTrue())
-			Expect(appCfg.Distributed.NatsURL).To(Equal(infra.NatsURL))
+			Expect(appCfg.Distributed.NatsURL).To(Equal(staleBusURL))
 		})
 
 		It("should generate unique frontend ID on startup", func() {

tests/e2e/distributed/gallery_distributed_test.go

@@ -70,8 +70,9 @@ var _ = Describe("Gallery Distributed", Label("Distributed"), func() {
 
 	// The gallery families ride the broadcast carrier, not NATS.
 	//
-	// These used to publish and subscribe on infra.NC, which asserted that NATS
-	// delivers to itself and nothing about this deployment: they would have
+	// These used to publish and subscribe on a message-bus client, which
+	// asserted that the bus delivers to itself and nothing about this
+	// deployment: they would have
 	// stayed green through the whole migration while the gallery service had
 	// already moved. Two carriers on the deployment's own database is the shape
 	// a fleet has, and it is the shape that fails when one end moves and the

tests/e2e/distributed/mcp_ci_job_helper_test.go

@@ -16,10 +16,20 @@ import (
 	"github.com/mudler/xlog"
 )
 
+// publishTestEvent is the one place this helper publishes from, so a carrier
+// that refuses a publish is reported rather than swallowed. A dropped progress
+// line has no other symptom: the spec's subscriber simply never fires and the
+// failure surfaces as a timeout that names the wrong thing.
+func publishTestEvent(pub messaging.Publisher, subject string, payload any) {
+	if err := pub.Publish(subject, payload); err != nil {
+		xlog.Error("test worker failed to publish", "subject", subject, "error", err)
+	}
+}
+
 // processMCPCIJobForTest replicates the logic of handleMCPCIJob from agent_worker.go
 // for testing purposes. This allows e2e testing of the full MCP CI job execution path
 // without needing to start an actual agent worker binary.
-func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *messaging.Client) {
+func processMCPCIJobForTest(data []byte, apiURL, apiToken string, pub messaging.Publisher) {
 	var evt jobs.JobEvent
 	if err := json.Unmarshal(data, &evt); err != nil {
 		xlog.Error("Failed to unmarshal job event", "error", err)
@@ -30,30 +40,30 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 	task := evt.Task
 	if job == nil || task == nil {
 		xlog.Error("MCP CI job missing enriched data", "jobID", evt.JobID)
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", "job or task data missing from NATS event")
+		publishTestJobResult(pub, evt.JobID, "failed", "", "job or task data missing from the job event")
 		return
 	}
 
 	modelCfg := evt.ModelConfig
 	if modelCfg == nil {
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", "model config missing from job event")
+		publishTestJobResult(pub, evt.JobID, "failed", "", "model config missing from job event")
 		return
 	}
 
 	// Publish running status
-	natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+	publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 		JobID: evt.JobID, Status: "running", Message: "Job started on test worker",
 	})
 
 	// Parse MCP config
 	if modelCfg.MCP.Servers == "" && modelCfg.MCP.Stdio == "" {
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", "no MCP servers configured for model")
+		publishTestJobResult(pub, evt.JobID, "failed", "", "no MCP servers configured for model")
 		return
 	}
 
 	remote, stdio, err := modelCfg.MCP.MCPConfigFromYAML()
 	if err != nil {
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", fmt.Sprintf("failed to parse MCP config: %v", err))
+		publishTestJobResult(pub, evt.JobID, "failed", "", fmt.Sprintf("failed to parse MCP config: %v", err))
 		return
 	}
 
@@ -64,7 +74,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 		if err != nil {
 			errMsg = fmt.Sprintf("failed to create MCP sessions: %v", err)
 		}
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", errMsg)
+		publishTestJobResult(pub, evt.JobID, "failed", "", errMsg)
 		return
 	}
 
@@ -92,7 +102,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 	defer cancel()
 
 	// Publish running status
-	publishTestJobStatus(natsClient, evt.JobID, "running", "")
+	publishTestJobStatus(pub, evt.JobID, "running", "")
 
 	// Buffer stream tokens
 	var reasoningBuf, contentBuf strings.Builder
@@ -100,13 +110,13 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 
 	flushStreamBuf := func() {
 		if reasoningBuf.Len() > 0 {
-			natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+			publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 				JobID: evt.JobID, TraceType: "reasoning", TraceContent: reasoningBuf.String(),
 			})
 			reasoningBuf.Reset()
 		}
 		if contentBuf.Len() > 0 {
-			natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+			publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 				JobID: evt.JobID, TraceType: "content", TraceContent: contentBuf.String(),
 			})
 			contentBuf.Reset()
@@ -119,13 +129,13 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 		cogito.WithMCPs(sessions...),
 		cogito.WithStatusCallback(func(status string) {
 			flushStreamBuf()
-			natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+			publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 				JobID: evt.JobID, TraceType: "status", TraceContent: status,
 			})
 		}),
 		cogito.WithToolCallResultCallback(func(t cogito.ToolStatus) {
 			flushStreamBuf()
-			natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+			publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 				JobID: evt.JobID, TraceType: "tool_result", TraceContent: fmt.Sprintf("%s: %s", t.Name, t.Result),
 			})
 		}),
@@ -140,7 +150,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 			case cogito.StreamEventContent:
 				contentBuf.WriteString(ev.Content)
 			case cogito.StreamEventToolCall:
-				natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
+				publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{
 					JobID: evt.JobID, TraceType: "tool_call", TraceContent: fmt.Sprintf("%s(%s)", ev.ToolName, ev.ToolArgs),
 				})
 			}
@@ -155,7 +165,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 	flushStreamBuf()
 
 	if err != nil {
-		publishTestJobResult(natsClient, evt.JobID, "failed", "", fmt.Sprintf("cogito execution failed: %v", err))
+		publishTestJobResult(pub, evt.JobID, "failed", "", fmt.Sprintf("cogito execution failed: %v", err))
 		return
 	}
 
@@ -163,27 +173,27 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me
 	if msg := f.LastMessage(); msg != nil {
 		result = msg.Content
 	}
-	publishTestJobResult(natsClient, evt.JobID, "completed", result, "")
+	publishTestJobResult(pub, evt.JobID, "completed", result, "")
 }
 
-func publishTestJobStatus(nc *messaging.Client, jobID, status, message string) {
-	nc.Publish(messaging.SubjectJobResult(jobID), jobs.JobResultEvent{
+func publishTestJobStatus(pub messaging.Publisher, jobID, status, message string) {
+	publishTestEvent(pub, messaging.SubjectJobResult(jobID), jobs.JobResultEvent{
 		JobID:  jobID,
 		Status: status,
 	})
-	nc.Publish(messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{
+	publishTestEvent(pub, messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{
 		JobID: jobID, Status: status, Message: message,
 	})
 }
 
-func publishTestJobResult(nc *messaging.Client, jobID, status, result, errMsg string) {
-	nc.Publish(messaging.SubjectJobResult(jobID), jobs.JobResultEvent{
+func publishTestJobResult(pub messaging.Publisher, jobID, status, result, errMsg string) {
+	publishTestEvent(pub, messaging.SubjectJobResult(jobID), jobs.JobResultEvent{
 		JobID:  jobID,
 		Status: status,
 		Result: result,
 		Error:  errMsg,
 	})
-	nc.Publish(messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{
+	publishTestEvent(pub, messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{
 		JobID: jobID, Status: status, Message: errMsg,
 	})
 }

tests/e2e/distributed/mcp_ci_job_test.go

@@ -280,10 +280,18 @@ func startMockLLMServer() (string, func()) {
 var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func() {
 	var (
 		infra *TestInfra
+		bus   messaging.Broadcaster
 	)
 
+	// One carrier per spec, on THIS spec's database. It is the PostgreSQL
+	// LISTEN/NOTIFY carrier the deployment runs, and it is what the SSE routes
+	// read: a spec that published its progress and result lines onto a message
+	// bus instead would keep passing after that carrier had gone silent,
+	// because a publish onto the wrong carrier succeeds and simply arrives
+	// nowhere.
 	BeforeEach(func() {
-		infra = SetupNATSOnly()
+		infra = SetupInfra("mcp_ci_job")
+		bus = infra.Bus()
 	})
 
 	Context("Full MCP CI Job Flow", func() {
@@ -303,7 +311,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			var resultEvent *jobs.JobResultEvent
 			var eventMu sync.Mutex
 
-			progressSub, err := infra.NC.Subscribe(messaging.SubjectJobProgress(jobID), func(data []byte) {
+			progressSub, err := bus.Subscribe(messaging.SubjectJobProgress(jobID), func(data []byte) {
 				var evt jobs.ProgressEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -314,7 +322,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer progressSub.Unsubscribe()
 
-			resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
+			resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
 				var evt jobs.JobResultEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -325,8 +333,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer resultSub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			// Build MCP config YAML pointing to mock MCP server
 			mcpRemoteJSON := fmt.Sprintf(`{"mcpServers":{"weather-api":{"url":"%s"}}}`, mcpURL)
 			modelCfg := &config.ModelConfig{
@@ -364,9 +370,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			raw, err := json.Marshal(evt)
 			Expect(err).ToNot(HaveOccurred())
 
-			FlushNATS(infra.NC)
-
-			go processMCPCIJobForTest(raw, llmURL, "test-token", infra.NC)
+			go processMCPCIJobForTest(raw, llmURL, "test-token", bus)
 
 			// Wait for result
 			Eventually(func() bool {
@@ -404,7 +408,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			var resultEvent *jobs.JobResultEvent
 			var eventMu sync.Mutex
 
-			resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
+			resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
 				var evt jobs.JobResultEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -415,8 +419,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer resultSub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			// MCP config pointing to unreachable server
 			mcpRemoteJSON := `{"mcpServers":{"bad-server":{"url":"http://127.0.0.1:1/mcp"}}}`
 			modelCfg := &config.ModelConfig{
@@ -448,7 +450,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 
 			// Process directly (no worker subscription needed)
 			evtData, _ := json.Marshal(evt)
-			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC)
+			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus)
 
 			// Wait for failure result
 			Eventually(func() bool {
@@ -477,7 +479,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			var resultEvent *jobs.JobResultEvent
 			var eventMu sync.Mutex
 
-			resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
+			resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
 				var evt jobs.JobResultEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -488,8 +490,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer resultSub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			mcpRemoteJSON := fmt.Sprintf(`{"mcpServers":{"weather-api":{"url":"%s"}}}`, mcpURL)
 			modelCfg := &config.ModelConfig{
 				MCP: config.MCPConfig{
@@ -523,7 +523,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			}
 
 			evtData, _ := json.Marshal(evt)
-			go processMCPCIJobForTest(evtData, llmURL, "test-token", infra.NC)
+			go processMCPCIJobForTest(evtData, llmURL, "test-token", bus)
 
 			Eventually(func() bool {
 				eventMu.Lock()
@@ -545,7 +545,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			var resultEvent *jobs.JobResultEvent
 			var eventMu sync.Mutex
 
-			resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
+			resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
 				var evt jobs.JobResultEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -556,8 +556,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer resultSub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			// Event with no Job or Task
 			evt := jobs.JobEvent{
 				JobID:  jobID,
@@ -566,7 +564,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			}
 
 			evtData, _ := json.Marshal(evt)
-			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC)
+			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus)
 
 			Eventually(func() bool {
 				eventMu.Lock()
@@ -586,7 +584,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			var resultEvent *jobs.JobResultEvent
 			var eventMu sync.Mutex
 
-			resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
+			resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) {
 				var evt jobs.JobResultEvent
 				if json.Unmarshal(data, &evt) == nil {
 					eventMu.Lock()
@@ -597,8 +595,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			Expect(err).ToNot(HaveOccurred())
 			defer resultSub.Unsubscribe()
 
-			FlushNATS(infra.NC)
-
 			// ModelConfig with empty MCP
 			modelCfg := &config.ModelConfig{}
 			modelCfg.Name = "no-mcp-model"
@@ -624,7 +620,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func(
 			}
 
 			evtData, _ := json.Marshal(evt)
-			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC)
+			go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus)
 
 			Eventually(func() bool {
 				eventMu.Lock()

tests/e2e/distributed/nats_jwt_helpers_test.go

@@ -1,156 +0,0 @@
-package distributed_test
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"strings"
-	"time"
-
-	"github.com/mudler/LocalAI/core/services/messaging"
-	"github.com/mudler/LocalAI/pkg/natsauth"
-	"github.com/nats-io/jwt/v2"
-	"github.com/nats-io/nkeys"
-
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
-
-	"github.com/testcontainers/testcontainers-go"
-	tcnats "github.com/testcontainers/testcontainers-go/modules/nats"
-)
-
-// JWTTestInfra holds a NATS server configured with JWT auth and minted worker credentials.
-type JWTTestInfra struct {
-	*TestInfra
-	AccountSeed string
-	NodeID      string
-	WorkerJWT   string
-	WorkerSeed  string
-}
-
-// SetupJWTInfra starts NATS with an in-memory JWT resolver and returns worker credentials
-// minted the same way as node registration (pkg/natsauth).
-func SetupJWTInfra() *JWTTestInfra {
-	GinkgoHelper()
-
-	infra := &JWTTestInfra{TestInfra: &TestInfra{Ctx: context.Background()}}
-
-	operatorJWT, accountJWT, accountSeed, err := jwtResolverMaterial()
-	Expect(err).ToNot(HaveOccurred())
-	infra.AccountSeed = accountSeed
-
-	conf := fmt.Sprintf(`listen: 0.0.0.0:4222
-
-operator: %s
-
-resolver: MEMORY
-resolver_preload: {
-	%s: %s
-}
-`, operatorJWT, accountPublicKeyFromSeed(accountSeed), accountJWT)
-
-	var natsContainer *tcnats.NATSContainer
-	// Override default testcontainers -js: JetStream fails without a system account in JWT mode.
-	natsContainer, err = tcnats.Run(infra.Ctx, "nats:2-alpine",
-		tcnats.WithConfigFile(bytes.NewBufferString(conf)),
-		testcontainers.WithCmd("-c", "/etc/nats.conf"),
-	)
-	Expect(err).ToNot(HaveOccurred())
-	infra.NATSContainer = natsContainer
-
-	infra.NatsURL, err = infra.NATSContainer.ConnectionString(infra.Ctx)
-	Expect(err).ToNot(HaveOccurred())
-
-	infra.NodeID = "550e8400-e29b-41d4-a716-446655440000"
-	cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour}
-	infra.WorkerJWT, infra.WorkerSeed, err = cfg.MintWorkerJWT(infra.NodeID, "backend")
-	Expect(err).ToNot(HaveOccurred())
-
-	infra.NC, err = messaging.New(infra.NatsURL, messaging.WithUserJWT(infra.WorkerJWT, infra.WorkerSeed))
-	Expect(err).ToNot(HaveOccurred())
-
-	DeferCleanup(func() {
-		if infra.NC != nil {
-			infra.NC.Close()
-		}
-		if infra.NATSContainer != nil {
-			_ = infra.NATSContainer.Terminate(context.Background())
-		}
-	})
-
-	return infra
-}
-
-// jwtResolverMaterial builds operator + account JWTs for a MEMORY resolver.
-// Follows the NATS JWT tutorial: self-signed account, then operator re-sign, with the
-// account identity key listed as a signing key so MintWorkerJWT can use the account seed.
-func jwtResolverMaterial() (operatorJWT, accountJWT, accountSeed string, err error) {
-	okp, err := nkeys.CreateOperator()
-	if err != nil {
-		return "", "", "", err
-	}
-	opk, err := okp.PublicKey()
-	if err != nil {
-		return "", "", "", err
-	}
-	oc := jwt.NewOperatorClaims(opk)
-	oc.Name = "localai-test-operator"
-	oskp, err := nkeys.CreateOperator()
-	if err != nil {
-		return "", "", "", err
-	}
-	ospk, err := oskp.PublicKey()
-	if err != nil {
-		return "", "", "", err
-	}
-	oc.SigningKeys.Add(ospk)
-	operatorJWT, err = oc.Encode(okp)
-	if err != nil {
-		return "", "", "", err
-	}
-
-	akp, err := nkeys.CreateAccount()
-	if err != nil {
-		return "", "", "", err
-	}
-	seed, err := akp.Seed()
-	if err != nil {
-		return "", "", "", err
-	}
-	accountSeed = string(seed)
-
-	apk, err := akp.PublicKey()
-	if err != nil {
-		return "", "", "", err
-	}
-	ac := jwt.NewAccountClaims(apk)
-	ac.Name = "localai-test-account"
-	ac.SigningKeys.Add(apk)
-	accountJWT, err = ac.Encode(akp)
-	if err != nil {
-		return "", "", "", err
-	}
-	ac, err = jwt.DecodeAccountClaims(accountJWT)
-	if err != nil {
-		return "", "", "", err
-	}
-	accountJWT, err = ac.Encode(oskp)
-	if err != nil {
-		return "", "", "", err
-	}
-	return operatorJWT, accountJWT, accountSeed, nil
-}
-
-func accountPublicKeyFromSeed(accountSeed string) string {
-	akp, err := nkeys.FromSeed([]byte(accountSeed))
-	Expect(err).ToNot(HaveOccurred())
-	pk, err := akp.PublicKey()
-	Expect(err).ToNot(HaveOccurred())
-	return pk
-}
-
-// nodeSubjectPrefix returns the sanitized nodes.* prefix for a node ID.
-func nodeSubjectPrefix(nodeID string) string {
-	tok := strings.NewReplacer(".", "-", "*", "-", ">", "-", " ", "-", "\t", "-", "\n", "-").Replace(nodeID)
-	return "nodes." + tok
-}

tests/e2e/distributed/nats_jwt_test.go

@@ -1,176 +0,0 @@
-package distributed_test
-
-import (
-	"time"
-
-	"github.com/mudler/LocalAI/core/services/messaging"
-	"github.com/mudler/LocalAI/pkg/natsauth"
-
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
-)
-
-var _ = Describe("NATS JWT Auth", Label("Distributed", "NatsJWT"), func() {
-	var infra *JWTTestInfra
-
-	BeforeEach(func() {
-		infra = SetupJWTInfra()
-	})
-
-	It("connects with a minted backend worker JWT and publishes on its one remaining allowed subject", func() {
-		// A backend worker's whole grant is `_INBOX.>` now, on both sides.
-		// Every verb a frontend gives it, file staging included, is an HTTP
-		// route on its tunnel, and it no longer opens a bus connection at all;
-		// the JWT is minted and unused. See pkg/natsauth.WorkerPermissions.
-		Expect(infra.NC.Publish("_INBOX.probe", map[string]string{"path": "/tmp/model"})).To(Succeed())
-		// ConfirmRoundTrip is the flush AND the server's verdict in one call.
-		// Read separately they were two assertions that could drift apart; the
-		// verdict is the one that matters, because a permission violation does
-		// not close the connection.
-		Expect(infra.NC.ConfirmRoundTrip(2 * time.Second)).To(Succeed())
-		Expect(infra.NC.IsConnected()).To(BeTrue())
-	})
-
-	It("denies a backend worker the file-staging subjects it no longer serves", func() {
-		// This spec used to assert the OPPOSITE, and kept passing after the
-		// grant was deleted. A NATS permission violation does not close the
-		// connection, so a spec that checks only FlushTimeout and IsConnected
-		// cannot tell an allowed publish from a denied one; LastError is what
-		// actually reads the server's verdict, which is why the sibling below
-		// has always used it.
-		subject := nodeSubjectPrefix(infra.NodeID) + ".files.stage"
-		Expect(infra.NC.Publish(subject, map[string]string{"path": "/tmp/model"})).To(Succeed())
-		Eventually(func() error {
-			return infra.NC.ConfirmRoundTrip(500 * time.Millisecond)
-		}, "3s", "50ms").Should(HaveOccurred())
-	})
-
-	It("denies backend subscribe on the node prefix it no longer listens to", func() {
-		// The node subtree was granted while a backend worker still held a
-		// connection with nothing under it subscribed. It does not hold one at
-		// all now, so the grant went too; asserting the denial is what would
-		// catch a subject quietly coming back to the bus.
-		wild := nodeSubjectPrefix(infra.NodeID) + ".>"
-		sub, err := infra.NC.Subscribe(wild, func(_ []byte) {})
-		if err == nil {
-			defer func() { _ = sub.Unsubscribe() }()
-			Eventually(func() error {
-				return infra.NC.ConfirmRoundTrip(500 * time.Millisecond)
-			}, "3s", "50ms").Should(HaveOccurred())
-		}
-	})
-
-	It("rejects anonymous publish on the JWT-enabled server", func() {
-		anon, err := messaging.New(infra.NatsURL)
-		Expect(err).ToNot(HaveOccurred())
-		defer anon.Close()
-
-		err = anon.Publish("nodes.any.files.x", map[string]string{"x": "1"})
-		Expect(err).ToNot(HaveOccurred())
-		Expect(anon.ConfirmRoundTrip(2 * time.Second)).To(HaveOccurred())
-	})
-
-	It("denies backend publish to another node's subjects", func() {
-		other := nodeSubjectPrefix("other-node-id") + ".files.stage"
-		Expect(infra.NC.Publish(other, map[string]string{"stage": "nope"})).To(Succeed())
-		Eventually(func() error {
-			return infra.NC.ConfirmRoundTrip(500 * time.Millisecond)
-		}, "3s", "50ms").Should(HaveOccurred())
-	})
-
-	It("mints agent JWT without backend.install in claims", func() {
-		cfg := natsauth.Config{AccountSeed: infra.AccountSeed}
-		token, _, err := cfg.MintWorkerJWT("agent-node-1", "agent")
-		Expect(err).ToNot(HaveOccurred())
-
-		claims, err := natsauth.DecodeUserClaims(token)
-		Expect(err).ToNot(HaveOccurred())
-		// agent.execute has left this list: agent execution is a streaming
-		// control verb on the tunnel now, driven by a claim a frontend replica
-		// took off the job store. What must still be here is the cancel
-		// broadcast, which cannot become an RPC.
-		Expect(claims.Permissions.Sub.Allow).To(ContainElement("agent.*.cancel"))
-		Expect(claims.Permissions.Sub.Allow).ToNot(ContainElement("agent.execute"))
-		for _, subj := range claims.Permissions.Sub.Allow {
-			Expect(subj).NotTo(ContainSubstring("backend.install"))
-		}
-	})
-
-	// Regression guard for the silent permission gaps: decoding the JWT claims
-	// (above) only proves the agent JWT is *restrictive*, not that it is
-	// *sufficient*. Stand a real agent connection up against the enforcing
-	// server and exercise every subscription core/cli/agent_worker.go actually
-	// makes — a denied SUB now surfaces synchronously via confirmSubscription,
-	// so a missing allow rule fails this test instead of silently dropping
-	// backend.stop / MCP-CI deliveries at runtime.
-	It("lets an agent-minted JWT establish all the subscriptions the agent worker uses", func() {
-		const nodeID = "agent-node-subs"
-		cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour}
-		token, seed, err := cfg.MintWorkerJWT(nodeID, "agent")
-		Expect(err).ToNot(HaveOccurred())
-
-		nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed))
-		Expect(err).ToNot(HaveOccurred())
-		DeferCleanup(nc.Close)
-
-		// Mirror core/cli/agent_worker.go exactly. MCP tool execution and
-		// discovery are absent, so is the per-node backend.stop, and so now are
-		// agent execution and MCP CI runs: none of them is a bus subject any
-		// more. The frontend selects an agent worker itself and reaches it with
-		// a control RPC over the tunnel that worker holds, and the work it
-		// hands over is a row it claimed on the job store.
-		//
-		// What is left is the cancel broadcast, which cannot become an RPC: the
-		// replica holding a run is not the one an API cancel lands on.
-		_, err = nc.Subscribe(messaging.SubjectAgentCancelWildcard, func([]byte) {})
-		Expect(err).ToNot(HaveOccurred(), "agent JWT must allow %s (cancellation)", messaging.SubjectAgentCancelWildcard)
-
-		_, err = nc.Subscribe(messaging.SubjectJobProgressWildcard, func([]byte) {})
-		Expect(err).ToNot(HaveOccurred(), "agent JWT must allow %s (progress bridging)", messaging.SubjectJobProgressWildcard)
-	})
-
-	// The narrowing, proved against the enforcing server rather than against
-	// the allow list that feeds it. The subject is written out by hand because
-	// its builder is deleted; that literal is what a worker from an older
-	// release would still send, and this is what the server now answers it.
-	//
-	// It is a narrowing and not a lockout: the two subscriptions above are made
-	// on a JWT minted the same way and both succeed, so the list this trims is
-	// demonstrably not the empty one NATS would read as unrestricted.
-	It("refuses an agent-minted JWT the retired per-node backend.stop subject", func() {
-		const nodeID = "agent-node-stop"
-		cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour}
-		token, seed, err := cfg.MintWorkerJWT(nodeID, "agent")
-		Expect(err).ToNot(HaveOccurred())
-
-		nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed))
-		Expect(err).ToNot(HaveOccurred())
-		DeferCleanup(nc.Close)
-
-		_, err = nc.Subscribe("nodes."+nodeID+".backend.stop", func([]byte) {})
-		Expect(err).To(HaveOccurred(),
-			"backend.stop is a control RPC on the worker's tunnel; the bus must not carry it")
-	})
-
-	// The same narrowing for the two queue subjects that became claim rows.
-	// Written out by hand for the same reason: those literals are what a worker
-	// from an older release would still subscribe to, and this is what the
-	// server now answers it.
-	DescribeTable("refuses an agent-minted JWT a retired queue subject",
-		func(subject string) {
-			cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour}
-			token, seed, err := cfg.MintWorkerJWT("agent-node-queues", "agent")
-			Expect(err).ToNot(HaveOccurred())
-
-			nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed))
-			Expect(err).ToNot(HaveOccurred())
-			DeferCleanup(nc.Close)
-
-			_, err = nc.Subscribe(subject, func([]byte) {})
-			Expect(err).To(HaveOccurred(),
-				"%s became a claim on the job store; the bus must not carry it", subject)
-		},
-		Entry("agent execution", "agent.execute"),
-		Entry("mcp ci jobs", "jobs.mcp-ci.new"),
-	)
-})

tests/e2e/distributed/testhelpers_test.go

@@ -8,14 +8,12 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/pgbus"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	"github.com/testcontainers/testcontainers-go"
-	tcnats "github.com/testcontainers/testcontainers-go/modules/nats"
 	tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres"
 	"github.com/testcontainers/testcontainers-go/wait"
 	"gorm.io/driver/postgres"
@@ -23,37 +21,44 @@ import (
 	gormlogger "gorm.io/gorm/logger"
 )
 
-// TestInfra holds shared test containers and connection strings.
+// TestInfra holds the shared test container and the connection strings derived
+// from it.
 //
-// PGContainer and NATSContainer are the SUITE-WIDE containers, shared by every
-// spec. Never call Terminate or Stop on them from a spec: it ends the run for
-// everything after it. They are exposed only because nats_jwt_helpers_test.go
-// builds its own TestInfra around a dedicated NATS container.
+// PGContainer is the SUITE-WIDE container, shared by every spec. Never call
+// Terminate or Stop on it from a spec: it ends the run for everything after it.
+//
+// There is one container and not two. Every carrier these specs exercise is a
+// PostgreSQL LISTEN/NOTIFY channel and every worker verb is an HTTP route on a
+// tunnel, so a message broker in this suite would be infrastructure no spec and
+// no production path can reach.
 type TestInfra struct {
-	Ctx           context.Context
-	PGContainer   *tcpostgres.PostgresContainer
-	NATSContainer *tcnats.NATSContainer
-	PGURL         string
-	NatsURL       string
-	NC            *messaging.Client
+	Ctx         context.Context
+	PGContainer *tcpostgres.PostgresContainer
+	PGURL       string
 }
 
-// Containers are suite-scoped, not spec-scoped. Starting a Postgres (~10s) and a
-// NATS (~3.5s) per spec cost roughly 48 minutes of pure startup across the 213
-// specs behind SetupInfra, which is why this suite was never wired into CI.
-// Isolation now comes from a database per spec (~67ms), which is what the dbName
-// argument was always describing.
+// staleBusURL is the address of a broker that is not running, and is not meant
+// to be. LOCALAI_NATS_URL and --nats-url are still accepted and ignored so an
+// operator's existing command line, unit file or Helm values file starts
+// unchanged after the broker is shut down; specs that exercise that promise
+// hand over THIS value, because a value pointing at a live server would let a
+// regression that dialled it pass unnoticed.
+const staleBusURL = "nats://127.0.0.1:1"
+
+// The container is suite-scoped, not spec-scoped. Starting a Postgres (~10s) per
+// spec cost roughly 36 minutes of pure startup across the 213 specs behind
+// SetupInfra, which is why this suite was never wired into CI. Isolation now
+// comes from a database per spec (~67ms), which is what the dbName argument was
+// always describing.
 //
 // Plain BeforeSuite rather than SynchronizedBeforeSuite is deliberate: under
-// `ginkgo -p` each process gets its own container pair, which keeps NATS subjects
-// isolated per process. A single shared NATS across parallel processes would let
-// specs on different processes see each other's messages on the same subject.
+// `ginkgo -p` each process gets its own container, and a database per spec on
+// top of that keeps two processes from reading each other's notifications on a
+// channel of the same name.
 var (
-	suitePG      *tcpostgres.PostgresContainer
-	suiteNATS    *tcnats.NATSContainer
-	suitePGDSN   string
-	suiteNatsURL string
-	dbCounter    atomic.Int64
+	suitePG    *tcpostgres.PostgresContainer
+	suitePGDSN string
+	dbCounter  atomic.Int64
 )
 
 var _ = BeforeSuite(func() {
@@ -74,12 +79,6 @@ var _ = BeforeSuite(func() {
 
 	suitePGDSN, err = suitePG.ConnectionString(ctx, "sslmode=disable")
 	Expect(err).ToNot(HaveOccurred())
-
-	suiteNATS, err = tcnats.Run(ctx, "nats:2-alpine")
-	Expect(err).ToNot(HaveOccurred())
-
-	suiteNatsURL, err = suiteNATS.ConnectionString(ctx)
-	Expect(err).ToNot(HaveOccurred())
 })
 
 var _ = AfterSuite(func() {
@@ -87,9 +86,6 @@ var _ = AfterSuite(func() {
 	if suitePG != nil {
 		_ = suitePG.Terminate(ctx)
 	}
-	if suiteNATS != nil {
-		_ = suiteNATS.Terminate(ctx)
-	}
 })
 
 // sanitizeDBName maps a spec-supplied label onto a legal unquoted Postgres
@@ -156,18 +152,16 @@ func closeDB(db *gorm.DB) {
 	}
 }
 
-// SetupInfra provisions a dedicated database on the suite-scoped Postgres and
-// returns a client connected to the suite-scoped NATS. Call in BeforeEach;
-// cleanup is registered with DeferCleanup.
+// SetupInfra provisions a dedicated database on the suite-scoped Postgres. Call
+// in BeforeEach; cleanup is registered with DeferCleanup. A spec that needs a
+// broadcast carrier opens one with Bus().
 func SetupInfra(dbName string) *TestInfra {
 	GinkgoHelper()
-	Expect(suitePG).ToNot(BeNil(), "SetupInfra called before BeforeSuite started the shared containers")
+	Expect(suitePG).ToNot(BeNil(), "SetupInfra called before BeforeSuite started the shared container")
 
 	infra := &TestInfra{
-		Ctx:           context.Background(),
-		PGContainer:   suitePG,
-		NATSContainer: suiteNATS,
-		NatsURL:       suiteNatsURL,
+		Ctx:         context.Background(),
+		PGContainer: suitePG,
 	}
 
 	db := fmt.Sprintf("%s_%d", sanitizeDBName(dbName), dbCounter.Add(1))
@@ -180,12 +174,9 @@ func SetupInfra(dbName string) *TestInfra {
 		Expect(admin.Exec(fmt.Sprintf("CREATE DATABASE %q", db)).Error).To(Succeed())
 	}()
 
-	// Registered before anything else can fail: a NATS connect error below would
-	// otherwise leave the database behind for the rest of the suite.
+	// Registered immediately after the CREATE, so no later failure in this
+	// helper can leave the database behind for the rest of the suite.
 	DeferCleanup(func() {
-		if infra.NC != nil {
-			infra.NC.Close()
-		}
 		drop, err := tryAdminDB()
 		if err != nil {
 			AddReportEntry("drop database skipped", fmt.Sprintf("%s: %v", db, err))
@@ -200,63 +191,23 @@ func SetupInfra(dbName string) *TestInfra {
 
 	infra.PGURL = replaceDBName(suitePGDSN, db)
 
-	var err error
-	infra.NC, err = messaging.New(infra.NatsURL)
-	Expect(err).ToNot(HaveOccurred())
-
 	return infra
 }
 
-// SetupNATSOnly returns a client on the suite-scoped NATS for specs that need no
-// database.
-func SetupNATSOnly() *TestInfra {
-	GinkgoHelper()
-	Expect(suiteNATS).ToNot(BeNil(), "SetupNATSOnly called before BeforeSuite started the shared containers")
-
-	infra := &TestInfra{
-		Ctx:           context.Background(),
-		NATSContainer: suiteNATS,
-		NatsURL:       suiteNatsURL,
-	}
-
-	var err error
-	infra.NC, err = messaging.New(infra.NatsURL)
-	Expect(err).ToNot(HaveOccurred())
-
-	DeferCleanup(func() {
-		if infra.NC != nil {
-			infra.NC.Close()
-		}
-	})
-
-	return infra
-}
-
-// FlushNATS ensures all subscriptions are registered server-side before publishing.
-//
-// It asserts the server's verdict too, not only that the round trip completed:
-// on a permission-enforcing server a denied SUB leaves the connection open and
-// the flush succeeding, so a helper that checked the flush alone would let a
-// spec proceed to publish into a subscription the server had already refused.
-func FlushNATS(nc *messaging.Client) {
-	GinkgoHelper()
-	Expect(nc.ConfirmRoundTrip(5 * time.Second)).To(Succeed())
-}
-
 // Bus opens a broadcast carrier on THIS spec's database.
 //
 // It is the carrier the job, agent and response families travel on, and it is
 // what these specs must build their dispatchers and bridges with. Publishing on
 // one carrier while the subscriber reads another is a defect with no error
 // anywhere: the publish succeeds and the SSE stream is simply empty, so a spec
-// that used the NATS client here would keep passing after production had gone
-// silent.
+// that reached for a message-bus client here would keep passing after
+// production had gone silent.
 //
 // Every call returns a SEPARATE carrier on the same database, so a spec can
 // build two and assert across them, which is the shape a deployment has.
 func (i *TestInfra) Bus() *pgbus.Bus {
 	GinkgoHelper()
-	Expect(i.PGURL).ToNot(BeEmpty(), "Bus needs a database; use SetupInfra rather than SetupNATSOnly")
+	Expect(i.PGURL).ToNot(BeEmpty(), "Bus needs a database; call SetupInfra first")
 
 	db, err := gorm.Open(postgres.Open(i.PGURL), &gorm.Config{Logger: gormlogger.Default.LogMode(gormlogger.Silent)})
 	Expect(err).ToNot(HaveOccurred())