diff --git a/CHANGELOG.md b/CHANGELOG.md index 994074410..ac358390b 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 660083a1a..501e49abb 100644 --- a/crates/broker/src/node_control.rs +++ b/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(); diff --git a/crates/broker/src/protocol.rs b/crates/broker/src/protocol.rs index 889fd3f6b..66e0e30ae 100644 --- a/crates/broker/src/protocol.rs +++ b/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, + }, /// 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); } diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 93c8f063e..069c54c0b 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -119,6 +119,12 @@ struct ActiveInjection { /// injection even while the interactive hold is active. A human asked /// for the backlog explicitly, so writing it is not a splice. hold_exempt: bool, + /// The exemption above came from an `event_id`-scoped flush rather than a + /// blanket one, so requeueing must return it to the event-id set instead of + /// the backlog counter — a targeted exemption stays bound to its one + /// delivery and must never turn into an allowance the next queued relay + /// message can spend. + targeted_hold_exemption: bool, } /// Result of an in-flight `write_pty` awaiting the PTY drainer: the request id @@ -297,6 +303,56 @@ fn injection_pop_allowed( !active_injection_present && (!interactive_hold || hold_exempt_remaining > 0) } +/// Index of the pending injection the loop may start this tick, if any. +/// +/// Normally that is just the front of the queue. Under an interactive hold with +/// no blanket allowance, only a delivery carrying a *targeted* exemption may go +/// — the broker released that one `event_id` explicitly (a (re)spawn's initial +/// task through a replayed hold) — so the queue is scanned for the first such +/// entry and everything ahead of it stays parked. Popping it out of order is +/// the point: a relay message retried into a restarted worker must not splice +/// into the human's session just because it happens to sit in front. +fn next_injection_index( + pending: &VecDeque, + active_injection_present: bool, + interactive_hold: bool, + hold_exempt_remaining: usize, + hold_exempt_event_ids: &HashSet, +) -> Option { + if active_injection_present { + return None; + } + if injection_pop_allowed( + active_injection_present, + interactive_hold, + hold_exempt_remaining, + ) { + return (!pending.is_empty()).then_some(0); + } + pending + .iter() + .position(|entry| hold_exempt_event_ids.contains(entry.delivery.event_id.as_str())) +} + +/// Return an interrupted injection's hold exemption so a requeued frame is +/// retried through the hold instead of freezing behind it. A targeted exemption +/// goes back to the event-id set — it must stay bound to that one delivery — +/// while a blanket `flush_injections` allowance goes back to the counter. +fn restore_hold_exemption( + injection: &ActiveInjection, + hold_exempt_injections: &mut usize, + hold_exempt_event_ids: &mut HashSet, +) { + if !injection.hold_exempt { + return; + } + if injection.targeted_hold_exemption { + hold_exempt_event_ids.insert(injection.pending.delivery.event_id.to_string()); + } else { + *hold_exempt_injections += 1; + } +} + /// Relay command detection must stay disabled for the entire injection lifecycle. /// The paced writer can echo command-like text before its post-write verification /// is queued, so checking only `pending_verifications` leaves a false-positive gap. @@ -481,6 +537,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // write so an explicit flush is never silently swallowed by a transient // error. Deliveries arriving after the flush get no exemption. let mut hold_exempt_injections: usize = 0; + // Event ids released through the hold individually by an `event_id`-scoped + // `flush_injections`. Unlike the counter above this grants nothing to the + // rest of the backlog: only the named delivery is popped (out of order if + // needed) and everything else stays parked. The broker uses it to start a + // (re)spawn's initial task under a replayed hold. An id is recorded even + // when the delivery has not been queued yet, so a retry that lands after + // the flush is still released; a hold boundary clears the set. + let mut hold_exempt_event_ids: HashSet = HashSet::new(); // Ack receiver for the in-flight injection's most recent Body/Enter write. // `submit_write` only enqueues to the bounded drainer queue; the oneshot // resolves once the drainer has actually written and flushed those bytes to @@ -856,8 +920,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // before a drive attach could keep injecting // into the human's session. hold_exempt_injections = 0; + hold_exempt_event_ids.clear(); if let Some(inj) = active_injection.as_mut() { inj.hold_exempt = false; + inj.targeted_hold_exemption = false; } } "flush_injections" => { @@ -871,18 +937,54 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // injections pop normally and an exemption // recorded now could pierce a hold that starts // before a slow backlog finishes draining. + // + // An `event_id` in the payload narrows the flush + // to that one delivery instead: the broker uses + // it to start a (re)spawn's initial task under a + // replayed hold, and a relay message retried into + // the restarted worker must keep parking rather + // than riding along on the same exemption. + let targeted_event_id = frame + .payload + .get("event_id") + .and_then(Value::as_str) + .filter(|event_id| !event_id.is_empty()) + .map(str::to_string); if pty_auto.interactive_hold { - let backlog = pending_worker_injections.len(); - hold_exempt_injections = hold_exempt_injections.max(backlog); - if let Some(inj) = active_injection.as_mut() { - inj.hold_exempt = true; + if let Some(event_id) = targeted_event_id { + if let Some(inj) = active_injection.as_mut() { + if inj.pending.delivery.event_id == event_id { + inj.hold_exempt = true; + inj.targeted_hold_exemption = true; + } + } + tracing::info!( + target: "agent_relay::worker::pty", + worker = %worker_name, + event_id = %event_id, + "releasing one delivery through interactive hold" + ); + hold_exempt_event_ids.insert(event_id); + } else { + let backlog = pending_worker_injections.len(); + hold_exempt_injections = + hold_exempt_injections.max(backlog); + if let Some(inj) = active_injection.as_mut() { + // Leave an existing targeted exemption + // targeted. Downgrading it would send a + // later requeue of *this* delivery back + // to the shared counter instead of to + // its own event id, so the credit could + // be spent on whatever else is queued. + inj.hold_exempt = true; + } + tracing::info!( + target: "agent_relay::worker::pty", + worker = %worker_name, + backlog, + "flushing queued injections through interactive hold" + ); } - tracing::info!( - target: "agent_relay::worker::pty", - worker = %worker_name, - backlog, - "flushing queued injections through interactive hold" - ); } } "ping" => { @@ -1352,14 +1454,22 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // Gated off while an interactive hold is active so queued deliveries // stay parked (not dropped) until the human releases the drive. _ = pending_injection_interval.tick() => { - if injection_pop_allowed(active_injection.is_some(), pty_auto.interactive_hold, hold_exempt_injections) { + if let Some(index) = next_injection_index( + &pending_worker_injections, + active_injection.is_some(), + pty_auto.interactive_hold, + hold_exempt_injections, + &hold_exempt_event_ids, + ) { let should_block = pending_worker_injections - .front() + .get(index) .map(|pending| should_block_pending_injection(pty_auto.auto_suggestion_visible, pending)) .unwrap_or(false); if !should_block { - if let Some(pending) = pending_worker_injections.pop_front() { - let hold_exempt = hold_exempt_injections > 0; + if let Some(pending) = pending_worker_injections.remove(index) { + let targeted_hold_exemption = + hold_exempt_event_ids.remove(pending.delivery.event_id.as_str()); + let hold_exempt = targeted_hold_exemption || hold_exempt_injections > 0; hold_exempt_injections = hold_exempt_injections.saturating_sub(1); active_injection = Some(ActiveInjection { pending, @@ -1367,6 +1477,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { next_at: tokio::time::Instant::now() + throttle.delay(), injection_text: None, hold_exempt, + targeted_hold_exemption, }); } } @@ -1398,9 +1509,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { error = %error, "steer mode ESC ESC write failed, re-queuing delivery" ); - if inj.hold_exempt { - hold_exempt_injections += 1; - } + restore_hold_exemption( + &inj, + &mut hold_exempt_injections, + &mut hold_exempt_event_ids, + ); pending_worker_injections.push_front(inj.pending); } else { inj.stage = InjectionStage::Body; @@ -1474,9 +1587,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { error = %e, "PTY injection write failed, re-queuing delivery" ); - if inj.hold_exempt { - hold_exempt_injections += 1; - } + restore_hold_exemption( + &inj, + &mut hold_exempt_injections, + &mut hold_exempt_event_ids, + ); pending_worker_injections.push_front(inj.pending); } } @@ -1490,9 +1605,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { delivery_id = %inj.pending.delivery.delivery_id, "injection reached Finalize in deadline arm without pending ack; re-queuing delivery" ); - if inj.hold_exempt { - hold_exempt_injections += 1; - } + restore_hold_exemption( + &inj, + &mut hold_exempt_injections, + &mut hold_exempt_event_ids, + ); pending_worker_injections.push_front(inj.pending); } } @@ -1569,9 +1686,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // the next pop retries the same delivery. A hold-exempt // injection keeps its exemption so an explicit flush is // retried rather than freezing under the hold. - if inj.hold_exempt { - hold_exempt_injections += 1; - } + restore_hold_exemption( + &inj, + &mut hold_exempt_injections, + &mut hold_exempt_event_ids, + ); pending_worker_injections.push_front(inj.pending); } } @@ -2045,6 +2164,134 @@ mod tests { assert!(injection_pop_allowed(false, false, 1)); } + fn test_pending_injection(event_id: &str) -> PendingWorkerInjection { + PendingWorkerInjection { + delivery: RelayDelivery { + delivery_id: format!("del_{event_id}").into(), + event_id: event_id.into(), + workspace_id: None, + workspace_alias: None, + from: "Lead".into(), + target: "Worker".into(), + body: "hello".into(), + thread_id: None, + priority: None, + injection_mode: MessageInjectionMode::Wait, + }, + request_id: None, + queued_at: Instant::now(), + } + } + + #[test] + fn targeted_flush_releases_only_its_own_delivery_through_the_hold() { + // A supervised restart keeps unacknowledged deliveries, so a retried + // relay message can be queued ahead of the (re)spawn's initial task. + // The targeted flush must reach past it — and leave it parked. + let pending: VecDeque = [ + test_pending_injection("evt_relay"), + test_pending_injection("init_task"), + ] + .into(); + let targeted = HashSet::from(["init_task".to_string()]); + + assert_eq!( + next_injection_index(&pending, false, true, 0, &targeted), + Some(1) + ); + // Nothing targeted: the whole queue stays parked under the hold. + assert_eq!( + next_injection_index(&pending, false, true, 0, &HashSet::new()), + None + ); + // A blanket flush still drains from the front, in order. + assert_eq!( + next_injection_index(&pending, false, true, 2, &HashSet::new()), + Some(0) + ); + // Unheld, the target is irrelevant — normal FIFO order applies. + assert_eq!( + next_injection_index(&pending, false, false, 0, &targeted), + Some(0) + ); + // An injection already in flight is never preempted, targeted or not. + assert_eq!( + next_injection_index(&pending, true, true, 0, &targeted), + None + ); + // An empty queue has nothing to start. + assert_eq!( + next_injection_index(&VecDeque::new(), false, false, 0, &targeted), + None + ); + } + + #[test] + fn requeued_targeted_exemption_returns_to_its_event_id() { + // A failed write requeues the injection. A targeted exemption must go + // back to the event-id set — turning it into a blanket allowance would + // let the next queued relay message spend it and splice into the hold. + let mut count = 0usize; + let mut event_ids = HashSet::new(); + let targeted = ActiveInjection { + pending: test_pending_injection("init_task"), + stage: InjectionStage::Escape, + next_at: tokio::time::Instant::now(), + injection_text: None, + hold_exempt: true, + targeted_hold_exemption: true, + }; + restore_hold_exemption(&targeted, &mut count, &mut event_ids); + assert_eq!(count, 0); + assert!(event_ids.contains("init_task")); + + // A blanket exemption still goes back to the counter. + let blanket = ActiveInjection { + targeted_hold_exemption: false, + ..targeted + }; + let mut event_ids = HashSet::new(); + restore_hold_exemption(&blanket, &mut count, &mut event_ids); + assert_eq!(count, 1); + assert!(event_ids.is_empty()); + + // An injection popped without any exemption restores nothing. + let unexempt = ActiveInjection { + hold_exempt: false, + ..blanket + }; + restore_hold_exemption(&unexempt, &mut count, &mut event_ids); + assert_eq!(count, 1); + assert!(event_ids.is_empty()); + } + + #[test] + fn blanket_flush_does_not_downgrade_an_in_flight_targeted_exemption() { + // A blanket `flush_injections` landing while the targeted initial task is + // in flight must not turn its exemption into a shared allowance: a later + // requeue would then hand the credit to whatever else is queued instead + // of back to this delivery. + let mut injection = ActiveInjection { + pending: test_pending_injection("init_task"), + stage: InjectionStage::Escape, + next_at: tokio::time::Instant::now(), + injection_text: None, + hold_exempt: true, + targeted_hold_exemption: true, + }; + // What the blanket branch does to an already-exempt in-flight injection. + injection.hold_exempt = true; + + let mut count = 0usize; + let mut event_ids = HashSet::new(); + restore_hold_exemption(&injection, &mut count, &mut event_ids); + assert_eq!( + count, 0, + "targeted exemption must not become blanket credit" + ); + assert!(event_ids.contains("init_task")); + } + #[test] fn injection_body_enter_ack_finalizes_when_confirmed() { // Combined Body+Enter write confirmed (stage advanced to Finalize) → diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 5044864b5..5603e2bf8 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -743,12 +743,12 @@ impl BrokerRuntime { .get(&name) .map(|handle| handle.spec.runtime == AgentRuntime::Pty) .unwrap_or(false); - if is_pty_worker + let interactive_hold_replayed = is_pty_worker && delivery_states .get(&name) .map(|s| s.mode == InboundDeliveryMode::ManualFlush) - .unwrap_or(false) - { + .unwrap_or(false); + if interactive_hold_replayed { if let Err(err) = workers .send_to_worker( &name, @@ -786,6 +786,37 @@ impl BrokerRuntime { { tracing::warn!(worker = %name, error = %e, "failed to deliver initial_task"); } + // The initial task bypasses the delivery-mode queue, but the + // hold replayed above freezes injection pops inside the PTY + // worker — so without this frame the task the spawn was asked + // to run sits invisibly in the worker's queue until the hold + // lifts. A worker that restarts while explicitly held reaches + // `worker_ready` in exactly that state. The spawn asked for + // this task, so a one-shot exemption releases it; later relay + // messages keep parking under the hold as usual. + // + // The flush is scoped to this task's own `event_id`. A blanket + // flush would exempt the worker's whole queue, and a restart + // keeps unacknowledged deliveries (see `maintenance.rs`), so a + // relay message retried into the new worker before + // `worker_ready` would splice into the human's session too. + if interactive_hold_replayed { + if let Err(err) = workers + .send_to_worker( + &name, + "flush_injections", + None, + json!({ "event_id": event_id }), + ) + .await + { + tracing::warn!( + worker = %name, + error = %err, + "failed to release initial task through interactive hold" + ); + } + } } let runtime = value .get("payload") diff --git a/packages/harness-driver/src/protocol.ts b/packages/harness-driver/src/protocol.ts index bb5d3fe7d..bc730fe35 100644 --- a/packages/harness-driver/src/protocol.ts +++ b/packages/harness-driver/src/protocol.ts @@ -646,9 +646,15 @@ export type 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. + * + * 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. */ type: 'flush_injections'; - payload: Record; + payload: { event_id?: string }; } | { type: 'native_harness_command';