Thread environment selection through core spawn

codex-rs/core/src/codex.rs

@@ -427,6 +427,7 @@ pub(crate) struct CodexSpawnArgs {
     pub(crate) auth_manager: Arc<AuthManager>,
     pub(crate) models_manager: Arc<ModelsManager>,
     pub(crate) environment_manager: Arc<EnvironmentManager>,
+    pub(crate) environment_id: Option<String>,
     pub(crate) skills_manager: Arc<SkillsManager>,
     pub(crate) plugins_manager: Arc<PluginsManager>,
     pub(crate) mcp_manager: Arc<McpManager>,
@@ -480,6 +481,7 @@ impl Codex {
             auth_manager,
             models_manager,
             environment_manager,
+            environment_id,
             skills_manager,
             plugins_manager,
             mcp_manager,
@@ -500,7 +502,7 @@ impl Codex {
         let (tx_event, rx_event) = async_channel::unbounded();
 
         let environment = environment_manager
-            .current()
+            .environment(environment_id.as_deref())
             .await
             .map_err(|err| CodexErr::Fatal(format!("failed to create environment: {err}")))?;
         let fs = environment

codex-rs/core/src/codex_delegate.rs

@@ -82,6 +82,7 @@ pub(crate) async fn run_codex_thread_interactive(
         environment_manager: Arc::new(EnvironmentManager::from_environment(
             parent_ctx.environment.as_deref(),
         )),
+        environment_id: None,
         skills_manager: Arc::clone(&parent_session.services.skills_manager),
         plugins_manager: Arc::clone(&parent_session.services.plugins_manager),
         mcp_manager: Arc::clone(&parent_session.services.mcp_manager),

codex-rs/core/src/codex_tests_guardian.rs

@@ -435,6 +435,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
         auth_manager,
         models_manager,
         environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
+        environment_id: None,
         skills_manager,
         plugins_manager,
         mcp_manager,

codex-rs/core/src/thread_manager.rs

@@ -471,6 +471,7 @@ impl ThreadManager {
             config,
             Vec::new(),
             /*persist_extended_history*/ false,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -480,6 +481,7 @@ impl ThreadManager {
         config: Config,
         dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
         persist_extended_history: bool,
+        environment_id: Option<String>,
     ) -> CodexResult<NewThread> {
         Box::pin(self.start_thread_with_tools_and_service_name(
             config,
@@ -488,6 +490,7 @@ impl ThreadManager {
             persist_extended_history,
             /*metrics_service_name*/ None,
             /*parent_trace*/ None,
+            environment_id,
         ))
         .await
     }
@@ -500,6 +503,7 @@ impl ThreadManager {
         persist_extended_history: bool,
         metrics_service_name: Option<String>,
         parent_trace: Option<W3cTraceContext>,
+        environment_id: Option<String>,
     ) -> CodexResult<NewThread> {
         Box::pin(self.state.spawn_thread(
             config,
@@ -511,6 +515,7 @@ impl ThreadManager {
             metrics_service_name,
             parent_trace,
             /*user_shell_override*/ None,
+            environment_id,
         ))
         .await
     }
@@ -551,6 +556,7 @@ impl ThreadManager {
             /*metrics_service_name*/ None,
             parent_trace,
             /*user_shell_override*/ None,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -570,6 +576,7 @@ impl ThreadManager {
             /*metrics_service_name*/ None,
             /*parent_trace*/ None,
             /*user_shell_override*/ Some(user_shell_override),
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -592,6 +599,7 @@ impl ThreadManager {
             /*metrics_service_name*/ None,
             /*parent_trace*/ None,
             /*user_shell_override*/ Some(user_shell_override),
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -700,6 +708,7 @@ impl ThreadManager {
             /*metrics_service_name*/ None,
             parent_trace,
             /*user_shell_override*/ None,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -801,6 +810,7 @@ impl ThreadManagerState {
             inherited_exec_policy,
             /*parent_trace*/ None,
             /*user_shell_override*/ None,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -828,6 +838,7 @@ impl ThreadManagerState {
             inherited_exec_policy,
             /*parent_trace*/ None,
             /*user_shell_override*/ None,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -856,6 +867,7 @@ impl ThreadManagerState {
             inherited_exec_policy,
             /*parent_trace*/ None,
             /*user_shell_override*/ None,
+            /*environment_id*/ None,
         ))
         .await
     }
@@ -873,6 +885,7 @@ impl ThreadManagerState {
         metrics_service_name: Option<String>,
         parent_trace: Option<W3cTraceContext>,
         user_shell_override: Option<crate::shell::Shell>,
+        environment_id: Option<String>,
     ) -> CodexResult<NewThread> {
         Box::pin(self.spawn_thread_with_source(
             config,
@@ -887,6 +900,7 @@ impl ThreadManagerState {
             /*inherited_exec_policy*/ None,
             parent_trace,
             user_shell_override,
+            environment_id,
         ))
         .await
     }
@@ -906,10 +920,11 @@ impl ThreadManagerState {
         inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
         parent_trace: Option<W3cTraceContext>,
         user_shell_override: Option<crate::shell::Shell>,
+        environment_id: Option<String>,
     ) -> CodexResult<NewThread> {
         let environment = self
             .environment_manager
-            .current()
+            .environment(environment_id.as_deref())
             .await
             .map_err(|err| CodexErr::Fatal(format!("failed to create environment: {err}")))?;
         let watch_registration = match environment.as_ref() {
@@ -932,6 +947,7 @@ impl ThreadManagerState {
             auth_manager,
             models_manager: Arc::clone(&self.models_manager),
             environment_manager: Arc::clone(&self.environment_manager),
+            environment_id,
             skills_manager: Arc::clone(&self.skills_manager),
             plugins_manager: Arc::clone(&self.plugins_manager),
             mcp_manager: Arc::clone(&self.mcp_manager),

codex-rs/core/tests/common/test_codex.rs

@@ -202,6 +202,8 @@ pub struct TestCodexBuilder {
     workspace_setups: Vec<Box<WorkspaceSetup>>,
     home: Option<Arc<TempDir>>,
     user_shell_override: Option<Shell>,
+    environment_manager_override: Option<Arc<codex_exec_server::EnvironmentManager>>,
+    thread_environment_id: Option<String>,
 }
 
 impl TestCodexBuilder {
@@ -253,6 +255,19 @@ impl TestCodexBuilder {
         self
     }
 
+    pub fn with_environment_manager(
+        mut self,
+        environment_manager: Arc<codex_exec_server::EnvironmentManager>,
+    ) -> Self {
+        self.environment_manager_override = Some(environment_manager);
+        self
+    }
+
+    pub fn with_thread_environment_id(mut self, environment_id: impl Into<String>) -> Self {
+        self.thread_environment_id = Some(environment_id.into());
+        self
+    }
+
     pub fn with_windows_cmd_shell(self) -> Self {
         if cfg!(windows) {
             self.with_user_shell(get_shell_by_model_provided_path(&PathBuf::from("cmd.exe")))
@@ -348,9 +363,17 @@ impl TestCodexBuilder {
         let (config, fallback_cwd) = self
             .prepare_config(base_url, &home, test_env.cwd().clone())
             .await?;
-        let environment_manager = Arc::new(codex_exec_server::EnvironmentManager::new(
-            test_env.exec_server_url().map(str::to_owned),
-        ));
+        let environment_manager = self
+            .environment_manager_override
+            .clone()
+            .unwrap_or_else(|| {
+                Arc::new(codex_exec_server::EnvironmentManager::new(
+                    test_env.exec_server_url().map(str::to_owned),
+                ))
+            });
+        let selected_environment = environment_manager
+            .environment(self.thread_environment_id.as_deref())
+            .await?;
         let file_system = test_env.environment().get_filesystem();
         let mut workspace_setups = vec![];
         swap(&mut self.workspace_setups, &mut workspace_setups);
@@ -365,6 +388,7 @@ impl TestCodexBuilder {
             resume_from,
             test_env,
             environment_manager,
+            selected_environment,
         ))
         .await
     }
@@ -377,6 +401,7 @@ impl TestCodexBuilder {
         resume_from: Option<PathBuf>,
         test_env: TestEnv,
         environment_manager: Arc<codex_exec_server::EnvironmentManager>,
+        selected_environment: Option<Arc<codex_exec_server::Environment>>,
     ) -> anyhow::Result<TestCodex> {
         let auth = self.auth.clone();
         let thread_manager = if config.model_catalog.is_some() {
@@ -398,8 +423,15 @@ impl TestCodexBuilder {
         };
         let thread_manager = Arc::new(thread_manager);
         let user_shell_override = self.user_shell_override.clone();
+        let thread_environment_id = self.thread_environment_id.clone();
 
         let new_conversation = match (resume_from, user_shell_override) {
+            (Some(_), _) if thread_environment_id.is_some() => {
+                anyhow::bail!("test harness does not support resuming with thread_environment_id")
+            }
+            (_, Some(_)) if thread_environment_id.is_some() => anyhow::bail!(
+                "test harness does not support user_shell_override with thread_environment_id"
+            ),
             (Some(path), Some(user_shell_override)) => {
                 let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
                 Box::pin(
@@ -433,7 +465,15 @@ impl TestCodexBuilder {
                 )
                 .await?
             }
-            (None, None) => Box::pin(thread_manager.start_thread(config.clone())).await?,
+            (None, None) => {
+                Box::pin(thread_manager.start_thread_with_tools(
+                    config.clone(),
+                    Vec::new(),
+                    /*persist_extended_history*/ false,
+                    thread_environment_id,
+                ))
+                .await?
+            }
         };
 
         Ok(TestCodex {
@@ -443,6 +483,7 @@ impl TestCodexBuilder {
             codex: new_conversation.thread,
             session_configured: new_conversation.session_configured,
             thread_manager,
+            selected_environment,
             _test_env: test_env,
         })
     }
@@ -533,6 +574,7 @@ pub struct TestCodex {
     pub session_configured: SessionConfiguredEvent,
     pub config: Config,
     pub thread_manager: Arc<ThreadManager>,
+    selected_environment: Option<Arc<codex_exec_server::Environment>>,
     _test_env: TestEnv,
 }
 
@@ -553,6 +595,10 @@ impl TestCodex {
         &self._test_env
     }
 
+    pub fn selected_environment(&self) -> Option<&codex_exec_server::Environment> {
+        self.selected_environment.as_deref()
+    }
+
     pub fn fs(&self) -> Arc<dyn ExecutorFileSystem> {
         self._test_env.environment().get_filesystem()
     }
@@ -879,6 +925,8 @@ pub fn test_codex() -> TestCodexBuilder {
         workspace_setups: vec![],
         home: None,
         user_shell_override: None,
+        environment_manager_override: None,
+        thread_environment_id: None,
     }
 }
 

codex-rs/core/tests/suite/code_mode.rs

@@ -2364,6 +2364,7 @@ async fn code_mode_can_call_hidden_dynamic_tools() -> Result<()> {
                 defer_loading: true,
             }],
             /*persist_extended_history*/ false,
+            /*environment_id*/ None,
         )
         .await?;
     let mut test = base_test;