From cc737efd65b9317f425d093669486dfc5658a20b Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 7 Sep 2026 17:52:18 +0000 Subject: [PATCH] Preserve the multi-agent version when forking at a turn cutoff (#43540) ## Why A fork cutoff can remove the only `TurnContext` recording the source thread's selected multi-agent version. Forking before the first turn must preserve that version, including when the child is resumed before its first turn. ## What changed Recover the version from the untrimmed source history or loaded parent, independently of permission overrides, and carry it into the fork's session metadata for `last_turn_id` and `before_turn_id` cutoffs. ## Testing Add regression coverage for forks before the first turn across legacy and paginated history modes, with loaded and unloaded parents and explicit permission overrides. Verify that the child persists `MultiAgentVersion::V2` and still uses the `collaboration` tools after restarting and resuming. GitOrigin-RevId: d7e30cf8bfdeb4f755fefab3d1d0884964d7c977 --- .../request_processors/thread_processor.rs | 42 ++++- codex-rs/app-server/tests/suite/v2/mod.rs | 2 + .../suite/v2/thread_fork_multi_agent_tests.rs | 151 ++++++++++++++++++ 3 files changed, 188 insertions(+), 7 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/thread_fork_multi_agent_tests.rs 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 9e28aef7b2..78ea161c26 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -4825,7 +4825,7 @@ impl ThreadRequestProcessor { .name .as_deref() .and_then(codex_core::util::normalize_thread_name); - let prepared_fork = if paginated_source { + let mut prepared_fork = if paginated_source { let boundary = match (last_turn_id.as_deref(), before_turn_id.as_deref()) { (Some(turn_id), None) => { codex_thread_store::ForkBoundary::ThroughTurn(turn_id.to_string()) @@ -4928,8 +4928,9 @@ impl ThreadRequestProcessor { !has_permission_override(request_overrides.as_ref(), &typesafe_overrides); let needs_latest_settings = restore_approval_policy || restore_approvals_reviewer || restore_permission_profile; + let loaded_parent = self.thread_manager.get_thread(source_thread_id).await.ok(); let loaded_parent_settings = if paginated_source && needs_latest_settings { - if let Ok(parent) = self.thread_manager.get_thread(source_thread_id).await { + if let Some(parent) = loaded_parent.as_ref() { let snapshot = parent.thread_settings_snapshot().await; Some(PersistedResumeSettings { approval_policy: snapshot.approval_policy, @@ -4943,11 +4944,10 @@ impl ThreadRequestProcessor { None }; let latest_context = if paginated_source - && needs_latest_settings - && loaded_parent_settings.is_none() + && loaded_parent.is_none() && (last_turn_id.is_some() || before_turn_id.is_some()) { - Some( + Some(Arc::new( self.thread_store .load_latest_model_context(StoreLoadThreadHistoryParams { thread_id: source_thread_id, @@ -4956,10 +4956,23 @@ impl ThreadRequestProcessor { .await .map_err(thread_store_resume_read_error)? .items, - ) + )) } else { None }; + // The fork cutoff can remove the only TurnContext that records the selected version. + // Recover it from the untrimmed source or live parent, independently of permission overrides. + let source_multi_agent_version = InitialHistory::Resumed(ResumedHistory { + conversation_id: source_thread_id, + history: Arc::clone(latest_context.as_ref().unwrap_or(&source_history_items)), + rollout_path: source_thread.rollout_path.clone(), + }) + .get_multi_agent_version() + .or_else(|| { + loaded_parent + .as_ref() + .and_then(|parent| parent.multi_agent_version()) + }); let persisted_settings = loaded_parent_settings.or_else(|| { latest_persisted_resume_settings( latest_context @@ -4992,7 +5005,7 @@ impl ThreadRequestProcessor { let parent_trace = self.request_trace_context(&request_id).await; let thread_source = thread_source.map(Into::into); - let history_items = if prepared_fork.is_some() { + let mut history_items = if prepared_fork.is_some() { source_history_items } else { let source_history_items = Arc::unwrap_or_clone(source_history_items); @@ -5010,6 +5023,21 @@ impl ThreadRequestProcessor { }; Arc::new(history_items) }; + if let Some(multi_agent_version) = source_multi_agent_version + && (last_turn_id.is_some() || before_turn_id.is_some()) + { + // Update only the fork's in-memory metadata; the source rollout stays unchanged. + for item in Arc::make_mut(&mut history_items) { + if let RolloutItem::SessionMeta(meta) = item + && meta.meta.id == source_thread_id + { + meta.meta.multi_agent_version = Some(multi_agent_version); + } + } + if let Some(prepared_fork) = prepared_fork.as_mut() { + prepared_fork.model_context = Arc::clone(&history_items); + } + } let ephemeral_preview = if ephemeral { if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() { diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index ee91cd9569..673dadf834 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -102,6 +102,8 @@ mod thread_archive; mod thread_delete; mod thread_environments; mod thread_fork; +#[path = "thread_fork_multi_agent_tests.rs"] +mod thread_fork_multi_agent; mod thread_inject_items; mod thread_list; mod thread_loaded_list; diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork_multi_agent_tests.rs b/codex-rs/app-server/tests/suite/v2/thread_fork_multi_agent_tests.rs new file mode 100644 index 0000000000..d8bde37558 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_fork_multi_agent_tests.rs @@ -0,0 +1,151 @@ +//! Regression coverage for preserving the source runtime across empty fork cutoffs. + +use anyhow::Result; +use app_test_support::MockResponsesConfig; +use app_test_support::TestAppServer; +use app_test_support::create_mock_responses_server_repeating_assistant; +use app_test_support::write_models_cache_with_models; +use codex_app_server_protocol::ApprovalsReviewer; +use codex_app_server_protocol::AskForApproval; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::SandboxMode; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::UserInput; +use codex_features::Feature; +use codex_protocol::protocol::MultiAgentVersion; +use codex_rollout::read_session_meta_line; +use core_test_support::load_default_config_for_test; +use pretty_assertions::assert_eq; +use serde_json::Value; +use tempfile::TempDir; + +enum SourceState { + Loaded, + Unloaded, +} + +#[test_case::test_case(ThreadHistoryMode::Legacy, SourceState::Loaded; "legacy_loaded")] +#[test_case::test_case(ThreadHistoryMode::Legacy, SourceState::Unloaded; "legacy_unloaded")] +#[test_case::test_case(ThreadHistoryMode::Paginated, SourceState::Loaded; "paginated_loaded")] +#[test_case::test_case(ThreadHistoryMode::Paginated, SourceState::Unloaded; "paginated_unloaded")] +#[tokio::test] +async fn fork_before_first_turn_preserves_model_selected_multi_agent_version( + history_mode: ThreadHistoryMode, + source_state: SourceState, +) -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .disable_feature(Feature::MultiAgentV2) + .write(codex_home.path())?; + let config = load_default_config_for_test(&codex_home).await; + let mut model = codex_core::test_support::construct_model_info_offline("mock-model", &config); + model.multi_agent_version = Some(MultiAgentVersion::V2); + write_models_cache_with_models(codex_home.path(), vec![model])?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let parent = mcp + .start_thread(ThreadStartParams { + history_mode: Some(history_mode), + ..Default::default() + }) + .await? + .thread; + let completed = mcp + .start_turn_and_wait_for_completion(TurnStartParams { + thread_id: parent.id.clone(), + input: vec![UserInput::Text { + text: "First message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + if matches!(source_state, SourceState::Unloaded) { + mcp.shutdown_gracefully().await?; + mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + } + let ThreadForkResponse { thread: child, .. } = mcp + .request(|request_id| ClientRequest::ThreadFork { + request_id, + params: ThreadForkParams { + thread_id: parent.id, + before_turn_id: Some(completed.turn.id), + exclude_turns: true, + // Version recovery must also work when no permission settings need restoring. + approval_policy: Some(AskForApproval::Never), + approvals_reviewer: Some(ApprovalsReviewer::User), + sandbox: Some(SandboxMode::ReadOnly), + ..Default::default() + }, + }) + .await?; + assert_eq!( + read_session_meta_line(child.path.as_ref().expect("fork rollout").as_path()) + .await? + .meta + .multi_agent_version, + Some(MultiAgentVersion::V2) + ); + + // Resume before the child's first turn can persist another version-bearing context. + mcp.shutdown_gracefully().await?; + mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let _: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: child.id.clone(), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + mcp.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: child.id, + input: vec![UserInput::Text { + text: "Forked message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + + let requests = server.received_requests().await.expect("response requests"); + let mut multi_agent_namespaces = Vec::new(); + for request in requests + .iter() + .filter(|request| request.url.path().ends_with("/responses")) + { + let body = request.body_json::()?; + multi_agent_namespaces.push( + body["tools"] + .as_array() + .expect("tools") + .iter() + .filter_map(|tool| tool["name"].as_str()) + .filter(|name| matches!(*name, "collaboration" | "multi_agent_v1")) + .map(str::to_owned) + .collect::>(), + ); + } + assert_eq!( + multi_agent_namespaces, + vec![vec!["collaboration"], vec!["collaboration"]] + ); + Ok(()) +}