From 130fc9e362097bb45445c880e0078de23c32e4b1 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Wed, 8 Jul 2026 18:23:51 -0700 Subject: [PATCH] model-provider: route model discovery through HTTP client factory --- codex-rs/Cargo.lock | 2 + codex-rs/app-server/Cargo.toml | 1 + codex-rs/app-server/src/message_processor.rs | 3 +- codex-rs/app-server/src/models.rs | 4 +- .../app-server/src/models_refresh_worker.rs | 13 +- .../src/models_refresh_worker_tests.rs | 9 +- .../request_processors/catalog_processor.rs | 18 ++- .../external_agent_session_import.rs | 1 + codex-rs/cli/src/main.rs | 5 +- codex-rs/codex-api/src/endpoint/models.rs | 42 ++--- .../codex-api/tests/models_integration.rs | 6 +- codex-rs/core/src/guardian/review.rs | 5 +- codex-rs/core/src/session/mod.rs | 5 +- codex-rs/core/src/session/review.rs | 5 +- codex-rs/core/src/session/turn.rs | 5 +- codex-rs/core/src/session/turn_context.rs | 5 +- codex-rs/core/src/test_support.rs | 6 + codex-rs/core/src/thread_manager.rs | 8 +- codex-rs/core/src/thread_manager_tests.rs | 7 +- .../src/tools/handlers/multi_agents_common.rs | 2 +- codex-rs/core/tests/suite/auto_review.rs | 5 +- .../tests/suite/model_runtime_selectors.rs | 5 +- codex-rs/core/tests/suite/model_switching.rs | 27 +++- codex-rs/core/tests/suite/models_cache_ttl.rs | 25 ++- codex-rs/core/tests/suite/personality.rs | 7 +- codex-rs/core/tests/suite/remote_models.rs | 51 +++++- codex-rs/core/tests/suite/rmcp_client.rs | 5 +- .../tests/suite/spawn_agent_description.rs | 7 +- codex-rs/http-client/src/outbound_proxy.rs | 40 ++++- .../http-client/src/outbound_proxy_tests.rs | 39 +++++ codex-rs/login/src/auth/default_client.rs | 22 +++ .../model-provider/src/models_endpoint.rs | 149 ++++++++++++++++-- codex-rs/model-provider/src/provider.rs | 28 +++- codex-rs/models-manager/Cargo.toml | 1 + codex-rs/models-manager/src/manager.rs | 84 ++++++++-- codex-rs/models-manager/src/manager_tests.rs | 142 +++++++++++++---- 36 files changed, 658 insertions(+), 131 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 3d9847974996..1c854338e302 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1995,6 +1995,7 @@ dependencies = [ "codex-guardian", "codex-home", "codex-hooks", + "codex-http-client", "codex-image-generation-extension", "codex-login", "codex-mcp", @@ -3588,6 +3589,7 @@ version = "0.0.0" dependencies = [ "chrono", "codex-collaboration-mode-templates", + "codex-http-client", "codex-login", "codex-otel", "codex-protocol", diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index bde8048e76d5..41de48a0b509 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -47,6 +47,7 @@ codex-guardian = { workspace = true } codex-git-utils = { workspace = true } codex-file-watcher = { workspace = true } codex-hooks = { workspace = true } +codex-http-client = { workspace = true } codex-otel = { workspace = true } codex-plugin = { workspace = true } codex-shell-command = { workspace = true } diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index ca122920bd64..62eff72660da 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -299,7 +299,8 @@ impl MessageProcessor { ) }); let models_manager = thread_manager.get_models_manager(); - let models_refresh_worker = crate::models_refresh_worker::spawn(&models_manager); + let models_refresh_worker = + crate::models_refresh_worker::spawn(&models_manager, config.http_client_factory()); thread_manager .plugins_manager() .set_analytics_events_client(analytics_events_client.clone()); diff --git a/codex-rs/app-server/src/models.rs b/codex-rs/app-server/src/models.rs index 7d460d5a97b4..36219e886f93 100644 --- a/codex-rs/app-server/src/models.rs +++ b/codex-rs/app-server/src/models.rs @@ -5,6 +5,7 @@ use codex_app_server_protocol::ModelServiceTier; use codex_app_server_protocol::ModelUpgradeInfo; use codex_app_server_protocol::ReasoningEffortOption; use codex_core::ThreadManager; +use codex_http_client::HttpClientFactory; use codex_models_manager::manager::RefreshStrategy; use codex_protocol::openai_models::ModelPreset; use codex_protocol::openai_models::ReasoningEffortPreset; @@ -12,9 +13,10 @@ use codex_protocol::openai_models::ReasoningEffortPreset; pub async fn supported_models( thread_manager: Arc, include_hidden: bool, + http_client_factory: HttpClientFactory, ) -> Vec { thread_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models(RefreshStrategy::OnlineIfUncached, http_client_factory) .await .into_iter() .filter(|preset| include_hidden || preset.show_in_picker) diff --git a/codex-rs/app-server/src/models_refresh_worker.rs b/codex-rs/app-server/src/models_refresh_worker.rs index 475c8acabf54..ab785867a6f4 100644 --- a/codex-rs/app-server/src/models_refresh_worker.rs +++ b/codex-rs/app-server/src/models_refresh_worker.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use std::time::Duration; +use codex_http_client::HttpClientFactory; use codex_models_manager::manager::RefreshStrategy; use codex_models_manager::manager::SharedModelsManager; use tokio::task::JoinHandle; @@ -26,12 +27,16 @@ impl Drop for ModelsRefreshWorker { } } -pub(crate) fn spawn(models_manager: &SharedModelsManager) -> ModelsRefreshWorker { - spawn_with_interval(models_manager, MODELS_REFRESH_INTERVAL) +pub(crate) fn spawn( + models_manager: &SharedModelsManager, + http_client_factory: HttpClientFactory, +) -> ModelsRefreshWorker { + spawn_with_interval(models_manager, http_client_factory, MODELS_REFRESH_INTERVAL) } fn spawn_with_interval( models_manager: &SharedModelsManager, + http_client_factory: HttpClientFactory, refresh_interval: Duration, ) -> ModelsRefreshWorker { let models_manager = Arc::downgrade(models_manager); @@ -45,7 +50,9 @@ fn spawn_with_interval( let Some(models_manager) = models_manager.upgrade() else { break; }; - models_manager.list_models(RefreshStrategy::Online).await; + models_manager + .list_models(RefreshStrategy::Online, http_client_factory.clone()) + .await; drop(models_manager); tokio::select! { diff --git a/codex-rs/app-server/src/models_refresh_worker_tests.rs b/codex-rs/app-server/src/models_refresh_worker_tests.rs index 4d7ff48d31a1..989bfd8608b0 100644 --- a/codex-rs/app-server/src/models_refresh_worker_tests.rs +++ b/codex-rs/app-server/src/models_refresh_worker_tests.rs @@ -3,6 +3,8 @@ use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::time::Duration; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_models_manager::manager::ModelsEndpointClient; use codex_models_manager::manager::ModelsEndpointFuture; use codex_models_manager::manager::OpenAiModelsManager; @@ -55,6 +57,7 @@ impl ModelsEndpointClient for TestModelsEndpoint { fn list_models<'a>( &'a self, _client_version: &'a str, + _http_client_factory: HttpClientFactory, ) -> ModelsEndpointFuture<'a, CoreResult<(Vec, Option)>> { Box::pin(async move { let fetch_index = self.fetch_count.fetch_add(1, Ordering::SeqCst); @@ -79,7 +82,11 @@ async fn refreshes_immediately_periodically_and_stops_when_dropped() { endpoint.clone(), /*auth_manager*/ None, )); - let worker = spawn_with_interval(&models_manager, Duration::from_millis(10)); + let worker = spawn_with_interval( + &models_manager, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + Duration::from_millis(10), + ); endpoint.wait_for_fetch_count(/*expected*/ 2).await; drop(worker); diff --git a/codex-rs/app-server/src/request_processors/catalog_processor.rs b/codex-rs/app-server/src/request_processors/catalog_processor.rs index 95423402003d..6cb4385151a5 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -157,9 +157,13 @@ impl CatalogRequestProcessor { &self, params: ModelListParams, ) -> Result, JSONRPCErrorError> { - Self::list_models(self.thread_manager.clone(), params) - .await - .map(|response| Some(response.into())) + Self::list_models( + self.thread_manager.clone(), + self.config.http_client_factory(), + params, + ) + .await + .map(|response| Some(response.into())) } pub(crate) async fn experimental_feature_list( @@ -247,6 +251,7 @@ impl CatalogRequestProcessor { async fn list_models( thread_manager: Arc, + http_client_factory: codex_http_client::HttpClientFactory, params: ModelListParams, ) -> Result { let ModelListParams { @@ -254,7 +259,12 @@ impl CatalogRequestProcessor { cursor, include_hidden, } = params; - let models = supported_models(thread_manager, include_hidden.unwrap_or(false)).await; + let models = supported_models( + thread_manager, + include_hidden.unwrap_or(false), + http_client_factory, + ) + .await; let total = models.len(); if total == 0 { diff --git a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs index e0a53a9c983e..e574b7a64b5d 100644 --- a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs +++ b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs @@ -189,6 +189,7 @@ impl ExternalAgentSessionImporter { &config.model, /*allow_provider_model_fallback*/ false, RefreshStrategy::Offline, + config.http_client_factory(), ) .await; let model_info = models_manager diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index c766baa04881..04d59c24abf4 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -2003,7 +2003,10 @@ async fn run_debug_models_command( AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ true).await; let models_manager = build_models_manager(&config, auth_manager); models_manager - .raw_model_catalog(RefreshStrategy::OnlineIfUncached) + .raw_model_catalog( + RefreshStrategy::OnlineIfUncached, + config.http_client_factory(), + ) .await }; diff --git a/codex-rs/codex-api/src/endpoint/models.rs b/codex-rs/codex-api/src/endpoint/models.rs index 2f1b799c3ba4..d210bb64c822 100644 --- a/codex-rs/codex-api/src/endpoint/models.rs +++ b/codex-rs/codex-api/src/endpoint/models.rs @@ -37,9 +37,15 @@ impl ModelsClient { req.url = format!("{}{}client_version={client_version}", req.url, separator); } + pub fn request_url(provider: &Provider, client_version: &str) -> String { + let mut request = provider.build_request(Method::GET, Self::path()); + Self::append_client_version_query(&mut request, client_version); + request.url + } + pub async fn list_models( &self, - client_version: &str, + request_url: String, extra_headers: HeaderMap, ) -> Result<(Vec, Option), ApiError> { let resp = self @@ -49,8 +55,8 @@ impl ModelsClient { Self::path(), extra_headers, /*body*/ None, - |req| { - Self::append_client_version_query(req, client_version); + move |req| { + req.url.clone_from(&request_url); }, ) .await?; @@ -161,14 +167,12 @@ mod tests { etag: None, }; - let client = ModelsClient::new( - transport.clone(), - provider("https://example.com/api/codex"), - Arc::new(DummyAuth), - ); + let provider = provider("https://example.com/api/codex"); + let request_url = ModelsClient::::request_url(&provider, "0.99.0"); + let client = ModelsClient::new(transport.clone(), provider, Arc::new(DummyAuth)); let (models, _) = client - .list_models("0.99.0", HeaderMap::new()) + .list_models(request_url, HeaderMap::new()) .await .expect("request should succeed"); @@ -225,14 +229,12 @@ mod tests { etag: None, }; - let client = ModelsClient::new( - transport, - provider("https://example.com/api/codex"), - Arc::new(DummyAuth), - ); + let provider = provider("https://example.com/api/codex"); + let request_url = ModelsClient::::request_url(&provider, "0.99.0"); + let client = ModelsClient::new(transport, provider, Arc::new(DummyAuth)); let (models, _) = client - .list_models("0.99.0", HeaderMap::new()) + .list_models(request_url, HeaderMap::new()) .await .expect("request should succeed"); @@ -252,14 +254,12 @@ mod tests { etag: Some("\"abc\"".to_string()), }; - let client = ModelsClient::new( - transport, - provider("https://example.com/api/codex"), - Arc::new(DummyAuth), - ); + let provider = provider("https://example.com/api/codex"); + let request_url = ModelsClient::::request_url(&provider, "0.1.0"); + let client = ModelsClient::new(transport, provider, Arc::new(DummyAuth)); let (models, etag) = client - .list_models("0.1.0", HeaderMap::new()) + .list_models(request_url, HeaderMap::new()) .await .expect("request should succeed"); diff --git a/codex-rs/codex-api/tests/models_integration.rs b/codex-rs/codex-api/tests/models_integration.rs index b5286043295f..8eddf01906b1 100644 --- a/codex-rs/codex-api/tests/models_integration.rs +++ b/codex-rs/codex-api/tests/models_integration.rs @@ -118,10 +118,12 @@ async fn models_client_hits_models_endpoint() { .await; let transport = ReqwestTransport::new(reqwest::Client::new()); - let client = ModelsClient::new(transport, provider(&base_url), Arc::new(DummyAuth)); + let provider = provider(&base_url); + let request_url = ModelsClient::::request_url(&provider, "0.1.0"); + let client = ModelsClient::new(transport, provider, Arc::new(DummyAuth)); let (models, _) = client - .list_models("0.1.0", HeaderMap::new()) + .list_models(request_url, HeaderMap::new()) .await .expect("models request should succeed"); diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 4756cc1fc404..e606c01d093e 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -687,7 +687,10 @@ pub(super) async fn guardian_review_session_config( let available_models = session .services .models_manager - .list_models(codex_models_manager::manager::RefreshStrategy::Offline) + .list_models( + codex_models_manager::manager::RefreshStrategy::Offline, + turn.config.http_client_factory(), + ) .await; let default_review_model_id = turn.provider.approval_review_preferred_model(); let preferred_reasoning_effort = |supports_low: bool, fallback| { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 02c8a6747364..c6d86b23ef42 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -572,13 +572,16 @@ impl Codex { codex_models_manager::manager::RefreshStrategy::Offline ) { - let _ = models_manager.list_models(refresh_strategy).await; + let _ = models_manager + .list_models(refresh_strategy, config.http_client_factory()) + .await; } let model = models_manager .get_default_model( &config.model, allow_provider_model_fallback, refresh_strategy, + config.http_client_factory(), ) .await; if allow_provider_model_fallback diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index 33de46ede081..3379a340fb39 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -27,7 +27,10 @@ pub(super) async fn spawn_review_thread( let available_models = sess .services .models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + config.http_client_factory(), + ) .await; let unified_exec_shell_mode = UnifiedExecShellMode::for_session( codex_tools::unified_exec_feature_mode_for_features(review_features.get()), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 41098b2c1e5f..a7959d136096 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -2252,7 +2252,10 @@ async fn try_run_sampling_request( } ResponseEvent::ModelsEtag(etag) => { // Update internal state with latest models etag - sess.services.models_manager.refresh_if_new_etag(etag).await; + sess.services + .models_manager + .refresh_if_new_etag(etag, turn_context.config.http_client_factory()) + .await; } ResponseEvent::Completed { token_usage, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index b9bf42073b3c..690fc501c2f3 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -244,7 +244,10 @@ impl TurnContext { /*developer_instructions*/ None, ); let available_models = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + config.http_client_factory(), + ) .await; Self { diff --git a/codex-rs/core/src/test_support.rs b/codex-rs/core/src/test_support.rs index 1f946602f5b6..8c8433085726 100644 --- a/codex-rs/core/src/test_support.rs +++ b/codex-rs/core/src/test_support.rs @@ -11,6 +11,8 @@ use codex_exec_server::EnvironmentManager; use codex_extension_api::LoadUserInstructionsFuture; use codex_extension_api::LoadedUserInstructions; use codex_extension_api::UserInstructionsProvider; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_model_provider::create_model_provider; @@ -158,6 +160,10 @@ pub fn models_manager_with_provider( provider.models_manager(codex_home, /*config_model_catalog*/ None) } +pub fn default_http_client_factory() -> HttpClientFactory { + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) +} + pub fn get_model_offline(model: Option<&str>) -> String { get_model_offline_for_tests(model) } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 669da4954775..de37484f8bee 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -542,10 +542,14 @@ impl ThreadManager { self.state.models_manager.clone() } - pub async fn list_models(&self, refresh_strategy: RefreshStrategy) -> Vec { + pub async fn list_models( + &self, + refresh_strategy: RefreshStrategy, + http_client_factory: codex_http_client::HttpClientFactory, + ) -> Vec { self.state .models_manager - .list_models(refresh_strategy) + .list_models(refresh_strategy, http_client_factory) .await } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 3c1d0146e7f4..bf8b7f70641e 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1336,7 +1336,12 @@ async fn new_uses_active_provider_for_model_refresh() { /*external_time_provider*/ None, ); - let _ = manager.list_models(RefreshStrategy::Online).await; + let _ = manager + .list_models( + RefreshStrategy::Online, + crate::test_support::default_http_client_factory(), + ) + .await; assert_eq!(models_mock.requests().len(), 1); } diff --git a/codex-rs/core/src/tools/handlers/multi_agents_common.rs b/codex-rs/core/src/tools/handlers/multi_agents_common.rs index 287e4b30d5d6..f0ebbd73692a 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_common.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_common.rs @@ -246,7 +246,7 @@ pub(crate) async fn apply_requested_spawn_agent_model_overrides( let available_models = session .services .models_manager - .list_models(RefreshStrategy::Offline) + .list_models(RefreshStrategy::Offline, config.http_client_factory()) .await; let selected_model_name = find_spawn_agent_model_name(&available_models, requested_model)?; let selected_model_info = session diff --git a/codex-rs/core/tests/suite/auto_review.rs b/codex-rs/core/tests/suite/auto_review.rs index dd39713ef2fc..30c4d4a595c1 100644 --- a/codex-rs/core/tests/suite/auto_review.rs +++ b/codex-rs/core/tests/suite/auto_review.rs @@ -131,7 +131,10 @@ async fn remote_model_override_uses_catalog_model_for_strict_auto_review() -> Re let models_manager = thread_manager.get_models_manager(); models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; let model_info = models_manager .get_model_info(model, &config.to_models_manager_config()) diff --git a/codex-rs/core/tests/suite/model_runtime_selectors.rs b/codex-rs/core/tests/suite/model_runtime_selectors.rs index 0c6a87605de5..6abb801da36c 100644 --- a/codex-rs/core/tests/suite/model_runtime_selectors.rs +++ b/codex-rs/core/tests/suite/model_runtime_selectors.rs @@ -74,7 +74,10 @@ async fn wait_for_model_available(manager: &SharedModelsManager, slug: &str) -> let deadline = Instant::now() + Duration::from_secs(2); loop { if let Some(model) = manager - .list_models(RefreshStrategy::Online) + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) .await .iter() .find(|model| model.model == slug) diff --git a/codex-rs/core/tests/suite/model_switching.rs b/codex-rs/core/tests/suite/model_switching.rs index 9ef3675b3ed2..2c54adcf51fa 100644 --- a/codex-rs/core/tests/suite/model_switching.rs +++ b/codex-rs/core/tests/suite/model_switching.rs @@ -546,7 +546,10 @@ async fn model_change_from_image_to_text_strips_prior_image_content() -> Result< let test = builder.build(&server).await?; let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; let image_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==" .to_string(); @@ -652,7 +655,10 @@ async fn generated_image_is_replayed_for_image_capable_models() -> Result<()> { let _ = std::fs::remove_file(&saved_path); let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; test.codex @@ -766,7 +772,10 @@ async fn model_change_from_generated_image_to_text_preserves_prior_generated_ima let _ = std::fs::remove_file(&saved_path); let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; test.codex @@ -884,7 +893,10 @@ async fn thread_rollback_after_generated_image_drops_entire_image_turn_history() let _ = std::fs::remove_file(&saved_path); let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; test.codex @@ -1044,7 +1056,12 @@ async fn model_switch_to_smaller_model_updates_token_context_window() -> Result< let test = builder.build(&server).await?; let models_manager = test.thread_manager.get_models_manager(); - let available_models = models_manager.list_models(RefreshStrategy::Online).await; + let available_models = models_manager + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) + .await; assert!( available_models .iter() diff --git a/codex-rs/core/tests/suite/models_cache_ttl.rs b/codex-rs/core/tests/suite/models_cache_ttl.rs index 4044ba8efa26..161a415008d0 100644 --- a/codex-rs/core/tests/suite/models_cache_ttl.rs +++ b/codex-rs/core/tests/suite/models_cache_ttl.rs @@ -71,7 +71,10 @@ async fn renews_cache_ttl_on_matching_models_etag() -> Result<()> { // Populate cache via initial refresh. let models_manager = test.thread_manager.get_models_manager(); let _ = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; let cache_path = config.codex_home.join(CACHE_FILE); @@ -135,7 +138,10 @@ async fn renews_cache_ttl_on_matching_models_etag() -> Result<()> { // Cached models remain usable offline. let offline_models = test .thread_manager - .list_models(RefreshStrategy::Offline) + .list_models( + RefreshStrategy::Offline, + codex_core::test_support::default_http_client_factory(), + ) .await; assert!( offline_models @@ -178,7 +184,10 @@ async fn uses_cache_when_version_matches() -> Result<()> { let test = builder.build(&server).await?; let models_manager = test.thread_manager.get_models_manager(); let models = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; assert!( @@ -225,7 +234,10 @@ async fn refreshes_when_cache_version_missing() -> Result<()> { let test = builder.build(&server).await?; let models_manager = test.thread_manager.get_models_manager(); let models = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; assert!( @@ -273,7 +285,10 @@ async fn refreshes_when_cache_version_differs() -> Result<()> { let test = builder.build(&server).await?; let models_manager = test.thread_manager.get_models_manager(); let models = models_manager - .list_models(RefreshStrategy::OnlineIfUncached) + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) .await; assert!( diff --git a/codex-rs/core/tests/suite/personality.rs b/codex-rs/core/tests/suite/personality.rs index 88733b466674..ed3927330028 100644 --- a/codex-rs/core/tests/suite/personality.rs +++ b/codex-rs/core/tests/suite/personality.rs @@ -802,7 +802,12 @@ async fn user_turn_personality_remote_model_template_includes_update_message() - async fn wait_for_model_available(manager: &SharedModelsManager, slug: &str) { let deadline = Instant::now() + Duration::from_secs(2); loop { - let models = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let models = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; if models.iter().any(|model| model.model == slug) { return; } diff --git a/codex-rs/core/tests/suite/remote_models.rs b/codex-rs/core/tests/suite/remote_models.rs index 16e31e50826d..895af1a64b49 100644 --- a/codex-rs/core/tests/suite/remote_models.rs +++ b/codex-rs/core/tests/suite/remote_models.rs @@ -103,7 +103,12 @@ async fn remote_models_get_model_info_uses_longest_matching_prefix() -> Result<( provider, ); - manager.list_models(RefreshStrategy::OnlineIfUncached).await; + manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; let model_info = manager .get_model_info("gpt-5.3-codex-test", &config.to_models_manager_config()) @@ -864,7 +869,12 @@ async fn remote_models_do_not_append_removed_builtin_presets() -> Result<()> { provider, ); - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; let remote = available .iter() .find(|model| model.model == "remote-alpha") @@ -925,7 +935,12 @@ async fn remote_models_merge_adds_new_high_priority_first() -> Result<()> { provider, ); - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; assert_eq!( available.first().map(|model| model.model.as_str()), Some("remote-top") @@ -972,7 +987,12 @@ async fn remote_models_merge_replaces_overlapping_model() -> Result<()> { provider, ); - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; let overridden = available .iter() .find(|model| model.model == slug) @@ -1016,7 +1036,12 @@ async fn remote_models_merge_preserves_bundled_models_on_empty_response() -> Res provider, ); - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; let bundled_slug = bundled_model_slug(); assert!( available.iter().any(|model| model.model == bundled_slug), @@ -1065,6 +1090,7 @@ async fn remote_models_request_times_out_after_5s() -> Result<()> { &None, /*allow_provider_model_fallback*/ false, RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), ), ) .await; @@ -1137,11 +1163,17 @@ async fn remote_models_hide_picker_only_models() -> Result<()> { &None, /*allow_provider_model_fallback*/ false, RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), ) .await; assert_eq!(selected, bundled_default_model_slug()); - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; let hidden = available .iter() .find(|model| model.model == "codex-auto-balanced") @@ -1162,7 +1194,12 @@ async fn wait_for_model_available(manager: &SharedModelsManager, slug: &str) -> let deadline = Instant::now() + Duration::from_secs(2); loop { if let Some(model) = { - let guard = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let guard = manager + .list_models( + RefreshStrategy::OnlineIfUncached, + codex_core::test_support::default_http_client_factory(), + ) + .await; guard.iter().find(|model| model.model == slug).cloned() } { return model; diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index e5bc6a34b653..d3c7ea81b453 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -1729,7 +1729,10 @@ async fn stdio_image_responses_are_sanitized_for_text_only_model() -> anyhow::Re fixture .thread_manager .get_models_manager() - .list_models(RefreshStrategy::Online) + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) .await; assert_eq!(models_mock.requests().len(), 1); diff --git a/codex-rs/core/tests/suite/spawn_agent_description.rs b/codex-rs/core/tests/suite/spawn_agent_description.rs index e7076e74650c..830e69bc29e8 100644 --- a/codex-rs/core/tests/suite/spawn_agent_description.rs +++ b/codex-rs/core/tests/suite/spawn_agent_description.rs @@ -94,7 +94,12 @@ fn test_model_info( async fn wait_for_model_available(manager: &SharedModelsManager, slug: &str) { let deadline = Instant::now() + Duration::from_secs(2); loop { - let available_models = manager.list_models(RefreshStrategy::Online).await; + let available_models = manager + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) + .await; if available_models.iter().any(|model| model.model == slug) { return; } diff --git a/codex-rs/http-client/src/outbound_proxy.rs b/codex-rs/http-client/src/outbound_proxy.rs index 71793025ede5..7bdfd745b7c4 100644 --- a/codex-rs/http-client/src/outbound_proxy.rs +++ b/codex-rs/http-client/src/outbound_proxy.rs @@ -352,12 +352,32 @@ enum SystemProxyDecision { } fn resolve_system_proxy(request_url: &str, origin: &RequestOrigin) -> SystemProxyDecision { - if let Some(decision) = cached_system_proxy_decision(request_url) { + let cache = SYSTEM_PROXY_CACHE.get_or_init(|| Mutex::new(HashMap::new())); + resolve_system_proxy_with(cache, request_url, origin, resolve_platform_system_proxy) +} + +fn resolve_system_proxy_with( + cache: &Mutex>, + request_url: &str, + origin: &RequestOrigin, + resolve_platform_system_proxy: impl FnOnce(&str, &RequestOrigin) -> SystemProxyDecision, +) -> SystemProxyDecision { + let mut cache = match cache.lock() { + Ok(cache) => cache, + Err(error) => panic!("system proxy cache lock should not be poisoned: {error}"), + }; + let cache_key = system_proxy_cache_key(request_url); + if let Some(decision) = + cached_system_proxy_decision_from_cache(&mut cache, &cache_key, Instant::now()) + { return decision; } + // Keep cache misses single-flight. Platform PAC/WPAD APIs are synchronous, so async callers + // run this work on the blocking pool; serializing misses prevents concurrent requests from + // consuming an unbounded number of blocking workers while system lookup is pending. let decision = resolve_platform_system_proxy(request_url, origin); - cache_system_proxy_decision(request_url, decision.clone()); + insert_system_proxy_cache_entry(&mut cache, &cache_key, decision.clone(), Instant::now()); decision } @@ -390,18 +410,28 @@ struct CachedSystemProxyDecision { static SYSTEM_PROXY_CACHE: OnceLock>> = OnceLock::new(); +#[cfg(test)] fn cached_system_proxy_decision(request_url: &str) -> Option { let cache = SYSTEM_PROXY_CACHE.get_or_init(|| Mutex::new(HashMap::new())); let mut cache = cache.lock().ok()?; let key = system_proxy_cache_key(request_url); - let cached = cache.get(&key)?; - if cached.expires_at > Instant::now() { + cached_system_proxy_decision_from_cache(&mut cache, &key, Instant::now()) +} + +fn cached_system_proxy_decision_from_cache( + cache: &mut HashMap, + cache_key: &str, + now: Instant, +) -> Option { + let cached = cache.get(cache_key)?; + if cached.expires_at > now { return Some(cached.decision.clone()); } - cache.remove(&key); + cache.remove(cache_key); None } +#[cfg(test)] fn cache_system_proxy_decision(request_url: &str, decision: SystemProxyDecision) { let cache = SYSTEM_PROXY_CACHE.get_or_init(|| Mutex::new(HashMap::new())); if let Ok(mut cache) = cache.lock() { diff --git a/codex-rs/http-client/src/outbound_proxy_tests.rs b/codex-rs/http-client/src/outbound_proxy_tests.rs index 3f8bd925c910..387973eec493 100644 --- a/codex-rs/http-client/src/outbound_proxy_tests.rs +++ b/codex-rs/http-client/src/outbound_proxy_tests.rs @@ -4,6 +4,7 @@ use super::*; use pretty_assertions::assert_eq; use std::io::Read; use std::io::Write; +use std::sync::Arc; struct MapEnv { values: HashMap, @@ -166,6 +167,44 @@ fn unavailable_system_proxy_decision_is_cached() { assert_eq!(cached_system_proxy_decision(request_url), Some(decision)); } +#[test] +fn system_proxy_resolution_is_single_flight() { + let cache = Arc::new(Mutex::new(HashMap::new())); + let request_url = "https://single-flight.test/models"; + let origin = RequestOrigin::parse(request_url).expect("valid request URL"); + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let worker_cache = Arc::clone(&cache); + let worker_origin = origin.clone(); + + let worker = std::thread::spawn(move || { + resolve_system_proxy_with(&worker_cache, request_url, &worker_origin, |_, _| { + started_tx.send(()).expect("test should still be running"); + release_rx.recv().expect("test should release resolver"); + SystemProxyDecision::Direct + }) + }); + + started_rx.recv().expect("resolver should start"); + assert!(matches!( + cache.try_lock(), + Err(std::sync::TryLockError::WouldBlock) + )); + release_tx + .send(()) + .expect("resolver should still be running"); + assert_eq!( + worker.join().expect("resolver should finish"), + SystemProxyDecision::Direct + ); + assert_eq!( + resolve_system_proxy_with(&cache, request_url, &origin, |_, _| { + panic!("cached waiter should not resolve the platform proxy again") + }), + SystemProxyDecision::Direct + ); +} + #[test] fn system_proxy_cache_is_bounded() { let mut cache = HashMap::new(); diff --git a/codex-rs/login/src/auth/default_client.rs b/codex-rs/login/src/auth/default_client.rs index 089e78340b65..0765d227ce1f 100644 --- a/codex-rs/login/src/auth/default_client.rs +++ b/codex-rs/login/src/auth/default_client.rs @@ -53,6 +53,8 @@ pub struct Originator { static ORIGINATOR: LazyLock>> = LazyLock::new(|| RwLock::new(None)); static REQUIREMENTS_RESIDENCY: LazyLock>> = LazyLock::new(|| RwLock::new(None)); +static ROUTE_AWARE_CLIENT_BUILD_PERMIT: tokio::sync::Semaphore = + tokio::sync::Semaphore::const_new(1); #[derive(Debug)] pub enum SetOriginatorError { @@ -265,6 +267,26 @@ pub fn build_default_reqwest_client_for_route( ) } +/// Builds the default Codex reqwest client for a concrete outbound route without blocking the +/// async runtime worker that initiated the request. +pub async fn build_default_reqwest_client_for_route_async( + http_client_factory: HttpClientFactory, + request_url: String, + route_class: ClientRouteClass, +) -> std::io::Result { + let permit = ROUTE_AWARE_CLIENT_BUILD_PERMIT + .acquire() + .await + .map_err(std::io::Error::other)?; + tokio::task::spawn_blocking(move || { + let _permit = permit; + build_default_reqwest_client_for_route(&http_client_factory, &request_url, route_class) + .map_err(std::io::Error::from) + }) + .await + .map_err(std::io::Error::other)? +} + fn default_reqwest_client_builder() -> reqwest::ClientBuilder { let mut builder = reqwest::Client::builder().default_headers(default_headers()); if is_sandboxed() { diff --git a/codex-rs/model-provider/src/models_endpoint.rs b/codex-rs/model-provider/src/models_endpoint.rs index 9a203c26e1c2..622cb8ec80ae 100644 --- a/codex-rs/model-provider/src/models_endpoint.rs +++ b/codex-rs/model-provider/src/models_endpoint.rs @@ -1,3 +1,6 @@ +use std::fmt; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::time::Duration; @@ -10,11 +13,13 @@ use codex_api::auth_header_telemetry; use codex_api::map_api_error; use codex_feedback::FeedbackRequestTags; use codex_feedback::emit_feedback_request_tags_with_auth_env; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; use codex_login::AuthEnvTelemetry; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_login::collect_auth_env_telemetry; -use codex_login::default_client::build_reqwest_client; +use codex_login::default_client::build_default_reqwest_client_for_route_async; use codex_model_provider_info::ModelProviderInfo; use codex_models_manager::manager::ModelsEndpointClient; use codex_models_manager::manager::ModelsEndpointFuture; @@ -38,6 +43,7 @@ const MODELS_ENDPOINT: &str = "/models"; pub(crate) struct OpenAiModelsEndpoint { provider_info: ModelProviderInfo, auth_manager: Option>, + transport_builder: Arc, } impl OpenAiModelsEndpoint { @@ -48,6 +54,7 @@ impl OpenAiModelsEndpoint { Self { provider_info, auth_manager, + transport_builder: Arc::new(RouteAwareModelsTransportBuilder), } } @@ -68,6 +75,7 @@ impl OpenAiModelsEndpoint { async fn list_models( &self, client_version: &str, + http_client_factory: HttpClientFactory, ) -> CoreResult<(Vec, Option)> { let _timer = codex_otel::start_global_timer("codex.remote_models.fetch_update.duration_ms", &[]); @@ -75,7 +83,8 @@ impl OpenAiModelsEndpoint { let auth_mode = auth.as_ref().map(CodexAuth::auth_mode); let api_provider = self.provider_info.to_api_provider(auth_mode)?; let api_auth = resolve_provider_auth(auth.as_ref(), &self.provider_info)?; - let transport = ReqwestTransport::new(build_reqwest_client()); + let request_url = + ModelsClient::::request_url(&api_provider, client_version); let auth_telemetry = auth_header_telemetry(api_auth.as_ref()); let agent_identity_telemetry = if let Some(CodexAuth::AgentIdentity(auth)) = auth.as_ref() { Some(agent_identity_telemetry(auth)) @@ -89,16 +98,20 @@ impl OpenAiModelsEndpoint { agent_identity_telemetry, auth_env: self.auth_env(), }); - let client = ModelsClient::new(transport, api_provider, api_auth) - .with_telemetry(Some(request_telemetry)); - - timeout( - MODELS_REFRESH_TIMEOUT, - client.list_models(client_version, HeaderMap::new()), - ) + timeout(MODELS_REFRESH_TIMEOUT, async { + let transport = self + .transport_builder + .build(http_client_factory, request_url.clone()) + .await?; + let client = ModelsClient::new(transport, api_provider, api_auth) + .with_telemetry(Some(request_telemetry)); + client + .list_models(request_url, HeaderMap::new()) + .await + .map_err(map_api_error) + }) .await .map_err(|_| CodexErr::Timeout)? - .map_err(map_api_error) } fn auth_env(&self) -> AuthEnvTelemetry { @@ -122,8 +135,48 @@ impl ModelsEndpointClient for OpenAiModelsEndpoint { fn list_models<'a>( &'a self, client_version: &'a str, + http_client_factory: HttpClientFactory, ) -> ModelsEndpointFuture<'a, CoreResult<(Vec, Option)>> { - Box::pin(OpenAiModelsEndpoint::list_models(self, client_version)) + Box::pin(OpenAiModelsEndpoint::list_models( + self, + client_version, + http_client_factory, + )) + } +} + +type ModelsTransportFuture<'a> = + Pin> + Send + 'a>>; + +/// Builds the concrete transport selected for one models request. +/// +/// Implementations must honor the supplied request-time client factory and exact request URL. +trait ModelsTransportBuilder: fmt::Debug + Send + Sync { + fn build( + &self, + http_client_factory: HttpClientFactory, + request_url: String, + ) -> ModelsTransportFuture<'_>; +} + +#[derive(Debug)] +struct RouteAwareModelsTransportBuilder; + +impl ModelsTransportBuilder for RouteAwareModelsTransportBuilder { + fn build( + &self, + http_client_factory: HttpClientFactory, + request_url: String, + ) -> ModelsTransportFuture<'_> { + Box::pin(async move { + build_default_reqwest_client_for_route_async( + http_client_factory, + request_url, + ClientRouteClass::Api, + ) + .await + .map(ReqwestTransport::new) + }) } } @@ -227,9 +280,42 @@ impl RequestTelemetry for ModelsRequestTelemetry { #[cfg(test)] mod tests { use std::num::NonZeroU64; + use std::sync::Mutex; use super::*; + use codex_http_client::OutboundProxyPolicy; + use codex_login::default_client::build_reqwest_client; use codex_protocol::config_types::ModelProviderAuthInfo; + use codex_protocol::openai_models::ModelsResponse; + use pretty_assertions::assert_eq; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::method; + use wiremock::matchers::path; + use wiremock::matchers::query_param; + + #[derive(Debug)] + struct RecordingTransportBuilder { + observed_request: Arc>>, + } + + impl ModelsTransportBuilder for RecordingTransportBuilder { + fn build( + &self, + http_client_factory: HttpClientFactory, + request_url: String, + ) -> ModelsTransportFuture<'_> { + let observed_request = Arc::clone(&self.observed_request); + Box::pin(async move { + *observed_request + .lock() + .expect("observed request lock should not be poisoned") = + Some((http_client_factory.outbound_proxy_policy(), request_url)); + Ok(ReqwestTransport::new(build_reqwest_client())) + }) + } + } fn provider_info_with_command_auth() -> ModelProviderInfo { ModelProviderInfo { @@ -267,4 +353,45 @@ mod tests { assert!(!endpoint.has_command_auth()); } + + #[tokio::test] + async fn model_request_uses_request_time_proxy_policy_and_exact_url() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/models")) + .and(query_param("client_version", "0.0.0")) + .respond_with( + ResponseTemplate::new(200).set_body_json(ModelsResponse { models: Vec::new() }), + ) + .expect(1) + .mount(&server) + .await; + + let observed_request = Arc::new(Mutex::new(None)); + let endpoint = OpenAiModelsEndpoint { + provider_info: ModelProviderInfo::create_openai_provider(Some(server.uri())), + auth_manager: None, + transport_builder: Arc::new(RecordingTransportBuilder { + observed_request: Arc::clone(&observed_request), + }), + }; + + endpoint + .list_models( + "0.0.0", + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ) + .await + .expect("models request should succeed"); + + assert_eq!( + *observed_request + .lock() + .expect("observed request lock should not be poisoned"), + Some(( + OutboundProxyPolicy::RespectSystemProxy, + format!("{}/models?client_version=0.0.0", server.uri()), + )) + ); + } } diff --git a/codex-rs/model-provider/src/provider.rs b/codex-rs/model-provider/src/provider.rs index f2fe59e6c913..6bc532ee50e1 100644 --- a/codex-rs/model-provider/src/provider.rs +++ b/codex-rs/model-provider/src/provider.rs @@ -340,6 +340,8 @@ impl ModelProvider for ConfiguredModelProvider { mod tests { use std::num::NonZeroU64; + use codex_http_client::HttpClientFactory; + use codex_http_client::OutboundProxyPolicy; use codex_login::auth::AgentIdentityAuthPolicy; use codex_login::auth::BedrockApiKeyAuth; use codex_model_provider_info::ModelProviderAwsAuthInfo; @@ -656,7 +658,12 @@ mod tests { let manager = provider.models_manager(test_codex_home(), /*config_model_catalog*/ None); - let catalog = manager.raw_model_catalog(RefreshStrategy::Online).await; + let catalog = manager + .raw_model_catalog( + RefreshStrategy::Online, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) + .await; let models = catalog .models .iter() @@ -675,7 +682,10 @@ mod tests { ); let default_model = manager - .list_models(RefreshStrategy::Online) + .list_models( + RefreshStrategy::Online, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) .await .into_iter() .find(|preset| preset.is_default) @@ -706,7 +716,12 @@ mod tests { }), ); - let catalog = manager.raw_model_catalog(RefreshStrategy::Online).await; + let catalog = manager + .raw_model_catalog( + RefreshStrategy::Online, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) + .await; assert_eq!(catalog.models.len(), 1); assert_eq!(catalog.models[0].slug, "gpt-5.5"); @@ -748,7 +763,12 @@ mod tests { let manager = provider.models_manager(test_codex_home(), /*config_model_catalog*/ None); - let catalog = manager.raw_model_catalog(RefreshStrategy::Online).await; + let catalog = manager + .raw_model_catalog( + RefreshStrategy::Online, + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ) + .await; assert!( catalog diff --git a/codex-rs/models-manager/Cargo.toml b/codex-rs/models-manager/Cargo.toml index fb5083c756c3..2655bb2e7fd8 100644 --- a/codex-rs/models-manager/Cargo.toml +++ b/codex-rs/models-manager/Cargo.toml @@ -15,6 +15,7 @@ workspace = true [dependencies] chrono = { workspace = true, features = ["serde"] } codex-collaboration-mode-templates = { workspace = true } +codex-http-client = { workspace = true } codex-login = { workspace = true } codex-otel = { workspace = true } codex-protocol = { workspace = true } diff --git a/codex-rs/models-manager/src/manager.rs b/codex-rs/models-manager/src/manager.rs index 67b939051b6a..3439ceafc5d3 100644 --- a/codex-rs/models-manager/src/manager.rs +++ b/codex-rs/models-manager/src/manager.rs @@ -2,6 +2,7 @@ use super::cache::ModelsCacheManager; use crate::collaboration_mode_presets::builtin_collaboration_mode_presets; use crate::config::ModelsManagerConfig; use crate::model_info; +use codex_http_client::HttpClientFactory; use codex_login::AuthManager; use codex_protocol::auth::AuthMode; use codex_protocol::config_types::CollaborationModeMask; @@ -41,6 +42,7 @@ pub trait ModelsEndpointClient: fmt::Debug + Send + Sync { fn list_models<'a>( &'a self, client_version: &'a str, + http_client_factory: HttpClientFactory, ) -> ModelsEndpointFuture<'a, CoreResult<(Vec, Option)>>; } @@ -83,10 +85,13 @@ pub trait ModelsManager: fmt::Debug + Send + Sync { fn list_models( &self, refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'_, Vec> { Box::pin( async move { - let catalog = self.raw_model_catalog(refresh_strategy).await; + let catalog = self + .raw_model_catalog(refresh_strategy, http_client_factory) + .await; self.build_available_models(catalog.models) } .instrument(tracing::info_span!( @@ -100,6 +105,7 @@ pub trait ModelsManager: fmt::Debug + Send + Sync { fn raw_model_catalog( &self, refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'_, ModelsResponse>; /// Return the current in-memory remote model catalog without refreshing or loading cache state. @@ -152,13 +158,17 @@ pub trait ModelsManager: fmt::Debug + Send + Sync { model: &'a Option, allow_provider_model_fallback: bool, refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'a, String> { Box::pin( async move { if let Some(model) = model.as_ref() { return model.to_string(); } - default_model_from_available(self.list_models(refresh_strategy).await) + default_model_from_available( + self.list_models(refresh_strategy, http_client_factory) + .await, + ) } .instrument(tracing::info_span!( "get_default_model", @@ -188,7 +198,11 @@ pub trait ModelsManager: fmt::Debug + Send + Sync { /// Refresh models if the provided ETag differs from the cached ETag. /// /// Uses `Online` strategy to fetch latest models when ETags differ. - fn refresh_if_new_etag(&self, etag: String) -> ModelsManagerFuture<'_, ()>; + fn refresh_if_new_etag( + &self, + etag: String, + http_client_factory: HttpClientFactory, + ) -> ModelsManagerFuture<'_, ()>; } pub type ModelsManagerFuture<'a, T> = Pin + Send + 'a>>; @@ -247,10 +261,12 @@ impl ModelsManager for OpenAiModelsManager { fn raw_model_catalog( &self, refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'_, ModelsResponse> { Box::pin(OpenAiModelsManager::raw_model_catalog( self, refresh_strategy, + http_client_factory, )) } @@ -270,14 +286,29 @@ impl ModelsManager for OpenAiModelsManager { builtin_collaboration_mode_presets() } - fn refresh_if_new_etag(&self, etag: String) -> ModelsManagerFuture<'_, ()> { - Box::pin(OpenAiModelsManager::refresh_if_new_etag(self, etag)) + fn refresh_if_new_etag( + &self, + etag: String, + http_client_factory: HttpClientFactory, + ) -> ModelsManagerFuture<'_, ()> { + Box::pin(OpenAiModelsManager::refresh_if_new_etag( + self, + etag, + http_client_factory, + )) } } impl OpenAiModelsManager { - async fn raw_model_catalog(&self, refresh_strategy: RefreshStrategy) -> ModelsResponse { - if let Err(err) = self.refresh_available_models(refresh_strategy).await { + async fn raw_model_catalog( + &self, + refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, + ) -> ModelsResponse { + if let Err(err) = self + .refresh_available_models(refresh_strategy, &http_client_factory) + .await + { error!("failed to refresh available models: {err}"); } ModelsResponse { @@ -285,7 +316,7 @@ impl OpenAiModelsManager { } } - async fn refresh_if_new_etag(&self, etag: String) { + async fn refresh_if_new_etag(&self, etag: String, http_client_factory: HttpClientFactory) { let current_etag = self.get_etag().await; if current_etag.clone().is_some() && current_etag.as_deref() == Some(etag.as_str()) { if let Err(err) = self.cache_manager.renew_cache_ttl().await { @@ -293,13 +324,20 @@ impl OpenAiModelsManager { } return; } - if let Err(err) = self.refresh_available_models(RefreshStrategy::Online).await { + if let Err(err) = self + .refresh_available_models(RefreshStrategy::Online, &http_client_factory) + .await + { error!("failed to refresh available models: {err}"); } } /// Refresh available models according to the specified strategy. - async fn refresh_available_models(&self, refresh_strategy: RefreshStrategy) -> CoreResult<()> { + async fn refresh_available_models( + &self, + refresh_strategy: RefreshStrategy, + http_client_factory: &HttpClientFactory, + ) -> CoreResult<()> { if !self.should_refresh_models().await { if matches!( refresh_strategy, @@ -323,18 +361,24 @@ impl OpenAiModelsManager { return Ok(()); } info!("models cache: cache miss, fetching remote models"); - self.fetch_and_update_models().await + self.fetch_and_update_models(http_client_factory).await } RefreshStrategy::Online => { // Always fetch from network - self.fetch_and_update_models().await + self.fetch_and_update_models(http_client_factory).await } } } - async fn fetch_and_update_models(&self) -> CoreResult<()> { + async fn fetch_and_update_models( + &self, + http_client_factory: &HttpClientFactory, + ) -> CoreResult<()> { let client_version = crate::client_version_to_whole(); - let (models, etag) = self.endpoint_client.list_models(&client_version).await?; + let (models, etag) = self + .endpoint_client + .list_models(&client_version, http_client_factory.clone()) + .await?; self.apply_remote_models(models.clone()).await; *self.etag.write().await = etag.clone(); self.cache_manager @@ -416,10 +460,13 @@ impl ModelsManager for StaticModelsManager { model: &'a Option, allow_provider_model_fallback: bool, refresh_strategy: RefreshStrategy, + http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'a, String> { Box::pin( async move { - let available_models = self.list_models(refresh_strategy).await; + let available_models = self + .list_models(refresh_strategy, http_client_factory) + .await; let requested_model = model.as_deref(); if allow_provider_model_fallback { @@ -447,6 +494,7 @@ impl ModelsManager for StaticModelsManager { fn raw_model_catalog( &self, _refresh_strategy: RefreshStrategy, + _http_client_factory: HttpClientFactory, ) -> ModelsManagerFuture<'_, ModelsResponse> { Box::pin(async move { ModelsResponse { @@ -471,7 +519,11 @@ impl ModelsManager for StaticModelsManager { builtin_collaboration_mode_presets() } - fn refresh_if_new_etag(&self, _etag: String) -> ModelsManagerFuture<'_, ()> { + fn refresh_if_new_etag( + &self, + _etag: String, + _http_client_factory: HttpClientFactory, + ) -> ModelsManagerFuture<'_, ()> { Box::pin(async {}) } } diff --git a/codex-rs/models-manager/src/manager_tests.rs b/codex-rs/models-manager/src/manager_tests.rs index f1678fe9f255..e74c7cac4cdf 100644 --- a/codex-rs/models-manager/src/manager_tests.rs +++ b/codex-rs/models-manager/src/manager_tests.rs @@ -1,6 +1,8 @@ use super::*; use crate::ModelsManagerConfig; use chrono::Utc; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; use codex_login::AuthCredentialsStoreMode; use codex_login::AuthKeyringBackendKind; use codex_login::AuthManager; @@ -23,6 +25,9 @@ use tempfile::tempdir; #[path = "model_info_overrides_tests.rs"] mod model_info_overrides_tests; +const DEFAULT_HTTP_CLIENT_FACTORY: HttpClientFactory = + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); + fn remote_model(slug: &str, display: &str, priority: i32) -> ModelInfo { remote_model_with_visibility(slug, display, priority, "list") } @@ -76,6 +81,7 @@ struct TestModelsEndpoint { uses_codex_backend: bool, responses: Mutex>>, fetch_count: AtomicUsize, + observed_proxy_policy: Mutex>, } impl TestModelsEndpoint { @@ -85,6 +91,7 @@ impl TestModelsEndpoint { uses_codex_backend: true, responses: Mutex::new(responses.into()), fetch_count: AtomicUsize::new(0), + observed_proxy_policy: Mutex::new(None), }) } @@ -94,6 +101,7 @@ impl TestModelsEndpoint { uses_codex_backend: false, responses: Mutex::new(responses.into()), fetch_count: AtomicUsize::new(0), + observed_proxy_policy: Mutex::new(None), }) } @@ -101,6 +109,13 @@ impl TestModelsEndpoint { self.fetch_count.load(Ordering::SeqCst) } + fn observed_proxy_policy(&self) -> Option { + *self + .observed_proxy_policy + .lock() + .expect("observed proxy policy lock should not be poisoned") + } + async fn list_models(&self) -> CoreResult<(Vec, Option)> { self.fetch_count.fetch_add(1, Ordering::SeqCst); let models = self @@ -157,8 +172,16 @@ impl ModelsEndpointClient for TestModelsEndpoint { fn list_models<'a>( &'a self, _client_version: &'a str, + http_client_factory: HttpClientFactory, ) -> ModelsEndpointFuture<'a, CoreResult<(Vec, Option)>> { - Box::pin(TestModelsEndpoint::list_models(self)) + Box::pin(async move { + *self + .observed_proxy_policy + .lock() + .expect("observed proxy policy lock should not be poisoned") = + Some(http_client_factory.outbound_proxy_policy()); + TestModelsEndpoint::list_models(self).await + }) } } @@ -241,6 +264,7 @@ async fn static_manager_preserves_supported_requested_model_when_fallback_is_all &requested_model, /*allow_provider_model_fallback*/ true, RefreshStrategy::Offline, + DEFAULT_HTTP_CLIENT_FACTORY, ) .await; @@ -262,6 +286,7 @@ async fn static_manager_falls_back_from_unsupported_requested_model_when_allowed &requested_model, /*allow_provider_model_fallback*/ true, RefreshStrategy::Offline, + DEFAULT_HTTP_CLIENT_FACTORY, ) .await; @@ -284,6 +309,7 @@ async fn static_manager_preserves_unsupported_requested_model_when_fallback_is_d &requested_model, /*allow_provider_model_fallback*/ false, RefreshStrategy::Offline, + DEFAULT_HTTP_CLIENT_FACTORY, ) .await; @@ -300,6 +326,7 @@ async fn static_manager_uses_empty_default_when_fallback_is_allowed_and_catalog_ &requested_model, /*allow_provider_model_fallback*/ true, RefreshStrategy::Offline, + DEFAULT_HTTP_CLIENT_FACTORY, ) .await; @@ -318,6 +345,7 @@ async fn dynamic_manager_preserves_requested_model_when_fallback_is_allowed() { &requested_model, /*allow_provider_model_fallback*/ true, RefreshStrategy::Online, + DEFAULT_HTTP_CLIENT_FACTORY, ) .await; @@ -439,14 +467,17 @@ async fn refresh_available_models_sorts_by_priority() { let endpoint = TestModelsEndpoint::new(vec![remote_models.clone()]); let manager = openai_manager_for_tests(codex_home.path().to_path_buf(), endpoint.clone()); - manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) - .await - .expect("refresh succeeds"); - let cached_remote = manager.get_remote_models().await; - assert_models_contain(&cached_remote, &remote_models); - - let available = manager.list_models(RefreshStrategy::OnlineIfUncached).await; + let available = manager + .list_models( + RefreshStrategy::Online, + HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + ) + .await; + assert_models_contain(&manager.get_remote_models().await, &remote_models); + assert_eq!( + endpoint.observed_proxy_policy(), + Some(OutboundProxyPolicy::RespectSystemProxy) + ); let high_idx = available .iter() .position(|model| model.model == "priority-high") @@ -474,7 +505,10 @@ async fn refresh_available_models_uses_remote_only_catalog_for_chatgpt_auth() { let manager = openai_manager_for_tests(codex_home.path().to_path_buf(), endpoint.clone()); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("refresh succeeds"); @@ -495,7 +529,10 @@ async fn refresh_available_models_uses_cached_remote_only_catalog_for_chatgpt_au openai_manager_for_tests(codex_home.path().to_path_buf(), fetch_endpoint.clone()); fetch_manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("initial refresh succeeds"); @@ -504,7 +541,10 @@ async fn refresh_available_models_uses_cached_remote_only_catalog_for_chatgpt_au openai_manager_for_tests(codex_home.path().to_path_buf(), cache_endpoint.clone()); cache_manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("cached refresh succeeds"); @@ -534,7 +574,10 @@ async fn get_model_info_uses_fallback_for_bundled_models_when_chatgpt_remote_is_ .clone(); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("refresh succeeds"); @@ -554,7 +597,10 @@ async fn refresh_available_models_preserves_bundled_catalog_for_empty_chatgpt_re let expected = load_remote_models_from_file().expect("bundled models should parse"); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("refresh succeeds"); @@ -576,7 +622,10 @@ async fn refresh_available_models_merges_hidden_only_chatgpt_remote_with_bundled expected.push(hidden_remote); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("refresh succeeds"); @@ -596,6 +645,7 @@ async fn refresh_available_models_keeps_merging_for_api_auth() { uses_codex_backend: false, responses: Mutex::new(vec![remote_models.clone()].into()), fetch_count: AtomicUsize::new(0), + observed_proxy_policy: Mutex::new(None), }); let manager = openai_manager_for_tests_with_auth( codex_home.path().to_path_buf(), @@ -608,7 +658,10 @@ async fn refresh_available_models_keeps_merging_for_api_auth() { expected.extend(remote_models); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("refresh succeeds"); @@ -624,14 +677,20 @@ async fn refresh_available_models_uses_cache_when_fresh() { let manager = openai_manager_for_tests(codex_home.path().to_path_buf(), endpoint.clone()); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("first refresh succeeds"); assert_models_contain(&manager.get_remote_models().await, &remote_models); // Second call should read from cache and avoid the network. manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("cached refresh succeeds"); assert_models_contain(&manager.get_remote_models().await, &remote_models); @@ -651,7 +710,10 @@ async fn refresh_available_models_refetches_when_cache_stale() { let manager = openai_manager_for_tests(codex_home.path().to_path_buf(), endpoint.clone()); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("initial refresh succeeds"); @@ -665,7 +727,10 @@ async fn refresh_available_models_refetches_when_cache_stale() { .expect("cache manipulation succeeds"); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("second refresh succeeds"); assert_models_contain(&manager.get_remote_models().await, &updated_models); @@ -685,7 +750,10 @@ async fn refresh_available_models_refetches_when_version_mismatch() { let manager = openai_manager_for_tests(codex_home.path().to_path_buf(), endpoint.clone()); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("initial refresh succeeds"); @@ -699,7 +767,10 @@ async fn refresh_available_models_refetches_when_version_mismatch() { .expect("cache mutation succeeds"); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("second refresh succeeds"); assert_models_contain(&manager.get_remote_models().await, &updated_models); @@ -728,12 +799,18 @@ async fn refresh_available_models_drops_removed_remote_models() { manager.cache_manager.set_ttl(Duration::ZERO); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("initial refresh succeeds"); manager - .refresh_available_models(RefreshStrategy::OnlineIfUncached) + .refresh_available_models( + RefreshStrategy::OnlineIfUncached, + &DEFAULT_HTTP_CLIENT_FACTORY, + ) .await .expect("second refresh succeeds"); @@ -771,7 +848,7 @@ async fn refresh_available_models_skips_network_without_chatgpt_auth() { ); manager - .refresh_available_models(RefreshStrategy::Online) + .refresh_available_models(RefreshStrategy::Online, &DEFAULT_HTTP_CLIENT_FACTORY) .await .expect("refresh should no-op without chatgpt auth"); let cached_remote = manager.get_remote_models().await; @@ -843,6 +920,7 @@ impl ModelsEndpointClient for TestAuthAwareModelsEndpoint { fn list_models<'a>( &'a self, _client_version: &'a str, + _http_client_factory: HttpClientFactory, ) -> ModelsEndpointFuture<'a, CoreResult<(Vec, Option)>> { Box::pin(TestAuthAwareModelsEndpoint::list_models(self)) } @@ -873,7 +951,7 @@ async fn refresh_available_models_skips_network_when_external_api_key_overrides_ ); manager - .refresh_available_models(RefreshStrategy::Online) + .refresh_available_models(RefreshStrategy::Online, &DEFAULT_HTTP_CLIENT_FACTORY) .await .expect("refresh should no-op with API key auth"); let cached_remote = manager.get_remote_models().await; @@ -916,7 +994,7 @@ async fn refresh_available_models_uses_cached_chatgpt_when_external_api_key_is_u ); manager - .refresh_available_models(RefreshStrategy::Online) + .refresh_available_models(RefreshStrategy::Online, &DEFAULT_HTTP_CLIENT_FACTORY) .await .expect("refresh should fall back to cached ChatGPT auth"); @@ -952,7 +1030,7 @@ async fn refresh_available_models_fetches_with_chatgpt_auth_tokens() { ); manager - .refresh_available_models(RefreshStrategy::Online) + .refresh_available_models(RefreshStrategy::Online, &DEFAULT_HTTP_CLIENT_FACTORY) .await .expect("refresh should fetch with ChatGPT auth tokens"); @@ -1006,7 +1084,9 @@ async fn static_manager_reads_latest_auth_mode() { }, ); - let chatgpt_models = manager.list_models(RefreshStrategy::Online).await; + let chatgpt_models = manager + .list_models(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY) + .await; assert_eq!( chatgpt_models .iter() @@ -1019,7 +1099,9 @@ async fn static_manager_reads_latest_auth_mode() { .set_external_auth(Arc::new(TestExternalApiKeyAuth)) .await .expect("external API key auth should resolve"); - let api_models = manager.list_models(RefreshStrategy::Online).await; + let api_models = manager + .list_models(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY) + .await; assert_eq!( api_models