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 a7754975b5..cdadea9724 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/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>, ) -> 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); 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 00faaafe95..e295fa2014 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/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(()) } 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 289859b26a..204018224d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs +++ b/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(()) }