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
This commit is contained in:
Eric Traut
2026-09-02 05:10:55 +00:00
committed by copyberry
parent 8813bd4b00
commit 27bf160f79
3 changed files with 51 additions and 5 deletions
+12
View File
@@ -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::<TypedRequestError>(),
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::<TypedRequestError>(),
@@ -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()
+16 -3
View File
@@ -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(())