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(()) +}