core-plugins: route curated startup sync through HTTP client factory

This commit is contained in:
Michael Bolin
2026-07-09 15:34:38 -07:00
parent fb2c0ac189
commit bc05be0f72
11 changed files with 226 additions and 39 deletions
+2
View File
@@ -2787,6 +2787,7 @@ dependencies = [
"codex-exec-server",
"codex-git-utils",
"codex-hooks",
"codex-http-client",
"codex-login",
"codex-mcp",
"codex-model-provider",
@@ -2800,6 +2801,7 @@ dependencies = [
"codex-utils-plugins",
"dirs",
"flate2",
"http 1.4.0",
"libc",
"pretty_assertions",
"regex",
+2
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,6 +37,7 @@ codex-utils-plugins = { workspace = true }
chrono = { workspace = true }
dirs = { workspace = true }
flate2 = { workspace = true }
http = { workspace = true }
reqwest = { workspace = true }
regex = { workspace = true }
semver = { workspace = true }
@@ -0,0 +1,18 @@
use codex_http_client::RouteAwareClientPool;
use codex_http_client::RouteAwareRequestBuilder;
use http::Method;
use std::fmt::Debug;
/// Builds requests whose URL is also used to resolve their outbound route.
///
/// Implementations must keep route selection coupled to the request URL. Returning a transport
/// client would let callers send a different URL than the one used for route selection.
pub(crate) trait HttpClientSelector: Debug + Send + Sync {
fn request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder;
}
impl HttpClientSelector for RouteAwareClientPool {
fn request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder {
RouteAwareClientPool::request(self, method, 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;
+11 -6
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,17 @@ 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,
}
}
}
/// Inputs used to select endpoint-backed plugin install candidates.
@@ -1942,7 +1947,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 +2303,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 +2311,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 +2336,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}");
@@ -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());
+80 -29
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::HttpClientFactory;
use codex_http_client::RouteAwareClientPool;
use codex_http_client::RouteAwareRequestBuilder;
use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_FINAL_METRIC;
use codex_otel::CURATED_PLUGINS_STARTUP_SYNC_METRIC;
use reqwest::Client;
use http::Method;
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,13 @@ 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_without_request_logging(
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 +350,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 +365,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 +374,13 @@ 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_without_request_logging(
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 +790,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 +810,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 +824,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 +862,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 +959,12 @@ 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 response = github_request(http_clients, url)
.send()
.await
.map_err(|err| format!("failed to {context} from {url}: {err}"))?;
@@ -939,8 +978,12 @@ 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 response = github_request(http_clients, url)
.send()
.await
.map_err(|err| format!("failed to {context} from {url}: {err}"))?;
@@ -958,9 +1001,13 @@ 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> {
let response = client
.get(url)
async fn fetch_public_text(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<String, String> {
let response = http_clients
.request(Method::GET, url)
.timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT)
.send()
.await
@@ -975,9 +1022,13 @@ 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> {
let response = client
.get(url)
async fn fetch_public_bytes(
http_clients: &dyn HttpClientSelector,
url: &str,
context: &str,
) -> Result<Vec<u8>, String> {
let response = http_clients
.request(Method::GET, url)
.timeout(CURATED_PLUGINS_BACKUP_ARCHIVE_TIMEOUT)
.send()
.await
@@ -996,9 +1047,9 @@ 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 {
client
.get(url)
fn github_request(http_clients: &dyn HttpClientSelector, url: &str) -> RouteAwareRequestBuilder {
http_clients
.request(Method::GET, url)
.timeout(CURATED_PLUGINS_HTTP_TIMEOUT)
.header("accept", GITHUB_API_ACCEPT_HEADER)
.header("x-github-api-version", GITHUB_API_VERSION_HEADER)
@@ -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);
+54
View File
@@ -1,19 +1,72 @@
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::HttpClientSelector;
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_http_client::RouteAwareRequestBuilder;
use codex_utils_absolute_path::AbsolutePathBuf;
use http::Method;
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 request(&self, method: Method, url: &str) -> RouteAwareRequestBuilder {
match self.selected_urls.lock() {
Ok(mut selected_urls) => selected_urls.push(url.to_string()),
Err(error) => panic!("selected URL recorder lock should not be poisoned: {error}"),
}
self.delegate.request(method, url)
}
}
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 +205,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(),
)
}
+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