From 48b40c22a94b76b3cc2e17b1ad1efdce0eae0d4c Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 8 Jul 2026 12:58:51 -0700 Subject: [PATCH 1/2] core: preserve Responses WebSockets with system proxy --- codex-rs/Cargo.lock | 1 + codex-rs/Cargo.toml | 1 + codex-rs/cli/src/doctor.rs | 2 + codex-rs/codex-api/Cargo.toml | 1 + .../src/endpoint/responses_websocket.rs | 42 ++++++++----------- codex-rs/core/src/client.rs | 1 + 6 files changed, 24 insertions(+), 24 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9856ef9cd374..3d9847974996 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1940,6 +1940,7 @@ dependencies = [ "codex-http-client", "codex-protocol", "codex-utils-rustls-provider", + "codex-websocket-client", "eventsource-stream", "futures", "http 1.4.0", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 0085b663efc9..7b153dfaa0b7 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -168,6 +168,7 @@ codex-code-mode = { path = "code-mode" } codex-code-mode-protocol = { path = "code-mode-protocol" } codex-home = { path = "codex-home" } codex-http-client = { path = "http-client" } +codex-websocket-client = { path = "websocket-client" } codex-config = { path = "config" } codex-connectors = { path = "connectors" } codex-connectors-extension = { path = "ext/connectors" } diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index 2e9c0094c4e8..7f06a611bcfa 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -2383,9 +2383,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, diff --git a/codex-rs/codex-api/Cargo.toml b/codex-rs/codex-api/Cargo.toml index 084b058e1263..21f6b674250c 100644 --- a/codex-rs/codex-api/Cargo.toml +++ b/codex-rs/codex-api/Cargo.toml @@ -13,6 +13,7 @@ codex-client = { workspace = true } codex-http-client = { workspace = true } codex-protocol = { workspace = true } codex-utils-rustls-provider = { workspace = true } +codex-websocket-client = { workspace = true } futures = { workspace = true } http = { workspace = true } reqwest = { workspace = true, features = ["json", "stream"] } diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index cce056f638b3..88e7fbb9b4e0 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -11,8 +11,9 @@ use crate::sse::ResponsesStreamEvent; use crate::sse::process_responses_event; use crate::telemetry::WebsocketTelemetry; use codex_client::TransportError; -use codex_http_client::maybe_build_rustls_client_config_with_custom_ca; -use codex_utils_rustls_provider::ensure_rustls_crypto_provider; +use codex_http_client::HttpClientFactory; +use codex_websocket_client::WebSocketConnection; +use codex_websocket_client::WebSocketConnector; use futures::SinkExt; use futures::StreamExt; use http::HeaderMap; @@ -25,14 +26,10 @@ 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::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 +59,7 @@ enum WsCommand { } impl WsStream { - fn new(inner: WebSocketStream>) -> Self { + fn new(inner: WebSocketConnection) -> Self { let (tx_command, mut rx_command) = mpsc::channel::(32); let (tx_message, rx_message) = mpsc::unbounded_channel::>(); @@ -334,6 +331,7 @@ impl ResponsesWebsocketClient { )] pub async fn connect( &self, + http_client_factory: &HttpClientFactory, extra_headers: HeaderMap, default_headers: HeaderMap, turn_state: Option>>, @@ -349,7 +347,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 +367,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 +382,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,9 +442,9 @@ fn merge_request_headers( async fn connect_websocket( url: Url, headers: HeaderMap, + http_client_factory: &HttpClientFactory, turn_state: Option>>, ) -> Result<(WsStream, StatusCode, bool, Option, Option), ApiError> { - ensure_rustls_crypto_provider(); info!("connecting to websocket: {url}"); let mut request = url @@ -448,20 +453,9 @@ 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; + let connector = WebSocketConnector::new(http_client_factory) + .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?; + let response = connector.connect(request, websocket_config()).await; let (stream, response) = match response { Ok((stream, response)) => { diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 8c98a3372eae..2bd4ae3bcf67 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -1006,6 +1006,7 @@ impl ModelClient { let result = match tokio::time::timeout( websocket_connect_timeout, ApiWebSocketResponsesClient::new(api_provider, api_auth).connect( + &self.http_client_factory, headers, codex_login::default_client::default_headers(), /*turn_state*/ None, From 2040588b019216b085c82121e2eb839f22b1b442 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 8 Jul 2026 13:03:27 -0700 Subject: [PATCH 2/2] core: test Responses WebSockets with system proxy --- .../core/tests/suite/client_websockets.rs | 56 ++++++++++++++++++- 1 file changed, 53 insertions(+), 3 deletions(-) diff --git a/codex-rs/core/tests/suite/client_websockets.rs b/codex-rs/core/tests/suite/client_websockets.rs index 05a93807b229..71d0965ef9d7 100755 --- a/codex-rs/core/tests/suite/client_websockets.rs +++ b/codex-rs/core/tests/suite/client_websockets.rs @@ -8,6 +8,7 @@ use codex_core::Prompt; use codex_core::ResponseEvent; use codex_core::X_RESPONSESAPI_INCLUDE_TIMING_METRICS_HEADER; use codex_features::Feature; +use codex_http_client::OutboundProxyPolicy; use codex_login::CodexAuth; use codex_login::auth::AgentIdentityAuthPolicy; use codex_model_provider_info::ModelProviderInfo; @@ -102,6 +103,7 @@ fn assert_request_trace_matches(body: &serde_json::Value, expected_trace: &W3cTr struct WebsocketTestHarness { _codex_home: TempDir, client: ModelClient, + outbound_proxy_policy: OutboundProxyPolicy, session_id: SessionId, thread_id: ThreadId, model_info: ModelInfo, @@ -230,6 +232,38 @@ async fn responses_websocket_streams_without_feature_flag_when_provider_supports server.shutdown().await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn responses_websocket_streams_with_system_proxy_feature() { + skip_if_no_network!(); + + let server = start_websocket_server(vec![vec![vec![ + ev_response_created("resp-1"), + ev_completed("resp-1"), + ]]]) + .await; + + let harness = websocket_harness_with_provider_options( + websocket_provider(&server), + /*runtime_metrics_enabled*/ false, + /*concurrent_reasoning_summaries_enabled*/ false, + /*enabled_features*/ &[Feature::RespectSystemProxy], + ) + .await; + assert_eq!( + harness.outbound_proxy_policy, + OutboundProxyPolicy::RespectSystemProxy + ); + let mut client_session = harness.client.new_session(); + let prompt = prompt_with_input(vec![message_item("hello")]); + + stream_until_complete(&mut client_session, &harness, &prompt).await; + + assert_eq!(server.handshakes().len(), 1); + assert_eq!(server.single_connection().len(), 1); + + server.shutdown().await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn responses_websocket_reuses_connection_with_per_turn_trace_payloads() { skip_if_no_network!(); @@ -389,8 +423,10 @@ async fn responses_websocket_request_prewarm_reuses_connection() { let mut provider = websocket_provider(&server); provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name; let harness = websocket_harness_with_provider_options( - provider, /*runtime_metrics_enabled*/ true, + provider, + /*runtime_metrics_enabled*/ true, /*concurrent_reasoning_summaries_enabled*/ true, + /*enabled_features*/ &[], ) .await; let mut client_session = harness.client.new_session(); @@ -1005,8 +1041,10 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() { let mut provider = websocket_provider(&server); provider.name = ModelProviderInfo::create_openai_provider(/*base_url*/ None).name; let harness = websocket_harness_with_provider_options( - provider, /*runtime_metrics_enabled*/ false, + provider, + /*runtime_metrics_enabled*/ false, /*concurrent_reasoning_summaries_enabled*/ false, + /*enabled_features*/ &[], ) .await; @@ -2216,6 +2254,7 @@ async fn websocket_harness_with_options( websocket_provider(server), runtime_metrics_enabled, /*concurrent_reasoning_summaries_enabled*/ false, + /*enabled_features*/ &[], ) .await } @@ -2224,6 +2263,7 @@ async fn websocket_harness_with_provider_options( provider: ModelProviderInfo, runtime_metrics_enabled: bool, concurrent_reasoning_summaries_enabled: bool, + enabled_features: &[Feature], ) -> WebsocketTestHarness { let codex_home = TempDir::new().unwrap(); let mut config = load_default_config_for_test(&codex_home).await; @@ -2240,6 +2280,15 @@ async fn websocket_harness_with_provider_options( .enable(Feature::ConcurrentReasoningSummaries) .expect("test config should allow feature update"); } + for feature in enabled_features { + config + .features + .enable(*feature) + .expect("test config should allow feature update"); + } + config.respect_system_proxy = config.features.enabled(Feature::RespectSystemProxy); + let http_client_factory = config.http_client_factory(); + let outbound_proxy_policy = http_client_factory.outbound_proxy_policy(); let config = Arc::new(config); let model_info = codex_core::test_support::construct_model_info_offline(MODEL, &config); let thread_id = ThreadId::new(); @@ -2284,12 +2333,13 @@ async fn websocket_harness_with_provider_options( .features .enabled(Feature::ConcurrentReasoningSummaries), /*attestation_provider*/ None, - config.http_client_factory(), + http_client_factory, ); WebsocketTestHarness { _codex_home: codex_home, client, + outbound_proxy_policy, session_id, thread_id, model_info,