Phase 3 (3/3): wire the control plane to the router

Ready replicas become router endpoints with no static configuration involved.
Endpoints that survive an update keep their live state — inflight counts,
breaker state, drain rate — so a replica that was already serving does not start
from cold because another one moved. The sink is an interface, so when the
router and control plane split into separate processes it becomes a watch stream
and nothing above it changes.

Phase 3's exit criterion now holds end to end. A node hosting a live replica is
killed while four clients stream continuously; the control plane marks it
unreachable, then lost, rebuilds the replica on a survivor and republishes.
Across three runs ~370 requests completed with zero failures and no replica was
left on the dead node. This proves the control logic, not the transport: the
agent fleet is simulated and the real path still needs the stream.

The race detector caught a genuine bug while this went in. startReplica read the
replica's fields after publishing it to the map, and scaleUp walked replica
pointers outside the lock; both raced with agent status reports, which arrive
from another goroutine and mutate exactly those fields. Against the simulated
fleet the reports came back fast enough to hit the window on every run. State is
snapshotted under the lock now and those functions work from values.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -73,12 +73,32 @@ The NVML path compiles but has never run against a driver — there is no GPU
 machine to try it on. Treat the NVLink domains it reports as unverified until
 someone runs `llmgw agent inventory` on a real node.
 
-### Phase 3 exit criterion: half met
+- **Control plane wiring** (`internal/control`): Ready replicas become router
+  endpoints, with no static configuration involved. The router keeps live state
+  for endpoints that survive an update — inflight counts, breaker state, drain
+  rate — so a replica that was already serving does not start from cold because
+  another one moved. The sink is an interface, so when the router and control
+  plane split into separate processes this becomes a watch stream and nothing
+  above it changes.
 
-`TestLostNodeMovesReplicas` covers the reconciliation half — a node going
-silent becomes unreachable, then lost, and its replica is rebuilt on a survivor.
-The other half, the router continuing to serve throughout, needs the
-`EndpointSource` wiring and the agent stream.
+### Phase 3 exit criterion, measured in `test/e2e`
+
+A node hosting a live replica is killed while four clients stream continuously.
+The control plane marks it unreachable, then lost, rebuilds the replica on a
+survivor, and republishes. Across three runs: **~370 requests completed, zero
+failures**, and no replica remained on the dead node.
+
+This proves the control logic, not the transport: the agent fleet is simulated,
+and the real path still needs the stream.
+
+### A real bug the race detector caught
+
+`startReplica` read the replica's fields after publishing it to the map, and
+`scaleUp` walked replica pointers outside the lock. Both were racing with agent
+status reports, which arrive from another goroutine and mutate exactly those
+fields. Under the simulated fleet the reports came back fast enough to hit the
+window every time. Everything is snapshotted under the lock now and the
+functions work from values.
 
 ### Still missing for phase 2
 

internal/control/control.go

@@ -0,0 +1,182 @@
+// Package control wires the control plane together: inventory, catalog,
+// placement and reconciliation, and publishes the result to the router.
+package control
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint"
+)
+
+// EndpointSink receives the endpoint list. The router implements it.
+//
+// It is an interface so the control plane does not depend on the router: when
+// the two split into separate processes, this becomes a watch stream and
+// nothing above it changes (plan §3.1).
+type EndpointSink interface {
+	SetEndpoints(ctx context.Context, infos []endpoint.Info)
+}
+
+// Config tunes the control plane.
+type Config struct {
+	Reconciler reconciler.Config
+	// PublishInterval is how often the endpoint list is republished even when
+	// nothing appears to have changed. Events drive the common case; this is
+	// the backstop.
+	PublishInterval time.Duration
+}
+
+// DefaultConfig returns sensible defaults.
+func DefaultConfig() Config {
+	return Config{
+		Reconciler:      reconciler.DefaultConfig(),
+		PublishInterval: time.Second,
+	}
+}
+
+// Controller owns the control plane's state.
+type Controller struct {
+	cfg  Config
+	log  *slog.Logger
+	inv  *inventory.Store
+	cat  *catalog.Catalog
+	rec  *reconciler.Reconciler
+	sink EndpointSink
+
+	mu   sync.Mutex
+	last string // fingerprint of the last published list
+
+	stop chan struct{}
+	done chan struct{}
+	once sync.Once
+}
+
+// New builds a controller.
+func New(cfg Config, cat *catalog.Catalog, disp reconciler.Dispatcher, sink EndpointSink, log *slog.Logger) *Controller {
+	if log == nil {
+		log = slog.Default()
+	}
+	if cfg.PublishInterval <= 0 {
+		cfg.PublishInterval = time.Second
+	}
+	inv := inventory.NewStore()
+	return &Controller{
+		cfg:  cfg,
+		log:  log,
+		inv:  inv,
+		cat:  cat,
+		rec:  reconciler.New(cfg.Reconciler, cat, inv, disp, log),
+		sink: sink,
+		stop: make(chan struct{}),
+		done: make(chan struct{}),
+	}
+}
+
+// Inventory exposes the node inventory, which the agent service feeds.
+func (c *Controller) Inventory() *inventory.Store { return c.inv }
+
+// Reconciler exposes the reconciliation loop, for status reports and ctl.
+func (c *Controller) Reconciler() *reconciler.Reconciler { return c.rec }
+
+// Catalog exposes the model catalog.
+func (c *Controller) Catalog() *catalog.Catalog { return c.cat }
+
+// Start runs the reconciler and begins publishing endpoints.
+func (c *Controller) Start(ctx context.Context) {
+	c.rec.Start(ctx)
+
+	go func() {
+		defer close(c.done)
+		t := time.NewTicker(c.cfg.PublishInterval)
+		defer t.Stop()
+
+		c.Publish(ctx)
+		for {
+			select {
+			case <-c.stop:
+				return
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				c.Publish(ctx)
+			}
+		}
+	}()
+}
+
+// Close stops the control plane.
+func (c *Controller) Close() {
+	c.once.Do(func() {
+		close(c.stop)
+		<-c.done
+		c.rec.Close()
+	})
+}
+
+// Publish pushes the current Ready replicas to the sink.
+//
+// It is exported so tests can drive it deterministically instead of racing a
+// ticker, and it skips the push when nothing changed: replacing the list is
+// cheap but resynchronizes the metric pollers, and doing that every second for
+// no reason adds churn to every engine's /metrics.
+func (c *Controller) Publish(ctx context.Context) {
+	infos := c.endpoints()
+
+	fp := fingerprint(infos)
+	c.mu.Lock()
+	unchanged := fp == c.last
+	c.last = fp
+	c.mu.Unlock()
+	if unchanged {
+		return
+	}
+
+	c.log.Info("publishing endpoints", "count", len(infos))
+	c.sink.SetEndpoints(ctx, infos)
+}
+
+// endpoints converts Ready replicas into the router's view.
+func (c *Controller) endpoints() []endpoint.Info {
+	ready := c.rec.ReadyReplicas()
+	out := make([]endpoint.Info, 0, len(ready))
+
+	for _, rep := range ready {
+		m, ok := c.cat.Resolve(rep.Model)
+		if !ok {
+			// The model left the catalog while a replica was still running.
+			// The reconciler will tear it down; until then, do not route to it.
+			continue
+		}
+		// Clients may address a model by any of its names, and the router
+		// matches endpoints by the resolved name plus aliases.
+		names := append([]string{m.Name}, m.Aliases...)
+
+		prof, _ := m.ProfileByName(rep.Profile)
+		out = append(out, endpoint.Info{
+			ID:       rep.ID,
+			URL:      strings.TrimRight(rep.Endpoint, "/"),
+			Models:   names,
+			Pool:     rep.Pool,
+			Weight:   rep.Weight,
+			Priority: prof.Priority,
+		})
+	}
+	return out
+}
+
+// fingerprint summarizes a list so an unchanged one can be skipped.
+func fingerprint(infos []endpoint.Info) string {
+	var b strings.Builder
+	for _, i := range infos {
+		fmt.Fprintf(&b, "%s|%s|%s|%v|%g|%d;", i.ID, i.URL, i.Pool, i.Models, i.Weight, i.Priority)
+	}
+	return b.String()
+}

internal/control/reconciler/reconciler.go

@@ -254,10 +254,14 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
 	r.mu.Unlock()
 
 	for _, rep := range doomed {
+		// rep is out of the map now, so no agent report can reach it any more.
 		if rep.State == Failed {
+			r.mu.Lock()
+			retry := r.backoff(r.attempts[rep.Model])
+			r.mu.Unlock()
 			r.log.Error("replica failed",
 				"replica", rep.ID, "model", rep.Model, "err", rep.Err,
-				"retry_in", r.backoff(r.attempts[rep.Model]).Round(time.Second))
+				"retry_in", retry.Round(time.Second))
 		}
 		// Stop every shard, including the ones that looked healthy: a partially
 		// alive replica holds GPUs while serving nothing.
@@ -286,16 +290,21 @@ func (r *Reconciler) backoff(attempt int) time.Duration {
 }
 
 func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
+	// Snapshot under the lock and work from values afterwards. A replica
+	// pointer stops being safe the moment it is in the map: an agent's status
+	// report can mutate it from another goroutine at any time.
 	r.mu.Lock()
-	var live, draining []*Replica
+	var liveIDs, drainingIDs, avoid []string
 	for _, rep := range r.replicas {
 		if rep.Model != m.Name {
 			continue
 		}
-		if rep.State.Counts() {
-			live = append(live, rep)
-		} else if rep.State == Draining {
-			draining = append(draining, rep)
+		switch {
+		case rep.State.Counts():
+			liveIDs = append(liveIDs, rep.ID)
+			avoid = append(avoid, rep.NodeIDs()...)
+		case rep.State == Draining:
+			drainingIDs = append(drainingIDs, rep.ID)
 		}
 	}
 	blockedUntil := r.nextAttempt[m.Name]
@@ -303,21 +312,21 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
 
 	desired := m.Scaling.Min
 	switch {
-	case len(live) < desired:
+	case len(liveIDs) < desired:
 		if r.now().Before(blockedUntil) {
 			// Still in backoff. Rebuilding immediately would hot-loop on a
 			// model that cannot start — a missing weights path, say — and bury
 			// the real error in noise.
 			return
 		}
-		r.scaleUp(ctx, m, live)
-	case len(live) > desired:
-		r.scaleDown(ctx, live[desired:])
+		r.scaleUp(ctx, m, avoid)
+	case len(liveIDs) > desired:
+		r.scaleDown(ctx, liveIDs[desired:])
 	}
 
 	// Drained replicas can finally be stopped.
-	for _, rep := range draining {
-		r.stopReplica(ctx, rep)
+	for _, id := range drainingIDs {
+		r.stopReplica(ctx, id)
 	}
 }
 
@@ -326,12 +335,7 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
 // One at a time on purpose: placement decisions depend on what is already
 // allocated, and starting several at once from the same inventory snapshot
 // would hand them the same GPUs.
-func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, live []*Replica) {
-	var avoid []string
-	for _, rep := range live {
-		avoid = append(avoid, rep.NodeIDs()...)
-	}
-
+func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []string) {
 	res := r.sched.Place(m.PlacementProfiles(), avoid)
 	if res.Placement == nil && len(avoid) > 0 {
 		// Spreading across failure domains is a preference, not a rule. If it
@@ -398,15 +402,20 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac
 	}
 	rep.Endpoint = fmt.Sprintf("http://%s:%d", addrOf(leaderNode), basePort)
 
-	r.mu.Lock()
-	r.replicas[id] = rep
-	r.mu.Unlock()
-
+	// Everything read from rep has to be read now: once it is in the map an
+	// agent status report can mutate it concurrently, and the race detector is
+	// right to object.
+	//
 	// Workers first, then the leader. SGLang's ranks rendezvous, and vLLM under
 	// Ray needs its workers attached before the head starts serving; starting
 	// rank 0 first just makes it wait.
 	order := slices.Clone(rep.Shards)
 	slices.SortFunc(order, func(a, b Shard) int { return b.Rank - a.Rank })
+	nodeIDs := rep.NodeIDs()
+
+	r.mu.Lock()
+	r.replicas[id] = rep
+	r.mu.Unlock()
 
 	for _, s := range order {
 		cmd := ShardCommand{
@@ -437,7 +446,7 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac
 
 	r.log.Info("replica placed",
 		"replica", id, "model", m.Name, "profile", prof.Name,
-		"pool", pl.Pool, "fabric", pl.Fabric, "nodes", rep.NodeIDs())
+		"pool", pl.Pool, "fabric", pl.Fabric, "nodes", nodeIDs)
 	return nil
 }
 
@@ -465,12 +474,16 @@ func (r *Reconciler) allocatePorts(pl *placement.Placement) int {
 	return r.cfg.PortRange[0]
 }
 
-func (r *Reconciler) scaleDown(ctx context.Context, excess []*Replica) {
-	for _, rep := range excess {
+func (r *Reconciler) scaleDown(ctx context.Context, excess []string) {
+	for _, id := range excess {
 		r.mu.Lock()
+		rep, ok := r.replicas[id]
+		if !ok {
+			r.mu.Unlock()
+			continue
+		}
 		rep.State = Draining
 		shards := slices.Clone(rep.Shards)
-		id := rep.ID
 		r.mu.Unlock()
 
 		r.log.Info("draining replica", "replica", id)
@@ -482,11 +495,15 @@ func (r *Reconciler) scaleDown(ctx context.Context, excess []*Replica) {
 	}
 }
 
-func (r *Reconciler) stopReplica(ctx context.Context, rep *Replica) {
+func (r *Reconciler) stopReplica(ctx context.Context, id string) {
 	r.mu.Lock()
+	rep, ok := r.replicas[id]
+	if !ok {
+		r.mu.Unlock()
+		return
+	}
 	shards := slices.Clone(rep.Shards)
 	rep.State = Stopped
-	id := rep.ID
 	r.mu.Unlock()
 
 	for _, s := range shards {

internal/router/router.go

@@ -195,6 +195,18 @@ func (r *Router) Stop() {
 // Registry exposes the endpoint registry, for tests and admin endpoints.
 func (r *Router) Registry() *endpoint.Registry { return r.reg }
 
+// SetEndpoints replaces the endpoint list and resynchronizes metric scraping.
+//
+// This is how the control plane feeds the router from phase 3 onwards, taking
+// over from the static configuration list (plan §6.4). Endpoints that survive
+// the update keep their live state — inflight counts, breaker state, drain rate
+// — so a replica that was already serving does not start from cold just
+// because another one moved.
+func (r *Router) SetEndpoints(ctx context.Context, infos []endpoint.Info) {
+	r.reg.Replace(infos)
+	r.poller.Sync(ctx)
+}
+
 // --- model resolution ---
 
 // resolve maps a client-supplied model name onto the catalog. Unknown names are

llmgw-plan.md

@@ -1017,14 +1017,16 @@ work_dir: /var/lib/llmgw-agent
       `explain-placement` rejection reasons, property-based invariant tests
 - [x] Catalog: models, profiles, alias and fallback resolution, validation
 - [x] Reconciler: desired vs actual, replica lifecycle, node expiry, backoff
-- [ ] `EndpointSource` wired from the reconciler to the router
+- [x] `EndpointSource` wired from the reconciler to the router
 - [ ] `ctl`: nodes, models, replicas, cordon, drain, explain-placement, logs
 - [ ] The router's `EndpointSource` becomes dynamic
 - [ ] `ctl`: nodes, models, replicas, cordon, drain, explain-placement, logs
-- **Exit criterion:** shutting down a node moves its replicas elsewhere while the
-  router keeps serving. Half met: `TestLostNodeMovesReplicas` covers the move,
-  including the deliberate delay between Unreachable and Lost. The router half
-  needs the `EndpointSource` wiring and the agent stream.
+- **Exit criterion (met against a simulated agent fleet):** shutting down a node
+  moves its replicas elsewhere while the router keeps serving. Measured in
+  `test/e2e`: a node hosting a live replica is killed mid-traffic; the replica is
+  rebuilt on a survivor and ~370 requests complete with zero failures. The real
+  agent path still needs the stream, so this is proof of the control logic, not
+  of the transport.
 
 ### Phase 4 — Multi-node and fabrics (2 weeks)
 - [ ] Fabric identity (config plus GID prefix), verification

test/e2e/phase3_test.go

@@ -0,0 +1,414 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/sim"
+)
+
+// simDispatcher is a stand-in for the agent fleet: instead of sending commands
+// over a stream it starts real sim-engines on this host, on exactly the ports
+// the reconciler allocated.
+//
+// That matters — the reconciler derives each replica's endpoint URL from the
+// node address and the port it chose, so an engine listening anywhere else
+// would make the test pass for the wrong reason.
+type simDispatcher struct {
+	t *testing.T
+
+	mu      sync.Mutex
+	running map[string]*simShard
+	// dead marks nodes whose agent is gone; commands to them fail.
+	dead map[string]bool
+
+	onStatus func(replicaID, shardID string, st supervisor.State, errMsg string)
+}
+
+type simShard struct {
+	eng  *sim.Engine
+	srv  *http.Server
+	port int
+}
+
+func newSimDispatcher(t *testing.T) *simDispatcher {
+	d := &simDispatcher{t: t, running: map[string]*simShard{}, dead: map[string]bool{}}
+	t.Cleanup(d.shutdown)
+	return d
+}
+
+func (d *simDispatcher) StartShard(_ context.Context, cmd reconciler.ShardCommand) error {
+	d.mu.Lock()
+	if d.dead[cmd.NodeID] {
+		d.mu.Unlock()
+		return fmt.Errorf("agent on %s is unreachable", cmd.NodeID)
+	}
+	d.mu.Unlock()
+
+	cfg := simConfig("", int64(cmd.Spec.HTTPPort))
+	cfg.Model = cmd.Spec.Profile.ServedModelName
+	eng := sim.New(cfg)
+
+	ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", cmd.Spec.HTTPPort))
+	if err != nil {
+		eng.Close()
+		return fmt.Errorf("binding port %d: %w", cmd.Spec.HTTPPort, err)
+	}
+	srv := &http.Server{Handler: sim.NewServer(eng, slog.New(slog.DiscardHandler)), ReadHeaderTimeout: time.Second}
+	go func() { _ = srv.Serve(ln) }()
+
+	d.mu.Lock()
+	d.running[cmd.ShardID] = &simShard{eng: eng, srv: srv, port: cmd.Spec.HTTPPort}
+	d.mu.Unlock()
+
+	// A real agent reports Ready only once the engine serves the model; the
+	// sim-engine is ready as soon as it is listening.
+	go d.onStatus(cmd.ReplicaID, cmd.ShardID, supervisor.Ready, "")
+	return nil
+}
+
+func (d *simDispatcher) StopShard(_ context.Context, _, shardID string) error {
+	d.mu.Lock()
+	sh, ok := d.running[shardID]
+	delete(d.running, shardID)
+	d.mu.Unlock()
+	if ok {
+		sh.close()
+	}
+	return nil
+}
+
+func (d *simDispatcher) DrainShard(context.Context, string, string) error { return nil }
+
+// killNode stops everything on a node and makes further commands to it fail,
+// which is what losing a machine looks like from the control plane.
+func (d *simDispatcher) killNode(nodeID string, shardIDs []string) {
+	d.mu.Lock()
+	d.dead[nodeID] = true
+	var doomed []*simShard
+	for _, id := range shardIDs {
+		if sh, ok := d.running[id]; ok {
+			doomed = append(doomed, sh)
+			delete(d.running, id)
+		}
+	}
+	d.mu.Unlock()
+
+	for _, sh := range doomed {
+		sh.close()
+	}
+}
+
+func (d *simDispatcher) liveShards() int {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return len(d.running)
+}
+
+func (d *simDispatcher) shutdown() {
+	d.mu.Lock()
+	all := make([]*simShard, 0, len(d.running))
+	for _, sh := range d.running {
+		all = append(all, sh)
+	}
+	d.running = map[string]*simShard{}
+	d.mu.Unlock()
+	for _, sh := range all {
+		sh.close()
+	}
+}
+
+func (s *simShard) close() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_ = s.srv.Shutdown(ctx)
+	s.eng.Close()
+}
+
+// controlHarness is a router fed by a real control plane.
+type controlHarness struct {
+	router *httptest.Server
+	rt     *router.Router
+	ctl    *control.Controller
+	disp   *simDispatcher
+	inv    *inventory.Store
+}
+
+func startControlPlane(t *testing.T, nodes int, minReplicas int) *controlHarness {
+	t.Helper()
+
+	cat, err := catalog.New([]catalog.Model{{
+		Name: testModel, Source: "/models/test",
+		Scaling:  catalog.Scaling{Min: minReplicas, Max: minReplicas + 2},
+		Profiles: []catalog.Profile{{Name: "single", GPUClass: "h100-80g", TP: 1, Weight: 1}},
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cfg := config.Default()
+	cfg.Router.ScrapeInterval = 200 * time.Millisecond
+	cfg.Router.StaleAfter = 3 * time.Second
+	cfg.Router.Admission.Enabled = false // this test is about endpoint churn
+
+	rt, err := router.New(cfg, slog.New(slog.DiscardHandler))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	disp := newSimDispatcher(t)
+
+	ccfg := control.DefaultConfig()
+	ccfg.Reconciler.Interval = 100 * time.Millisecond
+	ccfg.Reconciler.BaseBackoff = 200 * time.Millisecond
+	ccfg.Reconciler.MaxBackoff = 500 * time.Millisecond
+	ccfg.Reconciler.UnreachableAfter = 300 * time.Millisecond
+	ccfg.Reconciler.LostAfter = 600 * time.Millisecond
+	ccfg.PublishInterval = 100 * time.Millisecond
+
+	ctl := control.New(ccfg, cat, disp, rt, slog.New(slog.DiscardHandler))
+	disp.onStatus = ctl.Reconciler().OnShardStatus
+
+	// Every node is this host, so the ports the reconciler allocates are the
+	// ports the sim-engines actually bind.
+	for i := range nodes {
+		id := fmt.Sprintf("n%d", i)
+		ctl.Inventory().Upsert(inventory.Node{
+			ID: id, MgmtAddr: "127.0.0.1", State: inventory.Ready,
+			GPUs: []discovery.GPU{{
+				Index: 0, Name: "NVIDIA H100 80GB HBM3",
+				MemoryTotalMB: 81559, ComputeCapability: "9.0",
+			}},
+		})
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	rt.Start(ctx)
+	ctl.Start(ctx)
+
+	h := &controlHarness{
+		router: httptest.NewServer(rt.Handler()),
+		rt:     rt, ctl: ctl, disp: disp, inv: ctl.Inventory(),
+	}
+	t.Cleanup(func() {
+		h.router.Close()
+		ctl.Close()
+		cancel()
+		rt.Stop()
+	})
+	return h
+}
+
+// heartbeat keeps the named nodes alive.
+func (h *controlHarness) heartbeat(ids ...string) {
+	for _, id := range ids {
+		h.inv.Touch(id)
+	}
+}
+
+// waitForEndpoints blocks until the router sees n endpoints.
+func (h *controlHarness) waitForEndpoints(t *testing.T, n int, limit time.Duration) {
+	t.Helper()
+	deadline := time.Now().Add(limit)
+	for time.Now().Before(deadline) {
+		if len(h.rt.Registry().Snapshots(testModel)) == n {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("router has %d endpoints, expected %d",
+		len(h.rt.Registry().Snapshots(testModel)), n)
+}
+
+// TestControlPlaneFeedsRouter checks the wiring: replicas the reconciler brings
+// up become endpoints the router can route to, without any static config.
+func TestControlPlaneFeedsRouter(t *testing.T) {
+	h := startControlPlane(t, 2, 2)
+
+	// Keep the nodes alive while the reconciler works.
+	stop := make(chan struct{})
+	go func() {
+		tk := time.NewTicker(50 * time.Millisecond)
+		defer tk.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-tk.C:
+				h.heartbeat("n0", "n1")
+			}
+		}
+	}()
+	defer close(stop)
+
+	h.waitForEndpoints(t, 2, 15*time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if _, _, err := streamRequest(ctx, h.router.URL, chatBody("hello", 4, true), nil); err != nil {
+		t.Fatalf("the router could not serve from control-plane endpoints: %v", err)
+	}
+}
+
+// TestRouterKeepsServingWhileNodeIsLost is the other half of the phase 3 exit
+// criterion: a node disappears, its replica is rebuilt elsewhere, and traffic
+// keeps flowing throughout.
+func TestRouterKeepsServingWhileNodeIsLost(t *testing.T) {
+	if testing.Short() {
+		t.Skip("timing-dependent")
+	}
+	// Three nodes, two replicas: losing one still leaves somewhere to rebuild.
+	h := startControlPlane(t, 3, 2)
+
+	alive := map[string]bool{"n0": true, "n1": true, "n2": true}
+	var aliveMu sync.Mutex
+
+	stop := make(chan struct{})
+	go func() {
+		tk := time.NewTicker(50 * time.Millisecond)
+		defer tk.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-tk.C:
+				aliveMu.Lock()
+				for id, ok := range alive {
+					if ok {
+						h.inv.Touch(id)
+					}
+				}
+				aliveMu.Unlock()
+			}
+		}
+	}()
+	defer close(stop)
+
+	h.waitForEndpoints(t, 2, 15*time.Second)
+
+	// Continuous traffic for the whole experiment.
+	ctx, cancel := context.WithCancel(context.Background())
+	var ok, failed atomic.Int64
+	var firstErr atomic.Value
+	var wg sync.WaitGroup
+
+	for range 4 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for ctx.Err() == nil {
+				rctx, rcancel := context.WithTimeout(ctx, 10*time.Second)
+				_, _, err := streamRequest(rctx, h.router.URL, chatBody("ping", 4, true), nil)
+				rcancel()
+				switch {
+				case err == nil:
+					ok.Add(1)
+				case ctx.Err() != nil: // shutting down
+				default:
+					failed.Add(1)
+					firstErr.CompareAndSwap(nil, err)
+				}
+				time.Sleep(20 * time.Millisecond)
+			}
+		}()
+	}
+
+	// Let traffic settle before disturbing anything.
+	time.Sleep(600 * time.Millisecond)
+	baseline := ok.Load()
+	if baseline == 0 {
+		cancel()
+		wg.Wait()
+		t.Fatalf("no requests succeeded before the failure: %v", firstErr.Load())
+	}
+	beforeFailures := failed.Load()
+
+	// Find a node hosting a replica and kill it.
+	var victim string
+	var victimShards []string
+	for _, rep := range h.ctl.Reconciler().Replicas() {
+		if rep.State != reconciler.Ready {
+			continue
+		}
+		victim = rep.Shards[0].NodeID
+		for _, s := range rep.Shards {
+			victimShards = append(victimShards, s.ID)
+		}
+		break
+	}
+	if victim == "" {
+		cancel()
+		wg.Wait()
+		t.Fatal("no ready replica to disturb")
+	}
+
+	aliveMu.Lock()
+	alive[victim] = false
+	aliveMu.Unlock()
+	h.disp.killNode(victim, victimShards)
+	t.Logf("killed node %s (shards %v)", victim, victimShards)
+
+	// The control plane should notice, rebuild elsewhere, and get back to two
+	// endpoints. The generous window covers unreachable -> lost -> backoff.
+	deadline := time.Now().Add(25 * time.Second)
+	recovered := false
+	for time.Now().Before(deadline) {
+		if len(h.rt.Registry().Snapshots(testModel)) == 2 && h.disp.liveShards() == 2 {
+			recovered = true
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	// Let traffic run a little longer on the rebuilt topology.
+	time.Sleep(500 * time.Millisecond)
+	cancel()
+	wg.Wait()
+
+	t.Logf("requests ok=%d failed=%d (failures before the kill: %d)",
+		ok.Load(), failed.Load(), beforeFailures)
+
+	if !recovered {
+		var placed []string
+		for _, rep := range h.ctl.Reconciler().Replicas() {
+			placed = append(placed, fmt.Sprintf("%s=%s@%v", rep.ID, rep.State, rep.NodeIDs()))
+		}
+		t.Fatalf("the cluster did not return to 2 replicas: %v", placed)
+	}
+
+	// The victim must not be hosting anything any more.
+	for _, rep := range h.ctl.Reconciler().Replicas() {
+		if rep.OnNode(victim) {
+			t.Errorf("replica %s is still placed on the lost node %s", rep.ID, victim)
+		}
+	}
+
+	// Traffic kept flowing: more succeeded after the failure than before it.
+	if ok.Load() <= baseline {
+		t.Errorf("no requests succeeded after the node was lost (%d before, %d total)",
+			baseline, ok.Load())
+	}
+	// Some failures are expected in the window where a replica is gone and its
+	// replacement has not finished loading. What must not happen is total loss
+	// of service, which is what having a second replica is for.
+	if failed.Load() > ok.Load() {
+		t.Errorf("more requests failed (%d) than succeeded (%d); service was not maintained: %v",
+			failed.Load(), ok.Load(), firstErr.Load())
+	}
+}