agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add ThreadManager sample crate (#20141)

Browse Source 
Summary:
- Add codex-thread-manager-sample, a one-shot binary that starts a
ThreadManager thread, submits a prompt, and prints the final assistant
output.
- Pass ThreadStore into ThreadManager::new and expose
thread_store_from_config for existing callsites.
- Build the sample Config directly with only --model and prompt inputs.

Verification:
- just fmt
- cargo check -p codex-thread-manager-sample -p codex-app-server -p
codex-mcp-server
- git diff --check

Tests: Not run per request.
This commit is contained in:
pakrym-oai
2026-04-29 11:21:06 -07:00
committed by GitHub
Unverified
parent 47fba5df4a
commit 8356806fc9
33 changed files with 737 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+21
View File
@@ -3477,6 +3477,27 @@ dependencies = [
"tempfile",
]
[[package]]
name = "codex-thread-manager-sample"
version = "0.0.0"
dependencies = [
"anyhow",
"clap",
"codex-arg0",
"codex-config",
"codex-core",
"codex-exec-server",
"codex-features",
"codex-login",
"codex-model-provider-info",
"codex-models-manager",
"codex-protocol",
"codex-rollout",
"codex-thread-store",
"codex-utils-absolute-path",
"tracing",
]
[[package]]
name = "codex-thread-store"
version = "0.0.0"
codex-rs/Cargo.toml
+1
View File
@@ -97,6 +97,7 @@ members = [
"state",
"terminal-detection",
"test-binary-support",
"thread-manager-sample",
"thread-store",
"uds",
"codex-experimental-api-macros",
codex-rs/app-server/src/bespoke_event_handling.rs
+6 -1
View File
@@ -3591,7 +3591,12 @@ mod tests {
thread_id: conversation_id,
thread: conversation,
..
} = thread_manager.start_thread(config).await?;
} = thread_manager
.start_thread(
config.clone(),
codex_core::thread_store_from_config(&config),
)
.await?;
let thread_state = new_thread_state();
let thread_watch_manager = ThreadWatchManager::new();
let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY);
codex-rs/app-server/src/codex_message_processor.rs
+10 -5
View File
@@ -691,7 +691,7 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) log_db: Option<LogDbLayer>,
}
fn configured_thread_store(config: &Config) -> Arc<dyn ThreadStore> {
fn thread_store_from_config(config: &Config) -> Arc<dyn ThreadStore> {
match &config.experimental_thread_store {
ThreadStoreConfig::Local => Arc::new(configured_local_thread_store(config)),
ThreadStoreConfig::Remote { endpoint } => Arc::new(RemoteThreadStore::new(endpoint)),
@@ -790,7 +790,7 @@ impl CodexMessageProcessor {
outgoing: outgoing.clone(),
analytics_events_client,
arg0_paths,
thread_store: configured_thread_store(&config),
thread_store: thread_store_from_config(&config),
config,
config_manager,
active_login: Arc::new(Mutex::new(None)),
@@ -2509,6 +2509,7 @@ impl CodexMessageProcessor {
let imported_thread = self
.thread_manager
.start_thread_with_options(StartThreadOptions {
thread_store: thread_store_from_config(&config),
config,
initial_history: InitialHistory::Forked(rollout_items),
session_source: None,
@@ -2706,6 +2707,7 @@ impl CodexMessageProcessor {
} = listener_task_context
.thread_manager
.start_thread_with_options(StartThreadOptions {
thread_store: thread_store_from_config(&config),
config,
initial_history: match session_start_source
.unwrap_or(codex_app_server_protocol::ThreadStartSource::Startup)
@@ -4278,7 +4280,8 @@ impl CodexMessageProcessor {
match self
.thread_manager
.resume_thread_with_history(
config,
config.clone(),
thread_store_from_config(&config),
thread_history,
self.auth_manager.clone(),
persist_extended_history,
@@ -4852,7 +4855,7 @@ impl CodexMessageProcessor {
let fallback_model_provider = config.model_provider_id.clone();
let instruction_sources = Self::instruction_sources_from_config(&config).await;
let fork_thread_store = configured_thread_store(&config);
let fork_thread_store = thread_store_from_config(&config);
let NewThread {
thread_id,
@@ -4864,6 +4867,7 @@ impl CodexMessageProcessor {
.fork_thread_from_history(
ForkSnapshot::Interrupted,
config,
fork_thread_store.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: source_thread_id,
history: history_items.clone(),
@@ -7084,7 +7088,8 @@ impl CodexMessageProcessor {
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
config.clone(),
thread_store_from_config(&config),
rollout_path,
/*persist_extended_history*/ false,
self.request_trace_context(request_id).await,
codex-rs/core/src/agent/control.rs
+20 -7
View File
@@ -12,7 +12,9 @@ use crate::session::emit_subagent_session_started;
use crate::session_prefix::format_subagent_context_line;
use crate::session_prefix::format_subagent_notification_message;
use crate::shell_snapshot::ShellSnapshot;
use crate::thread_manager::ResumeThreadFromRolloutOptions;
use crate::thread_manager::ThreadManagerState;
use crate::thread_manager::thread_store_from_config;
use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns;
use codex_features::Feature;
use codex_protocol::AgentPath;
@@ -232,7 +234,8 @@ impl AgentControl {
(Some(session_source), None) => {
state
.spawn_new_thread_with_source(
config,
config.clone(),
thread_store_from_config(&config),
self.clone(),
session_source,
/*persist_extended_history*/ false,
@@ -243,7 +246,15 @@ impl AgentControl {
)
.await?
}
(None, _) => state.spawn_new_thread(config, self.clone()).await?,
(None, _) => {
state
.spawn_new_thread(
config.clone(),
thread_store_from_config(&config),
self.clone(),
)
.await?
}
};
agent_metadata.agent_id = Some(new_thread.thread_id);
reservation.commit(agent_metadata.clone());
@@ -424,7 +435,8 @@ impl AgentControl {
state
.fork_thread_with_source(
config,
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(forked_rollout_items),
self.clone(),
session_source,
@@ -578,14 +590,15 @@ impl AgentControl {
};
let resumed_thread = state
.resume_thread_from_rollout_with_source(
config,
.resume_thread_from_rollout_with_source(ResumeThreadFromRolloutOptions {
config: config.clone(),
thread_store: thread_store_from_config(&config),
rollout_path,
self.clone(),
agent_control: self.clone(),
session_source,
inherited_shell_snapshot,
inherited_exec_policy,
)
})
.await?;
let mut agent_metadata = agent_metadata;
agent_metadata.agent_id = Some(resumed_thread.thread_id);
codex-rs/core/src/agent/control_tests.rs
+11 -4
View File
@@ -7,6 +7,7 @@ use crate::config::Config;
use crate::config::ConfigBuilder;
use crate::context::ContextualUserFragment;
use crate::context::SubagentNotification;
use crate::thread_manager::thread_store_from_config;
use assert_matches::assert_matches;
use codex_features::Feature;
use codex_login::CodexAuth;
@@ -108,7 +109,7 @@ impl AgentControlHarness {
async fn start_thread(&self) -> (ThreadId, Arc<CodexThread>) {
let new_thread = self
.manager
.start_thread(self.config.clone())
.start_thread(self.config.clone(), thread_store_from_config(&self.config))
.await
.expect("start thread");
(new_thread.thread_id, new_thread.thread)
@@ -609,7 +610,10 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
Some("Child subagent guidance.".to_string());
let new_thread = harness
.manager
.start_thread(parent_config)
.start_thread(
parent_config.clone(),
thread_store_from_config(&parent_config),
)
.await
.expect("start parent thread");
let parent_thread_id = new_thread.thread_id;
@@ -952,7 +956,7 @@ async fn spawn_agent_respects_max_threads_limit() {
let control = manager.agent_control();
let _ = manager
.start_thread(config.clone())
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
@@ -1309,7 +1313,10 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
let _ = tester_config.features.enable(Feature::MultiAgentV2);
let tester_thread_id = harness
.manager
.start_thread(tester_config)
.start_thread(
tester_config.clone(),
thread_store_from_config(&tester_config),
)
.await
.expect("tester thread should start")
.thread_id;
codex-rs/core/src/config/managed_features.rs
+12
View File
@@ -27,6 +27,18 @@ pub struct ManagedFeatures {
pinned_features: BTreeMap<Feature, bool>,
}
impl Default for ManagedFeatures {
fn default() -> Self {
Self {
value: ConstrainedWithSource::new(
Constrained::allow_any(Features::default()),
/*source*/ None,
),
pinned_features: BTreeMap::new(),
}
}
}
impl ManagedFeatures {
pub(crate) fn from_configured(
configured_features: Features,
codex-rs/core/src/lib.rs
+1
View File
@@ -120,6 +120,7 @@ pub use thread_manager::NewThread;
pub use thread_manager::StartThreadOptions;
pub use thread_manager::ThreadManager;
pub use thread_manager::build_models_manager;
pub use thread_manager::thread_store_from_config;
pub use web_search::web_search_action_detail;
pub use web_search::web_search_detail;
pub use windows_sandbox_read_grants::grant_read_root_non_elevated;
codex-rs/core/src/prompt_debug.rs
+3 -1
View File
@@ -19,6 +19,7 @@ use crate::session::session::Session;
use crate::session::turn::build_prompt;
use crate::session::turn::built_tools;
use crate::thread_manager::ThreadManager;
use crate::thread_manager::thread_store_from_config;
/// Build the model-visible `input` list for a single debug turn.
#[doc(hidden)]
@@ -48,7 +49,8 @@ pub async fn build_prompt_input(
Arc::new(EnvironmentManager::new(EnvironmentManagerArgs::new(local_runtime_paths)).await),
/*analytics_events_client*/ None,
);
let thread = thread_manager.start_thread(config).await?;
let thread_store = thread_store_from_config(&config);
let thread = thread_manager.start_thread(config, thread_store).await?;
let output = build_prompt_input_from_session(thread.thread.codex.session.as_ref(), input).await;
let shutdown = thread.thread.shutdown_and_wait().await;
codex-rs/core/src/session/tests.rs
+4 -1
View File
@@ -1643,7 +1643,10 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result<
.thread_manager
.fork_thread(
usize::MAX,
fork_config,
fork_config.clone(),
std::sync::Arc::new(codex_thread_store::LocalThreadStore::new(
codex_rollout::RolloutConfig::from_view(&fork_config),
)),
rollout_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
codex-rs/core/src/test_support.rs
+5 -1
View File
@@ -20,6 +20,7 @@ use codex_models_manager::test_support::get_model_offline_for_tests;
use codex_protocol::config_types::CollaborationModeMask;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelPreset;
use codex_thread_store::ThreadStore;
use once_cell::sync::Lazy;
use crate::ThreadManager;
@@ -76,16 +77,18 @@ pub fn thread_manager_with_models_provider_and_home(
pub async fn start_thread_with_user_shell_override(
thread_manager: &ThreadManager,
config: Config,
thread_store: Arc<dyn ThreadStore>,
user_shell_override: crate::shell::Shell,
) -> codex_protocol::error::Result<crate::NewThread> {
thread_manager
.start_thread_with_user_shell_override_for_tests(config, user_shell_override)
.start_thread_with_user_shell_override_for_tests(config, thread_store, user_shell_override)
.await
}
pub async fn resume_thread_from_rollout_with_user_shell_override(
thread_manager: &ThreadManager,
config: Config,
thread_store: Arc<dyn ThreadStore>,
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
user_shell_override: crate::shell::Shell,
@@ -93,6 +96,7 @@ pub async fn resume_thread_from_rollout_with_user_shell_override(
thread_manager
.resume_thread_from_rollout_with_user_shell_override_for_tests(
config,
thread_store,
rollout_path,
auth_manager,
user_shell_override,
codex-rs/core/src/thread_manager.rs
+45 -17
View File
@@ -213,6 +213,7 @@ pub struct ThreadManager {
pub struct StartThreadOptions {
pub config: Config,
pub thread_store: Arc<dyn ThreadStore>,
pub initial_history: InitialHistory,
pub session_source: Option<SessionSource>,
pub dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
@@ -222,6 +223,16 @@ pub struct StartThreadOptions {
pub environments: Vec<TurnEnvironmentSelection>,
}
pub(crate) struct ResumeThreadFromRolloutOptions {
pub(crate) config: Config,
pub(crate) thread_store: Arc<dyn ThreadStore>,
pub(crate) rollout_path: PathBuf,
pub(crate) agent_control: AgentControl,
pub(crate) session_source: SessionSource,
pub(crate) inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
pub(crate) inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
}
/// Shared, `Arc`-owned state for [`ThreadManager`]. This `Arc` is required to have a single
/// `Arc` reference that can be downgraded to by `AgentControl` while preventing every single
/// function to require an `Arc<&Self>`.
@@ -254,7 +265,7 @@ pub fn build_models_manager(
)
}
fn configured_thread_store(config: &Config) -> Arc<dyn ThreadStore> {
pub fn thread_store_from_config(config: &Config) -> Arc<dyn ThreadStore> {
match &config.experimental_thread_store {
ThreadStoreConfig::Local => {
Arc::new(LocalThreadStore::new(RolloutConfig::from_view(config)))
@@ -520,11 +531,16 @@ impl ThreadManager {
Ok(subtree_thread_ids)
}
pub async fn start_thread(&self, config: Config) -> CodexResult<NewThread> {
pub async fn start_thread(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
) -> CodexResult<NewThread> {
// Box delegated thread-spawn futures so these convenience wrappers do
// not inline the full spawn path into every caller's async state.
Box::pin(self.start_thread_with_tools(
config,
thread_store,
Vec::new(),
/*persist_extended_history*/ false,
))
@@ -534,6 +550,7 @@ impl ThreadManager {
pub async fn start_thread_with_tools(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
persist_extended_history: bool,
) -> CodexResult<NewThread> {
@@ -543,6 +560,7 @@ impl ThreadManager {
);
Box::pin(self.start_thread_with_options(StartThreadOptions {
config,
thread_store,
initial_history: InitialHistory::New,
session_source: None,
dynamic_tools,
@@ -558,13 +576,12 @@ impl ThreadManager {
&self,
options: StartThreadOptions,
) -> CodexResult<NewThread> {
let thread_store = configured_thread_store(&options.config);
let session_source = options
.session_source
.unwrap_or_else(|| self.state.session_source.clone());
Box::pin(self.state.spawn_thread_with_source(
options.config,
thread_store,
options.thread_store,
options.initial_history,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
@@ -584,6 +601,7 @@ impl ThreadManager {
pub async fn resume_thread_from_rollout(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
parent_trace: Option<W3cTraceContext>,
@@ -591,6 +609,7 @@ impl ThreadManager {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
Box::pin(self.resume_thread_with_history(
config,
thread_store,
initial_history,
auth_manager,
/*persist_extended_history*/ false,
@@ -602,12 +621,12 @@ impl ThreadManager {
pub async fn resume_thread_with_history(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
let thread_store = configured_thread_store(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
@@ -631,9 +650,9 @@ impl ThreadManager {
pub(crate) async fn start_thread_with_user_shell_override_for_tests(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
user_shell_override: crate::shell::Shell,
) -> CodexResult<NewThread> {
let thread_store = configured_thread_store(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
@@ -657,12 +676,12 @@ impl ThreadManager {
pub(crate) async fn resume_thread_from_rollout_with_user_shell_override_for_tests(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
rollout_path: PathBuf,
auth_manager: Arc<AuthManager>,
user_shell_override: crate::shell::Shell,
) -> CodexResult<NewThread> {
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
let thread_store = configured_thread_store(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
@@ -749,6 +768,7 @@ impl ThreadManager {
&self,
snapshot: S,
config: Config,
thread_store: Arc<dyn ThreadStore>,
path: PathBuf,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
@@ -761,6 +781,7 @@ impl ThreadManager {
self.fork_thread_from_history(
snapshot,
config,
thread_store,
history,
persist_extended_history,
parent_trace,
@@ -773,6 +794,7 @@ impl ThreadManager {
&self,
snapshot: S,
config: Config,
thread_store: Arc<dyn ThreadStore>,
history: InitialHistory,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
@@ -783,6 +805,7 @@ impl ThreadManager {
self.fork_thread_with_initial_history(
snapshot.into(),
config,
thread_store,
history,
persist_extended_history,
parent_trace,
@@ -794,13 +817,13 @@ impl ThreadManager {
&self,
snapshot: ForkSnapshot,
config: Config,
thread_store: Arc<dyn ThreadStore>,
history: InitialHistory,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config);
let history = fork_history_from_snapshot(snapshot, history, interrupted_marker);
let thread_store = configured_thread_store(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
&config.cwd,
@@ -887,10 +910,12 @@ impl ThreadManagerState {
pub(crate) async fn spawn_new_thread(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
agent_control: AgentControl,
) -> CodexResult<NewThread> {
Box::pin(self.spawn_new_thread_with_source(
config,
thread_store,
agent_control,
self.session_source.clone(),
/*persist_extended_history*/ false,
@@ -906,6 +931,7 @@ impl ThreadManagerState {
pub(crate) async fn spawn_new_thread_with_source(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
agent_control: AgentControl,
session_source: SessionSource,
persist_extended_history: bool,
@@ -914,7 +940,6 @@ impl ThreadManagerState {
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
environments: Option<Vec<TurnEnvironmentSelection>>,
) -> CodexResult<NewThread> {
let thread_store = configured_thread_store(&config);
let environments = environments.unwrap_or_else(|| {
default_thread_environment_selections(self.environment_manager.as_ref(), &config.cwd)
});
@@ -939,15 +964,18 @@ impl ThreadManagerState {
pub(crate) async fn resume_thread_from_rollout_with_source(
&self,
config: Config,
rollout_path: PathBuf,
agent_control: AgentControl,
session_source: SessionSource,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
options: ResumeThreadFromRolloutOptions,
) -> CodexResult<NewThread> {
let ResumeThreadFromRolloutOptions {
config,
thread_store,
rollout_path,
agent_control,
session_source,
inherited_shell_snapshot,
inherited_exec_policy,
} = options;
let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?;
let thread_store = configured_thread_store(&config);
let environments =
default_thread_environment_selections(self.environment_manager.as_ref(), &config.cwd);
Box::pin(self.spawn_thread_with_source(
@@ -973,6 +1001,7 @@ impl ThreadManagerState {
pub(crate) async fn fork_thread_with_source(
&self,
config: Config,
thread_store: Arc<dyn ThreadStore>,
initial_history: InitialHistory,
agent_control: AgentControl,
session_source: SessionSource,
@@ -981,7 +1010,6 @@ impl ThreadManagerState {
inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
environments: Option<Vec<TurnEnvironmentSelection>>,
) -> CodexResult<NewThread> {
let thread_store = configured_thread_store(&config);
let environments = environments.unwrap_or_else(|| {
default_thread_environment_selections(self.environment_manager.as_ref(), &config.cwd)
});
codex-rs/core/src/thread_manager_tests.rs
+30 -7
View File
@@ -162,7 +162,8 @@ fn fork_thread_accepts_legacy_usize_snapshot_argument() {
) {
let _future = manager.fork_thread(
usize::MAX,
config,
config.clone(),
thread_store_from_config(&config),
path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -263,12 +264,12 @@ async fn shutdown_all_threads_bounded_submits_shutdown_to_every_thread() {
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
);
let thread_1 = manager
.start_thread(config.clone())
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start first thread")
.thread_id;
let thread_2 = manager
.start_thread(config)
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start second thread")
.thread_id;
@@ -314,6 +315,7 @@ async fn start_thread_accepts_explicit_environment_when_default_environment_is_d
let thread = manager
.start_thread_with_options(StartThreadOptions {
thread_store: thread_store_from_config(&config),
config: config.clone(),
initial_history: InitialHistory::New,
session_source: None,
@@ -346,8 +348,10 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
config.codex_home.to_path_buf(),
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
);
let thread_store = thread_store_from_config(&config);
let thread = manager
.start_thread_with_options(StartThreadOptions {
thread_store,
config,
initial_history: InitialHistory::New,
session_source: Some(SessionSource::Internal(
@@ -399,9 +403,11 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
cwd: selected_cwd.clone(),
}];
let default_cwd = config.cwd.clone();
let thread_store = thread_store_from_config(&config);
let source = manager
.start_thread_with_options(StartThreadOptions {
thread_store: Arc::clone(&thread_store),
config: config.clone(),
initial_history: InitialHistory::New,
session_source: None,
@@ -423,10 +429,17 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
.thread
.rollout_path()
.expect("source rollout path should exist");
source
.thread
.shutdown_and_wait()
.await
.expect("shutdown source thread before resume");
let _ = manager.remove_thread(&source.thread_id).await;
let resumed = manager
.resume_thread_from_rollout(
config.clone(),
Arc::clone(&thread_store),
rollout_path.clone(),
auth_manager,
/*parent_trace*/ None,
@@ -448,6 +461,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
.fork_thread(
ForkSnapshot::Interrupted,
config,
thread_store,
rollout_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -704,6 +718,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
let source = manager
.resume_thread_with_history(
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
@@ -729,7 +744,8 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
let forked = manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
config.clone(),
thread_store_from_config(&config),
source_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -806,6 +822,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
let source = manager
.resume_thread_with_history(
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(vec![
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: "turn-explicit".to_string(),
@@ -842,7 +859,8 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
let forked = manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
config.clone(),
thread_store_from_config(&config),
source_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -897,6 +915,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
let source = manager
.resume_thread_with_history(
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(vec![
RolloutItem::ResponseItem(user_msg("hello")),
RolloutItem::ResponseItem(assistant_msg("partial")),
@@ -921,6 +940,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
.fork_thread(
ForkSnapshot::Interrupted,
config.clone(),
thread_store_from_config(&config),
source_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -960,7 +980,8 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
let reforked = manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
config.clone(),
thread_store_from_config(&config),
forked_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -1033,6 +1054,7 @@ async fn resumed_thread_activates_paused_goal_and_continues_on_request() -> anyh
let source = manager
.resume_thread_with_history(
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("keep working"))]),
auth_manager.clone(),
/*persist_extended_history*/ false,
@@ -1061,7 +1083,8 @@ async fn resumed_thread_activates_paused_goal_and_continues_on_request() -> anyh
let resumed = manager
.resume_thread_from_rollout(
config,
config.clone(),
thread_store_from_config(&config),
source_path,
auth_manager,
/*parent_trace*/ None,
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+154 -41
View File
@@ -5,6 +5,7 @@ use crate::config::DEFAULT_AGENT_MAX_DEPTH;
use crate::function_tool::FunctionCallError;
use crate::session::tests::make_session_and_context;
use crate::session_prefix::format_subagent_notification_message;
use crate::thread_manager::thread_store_from_config;
use crate::tools::context::ToolOutput;
use crate::tools::handlers::multi_agents_v2::CloseAgentHandler as CloseAgentHandlerV2;
use crate::tools::handlers::multi_agents_v2::FollowupTaskHandler as FollowupTaskHandlerV2;
@@ -296,7 +297,10 @@ async fn spawn_agent_fork_context_rejects_agent_type_override() {
let role_name = install_role_with_model_override(&mut turn).await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -328,7 +332,10 @@ async fn spawn_agent_fork_context_rejects_child_model_overrides() {
let (mut session, turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -363,7 +370,10 @@ async fn multi_agent_v2_spawn_fork_turns_all_rejects_agent_type_override() {
let role_name = install_role_with_model_override(&mut turn).await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -406,7 +416,10 @@ async fn multi_agent_v2_spawn_defaults_to_full_fork_and_rejects_child_model_over
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -447,7 +460,10 @@ async fn multi_agent_v2_spawn_partial_fork_turns_allows_agent_type_override() {
let role_name = install_role_with_model_override(&mut turn).await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -530,7 +546,10 @@ async fn multi_agent_v2_spawn_requires_task_name() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -564,7 +583,10 @@ async fn multi_agent_v2_spawn_rejects_legacy_items_field() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -624,7 +646,10 @@ async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_pat
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -721,7 +746,10 @@ async fn multi_agent_v2_spawn_rejects_legacy_fork_context() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -760,7 +788,10 @@ async fn multi_agent_v2_spawn_rejects_invalid_fork_turns_string() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -799,7 +830,10 @@ async fn multi_agent_v2_spawn_rejects_zero_fork_turns() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -838,7 +872,10 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -914,7 +951,10 @@ async fn multi_agent_v2_followup_task_rejects_root_target_from_child() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -995,7 +1035,10 @@ async fn multi_agent_v2_list_agents_returns_completed_status_and_last_task_messa
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1089,7 +1132,10 @@ async fn multi_agent_v2_list_agents_filters_by_relative_path_prefix() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1176,7 +1222,10 @@ async fn multi_agent_v2_list_agents_omits_closed_agents() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1240,7 +1289,10 @@ async fn multi_agent_v2_send_message_rejects_legacy_items_field() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1296,7 +1348,10 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1369,7 +1424,10 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1504,7 +1562,10 @@ async fn multi_agent_v2_followup_task_rejects_legacy_items_field() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1557,7 +1618,10 @@ async fn multi_agent_v2_interrupted_turn_does_not_notify_parent() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1634,7 +1698,10 @@ async fn multi_agent_v2_spawn_omits_agent_id_when_named() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1673,7 +1740,10 @@ async fn multi_agent_v2_spawn_surfaces_task_name_validation_errors() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -1887,7 +1957,7 @@ async fn multi_agent_v2_spawn_agent_ignores_configured_max_depth() {
.enable(Feature::MultiAgentV2)
.expect("test config should allow feature update");
let root = manager
.start_thread(config.clone())
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2011,7 +2081,10 @@ async fn send_input_interrupts_before_prompt() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let invocation = invocation(
Arc::new(session),
@@ -2050,7 +2123,10 @@ async fn send_input_accepts_structured_items() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let invocation = invocation(
Arc::new(session),
@@ -2142,7 +2218,10 @@ async fn resume_agent_noops_for_active_agent() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let status_before = manager.agent_control().get_status(agent_id).await;
let invocation = invocation(
@@ -2180,7 +2259,8 @@ async fn resume_agent_restores_closed_agent_and_accepts_send_input() {
let config = turn.config.as_ref().clone();
let thread = manager
.resume_thread_with_history(
config,
config.clone(),
thread_store_from_config(&config),
InitialHistory::Forked(vec![RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
@@ -2345,7 +2425,10 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2521,7 +2604,10 @@ async fn wait_agent_times_out_when_status_is_not_final() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let invocation = invocation(
Arc::new(session),
@@ -2561,7 +2647,10 @@ async fn wait_agent_clamps_short_timeouts_to_minimum() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let invocation = invocation(
Arc::new(session),
@@ -2596,7 +2685,10 @@ async fn wait_agent_returns_final_status_without_timeout() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let mut status_rx = manager
.agent_control()
@@ -2644,7 +2736,10 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2735,7 +2830,10 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2813,7 +2911,10 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2901,7 +3002,10 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -2987,7 +3091,10 @@ async fn multi_agent_v2_close_agent_accepts_task_name_target() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -3046,7 +3153,10 @@ async fn multi_agent_v2_close_agent_rejects_root_target_and_id() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.start_thread(
(*turn.config).clone(),
thread_store_from_config(turn.config.as_ref()),
)
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
@@ -3095,7 +3205,10 @@ async fn close_agent_submits_shutdown_and_returns_previous_status() {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let config = turn.config.as_ref().clone();
let thread = manager.start_thread(config).await.expect("start thread");
let thread = manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("start thread");
let agent_id = thread.thread_id;
let status_before = manager.agent_control().get_status(agent_id).await;
@@ -3137,7 +3250,7 @@ async fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtr
.expect("test config should allow sqlite");
let parent = manager
.start_thread(config.clone())
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("parent thread should start");
let parent_thread_id = parent.thread_id;
@@ -3268,7 +3381,7 @@ async fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtr
);
let operator = manager
.start_thread(config)
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("operator thread should start");
let operator_session = operator.thread.codex.session.clone();
codex-rs/core/tests/common/test_codex.rs
+10 -1
View File
@@ -18,6 +18,7 @@ use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::shell::Shell;
use codex_core::shell::get_shell_by_model_provided_path;
use codex_core::thread_store_from_config;
use codex_exec_server::CreateDirectoryOptions;
use codex_exec_server::ExecutorFileSystem;
use codex_exec_server::RemoveOptions;
@@ -450,6 +451,7 @@ impl TestCodexBuilder {
codex_core::test_support::resume_thread_from_rollout_with_user_shell_override(
thread_manager.as_ref(),
config.clone(),
thread_store_from_config(&config),
path,
auth_manager,
user_shell_override,
@@ -461,6 +463,7 @@ impl TestCodexBuilder {
let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
Box::pin(thread_manager.resume_thread_from_rollout(
config.clone(),
thread_store_from_config(&config),
path,
auth_manager,
/*parent_trace*/ None,
@@ -472,12 +475,18 @@ impl TestCodexBuilder {
codex_core::test_support::start_thread_with_user_shell_override(
thread_manager.as_ref(),
config.clone(),
thread_store_from_config(&config),
user_shell_override,
),
)
.await?
}
(None, None) => Box::pin(thread_manager.start_thread(config.clone())).await?,
(None, None) => {
Box::pin(
thread_manager.start_thread(config.clone(), thread_store_from_config(&config)),
)
.await?
}
};
Ok(TestCodex {
codex-rs/core/tests/suite/client.rs
+2 -1
View File
@@ -5,6 +5,7 @@ use codex_core::NewThread;
use codex_core::Prompt;
use codex_core::ResponseEvent;
use codex_core::ThreadManager;
use codex_core::thread_store_from_config;
use codex_features::Feature;
use codex_login::AuthManager;
use codex_login::CodexAuth;
@@ -1114,7 +1115,7 @@ async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() {
/*analytics_events_client*/ None,
);
let NewThread { thread: codex, .. } = thread_manager
.start_thread(config)
.start_thread(config.clone(), thread_store_from_config(&config))
.await
.expect("create new conversation");
codex-rs/core/tests/suite/code_mode.rs
+1
View File
@@ -2556,6 +2556,7 @@ async fn code_mode_can_call_hidden_dynamic_tools() -> Result<()> {
.thread_manager
.start_thread_with_tools(
base_test.config.clone(),
codex_core::thread_store_from_config(&base_test.config),
vec![DynamicToolSpec {
namespace: Some("codex_app".to_string()),
name: "hidden_dynamic_tool".to_string(),
codex-rs/core/tests/suite/compact_remote.rs
+1
View File
@@ -441,6 +441,7 @@ async fn remote_compact_filters_deferred_dynamic_tools() -> Result<()> {
.thread_manager
.start_thread_with_tools(
test.config.clone(),
codex_core::thread_store_from_config(&test.config),
dynamic_tools,
/*persist_extended_history*/ false,
)
codex-rs/core/tests/suite/compact_resume_fork.rs
+2
View File
@@ -825,6 +825,7 @@ async fn resume_conversation(
);
Box::pin(manager.resume_thread_from_rollout(
config.clone(),
codex_core::thread_store_from_config(config),
path,
auth_manager,
/*parent_trace*/ None,
@@ -844,6 +845,7 @@ async fn fork_thread(
Box::pin(manager.fork_thread(
nth_user_message,
config.clone(),
codex_core::thread_store_from_config(config),
path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
codex-rs/core/tests/suite/fork_thread.rs
+4 -1
View File
@@ -100,6 +100,7 @@ async fn fork_thread_twice_drops_to_first_message() {
.fork_thread(
ForkSnapshot::TruncateBeforeNthUserMessage(1),
config_for_fork.clone(),
codex_core::thread_store_from_config(&config_for_fork),
base_path.clone(),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -124,6 +125,7 @@ async fn fork_thread_twice_drops_to_first_message() {
.fork_thread(
ForkSnapshot::TruncateBeforeNthUserMessage(0),
config_for_fork.clone(),
codex_core::thread_store_from_config(&config_for_fork),
fork1_path.clone(),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
@@ -191,7 +193,8 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() {
} = thread_manager
.fork_thread_from_history(
ForkSnapshot::Interrupted,
test.config,
test.config.clone(),
codex_core::thread_store_from_config(&test.config),
InitialHistory::Resumed(ResumedHistory {
conversation_id: test.session_configured.session_id,
history: source_items.clone(),
codex-rs/core/tests/suite/permissions_messages.rs
+2 -1
View File
@@ -495,7 +495,8 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> {
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
fork_config,
fork_config.clone(),
codex_core::thread_store_from_config(&fork_config),
rollout_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
codex-rs/core/tests/suite/realtime_conversation.rs
+1
View File
@@ -1623,6 +1623,7 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
.thread_manager
.resume_thread_with_history(
test.config.clone(),
codex_core::thread_store_from_config(&test.config),
InitialHistory::Forked(history),
auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
/*persist_extended_history*/ false,
codex-rs/core/tests/suite/resume_warning.rs
+2 -1
View File
@@ -105,7 +105,8 @@ async fn emits_warning_when_resumed_model_differs() {
..
} = thread_manager
.resume_thread_with_history(
config,
config.clone(),
codex_core::thread_store_from_config(&config),
initial_history,
auth_manager,
/*persist_extended_history*/ false,
codex-rs/core/tests/suite/search_tool.rs
+1
View File
@@ -793,6 +793,7 @@ async fn tool_search_returns_deferred_dynamic_tool_and_routes_follow_up_call() -
.thread_manager
.start_thread_with_tools(
base_test.config.clone(),
codex_core::thread_store_from_config(&base_test.config),
vec![dynamic_tool],
/*persist_extended_history*/ false,
)
codex-rs/core/tests/suite/skills.rs
+4 -1
View File
@@ -3,6 +3,7 @@
use anyhow::Result;
use codex_core::ThreadManager;
use codex_core::thread_store_from_config;
use codex_exec_server::CreateDirectoryOptions;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::ExecServerRuntimePaths;
@@ -248,7 +249,9 @@ async fn list_skills_skips_cwd_roots_when_environment_disabled() -> Result<()> {
)),
/*analytics_events_client*/ None,
);
let new_thread = thread_manager.start_thread(config.clone()).await?;
let new_thread = thread_manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await?;
let cwd = config.cwd.to_path_buf();
new_thread
codex-rs/core/tests/suite/unstable_features_warning.rs
+4 -2
View File
@@ -43,7 +43,8 @@ async fn emits_warning_when_unstable_features_enabled_via_config() {
..
} = thread_manager
.resume_thread_with_history(
config,
config.clone(),
codex_core::thread_store_from_config(&config),
InitialHistory::New,
auth_manager,
/*persist_extended_history*/ false,
@@ -90,7 +91,8 @@ async fn suppresses_warning_when_configured() {
..
} = thread_manager
.resume_thread_with_history(
config,
config.clone(),
codex_core::thread_store_from_config(&config),
InitialHistory::New,
auth_manager,
/*persist_extended_history*/ false,
codex-rs/core/tests/suite/window_headers.rs
+1
View File
@@ -71,6 +71,7 @@ async fn window_id_advances_after_compact_persists_on_resume_and_resets_on_fork(
.fork_thread(
/*snapshot*/ 0usize,
resumed.config.clone(),
codex_core::thread_store_from_config(&resumed.config),
rollout_path,
/*persist_extended_history*/ false,
/*parent_trace*/ None,
codex-rs/mcp-server/src/codex_tool_runner.rs
+5 -1
View File
@@ -13,6 +13,7 @@ use codex_core::CodexThread;
use codex_core::NewThread;
use codex_core::ThreadManager;
use codex_core::config::Config as CodexConfig;
use codex_core::thread_store_from_config;
use codex_protocol::ThreadId;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
@@ -68,7 +69,10 @@ pub async fn run_codex_tool_session(
thread_id,
thread,
session_configured,
} = match thread_manager.start_thread(config).await {
} = match thread_manager
.start_thread(config.clone(), thread_store_from_config(&config))
.await
{
Ok(res) => res,
Err(e) => {
let result = CallToolResult {
codex-rs/memories/write/src/runtime.rs
+2
View File
@@ -8,6 +8,7 @@ use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::content_items_to_text;
use codex_core::resolve_installation_id;
use codex_core::thread_store_from_config;
use codex_features::Feature;
use codex_login::AuthManager;
use codex_login::CodexAuth;
@@ -236,6 +237,7 @@ impl MemoryStartupContext {
} = self
.thread_manager
.start_thread_with_options(StartThreadOptions {
thread_store: thread_store_from_config(&config),
config,
initial_history: InitialHistory::New,
session_source: Some(SessionSource::Internal(
codex-rs/thread-manager-sample/BUILD.bazel
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "thread-manager-sample",
crate_name = "codex_thread_manager_sample",
)
codex-rs/thread-manager-sample/Cargo.toml
+25
View File
@@ -0,0 +1,25 @@
[package]
name = "codex-thread-manager-sample"
version.workspace = true
edition.workspace = true
license.workspace = true
[lints]
workspace = true
[dependencies]
anyhow = { workspace = true }
clap = { workspace = true, features = ["derive"] }
codex-arg0 = { workspace = true }
codex-config = { workspace = true }
codex-core = { workspace = true }
codex-exec-server = { workspace = true }
codex-features = { workspace = true }
codex-login = { workspace = true }
codex-model-provider-info = { workspace = true }
codex-models-manager = { workspace = true }
codex-protocol = { workspace = true }
codex-rollout = { workspace = true }
codex-thread-store = { workspace = true }
codex-utils-absolute-path = { workspace = true }
tracing = { workspace = true }
codex-rs/thread-manager-sample/README.md
+20
View File
@@ -0,0 +1,20 @@
# ThreadManager Sample
Small one-shot binary that starts a Codex thread with `ThreadManager`, submits a
single user turn, and prints the final assistant message.
```sh
cargo run -p codex-thread-manager-sample -- "Say hello"
```
Use `--model` to override the configured default model:
```sh
cargo run -p codex-thread-manager-sample -- --model gpt-5.2 "Say hello"
```
The prompt can also be piped through stdin:
```sh
printf 'Say hello\n' | cargo run -p codex-thread-manager-sample
```
codex-rs/thread-manager-sample/src/main.rs
+321
View File
@@ -0,0 +1,321 @@
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::io::IsTerminal;
use std::io::Read;
use std::io::Write;
use std::sync::Arc;
use anyhow::Context;
use anyhow::bail;
use clap::Parser;
use codex_arg0::Arg0DispatchPaths;
use codex_arg0::arg0_dispatch_or_else;
use codex_config::ConfigLayerStack;
use codex_config::config_toml::ProjectConfig;
use codex_config::config_toml::RealtimeAudioConfig;
use codex_config::config_toml::RealtimeConfig;
use codex_config::types::AuthCredentialsStoreMode;
use codex_config::types::History;
use codex_config::types::MemoriesConfig;
use codex_config::types::ModelAvailabilityNuxConfig;
use codex_config::types::Notice;
use codex_config::types::OAuthCredentialsStoreMode;
use codex_config::types::OtelConfig;
use codex_config::types::ToolSuggestConfig;
use codex_config::types::TuiKeymap;
use codex_config::types::TuiNotificationSettings;
use codex_config::types::UriBasedFileOpener;
use codex_core::CodexThread;
use codex_core::NewThread;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config::Constrained;
use codex_core::config::GhostSnapshotConfig;
use codex_core::config::MultiAgentV2Config;
use codex_core::config::Permissions;
use codex_core::config::TerminalResizeReflowConfig;
use codex_core::config::ThreadStoreConfig;
use codex_core::config::find_codex_home;
use codex_exec_server::EnvironmentManager;
use codex_exec_server::EnvironmentManagerArgs;
use codex_exec_server::ExecServerRuntimePaths;
use codex_features::Feature;
use codex_login::AuthManager;
use codex_login::default_client::set_default_originator;
use codex_model_provider_info::OPENAI_PROVIDER_ID;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_protocol::config_types::AltScreenMode;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ShellEnvironmentPolicy;
use codex_protocol::config_types::WebSearchMode;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionSource;
use codex_protocol::user_input::UserInput;
use codex_rollout::RolloutConfig;
use codex_thread_store::LocalThreadStore;
use codex_utils_absolute_path::AbsolutePathBuf;
#[derive(Debug, Parser)]
#[command(
name = "codex-thread-manager-sample",
about = "Run one Codex turn through ThreadManager and print the final assistant output."
)]
struct Args {
/// Override the model for this run.
#[arg(long, value_name = "MODEL")]
model: Option<String>,
/// Prompt text. If omitted, the prompt is read from piped stdin.
#[arg(value_name = "PROMPT", num_args = 0.., trailing_var_arg = true)]
prompt: Vec<String>,
}
fn main() -> anyhow::Result<()> {
arg0_dispatch_or_else(run_main)
}
async fn run_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
if let Err(err) = set_default_originator("codex_thread_manager_sample".to_string()) {
tracing::warn!("failed to set originator: {err:?}");
}
let args = Args::parse();
let prompt = if args.prompt.is_empty() {
if std::io::stdin().is_terminal() {
bail!("no prompt provided; pass a prompt argument or pipe one into stdin");
}
let mut prompt = String::new();
std::io::stdin()
.read_to_string(&mut prompt)
.context("read prompt from stdin")?;
let prompt = prompt.replace("\r\n", "\n").replace('\r', "\n");
if prompt.trim().is_empty() {
bail!("no prompt provided via stdin");
}
prompt
} else {
args.prompt.join(" ")
};
let config = new_config(args.model, arg0_paths)?;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
let local_runtime_paths = ExecServerRuntimePaths::from_optional_paths(
config.codex_self_exe.clone(),
config.codex_linux_sandbox_exe.clone(),
)?;
let thread_store = Arc::new(LocalThreadStore::new(RolloutConfig::from_view(&config)));
let environment_manager =
Arc::new(EnvironmentManager::new(EnvironmentManagerArgs::new(local_runtime_paths)).await);
let thread_manager = ThreadManager::new(
&config,
auth_manager,
SessionSource::Exec,
CollaborationModesConfig {
default_mode_request_user_input: config
.features
.enabled(Feature::DefaultModeRequestUserInput),
},
environment_manager,
/*analytics_events_client*/ None,
);
let NewThread {
thread_id, thread, ..
} = thread_manager
.start_thread(config, thread_store)
.await
.context("start Codex thread")?;
let turn_output = run_turn(&thread, prompt).await;
let shutdown_result = thread.shutdown_and_wait().await;
let _ = thread_manager.remove_thread(&thread_id).await;
let output = turn_output?;
shutdown_result.context("shut down Codex thread")?;
let mut stdout = std::io::stdout().lock();
stdout.write_all(output.as_bytes())?;
if !output.ends_with('\n') {
stdout.write_all(b"\n")?;
}
Ok(())
}
fn new_config(model: Option<String>, arg0_paths: Arg0DispatchPaths) -> anyhow::Result<Config> {
let codex_home = find_codex_home().context("find Codex home")?;
let cwd = AbsolutePathBuf::current_dir().context("resolve current directory")?;
let model_provider_id = OPENAI_PROVIDER_ID.to_string();
let model_providers = built_in_model_providers(/*openai_base_url*/ None);
let model_provider = model_providers
.get(&model_provider_id)
.context("OpenAI model provider should be available")?
.clone();
Ok(Config {
config_layer_stack: ConfigLayerStack::default(),
startup_warnings: Vec::new(),
model,
service_tier: None,
review_model: None,
model_context_window: None,
model_auto_compact_token_limit: None,
model_provider_id,
model_provider,
personality: None,
permissions: Permissions {
approval_policy: Constrained::allow_any(AskForApproval::Never),
permission_profile: Constrained::allow_any(PermissionProfile::default()),
network: None,
allow_login_shell: true,
shell_environment_policy: ShellEnvironmentPolicy::default(),
windows_sandbox_mode: None,
windows_sandbox_private_desktop: true,
},
approvals_reviewer: ApprovalsReviewer::User,
enforce_residency: Constrained::allow_any(/*initial_value*/ None),
hide_agent_reasoning: false,
show_raw_agent_reasoning: false,
user_instructions: None,
base_instructions: None,
developer_instructions: None,
guardian_policy_config: None,
include_permissions_instructions: false,
include_apps_instructions: false,
include_skill_instructions: false,
include_environment_context: false,
compact_prompt: None,
commit_attribution: None,
notify: None,
tui_notifications: TuiNotificationSettings::default(),
animations: true,
show_tooltips: true,
model_availability_nux: ModelAvailabilityNuxConfig::default(),
tui_alternate_screen: AltScreenMode::Auto,
tui_status_line: None,
tui_terminal_title: None,
tui_theme: None,
terminal_resize_reflow: TerminalResizeReflowConfig::default(),
tui_keymap: TuiKeymap::default(),
cwd,
cli_auth_credentials_store_mode: AuthCredentialsStoreMode::File,
mcp_servers: Constrained::allow_any(HashMap::new()),
mcp_oauth_credentials_store_mode: OAuthCredentialsStoreMode::File,
mcp_oauth_callback_port: None,
mcp_oauth_callback_url: None,
model_providers,
project_doc_max_bytes: 32 * 1024,
project_doc_fallback_filenames: Vec::new(),
tool_output_token_limit: None,
agent_max_threads: Some(6),
agent_job_max_runtime_seconds: None,
agent_interrupt_message_enabled: false,
agent_max_depth: 1,
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
sqlite_home: codex_home.to_path_buf(),
log_dir: codex_home.join("log").to_path_buf(),
codex_home,
history: History::default(),
ephemeral: true,
file_opener: UriBasedFileOpener::VsCode,
codex_self_exe: arg0_paths.codex_self_exe,
codex_linux_sandbox_exe: arg0_paths.codex_linux_sandbox_exe,
main_execve_wrapper_exe: arg0_paths.main_execve_wrapper_exe,
zsh_path: None,
model_reasoning_effort: None,
plan_mode_reasoning_effort: None,
model_reasoning_summary: None,
model_supports_reasoning_summaries: None,
model_catalog: None,
model_verbosity: None,
chatgpt_base_url: "https://chatgpt.com/backend-api/".to_string(),
realtime_audio: RealtimeAudioConfig::default(),
experimental_realtime_ws_base_url: None,
experimental_realtime_ws_model: None,
realtime: RealtimeConfig::default(),
experimental_realtime_ws_backend_prompt: None,
experimental_realtime_ws_startup_context: None,
experimental_realtime_start_instructions: None,
experimental_thread_config_endpoint: None,
experimental_thread_store: ThreadStoreConfig::Local,
forced_chatgpt_workspace_id: None,
forced_login_method: None,
include_apply_patch_tool: false,
web_search_mode: Constrained::allow_any(WebSearchMode::Disabled),
web_search_config: None,
use_experimental_unified_exec_tool: false,
background_terminal_max_timeout: 300_000,
ghost_snapshot: GhostSnapshotConfig::default(),
multi_agent_v2: MultiAgentV2Config::default(),
features: Default::default(),
suppress_unstable_features_warning: false,
active_profile: None,
active_project: ProjectConfig { trust_level: None },
windows_wsl_setup_acknowledged: false,
notices: Notice::default(),
check_for_update_on_startup: false,
disable_paste_burst: false,
analytics_enabled: Some(false),
feedback_enabled: false,
tool_suggest: ToolSuggestConfig::default(),
otel: OtelConfig::default(),
})
}
async fn run_turn(thread: &CodexThread, prompt: String) -> anyhow::Result<String> {
thread
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: prompt,
text_elements: Vec::new(),
}],
environments: None,
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await
.context("submit user input")?;
let mut last_agent_message = String::new();
loop {
let event = thread.next_event().await.context("read Codex event")?;
match event.msg {
EventMsg::TurnComplete(event) => {
return Ok(event.last_agent_message.unwrap_or(last_agent_message));
}
EventMsg::AgentMessage(event) => {
last_agent_message = event.message;
}
EventMsg::Error(event) => {
bail!(event.message);
}
EventMsg::TurnAborted(_) => {
bail!("turn aborted");
}
EventMsg::ExecApprovalRequest(_) => {
bail!("turn requested exec approval");
}
EventMsg::ApplyPatchApprovalRequest(_) => {
bail!("turn requested patch approval");
}
EventMsg::RequestPermissions(_) => {
bail!("turn requested permissions");
}
EventMsg::RequestUserInput(_) => {
bail!("turn requested user input");
}
EventMsg::DynamicToolCallRequest(_) => {
bail!("turn requested a dynamic tool call");
}
_ => {}
}
}
}