diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 77ef348efc..479b58e4fb 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 1746ceb7bc..4905a31d14 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -547,8 +547,8 @@ pub struct ThreadForkParams { /// Start from the loaded parent's effective config, tools, environments, and /// persisted history to maximize prompt-cache reuse. Requires a loaded, /// persisted parent and `ephemeral: true`. Enables `features.reasoning_effort_override` - /// on the fork. Cannot be combined with configuration overrides, `path`, `lastTurnId`, - /// or `beforeTurnId`. + /// on the fork. Only `config.thread_unload_delay_secs` may override the inherited + /// config. Cannot be combined with `path`, `lastTurnId`, or `beforeTurnId`. #[experimental("thread/fork.experimentalPredictionMode")] #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub experimental_prediction_mode: bool, 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 cdadea9724..e6864c4dca 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -15,7 +15,6 @@ pub(super) struct ListenerTaskContext { pub(super) pending_thread_unloads: PendingThreadUnloads, pub(super) thread_watch_manager: ThreadWatchManager, pub(super) codex_home: PathBuf, - pub(super) thread_unload_delay: Duration, pub(super) skills_watcher: Arc, pub(super) turn_cost_worker: Option, } @@ -238,25 +237,18 @@ pub(super) async fn ensure_listener_task_running( ) -> Result<(), JSONRPCErrorError> { let (cancel_tx, mut cancel_rx) = oneshot::channel(); let config_snapshot = conversation.config_snapshot().await; - let thread_unload_delay = if config_snapshot.ephemeral - && config_snapshot - .thread_source - .as_ref() - .is_some_and(|source| source.as_str() == "composer_predictions") - && !matches!(config_snapshot.session_source, SessionSource::SubAgent(_)) - { - Duration::ZERO - } else { - listener_task_context.thread_unload_delay - }; - let Some(mut unloading_state) = - UnloadingState::new(&listener_task_context, conversation_id, thread_unload_delay).await + let config = conversation.config().await; + let Some(mut unloading_state) = UnloadingState::new( + &listener_task_context, + conversation_id, + config.thread_unload_delay, + ) + .await else { return Err(invalid_request(format!( "thread {conversation_id} is closing; retry after the thread is closed" ))); }; - let config = conversation.config().await; let environments = conversation.environment_selections().await; let watch_registration = listener_task_context .skills_watcher 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 9563a250cc..8c17fb0340 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1102,7 +1102,6 @@ impl ThreadRequestProcessor { pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), thread_watch_manager: self.thread_watch_manager.clone(), codex_home: self.config.codex_home.to_path_buf(), - thread_unload_delay: self.config.thread_unload_delay, skills_watcher: Arc::clone(&self.skills_watcher), turn_cost_worker: self.turn_cost_worker.clone(), } @@ -1233,7 +1232,6 @@ impl ThreadRequestProcessor { pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), thread_watch_manager: self.thread_watch_manager.clone(), codex_home: self.config.codex_home.to_path_buf(), - thread_unload_delay: self.config.thread_unload_delay, skills_watcher: Arc::clone(&self.skills_watcher), turn_cost_worker: self.turn_cost_worker.clone(), }; @@ -5105,7 +5103,7 @@ impl ThreadRequestProcessor { || params .config .as_ref() - .is_some_and(|config| !config.is_empty()); + .is_some_and(|config| config.keys().any(|key| key != "thread_unload_delay_secs")); if params.experimental_prediction_mode && has_config_overrides { return Err(invalid_request( "`experimentalPredictionMode` cannot be combined with configuration overrides", @@ -5175,6 +5173,20 @@ impl ThreadRequestProcessor { _ => internal_error(format!("failed to inherit parent state: {err}")), })?; options.config.ephemeral = ephemeral; + // Runtime retention does not change the inherited prompt or cache key. + if let Some(value) = cli_overrides + .as_ref() + .and_then(|config| config.get("thread_unload_delay_secs")) + { + let seconds = value.as_u64().ok_or_else(|| { + invalid_request("`thread_unload_delay_secs` must be a non-negative integer") + })?; + let delay = Duration::from_secs(seconds); + if std::time::Instant::now().checked_add(delay).is_none() { + return Err(invalid_request("`thread_unload_delay_secs` is too large")); + } + options.config.thread_unload_delay = delay; + } options .thread_extension_init .insert(codex_core::compact::AutoCompactionDisabled); 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 de3bdf6f5d..1ce59c4cf4 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1691,7 +1691,6 @@ impl TurnRequestProcessor { pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), thread_watch_manager: self.thread_watch_manager.clone(), codex_home: self.config.codex_home.to_path_buf(), - thread_unload_delay: self.config.thread_unload_delay, skills_watcher: Arc::clone(&self.skills_watcher), turn_cost_worker: self.turn_cost_worker.clone(), } diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index e295fa2014..9182245cf1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1569,6 +1569,7 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { let fork_id = mcp .send_thread_fork_request(ThreadForkParams { thread_id: conversation_id.clone(), + config: Some([("thread_unload_delay_secs".into(), json!(0))].into()), ..Default::default() }) .await?; @@ -1673,6 +1674,40 @@ async fn thread_fork_creates_reference_backed_paginated_thread() -> Result<()> { .thread_id, ThreadId::from_string(forked_thread_id.as_str())? ); + let _: ThreadUnsubscribeResponse = mcp + .request(|request_id| ClientRequest::ThreadUnsubscribe { + request_id, + params: ThreadUnsubscribeParams { + thread_id: forked_thread_id.clone(), + }, + }) + .await?; + let closed: ThreadClosedNotification = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("thread/closed")).await??; + assert_eq!( + closed, + ThreadClosedNotification { + thread_id: forked_thread_id + } + ); + + // The nested fork did not inherit its parent's unload delay. + let _: ThreadUnsubscribeResponse = mcp + .request(|request_id| ClientRequest::ThreadUnsubscribe { + request_id, + params: ThreadUnsubscribeParams { + thread_id: nested_thread.id, + }, + }) + .await?; + assert!( + timeout( + std::time::Duration::from_millis(250), + mcp.read_stream_until_notification_message("thread/closed"), + ) + .await + .is_err() + ); Ok(()) } @@ -2779,13 +2814,37 @@ async fn inheriting_fork_preserves_parent_config(history_mode: ThreadHistoryMode }) .await?; + for (delay, message) in [ + ( + json!(-1), + "`thread_unload_delay_secs` must be a non-negative integer", + ), + (json!(u64::MAX), "`thread_unload_delay_secs` is too large"), + ] { + let request_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: parent.thread.id.clone(), + experimental_prediction_mode: true, + ephemeral: true, + config: Some([("thread_unload_delay_secs".into(), delay)].into()), + ..Default::default() + }) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(error.error.message, message); + } + let fork_id = mcp .send_thread_fork_request(ThreadForkParams { thread_id: parent.thread.id.clone(), experimental_prediction_mode: true, ephemeral: true, exclude_turns: true, - thread_source: Some(ThreadSource::Feature("composer_predictions".into())), + config: Some([("thread_unload_delay_secs".into(), json!(0))].into()), ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs index 204018224d..67e5d1bdda 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs @@ -199,15 +199,34 @@ async fn thread_unsubscribe_keeps_thread_loaded_until_idle_timeout() -> Result<( #[test_case(0, 0, ThreadStartParams::default(); "zero_delay")] #[test_case(1, 1, ThreadStartParams::default(); "one_second_delay")] +#[test_case( + 0, + 1, + ThreadStartParams { + ephemeral: Some(true), + config: Some([("thread_unload_delay_secs".into(), json!(1))].into()), + ..Default::default() + }; + "per_thread_one_second_delay" +)] #[test_case( 1800, 0, ThreadStartParams { ephemeral: Some(true), - thread_source: Some(ThreadSource::Feature("composer_predictions".into())), + config: Some([("thread_unload_delay_secs".into(), json!(0))].into()), ..Default::default() }; - "composer_prediction" + "immediate_ephemeral" +)] +#[test_case( + 1800, + 0, + ThreadStartParams { + config: Some([("thread_unload_delay_secs".into(), json!(0))].into()), + ..Default::default() + }; + "immediate_persistent" )] #[tokio::test] async fn thread_unsubscribe_during_turn_keeps_turn_running( @@ -215,6 +234,8 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running( expected_delay_secs: u64, thread_params: ThreadStartParams, ) -> Result<()> { + let check_cold_resume = + thread_params.config.is_some() && !thread_params.ephemeral.unwrap_or_default(); let call_id = "deterministic-wait-call"; let tool_name = "deterministic_wait"; let tool_args = json!({}); @@ -378,7 +399,41 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running( } let closed: ThreadClosedNotification = timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("thread/closed")).await??; - assert_eq!(closed, ThreadClosedNotification { thread_id }); + assert_eq!( + closed, + ThreadClosedNotification { + thread_id: thread_id.clone() + } + ); + if check_cold_resume { + // Immediate unloading preserves durable history. A cold resume uses the + // server's default delay rather than restoring the old runtime's override. + let resume: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread_id.clone(), + ..Default::default() + }, + }) + .await?; + assert_eq!(resume.thread.id, thread_id); + assert_eq!(resume.thread.turns.len(), 1); + let _: ThreadUnsubscribeResponse = mcp + .request(|request_id| ClientRequest::ThreadUnsubscribe { + request_id, + params: ThreadUnsubscribeParams { thread_id }, + }) + .await?; + assert!( + timeout( + std::time::Duration::from_millis(250), + mcp.read_stream_until_notification_message("thread/closed"), + ) + .await + .is_err() + ); + } server.shutdown().await; Ok(()) @@ -475,7 +530,10 @@ async fn thread_unsubscribe_preserves_cached_status_before_idle_unload() -> Resu } #[test_case(ThreadStartParams::default(); "persistent")] -#[test_case(ThreadStartParams { ephemeral: Some(true), ..Default::default() }; "ephemeral")] +#[test_case(ThreadStartParams { + ephemeral: Some(true), + ..Default::default() +}; "ephemeral")] #[test_case( ThreadStartParams { thread_source: Some(ThreadSource::Feature("composer_predictions".into())), diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 8c8c66ea1d..e1b08bf12b 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -348,7 +348,7 @@ pub struct ConfigToml { /// 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. + /// Resolved when each thread is loaded; request config overrides apply per thread. 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 e91aa7a3d5..e6ea77f28b 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -8164,7 +8164,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 1800 (30 minutes); 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. Resolved when each thread is loaded; request config overrides apply per thread.", "format": "uint64", "minimum": 0.0, "type": "integer" diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 2b99142597..ff6c9eca1d 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1127,7 +1127,7 @@ pub struct Config { /// Default: `300000` (5 minutes). pub background_terminal_max_timeout: u64, - /// Idle timeout for unsubscribed app-server threads, resolved at server startup. + /// Idle timeout for unsubscribed app-server threads. pub thread_unload_delay: Duration, /// Compatibility-only settings retained for legacy `ghost_snapshot` diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index dccb70d576..fd09c4091c 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2819,7 +2819,10 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - } // Cold-resume the persisted V2 root so its children are registered before selection. - app.config.thread_unload_delay = Duration::ZERO; + std::fs::write( + app.config.codex_home.join("config.toml"), + "thread_unload_delay_secs = 0\n", + )?; let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker( &app.config, )) diff --git a/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs b/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs index 0fed6fdf49..6daf28c9c6 100644 --- a/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs +++ b/codex-rs/tui/src/app/tests/resume_shutdown_tests.rs @@ -11,7 +11,10 @@ use tokio::net::TcpListener; #[tokio::test] async fn thread_close_exits_only_when_displayed_thread_is_unloaded() -> Result<()> { let (mut app, mut events, _commands) = make_test_app_with_channels().await; - app.config.thread_unload_delay = Duration::ZERO; + std::fs::write( + app.config.codex_home.join("config.toml"), + "thread_unload_delay_secs = 0\n", + )?; let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; let started = server.start_thread(&app.config).await?; let thread_id = started.session.thread_id;