core: preserve Responses WebSockets with system proxy

codex-rs/Cargo.toml

@@ -425,6 +425,7 @@ thiserror = "2.0.17"
 time = "0.3.47"
 tiny_http = "0.12"
 tokio = "1"
+tokio-rustls = "0.26.4"
 tokio-stream = "0.1.18"
 tokio-test = "0.4"
 tokio-tungstenite = { version = "0.28.0", features = [

codex-rs/cli/src/doctor.rs

@@ -2371,9 +2371,11 @@ async fn websocket_reachability_check(
         HeaderValue::from_static(RESPONSES_WEBSOCKETS_V2_BETA_HEADER_VALUE),
     );
     let client = ResponsesWebsocketClient::new(api_provider, api_auth);
+    let http_client_factory = config.http_client_factory();
     match tokio::time::timeout(
         provider.websocket_connect_timeout(),
         client.probe_handshake(
+            &http_client_factory,
             extra_headers,
             default_headers(),
             WEBSOCKET_IMMEDIATE_CLOSE_GRACE,

codex-rs/codex-api/Cargo.toml

@@ -16,11 +16,13 @@ codex-utils-rustls-provider = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
 reqwest = { workspace = true, features = ["json", "stream"] }
+rustls = { workspace = true }
 schemars = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["fs", "macros", "net", "rt", "sync", "time"] }
+tokio-rustls = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tungstenite = { workspace = true }
 tracing = { workspace = true }
@@ -33,6 +35,7 @@ url = { workspace = true }
 anyhow = { workspace = true }
 assert_matches = { workspace = true }
 pretty_assertions = { workspace = true }
+rcgen = { workspace = true }
 tokio-test = { workspace = true }
 wiremock = { workspace = true }
 reqwest = { workspace = true }

codex-rs/codex-api/src/endpoint/responses_websocket.rs

@@ -10,8 +10,11 @@ use crate::safety_buffering::treatment_from_headers;
 use crate::sse::ResponsesStreamEvent;
 use crate::sse::process_responses_event;
 use crate::telemetry::WebsocketTelemetry;
+use crate::websocket_connector;
+use crate::websocket_connector::ConnectedWebSocket;
 use codex_client::TransportError;
-use codex_http_client::maybe_build_rustls_client_config_with_custom_ca;
+use codex_http_client::HttpClientFactory;
+use codex_http_client::build_rustls_client_config_with_custom_ca;
 use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
 use futures::SinkExt;
 use futures::StreamExt;
@@ -25,14 +28,13 @@ use serde_json::map::Map as JsonMap;
 use std::sync::Arc;
 use std::sync::OnceLock;
 use std::time::Duration;
-use tokio::net::TcpStream;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::time::Instant;
-use tokio_tungstenite::MaybeTlsStream;
 use tokio_tungstenite::WebSocketStream;
-use tokio_tungstenite::connect_async_tls_with_config;
 use tokio_tungstenite::tungstenite::Error as WsError;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::tungstenite::client::IntoClientRequest;
@@ -62,7 +64,17 @@ enum WsCommand {
 }
 
 impl WsStream {
-    fn new(inner: WebSocketStream<MaybeTlsStream<TcpStream>>) -> Self {
+    fn new(inner: ConnectedWebSocket) -> Self {
+        match inner {
+            ConnectedWebSocket::TransportDefault(inner) => Self::new_inner(inner),
+            ConnectedWebSocket::Routed(inner) => Self::new_inner(inner),
+        }
+    }
+
+    fn new_inner<S>(inner: WebSocketStream<S>) -> Self
+    where
+        S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    {
         let (tx_command, mut rx_command) = mpsc::channel::<WsCommand>(32);
         let (tx_message, rx_message) = mpsc::unbounded_channel::<Result<Message, WsError>>();
 
@@ -334,6 +346,7 @@ impl ResponsesWebsocketClient {
     )]
     pub async fn connect(
         &self,
+        http_client_factory: &HttpClientFactory,
         extra_headers: HeaderMap,
         default_headers: HeaderMap,
         turn_state: Option<Arc<OnceLock<String>>>,
@@ -349,7 +362,7 @@ impl ResponsesWebsocketClient {
         self.auth.add_auth_headers(&mut headers);
 
         let (stream, _status, server_reasoning_included, models_etag, server_model) =
-            connect_websocket(ws_url, headers, turn_state.clone()).await?;
+            connect_websocket(ws_url, headers, http_client_factory, turn_state.clone()).await?;
         Ok(ResponsesWebsocketConnection::new(
             stream,
             self.provider.stream_idle_timeout,
@@ -369,6 +382,7 @@ impl ResponsesWebsocketClient {
     /// a usable connection from a policy rejection that closes right away.
     pub async fn probe_handshake(
         &self,
+        http_client_factory: &HttpClientFactory,
         extra_headers: HeaderMap,
         default_headers: HeaderMap,
         immediate_close_timeout: Duration,
@@ -383,7 +397,13 @@ impl ResponsesWebsocketClient {
         self.auth.add_auth_headers(&mut headers);
 
         let (mut stream, status, reasoning_included, models_etag, server_model) =
-            connect_websocket(ws_url.clone(), headers, /*turn_state*/ None).await?;
+            connect_websocket(
+                ws_url.clone(),
+                headers,
+                http_client_factory,
+                /*turn_state*/ None,
+            )
+            .await?;
         let immediate_close = tokio::time::timeout(immediate_close_timeout, stream.next())
             .await
             .ok()
@@ -437,6 +457,7 @@ fn merge_request_headers(
 async fn connect_websocket(
     url: Url,
     headers: HeaderMap,
+    http_client_factory: &HttpClientFactory,
     turn_state: Option<Arc<OnceLock<String>>>,
 ) -> Result<(WsStream, StatusCode, bool, Option<String>, Option<String>), ApiError> {
     ensure_rustls_crypto_provider();
@@ -448,20 +469,15 @@ async fn connect_websocket(
         .map_err(|err| ApiError::Stream(format!("failed to build websocket request: {err}")))?;
     request.headers_mut().extend(headers);
 
-    // Secure websocket traffic needs the same custom-CA policy as reqwest-based HTTPS traffic.
-    // If a Codex-specific CA bundle is configured, build an explicit rustls connector so this
-    // websocket path does not fall back to tungstenite's default native-roots-only behavior.
-    let connector = maybe_build_rustls_client_config_with_custom_ca()
-        .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?
-        .map(tokio_tungstenite::Connector::Rustls);
-
-    let response = connect_async_tls_with_config(
-        request,
-        Some(websocket_config()),
-        false, // `false` means "do not disable Nagle", which is tungstenite's recommended default.
-        connector,
-    )
-    .await;
+    // Secure websocket traffic and TLS-encrypted proxy connections need the same custom-CA policy
+    // as reqwest-based HTTPS traffic, so build one rustls config for both handshakes.
+    let tls_config = build_rustls_client_config_with_custom_ca()
+        .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?;
+
+    let proxy_route = http_client_factory.resolve_proxy_route(url.as_str());
+    let response =
+        websocket_connector::connect(request, Some(websocket_config()), tls_config, proxy_route)
+            .await;
 
     let (stream, response) = match response {
         Ok((stream, response)) => {

codex-rs/codex-api/src/lib.rs

@@ -12,6 +12,7 @@ pub(crate) mod safety_buffering;
 pub(crate) mod search;
 pub(crate) mod sse;
 pub(crate) mod telemetry;
+mod websocket_connector;
 
 pub use crate::requests::headers::build_session_headers;
 pub use codex_client::RequestTelemetry;