Finish idle thread unloads after slow shutdown (#51420)

## Why

Idle thread unloading stopped waiting after ten seconds, leaving the thread loaded when resource cleanup took longer.

## What changed

Keep awaiting the same shutdown future after the ten-second deadline, emitting `codex.app_server.thread_shutdown_slow` as a warning. Once shutdown completes, remove the thread and send `thread/closed` without requiring another unload request.

## Testing

Add a regression test that blocks writer-lock cleanup past the warning deadline, verifies the thread remains loaded while cleanup is blocked, then releases the lock and checks for `thread/closed` and an empty loaded-thread list.

GitOrigin-RevId: 18c411f86fd69e710e472a4e3572d75f18682147

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -443,8 +443,24 @@ pub(super) async fn unload_thread_without_subscribers(
     thread_state_manager.remove_thread_state(thread_id).await;
 
     tokio::spawn(async move {
-        match wait_for_thread_shutdown(&thread).await {
-            ThreadShutdownResult::Complete => {
+        // The deadline bounds our warning, not background cleanup. Keep polling the
+        // same future so even delayed shutdown submission can eventually finish.
+        let shutdown = thread.shutdown_and_wait();
+        tokio::pin!(shutdown);
+        let result = match tokio::time::timeout(Duration::from_secs(/*secs*/ 10), &mut shutdown)
+            .await
+        {
+            Ok(result) => result,
+            Err(_) => {
+                warn!(
+                    event.name = "codex.app_server.thread_shutdown_slow",
+                    "thread {thread_id} shutdown is taking longer than expected; continuing to wait"
+                );
+                shutdown.await
+            }
+        };
+        match result {
+            Ok(()) => {
                 // A delayed unload can finish after thread/revert replaces this runtime under
                 // the same thread ID. Only the runtime that scheduled this unload may remove it.
                 if thread_manager
@@ -467,14 +483,10 @@ pub(super) async fn unload_thread_without_subscribers(
                     .await;
                 pending_thread_unloads.lock().await.remove(&thread_id);
             }
-            ThreadShutdownResult::SubmitFailed => {
+            Err(_) => {
                 pending_thread_unloads.lock().await.remove(&thread_id);
                 warn!("failed to submit Shutdown to thread {thread_id}");
             }
-            ThreadShutdownResult::TimedOut => {
-                pending_thread_unloads.lock().await.remove(&thread_id);
-                warn!("thread {thread_id} shutdown timed out; leaving thread loaded");
-            }
         }
     });
 }

codex-rs/app-server/tests/common/json_logging.rs

@@ -26,8 +26,14 @@ impl JsonLogCapture {
         self.updated.notify_one();
     }
 
-    pub(crate) async fn wait_for_event(&self, event_name: &str) -> Result<Value> {
-        let mut events = self.wait_for_events(event_name, /*count*/ 1).await?;
+    pub(crate) async fn wait_for_event(
+        &self,
+        event_name: &str,
+        timeout: Duration,
+    ) -> Result<Value> {
+        let mut events = self
+            .wait_for_events(event_name, /*count*/ 1, timeout)
+            .await?;
         Ok(events.remove(0))
     }
 
@@ -35,8 +41,9 @@ impl JsonLogCapture {
         &self,
         event_name: &str,
         count: usize,
+        timeout: Duration,
     ) -> Result<Vec<Value>> {
-        let result = tokio::time::timeout(Duration::from_secs(10), async {
+        let result = tokio::time::timeout(timeout, async {
             loop {
                 let updated = self.updated.notified();
                 let events = self

codex-rs/app-server/tests/common/test_app_server.rs

@@ -243,12 +243,13 @@ impl TestAppServer {
         })
     }
 
-    /// Waits for a JSON stderr event whose structured `event.name` field matches.
+    /// Waits up to `timeout` for a JSON stderr event whose structured `event.name` field matches.
     pub async fn wait_for_json_log_event(
         &self,
         event_name: &str,
+        timeout: Duration,
     ) -> anyhow::Result<serde_json::Value> {
-        self.json_logs.wait_for_event(event_name).await
+        self.json_logs.wait_for_event(event_name, timeout).await
     }
 
     async fn new_with_program_env_and_args(

codex-rs/app-server/tests/suite/logging.rs

@@ -625,7 +625,7 @@ async fn app_server_emits_structured_tool_call_timing_event() -> Result<()> {
     .await??;
 
     let mut tool_call = app_server
-        .wait_for_json_log_event("codex.tool_call")
+        .wait_for_json_log_event("codex.tool_call", Duration::from_secs(/*secs*/ 10))
         .await?;
     let tool_call_object = tool_call
         .as_object_mut()

codex-rs/app-server/tests/suite/v2/code_mode_host.rs

@@ -149,7 +149,10 @@ async fn code_mode_model_output_uses_structured_host_timing(
         .await?;
     let _: TurnStartResponse = app_server.read_response(start_id).await?;
     let host_timing = app_server
-        .wait_for_json_log_event("codex.code_mode.host_timing")
+        .wait_for_json_log_event(
+            "codex.code_mode.host_timing",
+            Duration::from_secs(/*secs*/ 10),
+        )
         .await?;
     // Hold only the app-server after the host outcome, so local elapsed time
     // cannot round to the same displayed duration as the host measurement.
@@ -179,7 +182,7 @@ async fn code_mode_model_output_uses_structured_host_timing(
     assert_eq!(host_fields["call_id"], "timed-call");
     assert_eq!(host_fields["tool_name"], tool_name);
     let handler_timing = app_server
-        .wait_for_json_log_event("codex.tool_call")
+        .wait_for_json_log_event("codex.tool_call", Duration::from_secs(/*secs*/ 10))
         .await?;
     let handler_fields = &handler_timing["fields"];
     let handler_duration_ms = handler_fields["handler_duration_ms"]

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -156,6 +156,7 @@ mod thread_start;
 mod thread_status;
 mod thread_timeline;
 mod thread_unarchive;
+mod thread_unload;
 mod thread_unsubscribe;
 mod turn_cost_otel;
 mod turn_interrupt;

codex-rs/app-server/tests/suite/v2/otel.rs

@@ -101,7 +101,10 @@ async fn account_switch_reloads_telemetry_collectors_and_preserves_trace_context
     assert_eq!(response, LoginAccountResponse::ChatgptAuthTokens {});
     timeout(
         TEST_TIMEOUT,
-        app_server.wait_for_json_log_event("codex.app_server.otel_reloaded"),
+        app_server.wait_for_json_log_event(
+            "codex.app_server.otel_reloaded",
+            Duration::from_secs(/*secs*/ 10),
+        ),
     )
     .await??;
     timeout(TEST_TIMEOUT, async {

codex-rs/app-server/tests/suite/v2/thread_unload.rs

@@ -0,0 +1,107 @@
+//! Idle unloading must finish even when resource teardown exceeds the shutdown warning deadline.
+
+use anyhow::Context;
+use anyhow::Result;
+use app_test_support::MockResponsesConfig;
+use app_test_support::TestAppServer;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
+use codex_app_server_protocol::ClientRequest;
+use codex_app_server_protocol::ThreadClosedNotification;
+use codex_app_server_protocol::ThreadLoadedListParams;
+use codex_app_server_protocol::ThreadLoadedListResponse;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadUnsubscribeParams;
+use codex_app_server_protocol::ThreadUnsubscribeResponse;
+use codex_app_server_protocol::ThreadUnsubscribeStatus;
+use pretty_assertions::assert_eq;
+use std::fs::OpenOptions;
+use std::time::Duration;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+#[tokio::test]
+async fn idle_unload_finishes_after_slow_writer_shutdown() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
+    let home = TempDir::new()?;
+    MockResponsesConfig::new(&server.uri())
+        .with_root_config("thread_unload_delay_secs = 0\napprovals_reviewer = \"user\"")
+        .write(home.path())?;
+    let mut client = TestAppServer::builder()
+        .with_codex_home(home.path())
+        .with_json_logging("warn")
+        // Writer cleanup blocks one worker while the test queries the server.
+        .with_env_overrides(&[("TOKIO_WORKER_THREADS", Some("2"))])
+        .build_initialized()
+        .await?;
+    let started = client.start_thread(ThreadStartParams::default()).await?;
+    let thread_id = started.thread.id;
+
+    // Block the real cross-process writer-lock cleanup. Other app-server tasks,
+    // including the ten-second shutdown deadline and JSON-RPC, can still run.
+    let coordination_lock = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(home.path().join("thread-writer-locks/.coordination.lock"))?;
+    coordination_lock.lock()?;
+    let unsubscribed: ThreadUnsubscribeResponse = client
+        .request(|request_id| ClientRequest::ThreadUnsubscribe {
+            request_id,
+            params: ThreadUnsubscribeParams {
+                thread_id: thread_id.clone(),
+            },
+        })
+        .await?;
+    assert_eq!(
+        unsubscribed,
+        ThreadUnsubscribeResponse {
+            status: ThreadUnsubscribeStatus::Unsubscribed
+        }
+    );
+    // Unsubscribe only schedules shutdown. Keep the lock until the shutdown
+    // future has actually crossed its warning deadline.
+    client
+        .wait_for_json_log_event(
+            "codex.app_server.thread_shutdown_slow",
+            Duration::from_secs(/*secs*/ 30),
+        )
+        .await
+        .context("idle unload never reached the shutdown warning deadline")?;
+    let loaded: ThreadLoadedListResponse = client
+        .request(|request_id| ClientRequest::ThreadLoadedList {
+            request_id,
+            params: ThreadLoadedListParams::default(),
+        })
+        .await?;
+    assert_eq!(
+        loaded,
+        ThreadLoadedListResponse {
+            data: vec![thread_id.clone()],
+            next_cursor: None
+        }
+    );
+
+    // Once persistence can finish, delayed shutdown must still remove the runtime
+    // and report closure without another client subscription or unload request.
+    drop(coordination_lock);
+    let closed: ThreadClosedNotification = timeout(
+        Duration::from_secs(/*secs*/ 10),
+        client.read_notification("thread/closed"),
+    )
+    .await
+    .context("slow shutdown never finalized idle unload")??;
+    assert_eq!(closed, ThreadClosedNotification { thread_id });
+    let loaded: ThreadLoadedListResponse = client
+        .request(|request_id| ClientRequest::ThreadLoadedList {
+            request_id,
+            params: ThreadLoadedListParams::default(),
+        })
+        .await?;
+    assert_eq!(
+        loaded,
+        ThreadLoadedListResponse {
+            data: Vec::new(),
+            next_cursor: None
+        }
+    );
+    Ok(())
+}