backend-client: route requests through HTTP client factory

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

@@ -5,6 +5,7 @@ mod default_client;
 mod error;
 mod outbound_proxy;
 mod request;
+mod route_aware_client_pool;
 mod transport;
 
 pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store;
@@ -36,6 +37,8 @@ pub use crate::request::Request;
 pub use crate::request::RequestBody;
 pub use crate::request::RequestCompression;
 pub use crate::request::Response;
+pub use crate::route_aware_client_pool::RouteAwareClientPool;
+pub use crate::route_aware_client_pool::RouteAwareClientPoolError;
 pub use crate::transport::ByteStream;
 pub use crate::transport::HttpTransport;
 pub use crate::transport::ReqwestTransport;

codex-rs/http-client/src/outbound_proxy.rs

@@ -12,6 +12,7 @@ use std::sync::Mutex;
 use std::sync::OnceLock;
 use std::time::Duration;
 use std::time::Instant;
+use tokio::sync::Semaphore;
 
 use crate::custom_ca::BuildCustomCaTransportError;
 use crate::custom_ca::build_reqwest_client_with_custom_ca;
@@ -25,6 +26,7 @@ use thiserror::Error;
 const SYSTEM_PROXY_SUCCESS_CACHE_TTL: Duration = Duration::from_secs(60);
 const SYSTEM_PROXY_UNAVAILABLE_CACHE_TTL: Duration = Duration::from_secs(5);
 const SYSTEM_PROXY_CACHE_MAX_ENTRIES: usize = 256;
+static ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT: Semaphore = Semaphore::const_new(1);
 
 #[cfg(target_os = "macos")]
 mod macos;
@@ -102,7 +104,7 @@ pub enum OutboundProxyPolicy {
 ///
 /// `TransportDefault` delegates environment-proxy handling to the underlying transport. Proxy
 /// URLs are intentionally redacted from `Debug` output because they may contain credentials.
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Clone, Hash, PartialEq, Eq)]
 pub enum OutboundProxyRoute {
     /// Preserve the underlying transport's existing proxy behavior.
     TransportDefault,
@@ -159,6 +161,33 @@ impl HttpClientFactory {
         )
     }
 
+    /// Resolves the proxy route for a concrete destination without blocking a Tokio worker.
+    pub async fn resolve_proxy_route_async(
+        &self,
+        request_url: String,
+    ) -> io::Result<OutboundProxyRoute> {
+        if matches!(
+            self.outbound_proxy_policy,
+            OutboundProxyPolicy::ReqwestDefault
+        ) {
+            return Ok(OutboundProxyRoute::TransportDefault);
+        }
+
+        let permit = ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT
+            .acquire()
+            .await
+            .map_err(io::Error::other)?;
+        let factory = self.clone();
+        tokio::task::spawn_blocking(move || {
+            // Keep the permit with the blocking task: cancelling the caller must not allow a
+            // second PAC/WinHTTP lookup to start while this one is still running.
+            let _permit = permit;
+            factory.resolve_proxy_route(&request_url)
+        })
+        .await
+        .map_err(io::Error::other)
+    }
+
     /// Builds an HTTP client for a concrete outbound route.
     pub fn build_client(
         &self,
@@ -183,6 +212,22 @@ impl HttpClientFactory {
             self.outbound_proxy_policy,
         )
     }
+
+    pub(crate) fn build_reqwest_client_for_resolved_route(
+        &self,
+        builder: reqwest::ClientBuilder,
+        route_class: ClientRouteClass,
+        route: &OutboundProxyRoute,
+    ) -> Result<reqwest::Client, BuildRouteAwareHttpClientError> {
+        let builder = match route {
+            OutboundProxyRoute::TransportDefault => builder,
+            OutboundProxyRoute::Direct => builder.no_proxy(),
+            OutboundProxyRoute::Proxy { url } => {
+                configure_concrete_proxy(builder, route_class, url, /*no_proxy*/ None)?
+            }
+        };
+        build_reqwest_client_with_custom_ca(builder).map_err(Into::into)
+    }
 }
 
 fn resolve_proxy_route(

codex-rs/http-client/src/outbound_proxy_tests.rs

@@ -10,6 +10,48 @@ struct MapEnv {
     values: HashMap<String, String>,
 }
 
+fn spawn_proxy_listener() -> (
+    std::net::SocketAddr,
+    std::thread::JoinHandle<Option<String>>,
+) {
+    let listener =
+        std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
+    let proxy_addr = listener
+        .local_addr()
+        .expect("local proxy listener should have an address");
+    listener
+        .set_nonblocking(true)
+        .expect("proxy listener should become nonblocking");
+    let proxy_thread = std::thread::spawn(move || {
+        let deadline = Instant::now() + Duration::from_secs(2);
+        loop {
+            match listener.accept() {
+                Ok((mut stream, _)) => {
+                    stream
+                        .set_read_timeout(Some(Duration::from_secs(2)))
+                        .expect("proxy stream should get a read timeout");
+                    let mut buffer = [0_u8; 4096];
+                    let size = stream.read(&mut buffer).expect("proxy should read request");
+                    stream
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
+                        )
+                        .expect("proxy should write response");
+                    break Some(String::from_utf8_lossy(&buffer[..size]).into_owned());
+                }
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    if Instant::now() >= deadline {
+                        break None;
+                    }
+                    std::thread::sleep(Duration::from_millis(10));
+                }
+                Err(error) => panic!("proxy should accept a request: {error}"),
+            }
+        }
+    });
+    (proxy_addr, proxy_thread)
+}
+
 #[test]
 fn websocket_route_uses_http_equivalent_for_system_resolution() {
     let route = resolve_proxy_route(
@@ -97,20 +139,7 @@ fn environment_fallback_reads_injected_proxy_environment() {
 
 #[tokio::test]
 async fn enabled_environment_proxy_routes_request_through_proxy() {
-    let listener =
-        std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
-    let proxy_addr = listener
-        .local_addr()
-        .expect("local proxy listener should have an address");
-    let proxy_thread = std::thread::spawn(move || {
-        let (mut stream, _) = listener.accept().expect("proxy should accept a request");
-        let mut buffer = [0_u8; 4096];
-        let size = stream.read(&mut buffer).expect("proxy should read request");
-        stream
-            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
-            .expect("proxy should write response");
-        String::from_utf8_lossy(&buffer[..size]).into_owned()
-    });
+    let (proxy_addr, proxy_thread) = spawn_proxy_listener();
     let env = MapEnv {
         values: HashMap::from([("HTTP_PROXY".to_string(), format!("http://{proxy_addr}"))]),
     };
@@ -134,7 +163,10 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
         .send()
         .await
         .expect("request should use local proxy");
-    let proxy_request = proxy_thread.join().expect("proxy thread should finish");
+    let proxy_request = proxy_thread
+        .join()
+        .expect("proxy thread should finish")
+        .expect("proxy should receive request before timeout");
 
     assert_eq!(response.status(), reqwest::StatusCode::OK);
     assert_eq!(
@@ -143,6 +175,41 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
     );
 }
 
+#[tokio::test]
+async fn route_aware_pool_uses_respect_system_proxy_route_for_exact_url() {
+    let (proxy_addr, proxy_thread) = spawn_proxy_listener();
+    let request_url = "http://route-aware-proxy.test/proxy-check?pac=exact";
+    cache_system_proxy_decision(
+        request_url,
+        SystemProxyDecision::Proxy {
+            url: format!("http://{proxy_addr}"),
+        },
+    );
+    let pool = crate::RouteAwareClientPool::new(
+        HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
+        ClientRouteClass::Api,
+    );
+
+    let client = tokio::time::timeout(Duration::from_secs(2), pool.client_for_url(request_url))
+        .await
+        .expect("route resolution should finish")
+        .expect("route-aware client should build");
+    let response = tokio::time::timeout(Duration::from_secs(2), client.get(request_url).send())
+        .await
+        .expect("proxy request should finish")
+        .expect("request should use local proxy");
+    let proxy_request = proxy_thread
+        .join()
+        .expect("proxy thread should finish")
+        .expect("proxy should receive request before timeout");
+
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+    assert_eq!(
+        proxy_request.lines().next(),
+        Some("GET http://route-aware-proxy.test/proxy-check?pac=exact HTTP/1.1")
+    );
+}
+
 #[test]
 fn parses_pac_proxy_tokens() {
     assert_eq!(

codex-rs/http-client/src/route_aware_client_pool.rs

@@ -0,0 +1,140 @@
+use std::collections::HashMap;
+use std::fmt;
+use std::future::Future;
+use std::io;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use crate::BuildRouteAwareHttpClientError;
+use crate::ClientRouteClass;
+use crate::HttpClient;
+use crate::HttpClientFactory;
+use crate::OutboundProxyRoute;
+use crate::with_chatgpt_cloudflare_cookie_store;
+
+const MAX_CACHED_ROUTES: usize = 16;
+
+/// Reuses reqwest clients by resolved route while evaluating the exact URL of every request.
+///
+/// PAC and system-proxy selection remains request-specific, but requests that resolve to the same
+/// direct or proxy route share one connection pool. Named constructors apply transport-wide
+/// settings such as the ChatGPT Cloudflare cookie store without exposing reqwest to callers.
+#[derive(Clone)]
+pub struct RouteAwareClientPool {
+    http_client_factory: HttpClientFactory,
+    route_class: ClientRouteClass,
+    builder_factory: Arc<dyn Fn() -> reqwest::ClientBuilder + Send + Sync>,
+    clients: Arc<Mutex<HashMap<OutboundProxyRoute, HttpClient>>>,
+}
+
+impl fmt::Debug for RouteAwareClientPool {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("RouteAwareClientPool")
+            .field("http_client_factory", &self.http_client_factory)
+            .field("route_class", &self.route_class)
+            .finish_non_exhaustive()
+    }
+}
+
+/// Error returned when selecting a route or constructing its pooled HTTP client.
+#[derive(Debug, thiserror::Error)]
+pub enum RouteAwareClientPoolError {
+    #[error("failed to resolve the outbound proxy route: {0}")]
+    Resolve(#[source] io::Error),
+    #[error(transparent)]
+    Build(#[from] BuildRouteAwareHttpClientError),
+}
+
+impl RouteAwareClientPool {
+    /// Creates a pool with the shared default HTTP transport settings.
+    pub fn new(http_client_factory: HttpClientFactory, route_class: ClientRouteClass) -> Self {
+        Self::with_builder_factory(http_client_factory, route_class, reqwest::Client::builder)
+    }
+
+    /// Creates a pool that retains the Cloudflare cookies required by ChatGPT endpoints.
+    pub fn with_chatgpt_cloudflare_cookies(
+        http_client_factory: HttpClientFactory,
+        route_class: ClientRouteClass,
+    ) -> Self {
+        Self::with_builder_factory(http_client_factory, route_class, || {
+            with_chatgpt_cloudflare_cookie_store(reqwest::Client::builder())
+        })
+    }
+
+    pub(crate) fn with_builder_factory(
+        http_client_factory: HttpClientFactory,
+        route_class: ClientRouteClass,
+        builder_factory: impl Fn() -> reqwest::ClientBuilder + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            http_client_factory,
+            route_class,
+            builder_factory: Arc::new(builder_factory),
+            clients: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    /// Returns a pooled client configured for the proxy route selected for `request_url`.
+    pub async fn client_for_url(
+        &self,
+        request_url: &str,
+    ) -> Result<HttpClient, RouteAwareClientPoolError> {
+        let http_client_factory = self.http_client_factory.clone();
+        self.client_for_url_with_resolver(request_url, move |request_url| async move {
+            http_client_factory
+                .resolve_proxy_route_async(request_url)
+                .await
+        })
+        .await
+    }
+
+    async fn client_for_url_with_resolver<F, Fut>(
+        &self,
+        request_url: &str,
+        resolve_route: F,
+    ) -> Result<HttpClient, RouteAwareClientPoolError>
+    where
+        F: FnOnce(String) -> Fut,
+        Fut: Future<Output = io::Result<OutboundProxyRoute>>,
+    {
+        let route = resolve_route(request_url.to_string())
+            .await
+            .map_err(RouteAwareClientPoolError::Resolve)?;
+        let clients = match self.clients.lock() {
+            Ok(clients) => clients,
+            Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
+        };
+        if let Some(client) = clients.get(&route) {
+            return Ok(client.clone());
+        }
+        drop(clients);
+
+        let client = self
+            .http_client_factory
+            .build_reqwest_client_for_resolved_route(
+                (self.builder_factory)(),
+                self.route_class,
+                &route,
+            )?;
+        let client = HttpClient::new(client);
+        let mut clients = match self.clients.lock() {
+            Ok(clients) => clients,
+            Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
+        };
+        if let Some(existing_client) = clients.get(&route) {
+            return Ok(existing_client.clone());
+        }
+        if clients.len() >= MAX_CACHED_ROUTES
+            && let Some(route_to_evict) = clients.keys().next().cloned()
+        {
+            clients.remove(&route_to_evict);
+        }
+        clients.insert(route, client.clone());
+        Ok(client)
+    }
+}
+
+#[cfg(test)]
+#[path = "route_aware_client_pool_tests.rs"]
+mod tests;

codex-rs/http-client/src/route_aware_client_pool_tests.rs

@@ -0,0 +1,154 @@
+use std::collections::HashMap;
+use std::io;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+
+use pretty_assertions::assert_eq;
+
+use super::*;
+use crate::OutboundProxyPolicy;
+
+#[tokio::test]
+async fn forwards_exact_urls_and_reuses_clients_by_resolved_route() {
+    let builder_count = Arc::new(AtomicUsize::new(0));
+    let observed_builder_count = Arc::clone(&builder_count);
+    let pool = RouteAwareClientPool::with_builder_factory(
+        HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
+        ClientRouteClass::Api,
+        move || {
+            observed_builder_count.fetch_add(1, Ordering::SeqCst);
+            reqwest::Client::builder()
+        },
+    );
+
+    let direct_url = "https://example.com/first?target=direct";
+    let same_route_url = "https://example.com/second?target=direct%202";
+    let proxy_url = "https://example.com/third?target=proxy";
+    let resolver = FakeRouteResolver::new(HashMap::from([
+        (direct_url.to_string(), OutboundProxyRoute::Direct),
+        (same_route_url.to_string(), OutboundProxyRoute::Direct),
+        (
+            proxy_url.to_string(),
+            OutboundProxyRoute::Proxy {
+                url: "http://proxy.example".to_string(),
+            },
+        ),
+    ]));
+
+    resolve_with(&pool, &resolver, direct_url)
+        .await
+        .expect("first client should build");
+    resolve_with(&pool, &resolver, same_route_url)
+        .await
+        .expect("second client should reuse the route");
+    resolve_with(&pool, &resolver, proxy_url)
+        .await
+        .expect("proxy client should build separately");
+
+    assert_eq!(builder_count.load(Ordering::SeqCst), 2);
+    assert_eq!(
+        resolver.observed_urls(),
+        vec![
+            direct_url.to_string(),
+            same_route_url.to_string(),
+            proxy_url.to_string(),
+        ]
+    );
+}
+
+#[tokio::test]
+async fn bounds_cached_routes_and_rebuilds_an_evicted_route() {
+    let builder_count = Arc::new(AtomicUsize::new(0));
+    let observed_builder_count = Arc::clone(&builder_count);
+    let pool = RouteAwareClientPool::with_builder_factory(
+        HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
+        ClientRouteClass::Api,
+        move || {
+            observed_builder_count.fetch_add(1, Ordering::SeqCst);
+            reqwest::Client::builder()
+        },
+    );
+    let routes = (0..=MAX_CACHED_ROUTES)
+        .map(|index| {
+            (
+                format!("https://target-{index}.example"),
+                OutboundProxyRoute::Proxy {
+                    url: format!("http://proxy-{index}.example"),
+                },
+            )
+        })
+        .collect::<HashMap<_, _>>();
+    let resolver = FakeRouteResolver::new(routes.clone());
+
+    for request_url in routes.keys() {
+        resolve_with(&pool, &resolver, request_url)
+            .await
+            .expect("client should build");
+    }
+    let evicted_route = {
+        let clients = pool.clients.lock().expect("client cache lock");
+        assert_eq!(clients.len(), MAX_CACHED_ROUTES);
+        routes
+            .iter()
+            .find(|(_, route)| !clients.contains_key(*route))
+            .map(|(request_url, _)| request_url.clone())
+            .expect("one route should have been evicted")
+    };
+
+    resolve_with(&pool, &resolver, &evicted_route)
+        .await
+        .expect("evicted client should rebuild");
+
+    assert_eq!(builder_count.load(Ordering::SeqCst), MAX_CACHED_ROUTES + 2);
+    assert_eq!(
+        pool.clients.lock().expect("client cache lock").len(),
+        MAX_CACHED_ROUTES
+    );
+}
+
+#[derive(Clone)]
+struct FakeRouteResolver {
+    routes: Arc<HashMap<String, OutboundProxyRoute>>,
+    observed_urls: Arc<Mutex<Vec<String>>>,
+}
+
+impl FakeRouteResolver {
+    fn new(routes: HashMap<String, OutboundProxyRoute>) -> Self {
+        Self {
+            routes: Arc::new(routes),
+            observed_urls: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
+    async fn resolve(&self, request_url: String) -> io::Result<OutboundProxyRoute> {
+        self.observed_urls
+            .lock()
+            .expect("observed URL lock")
+            .push(request_url.clone());
+        self.routes
+            .get(&request_url)
+            .cloned()
+            .ok_or_else(|| io::Error::other(format!("no route for {request_url}")))
+    }
+
+    fn observed_urls(&self) -> Vec<String> {
+        self.observed_urls
+            .lock()
+            .expect("observed URL lock")
+            .clone()
+    }
+}
+
+async fn resolve_with(
+    pool: &RouteAwareClientPool,
+    resolver: &FakeRouteResolver,
+    request_url: &str,
+) -> Result<HttpClient, RouteAwareClientPoolError> {
+    let resolver = resolver.clone();
+    pool.client_for_url_with_resolver(request_url, move |request_url| async move {
+        resolver.resolve(request_url).await
+    })
+    .await
+}