diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index db4350e1f9..7809642880 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2092,12 +2092,12 @@ dependencies = [ "anyhow", "codex-app-server-protocol", "codex-app-server-transport", + "codex-http-client", "codex-uds", "codex-utils-home-dir", "futures", "libc", "pretty_assertions", - "reqwest 0.12.28", "serde", "serde_json", "sha2 0.10.9", @@ -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]] @@ -2346,6 +2347,7 @@ dependencies = [ "codex-features", "codex-git-utils", "codex-home", + "codex-http-client", "codex-install-context", "codex-login", "codex-mcp", @@ -2418,6 +2420,7 @@ dependencies = [ "codex-backend-client", "codex-config", "codex-core", + "codex-http-client", "codex-login", "codex-otel", "codex-protocol", @@ -2449,10 +2452,10 @@ dependencies = [ "codex-tui", "codex-utils-cli", "crossterm", + "http 1.4.0", "owo-colors", "pretty_assertions", "ratatui", - "reqwest 0.12.28", "serde", "serde_json", "supports-color 3.0.2", @@ -2472,6 +2475,7 @@ dependencies = [ "codex-api", "codex-backend-client", "codex-git-utils", + "codex-http-client", "serde", "serde_json", "thiserror 2.0.18", @@ -2782,6 +2786,7 @@ dependencies = [ "codex-exec-server", "codex-git-utils", "codex-hooks", + "codex-http-client", "codex-login", "codex-mcp", "codex-model-provider", @@ -2795,10 +2800,10 @@ dependencies = [ "codex-utils-plugins", "dirs", "flate2", + "http 1.4.0", "libc", "pretty_assertions", "regex", - "reqwest 0.12.28", "semver", "serde", "serde_json", diff --git a/codex-rs/app-server-daemon/Cargo.toml b/codex-rs/app-server-daemon/Cargo.toml index 24531b4c74..fee5732197 100644 --- a/codex-rs/app-server-daemon/Cargo.toml +++ b/codex-rs/app-server-daemon/Cargo.toml @@ -16,11 +16,11 @@ workspace = true anyhow = { workspace = true } codex-app-server-protocol = { workspace = true } codex-app-server-transport = { workspace = true } +codex-http-client = { workspace = true } codex-utils-home-dir = { workspace = true } codex-uds = { workspace = true } futures = { workspace = true } libc = { workspace = true } -reqwest = { workspace = true, features = ["rustls-tls"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha2 = { workspace = true } diff --git a/codex-rs/app-server-daemon/src/lib.rs b/codex-rs/app-server-daemon/src/lib.rs index 5870075fea..92e69c7b61 100644 --- a/codex-rs/app-server-daemon/src/lib.rs +++ b/codex-rs/app-server-daemon/src/lib.rs @@ -238,9 +238,11 @@ pub async fn set_remote_control(mode: RemoteControlMode) -> Result Result<()> { +pub async fn run_pid_update_loop( + http_client_factory: codex_http_client::HttpClientFactory, +) -> Result<()> { ensure_supported_platform()?; - update_loop::run().await + update_loop::run(http_client_factory).await } #[cfg(unix)] diff --git a/codex-rs/app-server-daemon/src/update_loop.rs b/codex-rs/app-server-daemon/src/update_loop.rs index 0fe21e0a6d..5b5ad3707c 100644 --- a/codex-rs/app-server-daemon/src/update_loop.rs +++ b/codex-rs/app-server-daemon/src/update_loop.rs @@ -11,6 +11,11 @@ use anyhow::Result; #[cfg(not(unix))] use anyhow::bail; #[cfg(unix)] +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +#[cfg(unix)] +use codex_http_client::RouteAwareClientPool; +#[cfg(unix)] use futures::FutureExt; #[cfg(unix)] use std::os::unix::process::CommandExt; @@ -48,17 +53,20 @@ const INITIAL_UPDATE_DELAY: Duration = Duration::from_secs(5 * 60); const RESTART_RETRY_INTERVAL: Duration = Duration::from_millis(50); #[cfg(unix)] const UPDATE_INTERVAL: Duration = Duration::from_secs(60 * 60); +#[cfg(unix)] +const INSTALL_URL: &str = "https://chatgpt.com/codex/install.sh"; #[cfg(unix)] -pub(crate) async fn run() -> Result<()> { +pub(crate) async fn run(http_client_factory: HttpClientFactory) -> Result<()> { let mut terminate = signal(SignalKind::terminate()).context("failed to install updater shutdown handler")?; let running_updater_identity = current_updater_identity().await?; + let http = RouteAwareClientPool::new(http_client_factory, ClientRouteClass::Other); if sleep_or_terminate(INITIAL_UPDATE_DELAY, &mut terminate).await { return Ok(()); } loop { - match update_once(&running_updater_identity, &mut terminate).await { + match update_once(&http, &running_updater_identity, &mut terminate).await { Ok(UpdateLoopControl::Continue) | Err(_) => {} Ok(UpdateLoopControl::Stop) => return Ok(()), } @@ -69,7 +77,7 @@ pub(crate) async fn run() -> Result<()> { } #[cfg(not(unix))] -pub(crate) async fn run() -> Result<()> { +pub(crate) async fn run(_http_client_factory: HttpClientFactory) -> Result<()> { bail!("pid-managed updater loop is unsupported on this platform") } @@ -89,10 +97,11 @@ enum UpdateLoopControl { #[cfg(unix)] async fn update_once( + http: &RouteAwareClientPool, running_updater_identity: &ExecutableIdentity, terminate: &mut Signal, ) -> Result { - install_latest_standalone().await?; + install_latest_standalone(http).await?; let daemon = Daemon::from_environment()?; let managed_codex_bin = resolved_managed_codex_bin(&daemon.managed_codex_bin).await?; @@ -154,15 +163,8 @@ pub(crate) fn reexec_managed_updater(managed_codex_bin: &std::path::Path) -> Res } #[cfg(unix)] -async fn install_latest_standalone() -> Result<()> { - let script = reqwest::get("https://chatgpt.com/codex/install.sh") - .await - .context("failed to fetch standalone Codex updater")? - .error_for_status() - .context("standalone Codex updater request failed")? - .bytes() - .await - .context("failed to read standalone Codex updater")?; +async fn install_latest_standalone(http: &RouteAwareClientPool) -> Result<()> { + let script = fetch_installer_script(http).await?; let mut child = Command::new("/bin/sh") .arg("-s") @@ -192,6 +194,54 @@ async fn install_latest_standalone() -> Result<()> { } } +#[cfg(unix)] +async fn fetch_installer_script(http: &impl InstallerHttp) -> Result> { + let response = http.get(INSTALL_URL).await?; + if !(200..300).contains(&response.status) { + anyhow::bail!( + "standalone Codex updater request failed with status {}", + response.status + ); + } + Ok(response.body) +} + +#[cfg(unix)] +#[derive(Clone, Debug, PartialEq, Eq)] +struct InstallerResponse { + status: u16, + body: Vec, +} + +#[cfg(unix)] +trait InstallerHttp: Send + Sync { + fn get<'a>( + &'a self, + url: &'a str, + ) -> impl std::future::Future> + Send + 'a; +} + +#[cfg(unix)] +impl InstallerHttp for RouteAwareClientPool { + async fn get(&self, url: &str) -> Result { + let response = self + .client_for_url(url) + .await + .context("failed to configure standalone Codex updater client")? + .get(url) + .send() + .await + .context("failed to fetch standalone Codex updater")?; + let status = response.status().as_u16(); + let body = response + .bytes() + .await + .context("failed to read standalone Codex updater")? + .to_vec(); + Ok(InstallerResponse { status, body }) + } +} + #[cfg(all(test, unix))] #[path = "update_loop_tests.rs"] mod tests; diff --git a/codex-rs/app-server-daemon/src/update_loop_tests.rs b/codex-rs/app-server-daemon/src/update_loop_tests.rs index cf270693aa..66ad6f8ffe 100644 --- a/codex-rs/app-server-daemon/src/update_loop_tests.rs +++ b/codex-rs/app-server-daemon/src/update_loop_tests.rs @@ -1,5 +1,11 @@ +use std::sync::Mutex; + use pretty_assertions::assert_eq; +use super::INSTALL_URL; +use super::InstallerHttp; +use super::InstallerResponse; +use super::fetch_installer_script; use super::update_modes_for_identities; use crate::RestartMode; use crate::UpdaterRefreshMode; @@ -29,3 +35,66 @@ fn changed_updater_forces_refresh_even_when_version_may_match() { ) ); } + +#[tokio::test] +async fn installer_fetch_uses_exact_url_and_preserves_bytes() { + let script = b"#!/bin/sh\nprintf 'update bytes'\n".to_vec(); + let http = FakeInstallerHttp::new(InstallerResponse { + status: 200, + body: script.clone(), + }); + + assert_eq!( + fetch_installer_script(&http) + .await + .expect("installer fetch should succeed"), + script + ); + assert_eq!(http.requested_urls(), vec![INSTALL_URL.to_string()]); +} + +#[tokio::test] +async fn installer_fetch_rejects_non_success_status() { + let http = FakeInstallerHttp::new(InstallerResponse { + status: 503, + body: b"unavailable".to_vec(), + }); + + let error = fetch_installer_script(&http) + .await + .expect_err("non-success response should fail"); + + assert!(error.to_string().contains("503")); + assert_eq!(http.requested_urls(), vec![INSTALL_URL.to_string()]); +} + +struct FakeInstallerHttp { + response: InstallerResponse, + requested_urls: Mutex>, +} + +impl FakeInstallerHttp { + fn new(response: InstallerResponse) -> Self { + Self { + response, + requested_urls: Mutex::new(Vec::new()), + } + } + + fn requested_urls(&self) -> Vec { + self.requested_urls + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +impl InstallerHttp for FakeInstallerHttp { + async fn get(&self, url: &str) -> anyhow::Result { + self.requested_urls + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(url.to_string()); + Ok(self.response.clone()) + } +} 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/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 688cfae917..9fef99572d 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -127,6 +127,13 @@ fn load_shared_plugin_ids_by_local_path( }) } +fn remote_plugin_service_config(config: &Config) -> RemotePluginServiceConfig { + RemotePluginServiceConfig::new( + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ) +} + fn share_context_for_source( source: &MarketplacePluginSource, shared_plugin_ids_by_local_path: &std::collections::BTreeMap, @@ -581,9 +588,7 @@ impl PluginRequestProcessor { !explicit_marketplace_kinds && config.features.enabled(Feature::RemotePlugin); let use_remote_global_catalog = include_global_remote && auth_mode.is_some_and(DomainAuthMode::uses_codex_backend); - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let refresh_global_remote_catalog_cache = use_remote_global_catalog && codex_core_plugins::remote::has_cached_global_remote_plugin_catalog( config.codex_home.as_path(), @@ -1030,9 +1035,7 @@ impl PluginRequestProcessor { ); let share_context = match share_context { Some(context) => { - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); match codex_core_plugins::remote::fetch_remote_plugin_share_context( &remote_plugin_service_config, auth.as_ref(), @@ -1138,9 +1141,7 @@ impl PluginRequestProcessor { "remote plugin read is not enabled for marketplace {remote_marketplace_name}" ))); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); validate_remote_plugin_id(&plugin_name)?; let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail( &remote_plugin_service_config, @@ -1196,9 +1197,7 @@ impl PluginRequestProcessor { } let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let remote_skill_detail = codex_core_plugins::remote::fetch_remote_plugin_skill_detail( &remote_plugin_service_config, auth.as_ref(), @@ -1249,9 +1248,7 @@ impl PluginRequestProcessor { validate_client_plugin_share_targets(share_targets)?; } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let access_policy = codex_core_plugins::remote::RemotePluginShareAccessPolicy { discoverability: discoverability.map(remote_plugin_share_discoverability), share_targets: share_targets.map(remote_plugin_share_targets), @@ -1292,9 +1289,7 @@ impl PluginRequestProcessor { } validate_client_plugin_share_targets(&share_targets)?; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let result = codex_core_plugins::remote::update_remote_plugin_share_targets( &remote_plugin_service_config, auth.as_ref(), @@ -1322,9 +1317,7 @@ impl PluginRequestProcessor { _params: PluginShareListParams, ) -> Result { let (config, auth) = self.load_plugin_share_config_and_auth().await?; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let data = codex_core_plugins::remote::list_remote_plugin_shares( &remote_plugin_service_config, auth.as_ref(), @@ -1361,9 +1354,7 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let result = codex_core_plugins::remote::checkout_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1394,9 +1385,7 @@ impl PluginRequestProcessor { return Err(invalid_request("invalid remote plugin id")); } - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); codex_core_plugins::remote::delete_remote_plugin_share( &remote_plugin_service_config, auth.as_ref(), @@ -1531,9 +1520,7 @@ impl PluginRequestProcessor { validate_remote_plugin_id(&remote_plugin_id)?; let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail_with_download_urls( &remote_plugin_service_config, @@ -1606,6 +1593,7 @@ impl PluginRequestProcessor { })?; let result = codex_core_plugins::remote_bundle::download_and_install_remote_plugin_bundle( + &remote_plugin_service_config, config.codex_home.to_path_buf(), validated_bundle, ) @@ -2020,9 +2008,7 @@ impl PluginRequestProcessor { validate_remote_plugin_id(&plugin_id)?; let auth = self.auth_manager.auth().await; - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; + let remote_plugin_service_config = remote_plugin_service_config(&config); let uninstall_target = codex_core_plugins::remote::resolve_remote_plugin_uninstall_target( &remote_plugin_service_config, auth.as_ref(), @@ -2296,6 +2282,7 @@ fn remote_plugin_catalog_error_type(err: &RemotePluginCatalogError) -> &'static RemotePluginCatalogError::AuthRequired => "remote_catalog_auth_required", RemotePluginCatalogError::UnsupportedAuthMode => "remote_catalog_unsupported_auth_mode", RemotePluginCatalogError::AuthToken(_) => "remote_catalog_auth_token", + RemotePluginCatalogError::HttpClient { .. } => "remote_catalog_http_client", RemotePluginCatalogError::Request { .. } => "remote_catalog_request", RemotePluginCatalogError::UnexpectedStatus { .. } => "remote_catalog_unexpected_status", RemotePluginCatalogError::Decode { .. } => "remote_catalog_decode", @@ -2342,6 +2329,7 @@ fn remote_plugin_bundle_install_error_type(err: &RemotePluginBundleInstallError) "remote_bundle_unsupported_download_url_scheme" } RemotePluginBundleInstallError::InvalidPluginId { .. } => "remote_bundle_invalid_plugin_id", + RemotePluginBundleInstallError::HttpClient { .. } => "remote_bundle_http_client", RemotePluginBundleInstallError::DownloadRequest { .. } => "remote_bundle_download_request", RemotePluginBundleInstallError::DownloadStatus { .. } => "remote_bundle_download_status", RemotePluginBundleInstallError::DownloadBody { .. } => "remote_bundle_download_body", @@ -2377,6 +2365,7 @@ fn remote_plugin_catalog_error_to_jsonrpc( | RemotePluginCatalogError::ArchiveTooLarge { .. } | RemotePluginCatalogError::UnknownMarketplace { .. } => invalid_request(message), RemotePluginCatalogError::AuthToken(_) + | RemotePluginCatalogError::HttpClient { .. } | RemotePluginCatalogError::Request { .. } | RemotePluginCatalogError::UnexpectedStatus { .. } | RemotePluginCatalogError::Decode { .. } 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/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index 9a8312ced2..614834d173 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -34,6 +34,7 @@ codex-config = { workspace = true } codex-core = { workspace = true } codex-core-plugins = { workspace = true } codex-home = { workspace = true } +codex-http-client = { workspace = true } codex-exec = { workspace = true } codex-exec-server = { workspace = true } codex-execpolicy = { workspace = true } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 04d59c24ab..06901ef4a4 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -1182,7 +1182,16 @@ async fn cli_main( print_app_server_daemon_output(AppServerLifecycleCommand::Version).await?; } AppServerDaemonSubcommand::PidUpdateLoop => { - codex_app_server_daemon::run_pid_update_loop().await?; + let cli_overrides = root_config_overrides + .parse_overrides() + .map_err(anyhow::Error::msg)?; + let config = ConfigBuilder::default() + .cli_overrides(cli_overrides) + .build() + .await + .map_err(anyhow::Error::from); + let http_client_factory = updater_http_client_factory(config); + codex_app_server_daemon::run_pid_update_loop(http_client_factory).await?; } }, Some(AppServerSubcommand::Proxy(proxy_cli)) => { @@ -2205,6 +2214,20 @@ async fn print_app_server_daemon_output(command: AppServerLifecycleCommand) -> a Ok(()) } +fn updater_http_client_factory( + config: anyhow::Result, +) -> codex_http_client::HttpClientFactory { + match config { + Ok(config) => config.http_client_factory(), + Err(error) => { + eprintln!("warning: failed to load updater network configuration: {error}"); + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::ReqwestDefault, + ) + } + } +} + async fn print_app_server_remote_control_output( mode: AppServerRemoteControlMode, ) -> anyhow::Result<()> { @@ -2501,6 +2524,34 @@ mod tests { use codex_tui::TokenUsage; use pretty_assertions::assert_eq; + #[tokio::test] + async fn updater_http_client_factory_honors_respect_system_proxy() { + let codex_home = tempfile::tempdir().expect("temporary Codex home"); + let config = ConfigBuilder::default() + .codex_home(codex_home.path().to_path_buf()) + .cli_overrides(vec![( + "features.respect_system_proxy".to_string(), + toml::Value::Boolean(true), + )]) + .build() + .await + .expect("config should load"); + + assert_eq!( + updater_http_client_factory(Ok(config)).outbound_proxy_policy(), + codex_http_client::OutboundProxyPolicy::RespectSystemProxy + ); + } + + #[test] + fn updater_http_client_factory_falls_back_when_config_load_fails() { + assert_eq!( + updater_http_client_factory(Err(anyhow::anyhow!("invalid config"))) + .outbound_proxy_policy(), + codex_http_client::OutboundProxyPolicy::ReqwestDefault + ); + } + #[test] fn exec_server_remote_auth_accepts_api_key_auth() { let auth = CodexAuth::from_api_key("sk-test"); 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/Cargo.toml b/codex-rs/cloud-tasks/Cargo.toml index d842a95a94..be3ec552c8 100644 --- a/codex-rs/cloud-tasks/Cargo.toml +++ b/codex-rs/cloud-tasks/Cargo.toml @@ -27,9 +27,9 @@ codex-model-provider = { workspace = true } codex-tui = { workspace = true } codex-utils-cli = { workspace = true } crossterm = { workspace = true, features = ["event-stream"] } +http = { workspace = true } owo-colors = { workspace = true, features = ["supports-colors"] } ratatui = { workspace = true } -reqwest = { workspace = true, features = ["json"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } supports-color = { workspace = true } diff --git a/codex-rs/cloud-tasks/src/env_detect.rs b/codex-rs/cloud-tasks/src/env_detect.rs index 6137f27ea8..bcb003d62c 100644 --- a/codex-rs/cloud-tasks/src/env_detect.rs +++ b/codex-rs/cloud-tasks/src/env_detect.rs @@ -1,6 +1,7 @@ -use codex_http_client::build_reqwest_client_with_custom_ca; -use reqwest::header::CONTENT_TYPE; -use reqwest::header::HeaderMap; +use codex_http_client::RouteAwareClientPool; +use http::StatusCode; +use http::header::CONTENT_TYPE; +use http::header::HeaderMap; use std::collections::HashMap; use tracing::info; use tracing::warn; @@ -16,22 +17,39 @@ struct CodeEnvironment { task_count: Option, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct AutodetectSelection { pub id: String, pub label: Option, } pub async fn autodetect_environment_id( + http: &RouteAwareClientPool, base_url: &str, headers: &HeaderMap, desired_label: Option, +) -> anyhow::Result { + autodetect_environment_id_with_origins( + http, + base_url, + headers, + desired_label, + &get_git_origins(), + ) + .await +} + +async fn autodetect_environment_id_with_origins( + http: &impl EnvironmentHttp, + base_url: &str, + headers: &HeaderMap, + desired_label: Option, + origins: &[String], ) -> anyhow::Result { // 1) Try repo-specific environments based on local git origins (GitHub only, like VSCode) - let origins = get_git_origins(); crate::append_error_log(format!("env: git origins: {origins:?}")); let mut by_repo_envs: Vec = Vec::new(); - for origin in &origins { + for origin in origins { if let Some((owner, repo)) = parse_owner_repo(origin) { let url = if base_url.contains("/backend-api") { format!( @@ -45,7 +63,7 @@ pub async fn autodetect_environment_id( ) }; crate::append_error_log(format!("env: GET {url}")); - match get_json::>(&url, headers).await { + match get_json::>(http, &url, headers).await { Ok(mut list) => { crate::append_error_log(format!( "env: by-repo returned {} env(s) for {owner}/{repo}", @@ -74,16 +92,10 @@ pub async fn autodetect_environment_id( }; crate::append_error_log(format!("env: GET {list_url}")); // Fetch and log the full environments JSON for debugging - let http = build_reqwest_client_with_custom_ca(reqwest::Client::builder())?; - let res = http.get(&list_url).headers(headers.clone()).send().await?; - let status = res.status(); - let ct = res - .headers() - .get(CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .unwrap_or("") - .to_string(); - let body = res.text().await.unwrap_or_default(); + let response = http.get(&list_url, headers).await?; + let status = response.status; + let ct = response.content_type; + let body = response.body; crate::append_error_log(format!("env: status={status} content-type={ct}")); match serde_json::from_str::(&body) { Ok(v) => { @@ -145,19 +157,14 @@ fn pick_environment_row( } async fn get_json( + http: &impl EnvironmentHttp, url: &str, headers: &HeaderMap, ) -> anyhow::Result { - let http = build_reqwest_client_with_custom_ca(reqwest::Client::builder())?; - let res = http.get(url).headers(headers.clone()).send().await?; - let status = res.status(); - let ct = res - .headers() - .get(CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .unwrap_or("") - .to_string(); - let body = res.text().await.unwrap_or_default(); + let response = http.get(url, headers).await?; + let status = response.status; + let ct = response.content_type; + let body = response.body; crate::append_error_log(format!("env: status={status} content-type={ct}")); if !status.is_success() { anyhow::bail!("GET {url} failed: {status}; content-type={ct}; body={body}"); @@ -168,6 +175,43 @@ async fn get_json( Ok(parsed) } +#[derive(Clone, Debug, PartialEq, Eq)] +struct EnvironmentResponse { + status: StatusCode, + content_type: String, + body: String, +} + +trait EnvironmentHttp: Send + Sync { + fn get<'a>( + &'a self, + url: &'a str, + headers: &'a HeaderMap, + ) -> impl std::future::Future> + Send + 'a; +} + +impl EnvironmentHttp for RouteAwareClientPool { + async fn get(&self, url: &str, headers: &HeaderMap) -> anyhow::Result { + let response = self + .client_for_url(url) + .await? + .get(url) + .headers(headers.clone()) + .send() + .await?; + Ok(EnvironmentResponse { + status: response.status(), + content_type: response + .headers() + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or("") + .to_string(), + body: response.text().await.unwrap_or_default(), + }) + } +} + fn get_git_origins() -> Vec { // Prefer: git config --get-regexp remote\..*\.url let out = std::process::Command::new("git") @@ -254,14 +298,23 @@ fn parse_owner_repo(url: &str) -> Option<(String, String)> { /// List environments for the current repo(s) with a fallback to the global list. /// Returns a de-duplicated, sorted set suitable for the TUI modal. pub async fn list_environments( + http: &RouteAwareClientPool, base_url: &str, headers: &HeaderMap, +) -> anyhow::Result> { + list_environments_with_origins(http, base_url, headers, &get_git_origins()).await +} + +async fn list_environments_with_origins( + http: &impl EnvironmentHttp, + base_url: &str, + headers: &HeaderMap, + origins: &[String], ) -> anyhow::Result> { let mut map: HashMap = HashMap::new(); // 1) By-repo lookup for each parsed GitHub origin - let origins = get_git_origins(); - for origin in &origins { + for origin in origins { if let Some((owner, repo)) = parse_owner_repo(origin) { let url = if base_url.contains("/backend-api") { format!( @@ -274,7 +327,7 @@ pub async fn list_environments( base_url, "github", owner, repo ) }; - match get_json::>(&url, headers).await { + match get_json::>(http, &url, headers).await { Ok(list) => { info!("env_tui: by-repo {}:{} -> {} envs", owner, repo, list.len()); for e in list { @@ -312,7 +365,7 @@ pub async fn list_environments( } else { format!("{base_url}/api/codex/environments") }; - match get_json::>(&list_url, headers).await { + match get_json::>(http, &list_url, headers).await { Ok(list) => { info!("env_tui: global list -> {} envs", list.len()); for e in list { @@ -360,3 +413,7 @@ pub async fn list_environments( }); Ok(rows) } + +#[cfg(test)] +#[path = "env_detect_tests.rs"] +mod tests; diff --git a/codex-rs/cloud-tasks/src/env_detect_tests.rs b/codex-rs/cloud-tasks/src/env_detect_tests.rs new file mode 100644 index 0000000000..21ff0e4cda --- /dev/null +++ b/codex-rs/cloud-tasks/src/env_detect_tests.rs @@ -0,0 +1,164 @@ +use std::collections::HashMap; +use std::sync::Mutex; + +use http::HeaderMap; +use http::StatusCode; +use pretty_assertions::assert_eq; + +use super::*; + +const BASE_URL: &str = "https://chatgpt.com/backend-api"; +const BY_REPO_URL: &str = + "https://chatgpt.com/backend-api/wham/environments/by-repo/github/openai/codex"; +const GLOBAL_URL: &str = "https://chatgpt.com/backend-api/wham/environments"; + +#[tokio::test] +async fn autodetect_requests_exact_repository_endpoint_and_decodes_selection() { + let http = FakeHttp::new(HashMap::from([( + BY_REPO_URL.to_string(), + json_response(r#"[{"id":"env-repo","label":"Repository","is_pinned":true}]"#), + )])); + + let selection = autodetect_environment_id_with_origins( + &http, + BASE_URL, + &HeaderMap::new(), + Some("Repository".to_string()), + &["git@github.com:openai/codex.git".to_string()], + ) + .await + .expect("repository environment should be selected"); + + assert_eq!( + selection, + AutodetectSelection { + id: "env-repo".to_string(), + label: Some("Repository".to_string()), + } + ); + assert_eq!(http.requested_urls(), vec![BY_REPO_URL.to_string()]); +} + +#[tokio::test] +async fn autodetect_falls_back_to_exact_global_endpoint_and_decodes_selection() { + let http = FakeHttp::new(HashMap::from([ + (BY_REPO_URL.to_string(), json_response("[]")), + ( + GLOBAL_URL.to_string(), + json_response(r#"[{"id":"env-global","label":"Global"}]"#), + ), + ])); + + let selection = autodetect_environment_id_with_origins( + &http, + BASE_URL, + &HeaderMap::new(), + /*desired_label*/ None, + &["git@github.com:openai/codex.git".to_string()], + ) + .await + .expect("global environment should be selected"); + + assert_eq!( + selection, + AutodetectSelection { + id: "env-global".to_string(), + label: Some("Global".to_string()), + } + ); + assert_eq!( + http.requested_urls(), + vec![BY_REPO_URL.to_string(), GLOBAL_URL.to_string()] + ); +} + +#[tokio::test] +async fn list_requests_exact_repository_and_global_endpoints_and_merges_results() { + let http = FakeHttp::new(HashMap::from([ + ( + BY_REPO_URL.to_string(), + json_response(r#"[{"id":"env-repo","label":"Repository"}]"#), + ), + ( + GLOBAL_URL.to_string(), + json_response( + r#"[{"id":"env-repo","is_pinned":true},{"id":"env-global","label":"Global"}]"#, + ), + ), + ])); + + let rows = list_environments_with_origins( + &http, + BASE_URL, + &HeaderMap::new(), + &["https://github.com/openai/codex.git".to_string()], + ) + .await + .expect("environment list should decode"); + + assert_eq!( + rows.into_iter() + .map(|row| (row.id, row.label, row.is_pinned, row.repo_hints)) + .collect::>(), + vec![ + ( + "env-repo".to_string(), + Some("Repository".to_string()), + true, + Some("openai/codex".to_string()), + ), + ( + "env-global".to_string(), + Some("Global".to_string()), + false, + None, + ), + ] + ); + assert_eq!( + http.requested_urls(), + vec![BY_REPO_URL.to_string(), GLOBAL_URL.to_string()] + ); +} + +struct FakeHttp { + responses: HashMap, + requested_urls: Mutex>, +} + +impl FakeHttp { + fn new(responses: HashMap) -> Self { + Self { + responses, + requested_urls: Mutex::new(Vec::new()), + } + } + + fn requested_urls(&self) -> Vec { + self.requested_urls + .lock() + .expect("requested URL lock") + .clone() + } +} + +impl EnvironmentHttp for FakeHttp { + async fn get(&self, url: &str, _headers: &HeaderMap) -> anyhow::Result { + self.requested_urls + .lock() + .expect("requested URL lock") + .push(url.to_string()); + self.responses + .get(url) + .cloned() + .ok_or_else(|| anyhow::anyhow!("unexpected URL: {url}")) + } +} + +fn json_response(body: &str) -> EnvironmentResponse { + EnvironmentResponse { + status: StatusCode::OK, + content_type: "application/json".to_string(), + body: body.to_string(), + } +} diff --git a/codex-rs/cloud-tasks/src/lib.rs b/codex-rs/cloud-tasks/src/lib.rs index 4b99bb8c87..b6ea9c5bdf 100644 --- a/codex-rs/cloud-tasks/src/lib.rs +++ b/codex-rs/cloud-tasks/src/lib.rs @@ -12,6 +12,10 @@ use chrono::Utc; use codex_cloud_tasks_client::TaskStatus; use codex_git_utils::current_branch_name; use codex_git_utils::default_branch_name; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; use codex_login::default_client::get_codex_user_agent; use owo_colors::OwoColorize; use owo_colors::Stream; @@ -38,6 +42,7 @@ struct ApplyJob { struct BackendContext { backend: Arc, base_url: String, + environment_http: RouteAwareClientPool, } async fn init_backend(user_agent_suffix: &str) -> anyhow::Result { @@ -56,11 +61,20 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result return Ok(BackendContext { backend: Arc::new(codex_cloud_tasks_mock_client::MockClient), base_url, + environment_http: RouteAwareClientPool::new( + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ClientRouteClass::Api, + ), }); } 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 environment_http = + RouteAwareClientPool::new(http_client_factory.clone(), ClientRouteClass::Api); + 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 +82,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, @@ -103,6 +116,7 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result Ok(BackendContext { backend: Arc::new(http), base_url, + environment_http, }) } @@ -190,7 +204,8 @@ async fn resolve_environment_id(ctx: &BackendContext, requested: &str) -> anyhow } let normalized = util::normalize_base_url(&ctx.base_url); let headers = util::build_chatgpt_headers().await; - let environments = crate::env_detect::list_environments(&normalized, &headers).await?; + let environments = + crate::env_detect::list_environments(&ctx.environment_http, &normalized, &headers).await?; if environments.is_empty() { return Err(anyhow!( "no cloud environments are available for this workspace" @@ -757,8 +772,11 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an .try_init(); info!("Launching Cloud Tasks list UI"); - let BackendContext { backend, .. } = init_backend("codex_cloud_tasks_tui").await?; - let backend = backend; + let BackendContext { + backend, + base_url, + environment_http, + } = init_backend("codex_cloud_tasks_tui").await?; // Terminal setup use crossterm::ExecutableCommand; @@ -838,34 +856,25 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an }); } // Fetch environment list in parallel so the header can show friendly names quickly. - { - let tx = tx.clone(); - tokio::spawn(async move { - let base_url = util::normalize_base_url( - &std::env::var("CODEX_CLOUD_TASKS_BASE_URL") - .unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string()), - ); - let headers = util::build_chatgpt_headers().await; - let res = crate::env_detect::list_environments(&base_url, &headers).await; - let _ = tx.send(app::AppEvent::EnvironmentsLoaded(res)); - }); - } + spawn_environment_load(tx.clone(), base_url.clone(), environment_http.clone()); // Try to auto-detect a likely environment id on startup and refresh if found. // Do this concurrently so the initial list shows quickly; on success we refetch with filter. { let tx = tx.clone(); + let base_url = base_url.clone(); + let environment_http = environment_http.clone(); tokio::spawn(async move { - let base_url = util::normalize_base_url( - &std::env::var("CODEX_CLOUD_TASKS_BASE_URL") - .unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string()), - ); + let base_url = util::normalize_base_url(&base_url); // Build headers: UA + ChatGPT auth if available let headers = util::build_chatgpt_headers().await; // Run autodetect. If it fails, we keep using "All". let res = crate::env_detect::autodetect_environment_id( - &base_url, &headers, /*desired_label*/ None, + &environment_http, + &base_url, + &headers, + /*desired_label*/ None, ) .await; let _ = tx.send(app::AppEvent::EnvironmentAutodetected(res)); @@ -1079,18 +1088,11 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an } // Proactively fetch environments to resolve a friendly name for the header. app.env_loading = true; - { - let tx = tx.clone(); - tokio::spawn(async move { - let base_url = crate::util::normalize_base_url( - &std::env::var("CODEX_CLOUD_TASKS_BASE_URL") - .unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string()), - ); - let headers = crate::util::build_chatgpt_headers().await; - let res = crate::env_detect::list_environments(&base_url, &headers).await; - let _ = tx.send(app::AppEvent::EnvironmentsLoaded(res)); - }); - } + spawn_environment_load( + tx.clone(), + base_url.clone(), + environment_http.clone(), + ); let _ = frame_tx.send(Instant::now()); } } @@ -1466,13 +1468,11 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an } needs_redraw = true; if should_fetch { - let tx = tx.clone(); - tokio::spawn(async move { - let base_url = crate::util::normalize_base_url(&std::env::var("CODEX_CLOUD_TASKS_BASE_URL").unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string())); - let headers = crate::util::build_chatgpt_headers().await; - let res = crate::env_detect::list_environments(&base_url, &headers).await; - let _ = tx.send(app::AppEvent::EnvironmentsLoaded(res)); - }); + spawn_environment_load( + tx.clone(), + base_url.clone(), + environment_http.clone(), + ); } // Render after opening env modal to show it instantly. render_if_needed(&mut terminal, &mut app, &mut needs_redraw)?; @@ -1652,16 +1652,11 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an if app.environments.is_empty() { app.env_loading = true; app.env_error = None; } needs_redraw = true; if app.environments.is_empty() { - let tx = tx.clone(); - tokio::spawn(async move { - let base_url = crate::util::normalize_base_url( - &std::env::var("CODEX_CLOUD_TASKS_BASE_URL") - .unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string()), - ); - let headers = crate::util::build_chatgpt_headers().await; - let res = crate::env_detect::list_environments(&base_url, &headers).await; - let _ = tx.send(app::AppEvent::EnvironmentsLoaded(res)); - }); + spawn_environment_load( + tx.clone(), + base_url.clone(), + environment_http.clone(), + ); } } KeyCode::Left => { @@ -1831,13 +1826,11 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an if should_fetch { app.env_loading = true; app.env_error = None; } needs_redraw = true; if should_fetch { - let tx = tx.clone(); - tokio::spawn(async move { - let base_url = crate::util::normalize_base_url(&std::env::var("CODEX_CLOUD_TASKS_BASE_URL").unwrap_or_else(|_| "https://chatgpt.com/backend-api".to_string())); - let headers = crate::util::build_chatgpt_headers().await; - let res = crate::env_detect::list_environments(&base_url, &headers).await; - let _ = tx.send(app::AppEvent::EnvironmentsLoaded(res)); - }); + spawn_environment_load( + tx.clone(), + base_url.clone(), + environment_http.clone(), + ); } } KeyCode::Char('n') => { @@ -2019,6 +2012,19 @@ pub async fn run_main(cli: Cli, _codex_linux_sandbox_exe: Option) -> an Ok(()) } +fn spawn_environment_load( + tx: UnboundedSender, + base_url: String, + http: RouteAwareClientPool, +) { + tokio::spawn(async move { + let base_url = util::normalize_base_url(&base_url); + let headers = util::build_chatgpt_headers().await; + let result = crate::env_detect::list_environments(&http, &base_url, &headers).await; + let _ = tx.send(app::AppEvent::EnvironmentsLoaded(result)); + }); +} + // extract_chatgpt_account_id moved to util.rs /// Build plain-text conversation lines: a labeled user prompt followed by assistant messages. diff --git a/codex-rs/cloud-tasks/src/util.rs b/codex-rs/cloud-tasks/src/util.rs index 2a836f8579..aff4cfe552 100644 --- a/codex-rs/cloud-tasks/src/util.rs +++ b/codex-rs/cloud-tasks/src/util.rs @@ -1,9 +1,11 @@ use chrono::DateTime; use chrono::Local; use chrono::Utc; -use reqwest::header::HeaderMap; +use http::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,28 +43,35 @@ 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`, /// and optional `ChatGPT-Account-Id`. pub async fn build_chatgpt_headers() -> HeaderMap { - use reqwest::header::HeaderValue; - use reqwest::header::USER_AGENT; + use http::header::HeaderValue; + use http::header::USER_AGENT; set_user_agent_suffix("codex_cloud_tasks_tui"); let ua = codex_login::default_client::get_codex_user_agent(); @@ -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/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index e85d16a126..1e9b2e9e58 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/codex-rs/core-plugins/Cargo.toml @@ -22,6 +22,7 @@ codex-core-skills = { workspace = true } codex-exec-server = { workspace = true } codex-git-utils = { workspace = true } codex-hooks = { workspace = true } +codex-http-client = { workspace = true } codex-login = { workspace = true } codex-mcp = { workspace = true } codex-model-provider = { workspace = true } @@ -36,7 +37,7 @@ codex-utils-plugins = { workspace = true } chrono = { workspace = true } dirs = { workspace = true } flate2 = { workspace = true } -reqwest = { workspace = true } +http = { workspace = true } regex = { workspace = true } semver = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/codex-rs/core-plugins/src/discoverable_tests.rs b/codex-rs/core-plugins/src/discoverable_tests.rs index 48482ed430..1dd88969ac 100644 --- a/codex-rs/core-plugins/src/discoverable_tests.rs +++ b/codex-rs/core-plugins/src/discoverable_tests.rs @@ -851,9 +851,10 @@ plugins = true let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf()); fetch_and_cache_global_remote_plugin_catalog( codex_home.path(), - &RemotePluginServiceConfig { - chatgpt_base_url: plugins.chatgpt_base_url.clone(), - }, + &RemotePluginServiceConfig::new( + plugins.chatgpt_base_url.clone(), + crate::test_support::test_http_client_factory(), + ), Some(&auth), ) .await diff --git a/codex-rs/core-plugins/src/http_client_selector.rs b/codex-rs/core-plugins/src/http_client_selector.rs new file mode 100644 index 0000000000..f7dcc349f9 --- /dev/null +++ b/codex-rs/core-plugins/src/http_client_selector.rs @@ -0,0 +1,23 @@ +use codex_http_client::HttpClient; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareClientPoolError; +use std::fmt::Debug; +use std::future::Future; +use std::pin::Pin; + +pub(crate) type HttpClientSelectionFuture<'a> = + Pin> + Send + 'a>>; + +/// Selects an HTTP client after resolving the outbound route for an exact request URL. +/// +/// Implementations are expected to preserve request-specific route selection while reusing +/// transport clients when multiple URLs resolve to the same route. +pub(crate) trait HttpClientSelector: Debug + Send + Sync { + fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a>; +} + +impl HttpClientSelector for RouteAwareClientPool { + fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a> { + Box::pin(RouteAwareClientPool::client_for_url(self, request_url)) + } +} diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index df30305c38..4998586516 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -1,5 +1,6 @@ mod app_mcp_routing; mod discoverable; +mod http_client_selector; pub mod installed_marketplaces; pub mod loader; mod manager; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 16fb4c8663..eda805d6d3 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -69,6 +69,7 @@ use codex_core_skills::SkillMetadata; use codex_core_skills::config_rules::SkillConfigRules; use codex_core_skills::config_rules::skill_config_rules_from_stack; use codex_hooks::plugin_hook_declarations; +use codex_http_client::HttpClientFactory; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_plugin::AppConnectorId; @@ -110,6 +111,7 @@ pub struct PluginsConfigInput { pub plugins_enabled: bool, pub remote_plugin_enabled: bool, pub chatgpt_base_url: String, + http_client_factory: HttpClientFactory, } impl PluginsConfigInput { @@ -118,14 +120,24 @@ impl PluginsConfigInput { plugins_enabled: bool, remote_plugin_enabled: bool, chatgpt_base_url: String, + http_client_factory: HttpClientFactory, ) -> Self { Self { config_layer_stack, plugins_enabled, remote_plugin_enabled, chatgpt_base_url, + http_client_factory, } } + + /// Builds route-aware service state for remote plugin requests. + pub fn remote_plugin_service_config(&self) -> RemotePluginServiceConfig { + RemotePluginServiceConfig::new( + self.chatgpt_base_url.clone(), + self.http_client_factory.clone(), + ) + } } /// Inputs used to select endpoint-backed plugin install candidates. @@ -223,9 +235,7 @@ struct ConfiguredMarketplaceUpgradeState { } fn remote_plugin_service_config(config: &PluginsConfigInput) -> RemotePluginServiceConfig { - RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - } + config.remote_plugin_service_config() } fn featured_plugin_ids_cache_key( @@ -1942,7 +1952,7 @@ impl PluginsManager { let use_remote_global_catalog = config.remote_plugin_enabled && auth_manager.current_auth_uses_codex_backend(); if !use_remote_global_catalog { - self.start_curated_repo_sync(); + self.start_curated_repo_sync(config.http_client_factory.clone()); } let should_spawn_marketplace_auto_upgrade = { let mut state = match self.configured_marketplace_upgrade_state.write() { @@ -2298,7 +2308,7 @@ impl PluginsManager { } } - fn start_curated_repo_sync(self: &Arc) { + fn start_curated_repo_sync(self: &Arc, http_client_factory: HttpClientFactory) { if CURATED_REPO_SYNC_STARTED.swap(true, Ordering::SeqCst) { return; } @@ -2306,8 +2316,8 @@ impl PluginsManager { let codex_home = self.codex_home.clone(); if let Err(err) = std::thread::Builder::new() .name("plugins-curated-repo-sync".to_string()) - .spawn( - move || match sync_openai_plugins_repo(codex_home.as_path()) { + .spawn(move || { + match sync_openai_plugins_repo(codex_home.as_path(), http_client_factory) { Ok(curated_plugin_version) => { let configured_curated_plugin_ids = configured_curated_plugin_ids_from_codex_home(codex_home.as_path()); @@ -2331,8 +2341,8 @@ impl PluginsManager { CURATED_REPO_SYNC_STARTED.store(false, Ordering::SeqCst); warn!("failed to sync curated plugins repo: {err}"); } - }, - ) + } + }) { CURATED_REPO_SYNC_STARTED.store(false, Ordering::SeqCst); warn!("failed to start curated plugins repo sync task: {err}"); @@ -2718,6 +2728,7 @@ fn remote_plugin_mutation_error_type(err: &RemotePluginMutationError) -> &'stati RemotePluginMutationError::AuthToken(_) => "remote_mutation_auth_token", RemotePluginMutationError::InvalidBaseUrl(_) => "remote_mutation_invalid_base_url", RemotePluginMutationError::InvalidBaseUrlPath => "remote_mutation_invalid_base_url_path", + RemotePluginMutationError::HttpClient { .. } => "remote_mutation_http_client", RemotePluginMutationError::Request { .. } => "remote_mutation_request", RemotePluginMutationError::UnexpectedStatus { .. } => "remote_mutation_unexpected_status", RemotePluginMutationError::Decode { .. } => "remote_mutation_decode", diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index 5e743416f2..ac5b046bc7 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -20,6 +20,7 @@ use crate::startup_sync::curated_plugins_repo_path; use crate::test_support::TEST_CURATED_PLUGIN_CACHE_VERSION; use crate::test_support::TEST_CURATED_PLUGIN_SHA; use crate::test_support::load_plugins_config as load_plugins_config_input; +use crate::test_support::test_http_client_factory; use crate::test_support::write_curated_plugin; use crate::test_support::write_curated_plugin_sha_with as write_curated_plugin_sha; use crate::test_support::write_file; @@ -110,6 +111,7 @@ fn plugins_config_input_with_requirements( /*plugins_enabled*/ true, /*remote_plugin_enabled*/ false, String::new(), + test_http_client_factory(), ) } @@ -2477,6 +2479,7 @@ async fn plugin_cache_ignores_unrelated_session_overrides() { /*plugins_enabled*/ true, /*remote_plugin_enabled*/ false, "https://chatgpt.com".to_string(), + test_http_client_factory(), ) }; let manager = PluginsManager::new(codex_home.path().to_path_buf()); diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index b9b94f4ded..f245eaf1c1 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -1,4 +1,5 @@ use crate::app_mcp_routing::apply_app_mcp_routing_policy; +use crate::http_client_selector::HttpClientSelector; use crate::loader::plugin_app_declarations_from_value; use crate::store::PLUGINS_CACHE_DIR; use crate::store::PluginStore; @@ -9,8 +10,13 @@ use codex_app_server_protocol::PluginInstallPolicy; use codex_app_server_protocol::PluginInstallPolicySource; use codex_app_server_protocol::PluginInterface; use codex_app_server_protocol::SkillInterface; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClient; +use codex_http_client::HttpClientFactory; +use codex_http_client::RequestBuilder; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareClientPoolError; use codex_login::CodexAuth; -use codex_login::default_client::build_reqwest_client; use codex_plugin::AppConnectorId; use codex_plugin::AppDeclaration; use codex_plugin::PluginCapabilitySummary; @@ -18,7 +24,7 @@ use codex_plugin::PluginId; use codex_plugin::app_connector_ids_from_declarations; use codex_plugin::prompt_safe_plugin_description; use codex_utils_absolute_path::AbsolutePathBuf; -use reqwest::RequestBuilder; +use http::StatusCode; use serde::Deserialize; use serde::Serialize; use serde_json::Value as JsonValue; @@ -29,6 +35,7 @@ use std::collections::HashSet; use std::fs; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use tracing::instrument; use url::Url; @@ -118,11 +125,59 @@ const REMOTE_INSTALLED_MARKETPLACE_DISPLAY_ORDER: [(&str, &str); 6] = [ ), ]; -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct RemotePluginServiceConfig { pub chatgpt_base_url: String, + pub(crate) http_client_factory: HttpClientFactory, + pub(crate) http_clients: Arc, } +impl RemotePluginServiceConfig { + /// Creates remote plugin service state from the effective application HTTP configuration. + /// + /// Keeping the factory mandatory ensures every catalog, mutation, upload, and bundle request + /// follows the same outbound proxy policy. + pub fn new(chatgpt_base_url: String, http_client_factory: HttpClientFactory) -> Self { + let http_clients = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + http_client_factory.clone(), + ClientRouteClass::Api, + ); + Self { + chatgpt_base_url, + http_client_factory, + http_clients: Arc::new(http_clients), + } + } + + pub(crate) async fn http_client_for_url( + &self, + request_url: &str, + ) -> Result { + self.http_clients.client_for_url(request_url).await + } + + async fn catalog_client_for_url( + &self, + request_url: &str, + ) -> Result { + self.http_client_for_url(request_url) + .await + .map_err(|source| RemotePluginCatalogError::HttpClient { + url: request_url.to_string(), + source, + }) + } +} + +impl PartialEq for RemotePluginServiceConfig { + fn eq(&self, other: &Self) -> bool { + self.chatgpt_base_url == other.chatgpt_base_url + && self.http_client_factory == other.http_client_factory + } +} + +impl Eq for RemotePluginServiceConfig {} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct RemotePluginUninstallTarget { pub plugin_id: PluginId, @@ -305,17 +360,24 @@ pub enum RemotePluginCatalogError { #[error("failed to read auth token for remote plugin catalog: {0}")] AuthToken(#[source] std::io::Error), + #[error("failed to build HTTP client for remote plugin request to {url}: {source}")] + HttpClient { + url: String, + #[source] + source: RouteAwareClientPoolError, + }, + #[error("failed to send remote plugin catalog request to {url}: {source}")] Request { url: String, #[source] - source: reqwest::Error, + source: codex_http_client::HttpError, }, #[error("remote plugin catalog request to {url} failed with status {status}: {body}")] UnexpectedStatus { url: String, - status: reqwest::StatusCode, + status: StatusCode, body: String, }, @@ -850,11 +912,13 @@ pub async fn fetch_recommended_plugins( ) -> Result { let auth = ensure_chatgpt_auth(auth)?; let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/suggested"); - let client = build_reqwest_client(); - let request = authenticated_request(client.get(&url), auth)? - .timeout(RECOMMENDED_PLUGINS_TIMEOUT) - .query(&[("scope", "GLOBAL")]); + let mut url = Url::parse(&format!("{base_url}/ps/plugins/suggested")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut().append_pair("scope", "GLOBAL"); + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = + authenticated_request(client.get(&url), auth)?.timeout(RECOMMENDED_PLUGINS_TIMEOUT); let response: RecommendedPluginsResponse = send_and_decode(request, &url).await?; Ok(recommended_plugins_mode(response)) } @@ -1160,7 +1224,7 @@ pub async fn fetch_remote_plugin_skill_detail( } let url = remote_plugin_skill_detail_url(config, plugin_id, skill_name)?; - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.get(&url), auth)?; let response: RemotePluginSkillDetailResponse = send_and_decode(request, &url).await?; if response.plugin_id != plugin_id { @@ -1315,14 +1379,13 @@ pub async fn install_remote_plugin( // marketplace name is not validated before sending the install mutation. let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/{plugin_id}/install"); - let client = build_reqwest_client(); - let request = authenticated_request( - client - .post(&url) - .query(&[("includeAppsNeedingAuth", "true")]), - auth, - )?; + let mut url = Url::parse(&format!("{base_url}/ps/plugins/{plugin_id}/install")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut() + .append_pair("includeAppsNeedingAuth", "true"); + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.post(&url), auth)?; let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?; if response.id != plugin_id { return Err(RemotePluginCatalogError::UnexpectedPluginId { @@ -1409,7 +1472,7 @@ pub async fn uninstall_remote_plugin( let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/ps/plugins/{remote_plugin_id}/uninstall"); - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.post(&url), auth)?; let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?; if response.id != remote_plugin_id { @@ -1827,17 +1890,20 @@ async fn get_remote_plugin_list_page( collection: Option<&str>, ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/list"); - let client = build_reqwest_client(); - let mut request = authenticated_request(client.get(&url), auth)?; - request = request.query(&[("scope", scope.api_value())]); - request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]); + let mut url = Url::parse(&format!("{base_url}/ps/plugins/list")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut() + .append_pair("scope", scope.api_value()) + .append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string()); if let Some(collection) = collection { - request = request.query(&[("collection", collection)]); + url.query_pairs_mut().append_pair("collection", collection); } if let Some(page_token) = page_token { - request = request.query(&[("pageToken", page_token)]); + url.query_pairs_mut().append_pair("pageToken", page_token); } + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.get(&url), auth)?; send_and_decode(request, &url).await } @@ -1847,13 +1913,16 @@ async fn get_remote_shared_workspace_plugins_page( page_token: Option<&str>, ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/workspace/shared"); - let client = build_reqwest_client(); - let mut request = authenticated_request(client.get(&url), auth)?; - request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]); + let mut url = Url::parse(&format!("{base_url}/ps/plugins/workspace/shared")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut() + .append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string()); if let Some(page_token) = page_token { - request = request.query(&[("pageToken", page_token)]); + url.query_pairs_mut().append_pair("pageToken", page_token); } + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.get(&url), auth)?; send_and_decode(request, &url).await } @@ -1865,16 +1934,20 @@ async fn get_remote_plugin_installed_page( include_download_urls: bool, ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/installed"); - let client = build_reqwest_client(); - let mut request = authenticated_request(client.get(&url), auth)?; - request = request.query(&[("scope", scope.api_value())]); + let mut url = Url::parse(&format!("{base_url}/ps/plugins/installed")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut() + .append_pair("scope", scope.api_value()); if include_download_urls { - request = request.query(&[("includeDownloadUrls", true)]); + url.query_pairs_mut() + .append_pair("includeDownloadUrls", "true"); } if let Some(page_token) = page_token { - request = request.query(&[("pageToken", page_token)]); + url.query_pairs_mut().append_pair("pageToken", page_token); } + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.get(&url), auth)?; send_and_decode(request, &url).await } @@ -1885,12 +1958,15 @@ async fn fetch_plugin_detail( include_download_urls: bool, ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/{plugin_id}"); - let client = build_reqwest_client(); - let mut request = authenticated_request(client.get(&url), auth)?; + let mut url = Url::parse(&format!("{base_url}/ps/plugins/{plugin_id}")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; if include_download_urls { - request = request.query(&[("includeDownloadUrls", true)]); + url.query_pairs_mut() + .append_pair("includeDownloadUrls", "true"); } + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.get(&url), auth)?; send_and_decode(request, &url).await } diff --git a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs index 2a23b066ed..24704acf6a 100644 --- a/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs +++ b/codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs @@ -248,6 +248,7 @@ pub async fn sync_remote_installed_plugin_bundles_once( }; match crate::remote_bundle::download_and_install_remote_plugin_bundle( + config, codex_home.clone(), bundle, ) @@ -542,9 +543,10 @@ mod tests { .expect(1) .mount(&server) .await; - let config = RemotePluginServiceConfig { - chatgpt_base_url: format!("{}/backend-api", server.uri()), - }; + let config = RemotePluginServiceConfig::new( + format!("{}/backend-api", server.uri()), + crate::test_support::test_http_client_factory(), + ); let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); let outcome = sync_remote_installed_plugin_bundles_once( diff --git a/codex-rs/core-plugins/src/remote/share.rs b/codex-rs/core-plugins/src/remote/share.rs index c400d27c47..6ef5aa88ed 100644 --- a/codex-rs/core-plugins/src/remote/share.rs +++ b/codex-rs/core-plugins/src/remote/share.rs @@ -1,17 +1,17 @@ use super::*; use crate::plugin_bundle_archive::PluginBundlePackError; use crate::plugin_bundle_archive::pack_plugin_bundle_tar_gz; +use codex_http_client::RequestBuilder; use codex_login::CodexAuth; -use codex_login::default_client::build_reqwest_client; use codex_utils_absolute_path::AbsolutePathBuf; -use reqwest::RequestBuilder; -use reqwest::StatusCode; +use http::StatusCode; use serde::Deserialize; use serde::Serialize; use std::collections::BTreeMap; use std::io; use std::path::Path; use tracing::warn; +use url::Url; mod checkout; mod local_paths; @@ -163,7 +163,7 @@ pub async fn save_remote_plugin_share( let etag = upload .etag .ok_or(RemotePluginCatalogError::MissingUploadEtag)?; - put_workspace_plugin_upload(&upload.upload_url, archive_bytes).await?; + put_workspace_plugin_upload(config, &upload.upload_url, archive_bytes).await?; let share_targets = access_policy.share_targets; let share_targets = ensure_unlisted_workspace_target(auth, access_policy.discoverability, share_targets)?; @@ -279,7 +279,7 @@ pub async fn delete_remote_plugin_share( let auth = ensure_chatgpt_auth(auth)?; let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/public/plugins/workspace/{remote_plugin_id}"); - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.delete(&url), auth)?; send_and_expect_status(request, &url, &[StatusCode::NO_CONTENT]).await?; if let Err(err) = local_paths::remove_plugin_share_local_path(codex_home, remote_plugin_id) { @@ -312,7 +312,7 @@ pub async fn update_remote_plugin_share_targets( .unwrap_or_default(); let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/ps/plugins/{remote_plugin_id}/shares"); - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.put(&url), auth)?.json( &RemotePluginShareUpdateTargetsRequest { discoverability, @@ -377,13 +377,16 @@ async fn get_created_workspace_plugins_page( page_token: Option<&str>, ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/ps/plugins/workspace/created"); - let client = build_reqwest_client(); - let mut request = authenticated_request(client.get(&url), auth)?; - request = request.query(&[("limit", REMOTE_PLUGIN_LIST_PAGE_LIMIT)]); + let mut url = Url::parse(&format!("{base_url}/ps/plugins/workspace/created")) + .map_err(RemotePluginCatalogError::InvalidBaseUrl)?; + url.query_pairs_mut() + .append_pair("limit", &REMOTE_PLUGIN_LIST_PAGE_LIMIT.to_string()); if let Some(page_token) = page_token { - request = request.query(&[("pageToken", page_token)]); + url.query_pairs_mut().append_pair("pageToken", page_token); } + let url = url.to_string(); + let client = config.catalog_client_for_url(&url).await?; + let request = authenticated_request(client.get(&url), auth)?; send_and_decode(request, &url).await } @@ -396,7 +399,7 @@ async fn create_workspace_plugin_upload( ) -> Result { let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/public/plugins/workspace/upload-url"); - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.post(&url), auth)?.json( &RemoteWorkspacePluginUploadUrlRequest { filename, @@ -409,10 +412,11 @@ async fn create_workspace_plugin_upload( } async fn put_workspace_plugin_upload( + config: &RemotePluginServiceConfig, upload_url: &str, archive_bytes: Vec, ) -> Result<(), RemotePluginCatalogError> { - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(upload_url).await?; let request = client .put(upload_url) .timeout(REMOTE_PLUGIN_CATALOG_TIMEOUT) @@ -450,7 +454,7 @@ async fn finalize_workspace_plugin_upload( } else { format!("{base_url}/public/plugins/workspace") }; - let client = build_reqwest_client(); + let client = config.catalog_client_for_url(&url).await?; let request = authenticated_request(client.post(&url), auth)?.json(&body); send_and_decode(request, &url).await } diff --git a/codex-rs/core-plugins/src/remote/share/checkout.rs b/codex-rs/core-plugins/src/remote/share/checkout.rs index 6b12d2588d..f846a9b90e 100644 --- a/codex-rs/core-plugins/src/remote/share/checkout.rs +++ b/codex-rs/core-plugins/src/remote/share/checkout.rs @@ -93,6 +93,7 @@ pub async fn checkout_remote_plugin_share( )) })?; crate::remote_bundle::download_and_extract_remote_plugin_bundle_to_path( + config, bundle, local_plugin_path.clone(), ) diff --git a/codex-rs/core-plugins/src/remote/share/tests.rs b/codex-rs/core-plugins/src/remote/share/tests.rs index e9887370c5..374a740789 100644 --- a/codex-rs/core-plugins/src/remote/share/tests.rs +++ b/codex-rs/core-plugins/src/remote/share/tests.rs @@ -1,4 +1,6 @@ use super::*; +use crate::test_support::recorded_http_client_urls; +use crate::test_support::recording_remote_plugin_service_config; use codex_app_server_protocol::PluginAuthPolicy; use codex_app_server_protocol::PluginInstallPolicy; use codex_app_server_protocol::PluginInterface; @@ -23,9 +25,10 @@ use wiremock::matchers::query_param; use wiremock::matchers::query_param_is_missing; fn test_config(server: &MockServer) -> RemotePluginServiceConfig { - RemotePluginServiceConfig { - chatgpt_base_url: format!("{}/backend-api", server.uri()), - } + RemotePluginServiceConfig::new( + format!("{}/backend-api", server.uri()), + crate::test_support::test_http_client_factory(), + ) } fn test_auth() -> CodexAuth { @@ -174,7 +177,8 @@ async fn save_remote_plugin_share_creates_workspace_plugin() { .unwrap() .len(); let server = MockServer::start().await; - let config = test_config(&server); + let (config, selected_urls) = + recording_remote_plugin_service_config(format!("{}/backend-api", server.uri())); let auth = test_auth(); Mock::given(method("POST")) @@ -260,6 +264,17 @@ async fn save_remote_plugin_share_creates_workspace_plugin() { local_paths::load_plugin_share_local_paths(codex_home.path()).unwrap(), BTreeMap::from([("plugins_123".to_string(), plugin_path)]) ); + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![ + format!( + "{}/backend-api/public/plugins/workspace/upload-url", + server.uri() + ), + format!("{}/upload/file_123", server.uri()), + format!("{}/backend-api/public/plugins/workspace", server.uri()), + ] + ); let requests = server.received_requests().await.unwrap_or_default(); let upload_request = requests diff --git a/codex-rs/core-plugins/src/remote_bundle.rs b/codex-rs/core-plugins/src/remote_bundle.rs index 7a302f2f36..6ad14f5873 100644 --- a/codex-rs/core-plugins/src/remote_bundle.rs +++ b/codex-rs/core-plugins/src/remote_bundle.rs @@ -1,18 +1,19 @@ use crate::plugin_bundle_archive::PluginBundleUnpackError; use crate::plugin_bundle_archive::unpack_plugin_bundle_tar_gz; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; +use crate::remote::RemotePluginServiceConfig; use crate::store::PluginInstallResult; use crate::store::PluginStore; use crate::store::PluginStoreError; use crate::store::error_context_sub_error_type; use crate::store::validate_plugin_version_segment; -use codex_login::default_client::build_reqwest_client; +use codex_http_client::HttpResponse; +use codex_http_client::RouteAwareClientPoolError; use codex_plugin::PluginId; use codex_plugin::PluginIdError; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_plugins::find_plugin_manifest_path; -use reqwest::Response; -use reqwest::StatusCode; +use http::StatusCode; use serde_json::Value as JsonValue; use std::fs; use std::io; @@ -83,11 +84,18 @@ pub enum RemotePluginBundleInstallError { source: PluginIdError, }, + #[error("failed to build HTTP client for remote plugin bundle download from {url}: {source}")] + HttpClient { + url: String, + #[source] + source: RouteAwareClientPoolError, + }, + #[error("failed to send remote plugin bundle download request to {url}: {source}")] DownloadRequest { url: String, #[source] - source: reqwest::Error, + source: codex_http_client::HttpError, }, #[error("remote plugin bundle download from {url} failed with status {status}: {body}")] @@ -101,7 +109,7 @@ pub enum RemotePluginBundleInstallError { DownloadBody { url: String, #[source] - source: reqwest::Error, + source: codex_http_client::HttpError, }, #[error("remote plugin bundle download from {url} exceeded maximum size of {max_bytes} bytes")] @@ -144,6 +152,7 @@ impl RemotePluginBundleInstallError { | Self::InvalidBundleDownloadUrl { .. } | Self::UnsupportedBundleDownloadUrlScheme { .. } | Self::InvalidPluginId { .. } + | Self::HttpClient { .. } | Self::DownloadRequest { .. } | Self::DownloadStatus { .. } | Self::DownloadBody { .. } @@ -247,10 +256,12 @@ fn is_loopback_url(url: &Url) -> bool { } pub async fn download_and_install_remote_plugin_bundle( + config: &RemotePluginServiceConfig, codex_home: PathBuf, bundle: ValidatedRemotePluginBundle, ) -> Result { let bundle_bytes = download_remote_plugin_bundle_with_limit( + config, &bundle.bundle_download_url, /*max_bytes*/ REMOTE_PLUGIN_BUNDLE_MAX_DOWNLOAD_BYTES, ) @@ -267,10 +278,12 @@ pub async fn download_and_install_remote_plugin_bundle( } pub(crate) async fn download_and_extract_remote_plugin_bundle_to_path( + config: &RemotePluginServiceConfig, bundle: ValidatedRemotePluginBundle, destination: AbsolutePathBuf, ) -> Result { let bundle_bytes = download_remote_plugin_bundle_with_limit( + config, &bundle.bundle_download_url, /*max_bytes*/ REMOTE_PLUGIN_BUNDLE_MAX_DOWNLOAD_BYTES, ) @@ -287,10 +300,17 @@ pub(crate) async fn download_and_extract_remote_plugin_bundle_to_path( } async fn download_remote_plugin_bundle_with_limit( + config: &RemotePluginServiceConfig, bundle_download_url: &str, max_bytes: u64, ) -> Result, RemotePluginBundleInstallError> { - let client = build_reqwest_client(); + let client = config + .http_client_for_url(bundle_download_url) + .await + .map_err(|source| RemotePluginBundleInstallError::HttpClient { + url: bundle_download_url.to_string(), + source, + })?; let response = client .get(bundle_download_url) .timeout(REMOTE_PLUGIN_BUNDLE_DOWNLOAD_TIMEOUT) @@ -354,7 +374,7 @@ async fn download_remote_plugin_bundle_with_limit( } async fn read_response_body_with_limit( - mut response: Response, + mut response: HttpResponse, url: &str, max_bytes: u64, ) -> Result, RemotePluginBundleInstallError> { @@ -623,11 +643,18 @@ fn is_standard_plugin_root(path: &Path) -> bool { #[cfg(test)] mod tests { use super::*; + use crate::test_support::recorded_http_client_urls; + use crate::test_support::recording_remote_plugin_service_config; use flate2::Compression; use flate2::write::GzEncoder; use pretty_assertions::assert_eq; use std::io::Write; use tempfile::tempdir; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::method; + use wiremock::matchers::path; const REMOTE_PLUGIN_ID: &str = "plugins~Plugin_00000000000000000000000000000000"; @@ -739,6 +766,34 @@ mod tests { )); } + #[tokio::test] + async fn bundle_download_routes_the_backend_supplied_url() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/signed/plugin-bundle")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"bundle")) + .expect(1) + .mount(&server) + .await; + let (config, selected_urls) = + recording_remote_plugin_service_config(format!("{}/backend-api", server.uri())); + let download_url = format!("{}/signed/plugin-bundle?sig=signed-token", server.uri()); + + let err = + download_remote_plugin_bundle_with_limit(&config, &download_url, /*max_bytes*/ 64) + .await + .expect_err("plain HTTP final URL should remain unsupported"); + + assert!(matches!( + err, + RemotePluginBundleInstallError::UnsupportedBundleDownloadFinalUrl { .. } + )); + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![download_url] + ); + } + #[test] fn install_rejects_invalid_tar_gz_bundle() { let codex_home = tempdir().expect("tempdir"); diff --git a/codex-rs/core-plugins/src/remote_legacy.rs b/codex-rs/core-plugins/src/remote_legacy.rs index 137c33753b..6cdc4162f9 100644 --- a/codex-rs/core-plugins/src/remote_legacy.rs +++ b/codex-rs/core-plugins/src/remote_legacy.rs @@ -1,7 +1,8 @@ use crate::remote::RemotePluginServiceConfig; +use codex_http_client::RouteAwareClientPoolError; use codex_login::CodexAuth; -use codex_login::default_client::build_reqwest_client; use codex_protocol::protocol::Product; +use http::StatusCode; use serde::Deserialize; use std::time::Duration; use url::Url; @@ -35,17 +36,24 @@ pub enum RemotePluginMutationError { #[error("chatgpt base url cannot be used for plugin mutation")] InvalidBaseUrlPath, + #[error("failed to build HTTP client for remote plugin mutation request to {url}: {source}")] + HttpClient { + url: String, + #[source] + source: RouteAwareClientPoolError, + }, + #[error("failed to send remote plugin mutation request to {url}: {source}")] Request { url: String, #[source] - source: reqwest::Error, + source: codex_http_client::HttpError, }, #[error("remote plugin mutation failed with status {status} from {url}: {body}")] UnexpectedStatus { url: String, - status: reqwest::StatusCode, + status: StatusCode, body: String, }, @@ -73,17 +81,27 @@ pub enum RemotePluginMutationError { #[derive(Debug, thiserror::Error)] pub enum RemotePluginFetchError { + #[error("invalid chatgpt base url for remote featured plugin request: {0}")] + InvalidBaseUrl(#[source] url::ParseError), + + #[error("failed to build HTTP client for remote featured plugin request to {url}: {source}")] + HttpClient { + url: String, + #[source] + source: RouteAwareClientPoolError, + }, + #[error("failed to send remote featured plugin request to {url}: {source}")] Request { url: String, #[source] - source: reqwest::Error, + source: codex_http_client::HttpError, }, #[error("remote featured plugin request to {url} failed with status {status}: {body}")] UnexpectedStatus { url: String, - status: reqwest::StatusCode, + status: StatusCode, body: String, }, @@ -101,14 +119,21 @@ pub async fn fetch_remote_featured_plugin_ids( product: Option, ) -> Result, RemotePluginFetchError> { let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/plugins/featured"); - let client = build_reqwest_client(); + let mut url = Url::parse(&format!("{base_url}/plugins/featured")) + .map_err(RemotePluginFetchError::InvalidBaseUrl)?; + url.query_pairs_mut().append_pair( + "platform", + product.unwrap_or(Product::Codex).to_app_platform(), + ); + let url = url.to_string(); + let client = config.http_client_for_url(&url).await.map_err(|source| { + RemotePluginFetchError::HttpClient { + url: url.clone(), + source, + } + })?; let mut request = client .get(&url) - .query(&[( - "platform", - product.unwrap_or(Product::Codex).to_app_platform(), - )]) .timeout(REMOTE_FEATURED_PLUGIN_FETCH_TIMEOUT); if let Some(auth) = auth.filter(|auth| auth.uses_codex_backend()) { @@ -173,7 +198,12 @@ async fn post_remote_plugin_mutation( ) -> Result { let auth = ensure_codex_backend_auth(auth)?; let url = remote_plugin_mutation_url(config, plugin_id, action)?; - let client = build_reqwest_client(); + let client = config.http_client_for_url(&url).await.map_err(|source| { + RemotePluginMutationError::HttpClient { + url: url.clone(), + source, + } + })?; let request = client .post(url.clone()) .timeout(REMOTE_PLUGIN_MUTATION_TIMEOUT) diff --git a/codex-rs/core-plugins/src/remote_tests.rs b/codex-rs/core-plugins/src/remote_tests.rs index 29fe8add19..f9b057e11d 100644 --- a/codex-rs/core-plugins/src/remote_tests.rs +++ b/codex-rs/core-plugins/src/remote_tests.rs @@ -1,5 +1,47 @@ use super::*; +use crate::test_support::recorded_http_client_urls; +use crate::test_support::recording_remote_plugin_service_config; use pretty_assertions::assert_eq; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; + +#[tokio::test] +async fn remote_plugin_list_routes_the_complete_query_url() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "plugins": [], + "pagination": {"next_page_token": null}, + }))) + .expect(1) + .mount(&server) + .await; + let (config, selected_urls) = + recording_remote_plugin_service_config(format!("{}/backend-api", server.uri())); + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + + get_remote_plugin_list_page( + &config, + &auth, + RemotePluginScope::Global, + Some("next page/+"), + Some("vertical & special"), + ) + .await + .expect("plugin list request should succeed"); + + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![format!( + "{}/backend-api/ps/plugins/list?scope=GLOBAL&limit=200&collection=vertical+%26+special&pageToken=next+page%2F%2B", + server.uri() + )] + ); +} #[test] fn build_remote_marketplace_preserves_directory_order_and_appends_installed_only_plugins() { diff --git a/codex-rs/core-plugins/src/startup_sync.rs b/codex-rs/core-plugins/src/startup_sync.rs index 0dd437af8a..a0234da9f2 100644 --- a/codex-rs/core-plugins/src/startup_sync.rs +++ b/codex-rs/core-plugins/src/startup_sync.rs @@ -6,16 +6,19 @@ use std::process::Output; use std::process::Stdio; use std::time::Duration; +use crate::http_client_selector::HttpClientSelector; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClient; +use codex_http_client::HttpClientFactory; +use codex_http_client::RequestBuilder; +use codex_http_client::RouteAwareClientPool; use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_FINAL_METRIC; use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_METRIC; -use reqwest::Client; use serde::Deserialize; use tempfile::TempDir; use tracing::warn; use zip::ZipArchive; -use codex_login::default_client::build_reqwest_client; - const GITHUB_API_BASE_URL: &str = "https://api.github.com"; const GITHUB_API_ACCEPT_HEADER: &str = "application/vnd.github+json"; const GITHUB_API_VERSION_HEADER: &str = "2022-11-28"; @@ -92,7 +95,10 @@ fn curated_plugins_sha_path(codex_home: &Path) -> PathBuf { codex_home.join(CURATED_PLUGINS_SHA_FILE) } -pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result { +pub fn sync_openai_plugins_repo( + codex_home: &Path, + http_client_factory: HttpClientFactory, +) -> Result { #[cfg(target_os = "macos")] let git_binary = match which::which("git") { Ok(git_path) => macos_git_binary_from_path(git_path, apple_developer_tools_available()), @@ -106,6 +112,7 @@ pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result { git_binary.as_deref(), GITHUB_API_BASE_URL, CURATED_PLUGINS_BACKUP_ARCHIVE_API_URL, + &http_client_factory, ) } @@ -114,6 +121,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( git_binary: Option<&Path>, api_base_url: &str, backup_archive_api_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let _file_guard = lock_curated_plugins_startup_sync(codex_home)?; @@ -134,7 +142,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( error = %err, "git sync failed for curated plugin sync; falling back to GitHub HTTP" ); - match sync_openai_plugins_repo_via_http(codex_home, api_base_url) { + match sync_openai_plugins_repo_via_http(codex_home, api_base_url, http_client_factory) { Ok(remote_sha) => { emit_curated_plugins_startup_sync_metric("http", "success"); emit_curated_plugins_startup_sync_final_metric("http", "success"); @@ -162,6 +170,7 @@ fn sync_openai_plugins_repo_with_transport_overrides( let result = sync_openai_plugins_repo_via_backup_archive( codex_home, backup_archive_api_url, + http_client_factory, ); let status = if result.is_ok() { "success" } else { "failure" }; emit_curated_plugins_startup_sync_metric("export_archive", status); @@ -319,6 +328,7 @@ fn run_git_in_repo( fn sync_openai_plugins_repo_via_http( codex_home: &Path, api_base_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let repo_path = curated_plugins_repo_path(codex_home); let sha_path = codex_home.join(CURATED_PLUGINS_SHA_FILE); @@ -326,7 +336,12 @@ fn sync_openai_plugins_repo_via_http( .enable_all() .build() .map_err(|err| format!("failed to create curated plugins sync runtime: {err}"))?; - let remote_sha = runtime.block_on(fetch_curated_repo_remote_sha(api_base_url))?; + let http_clients = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + http_client_factory.clone(), + ClientRouteClass::Api, + ); + let remote_sha = + runtime.block_on(fetch_curated_repo_remote_sha(&http_clients, api_base_url))?; let local_sha = read_sha_file(&sha_path); if local_sha.as_deref() == Some(remote_sha.as_str()) && repo_path.is_dir() { @@ -334,7 +349,11 @@ fn sync_openai_plugins_repo_via_http( } let staged_repo_dir = prepare_curated_repo_parent_and_temp_dir(&repo_path)?; - let zipball_bytes = runtime.block_on(fetch_curated_repo_zipball(api_base_url, &remote_sha))?; + let zipball_bytes = runtime.block_on(fetch_curated_repo_zipball( + &http_clients, + api_base_url, + &remote_sha, + ))?; extract_zipball_to_dir(&zipball_bytes, staged_repo_dir.path())?; ensure_marketplace_manifest_exists(staged_repo_dir.path())?; activate_curated_repo(&repo_path, staged_repo_dir)?; @@ -345,6 +364,7 @@ fn sync_openai_plugins_repo_via_http( fn sync_openai_plugins_repo_via_backup_archive( codex_home: &Path, backup_archive_api_url: &str, + http_client_factory: &HttpClientFactory, ) -> Result { let repo_path = curated_plugins_repo_path(codex_home); let sha_path = curated_plugins_sha_path(codex_home); @@ -353,7 +373,12 @@ fn sync_openai_plugins_repo_via_backup_archive( .build() .map_err(|err| format!("failed to create curated plugins sync runtime: {err}"))?; let staged_repo_dir = prepare_curated_repo_parent_and_temp_dir(&repo_path)?; + let http_clients = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + http_client_factory.clone(), + ClientRouteClass::Api, + ); let zipball_bytes = runtime.block_on(fetch_curated_repo_backup_archive_zip( + &http_clients, backup_archive_api_url, ))?; extract_zipball_to_dir(&zipball_bytes, staged_repo_dir.path())?; @@ -763,11 +788,14 @@ fn ensure_git_success(output: &Output, context: &str) -> Result<(), String> { } } -async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result { +async fn fetch_curated_repo_remote_sha( + http_clients: &dyn HttpClientSelector, + api_base_url: &str, +) -> Result { let api_base_url = api_base_url.trim_end_matches('/'); let repo_url = format!("{api_base_url}/repos/{OPENAI_PLUGINS_OWNER}/{OPENAI_PLUGINS_REPO}"); - let client = build_reqwest_client(); - let repo_body = fetch_github_text(&client, &repo_url, "get curated plugins repository").await?; + let repo_body = + fetch_github_text(http_clients, &repo_url, "get curated plugins repository").await?; let repo_summary: GitHubRepositorySummary = serde_json::from_str(&repo_body).map_err(|err| { format!("failed to parse curated plugins repository response from {repo_url}: {err}") @@ -780,7 +808,7 @@ async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result Result Result, String> { let api_base_url = api_base_url.trim_end_matches('/'); let repo_url = format!("{api_base_url}/repos/{OPENAI_PLUGINS_OWNER}/{OPENAI_PLUGINS_REPO}"); let zipball_url = format!("{repo_url}/zipball/{remote_sha}"); - let client = build_reqwest_client(); - fetch_github_bytes(&client, &zipball_url, "download curated plugins archive").await + fetch_github_bytes( + http_clients, + &zipball_url, + "download curated plugins archive", + ) + .await } async fn fetch_curated_repo_backup_archive_zip( + http_clients: &dyn HttpClientSelector, backup_archive_api_url: &str, ) -> Result, String> { - let client = build_reqwest_client(); let export_body = fetch_public_text( - &client, + http_clients, backup_archive_api_url, "get curated plugins export archive metadata", ) @@ -827,7 +860,7 @@ async fn fetch_curated_repo_backup_archive_zip( } fetch_public_bytes( - &client, + http_clients, &export_response.download_url, "download curated plugins export archive", ) @@ -924,8 +957,13 @@ fn read_git_ref_sha(git_dir: &Path, reference: &str) -> Result { )) } -async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result { - let response = github_request(client, url) +async fn fetch_github_text( + http_clients: &dyn HttpClientSelector, + url: &str, + context: &str, +) -> Result { + let client = client_for_url(http_clients, url, context).await?; + let response = github_request(&client, url) .send() .await .map_err(|err| format!("failed to {context} from {url}: {err}"))?; @@ -939,8 +977,13 @@ async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result< Ok(body) } -async fn fetch_github_bytes(client: &Client, url: &str, context: &str) -> Result, String> { - let response = github_request(client, url) +async fn fetch_github_bytes( + http_clients: &dyn HttpClientSelector, + url: &str, + context: &str, +) -> Result, String> { + let client = client_for_url(http_clients, url, context).await?; + let response = github_request(&client, url) .send() .await .map_err(|err| format!("failed to {context} from {url}: {err}"))?; @@ -958,7 +1001,12 @@ async fn fetch_github_bytes(client: &Client, url: &str, context: &str) -> Result Ok(body.to_vec()) } -async fn fetch_public_text(client: &Client, url: &str, context: &str) -> Result { +async fn fetch_public_text( + http_clients: &dyn HttpClientSelector, + url: &str, + context: &str, +) -> Result { + let client = client_for_url(http_clients, url, context).await?; let response = client .get(url) .timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT) @@ -975,7 +1023,12 @@ async fn fetch_public_text(client: &Client, url: &str, context: &str) -> Result< Ok(body) } -async fn fetch_public_bytes(client: &Client, url: &str, context: &str) -> Result, String> { +async fn fetch_public_bytes( + http_clients: &dyn HttpClientSelector, + url: &str, + context: &str, +) -> Result, String> { + let client = client_for_url(http_clients, url, context).await?; let response = client .get(url) .timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT) @@ -996,7 +1049,18 @@ async fn fetch_public_bytes(client: &Client, url: &str, context: &str) -> Result Ok(body.to_vec()) } -fn github_request(client: &Client, url: &str) -> reqwest::RequestBuilder { +async fn client_for_url( + http_clients: &dyn HttpClientSelector, + url: &str, + context: &str, +) -> Result { + http_clients + .client_for_url(url) + .await + .map_err(|err| format!("failed to build HTTP client to {context} from {url}: {err}")) +} + +fn github_request(client: &HttpClient, url: &str) -> RequestBuilder { client .get(url) .timeout(CURATED_PLUGINS_HTTP_TIMEOUT) diff --git a/codex-rs/core-plugins/src/startup_sync_tests.rs b/codex-rs/core-plugins/src/startup_sync_tests.rs index a4ea778443..2ec11f94a6 100644 --- a/codex-rs/core-plugins/src/startup_sync_tests.rs +++ b/codex-rs/core-plugins/src/startup_sync_tests.rs @@ -1,4 +1,6 @@ use super::*; +use crate::test_support::RecordingHttpClientSelector; +use crate::test_support::recorded_http_client_urls; use pretty_assertions::assert_eq; use std::ffi::OsStr; use std::io::Write; @@ -17,6 +19,46 @@ use zip::write::SimpleFileOptions; const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567"; +#[tokio::test] +async fn backup_archive_routes_metadata_and_backend_supplied_download_urls() { + let metadata_server = MockServer::start().await; + let download_server = MockServer::start().await; + let download_url = format!( + "{}/files/curated-plugins.zip?sig=signed", + download_server.uri() + ); + Mock::given(method("GET")) + .and(path("/backend-api/plugins/export/curated")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({"download_url": download_url.clone()})), + ) + .expect(1) + .mount(&metadata_server) + .await; + Mock::given(method("GET")) + .and(path("/files/curated-plugins.zip")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"archive".to_vec())) + .expect(1) + .mount(&download_server) + .await; + let metadata_url = format!( + "{}/backend-api/plugins/export/curated", + metadata_server.uri() + ); + let (http_clients, selected_urls) = RecordingHttpClientSelector::new(); + + let body = fetch_curated_repo_backup_archive_zip(http_clients.as_ref(), &metadata_url) + .await + .expect("backup archive download should succeed"); + + assert_eq!(body, b"archive"); + assert_eq!( + recorded_http_client_urls(&selected_urls), + vec![metadata_url, download_url] + ); +} + #[test] fn git_command_sanitizes_ambient_repository_environment() { let command = git_command(Path::new("git")); @@ -196,6 +238,7 @@ async fn run_sync_with_transport_overrides( Some(git_binary.as_path()), &api_base_url, &backup_archive_api_url, + &crate::test_support::test_http_client_factory(), ) }) .await @@ -215,6 +258,7 @@ async fn run_sync_without_git( /*git_binary*/ None, &api_base_url, &backup_archive_api_url, + &crate::test_support::test_http_client_factory(), ) }) .await @@ -227,7 +271,11 @@ async fn run_http_sync( ) -> Result { let api_base_url = api_base_url.into(); tokio::task::spawn_blocking(move || { - sync_openai_plugins_repo_via_http(codex_home.as_path(), &api_base_url) + sync_openai_plugins_repo_via_http( + codex_home.as_path(), + &api_base_url, + &crate::test_support::test_http_client_factory(), + ) }) .await .expect("sync task should join") @@ -366,6 +414,7 @@ exit 1 Some(git_path.as_path()), "http://127.0.0.1:9", "http://127.0.0.1:9/backend-api/plugins/export/curated", + &crate::test_support::test_http_client_factory(), ) }; let first = scope.spawn(run_sync); diff --git a/codex-rs/core-plugins/src/test_support.rs b/codex-rs/core-plugins/src/test_support.rs index 9e06de9d16..7c9ab17697 100644 --- a/codex-rs/core-plugins/src/test_support.rs +++ b/codex-rs/core-plugins/src/test_support.rs @@ -1,19 +1,87 @@ use std::fs; use std::path::Path; +use std::sync::Arc; +use std::sync::Mutex; use crate::OPENAI_API_CURATED_MARKETPLACE_NAME; use crate::OPENAI_CURATED_MARKETPLACE_NAME; use crate::PluginsConfigInput; +use crate::http_client_selector::HttpClientSelectionFuture; +use crate::http_client_selector::HttpClientSelector; +use crate::remote::RemotePluginServiceConfig; use codex_config::LoaderOverrides; use codex_config::NoopThreadConfigLoader; use codex_config::loader::load_config_layers_state; use codex_exec_server::LOCAL_FS; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_http_client::RouteAwareClientPool; use codex_utils_absolute_path::AbsolutePathBuf; use toml::Value; pub(crate) const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567"; pub(crate) const TEST_CURATED_PLUGIN_CACHE_VERSION: &str = "01234567"; +pub(crate) fn test_http_client_factory() -> HttpClientFactory { + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) +} + +#[derive(Debug)] +pub(crate) struct RecordingHttpClientSelector { + selected_urls: Arc>>, + delegate: RouteAwareClientPool, +} + +impl RecordingHttpClientSelector { + pub(crate) fn new() -> (Arc, Arc>>) { + let selected_urls = Arc::new(Mutex::new(Vec::new())); + let delegate = RouteAwareClientPool::with_chatgpt_cloudflare_cookies( + test_http_client_factory(), + ClientRouteClass::Api, + ); + ( + Arc::new(Self { + selected_urls: Arc::clone(&selected_urls), + delegate, + }), + selected_urls, + ) + } +} + +impl HttpClientSelector for RecordingHttpClientSelector { + fn client_for_url<'a>(&'a self, request_url: &'a str) -> HttpClientSelectionFuture<'a> { + match self.selected_urls.lock() { + Ok(mut selected_urls) => selected_urls.push(request_url.to_string()), + Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"), + } + Box::pin(self.delegate.client_for_url(request_url)) + } +} + +pub(crate) fn recording_remote_plugin_service_config( + chatgpt_base_url: String, +) -> (RemotePluginServiceConfig, Arc>>) { + let http_client_factory = test_http_client_factory(); + let (http_clients, selected_urls) = RecordingHttpClientSelector::new(); + ( + RemotePluginServiceConfig { + chatgpt_base_url, + http_client_factory, + http_clients, + }, + selected_urls, + ) +} + +pub(crate) fn recorded_http_client_urls(selected_urls: &Mutex>) -> Vec { + match selected_urls.lock() { + Ok(selected_urls) => selected_urls.clone(), + Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"), + } +} + pub(crate) fn write_file(path: &Path, contents: &str) { fs::create_dir_all(path.parent().expect("file should have a parent")).unwrap(); fs::write(path, contents).unwrap(); @@ -152,6 +220,7 @@ pub(crate) async fn load_plugins_config(codex_home: &Path, cwd: &Path) -> Plugin /*default_enabled*/ true, ), "https://chatgpt.com/backend-api/".to_string(), + test_http_client_factory(), ) } diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 1df2bb3e3f..2b83569906 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -1594,6 +1594,15 @@ respect_system_proxy = true config.http_client_factory().outbound_proxy_policy(), codex_http_client::OutboundProxyPolicy::RespectSystemProxy ); + assert_eq!( + config.plugins_config_input().remote_plugin_service_config(), + codex_core_plugins::remote::RemotePluginServiceConfig::new( + config.chatgpt_base_url, + codex_http_client::HttpClientFactory::new( + codex_http_client::OutboundProxyPolicy::RespectSystemProxy, + ), + ) + ); Ok(()) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 5cd6748c05..64b7d00e7c 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1503,6 +1503,7 @@ impl Config { self.features.enabled(Feature::Plugins), self.features.enabled(Feature::RemotePlugin), self.chatgpt_base_url.clone(), + self.http_client_factory(), ) } diff --git a/codex-rs/core/src/plugins/discoverable_tests.rs b/codex-rs/core/src/plugins/discoverable_tests.rs index 53d010d190..a5472f18db 100644 --- a/codex-rs/core/src/plugins/discoverable_tests.rs +++ b/codex-rs/core/src/plugins/discoverable_tests.rs @@ -221,9 +221,10 @@ plugins = true let plugins_manager = PluginsManager::new(config.codex_home.to_path_buf()); fetch_and_cache_global_remote_plugin_catalog( codex_home.path(), - &RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }, + &RemotePluginServiceConfig::new( + config.chatgpt_base_url.clone(), + config.http_client_factory(), + ), Some(&auth), ) .await diff --git a/codex-rs/deny.toml b/codex-rs/deny.toml index b15f7b2335..5f27f5542a 100644 --- a/codex-rs/deny.toml +++ b/codex-rs/deny.toml @@ -242,11 +242,7 @@ deny = [ "codex-agent-identity", "codex-api", "codex-app-server", - "codex-app-server-daemon", - "codex-backend-client", - "codex-cloud-tasks", "codex-core", - "codex-core-plugins", "codex-exec-server", "codex-lmstudio", "codex-login", diff --git a/codex-rs/http-client/src/default_client.rs b/codex-rs/http-client/src/default_client.rs index 73ff72f66f..4ed09d37d1 100644 --- a/codex-rs/http-client/src/default_client.rs +++ b/codex-rs/http-client/src/default_client.rs @@ -1,4 +1,4 @@ -use http::Error as HttpError; +use http::Error as HttpRequestBuildError; use http::HeaderMap; use http::HeaderName; use http::HeaderValue; @@ -6,13 +6,15 @@ use opentelemetry::global; use opentelemetry::propagation::Injector; use reqwest::IntoUrl; use reqwest::Method; -use reqwest::Response; use serde::Serialize; use std::fmt::Display; use std::time::Duration; use tracing::Span; use tracing_opentelemetry::OpenTelemetrySpanExt; +pub type HttpError = reqwest::Error; +pub type HttpResponse = reqwest::Response; + #[derive(Clone, Debug)] pub struct HttpClient { inner: reqwest::Client, @@ -52,6 +54,20 @@ impl HttpClient { self.request(Method::POST, url) } + pub fn put(&self, url: U) -> RequestBuilder + where + U: IntoUrl, + { + self.request(Method::PUT, url) + } + + pub fn delete(&self, url: U) -> RequestBuilder + where + U: IntoUrl, + { + self.request(Method::DELETE, url) + } + pub fn request(&self, method: Method, url: U) -> RequestBuilder where U: IntoUrl, @@ -112,9 +128,9 @@ impl RequestBuilder { pub fn header(self, key: K, value: V) -> Self where HeaderName: TryFrom, - >::Error: Into, + >::Error: Into, HeaderValue: TryFrom, - >::Error: Into, + >::Error: Into, { self.map(|builder| builder.header(key, value)) } @@ -144,7 +160,7 @@ impl RequestBuilder { self.map(|builder| builder.body(body)) } - pub async fn send(self) -> Result { + pub async fn send(self) -> Result { let headers = trace_headers(); match self.builder.headers(headers).send().await { diff --git a/codex-rs/http-client/src/lib.rs b/codex-rs/http-client/src/lib.rs index 49fc4225c3..854fd51e77 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; @@ -21,6 +22,8 @@ pub use crate::custom_ca::build_reqwest_client_with_custom_ca; pub use crate::custom_ca::build_rustls_client_config_with_custom_ca; pub use crate::custom_ca::maybe_build_rustls_client_config_with_custom_ca; pub use crate::default_client::HttpClient; +pub use crate::default_client::HttpError; +pub use crate::default_client::HttpResponse; pub use crate::default_client::RequestBuilder; pub use crate::error::StreamError; pub use crate::error::TransportError; @@ -36,6 +39,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 692da254fc..1247675361 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, @@ -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 { + 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()