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

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() {

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;

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::<Value>()?;
+        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::<Vec<_>>(),
+        );
+    }
+    assert_eq!(
+        multi_agent_namespaces,
+        vec![vec!["collaboration"], vec!["collaboration"]]
+    );
+    Ok(())
+}