From e36d6f47ee57696a97255be59d7b65efee69fe8b Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 08:15:17 +0100 Subject: [PATCH 1/6] fix(sync): judge agent silence by thread state, not a tool-duration guess MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The silence watchdog shipped in #73 disarmed permanently on the first agent event. That covers the production wedge (prompt accepted, zero events ever) but misses the other shape: streamed some output, then stalled before completing — observed in the E2E claude round, where the agent returned a correct answer and then never sent message_completed. The obvious fix — a second, longer idle timeout — would have required hard-coding an assumption about the longest plausible tool call, which is both unknowable and wrong the first time someone runs a 40-minute build. Instead the watchdog now asks the thread whether anything is actually outstanding. has_outstanding_work() reports true while any tool call is Pending / InProgress / WaitingForConfirmation. Busy threads are exempt for exactly as long as the work genuinely takes, with no duration constant anywhere; silence is only judged when nothing is running. Both wedge shapes then fall out of one rule: generating + no events for + nothing outstanding => wedged That also means the budget no longer has to cover tool calls at all — only model think-time between consecutive events — so a single modest 120s default is defensible. Renamed HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS -> HELIX_ACP_SILENCE_TIMEOUT_SECS to match the changed semantics (shipped hours ago, no deployment references it). Also fixes the E2E agent-version diagnostic, which queried @anthropic-ai/claude-agent-acp — a package that 404s. Zed installs @agentclientprotocol/claude-agent-acp (0.63.0 today, the same version that wedged in prod). The wrong scope silently reported 'unknown', disabling the one signal that distinguishes our regression from an agent-package change. Now queries the right scope, warns loudly if it cannot resolve, and states that the install is unpinned. cargo test -p external_websocket_sync: 56 passed. --- .../e2e-test/run_e2e.sh | 21 +- .../src/thread_service.rs | 195 +++++++++++++----- 2 files changed, 160 insertions(+), 56 deletions(-) diff --git a/crates/external_websocket_sync/e2e-test/run_e2e.sh b/crates/external_websocket_sync/e2e-test/run_e2e.sh index 5d8ae5d39f60d3..a793cf34f5755c 100755 --- a/crates/external_websocket_sync/e2e-test/run_e2e.sh +++ b/crates/external_websocket_sync/e2e-test/run_e2e.sh @@ -199,10 +199,23 @@ if echo "$E2E_AGENTS" | grep -q "claude"; then LOCAL_VERSION=$(node -e "console.log(require('/opt/claude-agent-acp/package.json').version)" 2>/dev/null || echo "unknown") echo "[setup] Using LOCAL claude-agent-acp v$LOCAL_VERSION from /opt/claude-agent-acp" else - # Log which version npx will install so we can correlate failures - # with claude-agent-acp upgrades. This is a quick check, not an install. - CLAUDE_ACP_VERSION=$(npm view @anthropic-ai/claude-agent-acp version 2>/dev/null || echo "unknown") - echo "[setup] Using npm-installed claude-agent-acp (auto-install, latest=$CLAUDE_ACP_VERSION)" + # Log which version npx will install so we can correlate failures with + # claude-agent-acp upgrades. This is a quick check, not an install. + # + # The scope matters and has been wrong before: Zed installs + # @agentclientprotocol/claude-agent-acp (see crates/agent_servers/), NOT + # @anthropic-ai/... . Querying the wrong scope silently yields "unknown", + # which quietly disables the one signal that distinguishes "we regressed" + # from "the agent package changed under us" when the claude round fails. + CLAUDE_ACP_PKG="@agentclientprotocol/claude-agent-acp" + CLAUDE_ACP_VERSION=$(npm view "$CLAUDE_ACP_PKG" version 2>/dev/null || echo "") + if [ -z "$CLAUDE_ACP_VERSION" ]; then + echo "[setup] WARNING: could not resolve a version for $CLAUDE_ACP_PKG." + echo "[setup] A claude-round failure will NOT be attributable to an agent-package change." + CLAUDE_ACP_VERSION="unknown" + fi + echo "[setup] Using npm-installed claude-agent-acp $CLAUDE_ACP_PKG (auto-install, latest=$CLAUDE_ACP_VERSION)" + echo "[setup] NOTE: this install is UNPINNED — the claude round is not reproducible across time." fi AGENT_SERVERS_JSON=$(cat << AGENTEOF "agent_servers": { diff --git a/crates/external_websocket_sync/src/thread_service.rs b/crates/external_websocket_sync/src/thread_service.rs index 930e84ac068d1d..1d2f944c2b7805 100644 --- a/crates/external_websocket_sync/src/thread_service.rs +++ b/crates/external_websocket_sync/src/thread_service.rs @@ -1238,19 +1238,58 @@ pub fn ensure_thread_subscription( spawn_silent_turn_watchdog(thread_entity, thread_id, cx); } -/// Origin-agnostic companion to the watchdog in [`handle_follow_up_message`]. +/// True while the thread is waiting on something that legitimately owns an +/// unbounded amount of wall-clock: a tool the agent is running, or a permission +/// prompt sitting in front of the user. /// -/// The send-path watchdog only wraps turns that Helix dispatched. A turn started -/// by the user typing directly into the Zed agent panel goes `agent_ui` → -/// `AcpThread::send()` and never passes through `handle_follow_up_message` — yet -/// that is exactly how the 2026-07-29 production wedge was triggered. This task -/// watches the thread itself, so it covers every turn regardless of origin. +/// This is what lets the silence watchdog avoid guessing "how long is the +/// longest tool call?". We do not need a duration assumption, because the thread +/// state already tells us whether anything is outstanding. A 40-minute +/// `cargo build` keeps an `InProgress` tool entry for its whole duration and is +/// therefore exempt for exactly as long as it actually takes; silence with +/// *nothing* outstanding is a different thing entirely, and that is the only +/// case the watchdog judges. +fn has_outstanding_work(thread: &AcpThread) -> bool { + thread.entries().iter().rev().any(|entry| { + matches!( + entry, + acp_thread::AgentThreadEntry::ToolCall(tc) + if matches!( + tc.status, + acp_thread::ToolCallStatus::Pending + | acp_thread::ToolCallStatus::InProgress + | acp_thread::ToolCallStatus::WaitingForConfirmation { .. } + ) + ) + }) +} + +/// Origin-agnostic silence watchdog for a thread. +/// +/// The send-path check in [`handle_follow_up_message`] only wraps turns that +/// Helix dispatched. A turn started by the user typing directly into the Zed +/// agent panel goes `agent_ui` → `AcpThread::send()` and never passes through it +/// — yet that is exactly how the 2026-07-29 production wedge was triggered. This +/// task watches the thread itself, so it covers every turn regardless of origin. /// -/// It observes `ThreadStatus::Generating` (i.e. `running_turn.is_some()`) and the -/// [`THREAD_ACTIVITY`] counter. A turn that is generating but has produced ZERO -/// events for longer than the first-event budget is reported to Helix as a -/// terminal `chat_response_error`, which marks the interaction errored and frees -/// the activation lane instead of leaving it in `waiting` forever. +/// **One rule:** while the thread is `Generating`, if no `AcpThreadEvent` has +/// been observed for longer than the silence budget AND nothing is outstanding +/// (see [`has_outstanding_work`]), the agent is wedged. Report it to Helix as a +/// terminal `chat_response_error` so the interaction is marked errored and the +/// activation lane freed, instead of sitting in `waiting` forever with nothing +/// surfaced in either UI. +/// +/// The state check is what makes a single modest budget safe, and is why there +/// is no "longest plausible tool call" constant anywhere here — an agent running +/// a long tool is *not silent by this definition*, it is busy, and busy is +/// exempt for however long the work genuinely takes. The budget only has to +/// cover model think-time between one event and the next, which is seconds. +/// +/// Both wedge shapes fall out of the same rule: +/// * prompt accepted, zero events ever → nothing outstanding, silent → caught +/// (this is the production wedge) +/// * streamed some output, then stalled → nothing outstanding, silent → caught +/// (this is the shape seen in the E2E claude round) /// /// Deliberately does NOT call `thread.cancel()`: the Stopped handler would then /// emit `message_completed` on top of this error and Helix would see the turn as @@ -1259,7 +1298,7 @@ pub fn ensure_thread_subscription( /// self-heals on the user's next message while Helix has already been told the /// truth. fn spawn_silent_turn_watchdog(thread_entity: &Entity, thread_id: &str, cx: &mut App) { - let Some(budget) = first_event_timeout() else { + let Some(budget) = silence_timeout() else { return; }; const POLL: Duration = Duration::from_secs(5); @@ -1268,10 +1307,10 @@ fn spawn_silent_turn_watchdog(thread_entity: &Entity, thread_id: &str let thread_id = thread_id.to_string(); cx.spawn(async move |cx| { - // Per-turn state: when the current generating turn was first seen, and - // the activity count at that moment. - let mut turn_started_at: Option = None; - let mut baseline: u64 = 0; + // `quiet_since` is the last moment we saw either an event or outstanding + // work — i.e. the last moment this thread was demonstrably alive. + let mut quiet_since: Option = None; + let mut last_activity: u64 = 0; let mut reported = false; loop { @@ -1280,36 +1319,43 @@ fn spawn_silent_turn_watchdog(thread_entity: &Entity, thread_id: &str let Some(thread) = weak.upgrade() else { return; // thread gone — stop watching }; - let status = cx.update(|cx| thread.read(cx).status()); + let (status, busy) = cx.update(|cx| { + let t = thread.read(cx); + (t.status(), has_outstanding_work(t)) + }); match status { acp_thread::ThreadStatus::Idle => { // Turn boundary: re-arm for the next one. - turn_started_at = None; + quiet_since = None; reported = false; } acp_thread::ThreadStatus::Generating => { - let started = *turn_started_at.get_or_insert_with(|| { - baseline = activity_count(&thread_id); - Instant::now() - }); - - if activity_count(&thread_id) > baseline { - // Proof of life — disarm for the rest of this turn. - reported = true; + let now = Instant::now(); + let seen = activity_count(&thread_id); + + // Alive if either an event landed since the last poll, or a + // tool / permission prompt is currently outstanding. Both + // reset the clock; `busy` is what makes a long tool call + // exempt for as long as it genuinely takes, with no + // assumption about how long that is. + if seen > last_activity || busy || quiet_since.is_none() { + last_activity = seen; + quiet_since = Some(now); + reported = false; continue; } - if !reported && started.elapsed() >= budget { + let quiet_for = now.duration_since(quiet_since.unwrap_or(now)); + if !reported && quiet_for >= budget { reported = true; let request_id = crate::get_thread_request_id(&thread_id) .unwrap_or_default(); let msg = format!( - "{}: agent has been generating for {:?} without emitting a single \ - event on thread {} — treating the agent session as wedged", - SILENT_PROMPT_WEDGE_MARKER, - started.elapsed(), - thread_id + "{}: thread {} has been generating with no agent events for {:?} \ + and nothing outstanding (no running tool, no pending permission) \ + — treating the agent session as wedged", + SILENT_PROMPT_WEDGE_MARKER, thread_id, quiet_for ); eprintln!("🛑 [THREAD_SERVICE] {}", msg); log::warn!("🛑 [THREAD_SERVICE] {}", msg); @@ -2212,7 +2258,7 @@ async fn handle_follow_up_message( // otherwise never resolve and this await would hang forever (the // 2026-07-29 production wedge). Once any event lands the watchdog // disarms and the turn runs unbounded, so long tool calls are unaffected. - let send_result = if let Some(budget) = first_event_timeout() { + let send_result = if let Some(budget) = silence_timeout() { let send_task = Box::pin(send_task); let watchdog = Box::pin(wait_for_first_agent_activity( &thread_id, @@ -2312,18 +2358,20 @@ pub(crate) fn is_silent_prompt_wedge(msg: &str) -> bool { msg.contains(SILENT_PROMPT_WEDGE_MARKER) } -/// How long a freshly-dispatched prompt may produce *zero* agent events before -/// we declare the session wedged. Override with -/// `HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS`; set to `0` to disable the watchdog. +/// How long a generating thread may go with **no agent events and nothing +/// outstanding** before we declare the session wedged. Override with +/// `HELIX_ACP_SILENCE_TIMEOUT_SECS`; set to `0` to disable the watchdog. /// -/// This is a time-to-FIRST-event budget, not a turn budget: once the agent emits -/// anything at all the watchdog disarms for the rest of the turn, so arbitrarily -/// long tool calls and slow generations are unaffected. A healthy agent emits -/// its first event within a couple of seconds, so 120s is ~2 orders of magnitude -/// of headroom while still bounding a wedge to a couple of minutes instead of -/// forever. -fn first_event_timeout() -> Option { - let secs = std::env::var("HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS") +/// This is a silence budget, not a turn budget, and crucially not a +/// "longest tool call" guess. Whether the agent is legitimately busy is answered +/// by thread *state* ([`has_outstanding_work`]) rather than by a duration +/// assumption: a running tool or a pending permission prompt is exempt for as +/// long as it genuinely takes. The budget therefore only has to cover model +/// think-time between one event and the next, which is normally seconds — 120s +/// is roughly two orders of magnitude of headroom while still bounding a wedge +/// to a couple of minutes instead of forever. +fn silence_timeout() -> Option { + let secs = std::env::var("HELIX_ACP_SILENCE_TIMEOUT_SECS") .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(120); @@ -3474,20 +3522,63 @@ mod silent_prompt_wedge_tests { ); } + /// A long-running tool must be exempt from the silence budget for as long as + /// it actually takes. This is the property that removes any need to guess a + /// "longest plausible tool call" duration: busy-ness is read from thread + /// state, not from a clock. + #[gpui::test] + async fn outstanding_tool_call_is_never_judged_silent(cx: &mut TestAppContext) { + use acp_thread::ToolCallStatus; + + // Every status that means "something else legitimately owns the time". + for status in [ + ToolCallStatus::Pending, + ToolCallStatus::InProgress, + ] { + assert!( + matches!( + status, + ToolCallStatus::Pending | ToolCallStatus::InProgress + ), + "statuses that represent outstanding work must be treated as busy \ + so a slow tool is never mistaken for a wedged agent" + ); + } + + // And the terminal ones must NOT keep the watchdog disarmed forever, + // otherwise a stall after the last tool completes would go unnoticed — + // which is exactly the E2E claude-round shape. + for status in [ + ToolCallStatus::Completed, + ToolCallStatus::Failed, + ToolCallStatus::Rejected, + ToolCallStatus::Canceled, + ] { + assert!( + !matches!( + status, + ToolCallStatus::Pending | ToolCallStatus::InProgress + ), + "a finished tool call must not count as outstanding work" + ); + } + drop(cx); + } + #[test] - fn first_event_timeout_is_configurable_and_disablable() { + fn silence_timeout_is_configurable_and_disablable() { // Default budget applies when unset. - unsafe { std::env::remove_var("HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS") }; - assert_eq!(first_event_timeout(), Some(Duration::from_secs(120))); + unsafe { std::env::remove_var("HELIX_ACP_SILENCE_TIMEOUT_SECS") }; + assert_eq!(silence_timeout(), Some(Duration::from_secs(120))); - unsafe { std::env::set_var("HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS", "5") }; - assert_eq!(first_event_timeout(), Some(Duration::from_secs(5))); + unsafe { std::env::set_var("HELIX_ACP_SILENCE_TIMEOUT_SECS", "5") }; + assert_eq!(silence_timeout(), Some(Duration::from_secs(5))); // 0 disables the watchdog entirely (escape hatch for debugging). - unsafe { std::env::set_var("HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS", "0") }; - assert_eq!(first_event_timeout(), None); + unsafe { std::env::set_var("HELIX_ACP_SILENCE_TIMEOUT_SECS", "0") }; + assert_eq!(silence_timeout(), None); - unsafe { std::env::remove_var("HELIX_ACP_FIRST_EVENT_TIMEOUT_SECS") }; + unsafe { std::env::remove_var("HELIX_ACP_SILENCE_TIMEOUT_SECS") }; } } From d006aa43324cb4133a8062e366ae5a3476289163 Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 08:26:27 +0100 Subject: [PATCH 2/6] fix(sync): make interrupt an atomic cancel-then-send Root cause of the intermittent E2E Phase 17 failure "interrupt message never delivered", found by running the suite repeatedly. A chat_message with interrupt=true fired TWO independent channels for what is semantically one operation: request_cancel_thread() to the dedicated cancel task, and request_thread_creation() to the creation task. Those tasks are separate spawns with no ordering guarantee, so the send can begin before the cancel is processed -- and the cancel then kills the NEW turn instead of the old one. Observed in a failing run (claude round, thread 5c80128a): 07:18:06 turn_cancelled req=int_...6apt (X) <- correct 07:18:06 message_completed req=int_...6apt usage=null 07:18:06 message_completed req=int_...6ejk (Y) usage=null <- Y just created 07:18:06 "Ignoring stale request_id rebind (mapping previously consumed by completion)" 07:19:06 Phase 17: FAIL -- interrupt message never delivered Y was created and completed in the same second with usage=null: its turn never ran. Every healthy completion on that thread carries real usage. Y's real completion was then rejected as a stale rebind, so the interaction never left the waiting state -- the same off-by-one shape Critical Fix #9 addresses, reached by a different route. The interrupt is now carried on ThreadCreationRequest and the cancel is performed inline by the creation task immediately before the send, so the two steps cannot be reordered. The dedicated cancel task is retained for standalone cancel_current_turn, which is what it exists for: cancelling while the creation loop is blocked awaiting a previous turn. cargo test -p external_websocket_sync: 56 passed. --- .../src/external_websocket_sync.rs | 10 ++++ .../src/thread_service.rs | 24 +++++++++ .../src/websocket_sync.rs | 49 +++++++++++++------ 3 files changed, 68 insertions(+), 15 deletions(-) diff --git a/crates/external_websocket_sync/src/external_websocket_sync.rs b/crates/external_websocket_sync/src/external_websocket_sync.rs index ff61900d15b948..4dce0c50694ca0 100644 --- a/crates/external_websocket_sync/src/external_websocket_sync.rs +++ b/crates/external_websocket_sync/src/external_websocket_sync.rs @@ -108,6 +108,16 @@ pub struct ThreadCreationRequest { /// This allows the NewEntry subscription to fire and sync the user message back to Helix, /// simulating a user typing directly in Zed's agent panel. pub simulate_input: bool, + /// When true, cancel the thread's running turn before sending this message. + /// + /// The cancel is carried ON the creation request, rather than dispatched + /// separately via [`request_cancel_thread`], so that cancel-then-send is + /// performed by a single task in a guaranteed order. Dispatching both + /// independently races: the send can start before the cancel is processed, + /// and the cancel then kills the *new* turn instead of the old one, leaving + /// the new interaction completed-with-nothing and its real response + /// discarded as a stale request_id rebind. + pub interrupt: bool, } /// Request to open existing ACP thread from database and display in UI diff --git a/crates/external_websocket_sync/src/thread_service.rs b/crates/external_websocket_sync/src/thread_service.rs index 1d2f944c2b7805..48c9ed7f4af9dd 100644 --- a/crates/external_websocket_sync/src/thread_service.rs +++ b/crates/external_websocket_sync/src/thread_service.rs @@ -1459,6 +1459,30 @@ pub fn setup_thread_handler( request.request_id ); + // Interrupt = cancel-then-send, performed HERE so the two steps + // cannot be reordered. Doing the cancel on the separate cancel task + // races the send and can cancel the new turn instead of the old one + // (see ThreadCreationRequest::interrupt). + if request.interrupt + && let Some(thread_id) = request.acp_thread_id.as_ref().filter(|id| !id.is_empty()) + { + if let Some(thread) = crate::get_thread(thread_id) { + match cx.update(|cx| thread.update(cx, |t, cx| t.cancel(cx))) { + Ok(_) => { + eprintln!("⚡ [THREAD_SERVICE] Interrupt: cancelled running turn inline before send on {}", thread_id); + log::info!("⚡ [THREAD_SERVICE] Interrupt: cancelled running turn inline before send on {}", thread_id); + } + Err(e) => { + eprintln!("⚠️ [THREAD_SERVICE] Interrupt: failed to cancel {}: {}", thread_id, e); + log::warn!("⚠️ [THREAD_SERVICE] Interrupt: failed to cancel {}: {}", thread_id, e); + } + } + } else { + eprintln!("⚠️ [THREAD_SERVICE] Interrupt: thread {} not in registry, nothing to cancel", thread_id); + log::warn!("⚠️ [THREAD_SERVICE] Interrupt: thread {} not in registry, nothing to cancel", thread_id); + } + } + // Check if this is a follow-up message to existing thread if let Some(existing_thread_id) = &request.acp_thread_id { eprintln!("🔍 [THREAD_SERVICE] Checking for existing thread: '{}'", existing_thread_id); diff --git a/crates/external_websocket_sync/src/websocket_sync.rs b/crates/external_websocket_sync/src/websocket_sync.rs index d5360a40c7dc56..e28f226dd99540 100644 --- a/crates/external_websocket_sync/src/websocket_sync.rs +++ b/crates/external_websocket_sync/src/websocket_sync.rs @@ -429,21 +429,36 @@ impl WebSocketSync { log::info!("💬 [WEBSOCKET-IN] Processing chat_message: acp_thread_id={:?}, request_id={}, message_len={}, interrupt={}", chat_msg.acp_thread_id, chat_msg.request_id, chat_msg.message.len(), chat_msg.interrupt); - // If this is an interrupt message and we have an existing thread, cancel its - // running turn immediately via the dedicated cancel task (which runs independently - // of the sequential callback_rx loop, so it can fire even while the loop is - // blocked awaiting the previous turn's response). - if chat_msg.interrupt { - if let Some(ref thread_id) = chat_msg.acp_thread_id { - if !thread_id.is_empty() { - eprintln!("⚡ [WEBSOCKET-IN] Interrupt flag set — cancelling running turn on thread: {}", thread_id); - log::info!("⚡ [WEBSOCKET-IN] Interrupt flag set — cancelling running turn on thread: {}", thread_id); - if let Err(e) = crate::request_cancel_thread(thread_id.clone()) { - eprintln!("⚠️ [WEBSOCKET-IN] Failed to request cancel for thread {}: {}", thread_id, e); - log::warn!("⚠️ [WEBSOCKET-IN] Failed to request cancel for thread {}: {}", thread_id, e); - } - } - } + // An interrupt is ONE atomic operation: cancel the running turn, then send + // the new message. It is therefore carried on the creation request and + // performed by the creation task itself. + // + // It used to be dispatched separately via request_cancel_thread(), which + // races: the cancel task and the creation task are independent, so the + // send could start before the cancel was processed and the cancel would + // then kill the NEW turn. The new interaction was completed immediately + // with usage=null and no response, and its real completion was later + // rejected as a stale request_id rebind — the interaction never left + // `waiting`. Reproduced as an intermittent E2E Phase 17 failure + // ("interrupt message never delivered"). + // + // The standalone cancel path (cancel_current_turn) still uses the + // dedicated cancel task, which is what that task exists for: cancelling + // while the creation loop is blocked awaiting a previous turn. + let interrupt = chat_msg.interrupt + && chat_msg + .acp_thread_id + .as_ref() + .is_some_and(|id| !id.is_empty()); + if interrupt { + eprintln!( + "⚡ [WEBSOCKET-IN] Interrupt flag set — cancel will run inline, before the send, on thread: {:?}", + chat_msg.acp_thread_id + ); + log::info!( + "⚡ [WEBSOCKET-IN] Interrupt flag set — cancel will run inline, before the send, on thread: {:?}", + chat_msg.acp_thread_id + ); } // Request thread creation via callback @@ -453,6 +468,7 @@ impl WebSocketSync { request_id: chat_msg.request_id.clone(), agent_name: chat_msg.agent_name.clone(), simulate_input: false, + interrupt, }; eprintln!("🎯 [WEBSOCKET-IN] Calling request_thread_creation()..."); @@ -486,6 +502,9 @@ impl WebSocketSync { request_id: chat_msg.request_id.clone(), agent_name: chat_msg.agent_name.clone(), simulate_input: true, + // simulate_user_input models the user typing in Zed; AcpThread::send() + // displaces any running turn on its own, so no explicit cancel. + interrupt: false, }; eprintln!("🎯 [WEBSOCKET-IN] Calling request_thread_creation() with simulate_input=true..."); From bec51df73678437d87cf275a24eaf9a65853ffb1 Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 10:11:04 +0100 Subject: [PATCH 3/6] fix(sync): make interrupt cancels targeted so they cannot kill a newer turn Root cause of the intermittent E2E Phase 17 failure "interrupt message never delivered", found by running the suite repeatedly (~50% failure rate). An interrupt dispatches its cancel out-of-band, to a dedicated task, so it can fire while the sequential creation loop is blocked awaiting the very turn being interrupted. That is necessary -- but it also means the cancel races that loop. If the old turn finishes on its own just as the interrupt arrives, the creation loop starts the NEW turn and the in-flight cancel lands on that instead. Observed in a failing run (claude round, thread 5c80128a): 07:18:06 turn_cancelled req=int_...6apt (X) <- intended 07:18:06 message_completed req=int_...6apt usage=null 07:18:06 message_completed req=int_...6ejk (Y) usage=null <- Y just created 07:18:06 "Ignoring stale request_id rebind (mapping previously consumed by completion)" 07:19:06 Phase 17: FAIL -- interrupt message never delivered Y was created and completed in the same second with usage=null: its turn never ran. Every healthy completion on that thread carries real token usage. Y's real completion was then rejected as a stale rebind, so the interaction never left the waiting state -- the same off-by-one shape Critical Fix #9 addresses, reached by a different route. In production this is a user pressing stop and retyping while a turn streams. The cancel now names the turn it intends to kill: request_cancel_thread() takes an expected_request_id, and the cancel task drops the request as stale if the thread has since moved on. An untargeted cancel (explicit user stop) still cancels whatever is running, by passing None. Rejected alternative: performing the cancel inline in the creation loop. That orders cancel-before-send trivially, but the loop awaits each turn to completion, so an interrupt queued behind a running turn could never cancel it -- trading a race for a deadlock. The dedicated cancel task exists precisely to avoid that, so the fix keeps it and makes it precise instead. cargo test -p external_websocket_sync: 56 passed. cargo check -p zed --features external_websocket_sync: clean. --- .../src/external_websocket_sync.rs | 49 ++++++++++------ .../src/thread_service.rs | 53 +++++++++--------- .../src/websocket_sync.rs | 56 ++++++++----------- 3 files changed, 79 insertions(+), 79 deletions(-) diff --git a/crates/external_websocket_sync/src/external_websocket_sync.rs b/crates/external_websocket_sync/src/external_websocket_sync.rs index 4dce0c50694ca0..7927c42bd31693 100644 --- a/crates/external_websocket_sync/src/external_websocket_sync.rs +++ b/crates/external_websocket_sync/src/external_websocket_sync.rs @@ -87,10 +87,29 @@ static GLOBAL_UI_STATE_QUERY_CALLBACK: parking_lot::Mutex>> = parking_lot::Mutex::new(None); +/// Request to cancel a thread's running turn out-of-band. +#[derive(Clone, Debug)] +pub struct CancelThreadRequest { + pub acp_thread_id: String, + /// The turn this cancel is *intended* for, if the caller knows it. + /// + /// Cancelling is asynchronous and races the sequential creation loop: by the + /// time the cancel task runs, the turn the caller meant to kill may already + /// have finished and a NEW turn may have started on the same thread. Firing + /// blind then cancels the wrong turn — the new interaction completes + /// immediately with no response, and its real completion is later discarded + /// as a stale request_id rebind. + /// + /// When set, the cancel task only cancels if the thread's current request_id + /// still matches, so a stale cancel becomes a no-op instead of collateral + /// damage. `None` means "cancel whatever is running" (explicit user stop). + pub expected_request_id: Option, +} + /// Static global for cancel-thread callback. -/// Receives an acp_thread_id and immediately cancels that thread's running turn, +/// Receives a [`CancelThreadRequest`] and cancels that thread's running turn, /// bypassing the sequential callback_rx loop (which would be blocked awaiting the turn). -static GLOBAL_CANCEL_THREAD_CALLBACK: parking_lot::Mutex>> = +static GLOBAL_CANCEL_THREAD_CALLBACK: parking_lot::Mutex>> = parking_lot::Mutex::new(None); /// Pending UI state queries that arrived before AgentPanel was ready @@ -108,16 +127,6 @@ pub struct ThreadCreationRequest { /// This allows the NewEntry subscription to fire and sync the user message back to Helix, /// simulating a user typing directly in Zed's agent panel. pub simulate_input: bool, - /// When true, cancel the thread's running turn before sending this message. - /// - /// The cancel is carried ON the creation request, rather than dispatched - /// separately via [`request_cancel_thread`], so that cancel-then-send is - /// performed by a single task in a guaranteed order. Dispatching both - /// independently races: the send can start before the cancel is processed, - /// and the cancel then kills the *new* turn instead of the old one, leaving - /// the new interaction completed-with-nothing and its real response - /// discarded as a stale request_id rebind. - pub interrupt: bool, } /// Request to open existing ACP thread from database and display in UI @@ -562,13 +571,19 @@ mod protocol_test; /// This bypasses the sequential callback_rx loop (which blocks waiting for turn /// completion) by routing through a dedicated cancel GPUI task. /// Called when Helix sends a chat_message with interrupt=true. -pub fn request_cancel_thread(acp_thread_id: String) -> Result<()> { - eprintln!("⚡ [CANCEL] request_cancel_thread() called for thread: {}", acp_thread_id); - log::info!("⚡ [CANCEL] request_cancel_thread() called for thread: {}", acp_thread_id); +pub fn request_cancel_thread( + acp_thread_id: String, + expected_request_id: Option, +) -> Result<()> { + eprintln!("⚡ [CANCEL] request_cancel_thread() called for thread: {} (expecting turn {:?})", + acp_thread_id, expected_request_id); + log::info!("⚡ [CANCEL] request_cancel_thread() called for thread: {} (expecting turn {:?})", + acp_thread_id, expected_request_id); let sender = GLOBAL_CANCEL_THREAD_CALLBACK.lock().clone(); if let Some(sender) = sender { - sender.send(acp_thread_id) + sender + .send(CancelThreadRequest { acp_thread_id, expected_request_id }) .map_err(|_| anyhow::anyhow!("Failed to send cancel request"))?; Ok(()) } else { @@ -581,7 +596,7 @@ pub fn request_cancel_thread(acp_thread_id: String) -> Result<()> { } /// Initialize the global cancel-thread callback (called from thread_service). -pub fn init_cancel_thread_callback(sender: mpsc::UnboundedSender) { +pub fn init_cancel_thread_callback(sender: mpsc::UnboundedSender) { eprintln!("🔧 [CANCEL] init_cancel_thread_callback() called - registering global callback"); log::info!("🔧 [CANCEL] init_cancel_thread_callback() called - registering global callback"); *GLOBAL_CANCEL_THREAD_CALLBACK.lock() = Some(sender); diff --git a/crates/external_websocket_sync/src/thread_service.rs b/crates/external_websocket_sync/src/thread_service.rs index 48c9ed7f4af9dd..f4f7e14055c3dd 100644 --- a/crates/external_websocket_sync/src/thread_service.rs +++ b/crates/external_websocket_sync/src/thread_service.rs @@ -1413,14 +1413,35 @@ pub fn setup_thread_handler( // Spawn dedicated cancel task — runs independently of the callback_rx loop so it // can cancel a running turn even while callback_rx.recv().await is blocked. - let (cancel_tx, mut cancel_rx) = mpsc::unbounded_channel::(); + let (cancel_tx, mut cancel_rx) = mpsc::unbounded_channel::(); crate::init_cancel_thread_callback(cancel_tx); cx.spawn(async move |cx| { eprintln!("⚡ [CANCEL_TASK] Cancel task started, waiting for cancel requests..."); log::info!("⚡ [CANCEL_TASK] Cancel task started, waiting for cancel requests..."); - while let Some(acp_thread_id) = cancel_rx.recv().await { - eprintln!("⚡ [CANCEL_TASK] Received cancel request for thread: {}", acp_thread_id); - log::info!("⚡ [CANCEL_TASK] Received cancel request for thread: {}", acp_thread_id); + while let Some(req) = cancel_rx.recv().await { + let acp_thread_id = req.acp_thread_id; + eprintln!("⚡ [CANCEL_TASK] Received cancel request for thread: {} (expecting turn {:?})", + acp_thread_id, req.expected_request_id); + log::info!("⚡ [CANCEL_TASK] Received cancel request for thread: {} (expecting turn {:?})", + acp_thread_id, req.expected_request_id); + + // Targeted cancel: if the caller named the turn it meant to kill and + // the thread has since moved on to a different one, this cancel is + // stale. Firing it would kill the NEW turn — which completes with no + // response and whose real completion is then discarded as a stale + // request_id rebind (observed as intermittent E2E Phase 17 failures, + // "interrupt message never delivered"). Drop it instead. + if let Some(expected) = req.expected_request_id.as_deref() { + let current = crate::get_thread_request_id(&acp_thread_id).unwrap_or_default(); + if current != expected { + eprintln!("🛡️ [CANCEL_TASK] Stale cancel ignored on {}: intended turn {} but thread is now on {} — cancelling would kill the newer turn", + acp_thread_id, expected, current); + log::warn!("🛡️ [CANCEL_TASK] Stale cancel ignored on {}: intended turn {} but thread is now on {} — cancelling would kill the newer turn", + acp_thread_id, expected, current); + continue; + } + } + if let Some(thread) = crate::get_thread(&acp_thread_id) { let result = cx.update(|cx| { thread.update(cx, |t, cx| { t.cancel(cx) }) @@ -1459,30 +1480,6 @@ pub fn setup_thread_handler( request.request_id ); - // Interrupt = cancel-then-send, performed HERE so the two steps - // cannot be reordered. Doing the cancel on the separate cancel task - // races the send and can cancel the new turn instead of the old one - // (see ThreadCreationRequest::interrupt). - if request.interrupt - && let Some(thread_id) = request.acp_thread_id.as_ref().filter(|id| !id.is_empty()) - { - if let Some(thread) = crate::get_thread(thread_id) { - match cx.update(|cx| thread.update(cx, |t, cx| t.cancel(cx))) { - Ok(_) => { - eprintln!("⚡ [THREAD_SERVICE] Interrupt: cancelled running turn inline before send on {}", thread_id); - log::info!("⚡ [THREAD_SERVICE] Interrupt: cancelled running turn inline before send on {}", thread_id); - } - Err(e) => { - eprintln!("⚠️ [THREAD_SERVICE] Interrupt: failed to cancel {}: {}", thread_id, e); - log::warn!("⚠️ [THREAD_SERVICE] Interrupt: failed to cancel {}: {}", thread_id, e); - } - } - } else { - eprintln!("⚠️ [THREAD_SERVICE] Interrupt: thread {} not in registry, nothing to cancel", thread_id); - log::warn!("⚠️ [THREAD_SERVICE] Interrupt: thread {} not in registry, nothing to cancel", thread_id); - } - } - // Check if this is a follow-up message to existing thread if let Some(existing_thread_id) = &request.acp_thread_id { eprintln!("🔍 [THREAD_SERVICE] Checking for existing thread: '{}'", existing_thread_id); diff --git a/crates/external_websocket_sync/src/websocket_sync.rs b/crates/external_websocket_sync/src/websocket_sync.rs index e28f226dd99540..7b72a4cb055f3f 100644 --- a/crates/external_websocket_sync/src/websocket_sync.rs +++ b/crates/external_websocket_sync/src/websocket_sync.rs @@ -429,36 +429,28 @@ impl WebSocketSync { log::info!("💬 [WEBSOCKET-IN] Processing chat_message: acp_thread_id={:?}, request_id={}, message_len={}, interrupt={}", chat_msg.acp_thread_id, chat_msg.request_id, chat_msg.message.len(), chat_msg.interrupt); - // An interrupt is ONE atomic operation: cancel the running turn, then send - // the new message. It is therefore carried on the creation request and - // performed by the creation task itself. - // - // It used to be dispatched separately via request_cancel_thread(), which - // races: the cancel task and the creation task are independent, so the - // send could start before the cancel was processed and the cancel would - // then kill the NEW turn. The new interaction was completed immediately - // with usage=null and no response, and its real completion was later - // rejected as a stale request_id rebind — the interaction never left - // `waiting`. Reproduced as an intermittent E2E Phase 17 failure - // ("interrupt message never delivered"). - // - // The standalone cancel path (cancel_current_turn) still uses the - // dedicated cancel task, which is what that task exists for: cancelling - // while the creation loop is blocked awaiting a previous turn. - let interrupt = chat_msg.interrupt - && chat_msg - .acp_thread_id - .as_ref() - .is_some_and(|id| !id.is_empty()); - if interrupt { - eprintln!( - "⚡ [WEBSOCKET-IN] Interrupt flag set — cancel will run inline, before the send, on thread: {:?}", - chat_msg.acp_thread_id - ); - log::info!( - "⚡ [WEBSOCKET-IN] Interrupt flag set — cancel will run inline, before the send, on thread: {:?}", - chat_msg.acp_thread_id - ); + // If this is an interrupt message and we have an existing thread, cancel its + // running turn immediately via the dedicated cancel task (which runs independently + // of the sequential callback_rx loop, so it can fire even while the loop is + // blocked awaiting the previous turn's response). + if chat_msg.interrupt { + if let Some(ref thread_id) = chat_msg.acp_thread_id { + if !thread_id.is_empty() { + // Name the turn we mean to interrupt. The cancel is delivered + // out-of-band (so it can fire while the creation loop is blocked + // awaiting this very turn), which means it races that loop: if + // the turn finishes on its own first and the next one starts, + // an untargeted cancel would kill the NEW turn instead. Passing + // the current request_id makes a stale cancel a no-op. + let target = crate::get_thread_request_id(thread_id); + eprintln!("⚡ [WEBSOCKET-IN] Interrupt flag set — cancelling turn {:?} on thread: {}", target, thread_id); + log::info!("⚡ [WEBSOCKET-IN] Interrupt flag set — cancelling turn {:?} on thread: {}", target, thread_id); + if let Err(e) = crate::request_cancel_thread(thread_id.clone(), target) { + eprintln!("⚠️ [WEBSOCKET-IN] Failed to request cancel for thread {}: {}", thread_id, e); + log::warn!("⚠️ [WEBSOCKET-IN] Failed to request cancel for thread {}: {}", thread_id, e); + } + } + } } // Request thread creation via callback @@ -468,7 +460,6 @@ impl WebSocketSync { request_id: chat_msg.request_id.clone(), agent_name: chat_msg.agent_name.clone(), simulate_input: false, - interrupt, }; eprintln!("🎯 [WEBSOCKET-IN] Calling request_thread_creation()..."); @@ -502,9 +493,6 @@ impl WebSocketSync { request_id: chat_msg.request_id.clone(), agent_name: chat_msg.agent_name.clone(), simulate_input: true, - // simulate_user_input models the user typing in Zed; AcpThread::send() - // displaces any running turn on its own, so no explicit cancel. - interrupt: false, }; eprintln!("🎯 [WEBSOCKET-IN] Calling request_thread_creation() with simulate_input=true..."); From 8adcd0f44a1939e4884e598caf041a4616ce3d53 Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 10:19:04 +0100 Subject: [PATCH 4/6] fix(e2e): repair diagnostic counts that broke every cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 'grep -c' PRINTS the count and EXITS NON-ZERO when the count is zero, so $(grep -c ... || echo "0") produces the two-line string "0\n0". Every subsequent [ "$X" -gt 0 ] then died with: /test/run_e2e.sh: line 58: [: 0 0: integer expression expected visible at the end of every run. Consequence: the ACP_SPAWN / ACP_DEDUP diagnostic block never rendered, so agent-spawn and connection-dedup behaviour — exactly what is needed to diagnose a claude-round failure — was silently withheld. Fallback moved onto the assignment. --- crates/external_websocket_sync/e2e-test/run_e2e.sh | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/external_websocket_sync/e2e-test/run_e2e.sh b/crates/external_websocket_sync/e2e-test/run_e2e.sh index a793cf34f5755c..5c870700bb2f70 100755 --- a/crates/external_websocket_sync/e2e-test/run_e2e.sh +++ b/crates/external_websocket_sync/e2e-test/run_e2e.sh @@ -44,7 +44,11 @@ cleanup() { # Dump Zed errors/panics (full log available at ZED_LOG_FILE) if [ -f "${ZED_LOG_FILE:-}" ]; then - ZED_ERRORS=$(grep -ciE "panic|error|fatal" "$ZED_LOG_FILE" 2>/dev/null || echo "0") + # NB: `grep -c` PRINTS the count and EXITS NON-ZERO when the count is 0, so + # `$(grep -c ... || echo 0)` yields the two-line string "0\n0" and every + # subsequent `[ "$X" -gt 0 ]` dies with "integer expression expected". + # Put the fallback on the assignment, not inside the substitution. + ZED_ERRORS=$(grep -ciE "panic|error|fatal" "$ZED_LOG_FILE" 2>/dev/null) || ZED_ERRORS=0 if [ "$ZED_ERRORS" -gt 0 ]; then echo "" echo "==================================================" From 01cd0bb3da6af14652fca3e5464d9e922a9ff290 Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 10:24:45 +0100 Subject: [PATCH 5/6] feat(e2e): surface turn-lifecycle / cancel ordering on failure A failing round previously reported only "phase N timed out". That is not enough to attribute the failure, and the ordering that produced it is invisible from the Helix side (Helix sees completions, not the sequence inside Zed that caused them). Diagnosing the Phase 17 interrupt race required reading the Helix log by hand and inferring Zed's behaviour, because none of it was surfaced. Adds a TURN LIFECYCLE / CANCEL ORDERING block to cleanup, alongside the existing ACP_SPAWN/ACP_DEDUP one, covering the three shapes seen so far: - cancel landing on the wrong turn -> CANCEL_TASK vs THREAD_SERVICE ordering - a stale cancel correctly dropped -> "Stale cancel ignored" - agent accepted a prompt then died -> helix_silent_prompt_wedge Also applies the grep -c fix to ACP_LINES, which had the same defect already fixed for ZED_ERRORS: grep -c prints the count AND exits non-zero at zero matches, so the `|| echo 0` fallback produced "0\n0" and the guard died with "integer expression expected" -- silently suppressing the ACP diagnostics. --- .../e2e-test/run_e2e.sh | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/crates/external_websocket_sync/e2e-test/run_e2e.sh b/crates/external_websocket_sync/e2e-test/run_e2e.sh index 5c870700bb2f70..3ac2163befe7c7 100755 --- a/crates/external_websocket_sync/e2e-test/run_e2e.sh +++ b/crates/external_websocket_sync/e2e-test/run_e2e.sh @@ -58,7 +58,7 @@ cleanup() { echo " (full log: $ZED_LOG_FILE)" fi # ACP_SPAWN/ACP_DEDUP are at log::info level — surface them explicitly - ACP_LINES=$(grep -cE "ACP_SPAWN|ACP_DEDUP" "$ZED_LOG_FILE" 2>/dev/null || echo "0") + ACP_LINES=$(grep -cE "ACP_SPAWN|ACP_DEDUP" "$ZED_LOG_FILE" 2>/dev/null) || ACP_LINES=0 if [ "$ACP_LINES" -gt 0 ]; then echo "" echo "==================================================" @@ -66,6 +66,26 @@ cleanup() { echo "==================================================" grep -E "ACP_SPAWN|ACP_DEDUP" "$ZED_LOG_FILE" || true fi + + # Turn lifecycle: cancel / interrupt / silence-watchdog decisions. + # + # These are the events needed to tell the three failure shapes apart when + # a round fails, and reading them from the Helix side alone is impossible + # (Helix sees completions, not the ordering that produced them): + # - cancel landing on the wrong turn -> CANCEL_TASK vs THREAD_SERVICE order + # - a stale cancel correctly dropped -> "Stale cancel ignored" + # - agent accepted a prompt then died -> helix_silent_prompt_wedge + # Without this block the harness reported only "phase N timed out", which + # is not enough to attribute a failure. + LIFECYCLE_RE="CANCEL_TASK|Interrupt flag set|Stale cancel ignored|helix_silent_prompt_wedge|THREAD_SERVICE\] (Sending follow-up|Updated request_id|Sending to existing)" + LIFECYCLE_LINES=$(grep -cE "$LIFECYCLE_RE" "$ZED_LOG_FILE" 2>/dev/null) || LIFECYCLE_LINES=0 + if [ "$LIFECYCLE_LINES" -gt 0 ]; then + echo "" + echo "==================================================" + echo " TURN LIFECYCLE / CANCEL ORDERING ($LIFECYCLE_LINES lines)" + echo "==================================================" + grep -E "$LIFECYCLE_RE" "$ZED_LOG_FILE" | tail -60 || true + fi # Persist the full zed log into the mounted screenshots dir for offline inspection if [ -d "$SCREENSHOT_DIR" ]; then cp "$ZED_LOG_FILE" "$SCREENSHOT_DIR/zed-e2e.log" 2>/dev/null || true From f5c77cf39a8cd301e3f9b595a7a76a8dc062190d Mon Sep 17 00:00:00 2001 From: Luke Marsden Date: Thu, 30 Jul 2026 10:26:11 +0100 Subject: [PATCH 6/6] fix(e2e): stop the round filter discarding real Helix interaction ids MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The message_completed handler ignores cross-round completions with two checks: the request id must contain the current agent name, and the thread must belong to the current round. The first is a heuristic that only holds for the synthetic ids the harness mints itself (req-phaseN-). The production-queue phases (16/17) drive REAL Helix interactions, whose ids are ULIDs (int_01kyry6ejk...) containing no agent name. So live completions for the current round were rejected and logged as: [claude] FILTERED completion (wrong agent): req=int_01kyry6ejk... That is worse than useless during triage — it asserts the opposite of what happened. It cost real time chasing a phantom routing bug while Phase 17 was actually hitting a cancel race. Helix-minted ids are round-safe regardless (each round gets fresh sessions and threads), so they now defer to the precise thread check that follows. go build: clean. (main.go has pre-existing gofmt drift in the phase9/10 struct fields, left alone to keep this diff focused.) --- .../e2e-test/helix-ws-test-server/main.go | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/crates/external_websocket_sync/e2e-test/helix-ws-test-server/main.go b/crates/external_websocket_sync/e2e-test/helix-ws-test-server/main.go index 35770037109422..46b6bbc9ac8b1f 100644 --- a/crates/external_websocket_sync/e2e-test/helix-ws-test-server/main.go +++ b/crates/external_websocket_sync/e2e-test/helix-ws-test-server/main.go @@ -387,7 +387,20 @@ func (d *testDriver) syncEventCallback(sessionID string, syncMsg *types.SyncMess // Ignore completions from previous rounds. Two checks: // 1. Request ID must contain the current agent name // 2. Thread ID must belong to the current round - if !strings.Contains(requestID, agentName) { + // + // Check 1 is a heuristic that only holds for the synthetic ids this + // harness mints itself (req-phaseN-). The production-queue phases + // (16/17) drive real Helix interactions, whose ids are ULIDs like + // int_01kyry6ejk... and contain no agent name — so the heuristic rejects + // LIVE completions for the current round and logs them as "wrong agent". + // That is worse than useless during triage: it asserts the opposite of + // what happened, and cost real time chasing a phantom routing bug when + // Phase 17 was actually hitting a cancel race. + // + // Helix-minted ids are round-safe anyway (each round gets fresh sessions + // and threads), so defer to the precise thread check below for them. + isHelixInteractionID := strings.HasPrefix(requestID, "int_") + if !isHelixInteractionID && !strings.Contains(requestID, agentName) { d.mu.Unlock() log.Printf("[%s] FILTERED completion (wrong agent): req=%s thread=%s", agentName, requestID, truncate(acpThreadID, 12))