diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ea08c04e96..ce08f95f99 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2274,10 +2274,11 @@ dependencies = [ "codex-login", "codex-model-provider", "codex-protocol", + "http 1.4.0", "pretty_assertions", - "reqwest 0.12.28", "serde", "serde_json", + "url", ] [[package]] @@ -2418,6 +2419,7 @@ dependencies = [ "codex-backend-client", "codex-config", "codex-core", + "codex-http-client", "codex-login", "codex-otel", "codex-protocol", @@ -2472,6 +2474,7 @@ dependencies = [ "codex-api", "codex-backend-client", "codex-git-utils", + "codex-http-client", "serde", "serde_json", "thiserror 2.0.18", diff --git a/codex-rs/app-server/src/config_manager.rs b/codex-rs/app-server/src/config_manager.rs index d3d7609d65..ee47f50cd5 100644 --- a/codex-rs/app-server/src/config_manager.rs +++ b/codex-rs/app-server/src/config_manager.rs @@ -95,9 +95,14 @@ impl ConfigManager { &self, auth_manager: Arc, chatgpt_base_url: String, + http_client_factory: codex_http_client::HttpClientFactory, ) { - let loader = - cloud_config_bundle_loader(auth_manager, chatgpt_base_url, self.codex_home.clone()); + let loader = cloud_config_bundle_loader( + auth_manager, + chatgpt_base_url, + self.codex_home.clone(), + http_client_factory, + ); if let Ok(mut guard) = self.cloud_config_bundle.write() { *guard = loader; } else { diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 91da0c1645..336e5d943a 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -505,8 +505,11 @@ pub async fn run_main_with_transport_options( .replace_thread_config_loader(Arc::clone(&discovered_thread_config_loader)); let auth_manager = AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await; - config_manager - .replace_cloud_config_bundle_loader(auth_manager, config.chatgpt_base_url); + config_manager.replace_cloud_config_bundle_loader( + auth_manager, + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ); } Err(err) => { warn!(error = %err, "Failed to preload config for cloud config bundle"); diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index f4acc5a8e4..64ac28655b 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -458,7 +458,7 @@ impl AccountRequestProcessor { let outgoing_clone = self.outgoing.clone(); let config_manager = self.config_manager.clone(); let thread_manager = Arc::clone(&self.thread_manager); - let chatgpt_base_url = self.config.chatgpt_base_url.clone(); + let config = Arc::clone(&self.config); let active_login = self.active_login.clone(); let auth_url = server.auth_url.clone(); tokio::spawn(async move { @@ -480,7 +480,7 @@ impl AccountRequestProcessor { &outgoing_clone, config_manager, thread_manager, - chatgpt_base_url, + config, login_id, success, error_msg, @@ -537,7 +537,7 @@ impl AccountRequestProcessor { let outgoing_clone = self.outgoing.clone(); let config_manager = self.config_manager.clone(); let thread_manager = Arc::clone(&self.thread_manager); - let chatgpt_base_url = self.config.chatgpt_base_url.clone(); + let config = Arc::clone(&self.config); let active_login = self.active_login.clone(); tokio::spawn(async move { let (success, error_msg) = tokio::select! { @@ -556,7 +556,7 @@ impl AccountRequestProcessor { &outgoing_clone, config_manager, thread_manager, - chatgpt_base_url, + config, login_id, success, error_msg, @@ -671,6 +671,7 @@ impl AccountRequestProcessor { self.config_manager.replace_cloud_config_bundle_loader( self.auth_manager.clone(), self.config.chatgpt_base_url.clone(), + self.config.http_client_factory(), ); self.config_manager .sync_default_client_residency_requirement() @@ -709,7 +710,7 @@ impl AccountRequestProcessor { outgoing: &OutgoingMessageSender, config_manager: ConfigManager, thread_manager: Arc, - chatgpt_base_url: String, + config: Arc, login_id: Uuid, success: bool, error_msg: Option, @@ -726,8 +727,11 @@ impl AccountRequestProcessor { if success { let auth_manager = thread_manager.auth_manager(); auth_manager.reload().await; - config_manager - .replace_cloud_config_bundle_loader(auth_manager.clone(), chatgpt_base_url); + config_manager.replace_cloud_config_bundle_loader( + auth_manager.clone(), + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ); config_manager .sync_default_client_residency_requirement() .await; @@ -933,8 +937,12 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; + let client = BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + ) + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; let (response, detailed_rate_limit_reset_credits) = tokio::join!( client.get_rate_limits_with_reset_credits(), @@ -1003,8 +1011,12 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; + let client = BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + ) + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; let profile = tokio::time::timeout( ACCOUNT_TOKEN_USAGE_FETCH_TIMEOUT, client.get_token_usage_profile(), @@ -1030,8 +1042,12 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; + let client = BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + ) + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; let messages = tokio::time::timeout( ACCOUNT_WORKSPACE_MESSAGES_FETCH_TIMEOUT, client.list_workspace_messages(), @@ -1118,8 +1134,12 @@ impl AccountRequestProcessor { )); } - let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; + let client = BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + ) + .map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?; match client .send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type)) diff --git a/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs b/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs index 919d30d770..18d953674c 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs @@ -109,8 +109,12 @@ impl AccountRequestProcessor { )); } - BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}"))) + BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + ) + .map_err(|err| internal_error(format!("failed to construct backend client: {err}"))) } } diff --git a/codex-rs/backend-client/Cargo.toml b/codex-rs/backend-client/Cargo.toml index 4ff5fb9b86..4ec3a16339 100644 --- a/codex-rs/backend-client/Cargo.toml +++ b/codex-rs/backend-client/Cargo.toml @@ -16,7 +16,8 @@ workspace = true anyhow = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" -reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +http = { workspace = true } +url = { workspace = true } codex-backend-openapi-models = { path = "../codex-backend-openapi-models" } codex-api = { workspace = true } codex-http-client = { workspace = true } diff --git a/codex-rs/backend-client/src/client.rs b/codex-rs/backend-client/src/client.rs index a5db25618b..97f8c811ee 100644 --- a/codex-rs/backend-client/src/client.rs +++ b/codex-rs/backend-client/src/client.rs @@ -9,8 +9,10 @@ use crate::types::TokenUsageProfile; use crate::types::TurnAttemptsSiblingTurnsResponse; use anyhow::Result; use codex_api::SharedAuthProvider; -use codex_http_client::build_reqwest_client_with_custom_ca; -use codex_http_client::with_chatgpt_cloudflare_cookie_store; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::RequestBuilder; +use codex_http_client::RouteAwareClientPool; use codex_login::CodexAuth; use codex_login::default_client::get_codex_user_agent; use codex_protocol::account::PlanType as AccountPlanType; @@ -19,13 +21,14 @@ use codex_protocol::protocol::RateLimitReachedType; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::SpendControlLimitSnapshot; -use reqwest::StatusCode; -use reqwest::header::CACHE_CONTROL; -use reqwest::header::CONTENT_TYPE; -use reqwest::header::HeaderMap; -use reqwest::header::HeaderName; -use reqwest::header::HeaderValue; -use reqwest::header::USER_AGENT; +use http::Method; +use http::StatusCode; +use http::header::CACHE_CONTROL; +use http::header::CONTENT_TYPE; +use http::header::HeaderMap; +use http::header::HeaderName; +use http::header::HeaderValue; +use http::header::USER_AGENT; use serde::Serialize; use serde::de::DeserializeOwned; use std::fmt; @@ -123,7 +126,7 @@ impl PathStyle { #[derive(Clone)] pub struct Client { base_url: String, - http: reqwest::Client, + http: RouteAwareClientPool, auth_provider: SharedAuthProvider, user_agent: Option, chatgpt_account_id: Option, @@ -148,7 +151,10 @@ impl fmt::Debug for Client { } impl Client { - pub fn new(base_url: impl Into) -> Result { + pub fn new( + base_url: impl Into, + http_client_factory: HttpClientFactory, + ) -> Result { let mut base_url = base_url.into(); // Normalize common ChatGPT hostnames to include /backend-api so we hit the WHAM paths. // Also trim trailing slashes for consistent URL building. @@ -161,9 +167,10 @@ impl Client { { base_url = format!("{base_url}/backend-api"); } - let http = build_reqwest_client_with_custom_ca(with_chatgpt_cloudflare_cookie_store( - reqwest::Client::builder(), - ))?; + let http = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + http_client_factory, + ClientRouteClass::Api, + ); let path_style = PathStyle::from_base_url(&base_url); Ok(Self { base_url, @@ -176,8 +183,12 @@ impl Client { }) } - pub fn from_auth(base_url: impl Into, auth: &CodexAuth) -> Result { - Ok(Self::new(base_url)? + pub fn from_auth( + base_url: impl Into, + auth: &CodexAuth, + http_client_factory: HttpClientFactory, + ) -> Result { + Ok(Self::new(base_url, http_client_factory)? .with_user_agent(get_codex_user_agent()) .with_auth_provider(codex_model_provider::auth_provider_from_auth(auth))) } @@ -231,9 +242,13 @@ impl Client { h } + async fn request(&self, method: Method, url: &str) -> Result { + Ok(self.http.client_for_url(url).await?.request(method, url)) + } + async fn exec_request( &self, - req: reqwest::RequestBuilder, + req: RequestBuilder, method: &str, url: &str, ) -> Result<(String, String)> { @@ -254,7 +269,7 @@ impl Client { async fn exec_request_detailed( &self, - req: reqwest::RequestBuilder, + req: RequestBuilder, method: &str, url: &str, ) -> std::result::Result<(String, String), RequestError> { @@ -306,14 +321,20 @@ impl Client { PathStyle::CodexApi => format!("{}/api/codex/accounts/check", self.base_url), PathStyle::ChatGptApi => format!("{}/wham/accounts/check", self.base_url), }; - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } pub async fn get_token_usage_profile(&self) -> Result { let url = self.token_usage_profile_url(); - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } @@ -331,8 +352,9 @@ impl Client { ) -> std::result::Result<(), RequestError> { let url = self.send_add_credits_nudge_email_url(); let req = self - .http - .post(&url) + .request(Method::POST, &url) + .await + .map_err(RequestError::from)? .headers(self.headers()) .header(CONTENT_TYPE, HeaderValue::from_static("application/json")) .json(&SendAddCreditsNudgeEmailRequest { credit_type }); @@ -347,33 +369,47 @@ impl Client { environment_id: Option<&str>, cursor: Option<&str>, ) -> Result { + let url = self.list_tasks_url(limit, task_filter, environment_id, cursor)?; + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); + let (body, ct) = self.exec_request(req, "GET", &url).await?; + self.decode_json::(&url, &ct, &body) + } + + fn list_tasks_url( + &self, + limit: Option, + task_filter: Option<&str>, + environment_id: Option<&str>, + cursor: Option<&str>, + ) -> Result { let url = match self.path_style { PathStyle::CodexApi => format!("{}/api/codex/tasks/list", self.base_url), PathStyle::ChatGptApi => format!("{}/wham/tasks/list", self.base_url), }; - let req = self.http.get(&url).headers(self.headers()); - let req = if let Some(lim) = limit { - req.query(&[("limit", lim)]) - } else { - req - }; - let req = if let Some(tf) = task_filter { - req.query(&[("task_filter", tf)]) - } else { - req - }; - let req = if let Some(c) = cursor { - req.query(&[("cursor", c)]) - } else { - req - }; - let req = if let Some(id) = environment_id { - req.query(&[("environment_id", id)]) - } else { - req - }; - let (body, ct) = self.exec_request(req, "GET", &url).await?; - self.decode_json::(&url, &ct, &body) + if limit.is_none() && task_filter.is_none() && environment_id.is_none() && cursor.is_none() + { + return Ok(url); + } + let mut url = url::Url::parse(&url)?; + { + let mut query = url.query_pairs_mut(); + if let Some(limit) = limit { + query.append_pair("limit", &limit.to_string()); + } + if let Some(task_filter) = task_filter { + query.append_pair("task_filter", task_filter); + } + if let Some(cursor) = cursor { + query.append_pair("cursor", cursor); + } + if let Some(environment_id) = environment_id { + query.append_pair("environment_id", environment_id); + } + } + Ok(url.to_string()) } pub async fn get_task_details(&self, task_id: &str) -> Result { @@ -389,7 +425,10 @@ impl Client { PathStyle::CodexApi => format!("{}/api/codex/tasks/{}", self.base_url, task_id), PathStyle::ChatGptApi => format!("{}/wham/tasks/{}", self.base_url, task_id), }; - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; let parsed: CodeTaskDetailsResponse = self.decode_json(&url, &ct, &body)?; Ok((parsed, body, ct)) @@ -410,7 +449,10 @@ impl Client { self.base_url, task_id, turn_id ), }; - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json::(&url, &ct, &body) } @@ -426,7 +468,11 @@ impl Client { PathStyle::CodexApi => format!("{}/api/codex/config/bundle", self.base_url), PathStyle::ChatGptApi => format!("{}/wham/config/bundle", self.base_url), }; - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await + .map_err(RequestError::from)? + .headers(self.headers()); let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?; self.decode_json::(&url, &ct, &body) .map_err(RequestError::from) @@ -437,8 +483,9 @@ impl Client { ) -> std::result::Result { let url = self.workspace_messages_url(); let req = self - .http - .get(&url) + .request(Method::GET, &url) + .await + .map_err(RequestError::from)? .headers(self.headers()) .header(CACHE_CONTROL, HeaderValue::from_static("no-store")); let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?; @@ -454,8 +501,8 @@ impl Client { PathStyle::ChatGptApi => format!("{}/wham/tasks", self.base_url), }; let req = self - .http - .post(&url) + .request(Method::POST, &url) + .await? .headers(self.headers()) .header(CONTENT_TYPE, HeaderValue::from_static("application/json")) .json(&request_body); @@ -974,10 +1021,39 @@ mod tests { ); } + #[test] + fn list_tasks_url_omits_empty_query_and_encodes_all_parameters() { + let client = test_client("https://example.test", PathStyle::CodexApi); + + assert_eq!( + client + .list_tasks_url( + /*limit*/ None, /*task_filter*/ None, /*environment_id*/ None, + /*cursor*/ None, + ) + .unwrap(), + "https://example.test/api/codex/tasks/list" + ); + assert_eq!( + client + .list_tasks_url( + /*limit*/ Some(10), + /*task_filter*/ Some("mine / shared"), + /*environment_id*/ Some("env&one"), + /*cursor*/ Some("next=page"), + ) + .unwrap(), + "https://example.test/api/codex/tasks/list?limit=10&task_filter=mine+%2F+shared&cursor=next%3Dpage&environment_id=env%26one" + ); + } + fn test_client(base_url: &str, path_style: PathStyle) -> Client { Client { base_url: base_url.to_string(), - http: reqwest::Client::new(), + http: RouteAwareClientPool::new( + HttpClientFactory::new(codex_http_client::OutboundProxyPolicy::ReqwestDefault), + ClientRouteClass::Api, + ), auth_provider: codex_model_provider::unauthenticated_auth_provider(), user_agent: None, chatgpt_account_id: None, diff --git a/codex-rs/backend-client/src/client/rate_limit_resets.rs b/codex-rs/backend-client/src/client/rate_limit_resets.rs index bed56ba648..93145b02fc 100644 --- a/codex-rs/backend-client/src/client/rate_limit_resets.rs +++ b/codex-rs/backend-client/src/client/rate_limit_resets.rs @@ -7,8 +7,9 @@ use crate::types::RateLimitResetCreditsDetails; use crate::types::RateLimitStatusWithResetCredits; use crate::types::RateLimitsWithResetCredits; use anyhow::Result; -use reqwest::header::CONTENT_TYPE; -use reqwest::header::HeaderValue; +use http::Method; +use http::header::CONTENT_TYPE; +use http::header::HeaderValue; use serde::Serialize; #[derive(Serialize)] @@ -29,14 +30,20 @@ impl Client { pub(super) async fn get_rate_limit_status(&self) -> Result { let url = self.rate_limit_status_url(); - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } pub async fn list_rate_limit_reset_credits(&self) -> Result { let url = self.rate_limit_reset_credits_url(); - let req = self.http.get(&url).headers(self.headers()); + let req = self + .request(Method::GET, &url) + .await? + .headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } @@ -65,8 +72,8 @@ impl Client { ) -> Result { let url = self.consume_rate_limit_reset_credit_url(); let req = self - .http - .post(&url) + .request(Method::POST, &url) + .await? .headers(self.headers()) .header(CONTENT_TYPE, HeaderValue::from_static("application/json")) .json(&ConsumeRateLimitResetCreditRequest { diff --git a/codex-rs/backend-client/src/client/rate_limit_resets_tests.rs b/codex-rs/backend-client/src/client/rate_limit_resets_tests.rs index f706be5da9..55703caab5 100644 --- a/codex-rs/backend-client/src/client/rate_limit_resets_tests.rs +++ b/codex-rs/backend-client/src/client/rate_limit_resets_tests.rs @@ -138,7 +138,12 @@ fn rate_limit_reset_contract_uses_expected_paths_and_payloads() { fn test_client(base_url: &str, path_style: PathStyle) -> Client { Client { base_url: base_url.to_string(), - http: reqwest::Client::new(), + http: codex_http_client::RouteAwareClientPool::new( + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ), + codex_http_client::ClientRouteClass::Api, + ), auth_provider: codex_model_provider::unauthenticated_auth_provider(), user_agent: None, chatgpt_account_id: None, diff --git a/codex-rs/cloud-config/Cargo.toml b/codex-rs/cloud-config/Cargo.toml index 363cb21b6f..ecb33d34be 100644 --- a/codex-rs/cloud-config/Cargo.toml +++ b/codex-rs/cloud-config/Cargo.toml @@ -12,6 +12,7 @@ base64 = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-backend-client = { workspace = true } codex-config = { workspace = true } +codex-http-client = { workspace = true } codex-core = { workspace = true } codex-login = { workspace = true } codex-otel = { workspace = true } diff --git a/codex-rs/cloud-config/src/backend.rs b/codex-rs/cloud-config/src/backend.rs index cb99316a70..6c1ecbc054 100644 --- a/codex-rs/cloud-config/src/backend.rs +++ b/codex-rs/cloud-config/src/backend.rs @@ -6,6 +6,7 @@ use codex_config::CloudConfigFragment; use codex_config::CloudConfigTomlBundle; use codex_config::CloudRequirementsFragment; use codex_config::CloudRequirementsTomlBundle; +use codex_http_client::HttpClientFactory; use codex_login::CodexAuth; use std::future::Future; @@ -46,24 +47,32 @@ pub(crate) trait BundleClient: Send + Sync { pub(crate) struct BackendBundleClient { base_url: String, + http_client_factory: HttpClientFactory, } impl BackendBundleClient { - pub(crate) fn new(base_url: String) -> Self { - Self { base_url } + pub(crate) fn new(base_url: String, http_client_factory: HttpClientFactory) -> Self { + Self { + base_url, + http_client_factory, + } } } impl BundleClient for BackendBundleClient { async fn get_bundle(&self, auth: &CodexAuth) -> Result { - let client = BackendClient::from_auth(self.base_url.clone(), auth) - .inspect_err(|err| { - tracing::warn!( - error = %err, - "Failed to construct backend client for cloud config bundle" - ); - }) - .map_err(|_| BundleRequestError::Retryable(RetryableFailureKind::BackendClientInit))?; + let client = BackendClient::from_auth( + self.base_url.clone(), + auth, + self.http_client_factory.clone(), + ) + .inspect_err(|err| { + tracing::warn!( + error = %err, + "Failed to construct backend client for cloud config bundle" + ); + }) + .map_err(|_| BundleRequestError::Retryable(RetryableFailureKind::BackendClientInit))?; let response = client .get_config_bundle() diff --git a/codex-rs/cloud-config/src/bundle_loader.rs b/codex-rs/cloud-config/src/bundle_loader.rs index e4266c6195..837001d015 100644 --- a/codex-rs/cloud-config/src/bundle_loader.rs +++ b/codex-rs/cloud-config/src/bundle_loader.rs @@ -5,6 +5,8 @@ use codex_config::CloudConfigBundleLoadError; use codex_config::CloudConfigBundleLoadErrorCode; use codex_config::CloudConfigBundleLoader; use codex_config::types::AuthCredentialsStoreMode; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_login::AuthKeyringBackendKind; use codex_login::AuthManager; use codex_login::AuthRouteConfig; @@ -23,10 +25,14 @@ pub fn cloud_config_bundle_loader( auth_manager: Arc, chatgpt_base_url: String, codex_home: PathBuf, + http_client_factory: HttpClientFactory, ) -> CloudConfigBundleLoader { let service = CloudConfigBundleService::new( auth_manager, - Arc::new(BackendBundleClient::new(chatgpt_base_url)), + Arc::new(BackendBundleClient::new( + chatgpt_base_url, + http_client_factory, + )), codex_home, CLOUD_CONFIG_BUNDLE_TIMEOUT, ); @@ -61,6 +67,10 @@ pub async fn cloud_config_bundle_loader_for_storage( chatgpt_base_url: String, auth_route_config: Option, ) -> CloudConfigBundleLoader { + let http_client_factory = auth_route_config.as_ref().map_or_else( + || HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + |config| config.http_client_factory().clone(), + ); let auth_manager = AuthManager::shared( codex_home.clone(), enable_codex_api_key_env, @@ -71,5 +81,10 @@ pub async fn cloud_config_bundle_loader_for_storage( auth_route_config, ) .await; - cloud_config_bundle_loader(auth_manager, chatgpt_base_url, codex_home) + cloud_config_bundle_loader( + auth_manager, + chatgpt_base_url, + codex_home, + http_client_factory, + ) } diff --git a/codex-rs/cloud-tasks-client/Cargo.toml b/codex-rs/cloud-tasks-client/Cargo.toml index dc07550d08..8efd78cb22 100644 --- a/codex-rs/cloud-tasks-client/Cargo.toml +++ b/codex-rs/cloud-tasks-client/Cargo.toml @@ -19,6 +19,7 @@ chrono = { workspace = true, features = ["serde"] } codex-api = { workspace = true } codex-backend-client = { workspace = true } codex-git-utils = { workspace = true } +codex-http-client = { workspace = true } serde = { version = "1", features = ["derive"] } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/codex-rs/cloud-tasks-client/src/http.rs b/codex-rs/cloud-tasks-client/src/http.rs index 2f0fd613f7..057e6c151a 100644 --- a/codex-rs/cloud-tasks-client/src/http.rs +++ b/codex-rs/cloud-tasks-client/src/http.rs @@ -28,9 +28,12 @@ pub struct HttpClient { } impl HttpClient { - pub fn new(base_url: impl Into) -> anyhow::Result { + pub fn new( + base_url: impl Into, + http_client_factory: codex_http_client::HttpClientFactory, + ) -> anyhow::Result { let base_url = base_url.into(); - let backend = backend::Client::new(base_url.clone())?; + let backend = backend::Client::new(base_url.clone(), http_client_factory)?; Ok(Self { base_url, backend }) } diff --git a/codex-rs/cloud-tasks/src/lib.rs b/codex-rs/cloud-tasks/src/lib.rs index 4b99bb8c87..525b2e5f51 100644 --- a/codex-rs/cloud-tasks/src/lib.rs +++ b/codex-rs/cloud-tasks/src/lib.rs @@ -60,7 +60,10 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result } let ua = get_codex_user_agent(); - let mut http = codex_cloud_tasks_client::HttpClient::new(base_url.clone())?.with_user_agent(ua); + let (auth_manager, http_client_factory) = util::load_auth_manager(Some(base_url.clone())).await; + let mut http = + codex_cloud_tasks_client::HttpClient::new(base_url.clone(), http_client_factory)? + .with_user_agent(ua); let style = if base_url.contains("/backend-api") { "wham" } else { @@ -68,7 +71,6 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result }; append_error_log(format!("startup: base_url={base_url} path_style={style}")); - let auth_manager = util::load_auth_manager(Some(base_url.clone())).await; let auth = match auth_manager.as_ref() { Some(manager) => manager.auth().await, None => None, diff --git a/codex-rs/cloud-tasks/src/util.rs b/codex-rs/cloud-tasks/src/util.rs index 2a836f8579..aa197c14af 100644 --- a/codex-rs/cloud-tasks/src/util.rs +++ b/codex-rs/cloud-tasks/src/util.rs @@ -4,6 +4,8 @@ use chrono::Utc; use reqwest::header::HeaderMap; use codex_core::config::Config; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_login::AuthManager; pub fn set_user_agent_suffix(suffix: &str) { @@ -41,21 +43,28 @@ pub fn normalize_base_url(input: &str) -> String { base_url } -pub async fn load_auth_manager(chatgpt_base_url: Option) -> Option { +pub async fn load_auth_manager( + chatgpt_base_url: Option, +) -> (Option, HttpClientFactory) { // TODO: pass in cli overrides once cloud tasks properly support them. - let config = Config::load_with_cli_overrides(Vec::new()).await.ok()?; - Some( - AuthManager::new( - config.codex_home.to_path_buf(), - /*enable_codex_api_key_env*/ false, - config.cli_auth_credentials_store_mode, - config.forced_chatgpt_workspace_id.clone(), - chatgpt_base_url.or(Some(config.chatgpt_base_url.clone())), - config.auth_keyring_backend_kind(), - config.auth_route_config(), - ) - .await, + let Some(config) = Config::load_with_cli_overrides(Vec::new()).await.ok() else { + return ( + None, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ); + }; + let http_client_factory = config.http_client_factory(); + let auth_manager = AuthManager::new( + config.codex_home.to_path_buf(), + /*enable_codex_api_key_env*/ false, + config.cli_auth_credentials_store_mode, + config.forced_chatgpt_workspace_id.clone(), + chatgpt_base_url.or(Some(config.chatgpt_base_url.clone())), + config.auth_keyring_backend_kind(), + config.auth_route_config(), ) + .await; + (Some(auth_manager), http_client_factory) } /// Build headers for ChatGPT-backed requests: `User-Agent`, optional `Authorization`, @@ -71,7 +80,7 @@ pub async fn build_chatgpt_headers() -> HeaderMap { USER_AGENT, HeaderValue::from_str(&ua).unwrap_or(HeaderValue::from_static("codex-cli")), ); - if let Some(am) = load_auth_manager(/*chatgpt_base_url*/ None).await + if let Some(am) = load_auth_manager(/*chatgpt_base_url*/ None).await.0 && let Some(auth) = am.auth().await && auth.uses_codex_backend() { diff --git a/codex-rs/deny.toml b/codex-rs/deny.toml index b15f7b2335..e420271c5c 100644 --- a/codex-rs/deny.toml +++ b/codex-rs/deny.toml @@ -243,7 +243,6 @@ deny = [ "codex-api", "codex-app-server", "codex-app-server-daemon", - "codex-backend-client", "codex-cloud-tasks", "codex-core", "codex-core-plugins", diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index 49fc4225c3..0195bb9b9f 100644 --- a/codex-rs/http-client/src/lib.rs +++ b/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; diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 9e29b19090..731c9d96f2 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/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 { + 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 { + 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( diff --git a/codex-rs/http-client/src/outbound_proxy_tests.rs b/codex-rs/http-client/src/outbound_proxy_tests.rs index 387973eec4..54677ba2a0 100644 --- a/codex-rs/http-client/src/outbound_proxy_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_tests.rs @@ -10,6 +10,48 @@ struct MapEnv { values: HashMap, } +fn spawn_proxy_listener() -> ( + std::net::SocketAddr, + std::thread::JoinHandle>, +) { + 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!( diff --git a/codex-rs/http-client/src/route_aware_client_pool.rs b/codex-rs/http-client/src/route_aware_client_pool.rs new file mode 100644 index 0000000000..43ba1327f1 --- /dev/null +++ b/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 reqwest::ClientBuilder + Send + Sync>, + clients: Arc>>, +} + +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 { + 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( + &self, + request_url: &str, + resolve_route: F, + ) -> Result + where + F: FnOnce(String) -> Fut, + Fut: Future>, + { + 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; diff --git a/codex-rs/http-client/src/route_aware_client_pool_tests.rs b/codex-rs/http-client/src/route_aware_client_pool_tests.rs new file mode 100644 index 0000000000..7d159847f5 --- /dev/null +++ b/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::>(); + 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>, + observed_urls: Arc>>, +} + +impl FakeRouteResolver { + fn new(routes: HashMap) -> Self { + Self { + routes: Arc::new(routes), + observed_urls: Arc::new(Mutex::new(Vec::new())), + } + } + + async fn resolve(&self, request_url: String) -> io::Result { + 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 { + self.observed_urls + .lock() + .expect("observed URL lock") + .clone() + } +} + +async fn resolve_with( + pool: &RouteAwareClientPool, + resolver: &FakeRouteResolver, + request_url: &str, +) -> Result { + let resolver = resolver.clone(); + pool.client_for_url_with_resolver(request_url, move |request_url| async move { + resolver.resolve(request_url).await + }) + .await +} diff --git a/codex-rs/login/src/outbound_proxy.rs b/codex-rs/login/src/outbound_proxy.rs index 79fcf76de1..c1fd6f9465 100644 --- a/codex-rs/login/src/outbound_proxy.rs +++ b/codex-rs/login/src/outbound_proxy.rs @@ -17,7 +17,8 @@ impl AuthRouteConfig { } } - pub(crate) fn http_client_factory(&self) -> &HttpClientFactory { + /// Returns the HTTP client factory represented by this routing configuration. + pub fn http_client_factory(&self) -> &HttpClientFactory { &self.http_client_factory } } diff --git a/codex-rs/memories/write/src/guard.rs b/codex-rs/memories/write/src/guard.rs index 4d75043f97..fcb2361e5a 100644 --- a/codex-rs/memories/write/src/guard.rs +++ b/codex-rs/memories/write/src/guard.rs @@ -18,9 +18,13 @@ async fn rate_limits_check(auth_manager: &AuthManager, config: &Config) -> Optio return None; } - let client = BackendClient::from_auth(config.chatgpt_base_url.clone(), &auth) - .map_err(|err| warn!(%err, "failed to construct backend client")) - .ok()?; + let client = BackendClient::from_auth( + config.chatgpt_base_url.clone(), + &auth, + config.http_client_factory(), + ) + .map_err(|err| warn!(%err, "failed to construct backend client")) + .ok()?; let snapshots = client .get_rate_limits_many()