feat(distributed): switch the cluster transport between NATS and the database at runtime

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

@@ -0,0 +1,97 @@
+---
+name: 'E2E Distributed Tests'
+
+on:
+  pull_request:
+    # The suite's dependency graph is large, so an allowlist of paths would
+    # silently stop guarding the moment code moves. It is cheap enough to run
+    # unless the diff is confined to paths it provably cannot reach.
+    # See .agents/ci-caching.md.
+    paths-ignore:
+      - 'gallery/**'
+      - 'docs/**'
+      - 'examples/**'
+      - '**/*.md'
+  push:
+    branches:
+      - master
+
+concurrency:
+  group: ci-tests-e2e-distributed-${{ github.event.pull_request.number || github.sha }}-${{ github.repository }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+jobs:
+  tests-e2e-distributed:
+    runs-on: ubuntu-latest
+    # Advisory: a failure is a visible red X on the pull request, and the job is
+    # deliberately not in branch protection. Promoting it to a required check
+    # is a repository-settings change, to be made once it has a track record.
+    continue-on-error: true
+    timeout-minutes: 45
+    steps:
+      - name: Clone
+        uses: actions/checkout@v7
+        with:
+          submodules: true
+      - name: Configure apt mirror on runner
+        uses: ./.github/actions/configure-apt-mirror
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.26.x'
+          cache: false
+      - name: Display Go version
+        run: go version
+      - name: Dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential libopus-dev
+      - name: Proto Dependencies
+        run: |
+          curl -L -s https://github.com/protocolbuffers/protobuf/releases/download/v26.1/protoc-26.1-linux-x86_64.zip -o protoc.zip && \
+          unzip -j -d /usr/local/bin protoc.zip bin/protoc && \
+          rm protoc.zip
+          go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.34.2
+          go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@1958fcbe2ca8bd93af633f11e97d44e567e945af
+          PATH="$PATH:$HOME/go/bin" make protogen-go
+      - name: Pre-pull test images
+        # The suite starts its own PostgreSQL and NATS servers with
+        # testcontainers. Pulling the images here keeps the pull time out of the
+        # spec timeouts and makes a registry outage read as a setup failure
+        # instead of a test failure.
+        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
+        # this runner is ephemeral and every container dies with the VM.
+        # Leaving it enabled would pull one more unpinned image mid-suite.
+        env:
+          TESTCONTAINERS_RYUK_DISABLED: "true"
+        run: |
+          PATH="$PATH:$HOME/go/bin" make test-e2e-distributed
+      - name: Distributed E2E, switch of the carrier
+        # Real processes: two frontends and two workers change the carrier
+        # between NATS and the database tunnel while a model serves requests.
+        # It runs apart from the suite above so that a failure names the
+        # switch and the two logs stay short. Like the suite, it needs only the
+        # public PostgreSQL and NATS images and no secret, so a pull request
+        # from a fork runs it the same way.
+        if: ${{ !cancelled() }}
+        env:
+          TESTCONTAINERS_RYUK_DISABLED: "true"
+        run: |
+          PATH="$PATH:$HOME/go/bin" make test-e2e-distributed-switch
+      # tmate keeps the runner busy until the 6 hour job limit, so a single
+      # failure costs a whole runner slot. Only open a session when someone
+      # asked for one by labelling the pull request `ci-debug`, and cap the
+      # session so a forgotten label cannot idle a runner either.
+      - name: Setup tmate session if tests fail
+        if: ${{ failure() && contains(github.event.pull_request.labels.*.name, 'ci-debug') }}
+        timeout-minutes: 30
+        uses: mxschmitt/action-tmate@v3.23
+        with:
+          detached: true
+          connect-timeout-seconds: 180
+          limit-access-to-actor: true

Makefile

@@ -349,6 +349,18 @@ test-e2e-distributed: protogen-go
 	@echo 'Running distributed e2e tests (label Distributed, incl. NatsJWT)'
 	$(GOCMD) run github.com/onsi/ginkgo/v2/ginkgo --label-filter='Distributed && !VLLMMultinode' --flake-attempts $(TEST_FLAKES) -v -r ./tests/e2e/distributed
 
+# Carrier switch with real processes. Two frontends and two workers (one
+# dual-capable, one tunnel-only) run as child processes against PostgreSQL and
+# NATS containers, and the carrier is changed from NATS to the tunnel and back.
+# The spec carries the CarrierSwitch label and not Distributed, so
+# test-e2e-distributed above does not run it: it starts many processes and takes
+# a few minutes. It builds local-ai when LOCALAI_E2E_BIN is not set, and skips
+# when docker is missing.
+test-e2e-distributed-switch: protogen-go build-mock-backend
+	@echo 'Running the carrier switch e2e (label CarrierSwitch)'
+	LOCALAI_E2E_MOCK_BACKEND=$(abspath tests/e2e/mock-backend/mock-backend) \
+	$(GOCMD) run github.com/onsi/ginkgo/v2/ginkgo --label-filter='CarrierSwitch' --flake-attempts $(TEST_FLAKES) -v -r ./tests/e2e/distributed
+
 # vLLM multi-node DP smoke (CPU). Builds local-ai:tests and the
 # cpu-vllm backend from the current working tree, then drives a
 # head + headless follower via testcontainers-go and asserts a chat

core/application/agent_jobs.go

@@ -38,7 +38,7 @@ func (a *Application) RestartAgentJobService() error {
 			agentJobService.SetDistributedJobStore(d.JobStore)
 		}
 		// Keep agent tasks consistent across replicas (same client the dispatcher uses).
-		agentJobService.SetTaskSyncNATS(d.Nats)
+		agentJobService.SetTaskSyncNATS(d.Broadcaster)
 	}
 
 	// Start the service

core/application/application.go

@@ -611,6 +611,9 @@ func (a *Application) start() error {
 		if a.distributed != nil && a.distributed.Unloader != nil {
 			assistantClient.LoadStopper = a.distributed.Unloader
 		}
+		if a.distributed != nil && a.distributed.Switch != nil {
+			assistantClient.Carrier = a.distributed.Switch
+		}
 		// PII filter — same nil-or-real wiring.
 		assistantClient.PIIRedactor = a.piiRedactor
 		assistantClient.PIIEvents = a.piiEvents
@@ -670,8 +673,8 @@ func (a *Application) agentPoolOptions() agentpool.AgentPoolOptions {
 		opts.WorkQueue = d.WorkQueue
 		opts.EventBridge = d.AgentBridge
 		opts.AgentStore = d.AgentStore
-		if d.Nats != nil {
-			opts.Bus = d.Nats
+		if d.Broadcaster != nil {
+			opts.Bus = d.Broadcaster
 		}
 	}
 	return opts
@@ -723,7 +726,7 @@ func (a *Application) StartAgentPool() {
 	}
 	// Keep per-user agent tasks consistent across replicas (nil in standalone).
 	if d := a.Distributed(); d != nil {
-		usm.SetJobSyncNATS(d.Nats)
+		usm.SetJobSyncNATS(d.Broadcaster)
 	}
 	aps.SetUserServicesManager(usm)