From bf7c87a7b7bdb1b65cbfdca06d15713d3aa20b01 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Tue, 22 Sep 2026 02:06:40 +0000 Subject: [PATCH] Ignore stale thread-close notifications after resuming a thread (#47155) ## Why A queued close notification from an old runtime can arrive after the same thread has been resumed. Because thread IDs survive reloads, handling that notification can incorrectly exit the TUI or switch away from the displayed thread. ## What changed Confirm the displayed thread's status with `thread/read` before handling its close notification. Ignore the notification if the thread is still loaded. Bound verification to five seconds and retry an internal server error once. If verification fails, reconnect remote sessions. Preserve close handling for embedded sessions, unreadable unpersisted threads, and servers that do not support the request. ## Testing Add regression tests for stale and genuine close notifications, plus disconnects, timeouts, transient and persistent server errors, embedded-session fallback, and unsupported requests. GitOrigin-RevId: 783433b11ed268aed521f3b2cc4d0aa3a36a3836 --- codex-rs/tui/src/app/app_server_events.rs | 59 +++++++ codex-rs/tui/src/app/tests.rs | 2 + .../src/app/tests/resume_shutdown_tests.rs | 149 ++++++++++++++++++ 3 files changed, 210 insertions(+) create mode 100644 codex-rs/tui/src/app/tests/resume_shutdown_tests.rs diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 80b9595f60..fbea229338 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -14,6 +14,7 @@ use crate::app_info::app_info_from_api; use crate::app_server_session::AppServerSession; use crate::app_server_session::status_account_display_from_auth_mode; use codex_app_server_client::AppServerEvent; +use codex_app_server_client::TypedRequestError; use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::RateLimitReachedType; @@ -25,8 +26,10 @@ use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadSource; +use codex_app_server_protocol::ThreadStatus; use codex_protocol::ThreadId; use codex_protocol::protocol::SubAgentSource; +use std::time::Duration; impl App { pub(super) fn refresh_mcp_startup_expected_servers_from_config(&mut self) { @@ -116,6 +119,62 @@ impl App { app_server_client: &AppServerSession, notification: ServerNotification, ) { + // A picker can leave an old runtime's close notification queued while the same thread + // is resumed. Thread IDs survive reloads, so confirm that the displayed thread is still + // unloaded before routing a close that would exit the TUI or switch away from it. + if let ServerNotification::ThreadClosed(closed) = ¬ification + && let Ok(thread_id) = ThreadId::from_string(&closed.thread_id) + && self.current_displayed_thread_id() == Some(thread_id) + { + let deadline = tokio::time::Instant::now() + Duration::from_secs(/*secs*/ 5); + for attempt in 0..2 { + let result = tokio::time::timeout_at( + deadline, + app_server_client + .request_handle() + .request_typed::(ClientRequest::ThreadRead { + request_id: RequestId::String(format!( + "thread-closed-{thread_id}-{attempt}" + )), + params: ThreadReadParams { + thread_id: closed.thread_id.clone(), + include_turns: false, + }, + }), + ) + .await; + match result { + Ok(Ok(response)) + if !matches!(response.thread.status, ThreadStatus::NotLoaded) => + { + return; + } + Ok(Ok(_)) => break, + // Unpersisted closed threads can no longer be read. Preserve that close + // and compatibility with servers that do not support this request. + Ok(Err(TypedRequestError::Server { source, .. })) + if matches!(source.code, -32602..=-32600) => + { + break; + } + Ok(Err(TypedRequestError::Server { source, .. })) + if source.code == -32603 && attempt == 0 => + { + continue; + } + Ok(Err(_)) | Err(_) => { + // A failed read does not confirm closure. Remote connections recover + // through the normal reconnect path, rather than exiting the TUI. + tracing::warn!("could not confirm displayed thread closure"); + if self.begin_reconnect() { + return; + } + break; + } + } + } + } + if let ServerNotification::ThreadStatusChanged(status) = ¬ification { let _ = self.dynamic_tool_status_updates.send(status.clone()); } diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 769afb0a1b..b111d7e068 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -71,6 +71,8 @@ mod realtime_start; mod reasoning_resume_tests; #[path = "tests/recap_generation_tests.rs"] mod recap_generation; +#[path = "tests/resume_shutdown_tests.rs"] +mod resume_shutdown_tests; mod safety_buffering; #[path = "tests/session_lifecycle_requests.rs"] mod session_lifecycle_requests; diff --git a/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs b/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs new file mode 100644 index 0000000000..654cc03742 --- /dev/null +++ b/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs @@ -0,0 +1,149 @@ +//! A delayed close must not shut down a thread that has already been reloaded. + +use super::disconnect::serve_reconnect_requests; +use super::*; +use crate::app_server_session::ThreadParamsMode; +use codex_app_server_client::AppServerEvent; +use pretty_assertions::assert_eq; +use serde_json::json; +use tokio::net::TcpListener; + +#[tokio::test] +async fn thread_close_exits_only_when_displayed_thread_is_unloaded() -> Result<()> { + let (mut app, mut events, _commands) = make_test_app_with_channels().await; + app.config.thread_unload_delay = Duration::ZERO; + let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; + let started = server.start_thread(&app.config).await?; + let thread_id = started.session.thread_id; + app.enqueue_primary_thread_session(started.session, started.turns) + .await?; + let mut tui = crate::tui::test_support::make_test_tui()?; + app.drain_active_thread_events(&mut tui).await?; + + // The old runtime's notification can arrive after this thread ID is loaded again. + app.handle_app_server_event( + &server, + AppServerEvent::ServerNotification(Box::new(thread_closed_notification(thread_id))), + ) + .await; + app.drain_active_thread_events(&mut tui).await?; + assert!( + !std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| matches!(event, AppEvent::Exit(_))) + ); + + // A genuine close must still exit. + server.thread_unsubscribe(thread_id).await?; + let closed = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + while let Some(event) = server.next_event().await { + if matches!(&event, + AppServerEvent::ServerNotification(notification) + if matches!(notification.as_ref(), ServerNotification::ThreadClosed(closed) + if closed.thread_id == thread_id.to_string())) + { + return event; + } + } + panic!("server event stream ended before the thread closed"); + }) + .await?; + app.handle_app_server_event(&server, closed).await; + app.drain_active_thread_events(&mut tui).await?; + assert!( + std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| matches!(event, AppEvent::Exit(ExitMode::Immediate))) + ); + server.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn failed_close_status_read_reconnects_remote_or_closes_embedded() -> Result<()> { + for (failure, expected_reads, offline) in [ + ("disconnect", 1, true), + ("timeout", 1, true), + ("transient", 2, false), + ("persistent", 2, true), + ("embedded", 2, false), + ("unsupported", 1, false), + ] { + let (mut app, mut events, _commands) = make_test_app_with_channels().await; + let id = ThreadId::new(); + app.enqueue_primary_thread_session( + test_thread_session(id, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + let listener = TcpListener::bind("127.0.0.1:0").await?; + let endpoint = crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?; + if failure != "embedded" { + app.app_server_target = AppServerTarget::Remote { + endpoint: endpoint.clone(), + }; + } + let reads = Arc::new(AtomicUsize::new(/*v*/ 0)); + let server_reads = reads.clone(); + let cwd = app.config.cwd.clone(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + serve_reconnect_requests( + tokio_tungstenite::accept_async(stream).await?, + move |request| { + assert_eq!(request.method, "thread/read"); + let attempt = server_reads.fetch_add(/*val*/ 1, Ordering::SeqCst); + let response = match failure { + "disconnect" | "timeout" => None, + "unsupported" => { + Some(json!({"error": {"code": -32601, "message": "method not found"}})) + } + "transient" if attempt == 1 => Some(json!({"result": {"thread": { + "id": id, "sessionId": id, "preview": "", "ephemeral": false, + "modelProvider": "test-provider", "createdAt": 1, "updatedAt": 2, + "status": {"type": "idle"}, "cwd": cwd, "cliVersion": "0.0.0", + "source": "cli", "turns": [] + }}})), + _ => { + Some(json!({"error": {"code": -32603, "message": "temporary failure"}})) + } + }; + async move { + if failure == "timeout" { + std::future::pending::<()>().await; + } + response + } + }, + ) + .await + }); + let session = AppServerSession::new( + crate::connect_remote_app_server(endpoint).await?, + ThreadParamsMode::Remote, + ); + tokio::time::timeout( + Duration::from_secs(/*secs*/ 10), + app.handle_app_server_event( + &session, + AppServerEvent::ServerNotification(Box::new(thread_closed_notification(id))), + ), + ) + .await?; + let mut tui = crate::tui::test_support::make_test_tui()?; + app.drain_active_thread_events(&mut tui).await?; + assert_eq!( + (reads.load(Ordering::SeqCst), app.reconnect.offline), + (expected_reads, offline), + "{failure}" + ); + assert_eq!( + std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| matches!(event, AppEvent::Exit(_) | AppEvent::FatalExitRequest(_))), + matches!(failure, "embedded" | "unsupported"), + "{failure}" + ); + server.abort(); + let _ = server.await; + session.shutdown().await?; + } + Ok(()) +}