From a1d4aea2652768445e9eb7eb1bfbf2f08f03f7ce Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 23 Aug 2026 03:17:18 +0000 Subject: [PATCH] Shut down resumed descendants when archiving thread trees (#40179) ## Why Collaboration can resume an archived descendant without unarchiving its rollout. Archiving the parent again previously skipped shutdown for that descendant because only rollouts newly marked as archived were prepared. ## What changed Prepare every loaded thread in the spawn subtree for archival, including descendants whose rollouts are already archived, before updating archive state. ## Testing Add an app-server integration test that resumes an archived child through its parent and verifies that archiving the parent unloads the entire subtree. GitOrigin-RevId: aca8c0d6ddb8900aad16becbeb594426cea1e3a6 --- .../request_processors/thread_processor.rs | 6 +- .../tests/suite/v2/thread_archive.rs | 178 ++++++++++++++++++ 2 files changed, 182 insertions(+), 2 deletions(-) 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 afea864aff..c648db95ab 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1644,8 +1644,10 @@ impl ThreadRequestProcessor { } archive_thread_ids[1..].reverse(); - for &thread_id_to_archive in &archive_thread_ids { - self.prepare_thread_for_archive(thread_id_to_archive).await; + // Collaboration may resume an archived descendant without unarchiving it. + self.prepare_thread_for_archive(thread_id).await; + for &descendant_thread_id in subtree_thread_ids.iter().skip(1).rev() { + self.prepare_thread_for_archive(descendant_thread_id).await; } let archived_thread_ids = self diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 48a239cc3e..290d0bc4f1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -1,6 +1,7 @@ use anyhow::Result; use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; +use app_test_support::create_fake_paginated_rollout; use app_test_support::create_fake_rollout; use app_test_support::create_mock_responses_server_repeating_assistant; use codex_app_server_protocol::ClientInfo; @@ -11,6 +12,8 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadArchivedNotification; use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadLoadedListParams; +use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSource; @@ -19,16 +22,19 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadUnarchiveParams; use codex_app_server_protocol::ThreadUnarchiveResponse; +use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_core::find_archived_thread_path_by_id_str; use codex_core::find_thread_path_by_id_str; +use codex_features::Feature; use codex_protocol::ThreadId; use codex_state::DirectionalThreadSpawnEdgeStatus; use codex_state::StateRuntime; use codex_utils_absolute_path::test_support::PathExt; +use core_test_support::responses; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; @@ -41,6 +47,12 @@ use super::analytics::wait_for_matching_analytics_event; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +fn body_contains(request: &wiremock::Request, text: &str) -> bool { + String::from_utf8(request.body.clone()) + .ok() + .is_some_and(|body| body.contains(text)) +} + #[tokio::test] async fn thread_archive_rejects_owned_unmaterialized_paginated_descendant() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; @@ -110,6 +122,172 @@ async fn thread_archive_rejects_owned_unmaterialized_paginated_descendant() -> R Ok(()) } +#[tokio::test] +async fn thread_archive_shuts_down_resumed_archived_descendant() -> Result<()> { + const RESUME_PROMPT: &str = "resume the child"; + const RESUME_CALL_ID: &str = "resume-call-1"; + + let server = responses::start_mock_server().await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .enable_feature(Feature::Collab) + .write(codex_home.path())?; + let parent_id = create_fake_paginated_rollout( + codex_home.path(), + "2025-01-01T00-00-00", + "2025-01-01T00:00:00Z", + "parent", + Some("mock_provider"), + /*git_info*/ None, + )?; + let child_id = create_fake_paginated_rollout( + codex_home.path(), + "2025-01-01T00-01-00", + "2025-01-01T00:01:00Z", + "child", + Some("mock_provider"), + /*git_info*/ None, + )?; + let parent_thread_id = ThreadId::from_string(&parent_id)?; + let child_thread_id = ThreadId::from_string(&child_id)?; + let state_db = StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + "mock_provider".into(), + ) + .await?; + state_db + .upsert_thread_spawn_edge( + parent_thread_id, + child_thread_id, + DirectionalThreadSpawnEdgeStatus::Closed, + ) + .await?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + + let _: ThreadArchiveResponse = mcp + .request(|request_id| ClientRequest::ThreadArchive { + request_id, + params: ThreadArchiveParams { + thread_id: parent_id.clone(), + }, + }) + .await?; + for _ in 0..2 { + let _: ThreadArchivedNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_notification("thread/archived"), + ) + .await??; + } + + let _: ThreadUnarchiveResponse = mcp + .request(|request_id| ClientRequest::ThreadUnarchive { + request_id, + params: ThreadUnarchiveParams { + thread_id: parent_id.clone(), + }, + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/unarchived"), + ) + .await??; + let _: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: parent_id.clone(), + model: Some("gpt-5.4".to_string()), + ..Default::default() + }, + }) + .await?; + + let resume_args = serde_json::to_string(&json!({ "id": child_thread_id }))?; + let _parent_resume = responses::mount_sse_once_match( + &server, + |request: &wiremock::Request| { + body_contains(request, RESUME_PROMPT) && !body_contains(request, RESUME_CALL_ID) + }, + responses::sse(vec![ + responses::ev_response_created("resp-parent-resume"), + responses::ev_function_call_with_namespace( + RESUME_CALL_ID, + "multi_agent_v1", + "resume_agent", + &resume_args, + ), + responses::ev_completed("resp-parent-resume"), + ]), + ) + .await; + let _parent_resume_follow_up = responses::mount_sse_once_match( + &server, + |request: &wiremock::Request| body_contains(request, RESUME_CALL_ID), + responses::sse(vec![ + responses::ev_response_created("resp-parent-resume-follow-up"), + responses::ev_assistant_message("msg-parent-resume", "parent done"), + responses::ev_completed("resp-parent-resume-follow-up"), + ]), + ) + .await; + + let _: TurnStartResponse = mcp + .request(|request_id| ClientRequest::TurnStart { + request_id, + params: TurnStartParams { + thread_id: parent_id.clone(), + input: vec![UserInput::Text { + text: RESUME_PROMPT.to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + }) + .await?; + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let completed: TurnCompletedNotification = + mcp.read_notification("turn/completed").await?; + if completed.thread_id == parent_id { + return Ok::<(), anyhow::Error>(()); + } + } + }) + .await??; + + let ThreadLoadedListResponse { data, .. } = mcp + .request(|request_id| ClientRequest::ThreadLoadedList { + request_id, + params: ThreadLoadedListParams::default(), + }) + .await?; + assert!(data.contains(&child_id)); + + let _: ThreadArchiveResponse = mcp + .request(|request_id| ClientRequest::ThreadArchive { + request_id, + params: ThreadArchiveParams { + thread_id: parent_id, + }, + }) + .await?; + let ThreadLoadedListResponse { data, .. } = mcp + .request(|request_id| ClientRequest::ThreadLoadedList { + request_id, + params: ThreadLoadedListParams::default(), + }) + .await?; + assert_eq!(data, Vec::::new()); + + Ok(()) +} + #[tokio::test] async fn thread_archive_requires_materialized_rollout() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await;