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
This commit is contained in:
vkg-oai
2026-09-09 04:07:40 +00:00
committed by copyberry
parent ec512d2347
commit 283f34387b
14 changed files with 51 additions and 149 deletions
@@ -574,13 +574,14 @@ impl ThreadRequestProcessor {
app_server_client_version: Option<String>,
client_mcp_extensions: ClientMcpExtensions,
) -> Result<Option<ClientResponsePayload>, 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
};
+1 -3
View File
@@ -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?;
+17 -54
View File
@@ -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<S>(
&self,
snapshot: S,
config: Config,
options: StartThreadOptions,
path: PathBuf,
thread_source: Option<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
where
S: Into<ForkSnapshot>,
{
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<S>(
&self,
snapshot: S,
config: Config,
options: StartThreadOptions,
history: InitialHistory,
thread_source: Option<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
client_mcp_extensions: ClientMcpExtensions,
reserved_thread_id: Option<ThreadId>,
) -> CodexResult<NewThread>
where
S: Into<ForkSnapshot>,
{
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<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
client_mcp_extensions: ClientMcpExtensions,
reserved_thread_id: Option<ThreadId>,
) -> CodexResult<NewThread> {
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<ThreadSource>,
parent_trace: Option<W3cTraceContext>,
client_mcp_extensions: ClientMcpExtensions,
reserved_thread_id: Option<ThreadId>,
) -> CodexResult<NewThread> {
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;
+7 -25
View File
@@ -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");
+1 -1
View File
@@ -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;
+1 -3
View File
@@ -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?;
@@ -929,10 +929,8 @@ async fn fork_thread(
) -> Arc<CodexThread> {
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")
+3 -11
View File
@@ -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");
+2 -10
View File
@@ -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?
};
@@ -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?
};
+1 -3
View File
@@ -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?;
@@ -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
@@ -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(
+1 -3
View File
@@ -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?;