use super::cache::ModelsCacheManager; use crate::collaboration_mode_presets::CollaborationModesConfig; use crate::collaboration_mode_presets::builtin_collaboration_mode_presets; use crate::config::ModelsManagerConfig; use crate::model_info; use codex_api::ModelsClient; use codex_api::RequestTelemetry; use codex_api::ReqwestTransport; use codex_api::TransportError; use codex_api::auth_header_telemetry; use codex_api::map_api_error; use codex_app_server_protocol::AuthMode; use codex_feedback::FeedbackRequestTags; use codex_feedback::emit_feedback_request_tags_with_auth_env; 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_model_provider::AuthorizationHeaderAuthProvider; use codex_model_provider::SharedModelProvider; use codex_model_provider::create_model_provider; use codex_model_provider_info::ModelProviderInfo; use codex_otel::TelemetryAuthMode; use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CoreResult; use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelPreset; use codex_protocol::openai_models::ModelsResponse; use codex_response_debug_context::extract_response_debug_context; use codex_response_debug_context::telemetry_transport_error_message; use http::HeaderMap; use std::fmt; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; use tokio::sync::TryLockError; use tokio::time::timeout; use tracing::error; use tracing::info; use tracing::instrument; const MODEL_CACHE_FILE: &str = "models_cache.json"; const DEFAULT_MODEL_CACHE_TTL: Duration = Duration::from_secs(300); const MODELS_REFRESH_TIMEOUT: Duration = Duration::from_secs(5); const MODELS_ENDPOINT: &str = "/models"; #[derive(Clone)] struct ModelsRequestTelemetry { auth_mode: Option, auth_header_attached: bool, auth_header_name: Option<&'static str>, auth_env: AuthEnvTelemetry, } impl RequestTelemetry for ModelsRequestTelemetry { fn on_request( &self, attempt: u64, status: Option, error: Option<&TransportError>, duration: Duration, ) { let success = status.is_some_and(|code| code.is_success()) && error.is_none(); let error_message = error.map(telemetry_transport_error_message); let response_debug = error .map(extract_response_debug_context) .unwrap_or_default(); let status = status.map(|status| status.as_u16()); tracing::event!( target: "codex_otel.log_only", tracing::Level::INFO, event.name = "codex.api_request", duration_ms = %duration.as_millis(), http.response.status_code = status, success = success, error.message = error_message.as_deref(), attempt = attempt, endpoint = MODELS_ENDPOINT, auth.header_attached = self.auth_header_attached, auth.header_name = self.auth_header_name, auth.env_openai_api_key_present = self.auth_env.openai_api_key_env_present, auth.env_codex_api_key_present = self.auth_env.codex_api_key_env_present, auth.env_codex_api_key_enabled = self.auth_env.codex_api_key_env_enabled, auth.env_provider_key_name = self.auth_env.provider_env_key_name.as_deref(), auth.env_provider_key_present = self.auth_env.provider_env_key_present, auth.env_refresh_token_url_override_present = self.auth_env.refresh_token_url_override_present, auth.request_id = response_debug.request_id.as_deref(), auth.cf_ray = response_debug.cf_ray.as_deref(), auth.error = response_debug.auth_error.as_deref(), auth.error_code = response_debug.auth_error_code.as_deref(), auth.mode = self.auth_mode.as_deref(), ); tracing::event!( target: "codex_otel.trace_safe", tracing::Level::INFO, event.name = "codex.api_request", duration_ms = %duration.as_millis(), http.response.status_code = status, success = success, error.message = error_message.as_deref(), attempt = attempt, endpoint = MODELS_ENDPOINT, auth.header_attached = self.auth_header_attached, auth.header_name = self.auth_header_name, auth.env_openai_api_key_present = self.auth_env.openai_api_key_env_present, auth.env_codex_api_key_present = self.auth_env.codex_api_key_env_present, auth.env_codex_api_key_enabled = self.auth_env.codex_api_key_env_enabled, auth.env_provider_key_name = self.auth_env.provider_env_key_name.as_deref(), auth.env_provider_key_present = self.auth_env.provider_env_key_present, auth.env_refresh_token_url_override_present = self.auth_env.refresh_token_url_override_present, auth.request_id = response_debug.request_id.as_deref(), auth.cf_ray = response_debug.cf_ray.as_deref(), auth.error = response_debug.auth_error.as_deref(), auth.error_code = response_debug.auth_error_code.as_deref(), auth.mode = self.auth_mode.as_deref(), ); emit_feedback_request_tags_with_auth_env( &FeedbackRequestTags { endpoint: MODELS_ENDPOINT, auth_header_attached: self.auth_header_attached, auth_header_name: self.auth_header_name, auth_mode: self.auth_mode.as_deref(), auth_retry_after_unauthorized: None, auth_recovery_mode: None, auth_recovery_phase: None, auth_connection_reused: None, auth_request_id: response_debug.request_id.as_deref(), auth_cf_ray: response_debug.cf_ray.as_deref(), auth_error: response_debug.auth_error.as_deref(), auth_error_code: response_debug.auth_error_code.as_deref(), auth_recovery_followup_success: None, auth_recovery_followup_status: None, }, &self.auth_env, ); } } /// Strategy for refreshing available models. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RefreshStrategy { /// Always fetch from the network, ignoring cache. Online, /// Only use cached data, never fetch from the network. Offline, /// Use cache if available and fresh, otherwise fetch from the network. OnlineIfUncached, } impl RefreshStrategy { const fn as_str(self) -> &'static str { match self { Self::Online => "online", Self::Offline => "offline", Self::OnlineIfUncached => "online_if_uncached", } } } impl fmt::Display for RefreshStrategy { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str(self.as_str()) } } /// How the manager's base catalog is sourced for the lifetime of the process. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum CatalogMode { /// Start from bundled `models.json` and allow cache/network refresh updates. Default, /// Use a caller-provided catalog as authoritative and do not mutate it via refresh. Custom, } /// Coordinates remote model discovery plus cached metadata on disk. #[derive(Debug)] pub struct ModelsManager { remote_models: RwLock>, catalog_mode: CatalogMode, collaboration_modes_config: CollaborationModesConfig, etag: RwLock>, cache_manager: ModelsCacheManager, provider: SharedModelProvider, } impl ModelsManager { /// Construct a manager scoped to the provided `AuthManager`. /// /// Uses `codex_home` to store cached model metadata and initializes with bundled catalog /// When `model_catalog` is provided, it becomes the authoritative remote model list and /// background refreshes from `/models` are disabled. pub fn new( codex_home: PathBuf, auth_manager: Arc, model_catalog: Option, collaboration_modes_config: CollaborationModesConfig, ) -> Self { Self::new_with_provider( codex_home, auth_manager, model_catalog, collaboration_modes_config, ModelProviderInfo::create_openai_provider(/*base_url*/ None), ) } /// Construct a manager with an explicit provider used for remote model refreshes. // TODO(celia-oai): Revisit this ownership direction: the model provider should likely // own or return the models manager instead of requiring the manager to construct and use // a provider from provider info. pub fn new_with_provider( codex_home: PathBuf, auth_manager: Arc, model_catalog: Option, collaboration_modes_config: CollaborationModesConfig, provider_info: ModelProviderInfo, ) -> Self { let model_provider = create_model_provider(provider_info, Some(auth_manager)); let cache_path = codex_home.join(MODEL_CACHE_FILE); let cache_manager = ModelsCacheManager::new(cache_path, DEFAULT_MODEL_CACHE_TTL); let catalog_mode = if model_catalog.is_some() { CatalogMode::Custom } else { CatalogMode::Default }; let remote_models = model_catalog .map(|catalog| catalog.models) .unwrap_or_else(|| Self::load_remote_models_from_file().unwrap_or_default()); Self { remote_models: RwLock::new(remote_models), catalog_mode, collaboration_modes_config, etag: RwLock::new(None), cache_manager, provider: model_provider, } } /// List all available models, refreshing according to the specified strategy. /// /// Returns model presets sorted by priority and filtered by auth mode and visibility. #[instrument( level = "info", skip(self), fields(refresh_strategy = %refresh_strategy) )] pub async fn list_models(&self, refresh_strategy: RefreshStrategy) -> Vec { if let Err(err) = self.refresh_available_models(refresh_strategy).await { error!("failed to refresh available models: {err}"); } let remote_models = self.get_remote_models().await; self.build_available_models(remote_models) } /// Return the active raw model catalog, refreshing according to the specified strategy. pub async fn raw_model_catalog(&self, refresh_strategy: RefreshStrategy) -> ModelsResponse { if let Err(err) = self.refresh_available_models(refresh_strategy).await { error!("failed to refresh available models: {err}"); } ModelsResponse { models: self.get_remote_models().await, } } /// List collaboration mode presets. /// /// Returns a static set of presets seeded with the configured model. pub fn list_collaboration_modes(&self) -> Vec { self.list_collaboration_modes_for_config(self.collaboration_modes_config) } pub fn list_collaboration_modes_for_config( &self, collaboration_modes_config: CollaborationModesConfig, ) -> Vec { builtin_collaboration_mode_presets(collaboration_modes_config) } /// Attempt to list models without blocking, using the current cached state. /// /// Returns an error if the internal lock cannot be acquired. pub fn try_list_models(&self) -> Result, TryLockError> { let remote_models = self.try_get_remote_models()?; Ok(self.build_available_models(remote_models)) } // todo(aibrahim): should be visible to core only and sent on session_configured event /// Get the model identifier to use, refreshing according to the specified strategy. /// /// If `model` is provided, returns it directly. Otherwise selects the default based on /// auth mode and available models. #[instrument( level = "info", skip(self, model), fields( model.provided = model.is_some(), refresh_strategy = %refresh_strategy ) )] pub async fn get_default_model( &self, model: &Option, refresh_strategy: RefreshStrategy, ) -> String { if let Some(model) = model.as_ref() { return model.to_string(); } if let Err(err) = self.refresh_available_models(refresh_strategy).await { error!("failed to refresh available models: {err}"); } let remote_models = self.get_remote_models().await; let available = self.build_available_models(remote_models); available .iter() .find(|model| model.is_default) .or_else(|| available.first()) .map(|model| model.model.clone()) .unwrap_or_default() } // todo(aibrahim): look if we can tighten it to pub(crate) /// Look up model metadata, applying remote overrides and config adjustments. #[instrument(level = "info", skip(self, config), fields(model = model))] pub async fn get_model_info(&self, model: &str, config: &ModelsManagerConfig) -> ModelInfo { let remote_models = self.get_remote_models().await; Self::construct_model_info_from_candidates(model, &remote_models, config) } fn find_model_by_longest_prefix(model: &str, candidates: &[ModelInfo]) -> Option { let mut best: Option = None; for candidate in candidates { if !model.starts_with(&candidate.slug) { continue; } let is_better_match = if let Some(current) = best.as_ref() { candidate.slug.len() > current.slug.len() } else { true }; if is_better_match { best = Some(candidate.clone()); } } best } /// Retry metadata lookup for a single namespaced slug like `namespace/model-name`. /// /// This only strips one leading namespace segment and only when the namespace is ASCII /// alphanumeric/underscore (`\\w+`) to avoid broadly matching arbitrary aliases. fn find_model_by_namespaced_suffix(model: &str, candidates: &[ModelInfo]) -> Option { let (namespace, suffix) = model.split_once('/')?; if suffix.contains('/') { return None; } if !namespace .chars() .all(|c| c.is_ascii_alphanumeric() || c == '_') { return None; } Self::find_model_by_longest_prefix(suffix, candidates) } fn construct_model_info_from_candidates( model: &str, candidates: &[ModelInfo], config: &ModelsManagerConfig, ) -> ModelInfo { // First use the normal longest-prefix match. If that misses, allow a narrowly scoped // retry for namespaced slugs like `custom/gpt-5.3-codex`. let remote = Self::find_model_by_longest_prefix(model, candidates) .or_else(|| Self::find_model_by_namespaced_suffix(model, candidates)); let model_info = if let Some(remote) = remote { ModelInfo { slug: model.to_string(), used_fallback_model_metadata: false, ..remote } } else { model_info::model_info_from_slug(model) }; model_info::with_config_overrides(model_info, config) } /// Refresh models if the provided ETag differs from the cached ETag. /// /// Uses `Online` strategy to fetch latest models when ETags differ. pub async fn refresh_if_new_etag(&self, etag: String) { 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 { error!("failed to renew cache TTL: {err}"); } return; } if let Err(err) = self.refresh_available_models(RefreshStrategy::Online).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<()> { // don't override the custom model catalog if one was provided by the user if matches!(self.catalog_mode, CatalogMode::Custom) { return Ok(()); } let auth_mode = self .provider .auth_manager() .and_then(|auth_manager| auth_manager.auth_mode()); if auth_mode != Some(AuthMode::Chatgpt) && !self.provider.info().has_command_auth() { if matches!( refresh_strategy, RefreshStrategy::Offline | RefreshStrategy::OnlineIfUncached ) { self.try_load_cache().await; } return Ok(()); } match refresh_strategy { RefreshStrategy::Offline => { // Only try to load from cache, never fetch self.try_load_cache().await; Ok(()) } RefreshStrategy::OnlineIfUncached => { // Try cache first, fall back to online if unavailable if self.try_load_cache().await { info!("models cache: using cached models for OnlineIfUncached"); return Ok(()); } info!("models cache: cache miss, fetching remote models"); self.fetch_and_update_models().await } RefreshStrategy::Online => { // Always fetch from network self.fetch_and_update_models().await } } } async fn fetch_and_update_models(&self) -> CoreResult<()> { let _timer = codex_otel::start_global_timer("codex.remote_models.fetch_update.duration_ms", &[]); let auth_manager = self.provider.auth_manager(); let codex_api_key_env_enabled = auth_manager .as_ref() .is_some_and(|auth_manager| auth_manager.codex_api_key_env_enabled()); let auth = self.provider.auth().await; let auth_mode = auth.as_ref().map(CodexAuth::auth_mode); let api_provider = self.provider.api_provider().await?; let mut api_auth = self.provider.api_auth().await?; if let Some(auth_manager) = auth_manager.as_ref() && let Some(auth) = auth.as_ref().filter(|auth| auth.is_chatgpt_auth()) && provider_uses_codex_login_auth(self.provider.info()) && let Some(authorization_header_value) = auth_manager .chatgpt_authorization_header_for_auth(auth) .await { let mut auth_provider = AuthorizationHeaderAuthProvider::new( Some(authorization_header_value), auth.get_account_id(), ); if auth.is_fedramp_account() { auth_provider = auth_provider.with_fedramp_routing_header(); } api_auth = Arc::new(auth_provider); } let auth_env = collect_auth_env_telemetry(self.provider.info(), codex_api_key_env_enabled); let transport = ReqwestTransport::new(build_reqwest_client()); let auth_telemetry = auth_header_telemetry(api_auth.as_ref()); let request_telemetry: Arc = Arc::new(ModelsRequestTelemetry { auth_mode: auth_mode.map(|mode| TelemetryAuthMode::from(mode).to_string()), auth_header_attached: auth_telemetry.attached, auth_header_name: auth_telemetry.name, auth_env, }); let client = ModelsClient::new(transport, api_provider, api_auth) .with_telemetry(Some(request_telemetry)); let client_version = crate::client_version_to_whole(); let (models, etag) = timeout( MODELS_REFRESH_TIMEOUT, client.list_models(&client_version, HeaderMap::new()), ) .await .map_err(|_| CodexErr::Timeout)? .map_err(map_api_error)?; self.apply_remote_models(models.clone()).await; *self.etag.write().await = etag.clone(); self.cache_manager .persist_cache(&models, etag, client_version) .await; Ok(()) } async fn get_etag(&self) -> Option { self.etag.read().await.clone() } /// Replace the cached remote models and rebuild the derived presets list. async fn apply_remote_models(&self, models: Vec) { let mut existing_models = Self::load_remote_models_from_file().unwrap_or_default(); for model in models { if let Some(existing_index) = existing_models .iter() .position(|existing| existing.slug == model.slug) { existing_models[existing_index] = model; } else { existing_models.push(model); } } *self.remote_models.write().await = existing_models; } fn load_remote_models_from_file() -> Result, std::io::Error> { Ok(crate::bundled_models_response()?.models) } /// Attempt to satisfy the refresh from the cache when it matches the provider and TTL. async fn try_load_cache(&self) -> bool { let _timer = codex_otel::start_global_timer("codex.remote_models.load_cache.duration_ms", &[]); let client_version = crate::client_version_to_whole(); info!(client_version, "models cache: evaluating cache eligibility"); let cache = match self.cache_manager.load_fresh(&client_version).await { Some(cache) => cache, None => { info!("models cache: no usable cache entry"); return false; } }; let models = cache.models.clone(); *self.etag.write().await = cache.etag.clone(); self.apply_remote_models(models.clone()).await; info!( models_count = models.len(), etag = ?cache.etag, "models cache: cache entry applied" ); true } /// Build picker-ready presets from the active catalog snapshot. fn build_available_models(&self, mut remote_models: Vec) -> Vec { remote_models.sort_by(|a, b| a.priority.cmp(&b.priority)); let mut presets: Vec = remote_models.into_iter().map(Into::into).collect(); let auth_mode = self .provider .auth_manager() .and_then(|auth_manager| auth_manager.auth_mode()); let chatgpt_mode = matches!(auth_mode, Some(AuthMode::Chatgpt)); presets = ModelPreset::filter_by_auth(presets, chatgpt_mode); ModelPreset::mark_default_by_picker_visibility(&mut presets); presets } async fn get_remote_models(&self) -> Vec { self.remote_models.read().await.clone() } fn try_get_remote_models(&self) -> Result, TryLockError> { Ok(self.remote_models.try_read()?.clone()) } /// Construct a manager with a specific provider for testing. pub fn with_provider_for_tests( codex_home: PathBuf, auth_manager: Arc, provider: ModelProviderInfo, ) -> Self { Self::new_with_provider( codex_home, auth_manager, /*model_catalog*/ None, CollaborationModesConfig::default(), provider, ) } /// Get model identifier without consulting remote state or cache. pub fn get_model_offline_for_tests(model: Option<&str>) -> String { if let Some(model) = model { return model.to_string(); } let mut models = Self::load_remote_models_from_file().unwrap_or_default(); models.sort_by(|a, b| a.priority.cmp(&b.priority)); let presets: Vec = models.into_iter().map(Into::into).collect(); presets .iter() .find(|preset| preset.show_in_picker) .or_else(|| presets.first()) .map(|preset| preset.model.clone()) .unwrap_or_default() } /// Build `ModelInfo` without consulting remote state or cache. pub fn construct_model_info_offline_for_tests( model: &str, config: &ModelsManagerConfig, ) -> ModelInfo { let candidates: &[ModelInfo] = if let Some(model_catalog) = config.model_catalog.as_ref() { &model_catalog.models } else { &[] }; Self::construct_model_info_from_candidates(model, candidates, config) } } fn provider_uses_codex_login_auth(provider: &ModelProviderInfo) -> bool { provider.env_key.is_none() && provider.experimental_bearer_token.is_none() } #[cfg(test)] #[path = "manager_tests.rs"] mod tests;