websocket-client: add proxy-aware connector (#31622)

## Why

The route-aware WebSocket connection setup in #31441 is transport
infrastructure rather than Responses API protocol logic. Landing it
first in a dedicated crate keeps `codex-api` focused on request and
response behavior and makes the transport reusable by future WebSocket
clients.

WebSockets must also apply the same effective outbound proxy and
custom-CA policy as HTTP without disabling the lower-latency WebSocket
path. Requiring an `HttpClientFactory` when constructing the connector
makes proxy-policy resolution part of the API instead of an optional
call-site convention.

This PR is an independent prerequisite based directly on `main`. After
it merges, #31441 can rebase onto it and replace its in-crate connector
with this API.

## What changed

- Add a new `codex-websocket-client` workspace crate with a
`WebSocketConnector` constructed from the effective `HttpClientFactory`.
- Resolve every destination through that factory before connecting, then
support direct connections, transport-default routing, HTTP proxies, and
TLS-encrypted HTTPS proxies.
- Preserve custom-CA trust for proxy and target TLS handshakes and
preserve Happy Eyeballs fallback for explicit direct and proxy routes.
- Expose an established `WebSocketConnection` as a uniform `Stream` and
`Sink`, hiding route-specific transport types from protocol clients.
- Add focused integration-style coverage for the public connector and
message stream, real WSS over direct and CONNECT routes, implicit and
explicit HTTPS proxy ports, and stalled-address-family fallback.

## Review guide

1. `codex-rs/websocket-client/src/lib.rs` defines the small public API
and the factory-required policy invariant.
2. `codex-rs/websocket-client/src/dialer.rs` contains DNS, TCP, proxy
tunneling, TLS, and WebSocket handshake setup.
3. `codex-rs/websocket-client/src/dialer_tests.rs` verifies the public
stream, direct and proxied WSS paths, HTTPS port preservation, and Happy
Eyeballs timing.
4. There is intentionally no consumer migration here; #31441 will become
the first consumer after this prerequisite merges.

## Test plan

- `cargo check -p codex-websocket-client --tests`
- `just test -p codex-websocket-client`
- `cargo shear`
- `just bazel-lock-check`

codex-rs/Cargo.lock

@@ -4485,6 +4485,22 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "codex-websocket-client"
+version = "0.0.0"
+dependencies = [
+ "codex-http-client",
+ "codex-utils-rustls-provider",
+ "futures",
+ "pretty_assertions",
+ "rcgen",
+ "rustls",
+ "tokio",
+ "tokio-rustls",
+ "tokio-tungstenite",
+ "url",
+]
+
 [[package]]
 name = "codex-windows-sandbox"
 version = "0.0.0"

codex-rs/Cargo.toml

@@ -90,6 +90,7 @@ members = [
     "tui",
     "tools",
     "v8-poc",
+    "websocket-client",
     "utils/absolute-path",
     "utils/path-uri",
     "utils/cargo-bin",
@@ -427,6 +428,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/websocket-client/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "websocket-client",
+    crate_name = "codex_websocket_client",
+)

codex-rs/websocket-client/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "codex-websocket-client"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+codex-http-client = { workspace = true }
+futures = { workspace = true }
+rustls = { workspace = true }
+tokio = { workspace = true, features = ["net", "rt", "time"] }
+tokio-rustls = { workspace = true }
+tokio-tungstenite = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+codex-utils-rustls-provider = { workspace = true }
+pretty_assertions = { workspace = true }
+rcgen = { workspace = true }
+tokio = { workspace = true, features = ["macros", "test-util"] }
+
+[lints]
+workspace = true
+
+[lib]
+doctest = false

codex-rs/websocket-client/src/dialer.rs

@@ -0,0 +1,256 @@
+use std::collections::VecDeque;
+use std::future::Future;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_http_client::OutboundProxyRoute;
+use futures::StreamExt;
+use futures::stream::FuturesUnordered;
+use rustls::ClientConfig;
+use rustls::pki_types::ServerName;
+use tokio::net::TcpStream;
+use tokio::time::Instant;
+use tokio::time::sleep_until;
+use tokio_rustls::TlsConnector;
+use tokio_tungstenite::Connector;
+use tokio_tungstenite::client_async_tls_with_config;
+use tokio_tungstenite::connect_async_tls_with_config;
+use tokio_tungstenite::proxy::connect_via_proxy;
+use tokio_tungstenite::tungstenite::Error as WebSocketError;
+use tokio_tungstenite::tungstenite::error::TlsError;
+use tokio_tungstenite::tungstenite::error::UrlError;
+use tokio_tungstenite::tungstenite::handshake::client::Request;
+use tokio_tungstenite::tungstenite::handshake::client::Response;
+use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
+use tokio_tungstenite::tungstenite::proxy::ProxyConfig;
+
+use crate::AsyncIo;
+use crate::ConnectionInner;
+
+const HAPPY_EYEBALLS_DELAY: Duration = Duration::from_millis(250);
+
+pub(crate) async fn connect(
+    request: Request,
+    config: WebSocketConfig,
+    tls_config: Arc<ClientConfig>,
+    proxy_route: OutboundProxyRoute,
+) -> Result<(ConnectionInner, Response), WebSocketError> {
+    let stream: Box<dyn AsyncIo> = match proxy_route {
+        OutboundProxyRoute::TransportDefault => {
+            // The workspace enables tokio-tungstenite's `proxy` feature, so its default dialer
+            // resolves HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, and NO_PROXY before opening the socket.
+            let (stream, response) = connect_async_tls_with_config(
+                request,
+                Some(config),
+                false, // Preserve Tungstenite's recommended Nagle default.
+                Some(Connector::Rustls(tls_config)),
+            )
+            .await?;
+            return Ok((ConnectionInner::TransportDefault(stream), response));
+        }
+        OutboundProxyRoute::Direct => {
+            let host = websocket_host(&request)?;
+            let port = websocket_port(&request)?;
+            Box::new(
+                connect_tcp(host_port(host, port))
+                    .await
+                    .map_err(WebSocketError::Io)?,
+            )
+        }
+        OutboundProxyRoute::Proxy { url } => {
+            let proxy = ProxyEndpoint::parse(&url)?;
+            let host = websocket_host(&request)?;
+            let port = websocket_port(&request)?;
+            let stream = connect_tcp(proxy.config.authority())
+                .await
+                .map_err(WebSocketError::Io)?;
+            let stream: Box<dyn AsyncIo> = if proxy.tls {
+                let server_name = ServerName::try_from(proxy.config.host.clone())
+                    .map_err(|_| WebSocketError::Tls(TlsError::InvalidDnsName))?;
+                let stream = TlsConnector::from(Arc::clone(&tls_config))
+                    .connect(server_name, stream)
+                    .await
+                    .map_err(WebSocketError::Io)?;
+                Box::new(stream)
+            } else {
+                Box::new(stream)
+            };
+            connect_via_proxy(stream, &proxy.config, host, port).await?
+        }
+    };
+
+    let (stream, response) = client_async_tls_with_config(
+        request,
+        stream,
+        Some(config),
+        Some(Connector::Rustls(tls_config)),
+    )
+    .await?;
+    Ok((ConnectionInner::Routed(stream), response))
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct ProxyEndpoint {
+    config: ProxyConfig,
+    tls: bool,
+}
+
+impl ProxyEndpoint {
+    fn parse(url: &str) -> Result<Self, WebSocketError> {
+        let mut parsed_url = url::Url::parse(url).map_err(|_| invalid_proxy_config())?;
+        let tls = parsed_url.scheme() == "https";
+        if tls {
+            // Capture the HTTPS default before changing schemes: `Url` normalizes default ports,
+            // so setting 443 before rewriting to HTTP would discard it and later imply port 80.
+            let port = parsed_url
+                .port_or_known_default()
+                .ok_or_else(invalid_proxy_config)?;
+            parsed_url
+                .set_scheme("http")
+                .map_err(|_| invalid_proxy_config())?;
+            parsed_url
+                .set_port(Some(port))
+                .map_err(|_| invalid_proxy_config())?;
+        }
+        let config = ProxyConfig::parse(parsed_url.as_str()).map_err(|error| match error {
+            WebSocketError::Url(UrlError::UnsupportedProxyScheme) => error,
+            _ => invalid_proxy_config(),
+        })?;
+        Ok(Self { config, tls })
+    }
+}
+
+fn invalid_proxy_config() -> WebSocketError {
+    WebSocketError::Url(UrlError::InvalidProxyConfig("<redacted>".to_string()))
+}
+
+fn websocket_host(request: &Request) -> Result<&str, WebSocketError> {
+    request
+        .uri()
+        .host()
+        .ok_or(WebSocketError::Url(UrlError::NoHostName))
+}
+
+fn websocket_port(request: &Request) -> Result<u16, WebSocketError> {
+    request
+        .uri()
+        .port_u16()
+        .or_else(|| match request.uri().scheme_str() {
+            Some("ws") => Some(80),
+            Some("wss") => Some(443),
+            _ => None,
+        })
+        .ok_or(WebSocketError::Url(UrlError::UnsupportedUrlScheme))
+}
+
+fn host_port(host: &str, port: u16) -> String {
+    if host.contains(':') && !host.starts_with('[') {
+        format!("[{host}]:{port}")
+    } else {
+        format!("{host}:{port}")
+    }
+}
+
+async fn connect_tcp(address: String) -> io::Result<TcpStream> {
+    let addresses = tokio::net::lookup_host(address).await?.collect::<Vec<_>>();
+    connect_happy_eyeballs(addresses, TcpStream::connect).await
+}
+
+async fn connect_happy_eyeballs<T, F, Fut>(
+    addresses: Vec<SocketAddr>,
+    mut connect: F,
+) -> io::Result<T>
+where
+    F: FnMut(SocketAddr) -> Fut,
+    Fut: Future<Output = io::Result<T>>,
+{
+    let mut addresses = addresses.into_iter();
+    let Some(first_address) = addresses.next() else {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "could not resolve to any address",
+        ));
+    };
+
+    let first_is_ipv4 = first_address.is_ipv4();
+    let mut preferred = VecDeque::new();
+    let mut alternate = VecDeque::new();
+    for address in addresses {
+        if address.is_ipv4() == first_is_ipv4 {
+            preferred.push_back(address);
+        } else {
+            alternate.push_back(address);
+        }
+    }
+
+    let mut addresses = VecDeque::new();
+    while !preferred.is_empty() || !alternate.is_empty() {
+        if let Some(address) = alternate.pop_front() {
+            addresses.push_back(address);
+        }
+        if let Some(address) = preferred.pop_front() {
+            addresses.push_back(address);
+        }
+    }
+
+    let mut attempts = FuturesUnordered::new();
+    attempts.push(connect(first_address));
+    let mut next_attempt_at = Instant::now() + HAPPY_EYEBALLS_DELAY;
+    let mut last_error = None;
+
+    loop {
+        if addresses.is_empty() {
+            match attempts.next().await {
+                Some(Ok(stream)) => return Ok(stream),
+                Some(Err(error)) => {
+                    if attempts.is_empty() {
+                        return Err(error);
+                    }
+                    last_error = Some(error);
+                }
+                None => {
+                    return Err(last_error.unwrap_or_else(|| {
+                        io::Error::other("connection attempts ended without an error")
+                    }));
+                }
+            }
+            continue;
+        }
+
+        tokio::select! {
+            result = attempts.next() => {
+                match result {
+                    Some(Ok(stream)) => return Ok(stream),
+                    Some(Err(error)) => {
+                        last_error = Some(error);
+                        let address = take_next_address(&mut addresses)?;
+                        attempts.push(connect(address));
+                        next_attempt_at = Instant::now() + HAPPY_EYEBALLS_DELAY;
+                    }
+                    None => {
+                        let address = take_next_address(&mut addresses)?;
+                        attempts.push(connect(address));
+                        next_attempt_at = Instant::now() + HAPPY_EYEBALLS_DELAY;
+                    }
+                }
+            }
+            _ = sleep_until(next_attempt_at) => {
+                let address = take_next_address(&mut addresses)?;
+                attempts.push(connect(address));
+                next_attempt_at = Instant::now() + HAPPY_EYEBALLS_DELAY;
+            }
+        }
+    }
+}
+
+fn take_next_address(addresses: &mut VecDeque<SocketAddr>) -> io::Result<SocketAddr> {
+    addresses
+        .pop_front()
+        .ok_or_else(|| io::Error::other("connection address queue unexpectedly empty"))
+}
+
+#[cfg(test)]
+#[path = "dialer_tests.rs"]
+mod tests;

codex-rs/websocket-client/src/dialer_tests.rs

@@ -0,0 +1,297 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_http_client::HttpClientFactory;
+use codex_http_client::OutboundProxyPolicy;
+use codex_http_client::OutboundProxyRoute;
+use codex_utils_rustls_provider::ensure_rustls_crypto_provider;
+use futures::SinkExt;
+use futures::StreamExt;
+use pretty_assertions::assert_eq;
+use rcgen::CertifiedKey;
+use rcgen::generate_simple_self_signed;
+use rustls::ClientConfig;
+use rustls::RootCertStore;
+use rustls::ServerConfig;
+use rustls::pki_types::PrivateKeyDer;
+use rustls::pki_types::PrivatePkcs8KeyDer;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
+use tokio_rustls::TlsAcceptor;
+use tokio_tungstenite::accept_async;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
+use tokio_tungstenite::tungstenite::proxy::ProxyScheme;
+
+use super::*;
+use crate::AsyncIo;
+use crate::WebSocketConnection;
+use crate::WebSocketConnector;
+
+#[tokio::test]
+async fn public_connector_uses_factory_and_exposes_stream_and_sink() {
+    let (target_addr, target_task) = start_plain_echo_websocket_server().await;
+    let request = format!("ws://localhost:{}/v1/responses", target_addr.port())
+        .into_client_request()
+        .expect("websocket request should build");
+    let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault);
+    let connector = WebSocketConnector::new(&factory).expect("connector should build");
+
+    let (mut websocket, _) = connector
+        .connect(request, WebSocketConfig::default())
+        .await
+        .expect("websocket handshake should succeed");
+    let expected = Message::Text("hello".into());
+    websocket
+        .send(expected.clone())
+        .await
+        .expect("websocket should send");
+    let actual = websocket
+        .next()
+        .await
+        .expect("websocket should receive a message")
+        .expect("websocket message should be valid");
+    assert_eq!(actual, expected);
+
+    target_task.await.expect("target task should finish");
+}
+
+#[tokio::test]
+async fn direct_route_connects_secure_websocket() {
+    let (tls_config, acceptor) = test_tls_configs();
+    let (target_addr, target_task) = start_tls_websocket_server(acceptor).await;
+    let request = format!("wss://localhost:{}/v1/responses", target_addr.port())
+        .into_client_request()
+        .expect("websocket request should build");
+
+    let (inner, _) = connect(
+        request,
+        WebSocketConfig::default(),
+        tls_config,
+        OutboundProxyRoute::Direct,
+    )
+    .await
+    .expect("direct websocket handshake should succeed");
+    drop(WebSocketConnection { inner });
+
+    target_task.await.expect("target task should finish");
+}
+
+#[tokio::test]
+async fn http_proxy_tunnels_secure_websocket_before_handshake() {
+    assert_proxy_tunnels_secure_websocket(/*proxy_tls*/ false).await;
+}
+
+#[tokio::test]
+async fn https_proxy_tunnels_secure_websocket_before_handshake() {
+    assert_proxy_tunnels_secure_websocket(/*proxy_tls*/ true).await;
+}
+
+#[test]
+fn https_proxy_defaults_to_port_443_and_preserves_explicit_port() {
+    let default_port = ProxyEndpoint::parse("https://proxy.example")
+        .expect("HTTPS proxy without a port should parse");
+    let explicit_port = ProxyEndpoint::parse("https://proxy.example:8443")
+        .expect("HTTPS proxy with a port should parse");
+
+    assert_eq!(
+        default_port,
+        ProxyEndpoint {
+            config: ProxyConfig {
+                scheme: ProxyScheme::Http,
+                host: "proxy.example".to_string(),
+                port: 443,
+                auth: None,
+            },
+            tls: true,
+        }
+    );
+    assert_eq!(
+        explicit_port,
+        ProxyEndpoint {
+            config: ProxyConfig {
+                scheme: ProxyScheme::Http,
+                host: "proxy.example".to_string(),
+                port: 8443,
+                auth: None,
+            },
+            tls: true,
+        }
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn happy_eyeballs_does_not_wait_for_stalled_preferred_family() {
+    let stalled = "[2001:db8::1]:443"
+        .parse::<SocketAddr>()
+        .expect("stalled address should parse");
+    let reachable = "127.0.0.1:443"
+        .parse::<SocketAddr>()
+        .expect("reachable address should parse");
+
+    let connected = tokio::time::timeout(
+        Duration::from_secs(1),
+        connect_happy_eyeballs(vec![stalled, reachable], |address| async move {
+            if address == stalled {
+                std::future::pending::<()>().await;
+            }
+            Ok(address)
+        }),
+    )
+    .await
+    .expect("alternate family should start before timeout")
+    .expect("alternate family should connect");
+
+    assert_eq!(connected, reachable);
+}
+
+async fn start_plain_echo_websocket_server() -> (SocketAddr, JoinHandle<()>) {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("target listener should bind");
+    let address = listener
+        .local_addr()
+        .expect("target listener should have an address");
+    let task = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("target should accept");
+        let mut websocket = accept_async(stream)
+            .await
+            .expect("target websocket handshake should succeed");
+        let message = websocket
+            .next()
+            .await
+            .expect("target should receive a message")
+            .expect("target websocket message should be valid");
+        websocket
+            .send(message)
+            .await
+            .expect("target should echo the message");
+    });
+    (address, task)
+}
+
+async fn assert_proxy_tunnels_secure_websocket(proxy_tls: bool) {
+    let (tls_config, acceptor) = test_tls_configs();
+    let (target_addr, target_task) = start_tls_websocket_server(acceptor.clone()).await;
+
+    let proxy_listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("proxy listener should bind");
+    let proxy_addr = proxy_listener
+        .local_addr()
+        .expect("proxy listener should have an address");
+    let connect_request = Arc::new(Mutex::new(None));
+    let proxy_connect_request = Arc::clone(&connect_request);
+    let proxy_task = tokio::spawn(async move {
+        let (client, _) = proxy_listener.accept().await.expect("proxy should accept");
+        let mut client: Box<dyn AsyncIo> = if proxy_tls {
+            Box::new(
+                acceptor
+                    .accept(client)
+                    .await
+                    .expect("proxy TLS handshake should succeed"),
+            )
+        } else {
+            Box::new(client)
+        };
+        let mut request = Vec::new();
+        let mut byte = [0_u8; 1];
+        while !request.ends_with(b"\r\n\r\n") {
+            client
+                .read_exact(&mut byte)
+                .await
+                .expect("proxy should read CONNECT request");
+            request.push(byte[0]);
+        }
+        *proxy_connect_request.lock().await =
+            Some(String::from_utf8(request).expect("CONNECT request should contain valid UTF-8"));
+
+        let mut target = tokio::net::TcpStream::connect(target_addr)
+            .await
+            .expect("proxy should connect to target");
+        client
+            .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")
+            .await
+            .expect("proxy should acknowledge CONNECT");
+        let _ = tokio::io::copy_bidirectional(&mut client, &mut target).await;
+    });
+
+    let target_authority = format!("localhost:{}", target_addr.port());
+    let proxy_scheme = if proxy_tls { "https" } else { "http" };
+    let request = format!("wss://{target_authority}/v1/responses")
+        .into_client_request()
+        .expect("websocket request should build");
+    let (inner, _) = connect(
+        request,
+        WebSocketConfig::default(),
+        tls_config,
+        OutboundProxyRoute::Proxy {
+            url: format!("{proxy_scheme}://localhost:{}", proxy_addr.port()),
+        },
+    )
+    .await
+    .expect("proxied websocket handshake should succeed");
+    drop(WebSocketConnection { inner });
+
+    target_task.await.expect("target task should finish");
+    proxy_task.await.expect("proxy task should finish");
+    let request = connect_request
+        .lock()
+        .await
+        .clone()
+        .expect("proxy should record CONNECT request");
+    let expected_request_line = format!("CONNECT {target_authority} HTTP/1.1");
+    assert_eq!(request.lines().next(), Some(expected_request_line.as_str()));
+}
+
+async fn start_tls_websocket_server(acceptor: TlsAcceptor) -> (SocketAddr, JoinHandle<()>) {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("target listener should bind");
+    let address = listener
+        .local_addr()
+        .expect("target listener should have an address");
+    let task = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("target should accept");
+        let stream = acceptor
+            .accept(stream)
+            .await
+            .expect("target TLS handshake should succeed");
+        let mut websocket = accept_async(stream)
+            .await
+            .expect("target websocket handshake should succeed");
+        let _ = websocket.close(None).await;
+    });
+    (address, task)
+}
+
+fn test_tls_configs() -> (Arc<ClientConfig>, TlsAcceptor) {
+    ensure_rustls_crypto_provider();
+    let CertifiedKey { cert, signing_key } =
+        generate_simple_self_signed(vec!["localhost".to_string()])
+            .expect("test certificate should generate");
+    let certificate = cert.der().clone();
+    let private_key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der()));
+    let server_config = ServerConfig::builder()
+        .with_no_client_auth()
+        .with_single_cert(vec![certificate.clone()], private_key)
+        .expect("test server config should build");
+
+    let mut roots = RootCertStore::empty();
+    roots
+        .add(certificate)
+        .expect("test certificate should be trusted");
+    let client_config = ClientConfig::builder()
+        .with_root_certificates(roots)
+        .with_no_client_auth();
+
+    (
+        Arc::new(client_config),
+        TlsAcceptor::from(Arc::new(server_config)),
+    )
+}

codex-rs/websocket-client/src/lib.rs

@@ -0,0 +1,133 @@
+//! Proxy-aware WebSocket connection setup shared by Codex API clients.
+
+mod dialer;
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::Context;
+use std::task::Poll;
+
+use codex_http_client::BuildCustomCaTransportError;
+use codex_http_client::HttpClientFactory;
+use codex_http_client::build_rustls_client_config_with_custom_ca;
+use futures::Sink;
+use futures::Stream;
+use rustls::ClientConfig;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncWrite;
+use tokio::net::TcpStream;
+use tokio_tungstenite::MaybeTlsStream;
+use tokio_tungstenite::WebSocketStream as TungsteniteStream;
+use tokio_tungstenite::tungstenite::Error as WebSocketError;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::handshake::client::Request;
+use tokio_tungstenite::tungstenite::handshake::client::Response;
+use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
+
+/// Connects WebSockets using the outbound proxy policy resolved by application configuration.
+///
+/// Construct this from the effective [`HttpClientFactory`] rather than selecting proxy behavior at
+/// individual call sites. Each connection resolves its destination through that factory before
+/// opening a socket.
+#[derive(Clone)]
+pub struct WebSocketConnector {
+    http_client_factory: HttpClientFactory,
+    tls_config: Arc<ClientConfig>,
+}
+
+impl WebSocketConnector {
+    /// Creates a connector using native roots and any configured Codex custom CA bundle.
+    pub fn new(
+        http_client_factory: &HttpClientFactory,
+    ) -> Result<Self, BuildCustomCaTransportError> {
+        Ok(Self {
+            http_client_factory: http_client_factory.clone(),
+            tls_config: build_rustls_client_config_with_custom_ca()?,
+        })
+    }
+
+    /// Connects a WebSocket after resolving the request destination through the configured proxy
+    /// policy.
+    pub async fn connect(
+        &self,
+        request: Request,
+        config: WebSocketConfig,
+    ) -> Result<(WebSocketConnection, Response), WebSocketError> {
+        let proxy_route = self
+            .http_client_factory
+            .resolve_proxy_route(&request.uri().to_string());
+        let (inner, response) =
+            dialer::connect(request, config, Arc::clone(&self.tls_config), proxy_route).await?;
+        Ok((WebSocketConnection { inner }, response))
+    }
+}
+
+/// An established WebSocket independent of its direct, proxy, and TLS transport layers.
+///
+/// This implements [`Stream`] and [`Sink`] so protocol clients can process Tungstenite messages
+/// without knowing which concrete network stream route selection produced.
+pub struct WebSocketConnection {
+    inner: ConnectionInner,
+}
+
+impl Stream for WebSocketConnection {
+    type Item = Result<Message, WebSocketError>;
+
+    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match &mut self.get_mut().inner {
+            ConnectionInner::TransportDefault(stream) => Pin::new(stream).poll_next(context),
+            ConnectionInner::Routed(stream) => Pin::new(stream).poll_next(context),
+        }
+    }
+}
+
+impl Sink<Message> for WebSocketConnection {
+    type Error = WebSocketError;
+
+    fn poll_ready(
+        self: Pin<&mut Self>,
+        context: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        match &mut self.get_mut().inner {
+            ConnectionInner::TransportDefault(stream) => Pin::new(stream).poll_ready(context),
+            ConnectionInner::Routed(stream) => Pin::new(stream).poll_ready(context),
+        }
+    }
+
+    fn start_send(self: Pin<&mut Self>, message: Message) -> Result<(), Self::Error> {
+        match &mut self.get_mut().inner {
+            ConnectionInner::TransportDefault(stream) => Pin::new(stream).start_send(message),
+            ConnectionInner::Routed(stream) => Pin::new(stream).start_send(message),
+        }
+    }
+
+    fn poll_flush(
+        self: Pin<&mut Self>,
+        context: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        match &mut self.get_mut().inner {
+            ConnectionInner::TransportDefault(stream) => Pin::new(stream).poll_flush(context),
+            ConnectionInner::Routed(stream) => Pin::new(stream).poll_flush(context),
+        }
+    }
+
+    fn poll_close(
+        self: Pin<&mut Self>,
+        context: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        match &mut self.get_mut().inner {
+            ConnectionInner::TransportDefault(stream) => Pin::new(stream).poll_close(context),
+            ConnectionInner::Routed(stream) => Pin::new(stream).poll_close(context),
+        }
+    }
+}
+
+pub(crate) enum ConnectionInner {
+    TransportDefault(TungsteniteStream<MaybeTlsStream<TcpStream>>),
+    Routed(TungsteniteStream<MaybeTlsStream<Box<dyn AsyncIo>>>),
+}
+
+/// Async network I/O carried through optional proxy and target TLS handshakes.
+pub(crate) trait AsyncIo: AsyncRead + AsyncWrite + Send + Unpin {}
+
+impl<T> AsyncIo for T where T: AsyncRead + AsyncWrite + Send + Unpin {}