Support multiple managed environments

codex-rs/app-server-client/src/lib.rs

@@ -46,6 +46,7 @@ use codex_core::config::Config;
 use codex_core::config_loader::CloudRequirementsLoader;
 use codex_core::config_loader::LoaderOverrides;
 pub use codex_exec_server::EnvironmentManager;
+pub use codex_exec_server::EnvironmentManagerArgs;
 pub use codex_exec_server::ExecServerRuntimePaths;
 use codex_feedback::CodexFeedback;
 use codex_protocol::protocol::SessionSource;
@@ -968,7 +969,7 @@ mod tests {
             cloud_requirements: CloudRequirementsLoader::default(),
             feedback: CodexFeedback::new(),
             log_db: None,
-            environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
+            environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
             config_warnings: Vec::new(),
             session_source,
             enable_codex_api_key_env: false,
@@ -1969,9 +1970,14 @@ mod tests {
     #[tokio::test]
     async fn runtime_start_args_forward_environment_manager() {
         let config = Arc::new(build_test_config().await);
-        let environment_manager = Arc::new(EnvironmentManager::new(Some(
-            "ws://127.0.0.1:8765".to_string(),
-        )));
+        let environment_manager = Arc::new(EnvironmentManager::new(EnvironmentManagerArgs {
+            exec_server_url: Some("ws://127.0.0.1:8765".to_string()),
+            local_runtime_paths: ExecServerRuntimePaths::new(
+                std::env::current_exe().expect("current exe"),
+                /*codex_linux_sandbox_exe*/ None,
+            )
+            .expect("runtime paths"),
+        }));
 
         let runtime_args = InProcessClientStartArgs {
             arg0_paths: Arg0DispatchPaths::default(),
@@ -1998,7 +2004,13 @@ mod tests {
             &runtime_args.environment_manager,
             &environment_manager
         ));
-        assert!(runtime_args.environment_manager.is_remote());
+        assert!(
+            runtime_args
+                .environment_manager
+                .default_environment()
+                .expect("default environment")
+                .is_remote()
+        );
     }
 
     #[tokio::test]

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -3497,9 +3497,7 @@ mod tests {
                 CodexAuth::create_dummy_chatgpt_auth_for_testing(),
                 config.model_provider.clone(),
                 config.codex_home.to_path_buf(),
-                Arc::new(codex_exec_server::EnvironmentManager::new(
-                    /*exec_server_url*/ None,
-                )),
+                Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
             ),
         );
         let codex_core::NewThread {

codex-rs/app-server/src/codex_message_processor.rs

@@ -260,6 +260,7 @@ use codex_core_plugins::loader::load_plugin_mcp_servers;
 use codex_core_plugins::manifest::PluginManifestInterface;
 use codex_core_plugins::marketplace::MarketplaceError;
 use codex_core_plugins::marketplace::MarketplacePluginSource;
+use codex_exec_server::EnvironmentManager;
 use codex_exec_server::LOCAL_FS;
 use codex_features::FEATURES;
 use codex_features::Feature;
@@ -5677,27 +5678,17 @@ impl CodexMessageProcessor {
             .to_mcp_config(self.thread_manager.plugins_manager().as_ref())
             .await;
         let auth = self.auth_manager.auth().await;
-        let runtime_environment = match self.thread_manager.environment_manager().current().await {
-            Ok(Some(environment)) => {
+        let environment_manager = self.thread_manager.environment_manager();
+        let runtime_environment = match environment_manager.default_environment() {
+            Some(environment) => {
                 // Status listing has no turn cwd. This fallback is used only
                 // by executor-backed stdio MCPs whose config omits `cwd`.
                 McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
             }
-            Ok(None) => McpRuntimeEnvironment::new(
-                Arc::new(codex_exec_server::Environment::default()),
+            None => McpRuntimeEnvironment::new(
+                environment_manager.local_environment(),
                 config.cwd.to_path_buf(),
             ),
-            Err(err) => {
-                // TODO(aibrahim): Investigate degrading MCP status listing when
-                // executor environment creation fails.
-                let error = JSONRPCErrorError {
-                    code: INTERNAL_ERROR_CODE,
-                    message: format!("failed to create environment: {err}"),
-                    data: None,
-                };
-                self.outgoing.send_error(request, error).await;
-                return;
-            }
         };
 
         tokio::spawn(async move {
@@ -5856,25 +5847,14 @@ impl CodexMessageProcessor {
             .to_mcp_config(self.thread_manager.plugins_manager().as_ref())
             .await;
         let auth = self.auth_manager.auth().await;
-        let runtime_environment = match self.thread_manager.environment_manager().current().await {
-            Ok(Some(environment)) => {
-                // Resource reads without a thread have no turn cwd. This fallback
-                // is used only by executor-backed stdio MCPs whose config omits `cwd`.
-                McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
-            }
-            Ok(None) => McpRuntimeEnvironment::new(
-                Arc::new(codex_exec_server::Environment::default()),
-                config.cwd.to_path_buf(),
-            ),
-            Err(err) => {
-                let error = JSONRPCErrorError {
-                    code: INTERNAL_ERROR_CODE,
-                    message: format!("failed to create environment: {err}"),
-                    data: None,
-                };
-                self.outgoing.send_error(request_id, error).await;
-                return;
-            }
+        let runtime_environment = {
+            let environment_manager = self.thread_manager.environment_manager();
+            let environment = environment_manager
+                .default_environment()
+                .unwrap_or_else(|| environment_manager.local_environment());
+            // Resource reads without a thread have no turn cwd. This fallback
+            // is used only by executor-backed stdio MCPs whose config omits `cwd`.
+            McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf())
         };
 
         tokio::spawn(async move {
@@ -6222,8 +6202,9 @@ impl CodexMessageProcessor {
 
         let request = request_id.clone();
         let outgoing = Arc::clone(&self.outgoing);
+        let environment_manager = self.thread_manager.environment_manager();
         tokio::spawn(async move {
-            Self::apps_list_task(outgoing, request, params, config).await;
+            Self::apps_list_task(outgoing, request, params, config, environment_manager).await;
         });
     }
 
@@ -6232,6 +6213,7 @@ impl CodexMessageProcessor {
         request_id: ConnectionRequestId,
         params: AppsListParams,
         config: Config,
+        environment_manager: Arc<EnvironmentManager>,
     ) {
         let AppsListParams {
             cursor,
@@ -6266,12 +6248,15 @@ impl CodexMessageProcessor {
         let accessible_config = config.clone();
         let accessible_tx = tx.clone();
         tokio::spawn(async move {
-            let result = connectors::list_accessible_connectors_from_mcp_tools_with_options(
-                &accessible_config,
-                force_refetch,
-            )
-            .await
-            .map_err(|err| format!("failed to load accessible apps: {err}"));
+            let result =
+                connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager(
+                    &accessible_config,
+                    force_refetch,
+                    &environment_manager,
+                )
+                .await
+                .map(|status| status.connectors)
+                .map_err(|err| format!("failed to load accessible apps: {err}"));
             let _ = accessible_tx.send(AppListLoadResult::Accessible(result));
         });
 
@@ -6461,23 +6446,11 @@ impl CodexMessageProcessor {
         };
         let skills_manager = self.thread_manager.skills_manager();
         let plugins_manager = self.thread_manager.plugins_manager();
-        let fs = match self.thread_manager.environment_manager().current().await {
-            Ok(Some(environment)) => Some(environment.get_filesystem()),
-            Ok(None) => None,
-            Err(err) => {
-                self.outgoing
-                    .send_error(
-                        request_id,
-                        JSONRPCErrorError {
-                            code: INTERNAL_ERROR_CODE,
-                            message: format!("failed to create environment: {err}"),
-                            data: None,
-                        },
-                    )
-                    .await;
-                return;
-            }
-        };
+        let fs = self
+            .thread_manager
+            .environment_manager()
+            .default_environment()
+            .map(|environment| environment.get_filesystem());
         let mut data = Vec::new();
         for cwd in cwds {
             let extra_roots = extra_roots_by_cwd
@@ -6793,8 +6766,13 @@ impl CodexMessageProcessor {
                 return;
             }
         };
-        let app_summaries =
-            plugin_app_helpers::load_plugin_app_summaries(&config, &outcome.plugin.apps).await;
+        let environment_manager = self.thread_manager.environment_manager();
+        let app_summaries = plugin_app_helpers::load_plugin_app_summaries(
+            &config,
+            &outcome.plugin.apps,
+            &environment_manager,
+        )
+        .await;
         let visible_skills = outcome
             .plugin
             .skills
@@ -6971,10 +6949,11 @@ impl CodexMessageProcessor {
                     ) {
                     Vec::new()
                 } else {
+                    let environment_manager = self.thread_manager.environment_manager();
                     let (all_connectors_result, accessible_connectors_result) = tokio::join!(
                         connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true),
-                        connectors::list_accessible_connectors_from_mcp_tools_with_options_and_status(
-                            &config, /*force_refetch*/ true
+                        connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager(
+                            &config, /*force_refetch*/ true, &environment_manager
                         ),
                     );
 

codex-rs/app-server/src/codex_message_processor/plugin_app_helpers.rs

@@ -5,11 +5,13 @@ use codex_app_server_protocol::AppSummary;
 use codex_chatgpt::connectors;
 use codex_core::config::Config;
 use codex_core::plugins::AppConnectorId;
+use codex_exec_server::EnvironmentManager;
 use tracing::warn;
 
 pub(super) async fn load_plugin_app_summaries(
     config: &Config,
     plugin_apps: &[AppConnectorId],
+    environment_manager: &EnvironmentManager,
 ) -> Vec<AppSummary> {
     if plugin_apps.is_empty() {
         return Vec::new();
@@ -29,8 +31,10 @@ pub(super) async fn load_plugin_app_summaries(
     let plugin_connectors = connectors::connectors_for_plugin_apps(connectors, plugin_apps);
 
     let accessible_connectors =
-        match connectors::list_accessible_connectors_from_mcp_tools_with_options_and_status(
-            config, /*force_refetch*/ false,
+        match connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager(
+            config,
+            /*force_refetch*/ false,
+            environment_manager,
         )
         .await
         {

codex-rs/app-server/src/fs_api.rs

@@ -20,7 +20,6 @@ use codex_app_server_protocol::FsWriteFileResponse;
 use codex_app_server_protocol::JSONRPCErrorError;
 use codex_exec_server::CopyOptions;
 use codex_exec_server::CreateDirectoryOptions;
-use codex_exec_server::Environment;
 use codex_exec_server::ExecutorFileSystem;
 use codex_exec_server::RemoveOptions;
 use std::io;
@@ -31,15 +30,11 @@ pub(crate) struct FsApi {
     file_system: Arc<dyn ExecutorFileSystem>,
 }
 
-impl Default for FsApi {
-    fn default() -> Self {
-        Self {
-            file_system: Environment::default().get_filesystem(),
-        }
+impl FsApi {
+    pub(crate) fn new(file_system: Arc<dyn ExecutorFileSystem>) -> Self {
+        Self { file_system }
     }
-}
 
-impl FsApi {
     pub(crate) async fn read_file(
         &self,
         params: FsReadFileParams,

codex-rs/app-server/src/in_process.rs

@@ -750,7 +750,7 @@ mod tests {
             thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
             feedback: CodexFeedback::new(),
             log_db: None,
-            environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
+            environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
             config_warnings: Vec::new(),
             session_source,
             enable_codex_api_key_env: false,

codex-rs/app-server/src/lib.rs

@@ -6,6 +6,7 @@ use codex_config::ThreadConfigLoader;
 use codex_core::config::Config;
 use codex_core::config_loader::ConfigLayerStackOrdering;
 use codex_core::config_loader::LoaderOverrides;
+use codex_exec_server::EnvironmentManagerArgs;
 use codex_features::Feature;
 use codex_login::AuthManager;
 use codex_utils_cli::CliConfigOverrides;
@@ -360,7 +361,7 @@ pub async fn run_main_with_transport(
     session_source: SessionSource,
     auth: AppServerWebsocketAuthSettings,
 ) -> IoResult<()> {
-    let environment_manager = Arc::new(EnvironmentManager::from_env_with_runtime_paths(Some(
+    let environment_manager = Arc::new(EnvironmentManager::new(EnvironmentManagerArgs::from_env(
         ExecServerRuntimePaths::from_optional_paths(
             arg0_paths.codex_self_exe.clone(),
             arg0_paths.codex_linux_sandbox_exe.clone(),

codex-rs/app-server/src/message_processor.rs

@@ -328,7 +328,12 @@ impl MessageProcessor {
         let device_key_api = DeviceKeyApi::default();
         let external_agent_config_api =
             ExternalAgentConfigApi::new(config.codex_home.to_path_buf());
-        let fs_api = FsApi::default();
+        let fs_api = FsApi::new(
+            thread_manager
+                .environment_manager()
+                .local_environment()
+                .get_filesystem(),
+        );
         let fs_watch_manager = FsWatchManager::new(outgoing.clone());
 
         Self {
@@ -1079,11 +1084,14 @@ impl MessageProcessor {
         }
 
         let outgoing = Arc::clone(&self.outgoing);
+        let environment_manager = self.thread_manager.environment_manager();
         tokio::spawn(async move {
             let (all_connectors_result, accessible_connectors_result) = tokio::join!(
                 connectors::list_all_connectors_with_options(&config, /*force_refetch*/ true),
-                connectors::list_accessible_connectors_from_mcp_tools_with_options(
-                    &config, /*force_refetch*/ true,
+                connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager(
+                    &config,
+                    /*force_refetch*/ true,
+                    &environment_manager,
                 ),
             );
             let all_connectors = match all_connectors_result {
@@ -1096,7 +1104,7 @@ impl MessageProcessor {
                 }
             };
             let accessible_connectors = match accessible_connectors_result {
-                Ok(connectors) => connectors,
+                Ok(status) => status.connectors,
                 Err(err) => {
                     tracing::warn!(
                         "failed to force-refresh accessible apps after experimental feature enablement: {err:#}"

codex-rs/app-server/src/message_processor/tracing_tests.rs

@@ -279,7 +279,7 @@ fn build_test_processor(
         arg0_paths: Arg0DispatchPaths::default(),
         config,
         config_manager,
-        environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
+        environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
         feedback: CodexFeedback::new(),
         log_db: None,
         config_warnings: Vec::new(),

codex-rs/app-server/tests/suite/v2/mcp_resource.rs

@@ -204,7 +204,7 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
         thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
         feedback: CodexFeedback::new(),
         log_db: None,
-        environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),
+        environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
         config_warnings: Vec::new(),
         session_source: SessionSource::Cli,
         enable_codex_api_key_env: false,

codex-rs/chatgpt/src/connectors.rs

@@ -13,6 +13,7 @@ use codex_connectors::merge::merge_connectors;
 use codex_connectors::merge::merge_plugin_connectors;
 use codex_core::config::Config;
 pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools;
+pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_environment_manager;
 pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_options;
 pub use codex_core::connectors::list_accessible_connectors_from_mcp_tools_with_options_and_status;
 pub use codex_core::connectors::list_cached_accessible_connectors_from_mcp_tools;