Honor network policy in remote control and recover remote execution (#47410)

## What changed

- Route remote-control enrollment, pairing, client management, and WebSocket connections through the authenticated HTTP client factory. Keep policy denials distinct from authentication failures and wait for policy changes before reconnecting.
- Preserve policy denials through exec-server connection failures so later connection attempts and capability discovery can recover. Keep direct executors running through temporary policy outages while rejecting permanent policy failures.
- Build cached HTTP transports on blocking workers, serialize construction, and retain successful builds after callers time out so request deadlines cover queued construction.

## Testing

Add regression coverage for policy recovery during remote startup, direct registration, accepted WebSocket reconnection, and capability discovery; queued transport construction timeouts; and remote-control WebSocket fallback with invalid custom CA files.

GitOrigin-RevId: c81db18e83de957175bb213d165281ac701089dd

codex-rs/exec-server/src/remote/direct.rs

@@ -2,6 +2,8 @@ use std::time::Instant;
 
 use codex_api::AuthError;
 use codex_api::AuthProvider;
+use codex_http_client::NetworkPolicy;
+use codex_http_client::NetworkPolicyDenied;
 use codex_http_client::Request;
 use codex_websocket_client::WebSocketConnection;
 use codex_websocket_client::WebSocketConnector;
@@ -43,6 +45,26 @@ struct DirectRegistrationResponse {
 }
 
 impl EnvironmentRegistryClient {
+    async fn register_direct_environment_when_available(
+        &self,
+        environment_id: &str,
+        policy: &NetworkPolicy,
+    ) -> Result<DirectRegistrationResponse, ExecServerError> {
+        let destination = super::endpoint_url(
+            &self.base_url,
+            &format!("/cloud/environment/{environment_id}/direct/register"),
+        );
+        let mut retry_attempt = 0;
+        loop {
+            match self.register_direct_environment(environment_id).await {
+                Err(error) if is_policy_recoverable(&error, policy, &destination) => {}
+                result => return result,
+            }
+            sleep(registry_recovery_retry_delay(environment_id, retry_attempt)).await;
+            retry_attempt = retry_attempt.saturating_add(1);
+        }
+    }
+
     #[tracing::instrument(
         name = "codex.exec_server.remote.register",
         skip_all,
@@ -122,7 +144,10 @@ pub(super) async fn run_direct_environment(
     require_tls_or_loopback(&config.base_url, "https")?;
     let mut retry_attempt = 0;
     let mut registration = client
-        .register_direct_environment(&config.environment_id)
+        .register_direct_environment_when_available(
+            &config.environment_id,
+            config.http_client_factory.network_policy(),
+        )
         .await?;
 
     loop {
@@ -155,7 +180,12 @@ pub(super) async fn run_direct_environment(
                 config.telemetry.remote_reconnect("disconnected");
             }
             Err(error)
-                if is_retryable_recovery_error(&error)
+                if (is_retryable_recovery_error(&error)
+                    || is_policy_recoverable(
+                        &error,
+                        config.http_client_factory.network_policy(),
+                        &registration.url,
+                    ))
                     && !matches!(
                         &error,
                         ExecServerError::WebSocketConnect {
@@ -180,7 +210,10 @@ pub(super) async fn run_direct_environment(
                     } if response.status() == StatusCode::CONFLICT
                 ) {
                     registration = client
-                        .register_direct_environment(&config.environment_id)
+                        .register_direct_environment_when_available(
+                            &config.environment_id,
+                            config.http_client_factory.network_policy(),
+                        )
                         .await?;
                 }
                 warn!("direct exec-server connection failed; retrying");
@@ -198,6 +231,24 @@ pub(super) async fn run_direct_environment(
     }
 }
 
+fn is_policy_recoverable(
+    error: &ExecServerError,
+    policy: &NetworkPolicy,
+    destination: &str,
+) -> bool {
+    match error.application_network_policy_denial() {
+        Some(NetworkPolicyDenied::Unavailable) => true,
+        Some(NetworkPolicyDenied::Revoked) => url::Url::parse(destination).is_ok_and(|url| {
+            matches!(
+                policy.acquire(&url),
+                Ok(_) | Err(NetworkPolicyDenied::Unavailable)
+            )
+        }),
+        Some(NetworkPolicyDenied::Destination | NetworkPolicyDenied::UnsupportedTransport)
+        | None => false,
+    }
+}
+
 async fn connect_direct(
     url: &str,
     auth_provider: &dyn AuthProvider,

codex-rs/exec-server/src/remote/direct_tests.rs

@@ -6,7 +6,9 @@ use std::time::Instant;
 
 use anyhow::Result;
 use codex_api::AuthProvider;
+use codex_http_client::DestinationPolicy;
 use codex_http_client::HttpClientFactory;
+use codex_http_client::NetworkPolicyController;
 use codex_http_client::OutboundProxyPolicy;
 use http::HeaderMap;
 use http::HeaderValue;
@@ -282,8 +284,14 @@ async fn direct_registration_uses_proxy_policy_without_logging_secrets() -> Resu
 }
 
 #[tokio::test]
-async fn direct_registration_failure_stops_initial_and_conflict_attempts() -> Result<()> {
-    for successful_registrations in [0, 1] {
+async fn direct_registration_handles_policy_outages_and_permanent_failures() -> Result<()> {
+    for (successful_registrations, policy_unavailable) in [(0, false), (1, false), (0, true)] {
+        let controller = NetworkPolicyController::default();
+        let policy = controller.policy();
+        controller.publish(policy.revision(), DestinationPolicy::Unrestricted);
+        if policy_unavailable && successful_registrations == 0 {
+            controller.unavailable(policy.revision());
+        }
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let websocket_url = format!("ws://{}/connect", listener.local_addr()?);
         let registry = MockServer::start().await;
@@ -294,7 +302,7 @@ async fn direct_registration_failure_stops_initial_and_conflict_attempts() -> Re
             ))
             .respond_with(move |_: &wiremock::Request| {
                 let attempt = registration_count.fetch_add(1, Ordering::Relaxed);
-                if attempt < successful_registrations {
+                if policy_unavailable || attempt < successful_registrations {
                     ResponseTemplate::new(200).set_body_json(serde_json::json!({
                         "environment_id": "environment-requested",
                         "transport": DIRECT_TRANSPORT,
@@ -313,7 +321,8 @@ async fn direct_registration_failure_stops_initial_and_conflict_attempts() -> Re
             "environment-requested".to_string(),
             RemoteEnvironmentTransport::Direct,
             Arc::new(StaticAuthProvider),
-            HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
+            HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault)
+                .with_network_policy(policy.clone().for_current_account()),
         )?;
         let runtime_paths = ExecServerRuntimePaths::new(
             std::env::current_exe()?,
@@ -329,6 +338,18 @@ async fn direct_registration_failure_stops_initial_and_conflict_attempts() -> Re
                 .await?;
             socket.shutdown().await?;
         }
+        if policy_unavailable {
+            timeout(Duration::from_secs(1), listener.accept())
+                .await
+                .expect_err("unavailable policy opened a connection");
+            controller.publish(policy.revision(), DestinationPolicy::Unrestricted);
+            let (socket, _) = timeout(Duration::from_secs(5), listener.accept()).await??;
+            let _websocket = tokio_tungstenite::accept_async(socket).await?;
+            registry.verify().await;
+            task.abort();
+            let _ = task.await;
+            continue;
+        }
         let error = timeout(Duration::from_secs(5), task)
             .await??
             .expect_err("registration failure should stop the runner");