fix(broker): keep respawned agents and held initial tasks receiving

CHANGELOG.md

@@ -23,6 +23,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- An agent released and respawned under the same name receives relay messages again. Release drops the agent's delivery cursor and the engine reuses the agent record, so every message after the respawn arrived past the start of the sequence, was classified as a gap, and was acknowledged without being delivered — discarding it and stopping the engine from retrying.
+- A PTY agent whose inbound delivery is held when it becomes ready now runs the initial task from its spawn, instead of leaving it parked in the worker's injection queue until the hold lifts.
 - The published CLI now actually reports telemetry. The npm package is plain `tsc` output with no key injection step, and the bun standalone's `--define` targeted a literal `process.env.AGENT_RELAY_POSTHOG_KEY` that the code never read (it used a computed `process.env[name]` lookup), so **both** installable artifacts shipped with telemetry silently disabled — every `cli_command_run`, `workflow_run`, `cloud_auth`, `agent_relay_tool_call`, `setup_init`, `swarm_run`, and `bridge_spawn` event was dropped. Only the Rust broker was reporting.
 - Opting out of telemetry (`AGENT_RELAY_TELEMETRY_DISABLED` or `DO_NOT_TRACK`) now keeps your cloud identity out of child process environments, including identity an ancestor process or your shell had already exported. The identity env vars — one of which carries your email — previously reached every spawned process, including third-party harness CLIs, even when opted out.
 - Identity forwarding to the Relaycast gateway is no longer gated on the local process carrying a PostHog key. An npm-installed CLI bakes no key, so it previously forwarded no identity at all and every hosted event fell back to being keyed on the workspace. Forwarding now follows the telemetry preference alone.

crates/broker/src/node_control.rs

@@ -571,6 +571,14 @@ struct AgentDeliveryCursor {
     acked_up_to_seq: u64,
     received_up_to_seq: u64,
     seen_msg_ids: SeenMsgIds,
+    /// Whether a sequenced (`seq >= 1`) position has been established for this
+    /// identity, either by a resume handshake seeding the cursor or by adopting
+    /// the first sequenced delivery. Tracked separately from the cursor's
+    /// existence because `seq:0` fan-out (reactions, receipts, action results)
+    /// creates a cursor at zero without establishing any position — without
+    /// this flag a single seq-0 frame would make the following resumed frame
+    /// look like a gap and leave a respawned agent deaf.
+    has_sequenced_position: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -721,6 +729,7 @@ impl FleetDeliveryBook {
                 acked_up_to_seq: up_to_seq,
                 received_up_to_seq: up_to_seq,
                 seen_msg_ids: SeenMsgIds::default(),
+                has_sequenced_position: true,
             },
         );
     }
@@ -782,12 +791,43 @@ impl FleetDeliveryBook {
             }
             return DeliveryDecision::Deliver { up_to_seq: 0 };
         }
-        let Some(cursor) = cursor else {
-            return if deliver.seq == 1 {
-                DeliveryDecision::Deliver { up_to_seq: 1 }
-            } else {
-                DeliveryDecision::Gap { up_to_seq: 0 }
-            };
+        // No sequenced position for this identity yet — either no cursor at all,
+        // or one created by `seq:0` fan-out, which never establishes a position.
+        // For an identity `agent.register` confirmed, that is the absence of a
+        // position rather than evidence of a gap: a release drops the cursor,
+        // and a respawn rebinds the same agent record, whose engine-side
+        // sequence keeps counting from where it left off (the engine reuses
+        // agent ids, and only seeds a cursor when it negotiates
+        // `relay:delivery-cursor-v1`). Treating that as a gap acks the message
+        // without surfacing it — see `plan_fleet_delivery` — which destroys it
+        // and stops the engine retrying, leaving the agent permanently deaf.
+        // Adopt this delivery as the starting position; `commit_received` seeds
+        // the cursor to match. A provisional binding gets no such benefit of the
+        // doubt: a second, unconfirmed identity claiming a live name must not be
+        // able to jump in mid-sequence.
+        let cursor = match cursor {
+            Some(cursor) if cursor.has_sequenced_position => cursor,
+            awaiting => {
+                let acked_up_to_seq = awaiting.map_or(0, |cursor| cursor.acked_up_to_seq);
+                if awaiting.is_some_and(|cursor| cursor.seen_msg_ids.contains(&deliver.msg_id)) {
+                    return DeliveryDecision::Duplicate {
+                        up_to_seq: acked_up_to_seq,
+                    };
+                }
+                let authoritative = self
+                    .active_agent_bindings_by_name
+                    .get(&deliver.agent)
+                    .is_some_and(|binding| binding.authoritative);
+                return if deliver.seq == 1 || authoritative {
+                    DeliveryDecision::Deliver {
+                        up_to_seq: deliver.seq,
+                    }
+                } else {
+                    DeliveryDecision::Gap {
+                        up_to_seq: acked_up_to_seq,
+                    }
+                };
+            }
         };
         if cursor.seen_msg_ids.contains(&deliver.msg_id) {
             return DeliveryDecision::Duplicate {
@@ -825,11 +865,23 @@ impl FleetDeliveryBook {
             });
         cursor.agent_name.clone_from(&deliver.agent);
         // seq:0 fan-out frames never advance either sequence cursor; they are
-        // deduped purely by msg_id.
+        // deduped purely by msg_id. They also leave the identity without a
+        // sequenced position, so the next sequenced frame is still adopted.
         if deliver.seq == 0 {
             cursor.seen_msg_ids.insert(&deliver.msg_id);
             return cursor.received_up_to_seq;
         }
+        if !cursor.has_sequenced_position {
+            // First sequenced delivery for this identity (`observe` adopted it):
+            // take the engine's position by starting one below it, so the
+            // advance below accepts this frame and every later one stays
+            // contiguous. Staying at zero instead would leave a respawned agent
+            // — whose engine sequence resumes mid-stream — permanently short of
+            // its own cursor, and every message would read as a gap.
+            cursor.acked_up_to_seq = deliver.seq.saturating_sub(1);
+            cursor.received_up_to_seq = deliver.seq.saturating_sub(1);
+            cursor.has_sequenced_position = true;
+        }
         if deliver.seq == cursor.received_up_to_seq.saturating_add(1) {
             cursor.seen_msg_ids.insert(&deliver.msg_id);
             cursor.received_up_to_seq = deliver.seq;
@@ -2388,6 +2440,104 @@ mod tests {
         );
     }
 
+    #[test]
+    fn delivery_book_respawn_keeps_delivering_when_the_engine_seq_continues() {
+        // Release drops the cursor; a respawn under the same name rebinds the
+        // same agent record, and `agent.register` does not always report a
+        // cumulative position, so no cursor is re-seeded. The engine's
+        // per-agent sequence keeps counting across the respawn, so the next
+        // live message arrives well past seq 1 — the agent must still get it.
+        let mut book = FleetDeliveryBook::default();
+        seed_authoritative_cursor(&mut book, "agent-a", "agent-a-id", 42);
+        book.remove_agent("agent-a");
+        book.bind_authoritative_identity("agent-a", "agent-a-id");
+
+        let resumed = test_delivery("agent-a", "agent-a-id", 43);
+        assert_eq!(
+            book.observe(&resumed),
+            DeliveryDecision::Deliver { up_to_seq: 43 }
+        );
+
+        // Adopting the position must also advance the cursor, or the next
+        // message reads as a gap and the agent goes deaf one frame later.
+        assert_eq!(book.commit_delivered(&resumed), 43);
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 44)),
+            DeliveryDecision::Deliver { up_to_seq: 44 }
+        );
+        // A redelivery of an adopted frame is still recognized, and a real hole
+        // in the sequence is still reported as a gap.
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 43)),
+            DeliveryDecision::Duplicate { up_to_seq: 43 }
+        );
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 99)),
+            DeliveryDecision::Gap { up_to_seq: 43 }
+        );
+    }
+
+    #[test]
+    fn delivery_book_respawn_adoption_survives_seq_zero_fan_out() {
+        // seq:0 fan-out (a reaction, a read receipt, an action result) creates
+        // a cursor without establishing any sequenced position. If the cursor's
+        // mere existence counted as a position, a single seq-0 frame landing
+        // before the respawned agent's next real message would put the cursor
+        // at zero, and the resumed frame would read as a gap — acked, never
+        // surfaced, and the agent deaf again for the rest of its life.
+        let mut book = FleetDeliveryBook::default();
+        seed_authoritative_cursor(&mut book, "agent-a", "agent-a-id", 42);
+        book.remove_agent("agent-a");
+        book.bind_authoritative_identity("agent-a", "agent-a-id");
+
+        let fan_out = test_delivery("agent-a", "agent-a-id", 0);
+        assert_eq!(
+            book.observe(&fan_out),
+            DeliveryDecision::Deliver { up_to_seq: 0 }
+        );
+        assert_eq!(book.commit_delivered(&fan_out), 0);
+        // The fan-out frame is still deduped by msg_id while the identity waits
+        // for its first sequenced delivery.
+        assert_eq!(
+            book.observe(&fan_out),
+            DeliveryDecision::Duplicate { up_to_seq: 0 }
+        );
+
+        let resumed = test_delivery("agent-a", "agent-a-id", 43);
+        assert_eq!(
+            book.observe(&resumed),
+            DeliveryDecision::Deliver { up_to_seq: 43 }
+        );
+        assert_eq!(book.commit_delivered(&resumed), 43);
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 44)),
+            DeliveryDecision::Deliver { up_to_seq: 44 }
+        );
+    }
+
+    #[test]
+    fn delivery_book_provisional_binding_still_gaps_after_seq_zero_fan_out() {
+        // The seq-0 relaxation must not become a back door for an unconfirmed
+        // identity: a provisional binding that has only seen fan-out still
+        // cannot claim a live name mid-sequence.
+        let mut book = FleetDeliveryBook::default();
+        let fan_out = test_delivery("agent-a", "agent-a-id", 0);
+        assert_eq!(
+            book.observe(&fan_out),
+            DeliveryDecision::Deliver { up_to_seq: 0 }
+        );
+        assert_eq!(book.commit_delivered(&fan_out), 0);
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 43)),
+            DeliveryDecision::Gap { up_to_seq: 0 }
+        );
+        // seq 1 is still the ordinary cold start and is delivered.
+        assert_eq!(
+            book.observe(&test_delivery("agent-a", "agent-a-id", 1)),
+            DeliveryDecision::Deliver { up_to_seq: 1 }
+        );
+    }
+
     #[test]
     fn delivery_book_retries_until_delivery_is_committed() {
         let mut book = FleetDeliveryBook::default();

crates/broker/src/protocol.rs

@@ -582,7 +582,17 @@ pub enum BrokerToWorker {
     /// asked for the backlog gets it injected immediately instead of it
     /// sitting frozen until the drive session detaches. Deliveries that
     /// arrive after the flush stay parked under the hold as usual.
-    FlushInjections {},
+    ///
+    /// With `event_id` set the flush is narrowed to that single delivery
+    /// instead of the whole backlog: it is popped through the hold even if
+    /// other injections sit in front of it, and they stay parked. The broker
+    /// sends it that way to start a (re)spawn's initial task under a hold
+    /// replayed onto a restarted worker, where relay messages retried into the
+    /// same queue must not ride along.
+    FlushInjections {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        event_id: Option<EventId>,
+    },
     /// Versioned control sent to a native harness sidecar. The
     /// envelope request id correlates the sidecar's command response.
     NativeHarnessCommand {
@@ -1159,12 +1169,28 @@ mod tests {
 
     #[test]
     fn broker_to_worker_flush_injections_round_trip() {
-        let msg = BrokerToWorker::FlushInjections {};
+        let msg = BrokerToWorker::FlushInjections { event_id: None };
         let encoded = serde_json::to_string(&msg).unwrap();
         let raw: Value = serde_json::from_str(&encoded).unwrap();
         // Wire tag must be snake_case and match the worker-side string match arm
         // in `pty_worker.rs` and `packages/harness-driver/src/protocol.ts`.
         assert_eq!(raw["type"], "flush_injections");
+        // A blanket flush carries no `event_id`, so older workers keep seeing
+        // the exact payload they always did.
+        assert!(raw["payload"].get("event_id").is_none());
+        let decoded: BrokerToWorker = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(decoded, msg);
+    }
+
+    #[test]
+    fn broker_to_worker_targeted_flush_injections_round_trip() {
+        let msg = BrokerToWorker::FlushInjections {
+            event_id: Some("init_abc123".into()),
+        };
+        let encoded = serde_json::to_string(&msg).unwrap();
+        let raw: Value = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(raw["type"], "flush_injections");
+        assert_eq!(raw["payload"]["event_id"], "init_abc123");
         let decoded: BrokerToWorker = serde_json::from_str(&encoded).unwrap();
         assert_eq!(decoded, msg);
     }