From 27bf160f7909704fb7e23d508f31900d90479699 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 2 Sep 2026 04:36:04 +0000 Subject: [PATCH] Retry TUI reconnects while threads are closing (#42207) ## Why An app-server thread can still be closing when the TUI reconnects. Its `thread/resume` request then uses the same `-32600` error code as an unavailable thread, even though the thread can be resumed once closing finishes. ## What changed Treat the thread-specific "is closing" response as a transient reconnect error so the existing reconnect loop retries instead of marking the conversation unavailable. ## Testing Extend the active reconnect and end-to-end reconnect tests to cover a closing thread that resumes successfully on the next attempt. GitOrigin-RevId: 30ce232b8894065fef346a86128b3ff9bf36ba1f --- codex-rs/tui/src/app/reconnect.rs | 12 +++++++++ .../src/app/tests/active_reconnect_tests.rs | 25 +++++++++++++++++-- codex-rs/tui/tests/suite/reconnect.rs | 19 +++++++++++--- 3 files changed, 51 insertions(+), 5 deletions(-) diff --git a/codex-rs/tui/src/app/reconnect.rs b/codex-rs/tui/src/app/reconnect.rs index 4be0cc49c0..4831fbb995 100644 --- a/codex-rs/tui/src/app/reconnect.rs +++ b/codex-rs/tui/src/app/reconnect.rs @@ -83,6 +83,18 @@ pub(super) async fn reconnect( { return Err(error); } + // Unloading threads use the same code as unavailable conversations, but + // ordinary resume can reattach once the unload finishes. + Err(error) + if matches!( + error.downcast_ref::(), + Some(TypedRequestError::Server { method, source }) + if method == "thread/resume" && source.code == -32600 + && source.message.starts_with(&format!("thread {thread_id} is closing;")) + ) => + { + return Err(error); + } Err(error) if matches!( error.downcast_ref::(), diff --git a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs index 41e2cf6775..8b9ca10b9b 100644 --- a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs @@ -11,7 +11,14 @@ use super::disconnect::serve_reconnect_requests; #[tokio::test] async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> Result<()> { - for (recovered_queue, edit_offline) in [(true, false), (false, false), (true, true)] { + for (recovered_queue, edit_offline, resume_error_code) in [ + (true, false, -32603), + (false, false, -32603), + (true, true, -32603), + (true, false, -32600), + (false, false, -32600), + (true, true, -32600), + ] { let (mut app, mut events, mut ops) = make_test_app_with_channels().await; let id = ThreadId::new(); let cwd = app.config.cwd.clone(); @@ -79,7 +86,13 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> for attempt in 0..2 { let (stream, _) = listener.accept().await?; methods.extend(serve_reconnect_requests(tokio_tungstenite::accept_async(stream).await?, |request| std::future::ready(match request.method.as_str() { - "thread/resume" if attempt == 0 => Some(json!({"error": {"code": -32603, "message": "temporarily unavailable"}})), + "thread/resume" if attempt == 0 => Some(json!({"error": {"code": resume_error_code, "message": + if resume_error_code == -32600 { + format!("thread {id} is closing; retry thread/resume after the thread is closed") + } else { + "temporarily unavailable".into() + } + }})), "thread/resume" => { let params = request.params.as_ref().unwrap(); assert_eq!(params["threadId"], id.to_string()); @@ -202,6 +215,7 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> app.finish_reconnect(&mut tui, &mut session, &mut events, connected) .await?; assert!(!app.reconnect.offline); + assert!(!app.thread_unavailable(id)); assert_eq!(app.last_subagent_backfill_attempt, None); assert!( app.rate_limit_refresh_state @@ -290,6 +304,13 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> } session.shutdown().await?; let methods = server.await??; + assert_eq!( + methods + .iter() + .filter(|method| *method == "thread/resume") + .count(), + 2 + ); assert_eq!( methods .iter() diff --git a/codex-rs/tui/tests/suite/reconnect.rs b/codex-rs/tui/tests/suite/reconnect.rs index fe2cc52cfd..925eac79cd 100644 --- a/codex-rs/tui/tests/suite/reconnect.rs +++ b/codex-rs/tui/tests/suite/reconnect.rs @@ -36,7 +36,7 @@ async fn automatic_reconnect_restores_draft_and_routes_new_notifications() -> Re "status": {"type": "active", "activeFlags": []}, "cwd": server_cwd, "cliVersion": "0.0.0", "source": "cli", "turns": [{"id": "running", "items": [], "status": "inProgress", "error": null}] }); - for connection in 0..2 { + for connection in 0..3 { let mut socket = loop { let (stream, _) = listener.accept().await?; // Startup probes the default daemon socket before opening its WebSocket. @@ -62,6 +62,19 @@ async fn automatic_reconnect_restores_draft_and_routes_new_notifications() -> Re if connection == 1 && request.method == "initialize" { restore_rx.take().unwrap().await?; } + if connection == 1 && request.method == "thread/resume" { + socket + .send(Message::Text( + json!({"id": request.id, "error": { + "code": -32600, + "message": format!("thread {id} is closing; retry after the thread is closed") + }}) + .to_string() + .into(), + )) + .await?; + continue; + } let result = match request.method.as_str() { "initialize" => json!({"userAgent": "reconnect-pty"}), "account/read" => { @@ -97,7 +110,7 @@ async fn automatic_reconnect_restores_draft_and_routes_new_notifications() -> Re .into(), )) .await?; - if connection == 1 && request.method == "thread/resume" { + if connection == 2 && request.method == "thread/resume" { // Keep the recovered turn running: its output must appear without waiting // for turn/completed or rebuilding the transcript. socket @@ -160,7 +173,7 @@ async fn automatic_reconnect_restores_draft_and_routes_new_notifications() -> Re .iter() .filter(|method| *method == "thread/resume") .count(), - 1 + 2 ); assert!(!methods.iter().any(|method| method == "turn/start")); Ok(())