Fix idle cleanup for disconnected multi-agent v2 children (#52081)

## Why

Persisted multi-agent v2 children created after the client disconnects need lifecycle listeners so their runtimes can be released once idle. Listener attachment must also wait for an earlier runtime's cleanup to finish.

## What changed

- Start lifecycle listeners for persisted v2 children even when no initialized clients are connected, and defer attachment until pending unload cleanup completes.
- Keep ephemeral children loaded so follow-up tasks retain their model context.
- Raise the default `thread_unload_delay_secs` from 60 seconds to 1800 seconds (30 minutes), updating the configuration documentation and sample.

## Testing

Add regression coverage for children spawned after disconnect with zero- and one-second unload delays, including waiting for pending cleanup, keeping running threads loaded, releasing idle runtimes, and preserving an ephemeral child's earlier reply across a follow-up. Update configuration tests for the new default and explicit overrides.

GitOrigin-RevId: 2fecc77b5c03237aa8fd7eacc4bd119b99d36e0a

codex-rs/app-server/src/message_processor.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::collections::HashSet;
 use std::future::Future;
 use std::sync::Arc;
@@ -394,7 +395,7 @@ impl MessageProcessor {
             outgoing.clone(),
         );
 
-        let pending_thread_unloads = Arc::new(Mutex::new(HashSet::new()));
+        let pending_thread_unloads = Arc::new(Mutex::new(HashMap::new()));
         let thread_watch_manager =
             crate::thread_status::ThreadWatchManager::new_with_outgoing(outgoing.clone());
         let thread_list_state_permit = Arc::new(Semaphore::new(/*permits*/ 1));

codex-rs/app-server/src/message_processor_thread_lifecycle_tests.rs

@@ -1,4 +1,14 @@
-//! Exercises thread cleanup when required MCP startup outlives its requesting connection.
+//! Exercises thread cleanup when startup or child creation outlives the client connection.
+
+use codex_app_server_protocol::TurnStartResponse;
+use codex_features::Feature;
+use codex_protocol::protocol::AgentStatus;
+use core_test_support::responses;
+use core_test_support::streaming_sse::StreamingSseChunk;
+use core_test_support::streaming_sse::start_streaming_sse_server;
+use pretty_assertions::assert_eq;
+use test_case::test_case;
+use tokio::sync::oneshot;
 
 use super::ConnectionSessionState;
 use super::MessageProcessor;
@@ -213,6 +223,293 @@ MCP_TEST_PID_FILE = {}
     Ok(())
 }
 
+#[test_case(0, false; "v2_zero_delay")]
+#[test_case(1, false; "v2_one_second_delay")]
+#[test_case(1, true; "ephemeral_v2_followup_after_idle_deadline")]
+#[tokio::test]
+async fn disconnected_v2_child_lifecycle(unload_delay_secs: u64, ephemeral: bool) -> Result<()> {
+    let (release_spawn, spawn_gate) = oneshot::channel();
+    let (release_spawn_completion, spawn_completion_gate) = oneshot::channel();
+    let (release_first_completion, first_completion_gate) = oneshot::channel();
+    let (release_second_completion, second_completion_gate) = oneshot::channel();
+    let done = responses::sse(vec![
+        responses::ev_response_created("done"),
+        responses::ev_assistant_message("done", "Done"),
+        responses::ev_completed("done"),
+    ]);
+    let (server, _completions) = start_streaming_sse_server(vec![
+        vec![
+            StreamingSseChunk {
+                gate: Some(spawn_gate),
+                body: responses::sse(vec![
+                    responses::ev_response_created("spawn"),
+                    responses::ev_function_call_with_namespace(
+                        "spawn",
+                        "collaboration",
+                        "spawn_agent",
+                        &json!({"task_name": "child", "message": "Reply done", "fork_turns": "none"})
+                            .to_string(),
+                    ),
+                ]),
+            },
+            StreamingSseChunk {
+                gate: Some(spawn_completion_gate),
+                body: responses::sse(vec![responses::ev_completed("spawn")]),
+            },
+        ],
+        vec![StreamingSseChunk {
+            gate: Some(first_completion_gate),
+            body: done.clone(),
+        }],
+        vec![StreamingSseChunk {
+            gate: Some(second_completion_gate),
+            body: if ephemeral {
+                responses::sse(vec![
+                    responses::ev_response_created("followup"),
+                    responses::ev_function_call_with_namespace(
+                        "followup",
+                        "collaboration",
+                        "followup_task",
+                        &json!({"target": "child", "message": "Recall your previous reply"})
+                            .to_string(),
+                    ),
+                    responses::ev_completed("followup"),
+                ])
+            } else {
+                done.clone()
+            },
+        }],
+        // Follow-up responses and the child's completion notification.
+        vec![StreamingSseChunk {
+            gate: None,
+            body: done.clone(),
+        }],
+        vec![StreamingSseChunk {
+            gate: None,
+            body: done.clone(),
+        }],
+        vec![StreamingSseChunk {
+            gate: None,
+            body: done,
+        }],
+    ])
+    .await;
+    let home = TempDir::new()?;
+    let config = MockResponsesConfig::new(server.uri())
+        .with_root_config(&format!(
+            "thread_unload_delay_secs = {unload_delay_secs}\napprovals_reviewer = \"user\""
+        ))
+        .enable_feature(Feature::Collab)
+        .enable_feature(Feature::MultiAgentV2);
+    config.write(home.path())?;
+    let config = Arc::new(
+        ConfigBuilder::default()
+            .codex_home(home.path().to_path_buf())
+            .build()
+            .await?,
+    );
+    let auth =
+        AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false)
+            .await?;
+    let (processor, mut outgoing) = build_test_processor(config, auth).await;
+    let mut created = processor.thread_created_receiver();
+    let owner_id = ConnectionId(1);
+    let owner = Arc::new(ConnectionSessionState::new(ConnectionOrigin::WebSocket));
+    let result: Result<()> = async {
+        send_request(
+            &processor,
+            owner_id,
+            &owner,
+            json!({"id": 1, "method": "initialize", "params": {
+                "clientInfo": {"name": "parent-owner", "version": "1"},
+                "capabilities": {"experimentalApi": true}
+            }}),
+        )
+        .await;
+        let _: InitializeResponse =
+            read_response_from(&mut outgoing, owner_id, /*request_id*/ 1).await;
+        processor
+            .connection_initialized(owner_id, owner.request_attestation())
+            .await;
+        send_request(
+            &processor,
+            owner_id,
+            &owner,
+            json!({"id": 2, "method": "thread/start", "params": {"ephemeral": ephemeral}}),
+        )
+        .await;
+        let parent: ThreadStartResponse =
+            read_response_from(&mut outgoing, owner_id, /*request_id*/ 2).await;
+        send_request(
+            &processor,
+            owner_id,
+            &owner,
+            json!({"id": 3, "method": "turn/start", "params": {
+                "threadId": parent.thread.id,
+                "input": [{"type": "text", "text": "Spawn a child and finish"}]
+            }}),
+        )
+        .await;
+        let _: TurnStartResponse =
+            read_response_from(&mut outgoing, owner_id, /*request_id*/ 3).await;
+        timeout(
+            Duration::from_secs(/*secs*/ 10),
+            server.wait_for_request_count(/*count*/ 1),
+        )
+        .await?;
+
+        // Finish connection cleanup before the model can create a child. Dispatch the
+        // core creation event exactly as the app-server does with no initialized clients.
+        processor.connection_closed(owner_id, &owner).await;
+        release_spawn.send(()).expect("parent response is waiting");
+        let child_id = timeout(Duration::from_secs(/*secs*/ 10), created.recv()).await??;
+        let child_runtime = Arc::downgrade(
+            &processor
+                .thread_processor
+                .thread_manager
+                .get_thread(child_id)
+                .await?,
+        );
+        let attach = processor.try_attach_thread_listener(child_id, Vec::new());
+        tokio::pin!(attach);
+        // The old runtime has left core, but its app-server cleanup still owns the ID.
+        let pending = &processor.thread_processor.pending_thread_unloads;
+        pending
+            .lock()
+            .await
+            .insert(child_id, tokio::sync::watch::channel(()).0);
+        assert!(
+            futures::poll!(&mut attach).is_pending(),
+            "attachment must wait for old cleanup"
+        );
+        pending.lock().await.remove(&child_id);
+        attach.await;
+        // Keep the spawn response open until the child's model request arrives, so
+        // the next gated response belongs to the parent regardless of scheduling.
+        timeout(
+            Duration::from_secs(/*secs*/ 10),
+            server.wait_for_request_count(/*count*/ 2),
+        )
+        .await?;
+        release_spawn_completion
+            .send(())
+            .expect("spawn completion is waiting");
+        timeout(
+            Duration::from_secs(/*secs*/ 10),
+            server.wait_for_request_count(/*count*/ 3),
+        )
+        .await?;
+
+        // Installing an unsubscribed listener must not unload a still-running child.
+        let premature_close = timeout(Duration::from_millis(/*millis*/ 1200), async {
+            loop {
+                if let OutgoingEnvelope::Broadcast {
+                    message: OutgoingMessage::AppServerNotification(notification),
+                } = outgoing.recv().await.expect("outgoing channel stays open")
+                    && matches!(
+                        notification.notification,
+                        ServerNotification::ThreadClosed(_)
+                    )
+                {
+                    break;
+                }
+            }
+        })
+        .await;
+        assert!(premature_close.is_err(), "running threads must stay loaded");
+        if ephemeral {
+            // Complete the child while the parent's follow-up response remains gated.
+            let child = processor
+                .thread_processor
+                .thread_manager
+                .get_thread(child_id)
+                .await?;
+            release_first_completion
+                .send(())
+                .expect("child completion is waiting");
+            timeout(Duration::from_secs(/*secs*/ 10), async {
+                while !matches!(child.agent_status().await, AgentStatus::Completed(_)) {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .context("child did not complete")?;
+            tokio::time::sleep(Duration::from_secs(unload_delay_secs + 1)).await;
+            release_second_completion
+                .send(())
+                .expect("parent completion is waiting");
+
+            // Exercise the parent's real followup_task call and inspect the child's next
+            // model request: accepting a follow-up must preserve its earlier model context.
+            timeout(
+                Duration::from_secs(/*secs*/ 10),
+                server.wait_for_request_count(/*count*/ 5),
+            )
+            .await
+            .context("ephemeral child did not receive its follow-up")?;
+            let requests = server.requests().await;
+            let followup = requests[3..]
+                .iter()
+                .map(|request| serde_json::from_slice::<serde_json::Value>(request))
+                .collect::<std::result::Result<Vec<_>, _>>()?
+                .into_iter()
+                .find(|request| {
+                    request["input"]
+                        .to_string()
+                        .contains("Recall your previous reply")
+                        && !request["input"].as_array().is_some_and(|input| {
+                            input.iter().any(|item| {
+                                item["type"] == "function_call_output" && item["call_id"] == "spawn"
+                            })
+                        })
+                })
+                .context("missing child follow-up request")?;
+            let previous_reply = followup["input"]
+                .as_array()
+                .context("follow-up input")?
+                .iter()
+                .find(|item| item["role"] == "assistant")
+                .context("child's previous reply must survive the idle deadline")?;
+            assert_eq!(
+                previous_reply["content"],
+                json!([{"type": "output_text", "text": "Done"}])
+            );
+            return Ok(());
+        }
+        release_first_completion
+            .send(())
+            .expect("first completion is waiting");
+        release_second_completion
+            .send(())
+            .expect("second completion is waiting");
+
+        // Check both registry cleanup and runtime release, regardless of the order
+        // in which the parent and child finish unloading.
+        timeout(Duration::from_secs(/*secs*/ 10), async {
+            loop {
+                let loaded = processor
+                    .thread_processor
+                    .thread_manager
+                    .list_thread_ids()
+                    .await;
+                if loaded.is_empty() && child_runtime.upgrade().is_none() {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .context("parent and disconnected child still retain their runtimes")?;
+        Ok(())
+    }
+    .await;
+    drop(outgoing);
+    processor.clear_runtime_references();
+    processor.shutdown_threads().await;
+    server.shutdown().await;
+    result
+}
+
 async fn send_request(
     processor: &Arc<MessageProcessor>,
     connection_id: ConnectionId,

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

@@ -607,6 +607,7 @@ use crate::error_code::invalid_request;
 use crate::filters::compute_source_filters;
 use crate::filters::source_kind_matches;
 use crate::thread_state::ConnectionCapabilities;
+use crate::thread_state::PendingThreadUnloads;
 use crate::thread_state::ThreadListenerCommand;
 use crate::thread_state::ThreadState;
 use crate::thread_state::ThreadStateManager;

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

@@ -16,7 +16,7 @@ impl ThreadRequestProcessor {
         let mut snapshot = RecoverySnapshot::default();
         for thread_id in self.thread_manager.list_thread_ids().await {
             let pending_unloads = self.pending_thread_unloads.lock().await;
-            if pending_unloads.contains(&thread_id) {
+            if pending_unloads.contains_key(&thread_id) {
                 continue;
             }
             let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {

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

@@ -12,7 +12,7 @@ pub(super) struct ListenerTaskContext {
     pub(super) thread_manager: Arc<ThreadManager>,
     pub(super) thread_state_manager: ThreadStateManager,
     pub(super) outgoing: Arc<OutgoingMessageSender>,
-    pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+    pub(super) pending_thread_unloads: PendingThreadUnloads,
     pub(super) thread_watch_manager: ThreadWatchManager,
     pub(super) codex_home: PathBuf,
     pub(super) thread_unload_delay: Duration,
@@ -160,7 +160,7 @@ pub(super) async fn ensure_conversation_listener(
     };
     let (thread_state, result) = {
         let pending_thread_unloads = listener_task_context.pending_thread_unloads.lock().await;
-        if pending_thread_unloads.contains(&conversation_id) {
+        if pending_thread_unloads.contains_key(&conversation_id) {
             return Err(invalid_request(format!(
                 "thread {conversation_id} is closing; retry after the thread is closed"
             )));
@@ -394,13 +394,67 @@ pub(super) async fn ensure_listener_task_running(
                     }
                     {
                         let mut pending_thread_unloads = pending_thread_unloads.lock().await;
-                        if pending_thread_unloads.contains(&conversation_id) {
+                        if pending_thread_unloads.contains_key(&conversation_id) {
                             continue;
                         }
                         if !unloading_state.should_unload_now() {
                             continue;
                         }
-                        pending_thread_unloads.insert(conversation_id);
+                        pending_thread_unloads.insert(conversation_id, watch::channel(()).0);
+                    }
+                    if is_v2_child {
+                        let eviction = thread_manager.try_evict_v2_thread(Arc::clone(&conversation));
+                        tokio::pin!(eviction);
+                        let result = match tokio::time::timeout(
+                            Duration::from_secs(/*secs*/ 10), &mut eviction,
+                        ).await {
+                            Ok(result) => result,
+                            Err(_) => {
+                                warn!(event.name = "codex.app_server.thread_shutdown_slow", "thread {conversation_id} shutdown is taking longer than expected; continuing to wait");
+                                eviction.await
+                            }
+                        };
+                        let outcome = match result {
+                            Ok(ThreadEvictionOutcome::Busy) => {
+                                pending_thread_unloads.lock().await.remove(&conversation_id);
+                                unloading_state.note_thread_activity_observed();
+                                continue;
+                            }
+                            Ok(outcome @ (ThreadEvictionOutcome::Evicted | ThreadEvictionOutcome::NotCurrent)) => outcome,
+                            Err(err) => {
+                                warn!("failed to evict thread {conversation_id}: {err}");
+                                pending_thread_unloads.lock().await.remove(&conversation_id);
+                                unloading_state.note_thread_activity_observed();
+                                continue;
+                            }
+                        };
+                        {
+                            let mut pending = pending_thread_unloads.lock().await;
+                            if !thread_state_manager.remove_unsubscribed_listener(
+                                conversation_id, &thread_state, listener_generation,
+                            ).await {
+                                pending.remove(&conversation_id);
+                                break;
+                            }
+                        }
+                        // Release the status receiver before removing its watch entry.
+                        drop(unloading_state);
+                        outgoing_for_task
+                            .cancel_requests_for_thread(conversation_id, /*error*/ None)
+                            .await;
+                        // A runtime removed by someone else may already have a replacement.
+                        if outcome == ThreadEvictionOutcome::Evicted {
+                            thread_watch_manager
+                                .remove_thread(&conversation_id.to_string())
+                                .await;
+                            outgoing_for_task
+                                .send_server_notification(ServerNotification::ThreadClosed(
+                                    ThreadClosedNotification { thread_id: conversation_id.to_string() },
+                                ))
+                                .await;
+                        }
+                        pending_thread_unloads.lock().await.remove(&conversation_id);
+                        return;
                     }
                     if is_v2_child {
                         let eviction = thread_manager.try_evict_v2_thread(Arc::clone(&conversation));
@@ -491,7 +545,7 @@ pub(super) async fn wait_for_thread_shutdown(thread: &Arc<CodexThread>) -> Threa
 pub(super) async fn unload_thread_without_subscribers(
     thread_manager: Arc<ThreadManager>,
     outgoing: Arc<OutgoingMessageSender>,
-    pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+    pending_thread_unloads: PendingThreadUnloads,
     thread_state_manager: ThreadStateManager,
     thread_watch_manager: ThreadWatchManager,
     thread_id: ThreadId,
@@ -564,7 +618,7 @@ pub(super) async fn handle_thread_listener_command(
     thread_state: &Arc<Mutex<ThreadState>>,
     thread_watch_manager: &ThreadWatchManager,
     outgoing: &Arc<OutgoingMessageSender>,
-    pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
+    pending_thread_unloads: &PendingThreadUnloads,
     listener_command: ThreadListenerCommand,
 ) {
     match listener_command {
@@ -658,7 +712,7 @@ pub(super) async fn handle_pending_thread_resume_request(
     thread_state: &Arc<Mutex<ThreadState>>,
     thread_watch_manager: &ThreadWatchManager,
     outgoing: &Arc<OutgoingMessageSender>,
-    pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
+    pending_thread_unloads: &PendingThreadUnloads,
     mut pending: crate::thread_state::PendingThreadResumeRequest,
 ) {
     let (active_turn_metadata, active_turn) = {
@@ -777,7 +831,7 @@ pub(super) async fn handle_pending_thread_resume_request(
 
     {
         let pending_thread_unloads = pending_thread_unloads.lock().await;
-        if pending_thread_unloads.contains(&conversation_id) {
+        if pending_thread_unloads.contains_key(&conversation_id) {
             drop(pending_thread_unloads);
             outgoing
                 .send_error(

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

@@ -30,6 +30,7 @@ use codex_protocol::SanitizedGitUrl;
 use codex_protocol::config_types::MultiAgentMode;
 use codex_protocol::error::CodexErrorDetails;
 use codex_protocol::mcp::ClientMcpExtensions;
+use codex_protocol::protocol::MultiAgentVersion;
 use codex_protocol::protocol::ThreadHistoryMode;
 use codex_thread_store::PersistContext;
 use std::ops::ControlFlow;
@@ -446,13 +447,13 @@ fn validate_dynamic_tools(tools: &[DynamicToolSpec]) -> Result<(), String> {
 #[derive(Clone)]
 pub(crate) struct ThreadRequestProcessor {
     pub(super) auth_manager: Arc<AuthManager>,
-    pub(super) thread_manager: Arc<ThreadManager>,
+    pub(crate) thread_manager: Arc<ThreadManager>,
     pub(super) outgoing: Arc<OutgoingMessageSender>,
     pub(super) arg0_paths: Arg0DispatchPaths,
     pub(super) config: Arc<Config>,
     pub(super) config_manager: ConfigManager,
     pub(super) thread_store: Arc<dyn ThreadStore>,
-    pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+    pub(crate) pending_thread_unloads: PendingThreadUnloads,
     pub(super) thread_state_manager: ThreadStateManager,
     pub(super) thread_watch_manager: ThreadWatchManager,
     pub(super) thread_list_state_permit: Arc<Semaphore>,
@@ -492,7 +493,7 @@ impl ThreadRequestProcessor {
         config: Arc<Config>,
         config_manager: ConfigManager,
         thread_store: Arc<dyn ThreadStore>,
-        pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+        pending_thread_unloads: PendingThreadUnloads,
         thread_state_manager: ThreadStateManager,
         thread_watch_manager: ThreadWatchManager,
         thread_list_state_permit: Arc<Semaphore>,
@@ -1007,7 +1008,6 @@ impl ThreadRequestProcessor {
     }
 
     async fn finalize_thread_teardown(&self, thread_id: ThreadId) {
-        self.pending_thread_unloads.lock().await.remove(&thread_id);
         self.outgoing
             .cancel_requests_for_thread(thread_id, /*error*/ None)
             .await;
@@ -1017,6 +1017,7 @@ impl ThreadRequestProcessor {
         self.thread_watch_manager
             .remove_thread(&thread_id.to_string())
             .await;
+        self.pending_thread_unloads.lock().await.remove(&thread_id);
     }
 
     async fn thread_unsubscribe_response_inner(
@@ -3621,12 +3622,54 @@ impl ThreadRequestProcessor {
         self.thread_watch_manager.subscribe_running_turn_count()
     }
 
-    /// Best-effort: ensure initialized connections are subscribed to this thread.
+    /// Best-effort: start the lifecycle listener and subscribe initialized connections.
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "listener state creation must be serialized against pending unloads"
+    )]
     pub(crate) async fn try_attach_thread_listener(
         &self,
         thread_id: ThreadId,
         connection_ids: Vec<ConnectionId>,
     ) {
+        if connection_ids.is_empty() {
+            let result = loop {
+                let pending = self.pending_thread_unloads.lock().await;
+                if let Some(unload) = pending.get(&thread_id) {
+                    // A replacement can be published before the old listener finishes cleanup.
+                    let mut completion = unload.subscribe();
+                    drop(pending);
+                    let _ = completion.changed().await;
+                    continue;
+                }
+                let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
+                    return;
+                };
+                // Only persisted V2 children can reload after idle eviction.
+                if thread.multi_agent_version() != Some(MultiAgentVersion::V2)
+                    || thread.config().await.ephemeral
+                {
+                    return;
+                }
+                self.thread_watch_manager
+                    .upsert_thread(&thread_id.to_string())
+                    .await;
+                let thread_state = self.thread_state_manager.thread_state(thread_id).await;
+                let result = self
+                    .ensure_listener_task_running(thread_id, thread, thread_state)
+                    .await;
+                drop(pending);
+                break result;
+            };
+            if let Err(err) = result {
+                warn!(
+                    "failed to start listener for thread {thread_id}: {message}",
+                    message = err.message
+                );
+            }
+            return;
+        }
+
         let mut raw_events_enabled = false;
         if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
             let config_snapshot = thread.config_snapshot().await;
@@ -3669,7 +3712,7 @@ impl ThreadRequestProcessor {
                 .pending_thread_unloads
                 .lock()
                 .await
-                .contains(&thread_id)
+                .contains_key(&thread_id)
         {
             return Err(invalid_request(format!(
                 "thread {thread_id} is closing; retry thread/resume after the thread is closed"
@@ -3782,7 +3825,7 @@ impl ThreadRequestProcessor {
                 .pending_thread_unloads
                 .lock()
                 .await
-                .contains(&resumed.conversation_id)
+                .contains_key(&resumed.conversation_id)
         {
             return Err(invalid_request(format!(
                 "thread {} is closing; retry thread/resume after the thread is closed",

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

@@ -87,7 +87,7 @@ pub(crate) struct TurnRequestProcessor {
     analytics_events_client: AnalyticsEventsClient,
     config: Arc<Config>,
     config_manager: ConfigManager,
-    pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+    pending_thread_unloads: PendingThreadUnloads,
     thread_state_manager: ThreadStateManager,
     thread_watch_manager: ThreadWatchManager,
     skills_watcher: Arc<SkillsWatcher>,
@@ -149,7 +149,7 @@ impl TurnRequestProcessor {
         analytics_events_client: AnalyticsEventsClient,
         config: Arc<Config>,
         config_manager: ConfigManager,
-        pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+        pending_thread_unloads: PendingThreadUnloads,
         thread_state_manager: ThreadStateManager,
         thread_watch_manager: ThreadWatchManager,
         skills_watcher: Arc<SkillsWatcher>,

codex-rs/app-server/src/thread_state.rs

@@ -35,6 +35,9 @@ use tracing::error;
 
 type PendingInterruptQueue = Vec<ConnectionRequestId>;
 
+// Removing an unload entry drops its sender and wakes deferred listener attachment.
+pub(crate) type PendingThreadUnloads = Arc<Mutex<HashMap<ThreadId, watch::Sender<()>>>>;
+
 pub(crate) struct PendingThreadResumeRequest {
     pub(crate) request_id: ConnectionRequestId,
     pub(crate) history_items: Vec<RolloutItem>,

codex-rs/config/src/config_toml.rs

@@ -344,7 +344,8 @@ pub struct ConfigToml {
     pub background_terminal_max_timeout: Option<u64>,
 
     /// Seconds a thread must have no subscribers and no activity before app-server
-    /// unloads it. Defaults to 60; zero unloads immediately. Changes require a server restart.
+    /// unloads it. Defaults to 1800 (30 minutes); zero unloads immediately.
+    /// Changes require a server restart.
     pub thread_unload_delay_secs: Option<u64>,
 
     /// Deprecated: ignored.

codex-rs/core/config.schema.json

@@ -7920,7 +7920,7 @@
       "type": "boolean"
     },
     "thread_unload_delay_secs": {
-      "description": "Seconds a thread must have no subscribers and no activity before app-server unloads it. Defaults to 60; zero unloads immediately. Changes require a server restart.",
+      "description": "Seconds a thread must have no subscribers and no activity before app-server unloads it. Defaults to 1800 (30 minutes); zero unloads immediately. Changes require a server restart.",
       "format": "uint64",
       "minimum": 0.0,
       "type": "integer"

codex-rs/core/src/config/config_tests.rs

@@ -255,9 +255,9 @@ async fn load_config_applies_optional_mcp_startup_grace() -> std::io::Result<()>
 async fn load_config_resolves_thread_unload_delay() -> anyhow::Result<()> {
     let codex_home = tempdir()?;
     for (toml, seconds) in [
-        ("", 60),
+        ("", 1800),
         ("thread_unload_delay_secs = 0", 0),
-        ("thread_unload_delay_secs = 1800", 1800),
+        ("thread_unload_delay_secs = 60", 60),
     ] {
         let config = Config::load_from_base_config_with_overrides(
             toml::from_str(toml)?,

codex-rs/core/src/config/mod.rs

@@ -3944,7 +3944,7 @@ impl Config {
             .unwrap_or(DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS)
             .max(MIN_EMPTY_YIELD_TIME_MS);
         let thread_unload_delay =
-            Duration::from_secs(cfg.thread_unload_delay_secs.unwrap_or(/*default*/ 60));
+            Duration::from_secs(cfg.thread_unload_delay_secs.unwrap_or(/*default*/ 1800));
         if std::time::Instant::now()
             .checked_add(thread_unload_delay)
             .is_none()

codex-rs/thread-manager-sample/src/main.rs

@@ -373,7 +373,7 @@ async fn new_config(
         tool_registry: Default::default(),
         code_mode: Default::default(),
         background_terminal_max_timeout: 300_000,
-        thread_unload_delay: std::time::Duration::from_secs(60),
+        thread_unload_delay: std::time::Duration::from_secs(/*secs*/ 1800),
         ghost_snapshot: GhostSnapshotConfig::default(),
         multi_agent_v2: MultiAgentV2Config::default(),
         max_goal_token_budget: None,