Merge 6c8b28d344 into sapling-pr-archive-bolinfest

codex-rs/core/src/thread_manager.rs

@@ -312,8 +312,9 @@ impl ThreadManager {
     }
 
     pub async fn start_thread(&self, config: Config) -> CodexResult<NewThread> {
-        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<codex_protocol::dynamic_tools::DynamicToolSpec>,
         persist_extended_history: bool,
     ) -> CodexResult<NewThread> {
-        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<String>,
     ) -> CodexResult<NewThread> {
-        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<AuthManager>,
     ) -> CodexResult<NewThread> {
         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<AuthManager>,
         persist_extended_history: bool,
     ) -> CodexResult<NewThread> {
-        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<NewThread> {
         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<NewThread> {
-        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<String>,
         inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
     ) -> CodexResult<NewThread> {
-        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<Arc<ShellSnapshot>>,
     ) -> CodexResult<NewThread> {
         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<Arc<ShellSnapshot>>,
     ) -> CodexResult<NewThread> {
-        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<String>,
     ) -> CodexResult<NewThread> {
-        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
     }
 

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<TempDir>,
         rollout_path: PathBuf,
     ) -> anyhow::Result<TestCodex> {
-        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<TestCodex> {
         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<PathBuf>,
     ) -> anyhow::Result<TestCodex> {
         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 {

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<CodexThread>) {
     wait_for_event(conversation, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
 }
 
-async fn fetch_conversation_path(conversation: &Arc<CodexThread>) -> std::path::PathBuf {
+fn fetch_conversation_path(conversation: &Arc<CodexThread>) -> 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<CodexThread> {
-    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