From 9f70e348e0227980de97e361cce830236fb18317 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 7 Sep 2026 15:16:44 +0000 Subject: [PATCH] Allow internal sessions to fork from selected history (#43495) ## What changed Add `ThreadManager::fork_internal_session` to start an internal session from caller-selected committed history without reading an in-flight parent turn or appending an interruption marker. Preserve parent authentication and budget sharing, and propagate fork lineage. Keep `spawn_internal_session` starting with fresh history. ## Testing Add a regression test verifying that the fork contains only the selected history, retains its parent association and authentication manager, and stays outside the public thread registry. GitOrigin-RevId: df4260ed3adfa0abe7a07c2b6d8b8c6ef6ba3d16 --- codex-rs/core/src/thread_manager.rs | 32 +++++++++- codex-rs/core/src/thread_manager_tests.rs | 71 +++++++++++++++++++++++ 2 files changed, 102 insertions(+), 1 deletion(-) diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index e2cfe7cb61..eb85a048b7 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -958,9 +958,37 @@ impl ThreadManager { /// Starts a fresh internal session associated with an existing parent thread. pub async fn spawn_internal_session( + &self, + parent_thread_id: ThreadId, + options: StartThreadOptions, + ) -> CodexResult { + self.spawn_internal_session_with_history(parent_thread_id, options, InitialHistory::New) + .await + } + + /// Starts an internal session from an explicitly selected, committed history. + /// + /// The caller selects the history; this never reads an in-flight parent turn or + /// appends an interruption marker. Authentication and budget still come from the parent. + pub async fn fork_internal_session( + &self, + parent_thread_id: ThreadId, + options: StartThreadOptions, + history: Vec, + ) -> CodexResult { + self.spawn_internal_session_with_history( + parent_thread_id, + options, + InitialHistory::Forked(history), + ) + .await + } + + async fn spawn_internal_session_with_history( &self, parent_thread_id: ThreadId, mut options: StartThreadOptions, + history: InitialHistory, ) -> CodexResult { if !matches!(options.session_source, Some(SessionSource::Internal(_))) { return Err(CodexErr::InvalidRequest( @@ -968,13 +996,15 @@ impl ThreadManager { )); } let parent = self.get_thread(parent_thread_id).await?; - options.initial_history = InitialHistory::New; + let forked_from_thread_id = history.forked_from_id(); + options.initial_history = history; let mut request = ThreadSpawnRequest::new( options, Arc::clone(&parent.session.services.auth_manager), parent.session.services.agent_control.clone(), ); request.parent_thread_id = Some(parent_thread_id); + request.forked_from_thread_id = forked_from_thread_id; Box::pin(self.state.spawn_thread(request)).await } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 1af560e2be..e78cee8d38 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -924,6 +924,77 @@ async fn spawn_internal_guardian_session_preserves_windows_sandbox_proxy_setting .await; } +#[tokio::test] +async fn fork_internal_session_uses_only_the_selected_history() { + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config().await; + config.codex_home = temp_dir.path().join("codex-home").abs(); + config.cwd = config.codex_home.abs(); + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + let manager = ThreadManager::with_models_provider_and_home_for_tests( + CodexAuth::from_api_key("dummy"), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ); + let parent = manager + .start_thread(StartThreadOptions::new(config.clone())) + .await + .expect("start parent"); + let selected = vec![ + user_msg("committed review"), + assistant_msg("completed assessment"), + ]; + parent + .thread + .inject_response_items(vec![user_msg("unrelated parent work")]) + .await + .expect("inject parent history"); + let reviewer = manager + .fork_internal_session( + parent.thread_id, + StartThreadOptions { + session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)), + ..StartThreadOptions::new(config) + }, + selected + .iter() + .cloned() + .map(|item| RolloutItem::ResponseItem(item.into())) + .collect(), + ) + .await + .expect("fork internal reviewer"); + let history = reviewer.thread.conversation_history_snapshot().await; + let mut actual = history.items().cloned().collect::>(); + // Recording history assigns message IDs and provenance; compare the selected content. + for item in &mut actual { + if let ResponseItem::Message { + id, + internal_chat_message_metadata_passthrough, + .. + } = item + { + *id = None; + *internal_chat_message_metadata_passthrough = None; + } + } + assert_eq!(actual, selected); + assert_eq!( + reviewer.thread.config_snapshot().await.parent_thread_id, + Some(parent.thread_id) + ); + assert!(Arc::ptr_eq( + &reviewer.thread.session.services.auth_manager, + &parent.thread.session.services.auth_manager, + )); + assert_eq!(manager.list_thread_ids().await, vec![parent.thread_id]); + assert!(manager.get_thread(reviewer.thread_id).await.is_err()); + manager + .shutdown_all_threads_bounded(Duration::from_secs(10)) + .await; +} + #[tokio::test] async fn spawn_internal_session_preserves_parent_lineage_without_forking_history() { struct ParentLifecycleContributor {