From d06c7ac055920c7cb140c25ebda3f3db20197b45 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 29 Jul 2026 03:44:51 +0000 Subject: [PATCH] Clean up side conversations in the background (#35887) ## Why Switching away from a side conversation should not wait for its interrupt and unsubscribe requests to finish. ## What changed - Remove a side conversation from local TUI state immediately when switching threads, then interrupt and unsubscribe it in a background task. - Retry interruption with the active turn ID when the original turn races with cleanup. - Ignore late notifications and reject server requests for abandoned side conversations. ## Testing - Added coverage that background cleanup removes local state and ignores late events and approval requests. GitOrigin-RevId: 7a1d1c34f16ab023d1d1e235451b55404731190c --- codex-rs/tui/src/app.rs | 3 + codex-rs/tui/src/app/app_server_events.rs | 17 ++++- codex-rs/tui/src/app/side.rs | 77 ++++++++++++++++++++--- codex-rs/tui/src/app/test_support.rs | 1 + codex-rs/tui/src/app/tests.rs | 58 +++++++++++++++++ codex-rs/tui/src/app/thread_routing.rs | 3 + 6 files changed, 148 insertions(+), 11 deletions(-) diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 401510d931..b201d8f020 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -180,6 +180,7 @@ use ratatui::widgets::Paragraph; use ratatui::widgets::Wrap; use std::collections::BTreeMap; use std::collections::HashMap; +use std::collections::HashSet; use std::collections::VecDeque; use std::io::Write; use std::path::Path; @@ -575,6 +576,7 @@ pub(crate) struct App { thread_event_listener_tasks: HashMap>, agent_navigation: AgentNavigationState, side_threads: HashMap, + abandoned_side_threads: HashSet, active_thread_id: Option, active_thread_rx: Option>, primary_thread_id: Option, @@ -1068,6 +1070,7 @@ See the Codex keymap documentation for supported actions and examples." thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), side_threads: HashMap::new(), + abandoned_side_threads: HashSet::new(), active_thread_id: None, active_thread_rx: None, primary_thread_id: None, diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index aae6b33871..8925ba8db2 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -195,6 +195,21 @@ impl App { app_server_client: &AppServerSession, request: ServerRequest, ) { + let thread_id = server_request_thread_id(&request); + if thread_id.is_some_and(|thread_id| self.abandoned_side_threads.contains(&thread_id)) { + if let Err(err) = self + .reject_app_server_request( + app_server_client, + request.id().clone(), + "side conversation was closed".to_string(), + ) + .await + { + tracing::warn!("{err}"); + } + return; + } + if let Some(unsupported) = self .pending_app_server_requests .note_server_request(&request) @@ -219,7 +234,7 @@ impl App { return; } - let Some(thread_id) = server_request_thread_id(&request) else { + let Some(thread_id) = thread_id else { tracing::warn!("ignoring threadless app-server request"); return; }; diff --git a/codex-rs/tui/src/app/side.rs b/codex-rs/tui/src/app/side.rs index 3e85ed65ef..477e1a3811 100644 --- a/codex-rs/tui/src/app/side.rs +++ b/codex-rs/tui/src/app/side.rs @@ -9,6 +9,10 @@ use super::*; use crate::chatwidget::InterruptedTurnNoticeMode; +use codex_app_server_protocol::ThreadUnsubscribeParams; +use codex_app_server_protocol::ThreadUnsubscribeResponse; +use codex_app_server_protocol::TurnInterruptParams; +use codex_app_server_protocol::TurnInterruptResponse; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; @@ -421,10 +425,70 @@ impl App { self.chat_widget.add_error_message(message); return false; } + self.abandoned_side_threads.insert(thread_id); self.discard_thread_local_state(thread_id).await; true } + pub(super) async fn discard_side_thread_in_background( + &mut self, + app_server: &mut AppServerSession, + thread_id: ThreadId, + ) { + let turn_id = self + .active_turn_id_for_thread(thread_id) + .await + .unwrap_or_default(); + let request_handle = app_server.request_handle(); + let interrupt_request_id = app_server.next_request_id(); + let retry_interrupt_request_id = app_server.next_request_id(); + let unsubscribe_request_id = app_server.next_request_id(); + + self.abandoned_side_threads.insert(thread_id); + self.discard_thread_local_state(thread_id).await; + + tokio::spawn(async move { + let interrupt_result = request_handle + .request_typed::(ClientRequest::TurnInterrupt { + request_id: interrupt_request_id, + params: TurnInterruptParams { + thread_id: thread_id.to_string(), + turn_id: turn_id.clone(), + }, + }) + .await; + let interrupt_result = if let Err(error) = &interrupt_result + && let Some(actual_turn_id) = active_turn_interrupt_race(error) + { + request_handle + .request_typed::(ClientRequest::TurnInterrupt { + request_id: retry_interrupt_request_id, + params: TurnInterruptParams { + thread_id: thread_id.to_string(), + turn_id: actual_turn_id, + }, + }) + .await + } else { + interrupt_result + }; + if let Err(error) = interrupt_result { + tracing::warn!(%error, "failed to interrupt side conversation"); + } + if let Err(error) = request_handle + .request_typed::(ClientRequest::ThreadUnsubscribe { + request_id: unsubscribe_request_id, + params: ThreadUnsubscribeParams { + thread_id: thread_id.to_string(), + }, + }) + .await + { + tracing::warn!(%error, "failed to unsubscribe side conversation"); + } + }); + } + pub(super) async fn discard_closed_side_thread(&mut self, thread_id: ThreadId) { self.discard_thread_local_state(thread_id).await; } @@ -578,17 +642,10 @@ impl App { if self.active_thread_id == Some(thread_id) && let Some(side_thread_id) = side_thread_to_discard { - if self.discard_side_thread(app_server, side_thread_id).await { - self.surface_pending_inactive_thread_interactive_requests() - .await?; - } else { - self.keep_side_thread_visible_after_cleanup_failure( - tui, - app_server, - side_thread_id, - ) + self.discard_side_thread_in_background(app_server, side_thread_id) .await; - } + self.surface_pending_inactive_thread_interactive_requests() + .await?; } Ok(()) } diff --git a/codex-rs/tui/src/app/test_support.rs b/codex-rs/tui/src/app/test_support.rs index d4beb21635..13d7702f5b 100644 --- a/codex-rs/tui/src/app/test_support.rs +++ b/codex-rs/tui/src/app/test_support.rs @@ -57,6 +57,7 @@ pub(super) async fn make_test_app() -> App { thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), side_threads: HashMap::new(), + abandoned_side_threads: HashSet::new(), active_thread_id: None, active_thread_rx: None, primary_thread_id: None, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index c542b2248e..042b08e0c4 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -4375,6 +4375,62 @@ async fn discard_side_thread_keeps_local_state_when_server_close_fails() -> Resu .await } +#[tokio::test] +async fn background_side_cleanup_removes_local_state_and_ignores_late_events() -> Result<()> { + let mut app = make_test_app().await; + let mut app_server = + crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?; + let parent_thread_id = ThreadId::new(); + let side_thread_id = ThreadId::new(); + app.active_thread_id = Some(parent_thread_id); + app.side_threads + .insert(side_thread_id, SideThreadState::new(parent_thread_id)); + app.thread_event_channels + .insert(side_thread_id, ThreadEventChannel::new(/*capacity*/ 4)); + app.agent_navigation.upsert( + side_thread_id, + Some("Side".to_string()), + Some("side".to_string()), + /*is_closed*/ false, + ); + app.discard_side_thread_in_background(&mut app_server, side_thread_id) + .await; + + assert_eq!(app.active_thread_id, Some(parent_thread_id)); + assert!(!app.side_threads.contains_key(&side_thread_id)); + assert!(!app.thread_event_channels.contains_key(&side_thread_id)); + assert_eq!(app.agent_navigation.get(&side_thread_id), None); + assert!(app.abandoned_side_threads.contains(&side_thread_id)); + + app.enqueue_thread_notification( + side_thread_id, + agent_message_delta_notification(side_thread_id, "turn-1", "item-1", "late"), + ) + .await?; + assert!(!app.thread_event_channels.contains_key(&side_thread_id)); + + app.handle_app_server_event( + &app_server, + codex_app_server_client::AppServerEvent::ServerRequest(exec_approval_request( + side_thread_id, + "turn-1", + "item-1", + Some("approval-1"), + )), + ) + .await; + let resolution = app + .pending_app_server_requests + .take_resolution(Op::ExecApproval { + id: "approval-1".to_string(), + turn_id: None, + decision: codex_app_server_protocol::CommandExecutionApprovalDecision::Accept, + }) + .expect("approval resolution should serialize"); + assert_eq!(resolution, None); + Ok(()) +} + #[tokio::test] async fn discard_closed_side_thread_removes_local_state_without_server_rpc() { let mut app = make_test_app().await; @@ -4698,6 +4754,7 @@ async fn make_test_app() -> App { thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), side_threads: HashMap::new(), + abandoned_side_threads: HashSet::new(), active_thread_id: None, active_thread_rx: None, primary_thread_id: None, @@ -4765,6 +4822,7 @@ async fn make_test_app_with_channels() -> ( thread_event_listener_tasks: HashMap::new(), agent_navigation: AgentNavigationState::default(), side_threads: HashMap::new(), + abandoned_side_threads: HashSet::new(), active_thread_id: None, active_thread_rx: None, primary_thread_id: None, diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 180abdb784..f1765b8171 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -954,6 +954,9 @@ impl App { thread_id: ThreadId, notification: ServerNotification, ) -> Result<()> { + if self.abandoned_side_threads.contains(&thread_id) { + return Ok(()); + } if matches!(notification, ServerNotification::ThreadSettingsUpdated(_)) && self.primary_thread_id.is_some() && self.primary_thread_id != Some(thread_id)