openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

backend-client: route requests through HTTP client factory

Browse Source
This commit is contained in:
Michael Bolin
2026-07-09 12:08:20 -07:00
parent fc0a88aa07
commit 1493a75867
24 changed files with 711 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+4 -1
View File
@@ -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",
codex-rs/app-server/src/config_manager.rs
+7 -2
View File
@@ -95,9 +95,14 @@ impl ConfigManager {
&self,
auth_manager: Arc<AuthManager>,
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 {
codex-rs/app-server/src/lib.rs
+5 -2
View File
@@ -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");
codex-rs/app-server/src/request_processors/account_processor.rs
+35 -15
View File
@@ -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<ThreadManager>,
chatgpt_base_url: String,
config: Arc<Config>,
login_id: Uuid,
success: bool,
error_msg: Option<String>,
@@ -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))
codex-rs/app-server/src/request_processors/account_processor/rate_limit_resets.rs
+6 -2
View File
@@ -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}")))
}
}
codex-rs/backend-client/Cargo.toml
+2 -1
View File
@@ -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 }
codex-rs/backend-client/src/client.rs
+129 -53
View File
@@ -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<HeaderValue>,
chatgpt_account_id: Option<String>,
@@ -148,7 +151,10 @@ impl fmt::Debug for Client {
}
impl Client {
pub fn new(base_url: impl Into<String>) -> Result<Self> {
pub fn new(
base_url: impl Into<String>,
http_client_factory: HttpClientFactory,
) -> Result<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,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<String>, auth: &CodexAuth) -> Result<Self> {
Ok(Self::new(base_url)?
pub fn from_auth(
base_url: impl Into<String>,
auth: &CodexAuth,
http_client_factory: HttpClientFactory,
) -> Result<Self> {
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<RequestBuilder> {
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<TokenUsageProfile> {
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<PaginatedListTaskListItem> {
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::<PaginatedListTaskListItem>(&url, &ct, &body)
}
fn list_tasks_url(
&self,
limit: Option<i32>,
task_filter: Option<&str>,
environment_id: Option<&str>,
cursor: Option<&str>,
) -> Result<String> {
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::<PaginatedListTaskListItem>(&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<CodeTaskDetailsResponse> {
@@ -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::<TurnAttemptsSiblingTurnsResponse>(&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::<ConfigBundleResponse>(&url, &ct, &body)
.map_err(RequestError::from)
@@ -437,8 +483,9 @@ impl Client {
) -> std::result::Result<CodexWorkspaceMessagesResponse, RequestError> {
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,
codex-rs/backend-client/src/client/rate_limit_resets.rs
+13 -6
View File
@@ -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<RateLimitStatusWithResetCredits> {
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<RateLimitResetCreditsDetails> {
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<ConsumeRateLimitResetCreditResponse> {
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 {
codex-rs/backend-client/src/client/rate_limit_resets_tests.rs
+6 -1
View File
@@ -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,
codex-rs/cloud-config/Cargo.toml
+1
View File
@@ -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 }
codex-rs/cloud-config/src/backend.rs
+19 -10
View File
@@ -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<CloudConfigBundle, BundleRequestError> {
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()
codex-rs/cloud-config/src/bundle_loader.rs
+17 -2
View File
@@ -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<AuthManager>,
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<AuthRouteConfig>,
) -> 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,
)
}
codex-rs/cloud-tasks-client/Cargo.toml
+1
View File
@@ -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 }
codex-rs/cloud-tasks-client/src/http.rs
+5 -2
View File
@@ -28,9 +28,12 @@ pub struct HttpClient {
}
impl HttpClient {
pub fn new(base_url: impl Into<String>) -> anyhow::Result<Self> {
pub fn new(
base_url: impl Into<String>,
http_client_factory: codex_http_client::HttpClientFactory,
) -> anyhow::Result<Self> {
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 })
}
codex-rs/cloud-tasks/src/lib.rs
+4 -2
View File
@@ -60,7 +60,10 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext>
}
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<BackendContext>
};
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,
codex-rs/cloud-tasks/src/util.rs
+23 -14
View File
@@ -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<String>) -> Option<AuthManager> {
pub async fn load_auth_manager(
chatgpt_base_url: Option<String>,
) -> (Option<AuthManager>, 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()
{
codex-rs/deny.toml
-1
View File
@@ -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",
codex-rs/http-client/src/lib.rs
+3
View File
@@ -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
+46 -1
View File
@@ -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,
@@ -193,6 +222,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
+82 -15
View File
@@ -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
+140
View File
@@ -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
+154
View File
@@ -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
}
codex-rs/login/src/outbound_proxy.rs
+2 -1
View File
@@ -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
}
}
codex-rs/memories/write/src/guard.rs
+7 -3
View File
@@ -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()