Unload idle ephemeral composer prediction threads without delay (#55255)

## What changed

Set the app server's unload delay to zero for ephemeral threads whose `thread_source` is `composer_predictions`, excluding subagent sessions. These threads unload once they are idle and have no subscribers, while active turns continue to completion.

## Testing

Extend integration coverage to check immediate unloading after an active prediction turn finishes and after unsubscribing from a fork that inherits the prediction configuration. Verify that persistent threads and ephemeral threads without the prediction source retain the configured delay.

GitOrigin-RevId: 04a0e405b666c689ccf6f482055d24692b18ead0

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

@@ -237,12 +237,20 @@ pub(super) async fn ensure_listener_task_running(
     thread_state: Arc<Mutex<ThreadState>>,
 ) -> Result<(), JSONRPCErrorError> {
     let (cancel_tx, mut cancel_rx) = oneshot::channel();
-    let Some(mut unloading_state) = UnloadingState::new(
-        &listener_task_context,
-        conversation_id,
-        listener_task_context.thread_unload_delay,
-    )
-    .await
+    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
     else {
         return Err(invalid_request(format!(
             "thread {conversation_id} is closing; retry after the thread is closed"
@@ -258,7 +266,6 @@ pub(super) async fn ensure_listener_task_running(
             &environments,
         )
         .await;
-    let config_snapshot = conversation.config_snapshot().await;
     let is_v2_child = conversation.multi_agent_version() == Some(MultiAgentVersion::V2)
         && matches!(config_snapshot.session_source, SessionSource::SubAgent(_));
     let thread_settings_baseline = thread_settings_from_config_snapshot(&config_snapshot);

codex-rs/app-server/tests/suite/v2/thread_fork.rs

@@ -32,6 +32,7 @@ use codex_app_server_protocol::ThreadAttachmentListParams;
 use codex_app_server_protocol::ThreadAttachmentListResponse;
 use codex_app_server_protocol::ThreadAttachmentRemoveParams;
 use codex_app_server_protocol::ThreadAttachmentRemoveResponse;
+use codex_app_server_protocol::ThreadClosedNotification;
 use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadForkResponse;
 use codex_app_server_protocol::ThreadHistoryMode;
@@ -52,6 +53,8 @@ use codex_app_server_protocol::ThreadStatus;
 use codex_app_server_protocol::ThreadStatusChangedNotification;
 use codex_app_server_protocol::ThreadTurnsListParams;
 use codex_app_server_protocol::ThreadTurnsListResponse;
+use codex_app_server_protocol::ThreadUnsubscribeParams;
+use codex_app_server_protocol::ThreadUnsubscribeResponse;
 use codex_app_server_protocol::TurnItemsView;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
@@ -2782,6 +2785,7 @@ async fn inheriting_fork_preserves_parent_config(history_mode: ThreadHistoryMode
             experimental_prediction_mode: true,
             ephemeral: true,
             exclude_turns: true,
+            thread_source: Some(ThreadSource::Feature("composer_predictions".into())),
             ..Default::default()
         })
         .await?;
@@ -2837,6 +2841,22 @@ async fn inheriting_fork_preserves_parent_config(history_mode: ThreadHistoryMode
             .to_string()
             .contains("parent_tool")
     );
+    let _: ThreadUnsubscribeResponse = mcp
+        .request(|request_id| ClientRequest::ThreadUnsubscribe {
+            request_id,
+            params: ThreadUnsubscribeParams {
+                thread_id: child.thread.id.clone(),
+            },
+        })
+        .await?;
+    let closed: ThreadClosedNotification =
+        timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("thread/closed")).await??;
+    assert_eq!(
+        closed,
+        ThreadClosedNotification {
+            thread_id: child.thread.id
+        }
+    );
     Ok(())
 }
 

codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs

@@ -19,6 +19,7 @@ use codex_app_server_protocol::ThreadReadParams;
 use codex_app_server_protocol::ThreadReadResponse;
 use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadResumeResponse;
+use codex_app_server_protocol::ThreadSource;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadStatus;
@@ -196,10 +197,24 @@ async fn thread_unsubscribe_keeps_thread_loaded_until_idle_timeout() -> Result<(
     Ok(())
 }
 
-#[test_case(0; "zero_delay")]
-#[test_case(1; "one_second_delay")]
+#[test_case(0, 0, ThreadStartParams::default(); "zero_delay")]
+#[test_case(1, 1, ThreadStartParams::default(); "one_second_delay")]
+#[test_case(
+    1800,
+    0,
+    ThreadStartParams {
+        ephemeral: Some(true),
+        thread_source: Some(ThreadSource::Feature("composer_predictions".into())),
+        ..Default::default()
+    };
+    "composer_prediction"
+)]
 #[tokio::test]
-async fn thread_unsubscribe_during_turn_keeps_turn_running(delay_secs: u64) -> Result<()> {
+async fn thread_unsubscribe_during_turn_keeps_turn_running(
+    delay_secs: u64,
+    expected_delay_secs: u64,
+    thread_params: ThreadStartParams,
+) -> Result<()> {
     let call_id = "deterministic-wait-call";
     let tool_name = "deterministic_wait";
     let tool_args = json!({});
@@ -253,7 +268,7 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running(delay_secs: u64) -> R
                 }),
                 defer_loading: false,
             })]),
-            ..Default::default()
+            ..thread_params
         })
         .await?;
     let thread_id = thread.id;
@@ -350,7 +365,7 @@ async fn thread_unsubscribe_during_turn_keeps_turn_running(delay_secs: u64) -> R
     )
     .await?;
     timeout(DEFAULT_READ_TIMEOUT, final_response_completed).await??;
-    if delay_secs > 0 {
+    if expected_delay_secs > 0 {
         // Once the turn finishes, inactivity starts a fresh countdown.
         assert!(
             timeout(
@@ -459,8 +474,19 @@ async fn thread_unsubscribe_preserves_cached_status_before_idle_unload() -> Resu
     Ok(())
 }
 
+#[test_case(ThreadStartParams::default(); "persistent")]
+#[test_case(ThreadStartParams { ephemeral: Some(true), ..Default::default() }; "ephemeral")]
+#[test_case(
+    ThreadStartParams {
+        thread_source: Some(ThreadSource::Feature("composer_predictions".into())),
+        ..Default::default()
+    };
+    "persistent_composer_prediction"
+)]
 #[tokio::test]
-async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload() -> Result<()> {
+async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload(
+    thread_params: ThreadStartParams,
+) -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     MockResponsesConfig::new(&server.uri())
@@ -475,7 +501,7 @@ async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload() -> Resul
     let ThreadStartResponse { thread, .. } = mcp
         .start_thread(ThreadStartParams {
             model: Some("mock-model".to_string()),
-            ..Default::default()
+            ..thread_params
         })
         .await?;
     let thread_id = thread.id;
@@ -503,6 +529,14 @@ async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload() -> Resul
         second_unsubscribe.status,
         ThreadUnsubscribeStatus::NotSubscribed
     );
+    assert!(
+        timeout(
+            std::time::Duration::from_millis(250),
+            mcp.read_stream_until_notification_message("thread/closed"),
+        )
+        .await
+        .is_err()
+    );
 
     Ok(())
 }