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,