diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 3570491d1a..b2718e3752 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -327,6 +327,15 @@ impl App { self.chat_widget.prepare_local_op_submission(&op); self.submit_active_thread_op(app_server, op).await?; } + AppEvent::TurnInterruptCompleted { thread_id, result } => { + if let Err(err) = result { + tracing::warn!(%thread_id, "turn interrupt request failed: {err}"); + if self.active_thread_id == Some(thread_id) { + self.chat_widget + .add_error_message(format!("Interrupt failed: {err}")); + } + } + } AppEvent::RestoreCancelledTurn(prompt) => { self.apply_cancelled_turn_edit(prompt); } diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index eb5e10c50f..2932f27451 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -5227,6 +5227,60 @@ async fn interrupt_without_active_turn_is_treated_as_handled() { .await; } +#[tokio::test] +async fn active_turn_interrupt_request_does_not_block_app_loop() { + Box::pin(async { + let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; + let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker( + app.chat_widget.config_ref(), + )) + .await + .expect("embedded app server"); + let started = app_server + .start_thread(app.chat_widget.config_ref()) + .await + .expect("thread/start should succeed"); + let thread_id = started.session.thread_id; + app.enqueue_primary_thread_session(started.session, started.turns) + .await + .expect("primary thread should be registered"); + while app_event_rx.try_recv().is_ok() {} + { + let channel = app + .thread_event_channels + .get(&thread_id) + .expect("primary thread event channel"); + let mut store = channel.store.lock().await; + store.active_turn_id = Some("turn-pending".to_string()); + } + let op = AppCommand::interrupt(); + + let handled = time::timeout( + std::time::Duration::from_millis(100), + app.try_submit_active_thread_op_via_app_server(&mut app_server, thread_id, &op), + ) + .await + .expect("interrupt submission should not wait for RPC completion") + .expect("interrupt submission should not fail"); + + assert_eq!(handled, true); + let event = time::timeout(std::time::Duration::from_secs(1), app_event_rx.recv()) + .await + .expect("background interrupt should report completion") + .expect("app event channel should remain open"); + let AppEvent::TurnInterruptCompleted { + thread_id: observed, + result, + } = event + else { + panic!("expected TurnInterruptCompleted, got {event:?}"); + }; + assert_eq!(observed, thread_id); + assert!(result.is_err()); + }) + .await; +} + #[tokio::test] async fn override_turn_context_sends_thread_settings_update() { Box::pin(async { diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 1bd28d5f65..8cec19380a 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -500,7 +500,15 @@ impl App { match op { AppCommand::Interrupt { .. } => { if let Some(turn_id) = self.active_turn_id_for_thread(thread_id).await { - app_server.turn_interrupt(thread_id, turn_id).await?; + let interrupt = app_server.spawn_turn_interrupt(thread_id, turn_id); + let app_event_tx = self.app_event_tx.clone(); + tokio::spawn(async move { + let result = match interrupt.await { + Ok(result) => result.map_err(|err| err.to_string()), + Err(err) => Err(format!("turn/interrupt task failed: {err}")), + }; + app_event_tx.send(AppEvent::TurnInterruptCompleted { thread_id, result }); + }); } else { app_server.startup_interrupt(thread_id).await?; } diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 228ee0c417..d578cf4566 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -234,6 +234,12 @@ pub(crate) enum AppEvent { /// bubbling channels through layers of widgets. CodexOp(AppCommand), + /// Result of a background turn interrupt request. + TurnInterruptCompleted { + thread_id: ThreadId, + result: Result<(), String>, + }, + /// Restore an output-free interrupted turn into the composer and roll it back. RestoreCancelledTurn(UserMessage), diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 2290640ab4..473da482be 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -737,18 +737,20 @@ impl AppServerSession { turn_id: String, ) -> Result<()> { let request_id = self.next_request_id(); - let _: TurnInterruptResponse = self - .client - .request_typed(ClientRequest::TurnInterrupt { - request_id, - params: TurnInterruptParams { - thread_id: thread_id.to_string(), - turn_id, - }, - }) - .await - .wrap_err("turn/interrupt failed in TUI")?; - Ok(()) + let request_handle = self.request_handle(); + send_turn_interrupt_request(request_handle, request_id, thread_id, turn_id).await + } + + pub(crate) fn spawn_turn_interrupt( + &mut self, + thread_id: ThreadId, + turn_id: String, + ) -> tokio::task::JoinHandle> { + let request_id = self.next_request_id(); + let request_handle = self.request_handle(); + tokio::spawn(async move { + send_turn_interrupt_request(request_handle, request_id, thread_id, turn_id).await + }) } pub(crate) async fn startup_interrupt(&mut self, thread_id: ThreadId) -> Result<()> { @@ -1134,6 +1136,25 @@ impl AppServerSession { } } +async fn send_turn_interrupt_request( + request_handle: AppServerRequestHandle, + request_id: RequestId, + thread_id: ThreadId, + turn_id: String, +) -> Result<()> { + let _: TurnInterruptResponse = request_handle + .request_typed(ClientRequest::TurnInterrupt { + request_id, + params: TurnInterruptParams { + thread_id: thread_id.to_string(), + turn_id, + }, + }) + .await + .wrap_err("turn/interrupt failed in TUI")?; + Ok(()) +} + pub(crate) async fn start_thread_with_request_handle( request_handle: AppServerRequestHandle, config: Config,