feat: More logging for command stream

.gitea/workflows/chart-release.yml

@@ -59,6 +59,23 @@ jobs:
                     --set server.replicaCount=3 \
                     --set web.replicaCount=3 > /dev/null
 
+            # The reaper deletes whole instances, so "does this env appear only
+            # in cloud mode" is worth asserting rather than eyeballing.
+            - name: Check the reaper is cloud-only
+              run: |
+                  set -eu
+                  if helm template test "$CHART_DIR" | grep -q FREE_INSTANCE_REAP_AFTER; then
+                      echo "FREE_INSTANCE_REAP_AFTER is set on a self-hosted render"
+                      exit 1
+                  fi
+                  if ! helm template test "$CHART_DIR" \
+                        --set server.env.deploymentType=cloud \
+                      | grep -q FREE_INSTANCE_REAP_AFTER; then
+                      echo "FREE_INSTANCE_REAP_AFTER is missing from a cloud render"
+                      exit 1
+                  fi
+                  echo "ok: reaper configured in cloud mode only"
+
             - name: Render against external Redis and MongoDB
               run: |
                   helm template test "$CHART_DIR" \
@@ -76,6 +93,20 @@ jobs:
                     --set ingress.tls.certResolver=letsencrypt \
                     --set server.env.grpcHost=agents.example.com:443 > /dev/null
 
+            # The shape the cloud deployment actually uses: a wildcard tenant
+            # namespace, /api and /auth routed at the edge, and no apex — that
+            # belongs to the marketing site, which this chart does not deploy.
+            - name: Render a wildcard host with edge-routed API paths
+              run: |
+                  helm template test "$CHART_DIR" \
+                    --set ingress.enabled=true \
+                    --set 'ingress.web.host=*.vantage.example.com' \
+                    --set ingress.api.enabled=true \
+                    --set ingress.grpc.host=agents.example.com \
+                    --set server.env.grpcHost=agents.example.com:443 \
+                    --set ingress.tls.secretName=vantage-tls \
+                    --set ingress.tls.grpcSecretName=agents-tls > /dev/null
+
             # The guards are load-bearing, so their absence is a regression the
             # same way a broken render is. Each of these must fail.
             - name: Check the guards still refuse bad values
@@ -99,6 +130,12 @@ jobs:
                       --set server.replicaCount=2 --set server.persistence.enabled=true
                   refuses "ingress with no web host" \
                       --set ingress.enabled=true
+                  refuses "edge-routed API with an empty path list" \
+                      --set ingress.enabled=true \
+                      --set ingress.web.host=vantage.example.com \
+                      --set ingress.grpc.enabled=false \
+                      --set ingress.api.enabled=true \
+                      --set 'ingress.api.paths=null'
                   refuses "gRPC ingress with no host" \
                       --set ingress.enabled=true \
                       --set ingress.web.host=vantage.example.com \

agent/internal/grpc/pb/vantage.pb.go

@@ -166,8 +166,14 @@ type ServerCommand struct {
 	RunStep          *RunStepCmd          `json:"run_step,omitempty"`
 	CleanupWorkspace *CleanupWorkspaceCmd `json:"cleanup_workspace,omitempty"`
 	OpenProxy        *OpenProxyCmd        `json:"open_proxy,omitempty"`
+	Ping             *PingCmd             `json:"ping,omitempty"`
 }
 
+// PingCmd is a server-originated liveness beat. It carries nothing and expects
+// no reply: its arrival is the entire message. See the .proto for why gRPC
+// keepalive is not sufficient on its own.
+type PingCmd struct{}
+
 
 
 type CleanupWorkspaceCmd struct {

agent/internal/sync/sync.go

@@ -118,9 +118,36 @@ func poll(client *grpcclient.Client, cfg *config.Config, version string) error {
 	return nil
 }
 
+// How long a command stream must survive before it counts as having worked.
+// Past this, the next drop is treated as a fresh incident rather than as the
+// continuation of a run of failures.
+const streamHealthyAfter = time.Minute
+
+// Stream staleness. The server beats every 20s, so 70s tolerates three missed
+// beats before the stream is written off — high enough that a slow network or a
+// briefly busy server does not cost a reconnect, low enough that an agent is
+// not uncommandable for minutes after a control-plane restart.
+const (
+	streamStaleAfter = 70 * time.Second
+	streamStaleCheck = 10 * time.Second
+
+	// How often a healthy stream reports itself. Also the interval at which an
+	// agent talking to a control plane too old to send heartbeats says so —
+	// that agent is running without a watchdog, and the journal should not be
+	// silent about it.
+	pingSummaryInterval = 5 * time.Minute
+)
+
 func runCommandStream(ctx context.Context, cfg *config.Config) {
 	backoff := time.Second
-	const maxBackoff = 2 * time.Minute
+
+	// Two minutes was the old ceiling, and it was reached far too easily. The
+	// command stream is what makes this agent controllable at all: while it is
+	// down, workflows and console sessions fail as "agent offline" even though
+	// SyncKeys keeps polling happily and the fleet list still shows the server
+	// active. A shorter ceiling costs a few reconnect attempts; the old one cost
+	// two minutes of an agent that looks fine and answers nothing.
+	const maxBackoff = 30 * time.Second
 
 	for {
 		select {
@@ -129,22 +156,45 @@ func runCommandStream(ctx context.Context, cfg *config.Config) {
 		default:
 		}
 
-		if err := connectAndHandleStream(ctx, cfg); err != nil {
-			if ctx.Err() != nil {
-				return
-			}
-			log.Printf("command stream error: %v, reconnecting in %s", err, backoff)
-			select {
-			case <-ctx.Done():
-				return
-			case <-time.After(backoff):
-			}
-			if backoff < maxBackoff {
-				backoff *= 2
-			}
-		} else {
+		started := time.Now()
+		err := connectAndHandleStream(ctx, cfg)
+		if ctx.Err() != nil {
+			return
+		}
+
+		// A stream that stayed up is evidence the control plane is reachable,
+		// whatever ended it. Without this the backoff only ever climbed:
+		// connectAndHandleStream returns an error on *every* stream end,
+		// including a healthy one dropped by a routine deploy, so an agent
+		// pinned itself at the ceiling after a handful of ordinary restarts and
+		// stayed there for the rest of its life.
+		if time.Since(started) >= streamHealthyAfter {
 			backoff = time.Second
 		}
+
+		// The uptime is in the line because it is what distinguishes a stream
+		// that never worked from one that ran for hours and was dropped by a
+		// deploy — and it is the same measure that decides whether the backoff
+		// resets, so a reader can see why the delay is what it is.
+		up := time.Since(started).Truncate(time.Second)
+		if err != nil {
+			log.Printf("command stream error after %s: %v, reconnecting in %s", up, err, backoff)
+		} else {
+			log.Printf("command stream closed after %s, reconnecting in %s", up, backoff)
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+
+		if backoff < maxBackoff {
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
 	}
 }
 
@@ -155,7 +205,13 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
 	}
 	defer client.Close()
 
-	stream, err := client.CommandStream(ctx)
+	// Cancelling this context is what unblocks Recv when the stream has gone
+	// quiet. Without it the watchdog below would have no way to interrupt a
+	// read that is never going to return.
+	streamCtx, abandon := context.WithCancel(ctx)
+	defer abandon()
+
+	stream, err := client.CommandStream(streamCtx)
 	if err != nil {
 		return fmt.Errorf("open stream: %w", err)
 	}
@@ -168,7 +224,7 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
 		return fmt.Errorf("send auth: %w", err)
 	}
 
-	log.Println("command stream connected")
+	log.Printf("command stream connected to %s", cfg.ServerURL)
 
 	var sendMu sync.Mutex
 	send := func(msg *pb.AgentMessage) error {
@@ -177,11 +233,93 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
 		return stream.Send(msg)
 	}
 
+	// Stream liveness, tracked here rather than left to gRPC keepalive.
+	//
+	// Keepalive operates on the transport, and behind an L7 proxy the transport
+	// ends at the proxy: it answers pings whether or not the server behind it
+	// is still running. A control-plane pod that dies therefore leaves this
+	// agent blocked in Recv on a stream that will never deliver another message
+	// and never error, while the control plane dispatches commands into it and
+	// the operator watches nothing happen.
+	//
+	// The watchdog only arms once a ping has actually been seen. A server too
+	// old to send them must not be treated as dead — that would put the agent
+	// in a reconnect loop against a control plane that is working perfectly.
+	var (
+		lastMu   sync.Mutex
+		lastRecv = time.Now()
+		pinged   bool
+		beats    int
+	)
+	markRecv := func(isPing bool) {
+		lastMu.Lock()
+		lastRecv = time.Now()
+		if isPing {
+			beats++
+			// Logged once per stream, because it is the moment the agent starts
+			// holding the control plane to account: before this the watchdog is
+			// disarmed and a dead stream would go unnoticed indefinitely.
+			if !pinged {
+				pinged = true
+				log.Printf("command stream heartbeat detected, watchdog armed (%s threshold)", streamStaleAfter)
+			}
+		}
+		lastMu.Unlock()
+	}
+
+	go func() {
+		t := time.NewTicker(streamStaleCheck)
+		defer t.Stop()
+
+		// Reported periodically rather than per beat: at one every 20s the
+		// journal would be nothing else. The count is what makes a partial
+		// failure visible — beats arriving but fewer than expected is a
+		// different problem from beats stopping altogether.
+		summary := time.NewTicker(pingSummaryInterval)
+		defer summary.Stop()
+
+		for {
+			select {
+			case <-streamCtx.Done():
+				return
+			case <-summary.C:
+				lastMu.Lock()
+				n, armed := beats, pinged
+				beats = 0
+				lastMu.Unlock()
+				if armed {
+					log.Printf("command stream healthy, %d heartbeats in the last %s", n, pingSummaryInterval)
+				} else {
+					log.Printf("command stream up but sending no heartbeats; "+
+						"control plane predates them, watchdog stays disarmed (last message %s ago)",
+						time.Since(lastRecv).Truncate(time.Second))
+				}
+			case <-t.C:
+				lastMu.Lock()
+				idle, armed := time.Since(lastRecv), pinged
+				lastMu.Unlock()
+				if armed && idle > streamStaleAfter {
+					log.Printf("command stream silent for %s (threshold %s), assuming it is dead and reconnecting",
+						idle.Truncate(time.Second), streamStaleAfter)
+					abandon()
+					return
+				}
+			}
+		}
+	}()
+
 	for {
 		cmd, err := stream.Recv()
 		if err != nil {
 			return fmt.Errorf("recv: %w", err)
 		}
+		markRecv(cmd.Ping != nil)
+
+		// Pings carry nothing and are not acknowledged; being received is their
+		// whole purpose.
+		if cmd.Ping != nil {
+			continue
+		}
 
 		if cmd.GenerateKey != nil {
 			go handleGenerateKey(cfg, cmd)

claude.md

@@ -150,6 +150,13 @@ server is behind NAT on a private address. It also means the console now
 **requires a live agent** on every deployment: `consoleConnect` answers 409
 `agent_offline` rather than hanging.
 
+**guacd's Service is headless on purpose.** The server resolves `GUACD_ADDR` to
+build the allow-list of sources permitted to claim a relay listener; a ClusterIP
+resolves to the Service's virtual address while guacd connects from its *pod*
+IP, so every relay connection is rejected and every session dies with
+`waiting for guacd: i/o timeout`. Compose is immune — there the name resolves to
+the address that connects.
+
 SSH connections authenticate with a stored private key; RDP/VNC credentials are
 encrypted, single-use, and consumed when the tunnel opens. None of them reach
 the agent — the session is negotiated end-to-end between guacd and the target
@@ -171,10 +178,10 @@ rare cross-pod branch that only fails under load.
 | Concern                | How it crosses replicas                                                                                                                                                                       |
 | ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | Which pod owns an agent | `vantage:agent:<server_id>` holds the owner's node ID with a 30s TTL, renewed every 10s. `Dispatcher.IsConnected` is an `EXISTS` on it                                                        |
-| Sending a command       | published to `vantage:cmd:<server_id>`; the owner pod acks on `vantage:ack:<command_id>`. **Request/ack, not a queue** — a command whose owner died must fail loudly (503) rather than queue |
+| Sending a command       | published to `vantage:cmd:<server_id>`; the owner pod acks on `vantage:ack:<command_id>`. **Request/ack, not a queue** — a command whose owner died must fail loudly (503) rather than queue. The envelope carries `node`, the presence holder resolved at publish time, and a pod ignores envelopes addressed elsewhere: the channel is a fan-out, and during a reconnect a half-open stream's pod is still subscribed. Unaddressed, it could ack first and queue the command onto a dead stream — the operator told it worked, the agent never seeing it. Presence renewal is owner-only (`RenewPresence`) for the same reason: a blind `SET` let the stale pod steal the key back every 10s |
 | Step results            | the owner pod publishes to `vantage:res:<command_id>`; the pod driving the run subscribes **before** dispatching, or a fast agent answers into a channel nobody has joined                    |
 | Step output             | never crosses. The dispatch envelope carries the secret mask list, so the owner pod masks and writes lines itself — unmasked bytes stay off the bus                                          |
-| Console relay           | the envelope asks the owner pod to bind the listener, and the ack returns **that pod's** address for guacd. The relay's failure reason comes back on `vantage:proxyend:<proxy_id>`           |
+| Console relay           | **not routed to the owner pod at all.** A `ProxyStream` is its own HTTP/2 request and an L7 proxy balances requests, not connections, so it does not follow the command stream — the listener therefore cannot be bound in advance. Whichever pod receives the stream binds it and announces **its own** address on `vantage:proxyaddr:<proxy_id>`; `vantage:proxypending:<proxy_id>` (30s, consumed atomically) is what authorises the claim, and the failure reason comes back on `vantage:proxyend:<proxy_id>` |
 | Background jobs         | `bus.RunAsLeader` — one Redis lock named `housekeeping`                                                                                                                                       |
 
 **Workflow logs are in MongoDB** (`workflow_log_lines`, one document per line,
@@ -188,6 +195,25 @@ marker is written and the rest is dropped. Without that cap a `yes` in a step
 is a database incident. **Nothing writes to `/data` any more**, which is why
 `server.persistence` now defaults to off and `VANTAGE_WORKFLOW_LOG_DIR` is gone.
 
+**Shutdown order is load-bearing.** `main` traps SIGTERM, stops gRPC
+(`GracefulStop`, 10s cap) and only then drains HTTP. Each `CommandStream`
+handler releases its agent's presence claim on return, so a killed process
+leaves `vantage:agent:<server_id>` behind for the rest of its 30s TTL — during
+which other replicas dispatch to a pod that has exited and the caller sees
+`agent offline` for a perfectly healthy agent. Draining HTTP first would hold
+those claims for the length of the drain, which is why gRPC goes first. The
+chart's `server.terminationGracePeriodSeconds` (30s) must stay above the
+10s + 10s the stop sequence needs, or the kubelet SIGKILLs mid-shutdown and the
+handling buys nothing.
+
+The agent side of the same failure: `runCommandStream` resets its backoff only
+after a stream that survived `streamHealthyAfter`. `connectAndHandleStream`
+returns an error on *every* stream end, healthy ones included, so without that
+reset the backoff only ever climbed — an agent pinned itself at the ceiling
+after a handful of ordinary deploys and stayed there. The ceiling is 30s, not
+minutes, because while the stream is down the agent still polls `SyncKeys` and
+still reads as `active` in the fleet list while answering no commands at all.
+
 **The leader lock is not an optimisation.** N replicas each running the monitor
 scheduler means each check fires N times, each incident notification reaches the
 customer N times, and each hourly rollup is written N times; N reapers race to
@@ -369,7 +395,18 @@ service Vantage {
 
 `CommandStream` is the only streaming RPC: the agent authenticates once with `AgentReady`, then the server pushes `ServerCommand`s and the agent replies with `CommandResult`, `StepResult`, or `StepOutputChunk`.
 
-`ServerCommand` variants: `GenerateKeyCmd`, `DeleteKeyCmd`, `UpdateAgentCmd`, `ApplyUpdatesCmd`, `RunStepCmd`, `CleanupWorkspaceCmd`.
+`ServerCommand` variants: `GenerateKeyCmd`, `DeleteKeyCmd`, `UpdateAgentCmd`, `ApplyUpdatesCmd`, `RunStepCmd`, `CleanupWorkspaceCmd`, `OpenProxyCmd`, `PingCmd`.
+
+**`PingCmd` is a liveness beat, and it is not redundant with gRPC keepalive.**
+The server sends one every 20s on an otherwise idle command stream; the agent
+treats 70s of silence as a dead stream and reconnects. Keepalive cannot do this
+job behind an L7 proxy: the agent's HTTP/2 connection terminates at the proxy,
+which answers pings on its own behalf, so a control-plane pod that dies leaves
+the agent blocked in `Recv` on a stream that never delivers another message and
+never errors — commands dispatched into it are silently lost while `SyncKeys`
+keeps succeeding and the fleet list still shows the server `active`. The agent's
+watchdog arms only **after** it has seen a first ping, so an older server that
+sends none is treated as working rather than put into a reconnect loop.
 
 Key-state polling stays on the 30s `SyncKeys` interval. Full message definitions live in `proto/vantage/v1/vantage.proto`.
 
@@ -610,12 +647,15 @@ Windows: MSI built by CI (WiX), or `installer/setup.ps1` registering the agent a
 
 `ingress.enabled` publishes **two** hostnames, because the two audiences arrive over different protocols:
 
-| Values                | Route                                                                                     |
-| --------------------- | ----------------------------------------------------------------------------------------- |
-| `ingress.web.host`    | browsers → `web:3000`. Everything, including `/api` — see below                            |
-| `ingress.grpc.host`   | agents → a dedicated `<release>-server-grpc` Service on 9090, annotated `serversscheme: h2c` |
+| Values                                       | Route                                                                                       |
+| -------------------------------------------- | --------------------------------------------------------------------------------------------- |
+| `ingress.web.host` (+ `web.extraHosts`)      | browsers → `web:3000`                                                                        |
+| `ingress.api.paths` (when `api.enabled`)     | `/api`, `/auth` → `<release>-server:8080`, bypassing the Next proxy                          |
+| `ingress.grpc.host`                          | agents → a dedicated `<release>-server-grpc` Service on 9090, annotated `serversscheme: h2c` |
 
-**The server's HTTP port is deliberately not publishable.** `web` already proxies `/api`, `/auth` and the install scripts to it (`web/next.config.ts`), so a second route would be a second front door to the same API with none of that routing — and the console WebSocket and ESO token path would then exist at two addresses with different behaviour.
+**`ingress.web.host` is normally a wildcard.** `*.vantage.example.com` is the per-tenant instance namespace — `APP_ROOT_LABEL` resolves the instance from the label. A Kubernetes wildcard host matches **exactly one** label, so it does not match the apex, and here that is correct rather than a gap: `vantage.hostxtra.co.uk` is the marketing site (`site/`, in `docker-compose.site.yml`), which this chart does not deploy. `extraHosts` is for a genuine second name; adding the apex to it would put the control plane on the marketing host. Every host in the list gets identical paths.
+
+**`ingress.api.enabled` routes `/api` and `/auth` straight to the server.** Both arrangements work — without it `web` proxies those prefixes onward itself (`web/next.config.ts`) — but edge routing is one hop shorter and matches what the Nginx Proxy Manager in front of the Docker deployment already does, so leaving it off makes the request path a different shape on Kubernetes than in production. It stays **off by default** because it only helps where the server is reachable on the same host and certificate as `web`; turning it on blindly moves the whole API onto a route that may not be provisioned. Traefik derives router priority from rule length, so `PathPrefix(/api)` outranks the catch-all `/` with no priority annotation needed.
 
 **The gRPC route needs its own Service.** The server terminates no TLS; it speaks plain h2c and always has, with TLS terminated by whatever sits in front. Traefik will not use h2c to a backend unless the *Service* says so, and that annotation applies to every port on the Service — so annotating the shared two-port `<release>-server` would force h2c on its HTTP port too.
 

deploy/chart/vantage/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: vantage
 description: Helm chart for the Vantage stack (Redis, MongoDB, guacd, server, web)
 type: application
-version: 1.0.2
-appVersion: "1.0.2"
+version: 1.0.7
+appVersion: "1.0.7"

deploy/chart/vantage/templates/NOTES.txt

@@ -37,9 +37,16 @@ Scaling (server.replicaCount / web.replicaCount):
 {{- if .Values.ingress.enabled }}
 
 Ingress (Traefik):
-  - UI and API:  https://{{ .Values.ingress.web.host }}
-    Everything browsers need goes here; web proxies /api, /auth and the install
-    scripts to the server, so the server's HTTP port is not published separately.
+  - Browsers:    https://{{ .Values.ingress.web.host }}
+{{- range .Values.ingress.web.extraHosts }}
+                 https://{{ . }}
+{{- end }}
+{{- if .Values.ingress.api.enabled }}
+    {{ join ", " .Values.ingress.api.paths }} go straight to the server; everything else to web.
+{{- else }}
+    Everything goes to web, which proxies /api and /auth onward. Set
+    ingress.api.enabled=true to route them at the edge instead.
+{{- end }}
 {{- if .Values.ingress.grpc.enabled }}
   - Agents:      {{ .Values.ingress.grpc.host }} (gRPC, h2c behind TLS)
     Agents dial server.env.grpcHost, currently {{ tpl .Values.server.env.grpcHost . }}.

deploy/chart/vantage/templates/_helpers.tpl

@@ -72,11 +72,12 @@ both read it.
   value: {{ .Values.server.env.proxyAdvertiseHost | quote }}
 - name: PROXY_LISTEN_HOST
   value: {{ .Values.server.env.proxyListenHost | quote }}
-# The address guacd dials to reach a console relay. It must name one pod, not
-# the Service: the relay listener is bound by whichever pod holds that agent's
-# command stream, and a Service would send guacd to a different one. POD_IP
-# takes precedence over PROXY_ADVERTISE_HOST in the server for exactly this
-# reason, so the setting above stays meaningful only outside Kubernetes.
+{{- if eq .Values.server.env.deploymentType "cloud" }}
+- name: VANTAGE_DEPLOYMENT
+  value: "cloud"
+- name: FREE_INSTANCE_REAP_AFTER
+  value: {{ .Values.server.env.freeInstanceReapAfter | quote }}
+{{- end }}
 - name: POD_IP
   valueFrom:
     fieldRef:

deploy/chart/vantage/templates/guacd.yaml

@@ -36,6 +36,9 @@ metadata:
     app.kubernetes.io/component: guacd
 spec:
   type: {{ .Values.guacd.service.type }}
+  {{- if eq .Values.guacd.service.type "ClusterIP" }}
+  clusterIP: None
+  {{- end }}
   selector:
     app.kubernetes.io/instance: {{ .Release.Name }}
     app.kubernetes.io/component: guacd

deploy/chart/vantage/templates/ingress.yaml

@@ -2,10 +2,24 @@
 {{/*
 Two hostnames, because the two audiences arrive over different protocols.
 
-Browsers reach `web`, and only `web`: it proxies /api, /auth and the install
-scripts through to the server itself (see web/next.config.ts), so publishing the
-server's HTTP port separately would be a second front door to the same API with
-none of the same routing.
+Browsers reach the web host. What answers there depends on the path: with
+ingress.api.enabled, /api and /auth go straight to the server and everything
+else goes to `web`. Without it, everything goes to `web`, which proxies those
+prefixes onward itself (web/next.config.ts).
+
+Both work. Routing at the edge is one hop shorter and is what the Nginx Proxy
+Manager deployment in front of the Docker install already does, so leaving it
+off changes the shape of the request path between the two deployments. It is
+still off by default, because turning it on where `web` is the only thing with
+a public certificate would strand /api behind a route nobody can reach.
+
+The web host is normally a wildcard — `*.vantage.example.com` — because that is
+the per-tenant instance namespace; APP_ROOT_LABEL resolves the instance from the
+label. Kubernetes wildcard hosts match exactly one label, so this does not match
+the apex, and on the deployment this chart was written for it must not: the apex
+is the marketing site, a separate application (see site/ and
+docker-compose.site.yml). extraHosts exists for a genuine second name, not for
+reclaiming the apex.
 
 Agents reach the server's gRPC port, which is plain h2c — the server holds no
 certificates of its own, TLS has always been terminated by whatever sits in
@@ -16,6 +30,11 @@ its HTTP port too.
 */}}
 {{- $tls := .Values.ingress.tls }}
 {{- $webHost := required "ingress.enabled requires ingress.web.host" .Values.ingress.web.host }}
+{{- $hosts := prepend .Values.ingress.web.extraHosts $webHost }}
+{{- $apiPaths := .Values.ingress.api.paths }}
+{{- if and .Values.ingress.api.enabled (not $apiPaths) }}
+{{- fail "ingress.api.enabled requires at least one path in ingress.api.paths" }}
+{{- end }}
 apiVersion: networking.k8s.io/v1
 kind: Ingress
 metadata:
@@ -41,20 +60,42 @@ spec:
   {{- if and $tls.enabled $tls.secretName }}
   tls:
     - hosts:
-        - {{ $webHost | quote }}
+        {{- range $hosts }}
+        - {{ . | quote }}
+        {{- end }}
       secretName: {{ $tls.secretName }}
   {{- end }}
   rules:
-    - host: {{ $webHost | quote }}
+    {{- range $host := $hosts }}
+    - host: {{ $host | quote }}
       http:
         paths:
+          {{- /*
+            The API paths come first and, more importantly, are longer. Traefik
+            derives router priority from rule length, so Host(x) &&
+            PathPrefix(/api) outranks Host(x) && PathPrefix(/) without anyone
+            having to assign priorities by hand. Order within the list is for
+            the reader; the length is what decides.
+          */}}
+          {{- if $.Values.ingress.api.enabled }}
+          {{- range $apiPaths }}
+          - path: {{ . | quote }}
+            pathType: Prefix
+            backend:
+              service:
+                name: {{ $.Release.Name }}-server
+                port:
+                  number: {{ $.Values.server.service.httpPort }}
+          {{- end }}
+          {{- end }}
           - path: /
             pathType: Prefix
             backend:
               service:
-                name: {{ .Release.Name }}-web
+                name: {{ $.Release.Name }}-web
                 port:
-                  number: {{ .Values.web.service.port }}
+                  number: {{ $.Values.web.service.port }}
+    {{- end }}
 {{- if .Values.ingress.grpc.enabled }}
 {{- $grpcHost := required "ingress.grpc.enabled requires ingress.grpc.host" .Values.ingress.grpc.host }}
 {{/*

deploy/chart/vantage/templates/server.yaml

@@ -48,6 +48,13 @@ spec:
         app.kubernetes.io/instance: {{ .Release.Name }}
         app.kubernetes.io/component: server
     spec:
+      # The server stops gRPC before draining HTTP, so that every CommandStream
+      # handler returns and releases its agent's presence claim. A claim left
+      # behind outlives the pod for its 30s TTL, and during that window other
+      # replicas dispatch commands to a process that has exited — surfacing to
+      # the operator as "agent offline" on an agent that is perfectly healthy.
+      # 10s for gRPC plus 10s for the HTTP drain, with headroom.
+      terminationGracePeriodSeconds: {{ .Values.server.terminationGracePeriodSeconds }}
       {{- if .Values.imagePullSecrets }}
       imagePullSecrets:
         {{- toYaml .Values.imagePullSecrets | nindent 8 }}

deploy/chart/vantage/values.yaml

@@ -1,9 +1,7 @@
 # Default values for the vantage chart.
 
 redis:
-    # false deploys no Redis and points the server at `redis.addr` instead.
     enabled: true
-    # Only read when enabled is false. host:port of an external Redis.
     addr: ""
     image:
         repository: redis
@@ -14,21 +12,14 @@ redis:
         storageClass: ""
         accessMode: ReadWriteOnce
     port: 6379
-    # Both empty for an unauthenticated Redis. Redis 6+ ACL auth takes both; a
-    # legacy `requirepass` instance takes the password alone and must leave the
-    # username empty. Set existingSecret to keep the password out of values.
     auth:
         username: ""
         password: ""
-        # Secret holding the credentials. When set, username/password above are
-        # ignored and these keys are read from the secret instead.
         existingSecret: ""
         usernameKey: username
         passwordKey: password
 
 mongo:
-    # false deploys no MongoDB. server.env.mongoUri must then point at an
-    # external one — the chart cannot guess it, and refuses to render without it.
     enabled: true
     image:
         repository: mongo
@@ -49,22 +40,14 @@ guacd:
         port: 4822
 
 server:
-    # Safe to raise. Agent commands, step results and console relays are routed
-    # between replicas over Redis, workflow logs live in MongoDB, and the
-    # background jobs (monitor scheduler, reaper, retention sweeps) run under a
-    # Redis leader lock so exactly one replica performs them.
-    #
-    # Two requirements come with raising it: server.persistence.enabled must be
-    # false (or the volume ReadWriteMany), and Redis must be shared by every
-    # replica — the bus is not optional and a per-pod Redis would partition it.
     replicaCount: 1
-    # Runs migrations, index builders and default-step seeding once, as a Helm
-    # pre-install/pre-upgrade hook, instead of in every starting pod. Leave it
-    # on for Kubernetes. Turning it off puts schema setup back in the pods.
+    # Must exceed the server's own stop sequence (10s gRPC GracefulStop + 10s
+    # HTTP drain) or the kubelet SIGKILLs mid-shutdown, which is exactly the
+    # abrupt exit that leaves agent presence claims stranded in Redis.
+    terminationGracePeriodSeconds: 30
     migrationJob:
         enabled: true
         backoffLimit: 0
-        # 15 minutes: the instance rename alone carries a 10-minute budget.
         activeDeadlineSeconds: 900
     image:
         repository: gitea.hostxtra.co.uk/mrhid6/vantage/server
@@ -78,16 +61,8 @@ server:
         grpcHost: "{{ .Release.Name }}-server:9090"
         keyEncryptionKey: ""
         appRootLabel: vantage
-        # Ignored under Kubernetes: the chart sets POD_IP from the downward API
-        # and the server prefers it, because a console relay listener belongs to
-        # one pod and a Service address cannot name one.
         proxyAdvertiseHost: "{{ .Release.Name }}-server"
         proxyListenHost: "0.0.0.0"
-    # Off by default: nothing in the server writes to disk any more. Workflow
-    # logs, the only thing that ever did, are in MongoDB so that every replica
-    # can read and write them. Turn this on only to reach files left behind by
-    # a release that predates that move — and note a ReadWriteOnce volume caps
-    # replicaCount at 1 while it is on.
     persistence:
         enabled: false
         size: 1Gi
@@ -96,8 +71,6 @@ server:
         hostPath: /data
 
 web:
-    # Stateless — safe to raise. Pin web.image.tag when you do: replicas on
-    # different builds serve mismatched chunk hashes and the UI 404s mid-session.
     replicaCount: 1
     image:
         repository: gitea.hostxtra.co.uk/mrhid6/vantage/web
@@ -108,35 +81,29 @@ web:
     env:
         apiUrl: "http://{{ .Release.Name }}-server:8080"
 
-# Traefik ingress. Two hostnames, because the two audiences arrive differently:
-# browsers reach `web` (which proxies /api, /auth and the install scripts to the
-# server), and agents reach the server's gRPC port directly.
-#
-# Publishing the server's HTTP port is deliberately not offered — it would be a
-# second door to the same API, bypassing the routing web already performs.
 ingress:
     enabled: false
     className: traefik
-    # Traefik entrypoint name. `websecure` is the default TLS entrypoint in the
-    # official chart; installs that renamed it must say so here.
     entrypoint: websecure
-    # Applied to the web router only. Middlewares, rate limits, IP allow lists.
     annotations: {}
     web:
-        # Required when ingress.enabled. The hostname users open in a browser.
         host: ""
+        extraHosts: []
+    api:
+        enabled: false
+        paths:
+            - /api
+            - /auth
+            - /update
+            - /install
+            - /update.ps1
+            - /install.ps1
     grpc:
-        # Agents dial this. Turning it off means agents reach gRPC some other
-        # way — a LoadBalancer Service, a node port, or an in-cluster path.
         enabled: true
         host: ""
         annotations: {}
     tls:
         enabled: true
-        # Either name pre-existing certificate Secrets, or leave both empty and
-        # set certResolver to have Traefik obtain them (ACME). Setting neither
-        # produces a TLS router with no certificate, which serves Traefik's
-        # self-signed default — valid-looking and trusted by nothing.
         secretName: ""
         grpcSecretName: ""
         certResolver: ""

proto/vantage/v1/vantage.proto

@@ -182,9 +182,22 @@ message ServerCommand {
     RunStepCmd      run_step      = 6;
     CleanupWorkspaceCmd cleanup_workspace = 7;
     OpenProxyCmd open_proxy = 8;
+    PingCmd ping = 9;
   }
 }
 
+// PingCmd is a liveness beat, carrying nothing and requiring no reply.
+//
+// It exists because gRPC keepalive cannot prove what the agent needs to know.
+// Behind an L7 proxy the agent's HTTP/2 connection terminates at the proxy, so
+// keepalive pings are answered by the proxy whether or not the server behind it
+// is still there. A pod that dies leaves the agent blocked in Recv on a stream
+// that will never produce another message and never error — commands are
+// dispatched into it and silently lost. Only traffic that originates at the
+// server itself distinguishes a live stream from an orphaned one.
+message PingCmd {
+}
+
 // CleanupWorkspaceCmd tells the agent to recursively remove the run's working
 // directory once all steps on that server have finished.
 message CleanupWorkspaceCmd {

server/cmd/main.go

@@ -2,9 +2,13 @@ package main
 
 import (
 	"context"
+	"errors"
 	"log"
+	"net/http"
 	"os"
+	"os/signal"
 	"strings"
+	"syscall"
 	"time"
 
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api"
@@ -141,13 +145,15 @@ func serve() {
 	}
 	log.Printf("message bus ready as node %s", bus.NodeID())
 
-	ctx := context.Background()
+	// Cancelled on SIGTERM/SIGINT. Everything below that takes a context — the
+	// housekeeping jobs, the leader lock — stops when the pod is asked to.
+	ctx, shutdown := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer shutdown()
 
-	go func() {
-		if err := grpcserver.StartGRPC(9090); err != nil {
-			log.Fatalf("gRPC server error: %v", err)
-		}
-	}()
+	stopGRPC, err := grpcserver.StartGRPC(9090)
+	if err != nil {
+		log.Fatalf("gRPC server error: %v", err)
+	}
 
 	// Everything below runs on exactly one replica at a time.
 	//
@@ -183,12 +189,37 @@ func serve() {
 	r.Use(corsMiddleware())
 	api.RegisterRoutes(r)
 
-	log.Println("REST server listening on :8080")
-	if err := r.Run(":8080"); err != nil {
-		log.Fatalf("REST server error: %v", err)
+	srv := &http.Server{Addr: ":8080", Handler: r}
+	go func() {
+		log.Println("REST server listening on :8080")
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			log.Fatalf("REST server error: %v", err)
+		}
+	}()
+
+	<-ctx.Done()
+	log.Println("shutdown signal received")
+
+	// gRPC first, and this ordering is the point of the whole exercise. Stopping
+	// it runs each CommandStream handler's deferred release, which clears that
+	// agent's presence claim; until that happens another replica will keep
+	// dispatching commands to this process. Draining HTTP first would leave the
+	// claims held for the length of the drain.
+	stopGRPC()
+
+	drainCtx, cancelDrain := context.WithTimeout(context.Background(), httpDrainTimeout)
+	defer cancelDrain()
+	if err := srv.Shutdown(drainCtx); err != nil {
+		log.Printf("REST server shutdown: %v", err)
 	}
+	log.Println("shutdown complete")
 }
 
+// How long in-flight REST requests are given to finish. Console tunnels are
+// long-lived WebSockets that will not end on their own, so this is a ceiling
+// rather than a target; the relays behind them are already gone by this point.
+const httpDrainTimeout = 10 * time.Second
+
 func corsMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		c.Header("Access-Control-Allow-Origin", "*")

server/internal/api/console.go

@@ -3,6 +3,7 @@ package api
 import (
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"net/http"
 	"os"
@@ -87,69 +88,109 @@ func queryIntDefault(r *http.Request, key string, def int) int {
 	return v
 }
 
+// consoleTunnel upgrades the browser's WebSocket and joins it to guacd.
+//
+// Every branch here logs. That is deliberate and worth keeping: this handler
+// spans four hops (session store, agent dispatch, relay announcement, guacd),
+// any of which can fail, and the client is told the same near-useless thing by
+// most of them — a 500 that guacamole then reports as an *upstream* error,
+// naming the wrong hop entirely. Without a line per branch the only evidence a
+// failure leaves is a GIN status code, and with several replicas you cannot
+// even tell which process produced it.
+//
+// Lines are prefixed with the session ID so one attempt can be followed across
+// pods, and the pod's own hostname so it is obvious which one served it.
 func consoleTunnel(c *gin.Context) {
+	host, _ := os.Hostname()
+
 	token := c.Query("token")
 	sessionID, err := services.VerifySessionToken(token)
 	if err != nil {
+		log.Printf("console[%s]: reject: invalid session token: %v", host, err)
 		c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid token"})
 		return
 	}
+
+	// Bound to the session from here on, so every later line correlates.
+	tlog := func(format string, args ...any) {
+		log.Printf("console[%s %s]: "+format, append([]any{host, sessionID}, args...)...)
+	}
+	tlog("tunnel opened by %s", actorFromCtx(c))
+
 	instanceID := auth.InstanceID(c)
 	sess, err := services.GetConsoleSession(instanceID, sessionID)
 	if err != nil {
+		tlog("reject: console session not found: %v", err)
 		c.JSON(http.StatusNotFound, gin.H{"error": "session not found"})
 		return
 	}
 
 	if actor := actorFromCtx(c); actor != sess.User {
+		tlog("reject: session belongs to %s, not %s", sess.User, actor)
 		c.JSON(http.StatusForbidden, gin.H{"error": "session belongs to another user"})
 		return
 	}
 
 	if err := services.ConsumeSessionToken(instanceID, sessionID); err != nil {
+		tlog("reject: token already consumed: %v", err)
 		c.JSON(http.StatusUnauthorized, gin.H{"error": "token already used"})
 		return
 	}
 
 	srv, err := services.GetServer(auth.InstanceID(c), sess.ServerID)
 	if err != nil {
+		tlog("reject: server %s not found: %v", sess.ServerID, err)
 		c.JSON(http.StatusNotFound, gin.H{"error": "server not found"})
 		return
 	}
+	tlog("server %s (%s), protocol %s", srv.ServerID, srv.Hostname, sess.Protocol)
 
 	var privKey, passphrase string
 	if sess.Protocol == "ssh" && sess.KeyID != "" {
 		privKey, err = services.GetPrivateKey(auth.InstanceID(c), sess.KeyID)
 		if err != nil {
+			tlog("reject: key %s has no private material: %v", sess.KeyID, err)
 			c.JSON(http.StatusBadRequest, gin.H{"error": "selected key has no private material"})
 			return
 		}
 		passphrase, _ = services.GetPassphrase(sess.KeyID)
+		tlog("ssh key %s loaded (passphrase=%t)", sess.KeyID, passphrase != "")
 	}
 
 	var rdpUser, rdpPass string
 	if sess.Protocol == "rdp" || sess.Protocol == "vnc" {
 		rdpUser, rdpPass, err = services.ConsumeConsoleRDPCreds(instanceID, sessionID)
 		if err != nil {
+			tlog("reject: could not consume %s credentials: %v", sess.Protocol, err)
 			c.JSON(http.StatusInternalServerError, gin.H{"error": "could not load credentials"})
 			return
 		}
+		tlog("%s credentials consumed (user=%t)", sess.Protocol, rdpUser != "")
 	}
+
 	targetPort, err := services.TargetPort(srv, sess.Protocol)
 	if err != nil {
+		tlog("reject: no target port for %s: %v", sess.Protocol, err)
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
 	}
+	tlog("opening relay to %s:%d", srv.ServerID, targetPort)
 
 	relay, err := services.OpenConsoleProxy(instanceID, srv.ServerID, targetPort)
 	if err != nil {
 		if errors.Is(err, services.ErrAgentOffline) {
+			tlog("reject: agent offline")
 			c.JSON(http.StatusConflict, gin.H{"error": "agent_offline"})
 			return
 		}
+		// The client is deliberately told nothing specific, so this is the only
+		// place the real reason exists — a failed dispatch and a relay that was
+		// never announced are the same generic 500 to the browser.
+		tlog("reject: open relay: %v", err)
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "could not open relay"})
 		return
 	}
+	tlog("relay %s ready at %s:%d", relay.ProxyID, relay.Host, relay.Port)
 	// guac.WebsocketServer.ServeHTTP returns before installing its
 	// OnDisconnect handler when the connect callback errors, which is exactly
 	// the path every relay failure this proxy introduces takes (the agent
@@ -161,10 +202,14 @@ func consoleTunnel(c *gin.Context) {
 	defer func() {
 		relay.Close()
 		if reason := relay.Reason(); reason != "" {
+			tlog("relay %s ended: %s", relay.ProxyID, reason)
 			services.LogEvent(instanceID, "console.proxy_failed", actorFromCtx(c), srv.ServerID, "",
 				fmt.Sprintf("console relay failed: %s (proxy_id=%s, port=%d)", reason, relay.ProxyID, relay.Port))
+		} else {
+			tlog("relay %s closed cleanly", relay.ProxyID)
 		}
 		_ = services.EndConsoleSession(instanceID, sessionID)
+		tlog("tunnel finished")
 	}()
 
 	services.LogEvent(instanceID, "console.proxy_opened", actorFromCtx(c), srv.ServerID, "",
@@ -173,6 +218,7 @@ func consoleTunnel(c *gin.Context) {
 	gp, err := services.BuildGuacParams(sess.Protocol, sess.SSHUsername, privKey, passphrase,
 		rdpUser, rdpPass, relay.Host, relay.Port)
 	if err != nil {
+		tlog("reject: build guacd parameters: %v", err)
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
 	}
@@ -192,18 +238,33 @@ func consoleTunnel(c *gin.Context) {
 		config.OptimalScreenHeight = queryIntDefault(r, "height", 768)
 		config.OptimalResolution = queryIntDefault(r, "dpi", 96)
 
+		// Resolution is logged separately from the dial: a headless guacd
+		// Service returns pod addresses, and which one was picked is the
+		// difference between "guacd refused" and "we called the wrong guacd".
 		addr, err := net.ResolveTCPAddr("tcp", guacdAddr)
 		if err != nil {
+			tlog("guacd: resolve %s: %v", guacdAddr, err)
 			return nil, err
 		}
+		tlog("guacd: dialling %s (%s)", guacdAddr, addr.String())
+
 		conn, err := net.DialTCP("tcp", nil, addr)
 		if err != nil {
+			tlog("guacd: dial %s: %v", addr.String(), err)
 			return nil, err
 		}
+
+		// The handshake is where guacd connects onward to the relay, so a
+		// failure here is guacd reporting it could not reach %s:%d — the hop
+		// that has been hardest to see from either end.
 		stream := guac.NewStream(conn, guac.SocketTimeout)
 		if err := stream.Handshake(config); err != nil {
+			tlog("guacd: handshake for %s to relay %s:%d: %v",
+				gp.Protocol, relay.Host, relay.Port, err)
 			return nil, err
 		}
+		tlog("guacd: tunnel established (%s %dx%d)",
+			gp.Protocol, config.OptimalScreenWidth, config.OptimalScreenHeight)
 		return guac.NewSimpleTunnel(stream), nil
 	}
 
@@ -211,5 +272,7 @@ func consoleTunnel(c *gin.Context) {
 	// func above, not here: this only fires once a tunnel was actually
 	// established, and letting both paths log would double the audit event.
 	wsServer := guac.NewWebsocketServer(connect)
+	tlog("serving websocket")
 	wsServer.ServeHTTP(c.Writer, c.Request)
+	tlog("websocket returned")
 }

server/internal/api/handlers.go

@@ -142,7 +142,6 @@ func newServer(c *gin.Context) {
 	}
 	services.LogEvent(auth.InstanceID(c), "server.created", actorFromCtx(c), s.ServerID, "", "pre-registration token issued")
 
-
 	host := publicHostFromRequest(c)
 
 	installCmd := fmt.Sprintf(
@@ -415,7 +414,7 @@ if [ -z "$LATEST" ]; then
 fi
 
 VERSION="${LATEST#agent/}"
-LATEST_ENCODED="${LATEST/\
+LATEST_ENCODED="${LATEST/\//%%2F}"
 BINARY_URL="https://${GITEA_HOST}/mrhid6/vantage/releases/download/${LATEST_ENCODED}/vantage-agent-linux-${ARCH}"
 CHECKSUM_URL="https://${GITEA_HOST}/mrhid6/vantage/releases/download/${LATEST_ENCODED}/checksums.txt"
 
@@ -522,7 +521,7 @@ if [ -z "$LATEST" ]; then
 fi
 
 VERSION="${LATEST#agent/}"
-LATEST_ENCODED="${LATEST/\
+LATEST_ENCODED="${LATEST/\//%%2F}"
 BINARY_URL="https://${GITEA_HOST}/mrhid6/vantage/releases/download/${LATEST_ENCODED}/vantage-agent-linux-${ARCH}"
 CHECKSUM_URL="https://${GITEA_HOST}/mrhid6/vantage/releases/download/${LATEST_ENCODED}/checksums.txt"
 

server/internal/bus/bus.go

@@ -92,9 +92,23 @@ const (
 	// ProxyEndChannel carries a console relay's terminal reason back to the pod
 	// serving the WebSocket, which is the pod that has to write the audit event.
 	ProxyEndChannel = prefix + "proxyend:"
+	// ProxyAddrChannel carries the address of a console relay listener back to
+	// the pod serving the WebSocket.
+	//
+	// The listener cannot be bound in advance on any particular pod. An agent's
+	// ProxyStream is a separate HTTP/2 request from its CommandStream, and an
+	// L7 proxy (Traefik) balances requests, not connections — so it may land on
+	// any replica, not the one holding the command stream. The pod it does land
+	// on binds the listener and announces it here.
+	ProxyAddrChannel = prefix + "proxyaddr:"
 
 	// PresenceKey records which node holds an agent's command stream.
 	PresenceKey = prefix + "agent:"
+	// ProxyPendingKey authorises one not-yet-opened ProxyStream. It is the only
+	// state tying a proxy_id to the instance and server it was minted for, and
+	// it must be visible to every replica because any of them may receive the
+	// stream.
+	ProxyPendingKey = prefix + "proxypending:"
 	// leaderKey records the holder of a named singleton job.
 	leaderKey = prefix + "leader:"
 )
@@ -190,6 +204,21 @@ func SetPresence(ctx context.Context, serverID string, ttl time.Duration) error
 	return rdb.Set(ctx, PresenceKey+serverID, nodeID, ttl).Err()
 }
 
+// RenewPresence extends serverID's claim, but only while this node still holds
+// it, and reports whether it did.
+//
+// A blind SET here is wrong, not merely untidy. When an agent reconnects, its
+// previous stream can stay half-open on another pod for the length of a
+// keepalive cycle, and that pod goes on renewing. Two processes then overwrite
+// each other's claim every renewal interval and the key names whichever wrote
+// last rather than whichever holds the live stream. A superseded pod must lose
+// quietly instead.
+func RenewPresence(ctx context.Context, serverID string, ttl time.Duration) bool {
+	n, err := renewPresenceIfOwner.Run(ctx, rdb,
+		[]string{PresenceKey + serverID}, nodeID, int64(ttl/time.Millisecond)).Int64()
+	return err == nil && n == 1
+}
+
 // ClearPresence releases serverID, but only if this node still holds it. A
 // blind DEL would let a pod whose stream had already been re-established
 // elsewhere delete the new owner's claim on its way out.
@@ -212,6 +241,61 @@ func IsConnected(ctx context.Context, serverID string) bool {
 	return err == nil && n > 0
 }
 
+// SetPendingProxy records that proxyID has been minted for instanceID and
+// serverID, for ttl. Written before the OpenProxyCmd is dispatched, so it is in
+// place before any agent can act on it.
+func SetPendingProxy(ctx context.Context, proxyID, instanceID, serverID string, ttl time.Duration) error {
+	b, err := json.Marshal(map[string]string{"instance_id": instanceID, "server_id": serverID})
+	if err != nil {
+		return err
+	}
+	return rdb.Set(ctx, ProxyPendingKey+proxyID, b, ttl).Err()
+}
+
+// ClaimPendingProxy consumes proxyID's pending record and returns the instance
+// and server it was minted for. Get and delete are one Lua call rather than two
+// round trips: single use is the whole security property, and two agents
+// racing the same proxy_id must not both be served.
+//
+// A missing record is reported as "", "" rather than an error — an unknown
+// proxy_id, an expired one and a second claim are all the same refusal.
+func ClaimPendingProxy(ctx context.Context, proxyID string) (instanceID, serverID string) {
+	v, err := claimPending.Run(ctx, rdb, []string{ProxyPendingKey + proxyID}).Text()
+	if err != nil || v == "" {
+		return "", ""
+	}
+	var rec struct {
+		InstanceID string `json:"instance_id"`
+		ServerID   string `json:"server_id"`
+	}
+	if err := json.Unmarshal([]byte(v), &rec); err != nil {
+		return "", ""
+	}
+	return rec.InstanceID, rec.ServerID
+}
+
+// ClearPendingProxy drops a pending record whose command never reached an
+// agent, so a dead proxy_id is not left claimable for the rest of its TTL.
+func ClearPendingProxy(ctx context.Context, proxyID string) {
+	_ = rdb.Del(ctx, ProxyPendingKey+proxyID).Err()
+}
+
+var claimPending = redis.NewScript(`
+local v = redis.call("GET", KEYS[1])
+if v then
+	redis.call("DEL", KEYS[1])
+	return v
+end
+return ""
+`)
+
+var renewPresenceIfOwner = redis.NewScript(`
+if redis.call("GET", KEYS[1]) == ARGV[1] then
+	return redis.call("PEXPIRE", KEYS[1], ARGV[2])
+end
+return 0
+`)
+
 var releaseIfOwner = redis.NewScript(`
 if redis.call("GET", KEYS[1]) == ARGV[1] then
 	return redis.call("DEL", KEYS[1])

server/internal/grpc/pb/vantage.pb.go

@@ -158,8 +158,14 @@ type ServerCommand struct {
 	RunStep          *RunStepCmd          `json:"run_step,omitempty"`
 	CleanupWorkspace *CleanupWorkspaceCmd `json:"cleanup_workspace,omitempty"`
 	OpenProxy        *OpenProxyCmd        `json:"open_proxy,omitempty"`
+	Ping             *PingCmd             `json:"ping,omitempty"`
 }
 
+// PingCmd is a server-originated liveness beat. It carries nothing and expects
+// no reply: its arrival is the entire message. See the .proto for why gRPC
+// keepalive is not sufficient on its own.
+type PingCmd struct{}
+
 type CleanupWorkspaceCmd struct {
 	WorkspaceId string `json:"workspace_id"`
 }

server/internal/grpc/proxystream.go

@@ -31,7 +31,7 @@ func (s *vantageServer) ProxyStream(stream pb.Vantage_ProxyStreamServer) error {
 		return status.Error(codes.PermissionDenied, "proxy session unavailable")
 	}
 
-	if err := serveProxy(proxy.Default, msg.Open, srv.InstanceID, stream); err != nil {
+	if err := serveProxy(msg.Open, srv.InstanceID, stream); err != nil {
 		// The reason is deliberately not returned to the agent: an unknown and a
 		// foreign proxy_id must be indistinguishable.
 		log.Printf("proxy %s (server %s): %v", msg.Open.ProxyId, msg.Open.ServerId, err)
@@ -40,12 +40,17 @@ func (s *vantageServer) ProxyStream(stream pb.Vantage_ProxyStreamServer) error {
 	return nil
 }
 
-// serveProxy claims the pending session and relays it. Split out from the gRPC
-// method so the authorisation matrix is testable without a real stream.
-func serveProxy(reg *proxy.Registry, open *pb.ProxyOpen, instanceID string, stream proxy.AgentStream) error {
-	entry, err := reg.Claim(instanceID, open.ServerId, open.ProxyId)
+// serveProxy claims the pending session, binds this pod's relay listener for
+// it, and relays. Split out from the gRPC method so the authorisation matrix is
+// testable without a real stream.
+//
+// The listener is bound here, on whichever replica the stream reached, rather
+// than in advance on the pod holding the agent's command stream — those are not
+// the same pod, because an L7 proxy balances HTTP/2 requests independently.
+func serveProxy(open *pb.ProxyOpen, instanceID string, stream proxy.AgentStream) error {
+	sess, err := services.ClaimProxyStream(instanceID, open.ServerId, open.ProxyId)
 	if err != nil {
 		return err
 	}
-	return entry.Session.Serve(stream)
+	return sess.Serve(stream)
 }

server/internal/grpc/server.go

@@ -212,11 +212,46 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
 		}
 	}()
 
+	// The heartbeat is what lets the agent tell a live stream from an orphaned
+	// one. gRPC keepalive cannot: behind an L7 proxy the agent's connection
+	// terminates at the proxy, which answers pings on its own behalf, so a dead
+	// pod leaves the agent blocked in Recv forever with commands vanishing into
+	// a stream nobody is serving. A message that originates here is the only
+	// thing that proves this process is still on the other end.
+	ping := time.NewTicker(pingInterval)
+	defer ping.Stop()
+
+	// Beats are counted and reported periodically rather than logged one by
+	// one: at one every 20s per agent, a fleet of any size would drown every
+	// other line in the log. What is worth a line of its own is the first beat
+	// (it tells the operator this stream's watchdog is now armed on the agent
+	// side) and any failure to send one.
+	var beats int
+	summary := time.NewTicker(pingSummaryInterval)
+	defer summary.Stop()
+
 	ctx := stream.Context()
 	for {
 		select {
 		case <-ctx.Done():
 			return nil
+		case <-summary.C:
+			log.Printf("agent %s command stream healthy, %d beats in the last %s",
+				srv.ServerID, beats, pingSummaryInterval)
+			beats = 0
+		case <-ping.C:
+			// A failed send is the point: it is how this side learns the stream
+			// is gone, which runs the deferred release and frees the agent's
+			// presence claim for whichever pod it reconnects to.
+			if err := stream.Send(&pb.ServerCommand{Ping: &pb.PingCmd{}}); err != nil {
+				log.Printf("agent %s command stream beat failed after %d beats: %v",
+					srv.ServerID, beats, err)
+				return err
+			}
+			beats++
+			if beats == 1 {
+				log.Printf("agent %s command stream beating every %s", srv.ServerID, pingInterval)
+			}
 		case cmd, ok := <-ch:
 			if !ok {
 				return nil
@@ -228,10 +263,29 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
 	}
 }
 
-func StartGRPC(port int) error {
+// How often the server beats on an idle command stream. Comfortably under the
+// agent's staleness threshold, so a single dropped beat does not cost a
+// reconnect.
+const pingInterval = 20 * time.Second
+
+// How often an otherwise silent healthy stream says so. Long enough that a
+// large fleet does not fill the log, short enough that "this pod is still
+// serving that agent" is answerable from the log rather than by inference.
+const pingSummaryInterval = 5 * time.Minute
+
+// StartGRPC serves the agent API until stop is called.
+//
+// It returns a stop function rather than serving forever because an abrupt exit
+// is not a neutral act here: every CommandStream handler holds an agent's
+// presence claim, released by a deferred call that a killed process never runs.
+// The claim then outlives its owner for the remainder of its 30s TTL, during
+// which dispatch believes the agent is reachable, publishes to a channel with
+// no subscriber, and fails as "agent offline" — a pod that has already exited
+// still answering for an agent it can no longer reach.
+func StartGRPC(port int) (stop func(), err error) {
 	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
 	if err != nil {
-		return fmt.Errorf("failed to listen: %w", err)
+		return nil, fmt.Errorf("failed to listen: %w", err)
 	}
 
 	s := grpc.NewServer(
@@ -248,6 +302,39 @@ func StartGRPC(port int) error {
 	)
 	pb.RegisterVantageServer(s, &vantageServer{})
 
-	log.Printf("gRPC server listening on :%d", port)
-	return s.Serve(lis)
+	go func() {
+		log.Printf("gRPC server listening on :%d", port)
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("gRPC server error: %v", err)
+		}
+	}()
+
+	// GracefulStop sends GOAWAY and waits for the handlers to return, which is
+	// what runs those deferred releases and, on the agent's side, ends the
+	// stream with a clean error it reconnects from immediately rather than
+	// waiting out a TCP timeout.
+	//
+	// It is bounded: an idle CommandStream returns as soon as its context is
+	// cancelled, but a console relay mid-transfer would otherwise hold the
+	// process past the pod's grace period and earn a SIGKILL — which is the
+	// abrupt exit this exists to avoid.
+	return func() {
+		done := make(chan struct{})
+		go func() {
+			s.GracefulStop()
+			close(done)
+		}()
+		select {
+		case <-done:
+			log.Println("gRPC server stopped gracefully")
+		case <-time.After(grpcStopTimeout):
+			log.Printf("gRPC server did not stop within %s, forcing", grpcStopTimeout)
+			s.Stop()
+		}
+	}, nil
 }
+
+// How long GracefulStop is given before outstanding streams are cut. Comfortably
+// inside the chart's termination grace period, so the forced stop below still
+// leaves time for the HTTP server to drain.
+const grpcStopTimeout = 10 * time.Second

server/internal/proxy/registry.go

@@ -7,7 +7,6 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"errors"
-	"sync"
 )
 
 var (
@@ -25,56 +24,14 @@ func NewID() (string, error) {
 	return hex.EncodeToString(b), nil
 }
 
-type Entry struct {
-	ProxyID    string
-	InstanceID string
-	ServerID   string
-	Session    *Session
-}
-
-type Registry struct {
-	mu      sync.Mutex
-	entries map[string]*Entry
-}
-
-func NewRegistry() *Registry {
-	return &Registry{entries: make(map[string]*Entry)}
-}
-
-var Default = NewRegistry()
-
-func (r *Registry) Add(e *Entry) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.entries[e.ProxyID] = e
-}
-
-// Claim removes and returns the entry. It is single-use: a second claim on the
-// same proxy_id gets ErrNotFound. A claim whose instance or server does not
-// match leaves the entry in place and gets ErrForbidden.
-func (r *Registry) Claim(instanceID, serverID, proxyID string) (*Entry, error) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	e, ok := r.entries[proxyID]
-	if !ok {
-		return nil, ErrNotFound
-	}
-	if e.InstanceID != instanceID || e.ServerID != serverID {
-		return nil, ErrForbidden
-	}
-	delete(r.entries, proxyID)
-	return e, nil
-}
-
-func (r *Registry) Remove(proxyID string) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	delete(r.entries, proxyID)
-}
-
-func (r *Registry) Len() int {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	return len(r.entries)
-}
+// There is deliberately no in-process registry of pending sessions here any
+// more. One existed, keyed by proxy_id, on the assumption that the pod which
+// bound a listener was the pod that would receive the matching ProxyStream.
+// That assumption holds only for a single replica: a ProxyStream is its own
+// HTTP/2 request and an L7 proxy routes it independently of the agent's
+// command stream, so with N replicas the lookup missed (N-1)/N of the time and
+// the console failed with "proxy session not found".
+//
+// The pending record lives in Redis instead (bus.SetPendingProxy /
+// ClaimPendingProxy), and the listener is bound by whichever pod the stream
+// actually reaches — see services.ClaimProxyStream.

server/internal/services/consoleproxy.go

@@ -21,18 +21,30 @@ import (
 // so this is fatal rather than a degraded mode.
 var ErrAgentOffline = errors.New("agent is not connected")
 
-// A console session spans two processes once there is more than one replica.
+// A console session spans up to three processes once there is more than one
+// replica, and no two of them can be assumed to be the same one:
 //
-// The browser's WebSocket lands on an arbitrary pod. The agent's ProxyStream
-// lands on the pod holding that agent's command stream. The relay listener has
-// to be on the latter — that is the only process that can match an incoming
-// ProxyStream to a waiting listener — while guacd is dialled from the former.
+//	the browser's WebSocket   lands on an arbitrary pod
+//	the agent's CommandStream lands on the pod holding presence for it
+//	the agent's ProxyStream   lands on an arbitrary pod
 //
-// So the WebSocket's pod asks, over the bus, for a relay to be bound on the
-// agent's pod, and gets back an address to hand to guacd. That address is the
-// owner pod's own, which is why it must resolve to a single pod (POD_IP under
-// Kubernetes) rather than to the Service, which would send guacd to a pod
-// holding no listener roughly (n-1)/n of the time.
+// That third line is the one that is easy to get wrong. A ProxyStream is a
+// separate HTTP/2 request, and an L7 proxy (Traefik, which the chart's gRPC
+// ingress uses) balances requests rather than connections — so it does not
+// follow the command stream. Binding the relay listener on the command
+// stream's pod therefore fails roughly (n-1)/n of the time with "proxy session
+// not found": the stream arrives at a pod whose registry is empty.
+//
+// So the listener is bound by whichever pod receives the ProxyStream, at the
+// moment it receives it, and that pod announces its own address on
+// ProxyAddrChannel. The WebSocket's pod subscribes before dispatching and
+// hands the announced address to guacd. The address is the announcing pod's
+// own, which is why it must resolve to a single pod (POD_IP under Kubernetes)
+// rather than to the Service.
+//
+// Authorisation cannot live in that pod's memory either, so a pending record
+// in Redis (bus.SetPendingProxy) carries the instance and server a proxy_id
+// was minted for, and is consumed atomically on first claim.
 //
 // Teardown needs no message of its own. When the browser goes away guac closes
 // its connection to the relay, the relay sees the read end, and the session
@@ -46,6 +58,15 @@ var ErrAgentOffline = errors.New("agent is not connected")
 // waited for longer.
 const proxyEndGrace = 2 * time.Second
 
+// How long a minted proxy_id stays claimable, and how long the WebSocket's pod
+// waits for the relay's address to be announced. The TTL is the longer of the
+// two on purpose: a record that expired while its own opener was still waiting
+// would turn a slow agent into an unexplained refusal.
+const (
+	proxyPendingTTL = 30 * time.Second
+	proxyAddrWait   = 15 * time.Second
+)
+
 // ConsoleProxy is a relay as seen by the pod serving the WebSocket.
 type ConsoleProxy struct {
 	ProxyID string
@@ -114,51 +135,63 @@ func guacdHosts(addr string) []string {
 	return ips
 }
 
-// localRelay is a listener bound by this process on behalf of a remote request.
-type localRelay struct {
-	proxyID string
-	host    string
-	port    int
-	session *proxy.Session
+// proxyAddr is what a relay's binding pod announces: the address guacd should
+// dial to reach the listener it has just bound.
+type proxyAddr struct {
+	Host string `json:"host"`
+	Port int    `json:"port"`
 }
 
-// openLocalRelay binds a listener here and registers it, so the agent's
-// ProxyStream — which will arrive at this process — can be matched to it.
-// Called on the owner pod, from the dispatch handler.
-func openLocalRelay(instanceID, serverID, proxyID string) (*localRelay, error) {
+// ClaimProxyStream authorises an incoming ProxyStream, binds a relay listener
+// for it on this pod, and announces the address to whichever pod is serving the
+// browser's WebSocket. It is called from the gRPC handler, on whichever replica
+// the stream happened to reach.
+//
+// instanceID and serverID are the *authenticated* identity of the calling
+// agent; they must match the pending record or the claim is refused, so an
+// agent cannot relay a console session minted for another server.
+func ClaimProxyStream(instanceID, serverID, proxyID string) (*proxy.Session, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+	defer cancel()
+
+	wantInstance, wantServer := bus.ClaimPendingProxy(ctx, proxyID)
+	if wantInstance == "" {
+		return nil, proxy.ErrNotFound
+	}
+	if wantInstance != instanceID || wantServer != serverID {
+		return nil, proxy.ErrForbidden
+	}
+
 	sess, err := proxy.NewSession(proxyListenHost(), guacdHosts(guacdAddr()))
 	if err != nil {
 		return nil, err
 	}
 
 	sess.OnEnd(func(reason string) {
-		ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
-		defer cancel()
-		if _, err := bus.Publish(ctx, bus.ProxyEndChannel+proxyID, proxyEnd{Reason: reason}); err != nil {
+		endCtx, endCancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+		defer endCancel()
+		if _, err := bus.Publish(endCtx, bus.ProxyEndChannel+proxyID, proxyEnd{Reason: reason}); err != nil {
 			log.Printf("proxy: publish end for %s: %v", proxyID, err)
 		}
 	})
 
-	proxy.Default.Add(&proxy.Entry{
-		ProxyID:    proxyID,
-		InstanceID: instanceID,
-		ServerID:   serverID,
-		Session:    sess,
-	})
+	addr := proxyAddr{Host: proxyAdvertiseHost(), Port: sess.Port()}
 
-	return &localRelay{
-		proxyID: proxyID,
-		host:    proxyAdvertiseHost(),
-		port:    sess.Port(),
-		session: sess,
-	}, nil
-}
+	// Logged on the success path, not just on failure. The address guacd is
+	// about to be sent to is chosen per pod (POD_IP), so when a console fails
+	// for some replicas and not others this line is the difference between
+	// seeing which one answered and inferring it from silence.
+	log.Printf("proxy %s (server %s): relay bound on %s:%d, node %s",
+		proxyID, serverID, addr.Host, addr.Port, bus.NodeID())
 
-// abandon tears down a relay that was bound but whose command never reached the
-// agent, so the listener does not sit out its rendezvous timeout for nothing.
-func (r *localRelay) abandon() {
-	proxy.Default.Remove(r.proxyID)
-	r.session.Close("dispatch_failed")
+	if _, err := bus.Publish(ctx, bus.ProxyAddrChannel+proxyID, addr); err != nil {
+		// Nobody will ever dial this listener, so it is closed now rather than
+		// left to sit out its rendezvous timeout.
+		sess.Close("announce_failed")
+		return nil, fmt.Errorf("announce relay address: %w", err)
+	}
+
+	return sess, nil
 }
 
 // OpenConsoleProxy asks the pod holding serverID's stream to bind a relay and
@@ -173,44 +206,85 @@ func OpenConsoleProxy(instanceID, serverID string, targetPort int) (*ConsoleProx
 		return nil, fmt.Errorf("generate proxy id: %w", err)
 	}
 
-	// Subscribed before the relay is asked for: a relay that fails immediately
-	// (the agent never claims it, the dial is refused) publishes its reason at
-	// once, and that reason is the whole content of the audit event.
+	// Both subscriptions are established before the command is dispatched: a
+	// fast agent binds and announces its relay within milliseconds, and a relay
+	// that fails immediately publishes its reason just as quickly. Either
+	// arriving before the subscriber is in place would be lost.
 	cp := &ConsoleProxy{ProxyID: proxyID, serverID: serverID, ended: make(chan struct{})}
 	endCtx, endCancel := context.WithCancel(context.Background())
-	ends, unsub, err := bus.Subscribe(endCtx, bus.ProxyEndChannel+proxyID)
+	ends, unsubEnd, err := bus.Subscribe(endCtx, bus.ProxyEndChannel+proxyID)
 	if err != nil {
 		endCancel()
 		return nil, fmt.Errorf("subscribe relay end: %w", err)
 	}
+	addrs, unsubAddr, err := bus.Subscribe(endCtx, bus.ProxyAddrChannel+proxyID)
+	if err != nil {
+		endCancel()
+		unsubEnd()
+		return nil, fmt.Errorf("subscribe relay address: %w", err)
+	}
 	cp.stop = func() {
 		endCancel()
-		unsub()
+		unsubAddr()
+		unsubEnd()
 	}
 	go cp.watchEnd(ends)
 
-	ack, err := Dispatcher.send(CommandEnvelope{
+	// The pending record authorises the ProxyStream the agent is about to open,
+	// and is written before the command so it cannot lose the race with it.
+	pendCtx, pendCancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+	if err := bus.SetPendingProxy(pendCtx, proxyID, instanceID, serverID, proxyPendingTTL); err != nil {
+		pendCancel()
+		cp.stop()
+		return nil, fmt.Errorf("register pending relay: %w", err)
+	}
+	pendCancel()
+
+	abandon := func() {
+		ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+		bus.ClearPendingProxy(ctx, proxyID)
+		cancel()
+		cp.stop()
+	}
+
+	if _, err := Dispatcher.send(CommandEnvelope{
 		ServerID: serverID,
 		Command: &pb.ServerCommand{
 			CommandId: proxyID,
 			OpenProxy: &pb.OpenProxyCmd{ProxyId: proxyID, Port: uint32(targetPort)},
 		},
-		Proxy: &ProxyRelayRequest{InstanceID: instanceID, ProxyID: proxyID},
-	})
-	if err != nil {
-		cp.stop()
+	}); err != nil {
+		abandon()
 		if errors.Is(err, ErrAgentNotConnected) {
 			return nil, ErrAgentOffline
 		}
 		return nil, err
 	}
-	if ack.ProxyHost == "" || ack.ProxyPort == 0 {
-		cp.stop()
+
+	// The command has reached the agent; the relay's address arrives only once
+	// the agent has actually opened its ProxyStream somewhere in the fleet.
+	var addr proxyAddr
+	select {
+	case b, ok := <-addrs:
+		if !ok {
+			abandon()
+			return nil, fmt.Errorf("relay address subscription closed")
+		}
+		if err := json.Unmarshal(b, &addr); err != nil {
+			abandon()
+			return nil, fmt.Errorf("undecodable relay address: %w", err)
+		}
+	case <-time.After(proxyAddrWait):
+		abandon()
+		return nil, fmt.Errorf("agent did not open a relay for %s", serverID)
+	}
+	if addr.Host == "" || addr.Port == 0 {
+		abandon()
 		return nil, fmt.Errorf("relay opened without an address")
 	}
 
-	cp.Host = ack.ProxyHost
-	cp.Port = ack.ProxyPort
+	cp.Host = addr.Host
+	cp.Port = addr.Port
 	return cp, nil
 }
 

server/internal/services/dispatch.go

@@ -38,11 +38,19 @@ const (
 // CommandEnvelope is what actually crosses the bus. It is the command plus the
 // small amount of context the owning pod needs to act on it locally.
 type CommandEnvelope struct {
-	ServerID string             `json:"server_id"`
-	Command  *pb.ServerCommand  `json:"command"`
-	ReplyTo  string             `json:"reply_to"`
-	Log      *LogRequest        `json:"log,omitempty"`
-	Proxy    *ProxyRelayRequest `json:"proxy,omitempty"`
+	ServerID string            `json:"server_id"`
+	Command  *pb.ServerCommand `json:"command"`
+	ReplyTo  string            `json:"reply_to"`
+	Log      *LogRequest       `json:"log,omitempty"`
+
+	// Node names the pod this envelope is for: the presence holder at the time
+	// it was published. The command channel is a fan-out, so during a reconnect
+	// two pods can be subscribed for one agent — the pod with the live stream,
+	// and a pod whose stream is half-open and has not yet noticed. Both would
+	// receive the envelope, and the first to ack wins the request. If that is
+	// the stale one, the command is queued onto a dead stream and acked OK: the
+	// operator is told it worked and the agent never sees it.
+	Node string `json:"node,omitempty"`
 }
 
 // LogRequest asks the owner pod to open a step log before it dispatches.
@@ -56,25 +64,17 @@ type LogRequest struct {
 	Mask     []string `json:"mask,omitempty"`
 }
 
-// ProxyRelayRequest asks the owner pod to bind a console relay listener and
-// register it before dispatching OpenProxyCmd.
+// CommandAck is the owner pod's answer. It reports only that the command
+// reached the agent's stream.
 //
-// The listener has to live on the owner pod: the agent's ProxyStream arrives
-// there, and only there can it be matched to a waiting listener. The pod
-// serving the browser's WebSocket learns the address from the ack and hands
-// that to guacd.
-type ProxyRelayRequest struct {
-	InstanceID string `json:"instance_id"`
-	ProxyID    string `json:"proxy_id"`
-}
-
-// CommandAck is the owner pod's answer.
+// A console relay listener used to be bound here and its address returned in
+// this ack. It no longer is: the agent's ProxyStream does not necessarily
+// arrive at the pod holding its command stream, so the listener is bound by
+// whichever pod receives that stream and announced on bus.ProxyAddrChannel.
 type CommandAck struct {
-	OK        bool   `json:"ok"`
-	Error     string `json:"error,omitempty"`
-	Node      string `json:"node,omitempty"`
-	ProxyHost string `json:"proxy_host,omitempty"`
-	ProxyPort int    `json:"proxy_port,omitempty"`
+	OK    bool   `json:"ok"`
+	Error string `json:"error,omitempty"`
+	Node  string `json:"node,omitempty"`
 }
 
 type commandDispatcher struct{}
@@ -117,8 +117,13 @@ func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan
 			case <-runCtx.Done():
 				return
 			case <-t.C:
-				if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil {
-					log.Printf("dispatch: renew presence for %s: %v", serverID, err)
+				// Renew only while this pod still holds the claim. Losing it
+				// means a newer stream for the same agent was established
+				// elsewhere, and this one is a half-open leftover: it must stop
+				// renewing rather than overwrite the live owner every 10s.
+				if !bus.RenewPresence(runCtx, serverID, presenceTTL) {
+					log.Printf("dispatch: presence for %s is held elsewhere, stopping renewal", serverID)
+					return
 				}
 			}
 		}
@@ -161,6 +166,15 @@ func (d *commandDispatcher) handleEnvelope(ctx context.Context, raw []byte, out
 		return
 	}
 
+	// Not addressed to this pod: stay silent rather than ack. Answering would
+	// win the race against the pod that actually holds the agent's stream, and
+	// the caller would be told a command succeeded that was queued onto a
+	// stream nobody is reading. Silence lets the real owner answer, or lets the
+	// request time out as ErrNoResponder, which fails loudly and correctly.
+	if env.Node != "" && env.Node != bus.NodeID() {
+		return
+	}
+
 	ack := CommandAck{OK: true, Node: bus.NodeID()}
 
 	if env.Log != nil {
@@ -169,27 +183,10 @@ func (d *commandDispatcher) handleEnvelope(ctx context.Context, raw []byte, out
 		}
 	}
 
-	var relay *localRelay
-	if env.Proxy != nil {
-		r, err := openLocalRelay(env.Proxy.InstanceID, env.ServerID, env.Proxy.ProxyID)
-		if err != nil {
-			ack = CommandAck{OK: false, Error: err.Error(), Node: bus.NodeID()}
-		} else {
-			relay = r
-			ack.ProxyHost = r.host
-			ack.ProxyPort = r.port
-		}
-	}
-
-	if ack.OK {
-		select {
-		case out <- env.Command:
-		default:
-			ack = CommandAck{OK: false, Error: "command queue full", Node: bus.NodeID()}
-			if relay != nil {
-				relay.abandon()
-			}
-		}
+	select {
+	case out <- env.Command:
+	default:
+		ack = CommandAck{OK: false, Error: "command queue full", Node: bus.NodeID()}
 	}
 
 	if err := bus.Reply(ctx, env.ReplyTo, ack); err != nil {
@@ -221,6 +218,15 @@ func (d *commandDispatcher) send(env CommandEnvelope) (CommandAck, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
 	defer cancel()
 
+	// Resolved once, here, and carried in the envelope. Reading it at publish
+	// time rather than letting subscribers self-select is what makes a stale
+	// subscriber harmless: it will see an envelope addressed elsewhere and
+	// ignore it.
+	env.Node = bus.PresenceHolder(ctx, env.ServerID)
+	if env.Node == "" {
+		return CommandAck{}, fmt.Errorf("%w: %s", ErrAgentNotConnected, env.ServerID)
+	}
+
 	raw, err := bus.Request(ctx, bus.CommandChannel+env.ServerID, env.ReplyTo, env, dispatchAckTimeout)
 	if err != nil {
 		if errors.Is(err, bus.ErrNoResponder) {

web/app/(app)/servers/[id]/console/page.tsx

@@ -6,7 +6,27 @@ import { useParams, useRouter, useSearchParams } from "next/navigation";
 import Link from "next/link";
 import { api } from "@/lib/api";
 import { Button, Card } from "@/components/ui";
-import { openConsole } from "@/lib/guacConsole";
+import { openConsole, type ConsoleFailure, type ConsoleState } from "@/lib/guacConsole";
+
+// The session's own lifecycle, which is not the same as the tunnel's: "idle"
+// means the form is showing, and everything else means a session has been
+// started and the viewport owns the page.
+type SessionPhase = "idle" | ConsoleState;
+
+const PHASE_LABEL: Record<Exclude<SessionPhase, "idle">, string> = {
+    connecting: "Connecting",
+    connected: "Connected",
+    disconnected: "Disconnected",
+    error: "Failed",
+};
+
+// Shape as well as colour: state must never read by colour alone.
+const PHASE_DOT: Record<Exclude<SessionPhase, "idle">, string> = {
+    connecting: "bg-warning animate-pulse",
+    connected: "bg-success",
+    disconnected: "bg-text-tertiary",
+    error: "bg-danger",
+};
 
 export default function ServerConsolePage() {
     const params = useParams();
@@ -29,8 +49,10 @@ export default function ServerConsolePage() {
     const [rdpPassword, setRdpPassword] = useState("");
     const [vncPassword, setVncPassword] = useState("");
     const [connecting, setConnecting] = useState(false);
-    const [connected, setConnected] = useState(false);
+    const [phase, setPhase] = useState<SessionPhase>("idle");
     const [error, setError] = useState<string | null>(null);
+    const [failure, setFailure] = useState<ConsoleFailure | null>(null);
+    const connected = phase !== "idle";
     const [pending, setPending] = useState<{ token: string; wsPath: string } | null>(null);
     const [zoom, setZoom] = useState(1);
     const dprRef = useRef(1);
@@ -74,6 +96,7 @@ export default function ServerConsolePage() {
 
     async function handleConnect() {
         setError(null);
+        setFailure(null);
         setConnecting(true);
         try {
             const body: Parameters<typeof api.connectConsole>[0] = {
@@ -93,10 +116,18 @@ export default function ServerConsolePage() {
             const { token, ws_path } = await api.connectConsole(body);
             
             
+            // "connecting", not "connected": all we have so far is a token. The
+            // real state now comes from the tunnel, which is the only thing that
+            // knows whether the far end ever answered.
             setPending({ token, wsPath: ws_path });
-            setConnected(true);
+            setPhase("connecting");
         } catch (e) {
-            setError(e instanceof Error ? e.message : "Failed to connect");
+            const message = e instanceof Error ? e.message : "Failed to connect";
+            setError(
+                message.includes("agent_offline")
+                    ? "The agent on this server is not connected, so a console session cannot be opened."
+                    : message
+            );
         } finally {
             setConnecting(false);
         }
@@ -121,7 +152,10 @@ export default function ServerConsolePage() {
             `&height=${Math.floor(rect.height * dpr)}` +
             `&dpi=96`;
 
-        connectionRef.current = openConsole(containerRef.current, wsUrl, connectData);
+        connectionRef.current = openConsole(containerRef.current, wsUrl, connectData, {
+            onState: (s) => setPhase(s),
+            onFailure: (f) => setFailure(f),
+        });
         connectionRef.current.setScale(zoom / dpr);
         setPending(null);
     }, [connected, pending]);
@@ -140,15 +174,23 @@ export default function ServerConsolePage() {
         connectionRef.current.setScale(zoom / dpr);
     }, [zoom]);
 
+    // Returns to the connection form. Used both by the Disconnect button and by
+    // Reconnect, which is the same teardown followed by a fresh dial.
     function handleDisconnect() {
         connectionRef.current?.disconnect();
         connectionRef.current = null;
-        setConnected(false);
+        setPhase("idle");
+        setFailure(null);
         if (containerRef.current) {
             containerRef.current.innerHTML = "";
         }
     }
 
+    function handleReconnect() {
+        handleDisconnect();
+        void handleConnect();
+    }
+
     if (serverLoading || keysLoading) {
         return (
             <div className="flex h-full items-center justify-center">
@@ -269,9 +311,18 @@ export default function ServerConsolePage() {
                 </Card>
             ) : (
                 <div className="mb-4 flex flex-wrap items-center gap-3">
+                    <span className="inline-flex items-center gap-2 rounded-full border border-border bg-surface-2 px-3 py-1.5 text-xs font-medium text-text-secondary">
+                        <span className={`h-2 w-2 rounded-full ${PHASE_DOT[phase as Exclude<SessionPhase, "idle">]}`} />
+                        {PHASE_LABEL[phase as Exclude<SessionPhase, "idle">]}
+                    </span>
                     <Button variant="danger" onClick={handleDisconnect}>
                         Disconnect
                     </Button>
+                    {(phase === "error" || phase === "disconnected") && (
+                        <Button variant="secondary" onClick={handleReconnect}>
+                            Reconnect
+                        </Button>
+                    )}
                     <label className="text-sm text-text-secondary">Scale</label>
                     <select
                         value={zoom}
@@ -292,10 +343,69 @@ export default function ServerConsolePage() {
                 </div>
             )}
 
-            <div
-                ref={containerRef}
-                className="min-h-[500px] flex-1 overflow-hidden rounded-lg border border-border bg-black"
-            />
+            {/* The viewport is always mounted — Guacamole attaches its display
+                element to it on connect, so it cannot be conditionally rendered.
+                Anything the operator needs to be told is layered over it
+                instead, which is what a bare black rectangle never did. */}
+            <div className="relative min-h-[500px] flex-1">
+                <div
+                    ref={containerRef}
+                    className="absolute inset-0 overflow-hidden rounded-lg border border-border bg-well"
+                />
+
+                {phase === "connecting" && (
+                    <div className="pointer-events-none absolute inset-0 flex flex-col items-center justify-center gap-3 rounded-lg bg-ground/70">
+                        <div className="h-8 w-8 animate-spin rounded-full border-2 border-border border-t-accent" />
+                        <p className="text-sm text-text-secondary">
+                            Opening {protocol.toUpperCase()} session on {server.hostname}…
+                        </p>
+                    </div>
+                )}
+
+                {(phase === "error" || phase === "disconnected") && (
+                    <div className="absolute inset-0 flex items-center justify-center rounded-lg bg-ground/80 p-6">
+                        <Card className="max-w-md">
+                            <div className="space-y-3">
+                                <div className="flex items-center gap-2">
+                                    <span
+                                        className={`h-2 w-2 rounded-full ${
+                                            phase === "error" ? "bg-danger" : "bg-text-tertiary"
+                                        }`}
+                                    />
+                                    <h2 className="text-sm font-semibold text-text-primary">
+                                        {phase === "error" ? "Console session failed" : "Console session ended"}
+                                    </h2>
+                                </div>
+
+                                <p className="text-sm text-text-secondary">
+                                    {failure?.message ??
+                                        (phase === "error"
+                                            ? "The session ended without reporting a reason."
+                                            : "The remote host closed the session.")}
+                                </p>
+
+                                {/* The numeric status is what makes a support
+                                    ticket actionable, so it is shown rather than
+                                    folded into the sentence above. */}
+                                {failure?.code !== undefined && (
+                                    <p className="font-mono text-xs text-text-tertiary">
+                                        {failure.source === "tunnel" ? "tunnel" : "session"} status {failure.code}
+                                    </p>
+                                )}
+
+                                <div className="flex gap-2 pt-1">
+                                    <Button variant="primary" onClick={handleReconnect}>
+                                        Reconnect
+                                    </Button>
+                                    <Button variant="secondary" onClick={handleDisconnect}>
+                                        Change settings
+                                    </Button>
+                                </div>
+                            </div>
+                        </Card>
+                    </div>
+                )}
+            </div>
         </div>
     );
 }

web/lib/guacConsole.ts

@@ -1,37 +1,135 @@
-
-
 declare const Guacamole: any;
 
+/**
+ * The lifecycle of a console session as the UI needs to talk about it.
+ *
+ * Deliberately not Guacamole's own state enum: "waiting" and "connecting" are
+ * one thing to an operator, and "disconnected because you clicked Disconnect"
+ * and "disconnected because the far end vanished" are two things that Guacamole
+ * reports identically.
+ */
+export type ConsoleState =
+  | "connecting"
+  | "connected"
+  | "disconnected"
+  | "error";
+
+export type ConsoleFailure = {
+  /** Guacamole status code, when the failure came with one. */
+  code?: number;
+  /** Operator-facing sentence. Always set. */
+  message: string;
+  /** Where the failure was reported: the tunnel or the client session. */
+  source: "tunnel" | "client";
+};
+
+/**
+ * Guacamole status codes, as operator-facing sentences.
+ *
+ * These are the only diagnosis anyone gets from a failed console: the relay,
+ * guacd and the target daemon are all invisible from the browser, and guacd
+ * deliberately reports upstream failures as a bare number. Leaving them
+ * unmapped is what makes a broken console indistinguishable from a slow one.
+ *
+ * Source: Guacamole protocol status codes (guacamole-common-js).
+ */
+const STATUS_TEXT: Record<number, string> = {
+  256: "The server does not support this operation.",
+  512: "The remote desktop server encountered an error.",
+  513: "The remote desktop server is busy.",
+  514: "The remote host did not respond in time.",
+  515: "The remote host encountered an error.",
+  516: "The requested resource was not found.",
+  517: "The requested resource is already in use.",
+  518: "The remote connection was closed.",
+  519: "The remote host could not be reached.",
+  520: "The remote host is not currently available.",
+  521: "The session conflicts with another session.",
+  522: "The session timed out.",
+  523: "The session was closed.",
+  768: "The server rejected the connection request.",
+  769: "Authentication failed — check the credentials or SSH key.",
+  771: "Access to this connection was refused.",
+  776: "The session was closed after a period of inactivity.",
+  781: "The connection was closed because the client fell behind.",
+  782: "The server sent data the client could not understand.",
+  783: "Too many concurrent connections.",
+};
+
+/**
+ * describeStatus turns a Guacamole status into a sentence, always keeping the
+ * numeric code available separately. The code is what makes a support ticket
+ * actionable, so it is never discarded in favour of the prose.
+ */
+export function describeStatus(code: number | undefined, fallback: string): string {
+  if (code === undefined) return fallback;
+  return STATUS_TEXT[code] ?? fallback;
+}
+
+export type ConsoleHandlers = {
+  onState?: (state: ConsoleState) => void;
+  onFailure?: (failure: ConsoleFailure) => void;
+};
+
 export function openConsole(
   container: HTMLElement,
   wsUrl: string,
-  connectData = ""
+  connectData = "",
+  handlers: ConsoleHandlers = {}
 ): {
   disconnect: () => void;
   setScale: (scale: number) => void;
   resize: (width: number, height: number) => void;
   focus: () => void;
 } {
-  
-  
   const tunnel = new Guacamole.WebSocketTunnel(wsUrl);
   const client = new Guacamole.Client(tunnel);
 
+  // Set once the caller tears the session down deliberately. Guacamole reports
+  // a user-initiated disconnect through exactly the same callbacks as a far-end
+  // failure, so without this flag closing the tab raises an error banner.
+  let closing = false;
+
+  const fail = (source: "tunnel" | "client", status: any, fallback: string) => {
+    if (closing) return;
+    const code: number | undefined =
+      typeof status?.code === "number" ? status.code : undefined;
+    const message = describeStatus(
+      code,
+      typeof status?.message === "string" && status.message ? status.message : fallback
+    );
+    handlers.onFailure?.({ code, message, source });
+    handlers.onState?.("error");
+  };
+
+  tunnel.onerror = (status: any) =>
+    fail("tunnel", status, "The connection to the control plane was lost.");
+
+  client.onerror = (status: any) =>
+    fail("client", status, "The remote session ended unexpectedly.");
+
+  // Guacamole.Client.State: 0 IDLE, 1 CONNECTING, 2 WAITING, 3 CONNECTED,
+  // 4 DISCONNECTING, 5 DISCONNECTED. CONNECTING and WAITING are one state to an
+  // operator — both mean "not usable yet".
+  client.onstatechange = (state: number) => {
+    if (closing) return;
+    if (state === 1 || state === 2) handlers.onState?.("connecting");
+    else if (state === 3) handlers.onState?.("connected");
+    else if (state === 5) handlers.onState?.("disconnected");
+  };
+
   container.innerHTML = "";
   container.appendChild(client.getDisplay().getElement());
-  
+
   container.tabIndex = 0;
   container.style.outline = "none";
 
+  handlers.onState?.("connecting");
   client.connect(connectData);
 
   const display = client.getDisplay();
   let scale = 1;
 
-  
-  
-  
-  
   const mouse = new Guacamole.Mouse(display.getElement());
   mouse.onmousedown = mouse.onmouseup = mouse.onmousemove = (state: any) => {
     const s = new Guacamole.Mouse.State(
@@ -45,18 +143,10 @@ export function openConsole(
     );
     client.sendMouseState(s);
   };
-  
-  
-  
-  
+
   const keyboard = new Guacamole.Keyboard(container);
   keyboard.onkeydown = (k: number) => client.sendKeyEvent(1, k);
   keyboard.onkeyup = (k: number) => client.sendKeyEvent(0, k);
-  
-  
-
-
-
 
   const refocus = () => {
     if (document.activeElement !== container) container.focus({ preventScroll: true });
@@ -65,8 +155,6 @@ export function openConsole(
   container.addEventListener("mousedown", refocus, true);
   container.addEventListener("touchstart", refocus, true);
 
-
-
   const onBlur = () => {
     if (typeof keyboard.reset === "function") keyboard.reset();
   };
@@ -77,6 +165,7 @@ export function openConsole(
 
   return {
     disconnect() {
+      closing = true;
       container.removeEventListener("pointerdown", refocus, true);
       container.removeEventListener("mousedown", refocus, true);
       container.removeEventListener("touchstart", refocus, true);