diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 2723ad955..68ccae177 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -312,8 +312,9 @@ impl ThreadManager { } pub async fn start_thread(&self, config: Config) -> CodexResult { - self.start_thread_with_tools(config, Vec::new(), false) - .await + // 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, Vec::new(), false)).await } pub async fn start_thread_with_tools( @@ -322,12 +323,12 @@ impl ThreadManager { dynamic_tools: Vec, persist_extended_history: bool, ) -> CodexResult { - self.start_thread_with_tools_and_service_name( + Box::pin(self.start_thread_with_tools_and_service_name( config, dynamic_tools, persist_extended_history, None, - ) + )) .await } @@ -338,17 +339,16 @@ impl ThreadManager { persist_extended_history: bool, metrics_service_name: Option, ) -> CodexResult { - self.state - .spawn_thread( - config, - InitialHistory::New, - Arc::clone(&self.state.auth_manager), - self.agent_control(), - dynamic_tools, - persist_extended_history, - metrics_service_name, - ) - .await + Box::pin(self.state.spawn_thread( + config, + InitialHistory::New, + Arc::clone(&self.state.auth_manager), + self.agent_control(), + dynamic_tools, + persist_extended_history, + metrics_service_name, + )) + .await } pub async fn resume_thread_from_rollout( @@ -358,7 +358,7 @@ impl ThreadManager { auth_manager: Arc, ) -> CodexResult { let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; - self.resume_thread_with_history(config, initial_history, auth_manager, false) + Box::pin(self.resume_thread_with_history(config, initial_history, auth_manager, false)) .await } @@ -369,17 +369,16 @@ impl ThreadManager { auth_manager: Arc, persist_extended_history: bool, ) -> CodexResult { - self.state - .spawn_thread( - config, - initial_history, - auth_manager, - self.agent_control(), - Vec::new(), - persist_extended_history, - None, - ) - .await + Box::pin(self.state.spawn_thread( + config, + initial_history, + auth_manager, + self.agent_control(), + Vec::new(), + persist_extended_history, + None, + )) + .await } /// Removes the thread from the manager's internal map, though the thread is stored @@ -411,17 +410,16 @@ impl ThreadManager { ) -> CodexResult { let history = RolloutRecorder::get_rollout_history(&path).await?; let history = truncate_before_nth_user_message(history, nth_user_message); - self.state - .spawn_thread( - config, - history, - Arc::clone(&self.state.auth_manager), - self.agent_control(), - Vec::new(), - persist_extended_history, - None, - ) - .await + Box::pin(self.state.spawn_thread( + config, + history, + Arc::clone(&self.state.auth_manager), + self.agent_control(), + Vec::new(), + persist_extended_history, + None, + )) + .await } pub(crate) fn agent_control(&self) -> AgentControl { @@ -474,14 +472,14 @@ impl ThreadManagerState { config: Config, agent_control: AgentControl, ) -> CodexResult { - self.spawn_new_thread_with_source( + Box::pin(self.spawn_new_thread_with_source( config, agent_control, self.session_source.clone(), false, None, None, - ) + )) .await } @@ -494,7 +492,7 @@ impl ThreadManagerState { metrics_service_name: Option, inherited_shell_snapshot: Option>, ) -> CodexResult { - self.spawn_thread_with_source( + Box::pin(self.spawn_thread_with_source( config, InitialHistory::New, Arc::clone(&self.auth_manager), @@ -504,7 +502,7 @@ impl ThreadManagerState { persist_extended_history, metrics_service_name, inherited_shell_snapshot, - ) + )) .await } @@ -517,7 +515,7 @@ impl ThreadManagerState { inherited_shell_snapshot: Option>, ) -> CodexResult { let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; - self.spawn_thread_with_source( + Box::pin(self.spawn_thread_with_source( config, initial_history, Arc::clone(&self.auth_manager), @@ -527,7 +525,7 @@ impl ThreadManagerState { false, None, inherited_shell_snapshot, - ) + )) .await } @@ -540,7 +538,7 @@ impl ThreadManagerState { persist_extended_history: bool, inherited_shell_snapshot: Option>, ) -> CodexResult { - self.spawn_thread_with_source( + Box::pin(self.spawn_thread_with_source( config, initial_history, Arc::clone(&self.auth_manager), @@ -550,7 +548,7 @@ impl ThreadManagerState { persist_extended_history, None, inherited_shell_snapshot, - ) + )) .await } @@ -566,7 +564,7 @@ impl ThreadManagerState { persist_extended_history: bool, metrics_service_name: Option, ) -> CodexResult { - self.spawn_thread_with_source( + Box::pin(self.spawn_thread_with_source( config, initial_history, auth_manager, @@ -576,7 +574,7 @@ impl ThreadManagerState { persist_extended_history, metrics_service_name, None, - ) + )) .await } diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index a79838df0..7bd5763d1 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -105,7 +105,7 @@ impl TestCodexBuilder { Some(home) => home, None => Arc::new(TempDir::new()?), }; - self.build_with_home(server, home, None).await + Box::pin(self.build_with_home(server, home, None)).await } pub async fn build_with_streaming_server( @@ -117,8 +117,7 @@ impl TestCodexBuilder { Some(home) => home, None => Arc::new(TempDir::new()?), }; - self.build_with_home_and_base_url(format!("{base_url}/v1"), home, None) - .await + Box::pin(self.build_with_home_and_base_url(format!("{base_url}/v1"), home, None)).await } pub async fn build_with_websocket_server( @@ -139,8 +138,7 @@ impl TestCodexBuilder { .enable(Feature::ResponsesWebsockets) .expect("test config should allow feature update"); })); - self.build_with_home_and_base_url(base_url, home, None) - .await + Box::pin(self.build_with_home_and_base_url(base_url, home, None)).await } pub async fn resume( @@ -149,7 +147,7 @@ impl TestCodexBuilder { home: Arc, rollout_path: PathBuf, ) -> anyhow::Result { - self.build_with_home(server, home, Some(rollout_path)).await + Box::pin(self.build_with_home(server, home, Some(rollout_path))).await } async fn build_with_home( @@ -160,7 +158,7 @@ impl TestCodexBuilder { ) -> anyhow::Result { let base_url = format!("{}/v1", server.uri()); let (config, cwd) = self.prepare_config(base_url, &home).await?; - self.build_from_config(config, cwd, home, resume_from).await + Box::pin(self.build_from_config(config, cwd, home, resume_from)).await } async fn build_with_home_and_base_url( @@ -170,7 +168,7 @@ impl TestCodexBuilder { resume_from: Option, ) -> anyhow::Result { let (config, cwd) = self.prepare_config(base_url, &home).await?; - self.build_from_config(config, cwd, home, resume_from).await + Box::pin(self.build_from_config(config, cwd, home, resume_from)).await } async fn build_from_config( @@ -201,11 +199,14 @@ impl TestCodexBuilder { let new_conversation = match resume_from { Some(path) => { let auth_manager = codex_core::test_support::auth_manager_from_auth(auth); - thread_manager - .resume_thread_from_rollout(config.clone(), path, auth_manager) - .await? + Box::pin(thread_manager.resume_thread_from_rollout( + config.clone(), + path, + auth_manager, + )) + .await? } - None => thread_manager.start_thread(config.clone()).await?, + None => Box::pin(thread_manager.start_thread(config.clone())).await?, }; Ok(TestCodex { diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index fe536077a..be79c1170 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -157,7 +157,7 @@ async fn compact_resume_and_fork_preserve_model_history_view() { user_turn(&base, "hello world").await; compact_conversation(&base).await; user_turn(&base, "AFTER_COMPACT").await; - let base_path = fetch_conversation_path(&base).await; + let base_path = fetch_conversation_path(&base); assert!( base_path.exists(), "compact+resume test expects base path {base_path:?} to exist", @@ -165,7 +165,7 @@ async fn compact_resume_and_fork_preserve_model_history_view() { let resumed = resume_conversation(&manager, &config, base_path).await; user_turn(&resumed, "AFTER_RESUME").await; - let resumed_path = fetch_conversation_path(&resumed).await; + let resumed_path = fetch_conversation_path(&resumed); assert!( resumed_path.exists(), "compact+resume test expects resumed path {resumed_path:?} to exist", @@ -292,33 +292,10 @@ async fn compact_resume_and_fork_preserve_model_history_view() { assert_eq!(requests.len(), 5); } -#[test] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] /// Scenario: after the forked branch is compacted, resuming again should reuse /// the compacted history and only append the new user message. -fn compact_resume_after_second_compaction_preserves_history() -> Result<()> { - const TEST_STACK_SIZE_BYTES: usize = 8 * 1024 * 1024; - - let handle = std::thread::Builder::new() - .name("compact_resume_after_second_compaction_preserves_history".to_string()) - .stack_size(TEST_STACK_SIZE_BYTES) - .spawn(|| -> Result<()> { - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .thread_stack_size(TEST_STACK_SIZE_BYTES) - .enable_all() - .build()?; - runtime.block_on(compact_resume_after_second_compaction_preserves_history_impl()) - })?; - - match handle.join() { - Ok(result) => result, - Err(_) => Err(anyhow::anyhow!( - "compact_resume_after_second_compaction_preserves_history thread panicked" - )), - } -} - -async fn compact_resume_after_second_compaction_preserves_history_impl() -> Result<()> { +async fn compact_resume_after_second_compaction_preserves_history() -> Result<()> { if network_disabled() { println!("Skipping test because network is disabled in this sandbox"); return Ok(()); @@ -335,7 +312,7 @@ async fn compact_resume_after_second_compaction_preserves_history_impl() -> Resu user_turn(&base, "hello world").await; compact_conversation(&base).await; user_turn(&base, "AFTER_COMPACT").await; - let base_path = fetch_conversation_path(&base).await; + let base_path = fetch_conversation_path(&base); assert!( base_path.exists(), "second compact test expects base path {base_path:?} to exist", @@ -343,7 +320,7 @@ async fn compact_resume_after_second_compaction_preserves_history_impl() -> Resu let resumed = resume_conversation(&manager, &config, base_path).await; user_turn(&resumed, "AFTER_RESUME").await; - let resumed_path = fetch_conversation_path(&resumed).await; + let resumed_path = fetch_conversation_path(&resumed); assert!( resumed_path.exists(), "second compact test expects resumed path {resumed_path:?} to exist", @@ -354,7 +331,7 @@ async fn compact_resume_after_second_compaction_preserves_history_impl() -> Resu compact_conversation(&forked).await; user_turn(&forked, "AFTER_COMPACT_2").await; - let forked_path = fetch_conversation_path(&forked).await; + let forked_path = fetch_conversation_path(&forked); assert!( forked_path.exists(), "second compact test expects forked path {forked_path:?} to exist", @@ -558,7 +535,9 @@ async fn start_test_conversation( config.model = Some(model); } }); - let test = builder.build(server).await.expect("create conversation"); + let test = Box::pin(builder.build(server)) + .await + .expect("create conversation"); (test.home, test.config, test.thread_manager, test.codex) } @@ -595,7 +574,7 @@ async fn compact_conversation(conversation: &Arc) { wait_for_event(conversation, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; } -async fn fetch_conversation_path(conversation: &Arc) -> std::path::PathBuf { +fn fetch_conversation_path(conversation: &Arc) -> std::path::PathBuf { conversation.rollout_path().expect("rollout path") } @@ -607,8 +586,7 @@ async fn resume_conversation( let auth_manager = codex_core::test_support::auth_manager_from_auth( codex_core::CodexAuth::from_api_key("dummy"), ); - manager - .resume_thread_from_rollout(config.clone(), path, auth_manager) + Box::pin(manager.resume_thread_from_rollout(config.clone(), path, auth_manager)) .await .expect("resume conversation") .thread @@ -621,8 +599,7 @@ async fn fork_thread( path: std::path::PathBuf, nth_user_message: usize, ) -> Arc { - manager - .fork_thread(nth_user_message, config.clone(), path, false) + Box::pin(manager.fork_thread(nth_user_message, config.clone(), path, false)) .await .expect("fork conversation") .thread