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

model-provider: route model discovery through HTTP client factory (#31361)

Browse Source
This commit is contained in:
Michael Bolin
2026-07-09 02:32:10 +00:00
committed by GitHub
parent 555aa79d5a
commit 5892c7b69d
36 changed files with 658 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -1998,6 +1998,7 @@ dependencies = [
"codex-guardian",
"codex-home",
"codex-hooks",
"codex-http-client",
"codex-image-generation-extension",
"codex-login",
"codex-mcp",
@@ -3591,6 +3592,7 @@ version = "0.0.0"
dependencies = [
"chrono",
"codex-collaboration-mode-templates",
"codex-http-client",
"codex-login",
"codex-otel",
"codex-protocol",
codex-rs/app-server/Cargo.toml
+1
View File
@@ -51,6 +51,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 }
codex-rs/app-server/src/message_processor.rs
+2 -1
View File
@@ -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());
codex-rs/app-server/src/models.rs
+3 -1
View File
@@ -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<ThreadManager>,
include_hidden: bool,
http_client_factory: HttpClientFactory,
) -> Vec<Model> {
thread_manager
.list_models(RefreshStrategy::OnlineIfUncached)
.list_models(RefreshStrategy::OnlineIfUncached, http_client_factory)
.await
.into_iter()
.filter(|preset| include_hidden || preset.show_in_picker)
codex-rs/app-server/src/models_refresh_worker.rs
+10 -3
View File
@@ -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! {
codex-rs/app-server/src/models_refresh_worker_tests.rs
+8 -1
View File
@@ -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<ModelInfo>, Option<String>)>> {
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);
codex-rs/app-server/src/request_processors/catalog_processor.rs
+14 -4
View File
@@ -157,9 +157,13 @@ impl CatalogRequestProcessor {
&self,
params: ModelListParams,
) -> Result<Option<ClientResponsePayload>, 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<ThreadManager>,
http_client_factory: codex_http_client::HttpClientFactory,
params: ModelListParams,
) -> Result<ModelListResponse, JSONRPCErrorError> {
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 {
codex-rs/app-server/src/request_processors/external_agent_session_import.rs
+1
View File
@@ -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
codex-rs/cli/src/main.rs
+4 -1
View File
@@ -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
};
codex-rs/codex-api/src/endpoint/models.rs
+21 -21
View File
@@ -37,9 +37,15 @@ impl<T: HttpTransport> ModelsClient<T> {
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<ModelInfo>, Option<String>), ApiError> {
let resp = self
@@ -49,8 +55,8 @@ impl<T: HttpTransport> ModelsClient<T> {
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::<CapturingTransport>::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::<CapturingTransport>::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::<CapturingTransport>::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");
codex-rs/codex-api/tests/models_integration.rs
+4 -2
View File
@@ -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::<ReqwestTransport>::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");
codex-rs/core/src/guardian/review.rs
+4 -1
View File
@@ -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| {
codex-rs/core/src/session/mod.rs
+4 -1
View File
@@ -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
codex-rs/core/src/session/review.rs
+4 -1
View File
@@ -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()),
codex-rs/core/src/session/turn.rs
+4 -1
View File
@@ -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,
codex-rs/core/src/session/turn_context.rs
+4 -1
View File
@@ -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 {
codex-rs/core/src/test_support.rs
+6
View File
@@ -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)
}
codex-rs/core/src/thread_manager.rs
+6 -2
View File
@@ -542,10 +542,14 @@ impl ThreadManager {
self.state.models_manager.clone()
}
pub async fn list_models(&self, refresh_strategy: RefreshStrategy) -> Vec<ModelPreset> {
pub async fn list_models(
&self,
refresh_strategy: RefreshStrategy,
http_client_factory: codex_http_client::HttpClientFactory,
) -> Vec<ModelPreset> {
self.state
.models_manager
.list_models(refresh_strategy)
.list_models(refresh_strategy, http_client_factory)
.await
}
codex-rs/core/src/thread_manager_tests.rs
+6 -1
View File
@@ -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);
}
codex-rs/core/src/tools/handlers/multi_agents_common.rs
+1 -1
View File
@@ -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
codex-rs/core/tests/suite/auto_review.rs
+4 -1
View File
@@ -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())
codex-rs/core/tests/suite/model_runtime_selectors.rs
+4 -1
View File
@@ -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)
codex-rs/core/tests/suite/model_switching.rs
+22 -5
View File
@@ -544,7 +544,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();
@@ -650,7 +653,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
@@ -764,7 +770,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
@@ -882,7 +891,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
@@ -1042,7 +1054,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()
codex-rs/core/tests/suite/models_cache_ttl.rs
+20 -5
View File
@@ -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!(
codex-rs/core/tests/suite/personality.rs
+6 -1
View File
@@ -792,7 +792,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;
}
codex-rs/core/tests/suite/remote_models.rs
+44 -7
View File
@@ -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;
codex-rs/core/tests/suite/rmcp_client.rs
+4 -1
View File
@@ -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);
codex-rs/core/tests/suite/spawn_agent_description.rs
+6 -1
View File
@@ -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;
}
codex-rs/http-client/src/outbound_proxy.rs
+35 -5
View File
@@ -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<HashMap<String, CachedSystemProxyDecision>>,
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<Mutex<HashMap<String, CachedSystemProxyDecision>>> =
OnceLock::new();
#[cfg(test)]
fn cached_system_proxy_decision(request_url: &str) -> Option<SystemProxyDecision> {
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<String, CachedSystemProxyDecision>,
cache_key: &str,
now: Instant,
) -> Option<SystemProxyDecision> {
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() {
codex-rs/http-client/src/outbound_proxy_tests.rs
+39
View File
@@ -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<String, String>,
@@ -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();
codex-rs/login/src/auth/default_client.rs
+22
View File
@@ -53,6 +53,8 @@ pub struct Originator {
static ORIGINATOR: LazyLock<RwLock<Option<Originator>>> = LazyLock::new(|| RwLock::new(None));
static REQUIREMENTS_RESIDENCY: LazyLock<RwLock<Option<ResidencyRequirement>>> =
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<reqwest::Client> {
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() {
codex-rs/model-provider/src/models_endpoint.rs
+138 -11
View File
@@ -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<Arc<AuthManager>>,
transport_builder: Arc<dyn ModelsTransportBuilder>,
}
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<ModelInfo>, Option<String>)> {
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::<ReqwestTransport>::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<ModelInfo>, Option<String>)>> {
Box::pin(OpenAiModelsEndpoint::list_models(self, client_version))
Box::pin(OpenAiModelsEndpoint::list_models(
self,
client_version,
http_client_factory,
))
}
}
type ModelsTransportFuture<'a> =
Pin<Box<dyn Future<Output = std::io::Result<ReqwestTransport>> + 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<Mutex<Option<(OutboundProxyPolicy, String)>>>,
}
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()),
))
);
}
}
codex-rs/model-provider/src/provider.rs
+24 -4
View File
@@ -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
codex-rs/models-manager/Cargo.toml
+1
View File
@@ -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 }
codex-rs/models-manager/src/manager.rs
+68 -16
View File
@@ -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<ModelInfo>, Option<String>)>>;
}
@@ -83,10 +85,13 @@ pub trait ModelsManager: fmt::Debug + Send + Sync {
fn list_models(
&self,
refresh_strategy: RefreshStrategy,
http_client_factory: HttpClientFactory,
) -> ModelsManagerFuture<'_, Vec<ModelPreset>> {
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<String>,
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<Box<dyn Future<Output = T> + 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<String>,
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 {})
}
}
codex-rs/models-manager/src/manager_tests.rs
+112 -30
View File
@@ -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<VecDeque<Vec<ModelInfo>>>,
fetch_count: AtomicUsize,
observed_proxy_policy: Mutex<Option<OutboundProxyPolicy>>,
}
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<OutboundProxyPolicy> {
*self
.observed_proxy_policy
.lock()
.expect("observed proxy policy lock should not be poisoned")
}
async fn list_models(&self) -> CoreResult<(Vec<ModelInfo>, Option<String>)> {
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<ModelInfo>, Option<String>)>> {
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<ModelInfo>, Option<String>)>> {
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