feat: More logging for command stream

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

@@ -123,6 +123,21 @@ func poll(client *grpcclient.Client, cfg *config.Config, version string) error {
 // 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
 
@@ -157,10 +172,15 @@ func runCommandStream(ctx context.Context, cfg *config.Config) {
 			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: %v, reconnecting in %s", err, backoff)
+			log.Printf("command stream error after %s: %v, reconnecting in %s", up, err, backoff)
 		} else {
-			log.Printf("command stream closed, reconnecting in %s", backoff)
+			log.Printf("command stream closed after %s, reconnecting in %s", up, backoff)
 		}
 
 		select {
@@ -185,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)
 	}
@@ -198,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 {
@@ -207,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

@@ -178,7 +178,7 @@ 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           | **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>` |
@@ -395,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`.
 

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/internal/bus/bus.go

@@ -204,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.
@@ -274,6 +289,13 @@ 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/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,6 +263,16 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
 	}
 }
 
+// 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

server/internal/services/dispatch.go

@@ -42,6 +42,15 @@ type CommandEnvelope struct {
 	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.
@@ -108,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
 				}
 			}
 		}
@@ -152,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 {
@@ -195,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) {