fix: Renew presence on sub/pub

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>` |

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/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) {