diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index db4350e1f947..323ac3dbac66 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2274,10 +2274,12 @@ dependencies = [ "codex-login", "codex-model-provider", "codex-protocol", + "http 1.4.0", "pretty_assertions", - "reqwest 0.12.28", "serde", "serde_json", + "tokio", + "url", ] [[package]] @@ -2418,6 +2420,7 @@ dependencies = [ "codex-backend-client", "codex-config", "codex-core", + "codex-http-client", "codex-login", "codex-otel", "codex-protocol", @@ -2472,6 +2475,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 d3d7609d6535..ee47f50cd54e 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 91da0c1645c9..336e5d943a5e 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 f4acc5a8e45a..3a044a9d7763 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,11 @@ 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(), + ); let (response, detailed_rate_limit_reset_credits) = tokio::join!( client.get_rate_limits_with_reset_credits(), @@ -1003,8 +1010,11 @@ 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(), + ); let profile = tokio::time::timeout( ACCOUNT_TOKEN_USAGE_FETCH_TIMEOUT, client.get_token_usage_profile(), @@ -1030,8 +1040,11 @@ 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(), + ); let messages = tokio::time::timeout( ACCOUNT_WORKSPACE_MESSAGES_FETCH_TIMEOUT, client.list_workspace_messages(), @@ -1118,8 +1131,11 @@ 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(), + ); 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 919d30d770a9..4c7930c10a41 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,11 @@ impl AccountRequestProcessor { )); } - BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth) - .map_err(|err| internal_error(format!("failed to construct backend client: {err}"))) + Ok(BackendClient::from_auth( + self.config.chatgpt_base_url.clone(), + &auth, + self.config.http_client_factory(), + )) } } diff --git a/codex-rs/backend-client/Cargo.toml b/codex-rs/backend-client/Cargo.toml index 4ff5fb9b863a..013628ccd502 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 } @@ -26,3 +27,4 @@ codex-protocol = { workspace = true } [dev-dependencies] pretty_assertions = "1" +tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/codex-rs/backend-client/src/client.rs b/codex-rs/backend-client/src/client.rs index a5db25618b4c..44b4b521d985 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::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestBuilder; 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,7 @@ 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) -> Self { 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,11 +164,12 @@ 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_without_request_logging( + http_client_factory, + ClientRouteClass::Api, + ); let path_style = PathStyle::from_base_url(&base_url); - Ok(Self { + Self { base_url, http, auth_provider: codex_model_provider::unauthenticated_auth_provider(), @@ -173,13 +177,17 @@ impl Client { chatgpt_account_id: None, chatgpt_account_is_fedramp: false, path_style, - }) + } } - 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, + ) -> Self { + 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))) + .with_auth_provider(codex_model_provider::auth_provider_from_auth(auth)) } pub fn with_auth_provider(mut self, auth: SharedAuthProvider) -> Self { @@ -231,9 +239,13 @@ impl Client { h } + fn request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder { + self.http.request(method, url) + } + async fn exec_request( &self, - req: reqwest::RequestBuilder, + req: RouteAwareRequestBuilder, method: &str, url: &str, ) -> Result<(String, String)> { @@ -254,7 +266,7 @@ impl Client { async fn exec_request_detailed( &self, - req: reqwest::RequestBuilder, + req: RouteAwareRequestBuilder, method: &str, url: &str, ) -> std::result::Result<(String, String), RequestError> { @@ -306,14 +318,14 @@ 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).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).headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } @@ -331,8 +343,7 @@ 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) .headers(self.headers()) .header(CONTENT_TYPE, HeaderValue::from_static("application/json")) .json(&SendAddCreditsNudgeEmailRequest { credit_type }); @@ -347,33 +358,44 @@ 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).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 +411,7 @@ 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).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 +432,7 @@ 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).headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json::(&url, &ct, &body) } @@ -426,7 +448,7 @@ 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).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 +459,7 @@ impl Client { ) -> std::result::Result { let url = self.workspace_messages_url(); let req = self - .http - .get(&url) + .request(Method::GET, &url) .headers(self.headers()) .header(CACHE_CONTROL, HeaderValue::from_static("no-store")); let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?; @@ -454,8 +475,7 @@ impl Client { PathStyle::ChatGptApi => format!("{}/wham/tasks", self.base_url), }; let req = self - .http - .post(&url) + .request(Method::POST, &url) .headers(self.headers()) .header(CONTENT_TYPE, HeaderValue::from_static("application/json")) .json(&request_body); @@ -666,6 +686,11 @@ impl Client { #[cfg(test)] mod tests { + use std::io::Read; + use std::io::Write; + use std::sync::Arc; + use std::time::Duration; + use super::*; use codex_backend_openapi_models::models::AdditionalRateLimitDetails; use codex_backend_openapi_models::models::RateLimitReachedKind; @@ -974,10 +999,132 @@ 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" + ); + } + + #[tokio::test] + async fn migrated_requests_preserve_query_auth_and_json_body() { + let listener = + std::net::TcpListener::bind("127.0.0.1:0").expect("HTTP listener should bind"); + let address = listener + .local_addr() + .expect("HTTP listener should have an address"); + let server = std::thread::spawn(move || { + let mut requests = Vec::new(); + for body in [r#"{"items":[]}"#, r#"{"task":{"id":"task-created"}}"#] { + let (mut stream, _) = listener.accept().expect("HTTP listener should accept"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("HTTP stream should get a read timeout"); + let mut request = Vec::new(); + let mut buffer = [0_u8; 4096]; + loop { + let size = stream.read(&mut buffer).expect("HTTP request should read"); + if size == 0 { + break; + } + request.extend_from_slice(&buffer[..size]); + let Some(headers_end) = request.windows(4).position(|part| part == b"\r\n\r\n") + else { + continue; + }; + let headers = String::from_utf8_lossy(&request[..headers_end]); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(0); + if request.len() >= headers_end + 4 + content_length { + break; + } + } + requests.push(String::from_utf8(request).expect("request should be UTF-8")); + write!( + stream, + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .expect("HTTP response should write"); + } + requests + }); + let client = Client::new( + format!("http://{address}"), + HttpClientFactory::new(codex_http_client::OutboundProxyPolicy::ReqwestDefault), + ) + .with_auth_provider(Arc::new(codex_model_provider::BearerAuthProvider::new( + "request-token".to_string(), + ))); + + let tasks = client + .list_tasks( + Some(10), + Some("mine / shared"), + Some("env&one"), + Some("next=page"), + ) + .await + .expect("list request should succeed"); + let task_id = client + .create_task(serde_json::json!({ "prompt": "hello" })) + .await + .expect("create request should succeed"); + let requests = server.join().expect("HTTP server should finish"); + + assert_eq!(tasks, PaginatedListTaskListItem::new(Vec::new())); + assert_eq!(task_id, "task-created"); + assert_eq!(requests.len(), 2); + assert!(requests[0].starts_with( + "GET /api/codex/tasks/list?limit=10&task_filter=mine+%2F+shared&cursor=next%3Dpage&environment_id=env%26one HTTP/1.1\r\n" + )); + assert!( + requests[0] + .to_ascii_lowercase() + .contains("authorization: bearer request-token\r\n") + ); + assert!(requests[1].starts_with("POST /api/codex/tasks HTTP/1.1\r\n")); + assert!( + requests[1] + .to_ascii_lowercase() + .contains("authorization: bearer request-token\r\n") + ); + assert!(requests[1].ends_with(r#"{"prompt":"hello"}"#)); + } + 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 bed56ba6488b..90bedfb24afd 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,14 @@ 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).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).headers(self.headers()); let (body, ct) = self.exec_request(req, "GET", &url).await?; self.decode_json(&url, &ct, &body) } @@ -65,8 +66,7 @@ impl Client { ) -> Result { let url = self.consume_rate_limit_reset_credit_url(); let req = self - .http - .post(&url) + .request(Method::POST, &url) .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 f706be5da90f..55703caab544 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 363cb21b6f13..ecb33d34beb8 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 cb99316a7053..b8b456a8ac12 100644 --- a/codex-rs/cloud-config/src/backend.rs +++ b/codex-rs/cloud-config/src/backend.rs @@ -6,19 +6,18 @@ 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; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum RetryableFailureKind { - BackendClientInit, Request { status_code: Option }, } impl RetryableFailureKind { pub(crate) fn status_code(self) -> Option { match self { - Self::BackendClientInit => None, Self::Request { status_code } => status_code, } } @@ -46,24 +45,25 @@ 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(), + ); 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 e4266c6195ba..837001d0150f 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 dc07550d086d..8efd78cb22d0 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 2f0fd613f79b..6ee2e5f632ae 100644 --- a/codex-rs/cloud-tasks-client/src/http.rs +++ b/codex-rs/cloud-tasks-client/src/http.rs @@ -28,10 +28,13 @@ 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, + ) -> Self { let base_url = base_url.into(); - let backend = backend::Client::new(base_url.clone())?; - Ok(Self { base_url, backend }) + let backend = backend::Client::new(base_url.clone(), http_client_factory); + Self { base_url, backend } } pub fn with_user_agent(mut self, ua: impl Into) -> Self { diff --git a/codex-rs/cloud-tasks/src/lib.rs b/codex-rs/cloud-tasks/src/lib.rs index 4b99bb8c87a1..64b4cb379c14 100644 --- a/codex-rs/cloud-tasks/src/lib.rs +++ b/codex-rs/cloud-tasks/src/lib.rs @@ -60,7 +60,9 @@ 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 +70,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 2a836f857931..aa197c14af36 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 b15f7b233517..e420271c5cc8 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/login/src/outbound_proxy.rs b/codex-rs/login/src/outbound_proxy.rs index 79fcf76de192..c1fd6f9465f6 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 4d75043f9717..a3b54876b312 100644 --- a/codex-rs/memories/write/src/guard.rs +++ b/codex-rs/memories/write/src/guard.rs @@ -18,9 +18,11 @@ 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(), + ); let snapshots = client .get_rate_limits_many()