From 3892450fa1c59cf36bcb3ba6228c2bdd8ea1480f Mon Sep 17 00:00:00 2001 From: amazd Date: Wed, 10 Jun 2026 12:55:16 -0400 Subject: [PATCH] feat(cli): cancel in-flight model requests on turn interruption Wire the runtime's TurnInterruptSignal through the CLI turn lifecycle: - AnthropicRuntimeClient races the streaming request against the interrupt flag with tokio::select!; when the flag trips, the request future is dropped, aborting the in-flight HTTP stream for every provider variant. - BuiltRuntime installs the shared signal into both the conversation loop and the API client. - The per-turn Ctrl+C monitor now interrupts the whole turn instead of only aborting running hooks, so Ctrl+C mid-turn returns to the REPL prompt with partial output instead of leaving the request running. - The REPL spinner reports an interrupted turn distinctly from a completed one. Part of the Esc-to-interrupt work (#3196); the Esc key listener lands separately. --- rust/crates/rusty-claude-cli/src/main.rs | 99 ++++++++++++++++++------ 1 file changed, 75 insertions(+), 24 deletions(-) diff --git a/rust/crates/rusty-claude-cli/src/main.rs b/rust/crates/rusty-claude-cli/src/main.rs index 665ce632..97d2090d 100644 --- a/rust/crates/rusty-claude-cli/src/main.rs +++ b/rust/crates/rusty-claude-cli/src/main.rs @@ -7194,6 +7194,18 @@ impl BuiltRuntime { self } + fn with_turn_interrupt_signal(mut self, signal: runtime::TurnInterruptSignal) -> Self { + let mut runtime = self + .runtime + .take() + .expect("runtime should exist before installing turn interrupt signal"); + runtime + .api_client_mut() + .set_turn_interrupt_signal(signal.clone()); + self.runtime = Some(runtime.with_turn_interrupt_signal(signal)); + self + } + fn shutdown_plugins(&mut self) -> Result<(), Box> { if self.plugins_active { self.plugin_registry.shutdown()?; @@ -7576,7 +7588,10 @@ struct HookAbortMonitor { } impl HookAbortMonitor { - fn spawn(abort_signal: runtime::HookAbortSignal) -> Self { + fn spawn( + abort_signal: runtime::HookAbortSignal, + turn_interrupt_signal: runtime::TurnInterruptSignal, + ) -> Self { Self::spawn_with_waiter(abort_signal, move |stop_rx, abort_signal| { let Ok(runtime) = tokio::runtime::Builder::new_current_thread() .enable_all() @@ -7593,7 +7608,11 @@ impl HookAbortMonitor { tokio::select! { result = tokio::signal::ctrl_c() => { if result.is_ok() { + // Ctrl+C stops the whole turn, not just running + // hooks: the conversation loop and the streaming + // client both poll the turn interrupt signal. abort_signal.abort(); + turn_interrupt_signal.interrupt(); } } _ = wait_for_stop => {} @@ -7725,6 +7744,7 @@ impl LiveCli { emit_output: bool, ) -> Result<(BuiltRuntime, HookAbortMonitor), Box> { let hook_abort_signal = runtime::HookAbortSignal::new(); + let turn_interrupt_signal = runtime::TurnInterruptSignal::new(); let runtime = build_runtime( self.runtime.session().clone(), &self.session.id, @@ -7736,8 +7756,9 @@ impl LiveCli { self.permission_mode, None, )? - .with_hook_abort_signal(hook_abort_signal.clone()); - let hook_abort_monitor = HookAbortMonitor::spawn(hook_abort_signal); + .with_hook_abort_signal(hook_abort_signal.clone()) + .with_turn_interrupt_signal(turn_interrupt_signal.clone()); + let hook_abort_monitor = HookAbortMonitor::spawn(hook_abort_signal, turn_interrupt_signal); Ok((runtime, hook_abort_monitor)) } @@ -7764,7 +7785,11 @@ impl LiveCli { Ok(summary) => { self.replace_runtime(runtime)?; spinner.finish( - "✨ Done", + if summary.interrupted { + "⏹ Interrupted" + } else { + "✨ Done" + }, TerminalRenderer::new().color_theme(), &mut stdout, )?; @@ -12542,6 +12567,7 @@ struct AnthropicRuntimeClient { tool_registry: GlobalToolRegistry, progress_reporter: Option, reasoning_effort: Option, + turn_interrupt_signal: Option, } impl AnthropicRuntimeClient { @@ -12607,12 +12633,17 @@ impl AnthropicRuntimeClient { tool_registry, progress_reporter, reasoning_effort: None, + turn_interrupt_signal: None, }) } fn set_reasoning_effort(&mut self, effort: Option) { self.reasoning_effort = effort; } + + fn set_turn_interrupt_signal(&mut self, signal: runtime::TurnInterruptSignal) { + self.turn_interrupt_signal = Some(signal); + } } fn resolve_cli_auth_source() -> Result> { @@ -12646,30 +12677,50 @@ impl ApiClient for AnthropicRuntimeClient { }; self.runtime.block_on(async { - // When resuming after tool execution, apply a stall timeout on the - // first stream event. If the model does not respond within the - // deadline we drop the stalled connection and re-send the request as - // a continuation nudge (one retry only). - let max_attempts: usize = if is_post_tool { 2 } else { 1 }; + let request_loop = async { + // When resuming after tool execution, apply a stall timeout on the + // first stream event. If the model does not respond within the + // deadline we drop the stalled connection and re-send the request as + // a continuation nudge (one retry only). + let max_attempts: usize = if is_post_tool { 2 } else { 1 }; - for attempt in 1..=max_attempts { - let result = self - .consume_stream(&message_request, is_post_tool && attempt == 1) - .await; - match result { - Ok(events) => return Ok(events), - Err(error) - if error.to_string().contains("post-tool stall") - && attempt < max_attempts => - { - // Stalled after tool completion — nudge the model by - // re-sending the same request. + for attempt in 1..=max_attempts { + let result = self + .consume_stream(&message_request, is_post_tool && attempt == 1) + .await; + match result { + Ok(events) => return Ok(events), + Err(error) + if error.to_string().contains("post-tool stall") + && attempt < max_attempts => + { + // Stalled after tool completion — nudge the model by + // re-sending the same request. + } + Err(error) => return Err(error), } - Err(error) => return Err(error), + } + + Err(RuntimeError::new("post-tool continuation nudge exhausted")) + }; + + let Some(interrupt) = self.turn_interrupt_signal.clone() else { + return request_loop.await; + }; + + tokio::select! { + result = request_loop => result, + () = async { + while !interrupt.is_interrupted() { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + } => { + // Dropping the request future aborts the in-flight HTTP + // stream. The conversation loop sees the interrupt flag and + // reports this as an interruption rather than a failure. + Err(RuntimeError::new("request interrupted by user")) } } - - Err(RuntimeError::new("post-tool continuation nudge exhausted")) }) } }