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
This commit is contained in:
Eric Traut
2026-07-29 03:54:05 +00:00
committed by copyberry
parent b96ebfb312
commit d06c7ac055
6 changed files with 148 additions and 11 deletions
+3
View File
@@ -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<ThreadId, JoinHandle<()>>,
agent_navigation: AgentNavigationState,
side_threads: HashMap<ThreadId, SideThreadState>,
abandoned_side_threads: HashSet<ThreadId>,
active_thread_id: Option<ThreadId>,
active_thread_rx: Option<mpsc::Receiver<ThreadBufferedEvent>>,
primary_thread_id: Option<ThreadId>,
@@ -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,
+16 -1
View File
@@ -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;
};
+67 -10
View File
@@ -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::<TurnInterruptResponse>(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::<TurnInterruptResponse>(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::<ThreadUnsubscribeResponse>(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(())
}
+1
View File
@@ -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,
+58
View File
@@ -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,
+3
View File
@@ -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)