openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-08-23 13:09:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Consolidate thread spawning behind a request object (#36862)

Browse Source 
## What changed

- Add `ThreadSpawnRequest` to carry thread options, authentication, agent
  control, fork metadata, inherited state, and shell overrides.
- Route new, resumed, and forked threads through one `spawn_thread` path.
- Centralize default session-source and environment selection when the request
  is consumed.

GitOrigin-RevId: 8fd1a8531a212000c7430218c2200aa4a047cb9a
This commit is contained in:
jif
2026-08-04 08:46:21 +00:00
committed by copyberry
parent d4fb78bfc5
commit 9873cba8ce
1 changed files with 144 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files

423
codex-rs/core/src/thread_manager.rs
View File

@@ -232,6 +232,38 @@ impl StartThreadOptions {
}
}
struct ThreadSpawnRequest {
options: StartThreadOptions,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
parent_thread_id: Option<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
fork_persistence: ForkPersistence,
inherited_environments: Option<TurnEnvironmentSnapshot>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
user_shell_override: Option<crate::shell::Shell>,
}
impl ThreadSpawnRequest {
fn new(
options: StartThreadOptions,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
) -> Self {
Self {
options,
auth_manager,
agent_control,
parent_thread_id: None,
forked_from_thread_id: None,
fork_persistence: ForkPersistence::Copied,
inherited_environments: None,
inherited_exec_policy: None,
user_shell_override: None,
}
}
}
fn originator_from_service_name(service_name: Option<&str>) -> Option<String> {
let service_name = service_name?.trim();
for originator in [
@@ -777,46 +809,25 @@ impl ThreadManager {
async fn start_thread_inner(
&self,
options: StartThreadOptions,
mut options: StartThreadOptions,
forked_from_thread_id: Option<ThreadId>,
) -> CodexResult<NewThread> {
let environments = options.environments.unwrap_or_else(|| {
default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&options.config.cwd,
&options.config.workspace_roots,
)
});
let agent_control = self.agent_control_for_config(&options.config);
let (resumed_session_source, resumed_thread_source) = options
.initial_history
.get_resumed_session_sources()
.unwrap_or_else(|| (self.state.session_source.clone(), None));
let session_source = options.session_source.unwrap_or(resumed_session_source);
let thread_source = options.thread_source.or(resumed_thread_source);
Box::pin(self.state.spawn_thread_with_source(
options.config,
options.initial_history,
options.history_mode,
options.allow_provider_model_fallback,
Arc::clone(&self.state.auth_manager),
agent_control,
session_source,
/*parent_thread_id*/ None,
forked_from_thread_id,
ForkPersistence::Copied,
thread_source,
options.dynamic_tools,
options.metrics_service_name,
/*inherited_environments*/ None,
/*inherited_exec_policy*/ None,
options.parent_trace,
environments,
options.thread_extension_init,
options.supports_openai_form_elicitation,
/*user_shell_override*/ None,
))
.await
options.session_source = Some(
options
.session_source
.take()
.unwrap_or(resumed_session_source),
);
options.thread_source = options.thread_source.take().or(resumed_thread_source);
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control);
request.forked_from_thread_id = forked_from_thread_id;
Box::pin(self.state.spawn_thread(request)).await
}
// TODO(jif) merge with fork_agent
@@ -885,11 +896,6 @@ impl ThreadManager {
supports_openai_form_elicitation: bool,
) -> CodexResult<NewThread> {
let agent_control = self.agent_control_for_config(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
);
let (session_source, thread_source) = initial_history
.get_resumed_session_sources()
.unwrap_or_else(|| (self.state.session_source.clone(), None));
@@ -901,28 +907,19 @@ impl ThreadManager {
.restore_v2_agent_metadata(&config, resumed.conversation_id)
.await;
}
Box::pin(self.state.spawn_thread_with_source(
config,
let options = StartThreadOptions {
initial_history,
/*history_mode*/ None,
/*allow_provider_model_fallback*/ false,
session_source: Some(session_source),
thread_source,
parent_trace,
supports_openai_form_elicitation,
..StartThreadOptions::new(config)
};
Box::pin(self.state.spawn_thread(ThreadSpawnRequest::new(
options,
auth_manager,
agent_control,
session_source,
/*parent_thread_id*/ None,
/*forked_from_thread_id*/ None,
ForkPersistence::Copied,
thread_source,
Vec::new(),
/*metrics_service_name*/ None,
/*inherited_environments*/ None,
/*inherited_exec_policy*/ None,
parent_trace,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
supports_openai_form_elicitation,
/*user_shell_override*/ None,
))
)))
.await
}
@@ -933,29 +930,14 @@ impl ThreadManager {
supports_openai_form_elicitation: bool,
) -> CodexResult<NewThread> {
let agent_control = self.agent_control_for_config(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
);
Box::pin(self.state.spawn_thread(
config,
InitialHistory::New,
Arc::clone(&self.state.auth_manager),
agent_control,
/*parent_thread_id*/ None,
/*forked_from_thread_id*/ None,
ForkPersistence::Copied,
/*thread_source*/ None,
Vec::new(),
/*metrics_service_name*/ None,
/*parent_trace*/ None,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
let options = StartThreadOptions {
supports_openai_form_elicitation,
/*user_shell_override*/ Some(user_shell_override),
))
.await
..StartThreadOptions::new(config)
};
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control);
request.user_shell_override = Some(user_shell_override);
Box::pin(self.state.spawn_thread(request)).await
}
pub(crate) async fn resume_thread_from_rollout_with_user_shell_override_for_tests(
@@ -968,37 +950,19 @@ impl ThreadManager {
) -> CodexResult<NewThread> {
let agent_control = self.agent_control_for_config(&config);
let initial_history = self.initial_history_from_rollout_path(rollout_path).await?;
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
);
let (session_source, thread_source) = initial_history
.get_resumed_session_sources()
.unwrap_or_else(|| (self.state.session_source.clone(), None));
Box::pin(self.state.spawn_thread_with_source(
config,
let options = StartThreadOptions {
initial_history,
/*history_mode*/ None,
/*allow_provider_model_fallback*/ false,
auth_manager,
agent_control,
session_source,
/*parent_thread_id*/ None,
/*forked_from_thread_id*/ None,
ForkPersistence::Copied,
session_source: Some(session_source),
thread_source,
Vec::new(),
/*metrics_service_name*/ None,
/*inherited_environments*/ None,
/*inherited_exec_policy*/ None,
/*parent_trace*/ None,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
supports_openai_form_elicitation,
/*user_shell_override*/ Some(user_shell_override),
))
.await
..StartThreadOptions::new(config)
};
let mut request = ThreadSpawnRequest::new(options, auth_manager, agent_control);
request.user_shell_override = Some(user_shell_override);
Box::pin(self.state.spawn_thread(request)).await
}
/// Removes the thread from the manager's internal map, though the thread is stored
@@ -1200,30 +1164,19 @@ impl ThreadManager {
let interrupted_marker =
InterruptedTurnHistoryMarker::from_config_and_version(&config, multi_agent_version);
let history = fork_history_from_snapshot(snapshot, history, interrupted_marker);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
);
let agent_control = self.agent_control_for_config(&config);
Box::pin(self.state.spawn_thread(
config,
history,
Arc::clone(&self.state.auth_manager),
agent_control,
/*parent_thread_id*/ None,
source_thread_id,
fork_persistence,
let options = StartThreadOptions {
initial_history: history,
thread_source,
Vec::new(),
/*metrics_service_name*/ None,
parent_trace,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
supports_openai_form_elicitation,
/*user_shell_override*/ None,
))
.await
..StartThreadOptions::new(config)
};
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control);
request.forked_from_thread_id = source_thread_id;
request.fork_persistence = fork_persistence;
Box::pin(self.state.spawn_thread(request)).await
}
pub(crate) fn agent_control(&self) -> AgentControl {
@@ -1551,36 +1504,21 @@ impl ThreadManagerState {
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
environments: Option<Vec<TurnEnvironmentSelection>>,
) -> CodexResult<NewThread> {
let environments = environments.unwrap_or_else(|| {
default_thread_environment_selections(
self.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
)
});
Box::pin(self.spawn_thread_with_source(
config,
InitialHistory::New,
let options = StartThreadOptions {
history_mode,
/*allow_provider_model_fallback*/ false,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
parent_thread_id,
forked_from_thread_id,
ForkPersistence::Copied,
session_source: Some(session_source),
thread_source,
Vec::new(),
metrics_service_name,
inherited_environments,
inherited_exec_policy,
/*parent_trace*/ None,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
/*supports_openai_form_elicitation*/ false,
/*user_shell_override*/ None,
))
.await
..StartThreadOptions::new(config)
};
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.auth_manager), agent_control);
request.parent_thread_id = parent_thread_id;
request.forked_from_thread_id = forked_from_thread_id;
request.inherited_environments = inherited_environments;
request.inherited_exec_policy = inherited_exec_policy;
Box::pin(self.spawn_thread(request)).await
}
pub(crate) async fn resume_thread_with_history_with_source(
@@ -1596,35 +1534,19 @@ impl ThreadManagerState {
inherited_environments,
inherited_exec_policy,
} = options;
let environments = default_thread_environment_selections(
self.environment_manager.as_ref(),
&config.cwd,
&config.workspace_roots,
);
let thread_source = initial_history.get_resumed_thread_source();
Box::pin(self.spawn_thread_with_source(
config,
let options = StartThreadOptions {
initial_history,
/*history_mode*/ None,
/*allow_provider_model_fallback*/ false,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
parent_thread_id,
/*forked_from_thread_id*/ None,
ForkPersistence::Copied,
session_source: Some(session_source),
thread_source,
Vec::new(),
/*metrics_service_name*/ None,
inherited_environments,
inherited_exec_policy,
/*parent_trace*/ None,
environments,
/*thread_extension_init*/ ExtensionDataInit::default(),
/*supports_openai_form_elicitation*/ false,
/*user_shell_override*/ None,
))
.await
..StartThreadOptions::new(config)
};
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.auth_manager), agent_control);
request.parent_thread_id = parent_thread_id;
request.inherited_environments = inherited_environments;
request.inherited_exec_policy = inherited_exec_policy;
Box::pin(self.spawn_thread(request)).await
}
#[allow(clippy::too_many_arguments)]
@@ -1643,6 +1565,52 @@ impl ThreadManagerState {
environments: Option<Vec<TurnEnvironmentSelection>>,
thread_extension_init: ExtensionDataInit,
) -> CodexResult<NewThread> {
let options = StartThreadOptions {
initial_history,
history_mode,
session_source: Some(session_source),
thread_source,
environments,
thread_extension_init,
..StartThreadOptions::new(config)
};
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.auth_manager), agent_control);
request.parent_thread_id = parent_thread_id;
request.forked_from_thread_id = forked_from_thread_id;
request.inherited_environments = inherited_environments;
request.inherited_exec_policy = inherited_exec_policy;
Box::pin(self.spawn_thread(request)).await
}
/// Spawn a new thread with optional history and register it with the manager.
async fn spawn_thread(&self, request: ThreadSpawnRequest) -> CodexResult<NewThread> {
let ThreadSpawnRequest {
options,
auth_manager,
agent_control,
parent_thread_id,
forked_from_thread_id,
fork_persistence,
inherited_environments,
inherited_exec_policy,
user_shell_override,
} = request;
let StartThreadOptions {
config,
allow_provider_model_fallback,
initial_history,
history_mode,
session_source,
thread_source,
dynamic_tools,
metrics_service_name,
parent_trace,
environments,
thread_extension_init,
supports_openai_form_elicitation,
} = options;
let session_source = session_source.unwrap_or_else(|| self.session_source.clone());
let environments = environments.unwrap_or_else(|| {
default_thread_environment_selections(
self.environment_manager.as_ref(),
@@ -1650,109 +1618,6 @@ impl ThreadManagerState {
&config.workspace_roots,
)
});
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
history_mode,
/*allow_provider_model_fallback*/ false,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
parent_thread_id,
forked_from_thread_id,
ForkPersistence::Copied,
thread_source,
Vec::new(),
/*metrics_service_name*/ None,
inherited_environments,
inherited_exec_policy,
/*parent_trace*/ None,
environments,
thread_extension_init,
/*supports_openai_form_elicitation*/ false,
/*user_shell_override*/ None,
))
.await
}
/// Spawn a new thread with optional history and register it with the manager.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn spawn_thread(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
parent_thread_id: Option<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
fork_persistence: ForkPersistence,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
metrics_service_name: Option<String>,
parent_trace: Option<W3cTraceContext>,
environments: Vec<TurnEnvironmentSelection>,
thread_extension_init: ExtensionDataInit,
supports_openai_form_elicitation: bool,
user_shell_override: Option<crate::shell::Shell>,
) -> CodexResult<NewThread> {
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
/*history_mode*/ None,
/*allow_provider_model_fallback*/ false,
auth_manager,
agent_control,
self.session_source.clone(),
parent_thread_id,
forked_from_thread_id,
fork_persistence,
thread_source,
dynamic_tools,
metrics_service_name,
/*inherited_environments*/ None,
/*inherited_exec_policy*/ None,
parent_trace,
environments,
thread_extension_init,
supports_openai_form_elicitation,
user_shell_override,
))
.await
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn spawn_thread_with_source(
&self,
config: Config,
initial_history: InitialHistory,
history_mode: Option<ThreadHistoryMode>,
allow_provider_model_fallback: bool,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
session_source: SessionSource,
parent_thread_id: Option<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
fork_persistence: ForkPersistence,
thread_source: Option<ThreadSource>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
metrics_service_name: Option<String>,
inherited_environments: Option<TurnEnvironmentSnapshot>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
parent_trace: Option<W3cTraceContext>,
environments: Vec<TurnEnvironmentSelection>,
thread_extension_init: ExtensionDataInit,
supports_openai_form_elicitation: bool,
user_shell_override: Option<crate::shell::Shell>,
) -> CodexResult<NewThread> {
let source_changed_during_startup = Arc::new(AtomicBool::new(false));
{
let mut starting = self
.starting_mcp_runtimes
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
starting.retain(|runtime| runtime.strong_count() != 0);
starting.push(Arc::downgrade(&source_changed_during_startup));
}
let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_));
if let InitialHistory::Resumed(resumed) = &initial_history {
let mut threads = self.threads.write().await;