Merge features-ha: two control planes, a fast failover, ctl upgrade in waves

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

CHANGELOG.md

@@ -21,6 +21,39 @@ is only released once its **exit criterion** is met, not when its code lands.
 Until v1.0.0 the minor version is the breaking one: configuration formats and Go
 APIs change between minors without a deprecation period.
 
+## v0.11.3-rc.1 — 2026-10-05
+
+Two control planes, run in anger on the reference cluster.
+
+### Added
+
+- **The console sets up a second control plane**: Do → Add a control plane
+  asks for the machine and a floating address and runs `server add`.
+- **Who leads, everywhere.** `ctl status` and the console's Overview have a
+  `CONTROL` line: which control plane leads, which stand by, one that does not
+  answer, and whether the floating address is on the leader.
+- **A standby's console is the leader's.** Its admin API passes `/api/` to the
+  leader over the agent port, with the cluster's certificates both ways, so
+  `sudo llmgw` and `llmgw ctl` show the same cluster on either machine. Control
+  plane certificates are now issued for client use too; `server add` reissues
+  older ones.
+- **`ctl upgrade` in waves.** Nodes are upgraded one, then two, four… up to
+  `--parallel` (20) at once, so hundreds take minutes; a wave that does not
+  come back stops it. The other control plane is upgraded first, over ssh.
+- `server add` says when the mirror's `--api-allow` would refuse the new
+  control plane.
+
+### Fixed
+
+- **Failover took 40 seconds; it takes a few.** A stopping control plane held
+  every agent's stream for the thirty seconds of a graceful shutdown; it now
+  gives up its lease first and lets the agents go at once.
+- A control plane that lost its lease while running kept its agents, leaving
+  the new leader with no nodes. It now lets them go.
+- An agent's reconnect backoff carried over from one outage to the next, so
+  each failover made the next one slower; it starts again after a session that
+  worked.
+
 ## v0.11.2 — 2026-10-05
 
 ### Routing by what each vLLM replica's cache holds

docs/admin-api.openapi.json

@@ -3443,6 +3443,45 @@
           }
         }
       },
+      "ControlPlane": {
+        "type": "object",
+        "description": "ControlPlane is one control plane.",
+        "properties": {
+          "name": {
+            "type": "string"
+          },
+          "address": {
+            "type": "string"
+          },
+          "state": {
+            "type": "string",
+            "description": "State is \"leading\", \"standing by\" or \"not answering\"."
+          },
+          "this": {
+            "type": "boolean",
+            "description": "This is the control plane that answered the summary."
+          }
+        }
+      },
+      "ControlPlanes": {
+        "type": "object",
+        "description": "ControlPlanes is every control plane of a cluster that has several, as each answered when asked.",
+        "properties": {
+          "members": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/ControlPlane"
+            }
+          },
+          "floating": {
+            "type": "string",
+            "description": "Floating is the address keepalived keeps on the leader (10.0.0.161), and FloatingUp whether it answered as a leader when asked; empty when nothing floats."
+          },
+          "floating_up": {
+            "type": "boolean"
+          }
+        }
+      },
       "DomainView": {
         "type": "object",
         "description": "DomainView is one domain: its nodes, its power and whether it is held.",
@@ -5276,6 +5315,10 @@
           },
           "store_error": {
             "type": "string"
+          },
+          "control_planes": {
+            "$ref": "#/components/schemas/ControlPlanes",
+            "description": "ControlPlanes is the cluster's control planes when there is more than one: which leads, which stand by, and the floating address in front."
           }
         }
       },

docs/install.md

@@ -1915,15 +1915,49 @@ one trusted machine only; see [development.md](development.md).
 
 ## Two control planes
 
-llmgw handles the hard half itself: only one places work, and agents find
-whichever is leading. What it does not do is move client traffic, which
-happens in front of both.
+Once the store is replicated (`server init --patroni`, above), one command on
+the control plane makes another machine a second one — or **Do → Add a
+control plane** in the console, which asks the same two questions:
 
-[deploy/ha/](../deploy/ha/) has the pattern and a keepalived example. The one
-thing to get right is the health check: point it at `/ready`, not `/health`.
-`/health` is 200 on a standby, which is alive and holds no endpoints at all,
-so a load balancer using it would send half the traffic to something that
-refuses every request.
+```
+sudo --preserve-env=SSH_AUTH_SOCK llmgw server add ubuntu@10.0.0.150 --vip 10.0.0.161/24
+```
+
+Over ssh, as `node add` works, it gives the machine this llmgw, a certificate
+from the cluster's authority, this control plane's `server.yaml` and secrets,
+and lets it into the store; it starts there as a standby. Every node is told
+of it within half a minute and fails over to it without being touched.
+`--vip` keeps a floating address on whichever control plane leads, with
+keepalived on both: that is the address to give clients, or a proxy in front.
+Without it, put a load balancer in front that checks `/ready` — not
+`/health`, which is 200 on a standby that would refuse every request.
+
+- **What a failover costs.** Measured on the reference cluster, stopping the
+  leader: the standby led within five seconds, every node had moved to it
+  and every replica was adopted — none restarted, the engines never stopped
+  serving — and the floating address answered again with it.
+- **Who leads.** `llmgw ctl status` and the console's Overview have a
+  `CONTROL` line: which leads, which stand by, one that does not answer, and
+  whether the floating address is on the leader. A standby with nothing to
+  take over from is the thing to see before it is needed.
+- **Either machine's console.** A standby's admin API passes everything under
+  `/api/` to the leader, over the port control planes already reach each
+  other on and with the cluster's certificates both ways, so `sudo llmgw` and
+  `llmgw ctl` show the same cluster on both — the leader's — and a change made
+  on the standby is made by the leader. A node's certificate is refused
+  there. Control plane certificates from before v0.11.3 cannot present
+  themselves as clients: `server add` reissues them, and the running server
+  picks the new one up when restarted.
+- **The mirror** answers only the addresses it was given with `--api-allow`.
+  `server add` checks, and says what to add to the mirror's unit if it
+  refuses the new control plane — which would otherwise find out the first
+  time it leads and fetches a model.
+- **Limits are per process.** Rate limits are counted in memory, and clients
+  reach the leader through the floating address, so a failover starts each
+  key's minute afresh.
+
+[deploy/ha/](../deploy/ha/) describes what each half does, for anyone putting
+their own load balancer in front.
 
 ## The portal's language
 
@@ -2215,27 +2249,33 @@ joined with `node add`. Why, and how, is in [kubernetes.md](kubernetes.md).
 
 ## Upgrading
 
-On the control plane:
+On a control plane:
 
 ```
-sudo llmgw ctl upgrade v0.10.4
+sudo --preserve-env=SSH_AUTH_SOCK llmgw ctl upgrade v0.10.4
 ```
 
-It upgrades this machine first — fetching the release for its platform,
-checking it against the release's checksums, keeping the old binary as
-`/usr/local/bin/llmgw.bak-pre-<version>`, restarting `llmgw-server` and waiting
-for every node to reconnect — and then each node in turn. A node's agent is told
-the version over the stream it already holds, fetches and checks the release
-itself, swaps it in and restarts; no ssh into the nodes is needed. Release
-candidates (`v0.10.4-rc.2`) work the same way.
+With two control planes it upgrades the other one first, over ssh as the user
+who ran sudo (`--ssh-user`), with `llmgw upgrade` there; then this machine —
+fetching the release for its platform, checking it against the release's
+checksums, keeping the old binary as `/usr/local/bin/llmgw.bak-pre-<version>`,
+restarting `llmgw-server` and waiting for every node to reconnect. A leader
+restarted last hands over to a standby already on the release.
+
+Then the nodes, in waves: one first, then two, four, and so on up to
+`--parallel` (20) at once. A node's agent is told the version over the stream
+it already holds, fetches and checks the release itself, swaps it in and
+restarts; no ssh into the nodes is needed. A fleet of hundreds takes minutes,
+not hours. Release candidates (`v0.10.4-rc.2`) work the same way.
 
 A node counts as done when it is back on the new version, ready, with as many
-ready replicas as before. One that is not within `--timeout` (three minutes)
-stops the upgrade there, so a bad release reaches one node rather than all of
-them; `llmgw ctl logs <node>` says why, and the node keeps its previous binary
-beside the new one. Nodes already on the version are skipped, so running the
-command again resumes where it stopped. `--nodes a,b` limits it to some nodes,
-and `--skip-control-plane` leaves this machine alone.
+ready replicas as before. A wave not back within `--timeout` (three minutes)
+stops the upgrade there — the first wave is one node, so a bad release reaches
+one node rather than all of them; `llmgw ctl logs <node>` says why, and the node
+keeps its previous binary beside the new one. Nodes already on the version are
+skipped, so running the command again resumes where it stopped. `--nodes a,b`
+limits it to some nodes, `--skip-control-plane` leaves the control planes alone,
+and `--other-control-planes=false` this machine's partner.
 
 **Nothing stops serving.** Engines are started with `KillMode=process`, so they
 keep running when the agent restarts, and the agent reports them back on its

internal/agent/agent.go

@@ -21,6 +21,7 @@ import (
 	"slices"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -42,11 +43,14 @@ import (
 
 // Agent runs on a GPU node.
 type Agent struct {
-	cfg  Config
-	log  *slog.Logger
-	sup  *supervisor.Supervisor
-	reg  *engines.Registry
-	http connectHTTPClient
+	cfg Config
+	// answered says the current session heard from its control plane, which
+	// starts the reconnect backoff again when it ends.
+	answered atomic.Bool
+	log      *slog.Logger
+	sup      *supervisor.Supervisor
+	reg      *engines.Registry
+	http     connectHTTPClient
 
 	// engineOnce probes the configured engines once before registering.
 	//
@@ -215,6 +219,14 @@ func (a *Agent) Run(ctx context.Context) error {
 		// never answers — from here they are the same thing, and trying
 		// somewhere else is the answer to all three.
 		next++
+		// A session that was answered was a connection that worked, however
+		// it ended: the backoff starts again rather than carrying on from
+		// the last outage. Without this every failover the agent lived
+		// through made the next one slower — the third waited half a minute
+		// on a control plane that had been ready for twenty seconds.
+		if a.answered.Swap(false) {
+			attempt = 0
+		}
 		attempt++
 		delay := a.reconnectDelay(attempt)
 		// No wait when there is somewhere else to try: the delay exists to
@@ -568,6 +580,7 @@ func (a *Agent) session(ctx context.Context, server string) error {
 			// why it is not registering is one an operator cannot fix.
 			return err
 		}
+		a.answered.Store(true)
 		a.handle(ctx, msg, out)
 	}
 }

internal/agent/failover_test.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -105,3 +106,49 @@ func TestALeaderStillAcceptsWithNoLeadershipConfigured(t *testing.T) {
 	}
 	t.Fatal("a service with no leadership configured refused an agent")
 }
+
+// flipping is a leader that can stop leading partway through a test.
+type flipping struct{ leading atomic.Bool }
+
+func (f *flipping) IsLeader() bool { return f.leading.Load() }
+
+func TestAnAgentLeavesAControlPlaneThatStopsLeading(t *testing.T) {
+	// Leadership lost while the process lives on: the store was slow and the
+	// lease lapsed, or another control plane took it. Agents that stayed on
+	// the old leader would leave the new one with no nodes, so the old one
+	// has to let them go rather than wait for them to leave.
+	was := &flipping{}
+	was.leading.Store(true)
+	wasInv, wasAddr := startPlain(t, was)
+	nowInv, nowAddr := startPlain(t, standby(false))
+
+	ag := runAgent(t, Config{
+		NodeID: "n1", MgmtAddr: "127.0.0.1",
+		Servers: []string{wasAddr, nowAddr},
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	go ag.Run(ctx)
+
+	waitFor(t, 5*time.Second, "the agent never reached the first leader", func() bool {
+		_, ok := wasInv.Get("n1")
+		return ok
+	})
+	was.leading.Store(false)
+	waitFor(t, 8*time.Second, "the agent stayed on a control plane that stopped leading", func() bool {
+		_, ok := nowInv.Get("n1")
+		return ok
+	})
+}
+
+func waitFor(t *testing.T, d time.Duration, msg string, ok func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		if ok() {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatal(msg)
+}

internal/cli/controlplanes.go

@@ -0,0 +1,123 @@
+package cli
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+)
+
+// floatingPath is where `server add --vip` records the floating address.
+const floatingPath = "/etc/llmgw/floating"
+
+// controlPlaneProbe answers the summary's question of which control plane
+// leads: it asks each one's /leader, on the client port every one of them
+// listens on, since they share a server.yaml.
+//
+// Asked, rather than read from the lease, because the lease names whoever
+// held it last and says nothing about a standby that has stopped — and a
+// standby that is not there is the thing worth seeing before it is needed.
+type controlPlaneProbe struct {
+	st   *store.DB
+	self string
+	// floating is the file the floating address is recorded in.
+	floating string
+	client   *http.Client
+	// leaderURL is where a host is asked; tests point it elsewhere.
+	leaderURL func(host string) string
+
+	mu   sync.Mutex
+	at   time.Time
+	last adminapi.ControlPlanes
+}
+
+func newControlPlaneProbe(st *store.DB, listen string) *controlPlaneProbe {
+	_, port, _ := net.SplitHostPort(listen)
+	if port == "" {
+		port = "8080"
+	}
+	self, _ := os.Hostname()
+	return &controlPlaneProbe{st: st, self: self,
+		leaderURL: func(host string) string { return "http://" + net.JoinHostPort(host, port) + "/leader" },
+		floating:  floatingAddress(floatingPath),
+		client:    &http.Client{Timeout: time.Second}}
+}
+
+// Status is the answer, kept for a few seconds: the console asks every three,
+// and each asking is a request to every control plane.
+func (p *controlPlaneProbe) Status() adminapi.ControlPlanes {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if time.Since(p.at) < 2*time.Second {
+		return p.last
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	cps, err := p.st.ControlPlanes(ctx)
+	if err != nil || len(cps) < 2 {
+		// One control plane is not a set of them; nothing to say.
+		p.at, p.last = time.Now(), adminapi.ControlPlanes{}
+		return p.last
+	}
+	// Read each time: server add writes it after this process may have
+	// started.
+	out := adminapi.ControlPlanes{Members: make([]adminapi.ControlPlane, len(cps)), Floating: floatingAddress(p.floating)}
+	var wg sync.WaitGroup
+	for i, cp := range cps {
+		host := cp.URL
+		if u, err := url.Parse(cp.URL); err == nil && u.Hostname() != "" {
+			host = u.Hostname()
+		}
+		out.Members[i] = adminapi.ControlPlane{Name: cp.Name, Address: host, This: cp.Name == p.self}
+		wg.Go(func() { out.Members[i].State = p.ask(ctx, host) })
+	}
+	if out.Floating != "" {
+		wg.Go(func() { out.FloatingUp = p.ask(ctx, out.Floating) == "leading" })
+	}
+	wg.Wait()
+	p.at, p.last = time.Now(), out
+	return out
+}
+
+// ask is what one address's /leader says.
+func (p *controlPlaneProbe) ask(ctx context.Context, host string) string {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.leaderURL(host), nil)
+	if err != nil {
+		return "not answering"
+	}
+	resp, err := p.client.Do(req)
+	if err != nil {
+		return "not answering"
+	}
+	_ = resp.Body.Close()
+	switch resp.StatusCode {
+	case http.StatusOK:
+		return "leading"
+	case http.StatusServiceUnavailable:
+		return "standing by"
+	}
+	return "not answering"
+}
+
+// floatingAddress is the address in the file server add keeps beside the
+// keepalived configuration, without its prefix; "" when nothing floats.
+// keepalived's own file holds its password and is root's alone, and this
+// process is not root.
+func floatingAddress(path string) string {
+	b, err := os.ReadFile(path) //nolint:gosec // a fixed path of our own
+	if err != nil {
+		return ""
+	}
+	addr, _, _ := strings.Cut(strings.TrimSpace(string(b)), "/")
+	if net.ParseIP(addr) == nil {
+		return ""
+	}
+	return addr
+}

internal/cli/controlplanes_test.go

@@ -0,0 +1,134 @@
+package cli
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/spf13/cobra"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+)
+
+func TestStatusSaysWhichControlPlaneLeadsAndWhetherTheOthersStandBy(t *testing.T) {
+	render := func(s adminapi.Summary) string {
+		cmd := &cobra.Command{}
+		var b bytes.Buffer
+		cmd.SetOut(&b)
+		printSummary(cmd, s)
+		return b.String()
+	}
+	out := render(adminapi.Summary{ControlPlanes: adminapi.ControlPlanes{
+		Members: []adminapi.ControlPlane{
+			{Name: "llmgw-ctl", State: "leading", This: true},
+			{Name: "llmgw-ctl02", State: "standing by"},
+		},
+		Floating: "10.0.0.161", FloatingUp: true,
+	}})
+	if !strings.Contains(out, "CONTROL   llmgw-ctl leads (this one), llmgw-ctl02 standing by · 10.0.0.161 on the leader") {
+		t.Errorf("status:\n%s", out)
+	}
+	out = render(adminapi.Summary{ControlPlanes: adminapi.ControlPlanes{
+		Members: []adminapi.ControlPlane{
+			{Name: "llmgw-ctl", State: "not answering"},
+			{Name: "llmgw-ctl02", State: "standing by", This: true},
+		},
+		Floating: "10.0.0.161",
+	}})
+	for _, want := range []string{"NO LEADER", "llmgw-ctl not answering", "10.0.0.161 NOT ANSWERING"} {
+		if !strings.Contains(out, want) {
+			t.Errorf("%q missing:\n%s", want, out)
+		}
+	}
+	if out := render(adminapi.Summary{}); strings.Contains(out, "CONTROL") {
+		t.Error("a single control plane got a control line")
+	}
+}
+
+func TestTheProbeAsksEachControlPlaneWhetherItLeads(t *testing.T) {
+	leading := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
+	defer leading.Close()
+	standby := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer standby.Close()
+	gone := httptest.NewServer(http.NotFoundHandler())
+	gone.Close()
+
+	st, err := store.Open(t.Context(), filepath.Join(t.TempDir(), "s.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	for _, cp := range []store.ControlPlane{
+		{URL: "https://10.0.0.1:7443", Name: "one"},
+		{URL: "https://10.0.0.2:7443", Name: "two"},
+		{URL: "https://10.0.0.3:7443", Name: "three"},
+	} {
+		if err := st.PutControlPlane(t.Context(), cp); err != nil {
+			t.Fatal(err)
+		}
+	}
+	at := map[string]string{"10.0.0.1": leading.URL, "10.0.0.2": standby.URL, "10.0.0.3": gone.URL, "10.0.0.161": leading.URL}
+	p := newControlPlaneProbe(st, ":8080")
+	floating := filepath.Join(t.TempDir(), "floating")
+	if err := os.WriteFile(floating, []byte("10.0.0.161/24\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	p.self, p.floating = "two", floating
+	p.leaderURL = func(host string) string { return at[host] + "/leader" }
+
+	s := p.Status()
+	got := map[string]adminapi.ControlPlane{}
+	for _, m := range s.Members {
+		got[m.Name] = m
+	}
+	if got["one"].State != "leading" || got["two"].State != "standing by" || got["three"].State != "not answering" {
+		t.Errorf("states: %+v", got)
+	}
+	if !got["two"].This || got["one"].This || got["one"].Address != "10.0.0.1" {
+		t.Errorf("which and where: %+v", got)
+	}
+	if !s.FloatingUp {
+		t.Error("a floating address on the leader read as down")
+	}
+}
+
+func TestOneControlPlaneIsNotASetOfThem(t *testing.T) {
+	st, err := store.Open(t.Context(), filepath.Join(t.TempDir(), "s.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	if err := st.PutControlPlane(t.Context(), store.ControlPlane{URL: "https://10.0.0.1:7443", Name: "one"}); err != nil {
+		t.Fatal(err)
+	}
+	if s := newControlPlaneProbe(st, ":8080").Status(); len(s.Members) != 0 {
+		t.Errorf("one control plane reported as a set: %+v", s)
+	}
+}
+
+func TestTheFloatingAddressIsReadFromWhatServerAddWrote(t *testing.T) {
+	dir := t.TempDir()
+	file := filepath.Join(dir, "floating")
+	if err := os.WriteFile(file, []byte("10.0.0.161/24\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if got := floatingAddress(file); got != "10.0.0.161" {
+		t.Errorf("read %q", got)
+	}
+	if err := os.WriteFile(file, []byte("not an address\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if got := floatingAddress(file); got != "" {
+		t.Errorf("nonsense read as %q", got)
+	}
+	if got := floatingAddress(filepath.Join(dir, "none")); got != "" {
+		t.Errorf("no file read as %q", got)
+	}
+}

internal/cli/ctl_upgrade.go

@@ -16,6 +16,7 @@ import (
 	"github.com/spf13/cobra"
 
 	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/release"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/version"
 )
@@ -27,19 +28,25 @@ func newUpgradeCmd(client func() *ctlClient) *cobra.Command {
 		releases  string
 		perNode   time.Duration
 		serverSvc string
+		parallel  int
+		sshUser   string
+		othersCP  bool
 	)
 	c := &cobra.Command{
 		Use:   "upgrade VERSION",
-		Short: "Upgrade this control plane, then every node, one at a time",
+		Short: "Upgrade the control planes, then every node, in waves",
 		Long: "Moves the cluster to a release: v0.10.4, or a candidate like v0.10.4-rc.2.\n\n" +
-			"This machine first — run it with sudo on the control plane, since it\n" +
-			"replaces /usr/local/bin/llmgw and restarts llmgw-server. Then each node, in\n" +
-			"turn: its agent fetches the release for its own platform, checks it\n" +
-			"against the release's checksums, swaps it in and restarts. Engines keep\n" +
-			"serving through all of it and are adopted again; no ssh into the nodes is\n" +
-			"needed.\n\n" +
+			"The other control planes first, over ssh as --ssh-user (the one sudo was\n" +
+			"run by), then this machine — run it with sudo on a control plane, since it\n" +
+			"replaces /usr/local/bin/llmgw and restarts llmgw-server. A leader restarted\n" +
+			"last hands over to a standby already on the release.\n\n" +
+			"Then the nodes, in waves: one first, then twice as many each time a wave\n" +
+			"comes back whole, up to --parallel at once. Each agent fetches the release\n" +
+			"for its own platform, checks it against the release's checksums, swaps it\n" +
+			"in and restarts; engines keep serving through all of it and are adopted\n" +
+			"again, and no ssh into the nodes is needed.\n\n" +
 			"A node counts as done when it is back on the new version, ready, with as\n" +
-			"many ready replicas as before. One that is not within --timeout stops the\n" +
+			"many ready replicas as before. A wave not back within --timeout stops the\n" +
 			"upgrade there, so a bad release reaches one node rather than all of them.\n" +
 			"Nodes already on the version are skipped, so running it again resumes.",
 		Args: cobra.ExactArgs(1),
@@ -61,11 +68,16 @@ func newUpgradeCmd(client func() *ctlClient) *cobra.Command {
 				}
 			}
 			if !skipCP {
+				if othersCP {
+					if err := upgradeOtherControlPlanes(ctx, out, client, v, releases, sshUser); err != nil {
+						return err
+					}
+				}
 				if err := upgradeControlPlane(ctx, out, client, v, releases, serverSvc); err != nil {
 					return err
 				}
 			}
-			return upgradeNodes(ctx, out, client, v, nodes, perNode)
+			return upgradeNodes(ctx, out, client, v, nodes, perNode, parallel)
 		},
 	}
 	c.Flags().StringSliceVar(&nodes, "nodes", nil, "only these nodes (default: all)")
@@ -74,9 +86,47 @@ func newUpgradeCmd(client func() *ctlClient) *cobra.Command {
 		"(default: as server.yaml says, the mirror's copy on an offline cluster, else the project's)")
 	c.Flags().DurationVar(&perNode, "timeout", 3*time.Minute, "how long a node may take to come back")
 	c.Flags().StringVar(&serverSvc, "service", "llmgw-server", "this machine's systemd unit")
+	c.Flags().IntVar(&parallel, "parallel", 20, "the most nodes upgraded at once, once the first waves are back")
+	c.Flags().StringVar(&sshUser, "ssh-user", os.Getenv("SUDO_USER"), "who to log in to the other control planes as")
+	c.Flags().BoolVar(&othersCP, "other-control-planes", true, "upgrade the cluster's other control planes too, over ssh")
 	return c
 }
 
+// upgradeOtherControlPlanes moves every other control plane of the cluster to
+// v, over ssh with `llmgw upgrade`, before this machine: each is a standby, or
+// becomes one when this one is restarted, and the leadership should move to
+// a machine already on the release. Nothing is changed here if one cannot be
+// reached.
+func upgradeOtherControlPlanes(ctx context.Context, out io.Writer, client func() *ctlClient, v, releases, user string) error {
+	var sum adminapi.Summary
+	if err := client().get(ctx, "/api/v1/summary", &sum); err != nil {
+		return err
+	}
+	for _, cp := range sum.ControlPlanes.Members {
+		if cp.This {
+			continue
+		}
+		r := &onboard.SSH{Addr: cp.Address, User: user}
+		have, err := r.Run(ctx, "llmgw --version", nil)
+		if err != nil {
+			return fmt.Errorf("%s (%s) cannot be reached over ssh as %q: %v\n"+
+				"run `sudo llmgw upgrade %s --service llmgw-server` there, then this again with "+
+				"--other-control-planes=false", cp.Name, cp.Address, user, err, v)
+		}
+		if strings.Contains(string(have), " "+v+"\n") || strings.HasSuffix(strings.TrimSpace(string(have)), " "+v) {
+			fmt.Fprintf(out, "== %s already runs %s\n", cp.Name, v)
+			continue
+		}
+		fmt.Fprintf(out, "== %s (%s): %s -> %s\n", cp.Name, cp.State, strings.TrimSpace(string(have)), v)
+		res, err := r.Run(ctx, "sudo llmgw upgrade "+shellQuote(v)+" --service llmgw-server --releases "+shellQuote(releases), nil)
+		if err != nil {
+			return fmt.Errorf("upgrading %s: %v: %s", cp.Name, err, strings.TrimSpace(string(res)))
+		}
+		fmt.Fprintf(out, "   upgraded and restarted\n")
+	}
+	return nil
+}
+
 func upgradeControlPlane(ctx context.Context, out io.Writer, client func() *ctlClient, v, releases, svc string) error {
 	if version.Version == v {
 		fmt.Fprintf(out, "this machine already runs %s\n", v)
@@ -133,13 +183,23 @@ func upgradeControlPlane(ctx context.Context, out io.Writer, client func() *ctlC
 	}
 }
 
-func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient, v string, only []string, perNode time.Duration) error {
+// upgradeNodes moves nodes to v in waves: one node first, then twice as
+// many each time one comes back whole, up to parallel at once.
+//
+// Waves rather than one at a time, because one at a time is hours on a
+// fleet of hundreds, and an agent's restart costs nothing served — the
+// engines keep running and are adopted again. One first, because a release
+// that does not come back should reach one node rather than a wave of them.
+func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient, v string, only []string,
+	perNode time.Duration, parallel int,
+) error {
 	var nodes []adminapi.NodeView
 	if err := client().get(ctx, "/api/v1/nodes", &nodes); err != nil {
 		return err
 	}
 	slices.SortFunc(nodes, func(a, b adminapi.NodeView) int { return strings.Compare(a.ID, b.ID) })
-	done, skipped := 0, 0
+	var todo []adminapi.NodeView
+	skipped := 0
 	for _, n := range nodes {
 		if len(only) > 0 && !slices.Contains(only, n.ID) {
 			continue
@@ -148,41 +208,21 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient,
 			skipped++
 			continue
 		}
-		start := time.Now()
-		before := readyReplicasOn(ctx, client, n.ID)
-		fmt.Fprintf(out, "== %s: %s -> %s\n", n.ID, n.AgentVersion, v)
-		body, _ := json.Marshal(adminapi.UpgradeRequest{Version: v})
-		if err := client().post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil); err != nil {
-			return fmt.Errorf("%s: %w\nstopped here; `llmgw ctl logs %s` says more", n.ID, err, n.ID)
-		}
-		fmt.Fprintf(out, "   new binary in place; waiting for it to come back with its %d replica(s)\n", before)
-		deadline := time.Now().Add(perNode)
-		for {
-			var now []adminapi.NodeView
-			_ = client().get(ctx, "/api/v1/nodes", &now)
-			var cur adminapi.NodeView
-			for _, x := range now {
-				if x.ID == n.ID {
-					cur = x
-				}
-			}
-			after := readyReplicasOn(ctx, client, n.ID)
-			if cur.AgentVersion == v && cur.State == "ready" && after >= before {
-				fmt.Fprintf(out, "   done in %s: ready, replicas %d/%d\n",
-					time.Since(start).Round(time.Second), after, before)
-				done++
-				break
-			}
-			if time.Now().After(deadline) {
-				return fmt.Errorf("%s did not come back within %s (version %q, state %q, replicas %d/%d)\n"+
-					"stopped here, before the next node; `llmgw ctl logs %s` says why, and the node "+
-					"keeps its previous binary beside the new one", n.ID, perNode, cur.AgentVersion,
-					cur.State, after, before, n.ID)
-			}
-			time.Sleep(2 * time.Second)
-		}
+		todo = append(todo, n)
 	}
-	fmt.Fprintf(out, "%d node(s) upgraded to %s, %d already on it\n", done, v, skipped)
+	parallel = max(parallel, 1)
+	start, size, done := time.Now(), 1, 0
+	for len(todo) > 0 {
+		wave := todo[:min(size, len(todo))]
+		todo = todo[len(wave):]
+		if err := upgradeWave(ctx, out, client, v, wave, perNode); err != nil {
+			return fmt.Errorf("%w\nstopped after this wave: %d upgraded, %d not tried", err, done, len(todo))
+		}
+		done += len(wave)
+		size = min(size*2, parallel)
+	}
+	fmt.Fprintf(out, "%d node(s) upgraded to %s in %s, %d already on it\n",
+		done, v, time.Since(start).Round(time.Second), skipped)
 	// The one machine this does not reach. Said every time, because it is
 	// exactly the kind of thing that stays three versions behind unnoticed.
 	fmt.Fprintf(out, "The model mirror's host is not upgraded by this; on it, run:\n"+
@@ -190,23 +230,87 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient,
 	return nil
 }
 
-// readyReplicasOn counts the ready replicas with a shard on a node.
-func readyReplicasOn(ctx context.Context, client func() *ctlClient, node string) int {
-	var reps []adminapi.ReplicaView
-	if err := client().get(ctx, "/api/v1/replicas", &reps); err != nil {
-		return 0
+// upgradeWave tells every node in the wave at once, and waits until each is
+// back on v, ready, with as many ready replicas as it had.
+func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v string,
+	wave []adminapi.NodeView, perNode time.Duration,
+) error {
+	start := time.Now()
+	before := readyReplicasByNode(ctx, client)
+	ids := make([]string, len(wave))
+	for i, n := range wave {
+		ids[i] = n.ID
+	}
+	fmt.Fprintf(out, "== %d node(s): %s\n", len(wave), abbreviate(ids, 8))
+	body, _ := json.Marshal(adminapi.UpgradeRequest{Version: v})
+	var failed []string
+	for _, n := range wave {
+		if err := client().post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil); err != nil {
+			failed = append(failed, fmt.Sprintf("%s: %v", n.ID, err))
+		}
+	}
+	if len(failed) > 0 {
+		return fmt.Errorf("not told to upgrade:\n  %s\n`llmgw ctl logs NODE` says more", strings.Join(failed, "\n  "))
+	}
+	waiting := slices.Clone(ids)
+	deadline := time.Now().Add(perNode)
+	for {
+		var now []adminapi.NodeView
+		_ = client().get(ctx, "/api/v1/nodes", &now)
+		after := readyReplicasByNode(ctx, client)
+		state := map[string]adminapi.NodeView{}
+		for _, x := range now {
+			state[x.ID] = x
+		}
+		waiting = slices.DeleteFunc(waiting, func(id string) bool {
+			cur := state[id]
+			return cur.AgentVersion == v && cur.State == "ready" && after[id] >= before[id]
+		})
+		if len(waiting) == 0 {
+			fmt.Fprintf(out, "   done in %s\n", time.Since(start).Round(time.Second))
+			return nil
+		}
+		if time.Now().After(deadline) {
+			var why []string
+			for _, id := range waiting {
+				cur := state[id]
+				why = append(why, fmt.Sprintf("%s (version %q, state %q, replicas %d/%d)",
+					id, cur.AgentVersion, cur.State, after[id], before[id]))
+			}
+			return fmt.Errorf("not back within %s:\n  %s\n`llmgw ctl logs NODE` says why, and each keeps its "+
+				"previous binary beside the new one", perNode, strings.Join(why, "\n  "))
+		}
+		time.Sleep(2 * time.Second)
+	}
+}
+
+// abbreviate lists up to n names, and says how many more.
+func abbreviate(names []string, n int) string {
+	if len(names) <= n {
+		return strings.Join(names, ", ")
+	}
+	return strings.Join(names[:n], ", ") + fmt.Sprintf(" and %d more", len(names)-n)
+}
+
+// readyReplicasByNode counts, for each node, the ready replicas with a
+// shard on it.
+func readyReplicasByNode(ctx context.Context, client func() *ctlClient) map[string]int {
+	var reps []adminapi.ReplicaView
+	out := map[string]int{}
+	if err := client().get(ctx, "/api/v1/replicas", &reps); err != nil {
+		return out
 	}
-	n := 0
 	for _, r := range reps {
 		if r.State != "ready" {
 			continue
 		}
-		for _, s := range r.Shards {
-			if s.Node == node {
-				n++
-				break
+		seen := map[string]bool{}
+		for _, sh := range r.Shards {
+			if !seen[sh.Node] {
+				seen[sh.Node] = true
+				out[sh.Node]++
 			}
 		}
 	}
-	return n
+	return out
 }

internal/cli/ctl_upgrade_test.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -57,32 +58,56 @@ func (f *fakeFleet) server(t *testing.T) *httptest.Server {
 	return srv
 }
 
-func TestAnUpgradeGoesNodeByNodeAndSkipsTheDone(t *testing.T) {
+func TestAnUpgradeSkipsTheNodesAlreadyDone(t *testing.T) {
 	f := &fakeFleet{versions: map[string]string{"a": "v1.0.0", "b": "v1.0.1", "c": "v1.0.0"}}
 	srv := f.server(t)
 	var out bytes.Buffer
-	err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 5*time.Second)
+	err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 5*time.Second, 1)
 	if err != nil {
 		t.Fatal(err)
 	}
 	if strings.Join(f.told, ",") != "a,c" {
 		t.Errorf("told %v; b was already on the version", f.told)
 	}
-	if !strings.Contains(out.String(), "2 node(s) upgraded to v1.0.1, 1 already on it") {
+	if !strings.Contains(out.String(), "2 node(s) upgraded to v1.0.1 in") || !strings.Contains(out.String(), "1 already on it") {
 		t.Errorf("output:\n%s", out.String())
 	}
 }
 
-func TestAnUpgradeStopsAtTheFirstNodeThatDoesNotComeBack(t *testing.T) {
+func TestAnUpgradeStopsAtTheFirstWaveThatDoesNotComeBack(t *testing.T) {
 	// A bad release reaches one node, not all of them.
 	f := &fakeFleet{versions: map[string]string{"a": "v1.0.0", "b": "v1.0.0"}, stuck: map[string]bool{"a": true}}
 	srv := f.server(t)
 	var out bytes.Buffer
-	err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 3*time.Second)
-	if err == nil || !strings.Contains(err.Error(), "a did not come back") {
+	err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 3*time.Second, 1)
+	if err == nil || !strings.Contains(err.Error(), "not back within") || !strings.Contains(err.Error(), "a (version") {
 		t.Fatalf("err %v", err)
 	}
 	if len(f.told) != 1 {
 		t.Errorf("went on to %v after a node failed to come back", f.told)
 	}
 }
+
+func TestAnUpgradeGoesInWavesThatGrowUpToTheLimit(t *testing.T) {
+	// One first, then twice as many each time, up to --parallel: hundreds of
+	// nodes in minutes rather than hours, and a bad release still on one.
+	versions := map[string]string{}
+	for i := range 12 {
+		versions[fmt.Sprintf("n%02d", i)] = "v1.0.0"
+	}
+	f := &fakeFleet{versions: versions}
+	srv := f.server(t)
+	var out bytes.Buffer
+	if err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 5*time.Second, 4); err != nil {
+		t.Fatal(err)
+	}
+	var sizes []string
+	for _, line := range strings.Split(out.String(), "\n") {
+		if n, ok := strings.CutPrefix(line, "== "); ok {
+			sizes = append(sizes, strings.Fields(n)[0])
+		}
+	}
+	if got := strings.Join(sizes, ","); got != "1,2,4,4,1" {
+		t.Errorf("waves %s:\n%s", got, out.String())
+	}
+}

internal/cli/leaderforward.go

@@ -0,0 +1,168 @@
+package cli
+
+import (
+	"context"
+	"crypto/x509"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"slices"
+	"strings"
+	"time"
+
+	"golang.org/x/net/http2"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+// forwardPrefix is where the leader answers a standby's admin API, on the
+// agent port: the one port control planes already reach each other on, with
+// the cluster's certificates both ways. The admin port itself listens on
+// loopback, and should — it serves pprof.
+const forwardPrefix = "/_admin"
+
+// Headers a forwarded request and its answer carry: which standby asked, and
+// which control plane answered.
+const (
+	hdrForwardedBy = "X-Llmgw-Forwarded-By"
+	hdrAnsweredBy  = "X-Llmgw-Answered-By"
+)
+
+// leaderForward sends a standby's admin API to the leader.
+//
+// Everything the console and ctl show is the leader's: nodes, replicas and
+// jobs live in its memory, put there by the agent streams only it holds. A
+// standby answering from its own would show an empty cluster, and accept
+// changes it cannot carry out. So a standby answers nothing under /api/
+// itself; the leader does, and the console looks the same on either machine.
+type leaderForward struct {
+	st   *store.DB
+	lead interface{ IsLeader() bool }
+	self string
+	tls  tlsconf.Config
+	log  *slog.Logger
+}
+
+// standby wraps the admin handler: /api/ goes to the leader while this
+// process is not it.
+func (f *leaderForward) standby(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.HasPrefix(r.URL.Path, "/api/") || f.lead.IsLeader() {
+			next.ServeHTTP(w, r)
+			return
+		}
+		target, name, err := f.leader(r.Context())
+		if err != nil {
+			writeForwardError(w, http.StatusServiceUnavailable,
+				"this control plane is standing by, and cannot reach the leader: "+err.Error())
+			return
+		}
+		tc, err := f.tls.ClientConfig(target.Hostname())
+		if err != nil {
+			writeForwardError(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+		p := &httputil.ReverseProxy{
+			Rewrite: func(pr *httputil.ProxyRequest) {
+				pr.Out.URL.Scheme, pr.Out.URL.Host = "https", target.Host
+				pr.Out.URL.Path = forwardPrefix + pr.In.URL.Path
+				pr.Out.URL.RawPath = ""
+				pr.Out.Host = target.Host
+				// The leader's own token is not this one's, and the
+				// certificate is what it believes.
+				pr.Out.Header.Del("Authorization")
+				pr.Out.Header.Set(hdrForwardedBy, f.self)
+			},
+			Transport: &http2.Transport{TLSClientConfig: tc},
+			// Streams, such as a job's progress, pass through as they come.
+			FlushInterval: -1,
+			ErrorHandler: func(w http.ResponseWriter, _ *http.Request, err error) {
+				writeForwardError(w, http.StatusBadGateway, fmt.Sprintf("the leader, %s, did not answer: %v", name, err))
+			},
+		}
+		p.ServeHTTP(w, r)
+	})
+}
+
+// leader is where the lease's holder is reached, and its name.
+func (f *leaderForward) leader(ctx context.Context) (*url.URL, string, error) {
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+	lease, err := f.st.Leadership(ctx, control.ControlRole)
+	if err != nil || lease.Owner == "" || time.Now().After(lease.ExpiresAt) {
+		return nil, "", errors.New("no control plane holds the lease right now")
+	}
+	name := control.HolderName(lease.Owner)
+	cps, err := f.st.ControlPlanes(ctx)
+	if err != nil {
+		return nil, name, err
+	}
+	for _, cp := range cps {
+		if cp.Name == name {
+			u, err := url.Parse(cp.URL)
+			return u, name, err
+		}
+	}
+	return nil, name, fmt.Errorf("%s leads, and is not among the control planes the store lists", name)
+}
+
+// leaderSide answers forwarded requests, on the agent port: only from another
+// control plane of this cluster, only while this one leads, and only the API.
+func (f *leaderForward) leaderSide(admin http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		path := strings.TrimPrefix(r.URL.Path, forwardPrefix)
+		if !strings.HasPrefix(path, "/api/") {
+			http.NotFound(w, r)
+			return
+		}
+		if err := f.fromControlPlane(r); err != nil {
+			writeForwardError(w, http.StatusForbidden, err.Error())
+			return
+		}
+		if !f.lead.IsLeader() {
+			// Not passed on again: two standbys must not bounce a request
+			// between them while the lease moves.
+			writeForwardError(w, http.StatusServiceUnavailable, "this control plane no longer leads; ask again")
+			return
+		}
+		r2 := r.Clone(r.Context())
+		r2.URL.Path, r2.URL.RawPath = path, ""
+		w.Header().Set(hdrAnsweredBy, f.self)
+		admin.ServeHTTP(w, r2)
+	})
+}
+
+// fromControlPlane checks the client certificate names a control plane the
+// store lists. A node's certificate is from the same authority, and a node
+// must not reach the admin API by presenting it.
+func (f *leaderForward) fromControlPlane(r *http.Request) error {
+	if r.TLS == nil || len(r.TLS.VerifiedChains) == 0 {
+		return errors.New("a control plane's certificate is required")
+	}
+	cert := r.TLS.VerifiedChains[0][0]
+	if !slices.Contains(cert.ExtKeyUsage, x509.ExtKeyUsageServerAuth) {
+		return errors.New("not a control plane's certificate")
+	}
+	cps, err := f.st.ControlPlanes(r.Context())
+	if err != nil {
+		return err
+	}
+	for _, cp := range cps {
+		if cp.Name == cert.Subject.CommonName {
+			return nil
+		}
+	}
+	return fmt.Errorf("%s is not a control plane of this cluster", cert.Subject.CommonName)
+}
+
+func writeForwardError(w http.ResponseWriter, code int, msg string) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(code)
+	_ = json.NewEncoder(w).Encode(map[string]string{"error": msg})
+}

internal/cli/leaderforward_test.go

@@ -0,0 +1,188 @@
+package cli
+
+import (
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
+)
+
+type leading struct{ v atomic.Bool }
+
+func (l *leading) IsLeader() bool { return l.v.Load() }
+
+// forwardCluster is two control planes of one cluster on one store: a
+// leader whose agent port answers forwarded admin requests, and a standby.
+type forwardCluster struct {
+	st              *store.DB
+	dir             string
+	leaderLeads     *leading
+	standby, leader *leaderForward
+	leaderURL       string
+}
+
+func newForwardCluster(t *testing.T) *forwardCluster {
+	t.Helper()
+	dir := t.TempDir()
+	ca, err := tlsconf.NewCA("test CA")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := ca.Write(dir); err != nil {
+		t.Fatal(err)
+	}
+	for _, name := range []string{"ctl-a", "ctl-b"} {
+		leaf, err := ca.IssuePeer(name, []string{"127.0.0.1"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := leaf.Write(dir, name); err != nil {
+			t.Fatal(err)
+		}
+	}
+	node, err := ca.IssueClient("node1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := node.Write(dir, "node1"); err != nil {
+		t.Fatal(err)
+	}
+	cfgFor := func(name string) tlsconf.Config {
+		return tlsconf.Config{CA: filepath.Join(dir, "ca.pem"), Cert: filepath.Join(dir, name+".pem"), Key: filepath.Join(dir, name+"-key.pem")}
+	}
+
+	st, err := store.Open(t.Context(), filepath.Join(dir, "s.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { st.Close() })
+
+	c := &forwardCluster{st: st, dir: dir, leaderLeads: &leading{}}
+	c.leaderLeads.v.Store(true)
+	log := slog.New(slog.DiscardHandler)
+	c.leader = &leaderForward{st: st, lead: c.leaderLeads, self: "ctl-a", tls: cfgFor("ctl-a"), log: log}
+	standbyLeads := &leading{}
+	c.standby = &leaderForward{st: st, lead: standbyLeads, self: "ctl-b", tls: cfgFor("ctl-b"), log: log}
+
+	// The leader's admin API: says whose it is, and who it was asked by.
+	admin := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = json.NewEncoder(w).Encode(map[string]string{"path": r.URL.Path, "by": r.Header.Get(hdrForwardedBy),
+			"auth": r.Header.Get("Authorization"), "actor": r.Header.Get("X-Llmgw-Actor")})
+	})
+	mux := http.NewServeMux()
+	mux.Handle(forwardPrefix+"/", c.leader.leaderSide(admin))
+	srv := httptest.NewUnstartedServer(mux)
+	tc, err := cfgFor("ctl-a").ServerConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv.TLS = tc
+	srv.EnableHTTP2 = true
+	srv.StartTLS()
+	t.Cleanup(srv.Close)
+	c.leaderURL = srv.URL
+
+	for _, cp := range []store.ControlPlane{{URL: srv.URL, Name: "ctl-a"}, {URL: "https://127.0.0.1:1", Name: "ctl-b"}} {
+		if err := st.PutControlPlane(t.Context(), cp); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := st.AcquireLeadership(t.Context(), control.ControlRole, "ctl-a-0123456789abcdef", time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	return c
+}
+
+// ask sends one request through the standby's admin API.
+func (c *forwardCluster) ask(t *testing.T, path string) (*http.Response, map[string]string) {
+	t.Helper()
+	local := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_ = json.NewEncoder(w).Encode(map[string]string{"path": "local"})
+	})
+	srv := httptest.NewServer(c.standby.standby(local))
+	defer srv.Close()
+	req, _ := http.NewRequestWithContext(t.Context(), http.MethodGet, srv.URL+path, nil)
+	req.Header.Set("Authorization", "Bearer the-standbys-own-token")
+	req.Header.Set("X-Llmgw-Actor", "root@ctl-b")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(resp.Body)
+	var got map[string]string
+	_ = json.Unmarshal(b, &got)
+	return resp, got
+}
+
+func TestAStandbysAdminAPIIsTheLeaders(t *testing.T) {
+	c := newForwardCluster(t)
+	resp, got := c.ask(t, "/api/v1/summary")
+	if resp.StatusCode != http.StatusOK || got["path"] != "/api/v1/summary" {
+		t.Fatalf("%d %v", resp.StatusCode, got)
+	}
+	if resp.Header.Get(hdrAnsweredBy) != "ctl-a" || got["by"] != "ctl-b" {
+		t.Errorf("answered by %q, asked by %q", resp.Header.Get(hdrAnsweredBy), got["by"])
+	}
+	if got["auth"] != "" {
+		t.Error("the standby's token was passed on to the leader")
+	}
+	if got["actor"] != "root@ctl-b" {
+		t.Errorf("who asked was lost: %q", got["actor"])
+	}
+	// What is not the API stays on the machine: the dashboard, pprof.
+	if _, got := c.ask(t, "/debug/pprof/"); got["path"] != "local" {
+		t.Errorf("a non-API path was forwarded: %v", got)
+	}
+}
+
+func TestTheLeaderAnswersOnlyControlPlanesAndOnlyWhileItLeads(t *testing.T) {
+	c := newForwardCluster(t)
+	call := func(cert string) int {
+		cfg := tlsconf.Config{CA: filepath.Join(c.dir, "ca.pem"), Cert: filepath.Join(c.dir, cert+".pem"), Key: filepath.Join(c.dir, cert+"-key.pem")}
+		tc, err := cfg.ClientConfig("127.0.0.1")
+		if err != nil {
+			t.Fatal(err)
+		}
+		tc.NextProtos = nil
+		client := &http.Client{Transport: &http.Transport{TLSClientConfig: tc}}
+		resp, err := client.Get(c.leaderURL + forwardPrefix + "/api/v1/summary")
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp.Body.Close()
+		return resp.StatusCode
+	}
+	if code := call("ctl-b"); code != http.StatusOK {
+		t.Errorf("a control plane was refused: %d", code)
+	}
+	// A node's certificate is from the same authority, and is no way in.
+	if code := call("node1"); code != http.StatusForbidden {
+		t.Errorf("a node reached the admin API: %d", code)
+	}
+	c.leaderLeads.v.Store(false)
+	if code := call("ctl-b"); code != http.StatusServiceUnavailable {
+		t.Errorf("a control plane that no longer leads answered: %d", code)
+	}
+}
+
+func TestAStandbyWithNoLeaderSaysSo(t *testing.T) {
+	c := newForwardCluster(t)
+	if err := c.st.ReleaseLeadership(t.Context(), control.ControlRole, "ctl-a-0123456789abcdef"); err != nil {
+		t.Fatal(err)
+	}
+	resp, got := c.ask(t, "/api/v1/summary")
+	if resp.StatusCode != http.StatusServiceUnavailable || !strings.Contains(got["error"], "standing by") {
+		t.Errorf("%d %v", resp.StatusCode, got)
+	}
+}

internal/cli/server.go

@@ -330,6 +330,9 @@ func newServerCmd() *cobra.Command {
 						return out, err
 					})
 				}
+				if st := rt.Store(); st != nil {
+					api.SetControlPlanes(newControlPlaneProbe(st, cfg.Listen).Status)
+				}
 				adminAPI = api
 				api.SetReleases(cfg.Releases)
 				api.SetNotifier(notifier)
@@ -499,8 +502,20 @@ func newServerCmd() *cobra.Command {
 			if err != nil {
 				return err
 			}
+			adminHandler := auditAPI(rt, adminMux, log)
+			// With a rival, a standby's admin API is the leader's: what the
+			// console and ctl show lives in the leader's memory.
+			var forward *leaderForward
+			if admin != nil && admin.Leadership() != nil && rt.Store() != nil && cfg.AgentTLS.Enabled() {
+				self, _ := os.Hostname()
+				forward = &leaderForward{st: rt.Store(), lead: admin.Leadership(), self: self, tls: cfg.AgentTLS, log: log}
+			}
+			served := adminHandler
+			if forward != nil {
+				served = forward.standby(adminHandler)
+			}
 			adminSrv := withLimits(&http.Server{
-				Handler:     router.AdminGuard(router.AdminAuth(adminToken, cfg.DebugPprof, auditAPI(rt, adminMux, log))),
+				Handler:     router.AdminGuard(router.AdminAuth(adminToken, cfg.DebugPprof, served)),
 				ReadTimeout: time.Minute,
 			})
 
@@ -566,6 +581,9 @@ func newServerCmd() *cobra.Command {
 			if agentHandler != nil {
 				mux := http.NewServeMux()
 				mux.Handle(agentPath, agentHandler)
+				if forward != nil {
+					mux.Handle(forwardPrefix+"/", router.AdminGuard(forward.leaderSide(adminHandler)))
+				}
 				// No read timeout: an agent's stream is one request that
 				// lasts as long as the agent does.
 				agentSrv = withLimits(&http.Server{Handler: mux})
@@ -617,8 +635,19 @@ func newServerCmd() *cobra.Command {
 				log.Info("shutting down")
 				sctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 				defer cancel()
+				// The lease first: a standby polls for it, and every agent
+				// about to be let go will be refused by a standby that has
+				// not taken it yet.
+				if admin != nil && admin.Leadership() != nil {
+					admin.Leadership().StepDown()
+				}
 				if agentSrv != nil {
-					_ = agentSrv.Shutdown(sctx)
+					// Closed, not shut down gracefully: an agent's stream
+					// never goes idle, so Shutdown would wait out its whole
+					// timeout while every node stayed on a control plane
+					// that is leaving, and a standby could not be found
+					// until it gave up. Closed, they move at once.
+					_ = agentSrv.Close()
 				}
 				if metricsSrv != nil {
 					_ = metricsSrv.Shutdown(sctx)

internal/cli/serveradd.go

@@ -121,6 +121,7 @@ func (sa *serverAdd) run(ctx context.Context) error {
 		{"connect", sa.connect},
 		{"binary", sa.binary},
 		{"directories", sa.dirs},
+		{"this certificate", sa.ownCertificate},
 		{"certificate", sa.certificate},
 		{"configuration", sa.configuration},
 		{"store", sa.storeAccess},
@@ -132,6 +133,10 @@ func (sa *serverAdd) run(ctx context.Context) error {
 			fn   func(context.Context) error
 		}{"firewall", sa.firewall})
 	}
+	steps = append(steps, struct {
+		name string
+		fn   func(context.Context) error
+	}{"mirror", sa.mirror})
 	if sa.vip != "" {
 		steps = append(steps, struct {
 			name string
@@ -227,6 +232,40 @@ sudo install -d -o llmgw -g llmgw -m 0750 /var/lib/llmgw`, nil)
 	return err
 }
 
+// ownCertificate makes this control plane's certificate one it can also
+// present as a client — a standby asks the leader for the console's data
+// with it — reissuing one from before that with the same names. The running
+// server reads it as it starts.
+func (sa *serverAdd) ownCertificate(context.Context) error {
+	path := filepath.Join(sa.etcDir, "server.pem")
+	cert, err := readCert(path)
+	if err != nil {
+		return err
+	}
+	if clientCapable(cert) {
+		sa.kept("this control plane's certificate")
+		return nil
+	}
+	ca, err := loadCA(sa.pkiDir)
+	if err != nil {
+		return fmt.Errorf("the authority's key is not in %s: run this where it is (%v)", sa.pkiDir, err)
+	}
+	hosts := strings.Split(sanNames(cert), ", ")
+	leaf, err := ca.IssuePeer(cert.Subject.CommonName, hosts)
+	if err != nil {
+		return err
+	}
+	if err := os.WriteFile(path, leaf.CertPEM, 0o644); err != nil { //nolint:gosec // a certificate
+		return err
+	}
+	if err := os.WriteFile(filepath.Join(sa.etcDir, "server-key.pem"), leaf.KeyPEM, 0o640); err != nil {
+		return err
+	}
+	sa.done("this control plane's certificate reissued for %s, to ask the leader as well as answer;\n"+
+		"        restart llmgw-server here for it to take effect — %s takes over meanwhile", strings.Join(hosts, ", "), sa.name)
+	return nil
+}
+
 // certificate issues it a server certificate from this cluster's authority,
 // for its name and addresses, unless it holds one that already names them.
 func (sa *serverAdd) certificate(ctx context.Context) error {
@@ -251,14 +290,14 @@ func (sa *serverAdd) certificate(ctx context.Context) error {
 				for _, w := range want {
 					missing = missing || !slices.Contains(names, w)
 				}
-				if !missing {
+				if !missing && clientCapable(cert) {
 					sa.kept("certificate for %s", strings.Join(names, ", "))
 					return sa.put(ctx, "/etc/llmgw/ca.pem", "0644", "root:root", ca.PEM)
 				}
 			}
 		}
 	}
-	leaf, err := ca.IssueServer(sa.name, want)
+	leaf, err := ca.IssuePeer(sa.name, want)
 	if err != nil {
 		return err
 	}
@@ -419,6 +458,38 @@ func (sa *serverAdd) service(ctx context.Context) error {
 	}
 }
 
+// mirror checks that the mirror's management API answers the new control
+// plane. It answers only the addresses and certificates it was told to, and
+// one told only of the first control plane refuses the second the moment it
+// leads: models cannot be fetched, updated or served from the registry.
+//
+// Said, not fixed: the mirror is its own machine with its own unit, and
+// llmgw does not log in to it.
+func (sa *serverAdd) mirror(ctx context.Context) error {
+	cfg, err := config.Load(filepath.Join(sa.etcDir, "server.yaml"))
+	if err != nil || cfg.Mirror.API == "" {
+		return nil
+	}
+	if cfg.Mirror.TLS.Enabled() {
+		fmt.Fprintf(sa.out, "  note  the mirror at %s answers only the certificates it names: add\n"+
+			"        --api-client %s to its llmgw-mirror unit, or it refuses %s when it leads\n",
+			cfg.Mirror.API, sa.name, sa.name)
+		return nil
+	}
+	code, _ := sa.sh(ctx, "curl -s -o /dev/null -m 3 -w '%{http_code}' "+shellQuote(strings.TrimRight(cfg.Mirror.API, "/")+"/")+" || true", nil)
+	switch code = strings.TrimSpace(code); code {
+	case "403":
+		fmt.Fprintf(sa.out, "  !!    the mirror at %s refuses %s: add --api-allow %s to its\n"+
+			"        llmgw-mirror unit and restart it, or models cannot be fetched while %s leads\n",
+			cfg.Mirror.API, sa.name, sa.ip, sa.name)
+	case "", "000":
+		fmt.Fprintf(sa.out, "  !!    the mirror at %s does not answer %s\n", cfg.Mirror.API, sa.name)
+	default:
+		sa.done("the mirror at %s answers it", cfg.Mirror.API)
+	}
+	return nil
+}
+
 // floating keeps an address on whichever of the two leads, with keepalived
 // on both, following /leader.
 func (sa *serverAdd) floating(ctx context.Context) error {
@@ -453,6 +524,12 @@ sudo systemctl restart keepalived`, []byte(conf)); err != nil {
 			return fmt.Errorf("on %s: %v: %s", m.name, err, strings.TrimSpace(string(out)))
 		}
 	}
+	// Where the summary reads it, on both: readable, unlike keepalived's.
+	for _, r := range []onboard.Remote{here, sa.remote} {
+		if out, err := r.Run(ctx, "sudo sh -c 'cat > "+floatingPath+" && chmod 0644 "+floatingPath+"'", []byte(sa.vip+"\n")); err != nil {
+			return fmt.Errorf("recording it: %v: %s", err, strings.TrimSpace(string(out)))
+		}
+	}
 	sa.done("%s follows whichever control plane leads (keepalived on both)", sa.vip)
 	return nil
 }
@@ -503,3 +580,9 @@ vrrp_instance llmgw {
 }
 `, iface, priority, pass, vip, prefix)
 }
+
+// clientCapable says a control plane's certificate can also present itself
+// as a client, as a standby does to the leader.
+func clientCapable(cert *x509.Certificate) bool {
+	return slices.Contains(cert.ExtKeyUsage, x509.ExtKeyUsageClientAuth)
+}

internal/cli/serverinit.go

@@ -462,7 +462,10 @@ func (in *serverInit) ensureCerts(context.Context) error {
 		// A certificate from this authority naming every address is left
 		// alone. One naming fewer is reissued with the union, so an address
 		// that worked yesterday is never dropped by adding one today.
-		if len(missing) == 0 && cert.CheckSignatureFrom(ca.Cert) == nil {
+		// And one that can also present itself as a client, which a
+		// standby's admin API needs to reach the leader's; certificates
+		// from before that are reissued.
+		if len(missing) == 0 && cert.CheckSignatureFrom(ca.Cert) == nil && clientCapable(cert) {
 			in.kept("server certificate for %s", strings.Join(have, ", "))
 			return nil
 		}
@@ -477,7 +480,7 @@ func (in *serverInit) ensureCerts(context.Context) error {
 	if name == "" {
 		name = "llmgw-control"
 	}
-	leaf, err := ca.IssueServer(name, want)
+	leaf, err := ca.IssuePeer(name, want)
 	if err != nil {
 		return err
 	}

internal/cli/status.go

@@ -95,6 +95,39 @@ func printSummary(cmd *cobra.Command, s adminapi.Summary) {
 	fmt.Fprintf(out, "\nMODELS    %d of %d have the replicas they asked for\n",
 		s.Models.Satisfied, s.Models.Total)
 
+	// The control planes, when there are several: who leads, whether the
+	// others would be there to take over, and the address clients use.
+	if cps := s.ControlPlanes; len(cps.Members) > 0 {
+		var parts []string
+		leader := false
+		for _, cp := range cps.Members {
+			p := cp.Name
+			switch cp.State {
+			case "leading":
+				leader = true
+				p += " leads"
+			default:
+				p += " " + cp.State
+			}
+			if cp.This {
+				p += " (this one)"
+			}
+			parts = append(parts, p)
+		}
+		if !leader {
+			parts = append([]string{"NO LEADER — nothing is placed"}, parts...)
+		}
+		line := strings.Join(parts, ", ")
+		if cps.Floating != "" {
+			if cps.FloatingUp {
+				line += " · " + cps.Floating + " on the leader"
+			} else {
+				line += " · " + cps.Floating + " NOT ANSWERING"
+			}
+		}
+		fmt.Fprintf(out, "CONTROL   %s\n", line)
+	}
+
 	// The store, when it is a replicated cluster: one line, saying who leads
 	// and whether every other member keeps up.
 	switch {

internal/cli/tui_console.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/url"
 	"os"
 	"os/exec"
@@ -53,6 +54,7 @@ const (
 	cManual
 	cSettings
 	cAddNode
+	cAddControl
 	cReplicate
 	cServe
 	cUpgrade
@@ -63,7 +65,7 @@ const (
 )
 
 var consoleLabels = [cCount]string{"Overview", "Monitor", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Federation", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Audit", "Manual",
-	"server.yaml", "Add a node", "Replicate the store", "Serve a model", "Upgrade the cluster",
+	"server.yaml", "Add a node", "Add a control plane", "Replicate the store", "Serve a model", "Upgrade the cluster",
 	"Restart all replicas", "Back up now", "Leave"}
 
 // consoleAction is what is done once the screen closes: each runs its own
@@ -447,6 +449,14 @@ func (m consoleModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) {
 			switch m.menu {
 			case cAddNode:
 				return m.ask("Add a node", "user@address of the machine", 1)
+			case cAddControl:
+				if !m.storeReplicated() {
+					m.dialog, m.dialogTitle = "error", "Add a control plane"
+					m.dialogText = "A second control plane needs a store both can reach, and this one's\n" +
+						"is on one machine. Replicate the store first (Do → Replicate the store)."
+					return m, nil
+				}
+				return m.ask("Add a control plane", "user@address of the machine", 2)
 			case cReplicate:
 				return m.ask("Replicate the store", "user@address of database machine", 2)
 			case cLeave:
@@ -659,19 +669,38 @@ func (m consoleModel) dialogKey(k tea.KeyMsg) (tea.Model, tea.Cmd) {
 		v := strings.TrimSpace(m.prompt.Value())
 		switch m.dialog {
 		case "target":
-			if v == "" {
+			// The floating address is the one answer that may be left empty.
+			floating := m.menu == cAddControl && len(m.asked) == 1
+			if v == "" && !floating {
 				return m, nil
 			}
+			if floating && v != "" {
+				if _, _, err := net.ParseCIDR(v); err != nil {
+					m.dialogText = floatingQuestion + "\n\n" + sBad.Render("not an address with its prefix, like 10.0.0.161/24")
+					return m, nil
+				}
+			}
 			m.asked = append(m.asked, v)
 			m.prompt.SetValue("")
 			if len(m.asked) == int(m.editKind) {
-				kind := "node"
-				if m.menu == cReplicate {
-					kind = "patroni"
+				switch m.menu {
+				case cAddControl:
+					args := []string{"server", "add", m.asked[0]}
+					if m.asked[1] != "" {
+						args = append(args, "--vip", m.asked[1])
+					}
+					m.action = consoleAction{kind: "run", targets: args}
+				case cReplicate:
+					m.action = consoleAction{kind: "patroni", targets: m.asked}
+				default:
+					m.action = consoleAction{kind: "node", targets: m.asked}
 				}
-				m.action = consoleAction{kind: kind, targets: m.asked}
 				return m, tea.Quit
 			}
+			if m.menu == cAddControl {
+				m.dialogText = floatingQuestion
+				m.prompt.Placeholder = "10.0.0.161/24"
+			}
 		case "edit":
 			if err := m.doc.Set(m.editPath, m.editKind, m.prompt.Value()); err != nil {
 				m.dialogText = m.editQuestion() + "\n\n" + sBad.Render(err.Error())
@@ -1036,7 +1065,7 @@ func (m consoleModel) panel() (title, body, keys string) {
 		c := &cobra.Command{}
 		c.SetOut(&b)
 		printSummary(c, m.data.sum)
-		return "Overview", "\n" + indent(b.String(), "  "), keys
+		return "Overview", m.standbyNote() + "\n" + indent(b.String(), "  "), keys
 	case cNodes, cLogs:
 		if m.detail && m.row < len(m.data.nodes) {
 			n := m.data.nodes[m.row]
@@ -1173,6 +1202,10 @@ func (m consoleModel) panel() (title, body, keys string) {
 	case cAddNode:
 		item = "\n  Adds a machine as a node: installs llmgw on it over ssh, checks it,\n  gives it a certificate and starts its agent.\n\n  " +
 			sDim.Render("On the command line:") + "\n  sudo llmgw node add <user>@<address>"
+	case cAddControl:
+		item = "\n  Makes another machine a control plane of this cluster: a standby\n  that takes over within seconds when this one is lost. Every node is\n  told of it; none is touched. With a floating address, keepalived on\n  both keeps it on whichever leads — the address to give clients.\n\n  " +
+			sDim.Render("Needs the store replicated first, so both reach it.") + "\n\n  " +
+			sDim.Render("On the command line:") + "\n  sudo --preserve-env=SSH_AUTH_SOCK llmgw server add <user>@<address> --vip <address>/<prefix>"
 	case cReplicate:
 		item = "\n  Moves the store onto two machines under Patroni, replicated,\n  so losing one loses nothing.\n\n  " +
 			sDim.Render("On the command line:") +
@@ -1531,3 +1564,26 @@ func runConsole(ctx context.Context, out io.Writer, file string) error {
 		_, _ = bufio.NewReader(os.Stdin).ReadString('\n')
 	}
 }
+
+// floatingQuestion is the second question when adding a control plane.
+const floatingQuestion = "A floating address for clients, kept on whichever control plane leads,\n" +
+	"as 10.0.0.161/24 — a free one on this network. Empty for none, when a load\n" +
+	"balancer in front checks /ready instead."
+
+// standbyNote says, on a control plane standing by, whose cluster the
+// console shows: the leader's, which this one's admin API passes it on to.
+func (m consoleModel) standbyNote() string {
+	var here, leader string
+	for _, cp := range m.data.sum.ControlPlanes.Members {
+		if cp.This {
+			here = cp.State
+		}
+		if cp.State == "leading" {
+			leader = cp.Name
+		}
+	}
+	if here != "standing by" || leader == "" {
+		return ""
+	}
+	return "\n  " + sDim.Render("Standing by. Shown, and changed, on the leader: "+leader) + "\n"
+}

internal/cli/tui_setup_test.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"
 
@@ -171,3 +172,49 @@ func TestTheConsoleWatchesTheClusterAndEditsItsSettings(t *testing.T) {
 		t.Errorf("server.yaml:\n%s", b)
 	}
 }
+
+func TestTheConsoleAddsAControlPlaneWithAFloatingAddress(t *testing.T) {
+	replicated := consoleModel{menu: cAddControl, loaded: true, width: 100, height: 30}
+	replicated.data.sum.Store = []adminapi.StoreMember{{Name: "a", Role: "leader"}}
+
+	// The address, then the floating address; the console closes to run it.
+	var mm tea.Model = replicated
+	mm = press(mm, "enter", "ubuntu@10.0.0.150", "enter", "10.0.0.161/24", "enter")
+	want := []string{"server", "add", "ubuntu@10.0.0.150", "--vip", "10.0.0.161/24"}
+	if a := mm.(consoleModel).action; a.kind != "run" || !slices.Equal(a.targets, want) {
+		t.Errorf("%+v", a)
+	}
+
+	// No floating address is an answer too.
+	mm = press(replicated, "enter", "ubuntu@10.0.0.150", "enter", "enter")
+	if a := mm.(consoleModel).action; !slices.Equal(a.targets, want[:3]) {
+		t.Errorf("without one: %+v", a)
+	}
+
+	// A mistyped one is caught here rather than half way through.
+	mm = press(replicated, "enter", "ubuntu@10.0.0.150", "enter", "10.0.0.161", "enter")
+	if c := mm.(consoleModel); c.action.kind != "" || !strings.Contains(c.dialogText, "prefix") {
+		t.Errorf("an address without its prefix went through: %+v", c.action)
+	}
+
+	// On a store on one machine it says what to do first.
+	mm = press(consoleModel{menu: cAddControl, loaded: true, width: 100, height: 30}, "enter")
+	if c := mm.(consoleModel); c.dialog != "error" || !strings.Contains(c.dialogText, "Replicate the store") {
+		t.Errorf("dialog %q: %s", c.dialog, c.dialogText)
+	}
+}
+
+func TestTheConsoleOnAStandbySaysWhoseClusterItShows(t *testing.T) {
+	m := consoleModel{menu: cOverview, loaded: true, width: 120, height: 40}
+	m.data.sum.ControlPlanes = adminapi.ControlPlanes{Members: []adminapi.ControlPlane{
+		{Name: "llmgw-ctl", State: "leading"},
+		{Name: "llmgw-ctl02", State: "standing by", This: true},
+	}}
+	if v := m.View(); !strings.Contains(v, "Standing by. Shown, and changed, on the leader: llmgw-ctl") {
+		t.Errorf("the overview on a standby:\n%s", v)
+	}
+	m.data.sum.ControlPlanes.Members[0].This, m.data.sum.ControlPlanes.Members[1].This = true, false
+	if v := m.View(); strings.Contains(v, "Standing by.") {
+		t.Error("the leader's console says it is standing by")
+	}
+}

internal/control/adminapi/api.go

@@ -54,6 +54,9 @@ type API struct {
 	releases string
 	// storeStatus reads a replicated store's members; nil for one database.
 	storeStatus func() ([]StoreMember, error)
+	// controlPlanes says what each control plane answers; nil for a cluster
+	// of one.
+	controlPlanes func() ControlPlanes
 	// serve runs registry-to-replica jobs; nil without a mirror and a store.
 	serve *modelserve.Server
 	// history is the Prometheus the console's charts read from.
@@ -870,6 +873,10 @@ func (a *API) SetUsage(u Usage) { a.usage = u }
 // SetStoreStatus supplies the state of a replicated store, for the summary.
 func (a *API) SetStoreStatus(f func() ([]StoreMember, error)) { a.storeStatus = f }
 
+// SetControlPlanes sets where the summary learns which control plane leads
+// and which stand by.
+func (a *API) SetControlPlanes(f func() ControlPlanes) { a.controlPlanes = f }
+
 // ScopeUsageView is one group's spend, with its quota when it has one.
 type ScopeUsageView struct {
 	Scope string `json:"scope"`
@@ -1155,8 +1162,18 @@ func (a *API) removeModel(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, Removed{Removed: name})
 }
 
-func (a *API) summary(w http.ResponseWriter, _ *http.Request) {
-	writeJSON(w, http.StatusOK, a.Summary())
+func (a *API) summary(w http.ResponseWriter, r *http.Request) {
+	s := a.Summary()
+	// Asked through a standby, which passes on the console's questions to
+	// the leader: "this one" is the standby the console is on.
+	if by := r.Header.Get("X-Llmgw-Forwarded-By"); by != "" && len(s.ControlPlanes.Members) > 0 {
+		members := slices.Clone(s.ControlPlanes.Members)
+		for i := range members {
+			members[i].This = members[i].Name == by
+		}
+		s.ControlPlanes.Members = members
+	}
+	writeJSON(w, http.StatusOK, s)
 }
 
 func (a *API) explainPlacement(w http.ResponseWriter, r *http.Request) {

internal/control/adminapi/summary.go

@@ -35,6 +35,9 @@ type Summary struct {
 	// StoreError why it could not be read; both empty for a single database.
 	Store      []StoreMember `json:"store,omitempty"`
 	StoreError string        `json:"store_error,omitempty"`
+	// ControlPlanes is the cluster's control planes when there is more than
+	// one: which leads, which stand by, and the floating address in front.
+	ControlPlanes ControlPlanes `json:"control_planes,omitzero"`
 }
 
 // StoreMember is one member of a replicated store, as its Patroni reports it.
@@ -47,6 +50,27 @@ type StoreMember struct {
 	Lag int64 `json:"lag"`
 }
 
+// ControlPlanes is every control plane of a cluster that has several, as
+// each answered when asked.
+type ControlPlanes struct {
+	Members []ControlPlane `json:"members,omitempty"`
+	// Floating is the address keepalived keeps on the leader (10.0.0.161),
+	// and FloatingUp whether it answered as a leader when asked; empty when
+	// nothing floats.
+	Floating   string `json:"floating,omitempty"`
+	FloatingUp bool   `json:"floating_up,omitempty"`
+}
+
+// ControlPlane is one control plane.
+type ControlPlane struct {
+	Name    string `json:"name"`
+	Address string `json:"address"`
+	// State is "leading", "standing by" or "not answering".
+	State string `json:"state"`
+	// This is the control plane that answered the summary.
+	This bool `json:"this,omitempty"`
+}
+
 // NodeTotals counts nodes by the states that need different responses.
 type NodeTotals struct {
 	Total       int `json:"total"`
@@ -128,6 +152,9 @@ func (a *API) Summary() Summary {
 			s.StoreError = err.Error()
 		}
 	}
+	if a.controlPlanes != nil {
+		s.ControlPlanes = a.controlPlanes()
+	}
 
 	nodes := a.Nodes()
 	byClass := map[string]*ClassUse{}

internal/control/agentsvc/service.go

@@ -302,16 +302,58 @@ func (s *Service) Connect(
 	}()
 	defer wg.Wait()
 
+	// Read in a goroutine of its own, so the stream can also be ended from
+	// this side: Receive blocks until the agent sends, and an agent that is
+	// only heartbeating would keep a stream this process should no longer
+	// hold for as long as it likes. Returning ends it; the reader's Receive
+	// then fails and it follows.
+	read := make(chan struct{})
+	go func() {
+		defer close(read)
+		s.readLoop(ctx, stream, c, nodeID)
+	}()
+	tick := time.NewTicker(leaderCheck)
+	defer tick.Stop()
+	for {
+		select {
+		case <-read:
+			return nil
+		case <-c.closed:
+			// Replaced by a newer stream from the same node, or the writer
+			// failed.
+			return nil
+		case <-tick.C:
+			// Leadership lost while the process lives on — the store was
+			// slow, or another took the lease. A standby holding agents is
+			// a cluster whose leader cannot reach its nodes, so let them go
+			// and they find the leader.
+			if s.leader != nil && !s.leader.IsLeader() {
+				s.log.Info("no longer leading; letting the agent go", "node", nodeID)
+				c.close()
+				return connect.NewError(connect.CodeUnavailable,
+					errors.New("this control plane is a standby; connect to the leader"))
+			}
+		}
+	}
+}
+
+// leaderCheck is how often an open stream checks that this process still
+// leads.
+var leaderCheck = time.Second
+
+func (s *Service) readLoop(ctx context.Context,
+	stream *connect.BidiStream[llmgwv1.AgentMessage, llmgwv1.ControlMessage], c *conn, nodeID string,
+) {
 	for {
 		msg, err := stream.Receive()
 		if err != nil {
 			c.close()
 			if ctx.Err() != nil || errors.Is(err, io.EOF) {
 				s.log.Info("agent disconnected", "node", nodeID)
-				return nil
+				return
 			}
 			s.log.Warn("agent stream failed", "node", nodeID, "err", err)
-			return nil
+			return
 		}
 		if kv := msg.GetKvEvents(); kv != nil {
 			s.kvEvents(c, kv)

internal/control/control.go

@@ -231,6 +231,10 @@ func (c *Controller) SetLeadership(l *Leadership) {
 	c.rec.SetLeadership(l)
 }
 
+// Leadership is what SetLeadership set; nil when there is nobody to compete
+// with.
+func (c *Controller) Leadership() *Leadership { return c.lead }
+
 // IsLeader reports whether this process is placing work. True when leadership
 // was never configured, because then there is nobody to compete with.
 func (c *Controller) IsLeader() bool { return c.lead == nil || c.lead.IsLeader() }

internal/control/leadership.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"log/slog"
 	"os"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -51,6 +52,8 @@ type Leadership struct {
 	renew time.Duration
 
 	leader atomic.Bool
+	// stopped is set by StepDown: nothing takes the lease again after it.
+	stopped atomic.Bool
 }
 
 // NewLeadership prepares leadership over a store.
@@ -117,8 +120,16 @@ func (l *Leadership) Run(ctx context.Context) {
 // process: a line per attempt would bury everything else, and the thing an
 // operator needs to see is the moment leadership moved.
 func (l *Leadership) attempt(ctx context.Context) {
+	if l.stopped.Load() {
+		return
+	}
 	lease, err := l.store.AcquireLeadership(ctx, controlRole, l.owner, l.ttl)
 	switch {
+	case err == nil && l.stopped.Load():
+		// Stepped down while this was on its way to the store: give back
+		// what it just took.
+		l.leader.Store(true)
+		l.step(ctx)
 	case err == nil:
 		if l.leader.CompareAndSwap(false, true) {
 			l.log.Info("this control plane is now placing work", "owner", l.owner)
@@ -145,10 +156,24 @@ func (l *Leadership) attempt(ctx context.Context) {
 	}
 }
 
+// StepDown gives the lease up now, for a process on its way out: called
+// before the agents are let go, so the standby they move to is already the
+// leader, and before the process exits, which would otherwise race the
+// release in Run and leave the standby to wait out the lease. Run's own
+// release after it is then nothing.
+func (l *Leadership) StepDown() {
+	if l.store == nil {
+		return
+	}
+	l.stopped.Store(true)
+	l.step(context.Background())
+}
+
 // step gives the lease up on the way out, so a standby takes over in seconds
 // rather than waiting out the clock.
 func (l *Leadership) step(ctx context.Context) {
-	if !l.leader.Load() {
+	// Once: StepDown and Run's own way out both come here.
+	if !l.leader.CompareAndSwap(true, false) {
 		return
 	}
 	// A fresh context: the one that just ended is the reason this is running.
@@ -180,3 +205,16 @@ func newOwnerID() string {
 	}
 	return host + "-" + hex.EncodeToString(b[:])
 }
+
+// ControlRole is the lease's name in the store, for readers that ask who
+// holds it without competing.
+const ControlRole = controlRole
+
+// HolderName is the machine an owner id names: the hostname it was made
+// from, without the random part.
+func HolderName(owner string) string {
+	if i := strings.LastIndexByte(owner, '-'); i > 0 {
+		return owner[:i]
+	}
+	return owner
+}

internal/control/leadership_test.go

@@ -0,0 +1,51 @@
+package control
+
+import (
+	"log/slog"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+)
+
+func TestAStoppingLeaderHandsOverAtOnceAndNeverTakesItBack(t *testing.T) {
+	st, err := store.Open(t.Context(), filepath.Join(t.TempDir(), "s.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer st.Close()
+	log := slog.New(slog.DiscardHandler)
+	a := NewLeadership(st, time.Minute, log)
+	b := NewLeadership(st, time.Minute, log)
+
+	a.attempt(t.Context())
+	b.attempt(t.Context())
+	if !a.IsLeader() || b.IsLeader() {
+		t.Fatalf("a %v, b %v", a.IsLeader(), b.IsLeader())
+	}
+
+	// Stepping down gives the lease back, rather than leaving b to wait out
+	// the minute it was taken for.
+	a.StepDown()
+	if a.IsLeader() {
+		t.Error("still leading after stepping down")
+	}
+	// A renewal already on its way when the process began to stop must not
+	// take it back.
+	a.attempt(t.Context())
+	if a.IsLeader() {
+		t.Error("took the lease again after stepping down")
+	}
+	b.attempt(t.Context())
+	if !b.IsLeader() {
+		t.Error("the standby could not take the lease that was given up")
+	}
+	lease, err := st.Leadership(t.Context(), ControlRole)
+	if err != nil || lease.Owner != b.Owner() {
+		t.Errorf("the lease is %q (%v), not b's", lease.Owner, err)
+	}
+	if got := HolderName(b.Owner()); got == "" || got == b.Owner() {
+		t.Errorf("holder name %q from %q", got, b.Owner())
+	}
+}