From efb9309ac496f21378b5f1674472e2e7c256c94c Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Mon, 27 Jul 2026 12:18:23 -0400 Subject: [PATCH 1/3] fix(broker): keep respawned agents and held initial tasks receiving MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An agent released and respawned under the same name stopped receiving relay messages. Release drops the agent's delivery cursor, and the engine reuses the agent record without reporting a cumulative position, so the next message arrived past the start of the sequence, was classified as a gap, and was acknowledged without being surfaced — discarding it and stopping the engine from retrying. An identity confirmed by `agent.register` now adopts the observed sequence as its cursor instead of treating a missing position as a gap. A provisional binding keeps the previous response so an unconfirmed second identity cannot claim a live name mid-sequence. A PTY worker that becomes ready while its inbound delivery is held also now releases the initial task from its spawn through the hold, instead of leaving it parked in the injection queue until the hold lifts. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 2 + crates/broker/src/node_control.rs | 68 +++++++++++++++++++++- crates/broker/src/runtime/worker_events.rs | 26 ++++++++- 3 files changed, 91 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e7ce1aabe..44b780f9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,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. - `node agent attach --mode view` now exits on the first Ctrl-C instead of waiting for a WebSocket close handshake. - The broker now sends its anonymous telemetry id (`X-Agent-Relay-Distinct-Id`) and origin actor with its Relaycast requests, so hosted usage can be attributed to an install instead of only to a workspace. The id header is omitted when telemetry is opted out; requests and origin actor are unaffected. - The broker now reads its telemetry preference and machine-id files from `AGENT_RELAY_DATA_DIR` when set, matching the CLI. It previously only read `~/.agentworkforce/relay/telemetry.json`, so an opt-out written by `agent-relay telemetry disable` under a configured data directory was ignored. diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index f7d73427e..b97c1f897 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -780,8 +780,27 @@ 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 } + // No cursor for this identity yet. 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 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: 0 } }; @@ -818,6 +837,14 @@ impl FleetDeliveryBook { .entry(deliver.agent_id.clone()) .or_insert_with(|| AgentDeliveryCursor { agent_name: deliver.agent.clone(), + // First delivery seen for this identity: adopt the engine's + // position by starting one below it, so the advance below + // accepts this frame and every later one stays contiguous. + // Starting at zero instead would leave a respawned agent — whose + // engine sequence resumes mid-stream — permanently one short of + // its own cursor, and every message would read as a gap. + acked_up_to_seq: deliver.seq.saturating_sub(1), + received_up_to_seq: deliver.seq.saturating_sub(1), ..AgentDeliveryCursor::default() }); cursor.agent_name.clone_from(&deliver.agent); @@ -2384,6 +2411,43 @@ 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_retries_until_delivery_is_committed() { let mut book = FleetDeliveryBook::default(); diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 5044864b5..1241736e2 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,26 @@ 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. + if interactive_hold_replayed { + if let Err(err) = workers + .send_to_worker(&name, "flush_injections", None, json!({})) + .await + { + tracing::warn!( + worker = %name, + error = %err, + "failed to release initial task through interactive hold" + ); + } + } } let runtime = value .get("payload") From 95d6bde276fa55a52f93b706d185668d8bdff57e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 14:57:06 +0000 Subject: [PATCH 2/3] fix(broker): survive seq-0 fan-out and scope the initial-task flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P1s from review on the respawn fix. Restart adoption survived only while the respawned identity had no cursor at all. A `seq:0` fan-out frame — a reaction, a read receipt, an `action.completed` result — creates a cursor without establishing any position, so one landing before the agent's next real message put the cursor at zero and the resumed frame (seq 43) read as a gap again: acked, never surfaced, agent deaf. Adoption is now keyed on an explicit `has_sequenced_position`, set by a seeded cursor or the first sequenced delivery, instead of on the cursor's existence. The `flush_injections` that releases a (re)spawn's initial task through a replayed interactive hold granted an allowance for the worker's whole queue. A supervised restart retains unacknowledged deliveries, so a relay message retried into the new worker before `worker_ready` could spend that allowance and splice into the human's session. The frame now carries the initial task's `event_id` and the worker releases only that delivery — popping it past anything queued in front, which stays parked. A blanket flush (`POST /api/spawned/{name}/flush`) is unchanged. --- crates/broker/src/node_control.rs | 152 +++++++++--- crates/broker/src/protocol.rs | 30 ++- crates/broker/src/pty_worker.rs | 267 +++++++++++++++++++-- crates/broker/src/runtime/worker_events.rs | 13 +- packages/harness-driver/src/protocol.ts | 8 +- 5 files changed, 407 insertions(+), 63 deletions(-) diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index c3f2ec7b8..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,31 +791,43 @@ impl FleetDeliveryBook { } return DeliveryDecision::Deliver { up_to_seq: 0 }; } - let Some(cursor) = cursor else { - // No cursor for this identity yet. 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 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, + // 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, + }; } - } else { - DeliveryDecision::Gap { up_to_seq: 0 } - }; + 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 { @@ -840,23 +861,27 @@ impl FleetDeliveryBook { .entry(deliver.agent_id.clone()) .or_insert_with(|| AgentDeliveryCursor { agent_name: deliver.agent.clone(), - // First delivery seen for this identity: adopt the engine's - // position by starting one below it, so the advance below - // accepts this frame and every later one stays contiguous. - // Starting at zero instead would leave a respawned agent — whose - // engine sequence resumes mid-stream — permanently one short of - // its own cursor, and every message would read as a gap. - acked_up_to_seq: deliver.seq.saturating_sub(1), - received_up_to_seq: deliver.seq.saturating_sub(1), ..AgentDeliveryCursor::default() }); 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; @@ -2452,6 +2477,67 @@ mod tests { ); } + #[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..99bdc33a5 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,49 @@ 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() { + inj.hold_exempt = true; + inj.targeted_hold_exemption = false; + } + 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 +1449,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 +1472,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 +1504,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 +1582,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 +1600,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 +1681,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 +2159,107 @@ 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 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 1241736e2..5603e2bf8 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -794,9 +794,20 @@ impl BrokerRuntime { // `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!({})) + .send_to_worker( + &name, + "flush_injections", + None, + json!({ "event_id": event_id }), + ) .await { tracing::warn!( 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'; From 9529071fe8698a4de6c067d124cbad99561a306b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 15:12:24 +0000 Subject: [PATCH 3/3] fix(broker): keep a targeted hold exemption targeted across a blanket flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A blanket `flush_injections` arriving while the initial task was in flight downgraded its targeted exemption. If that injection's write then failed, `restore_hold_exemption` returned the credit to the shared counter instead of to the delivery's own event id, so it could be spent on whatever else was queued — the failure mode the targeted exemption exists to prevent. The blanket branch now only grants `hold_exempt` and leaves an existing targeted exemption alone. --- crates/broker/src/pty_worker.rs | 34 ++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 99bdc33a5..069c54c0b 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -970,8 +970,13 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { 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; - inj.targeted_hold_exemption = false; } tracing::info!( target: "agent_relay::worker::pty", @@ -2260,6 +2265,33 @@ mod tests { 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) →