From 4cd85b28d2713289d5940b6cb967894277a17419 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Thu, 16 Apr 2026 00:03:50 -0700 Subject: [PATCH] Fix MCP startup cancellation through app server (#18078) Addresses https://github.com/openai/codex/issues/17143 Problem: TUI interrupts without an active turn stopped cancelling slow MCP startup after routing through the app-server APIs. Solution: Route no-active-turn interrupts through app-server as startup cancels, acknowledge them immediately, and emit cancelled MCP startup updates. Testing: I manually confirmed that MCP cancellation didn't work prior to this PR and works after the fix was in place. --- .../app-server/src/codex_message_processor.rs | 51 +++++++++++++++---- .../codex-mcp/src/mcp_connection_manager.rs | 5 +- codex-rs/tui/src/app.rs | 18 +++++-- codex-rs/tui/src/app_server_session.rs | 4 ++ 4 files changed, 61 insertions(+), 17 deletions(-) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 94c1e5664..2585383cd 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -186,6 +186,7 @@ use codex_app_server_protocol::ThreadUnsubscribeStatus; use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnError; use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; @@ -7559,9 +7560,12 @@ impl CodexMessageProcessor { async fn turn_interrupt(&self, request_id: ConnectionRequestId, params: TurnInterruptParams) { let TurnInterruptParams { thread_id, turn_id } = params; - self.outgoing - .record_request_turn_id(&request_id, &turn_id) - .await; + let is_startup_interrupt = turn_id.is_empty(); + if !is_startup_interrupt { + self.outgoing + .record_request_turn_id(&request_id, &turn_id) + .await; + } let (thread_uuid, thread) = match self.load_thread(&thread_id).await { Ok(v) => v, @@ -7571,21 +7575,48 @@ impl CodexMessageProcessor { } }; - let request = request_id.clone(); - - // Record the pending interrupt so we can reply when TurnAborted arrives. - { + // Record turn interrupts so we can reply when TurnAborted arrives. Startup + // interrupts do not have a turn and are acknowledged after submission. + if !is_startup_interrupt { let thread_state = self.thread_state_manager.thread_state(thread_uuid).await; let mut thread_state = thread_state.lock().await; thread_state .pending_interrupts - .push((request, ApiVersion::V2)); + .push((request_id.clone(), ApiVersion::V2)); } - // Submit the interrupt; we'll respond upon TurnAborted. - let _ = self + // Submit the interrupt. Turn interrupts respond upon TurnAborted; startup + // interrupts respond here because startup cancellation has no turn event. + let submit_result = self .submit_core_op(&request_id, thread.as_ref(), Op::Interrupt) .await; + match submit_result { + Ok(_) if is_startup_interrupt => { + self.outgoing + .send_response(request_id, TurnInterruptResponse {}) + .await; + } + Ok(_) => {} + Err(err) => { + if !is_startup_interrupt { + let thread_state = self.thread_state_manager.thread_state(thread_uuid).await; + let mut thread_state = thread_state.lock().await; + thread_state + .pending_interrupts + .retain(|(pending_request_id, _)| pending_request_id != &request_id); + } + let interrupt_target = if is_startup_interrupt { + "startup" + } else { + "turn" + }; + self.send_internal_error( + request_id, + format!("failed to interrupt {interrupt_target}: {err}"), + ) + .await; + } + } } async fn ensure_conversation_listener( diff --git a/codex-rs/codex-mcp/src/mcp_connection_manager.rs b/codex-rs/codex-mcp/src/mcp_connection_manager.rs index 1542c7b30..d1696b593 100644 --- a/codex-rs/codex-mcp/src/mcp_connection_manager.rs +++ b/codex-rs/codex-mcp/src/mcp_connection_manager.rs @@ -758,12 +758,13 @@ impl McpConnectionManager { let submit_id = startup_submit_id.clone(); let auth_entry = auth_entries.get(&server_name).cloned(); join_set.spawn(async move { - let outcome = async_managed_client.client().await; + let mut outcome = async_managed_client.client().await; if cancel_token.is_cancelled() { - return (server_name, Err(StartupOutcomeError::Cancelled)); + outcome = Err(StartupOutcomeError::Cancelled); } let status = match &outcome { Ok(_) => McpStartupStatus::Ready, + Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled, Err(error) => { let error_str = mcp_init_error_display( server_name.as_str(), diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 03a99e3aa..a60856b9c 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -2378,10 +2378,11 @@ impl App { ) -> Result { match op.view() { AppCommandView::Interrupt => { - let Some(turn_id) = self.active_turn_id_for_thread(thread_id).await else { - return Ok(true); - }; - app_server.turn_interrupt(thread_id, turn_id).await?; + if let Some(turn_id) = self.active_turn_id_for_thread(thread_id).await { + app_server.turn_interrupt(thread_id, turn_id).await?; + } else { + app_server.startup_interrupt(thread_id).await?; + } Ok(true) } AppCommandView::UserTurn { @@ -11410,11 +11411,18 @@ guardian_approval = true #[tokio::test] async fn interrupt_without_active_turn_is_treated_as_handled() { let mut app = make_test_app().await; - let thread_id = ThreadId::new(); let mut app_server = 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"); let op = AppCommand::interrupt(); let handled = app diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 47a06738e..5e11a5334 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -476,6 +476,10 @@ impl AppServerSession { Ok(()) } + pub(crate) async fn startup_interrupt(&mut self, thread_id: ThreadId) -> Result<()> { + self.turn_interrupt(thread_id, String::new()).await + } + pub(crate) async fn turn_steer( &mut self, thread_id: ThreadId,