diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 06fc7c636f..7f6d6c2666 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -907,6 +907,9 @@ "steer": { "type": "boolean" }, + "step_model_switching": { + "type": "boolean" + }, "telepathy": { "type": "boolean" }, @@ -5916,6 +5919,9 @@ "steer": { "type": "boolean" }, + "step_model_switching": { + "type": "boolean" + }, "telepathy": { "type": "boolean" }, diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index eaf2826c84..99bb8ae95a 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -35,6 +35,7 @@ pub(crate) use approval_request::GuardianNetworkAccessTrigger; #[cfg(test)] pub(crate) use approval_request::guardian_approval_request_to_json; pub(crate) use prompt::BUNDLED_GUARDIAN_POLICY; +pub(crate) use prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE; pub(crate) use prompt::guardian_truncate_text; pub(crate) use review::GuardianReviewOptions; pub(crate) use review::guardian_timeout_message; diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 008370baef..5e97f258f1 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -817,7 +817,7 @@ For anything else, use this JSON schema: } pub(crate) const BUNDLED_GUARDIAN_POLICY: &str = include_str!("policy.md"); -pub(super) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str = include_str!("policy_template.md"); +pub(crate) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str = include_str!("policy_template.md"); const TENANT_POLICY_CONFIG_PLACEHOLDER: &str = "{{ tenant_policy_config }}"; /// Guardian policy prompt. diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index de495d99c9..98d02580a4 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -596,6 +596,15 @@ pub(super) async fn submission_loop( thread_settings::update(&sess, sub.id.clone(), thread_settings).await; false } + Op::TurnSettings { + turn_id, + update, + reply, + } => { + let outcome = sess.apply_turn_settings(&turn_id, update).await; + let _ = reply.send(outcome); + false + } Op::InterAgentCommunication { communication } => { inter_agent_communication( &sess, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 2373476fbc..bf07f114d9 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -229,6 +229,7 @@ mod rollout_budget; mod rollout_reconstruction; #[allow(clippy::module_inception)] pub(crate) mod session; +mod step_activation; pub(crate) mod step_context; pub(crate) mod step_settings; mod thread_settings; diff --git a/codex-rs/core/src/session/step_activation.rs b/codex-rs/core/src/session/step_activation.rs new file mode 100644 index 0000000000..ffbe0ebd19 --- /dev/null +++ b/codex-rs/core/src/session/step_activation.rs @@ -0,0 +1,385 @@ +//! Restricted updates to a running turn's immutable settings snapshots. + +use super::session::Session; +use super::session::SessionConfiguration; +use super::step_settings::ResolvedStepSettings; +use super::step_settings::StepSettingsConstraints; +use super::step_settings::StepSettingsUpdate; +use super::turn_context::TurnContext; +use crate::config::Config; +use crate::config::ConstraintResult; +use crate::exec_policy::AllowPrefixRules; +use crate::guardian::BUNDLED_GUARDIAN_POLICY_TEMPLATE; +use codex_features::Feature; +use codex_protocol::openai_models::GuardianV2ModelConfig; +use codex_protocol::openai_models::GuardianV2TranscriptModelConfig; +use codex_protocol::openai_models::MODEL_SPECIALTY_CYBER; +use codex_protocol::openai_models::ModelInfo; +use codex_protocol::protocol::TurnSettingsUpdate; +use codex_protocol::protocol::TurnSettingsUpdateOutcome; +use std::sync::Arc; + +/// Temporary restrictions while approvals and Guardian still read the admitted +/// `TurnContext`. Ordinary live authorization is validated separately. Remove +/// these restrictions as their consumers migrate to captured step settings. +fn check_legacy_turn_safety( + turn_context: &TurnContext, + current: &ResolvedStepSettings, + destination: &ResolvedStepSettings, + live_config: &Config, +) -> Result<(), String> { + let stack = &live_config.config_layer_stack; + let requirements = stack.requirements(); + let required_review = requirements + .auto_review_required_for_model(destination.selected_collaboration_mode().model()); + let admitted_required_review = turn_context + .config + .config_layer_stack + .requirements() + .auto_review_required_for_model(&turn_context.model_info().slug); + let ignored_models = stack + .requirements_toml() + .auto_review + .as_ref() + .and_then(|review| review.ignore_rules.as_ref()); + let ignores_prefix_rules = |model: &ModelInfo| { + model.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER) + || ignored_models.is_some_and(|models| models.contains(&model.slug)) + }; + + // tools::approvals and guardian::review still route approvals using the + // originating turn's policy, reviewer, and required-model classification. + if destination.constrained_approval_policy() != current.constrained_approval_policy() + || destination.approval_policy() != turn_context.approval_policy() + { + return Err("the destination changes the admitted approval policy".to_string()); + } + if destination.approvals_reviewer() != current.approvals_reviewer() + || destination.approvals_reviewer() != turn_context.config.approvals_reviewer + { + return Err("the destination changes the admitted approvals reviewer".to_string()); + } + if required_review + != requirements + .auto_review_required_for_model(current.selected_collaboration_mode().model()) + || required_review != admitted_required_review + { + return Err("the destination changes model-required approval authority".to_string()); + } + // Command approval continues to use TurnContext::allow_prefix_rules. + if ignores_prefix_rules(&destination.model_info) != ignores_prefix_rules(¤t.model_info) + || ignores_prefix_rules(&destination.model_info) + != (turn_context.allow_prefix_rules() == AllowPrefixRules::IgnoreForCyberModel) + { + return Err("the destination changes the admitted prefix-rule policy".to_string()); + } + + check_legacy_model_safety( + turn_context.model_info(), + ¤t.model_info, + &destination.model_info, + &turn_context.config, + live_config, + ) +} + +/// Model-owned portion of the temporary legacy-turn safety check. Ordinary +/// model metadata may differ so diagnostics can expose unmigrated consumers. +fn check_legacy_model_safety( + admitted: &ModelInfo, + current: &ModelInfo, + destination: &ModelInfo, + admitted_config: &Config, + live_config: &Config, +) -> Result<(), String> { + if admitted.used_fallback_model_metadata || current.used_fallback_model_metadata { + return Err("the active model has only fallback metadata".to_string()); + } + if destination.used_fallback_model_metadata { + return Err("the destination model has only fallback metadata".to_string()); + } + let retained_models = [admitted, current]; + // guardian::review::record_guardian_denial still selects its circuit-breaker + // policy from the admitted model's Cyber classification. + let destination_is_cyber = + destination.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER); + if retained_models.iter().any(|model| { + (model.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER)) != destination_is_cyber + }) { + return Err("the destination changes the admitted Guardian rejection policy".to_string()); + } + // TurnMetadataState pins both node REPL flags. Guardian prompt/evidence + // construction also reads node_repl_auto_review_required from the turn. + if retained_models.iter().any(|model| { + model.node_repl_auto_review_required != destination.node_repl_auto_review_required + }) { + return Err( + "the destination changes the admitted node REPL review requirement".to_string(), + ); + } + if retained_models + .iter() + .any(|model| model.node_repl_disabled != destination.node_repl_disabled) + { + return Err( + "the destination changes the admitted node REPL availability restriction".to_string(), + ); + } + // guardian::review::guardian_review_session_config and Guardian V2 still + // select the reviewer from the retained parent metadata. + if retained_models + .iter() + .any(|model| model.auto_review_model_override != destination.auto_review_model_override) + { + return Err("the destination changes the explicit Guardian reviewer model".to_string()); + } + + if admitted_config.features.enabled(Feature::GuardianV2) + && admitted_config.features.enabled(Feature::GuardianApproval) + { + // GuardianV2Extension::on_tool_start reads the parent ModelInfo from + // thread_store. Its classifier settings are independent of the reviewer + // override. Local overrides may mask differences, but resolving those + // overrides remains the extension's responsibility. + let classification_settings = + |model: &ModelInfo| -> GuardianV2ModelConfig { + let mut settings = model + .model_messages + .as_ref() + .and_then(|messages| messages.guardian_v2.clone()) + .unwrap_or_default(); + // Missing and empty transcript records supply the same defaults. + if settings.transcript.as_ref().is_some_and(|transcript| { + *transcript == GuardianV2TranscriptModelConfig::default() + }) { + settings.transcript = None; + } + settings + }; + let destination_settings = classification_settings(destination); + if retained_models + .iter() + .any(|model| classification_settings(model) != destination_settings) + { + return Err( + "the destination changes the admitted Guardian V2 classification settings" + .to_string(), + ); + } + } + + // guardian_review_session_config and Guardian V2 can fall back to parent + // metadata if their preferred reviewer is unavailable, including after a + // catalog refresh. V1 uses the admitted config; V2 can use the live config. + // An unchanged explicit reviewer override prevents both fallback paths. + if destination.auto_review_model_override.is_none() { + for config in [admitted_config, live_config] { + let destination_policy = + config.resolve_guardian_policy(destination.model_messages.as_ref()); + if retained_models.iter().any(|model| { + config.resolve_guardian_policy(model.model_messages.as_ref()) != destination_policy + }) { + return Err( + "the destination changes the Guardian parent-fallback policy".to_string(), + ); + } + } + let destination_template = destination + .model_messages + .as_ref() + .and_then(|messages| messages.auto_review.as_ref()) + .and_then(|messages| messages.policy_template.as_deref()) + .unwrap_or(BUNDLED_GUARDIAN_POLICY_TEMPLATE) + .trim_end(); + if retained_models.iter().any(|model| { + model + .model_messages + .as_ref() + .and_then(|messages| messages.auto_review.as_ref()) + .and_then(|messages| messages.policy_template.as_deref()) + .unwrap_or(BUNDLED_GUARDIAN_POLICY_TEMPLATE) + .trim_end() + != destination_template + }) { + return Err( + "the destination changes the Guardian parent-fallback policy template".to_string(), + ); + } + } + Ok(()) +} + +impl Session { + /// Publishes settings to the named, originally captured live task, regardless + /// of task kind. Publication does not propagate to child sessions or require + /// the task to sample; consumers using initial settings remain unchanged. + /// + /// Callers must serialize updates through completion, including model + /// resolution, so each sparse patch sees the preceding publication. + #[expect( + clippy::await_holding_invalid_type, + reason = "the final managed-policy check and active settings publication must remain atomic" + )] + pub(super) async fn apply_turn_settings( + &self, + turn_id: &str, + update: TurnSettingsUpdate, + ) -> TurnSettingsUpdateOutcome { + if !self.features.enabled(Feature::StepModelSwitching) { + return TurnSettingsUpdateOutcome::Rejected { + reason: "turn settings updates require the step_model_switching feature" + .to_string(), + }; + } + + // Capture the exact live task and its settings, then release the + // lock. A task that starts during preparation is never a new target. + let target = { + let active = self.active_turn.lock().await; + active.as_ref().and_then(|active| { + active.task.as_ref().and_then(|task| { + (task.turn_context.sub_id == turn_id && !task.cancellation_token.is_cancelled()) + .then(|| { + ( + Arc::clone(&task.turn_context), + Arc::clone(&task.done), + task.turn_context.current_settings.load_full(), + ) + }) + }) + }) + }; + let Some((turn_context, task_done, current)) = target else { + return TurnSettingsUpdateOutcome::TargetUnavailable; + }; + let TurnSettingsUpdate { + model, + effort, + summary, + service_tier, + } = update; + let update = StepSettingsUpdate { + model, + effort, + reasoning_summary: summary, + service_tier, + ..Default::default() + }; + // Apply the sparse patch to the captured active base using the shared + // settings rules. The task can progress, finish, or be cancelled while + // preparation awaits; no publication locks are held here. + let prepared = self + .prepare_step_settings_activation(&turn_context, ¤t, &update) + .await; + let active = self.active_turn.lock().await; + let Some(task) = active.as_ref().and_then(|active| active.task.as_ref()) else { + return TurnSettingsUpdateOutcome::TargetUnavailable; + }; + // A later task may reuse the same context and turn ID. `done` is + // allocated per task, so matching only the ID/context is insufficient. + // A mismatch abandons the update without retrying or retargeting. + if !Arc::ptr_eq(&task.done, &task_done) + || !Arc::ptr_eq(&task.turn_context, &turn_context) + || !Arc::ptr_eq(&task.turn_context.current_settings.load_full(), ¤t) + || task.cancellation_token.is_cancelled() + { + return TurnSettingsUpdateOutcome::TargetUnavailable; + } + let destination = match prepared { + Ok(destination) => destination, + Err(reason) => return TurnSettingsUpdateOutcome::Rejected { reason }, + }; + // Managed requirements can change during resolution. Keep the live + // authorization and safety checks atomic with publication under state + // and active_turn; no asynchronous preparation runs under these locks. + let state = self.state.lock().await; + if let Err(reason) = self + .validate_active_step_settings( + &turn_context, + &destination, + &state.session_configuration, + ) + .map_err(|error| error.to_string()) + .and_then(|()| { + check_legacy_turn_safety( + &turn_context, + ¤t, + &destination, + &state.session_configuration.original_config_do_not_use, + ) + }) + { + return TurnSettingsUpdateOutcome::Rejected { reason }; + } + // Publish the immutable snapshot. Frozen initial settings, existing step + // captures, and future thread settings are not changed. + task.turn_context + .current_settings + .store(Arc::new(destination)); + TurnSettingsUpdateOutcome::Applied + } + + async fn prepare_step_settings_activation( + &self, + turn_context: &TurnContext, + current: &ResolvedStepSettings, + update: &StepSettingsUpdate, + ) -> Result { + let (requirements, overrides, trusted_guardian_reviewer) = { + let state = self.state.lock().await; + let configuration = &state.session_configuration; + let stack = &configuration.original_config_do_not_use.config_layer_stack; + ( + stack.requirements().clone(), + configuration.model_info_overrides.clone(), + configuration.trusted_guardian_reviewer, + ) + }; + let constraints = StepSettingsConstraints { + requirements: &requirements, + guardian_approval_enabled: self.features.enabled(Feature::GuardianApproval), + trusted_guardian_reviewer, + has_full_disk_write_access: turn_context + .file_system_sandbox_policy() + .has_full_disk_write_access(), + }; + current + .apply_update( + update, + &constraints, + self.services.models_manager.as_ref(), + &overrides, + self.features.enabled(Feature::Personality), + self.features.enabled(Feature::FastMode), + ) + .await + .map_err(|error| error.to_string()) + } + + /// Rechecks ordinary managed authorization after asynchronous resolution. + /// Unlike the temporary legacy-turn check, these requirements also apply + /// once all execution consumers read their captured `StepContext`. + fn validate_active_step_settings( + &self, + turn_context: &TurnContext, + settings: &ResolvedStepSettings, + configuration: &SessionConfiguration, + ) -> ConstraintResult<()> { + let requirements = configuration + .original_config_do_not_use + .config_layer_stack + .requirements(); + settings.revalidate(&StepSettingsConstraints { + requirements, + guardian_approval_enabled: self.features.enabled(Feature::GuardianApproval), + trusted_guardian_reviewer: configuration.trusted_guardian_reviewer, + has_full_disk_write_access: turn_context + .file_system_sandbox_policy() + .has_full_disk_write_access(), + }) + } +} + +#[cfg(test)] +#[path = "step_activation_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/session/step_activation_tests.rs b/codex-rs/core/src/session/step_activation_tests.rs new file mode 100644 index 0000000000..3fc026b3ca --- /dev/null +++ b/codex-rs/core/src/session/step_activation_tests.rs @@ -0,0 +1,1142 @@ +use super::*; +use crate::guardian::BUNDLED_GUARDIAN_POLICY; +use crate::session::TurnInput; +use crate::session::handlers::submission_loop; +use crate::session::step_context::StepContext; +use crate::session::step_settings::StepSettings; +use crate::session::tests::make_session_and_context; +use crate::session::tests::update_selected_settings_for_test; +use crate::session::tests::update_turn_settings_for_test; +use crate::state::TaskKind; +use crate::tasks::SessionTask; +use crate::tasks::SessionTaskResult; +use codex_config::AutoReviewRequirementsToml; +use codex_config::ConfigLayerStack; +use codex_config::ConfigRequirements; +use codex_config::ConfigRequirementsToml; +use codex_config::ConfigRequirementsWithSources; +use codex_config::RequirementSource; +use codex_config::Sourced; +use codex_http_client::HttpClientFactory; +use codex_login::AuthManager; +use codex_models_manager::ModelsManagerConfig; +use codex_models_manager::bundled_models_response; +use codex_models_manager::manager::ModelsManager; +use codex_models_manager::manager::ModelsManagerFuture; +use codex_models_manager::manager::RefreshStrategy; +use codex_models_manager::manager::StaticModelsManager; +use codex_models_manager::model_info::with_config_overrides; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::config_types::CollaborationModeMask; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::AutoReviewMessages; +use codex_protocol::openai_models::GuardianV2ModelConfig; +use codex_protocol::openai_models::GuardianV2TranscriptModelConfig; +use codex_protocol::openai_models::ModelsResponse; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::Submission; +use codex_protocol::protocol::TurnAbortReason; +use pretty_assertions::assert_eq; +use std::collections::BTreeSet; +use std::sync::Mutex as StdMutex; +use std::time::Duration; +use test_case::test_case; +use tokio::sync::Notify; +use tokio::sync::TryLockError; +use tokio::sync::oneshot; +use tokio::time::timeout; +use tokio_util::sync::CancellationToken; + +const MODEL_A: &str = "step-activation-a"; +const MODEL_B: &str = "step-activation-b"; + +fn activation_models() -> Vec { + let model = bundled_models_response() + .expect("bundled models") + .models + .into_iter() + .find(|model| model.slug == "gpt-5.4") + .expect("bundled gpt-5.4"); + [MODEL_A, MODEL_B] + .into_iter() + .map(|slug| ModelInfo { + slug: slug.to_string(), + ..model.clone() + }) + .collect() +} + +#[derive(Debug, Default)] +struct ModelLookupGate { + started: Notify, + resume: Notify, +} + +impl ModelLookupGate { + async fn wait_until_blocked(&self) { + timeout(Duration::from_secs(/*secs*/ 10), self.started.notified()) + .await + .expect("model lookup started"); + } + + fn release(&self) { + self.resume.notify_one(); + } +} + +#[derive(Debug)] +struct GatedModelsManager { + inner: StaticModelsManager, + first_b_lookup: StdMutex>>, +} + +impl GatedModelsManager { + fn new(models: Vec) -> (Arc, Arc) { + let lookup = Arc::new(ModelLookupGate::default()); + ( + Arc::new(Self { + inner: StaticModelsManager::new( + /*auth_manager*/ None, + ModelsResponse { models }, + ), + first_b_lookup: StdMutex::new(Some(Arc::clone(&lookup))), + }), + lookup, + ) + } +} + +impl ModelsManager for GatedModelsManager { + fn raw_model_catalog( + &self, + strategy: RefreshStrategy, + factory: HttpClientFactory, + ) -> ModelsManagerFuture<'_, ModelsResponse> { + self.inner.raw_model_catalog(strategy, factory) + } + + fn get_remote_models(&self) -> ModelsManagerFuture<'_, Vec> { + self.inner.get_remote_models() + } + + fn try_get_remote_models(&self) -> Result, TryLockError> { + self.inner.try_get_remote_models() + } + + fn auth_manager(&self) -> Option<&AuthManager> { + self.inner.auth_manager() + } + + fn list_collaboration_modes(&self) -> Vec { + self.inner.list_collaboration_modes() + } + + fn refresh_if_new_etag( + &self, + etag: String, + factory: HttpClientFactory, + ) -> ModelsManagerFuture<'_, ()> { + self.inner.refresh_if_new_etag(etag, factory) + } + + fn get_model_info<'a>( + &'a self, + model: &'a str, + config: &'a ModelsManagerConfig, + ) -> ModelsManagerFuture<'a, ModelInfo> { + Box::pin(async move { + let pause = if model == MODEL_B { + self.first_b_lookup.lock().expect("lookup gate").take() + } else { + None + }; + if let Some(pause) = pause { + pause.started.notify_one(); + pause.resume.notified().await; + } + self.inner.get_model_info(model, config).await + }) + } +} + +struct HeldStepTask { + finish: Arc, +} + +impl SessionTask for HeldStepTask { + fn kind(&self) -> TaskKind { + // Publication also works for a non-regular task that never samples. + TaskKind::Compact + } + + fn span_name(&self) -> &'static str { + "session_task.step_activation_test" + } + + async fn run( + self: Arc, + _session: Arc, + _turn: Arc, + _input: Vec, + cancellation_token: CancellationToken, + ) -> SessionTaskResult { + tokio::select! { + _ = cancellation_token.cancelled() => {}, + _ = self.finish.notified() => {}, + } + Ok(None) + } +} + +struct ActivationFixture { + session: Arc, + turn: Arc, + finish: Arc, + lookup: Arc, +} + +async fn activation_fixture(models: Vec) -> ActivationFixture { + let (session, _) = make_session_and_context().await; + let mut session = Arc::new(session); + let mutable = Arc::get_mut(&mut session).expect("unshared test session"); + let (models, lookup) = GatedModelsManager::new(models); + mutable.services.models_manager = models; + for feature in [Feature::StepModelSwitching, Feature::FastMode] { + mutable + .features + .enable(feature) + .expect("enable test feature"); + } + let configuration = &mut mutable.state.get_mut().session_configuration; + let config = Arc::make_mut(&mut configuration.original_config_do_not_use); + config.model = Some(MODEL_A.to_string()); + config.features = mutable.features.clone(); + let settings = Arc::make_mut(&mut configuration.step_settings); + settings.collaboration_mode = settings.collaboration_mode.with_updates( + Some(MODEL_A.to_string()), + Some(Some(ReasoningEffort::Low)), + /*developer_instructions*/ None, + ); + settings.reasoning_summary = Some(ReasoningSummary::Concise); + settings.service_tier = None; + let prepared = session + .new_default_turn_with_sub_id("step-activation-turn".to_string()) + .await; + let turn = Arc::clone(&prepared); + let finish = Arc::new(Notify::new()); + session + .spawn_task( + prepared, + Vec::new(), + HeldStepTask { + finish: Arc::clone(&finish), + }, + ) + .await; + ActivationFixture { + session, + turn, + finish, + lookup, + } +} + +fn step_values( + step: &StepContext, +) -> ( + &str, + Option, + ReasoningSummary, + Option<&str>, +) { + ( + step.settings.model_info.slug.as_str(), + step.settings.reasoning_effort().cloned(), + step.settings.reasoning_summary, + step.settings.service_tier.as_deref(), + ) +} + +async fn desired_step_settings(session: &Session) -> Arc { + Arc::clone( + &session + .state + .lock() + .await + .session_configuration + .step_settings, + ) +} + +fn settings_submission( + id: &str, + turn_id: &str, + update: TurnSettingsUpdate, +) -> (Submission, oneshot::Receiver) { + let (reply, receiver) = oneshot::channel(); + ( + Submission { + id: id.to_string(), + op: Op::TurnSettings { + turn_id: turn_id.to_string(), + update, + reply, + }, + trace: None, + parent_turn_id: None, + root_turn_id: None, + }, + receiver, + ) +} + +#[tokio::test] +async fn submitted_sparse_updates_preserve_captured_steps_and_ordering() { + let mut models = activation_models(); + let destination = models + .iter_mut() + .find(|model| model.slug == MODEL_B) + .expect("destination model"); + destination.context_window = Some(190_000); + destination.max_context_window = Some(190_000); + destination.auto_compact_token_limit = Some(150_000); + destination.default_reasoning_summary = ReasoningSummary::Detailed; + destination + .model_messages + .as_mut() + .expect("model messages") + .instructions_template = Some("Destination-model instructions.".to_string()); + let expected_destination = destination.clone(); + let ActivationFixture { + session, + turn, + lookup, + finish, + } = activation_fixture(models).await; + let model_manager_config = { + let state = session.state.lock().await; + let configuration = &state.session_configuration; + configuration.model_info_overrides.models_manager_config( + configuration.step_settings.personality, + session.features.enabled(Feature::Personality), + ) + }; + let expected_destination = with_config_overrides(expected_destination, &model_manager_config); + let desired = desired_step_settings(&session).await; + let (submissions, receiver) = async_channel::unbounded(); + let loop_task = tokio::spawn(submission_loop( + Arc::clone(&session), + session.get_config().await, + receiver, + )); + let before = session + .capture_step_context(Arc::clone(&turn), &CancellationToken::new()) + .await + .expect("capture initial step"); + let (submission, first_reply) = settings_submission( + "activate-model", + &turn.sub_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + effort: Some(Some(ReasoningEffort::Low)), + ..Default::default() + }, + ); + submissions + .send(submission) + .await + .expect("submit model update"); + lookup.wait_until_blocked().await; + let refresh = session + .mcp_refresh + .acquire() + .await + .expect("MCP refresh lock"); + session.mark_mcp_runtime_dirty(); + let capture_cancel = CancellationToken::new(); + let mut during = Box::pin(tokio::task::unconstrained( + session.capture_step_context(Arc::clone(&turn), &capture_cancel), + )); + // Capture A, then hold asynchronous planning across publication of B. + { + let mut context = std::task::Context::from_waker(futures::task::noop_waker_ref()); + assert!(std::future::Future::poll(during.as_mut(), &mut context).is_pending()); + } + let priority = ServiceTier::Fast.request_value(); + let mut replies = vec![first_reply]; + for (id, update) in [ + ( + "activate-reasoning", + TurnSettingsUpdate { + effort: Some(Some(ReasoningEffort::High)), + summary: Some(ReasoningSummary::Detailed), + ..Default::default() + }, + ), + ( + "activate-tier", + TurnSettingsUpdate { + service_tier: Some(Some(priority.to_string())), + ..Default::default() + }, + ), + ] { + let (submission, reply) = settings_submission(id, &turn.sub_id, update); + submissions.send(submission).await.expect("queue update"); + replies.push(reply); + } + lookup.release(); + // Await each operation's publication result, including the patches queued + // behind the blocked model lookup. + for reply in replies { + assert_eq!( + timeout(Duration::from_secs(/*secs*/ 10), reply) + .await + .expect("settings completion") + .expect("settings reply"), + TurnSettingsUpdateOutcome::Applied, + ); + } + drop(refresh); + let during = during.await.expect("capture spanning activation"); + assert!(Arc::ptr_eq( + &before.settings.model_info, + &during.settings.model_info + )); + let after = session + .capture_step_context(Arc::clone(&turn), &CancellationToken::new()) + .await + .expect("capture published settings"); + let initial = ( + MODEL_A, + Some(ReasoningEffort::Low), + ReasoningSummary::Concise, + None, + ); + assert_eq!( + [ + step_values(&before), + step_values(&during), + step_values(&after) + ], + [ + initial.clone(), + initial, + ( + MODEL_B, + Some(ReasoningEffort::High), + ReasoningSummary::Detailed, + Some(priority), + ), + ] + ); + assert!(Arc::ptr_eq(&before.turn, &after.turn)); + assert_eq!(after.settings.model_info.as_ref(), &expected_destination); + assert_ne!( + before.settings.model_info.context_window, + after.settings.model_info.context_window + ); + assert_eq!(desired_step_settings(&session).await, desired); + assert!(Arc::ptr_eq(&before.settings, &during.settings)); + assert!(Arc::ptr_eq(&before.settings, &turn.initial_settings)); + assert_eq!(turn.model_info().slug, MODEL_A); + + let done = { + let active = session.active_turn.lock().await; + Arc::clone(&active.as_ref().unwrap().task.as_ref().unwrap().done) + }; + let completed = done.notified(); + finish.notify_one(); + timeout(Duration::from_secs(/*secs*/ 10), completed) + .await + .expect("original task completed"); + assert!(session.active_turn.lock().await.is_none()); + + // A retained context owns its last published snapshot even after the task + // is unregistered. Capture must not fall back to the initial turn model. + let retained = session + .capture_step_context(Arc::clone(&turn), &CancellationToken::new()) + .await + .expect("capture retained context after task completion"); + assert!(Arc::ptr_eq(&retained.settings, &after.settings)); + assert_eq!(step_values(&retained), step_values(&after)); + assert_eq!(before.settings.model_info.slug, MODEL_A); + assert_eq!(turn.model_info().slug, MODEL_A); + drop(submissions); + loop_task.await.expect("submission loop teardown"); +} + +#[derive(Clone, Copy)] +enum TaskChangeDuringLookup { + CancelledWithRejectedDestination, + FinishedAndReplaced, + FinishedAndReusedContext, +} + +#[test_case(TaskChangeDuringLookup::CancelledWithRejectedDestination; "cancelled task is unavailable even when destination is rejected")] +#[test_case(TaskChangeDuringLookup::FinishedAndReplaced; "completed task is replaced")] +#[test_case(TaskChangeDuringLookup::FinishedAndReusedContext; "completed context is reused by another task")] +#[tokio::test] +async fn delayed_activation_does_not_retarget_a_task(change: TaskChangeDuringLookup) { + let mut models = activation_models(); + if matches!( + change, + TaskChangeDuringLookup::CancelledWithRejectedDestination + ) { + models + .iter_mut() + .find(|model| model.slug == MODEL_B) + .expect("destination model") + .node_repl_disabled = true; + } + let ActivationFixture { + session, + turn, + finish, + lookup, + } = activation_fixture(models).await; + let desired = desired_step_settings(&session).await; + let original = turn.current_settings.load_full(); + let update_session = Arc::clone(&session); + let turn_id = turn.sub_id.clone(); + let update = tokio::spawn(async move { + update_session + .apply_turn_settings( + &turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await + }); + lookup.wait_until_blocked().await; + assert_eq!(desired_step_settings(&session).await, desired); + let (cancellation_token, done) = { + let active = session.active_turn.lock().await; + let task = active + .as_ref() + .and_then(|active| active.task.as_ref()) + .expect("active task"); + (task.cancellation_token.clone(), Arc::clone(&task.done)) + }; + let (expected_turn, expected_settings) = match change { + TaskChangeDuringLookup::CancelledWithRejectedDestination => { + cancellation_token.cancel(); + (Arc::clone(&turn), original) + } + TaskChangeDuringLookup::FinishedAndReplaced + | TaskChangeDuringLookup::FinishedAndReusedContext => { + let completed = done.notified(); + finish.notify_one(); + timeout(Duration::from_secs(/*secs*/ 10), completed) + .await + .expect("original task completed"); + let replacement = match change { + TaskChangeDuringLookup::FinishedAndReplaced => { + session + .new_default_turn_with_sub_id("replacement-turn".to_string()) + .await + } + TaskChangeDuringLookup::FinishedAndReusedContext => Arc::clone(&turn), + TaskChangeDuringLookup::CancelledWithRejectedDestination => unreachable!(), + }; + let settings = replacement.current_settings.load_full(); + session + .spawn_task( + Arc::clone(&replacement), + Vec::new(), + HeldStepTask { + finish: Arc::new(Notify::new()), + }, + ) + .await; + assert!(Arc::ptr_eq(&turn.current_settings.load_full(), &original)); + (replacement, settings) + } + }; + lookup.release(); + assert_eq!( + update.await.expect("activation task"), + TurnSettingsUpdateOutcome::TargetUnavailable + ); + assert!(Arc::ptr_eq( + &expected_turn.current_settings.load_full(), + &expected_settings, + )); + assert_eq!(desired_step_settings(&session).await, desired); + session.abort_all_tasks(TurnAbortReason::Replaced).await; +} + +#[derive(Clone, Copy)] +enum ManagedAuthorizationChange { + ApprovalPolicy, + ApprovalsReviewer, +} + +#[test_case(ManagedAuthorizationChange::ApprovalPolicy; "approval policy refreshed during lookup")] +#[test_case(ManagedAuthorizationChange::ApprovalsReviewer; "approvals reviewer refreshed during lookup")] +#[tokio::test] +async fn delayed_activation_rechecks_live_managed_authorization( + change: ManagedAuthorizationChange, +) { + let ActivationFixture { + session, + turn, + lookup, + .. + } = activation_fixture(activation_models()).await; + let original = turn.current_settings.load_full(); + let desired = desired_step_settings(&session).await; + let update_session = Arc::clone(&session); + let turn_id = turn.sub_id.clone(); + let update = tokio::spawn(async move { + update_session + .apply_turn_settings( + &turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await + }); + lookup.wait_until_blocked().await; + assert_eq!(desired_step_settings(&session).await, desired); + + // Refresh only a live constraint, leaving the admitted settings and the + // legacy model-authority classifications unchanged. + let source = RequirementSource::EnterpriseManaged { + id: "refreshed-policy".to_string(), + name: "Refreshed policy".to_string(), + }; + let sourced = match change { + ManagedAuthorizationChange::ApprovalPolicy => { + let allowed = if original.approval_policy() == AskForApproval::Never { + AskForApproval::OnRequest + } else { + AskForApproval::Never + }; + ConfigRequirementsWithSources { + allowed_approval_policies: Some(Sourced::new(vec![allowed], source)), + ..Default::default() + } + } + ManagedAuthorizationChange::ApprovalsReviewer => { + let allowed = if original.approvals_reviewer() == ApprovalsReviewer::User { + ApprovalsReviewer::AutoReview + } else { + ApprovalsReviewer::User + }; + ConfigRequirementsWithSources { + allowed_approvals_reviewers: Some(Sourced::new(vec![allowed], source)), + ..Default::default() + } + } + }; + let requirements = + ConfigRequirements::try_from(sourced.clone()).expect("normalize refreshed requirements"); + let expected_error = { + let mut state = session.state.lock().await; + let config = Arc::make_mut(&mut state.session_configuration.original_config_do_not_use); + config.config_layer_stack = ConfigLayerStack::new( + config + .config_layer_stack + .all_layers_low_to_high() + .cloned() + .collect(), + requirements, + sourced.into_toml(), + ) + .expect("build refreshed requirements"); + let requirements = config.config_layer_stack.requirements(); + match change { + ManagedAuthorizationChange::ApprovalPolicy => requirements + .approval_policy + .can_set(&original.approval_policy()), + ManagedAuthorizationChange::ApprovalsReviewer => requirements + .approvals_reviewer + .can_set(&original.approvals_reviewer()), + } + .expect_err("the refreshed constraint rejects the admitted value") + }; + lookup.release(); + assert_eq!( + update.await.expect("activation task"), + TurnSettingsUpdateOutcome::Rejected { + reason: expected_error.to_string(), + } + ); + assert!(Arc::ptr_eq(&turn.current_settings.load_full(), &original,)); + assert_eq!(desired_step_settings(&session).await, desired); + session.abort_all_tasks(TurnAbortReason::Replaced).await; +} + +#[derive(Clone, Copy)] +enum ManagedPolicyChange { + RequireReview, + IgnorePrefixRules, +} + +#[test_case(ManagedPolicyChange::RequireReview; "required review changed since admission")] +#[test_case(ManagedPolicyChange::IgnorePrefixRules; "prefix rule policy changed since admission")] +#[tokio::test] +async fn activation_must_match_the_retained_turn_authority(change: ManagedPolicyChange) { + let (mut session, mut turn) = make_session_and_context().await; + session + .features + .enable(Feature::GuardianApproval) + .expect("enable Guardian for the test"); + let config = Arc::make_mut(&mut turn.config); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config.config_layer_stack = ConfigLayerStack::new( + config + .config_layer_stack + .all_layers_low_to_high() + .cloned() + .collect(), + ConfigRequirements::default(), + ConfigRequirementsToml::default(), + ) + .expect("build admitted requirements"); + update_turn_settings_for_test(&mut turn, |settings| { + update_selected_settings_for_test(settings, |selected| { + selected.approvals_reviewer = ApprovalsReviewer::AutoReview; + }); + let model_info = Arc::make_mut(&mut settings.model_info); + model_info.model_specialty = None; + model_info.used_fallback_model_metadata = false; + }); + assert!( + !turn + .file_system_sandbox_policy() + .has_full_disk_write_access() + ); + + let prepared = Arc::new(turn); + let current = &prepared.initial_settings; + let mut model_info = current.model_info.as_ref().clone(); + model_info.slug = "step-settings-policy-destination".to_string(); + let mut selected = current.selected().clone(); + selected.collaboration_mode.settings.model = model_info.slug.clone(); + let destination = ResolvedStepSettings::new( + Arc::new(selected), + Arc::new(model_info), + session.features.enabled(Feature::FastMode), + ); + let mut live = session.state.lock().await.session_configuration.clone(); + live.original_config_do_not_use = Arc::clone(&prepared.config); + assert_eq!( + session.validate_active_step_settings(&prepared, &destination, &live,), + Ok(()) + ); + assert_eq!( + check_legacy_turn_safety( + &prepared, + current, + &destination, + &live.original_config_do_not_use, + ), + Ok(()) + ); + + // Both model names acquire the same live classification. Comparing only + // those names in the latest requirements would therefore accept the switch, + // even though remaining consumers still use the admitted turn's authority. + let models = vec![ + current.model_info.slug.clone(), + destination.model_info.slug.clone(), + ]; + let mut requirements = ConfigRequirements::default(); + let auto_review = match change { + ManagedPolicyChange::RequireReview => { + requirements.auto_review_required_models = Some(Sourced::new( + BTreeSet::from_iter(models.iter().cloned()), + RequirementSource::Unknown, + )); + AutoReviewRequirementsToml { + required_on_models: Some(models), + ..Default::default() + } + } + ManagedPolicyChange::IgnorePrefixRules => AutoReviewRequirementsToml { + ignore_rules: Some(models), + ..Default::default() + }, + }; + let config = Arc::make_mut(&mut live.original_config_do_not_use); + config.config_layer_stack = ConfigLayerStack::new( + config + .config_layer_stack + .all_layers_low_to_high() + .cloned() + .collect(), + requirements, + ConfigRequirementsToml { + auto_review: Some(auto_review), + ..Default::default() + }, + ) + .expect("build refreshed requirements"); + // The destination satisfies today's managed policy. Only the temporary + // compatibility check rejects the mismatch with retained turn consumers. + assert_eq!( + session.validate_active_step_settings(&prepared, &destination, &live,), + Ok(()) + ); + assert_eq!( + check_legacy_turn_safety( + &prepared, + current, + &destination, + &live.original_config_do_not_use, + ), + Err(match change { + ManagedPolicyChange::RequireReview => { + "the destination changes model-required approval authority".to_string() + } + ManagedPolicyChange::IgnorePrefixRules => { + "the destination changes the admitted prefix-rule policy".to_string() + } + }) + ); +} + +fn safety_models() -> (ModelInfo, ModelInfo) { + let [mut admitted, mut destination]: [ModelInfo; 2] = activation_models() + .try_into() + .expect("two activation models"); + for model in [&mut admitted, &mut destination] { + model.model_specialty = None; + model.node_repl_auto_review_required = false; + model.node_repl_disabled = false; + model.auto_review_model_override = None; + model.used_fallback_model_metadata = false; + let messages = model.model_messages.as_mut().expect("model messages"); + messages.auto_review = None; + messages.guardian_v2 = None; + } + (admitted, destination) +} + +fn parent_review_messages(model: &mut ModelInfo) -> &mut AutoReviewMessages { + model + .model_messages + .as_mut() + .expect("model messages") + .auto_review + .get_or_insert(AutoReviewMessages { + policy: None, + policy_template: None, + rejection_instructions: None, + timeout_instructions: None, + }) +} + +#[derive(Clone, Copy)] +enum ModelSafetyChange { + Cyber, + NodeReplReview, + NodeReplDisabled, + ReviewerOverride, +} + +#[test_case(ModelSafetyChange::Cyber, "the destination changes the admitted Guardian rejection policy"; "Cyber rejection policy")] +#[test_case(ModelSafetyChange::NodeReplReview, "the destination changes the admitted node REPL review requirement"; "node REPL review")] +#[test_case(ModelSafetyChange::NodeReplDisabled, "the destination changes the admitted node REPL availability restriction"; "node REPL availability")] +#[test_case(ModelSafetyChange::ReviewerOverride, "the destination changes the explicit Guardian reviewer model"; "explicit reviewer")] +#[tokio::test] +async fn model_safety_changes_must_match_the_admitted_authority( + change: ModelSafetyChange, + reason: &str, +) { + let (_, turn) = make_session_and_context().await; + let (admitted, mut destination) = safety_models(); + match change { + ModelSafetyChange::Cyber => { + destination.model_specialty = Some(MODEL_SPECIALTY_CYBER.to_string()); + } + ModelSafetyChange::NodeReplReview => destination.node_repl_auto_review_required = true, + ModelSafetyChange::NodeReplDisabled => destination.node_repl_disabled = true, + ModelSafetyChange::ReviewerOverride => { + destination.auto_review_model_override = Some("required-reviewer".to_string()); + } + } + for current in [&admitted, &destination] { + assert_eq!( + check_legacy_model_safety(&admitted, current, &destination, &turn.config, &turn.config,), + Err(reason.to_string()) + ); + } +} + +#[tokio::test] +async fn model_safety_rejects_fallback_metadata() { + let (_, turn) = make_session_and_context().await; + let (admitted, destination) = safety_models(); + let mut fallback = admitted.clone(); + fallback.used_fallback_model_metadata = true; + for (old, current, next, reason) in [ + ( + &fallback, + &admitted, + &destination, + "the active model has only fallback metadata", + ), + ( + &admitted, + &fallback, + &destination, + "the active model has only fallback metadata", + ), + ( + &admitted, + &admitted, + &fallback, + "the destination model has only fallback metadata", + ), + ] { + assert_eq!( + check_legacy_model_safety(old, current, next, &turn.config, &turn.config), + Err(reason.to_string()) + ); + } +} + +#[test_case(None, None, false; "both catalog policies")] +#[test_case(Some("admitted policy"), None, false; "live config unmasks catalog")] +#[test_case(None, Some("live policy"), false; "admitted config unmasks catalog")] +#[test_case(Some("admitted policy"), Some("live policy"), true; "both configured policies mask catalog")] +#[tokio::test] +async fn parent_fallback_policy_uses_both_config_lifetimes( + admitted_policy: Option<&str>, + live_policy: Option<&str>, + allowed: bool, +) { + let (_, turn) = make_session_and_context().await; + let mut admitted_config = turn.config.as_ref().clone(); + admitted_config.guardian_policy_config = admitted_policy.map(str::to_string); + let mut live_config = admitted_config.clone(); + live_config.guardian_policy_config = live_policy.map(str::to_string); + let (mut admitted, mut destination) = safety_models(); + parent_review_messages(&mut admitted).policy = Some("catalog policy A".to_string()); + parent_review_messages(&mut destination).policy = Some("catalog policy B".to_string()); + assert_eq!( + check_legacy_model_safety( + &admitted, + &admitted, + &destination, + &admitted_config, + &live_config, + ), + if allowed { + Ok(()) + } else { + Err("the destination changes the Guardian parent-fallback policy".to_string()) + } + ); +} + +#[tokio::test] +async fn parent_fallback_preserves_explicit_empty_and_bundled_defaults() { + let (_, turn) = make_session_and_context().await; + let mut config = turn.config.as_ref().clone(); + config.guardian_policy_config = None; + let (admitted, mut destination) = safety_models(); + let check = |destination: &ModelInfo| { + check_legacy_model_safety(&admitted, &admitted, destination, &config, &config) + }; + parent_review_messages(&mut destination).policy = Some(BUNDLED_GUARDIAN_POLICY.to_string()); + parent_review_messages(&mut destination).policy_template = + Some(BUNDLED_GUARDIAN_POLICY_TEMPLATE.to_string()); + assert_eq!(check(&destination), Ok(())); + parent_review_messages(&mut destination).policy_template = Some(String::new()); + assert_eq!( + check(&destination), + Err("the destination changes the Guardian parent-fallback policy template".to_string()) + ); + parent_review_messages(&mut destination).policy_template = None; + parent_review_messages(&mut destination).policy = Some(String::new()); + assert_eq!( + check(&destination), + Err("the destination changes the Guardian parent-fallback policy".to_string()) + ); + parent_review_messages(&mut destination).policy = None; + assert_eq!(check(&destination), Ok(())); +} + +#[tokio::test] +async fn unchanged_explicit_reviewer_does_not_use_parent_policy() { + let (_, turn) = make_session_and_context().await; + let (mut admitted, mut destination) = safety_models(); + for model in [&mut admitted, &mut destination] { + model.auto_review_model_override = Some("shared-reviewer".to_string()); + } + parent_review_messages(&mut admitted).policy = Some("catalog policy A".to_string()); + parent_review_messages(&mut destination).policy = Some("catalog policy B".to_string()); + parent_review_messages(&mut destination).policy_template = Some(String::new()); + assert_eq!( + check_legacy_model_safety( + &admitted, + &admitted, + &destination, + &turn.config, + &turn.config, + ), + Ok(()) + ); +} + +#[test_case(GuardianV2ModelConfig { + classifier_instructions: Some("Review the complete action.".to_string()), + ..Default::default() +}; "classifier instructions")] +#[test_case(GuardianV2ModelConfig { + review_threshold_basis_points: Some(6_000), + ..Default::default() +}; "review threshold")] +#[test_case(GuardianV2ModelConfig { + reasoning_effort: Some(ReasoningEffort::High), + ..Default::default() +}; "classifier effort")] +#[test_case(GuardianV2ModelConfig { + transcript: Some(GuardianV2TranscriptModelConfig { + sources: Some(vec!["reasoning".to_string()]), + ..Default::default() + }), + ..Default::default() +}; "transcript sources")] +#[test_case(GuardianV2ModelConfig { + transcript: Some(GuardianV2TranscriptModelConfig { + max_message_entry_tokens: Some(128), + ..Default::default() + }), + ..Default::default() +}; "transcript bounds")] +#[test_case(GuardianV2ModelConfig { + max_action_tokens: Some(128), + ..Default::default() +}; "action limit")] +#[test_case(GuardianV2ModelConfig { + max_classifier_instruction_tokens: Some(256), + ..Default::default() +}; "classifier instruction limit")] +#[test_case(GuardianV2ModelConfig { + max_parent_compaction_tokens: Some(384), + ..Default::default() +}; "parent compaction limit")] +#[tokio::test] +async fn guardian_v2_classification_settings_match_the_admitted_authority( + destination_settings: GuardianV2ModelConfig, +) { + let (_, turn) = make_session_and_context().await; + let mut config = turn.config.as_ref().clone(); + for feature in [Feature::GuardianV2, Feature::GuardianApproval] { + config + .features + .set_enabled(feature, /*enabled*/ true) + .expect("enable Guardian V2"); + } + // V2 classification does not depend on the default approval reviewer or + // the explicitly selected model used to resolve the security policy. + config.approvals_reviewer = ApprovalsReviewer::User; + let (mut admitted, mut destination) = safety_models(); + destination + .model_messages + .as_mut() + .expect("model messages") + .guardian_v2 = Some(destination_settings); + for reviewer_override in [None, Some("shared-reviewer".to_string())] { + admitted.auto_review_model_override = reviewer_override.clone(); + destination.auto_review_model_override = reviewer_override; + for current in [&admitted, &destination] { + assert_eq!( + check_legacy_model_safety(&admitted, current, &destination, &config, &config,), + Err( + "the destination changes the admitted Guardian V2 classification settings" + .to_string() + ) + ); + } + } +} + +#[test_case(false, true; "Guardian V2 disabled")] +#[test_case(true, false; "Guardian approval disabled")] +#[tokio::test] +async fn inactive_guardian_v2_does_not_restrict_model_defaults( + guardian_v2_enabled: bool, + guardian_approval_enabled: bool, +) { + let (_, turn) = make_session_and_context().await; + let mut admitted_config = turn.config.as_ref().clone(); + admitted_config + .features + .set_enabled(Feature::GuardianV2, guardian_v2_enabled) + .expect("configure admitted Guardian V2"); + admitted_config + .features + .set_enabled(Feature::GuardianApproval, guardian_approval_enabled) + .expect("configure admitted Guardian approval"); + let mut live_config = admitted_config.clone(); + for feature in [Feature::GuardianV2, Feature::GuardianApproval] { + live_config + .features + .set_enabled(feature, /*enabled*/ true) + .expect("enable live Guardian V2"); + } + let (admitted, mut destination) = safety_models(); + let destination_settings = GuardianV2ModelConfig { + review_threshold_basis_points: Some(6_000), + ..Default::default() + }; + destination + .model_messages + .as_mut() + .expect("model messages") + .guardian_v2 = Some(destination_settings); + assert_eq!( + check_legacy_model_safety( + &admitted, + &admitted, + &destination, + &admitted_config, + &live_config, + ), + Ok(()) + ); +} + +#[tokio::test] +async fn guardian_v2_empty_model_defaults_are_equivalent() { + let (_, turn) = make_session_and_context().await; + let mut config = turn.config.as_ref().clone(); + for feature in [Feature::GuardianV2, Feature::GuardianApproval] { + config + .features + .set_enabled(feature, /*enabled*/ true) + .expect("enable Guardian V2"); + } + let (admitted, mut destination) = safety_models(); + destination.context_window = Some(123_456); + for settings in [ + GuardianV2ModelConfig::default(), + GuardianV2ModelConfig { + transcript: Some(GuardianV2TranscriptModelConfig::default()), + ..Default::default() + }, + ] { + destination + .model_messages + .as_mut() + .expect("model messages") + .guardian_v2 = Some(settings); + assert_eq!( + check_legacy_model_safety(&admitted, &admitted, &destination, &config, &config,), + Ok(()) + ); + } +} diff --git a/codex-rs/core/src/session/step_settings.rs b/codex-rs/core/src/session/step_settings.rs index 7d52a2ad0e..2e8fbe6753 100644 --- a/codex-rs/core/src/session/step_settings.rs +++ b/codex-rs/core/src/session/step_settings.rs @@ -101,6 +101,57 @@ impl ResolvedStepSettings { self.selected.personality } + pub(super) fn constrained_approval_policy(&self) -> &Constrained { + &self.selected.approval_policy + } + + /// Applies sparse edits to the retained selection, preserving pinned metadata + /// unless model or personality selection changes, then resolves request values. + pub(super) async fn apply_update( + &self, + update: &StepSettingsUpdate, + constraints: &StepSettingsConstraints<'_>, + models_manager: &dyn ModelsManager, + overrides: &ModelInfoOverrides, + personality_enabled: bool, + fast_mode_enabled: bool, + ) -> ConstraintResult { + let selected = self.selected.apply(update, constraints)?; + let model_info = if selected.collaboration_mode.model() + == self.selected.collaboration_mode.model() + && selected.personality == self.selected.personality + { + Arc::clone(&self.model_info) + } else { + Arc::new( + selected + .resolve_model_info(models_manager, overrides, personality_enabled) + .await, + ) + }; + Ok(Self::new(Arc::new(selected), model_info, fast_mode_enabled)) + } + + /// Rechecks the retained selection against current managed constraints. + pub(super) fn revalidate( + &self, + constraints: &StepSettingsConstraints<'_>, + ) -> ConstraintResult<()> { + // TODO: Revisit the split with StepSettings::validate, which does not + // check inherited approval values against the supplied managed allow-lists. + // Sharing these checks would also make ordinary settings updates stricter + // after a requirements refresh. + self.selected.validate(constraints)?; + constraints + .requirements + .approval_policy + .can_set(&self.approval_policy())?; + constraints + .requirements + .approvals_reviewer + .can_set(&self.approvals_reviewer()) + } + pub(super) fn telemetry(&self, base: &SessionTelemetry) -> SessionTelemetry { base.clone().with_model( self.selected.collaboration_mode.model(), diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index c60d695a96..584890e37a 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -153,6 +153,7 @@ mod skills; mod skills_extension; mod spawn_agent_description; mod sqlite_state; +mod step_settings; mod step_settings_snapshots; mod stream_error_allows_next_turn; mod stream_no_completed; diff --git a/codex-rs/core/tests/suite/step_settings.rs b/codex-rs/core/tests/suite/step_settings.rs new file mode 100644 index 0000000000..8ab26c4bea --- /dev/null +++ b/codex-rs/core/tests/suite/step_settings.rs @@ -0,0 +1,911 @@ +use anyhow::Result; +use codex_core::CodexThread; +use codex_core::TurnInputRequest; +use codex_core::config::Constrained; +use codex_features::Feature; +use codex_login::CodexAuth; +use codex_models_manager::bundled_models_response; +use codex_models_manager::manager::RefreshStrategy; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::ModelInfo; +use codex_protocol::openai_models::ModelsResponse; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::SafetyBufferingEvent; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::TurnSettingsUpdate; +use codex_protocol::protocol::TurnSettingsUpdateOutcome; +use codex_protocol::request_user_input::RequestUserInputAnswer; +use codex_protocol::request_user_input::RequestUserInputEvent; +use codex_protocol::request_user_input::RequestUserInputResponse; +use codex_protocol::user_input::UserInput; +use core_test_support::responses::ResponsesRequest; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_models_once; +use core_test_support::responses::mount_response_sequence; +use core_test_support::responses::mount_sse_sequence; +use core_test_support::responses::sse; +use core_test_support::responses::sse_completed; +use core_test_support::responses::sse_response; +use core_test_support::responses::start_mock_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::TestCodexBuilder; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::collections::HashMap; +use test_case::test_case; + +const MODEL_A: &str = "step-settings-a"; +const MODEL_B: &str = "step-settings-b"; +const MODEL_C: &str = "step-settings-c"; +const TURN_STATE_HEADER: &str = "x-codex-turn-state"; + +fn step_settings_models() -> Vec { + let model = bundled_models_response() + .expect("bundled models should parse") + .models + .into_iter() + .find(|model| model.slug == "gpt-5.4") + .expect("bundled gpt-5.4 model"); + [MODEL_A, MODEL_B, MODEL_C] + .into_iter() + .map(|slug| { + let mut model = model.clone(); + // Tests add model-owned differences when they need to exercise + // activation or an explicit safety restriction. + model.slug = slug.to_string(); + model + }) + .collect() +} + +fn step_settings_test() -> TestCodexBuilder { + test_codex().with_model(MODEL_A).with_config(move |config| { + for feature in [ + Feature::StepModelSwitching, + Feature::DefaultModeRequestUserInput, + Feature::FastMode, + ] { + config + .features + .enable(feature) + .expect("test config should allow feature update"); + } + config.model_catalog = Some(ModelsResponse { + models: step_settings_models(), + }); + config.model_reasoning_effort = Some(ReasoningEffort::Low); + config.model_reasoning_summary = Some(ReasoningSummary::Concise); + config.service_tier = None; + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::User; + }) +} + +fn paused_response(response_id: &str, call_id: &str) -> String { + sse(vec![ + ev_response_created(response_id), + ev_function_call( + call_id, + "request_user_input", + &json!({ + "questions": [{ + "id": "continue", + "header": "Continue", + "question": "Continue after the settings update?", + "options": [{ + "label": "Yes (Recommended)", + "description": "Continue the current turn." + }, { + "label": "No", + "description": "Stop the current turn." + }] + }] + }) + .to_string(), + ), + ev_completed(response_id), + ]) +} + +async fn start_paused_turn(thread: &CodexThread) -> Result { + thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "pause before continuing".into(), + text_elements: Vec::new(), + }])) + .await?; + Ok(wait_for_event_match(thread, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await) +} + +async fn answer_paused_turn(thread: &CodexThread, turn_id: &str) -> Result<()> { + thread + .submit(Op::UserInputAnswer { + id: turn_id.to_string(), + response: RequestUserInputResponse { + answers: HashMap::from([( + "continue".to_string(), + RequestUserInputAnswer { + answers: vec!["Yes (Recommended)".to_string()], + }, + )]), + }, + }) + .await?; + Ok(()) +} + +async fn submit_turn_settings( + thread: &CodexThread, + turn_id: &str, + update: TurnSettingsUpdate, +) -> Result { + let (reply, outcome) = tokio::sync::oneshot::channel(); + thread + .submit(Op::TurnSettings { + turn_id: turn_id.to_string(), + update, + reply, + }) + .await?; + Ok(tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), outcome).await??) +} + +fn request_settings(request: &ResponsesRequest) -> Value { + let body = request.body_json(); + json!({ + "model": body["model"], + "reasoning": body["reasoning"], + "service_tier": body.get("service_tier"), + }) +} + +fn request_turn_id(request: &ResponsesRequest) -> String { + let metadata: Value = serde_json::from_str( + &request + .header("x-codex-turn-metadata") + .expect("request should include turn metadata"), + ) + .expect("valid turn metadata"); + metadata["turn_id"] + .as_str() + .expect("request should include turn_id") + .to_string() +} + +#[derive(Clone, Copy)] +enum SettingsTarget { + Thread, + Turn, +} + +#[test_case(SettingsTarget::Thread; "thread updates stay next-turn-only with the feature enabled")] +#[test_case(SettingsTarget::Turn; "turn updates leave future settings unchanged")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn settings_updates_preserve_turn_identity_and_target(target: SettingsTarget) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response_mock = mount_response_sequence( + &server, + vec![ + sse_response(paused_response("resp-1", "pause-turn")) + .insert_header(TURN_STATE_HEADER, "original-turn-state"), + sse_response(sse_completed("resp-2")), + sse_response(sse_completed("resp-3")), + ], + ) + .await; + let test = step_settings_test().build_with_auto_env(&server).await?; + let original_settings = test.codex.thread_settings_snapshot().await; + let request = start_paused_turn(&test.codex).await?; + + match target { + SettingsTarget::Thread => { + test.codex + .submit(Op::ThreadSettings { + thread_settings: ThreadSettingsOverrides { + model: Some(MODEL_B.to_string()), + effort: Some(Some(ReasoningEffort::High)), + summary: Some(ReasoningSummary::Detailed), + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + }, + }) + .await?; + } + SettingsTarget::Turn => { + let update = TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + effort: Some(Some(ReasoningEffort::High)), + summary: Some(ReasoningSummary::Detailed), + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + }; + assert_eq!( + submit_turn_settings(&test.codex, "different-turn", update.clone()).await?, + TurnSettingsUpdateOutcome::TargetUnavailable + ); + assert_eq!( + submit_turn_settings(&test.codex, &request.turn_id, update).await?, + TurnSettingsUpdateOutcome::Applied + ); + } + } + answer_paused_turn(&test.codex, &request.turn_id).await?; + let mut settings_events = Vec::new(); + let mut new_turns = Vec::new(); + let completion = wait_for_event(&test.codex, |event| match event { + EventMsg::ThreadSettingsApplied(event) => { + settings_events.push(event.thread_settings.clone()); + false + } + EventMsg::TurnStarted(event) => { + new_turns.push(event.turn_id.clone()); + false + } + EventMsg::Error(error) => panic!("settings activation failed: {}", error.message), + EventMsg::TurnComplete(_) => true, + _ => false, + }) + .await; + let EventMsg::TurnComplete(completion) = completion else { + unreachable!("waited for turn completion") + }; + assert_eq!(completion.turn_id, request.turn_id); + assert_eq!(new_turns, Vec::::new()); + + let mut changed_settings = original_settings.clone(); + changed_settings.model = MODEL_B.to_string(); + changed_settings.reasoning_effort = Some(ReasoningEffort::High); + changed_settings.reasoning_summary = Some(ReasoningSummary::Detailed); + changed_settings.service_tier = Some(ServiceTier::Fast.request_value().to_string()); + changed_settings.collaboration_mode = changed_settings.collaboration_mode.with_updates( + Some(MODEL_B.to_string()), + Some(Some(ReasoningEffort::High)), + /*developer_instructions*/ None, + ); + let expected_future_settings = match target { + SettingsTarget::Turn => { + assert_eq!(settings_events, Vec::new()); + original_settings + } + SettingsTarget::Thread => { + assert_eq!(settings_events, vec![changed_settings.clone()]); + changed_settings + } + }; + assert_eq!( + test.codex.thread_settings_snapshot().await, + expected_future_settings + ); + test.submit_text_turn("start the next turn").await?; + + let original_request_settings = json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }); + let changed_request_settings = json!({ + "model": MODEL_B, + "reasoning": { "effort": "high", "summary": "detailed" }, + "service_tier": "priority", + }); + let (continued_settings, next_turn_settings) = match target { + SettingsTarget::Thread => (original_request_settings.clone(), changed_request_settings), + SettingsTarget::Turn => (changed_request_settings, original_request_settings.clone()), + }; + let requests = response_mock.requests(); + assert_eq!( + requests.iter().map(request_settings).collect::>(), + vec![ + original_request_settings, + continued_settings, + next_turn_settings, + ] + ); + assert_eq!(request_turn_id(&requests[0]), request.turn_id); + assert_eq!(request_turn_id(&requests[1]), request.turn_id); + assert_ne!(request_turn_id(&requests[2]), request.turn_id); + let session_id = requests[0] + .header("session-id") + .expect("initial request session id"); + assert_eq!(requests[1].header("session-id"), Some(session_id)); + assert_eq!(requests[0].header(TURN_STATE_HEADER), None); + assert_eq!( + requests[1].header(TURN_STATE_HEADER), + Some("original-turn-state".to_string()) + ); + assert_eq!(requests[2].header(TURN_STATE_HEADER), None); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn response_metadata_uses_the_captured_model_after_a_turn_update() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let mut created = ev_response_created("resp-2"); + created["safety_buffering"] = json!({ + "use_cases": ["cyber"], + "reasons": ["policy-check"], + "retry_model": MODEL_C, + }); + let response_mock = mount_response_sequence( + &server, + vec![ + sse_response(paused_response("resp-1", "pause-turn")), + sse_response(sse(vec![created, ev_completed("resp-2")])) + .insert_header("OpenAI-Model", MODEL_B), + ], + ) + .await; + let test = step_settings_test().build_with_auto_env(&server).await?; + let request = start_paused_turn(&test.codex).await?; + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + answer_paused_turn(&test.codex, &request.turn_id).await?; + + let mut reroutes = Vec::new(); + let mut buffering_events = Vec::new(); + wait_for_event(&test.codex, |event| match event { + EventMsg::ModelReroute(event) => { + reroutes.push(event.clone()); + false + } + EventMsg::SafetyBuffering(event) => { + buffering_events.push(event.clone()); + false + } + EventMsg::Error(error) => panic!("sampling failed: {}", error.message), + EventMsg::TurnComplete(_) => true, + _ => false, + }) + .await; + + assert_eq!( + response_mock + .requests() + .iter() + .map(|request| request.body_json()["model"].clone()) + .collect::>(), + vec![json!(MODEL_A), json!(MODEL_B)] + ); + // B's matching response header is not a reroute from the turn's initial A. + // Buffering metadata likewise belongs to the captured B step. + assert_eq!( + (reroutes, buffering_events), + ( + vec![], + vec![SafetyBufferingEvent { + model: MODEL_B.to_string(), + use_cases: vec!["cyber".to_string()], + reasons: vec!["policy-check".to_string()], + show_buffering_ui: true, + faster_model: Some(MODEL_C.to_string()), + }], + ) + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sparse_updates_preserve_divergent_active_and_future_models() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + paused_response("resp-1", "pause-first-step"), + paused_response("resp-2", "pause-second-step"), + sse_completed("resp-3"), + sse_completed("resp-4"), + ], + ) + .await; + let test = step_settings_test().build_with_auto_env(&server).await?; + let request = start_paused_turn(&test.codex).await?; + + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_C.to_string()), + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + answer_paused_turn(&test.codex, &request.turn_id).await?; + let second_request = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + assert_eq!(second_request.turn_id, request.turn_id); + assert_eq!(second_request.call_id, "pause-second-step"); + + test.codex + .submit(Op::ThreadSettings { + thread_settings: ThreadSettingsOverrides { + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + }, + }) + .await?; + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + let durable_settings = wait_for_event_match(&test.codex, |event| match event { + EventMsg::ThreadSettingsApplied(event) => Some(event.thread_settings.clone()), + _ => None, + }) + .await; + assert_eq!(durable_settings.model, MODEL_B); + assert_eq!( + durable_settings.reasoning_effort, + Some(ReasoningEffort::Low) + ); + assert_eq!( + durable_settings.service_tier, + Some(ServiceTier::Fast.request_value().to_string()) + ); + answer_paused_turn(&test.codex, &second_request.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_text_turn("start the next turn").await?; + + let requests = response_mock.requests(); + assert_eq!( + requests.iter().map(request_settings).collect::>(), + vec![ + json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }), + json!({ + "model": MODEL_C, + "reasoning": { "effort": "high", "summary": "concise" }, + "service_tier": null, + }), + json!({ + "model": MODEL_C, + "reasoning": { "effort": "high", "summary": "concise" }, + "service_tier": "priority", + }), + json!({ + "model": MODEL_B, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": "priority", + }), + ] + ); + assert_eq!( + requests[..3] + .iter() + .map(request_turn_id) + .collect::>(), + vec![request.turn_id.clone(); 3] + ); + assert_ne!(request_turn_id(&requests[3]), request.turn_id); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn turn_settings_do_not_target_idle_or_finished_turns() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response_mock = mount_sse_sequence( + &server, + vec![sse_completed("resp-1"), sse_completed("resp-2")], + ) + .await; + let test = step_settings_test().build_with_auto_env(&server).await?; + + for (discarded_model, model) in [(MODEL_C, MODEL_B), (MODEL_A, MODEL_C)] { + let turn_id = response_mock + .requests() + .last() + .map(request_turn_id) + .unwrap_or_else(|| "never-started".to_string()); + let before = test.codex.thread_settings_snapshot().await; + assert_eq!( + submit_turn_settings( + &test.codex, + &turn_id, + TurnSettingsUpdate { + model: Some(discarded_model.to_string()), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::TargetUnavailable + ); + assert_eq!(test.codex.thread_settings_snapshot().await, before); + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + model: Some(model.to_string()), + ..Default::default() + }, + ) + .await?; + test.submit_text_turn("start the next turn").await?; + } + + assert_eq!( + response_mock + .requests() + .iter() + .map(request_settings) + .collect::>(), + vec![ + json!({ + "model": MODEL_B, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }), + json!({ + "model": MODEL_C, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }), + ] + ); + + Ok(()) +} + +#[test_case(None, "detailed"; "unset summary follows the destination model")] +#[test_case(Some(ReasoningSummary::Concise), "concise"; "explicit summary is preserved")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn model_activation_uses_destination_metadata_defaults( + configured_summary: Option, + expected_summary: &str, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + paused_response("resp-1", "pause-before-model-change"), + paused_response("resp-2", "pause-before-restoring-model"), + sse_completed("resp-3"), + ], + ) + .await; + let test = step_settings_test() + .with_config(move |config| { + config.model_context_window = None; + config.model_auto_compact_token_limit = None; + config.model_reasoning_summary = configured_summary; + config.service_tier = Some(ServiceTier::Fast.request_value().to_string()); + config.base_instructions = None; + for model in &mut config + .model_catalog + .as_mut() + .expect("controlled model catalog") + .models + { + model.context_window = Some(256_000); + model.auto_compact_token_limit = Some(200_000); + model.default_reasoning_summary = ReasoningSummary::Concise; + if model.slug == MODEL_B { + model.context_window = Some(128_000); + model.auto_compact_token_limit = Some(100_000); + model.default_reasoning_summary = ReasoningSummary::Detailed; + model + .service_tiers + .retain(|tier| tier.id != ServiceTier::Fast.request_value()); + model + .model_messages + .as_mut() + .expect("model instruction metadata") + .instructions_template = + Some("Destination model instructions.".to_string()); + } + } + }) + .build_with_auto_env(&server) + .await?; + let request = start_paused_turn(&test.codex).await?; + + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + answer_paused_turn(&test.codex, &request.turn_id).await?; + let paused = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + // B cannot use the requested tier. Switching back must recover that + // selection and preserve an unset summary, not reuse B's effective values. + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_A.to_string()), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + answer_paused_turn(&test.codex, &paused.turn_id).await?; + wait_for_event(&test.codex, |event| match event { + EventMsg::Error(error) => panic!("model activation failed: {}", error.message), + EventMsg::TurnComplete(_) => true, + _ => false, + }) + .await; + + let requests = response_mock.requests(); + assert_eq!( + requests.iter().map(request_settings).collect::>(), + vec![ + json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": "priority", + }), + json!({ + "model": MODEL_B, + "reasoning": { "effort": "low", "summary": expected_summary }, + "service_tier": null, + }), + json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": "priority", + }), + ] + ); + assert_eq!( + requests.iter().map(request_turn_id).collect::>(), + vec![request.turn_id; 3], + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn turn_settings_rejection_preserves_independent_future_settings() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + paused_response("resp-1", "pause-before-node-repl-restriction"), + sse_completed("resp-2"), + sse_completed("resp-3"), + ], + ) + .await; + let test = step_settings_test() + .with_config(|config| { + for model in &mut config + .model_catalog + .as_mut() + .expect("controlled model catalog") + .models + { + model.node_repl_disabled = model.slug == MODEL_B; + } + }) + .build_with_auto_env(&server) + .await?; + let request = start_paused_turn(&test.codex).await?; + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + model: Some(MODEL_B.to_string()), + ..Default::default() + }, + ) + .await?; + let future = test.codex.thread_settings_snapshot().await; + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + model: Some(MODEL_B.to_string()), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Rejected { + reason: "the destination changes the admitted node REPL availability restriction" + .to_string(), + } + ); + assert_eq!(test.codex.thread_settings_snapshot().await, future); + answer_paused_turn(&test.codex, &request.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_text_turn("admit the next turn").await?; + + let requests = response_mock.requests(); + assert_eq!( + requests.iter().map(request_settings).collect::>(), + [MODEL_A, MODEL_A, MODEL_B] + .into_iter() + .map(|model| { + json!({ + "model": model, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }) + }) + .collect::>() + ); + assert_eq!(request_turn_id(&requests[1]), request.turn_id); + assert_ne!(request_turn_id(&requests[2]), request.turn_id); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn request_preference_activation_keeps_admitted_model_metadata() -> Result<()> { + skip_if_no_network!(Ok(())); + + // This test owns both catalog responses. The generic mock-server helper + // installs an extra one-shot /models response that would shift the sequence. + let server = wiremock::MockServer::start().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + paused_response("resp-1", "pause-before-models-refresh"), + sse_completed("resp-2"), + sse_completed("resp-3"), + ], + ) + .await; + let mut models = step_settings_models(); + for model in &mut models { + model.default_reasoning_summary = ReasoningSummary::Concise; + } + let initial_catalog = mount_models_once( + &server, + ModelsResponse { + models: models.clone(), + }, + ) + .await; + let test = step_settings_test() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + // Use the real refreshable models manager, and retain an unset + // preference so the selected model's default is observable. + config.model_catalog = None; + config.model_reasoning_summary = None; + }) + .build_with_auto_env(&server) + .await?; + let request = start_paused_turn(&test.codex).await?; + assert_eq!(initial_catalog.requests().len(), 1); + + for model in &mut models { + model.default_reasoning_summary = ReasoningSummary::Detailed; + } + let refresh = mount_models_once(&server, ModelsResponse { models }).await; + test.thread_manager + .get_models_manager() + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) + .await; + assert_eq!(refresh.requests().len(), 1); + + assert_eq!( + submit_turn_settings( + &test.codex, + &request.turn_id, + TurnSettingsUpdate { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + } + ) + .await?, + TurnSettingsUpdateOutcome::Applied + ); + answer_paused_turn(&test.codex, &request.turn_id).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_text_turn("admit the next turn").await?; + + assert_eq!( + response_mock + .requests() + .iter() + .map(request_settings) + .collect::>(), + vec![ + json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "concise" }, + "service_tier": null, + }), + json!({ + "model": MODEL_A, + "reasoning": { "effort": "high", "summary": "concise" }, + "service_tier": null, + }), + json!({ + "model": MODEL_A, + "reasoning": { "effort": "low", "summary": "detailed" }, + "service_tier": null, + }), + ] + ); + + Ok(()) +} diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 58c697b88e..29abf70c52 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -309,6 +309,8 @@ pub enum Feature { Artifact, /// Enable Fast mode selection in the TUI and request layer. FastMode, + /// Enable explicitly requested model changes for later step captures. + StepModelSwitching, /// Enable experimental realtime voice conversation mode in the TUI. RealtimeConversation, /// Prevent idle system sleep while a turn is actively running. @@ -1515,6 +1517,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::Stable, default_enabled: true, }, + FeatureSpec { + id: Feature::StepModelSwitching, + key: "step_model_switching", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::RealtimeConversation, key: "realtime_conversation", diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 2f089e7176..29f54c42f9 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -468,8 +468,33 @@ pub struct ConversationSpeechParams { pub text: String, } -/// Persistent thread-settings overrides that can be applied before user input or -/// on their own. +/// Supported sparse changes to one live task's current settings, regardless of +/// task kind. Child sessions and consumers of frozen initial settings are unchanged. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TurnSettingsUpdate { + pub model: Option, + /// `None` preserves the selection; `Some(None)` clears it. + pub effort: Option>, + pub summary: Option, + /// `None` preserves the requested tier; `Some(None)` clears it. + pub service_tier: Option>, +} + +/// The result of processing a turn-settings update, not merely queueing it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TurnSettingsUpdateOutcome { + /// Published for subsequent captures; already captured steps are unchanged. + /// The task need not sample or consume every selected preference. + Applied, + /// The named live task was absent or lost before publication. + TargetUnavailable, + Rejected { + reason: String, + }, +} + +/// Thread-settings overrides that can be applied before user input or on their +/// own. Standalone updates change the settings inherited by future turns. #[derive(Debug, Clone, Default, PartialEq)] pub struct ThreadSettingsOverrides { /// Updated fallback `cwd` and environments supplied together as a complete pair. @@ -587,15 +612,23 @@ pub enum Op { reply: oneshot::Sender>, }, - /// Apply persistent thread-settings overrides without starting a turn. + /// Apply thread-settings overrides without starting a turn. /// /// This uses the same submission queue as turn starts so app-server can /// preserve caller order between both kinds of mutation. ThreadSettings { - /// Persistent thread-settings overrides to apply. + /// Sparse thread-settings overrides to apply. thread_settings: ThreadSettingsOverrides, }, + /// Update only the named running turn, without changing future settings. + /// The reply reports the actual publication or why it did not occur. + TurnSettings { + turn_id: String, + update: TurnSettingsUpdate, + reply: oneshot::Sender, + }, + /// Inter-agent communication that should be recorded as agent-message history /// while still using the normal thread submission lifecycle. InterAgentCommunication { @@ -883,6 +916,7 @@ impl Op { Self::RecoverTurn { .. } => "recover_turn", Self::SuspendTurnAndShutdown { .. } => "suspend_turn_and_shutdown", Self::ThreadSettings { .. } => "thread_settings", + Self::TurnSettings { .. } => "turn_settings", Self::InterAgentCommunication { .. } => "inter_agent_communication", Self::ExecApproval { .. } => "exec_approval", Self::PatchApproval { .. } => "patch_approval",