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.
This commit is contained in:
amazd 2026-06-10 12:55:16 -04:00
parent 5a4432bfd6
commit 3892450fa1
1 changed files with 75 additions and 24 deletions

View File

@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<InternalPromptProgressReporter>,
reasoning_effort: Option<String>,
turn_interrupt_signal: Option<runtime::TurnInterruptSignal>,
}
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<String>) {
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<AuthSource, Box<dyn std::error::Error>> {
@ -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"))
})
}
}