Keep Guardian v2 WebSocket pools warm with concurrent replenishment (#51378)

## Why

Serial connection opens let a stalled handshake delay healthy sockets. Waiting until the idle pool is empty to refill also leaves classification bursts using HTTP fallback.

## What changed

- Open up to four WebSockets concurrently and start replenishing when the idle pool falls to half capacity. Preserve successful in-flight opens when another open fails or times out.
- Make explicit prewarming wait for an existing refill before checking readiness.
- Add `ResponsesWebsocketClient::connect_with_connector` and reuse one connector per sampler, initializing TLS configuration on a blocking task while retaining per-connection authentication and destination policy checks.
- Record connection lease durations for admission, setup, and total time, tagged by transport.

## Testing

Add regression tests for a stalled handshake alongside a healthy socket and replenishment before the last idle socket is leased. Update cooldown, authentication, and transport retry tests to account for concurrent handshake completion.

GitOrigin-RevId: f4a8258977c1f2a58e00c6badc7b599280ef22ad

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

@@ -349,6 +349,8 @@ impl ResponsesWebsocketConnection {
     }
 }
 
+mod connector;
+
 /// Client for connecting to the Responses WebSocket endpoint for one provider.
 pub struct ResponsesWebsocketClient {
     provider: Provider,
@@ -385,12 +387,6 @@ impl ResponsesWebsocketClient {
         Self { provider, auth }
     }
 
-    #[instrument(
-        name = "responses_websocket.connect",
-        level = "info",
-        skip_all,
-        fields(transport = "responses_websocket", api.path = "/responses")
-    )]
     pub async fn connect(
         &self,
         http_client_factory: &HttpClientFactory,
@@ -399,24 +395,16 @@ impl ResponsesWebsocketClient {
         turn_state: Option<Arc<OnceLock<String>>>,
         telemetry: Option<Arc<dyn WebsocketTelemetry>>,
     ) -> Result<ResponsesWebsocketConnection, ApiError> {
-        let ws_url = self
-            .provider
-            .websocket_url_for_path("/responses")
-            .map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?;
-
-        let mut headers =
-            merge_request_headers(&self.provider.headers, extra_headers, default_headers);
-        self.auth.add_auth_headers(&mut headers);
-
-        let (stream, _status, server_reasoning_included, server_model) =
-            connect_websocket(ws_url, headers, http_client_factory, turn_state.clone()).await?;
-        Ok(ResponsesWebsocketConnection::new(
-            stream,
-            self.provider.stream_idle_timeout,
-            server_reasoning_included,
-            server_model,
+        let connector = WebSocketConnector::new(http_client_factory)
+            .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?;
+        self.connect_with_connector(
+            &connector,
+            extra_headers,
+            default_headers,
+            turn_state,
             telemetry,
-        ))
+        )
+        .await
     }
 
     /// Opens a WebSocket connection long enough to validate the upgrade response.
@@ -442,10 +430,12 @@ impl ResponsesWebsocketClient {
             merge_request_headers(&self.provider.headers, extra_headers, default_headers);
         self.auth.add_auth_headers(&mut headers);
 
+        let connector = WebSocketConnector::new(http_client_factory)
+            .map_err(|err| ApiError::Stream(format!("failed to configure websocket TLS: {err}")))?;
         let (mut stream, status, reasoning_included, server_model) = connect_websocket(
             ws_url.clone(),
             headers,
-            http_client_factory,
+            &connector,
             /*turn_state*/ None,
         )
         .await?;
@@ -499,7 +489,7 @@ fn merge_request_headers(
 async fn connect_websocket(
     url: Url,
     headers: HeaderMap,
-    http_client_factory: &HttpClientFactory,
+    connector: &WebSocketConnector,
     turn_state: Option<Arc<OnceLock<String>>>,
 ) -> Result<(WsStream, StatusCode, bool, Option<String>), ApiError> {
     info!("connecting to websocket: {url}");
@@ -510,8 +500,6 @@ async fn connect_websocket(
         .map_err(|err| ApiError::Stream(format!("failed to build websocket request: {err}")))?;
     request.headers_mut().extend(headers);
 
-    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 {

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

@@ -0,0 +1,49 @@
+//! Establishes Responses sockets with caller-owned TLS and proxy configuration.
+//! Connector reuse never caches authentication or bypasses per-connection destination policy.
+
+use super::ResponsesWebsocketClient;
+use super::ResponsesWebsocketConnection;
+use super::connect_websocket;
+use super::merge_request_headers;
+use crate::ApiError;
+use crate::WebsocketTelemetry;
+use codex_websocket_client::WebSocketConnector;
+use http::HeaderMap;
+use std::sync::Arc;
+use std::sync::OnceLock;
+
+impl ResponsesWebsocketClient {
+    /// Opens a socket using the caller's reusable connector and this client's current auth.
+    /// The caller owns the lifetime of the connector's TLS configuration snapshot.
+    #[tracing::instrument(
+        name = "responses_websocket.connect",
+        level = "info",
+        skip_all,
+        fields(transport = "responses_websocket", api.path = "/responses")
+    )]
+    pub async fn connect_with_connector(
+        &self,
+        connector: &WebSocketConnector,
+        extra_headers: HeaderMap,
+        default_headers: HeaderMap,
+        turn_state: Option<Arc<OnceLock<String>>>,
+        telemetry: Option<Arc<dyn WebsocketTelemetry>>,
+    ) -> Result<ResponsesWebsocketConnection, ApiError> {
+        let ws_url = self
+            .provider
+            .websocket_url_for_path("/responses")
+            .map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?;
+        let mut headers =
+            merge_request_headers(&self.provider.headers, extra_headers, default_headers);
+        self.auth.add_auth_headers(&mut headers);
+        let (stream, _status, server_reasoning_included, server_model) =
+            connect_websocket(ws_url, headers, connector, turn_state).await?;
+        Ok(ResponsesWebsocketConnection::new(
+            stream,
+            self.provider.stream_idle_timeout,
+            server_reasoning_included,
+            server_model,
+            telemetry,
+        ))
+    }
+}