From 283f34387b7e16bd524d8f3a431f77aa7395471d Mon Sep 17 00:00:00 2001 From: vkg-oai Date: Wed, 9 Sep 2026 03:53:45 +0000 Subject: [PATCH] Use `StartThreadOptions` across thread fork APIs (#44043) ## What changed Pass `StartThreadOptions` through rollout, loaded-history, and prepared forks instead of separate configuration and startup arguments. Replace `options.initial_history` with the fork snapshot while preserving the other supplied startup options, and update app-server and test callers. Box the app-server's `thread_fork_inner` future to keep the large fork future out of the shared request dispatcher's stack frame. GitOrigin-RevId: d54eb8b212dedabaf2c5d99d155637cefd7deb7a --- .../request_processors/thread_processor.rs | 27 ++++--- codex-rs/core/src/session/tests.rs | 4 +- codex-rs/core/src/thread_manager.rs | 71 +++++-------------- codex-rs/core/src/thread_manager_tests.rs | 32 ++------- codex-rs/core/tests/common/test_codex.rs | 2 +- codex-rs/core/tests/suite/agents_md.rs | 4 +- .../core/tests/suite/compact_resume_fork.rs | 4 +- codex-rs/core/tests/suite/fork_thread.rs | 14 +--- codex-rs/core/tests/suite/guardian_history.rs | 12 +--- .../tests/suite/guardian_retained_context.rs | 12 +--- codex-rs/core/tests/suite/model_switching.rs | 4 +- .../core/tests/suite/permissions_messages.rs | 4 +- .../core/tests/suite/rollout_compression.rs | 6 +- codex-rs/core/tests/suite/window_headers.rs | 4 +- 14 files changed, 51 insertions(+), 149 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 c4cce3011d..cea25713f1 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -574,13 +574,14 @@ impl ThreadRequestProcessor { app_server_client_version: Option, client_mcp_extensions: ClientMcpExtensions, ) -> Result, JSONRPCErrorError> { - self.thread_fork_inner( + // Keep the large fork future out of the shared request dispatcher's stack frame. + Box::pin(self.thread_fork_inner( request_id, params, app_server_client_name, app_server_client_version, client_mcp_extensions, - ) + )) .await .map(|()| None) } @@ -5162,31 +5163,27 @@ impl ThreadRequestProcessor { .await? }; + let fork_options = StartThreadOptions { + thread_source, + parent_trace, + client_mcp_extensions, + reserved_thread_id, + ..StartThreadOptions::new(config) + }; let new_thread = if let Some(prepared_fork) = prepared_fork { self.thread_manager - .fork_prepared_thread( - config, - prepared_fork, - thread_source, - parent_trace, - client_mcp_extensions, - reserved_thread_id, - ) + .fork_prepared_thread(fork_options, prepared_fork) .await } else { self.thread_manager .fork_thread_from_history( ForkSnapshot::Interrupted, - config, + fork_options, InitialHistory::Resumed(ResumedHistory { conversation_id: source_thread_id, history: history_items, rollout_path: source_thread.rollout_path.clone(), }), - thread_source, - parent_trace, - client_mcp_extensions, - reserved_thread_id, ) .await }; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index b5969c64ce..3b2d0bfb1f 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -3708,10 +3708,8 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result< .thread_manager .fork_thread( usize::MAX, - fork_config.clone(), + core_test_support::test_codex::StartThreadOptions::new(fork_config.clone()), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await?; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index d6e49f8667..8025e00b5b 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1308,31 +1308,21 @@ impl ThreadManager { /// Fork an existing thread by snapshotting rollout history according to /// `snapshot` and starting a new thread with identical configuration - /// (unless overridden by the caller's `config`). The new thread will have - /// a fresh id. + /// (unless overridden by the caller's options). The new thread has a fresh id. + /// Fork history replaces `options.initial_history`. pub async fn fork_thread( &self, snapshot: S, - config: Config, + options: StartThreadOptions, path: PathBuf, - thread_source: Option, - parent_trace: Option, ) -> CodexResult where S: Into, { let snapshot = snapshot.into(); let history = self.initial_history_from_rollout_path(path).await?; - self.fork_thread_from_history( - snapshot, - config, - history, - thread_source, - parent_trace, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, - ) - .await + self.fork_thread_from_history(snapshot, options, history) + .await } async fn initial_history_from_rollout_path( @@ -1354,31 +1344,22 @@ impl ThreadManager { } /// Fork an existing thread from already-loaded store history. - #[allow(clippy::too_many_arguments)] pub async fn fork_thread_from_history( &self, snapshot: S, - config: Config, + options: StartThreadOptions, history: InitialHistory, - thread_source: Option, - parent_trace: Option, - client_mcp_extensions: ClientMcpExtensions, - reserved_thread_id: Option, ) -> CodexResult where S: Into, { self.fork_thread_with_initial_history( - config, + options, ForkHistory { snapshot: snapshot.into(), initial_history: history, persistence: ForkPersistence::Copied, }, - thread_source, - parent_trace, - client_mcp_extensions, - reserved_thread_id, ) .await } @@ -1386,12 +1367,8 @@ impl ThreadManager { /// Fork prepared reference-backed history using the same snapshot semantics as copied forks. pub async fn fork_prepared_thread( &self, - config: Config, + options: StartThreadOptions, prepared: PreparedFork, - thread_source: Option, - parent_trace: Option, - client_mcp_extensions: ClientMcpExtensions, - reserved_thread_id: Option, ) -> CodexResult { let history = InitialHistory::Resumed(ResumedHistory { conversation_id: prepared.source_thread_id, @@ -1404,16 +1381,12 @@ impl ThreadManager { }; let result = self .fork_thread_with_initial_history( - config, + options, ForkHistory { snapshot: ForkSnapshot::Interrupted, initial_history: history, persistence: fork_persistence, }, - thread_source, - parent_trace, - client_mcp_extensions, - reserved_thread_id, ) .await; drop(prepared); @@ -1422,12 +1395,8 @@ impl ThreadManager { async fn fork_thread_with_initial_history( &self, - config: Config, + mut options: StartThreadOptions, fork_history: ForkHistory, - thread_source: Option, - parent_trace: Option, - client_mcp_extensions: ClientMcpExtensions, - reserved_thread_id: Option, ) -> CodexResult { let ForkHistory { snapshot, @@ -1448,21 +1417,15 @@ impl ThreadManager { /*session_source*/ None, /*parent_thread_id*/ None, source_thread_id, - &config, + &options.config, ) .await; - let interrupted_marker = - InterruptedTurnHistoryMarker::from_config_and_version(&config, multi_agent_version); - let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); - let agent_control = self.agent_control_for_config(&config); - let options = StartThreadOptions { - initial_history: history, - thread_source, - parent_trace, - client_mcp_extensions, - reserved_thread_id, - ..StartThreadOptions::new(config) - }; + let interrupted_marker = InterruptedTurnHistoryMarker::from_config_and_version( + &options.config, + multi_agent_version, + ); + options.initial_history = fork_history_from_snapshot(snapshot, history, interrupted_marker); + let agent_control = self.agent_control_for_config(&options.config); let mut request = ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control); request.forked_from_thread_id = source_thread_id; diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index e78cee8d38..a9d7077e98 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -548,13 +548,7 @@ fn fork_thread_accepts_legacy_usize_snapshot_argument() { config: Config, path: std::path::PathBuf, ) { - let _future = manager.fork_thread( - usize::MAX, - config, - path, - /*thread_source*/ None, - /*parent_trace*/ None, - ); + let _future = manager.fork_thread(usize::MAX, crate::StartThreadOptions::new(config), path); } let _: fn(&ThreadManager, Config, std::path::PathBuf) = assert_legacy_snapshot_callsite; @@ -1668,10 +1662,8 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { let forked = manager .fork_thread( ForkSnapshot::Interrupted, - config, + crate::StartThreadOptions::new(config), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork source thread"); @@ -2098,10 +2090,8 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() { let forked = manager .fork_thread( ForkSnapshot::Interrupted, - config, + crate::StartThreadOptions::new(config), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork from rollout path"); @@ -2622,10 +2612,8 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor let forked = manager .fork_thread( ForkSnapshot::Interrupted, - config.clone(), + crate::StartThreadOptions::new(config.clone()), source_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); @@ -2749,10 +2737,8 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() { let forked = manager .fork_thread( ForkSnapshot::Interrupted, - config.clone(), + crate::StartThreadOptions::new(config.clone()), source_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); @@ -2838,10 +2824,8 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ let forked = manager .fork_thread( ForkSnapshot::Interrupted, - config.clone(), + crate::StartThreadOptions::new(config.clone()), source_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork interrupted snapshot"); @@ -2879,10 +2863,8 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ let reforked = manager .fork_thread( ForkSnapshot::Interrupted, - config.clone(), + crate::StartThreadOptions::new(config.clone()), forked_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("re-fork interrupted snapshot"); diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index c886a86656..750445cdd2 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -15,7 +15,7 @@ use anyhow::Result; use anyhow::anyhow; use codex_config::CloudConfigBundleLoader; use codex_core::CodexThread; -use codex_core::StartThreadOptions; +pub use codex_core::StartThreadOptions; use codex_core::ThreadManager; use codex_core::TimeProvider; pub use codex_core::TurnInputRequest; diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index 9790773706..e0ca380b37 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/codex-rs/core/tests/suite/agents_md.rs @@ -1392,10 +1392,8 @@ async fn fork_injects_changed_agents_md_once() -> Result<()> { .thread_manager .fork_thread( ForkSnapshot::Interrupted, - fork_config, + codex_core::StartThreadOptions::new(fork_config), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await?; diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index 522ef8c047..d937ad13c4 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -929,10 +929,8 @@ async fn fork_thread( ) -> Arc { Box::pin(manager.fork_thread( nth_user_message, - config.clone(), + codex_core::StartThreadOptions::new(config.clone()), path, - /*thread_source*/ None, - /*parent_trace*/ None, )) .await .expect("fork conversation") diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index ad9434a50a..51471b44e2 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -101,10 +101,8 @@ async fn fork_thread_twice_drops_to_first_message() { } = thread_manager .fork_thread( ForkSnapshot::TruncateBeforeNthUserMessage(1), - config_for_fork.clone(), + codex_core::StartThreadOptions::new(config_for_fork.clone()), base_path.clone(), - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork 1"); @@ -130,10 +128,8 @@ async fn fork_thread_twice_drops_to_first_message() { } = thread_manager .fork_thread( ForkSnapshot::TruncateBeforeNthUserMessage(0), - config_for_fork.clone(), + codex_core::StartThreadOptions::new(config_for_fork.clone()), fork1_path.clone(), - /*thread_source*/ None, - /*parent_trace*/ None, ) .await .expect("fork 2"); @@ -227,16 +223,12 @@ async fn assert_copied_fork_persists_inherited_history(history_mode: ThreadHisto } = thread_manager .fork_thread_from_history( ForkSnapshot::Interrupted, - test.config.clone(), + codex_core::StartThreadOptions::new(test.config.clone()), InitialHistory::Resumed(ResumedHistory { conversation_id: test.session_configured.thread_id, history: Arc::new(supplied_history), rollout_path: None, }), - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await .expect("fork from stored history"); diff --git a/codex-rs/core/tests/suite/guardian_history.rs b/codex-rs/core/tests/suite/guardian_history.rs index 6f50f1cd87..1438fd937e 100644 --- a/codex-rs/core/tests/suite/guardian_history.rs +++ b/codex-rs/core/tests/suite/guardian_history.rs @@ -123,12 +123,8 @@ async fn guardian_history_survives_restart_and_user_fork( initial .thread_manager .fork_prepared_thread( - initial.config.clone(), + codex_core::StartThreadOptions::new(initial.config.clone()), prepared, - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await? } else { @@ -136,12 +132,8 @@ async fn guardian_history_survives_restart_and_user_fork( .thread_manager .fork_thread_from_history( ForkSnapshot::Interrupted, - initial.config.clone(), + codex_core::StartThreadOptions::new(initial.config.clone()), history.clone(), - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await? }; diff --git a/codex-rs/core/tests/suite/guardian_retained_context.rs b/codex-rs/core/tests/suite/guardian_retained_context.rs index 2e2e2cb623..30d63fe815 100644 --- a/codex-rs/core/tests/suite/guardian_retained_context.rs +++ b/codex-rs/core/tests/suite/guardian_retained_context.rs @@ -1061,28 +1061,20 @@ async fn standalone_fork_retains_inherited_user_instructions( .await?; test.thread_manager .fork_prepared_thread( - test.config.clone(), + codex_core::StartThreadOptions::new(test.config.clone()), prepared, - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await? } else { test.thread_manager .fork_thread_from_history( ForkSnapshot::Interrupted, - test.config.clone(), + codex_core::StartThreadOptions::new(test.config.clone()), InitialHistory::Resumed(ResumedHistory { conversation_id: worker.session_configured().thread_id, history: Arc::new(load_context(&test, &worker).await?), rollout_path: None, }), - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await? }; diff --git a/codex-rs/core/tests/suite/model_switching.rs b/codex-rs/core/tests/suite/model_switching.rs index 7443b63188..cd9b33f9b2 100644 --- a/codex-rs/core/tests/suite/model_switching.rs +++ b/codex-rs/core/tests/suite/model_switching.rs @@ -265,10 +265,8 @@ async fn first_turn_after_empty_prefix_fork_preserves_inherited_base_instruction .thread_manager .fork_thread( ForkSnapshot::TruncateBeforeNthUserMessage(0), - fork_config, + codex_core::StartThreadOptions::new(fork_config), source_rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await?; submit_model_turn(&fork.thread, turn_model, ThreadSettingsOverrides::default()).await?; diff --git a/codex-rs/core/tests/suite/permissions_messages.rs b/codex-rs/core/tests/suite/permissions_messages.rs index f58dba2c73..116a0fdc04 100644 --- a/codex-rs/core/tests/suite/permissions_messages.rs +++ b/codex-rs/core/tests/suite/permissions_messages.rs @@ -742,10 +742,8 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> { .thread_manager .fork_thread( ForkSnapshot::Interrupted, - fork_config.clone(), + codex_core::StartThreadOptions::new(fork_config.clone()), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await?; forked diff --git a/codex-rs/core/tests/suite/rollout_compression.rs b/codex-rs/core/tests/suite/rollout_compression.rs index e361cd375c..1ab5986ae2 100644 --- a/codex-rs/core/tests/suite/rollout_compression.rs +++ b/codex-rs/core/tests/suite/rollout_compression.rs @@ -88,12 +88,8 @@ async fn compressed_shared_fork_resume_preserves_checkpoint_and_frozen_history() let child = test .thread_manager .fork_prepared_thread( - test.config.clone(), + codex_core::StartThreadOptions::new(test.config.clone()), prepared, - /*thread_source*/ None, - /*parent_trace*/ None, - ClientMcpExtensions::default(), - /*reserved_thread_id*/ None, ) .await?; turn( diff --git a/codex-rs/core/tests/suite/window_headers.rs b/codex-rs/core/tests/suite/window_headers.rs index 22bbd2d8e0..3cd3aabcf5 100644 --- a/codex-rs/core/tests/suite/window_headers.rs +++ b/codex-rs/core/tests/suite/window_headers.rs @@ -69,10 +69,8 @@ async fn window_id_advances_after_compact_persists_on_resume_and_resets_on_fork( .thread_manager .fork_thread( /*snapshot*/ 0usize, - resumed.config.clone(), + codex_core::StartThreadOptions::new(resumed.config.clone()), rollout_path, - /*thread_source*/ None, - /*parent_trace*/ None, ) .await?; submit_user_turn(&forked.thread, "after fork").await?;