Honor per-thread unload delays in app-server (#55270)

## Why

Thread unload timing used the server-wide delay, ignoring per-thread configuration overrides. Clients need to control how long individual idle, unsubscribed threads remain loaded.

## What changed

- Resolve unload timing from each loaded thread's `thread_unload_delay_secs` configuration. Setting it to `0` allows immediate unloading once the thread has no subscribers or activity.
- Allow this runtime retention override in `experimentalPredictionMode` forks while continuing to reject other configuration overrides. Reject negative values and delays too large to schedule.
- Replace the source-specific immediate-unload behavior with explicit configuration.

## Testing

Extend integration coverage for per-thread delays, unloading after active turns finish, fork overrides, and invalid prediction-fork delays. Verify that immediate unloading preserves persistent history and that cold resumes and ordinary nested forks use their default delay instead of retaining the previous runtime's override.

GitOrigin-RevId: fcb82bd73a101260c51e2a8476f94320c23ac929

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<SkillsWatcher>,
     pub(super) turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
 }
@@ -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

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);

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(),
         }