diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 0643950835..0a30f470ef 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -77,6 +77,7 @@ codex-memories-write = { workspace = true } codex-mcp = { workspace = true } codex-mcp-extension = { workspace = true } codex-model-provider = { workspace = true } +codex-model-provider-info = { workspace = true } codex-models-manager = { workspace = true } codex-protocol = { workspace = true } codex-queue-extension = { workspace = true } @@ -131,7 +132,6 @@ axum = { workspace = true, default-features = false, features = [ "tokio", ] } base64 = { workspace = true } -codex-model-provider-info = { workspace = true } codex-utils-cargo-bin = { workspace = true } core_test_support = { workspace = true } flate2 = { workspace = true } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index e1fc57eaa0..507c7ec0d1 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -113,6 +113,15 @@ Failures use the normal JSON-RPC error envelope with closed `{type, reason}` dat `invalidRequest`, `unavailable`, `cancelled`, or `failed`. UI clients branch on these values rather than message text. Native diagnostic payloads stay private. +## Managed model provider requirements + +Existing threads retain their provider configuration. Input RPCs reject requests when managed +`model_provider` or `model_providers` requirements no longer match that configuration, or cannot +be loaded. This covers turn start/steer, review, compaction, manual queue start, and active goal +updates. Realtime connections use separate routing configuration and are not checked here. +Interrupt, realtime stop, and goal pause/clear remain available. User and project +configuration changes alone do not invalidate existing threads. + # Amazon Bedrock authentication If `model_providers.amazon-bedrock.aws.credential_export` is configured, Bedrock setup and diff --git a/codex-rs/app-server/src/config_manager.rs b/codex-rs/app-server/src/config_manager.rs index f62118d292..3dca9dd143 100644 --- a/codex-rs/app-server/src/config_manager.rs +++ b/codex-rs/app-server/src/config_manager.rs @@ -5,6 +5,7 @@ use codex_config::ConfigLayerStack; use codex_config::LoaderOverrides; use codex_config::ThreadConfigLoader; use codex_config::loader::load_config_layers_state; +use codex_config::loader::load_managed_requirements_state; use codex_core::config::Config; use codex_core::config::ConfigBuilder; use codex_core::config::ConfigOverrides; @@ -12,6 +13,10 @@ use codex_exec_server::LOCAL_FS; use codex_features::feature_for_key; use codex_login::AuthManager; use codex_login::default_client::set_default_client_residency_requirement; +use codex_model_provider_info::AMAZON_BEDROCK_PROVIDER_ID; +use codex_model_provider_info::AMAZON_BEDROCK_RUNTIME_PROVIDER_ID; +use codex_model_provider_info::built_in_model_providers; +use codex_model_provider_info::merge_configured_model_providers; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_json_to_toml::json_to_toml; use std::collections::BTreeMap; @@ -25,6 +30,12 @@ use toml::Value as TomlValue; use tracing::instrument; use tracing::warn; +#[derive(Debug, thiserror::Error)] +#[error( + "Your organization's required model provider settings changed. Restart Codex to apply them; this request was not sent" +)] +pub(crate) struct ModelProviderRequirementsChanged; + /// Shared app-server entry point for loading effective Codex configuration. #[derive(Clone)] pub(crate) struct ConfigManager { @@ -167,6 +178,62 @@ impl ConfigManager { Ok(config) } + /// Checks the retained session route against current managed provider requirements. + pub(crate) async fn check_thread_model_provider( + &self, + current: &Config, + ) -> std::io::Result<()> { + // Existing threads retain their session route; only managed + // requirements can invalidate it. + let requirements = load_managed_requirements_state( + LOCAL_FS.as_ref(), + &self.codex_home, + codex_config::ConfigLoadOptions { + loader_overrides: self.loader_overrides.clone(), + strict_config: self.strict_config, + cloud_config_bundle: self.current_cloud_config_bundle(), + }, + ) + .await?; + let selection_changed = requirements + .model_provider + .as_ref() + .is_some_and(|provider_id| provider_id != ¤t.model_provider_id); + let required_provider = requirements + .model_providers + .as_ref() + .and_then(|providers| providers.get(¤t.model_provider_id)) + .cloned(); + let required_provider = match required_provider { + Some(provider) + if matches!( + current.model_provider_id.as_str(), + AMAZON_BEDROCK_PROVIDER_ID | AMAZON_BEDROCK_RUNTIME_PROVIDER_ID + ) => + { + // Bedrock requirements contain overrides of the built-in provider. + merge_configured_model_providers( + built_in_model_providers(/*openai_base_url*/ None), + HashMap::from([(current.model_provider_id.clone(), provider)]), + ) + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidData, message))? + .remove(¤t.model_provider_id) + } + Some(provider) => Some(provider), + None => None, + }; + let definition_changed = required_provider + .as_ref() + .is_some_and(|provider| provider != ¤t.model_provider); + if selection_changed || definition_changed { + return Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + ModelProviderRequirementsChanged, + )); + } + Ok(()) + } + pub(crate) async fn load_default_config(&self) -> std::io::Result { let mut loader_overrides = self.loader_overrides.clone(); loader_overrides.ignore_user_config = true; @@ -326,6 +393,10 @@ impl ConfigManager { } } +#[cfg(test)] +#[path = "config_manager_provider_tests.rs"] +mod provider_tests; + pub(crate) fn protected_feature_keys(config_layer_stack: &ConfigLayerStack) -> BTreeSet { let mut protected_features = config_layer_stack .effective_config() diff --git a/codex-rs/app-server/src/config_manager_provider_tests.rs b/codex-rs/app-server/src/config_manager_provider_tests.rs new file mode 100644 index 0000000000..afae98559f --- /dev/null +++ b/codex-rs/app-server/src/config_manager_provider_tests.rs @@ -0,0 +1,196 @@ +//! Checks retained providers against managed policy without reloading session configuration. + +use super::*; +use anyhow::Result; +use codex_config::CloudConfigBundleLoadError; +use codex_config::CloudConfigBundleLoadErrorCode; +use codex_config::ThreadConfigContext; +use codex_config::ThreadConfigLoadError; +use codex_config::ThreadConfigLoadErrorCode; +use codex_config::ThreadConfigLoaderFuture; +use codex_config::ThreadConfigSource; +use codex_config::test_support::CloudConfigBundleFixture; +use pretty_assertions::assert_eq; +use tempfile::tempdir; +use test_case::test_case; + +#[test_case(AMAZON_BEDROCK_PROVIDER_ID; "bedrock")] +#[test_case(AMAZON_BEDROCK_RUNTIME_PROVIDER_ID; "bedrock_runtime")] +#[tokio::test] +async fn provider_requirements_resolve_bedrock_overrides(provider_id: &str) -> Result<()> { + let home = tempdir()?; + let requirements = format!( + "model_provider = '{provider_id}'\n[model_providers.{provider_id}.aws]\nregion = 'us-west-2'" + ); + let manager = ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + LoaderOverrides::without_managed_config_for_tests(), + CloudConfigBundleFixture::loader_with_enterprise_requirement(&requirements), + ); + let current = manager.load_latest_config(/*fallback_cwd*/ None).await?; + manager.check_thread_model_provider(¤t).await?; + + let changed = ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + LoaderOverrides::without_managed_config_for_tests(), + CloudConfigBundleFixture::loader_with_enterprise_requirement( + requirements.replace("us-west-2", "us-east-1"), + ), + ); + assert_eq!( + changed + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::PermissionDenied, + ); + Ok(()) +} + +struct UnavailableThreadConfig; + +impl ThreadConfigLoader for UnavailableThreadConfig { + fn load( + &self, + _context: ThreadConfigContext, + ) -> ThreadConfigLoaderFuture<'_, Vec> { + Box::pin(async { + Err(ThreadConfigLoadError::new( + ThreadConfigLoadErrorCode::RequestFailed, + /*status_code*/ None, + "thread config service unavailable", + )) + }) + } +} + +#[tokio::test] +async fn provider_requirements_do_not_reload_thread_config() -> Result<()> { + let home = tempdir()?; + let mut manager = ConfigManager::without_managed_config_for_tests(home.path().to_path_buf()); + let current = manager.load_latest_config(/*fallback_cwd*/ None).await?; + manager.thread_config_loader = Arc::new(UnavailableThreadConfig); + assert!( + manager + .load_latest_config(/*fallback_cwd*/ None) + .await + .is_err() + ); + manager.check_thread_model_provider(¤t).await?; + + manager.cloud_config_bundle = Arc::new(RwLock::new( + CloudConfigBundleFixture::loader_with_enterprise_requirement("model_provider = 'other'"), + )); + assert_eq!( + manager + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::PermissionDenied, + ); + Ok(()) +} + +#[tokio::test] +async fn provider_requirements_ignore_system_defaults_but_reject_requirement_changes() -> Result<()> +{ + let home = tempdir()?; + let system_config_path = home.path().join("system-config.toml"); + let requirements_path = home.path().join("requirements.toml"); + let managed_config_path = home.path().join("managed-config.toml"); + let mut overrides = LoaderOverrides::without_managed_config_for_tests(); + overrides.system_config_path = Some(system_config_path.clone()); + overrides.system_requirements_path = Some(requirements_path.clone()); + overrides.managed_config_path = Some(managed_config_path.clone()); + let manager = ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + overrides, + CloudConfigBundleLoader::default(), + ); + let current = manager.load_latest_config(/*fallback_cwd*/ None).await?; + + std::fs::write(&system_config_path, "invalid toml !!!")?; + manager.check_thread_model_provider(¤t).await?; + + std::fs::write(&requirements_path, "model_provider = 'other'")?; + assert_eq!( + manager + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::PermissionDenied, + ); + + std::fs::write(&requirements_path, "invalid toml !!!")?; + assert_eq!( + manager + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::InvalidData, + ); + + std::fs::write( + &requirements_path, + "[model_providers.gateway]\nbase_url = 'https://example.test'", + )?; + assert_eq!( + manager + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::InvalidData, + ); + std::fs::remove_file(requirements_path)?; + + std::fs::write(managed_config_path, "invalid toml !!!")?; + assert_eq!( + manager + .check_thread_model_provider(¤t) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::InvalidData, + ); + Ok(()) +} + +#[tokio::test] +async fn provider_requirement_load_errors_reject_input() -> Result<()> { + let home = tempdir()?; + let previous = ConfigManager::without_managed_config_for_tests(home.path().to_path_buf()) + .load_latest_config(/*fallback_cwd*/ None) + .await?; + for loader in [ + CloudConfigBundleFixture::loader_with_enterprise_requirement("invalid toml !!!"), + CloudConfigBundleLoader::new(async { + Err(CloudConfigBundleLoadError::new( + CloudConfigBundleLoadErrorCode::RequestFailed, + /*status_code*/ None, + "policy unavailable", + )) + }), + ] { + let manager = ConfigManager::new_for_tests( + home.path().to_path_buf(), + Vec::new(), + LoaderOverrides::without_managed_config_for_tests(), + loader, + ); + assert!( + manager + .check_thread_model_provider(&previous) + .await + .is_err() + ); + } + Ok(()) +} diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index c0d2818bb9..5798a5f602 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -486,11 +486,13 @@ impl MessageProcessor { thread_state_manager.clone(), state_db.clone(), Arc::clone(&goal_service), + config_manager.clone(), ); let thread_queue_processor = ThreadQueueRequestProcessor::new( Arc::clone(&thread_manager), Arc::clone(&thread_store), outgoing.clone(), + config_manager.clone(), queue_service, ); let project_processor = ProjectRequestProcessor::new( diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index c277dd9f14..6d52f5b52f 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -26,6 +26,7 @@ pub(crate) struct ThreadGoalRequestProcessor { thread_state_manager: ThreadStateManager, state_db: Option, goal_service: Arc, + config_manager: ConfigManager, } impl ThreadGoalRequestProcessor { @@ -36,6 +37,7 @@ impl ThreadGoalRequestProcessor { thread_state_manager: ThreadStateManager, state_db: Option, goal_service: Arc, + config_manager: ConfigManager, ) -> Self { Self { thread_manager, @@ -44,6 +46,7 @@ impl ThreadGoalRequestProcessor { thread_state_manager, state_db, goal_service, + config_manager, } } @@ -135,6 +138,26 @@ impl ThreadGoalRequestProcessor { .await?; self.reconcile_thread_goal_rollout(thread_id, &state_db) .await?; + // Active goals can immediately inject an objective or start an idle turn. + // Stopping a goal must remain possible after managed policy changes. + let existing_goal = self + .goal_service + .get_thread_goal(&state_db, thread_id) + .await + .map_err(goal_service_error)?; + let resulting_status = params + .status + .map(ThreadGoalStatus::to_core) + .or_else(|| existing_goal.as_ref().map(|goal| goal.status)) + .unwrap_or(codex_protocol::protocol::ThreadGoalStatus::Active); + if resulting_status == codex_protocol::protocol::ThreadGoalStatus::Active + && let Ok(thread) = self.thread_manager.get_thread(thread_id).await + { + self.config_manager + .check_thread_model_provider(thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; + } let max_goal_token_budget = match self.thread_manager.get_thread(thread_id).await { Ok(thread) => thread.config().await.max_goal_token_budget, Err(_) => self.config.max_goal_token_budget, diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 0b5f502bd8..c6e0524ba3 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2329,6 +2329,10 @@ impl ThreadRequestProcessor { let (_, thread) = self.load_thread(&thread_id).await?; ensure_direct_input_allowed(thread.as_ref()).await?; + self.config_manager + .check_thread_model_provider(thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; self.submit_core_op(request_id, thread.as_ref(), Op::Compact) .await .map_err(|err| internal_error(format!("failed to start compaction: {err}")))?; diff --git a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs index a561272e3f..ea6ec7ddab 100644 --- a/codex-rs/app-server/src/request_processors/thread_queue_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_queue_processor.rs @@ -1,3 +1,4 @@ +use crate::config_manager::ConfigManager; use std::sync::Arc; use crate::error_code::internal_error; @@ -38,6 +39,7 @@ use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStoreError; use super::TurnRequestProcessor; +use super::config_load_error; use super::thread_input::DIRECT_INPUT_TO_MULTI_AGENT_V2_SUBAGENT_ERROR; use super::thread_input::can_accept_direct_input; use super::thread_processor::THREAD_LIST_DEFAULT_LIMIT; @@ -51,6 +53,7 @@ pub(crate) struct ThreadQueueRequestProcessor { thread_manager: Arc, thread_store: Arc, outgoing: Arc, + config_manager: ConfigManager, service: Option>, } @@ -59,12 +62,14 @@ impl ThreadQueueRequestProcessor { thread_manager: Arc, thread_store: Arc, outgoing: Arc, + config_manager: ConfigManager, service: Option>, ) -> Self { Self { thread_manager, thread_store, outgoing, + config_manager, service, } } @@ -187,6 +192,10 @@ impl ThreadQueueRequestProcessor { ensure_direct_input_allowed(loaded_thread.as_deref(), &source)?; let thread = loaded_thread .ok_or_else(|| invalid_request("resume the thread before starting a queued message"))?; + self.config_manager + .check_thread_model_provider(thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; let submission = self .service()? .start( diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 78232498a5..95e1d62f5c 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -531,6 +531,10 @@ impl TurnRequestProcessor { })?; self.ensure_direct_input_allowed(&request_id, thread.as_ref()) .await?; + self.config_manager + .check_thread_model_provider(thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; if let Some(tool_output) = ¶ms.tool_output { if !params.input.is_empty() { return Err(invalid_request( @@ -1032,6 +1036,10 @@ impl TurnRequestProcessor { })?; self.ensure_direct_input_allowed(request_id, thread.as_ref()) .await?; + self.config_manager + .check_thread_model_provider(thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; if params.expected_turn_id.is_empty() { return Err(invalid_request("expectedTurnId must not be empty")); @@ -1457,7 +1465,7 @@ impl TurnRequestProcessor { "paginated threads do not support detached review", )); } - let mut config = self.config.as_ref().clone(); + let mut config = parent_thread.config().await.as_ref().clone(); if let Some(review_model) = &config.review_model { config.model = Some(review_model.clone()); } @@ -1550,6 +1558,10 @@ impl TurnRequestProcessor { let (_, parent_thread) = self.load_thread(&thread_id).await?; self.ensure_direct_input_allowed(request_id, parent_thread.as_ref()) .await?; + self.config_manager + .check_thread_model_provider(parent_thread.config().await.as_ref()) + .await + .map_err(|error| config_load_error(&error))?; let (review_request, display_text, target_prompt) = Self::review_request_from_target(target)?; match delivery.unwrap_or(ApiReviewDelivery::Inline).to_core() { diff --git a/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs b/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs index cf797d693b..def85ba0a4 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs @@ -608,6 +608,7 @@ async fn start_elicitation_services( struct ElicitationRoundTripFixture { mcp: TestAppServer, + _codex_home: TempDir, response_mock: ResponseMock, _responses_server: wiremock::MockServer, scenario: ElicitationScenario, @@ -792,6 +793,7 @@ impl ElicitationRoundTripFixture { Ok(Self { mcp, + _codex_home: codex_home, response_mock, _responses_server: responses_server, scenario, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index d545561e74..73928ce666 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -72,6 +72,8 @@ mod misalignment_policy; mod model_auto_review; mod model_list; mod model_provider_capabilities_read; +#[path = "model_provider_enforcement_tests.rs"] +mod model_provider_enforcement; mod multi_agent_v2_developer_instructions; mod otel; mod output_schema; diff --git a/codex-rs/app-server/tests/suite/v2/model_provider_enforcement_tests.rs b/codex-rs/app-server/tests/suite/v2/model_provider_enforcement_tests.rs new file mode 100644 index 0000000000..6b2f5f8e3c --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/model_provider_enforcement_tests.rs @@ -0,0 +1,387 @@ +//! Verifies provider requirements at app-server input and queue admission boundaries. + +use anyhow::Result; +use app_test_support::MockResponsesConfig; +use app_test_support::TestAppServer; +use codex_app_server::in_process; +use codex_app_server::in_process::InProcessServerEvent; +use codex_app_server::in_process::InProcessStartArgs; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadQueueAddParams; +use codex_app_server_protocol::ThreadQueueAddResponse; +use codex_app_server_protocol::ThreadQueueListParams; +use codex_app_server_protocol::ThreadQueueListResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; +use codex_arg0::Arg0DispatchPaths; +use codex_config::CloudConfigBundleLoader; +use codex_config::LoaderOverrides; +use codex_config::NoopThreadConfigLoader; +use codex_core::config::ConfigBuilder; +use codex_exec_server::EnvironmentManager; +use codex_feedback::CodexFeedback; +use codex_protocol::protocol::SessionSource; +use core_test_support::responses; +use core_test_support::streaming_sse::StreamingSseChunk; +use core_test_support::streaming_sse::start_streaming_sse_server; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; +use std::time::Duration; +use tempfile::TempDir; +use test_case::test_case; +use tokio::sync::oneshot; +use tokio::time::timeout; +use wiremock::MockServer; + +#[test_case("other"; "selection_changes")] +#[test_case("gateway"; "definition_changes")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn provider_requirement_changes_reject_inputs_to_existing_threads( + required_selection: &str, +) -> Result<()> { + let (release_response, response_gate) = oneshot::channel(); + let (gateway, _) = start_streaming_sse_server(vec![vec![ + StreamingSseChunk { + gate: None, + body: responses::sse(vec![responses::ev_response_created("response-1")]), + }, + StreamingSseChunk { + gate: Some(response_gate), + body: responses::sse(vec![responses::ev_completed("response-1")]), + }, + ]]) + .await; + let other = MockServer::start().await; + let home = TempDir::new()?; + MockResponsesConfig::new(gateway.uri()) + .enable_feature(codex_features::Feature::Goals) + .write(home.path())?; + let requirements = format!( + r#" +model_provider = "gateway" +[model_providers.gateway] +name = "Gateway" +base_url = "{}/v1" +[model_providers.other] +name = "Other" +base_url = "{}/v1" +"#, + gateway.uri(), + other.uri(), + ); + let requirements_path = home.path().join("requirements.toml"); + std::fs::write(&requirements_path, &requirements)?; + let mut server = TestAppServer::builder() + .with_codex_home(home.path()) + .with_json_logging("warn") + .build_initialized() + .await?; + let started = server.start_thread(ThreadStartParams::default()).await?; + let input = vec![UserInput::Text { + text: "Continue.".to_string(), + text_elements: Vec::new(), + }]; + let id = server + .send_turn_start_request(TurnStartParams { + thread_id: started.thread.id.clone(), + input: input.clone(), + ..Default::default() + }) + .await?; + let wait = Duration::from_secs(/*secs*/ 60); + let active: TurnStartResponse = timeout(wait, server.read_response(id)).await??; + timeout(wait, gateway.wait_for_request_count(/*count*/ 1)).await?; + let queued: ThreadQueueAddResponse = server + .request(|request_id| ClientRequest::ThreadQueueAdd { + request_id, + params: ThreadQueueAddParams { + thread_id: started.thread.id.clone(), + input: input.clone(), + client_user_message_id: "queued-message".into(), + }, + }) + .await?; + + let id = server + .send_raw_request( + "thread/goal/set", + Some(json!({ + "threadId": started.thread.id, "objective": "Original goal" + })), + ) + .await?; + let _: codex_app_server_protocol::ThreadGoalSetResponse = server.read_response(id).await?; + + let updated = if required_selection == "other" { + requirements.replace("model_provider = \"gateway\"", "model_provider = \"other\"") + } else { + requirements.replace(gateway.uri(), &other.uri()) + }; + std::fs::write(requirements_path, updated)?; + let expected_error = 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, + }; + for (method, params) in [ + ( + "thread/goal/set", + json!({"threadId": started.thread.id, "objective": "Blocked objective"}), + ), + ( + "turn/start", + json!({ "threadId": started.thread.id, "input": input }), + ), + ( + "turn/steer", + json!({ "threadId": started.thread.id, "input": input, "expectedTurnId": active.turn.id }), + ), + ( + "review/start", + json!({ "threadId": started.thread.id, "target": { "type": "custom", "instructions": "Review this change." }, "delivery": "inline" }), + ), + ( + "review/start", + json!({ "threadId": started.thread.id, "target": { "type": "custom", "instructions": "Review this change." }, "delivery": "detached" }), + ), + ] { + let id = server.send_raw_request(method, Some(params)).await?; + let error = server + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert_eq!(error.error, expected_error); + } + + let id = server + .send_raw_request( + "thread/goal/get", + Some(json!({"threadId": started.thread.id})), + ) + .await?; + let goal: codex_app_server_protocol::ThreadGoalGetResponse = server.read_response(id).await?; + assert_eq!(goal.goal.expect("existing goal").objective, "Original goal"); + let id = server + .send_raw_request( + "thread/goal/set", + Some(json!({"threadId": started.thread.id, "status": "paused"})), + ) + .await?; + let _: codex_app_server_protocol::ThreadGoalSetResponse = server.read_response(id).await?; + let id = server + .send_raw_request( + "thread/goal/clear", + Some(json!({"threadId": started.thread.id})), + ) + .await?; + let _: codex_app_server_protocol::ThreadGoalClearResponse = server.read_response(id).await?; + let id = server + .send_raw_request( + "thread/queue/start", + Some(json!({ + "threadId": started.thread.id, + "queuedSubmissionId": queued.queued_submission.id, + })), + ) + .await?; + let error = server + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert_eq!(error.error, expected_error); + let queue: ThreadQueueListResponse = server + .request(|request_id| ClientRequest::ThreadQueueList { + request_id, + params: ThreadQueueListParams { + thread_id: started.thread.id.clone(), + cursor: None, + limit: None, + }, + }) + .await?; + assert_eq!( + queue, + ThreadQueueListResponse { + data: vec![queued.queued_submission.clone()], + next_cursor: None, + } + ); + let id = server + .send_raw_request( + "thread/queue/delete", + Some(json!({ + "threadId": started.thread.id, "queuedSubmissionId": queued.queued_submission.id + })), + ) + .await?; + let _: codex_app_server_protocol::ThreadQueueDeleteResponse = server.read_response(id).await?; + release_response.send(()).expect("release active response"); + timeout( + wait, + server.read_stream_until_notification_message("turn/completed"), + ) + .await??; + for (method, params) in [ + ( + "thread/compact/start", + json!({"threadId": started.thread.id}), + ), + ( + "thread/goal/set", + json!({"threadId": started.thread.id, "objective": "Blocked new goal"}), + ), + ] { + let id = server.send_raw_request(method, Some(params)).await?; + let error = server + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert_eq!(error.error, expected_error); + } + assert_eq!(gateway.requests().await.len(), 1); + assert!( + other + .received_requests() + .await + .expect("recorded requests") + .is_empty() + ); + gateway.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn local_config_changes_do_not_block_existing_threads() -> Result<()> { + let provider = MockServer::start().await; + let home = TempDir::new()?; + MockResponsesConfig::new(&provider.uri()).write(home.path())?; + let mut server = TestAppServer::builder() + .with_codex_home(home.path()) + .build_initialized() + .await?; + let started = server.start_thread(ThreadStartParams::default()).await?; + let requests = responses::mount_sse_sequence( + &provider, + vec![ + responses::sse(vec![responses::ev_completed("first")]), + responses::sse(vec![responses::ev_completed("second")]), + ], + ) + .await; + for local_settings in ["model_provider = 'openai'", "invalid toml !!!"] { + std::fs::write(home.path().join("config.toml"), local_settings)?; + let id = server + .send_turn_start_request(TurnStartParams { + thread_id: started.thread.id.clone(), + input: vec![UserInput::Text { + text: "Continue on the original provider.".into(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let _: TurnStartResponse = server.read_response(id).await?; + server + .read_stream_until_notification_message("turn/completed") + .await?; + } + assert_eq!(requests.requests().len(), 2); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn malformed_system_defaults_do_not_block_existing_thread_turn() -> Result<()> { + let provider = MockServer::start().await; + let home = TempDir::new()?; + MockResponsesConfig::new(&provider.uri()).write(home.path())?; + let system_config_path = home.path().join("system-config.toml"); + let mut overrides = LoaderOverrides::without_managed_config_for_tests(); + overrides.system_config_path = Some(system_config_path.clone()); + let config = Arc::new( + ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .fallback_cwd(Some(home.path().to_path_buf())) + .loader_overrides(overrides.clone()) + .build() + .await?, + ); + let mut client = in_process::start(InProcessStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config, + cli_overrides: Vec::new(), + loader_overrides: overrides, + strict_config: false, + cloud_config_bundle: CloudConfigBundleLoader::default(), + thread_config_loader: Arc::new(NoopThreadConfigLoader), + feedback: CodexFeedback::new(), + log_db: None, + state_db: None, + environment_manager: Arc::new(EnvironmentManager::default_for_tests()), + config_warnings: Vec::new(), + session_source: SessionSource::Cli, + enable_codex_api_key_env: false, + initialize: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-tests".to_string(), + title: None, + version: "0.1.0".to_string(), + }, + capabilities: None, + }, + channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await?; + let response = client + .request(ClientRequest::ThreadStart { + request_id: RequestId::Integer(1), + params: ThreadStartParams::default(), + }) + .await? + .expect("thread/start should succeed"); + let started: ThreadStartResponse = serde_json::from_value(response)?; + + let requests = responses::mount_sse_once( + &provider, + responses::sse(vec![responses::ev_completed("response-1")]), + ) + .await; + std::fs::write(system_config_path, "invalid toml !!!")?; + let response = client + .request(ClientRequest::TurnStart { + request_id: RequestId::Integer(2), + params: TurnStartParams { + thread_id: started.thread.id.clone(), + input: vec![UserInput::Text { + text: "Continue on the original provider.".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + }) + .await? + .expect("turn/start should succeed"); + let _: TurnStartResponse = serde_json::from_value(response)?; + timeout(Duration::from_secs(/*secs*/ 60), async { + loop { + let Some(event) = client.next_event().await else { + anyhow::bail!("app-server stopped before turn/completed"); + }; + if let InProcessServerEvent::ServerNotification(notification) = event + && let ServerNotification::TurnCompleted(completed) = notification.as_ref() + && completed.thread_id == started.thread.id + { + return Ok::<(), anyhow::Error>(()); + } + } + }) + .await??; + assert_eq!(requests.requests().len(), 1); + client.shutdown().await?; + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/remote_control.rs b/codex-rs/app-server/tests/suite/v2/remote_control.rs index 99b343eba3..85ad2c8fae 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_control.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_control.rs @@ -956,6 +956,7 @@ struct BlockingRemoteControlBackend { struct ConnectedRemoteControlBackend { initialized_rx: Option>>, server_task: JoinHandle>, + _models_server: wiremock::MockServer, } struct ClientManagementRemoteControlBackend { @@ -966,6 +967,20 @@ struct ClientManagementRemoteControlBackend { impl ConnectedRemoteControlBackend { async fn start(codex_home: &std::path::Path) -> Result { let listener = configured_remote_control_listener(codex_home).await?; + // Model refreshes can arrive after enrollment, when this listener expects a WebSocket. + let models_server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .and(wiremock::matchers::path("/v1/models")) + .respond_with( + wiremock::ResponseTemplate::new(/*s*/ 200) + .set_body_json(serde_json::json!({ "models": [] })), + ) + .mount(&models_server) + .await; + let remote_control_url = format!("http://{}/backend-api/", listener.local_addr()?); + MockResponsesConfig::new(&models_server.uri()) + .with_root_config(&format!("chatgpt_base_url = \"{remote_control_url}\"")) + .write(codex_home)?; let (initialized_tx, initialized_rx) = oneshot::channel(); let server_task = tokio::spawn(async move { let mut initialized_tx = Some(initialized_tx); @@ -1061,6 +1076,7 @@ impl ConnectedRemoteControlBackend { Ok(Self { initialized_rx: Some(initialized_rx), server_task, + _models_server: models_server, }) } diff --git a/codex-rs/app-server/tests/suite/v2/review.rs b/codex-rs/app-server/tests/suite/v2/review.rs index 0ae2b5da1b..077590aa9c 100644 --- a/codex-rs/app-server/tests/suite/v2/review.rs +++ b/codex-rs/app-server/tests/suite/v2/review.rs @@ -362,7 +362,8 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<( .await; let codex_home = TempDir::new()?; - create_config_toml(codex_home.path(), &server.uri())?; + let startup_provider = responses::start_mock_server().await; + create_config_toml(codex_home.path(), &startup_provider.uri())?; let colliding_skill_dir = codex_home.path().join("skills/review-agent-collision"); std::fs::create_dir_all(&colliding_skill_dir)?; std::fs::write( @@ -384,6 +385,14 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<( .with_codex_home(codex_home.path()) .build_initialized() .await?; + // New threads use the refreshed route; detached review must inherit that route. + std::fs::write( + codex_home.path().join("requirements.toml"), + format!( + "model_provider = 'review-gateway'\n[model_providers.review-gateway]\nname = 'Review Gateway'\nbase_url = '{}/v1'\n", + server.uri() + ), + )?; let ThreadStartResponse { thread, .. } = mcp .start_thread(ThreadStartParams { model: Some("mock-model".to_string()), diff --git a/codex-rs/config/src/cloud_config_bundle.rs b/codex-rs/config/src/cloud_config_bundle.rs index 7982d11ea4..847b67725e 100644 --- a/codex-rs/config/src/cloud_config_bundle.rs +++ b/codex-rs/config/src/cloud_config_bundle.rs @@ -54,6 +54,29 @@ pub struct CloudRequirementsTomlBundle { pub enterprise_managed: Vec, } +impl CloudRequirementsTomlBundle { + pub(crate) fn into_layers(self, base_dir: &AbsolutePathBuf) -> Vec { + let Self { enterprise_managed } = self; + let mut layers = enterprise_managed + .into_iter() + .map(|fragment| { + RequirementsLayerEntry::from_toml( + RequirementSource::EnterpriseManaged { + id: fragment.id, + name: fragment.name, + }, + fragment.contents, + ) + .with_base_dir(base_dir.clone()) + }) + .collect::>(); + // Bundle fragments arrive highest-priority first, while requirements + // layers are merged lowest-priority to highest-priority. + layers.reverse(); + layers + } +} + #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct CloudRequirementsFragment { pub id: String, @@ -100,10 +123,7 @@ impl CloudConfigBundleLayers { CloudConfigTomlBundle { enterprise_managed: config_enterprise_managed, }, - requirements_toml: - CloudRequirementsTomlBundle { - enterprise_managed: requirements_enterprise_managed, - }, + requirements_toml, } = bundle; let enterprise_managed_config = if strict_config { @@ -112,22 +132,7 @@ impl CloudConfigBundleLayers { cloud_config_layers_from_fragments(config_enterprise_managed, base_dir)? }; - let mut enterprise_managed_requirements = requirements_enterprise_managed - .into_iter() - .map(|fragment| { - RequirementsLayerEntry::from_toml( - RequirementSource::EnterpriseManaged { - id: fragment.id, - name: fragment.name, - }, - fragment.contents, - ) - .with_base_dir(base_dir.clone()) - }) - .collect::>(); - // Bundle fragments arrive highest-priority first, while requirements - // layers are merged lowest-priority to highest-priority. - enterprise_managed_requirements.reverse(); + let enterprise_managed_requirements = requirements_toml.into_layers(base_dir); Ok(Self { enterprise_managed_config, diff --git a/codex-rs/config/src/loader/managed_requirements.rs b/codex-rs/config/src/loader/managed_requirements.rs new file mode 100644 index 0000000000..c3efe1abee --- /dev/null +++ b/codex-rs/config/src/loader/managed_requirements.rs @@ -0,0 +1,121 @@ +//! Loads managed requirements independently of user and project configuration, +//! retaining source layers for diagnostics during full configuration loads. + +use super::layer_io; +use super::layer_io::LoadedConfigLayers; +use super::load_requirements_toml; +#[cfg(target_os = "macos")] +use super::macos; +use super::requirements_layers_from_legacy_scheme; +use super::system_requirements_toml_file_with_overrides; +use crate::ConfigRequirements; +use crate::ConfigRequirementsToml; +use crate::RequirementsLayerEntry; +use crate::compose_requirements; +use crate::config_requirements::ConfigRequirementsWithSources; +use crate::state::ConfigLoadOptions; +use crate::state::LoaderOverrides; +use codex_file_system::ExecutorFileSystem; +use codex_utils_absolute_path::AbsolutePathBuf; +use std::io; +use std::path::Path; + +/// Loads current managed requirements without reading system defaults, user or +/// project settings, or thread-provided configuration. Legacy managed config +/// remains a requirements source. +pub async fn load_managed_requirements_state( + fs: &dyn ExecutorFileSystem, + codex_home: &Path, + options: impl Into, +) -> io::Result { + let ConfigLoadOptions { + loader_overrides: overrides, + strict_config, + cloud_config_bundle, + } = options.into(); + if overrides.ignore_managed_requirements { + return Ok(ConfigRequirementsToml::default()); + } + + let mut bundle_requirements = Vec::new(); + if let Some(bundle) = cloud_config_bundle.get().await.map_err(io::Error::other)? { + let base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; + bundle_requirements = bundle.requirements_toml.into_layers(&base_dir); + } + let (requirements, _, _) = load_requirements_from_sources( + fs, + codex_home, + &overrides, + strict_config, + bundle_requirements, + ) + .await?; + let _: ConfigRequirements = requirements.clone().try_into()?; + let requirements_toml = requirements.into_toml(); + // Full config loads also validate required provider definitions when + // constructing ConfigLayerStack, including providers not in use by this thread. + let _ = crate::model_provider_requirements::to_config(&requirements_toml)?; + Ok(requirements_toml) +} + +/// Composes the same requirements for full config loads and policy-only checks. +/// The loaded legacy config and source layers are also returned for the full +/// config stack and its ignored-field diagnostics. +pub(super) async fn load_requirements_from_sources( + fs: &dyn ExecutorFileSystem, + codex_home: &Path, + overrides: &LoaderOverrides, + strict_config: bool, + bundle_requirements: Vec, +) -> io::Result<( + ConfigRequirementsWithSources, + LoadedConfigLayers, + Vec, +)> { + let mut requirements_layers = Vec::new(); + let mut system_requirements_layer = None; + let managed_preferences_requirements_layer; + if !overrides.ignore_managed_requirements { + #[cfg(target_os = "macos")] + { + let base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; + managed_preferences_requirements_layer = macos::load_managed_admin_requirements_layer( + overrides + .macos_managed_config_requirements_base64 + .as_deref(), + ) + .await? + .map(|layer| layer.with_base_dir(base_dir)); + } + #[cfg(not(target_os = "macos"))] + { + managed_preferences_requirements_layer = None; + } + + let requirements_file = system_requirements_toml_file_with_overrides(overrides)?; + system_requirements_layer = load_requirements_toml(fs, &requirements_file).await?; + } else { + managed_preferences_requirements_layer = None; + } + + let loaded_config_layers = + layer_io::load_config_layers_internal(fs, codex_home, overrides.clone(), strict_config) + .await?; + if !overrides.ignore_managed_requirements { + requirements_layers.extend(system_requirements_layer); + requirements_layers.extend(bundle_requirements); + // Legacy managed_config.toml contributes approval and sandbox requirements. + requirements_layers.extend(requirements_layers_from_legacy_scheme( + loaded_config_layers.clone(), + codex_home, + )?); + requirements_layers.extend(managed_preferences_requirements_layer); + } + + let mut requirements = compose_requirements(requirements_layers.clone())?.unwrap_or_default(); + if overrides.ignore_login_requirements { + requirements.allowed_login_methods = None; + requirements.allowed_chatgpt_workspaces = None; + } + Ok((requirements, loaded_config_layers, requirements_layers)) +} diff --git a/codex-rs/config/src/loader/mod.rs b/codex-rs/config/src/loader/mod.rs index 6f6a7bef03..d375e988fe 100644 --- a/codex-rs/config/src/loader/mod.rs +++ b/codex-rs/config/src/loader/mod.rs @@ -2,6 +2,7 @@ mod layer_io; mod local; #[cfg(target_os = "macos")] mod macos; +mod managed_requirements; mod project_discovery; #[cfg(test)] mod tests; @@ -14,7 +15,6 @@ use crate::CloudConfigBundleLayers; use crate::ConfigLayerSource; use crate::ProfileV2Name; use crate::RequirementsLayerEntry; -use crate::compose_requirements; use crate::config_requirements::RequirementSource; use crate::config_requirements::SandboxModeRequirement; use crate::config_toml::ConfigToml; @@ -61,6 +61,7 @@ pub use local::LocalConfigLayers; pub use local::LocalTomlLayer; pub use local::LocalTomlLayerStack; pub use local::load_local_config_layers; +pub use managed_requirements::load_managed_requirements_state; #[cfg(windows)] pub use windows::WindowsSystemConfigNamespaceProbe; #[cfg(windows)] @@ -190,80 +191,40 @@ pub async fn load_config_layers_state( ) }; let active_user_profile = overrides.user_config_profile.clone(); - let ignore_managed_requirements = overrides.ignore_managed_requirements; let ignore_user_config = overrides.ignore_user_config; let ignore_project_config = overrides.ignore_project_config; let ignore_user_and_project_exec_policy_rules = overrides.ignore_user_and_project_exec_policy_rules; - let mut requirements_layers = Vec::new(); let mut bundle_requirements_layers = Vec::new(); - let mut system_requirements_layer = None; - let managed_preferences_requirements_layer; let mut cloud_config_layers = Vec::new(); - if !ignore_managed_requirements { - if let Some(bundle) = cloud_config_bundle.get().await.map_err(io::Error::other)? { - let cloud_config_base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; - let bundle_layers = if strict_config { - CloudConfigBundleLayers::from_bundle_strict_config(bundle, &cloud_config_base_dir)? - } else { - CloudConfigBundleLayers::from_bundle(bundle, &cloud_config_base_dir)? - }; - let CloudConfigBundleLayers { - enterprise_managed_config, - enterprise_managed_requirements, - } = bundle_layers; - bundle_requirements_layers = enterprise_managed_requirements; - cloud_config_layers = enterprise_managed_config; - } - - #[cfg(target_os = "macos")] - { - let managed_preferences_base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; - managed_preferences_requirements_layer = macos::load_managed_admin_requirements_layer( - overrides - .macos_managed_config_requirements_base64 - .as_deref(), - ) - .await? - .map(|layer| layer.with_base_dir(managed_preferences_base_dir)); - } - #[cfg(not(target_os = "macos"))] - { - managed_preferences_requirements_layer = None; - } - - // Honor the system requirements.toml location. - let requirements_toml_file = system_requirements_toml_file_with_overrides(&overrides)?; - system_requirements_layer = load_requirements_toml(fs, &requirements_toml_file).await?; - } else { - managed_preferences_requirements_layer = None; + if !overrides.ignore_managed_requirements + && let Some(bundle) = cloud_config_bundle.get().await.map_err(io::Error::other)? + { + let cloud_config_base_dir = AbsolutePathBuf::from_absolute_path(codex_home)?; + let bundle_layers = if strict_config { + CloudConfigBundleLayers::from_bundle_strict_config(bundle, &cloud_config_base_dir)? + } else { + CloudConfigBundleLayers::from_bundle(bundle, &cloud_config_base_dir)? + }; + let CloudConfigBundleLayers { + enterprise_managed_config, + enterprise_managed_requirements, + } = bundle_layers; + bundle_requirements_layers = enterprise_managed_requirements; + cloud_config_layers = enterprise_managed_config; } - let loaded_config_layers = - layer_io::load_config_layers_internal(fs, codex_home, overrides.clone(), strict_config) - .await?; - let mut startup_warnings = loaded_config_layers.startup_warnings.clone(); - if !ignore_managed_requirements { - requirements_layers.extend(system_requirements_layer); - requirements_layers.extend(bundle_requirements_layers); - // Continue to support loaded legacy `managed_config.toml` sources as - // requirements layers for backwards compatibility. - requirements_layers.extend(requirements_layers_from_legacy_scheme( - loaded_config_layers.clone(), + let (config_requirements_toml, loaded_config_layers, requirements_layers) = + managed_requirements::load_requirements_from_sources( + fs, codex_home, - )?); - requirements_layers.extend(managed_preferences_requirements_layer); - } - - let mut config_requirements_toml = - compose_requirements(requirements_layers.clone())?.unwrap_or_default(); - // Remote app servers enforce auth policy for their workspaces; do not let local - // requirements reintroduce authentication restrictions for those workspaces. - if overrides.ignore_login_requirements { - config_requirements_toml.allowed_login_methods = None; - config_requirements_toml.allowed_chatgpt_workspaces = None; - } + &overrides, + strict_config, + bundle_requirements_layers, + ) + .await?; + let mut startup_warnings = loaded_config_layers.startup_warnings.clone(); let thread_config_context = ThreadConfigContext { thread_id: None,