[codex] support executor registry remote environments (#21323)

## Summary

Support registry-backed remote executors end to end so downstream
services can resolve an executor id into an exec-server URL and make
that environment available to Codex without relying on the legacy cloud
environments flow.

## What changed

- switch remote executor registration to the executor registry bootstrap
contract
- allow named remote environments to be inserted into
`EnvironmentManager` at runtime
- add the experimental app-server RPC `environment/add` so initialized
experimental clients can register those remote environments for later
`thread/start` and `turn/start` selection

## Validation

Ran focused validation locally:
- `cargo test -p codex-exec-server environment_manager_`
- `cargo test -p codex-exec-server
register_executor_posts_with_bearer_token_header`
- `cargo test -p codex-app-server-protocol`

codex-rs/exec-server/src/environment.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::RwLock;
 
 use crate::ExecServerError;
 use crate::ExecServerRuntimePaths;
@@ -40,7 +41,7 @@ pub const CODEX_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_EXEC_SERVER_URL";
 #[derive(Debug)]
 pub struct EnvironmentManager {
     default_environment: Option<String>,
-    environments: HashMap<String, Arc<Environment>>,
+    environments: RwLock<HashMap<String, Arc<Environment>>>,
     local_environment: Arc<Environment>,
 }
 
@@ -65,10 +66,10 @@ impl EnvironmentManager {
     pub fn default_for_tests() -> Self {
         Self {
             default_environment: Some(LOCAL_ENVIRONMENT_ID.to_string()),
-            environments: HashMap::from([(
+            environments: RwLock::new(HashMap::from([(
                 LOCAL_ENVIRONMENT_ID.to_string(),
                 Arc::new(Environment::default_for_tests()),
-            )]),
+            )])),
             local_environment: Arc::new(Environment::default_for_tests()),
         }
     }
@@ -77,7 +78,7 @@ impl EnvironmentManager {
     pub fn disabled_for_tests(local_runtime_paths: ExecServerRuntimePaths) -> Self {
         Self {
             default_environment: None,
-            environments: HashMap::new(),
+            environments: RwLock::new(HashMap::new()),
             local_environment: Arc::new(Environment::local(local_runtime_paths)),
         }
     }
@@ -196,7 +197,7 @@ impl EnvironmentManager {
         };
         Ok(Self {
             default_environment,
-            environments: environment_map,
+            environments: RwLock::new(environment_map),
             local_environment,
         })
     }
@@ -218,10 +219,14 @@ impl EnvironmentManager {
         let Some(default_environment_id) = self.default_environment.as_ref() else {
             return Vec::new();
         };
-        let mut environment_ids = Vec::with_capacity(self.environments.len());
+        let environments = self
+            .environments
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut environment_ids = Vec::with_capacity(environments.len());
         environment_ids.push(default_environment_id.clone());
         environment_ids.extend(
-            self.environments
+            environments
                 .keys()
                 .filter(|environment_id| *environment_id != default_environment_id)
                 .cloned(),
@@ -236,7 +241,45 @@ impl EnvironmentManager {
 
     /// Returns a named environment instance.
     pub fn get_environment(&self, environment_id: &str) -> Option<Arc<Environment>> {
-        self.environments.get(environment_id).cloned()
+        self.environments
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(environment_id)
+            .cloned()
+    }
+
+    /// Adds or replaces a named remote environment without changing the
+    /// manager's default environment selection.
+    pub fn upsert_environment(
+        &self,
+        environment_id: String,
+        exec_server_url: String,
+    ) -> Result<(), ExecServerError> {
+        if environment_id.is_empty() {
+            return Err(ExecServerError::Protocol(
+                "environment id cannot be empty".to_string(),
+            ));
+        }
+        let (exec_server_url, disabled) = normalize_exec_server_url(Some(exec_server_url));
+        if disabled {
+            return Err(ExecServerError::Protocol(
+                "remote environment cannot use disabled exec-server url".to_string(),
+            ));
+        }
+        let Some(exec_server_url) = exec_server_url else {
+            return Err(ExecServerError::Protocol(
+                "remote environment requires an exec-server url".to_string(),
+            ));
+        };
+        let environment = Environment::remote_inner(
+            exec_server_url,
+            self.local_environment.local_runtime_paths.clone(),
+        );
+        self.environments
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(environment_id, Arc::new(environment));
+        Ok(())
     }
 }
 
@@ -717,6 +760,45 @@ mod tests {
         assert!(manager.get_environment("does-not-exist").is_none());
     }
 
+    #[tokio::test]
+    async fn environment_manager_upserts_named_remote_environment() {
+        let manager = EnvironmentManager::disabled_for_tests(test_runtime_paths());
+
+        manager
+            .upsert_environment("executor-a".to_string(), "ws://127.0.0.1:8765".to_string())
+            .expect("remote environment");
+        let first = manager
+            .get_environment("executor-a")
+            .expect("first remote environment");
+        assert!(first.is_remote());
+        assert_eq!(first.exec_server_url(), Some("ws://127.0.0.1:8765"));
+        assert_eq!(manager.default_environment_id(), None);
+
+        manager
+            .upsert_environment("executor-a".to_string(), "ws://127.0.0.1:9876".to_string())
+            .expect("updated remote environment");
+        let second = manager
+            .get_environment("executor-a")
+            .expect("second remote environment");
+        assert!(second.is_remote());
+        assert_eq!(second.exec_server_url(), Some("ws://127.0.0.1:9876"));
+        assert!(!Arc::ptr_eq(&first, &second));
+    }
+
+    #[tokio::test]
+    async fn environment_manager_rejects_empty_remote_environment_url() {
+        let manager = EnvironmentManager::disabled_for_tests(test_runtime_paths());
+
+        let err = manager
+            .upsert_environment("executor-a".to_string(), String::new())
+            .expect_err("empty URL should fail");
+
+        assert_eq!(
+            err.to_string(),
+            "exec-server protocol error: remote environment requires an exec-server url"
+        );
+    }
+
     #[tokio::test]
     async fn default_environment_has_ready_local_executor() {
         let environment = Environment::default_for_tests();