Merge 20d626c6a7 into sapling-pr-archive-bolinfest

This commit is contained in:
Michael Bolin
2026-07-09 12:09:14 -07:00
committed by GitHub
55 changed files with 1847 additions and 408 deletions
+9 -4
View File
@@ -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",
+1 -1
View File
@@ -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 }
+4 -2
View File
@@ -238,9 +238,11 @@ pub async fn set_remote_control(mode: RemoteControlMode) -> Result<RemoteControl
Daemon::from_environment()?.set_remote_control(mode).await
}
pub async fn run_pid_update_loop() -> 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)]
+63 -13
View File
@@ -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<UpdateLoopControl> {
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<Vec<u8>> {
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<u8>,
}
#[cfg(unix)]
trait InstallerHttp: Send + Sync {
fn get<'a>(
&'a self,
url: &'a str,
) -> impl std::future::Future<Output = Result<InstallerResponse>> + Send + 'a;
}
#[cfg(unix)]
impl InstallerHttp for RouteAwareClientPool {
async fn get(&self, url: &str) -> Result<InstallerResponse> {
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;
@@ -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<Vec<String>>,
}
impl FakeInstallerHttp {
fn new(response: InstallerResponse) -> Self {
Self {
response,
requested_urls: Mutex::new(Vec::new()),
}
}
fn requested_urls(&self) -> Vec<String> {
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<InstallerResponse> {
self.requested_urls
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push(url.to_string());
Ok(self.response.clone())
}
}
+7 -2
View File
@@ -95,9 +95,14 @@ impl ConfigManager {
&self,
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
http_client_factory: codex_http_client::HttpClientFactory,
) {
let loader =
cloud_config_bundle_loader(auth_manager, chatgpt_base_url, self.codex_home.clone());
let loader = cloud_config_bundle_loader(
auth_manager,
chatgpt_base_url,
self.codex_home.clone(),
http_client_factory,
);
if let Ok(mut guard) = self.cloud_config_bundle.write() {
*guard = loader;
} else {
+5 -2
View File
@@ -505,8 +505,11 @@ pub async fn run_main_with_transport_options(
.replace_thread_config_loader(Arc::clone(&discovered_thread_config_loader));
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
config_manager
.replace_cloud_config_bundle_loader(auth_manager, config.chatgpt_base_url);
config_manager.replace_cloud_config_bundle_loader(
auth_manager,
config.chatgpt_base_url.clone(),
config.http_client_factory(),
);
}
Err(err) => {
warn!(error = %err, "Failed to preload config for cloud config bundle");
@@ -458,7 +458,7 @@ impl AccountRequestProcessor {
let outgoing_clone = self.outgoing.clone();
let config_manager = self.config_manager.clone();
let thread_manager = Arc::clone(&self.thread_manager);
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let config = Arc::clone(&self.config);
let active_login = self.active_login.clone();
let auth_url = server.auth_url.clone();
tokio::spawn(async move {
@@ -480,7 +480,7 @@ impl AccountRequestProcessor {
&outgoing_clone,
config_manager,
thread_manager,
chatgpt_base_url,
config,
login_id,
success,
error_msg,
@@ -537,7 +537,7 @@ impl AccountRequestProcessor {
let outgoing_clone = self.outgoing.clone();
let config_manager = self.config_manager.clone();
let thread_manager = Arc::clone(&self.thread_manager);
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let config = Arc::clone(&self.config);
let active_login = self.active_login.clone();
tokio::spawn(async move {
let (success, error_msg) = tokio::select! {
@@ -556,7 +556,7 @@ impl AccountRequestProcessor {
&outgoing_clone,
config_manager,
thread_manager,
chatgpt_base_url,
config,
login_id,
success,
error_msg,
@@ -671,6 +671,7 @@ impl AccountRequestProcessor {
self.config_manager.replace_cloud_config_bundle_loader(
self.auth_manager.clone(),
self.config.chatgpt_base_url.clone(),
self.config.http_client_factory(),
);
self.config_manager
.sync_default_client_residency_requirement()
@@ -709,7 +710,7 @@ impl AccountRequestProcessor {
outgoing: &OutgoingMessageSender,
config_manager: ConfigManager,
thread_manager: Arc<ThreadManager>,
chatgpt_base_url: String,
config: Arc<Config>,
login_id: Uuid,
success: bool,
error_msg: Option<String>,
@@ -726,8 +727,11 @@ impl AccountRequestProcessor {
if success {
let auth_manager = thread_manager.auth_manager();
auth_manager.reload().await;
config_manager
.replace_cloud_config_bundle_loader(auth_manager.clone(), chatgpt_base_url);
config_manager.replace_cloud_config_bundle_loader(
auth_manager.clone(),
config.chatgpt_base_url.clone(),
config.http_client_factory(),
);
config_manager
.sync_default_client_residency_requirement()
.await;
@@ -933,8 +937,12 @@ impl AccountRequestProcessor {
));
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let client = BackendClient::from_auth(
self.config.chatgpt_base_url.clone(),
&auth,
self.config.http_client_factory(),
)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let (response, detailed_rate_limit_reset_credits) = tokio::join!(
client.get_rate_limits_with_reset_credits(),
@@ -1003,8 +1011,12 @@ impl AccountRequestProcessor {
));
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let client = BackendClient::from_auth(
self.config.chatgpt_base_url.clone(),
&auth,
self.config.http_client_factory(),
)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let profile = tokio::time::timeout(
ACCOUNT_TOKEN_USAGE_FETCH_TIMEOUT,
client.get_token_usage_profile(),
@@ -1030,8 +1042,12 @@ impl AccountRequestProcessor {
));
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let client = BackendClient::from_auth(
self.config.chatgpt_base_url.clone(),
&auth,
self.config.http_client_factory(),
)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let messages = tokio::time::timeout(
ACCOUNT_WORKSPACE_MESSAGES_FETCH_TIMEOUT,
client.list_workspace_messages(),
@@ -1118,8 +1134,12 @@ impl AccountRequestProcessor {
));
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
let client = BackendClient::from_auth(
self.config.chatgpt_base_url.clone(),
&auth,
self.config.http_client_factory(),
)
.map_err(|err| internal_error(format!("failed to construct backend client: {err}")))?;
match client
.send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type))
@@ -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}")))
}
}
@@ -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<AbsolutePathBuf, String>,
@@ -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<PluginShareListResponse, JSONRPCErrorError> {
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 { .. }
+2 -1
View File
@@ -16,7 +16,8 @@ workspace = true
anyhow = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
http = { workspace = true }
url = { workspace = true }
codex-backend-openapi-models = { path = "../codex-backend-openapi-models" }
codex-api = { workspace = true }
codex-http-client = { workspace = true }
+129 -53
View File
@@ -9,8 +9,10 @@ use crate::types::TokenUsageProfile;
use crate::types::TurnAttemptsSiblingTurnsResponse;
use anyhow::Result;
use codex_api::SharedAuthProvider;
use codex_http_client::build_reqwest_client_with_custom_ca;
use codex_http_client::with_chatgpt_cloudflare_cookie_store;
use codex_http_client::ClientRouteClass;
use codex_http_client::HttpClientFactory;
use codex_http_client::RequestBuilder;
use codex_http_client::RouteAwareClientPool;
use codex_login::CodexAuth;
use codex_login::default_client::get_codex_user_agent;
use codex_protocol::account::PlanType as AccountPlanType;
@@ -19,13 +21,14 @@ use codex_protocol::protocol::RateLimitReachedType;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::RateLimitWindow;
use codex_protocol::protocol::SpendControlLimitSnapshot;
use reqwest::StatusCode;
use reqwest::header::CACHE_CONTROL;
use reqwest::header::CONTENT_TYPE;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderName;
use reqwest::header::HeaderValue;
use reqwest::header::USER_AGENT;
use http::Method;
use http::StatusCode;
use http::header::CACHE_CONTROL;
use http::header::CONTENT_TYPE;
use http::header::HeaderMap;
use http::header::HeaderName;
use http::header::HeaderValue;
use http::header::USER_AGENT;
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::fmt;
@@ -123,7 +126,7 @@ impl PathStyle {
#[derive(Clone)]
pub struct Client {
base_url: String,
http: reqwest::Client,
http: RouteAwareClientPool,
auth_provider: SharedAuthProvider,
user_agent: Option<HeaderValue>,
chatgpt_account_id: Option<String>,
@@ -148,7 +151,10 @@ impl fmt::Debug for Client {
}
impl Client {
pub fn new(base_url: impl Into<String>) -> Result<Self> {
pub fn new(
base_url: impl Into<String>,
http_client_factory: HttpClientFactory,
) -> Result<Self> {
let mut base_url = base_url.into();
// Normalize common ChatGPT hostnames to include /backend-api so we hit the WHAM paths.
// Also trim trailing slashes for consistent URL building.
@@ -161,9 +167,10 @@ impl Client {
{
base_url = format!("{base_url}/backend-api");
}
let http = build_reqwest_client_with_custom_ca(with_chatgpt_cloudflare_cookie_store(
reqwest::Client::builder(),
))?;
let http = RouteAwareClientPool::with_chatgpt_cloudflare_cookies(
http_client_factory,
ClientRouteClass::Api,
);
let path_style = PathStyle::from_base_url(&base_url);
Ok(Self {
base_url,
@@ -176,8 +183,12 @@ impl Client {
})
}
pub fn from_auth(base_url: impl Into<String>, auth: &CodexAuth) -> Result<Self> {
Ok(Self::new(base_url)?
pub fn from_auth(
base_url: impl Into<String>,
auth: &CodexAuth,
http_client_factory: HttpClientFactory,
) -> Result<Self> {
Ok(Self::new(base_url, http_client_factory)?
.with_user_agent(get_codex_user_agent())
.with_auth_provider(codex_model_provider::auth_provider_from_auth(auth)))
}
@@ -231,9 +242,13 @@ impl Client {
h
}
async fn request(&self, method: Method, url: &str) -> Result<RequestBuilder> {
Ok(self.http.client_for_url(url).await?.request(method, url))
}
async fn exec_request(
&self,
req: reqwest::RequestBuilder,
req: RequestBuilder,
method: &str,
url: &str,
) -> Result<(String, String)> {
@@ -254,7 +269,7 @@ impl Client {
async fn exec_request_detailed(
&self,
req: reqwest::RequestBuilder,
req: RequestBuilder,
method: &str,
url: &str,
) -> std::result::Result<(String, String), RequestError> {
@@ -306,14 +321,20 @@ impl Client {
PathStyle::CodexApi => format!("{}/api/codex/accounts/check", self.base_url),
PathStyle::ChatGptApi => format!("{}/wham/accounts/check", self.base_url),
};
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json(&url, &ct, &body)
}
pub async fn get_token_usage_profile(&self) -> Result<TokenUsageProfile> {
let url = self.token_usage_profile_url();
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json(&url, &ct, &body)
}
@@ -331,8 +352,9 @@ impl Client {
) -> std::result::Result<(), RequestError> {
let url = self.send_add_credits_nudge_email_url();
let req = self
.http
.post(&url)
.request(Method::POST, &url)
.await
.map_err(RequestError::from)?
.headers(self.headers())
.header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.json(&SendAddCreditsNudgeEmailRequest { credit_type });
@@ -347,33 +369,47 @@ impl Client {
environment_id: Option<&str>,
cursor: Option<&str>,
) -> Result<PaginatedListTaskListItem> {
let url = self.list_tasks_url(limit, task_filter, environment_id, cursor)?;
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json::<PaginatedListTaskListItem>(&url, &ct, &body)
}
fn list_tasks_url(
&self,
limit: Option<i32>,
task_filter: Option<&str>,
environment_id: Option<&str>,
cursor: Option<&str>,
) -> Result<String> {
let url = match self.path_style {
PathStyle::CodexApi => format!("{}/api/codex/tasks/list", self.base_url),
PathStyle::ChatGptApi => format!("{}/wham/tasks/list", self.base_url),
};
let req = self.http.get(&url).headers(self.headers());
let req = if let Some(lim) = limit {
req.query(&[("limit", lim)])
} else {
req
};
let req = if let Some(tf) = task_filter {
req.query(&[("task_filter", tf)])
} else {
req
};
let req = if let Some(c) = cursor {
req.query(&[("cursor", c)])
} else {
req
};
let req = if let Some(id) = environment_id {
req.query(&[("environment_id", id)])
} else {
req
};
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json::<PaginatedListTaskListItem>(&url, &ct, &body)
if limit.is_none() && task_filter.is_none() && environment_id.is_none() && cursor.is_none()
{
return Ok(url);
}
let mut url = url::Url::parse(&url)?;
{
let mut query = url.query_pairs_mut();
if let Some(limit) = limit {
query.append_pair("limit", &limit.to_string());
}
if let Some(task_filter) = task_filter {
query.append_pair("task_filter", task_filter);
}
if let Some(cursor) = cursor {
query.append_pair("cursor", cursor);
}
if let Some(environment_id) = environment_id {
query.append_pair("environment_id", environment_id);
}
}
Ok(url.to_string())
}
pub async fn get_task_details(&self, task_id: &str) -> Result<CodeTaskDetailsResponse> {
@@ -389,7 +425,10 @@ impl Client {
PathStyle::CodexApi => format!("{}/api/codex/tasks/{}", self.base_url, task_id),
PathStyle::ChatGptApi => format!("{}/wham/tasks/{}", self.base_url, task_id),
};
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
let parsed: CodeTaskDetailsResponse = self.decode_json(&url, &ct, &body)?;
Ok((parsed, body, ct))
@@ -410,7 +449,10 @@ impl Client {
self.base_url, task_id, turn_id
),
};
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json::<TurnAttemptsSiblingTurnsResponse>(&url, &ct, &body)
}
@@ -426,7 +468,11 @@ impl Client {
PathStyle::CodexApi => format!("{}/api/codex/config/bundle", self.base_url),
PathStyle::ChatGptApi => format!("{}/wham/config/bundle", self.base_url),
};
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await
.map_err(RequestError::from)?
.headers(self.headers());
let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?;
self.decode_json::<ConfigBundleResponse>(&url, &ct, &body)
.map_err(RequestError::from)
@@ -437,8 +483,9 @@ impl Client {
) -> std::result::Result<CodexWorkspaceMessagesResponse, RequestError> {
let url = self.workspace_messages_url();
let req = self
.http
.get(&url)
.request(Method::GET, &url)
.await
.map_err(RequestError::from)?
.headers(self.headers())
.header(CACHE_CONTROL, HeaderValue::from_static("no-store"));
let (body, ct) = self.exec_request_detailed(req, "GET", &url).await?;
@@ -454,8 +501,8 @@ impl Client {
PathStyle::ChatGptApi => format!("{}/wham/tasks", self.base_url),
};
let req = self
.http
.post(&url)
.request(Method::POST, &url)
.await?
.headers(self.headers())
.header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.json(&request_body);
@@ -974,10 +1021,39 @@ mod tests {
);
}
#[test]
fn list_tasks_url_omits_empty_query_and_encodes_all_parameters() {
let client = test_client("https://example.test", PathStyle::CodexApi);
assert_eq!(
client
.list_tasks_url(
/*limit*/ None, /*task_filter*/ None, /*environment_id*/ None,
/*cursor*/ None,
)
.unwrap(),
"https://example.test/api/codex/tasks/list"
);
assert_eq!(
client
.list_tasks_url(
/*limit*/ Some(10),
/*task_filter*/ Some("mine / shared"),
/*environment_id*/ Some("env&one"),
/*cursor*/ Some("next=page"),
)
.unwrap(),
"https://example.test/api/codex/tasks/list?limit=10&task_filter=mine+%2F+shared&cursor=next%3Dpage&environment_id=env%26one"
);
}
fn test_client(base_url: &str, path_style: PathStyle) -> Client {
Client {
base_url: base_url.to_string(),
http: reqwest::Client::new(),
http: RouteAwareClientPool::new(
HttpClientFactory::new(codex_http_client::OutboundProxyPolicy::ReqwestDefault),
ClientRouteClass::Api,
),
auth_provider: codex_model_provider::unauthenticated_auth_provider(),
user_agent: None,
chatgpt_account_id: None,
@@ -7,8 +7,9 @@ use crate::types::RateLimitResetCreditsDetails;
use crate::types::RateLimitStatusWithResetCredits;
use crate::types::RateLimitsWithResetCredits;
use anyhow::Result;
use reqwest::header::CONTENT_TYPE;
use reqwest::header::HeaderValue;
use http::Method;
use http::header::CONTENT_TYPE;
use http::header::HeaderValue;
use serde::Serialize;
#[derive(Serialize)]
@@ -29,14 +30,20 @@ impl Client {
pub(super) async fn get_rate_limit_status(&self) -> Result<RateLimitStatusWithResetCredits> {
let url = self.rate_limit_status_url();
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json(&url, &ct, &body)
}
pub async fn list_rate_limit_reset_credits(&self) -> Result<RateLimitResetCreditsDetails> {
let url = self.rate_limit_reset_credits_url();
let req = self.http.get(&url).headers(self.headers());
let req = self
.request(Method::GET, &url)
.await?
.headers(self.headers());
let (body, ct) = self.exec_request(req, "GET", &url).await?;
self.decode_json(&url, &ct, &body)
}
@@ -65,8 +72,8 @@ impl Client {
) -> Result<ConsumeRateLimitResetCreditResponse> {
let url = self.consume_rate_limit_reset_credit_url();
let req = self
.http
.post(&url)
.request(Method::POST, &url)
.await?
.headers(self.headers())
.header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.json(&ConsumeRateLimitResetCreditRequest {
@@ -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,
+1
View File
@@ -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 }
+52 -1
View File
@@ -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_core::config::Config>,
) -> 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");
+1
View File
@@ -12,6 +12,7 @@ base64 = { workspace = true }
chrono = { workspace = true, features = ["serde"] }
codex-backend-client = { workspace = true }
codex-config = { workspace = true }
codex-http-client = { workspace = true }
codex-core = { workspace = true }
codex-login = { workspace = true }
codex-otel = { workspace = true }
+19 -10
View File
@@ -6,6 +6,7 @@ use codex_config::CloudConfigFragment;
use codex_config::CloudConfigTomlBundle;
use codex_config::CloudRequirementsFragment;
use codex_config::CloudRequirementsTomlBundle;
use codex_http_client::HttpClientFactory;
use codex_login::CodexAuth;
use std::future::Future;
@@ -46,24 +47,32 @@ pub(crate) trait BundleClient: Send + Sync {
pub(crate) struct BackendBundleClient {
base_url: String,
http_client_factory: HttpClientFactory,
}
impl BackendBundleClient {
pub(crate) fn new(base_url: String) -> Self {
Self { base_url }
pub(crate) fn new(base_url: String, http_client_factory: HttpClientFactory) -> Self {
Self {
base_url,
http_client_factory,
}
}
}
impl BundleClient for BackendBundleClient {
async fn get_bundle(&self, auth: &CodexAuth) -> Result<CloudConfigBundle, BundleRequestError> {
let client = BackendClient::from_auth(self.base_url.clone(), auth)
.inspect_err(|err| {
tracing::warn!(
error = %err,
"Failed to construct backend client for cloud config bundle"
);
})
.map_err(|_| BundleRequestError::Retryable(RetryableFailureKind::BackendClientInit))?;
let client = BackendClient::from_auth(
self.base_url.clone(),
auth,
self.http_client_factory.clone(),
)
.inspect_err(|err| {
tracing::warn!(
error = %err,
"Failed to construct backend client for cloud config bundle"
);
})
.map_err(|_| BundleRequestError::Retryable(RetryableFailureKind::BackendClientInit))?;
let response = client
.get_config_bundle()
+17 -2
View File
@@ -5,6 +5,8 @@ use codex_config::CloudConfigBundleLoadError;
use codex_config::CloudConfigBundleLoadErrorCode;
use codex_config::CloudConfigBundleLoader;
use codex_config::types::AuthCredentialsStoreMode;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use codex_login::AuthKeyringBackendKind;
use codex_login::AuthManager;
use codex_login::AuthRouteConfig;
@@ -23,10 +25,14 @@ pub fn cloud_config_bundle_loader(
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
codex_home: PathBuf,
http_client_factory: HttpClientFactory,
) -> CloudConfigBundleLoader {
let service = CloudConfigBundleService::new(
auth_manager,
Arc::new(BackendBundleClient::new(chatgpt_base_url)),
Arc::new(BackendBundleClient::new(
chatgpt_base_url,
http_client_factory,
)),
codex_home,
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
@@ -61,6 +67,10 @@ pub async fn cloud_config_bundle_loader_for_storage(
chatgpt_base_url: String,
auth_route_config: Option<AuthRouteConfig>,
) -> CloudConfigBundleLoader {
let http_client_factory = auth_route_config.as_ref().map_or_else(
|| HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
|config| config.http_client_factory().clone(),
);
let auth_manager = AuthManager::shared(
codex_home.clone(),
enable_codex_api_key_env,
@@ -71,5 +81,10 @@ pub async fn cloud_config_bundle_loader_for_storage(
auth_route_config,
)
.await;
cloud_config_bundle_loader(auth_manager, chatgpt_base_url, codex_home)
cloud_config_bundle_loader(
auth_manager,
chatgpt_base_url,
codex_home,
http_client_factory,
)
}
+1
View File
@@ -19,6 +19,7 @@ chrono = { workspace = true, features = ["serde"] }
codex-api = { workspace = true }
codex-backend-client = { workspace = true }
codex-git-utils = { workspace = true }
codex-http-client = { workspace = true }
serde = { version = "1", features = ["derive"] }
serde_json = { workspace = true }
thiserror = { workspace = true }
+5 -2
View File
@@ -28,9 +28,12 @@ pub struct HttpClient {
}
impl HttpClient {
pub fn new(base_url: impl Into<String>) -> anyhow::Result<Self> {
pub fn new(
base_url: impl Into<String>,
http_client_factory: codex_http_client::HttpClientFactory,
) -> anyhow::Result<Self> {
let base_url = base_url.into();
let backend = backend::Client::new(base_url.clone())?;
let backend = backend::Client::new(base_url.clone(), http_client_factory)?;
Ok(Self { base_url, backend })
}
+1 -1
View File
@@ -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 }
+88 -31
View File
@@ -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<i64>,
}
#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AutodetectSelection {
pub id: String,
pub label: Option<String>,
}
pub async fn autodetect_environment_id(
http: &RouteAwareClientPool,
base_url: &str,
headers: &HeaderMap,
desired_label: Option<String>,
) -> anyhow::Result<AutodetectSelection> {
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<String>,
origins: &[String],
) -> anyhow::Result<AutodetectSelection> {
// 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<CodeEnvironment> = 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::<Vec<CodeEnvironment>>(&url, headers).await {
match get_json::<Vec<CodeEnvironment>>(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::<serde_json::Value>(&body) {
Ok(v) => {
@@ -145,19 +157,14 @@ fn pick_environment_row(
}
async fn get_json<T: serde::de::DeserializeOwned>(
http: &impl EnvironmentHttp,
url: &str,
headers: &HeaderMap,
) -> anyhow::Result<T> {
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<T: serde::de::DeserializeOwned>(
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<Output = anyhow::Result<EnvironmentResponse>> + Send + 'a;
}
impl EnvironmentHttp for RouteAwareClientPool {
async fn get(&self, url: &str, headers: &HeaderMap) -> anyhow::Result<EnvironmentResponse> {
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<String> {
// 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<Vec<crate::app::EnvironmentRow>> {
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<Vec<crate::app::EnvironmentRow>> {
let mut map: HashMap<String, crate::app::EnvironmentRow> = 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::<Vec<CodeEnvironment>>(&url, headers).await {
match get_json::<Vec<CodeEnvironment>>(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::<Vec<CodeEnvironment>>(&list_url, headers).await {
match get_json::<Vec<CodeEnvironment>>(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;
@@ -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<_>>(),
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<String, EnvironmentResponse>,
requested_urls: Mutex<Vec<String>>,
}
impl FakeHttp {
fn new(responses: HashMap<String, EnvironmentResponse>) -> Self {
Self {
responses,
requested_urls: Mutex::new(Vec::new()),
}
}
fn requested_urls(&self) -> Vec<String> {
self.requested_urls
.lock()
.expect("requested URL lock")
.clone()
}
}
impl EnvironmentHttp for FakeHttp {
async fn get(&self, url: &str, _headers: &HeaderMap) -> anyhow::Result<EnvironmentResponse> {
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(),
}
}
+64 -58
View File
@@ -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<dyn codex_cloud_tasks_client::CloudBackend>,
base_url: String,
environment_http: RouteAwareClientPool,
}
async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext> {
@@ -56,11 +61,20 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext>
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<BackendContext>
};
append_error_log(format!("startup: base_url={base_url} path_style={style}"));
let auth_manager = util::load_auth_manager(Some(base_url.clone())).await;
let auth = match auth_manager.as_ref() {
Some(manager) => manager.auth().await,
None => None,
@@ -103,6 +116,7 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext>
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<PathBuf>) -> 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<PathBuf>) -> 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<PathBuf>) -> 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<PathBuf>) -> 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<PathBuf>) -> 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<PathBuf>) -> 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<PathBuf>) -> an
Ok(())
}
fn spawn_environment_load(
tx: UnboundedSender<app::AppEvent>,
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.
+26 -17
View File
@@ -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<String>) -> Option<AuthManager> {
pub async fn load_auth_manager(
chatgpt_base_url: Option<String>,
) -> (Option<AuthManager>, HttpClientFactory) {
// TODO: pass in cli overrides once cloud tasks properly support them.
let config = Config::load_with_cli_overrides(Vec::new()).await.ok()?;
Some(
AuthManager::new(
config.codex_home.to_path_buf(),
/*enable_codex_api_key_env*/ false,
config.cli_auth_credentials_store_mode,
config.forced_chatgpt_workspace_id.clone(),
chatgpt_base_url.or(Some(config.chatgpt_base_url.clone())),
config.auth_keyring_backend_kind(),
config.auth_route_config(),
)
.await,
let Some(config) = Config::load_with_cli_overrides(Vec::new()).await.ok() else {
return (
None,
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
);
};
let http_client_factory = config.http_client_factory();
let auth_manager = AuthManager::new(
config.codex_home.to_path_buf(),
/*enable_codex_api_key_env*/ false,
config.cli_auth_credentials_store_mode,
config.forced_chatgpt_workspace_id.clone(),
chatgpt_base_url.or(Some(config.chatgpt_base_url.clone())),
config.auth_keyring_backend_kind(),
config.auth_route_config(),
)
.await;
(Some(auth_manager), http_client_factory)
}
/// Build headers for ChatGPT-backed requests: `User-Agent`, optional `Authorization`,
/// 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()
{
+2 -1
View File
@@ -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"] }
@@ -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
@@ -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<Box<dyn Future<Output = Result<HttpClient, RouteAwareClientPoolError>> + 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))
}
}
+1
View File
@@ -1,5 +1,6 @@
mod app_mcp_routing;
mod discoverable;
mod http_client_selector;
pub mod installed_marketplaces;
pub mod loader;
mod manager;
+20 -9
View File
@@ -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<Self>) {
fn start_curated_repo_sync(self: &Arc<Self>, 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",
@@ -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());
+118 -42
View File
@@ -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<dyn HttpClientSelector>,
}
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<HttpClient, RouteAwareClientPoolError> {
self.http_clients.client_for_url(request_url).await
}
async fn catalog_client_for_url(
&self,
request_url: &str,
) -> Result<HttpClient, RemotePluginCatalogError> {
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<RecommendedPluginsMode, RemotePluginCatalogError> {
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<RemotePluginListResponse, RemotePluginCatalogError> {
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<RemotePluginListResponse, RemotePluginCatalogError> {
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<RemotePluginInstalledResponse, RemotePluginCatalogError> {
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<RemotePluginDirectoryItem, RemotePluginCatalogError> {
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
}
@@ -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(
+18 -14
View File
@@ -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<RemotePluginListResponse, RemotePluginCatalogError> {
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<RemoteWorkspacePluginUploadUrlResponse, RemotePluginCatalogError> {
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<u8>,
) -> 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
}
@@ -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(),
)
@@ -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
+62 -7
View File
@@ -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<PluginInstallResult, RemotePluginBundleInstallError> {
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<AbsolutePathBuf, RemotePluginBundleInstallError> {
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<Vec<u8>, 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<Vec<u8>, 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");
+42 -12
View File
@@ -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<Product>,
) -> Result<Vec<String>, 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<RemotePluginMutationResponse, RemotePluginMutationError> {
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)
+42
View File
@@ -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() {
+87 -23
View File
@@ -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<String, String> {
pub fn sync_openai_plugins_repo(
codex_home: &Path,
http_client_factory: HttpClientFactory,
) -> Result<String, String> {
#[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<String, String> {
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<String, String> {
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<String, String> {
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<String, String> {
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<String, String> {
async fn fetch_curated_repo_remote_sha(
http_clients: &dyn HttpClientSelector,
api_base_url: &str,
) -> Result<String, 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 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<String, Str
let git_ref_url = format!("{repo_url}/git/ref/heads/{}", repo_summary.default_branch);
let git_ref_body =
fetch_github_text(&client, &git_ref_url, "get curated plugins HEAD ref").await?;
fetch_github_text(http_clients, &git_ref_url, "get curated plugins HEAD ref").await?;
let git_ref: GitHubGitRefSummary = serde_json::from_str(&git_ref_body).map_err(|err| {
format!("failed to parse curated plugins ref response from {git_ref_url}: {err}")
})?;
@@ -794,22 +822,27 @@ async fn fetch_curated_repo_remote_sha(api_base_url: &str) -> Result<String, Str
}
async fn fetch_curated_repo_zipball(
http_clients: &dyn HttpClientSelector,
api_base_url: &str,
remote_sha: &str,
) -> Result<Vec<u8>, 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<Vec<u8>, 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<String, String> {
))
}
async fn fetch_github_text(client: &Client, url: &str, context: &str) -> Result<String, String> {
let response = github_request(client, url)
async fn fetch_github_text(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<String, 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}"))?;
@@ -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<Vec<u8>, String> {
let response = github_request(client, url)
async fn fetch_github_bytes(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<Vec<u8>, 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<String, String> {
async fn fetch_public_text(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<String, String> {
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<Vec<u8>, String> {
async fn fetch_public_bytes(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<Vec<u8>, 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<HttpClient, String> {
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)
@@ -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<String, String> {
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);
+69
View File
@@ -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<Mutex<Vec<String>>>,
delegate: RouteAwareClientPool,
}
impl RecordingHttpClientSelector {
pub(crate) fn new() -> (Arc<Self>, Arc<Mutex<Vec<String>>>) {
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<Mutex<Vec<String>>>) {
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<String>>) -> Vec<String> {
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(),
)
}
+9
View File
@@ -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(())
}
+1
View File
@@ -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(),
)
}
@@ -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
-4
View File
@@ -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",
+21 -5
View File
@@ -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<U>(&self, url: U) -> RequestBuilder
where
U: IntoUrl,
{
self.request(Method::PUT, url)
}
pub fn delete<U>(&self, url: U) -> RequestBuilder
where
U: IntoUrl,
{
self.request(Method::DELETE, url)
}
pub fn request<U>(&self, method: Method, url: U) -> RequestBuilder
where
U: IntoUrl,
@@ -112,9 +128,9 @@ impl RequestBuilder {
pub fn header<K, V>(self, key: K, value: V) -> Self
where
HeaderName: TryFrom<K>,
<HeaderName as TryFrom<K>>::Error: Into<HttpError>,
<HeaderName as TryFrom<K>>::Error: Into<HttpRequestBuildError>,
HeaderValue: TryFrom<V>,
<HeaderValue as TryFrom<V>>::Error: Into<HttpError>,
<HeaderValue as TryFrom<V>>::Error: Into<HttpRequestBuildError>,
{
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<Response, reqwest::Error> {
pub async fn send(self) -> Result<HttpResponse, HttpError> {
let headers = trace_headers();
match self.builder.headers(headers).send().await {
+5
View File
@@ -5,6 +5,7 @@ mod default_client;
mod error;
mod outbound_proxy;
mod request;
mod route_aware_client_pool;
mod transport;
pub use crate::chatgpt_cloudflare_cookies::with_chatgpt_cloudflare_cookie_store;
@@ -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;
+46 -1
View File
@@ -12,6 +12,7 @@ use std::sync::Mutex;
use std::sync::OnceLock;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Semaphore;
use crate::custom_ca::BuildCustomCaTransportError;
use crate::custom_ca::build_reqwest_client_with_custom_ca;
@@ -25,6 +26,7 @@ use thiserror::Error;
const SYSTEM_PROXY_SUCCESS_CACHE_TTL: Duration = Duration::from_secs(60);
const SYSTEM_PROXY_UNAVAILABLE_CACHE_TTL: Duration = Duration::from_secs(5);
const SYSTEM_PROXY_CACHE_MAX_ENTRIES: usize = 256;
static ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT: Semaphore = Semaphore::const_new(1);
#[cfg(target_os = "macos")]
mod macos;
@@ -102,7 +104,7 @@ pub enum OutboundProxyPolicy {
///
/// `TransportDefault` delegates environment-proxy handling to the underlying transport. Proxy
/// URLs are intentionally redacted from `Debug` output because they may contain credentials.
#[derive(Clone, PartialEq, Eq)]
#[derive(Clone, Hash, PartialEq, Eq)]
pub enum OutboundProxyRoute {
/// Preserve the underlying transport's existing proxy behavior.
TransportDefault,
@@ -159,6 +161,33 @@ impl HttpClientFactory {
)
}
/// Resolves the proxy route for a concrete destination without blocking a Tokio worker.
pub async fn resolve_proxy_route_async(
&self,
request_url: String,
) -> io::Result<OutboundProxyRoute> {
if matches!(
self.outbound_proxy_policy,
OutboundProxyPolicy::ReqwestDefault
) {
return Ok(OutboundProxyRoute::TransportDefault);
}
let permit = ASYNC_SYSTEM_PROXY_RESOLUTION_PERMIT
.acquire()
.await
.map_err(io::Error::other)?;
let factory = self.clone();
tokio::task::spawn_blocking(move || {
// Keep the permit with the blocking task: cancelling the caller must not allow a
// second PAC/WinHTTP lookup to start while this one is still running.
let _permit = permit;
factory.resolve_proxy_route(&request_url)
})
.await
.map_err(io::Error::other)
}
/// Builds an HTTP client for a concrete outbound route.
pub fn build_client(
&self,
@@ -193,6 +222,22 @@ impl HttpClientFactory {
self.outbound_proxy_policy,
)
}
pub(crate) fn build_reqwest_client_for_resolved_route(
&self,
builder: reqwest::ClientBuilder,
route_class: ClientRouteClass,
route: &OutboundProxyRoute,
) -> Result<reqwest::Client, BuildRouteAwareHttpClientError> {
let builder = match route {
OutboundProxyRoute::TransportDefault => builder,
OutboundProxyRoute::Direct => builder.no_proxy(),
OutboundProxyRoute::Proxy { url } => {
configure_concrete_proxy(builder, route_class, url, /*no_proxy*/ None)?
}
};
build_reqwest_client_with_custom_ca(builder).map_err(Into::into)
}
}
fn resolve_proxy_route(
@@ -10,6 +10,48 @@ struct MapEnv {
values: HashMap<String, String>,
}
fn spawn_proxy_listener() -> (
std::net::SocketAddr,
std::thread::JoinHandle<Option<String>>,
) {
let listener =
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
let proxy_addr = listener
.local_addr()
.expect("local proxy listener should have an address");
listener
.set_nonblocking(true)
.expect("proxy listener should become nonblocking");
let proxy_thread = std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(2);
loop {
match listener.accept() {
Ok((mut stream, _)) => {
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("proxy stream should get a read timeout");
let mut buffer = [0_u8; 4096];
let size = stream.read(&mut buffer).expect("proxy should read request");
stream
.write_all(
b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok",
)
.expect("proxy should write response");
break Some(String::from_utf8_lossy(&buffer[..size]).into_owned());
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
if Instant::now() >= deadline {
break None;
}
std::thread::sleep(Duration::from_millis(10));
}
Err(error) => panic!("proxy should accept a request: {error}"),
}
}
});
(proxy_addr, proxy_thread)
}
#[test]
fn websocket_route_uses_http_equivalent_for_system_resolution() {
let route = resolve_proxy_route(
@@ -97,20 +139,7 @@ fn environment_fallback_reads_injected_proxy_environment() {
#[tokio::test]
async fn enabled_environment_proxy_routes_request_through_proxy() {
let listener =
std::net::TcpListener::bind(("127.0.0.1", 0)).expect("local proxy listener should bind");
let proxy_addr = listener
.local_addr()
.expect("local proxy listener should have an address");
let proxy_thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("proxy should accept a request");
let mut buffer = [0_u8; 4096];
let size = stream.read(&mut buffer).expect("proxy should read request");
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
.expect("proxy should write response");
String::from_utf8_lossy(&buffer[..size]).into_owned()
});
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
let env = MapEnv {
values: HashMap::from([("HTTP_PROXY".to_string(), format!("http://{proxy_addr}"))]),
};
@@ -134,7 +163,10 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
.send()
.await
.expect("request should use local proxy");
let proxy_request = proxy_thread.join().expect("proxy thread should finish");
let proxy_request = proxy_thread
.join()
.expect("proxy thread should finish")
.expect("proxy should receive request before timeout");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(
@@ -143,6 +175,41 @@ async fn enabled_environment_proxy_routes_request_through_proxy() {
);
}
#[tokio::test]
async fn route_aware_pool_uses_respect_system_proxy_route_for_exact_url() {
let (proxy_addr, proxy_thread) = spawn_proxy_listener();
let request_url = "http://route-aware-proxy.test/proxy-check?pac=exact";
cache_system_proxy_decision(
request_url,
SystemProxyDecision::Proxy {
url: format!("http://{proxy_addr}"),
},
);
let pool = crate::RouteAwareClientPool::new(
HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy),
ClientRouteClass::Api,
);
let client = tokio::time::timeout(Duration::from_secs(2), pool.client_for_url(request_url))
.await
.expect("route resolution should finish")
.expect("route-aware client should build");
let response = tokio::time::timeout(Duration::from_secs(2), client.get(request_url).send())
.await
.expect("proxy request should finish")
.expect("request should use local proxy");
let proxy_request = proxy_thread
.join()
.expect("proxy thread should finish")
.expect("proxy should receive request before timeout");
assert_eq!(response.status(), reqwest::StatusCode::OK);
assert_eq!(
proxy_request.lines().next(),
Some("GET http://route-aware-proxy.test/proxy-check?pac=exact HTTP/1.1")
);
}
#[test]
fn parses_pac_proxy_tokens() {
assert_eq!(
@@ -0,0 +1,140 @@
use std::collections::HashMap;
use std::fmt;
use std::future::Future;
use std::io;
use std::sync::Arc;
use std::sync::Mutex;
use crate::BuildRouteAwareHttpClientError;
use crate::ClientRouteClass;
use crate::HttpClient;
use crate::HttpClientFactory;
use crate::OutboundProxyRoute;
use crate::with_chatgpt_cloudflare_cookie_store;
const MAX_CACHED_ROUTES: usize = 16;
/// Reuses reqwest clients by resolved route while evaluating the exact URL of every request.
///
/// PAC and system-proxy selection remains request-specific, but requests that resolve to the same
/// direct or proxy route share one connection pool. Named constructors apply transport-wide
/// settings such as the ChatGPT Cloudflare cookie store without exposing reqwest to callers.
#[derive(Clone)]
pub struct RouteAwareClientPool {
http_client_factory: HttpClientFactory,
route_class: ClientRouteClass,
builder_factory: Arc<dyn Fn() -> reqwest::ClientBuilder + Send + Sync>,
clients: Arc<Mutex<HashMap<OutboundProxyRoute, HttpClient>>>,
}
impl fmt::Debug for RouteAwareClientPool {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RouteAwareClientPool")
.field("http_client_factory", &self.http_client_factory)
.field("route_class", &self.route_class)
.finish_non_exhaustive()
}
}
/// Error returned when selecting a route or constructing its pooled HTTP client.
#[derive(Debug, thiserror::Error)]
pub enum RouteAwareClientPoolError {
#[error("failed to resolve the outbound proxy route: {0}")]
Resolve(#[source] io::Error),
#[error(transparent)]
Build(#[from] BuildRouteAwareHttpClientError),
}
impl RouteAwareClientPool {
/// Creates a pool with the shared default HTTP transport settings.
pub fn new(http_client_factory: HttpClientFactory, route_class: ClientRouteClass) -> Self {
Self::with_builder_factory(http_client_factory, route_class, reqwest::Client::builder)
}
/// Creates a pool that retains the Cloudflare cookies required by ChatGPT endpoints.
pub fn with_chatgpt_cloudflare_cookies(
http_client_factory: HttpClientFactory,
route_class: ClientRouteClass,
) -> Self {
Self::with_builder_factory(http_client_factory, route_class, || {
with_chatgpt_cloudflare_cookie_store(reqwest::Client::builder())
})
}
pub(crate) fn with_builder_factory(
http_client_factory: HttpClientFactory,
route_class: ClientRouteClass,
builder_factory: impl Fn() -> reqwest::ClientBuilder + Send + Sync + 'static,
) -> Self {
Self {
http_client_factory,
route_class,
builder_factory: Arc::new(builder_factory),
clients: Arc::new(Mutex::new(HashMap::new())),
}
}
/// Returns a pooled client configured for the proxy route selected for `request_url`.
pub async fn client_for_url(
&self,
request_url: &str,
) -> Result<HttpClient, RouteAwareClientPoolError> {
let http_client_factory = self.http_client_factory.clone();
self.client_for_url_with_resolver(request_url, move |request_url| async move {
http_client_factory
.resolve_proxy_route_async(request_url)
.await
})
.await
}
async fn client_for_url_with_resolver<F, Fut>(
&self,
request_url: &str,
resolve_route: F,
) -> Result<HttpClient, RouteAwareClientPoolError>
where
F: FnOnce(String) -> Fut,
Fut: Future<Output = io::Result<OutboundProxyRoute>>,
{
let route = resolve_route(request_url.to_string())
.await
.map_err(RouteAwareClientPoolError::Resolve)?;
let clients = match self.clients.lock() {
Ok(clients) => clients,
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
};
if let Some(client) = clients.get(&route) {
return Ok(client.clone());
}
drop(clients);
let client = self
.http_client_factory
.build_reqwest_client_for_resolved_route(
(self.builder_factory)(),
self.route_class,
&route,
)?;
let client = HttpClient::new(client);
let mut clients = match self.clients.lock() {
Ok(clients) => clients,
Err(error) => panic!("route-aware client cache lock should not be poisoned: {error}"),
};
if let Some(existing_client) = clients.get(&route) {
return Ok(existing_client.clone());
}
if clients.len() >= MAX_CACHED_ROUTES
&& let Some(route_to_evict) = clients.keys().next().cloned()
{
clients.remove(&route_to_evict);
}
clients.insert(route, client.clone());
Ok(client)
}
}
#[cfg(test)]
#[path = "route_aware_client_pool_tests.rs"]
mod tests;
@@ -0,0 +1,154 @@
use std::collections::HashMap;
use std::io;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use pretty_assertions::assert_eq;
use super::*;
use crate::OutboundProxyPolicy;
#[tokio::test]
async fn forwards_exact_urls_and_reuses_clients_by_resolved_route() {
let builder_count = Arc::new(AtomicUsize::new(0));
let observed_builder_count = Arc::clone(&builder_count);
let pool = RouteAwareClientPool::with_builder_factory(
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
ClientRouteClass::Api,
move || {
observed_builder_count.fetch_add(1, Ordering::SeqCst);
reqwest::Client::builder()
},
);
let direct_url = "https://example.com/first?target=direct";
let same_route_url = "https://example.com/second?target=direct%202";
let proxy_url = "https://example.com/third?target=proxy";
let resolver = FakeRouteResolver::new(HashMap::from([
(direct_url.to_string(), OutboundProxyRoute::Direct),
(same_route_url.to_string(), OutboundProxyRoute::Direct),
(
proxy_url.to_string(),
OutboundProxyRoute::Proxy {
url: "http://proxy.example".to_string(),
},
),
]));
resolve_with(&pool, &resolver, direct_url)
.await
.expect("first client should build");
resolve_with(&pool, &resolver, same_route_url)
.await
.expect("second client should reuse the route");
resolve_with(&pool, &resolver, proxy_url)
.await
.expect("proxy client should build separately");
assert_eq!(builder_count.load(Ordering::SeqCst), 2);
assert_eq!(
resolver.observed_urls(),
vec![
direct_url.to_string(),
same_route_url.to_string(),
proxy_url.to_string(),
]
);
}
#[tokio::test]
async fn bounds_cached_routes_and_rebuilds_an_evicted_route() {
let builder_count = Arc::new(AtomicUsize::new(0));
let observed_builder_count = Arc::clone(&builder_count);
let pool = RouteAwareClientPool::with_builder_factory(
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
ClientRouteClass::Api,
move || {
observed_builder_count.fetch_add(1, Ordering::SeqCst);
reqwest::Client::builder()
},
);
let routes = (0..=MAX_CACHED_ROUTES)
.map(|index| {
(
format!("https://target-{index}.example"),
OutboundProxyRoute::Proxy {
url: format!("http://proxy-{index}.example"),
},
)
})
.collect::<HashMap<_, _>>();
let resolver = FakeRouteResolver::new(routes.clone());
for request_url in routes.keys() {
resolve_with(&pool, &resolver, request_url)
.await
.expect("client should build");
}
let evicted_route = {
let clients = pool.clients.lock().expect("client cache lock");
assert_eq!(clients.len(), MAX_CACHED_ROUTES);
routes
.iter()
.find(|(_, route)| !clients.contains_key(*route))
.map(|(request_url, _)| request_url.clone())
.expect("one route should have been evicted")
};
resolve_with(&pool, &resolver, &evicted_route)
.await
.expect("evicted client should rebuild");
assert_eq!(builder_count.load(Ordering::SeqCst), MAX_CACHED_ROUTES + 2);
assert_eq!(
pool.clients.lock().expect("client cache lock").len(),
MAX_CACHED_ROUTES
);
}
#[derive(Clone)]
struct FakeRouteResolver {
routes: Arc<HashMap<String, OutboundProxyRoute>>,
observed_urls: Arc<Mutex<Vec<String>>>,
}
impl FakeRouteResolver {
fn new(routes: HashMap<String, OutboundProxyRoute>) -> Self {
Self {
routes: Arc::new(routes),
observed_urls: Arc::new(Mutex::new(Vec::new())),
}
}
async fn resolve(&self, request_url: String) -> io::Result<OutboundProxyRoute> {
self.observed_urls
.lock()
.expect("observed URL lock")
.push(request_url.clone());
self.routes
.get(&request_url)
.cloned()
.ok_or_else(|| io::Error::other(format!("no route for {request_url}")))
}
fn observed_urls(&self) -> Vec<String> {
self.observed_urls
.lock()
.expect("observed URL lock")
.clone()
}
}
async fn resolve_with(
pool: &RouteAwareClientPool,
resolver: &FakeRouteResolver,
request_url: &str,
) -> Result<HttpClient, RouteAwareClientPoolError> {
let resolver = resolver.clone();
pool.client_for_url_with_resolver(request_url, move |request_url| async move {
resolver.resolve(request_url).await
})
.await
}
+2 -1
View File
@@ -17,7 +17,8 @@ impl AuthRouteConfig {
}
}
pub(crate) fn http_client_factory(&self) -> &HttpClientFactory {
/// Returns the HTTP client factory represented by this routing configuration.
pub fn http_client_factory(&self) -> &HttpClientFactory {
&self.http_client_factory
}
}
+7 -3
View File
@@ -18,9 +18,13 @@ async fn rate_limits_check(auth_manager: &AuthManager, config: &Config) -> Optio
return None;
}
let client = BackendClient::from_auth(config.chatgpt_base_url.clone(), &auth)
.map_err(|err| warn!(%err, "failed to construct backend client"))
.ok()?;
let client = BackendClient::from_auth(
config.chatgpt_base_url.clone(),
&auth,
config.http_client_factory(),
)
.map_err(|err| warn!(%err, "failed to construct backend client"))
.ok()?;
let snapshots = client
.get_rate_limits_many()