diff --git a/codex-rs/app-server/src/config_manager.rs b/codex-rs/app-server/src/config_manager.rs index d5f2c2b89a..d2c2db7481 100644 --- a/codex-rs/app-server/src/config_manager.rs +++ b/codex-rs/app-server/src/config_manager.rs @@ -56,6 +56,12 @@ pub(crate) struct ConfigManager { network_policy_snapshot: Arc>>>, } +/// Startup configuration, including whether loading had to fall back to defaults. +pub(crate) struct StartupConfig { + pub(crate) config: Config, + pub(crate) used_defaults: bool, +} + /// Configuration and policy must finish loading against the same account and cloud snapshot. pub(crate) struct ApplicationPolicyLoad { cloud_config: CloudConfigBundleLoader, @@ -336,9 +342,12 @@ impl ConfigManager { pub(crate) async fn load_startup_config( &self, fallback_cwd: Option, - ) -> std::io::Result { + ) -> std::io::Result { match self.load_latest_config(fallback_cwd).await { - Ok(config) => Ok(config), + Ok(config) => Ok(StartupConfig { + config, + used_defaults: false, + }), Err(error) if self.strict_config || crate::is_unsupported_untrusted_approval_policy_error(&error) => @@ -347,7 +356,10 @@ impl ConfigManager { } Err(error) => { warn!(%error, "configuration is unavailable; using default settings"); - self.load_default_config().await + Ok(StartupConfig { + config: self.load_default_config().await?, + used_defaults: true, + }) } } } diff --git a/codex-rs/app-server/src/config_manager_provider_tests.rs b/codex-rs/app-server/src/config_manager_provider_tests.rs index 20fbfb4989..86ee457514 100644 --- a/codex-rs/app-server/src/config_manager_provider_tests.rs +++ b/codex-rs/app-server/src/config_manager_provider_tests.rs @@ -210,3 +210,26 @@ async fn provider_requirement_load_errors_reject_input() -> Result<()> { } Ok(()) } + +/// A managed startup selection must disappear when its policy is removed. +#[tokio::test] +async fn catalog_reload_uses_latest_model_provider() -> Result<()> { + let home = tempdir()?; + let manager = ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + LoaderOverrides::without_managed_config_for_tests(), + CloudConfigBundleFixture::loader_with_enterprise_requirement( + "model_provider = 'gateway'\n[model_providers.gateway]\nname = 'Gateway'\nbase_url = 'https://example.test'", + ), + ); + let startup = manager.load_non_project_config().await?; + assert_eq!(startup.model_provider_id, "gateway"); + manager.clear_cloud_config_bundle_loader(); + let refreshed = manager + .load_startup_config(/*fallback_cwd*/ None) + .await? + .config; + assert_eq!(refreshed.model_provider_id, "openai"); + Ok(()) +} diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 9673442d38..64d37f14df 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -435,6 +435,9 @@ async fn start_uninitialized(mut args: InProcessStartArgs) -> IoResult IoResult IoResult> { let bootstrap_config = config_manager .load_startup_config(Some(config.cwd.to_path_buf())) - .await?; + .await? + .config; let caller_auth = config.auth_config(); let bootstrap_auth = codex_login::AuthConfig { chatgpt_base_url: Some(bootstrap_config.chatgpt_base_url.clone()), @@ -132,7 +133,8 @@ pub(super) async fn configure( ); let policy_config = config_manager .load_startup_config(Some(config.cwd.to_path_buf())) - .await?; + .await? + .config; config.validate_windows_mxc_requirement_from(&policy_config)?; let auth_config = codex_login::AuthConfig { chatgpt_base_url: Some(policy_config.chatgpt_base_url.clone()), diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 70e8519ab8..9ef1b73a82 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -546,7 +546,8 @@ pub async fn run_main_with_transport_options( ); let bootstrap_config = config_manager .load_startup_config(startup_cwd.clone()) - .await?; + .await? + .config; let bootstrap_auth = AuthManager::shared_from_config(&bootstrap_config, /*enable_codex_api_key_env*/ false) .await @@ -558,7 +559,7 @@ pub async fn run_main_with_transport_options( ); let mut config_warnings = Vec::new(); let mut plugin_startup_config = PluginStartupConfig::Current; - let config = match config_manager.load_latest_config(startup_cwd).await { + let config = match config_manager.load_latest_config(startup_cwd.clone()).await { Ok(config) => config, Err(err) if is_unsupported_untrusted_approval_policy_error(&err) => { return Err(err); @@ -988,6 +989,7 @@ pub async fn run_main_with_transport_options( analytics_events_client, arg0_paths, config: Arc::new(config), + startup_cwd, config_manager, environment_manager, feedback: feedback.clone(), diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index c9858cd9d5..767c600e07 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -252,6 +252,8 @@ pub(crate) struct MessageProcessorArgs { pub(crate) analytics_events_client: AnalyticsEventsClient, pub(crate) arg0_paths: Arg0DispatchPaths, pub(crate) config: Arc, + /// Original project context, retained even when startup falls back to defaults. + pub(crate) startup_cwd: Option, pub(crate) config_manager: ConfigManager, pub(crate) environment_manager: Arc, pub(crate) feedback: CodexFeedback, @@ -278,6 +280,7 @@ impl MessageProcessor { analytics_events_client, arg0_paths, config, + startup_cwd, config_manager, environment_manager, feedback, @@ -380,8 +383,10 @@ impl MessageProcessor { }); let model_catalog = Arc::new(crate::model_catalog::ModelCatalog::new( config_manager.clone(), + startup_cwd, Arc::clone(&config), thread_manager.get_models_manager(), + Arc::clone(&auth_manager), )); let models_refresh_worker = crate::models_refresh_worker::spawn(&model_catalog); let turn_cost_worker = @@ -405,6 +410,7 @@ impl MessageProcessor { outgoing.clone(), config_manager.clone(), thread_manager.clone(), + Arc::clone(&model_catalog), analytics_events_client.clone(), ); let on_effective_plugins_changed = @@ -421,6 +427,7 @@ impl MessageProcessor { outgoing.clone(), Arc::clone(&config), config_manager.clone(), + Arc::clone(&model_catalog), ); let apps_processor = AppsRequestProcessor::new( auth_manager.clone(), diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index ca4aeeda37..f9b1c8cea6 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -264,6 +264,7 @@ pub(super) async fn build_test_processor( outgoing, analytics_events_client, arg0_paths: Arg0DispatchPaths::default(), + startup_cwd: Some(config.cwd.to_path_buf()), config, config_manager, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), diff --git a/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs b/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs index db478a9960..e1d4c4a4da 100644 --- a/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs +++ b/codex-rs/app-server/src/message_processor_workspace_routing_tests.rs @@ -67,6 +67,7 @@ async fn background_routing_call_sites_skip_config(api_key: Option<&str>) -> any outgoing, analytics_events_client, arg0_paths: Arg0DispatchPaths::default(), + startup_cwd: Some(config.cwd.to_path_buf()), config, config_manager, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), diff --git a/codex-rs/app-server/src/model_catalog.rs b/codex-rs/app-server/src/model_catalog.rs index 54d7ff8501..ff9704fa30 100644 --- a/codex-rs/app-server/src/model_catalog.rs +++ b/codex-rs/app-server/src/model_catalog.rs @@ -1,17 +1,29 @@ -//! Gates access to the retained startup model catalog on current managed provider requirements. +//! Loads the model catalog for the current effective provider configuration. +use std::path::PathBuf; use std::sync::Arc; use codex_core::config::Config; +use codex_features::Feature; +use codex_login::AuthManager; use codex_models_manager::manager::RefreshStrategy; use codex_models_manager::manager::SharedModelsManager; use codex_protocol::openai_models::ModelPreset; +use tokio::sync::Mutex; use crate::config_manager::ConfigManager; +use crate::config_manager::StartupConfig; -/// Checks the retained catalog route before model/list and background refreshes. +/// Refreshes catalog routing without changing the routes retained by existing threads. pub(crate) struct ModelCatalog { config_manager: ConfigManager, + auth_manager: Arc, + startup_cwd: Option, + state: Mutex, +} + +struct CatalogState { + needs_reload: bool, config: Arc, models_manager: SharedModelsManager, } @@ -19,28 +31,92 @@ pub(crate) struct ModelCatalog { impl ModelCatalog { pub(crate) fn new( config_manager: ConfigManager, + startup_cwd: Option, config: Arc, models_manager: SharedModelsManager, + auth_manager: Arc, ) -> Self { Self { config_manager, - config, - models_manager, + auth_manager, + startup_cwd, + state: Mutex::new(CatalogState { + needs_reload: false, + config, + models_manager, + }), } } + /// Marks catalog configuration stale after an account transition. + pub(crate) async fn invalidate(&self) { + self.state.lock().await.needs_reload = true; + } + + pub(crate) async fn set_api_key_model_discovery_enabled(&self, enabled: bool) { + self.state + .lock() + .await + .models_manager + .set_api_key_model_discovery_enabled(enabled); + } + + #[expect( + clippy::await_holding_invalid_type, + reason = "catalog reloads are serialized with account invalidation; model fetching holds no lock" + )] + pub(crate) async fn current(&self) -> std::io::Result<(Arc, SharedModelsManager)> { + let mut state = self.state.lock().await; + if state.needs_reload { + // Keep the large config-loading future off callers' stacks. + let config = Box::pin( + self.config_manager + .load_startup_config(self.startup_cwd.clone()), + ) + .await?; + let StartupConfig { + config, + used_defaults, + } = config; + if config.model_provider_id != state.config.model_provider_id + || config.model_provider != state.config.model_provider + || config.model_catalog != state.config.model_catalog + { + state.models_manager = + codex_core::build_models_manager(&config, Arc::clone(&self.auth_manager)); + } + state.models_manager.set_api_key_model_discovery_enabled( + config.features.enabled(Feature::ApiKeyModelDiscovery), + ); + state.config = Arc::new(config); + // Try again on the next access so a temporary fallback cannot pin the provider. + state.needs_reload = used_defaults; + } + Ok((Arc::clone(&state.config), Arc::clone(&state.models_manager))) + } + pub(crate) async fn list_models( &self, refresh_strategy: RefreshStrategy, ) -> std::io::Result> { - // Check before consulting even a warm cache, just as turn admission checks - // the retained session route before using it. + let (config, models_manager) = self.current().await?; + self.list_models_from_snapshot(config, models_manager, refresh_strategy) + .await + } + + pub(crate) async fn list_models_from_snapshot( + &self, + config: Arc, + models_manager: SharedModelsManager, + refresh_strategy: RefreshStrategy, + ) -> std::io::Result> { + // Keep the existing policy check even for a warm catalog. A policy + // edit alone must not make the retained provider silently permissible. self.config_manager - .check_thread_model_provider(&self.config) + .check_thread_model_provider(&config) .await?; - Ok(self - .models_manager - .list_models(refresh_strategy, self.config.http_client_factory()) + Ok(models_manager + .list_models(refresh_strategy, config.http_client_factory()) .await) } } 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 5356eb2af3..12b1c0834c 100644 --- a/codex-rs/app-server/src/models_refresh_worker_tests.rs +++ b/codex-rs/app-server/src/models_refresh_worker_tests.rs @@ -18,6 +18,7 @@ use codex_models_manager::manager::OpenAiModelsManager; use codex_models_manager::manager::SharedModelsManager; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CoreResult; +use codex_utils_absolute_path::AbsolutePathBuf; use pretty_assertions::assert_eq; use tempfile::tempdir; use tokio::sync::Notify; @@ -98,13 +99,13 @@ async fn refreshes_immediately_periodically_and_stops_when_dropped() { std::fs::write(&requirements_path, "model_provider = 'ollama'").unwrap(); let config_manager = crate::config_manager::ConfigManager::new_for_tests( codex_home.path().to_path_buf(), - Vec::new(), + vec![("model_provider".to_string(), "ollama".into())], LoaderOverrides::with_managed_config_path_for_tests( codex_home.path().join("managed_config.toml"), ), CloudConfigBundleLoader::default(), ); - // --oss selects the provider through typed overrides, outside ConfigManager's raw -c flags. + // The TUI forwards its typed --oss selection through the raw CLI overrides too. let config = Arc::new( ConfigBuilder::default() .codex_home(codex_home.path().to_path_buf()) @@ -122,7 +123,19 @@ async fn refreshes_immediately_periodically_and_stops_when_dropped() { endpoint.clone(), /*auth_manager*/ None, )); - let catalog = Arc::new(ModelCatalog::new(config_manager, config, models_manager)); + let auth_manager = codex_login::AuthManager::shared_from_config( + config.as_ref(), + /*enable_codex_api_key_env*/ false, + ) + .await + .expect("test auth manager"); + let catalog = Arc::new(ModelCatalog::new( + config_manager, + Some(config.cwd.to_path_buf()), + config, + Arc::clone(&models_manager), + auth_manager, + )); let worker = spawn_with_interval(&catalog, Duration::from_millis(/*millis*/ 10)); endpoint.wait_for_fetch_count(/*expected*/ 2).await; @@ -132,26 +145,37 @@ async fn refreshes_immediately_periodically_and_stops_when_dropped() { assert_eq!(endpoint.fetch_count.load(Ordering::SeqCst), 2); - // Reloading config would select the newly required provider. The catalog must - // keep validating the typed startup selection, even when only reading its cache. + // Reloading after login preserves an embedded caller's typed --oss override. + catalog.invalidate().await; + assert_eq!( + catalog.current().await.unwrap().0.model_provider_id, + "ollama" + ); + + assert!(Arc::ptr_eq( + &catalog.current().await.unwrap().1, + &models_manager + )); + + // The catalog follows a new managed route without reusing the old endpoint. std::fs::write(&requirements_path, "model_provider = 'openai'").unwrap(); - let error = catalog - .list_models(RefreshStrategy::Offline) - .await - .unwrap_err(); - assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied); - assert!( - error - .get_ref() - .unwrap() - .is::() + catalog.invalidate().await; + catalog.list_models(RefreshStrategy::Offline).await.unwrap(); + assert_eq!( + catalog.current().await.unwrap().0.model_provider_id, + "openai" + ); + std::fs::remove_file(&requirements_path).unwrap(); + catalog.invalidate().await; + assert_eq!( + catalog.current().await.unwrap().0.model_provider_id, + "ollama" ); assert_eq!(endpoint.fetch_count.load(Ordering::SeqCst), 2); } #[tokio::test] -async fn background_refresh_blocks_changed_and_invalid_requirements_then_recovers() --> anyhow::Result<()> { +async fn background_refresh_blocks_invalid_requirements_then_recovers() -> anyhow::Result<()> { let home = tempdir()?; let config = Arc::new( crate::config_manager::ConfigManager::without_managed_config_for_tests( @@ -180,7 +204,19 @@ async fn background_refresh_blocks_changed_and_invalid_requirements_then_recover endpoint.clone(), /*auth_manager*/ None, )); - let catalog = Arc::new(ModelCatalog::new(config_manager, config, models_manager)); + let auth_manager = codex_login::AuthManager::shared_from_config( + config.as_ref(), + /*enable_codex_api_key_env*/ false, + ) + .await + .expect("test auth manager"); + let catalog = Arc::new(ModelCatalog::new( + config_manager, + Some(config.cwd.to_path_buf()), + config, + models_manager, + auth_manager, + )); let feedback = CodexFeedback::new(); // This test uses Tokio's current-thread runtime, so spawned refreshes share the subscriber. let _subscriber_guard = tracing::subscriber::set_default( @@ -224,8 +260,118 @@ async fn background_refresh_blocks_changed_and_invalid_requirements_then_recover .buffer, )?; assert!(logs.contains("model catalog refresh blocked by provider requirements")); - assert!(logs.contains("error_kind=PermissionDenied")); assert!(logs.contains("error_kind=InvalidData")); assert!(!logs.contains("private-provider-token-marker")); Ok(()) } + +/// A transient cloud failure must not pin the fallback catalog after policy recovers. +#[tokio::test] +async fn catalog_reload_recovers_after_cloud_failure() -> anyhow::Result<()> { + let home = tempdir()?; + let failed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let loader_failed = Arc::clone(&failed); + let loader = CloudConfigBundleLoader::from_getter(move || { + let failed = Arc::clone(&loader_failed); + async move { + if failed.load(Ordering::SeqCst) { + Err(codex_config::CloudConfigBundleLoadError::new( + codex_config::CloudConfigBundleLoadErrorCode::RequestFailed, + /*status_code*/ None, + "test cloud timeout", + )) + } else { + Ok(Some( + CloudConfigBundleFixture::enterprise_requirement("model_provider = 'ollama'") + .into_bundle(), + )) + } + } + }); + let manager = crate::config_manager::ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + LoaderOverrides::without_managed_config_for_tests(), + loader, + ); + let config = Arc::new(manager.load_non_project_config().await?); + let auth = codex_login::AuthManager::shared_from_config( + config.as_ref(), + /*enable_codex_api_key_env*/ false, + ) + .await?; + let models = codex_core::build_models_manager(&config, Arc::clone(&auth)); + let catalog = Arc::new(ModelCatalog::new( + manager, /*startup_cwd*/ None, config, models, auth, + )); + failed.store(true, Ordering::SeqCst); + catalog.invalidate().await; + assert_eq!(catalog.current().await?.0.model_provider_id, "openai"); + assert!(catalog.list_models(RefreshStrategy::Offline).await.is_err()); + failed.store(false, Ordering::SeqCst); + // Exercise the same online request used by the periodic background worker. + catalog.list_models(RefreshStrategy::Online).await?; + assert_eq!(catalog.current().await?.0.model_provider_id, "ollama"); + assert!( + !catalog + .list_models(RefreshStrategy::Offline) + .await? + .is_empty() + ); + Ok(()) +} + +/// Repairing a project config after fallback must restore that project's local model catalog. +#[tokio::test] +async fn catalog_reload_recovers_original_project_after_startup_fallback() -> anyhow::Result<()> { + let home = tempdir()?; + let project = tempdir()?; + let project_path = AbsolutePathBuf::from_absolute_path(project.path())? + .canonicalize()? + .to_path_buf(); + std::fs::create_dir(project_path.join(".codex"))?; + std::fs::write( + home.path().join("config.toml"), + toml::to_string(&serde_json::json!({ + "projects": {project_path.to_string_lossy(): {"trust_level": "trusted"}} + }))?, + )?; + let project_config = project_path.join(".codex/config.toml"); + std::fs::write(&project_config, "model_catalog_json = [")?; + let manager = crate::config_manager::ConfigManager::without_managed_config_for_tests( + home.path().to_path_buf(), + ); + let startup = manager + .load_startup_config(Some(project_path.clone())) + .await?; + assert!(startup.used_defaults); + let config = Arc::new(startup.config); + let auth = codex_login::AuthManager::shared_from_config( + config.as_ref(), + /*enable_codex_api_key_env*/ false, + ) + .await?; + let models = codex_core::build_models_manager(&config, Arc::clone(&auth)); + let catalog = ModelCatalog::new(manager, Some(project_path.clone()), config, models, auth); + let mut model = codex_models_manager::bundled_models_response()? + .models + .remove(0); + model.slug = "project-model".into(); + std::fs::write( + project_path.join(".codex/catalog.json"), + serde_json::to_vec(&serde_json::json!({"models": [model]}))?, + )?; + std::fs::write(&project_config, "model_catalog_json = 'catalog.json'")?; + catalog.invalidate().await; + let (config, _) = catalog.current().await?; + assert_eq!(config.cwd.as_path(), project_path); + let models = catalog.list_models(RefreshStrategy::Offline).await?; + assert_eq!( + models + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["project-model"] + ); + Ok(()) +} diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index b8d56053b1..5fc3c75ac3 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -5,6 +5,7 @@ use super::bedrock_auth::ensure_user_model_provider_can_be_bedrock; use super::*; use crate::auth_mode::auth_mode_to_api; use crate::external_auth::ExternalAuthBridge; +use crate::model_catalog::ModelCatalog; use crate::outgoing_message::AccountNotification; use chrono::DateTime; use codex_app_server_protocol::DesktopOnboardingEntrypoint; @@ -101,6 +102,7 @@ pub(crate) struct AccountRequestProcessor { outgoing: Arc, config: Arc, config_manager: ConfigManager, + model_catalog: Arc, active_login: Arc>>, workspace_routing: Arc>>, workspace_routing_fetches: Arc>, @@ -118,6 +120,7 @@ impl AccountRequestProcessor { outgoing: Arc, config: Arc, config_manager: ConfigManager, + model_catalog: Arc, ) -> Arc { let gateway_notifications = crate::gateway_oauth_notifications::spawn( Arc::clone(&auth_manager), @@ -136,6 +139,7 @@ impl AccountRequestProcessor { outgoing, config, config_manager, + model_catalog, active_login: Arc::new(Mutex::new(None)), gateway_login: Arc::new(std::sync::Mutex::new(/*t*/ None)), gateway_client: Arc::new(std::sync::Mutex::new(/*t*/ None)), @@ -485,6 +489,9 @@ impl AccountRequestProcessor { .await .map(|()| LoginAccountResponse::ApiKey {}); let logged_in = result.is_ok(); + if logged_in { + self.model_catalog.invalidate().await; + } self.outgoing.send_result(request_id, result).await; if logged_in { @@ -573,6 +580,9 @@ impl AccountRequestProcessor { } .await; let logged_in = result.is_ok(); + if logged_in { + self.model_catalog.invalidate().await; + } self.outgoing.send_result(request_id, result).await; if logged_in { @@ -845,6 +855,9 @@ impl AccountRequestProcessor { .login_chatgpt_auth_tokens_response(access_token, chatgpt_account_id, chatgpt_plan_type) .await; let logged_in = result.is_ok(); + if logged_in { + self.model_catalog.invalidate().await; + } self.outgoing.send_result(request_id, result).await; if logged_in { @@ -986,6 +999,9 @@ impl AccountRequestProcessor { payload_v2.error = Some("account changed before sign-in completed".into()); } } + if payload_v2.success { + self.model_catalog.invalidate().await; + } self.send_account_login_notifications(payload_v2).await; } @@ -1059,6 +1075,7 @@ impl AccountRequestProcessor { clear_user_model_provider_if_bedrock(&self.config_manager, &config).await?; } + self.model_catalog.invalidate().await; *self.workspace_routing.lock().await = None; Self::maybe_refresh_plugin_caches_for_current_config( diff --git a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing_tests.rs b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing_tests.rs index 7ae5fff4da..dafe4e8e98 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/workspace_routing_tests.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/workspace_routing_tests.rs @@ -98,12 +98,20 @@ async fn background_routing_skips_config_until_chatgpt_sign_in( sender, codex_analytics::AnalyticsEventsClient::disabled(), )); + let model_catalog = Arc::new(crate::model_catalog::ModelCatalog::new( + manager.clone(), + Some(config.cwd.to_path_buf()), + Arc::clone(&config), + threads.get_models_manager(), + Arc::clone(&auth_manager), + )); let processor = AccountRequestProcessor::new( Arc::clone(&auth_manager), threads, outgoing, config, manager, + model_catalog, ); let initial_loads = loads.load(Ordering::SeqCst); for _ in 0..3 { 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 e8ea00fd47..efbed26c5c 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -178,10 +178,15 @@ impl CatalogRequestProcessor { &self, params: ModelListParams, ) -> Result, JSONRPCErrorError> { + let (config, models_manager) = self + .model_catalog + .current() + .await + .map_err(|err| config_load_error(&err))?; // Gate the same provider used by the catalog, including when its model cache is warm. // Resolving credentials may refresh them, but explicit host policy prevents browser login. if let Some(gateway) = codex_model_provider::create_model_provider( - self.config.model_provider.clone(), + config.model_provider.clone(), Some(self.thread_manager.auth_manager()), ) .gateway_auth_manager() @@ -189,7 +194,7 @@ impl CatalogRequestProcessor { { // Refreshing credentials can contact the gateway before the catalog's own check. self.config_manager - .check_thread_model_provider(&self.config) + .check_thread_model_provider(&config) .await .map_err(|err| config_load_error(&err))?; if gateway.resolve_access_token().await.is_err() { @@ -198,9 +203,9 @@ impl CatalogRequestProcessor { .load_latest_config(/*fallback_cwd*/ None) .await .map_err(|err| config_load_error(&err))?; - // Login RPCs use current config, while the catalog retains its startup provider. - if current.model_provider_id != self.config.model_provider_id - || current.model_provider != self.config.model_provider + // Configuration edits without an account transition still require a restart to refresh the model catalog. + if current.model_provider_id != config.model_provider_id + || current.model_provider != config.model_provider { return Err(invalid_request( "Model provider settings changed. Restart Codex to apply them, then retry fetching the model list", @@ -211,7 +216,7 @@ impl CatalogRequestProcessor { )); } } - self.list_models(params) + self.list_models(params, config, models_manager) .await .map(|response| Some(response.into())) } @@ -280,6 +285,8 @@ impl CatalogRequestProcessor { async fn list_models( &self, params: ModelListParams, + config: Arc, + models_manager: codex_models_manager::manager::SharedModelsManager, ) -> Result { let ModelListParams { limit, @@ -288,7 +295,11 @@ impl CatalogRequestProcessor { } = params; let presets = self .model_catalog - .list_models(codex_models_manager::manager::RefreshStrategy::OnlineIfUncached) + .list_models_from_snapshot( + config, + models_manager, + codex_models_manager::manager::RefreshStrategy::OnlineIfUncached, + ) .await .map_err(|err| config_load_error(&err))?; let models = supported_models(presets, include_hidden.unwrap_or(false)); diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 1343329ae1..d903a6fce2 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -4,6 +4,7 @@ use crate::config_manager::ConfigManager; use crate::config_manager_service::ConfigManagerError; use crate::error_code::internal_error; use crate::error_code::invalid_request; +use crate::model_catalog::ModelCatalog; use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::OutgoingMessageSender; use codex_analytics::AnalyticsEventsClient; @@ -83,6 +84,7 @@ pub(crate) struct ConfigRequestProcessor { outgoing: Arc, config_manager: ConfigManager, thread_manager: Arc, + model_catalog: Arc, analytics_events_client: AnalyticsEventsClient, } @@ -91,12 +93,14 @@ impl ConfigRequestProcessor { outgoing: Arc, config_manager: ConfigManager, thread_manager: Arc, + model_catalog: Arc, analytics_events_client: AnalyticsEventsClient, ) -> Self { Self { outgoing, config_manager, thread_manager, + model_catalog, analytics_events_client, } } @@ -328,6 +332,11 @@ impl ConfigRequestProcessor { .set_api_key_model_discovery_enabled( config.features.enabled(Feature::ApiKeyModelDiscovery), ); + self.model_catalog + .set_api_key_model_discovery_enabled( + config.features.enabled(Feature::ApiKeyModelDiscovery), + ) + .await; if should_start_background_rollout_migration && config.features.enabled(feature) { self.thread_manager.start_background_rollout_migration(); } diff --git a/codex-rs/app-server/src/user_verification_test_support.rs b/codex-rs/app-server/src/user_verification_test_support.rs index d2096bc1f1..6367af1bd7 100644 --- a/codex-rs/app-server/src/user_verification_test_support.rs +++ b/codex-rs/app-server/src/user_verification_test_support.rs @@ -157,6 +157,7 @@ impl Harness { outgoing: Arc::clone(&outgoing), analytics_events_client: AnalyticsEventsClient::disabled(), arg0_paths: Arg0DispatchPaths::default(), + startup_cwd: Some(config.cwd.to_path_buf()), config, config_manager, environment_manager: Arc::new(EnvironmentManager::default_for_tests()), diff --git a/codex-rs/app-server/tests/suite/v2/account.rs b/codex-rs/app-server/tests/suite/v2/account.rs index 94a3eb2e45..fa6159d6bd 100644 --- a/codex-rs/app-server/tests/suite/v2/account.rs +++ b/codex-rs/app-server/tests/suite/v2/account.rs @@ -2687,12 +2687,44 @@ async fn login_account_chatgpt_start_can_be_cancelled() -> Result<()> { #[serial(login_port)] async fn login_account_chatgpt_uses_oauth_overrides() -> Result<()> { let codex_home = TempDir::new()?; - create_config_toml(codex_home.path(), CreateConfigTomlParams::default())?; let mock_server = MockServer::start().await; + create_config_toml( + codex_home.path(), + CreateConfigTomlParams { + chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())), + ..Default::default() + }, + )?; + app_test_support::mount_workspace_routing(&mock_server).await; + let requirements = format!( + "model_provider = 'gateway'\n[model_providers.gateway]\nname = 'Gateway'\nrequires_openai_auth = true\nbase_url = '{0}/v1'\nmodel_catalog_url = '{0}/v1/models'", + mock_server.uri(), + ); + Mock::given(method("GET")) + .and(path("/backend-api/wham/config/bundle")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "requirements_toml": {"enterprise_managed": [{ + "id": "gateway", "name": "Gateway", "contents": requirements, + }]} + }))) + .mount(&mock_server) + .await; + let mut model = codex_models_manager::bundled_models_response()? + .models + .remove(0); + model.slug = "workspace-gateway-model".into(); + model.visibility = codex_protocol::openai_models::ModelVisibility::List; + Mock::given(method("GET")) + .and(path("/v1/models")) + .and(header("authorization", "Bearer access-token-123")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"models": [model]}))) + .expect(1..) + .mount(&mock_server) + .await; let id_token = encode_id_token( &ChatGptIdTokenClaims::new() .email("staging@example.com") - .plan_type("pro") + .plan_type("enterprise") .chatgpt_account_id(WORKSPACE_ID_EMBEDDED), )?; mock_oauth_token(&mock_server, &id_token).await; @@ -2711,6 +2743,15 @@ async fn login_account_chatgpt_uses_oauth_overrides() -> Result<()> { .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) .await?; + let request = mcp.send_list_models_request(Default::default()).await?; + let initial: codex_app_server_protocol::ModelListResponse = mcp.read_response(request).await?; + assert!( + !initial + .data + .iter() + .any(|model| model.id == "workspace-gateway-model") + ); + let request_id = mcp.send_login_account_chatgpt_request().await?; let login: LoginAccountResponse = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; @@ -2789,6 +2830,18 @@ async fn login_account_chatgpt_uses_oauth_overrides() -> Result<()> { onboarding_entrypoint: None, } ); + // Listing immediately after asynchronous login completion must use the new cloud provider. + let request = mcp.send_list_models_request(Default::default()).await?; + let models: codex_app_server_protocol::ModelListResponse = mcp.read_response(request).await?; + assert_eq!( + models + .data + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["workspace-gateway-model"] + ); + mock_server.verify().await; Ok(()) } diff --git a/codex-rs/app-server/tests/suite/v2/gateway_oauth_tests.rs b/codex-rs/app-server/tests/suite/v2/gateway_oauth_tests.rs index d28243b7ad..1b9de1cd96 100644 --- a/codex-rs/app-server/tests/suite/v2/gateway_oauth_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/gateway_oauth_tests.rs @@ -136,6 +136,42 @@ async fn model_list_requests_restart_after_gateway_provider_changes() -> Result< Ok(()) } +/// Account changes reload the catalog before validating a retained OAuth gateway. +#[tokio::test] +async fn model_list_leaves_oauth_gateway_after_account_change() -> Result<()> { + let oauth = MockServer::start().await; + let home = TempDir::new()?; + std::fs::write(home.path().join("config.toml"), gateway_config(&oauth))?; + let mut server = gateway_server(home.path()).await?; + std::fs::write(home.path().join("config.toml"), "")?; + std::fs::write( + home.path().join("requirements.toml"), + "model_provider = 'openai'", + )?; + let id = server + .send_login_account_api_key_request("sk-catalog-test") + .await?; + let _: codex_app_server_protocol::LoginAccountResponse = server.read_response(id).await?; + server + .read_stream_until_notification_message("account/updated") + .await?; + let models: codex_app_server_protocol::ModelListResponse = server + .request(|request_id| ClientRequest::ModelList { + request_id, + params: codex_app_server_protocol::ModelListParams::default(), + }) + .await?; + assert!(!models.data.is_empty()); + assert!( + oauth + .received_requests() + .await + .expect("gateway requests") + .is_empty() + ); + Ok(()) +} + #[tokio::test] async fn token_exchange_failure_completes_login_and_updates_readiness() -> Result<()> { let home = TempDir::new()?; diff --git a/codex-rs/app-server/tests/suite/v2/model_list.rs b/codex-rs/app-server/tests/suite/v2/model_list.rs index 9a78ca5e34..c9604f1b09 100644 --- a/codex-rs/app-server/tests/suite/v2/model_list.rs +++ b/codex-rs/app-server/tests/suite/v2/model_list.rs @@ -102,6 +102,31 @@ requires_openai_auth = true .with_env_overrides(&[("OPENAI_API_KEY", None), ("CODEX_API_KEY", None)]) .build_initialized() .await?; + // Exercise runtime toggles after login has replaced the startup catalog manager. + if let Some(enabled) = app_enablement { + let _: ExperimentalFeatureEnablementSetResponse = mcp + .request( + |request_id| ClientRequest::ExperimentalFeatureEnablementSet { + request_id, + params: ExperimentalFeatureEnablementSetParams { + enablement: [("api_key_model_discovery".to_string(), !enabled)].into(), + }, + }, + ) + .await?; + } + let config_path = codex_home.path().join("config.toml"); + let config = std::fs::read_to_string(&config_path)? + .replace("name = \"OpenAI\"", "name = \"After login\""); + std::fs::write(config_path, config)?; + let login = mcp.send_login_account_api_key_request("test-key").await?; + let _: codex_app_server_protocol::LoginAccountResponse = mcp.read_response(login).await?; + let _: ModelListResponse = mcp + .request(|request_id| ClientRequest::ModelList { + request_id, + params: ModelListParams::default(), + }) + .await?; let mut bundled = codex_models_manager::bundled_models_response()?.models; bundled.sort_by_key(|model| model.priority); let mut bundled = ModelPreset::filter_by_auth( diff --git a/codex-rs/app-server/tests/suite/v2/model_list_requirements_tests.rs b/codex-rs/app-server/tests/suite/v2/model_list_requirements_tests.rs index 77964124dd..6c2f24724d 100644 --- a/codex-rs/app-server/tests/suite/v2/model_list_requirements_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/model_list_requirements_tests.rs @@ -1,23 +1,26 @@ //! Exercises live catalog routing through the public model/list RPC. use anyhow::Result; -use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; -use app_test_support::write_chatgpt_auth; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::LoginAccountResponse; +use codex_app_server_protocol::LogoutAccountResponse; use codex_app_server_protocol::ModelListParams; use codex_app_server_protocol::ModelListResponse; use codex_app_server_protocol::RequestId; use codex_config::types::AuthCredentialsStoreMode; use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelsResponse; -use core_test_support::responses::mount_models_once; use pretty_assertions::assert_eq; use serde_json::json; use std::time::Duration; use tempfile::TempDir; use test_case::test_case; +use wiremock::Mock; use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path_regex; fn catalog(slug: &str) -> Result { Ok(ModelsResponse { @@ -45,39 +48,61 @@ async fn list_models(server: &mut TestAppServer) -> Result { .await } +async fn sign_in(server: &mut TestAppServer) -> Result<()> { + let id = server + .send_login_account_api_key_request("sk-catalog-test") + .await?; + let _: LoginAccountResponse = server.read_response(id).await?; + Ok(()) +} + #[test_case("other"; "selection_changes")] #[test_case("gateway"; "definition_changes")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn model_list_blocks_noncompliant_cached_provider_until_requirements_allow_it( +/// Switching managed provider selection or definition refreshes the catalog in +/// the same server, while invalid requirements still block catalog access. +async fn model_list_follows_changed_provider_without_restart( required_selection: &str, ) -> Result<()> { let gateway = MockServer::start().await; let other = MockServer::start().await; - mount_models_once(&gateway, catalog("gateway-model")?).await; - mount_models_once(&other, catalog("other-model")?).await; + for (server, slug) in [(&gateway, "gateway-model"), (&other, "other-model")] { + Mock::given(method("GET")) + .and(path_regex(".*/models$")) + .respond_with(ResponseTemplate::new(200).set_body_json(catalog(slug)?)) + .mount(server) + .await; + } let home = TempDir::new()?; let providers = format!( r#" [model_providers.gateway] name = "Gateway" base_url = "{}/v1" +model_catalog_url = "{}/v1/models" requires_openai_auth = true [model_providers.other] name = "Other" base_url = "{}/v1" +model_catalog_url = "{}/v1/models" requires_openai_auth = true "#, gateway.uri(), + gateway.uri(), + other.uri(), other.uri(), ); std::fs::write( home.path().join("config.toml"), - format!("model_provider = 'gateway'\n{providers}"), + format!( + "model_provider = 'gateway'\n[features]\napi_key_model_discovery = true\n{providers}" + ), )?; - write_chatgpt_auth( + codex_login::login_with_api_key( home.path(), - ChatGptAuthFixture::new("chatgpt-access-token").plan_type("pro"), + "sk-initial-catalog-test", AuthCredentialsStoreMode::File, + codex_login::AuthKeyringBackendKind::default(), )?; let requirements_path = home.path().join("requirements.toml"); std::fs::write( @@ -85,25 +110,12 @@ requires_openai_auth = true format!("model_provider = 'gateway'\n{providers}"), )?; let mut server = TestAppServer::builder() + .with_mock_chatgpt_backend() .with_codex_home(home.path()) .with_env_overrides(&[("OPENAI_API_KEY", None)]) - .build_initialized() + .build_initialized_with_timeout(Duration::from_secs(/*secs*/ 60)) .await?; - // Let the startup refresh populate the cache before exercising cached RPCs. - tokio::time::timeout(Duration::from_secs(/*secs*/ 10), async { - loop { - if let Ok(bytes) = tokio::fs::read(home.path().join("models_cache.json")).await - && let Ok(cache) = serde_json::from_slice::(&bytes) - && cache["models"][0]["slug"] == "gateway-model" - { - break; - } - tokio::time::sleep(Duration::from_millis(/*millis*/ 10)).await; - } - }) - .await?; - let initial = list_models(&mut server).await?; assert_eq!( initial @@ -113,23 +125,6 @@ requires_openai_auth = true .collect::>(), vec!["gateway-model"] ); - assert_eq!(list_models(&mut server).await?, initial); - assert_eq!( - gateway - .received_requests() - .await - .expect("request recording enabled") - .len(), - 1 - ); - assert!( - other - .received_requests() - .await - .expect("request recording enabled") - .is_empty() - ); - let updated_providers = if required_selection == "gateway" { providers.replace(&gateway.uri(), &other.uri()) } else { @@ -137,34 +132,29 @@ requires_openai_auth = true }; let updated = format!("model_provider = '{required_selection}'\n{updated_providers}"); std::fs::write(&requirements_path, &updated)?; - let id = server - .send_raw_request("model/list", Some(json!({}))) - .await?; - let error = server - .read_stream_until_error_message(RequestId::Integer(id)) - .await?; - assert_eq!(error.error, codex_app_server_protocol::JSONRPCErrorError { - code: -32600, - message: "failed to load configuration: Your organization's required model provider settings changed. Restart Codex to apply them; this request was not sent".to_string(), - data: None, - }); + sign_in(&mut server).await?; + let switched = list_models(&mut server).await?; assert_eq!( + switched + .data + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["other-model"] + ); + let requests_before = [ gateway .received_requests() .await - .expect("request recording enabled") + .expect("recorded provider requests") .len(), - 1 - ); - assert!( other .received_requests() .await - .expect("request recording enabled") - .is_empty() - ); - - // A warm catalog must not hide a policy load failure or make any stale request. + .expect("recorded provider requests") + .len(), + ]; + // A warm catalog must not hide a policy load failure or send to either provider. std::fs::write(&requirements_path, "model_provider = []")?; let id = server .send_raw_request("model/list", Some(json!({}))) @@ -175,31 +165,107 @@ requires_openai_auth = true assert_eq!(error.error.code, -32600); assert!(error.error.message.contains("failed to load configuration")); assert_eq!( - gateway - .received_requests() - .await - .expect("request recording enabled") - .len(), - 1 + [ + gateway + .received_requests() + .await + .expect("recorded provider requests") + .len(), + other + .received_requests() + .await + .expect("recorded provider requests") + .len(), + ], + requests_before, ); - assert!( - other - .received_requests() - .await - .expect("request recording enabled") - .is_empty() - ); - - // Removing the requirement makes the retained provider permissible again. + // Removing the requirement restores the user-configured provider in this process. std::fs::remove_file(requirements_path)?; + sign_in(&mut server).await?; assert_eq!(list_models(&mut server).await?, initial); + Ok(()) +} + +/// Warm lists reuse their configuration; a completed login reloads the catalog once. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn model_list_reuses_catalog_until_account_change() -> Result<()> { + let home = TempDir::new()?; + let catalog_path = home.path().join("catalog.json"); + std::fs::write(&catalog_path, serde_json::to_vec(&catalog("first-model")?)?)?; + let mut config = toml::map::Map::new(); + config.insert( + "model_catalog_json".to_string(), + toml::Value::String(catalog_path.to_string_lossy().into_owned()), + ); + std::fs::write(home.path().join("config.toml"), toml::to_string(&config)?)?; + let mut server = TestAppServer::builder() + .with_mock_chatgpt_backend() + .with_codex_home(home.path()) + .with_env_overrides(&[("OPENAI_API_KEY", None)]) + .build_initialized() + .await?; + let initial = list_models(&mut server).await?; assert_eq!( - gateway - .received_requests() - .await - .expect("request recording enabled") - .len(), - 1 + initial + .data + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["first-model"] + ); + // A full config reload would try to parse this invalid catalog. + std::fs::write(&catalog_path, "not JSON")?; + assert_eq!(list_models(&mut server).await?, initial); + std::fs::write( + &catalog_path, + serde_json::to_vec(&catalog("second-model")?)?, + )?; + assert_eq!(list_models(&mut server).await?, initial); + sign_in(&mut server).await?; + let changed = list_models(&mut server).await?; + assert_eq!( + changed + .data + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["second-model"] + ); + std::fs::write( + &catalog_path, + serde_json::to_vec(&catalog("logged-out-model")?)?, + )?; + let id = server.send_logout_account_request().await?; + let _: LogoutAccountResponse = server.read_response(id).await?; + assert_eq!( + list_models(&mut server) + .await? + .data + .iter() + .map(|model| model.id.as_str()) + .collect::>(), + vec!["logged-out-model"] ); Ok(()) } + +/// Account transitions preserve the non-strict startup fallback for malformed user config. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn model_list_preserves_invalid_startup_config_fallback() -> Result<()> { + let home = TempDir::new()?; + std::fs::write(home.path().join("config.toml"), "model = [")?; + let mut server = TestAppServer::builder() + .with_mock_chatgpt_backend() + .with_codex_home(home.path()) + .with_env_overrides(&[("OPENAI_API_KEY", None)]) + .build_initialized_with_timeout(Duration::from_secs(/*secs*/ 60)) + .await?; + let initial = list_models(&mut server).await?; + assert!(!initial.data.is_empty()); + sign_in(&mut server).await?; + assert!(!list_models(&mut server).await?.data.is_empty()); + let id = server.send_logout_account_request().await?; + let _: LogoutAccountResponse = server.read_response(id).await?; + assert!(!list_models(&mut server).await?.data.is_empty()); + Ok(()) +} diff --git a/codex-rs/tui/src/startup_orchestration.rs b/codex-rs/tui/src/startup_orchestration.rs index a96e25bbfd..ca871087c4 100644 --- a/codex-rs/tui/src/startup_orchestration.rs +++ b/codex-rs/tui/src/startup_orchestration.rs @@ -76,7 +76,7 @@ pub(super) async fn run_main_inner( let raw_overrides = cli.config_overrides.raw_overrides.clone(); // `oss` model provider. let overrides_cli = codex_utils_cli::CliConfigOverrides { raw_overrides }; - let cli_kv_overrides = match overrides_cli.parse_overrides() { + let mut cli_kv_overrides = match overrides_cli.parse_overrides() { // Parse `-c` overrides from the CLI. Ok(v) => v, #[allow(clippy::print_stderr)] @@ -421,6 +421,15 @@ pub(super) async fn run_main_inner( None }; + // Pass the --oss provider to the embedded app-server so catalog reloads keep + // the launch choice after workspace requirements are removed. + if let Some(provider) = &model_provider_override { + cli_kv_overrides.push(( + "model_provider".to_string(), + toml::Value::String(provider.clone()), + )); + } + // When using `--oss`, let the bootstrapper pick the model based on selected provider let model = if let Some(model) = &cli.model { Some(model.clone())