test(distributed): prove phase 3 end to end, under real processes

Five cluster specs that run the binaries an operator runs, plus the repair
of eighteen specs phase 2 left red.

The eighteen were router_tracking and full_flow, failing since 1cf847f29 on
"reported backend installed but named no address for the process". Two
contracts had changed under them: an install reply that names no
worker-local address is refused rather than substituted, and a frontend with
no worker dialer reaches no backend at all. Nobody noticed for a phase
because phase 2 verified with --label-filter='Cluster', which excludes both
suites. ServeBackendLifecycle and tunnelBackendClients state both facts once
for every spec.

The transport double is the part that matters. It translates a refused
connect into cluster.ErrStreamTargetUnavailable, which is what a real worker
answers when its backend process has died and what IsWorkerAnswer lets a
reap guard act on. A bare ECONNREFUSED reaches those guards as "no route" and
reaps nothing, so a double returning the raw syscall error could never fail
the way production fails; putting it back reddens the stale-record spec and
nothing else.

The new specs cover: a backend worker with no bus URL in its /proc environ
registering, being scheduled onto and serving inference; a backend install
and a backend listing driven through the replica that does NOT own the
worker, with the owner read through the production Owner query and re-read
after; that install's progress proven to arrive before its terminal reply,
made deterministic by a gallery server that holds the worker's fetch open so
a reply cannot exist yet; a worker whose tunnel is genuinely gone, waited for
rather than assumed, losing nothing inside the reconnect grace and re-homing
after; a heartbeating worker with a permanently dead tunnel losing its
healthy status while an agent worker in the same cluster keeps it; and the
suite's negative control, where a control RPC to a tunnel-less worker fails
naming the missing route, reaps nothing, and succeeds the moment the tunnel
returns.

Every scenario was attacked. The churn one was WRONG on the first attempt
and only the attack found it: its hold window sat entirely inside
cluster.InstanceLiveness, so a killed replica still read as a live owner
throughout, presence was "connected", and the spec passed with the reconnect
grace set to a nanosecond. It now blocks the tunnel before the kill and waits
for the ownership row to actually empty. Attacks that redden the rest:
posting at the owner, writing the install reply before the work, collapsing
PresenceReconnecting into PresenceGone, removing the non-backend node-type
guard, and not blocking the tunnel. Agent workers turn out to be protected
twice over; no single mutation reaches them.

Harness: Options.AgentWorkers and Options.ReconnectGrace, WorkerEnviron
(read from /proc, because Cmd.Env is the harness agreeing with itself),
NatsURL, FrontendBackendsDir, AgentWorkerName, PostJSON, and a node String()
so a failing roster assertion is readable instead of several hundred bytes
rendered as numbers.

Budget: 20 specs at 787 to 808 seconds over three runs, up from phase 2's 591
to 612. --timeout goes to 30m so a loaded runner reports a cause rather than
a spec name.

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

Makefile

@@ -378,9 +378,21 @@ test-e2e-distributed: protogen-go
 # --flake-attempts is pinned to 1 rather than $(DISTRIBUTED_TEST_FLAKES), and
 # should stay there: this suite exists to catch nondeterministic cluster
 # behaviour, and a retry turns exactly that signal into a green run.
+#
+# Budget: 20 specs, measured at 800 to 830 seconds of Ginkgo time (13 to 14
+# minutes wall including the compile) on a fast developer box. It was 591 to 612
+# seconds before the phase 3 control-plane specs; those five added roughly 200
+# seconds, most of it in the two that wait out real windows rather than poll for
+# a state change (cluster.InstanceLiveness is 30s, and a departed worker cannot
+# be demoted before its reconnect grace).
+#
+# --timeout is 30m rather than 20m because of that. The margin is not slack: a
+# Ginkgo timeout kills the suite mid-spec and reports a spec name rather than a
+# cause, and 20m on a loaded CI runner was one slow health tick away from
+# turning a green suite into an unreadable red one.
 test-e2e-cluster: protogen-go build-mock-backend
 	@echo 'Running cluster e2e tests (label Cluster, real local-ai processes)'
-	$(GOCMD) run github.com/onsi/ginkgo/v2/ginkgo --label-filter='Cluster' --fail-on-empty --flake-attempts 1 --timeout=20m -v ./tests/e2e/distributed
+	$(GOCMD) run github.com/onsi/ginkgo/v2/ginkgo --label-filter='Cluster' --fail-on-empty --flake-attempts 1 --timeout=30m -v ./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

tests/e2e/distributed/cluster/admin.go

@@ -128,6 +128,38 @@ func (c *Cluster) GetJSON(client *http.Client, frontend int, path string, out an
 	return nil
 }
 
+// PostJSON performs an authenticated POST against a frontend and decodes the
+// body, reporting the status it got so a caller can act on it.
+//
+// It returns the status rather than requiring 200 the way GetJSON does,
+// because the control-plane endpoints a spec drives through here answer 202
+// (an install was accepted and runs asynchronously) and the failure cases a
+// negative control is about are statuses, not transport errors. out may be nil
+// for a caller that only wants the status.
+func (c *Cluster) PostJSON(client *http.Client, frontend int, path string, body any, out any) (int, error) {
+	base, err := c.frontendBaseURL(frontend)
+	if err != nil {
+		return 0, err
+	}
+	encoded, err := json.Marshal(body)
+	if err != nil {
+		return 0, fmt.Errorf("marshalling the body for %s: %w", path, err)
+	}
+	resp, err := client.Post(base+path, "application/json", bytes.NewReader(encoded))
+	if err != nil {
+		return 0, fmt.Errorf("POST %s on frontend %d: %w", path, frontend, err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if out == nil {
+		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, bodyExcerptLimit))
+		return resp.StatusCode, nil
+	}
+	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
+		return resp.StatusCode, fmt.Errorf("decoding the %s response from frontend %d (status %d): %w", path, frontend, resp.StatusCode, err)
+	}
+	return resp.StatusCode, nil
+}
+
 // frontendBaseURL validates the index before FrontendURL indexes the slice: a
 // bare index panic in a helper every failover spec calls is far harder to read
 // than a named error.

tests/e2e/distributed/cluster/cluster.go

@@ -15,6 +15,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/mudler/LocalAI/pkg/httpclient"
@@ -43,6 +44,35 @@ type Options struct {
 	Frontends int
 	Workers   int
 
+	// AgentWorkers is how many `local-ai agent-worker` processes to start
+	// alongside the backend workers.
+	//
+	// They exist so a spec can hold the two kinds of worker side by side in one
+	// cluster. An agent worker still speaks NATS and holds no tunnel at all,
+	// which is exactly the shape the tunnel-departure rules must not act on: it
+	// has no node_connections row, so its presence is PresenceUnknown forever.
+	// A spec that asserted only on backend workers could not tell "agent
+	// workers are unaffected" from "nothing here looks at them".
+	//
+	// They register through the same WorkerFrontendURL hook as backend workers,
+	// so a spec that puts a balancer in front of the fleet gets one for its
+	// agent workers too. Without that, killing a replica would orphan the agent
+	// worker's heartbeats and it would go unhealthy for a reason that has
+	// nothing to do with what the spec is about.
+	AgentWorkers int
+
+	// ReconnectGrace sets LOCALAI_WORKER_RECONNECT_GRACE on every frontend: how
+	// long a worker whose tunnel was lost is read as reconnecting rather than
+	// gone. Zero leaves the binary's own default (90s).
+	//
+	// It is an Option rather than a per-spec environment edit because two
+	// specs need it pulled in OPPOSITE directions and neither can use the
+	// default: one has to prove nothing is reaped inside the window and needs
+	// it longer than a re-home takes, and one has to prove a departed worker
+	// stops being reported healthy and would otherwise wait 90 seconds to say
+	// so.
+	ReconnectGrace time.Duration
+
 	// SpreadWorkerRegistrations sends worker i to frontend i%Frontends instead
 	// of sending every worker to frontend 0.
 	//
@@ -113,7 +143,11 @@ type Cluster struct {
 	opts      Options
 	frontends []*Process
 	workers   []*Process
-	baseDir   string
+	// agentWorkers are kept apart from workers rather than appended to it. Every
+	// index-taking method on this type means "backend worker i", and folding the
+	// two together would silently renumber them for every existing spec.
+	agentWorkers []*Process
+	baseDir      string
 }
 
 const (
@@ -185,6 +219,14 @@ func Start(opts Options) (*Cluster, error) {
 		}
 		c.workers = append(c.workers, p)
 	}
+	for i := 0; i < opts.AgentWorkers; i++ {
+		p, err := c.startAgentWorker(i)
+		if err != nil {
+			c.Stop()
+			return nil, err
+		}
+		c.agentWorkers = append(c.agentWorkers, p)
+	}
 	return c, nil
 }
 
@@ -258,6 +300,9 @@ func (c *Cluster) startFrontend(i int, port int) (*Process, error) {
 		"LOCALAI_AUTO_APPROVE_NODES=true",
 		"DEBUG=true",
 	)
+	if c.opts.ReconnectGrace > 0 {
+		cmd.Env = append(cmd.Env, "LOCALAI_WORKER_RECONNECT_GRACE="+c.opts.ReconnectGrace.String())
+	}
 
 	p, err := c.spawn(name, cmd, port)
 	if err != nil {
@@ -346,6 +391,77 @@ func (c *Cluster) startWorker(i int) (*Process, error) {
 	return c.spawn(name, cmd, grpcPort)
 }
 
+// startAgentWorker starts agent worker i.
+//
+// It is `local-ai agent-worker`, not `local-ai worker`, and the difference is
+// the whole point of having it here: an agent worker REQUIRES a NATS URL, dials
+// no tunnel, and runs no backend, so it is the control for every rule this
+// phase added about a worker whose tunnel is gone. It binds nothing, so there
+// is no port to reserve and no readiness endpoint to wait on; a spec learns it
+// is up by finding it in the roster.
+func (c *Cluster) startAgentWorker(i int) (*Process, error) {
+	name := agentWorkerName(i)
+	cmd := exec.Command(c.opts.Binary, "agent-worker")
+	cmd.Env = append(cmd.Environ(),
+		// The bus, which is what makes this worker the control: a backend
+		// worker in this same cluster is given none.
+		"LOCALAI_NATS_URL="+c.opts.NatsURL,
+		"LOCALAI_REGISTER_TO="+c.workerFrontendURL(i),
+		"LOCALAI_NODE_NAME="+name,
+		"LOCALAI_REGISTRATION_TOKEN="+c.opts.RegistrationToken,
+		"DEBUG=true",
+	)
+	return c.spawn(name, cmd, 0)
+}
+
+// WorkerEnviron is the environment of worker i's RUNNING PROCESS, read from
+// /proc.
+//
+// Not Cmd.Env, deliberately. A spec asserting that a worker runs with no bus
+// URL is asserting about the process, and Cmd.Env is the harness telling the
+// spec what the harness meant to do: the two agree by construction, so a spec
+// reading it proves the harness consistent with itself and nothing about the
+// binary. /proc/<pid>/environ is what the kernel handed the process.
+//
+// Linux only, which this package already is (it signals with syscall.SIGKILL
+// and reserves ports by binding loopback). A platform without /proc returns the
+// read error rather than falling back to Cmd.Env, so the assertion fails loudly
+// instead of quietly becoming the weaker one.
+func (c *Cluster) WorkerEnviron(i int) ([]string, error) {
+	if err := c.checkWorkerIndex(i); err != nil {
+		return nil, err
+	}
+	p := c.workers[i]
+	if p == nil || p.Cmd == nil || p.Cmd.Process == nil {
+		return nil, fmt.Errorf("worker %d is not running", i)
+	}
+	raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/environ", p.Cmd.Process.Pid))
+	if err != nil {
+		return nil, fmt.Errorf("reading the environment of %s from /proc: %w", p.Name, err)
+	}
+	// NUL separated, with a trailing NUL on a non-empty environment.
+	entries := strings.Split(string(raw), "\x00")
+	out := make([]string, 0, len(entries))
+	for _, e := range entries {
+		if e != "" {
+			out = append(out, e)
+		}
+	}
+	return out, nil
+}
+
+// AgentWorkerName is the node name agent worker i registered under.
+func (c *Cluster) AgentWorkerName(i int) string {
+	if i < 0 || i >= len(c.agentWorkers) {
+		return ""
+	}
+	return c.agentWorkers[i].Name
+}
+
+func agentWorkerName(i int) string {
+	return fmt.Sprintf("agent-worker-%d", i)
+}
+
 const (
 	// workerPortBlockSize is how many ports one worker reserves: one for its
 	// HTTP file-transfer server and the rest for backend processes. A spec that
@@ -488,6 +604,30 @@ func (c *Cluster) RegistrationToken() string {
 	return c.opts.RegistrationToken
 }
 
+// FrontendBackendsDir is the directory frontend i installs its OWN backends
+// into.
+//
+// It is exported for one assertion, and a filesystem one rather than an API
+// one: a spec proving a node backend listing came from the WORKER has to show
+// the frontend that answered does not have that backend itself, and the
+// /backends endpoint cannot say so, because in distributed mode it reports the
+// cluster's backends rather than this process's.
+func (c *Cluster) FrontendBackendsDir(i int) (string, error) {
+	if err := c.checkFrontendIndex(i); err != nil {
+		return "", err
+	}
+	return filepath.Join(c.frontendDir(i), "backends"), nil
+}
+
+// NatsURL is the bus this cluster's frontends were given.
+//
+// It is exported for one assertion: a spec proving a WORKER runs with no bus
+// has to show the deployment it joined has one, or "no NATS anywhere" would
+// satisfy it just as well.
+func (c *Cluster) NatsURL() string {
+	return c.opts.NatsURL
+}
+
 // WorkerName is the node name worker i registered under.
 func (c *Cluster) WorkerName(i int) string {
 	return c.workers[i].Name
@@ -528,7 +668,7 @@ func (c *Cluster) Stop() {
 	if c == nil {
 		return
 	}
-	for _, p := range append(append([]*Process{}, c.workers...), c.frontends...) {
+	for _, p := range append(append(append([]*Process{}, c.workers...), c.agentWorkers...), c.frontends...) {
 		p.terminate()
 	}
 	if c.baseDir != "" {
@@ -539,7 +679,7 @@ func (c *Cluster) Stop() {
 // DumpLogs writes every process log to stdout. Call from an AfterEach guarded by
 // CurrentSpecReport().Failed().
 func (c *Cluster) DumpLogs() {
-	for _, p := range append(append([]*Process{}, c.frontends...), c.workers...) {
+	for _, p := range append(append(append([]*Process{}, c.frontends...), c.workers...), c.agentWorkers...) {
 		if p == nil {
 			continue
 		}

tests/e2e/distributed/cluster_baseline_test.go

@@ -44,6 +44,11 @@ type node struct {
 	// have dialled instead of the tunnel.
 	Address     string `json:"address"`
 	HTTPAddress string `json:"http_address"`
+	// LastHeartbeat is what separates "the worker is gone" from "the worker is
+	// here and this deployment cannot reach it". A spec asserting the second
+	// has to show the first is false, and the heartbeat is the only evidence
+	// of that in this payload.
+	LastHeartbeat time.Time `json:"last_heartbeat"`
 	// keys is what the payload actually carried, which a decoded struct cannot
 	// tell you. Both fields above are the zero value when a worker advertises
 	// nothing AND when the key was renamed or dropped, and the whole point of
@@ -53,6 +58,17 @@ type node struct {
 	keys map[string]json.RawMessage `json:"-"`
 }
 
+// String is what a failing assertion prints for a node.
+//
+// Without it %+v renders keys, whose values are json.RawMessage, as slices of
+// byte VALUES: one node becomes several hundred numbers and a roster of two
+// buries the assertion that failed. The key set is still what advertisementOf
+// reads; it is just not something a human ever needs to see.
+func (n node) String() string {
+	return fmt.Sprintf("{name:%s status:%s id:%s lastHeartbeat:%s advertised:%q/%q}",
+		n.Name, n.Status, n.ID, n.LastHeartbeat.Format(time.RFC3339), n.Address, n.HTTPAddress)
+}
+
 // UnmarshalJSON decodes the fields above and keeps the raw key set beside them.
 func (n *node) UnmarshalJSON(data []byte) error {
 	// A distinct type, or this method calls itself.
@@ -272,6 +288,29 @@ func (p *rosterProbe) advertisementOf(name string) (string, bool) {
 	return "", false
 }
 
+// heartbeatOf is the last heartbeat the roster reported for a node, refreshed
+// on every call.
+//
+// A zero time is returned for a node the roster does not carry, which no
+// freshness assertion can accept: the spec that reads this is proving the
+// worker is still alive, and a missing node must not read as a recent
+// heartbeat.
+func (p *rosterProbe) heartbeatOf(name string) time.Time {
+	var roster []node
+	if err := p.cluster.GetJSON(p.client, p.frontend, "/api/nodes", &roster); err != nil {
+		p.lastErr = err
+		return time.Time{}
+	}
+	p.lastErr = nil
+	p.lastSeen = roster
+	for _, n := range roster {
+		if n.Name == name {
+			return n.LastHeartbeat
+		}
+	}
+	return time.Time{}
+}
+
 // describe is handed to Should as the failure message. Gomega calls a
 // func() string description lazily, so this runs only on failure and reports
 // whichever of the two distinct causes actually occurred.

tests/e2e/distributed/cluster_control_test.go

@@ -0,0 +1,797 @@
+package distributed_test
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/mudler/LocalAI/tests/e2e/distributed/cluster"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// End-to-end proof that phase 3's control plane works under real processes.
+//
+// Phase 3 moved fourteen control verbs off NATS and onto the tunnel the worker
+// already dials, made absence a database fact rather than a bus timeout, and
+// took the backend worker off the bus entirely. Every one of those changes is
+// proven by unit and integration specs only; these are the first that run the
+// binaries an operator runs and let the real transports fail.
+//
+// Read the LAST spec before trusting the others. Frontend and worker share a
+// host here, so almost anything that looks like it went over the tunnel could
+// have gone some other way. The last one takes the tunnel away and requires the
+// control plane to become unreachable, with the refusal naming the missing
+// route rather than claiming the worker is absent; without it the rest could
+// pass with the tunnel doing nothing.
+
+const (
+	// controlRPCTimeout bounds one admin control call that has to cross a
+	// relay and reach a worker. Generous because the install verb behind it
+	// copies an artifact and starts a process.
+	controlRPCTimeout = 3 * time.Minute
+
+	// installJobTimeout bounds one node-scoped backend install from accepted to
+	// terminal. The gated spec spends most of it deliberately blocked.
+	installJobTimeout = "150s"
+	installJobPoll    = "250ms"
+
+	// gateHoldWindow is how long a spec requires an install to stay
+	// unfinished while its gallery fetch is blocked.
+	//
+	// It is not a tolerance. The worker is stopped inside an HTTP read that
+	// only this spec can complete, so a terminal reply arriving inside this
+	// window is a reply written before the work it reports on, which is the
+	// exact defect the envelope ordering exists to prevent.
+	gateHoldWindow = "3s"
+	gateHoldPoll   = "250ms"
+
+	// departedTimeout bounds the wait for a worker whose tunnel is gone past
+	// the grace to stop being reported healthy. It has to clear the grace the
+	// spec sets plus one health-monitor tick (15s by default).
+	departedTimeout = "90s"
+	departedPoll    = "1s"
+
+	// churnGrace is the reconnect grace the churn spec runs with: long enough
+	// that the whole scenario, including the worker's capped 30s reconnect
+	// backoff and the re-home after it, happens INSIDE the window. The claim
+	// under test is "nothing is reaped inside the grace", so the window has to
+	// be the thing that is generous, not the assertion.
+	churnGrace = 10 * time.Minute
+
+	// churnHoldWindow is how long the churn spec requires the fleet to survive
+	// with its tunnel genuinely gone. It starts only after the tunnel has been
+	// OBSERVED unowned, and it is past the health monitor's 15s tick, so at
+	// least one sweep runs against a worker nothing can reach and finds nothing
+	// to reap.
+	churnHoldWindow = "20s"
+	churnHoldPoll   = "1s"
+
+	// churnDropTimeout bounds the wait for a killed replica's claim to stop
+	// reading as a live owner.
+	//
+	// It has to outlast cluster.InstanceLiveness (30s): the ownership query
+	// joins against instances the DATABASE still considers live, so for the
+	// half minute after a SIGKILL a dead replica is still reported as holding
+	// every tunnel it held. That window is the reason this spec waits for the
+	// drop instead of assuming the kill produced one, and it is what an earlier
+	// version of this spec got wrong: its hold window sat entirely inside the
+	// liveness window, so it never reached the branch it claims to test and
+	// stayed green with the reconnect grace set to a nanosecond.
+	churnDropTimeout = "90s"
+	churnDropPoll    = "1s"
+
+	// probeBackend is the backend a control spec installs on a worker.
+	//
+	// A name nothing else knows, and that is the point: the frontends' own
+	// backends directory is empty and the worker's holds only mock-backend, so
+	// a node backend listing that names this one can only have come from the
+	// worker, and only after an install that reached it.
+	probeBackend = "relay-probe"
+
+	// probeGalleryName is the gallery the probe backend is served from.
+	probeGalleryName = "e2e-control"
+)
+
+// nodeModel is the subset of a /api/nodes/:id/models row these specs assert on.
+type nodeModel struct {
+	ModelName string `json:"model_name"`
+	State     string `json:"state"`
+	Address   string `json:"address"`
+}
+
+// nodeBackend is the subset of a /api/nodes/:id/backends row these specs assert
+// on. It is the worker's own answer to the backend.list control verb, relayed
+// back through whichever replica took the request.
+type nodeBackend struct {
+	Name string `json:"name"`
+}
+
+// installJob is the subset of galleryop.OpStatus these specs assert on.
+//
+// It is read from GET /backends/jobs/:uuid and not from the UI's
+// /api/backends/job/:uid, because only the former carries the per-node
+// breakdown, and the per-node breakdown is where a worker's progress line
+// lands. The UI endpoint drops it.
+type installJob struct {
+	Processed bool             `json:"processed"`
+	Message   string           `json:"message"`
+	Error     string           `json:"error"`
+	Nodes     []installJobNode `json:"nodes"`
+}
+
+type installJobNode struct {
+	NodeID string `json:"node_id"`
+	Status string `json:"status"`
+	Phase  string `json:"phase"`
+	Error  string `json:"error"`
+}
+
+// nodeEntry returns the per-node row for nodeID, or nil.
+func (j installJob) nodeEntry(nodeID string) *installJobNode {
+	for i := range j.Nodes {
+		if j.Nodes[i].NodeID == nodeID {
+			return &j.Nodes[i]
+		}
+	}
+	return nil
+}
+
+// controlSession is one admin session for a whole cluster, with a budget long
+// enough for a control RPC that crosses a relay.
+//
+// One per spec and never one per frontend, for the reason inferenceClient gives:
+// the auth routes share a five-per-minute-per-IP budget and every request here
+// comes from 127.0.0.1.
+func controlSession(c *cluster.Cluster) *http.Client {
+	GinkgoHelper()
+	client, err := c.AdminSession(0)
+	Expect(err).ToNot(HaveOccurred())
+	client.Timeout = controlRPCTimeout
+	return client
+}
+
+// nodeBackendNames lists the backends a worker reports installed, asked at one
+// specific frontend.
+//
+// The frontend answers this by issuing the backend.list control verb over that
+// worker's tunnel, relayed through the owner when this replica is not it, so
+// the returned names are the WORKER's and not this process's.
+func nodeBackendNames(c *cluster.Cluster, client *http.Client, frontend int, nodeID string) ([]string, error) {
+	var listed []nodeBackend
+	if err := c.GetJSON(client, frontend, "/api/nodes/"+nodeID+"/backends", &listed); err != nil {
+		return nil, err
+	}
+	names := []string{}
+	for _, b := range listed {
+		names = append(names, b.Name)
+	}
+	return names, nil
+}
+
+// nodeModelNames lists the models a frontend records as loaded on a worker.
+func nodeModelNames(c *cluster.Cluster, client *http.Client, frontend int, nodeID string) ([]string, error) {
+	var rows []nodeModel
+	if err := c.GetJSON(client, frontend, "/api/nodes/"+nodeID+"/models", &rows); err != nil {
+		return nil, err
+	}
+	names := []string{}
+	for _, m := range rows {
+		names = append(names, m.ModelName)
+	}
+	return names, nil
+}
+
+// gatedGallery serves one backend gallery index, and can hold the request open
+// until a spec lets it go.
+//
+// The gate is what makes the ordering assertion deterministic rather than
+// tolerant. A worker install that finishes in milliseconds gives a poller no
+// window to observe progress in, and a spec that "usually" sees a tick before
+// the reply is a spec that passes for timing reasons. Here the worker is
+// stopped inside the gallery fetch, which happens AFTER it emits its first
+// progress line and BEFORE it can produce any reply, so the two are separated
+// by something the spec controls instead of by luck.
+//
+// fetches counts what the worker actually asked for, so a spec can prove the
+// gate is on the path it thinks it is rather than assuming it.
+type gatedGallery struct {
+	server  *httptest.Server
+	index   string
+	open    chan struct{}
+	once    sync.Once
+	fetches atomic.Int64
+}
+
+// newGatedGallery starts a gallery serving index. When gated, the first and
+// every subsequent fetch blocks until release is called.
+func newGatedGallery(index string, gated bool) *gatedGallery {
+	GinkgoHelper()
+	g := &gatedGallery{index: index, open: make(chan struct{})}
+	if !gated {
+		g.release()
+	}
+	mux := http.NewServeMux()
+	mux.HandleFunc("/index.yaml", func(w http.ResponseWriter, r *http.Request) {
+		g.fetches.Add(1)
+		select {
+		case <-g.open:
+		case <-r.Context().Done():
+			// The worker gave up, or the spec ended. Answering nothing is
+			// right: writing a body here would let a spec that failed its own
+			// gate assertion still see a successful install.
+			return
+		}
+		w.Header().Set("Content-Type", "application/yaml")
+		_, _ = io.WriteString(w, g.index)
+	})
+	g.server = httptest.NewServer(mux)
+	DeferCleanup(func() {
+		// Released before close so a handler still parked on the gate returns
+		// instead of leaking until the test binary exits.
+		g.release()
+		g.server.Close()
+	})
+	return g
+}
+
+func (g *gatedGallery) release() { g.once.Do(func() { close(g.open) }) }
+
+// URL is the index URL a worker fetches.
+func (g *gatedGallery) URL() string { return g.server.URL + "/index.yaml" }
+
+// galleriesJSON is the backend_galleries override an install request carries.
+// It is a JSON string INSIDE the request body, which is the shape
+// InstallBackendOnNodeEndpoint binds.
+func (g *gatedGallery) galleriesJSON() string {
+	return fmt.Sprintf(`[{"name":%q,"url":%q}]`, probeGalleryName, g.URL())
+}
+
+// probeBackendSource lays out a directory a worker can install as a backend and
+// then RUN, and returns its path.
+//
+// It is a real backend by the two rules core/gallery enforces on a directory
+// URI: a run.sh, which is the validation gate, and an executable named after
+// the backend, which is what the worker's findBackend resolves and starts. The
+// executable is the mock backend, so the install ends with a live gRPC process
+// rather than with a process that dies and turns a relay spec into a spec about
+// a broken artifact.
+func probeBackendSource() string {
+	GinkgoHelper()
+	dir := filepath.Join(GinkgoT().TempDir(), probeBackend)
+	Expect(os.MkdirAll(dir, 0o755)).To(Succeed())
+
+	binary, err := os.ReadFile(mockBackendBinary())
+	Expect(err).ToNot(HaveOccurred())
+	Expect(os.WriteFile(filepath.Join(dir, probeBackend), binary, 0o755)).To(Succeed())
+	Expect(os.WriteFile(filepath.Join(dir, "run.sh"),
+		[]byte("#!/bin/sh\nexec \"$(dirname \"$0\")/"+probeBackend+"\" \"$@\"\n"), 0o755)).To(Succeed())
+	return dir
+}
+
+// probeGalleryIndex is the one-entry gallery index that points at src.
+func probeGalleryIndex(src string) string {
+	return fmt.Sprintf("- name: %s\n  uri: %s\n  description: e2e control-plane probe\n", probeBackend, src)
+}
+
+// startNodeInstall posts a node-scoped backend install at one frontend and
+// returns the job id it was given.
+//
+// The install is asynchronous by design (202 plus a job id), so this is where
+// the request stops and the job polling below takes over.
+func startNodeInstall(c *cluster.Cluster, client *http.Client, frontend int, nodeID, backend, galleriesJSON string) string {
+	GinkgoHelper()
+	var accepted struct {
+		JobID string `json:"jobID"`
+	}
+	status, err := c.PostJSON(client, frontend, "/api/nodes/"+nodeID+"/backends/install", map[string]string{
+		"backend":           backend,
+		"backend_galleries": galleriesJSON,
+	}, &accepted)
+	Expect(err).ToNot(HaveOccurred())
+	Expect(status).To(Equal(http.StatusAccepted),
+		"frontend %d refused the install outright, so nothing was ever sent to the worker", frontend)
+	Expect(accepted.JobID).ToNot(BeEmpty())
+	return accepted.JobID
+}
+
+// jobProbe polls one install job at one frontend and keeps what it last saw, so
+// a failing Eventually can name it.
+type jobProbe struct {
+	cluster  *cluster.Cluster
+	client   *http.Client
+	frontend int
+	jobID    string
+	nodeID   string
+
+	lastErr error
+	last    installJob
+}
+
+func newJobProbe(c *cluster.Cluster, client *http.Client, frontend int, jobID, nodeID string) *jobProbe {
+	return &jobProbe{cluster: c, client: client, frontend: frontend, jobID: jobID, nodeID: nodeID}
+}
+
+// read fetches the job once, keeping the error rather than raising it.
+func (p *jobProbe) read() installJob {
+	var job installJob
+	if err := p.cluster.GetJSON(p.client, p.frontend, "/backends/jobs/"+p.jobID, &job); err != nil {
+		p.lastErr = err
+		return installJob{}
+	}
+	p.lastErr = nil
+	p.last = job
+	return job
+}
+
+// nodePhase is the phase the worker last reported for this node, or "" when the
+// job carries no row for it yet.
+func (p *jobProbe) nodePhase() string {
+	job := p.read()
+	if entry := job.nodeEntry(p.nodeID); entry != nil {
+		return entry.Phase
+	}
+	return ""
+}
+
+// nodeStatus is the status the job last recorded for this node.
+func (p *jobProbe) nodeStatus() string {
+	job := p.read()
+	if entry := job.nodeEntry(p.nodeID); entry != nil {
+		return entry.Status
+	}
+	return ""
+}
+
+// processed reports whether the job has reached a terminal state.
+func (p *jobProbe) processed() bool { return p.read().Processed }
+
+// jobError is the job's error text, empty when it has none.
+func (p *jobProbe) jobError() string { return p.read().Error }
+
+// explain builds a lazy failure description.
+//
+// Gomega formats a (string, args...) description when the assertion is
+// CONSTRUCTED, which for an Eventually or a Consistently is before anything has
+// gone wrong: the job it quoted would be the one from before the wait. A
+// func() string is called only on failure. The same trap, and the same fix, as
+// rosterProbe.explain.
+func (p *jobProbe) explain(format string, args ...any) func() string {
+	return func() string {
+		return fmt.Sprintf(format, args...) + ": " + p.describe()
+	}
+}
+
+func (p *jobProbe) describe() string {
+	if p.lastErr != nil {
+		return fmt.Sprintf("frontend %d: the last read of job %s failed: %v", p.frontend, p.jobID, p.lastErr)
+	}
+	return fmt.Sprintf("frontend %d: job %s last read as %+v", p.frontend, p.jobID, p.last)
+}
+
+// withReconnectGrace pins how long a lost worker tunnel is read as reconnecting
+// rather than gone.
+func withReconnectGrace(d time.Duration) func(*cluster.Options) {
+	return func(o *cluster.Options) { o.ReconnectGrace = d }
+}
+
+// withAgentWorkers adds agent workers, which still speak NATS and hold no
+// tunnel, to a cluster.
+func withAgentWorkers(n int) func(*cluster.Options) {
+	return func(o *cluster.Options) { o.AgentWorkers = n }
+}
+
+var _ = Describe("Control plane over the worker tunnel", Label("Distributed"), Label("Cluster"), func() {
+
+	// Scenario 1, the headline. A backend worker that is connected to no bus at
+	// all registers, is scheduled onto, and serves inference.
+	//
+	// A wrong implementation leaves the worker up and inert, because nothing
+	// reaches it: before phase 3 every control verb travelled on NATS, so a
+	// worker with no NATS URL would register and heartbeat and never be given a
+	// backend or a model.
+	It("registers, schedules onto and serves a worker that is connected to no bus", func() {
+		c, dsn := startClusterOnFreshDB(1, 1, withMockModel("busless-model"))
+		client := inferenceClient(c)
+
+		// The environment of the RUNNING process, not the one the harness
+		// assembled. LOCALAI_REGISTER_TO is asserted alongside so an absent
+		// LOCALAI_NATS_URL is a fact about the worker rather than a read that
+		// returned nothing: a broken read would lose both.
+		environ, err := c.WorkerEnviron(0)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(environ).To(ContainElement(HavePrefix("LOCALAI_REGISTER_TO=")),
+			"the worker's environment could not be read, so the absence below proves nothing")
+		for _, entry := range environ {
+			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.
+		Expect(c.NatsURL()).ToNot(BeEmpty())
+
+		probe := newRosterProbe(c, client, 0)
+		Eventually(probe.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(ContainElement(c.WorkerName(0)), probe.describe)
+		nodeID := probe.idOf(c.WorkerName(0))
+		Expect(nodeID).ToNot(BeEmpty())
+
+		// Its tunnel is held here, and it advertises nothing. Both matter: the
+		// first says there is a route, the second says there is no other one.
+		owners := newTunnelOwners(openClusterDB(dsn))
+		Eventually(func() int { return owners.ownerIndexOf(c, 1, nodeID) }, tunnelOwnershipTimeout, tunnelOwnershipPoll).
+			Should(Equal(0), owners.describe)
+		advertised, carriedKeys := probe.advertisementOf(c.WorkerName(0))
+		Expect(carriedKeys).To(BeTrue(),
+			"the roster payload no longer carries the advertisement keys, so this spec cannot tell a worker that advertises nothing from one it cannot read")
+		Expect(advertised).To(BeEmpty(),
+			"the worker advertised %q, so a frontend could have reached it without the tunnel", advertised)
+
+		// The inference is what drives the whole control plane: the frontend
+		// installs the backend on the worker, stages the model artifact to it
+		// and loads the model, all over the tunnel and all with no bus on the
+		// worker's side.
+		expectMockedInference(client, c.FrontendURL(0), "busless-model",
+			"a worker with no bus connection must still be scheduled onto and serve inference")
+
+		// And the frontend recorded the model as loaded THERE, naming the
+		// process the worker started. An empty address would mean the install
+		// reply named none, which is refused now rather than substituted.
+		var rows []nodeModel
+		Expect(c.GetJSON(client, 0, "/api/nodes/"+nodeID+"/models", &rows)).To(Succeed())
+		Expect(rows).ToNot(BeEmpty(), "no model is recorded on the worker that just served the request")
+		Expect(rows[0].ModelName).To(Equal("busless-model"))
+		Expect(rows[0].State).To(Equal("loaded"))
+		Expect(rows[0].Address).ToNot(BeEmpty(),
+			"the row names no backend process on the worker, so nothing could address it again")
+	})
+
+	// Scenario 2, the relay, and scenario 3, the ordering, in one cluster.
+	//
+	// They are one spec because a relay spec needs something to ask a worker
+	// that only the worker can answer, and a fresh worker's backend list is
+	// empty: the install is what puts something there. Splitting them would
+	// have cost a second cluster to assert less.
+	//
+	// A wrong implementation answers the control RPC by reaching the worker
+	// from the replica that took it, which works on one host and nowhere else,
+	// or refuses it as a worker that is not connected. A wrong STREAM
+	// implementation writes the terminal reply before the progress it reports
+	// on, or drops the progress entirely; both pass every unit spec.
+	It("installs a backend, streams its progress in order, and lists it back, all through the replica that does not own the worker", func() {
+		c, dsn := startClusterOnFreshDB(2, 1)
+		client := controlSession(c)
+
+		probe := newRosterProbe(c, client, 0)
+		Eventually(probe.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(ContainElement(c.WorkerName(0)), probe.describe)
+		nodeID := probe.idOf(c.WorkerName(0))
+		Expect(nodeID).ToNot(BeEmpty())
+
+		// Which replica owns the tunnel is READ through the production Owner
+		// query, not assumed. The harness sends the worker to frontend 0 by
+		// default, and a spec that wrote that down would keep passing after the
+		// default changed while quietly testing the owner path instead.
+		owners := newTunnelOwners(openClusterDB(dsn))
+		var owner int
+		Eventually(func() int {
+			owner = owners.ownerIndexOf(c, 2, nodeID)
+			return owner
+		}, tunnelOwnershipTimeout, tunnelOwnershipPoll).Should(BeNumerically(">=", 0), owners.describe)
+
+		nonOwner := 1 - owner
+		Expect(owners.ownerIndexOf(c, 2, nodeID)).ToNot(Equal(nonOwner),
+			"frontend %d owns the worker's tunnel, so a request to it would not be relayed and this spec would prove nothing", nonOwner)
+
+		// Nothing is installed on the worker yet, and the request below is the
+		// only thing that could change that.
+		Expect(nodeBackendNames(c, client, nonOwner, nodeID)).ToNot(ContainElement(probeBackend),
+			"the worker already has %q, so a listing that names it later says nothing about this install", probeBackend)
+
+		gallery := newGatedGallery(probeGalleryIndex(probeBackendSource()), true)
+		jobID := startNodeInstall(c, client, nonOwner, nodeID, probeBackend, gallery.galleriesJSON())
+		job := newJobProbe(c, client, nonOwner, jobID, nodeID)
+
+		// The worker has emitted its first progress line and is now stopped
+		// inside the gallery fetch. Seeing the phase here proves a progress
+		// envelope crossed the relay and was decoded, and it proves it BEFORE
+		// any reply could exist, because the worker cannot produce one until
+		// this spec releases the gate.
+		Eventually(job.nodePhase, installJobTimeout, installJobPoll).Should(Equal("resolving"), job.describe)
+		Eventually(gallery.fetches.Load, installJobTimeout, installJobPoll).Should(BeNumerically(">", 0),
+			"the worker never fetched the gated gallery, so the gate is not on the path this spec thinks it is")
+
+		// Held, not sampled. A reply arriving in this window is a reply written
+		// ahead of the work it reports on.
+		Consistently(job.processed, gateHoldWindow, gateHoldPoll).Should(BeFalse(),
+			job.explain("the install reported a terminal result while the worker was still blocked fetching its gallery"))
+
+		gallery.release()
+
+		Eventually(job.processed, installJobTimeout, installJobPoll).Should(BeTrue(), job.describe)
+		Expect(job.jobError()).To(BeEmpty(), "the relayed install failed: %s", job.describe())
+		Expect(job.nodeStatus()).To(Equal("success"), job.describe)
+
+		// Nothing follows the terminal reply. A worker that appended a late
+		// progress line after its reply would move this row back to
+		// "downloading"; the guard against that is ndjsonStream.done, and this
+		// is the only place it is exercised over a real stream.
+		Consistently(job.nodeStatus, gateHoldWindow, gateHoldPoll).Should(Equal("success"),
+			job.explain("the node's status moved after the install's terminal reply"))
+
+		// A SECOND control verb across the relay, and the one whose answer
+		// could only have come from the worker: the frontends' own backends
+		// directories are empty, and this backend was installed on the worker
+		// alone.
+		Eventually(func() ([]string, error) { return nodeBackendNames(c, client, nonOwner, nodeID) },
+			installJobTimeout, installJobPoll).Should(ContainElement(probeBackend))
+
+		// And the frontend that answered does not have it itself, so it cannot
+		// have been reporting its own installation as the worker's.
+		//
+		// Read off the filesystem and not from GET /backends: in distributed
+		// mode that endpoint reports the CLUSTER's backends, so it names this
+		// one whether the frontend has it or not, and an assertion on it fails
+		// for a reason that has nothing to do with what is being proven.
+		backendsDir, err := c.FrontendBackendsDir(nonOwner)
+		Expect(err).ToNot(HaveOccurred())
+		entries, err := os.ReadDir(backendsDir)
+		Expect(err).ToNot(HaveOccurred())
+		for _, e := range entries {
+			Expect(e.Name()).ToNot(Equal(probeBackend),
+				"frontend %d installed %q locally, so its answer about the worker may be about itself", nonOwner, probeBackend)
+		}
+
+		// THIS is what makes every request above a relayed one. It rules out
+		// the two ways they could have succeeded without a relay: a replica
+		// that took the tunnel for itself, and the tunnel moving to nonOwner
+		// mid-spec so that it served directly. Both leave the owner changed.
+		Expect(owners.ownerIndexOf(c, 2, nodeID)).To(Equal(owner),
+			"the tunnel is no longer held by frontend %d, so the requests to frontend %d were not necessarily relayed", owner, nonOwner)
+	})
+
+	// Scenario 4, absence under replica churn. The catastrophe control.
+	//
+	// A worker whose tunnel drops and returns INSIDE the grace must not be
+	// reaped. A wrong implementation reads a lost tunnel as a departed worker
+	// and evicts the models it is serving in the seconds before the re-home,
+	// which is the fleet-wide outage the four-valued presence answer exists to
+	// prevent.
+	It("reaps nothing when the replica holding a worker's tunnel dies and the worker re-homes inside the grace", func() {
+		var balancer *frontendBalancer
+		c, dsn := startClusterOnFreshDB(2, 1, withMockModel("churn-model"),
+			withBalancer(&balancer), withReconnectGrace(churnGrace))
+
+		client := inferenceClient(c)
+		probe := newRosterProbe(c, client, 0)
+		Eventually(probe.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(ContainElement(c.WorkerName(0)), probe.describe)
+		nodeID := probe.idOf(c.WorkerName(0))
+		Expect(nodeID).ToNot(BeEmpty())
+
+		owners := newTunnelOwners(openClusterDB(dsn))
+		var owner int
+		Eventually(func() int {
+			owner = owners.ownerIndexOf(c, 2, nodeID)
+			return owner
+		}, tunnelOwnershipTimeout, tunnelOwnershipPoll).Should(BeNumerically(">=", 0), owners.describe)
+		survivor := 1 - owner
+
+		// A model actually loaded on the worker, or there is nothing to reap
+		// and the assertions below hold vacuously.
+		expectMockedInference(client, c.FrontendURL(owner), "churn-model",
+			"inference must work before the owner is killed, or nothing here is at risk")
+		Expect(nodeModelNames(c, client, survivor, nodeID)).To(ContainElement("churn-model"))
+
+		// The tunnel is taken away BEFORE the owner is killed, and held away
+		// until the assertions below have run.
+		//
+		// Without the block the worker re-homes onto the survivor within
+		// seconds, and a spec that asserted over those seconds would never
+		// reach the state it is about: for the first half minute after a
+		// SIGKILL the dead replica still reads as a live owner, so presence is
+		// "connected" and no rule about a departed worker has anything to act
+		// on. Blocking makes the outage last long enough for a departure to be
+		// recorded, which is the only way the grace is consulted at all.
+		balancer.blockTunnel.Store(true)
+		Expect(c.KillFrontend(owner)).To(Succeed())
+		Eventually(func() bool { return c.FrontendAlive(owner) }, "20s", "500ms").Should(BeFalse())
+
+		// Read at the SURVIVOR from here on: the probe above is bound to a dead
+		// process.
+		atSurvivor := newRosterProbe(c, client, survivor)
+
+		// Wait for the drop to be REAL rather than assume the kill produced
+		// one. Until this reads empty the ownership query still reports the
+		// dead replica, and everything below would be asserting about a worker
+		// the deployment believes is connected.
+		Eventually(func() string { return owners.ownerOf(nodeID) }, churnDropTimeout, churnDropPoll).
+			Should(BeEmpty(), owners.describe)
+
+		// THIS is the window the phase is about: the tunnel is gone, the
+		// departure is recorded, and the grace has not run out. Nothing may be
+		// reaped in it.
+		Consistently(func() []string {
+			names, err := nodeModelNames(c, client, survivor, nodeID)
+			if err != nil {
+				return nil
+			}
+			return names
+		}, churnHoldWindow, churnHoldPoll).Should(ContainElement("churn-model"),
+			"the model loaded on the worker was reaped while its tunnel was inside the reconnect grace")
+		Consistently(func() string { return atSurvivor.statusOf(c.WorkerName(0)) }, churnHoldWindow, churnHoldPoll).
+			ShouldNot(Equal("unhealthy"),
+				atSurvivor.explain("the worker was demoted while its tunnel was inside the reconnect grace"))
+
+		// And it comes back: with the block lifted the tunnel re-homes onto the
+		// survivor and the same model serves again. Without this the assertions
+		// above would be satisfied by a fleet that was simply never touched.
+		balancer.blockTunnel.Store(false)
+		Eventually(func() int { return owners.ownerIndexOf(c, 2, nodeID) }, tunnelOwnershipTimeout, tunnelOwnershipPoll).
+			Should(Equal(survivor), owners.describe)
+		Eventually(atSurvivor.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(ContainElement(c.WorkerName(0)), atSurvivor.describe)
+		Expect(atSurvivor.idOf(c.WorkerName(0))).To(Equal(nodeID),
+			"the worker re-registered rather than re-homing, so this proves nothing about a tunnel surviving its replica")
+
+		eventuallyMockedInference(client, c.FrontendURL(survivor), "churn-model",
+			"the survivor must serve the re-homed worker")
+	})
+
+	// Scenario 5, the wedge task 6 fixed, plus the agent-worker control.
+	//
+	// A worker with a FRESH heartbeat and a tunnel that is gone past the grace
+	// must stop being reported healthy. Before task 6 it stayed healthy
+	// forever, with every request for a model on it failing "no route", because
+	// every reaper keyed on the heartbeat and the heartbeat was fine.
+	//
+	// The agent worker in the same cluster is the control for the other
+	// direction. It still speaks NATS, holds no tunnel and never will, so a
+	// rule that read "no tunnel" as "gone" would take the whole agent fleet
+	// down with it.
+	It("stops reporting a heartbeating worker healthy once its tunnel is gone past the grace, and leaves agent workers alone", func() {
+		var balancer *frontendBalancer
+		c, dsn := startClusterOnFreshDB(2, 1, withBalancer(&balancer),
+			withAgentWorkers(1), withReconnectGrace(10*time.Second))
+
+		client := controlSession(c)
+		probe := newRosterProbe(c, client, 0)
+		Eventually(probe.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(And(ContainElement(c.WorkerName(0)), ContainElement(c.AgentWorkerName(0))), probe.describe)
+		nodeID := probe.idOf(c.WorkerName(0))
+		Expect(nodeID).ToNot(BeEmpty())
+
+		owners := newTunnelOwners(openClusterDB(dsn))
+		var owner int
+		Eventually(func() int {
+			owner = owners.ownerIndexOf(c, 2, nodeID)
+			return owner
+		}, tunnelOwnershipTimeout, tunnelOwnershipPoll).Should(BeNumerically(">=", 0), owners.describe)
+		survivor := 1 - owner
+
+		// Take the tunnel away permanently: block the dial, then kill the
+		// replica holding the live session. The worker keeps registering and
+		// heartbeating through the balancer, which still proxies everything
+		// except the tunnel connect.
+		balancer.blockTunnel.Store(true)
+		Expect(c.KillFrontend(owner)).To(Succeed())
+		Eventually(func() bool { return c.FrontendAlive(owner) }, "20s", "500ms").Should(BeFalse())
+
+		atSurvivor := newRosterProbe(c, client, survivor)
+
+		// The worker really is trying and really is being refused, so what
+		// makes it unreachable below is the tunnel and not the worker dying.
+		before := balancer.tunnelDials.Load()
+		Eventually(balancer.tunnelDials.Load, "90s", "500ms").Should(BeNumerically(">", before),
+			"the worker stopped dialling its tunnel, so blocking the dial is not what keeps it away")
+
+		// The verdict. Fresh heartbeat, no tunnel, past the grace.
+		Eventually(func() string { return atSurvivor.statusOf(c.WorkerName(0)) }, departedTimeout, departedPoll).
+			Should(Equal("unhealthy"), atSurvivor.describe)
+
+		// The heartbeat was NOT what demoted it. Without this the assertion
+		// above is satisfied by a worker that simply died, which says nothing
+		// about tunnel departure.
+		Expect(atSurvivor.heartbeatOf(c.WorkerName(0))).To(BeTemporally(">", time.Now().Add(-1*time.Minute)),
+			"the worker's heartbeat is stale, so it was demoted for being gone rather than for having no route")
+
+		// The agent worker, in the same cluster, under the same grace, on the
+		// same health monitor, is untouched. It holds no tunnel either.
+		Consistently(func() string { return atSurvivor.statusOf(c.AgentWorkerName(0)) }, "20s", "2s").
+			Should(Equal("healthy"),
+				atSurvivor.explain("an agent worker was demoted by a rule about tunnels, and agent workers never hold one"))
+
+		// And the demotion reverses when the route comes back, so it is a
+		// statement about the route rather than a one-way condemnation.
+		balancer.blockTunnel.Store(false)
+		Eventually(func() int { return owners.ownerIndexOf(c, 2, nodeID) }, tunnelOwnershipTimeout, tunnelOwnershipPoll).
+			Should(Equal(survivor), owners.describe)
+		Eventually(func() string { return atSurvivor.statusOf(c.WorkerName(0)) }, departedTimeout, departedPoll).
+			Should(Equal("healthy"), atSurvivor.describe)
+	})
+
+	// Scenario 6. THE NEGATIVE CONTROL FOR THE WHOLE SUITE.
+	//
+	// Frontend and worker share a host, so every address the control plane
+	// names is one the frontend could also have reached directly. If it did,
+	// every spec above would pass with the tunnel doing nothing. This one takes
+	// the tunnel away and requires the control plane to become unreachable,
+	// with the refusal naming the ROUTE, while registration, heartbeats and the
+	// roster stay exactly as they were.
+	//
+	// It cannot use LOCALAI_WORKER_TUNNEL=false: that is a fatal startup error
+	// after phase 2, and a worker that never started says nothing about a
+	// worker reachable by some other path.
+	It("cannot drive the control plane on a worker whose tunnel is refused, reaps nothing for it, and can as soon as it is not", func() {
+		var balancer *frontendBalancer
+		c, dsn := startClusterOnFreshDB(1, 1,
+			withBalancer(&balancer, func(b *frontendBalancer) { b.blockTunnel.Store(true) }))
+
+		client := controlSession(c)
+		probe := newRosterProbe(c, client, 0)
+		Eventually(probe.healthyNames, nodeRosterTimeout, nodeRosterPoll).
+			Should(ContainElement(c.WorkerName(0)), probe.describe)
+		nodeID := probe.idOf(c.WorkerName(0))
+		Expect(nodeID).ToNot(BeEmpty())
+
+		// In every respect the worker the specs above used, except the tunnel,
+		// and both halves of that are asserted rather than assumed: it tried,
+		// and no replica holds it.
+		Eventually(balancer.tunnelDials.Load, "60s", "500ms").Should(BeNumerically(">", 0),
+			"the worker never dialled its tunnel, so blocking the dial is not what makes it unreachable below")
+		owners := newTunnelOwners(openClusterDB(dsn))
+		Consistently(func() string { return owners.ownerOf(nodeID) }, "5s", "500ms").Should(BeEmpty(),
+			"a replica holds this worker's tunnel, so the blocker is not blocking")
+
+		gallery := newGatedGallery(probeGalleryIndex(probeBackendSource()), false)
+		jobID := startNodeInstall(c, client, 0, nodeID, probeBackend, gallery.galleriesJSON())
+		job := newJobProbe(c, client, 0, jobID, nodeID)
+
+		Eventually(job.processed, installJobTimeout, installJobPoll).Should(BeTrue(), job.describe)
+
+		// It fails, and it fails for the ROUTING reason. A refusal for any
+		// other cause (a gallery it could not read, a node it thought absent, a
+		// backend that would not install) would satisfy "it failed" just as
+		// well and would leave every spec above unproven.
+		//
+		// One substring and not a disjunction: "no route" is what
+		// cluster.ErrNoRoute reads as, and nothing else on this path produces
+		// it. In particular it is NOT what an absent worker produces, and that
+		// distinction is the phase's whole absence contract: a worker this
+		// replica cannot reach must never be reported as one that has gone.
+		Expect(job.jobError()).To(ContainSubstring("no route"),
+			"the refusal does not name the missing route, so this spec cannot tell a worker with no tunnel from an ordinary install failure: %s", job.describe())
+
+		// And nothing was reaped for it. The worker never stopped
+		// heartbeating, so a control RPC that could not be delivered must not
+		// have cost it its row or its health.
+		_, listErr := nodeBackendNames(c, client, 0, nodeID)
+		Expect(listErr).To(HaveOccurred(),
+			"the frontend answered a backend listing for a worker it has no route to, so something other than the tunnel reaches it")
+		Consistently(probe.healthyNames, "10s", "1s").
+			Should(ContainElement(c.WorkerName(0)),
+				probe.explain("the worker was demoted or removed because a control RPC could not be routed to it"))
+
+		// The control's own control: put the tunnel back, change nothing else,
+		// and the SAME install must now succeed. Without this the failure above
+		// could be any of the ordinary reasons an e2e install fails.
+		balancer.blockTunnel.Store(false)
+		Eventually(func() int { return owners.ownerIndexOf(c, 1, nodeID) }, tunnelOwnershipTimeout, tunnelOwnershipPoll).
+			Should(Equal(0), owners.describe)
+
+		retryID := startNodeInstall(c, client, 0, nodeID, probeBackend, gallery.galleriesJSON())
+		retry := newJobProbe(c, client, 0, retryID, nodeID)
+		Eventually(retry.processed, installJobTimeout, installJobPoll).Should(BeTrue(), retry.describe)
+		Expect(retry.jobError()).To(BeEmpty(),
+			"the only thing that changed is the tunnel, so the failure above was the missing tunnel and nothing else: %s", retry.describe())
+		Eventually(func() ([]string, error) { return nodeBackendNames(c, client, 0, nodeID) },
+			installJobTimeout, installJobPoll).Should(ContainElement(probeBackend))
+	})
+})

tests/e2e/distributed/control_workers_test.go

@@ -3,6 +3,7 @@ package distributed_test
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -13,6 +14,8 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
+	clustersvc "github.com/mudler/LocalAI/core/services/cluster"
+	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/workerctl"
 )
@@ -110,3 +113,78 @@ func (c *ControlWorkers) serve(w http.ResponseWriter, r *http.Request) {
 	// The reply line, and it is the last thing on the body by contract.
 	Expect(json.NewEncoder(w).Encode(workerctl.Envelope{Reply: raw})).To(Succeed())
 }
+
+// ServeBackendLifecycle registers the three verbs every routing spec needs from
+// a worker: install, the running-model list, and the backend list.
+//
+// The install reply NAMES the address of the backend process the fake worker
+// started, and that is the half these suites used to leave out. Since phase 2 a
+// worker advertises no address of its own, so this string is the only thing
+// that tells the frontend WHICH process on that worker a model was loaded into,
+// and installBackendOnNode refuses a success reply that omits it rather than
+// substituting anything. Every one of these suites runs its mock gRPC backend
+// on loopback and records where it listens in the node row it registers, so
+// that row is where this reads it back from. It is the spec's own bookkeeping
+// standing in for what a real worker reports about its own process; nothing in
+// production reads BackendNode.Address any more.
+func (c *ControlWorkers) ServeBackendLifecycle(registry *nodes.NodeRegistry) {
+	c.On(AnyNode, workerctl.PathBackendInstall, func(nodeID string, _ []byte) any {
+		node, err := registry.Get(context.Background(), nodeID)
+		if err != nil {
+			return messaging.BackendInstallReply{Success: false, Error: err.Error()}
+		}
+		return messaging.BackendInstallReply{Success: true, WorkerLocalAddress: node.Address}
+	})
+	c.On(AnyNode, workerctl.PathModelsRunning, func(string, []byte) any {
+		return messaging.ModelsRunningReply{}
+	})
+	c.On(AnyNode, workerctl.PathBackendList, func(string, []byte) any {
+		return messaging.BackendListReply{}
+	})
+}
+
+// workerBackendDialerFor stands in for the worker tunnel on the gRPC path.
+//
+// A frontend no longer dials a backend process: it opens a stream on the
+// worker's tunnel and names the process by its worker-local address. These
+// suites run the process on loopback, so a TCP dial to that address is the
+// stand-in, and NewTunnelClientFactory below is what makes the specs go through
+// a dialer at all rather than through the direct dial the default factory now
+// refuses.
+//
+// The refusal is translated, and that is the load-bearing half. A real worker
+// whose backend process has died answers the stream with
+// ErrStreamTargetUnavailable, which cluster.IsWorkerAnswer reads as the WORKER
+// speaking about its backend, and every reap guard in core/services/nodes acts
+// only on that. A bare ECONNREFUSED from net.Dialer carries no such thing and
+// reaches those guards as "no route", which reaps nothing. A double that
+// reported the raw syscall error could therefore never fail the way production
+// fails, and the stale-record spec in router_tracking_test.go would be
+// asserting against a transport that cannot produce the condition it is about.
+// That is not a hypothetical: replacing this translation with the raw error
+// reddens exactly that spec and nothing else.
+func workerBackendDialerFor(_ string) func(ctx context.Context, addr string) (net.Conn, error) {
+	return func(ctx context.Context, addr string) (net.Conn, error) {
+		var d net.Dialer
+		conn, err := d.DialContext(ctx, "tcp", addr)
+		if err != nil {
+			return nil, fmt.Errorf("%w: %v", clustersvc.ErrStreamTargetUnavailable, err)
+		}
+		return conn, nil
+	}
+}
+
+// tunnelBackendClients is the BackendClientFactory a SmartRouter needs in these
+// suites.
+//
+// It is NewTunnelClientFactory and not a bespoke double on purpose: the default
+// factory refuses every request now (see nodes.ErrNoWorkerDialer), so a spec
+// that omitted this would fail at the first inference with a boot-time
+// misconfiguration rather than testing anything, and a bespoke factory that
+// dialled directly would put back the bypass the phase removed.
+func tunnelBackendClients() nodes.BackendClientFactory {
+	GinkgoHelper()
+	factory, err := nodes.NewTunnelClientFactory("", workerBackendDialerFor)
+	Expect(err).ToNot(HaveOccurred())
+	return factory
+}

tests/e2e/distributed/distributed_full_flow_test.go

@@ -10,7 +10,6 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/workerctl"
 	"github.com/mudler/LocalAI/pkg/grpc/base"
@@ -226,22 +225,22 @@ var _ = Describe("Full Distributed Inference Flow", Label("Distributed"), func()
 
 	// newTestSmartRouter creates a SmartRouter reaching a fleet of fake workers
 	// over the tunnelled control plane, with a backend.install handler that
-	// always replies success for every registered node.
+	// replies success for every registered node and names where that node's
+	// backend process listens.
+	//
+	// Both halves of the post-tunnel contract are here rather than in the
+	// specs, because they are the same two facts in every one of them: an
+	// install reply that names no process address is refused
+	// (installBackendOnNode), and a frontend with no worker dialer reaches no
+	// backend at all (nodes.ErrNoWorkerDialer).
 	newTestSmartRouter := func(reg *nodes.NodeRegistry, extraOpts ...nodes.SmartRouterOptions) *nodes.SmartRouter {
 		workers := NewControlWorkers()
-		workers.On(AnyNode, workerctl.PathBackendInstall, func(string, []byte) any {
-			return messaging.BackendInstallReply{Success: true}
-		})
-		workers.On(AnyNode, workerctl.PathModelsRunning, func(string, []byte) any {
-			return messaging.ModelsRunningReply{}
-		})
-		workers.On(AnyNode, workerctl.PathBackendList, func(string, []byte) any {
-			return messaging.BackendListReply{}
-		})
+		workers.ServeBackendLifecycle(reg)
 		unloader := nodes.NewRemoteUnloaderAdapter(reg, infra.NC, workers.Client(), 3*time.Minute, 15*time.Minute)
 
 		opts := nodes.SmartRouterOptions{
-			Unloader: unloader,
+			Unloader:      unloader,
+			ClientFactory: tunnelBackendClients(),
 		}
 		if len(extraOpts) > 0 {
 			o := extraOpts[0]
@@ -332,8 +331,12 @@ var _ = Describe("Full Distributed Inference Flow", Label("Distributed"), func()
 		Expect(registry.Register(context.Background(), node2, true)).To(Succeed())
 
 		// Set both as having the model loaded
-		Expect(registry.SetNodeModel(context.Background(), node1.ID, "test-model", 0, "loaded", "", 0)).To(Succeed())
-		Expect(registry.SetNodeModel(context.Background(), node2.ID, "test-model", 0, "loaded", "", 0)).To(Succeed())
+		// The address is where that node's backend process listens, which is
+		// what a real install reply would have recorded on the row. A row
+		// carrying none names no process, so the router would re-install rather
+		// than route to it and this spec would be measuring the install path.
+		Expect(registry.SetNodeModel(context.Background(), node1.ID, "test-model", 0, "loaded", addr1, 0)).To(Succeed())
+		Expect(registry.SetNodeModel(context.Background(), node2.ID, "test-model", 0, "loaded", addr2, 0)).To(Succeed())
 
 		// Set node-1 with high in-flight (5), node-2 with low in-flight (1)
 		for range 5 {

tests/e2e/distributed/nats_jwt_helpers_test.go

@@ -153,4 +153,4 @@ func accountPublicKeyFromSeed(accountSeed string) string {
 func nodeSubjectPrefix(nodeID string) string {
 	tok := strings.NewReplacer(".", "-", "*", "-", ">", "-", " ", "-", "\t", "-", "\n", "-").Replace(nodeID)
 	return "nodes." + tok
-}
+}

tests/e2e/distributed/router_tracking_test.go

@@ -4,9 +4,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/nodes"
-	"github.com/mudler/LocalAI/core/services/workerctl"
 	"github.com/mudler/LocalAI/pkg/grpc/base"
 	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
 
@@ -58,24 +56,21 @@ var _ = Describe("SmartRouter trackingKey", Label("Distributed"), func() {
 		registry, err = nodes.NewNodeRegistry(db)
 		Expect(err).ToNot(HaveOccurred())
 
-		// Mock control plane — backend.install always replies success.
+		// Mock control plane. The install reply names where the backend process
+		// listens on that worker, which is what the frontend routes to now that
+		// a worker advertises no address of its own.
 		workers := NewControlWorkers()
-		workers.On(AnyNode, workerctl.PathBackendInstall, func(string, []byte) any {
-			return messaging.BackendInstallReply{Success: true}
-		})
-		workers.On(AnyNode, workerctl.PathModelsRunning, func(string, []byte) any {
-			return messaging.ModelsRunningReply{}
-		})
-		workers.On(AnyNode, workerctl.PathBackendList, func(string, []byte) any {
-			return messaging.BackendListReply{}
-		})
+		workers.ServeBackendLifecycle(registry)
 
 		// Start a mock gRPC backend using the same helper as full flow tests
 		llm := &trackingTestLLM{}
 		grpcAddr, grpcCleanup, err = startTestGRPCServer(grpcPkg.AIModel(llm))
 		Expect(err).ToNot(HaveOccurred())
 
-		// Register a node pointing to the mock backend
+		// Register a node whose backend process is the mock server above. The
+		// address is the spec's own record of where that process listens; the
+		// fake worker reports it back on install, and nothing in production
+		// reads this column any more.
 		node := &nodes.BackendNode{
 			Name: "tracking-node", Address: grpcAddr,
 		}
@@ -85,6 +80,11 @@ var _ = Describe("SmartRouter trackingKey", Label("Distributed"), func() {
 		unloader := nodes.NewRemoteUnloaderAdapter(registry, infra.NC, workers.Client(), 3*time.Minute, 15*time.Minute)
 		router = nodes.NewSmartRouter(registry, nodes.SmartRouterOptions{
 			Unloader: unloader,
+			// Without a worker dialer the default factory refuses every
+			// request (nodes.ErrNoWorkerDialer), which is a boot-time
+			// misconfiguration rather than a routing outcome any of these
+			// specs is about.
+			ClientFactory: tunnelBackendClients(),
 		})
 	})