From 2fdf047c9631c9ed01a31b62efb7891718a931a8 Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 8 Oct 2026 11:41:59 +0000 Subject: [PATCH] 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 | 3 +- ...essage_processor_thread_lifecycle_tests.rs | 299 +++++++++++++++++- codex-rs/app-server/src/request_processors.rs | 1 + .../src/request_processors/daemon_snapshot.rs | 2 +- .../request_processors/thread_lifecycle.rs | 70 +++- .../request_processors/thread_processor.rs | 57 +++- .../src/request_processors/turn_processor.rs | 4 +- codex-rs/app-server/src/thread_state.rs | 3 + codex-rs/config/src/config_toml.rs | 3 +- codex-rs/core/config.schema.json | 2 +- codex-rs/core/src/config/config_tests.rs | 4 +- codex-rs/core/src/config/mod.rs | 2 +- codex-rs/thread-manager-sample/src/main.rs | 2 +- 13 files changed, 426 insertions(+), 26 deletions(-) diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index e5434177b2..90f5146ebe 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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)); diff --git a/codex-rs/app-server/src/message_processor_thread_lifecycle_tests.rs b/codex-rs/app-server/src/message_processor_thread_lifecycle_tests.rs index 930a4d7a32..752452815d 100644 --- a/codex-rs/app-server/src/message_processor_thread_lifecycle_tests.rs +++ b/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::(request)) + .collect::, _>>()? + .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, connection_id: ConnectionId, diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 3d3a1df988..22249329ed 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/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; diff --git a/codex-rs/app-server/src/request_processors/daemon_snapshot.rs b/codex-rs/app-server/src/request_processors/daemon_snapshot.rs index 038373f3a1..81462202d1 100644 --- a/codex-rs/app-server/src/request_processors/daemon_snapshot.rs +++ b/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 { diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 1c7535455a..4418577ee6 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -12,7 +12,7 @@ pub(super) struct ListenerTaskContext { pub(super) thread_manager: Arc, pub(super) thread_state_manager: ThreadStateManager, pub(super) outgoing: Arc, - pub(super) pending_thread_unloads: Arc>>, + 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) -> Threa pub(super) async fn unload_thread_without_subscribers( thread_manager: Arc, outgoing: Arc, - pending_thread_unloads: Arc>>, + 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>, thread_watch_manager: &ThreadWatchManager, outgoing: &Arc, - pending_thread_unloads: &Arc>>, + 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>, thread_watch_manager: &ThreadWatchManager, outgoing: &Arc, - pending_thread_unloads: &Arc>>, + 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( diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 123a52e9ca..efad426991 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/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, - pub(super) thread_manager: Arc, + pub(crate) thread_manager: Arc, pub(super) outgoing: Arc, pub(super) arg0_paths: Arg0DispatchPaths, pub(super) config: Arc, pub(super) config_manager: ConfigManager, pub(super) thread_store: Arc, - pub(super) pending_thread_unloads: Arc>>, + 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, @@ -492,7 +493,7 @@ impl ThreadRequestProcessor { config: Arc, config_manager: ConfigManager, thread_store: Arc, - pending_thread_unloads: Arc>>, + pending_thread_unloads: PendingThreadUnloads, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, thread_list_state_permit: Arc, @@ -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, ) { + 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", diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 67946c8ebc..9316aa3f0b 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/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_manager: ConfigManager, - pending_thread_unloads: Arc>>, + pending_thread_unloads: PendingThreadUnloads, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, skills_watcher: Arc, @@ -149,7 +149,7 @@ impl TurnRequestProcessor { analytics_events_client: AnalyticsEventsClient, config: Arc, config_manager: ConfigManager, - pending_thread_unloads: Arc>>, + pending_thread_unloads: PendingThreadUnloads, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, skills_watcher: Arc, diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 7ef7e0f20e..6dfe969ab3 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -35,6 +35,9 @@ use tracing::error; type PendingInterruptQueue = Vec; +// Removing an unload entry drops its sender and wakes deferred listener attachment. +pub(crate) type PendingThreadUnloads = Arc>>>; + pub(crate) struct PendingThreadResumeRequest { pub(crate) request_id: ConnectionRequestId, pub(crate) history_items: Vec, diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 3e5cc07e6c..5f56c919ea 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -344,7 +344,8 @@ pub struct ConfigToml { pub background_terminal_max_timeout: Option, /// 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, /// Deprecated: ignored. diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 4133cc374e..9fdcc89e7f 100644 --- a/codex-rs/core/config.schema.json +++ b/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" diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 81000cb7c7..2ec201aff2 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/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)?, diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 062e7db10b..0537ee4adb 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/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() diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index a70b8c9817..640f3a1855 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/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,