From 8356806fc93469956b4a528d29469ab74cb80280 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Wed, 29 Apr 2026 11:21:06 -0700 Subject: [PATCH] Add ThreadManager sample crate (#20141) 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. --- codex-rs/Cargo.lock | 21 ++ codex-rs/Cargo.toml | 1 + .../app-server/src/bespoke_event_handling.rs | 7 +- .../app-server/src/codex_message_processor.rs | 15 +- codex-rs/core/src/agent/control.rs | 27 +- codex-rs/core/src/agent/control_tests.rs | 15 +- codex-rs/core/src/config/managed_features.rs | 12 + codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/prompt_debug.rs | 4 +- codex-rs/core/src/session/tests.rs | 5 +- codex-rs/core/src/test_support.rs | 6 +- codex-rs/core/src/thread_manager.rs | 62 +++- codex-rs/core/src/thread_manager_tests.rs | 37 +- .../src/tools/handlers/multi_agents_tests.rs | 195 ++++++++--- codex-rs/core/tests/common/test_codex.rs | 11 +- codex-rs/core/tests/suite/client.rs | 3 +- codex-rs/core/tests/suite/code_mode.rs | 1 + codex-rs/core/tests/suite/compact_remote.rs | 1 + .../core/tests/suite/compact_resume_fork.rs | 2 + codex-rs/core/tests/suite/fork_thread.rs | 5 +- .../core/tests/suite/permissions_messages.rs | 3 +- .../core/tests/suite/realtime_conversation.rs | 1 + codex-rs/core/tests/suite/resume_warning.rs | 3 +- codex-rs/core/tests/suite/search_tool.rs | 1 + codex-rs/core/tests/suite/skills.rs | 5 +- .../tests/suite/unstable_features_warning.rs | 6 +- codex-rs/core/tests/suite/window_headers.rs | 1 + codex-rs/mcp-server/src/codex_tool_runner.rs | 6 +- codex-rs/memories/write/src/runtime.rs | 2 + codex-rs/thread-manager-sample/BUILD.bazel | 6 + codex-rs/thread-manager-sample/Cargo.toml | 25 ++ codex-rs/thread-manager-sample/README.md | 20 ++ codex-rs/thread-manager-sample/src/main.rs | 321 ++++++++++++++++++ 33 files changed, 737 insertions(+), 94 deletions(-) create mode 100644 codex-rs/thread-manager-sample/BUILD.bazel create mode 100644 codex-rs/thread-manager-sample/Cargo.toml create mode 100644 codex-rs/thread-manager-sample/README.md create mode 100644 codex-rs/thread-manager-sample/src/main.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f219096e3..a37e13bbf 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index a9d6ea6c2..217c179be 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -97,6 +97,7 @@ members = [ "state", "terminal-detection", "test-binary-support", + "thread-manager-sample", "thread-store", "uds", "codex-experimental-api-macros", diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index d67efed16..f051aca7c 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -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); diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 77fe5f9ec..4a7d61ae4 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -691,7 +691,7 @@ pub(crate) struct CodexMessageProcessorArgs { pub(crate) log_db: Option, } -fn configured_thread_store(config: &Config) -> Arc { +fn thread_store_from_config(config: &Config) -> Arc { 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, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 339b99787..09a9d4c14 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -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); diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 8ce7fa2ea..6a86000f9 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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) { 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; diff --git a/codex-rs/core/src/config/managed_features.rs b/codex-rs/core/src/config/managed_features.rs index fd3032920..a575b0a45 100644 --- a/codex-rs/core/src/config/managed_features.rs +++ b/codex-rs/core/src/config/managed_features.rs @@ -27,6 +27,18 @@ pub struct ManagedFeatures { pinned_features: BTreeMap, } +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, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 35d2f8410..bd0cdd885 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -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; diff --git a/codex-rs/core/src/prompt_debug.rs b/codex-rs/core/src/prompt_debug.rs index cc1203c2d..46431c59d 100644 --- a/codex-rs/core/src/prompt_debug.rs +++ b/codex-rs/core/src/prompt_debug.rs @@ -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; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 3062f8f6f..aa4b0147d 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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, diff --git a/codex-rs/core/src/test_support.rs b/codex-rs/core/src/test_support.rs index 0cb0e9d0c..6c2f29d9c 100644 --- a/codex-rs/core/src/test_support.rs +++ b/codex-rs/core/src/test_support.rs @@ -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, user_shell_override: crate::shell::Shell, ) -> codex_protocol::error::Result { 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, rollout_path: PathBuf, auth_manager: Arc, 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, diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 5d463f934..2a83eb451 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -213,6 +213,7 @@ pub struct ThreadManager { pub struct StartThreadOptions { pub config: Config, + pub thread_store: Arc, pub initial_history: InitialHistory, pub session_source: Option, pub dynamic_tools: Vec, @@ -222,6 +223,16 @@ pub struct StartThreadOptions { pub environments: Vec, } +pub(crate) struct ResumeThreadFromRolloutOptions { + pub(crate) config: Config, + pub(crate) thread_store: Arc, + pub(crate) rollout_path: PathBuf, + pub(crate) agent_control: AgentControl, + pub(crate) session_source: SessionSource, + pub(crate) inherited_shell_snapshot: Option>, + pub(crate) inherited_exec_policy: Option>, +} + /// 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 { +pub fn thread_store_from_config(config: &Config) -> Arc { 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 { + pub async fn start_thread( + &self, + config: Config, + thread_store: Arc, + ) -> CodexResult { // 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, dynamic_tools: Vec, persist_extended_history: bool, ) -> CodexResult { @@ -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 { - 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, rollout_path: PathBuf, auth_manager: Arc, parent_trace: Option, @@ -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, initial_history: InitialHistory, auth_manager: Arc, persist_extended_history: bool, parent_trace: Option, ) -> CodexResult { - 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, user_shell_override: crate::shell::Shell, ) -> CodexResult { - 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, rollout_path: PathBuf, auth_manager: Arc, user_shell_override: crate::shell::Shell, ) -> CodexResult { 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, path: PathBuf, persist_extended_history: bool, parent_trace: Option, @@ -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, history: InitialHistory, persist_extended_history: bool, parent_trace: Option, @@ -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, history: InitialHistory, persist_extended_history: bool, parent_trace: Option, ) -> CodexResult { 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, agent_control: AgentControl, ) -> CodexResult { 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, agent_control: AgentControl, session_source: SessionSource, persist_extended_history: bool, @@ -914,7 +940,6 @@ impl ThreadManagerState { inherited_exec_policy: Option>, environments: Option>, ) -> CodexResult { - 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>, - inherited_exec_policy: Option>, + options: ResumeThreadFromRolloutOptions, ) -> CodexResult { + 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, initial_history: InitialHistory, agent_control: AgentControl, session_source: SessionSource, @@ -981,7 +1010,6 @@ impl ThreadManagerState { inherited_exec_policy: Option>, environments: Option>, ) -> CodexResult { - let thread_store = configured_thread_store(&config); let environments = environments.unwrap_or_else(|| { default_thread_environment_selections(self.environment_manager.as_ref(), &config.cwd) }); diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 54d7ea4f7..348edaa22 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -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, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 8804e6d25..e55bd4fa2 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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(); diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 60002eff4..600a58c72 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -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 { diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 67340f316..b8871cf52 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -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"); diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index af94252c0..c000469aa 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -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(), diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index 24e86881b..57579ec0a 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -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, ) diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index 6e6dcd40f..24f77fefd 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -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, diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 7648e97f8..50e0dc186 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -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(), diff --git a/codex-rs/core/tests/suite/permissions_messages.rs b/codex-rs/core/tests/suite/permissions_messages.rs index 8ec48e4b1..d64349c7f 100644 --- a/codex-rs/core/tests/suite/permissions_messages.rs +++ b/codex-rs/core/tests/suite/permissions_messages.rs @@ -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, diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index b4a7e5739..82b0c0fdd 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -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, diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index f5810956a..9b2faba36 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -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, diff --git a/codex-rs/core/tests/suite/search_tool.rs b/codex-rs/core/tests/suite/search_tool.rs index b4edf2466..5b3c540ef 100644 --- a/codex-rs/core/tests/suite/search_tool.rs +++ b/codex-rs/core/tests/suite/search_tool.rs @@ -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, ) diff --git a/codex-rs/core/tests/suite/skills.rs b/codex-rs/core/tests/suite/skills.rs index 183881013..87b5eb46f 100644 --- a/codex-rs/core/tests/suite/skills.rs +++ b/codex-rs/core/tests/suite/skills.rs @@ -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 diff --git a/codex-rs/core/tests/suite/unstable_features_warning.rs b/codex-rs/core/tests/suite/unstable_features_warning.rs index 1b1869729..973a23a46 100644 --- a/codex-rs/core/tests/suite/unstable_features_warning.rs +++ b/codex-rs/core/tests/suite/unstable_features_warning.rs @@ -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, diff --git a/codex-rs/core/tests/suite/window_headers.rs b/codex-rs/core/tests/suite/window_headers.rs index de5282183..eca8a1fce 100644 --- a/codex-rs/core/tests/suite/window_headers.rs +++ b/codex-rs/core/tests/suite/window_headers.rs @@ -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, diff --git a/codex-rs/mcp-server/src/codex_tool_runner.rs b/codex-rs/mcp-server/src/codex_tool_runner.rs index f759f8bb8..46b786eba 100644 --- a/codex-rs/mcp-server/src/codex_tool_runner.rs +++ b/codex-rs/mcp-server/src/codex_tool_runner.rs @@ -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 { diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index 737fb6787..255fb718f 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -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( diff --git a/codex-rs/thread-manager-sample/BUILD.bazel b/codex-rs/thread-manager-sample/BUILD.bazel new file mode 100644 index 000000000..5f20a3133 --- /dev/null +++ b/codex-rs/thread-manager-sample/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "thread-manager-sample", + crate_name = "codex_thread_manager_sample", +) diff --git a/codex-rs/thread-manager-sample/Cargo.toml b/codex-rs/thread-manager-sample/Cargo.toml new file mode 100644 index 000000000..52dc65e09 --- /dev/null +++ b/codex-rs/thread-manager-sample/Cargo.toml @@ -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 } diff --git a/codex-rs/thread-manager-sample/README.md b/codex-rs/thread-manager-sample/README.md new file mode 100644 index 000000000..c59b11265 --- /dev/null +++ b/codex-rs/thread-manager-sample/README.md @@ -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 +``` diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs new file mode 100644 index 000000000..e78823ec3 --- /dev/null +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -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, + + /// Prompt text. If omitted, the prompt is read from piped stdin. + #[arg(value_name = "PROMPT", num_args = 0.., trailing_var_arg = true)] + prompt: Vec, +} + +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, arg0_paths: Arg0DispatchPaths) -> anyhow::Result { + 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 { + 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"); + } + _ => {} + } + } +}