diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index d9d4adedee..75866e17c3 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -522,7 +522,7 @@ impl CodexThread { &self, overrides: CodexThreadSettingsOverrides, ) -> ConstraintResult { - let updates = self.thread_settings_update(overrides).await; + let updates = Self::thread_settings_update(overrides); self.session.preview_settings(&updates).await } @@ -534,14 +534,11 @@ impl CodexThread { &self, settings: CodexThreadSettingsOverrides, ) -> ConstraintResult<()> { - let updates = self.thread_settings_update(settings).await; + let updates = Self::thread_settings_update(settings); self.session.update_settings(updates).await.map(|_| ()) } - async fn thread_settings_update( - &self, - overrides: CodexThreadSettingsOverrides, - ) -> SessionSettingsUpdate { + fn thread_settings_update(overrides: CodexThreadSettingsOverrides) -> SessionSettingsUpdate { let CodexThreadSettingsOverrides { environments, profile_workspace_roots, @@ -558,18 +555,11 @@ impl CodexThread { collaboration_mode, personality, } = overrides; - let collaboration_mode = if let Some(collaboration_mode) = collaboration_mode { - collaboration_mode - } else { - self.session - .collaboration_mode() - .await - .with_updates(model, effort, /*developer_instructions*/ None) - }; - SessionSettingsUpdate { step_settings: StepSettingsUpdate { - collaboration_mode: Some(collaboration_mode), + model, + effort, + collaboration_mode, reasoning_summary: summary, service_tier, personality, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 6ced24ca68..3d180b03a9 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1602,6 +1602,24 @@ impl Session { &self, updates: SessionSettingsUpdate, ) -> ConstraintResult { + let Some(commit) = self.update_settings_if(updates, |_, _| true).await? else { + unreachable!("unconditional settings updates must commit"); + }; + Ok(commit) + } + + /// Evaluates the caller's synchronous predicate against the current configuration + /// and validated candidate under the same state lock used for publication. + /// This prevents a stale admission decision. Rejection has no settings or runtime + /// effects. + /// + /// The predicate must be fast and side-effect-free. It must not block, acquire + /// other locks, or call back into `Session`. + async fn update_settings_if( + &self, + updates: SessionSettingsUpdate, + should_commit: impl FnOnce(&SessionConfiguration, &SessionConfiguration) -> bool + Send, + ) -> ConstraintResult> { let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); let (commit, previous_config, new_config, permission_profile_changed, mcp_inputs_changed) = { let mut state = self.state.lock().await; @@ -1614,6 +1632,10 @@ impl Session { } }; + if !should_commit(&state.session_configuration, &updated) { + return Ok(None); + } + let previous_config = notify_config_contributors .then(|| self.build_effective_session_config(&state.session_configuration)); let previous_permission_profile = state.session_configuration.permission_profile(); @@ -1662,7 +1684,7 @@ impl Session { if mcp_inputs_changed { self.schedule_mcp_prewarm(); } - Ok(commit) + Ok(Some(commit)) } pub(crate) async fn preview_settings( diff --git a/codex-rs/core/src/session/step_settings.rs b/codex-rs/core/src/session/step_settings.rs index e05bc7d472..484ee756bb 100644 --- a/codex-rs/core/src/session/step_settings.rs +++ b/codex-rs/core/src/session/step_settings.rs @@ -11,6 +11,7 @@ use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE; use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; /// Model and execution settings selected for an individual model step within @@ -64,9 +65,15 @@ impl ModelInfoOverrides { } } -/// Partial edits to configured step settings. +/// Sparse edits applied independently to each settings owner. +/// +/// Do not materialize a partial update against one settings owner and reuse +/// that full value for another. Merge the requested edits with each target. #[derive(Clone, Debug, Default, PartialEq)] pub(crate) struct StepSettingsUpdate { + pub(crate) model: Option, + pub(crate) effort: Option>, + /// A complete collaboration mode takes precedence over model and effort edits. pub(crate) collaboration_mode: Option, pub(crate) reasoning_summary: Option, pub(crate) service_tier: Option>, @@ -92,9 +99,13 @@ impl StepSettings { constraints: &StepSettingsConstraints<'_>, ) -> ConstraintResult { let mut next = self.clone(); - if let Some(collaboration_mode) = update.collaboration_mode.clone() { - next.collaboration_mode = collaboration_mode; - } + next.collaboration_mode = update.collaboration_mode.clone().unwrap_or_else(|| { + self.collaboration_mode.with_updates( + update.model.clone(), + update.effort.clone(), + /*developer_instructions*/ None, + ) + }); if let Some(summary) = update.reasoning_summary { next.reasoning_summary = Some(summary); } diff --git a/codex-rs/core/src/session/step_settings_tests.rs b/codex-rs/core/src/session/step_settings_tests.rs index b169b9219d..778a6edc52 100644 --- a/codex-rs/core/src/session/step_settings_tests.rs +++ b/codex-rs/core/src/session/step_settings_tests.rs @@ -7,6 +7,8 @@ use codex_config::ConfigLayerStack; use codex_config::RequirementSource; use codex_config::Sourced; use codex_features::Feature; +use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Settings; use codex_protocol::models::PermissionProfile; use codex_protocol::permissions::FileSystemAccessMode; use codex_protocol::permissions::FileSystemPath; @@ -270,3 +272,114 @@ async fn environment_only_update_revalidates_existing_step_settings() { }), ); } + +fn configured_settings() -> StepSettings { + StepSettings { + collaboration_mode: CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model: "model-a".to_string(), + reasoning_effort: Some(ReasoningEffort::Low), + developer_instructions: Some("keep these instructions".to_string()), + }, + }, + reasoning_summary: Some(ReasoningSummary::Concise), + service_tier: None, + personality: Some(Personality::Friendly), + approval_policy: Constrained::allow_any(AskForApproval::OnRequest), + approvals_reviewer: ApprovalsReviewer::User, + } +} + +fn step_settings_constraints(requirements: &ConfigRequirements) -> StepSettingsConstraints<'_> { + StepSettingsConstraints { + requirements, + guardian_approval_enabled: false, + trusted_guardian_reviewer: false, + has_full_disk_write_access: false, + } +} + +#[test] +fn sparse_patch_uses_the_settings_version_being_updated() { + let requirements = ConfigRequirements::default(); + let constraints = step_settings_constraints(&requirements); + let initial = configured_settings(); + let tier_update = StepSettingsUpdate { + service_tier: Some(Some("fast".to_string())), + ..Default::default() + }; + let latest = initial + .apply( + &StepSettingsUpdate { + model: Some("model-b".to_string()), + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + &constraints, + ) + .expect("model update should apply"); + + let updated = latest + .apply(&tier_update, &constraints) + .expect("a previously prepared sparse update should apply"); + let expected = StepSettings { + service_tier: Some(ServiceTier::Fast.request_value().to_string()), + ..latest + }; + assert_eq!(updated, expected); +} + +#[test] +fn collaboration_replacement_wins_and_effort_clear_remains_sparse() { + let requirements = ConfigRequirements::default(); + let constraints = step_settings_constraints(&requirements); + let initial = configured_settings(); + let replacement = CollaborationMode { + mode: ModeKind::Plan, + settings: Settings { + model: "model-b".to_string(), + reasoning_effort: Some(ReasoningEffort::Medium), + developer_instructions: None, + }, + }; + let replaced = initial + .apply( + &StepSettingsUpdate { + model: Some("ignored-model".to_string()), + effort: Some(Some(ReasoningEffort::High)), + collaboration_mode: Some(replacement.clone()), + ..Default::default() + }, + &constraints, + ) + .expect("collaboration mode should replace model and effort edits"); + assert_eq!( + replaced, + StepSettings { + collaboration_mode: replacement.clone(), + ..initial + } + ); + + let cleared = replaced + .apply( + &StepSettingsUpdate { + effort: Some(None), + ..Default::default() + }, + &constraints, + ) + .expect("effort clear should apply"); + assert_eq!( + cleared, + StepSettings { + collaboration_mode: replacement.with_updates( + /*model*/ None, + Some(None), + /*developer_instructions*/ None, + ), + ..replaced + } + ); +} diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index 181ab427d7..82126153ce 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -21,7 +21,7 @@ pub(super) async fn update( submission_id: String, overrides: ThreadSettingsOverrides, ) { - let updates = prepare_update(session, overrides).await; + let updates = prepare_update(overrides); if let Err(error) = apply_update(session, submission_id.clone(), updates).await { session .send_event_raw(Event { @@ -36,10 +36,7 @@ pub(super) async fn update( } /// Converts protocol overrides into the internal settings update shape. -pub(super) async fn prepare_update( - session: &Session, - overrides: ThreadSettingsOverrides, -) -> SessionSettingsUpdate { +pub(super) fn prepare_update(overrides: ThreadSettingsOverrides) -> SessionSettingsUpdate { let ThreadSettingsOverrides { environments, profile_workspace_roots, @@ -56,22 +53,11 @@ pub(super) async fn prepare_update( collaboration_mode, personality, } = overrides; - let collaboration_mode = match collaboration_mode { - Some(collaboration_mode) => collaboration_mode, - None => { - let state = session.state.lock().await; - // Model and reasoning effort live in CollaborationMode settings today, so - // partial thread-settings updates refresh those fields on the active mode. - state - .session_configuration - .step_settings - .collaboration_mode - .with_updates(model, effort, /*developer_instructions*/ None) - } - }; SessionSettingsUpdate { step_settings: StepSettingsUpdate { - collaboration_mode: Some(collaboration_mode), + model, + effort, + collaboration_mode, reasoning_summary: summary, service_tier, personality, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index ba506f2dc2..5041644280 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -758,18 +758,36 @@ impl Session { sub_id: String, updates: SessionSettingsUpdate, ) -> CodexResult<(Arc, ThreadSettingsSnapshot)> { + let Some(turn) = self + .new_turn_with_sub_id_if(sub_id, updates, |_, _| true) + .await? + else { + unreachable!("unconditional turn construction must accept valid settings"); + }; + Ok(turn) + } + + /// Commits accepted settings atomically, then constructs the turn without holding + /// the state lock. The caller owns admission policy. + /// + /// `should_start` runs under the state lock against the current and validated + /// proposed configurations. It must be fast and side-effect-free, and must not + /// block, acquire other locks, or call back into `Session`. + pub(super) async fn new_turn_with_sub_id_if( + &self, + sub_id: String, + updates: SessionSettingsUpdate, + should_start: impl FnOnce(&SessionConfiguration, &SessionConfiguration) -> bool + Send, + ) -> CodexResult, ThreadSettingsSnapshot)>> { let final_output_json_schema = updates.final_output_json_schema.clone(); let service_tier_for_turn = updates.service_tier_for_turn.clone(); - let update_result = self - .update_settings(updates) - .await - .map_err(|err| CodexErr::InvalidRequest(err.to_string())); - let commit = match update_result { - Ok(commit) => commit, - Err(err) => { - let message = err.to_string(); + let commit = match self.update_settings_if(updates, should_start).await { + Ok(Some(commit)) => commit, + Ok(None) => return Ok(None), + Err(error) => { + let message = CodexErr::InvalidRequest(error.to_string()).to_string(); self.send_event_raw(Event { - id: sub_id.clone(), + id: sub_id, msg: EventMsg::Error(ErrorEvent { message: message.clone(), codex_error_info: Some(CodexErrorInfo::BadRequest), @@ -787,9 +805,10 @@ impl Session { let turn_context = self .new_turn_from_configuration(sub_id, configuration, final_output_json_schema) .await; - Ok((turn_context, commit.snapshot)) + Ok(Some((turn_context, commit.snapshot))) } + /// Builds a turn from the exact committed settings without starting a task. async fn new_turn_from_configuration( &self, sub_id: String, diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 704af6c1ff..08ca3886bd 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -10,6 +10,7 @@ use super::TurnInput; use super::session::Session; +use super::session::SessionConfiguration; use super::session::SessionSettingsUpdate; use super::thread_settings; use super::turn_context::TurnContext; @@ -44,6 +45,33 @@ use uuid::Uuid; #[path = "turn_input_tests.rs"] mod tests; +/// Why input is starting a turn; shared by admission and input delivery. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TurnStartKind { + User, + Automatic, + Recovery, +} + +impl TurnStartKind { + fn permits_mode(self, mode: ModeKind) -> bool { + match self { + Self::User | Self::Recovery => true, + Self::Automatic => mode != ModeKind::Plan, + } + } + + /// Automatic work may neither leave an existing Plan mode nor enter it. + fn permits_settings( + self, + current: &SessionConfiguration, + proposed: &SessionConfiguration, + ) -> bool { + self.permits_mode(current.step_settings.collaboration_mode.mode) + && self.permits_mode(proposed.step_settings.collaboration_mode.mode) + } +} + /// Thread settings and start-only options prepared before Core knows whether /// turn input starts or steers. /// @@ -65,7 +93,7 @@ impl PreparedTurnInputSettings { let thread_settings_update = if thread_settings == ThreadSettingsOverrides::default() { None } else { - let updates = thread_settings::prepare_update(session, thread_settings).await; + let updates = thread_settings::prepare_update(thread_settings); session .preview_settings(&updates) .await @@ -82,20 +110,15 @@ impl PreparedTurnInputSettings { self.start_options.final_output_json_schema.as_ref() } - fn would_enter_plan_mode(&self) -> bool { - self.thread_settings_update - .as_ref() - .and_then(|updates| updates.step_settings.collaboration_mode.as_ref()) - .is_some_and(|collaboration_mode| collaboration_mode.mode == ModeKind::Plan) - } - /// Applies persistent settings and start-only options before creating a - /// new turn context. + /// new turn context. Returns `None` if admission rejects the candidate, + /// without committing its settings. async fn apply_started( self, session: &Arc, submission_id: String, - ) -> CodexResult> { + kind: TurnStartKind, + ) -> CodexResult>> { let TurnStartOptions { final_output_json_schema, service_tier, @@ -107,10 +130,23 @@ impl PreparedTurnInputSettings { updates.final_output_json_schema = Some(final_output_json_schema); updates.service_tier_for_turn = service_tier; - // new_turn_with_sub_id already emits an error event when settings are invalid. - let (turn_context, settings_snapshot) = session - .new_turn_with_sub_id(submission_id.clone(), updates) - .await?; + let turn = match kind { + TurnStartKind::User | TurnStartKind::Recovery => Some( + session + .new_turn_with_sub_id(submission_id.clone(), updates) + .await?, + ), + TurnStartKind::Automatic => { + session + .new_turn_with_sub_id_if(submission_id.clone(), updates, |current, proposed| { + kind.permits_settings(current, proposed) + }) + .await? + } + }; + let Some((turn_context, settings_snapshot)) = turn else { + return Ok(None); + }; if emit_thread_settings_applied { thread_settings::emit_applied(session, submission_id, settings_snapshot).await; } @@ -124,7 +160,7 @@ impl PreparedTurnInputSettings { .turn_metadata_state .set_root_turn_id(root_turn_id); } - Ok(turn_context) + Ok(Some(turn_context)) } /// Applies only persistent settings after steering succeeds. The active @@ -148,7 +184,15 @@ pub(super) async fn handle( match mode { TurnInputMode::StartOrSteer => start_or_steer(session, request, submission_id).await, TurnInputMode::StartIfIdle => { - start_if_idle(session, request, submission_id, /*is_recovery*/ false).await + let kind = match &request.input { + SubmittedTurnInput::UserInput { content, .. } if !content.is_empty() => { + TurnStartKind::User + } + SubmittedTurnInput::UserInput { .. } + | SubmittedTurnInput::ResponseItem(_) + | SubmittedTurnInput::InterAgentCommunication(_) => TurnStartKind::Automatic, + }; + start_if_idle(session, request, submission_id, kind).await } TurnInputMode::Steer { expected_turn_id } => { steer(session, request, expected_turn_id, submission_id).await @@ -162,7 +206,7 @@ pub(super) async fn handle_recovery( submission_id: String, ) -> CodexResult { let request = TurnInputRequest::user_input(Vec::new()).with_thread_settings(thread_settings); - start_if_idle(session, request, submission_id, /*is_recovery*/ true).await + start_if_idle(session, request, submission_id, TurnStartKind::Recovery).await } async fn start_or_steer( @@ -210,9 +254,12 @@ async fn start_or_steer( Ok(TurnInputSubmission::Steered { turn_id }) } Err(NotSubmittedReason::NoActiveTurn) => { - let turn_context = settings - .apply_started(session, submission_id.clone()) - .await?; + let Some(turn_context) = settings + .apply_started(session, submission_id.clone(), TurnStartKind::User) + .await? + else { + unreachable!("explicit user input can enter Plan mode"); + }; if can_start_root_turn && !items.is_empty() && turn_context @@ -254,7 +301,7 @@ async fn start_if_idle( session: &Arc, request: TurnInputRequest, submission_id: String, - is_recovery: bool, + kind: TurnStartKind, ) -> CodexResult { let TurnInputRequest { input, @@ -264,17 +311,17 @@ async fn start_if_idle( responsesapi_client_metadata, .. } = request; - let has_user_input = has_nonempty_user_input(&input); - let is_automatic_idle_work = !has_user_input && !is_recovery; let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none(); if session.input_queue.has_trigger_turn_mailbox_items().await { return Ok(TurnInputSubmission::NotSubmitted { reason: NotSubmittedReason::PendingTriggerTurn, }); } - // Empty non-recovery starts are automatic wakeups, not explicit user requests. - // Do not let them start a Plan turn. - if is_automatic_idle_work && session.collaboration_mode().await.mode == ModeKind::Plan { + // Preserve current-Plan rejection before reservation and settings errors. + // The commit-time decision also checks the proposed mode. + if kind == TurnStartKind::Automatic + && !kind.permits_mode(session.collaboration_mode().await.mode) + { return Ok(TurnInputSubmission::NotSubmitted { reason: NotSubmittedReason::PlanMode, }); @@ -306,17 +353,17 @@ async fn start_if_idle( return Err(error); } }; - // Automatic work must not use persistent settings to start a turn - // whose effective collaboration mode is Plan. - if is_automatic_idle_work && settings.would_enter_plan_mode() { - session.clear_reserved_idle_turn(&turn_state).await; - return Ok(TurnInputSubmission::NotSubmitted { - reason: NotSubmittedReason::PlanMode, - }); - } - - let turn_context = match settings.apply_started(session, submission_id.clone()).await { - Ok(turn_context) => turn_context, + let turn_context = match settings + .apply_started(session, submission_id.clone(), kind) + .await + { + Ok(Some(turn_context)) => turn_context, + Ok(None) => { + session.clear_reserved_idle_turn(&turn_state).await; + return Ok(TurnInputSubmission::NotSubmitted { + reason: NotSubmittedReason::PlanMode, + }); + } Err(error) => { session.clear_reserved_idle_turn(&turn_state).await; return Err(error); @@ -327,7 +374,7 @@ async fn start_if_idle( .turn_metadata_state .set_responsesapi_client_metadata(responsesapi_client_metadata); } - if has_user_input + if kind == TurnStartKind::User && can_start_root_turn && turn_context .turn_metadata_state @@ -342,22 +389,29 @@ async fn start_if_idle( .await; let mut task_input = merge_additional_context_input(session, additional_context).await; - if has_user_input { - session.clear_connector_selection().await; - if let SubmittedTurnInput::UserInput { content, .. } = &input { - turn_context.session_telemetry.user_prompt(content); + match kind { + TurnStartKind::User => { + session.clear_connector_selection().await; + if let SubmittedTurnInput::UserInput { content, .. } = &input { + turn_context.session_telemetry.user_prompt(content); + } + task_input.push(pending_turn_input(input)); + } + TurnStartKind::Automatic => { + // Empty automatic user input resumes sampling without a new message. + if !matches!(&input, SubmittedTurnInput::UserInput { .. }) { + session + .input_queue + .extend_pending_input_for_turn_state( + turn_state.as_ref(), + vec![pending_turn_input(input)], + ) + .await; + } + } + TurnStartKind::Recovery => { + // Recovery resumes an existing turn without a new empty user message. } - task_input.push(pending_turn_input(input)); - } else if is_automatic_idle_work && !matches!(&input, SubmittedTurnInput::UserInput { .. }) { - // Automatic response-item work still needs to be queued, but an empty - // user-input request should start sampling without adding a message. - session - .input_queue - .extend_pending_input_for_turn_state( - turn_state.as_ref(), - vec![pending_turn_input(input)], - ) - .await; } session .start_task( @@ -566,10 +620,6 @@ impl Session { } } -fn has_nonempty_user_input(input: &SubmittedTurnInput) -> bool { - matches!(input, SubmittedTurnInput::UserInput { content, .. } if !content.is_empty()) -} - async fn merge_additional_context_input( session: &Session, additional_context: BTreeMap, diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index b16d8be2d6..06c1acf343 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -1,22 +1,30 @@ use super::*; +use crate::config::Constrained; +use crate::session::step_settings::StepSettingsUpdate; use crate::session::tests::make_session_and_context_with_rx; use crate::state::TaskKind; use crate::tasks::SessionTask; use crate::tasks::SessionTaskResult; use codex_protocol::AgentPath; +use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::ServiceTier; use codex_protocol::config_types::Settings; +use codex_protocol::error::CodexErrorDetails; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; +use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::turn_input::TurnInput as SubmittedTurnInput; use codex_protocol::user_input::UserInput; +use codex_utils_absolute_path::AbsolutePathBuf; use core_test_support::test_codex::local_selections; use pretty_assertions::assert_eq; +use test_case::test_case; use tokio::time::sleep; use tokio_util::sync::CancellationToken; @@ -234,28 +242,71 @@ async fn recovery_rejects_active_turn_without_injecting_or_applying_settings() { } #[tokio::test] -async fn start_only_rejects_plan_mode_without_injecting() { +async fn start_only_rejects_current_plan_before_validating_settings() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; - let mut collaboration_mode = session.collaboration_mode().await; - collaboration_mode.mode = ModeKind::Plan; + let default_mode = session.collaboration_mode().await; { let mut state = session.state.lock().await; - Arc::make_mut(&mut state.session_configuration.step_settings).collaboration_mode = - collaboration_mode; + let settings = Arc::make_mut(&mut state.session_configuration.step_settings); + settings.collaboration_mode.mode = ModeKind::Plan; + settings.approval_policy = Constrained::allow_only(AskForApproval::OnRequest); } + let desired_settings = session.thread_settings_snapshot().await; + let invalid_override = ThreadSettingsOverrides { + collaboration_mode: Some(default_mode.clone()), + approval_policy: Some(AskForApproval::Never), + ..Default::default() + }; - let submission = submit_start_only( + // Current Plan takes precedence even when the request would leave Plan or + // fail settings validation. Nothing has been reserved or applied yet. + let submission = handle( &session, - SubmittedTurnInput::ResponseItem(user_message("synthetic idle input")), + TurnInputRequest::new(SubmittedTurnInput::ResponseItem(user_message( + "synthetic idle input", + ))) + .with_thread_settings(invalid_override.clone()), + TurnInputMode::StartIfIdle, + "automatic-plan-submission".to_string(), ) - .await; - + .await + .expect("current Plan must reject before settings validation"); assert_eq!( + submission, TurnInputSubmission::NotSubmitted { reason: NotSubmittedReason::PlanMode, - }, - submission + } ); + assert_eq!(session.thread_settings_snapshot().await, desired_settings); + assert!(session.active_turn.lock().await.is_none()); + + session + .update_settings(SessionSettingsUpdate { + step_settings: StepSettingsUpdate { + collaboration_mode: Some(default_mode), + ..Default::default() + }, + ..Default::default() + }) + .await + .expect("explicit settings may leave Plan mode"); + let desired_settings = session.thread_settings_snapshot().await; + let result = handle( + &session, + TurnInputRequest::new(SubmittedTurnInput::ResponseItem(user_message( + "invalid automatic input", + ))) + .with_thread_settings(invalid_override), + TurnInputMode::StartIfIdle, + "invalid-automatic-submission".to_string(), + ) + .await; + let error = result.expect_err("invalid automatic settings must be rejected"); + assert!(matches!( + error.details(), + CodexErrorDetails::InvalidRequest(_) + )); + assert_eq!(session.thread_settings_snapshot().await, desired_settings); assert!(session.active_turn.lock().await.is_none()); assert_eq!( (Vec::::new(), None, None), @@ -266,6 +317,338 @@ async fn start_only_rejects_plan_mode_without_injecting() { ); } +#[tokio::test] +async fn prepared_user_updates_merge_with_settings_at_turn_start() { + for (requested, intervening) in [ + ( + ThreadSettingsOverrides { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ThreadSettingsOverrides { + model: Some("gpt-5.2".to_string()), + ..Default::default() + }, + ), + ( + ThreadSettingsOverrides { + model: Some("gpt-5.2".to_string()), + ..Default::default() + }, + ThreadSettingsOverrides { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + ), + ] { + let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; + let initial = CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model: "gpt-5.4".to_string(), + reasoning_effort: Some(ReasoningEffort::Low), + developer_instructions: None, + }, + }; + session + .update_settings(thread_settings::prepare_update(ThreadSettingsOverrides { + collaboration_mode: Some(initial.clone()), + ..Default::default() + })) + .await + .expect("set initial model and effort"); + let prepared = + PreparedTurnInputSettings::prepare(&session, requested, TurnStartOptions::default()) + .await + .expect("prepare partial protocol update"); + session + .update_settings(thread_settings::prepare_update(intervening)) + .await + .expect("commit intervening settings"); + + let turn_context = prepared + .apply_started( + &session, + "sparse-user-start".to_string(), + TurnStartKind::User, + ) + .await + .expect("apply prepared settings") + .expect("user start is permitted"); + let expected = initial.with_updates( + Some("gpt-5.2".to_string()), + Some(Some(ReasoningEffort::High)), + /*developer_instructions*/ None, + ); + assert_eq!(session.collaboration_mode().await, expected); + assert_eq!(turn_context.collaboration_mode(), expected); + } +} + +#[tokio::test] +async fn automatic_admission_uses_current_candidate_after_plan_preview() { + let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; + let default_mode = session.collaboration_mode().await; + let mut plan_mode = default_mode.clone(); + plan_mode.mode = ModeKind::Plan; + session + .update_settings(thread_settings::prepare_update(ThreadSettingsOverrides { + collaboration_mode: Some(plan_mode), + ..Default::default() + })) + .await + .expect("enter Plan after the initial admission check"); + let prepared = PreparedTurnInputSettings::prepare( + &session, + ThreadSettingsOverrides { + effort: Some(Some(ReasoningEffort::High)), + ..Default::default() + }, + TurnStartOptions::default(), + ) + .await + .expect("validate the patch while the preview is Plan"); + session + .update_settings(thread_settings::prepare_update(ThreadSettingsOverrides { + collaboration_mode: Some(default_mode.clone()), + ..Default::default() + })) + .await + .expect("leave Plan before atomic admission"); + + let turn_context = prepared + .apply_started( + &session, + "automatic-after-plan-preview".to_string(), + TurnStartKind::Automatic, + ) + .await + .expect("automatic admission should succeed") + .expect("current and proposed modes are both Default"); + let expected = default_mode.with_updates( + /*model*/ None, + Some(Some(ReasoningEffort::High)), + /*developer_instructions*/ None, + ); + assert_eq!(session.collaboration_mode().await, expected); + assert_eq!(turn_context.collaboration_mode(), expected); +} + +#[tokio::test] +async fn automatic_admission_rechecks_plan_mode_without_committing_sparse_settings() { + struct ConfigRecorder(Arc>>); + + impl codex_extension_api::ConfigContributor for ConfigRecorder { + fn on_config_changed( + &self, + _session_store: &codex_extension_api::ExtensionData, + _thread_store: &codex_extension_api::ExtensionData, + _previous_config: &crate::config::Config, + new_config: &crate::config::Config, + ) { + self.0.lock().expect("config records lock").push(( + new_config.permissions.approval_policy.value(), + new_config.approvals_reviewer, + )); + } + } + + let (mut session, _turn_context, rx) = make_session_and_context_with_rx().await; + let records = Arc::new(std::sync::Mutex::new(Vec::new())); + let mut extensions = + codex_extension_api::ExtensionRegistryBuilder::::new(); + extensions.config_contributor(Arc::new(ConfigRecorder(Arc::clone(&records)))); + Arc::get_mut(&mut session) + .expect("unique test session") + .services + .extensions = Arc::new(extensions.build()); + { + let mut state = session.state.lock().await; + let settings = Arc::make_mut(&mut state.session_configuration.step_settings); + settings.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + settings.approvals_reviewer = ApprovalsReviewer::User; + } + let original_environments = session.services.turn_environments.selections(); + let workspace = tempfile::tempdir().expect("create proposed workspace"); + let proposed_environments = local_selections( + AbsolutePathBuf::try_from(workspace.path()).expect("absolute workspace path"), + ); + assert_ne!(original_environments, proposed_environments.environments); + let default_mode = session.collaboration_mode().await; + let overrides = ThreadSettingsOverrides { + model: Some("automatic-model-must-not-be-applied".to_string()), + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + environments: Some(proposed_environments.clone()), + approval_policy: Some(AskForApproval::Never), + approvals_reviewer: Some(ApprovalsReviewer::AutoReview), + ..Default::default() + }; + let default_override = ThreadSettingsOverrides { + collaboration_mode: Some(default_mode.clone()), + ..overrides.clone() + }; + let prepared = PreparedTurnInputSettings::prepare( + &session, + overrides.clone(), + TurnStartOptions::default(), + ) + .await + .expect("sparse settings should preview successfully"); + let prepared_default = + PreparedTurnInputSettings::prepare(&session, default_override, TurnStartOptions::default()) + .await + .expect("Default replacement should preview successfully"); + + // Another settings writer changes the effective mode after preview. The + // sparse patch must not commit in Plan, and a full Default replacement + // must not let automatic work escape the now-current Plan configuration. + let mut collaboration_mode = default_mode; + collaboration_mode.mode = ModeKind::Plan; + session + .update_settings(SessionSettingsUpdate { + step_settings: StepSettingsUpdate { + collaboration_mode: Some(collaboration_mode), + ..Default::default() + }, + ..Default::default() + }) + .await + .expect("Plan mode should be allowed for explicit settings updates"); + let desired_settings = session.thread_settings_snapshot().await; + records.lock().expect("config records lock").clear(); + // Keep the existing refresh worker from consuming invalidation while the + // rejection and its positive control inspect MCP's dirty state. + let _mcp_refresh = session + .mcp_refresh + .acquire() + .await + .expect("acquire MCP refresh gate"); + session.mcp_refresh.claim(); + assert!(!session.mcp_refresh.is_pending()); + + for (submission_id, prepared) in [ + ("automatic-after-preview", prepared), + ("automatic-default-after-preview", prepared_default), + ] { + let outcome = prepared + .apply_started( + &session, + submission_id.to_string(), + TurnStartKind::Automatic, + ) + .await + .expect("automatic admission should return a typed rejection"); + assert!(outcome.is_none()); + assert_eq!(session.thread_settings_snapshot().await, desired_settings); + assert!(session.active_turn.lock().await.is_none()); + assert_eq!( + session.services.turn_environments.selections(), + original_environments + ); + assert!(!session.mcp_refresh.is_pending()); + assert_eq!(*records.lock().expect("config records lock"), Vec::new()); + while let Ok(event) = rx.try_recv() { + assert_ne!(event.id, submission_id); + } + } + + assert_eq!( + session + .input_queue + .get_pending_input(&session.active_turn) + .await, + (Vec::::new(), None, None) + ); + + // The rejected candidate is valid and would have real runtime effects if + // accepted by an ordinary settings update. + session + .update_settings(thread_settings::prepare_update(overrides)) + .await + .expect("explicit settings update accepts the same patch"); + assert_eq!( + session.services.turn_environments.selections(), + proposed_environments.environments + ); + assert!(session.mcp_refresh.is_pending()); + assert_eq!( + *records.lock().expect("config records lock"), + vec![(AskForApproval::Never, ApprovalsReviewer::AutoReview)] + ); +} + +#[test_case(TurnStartKind::User; "ordinary constructor")] +#[test_case(TurnStartKind::Automatic; "conditional constructor")] +#[tokio::test] +async fn admission_revalidates_constraints_before_committing(kind: TurnStartKind) { + let (session, _turn_context, rx) = make_session_and_context_with_rx().await; + { + let mut state = session.state.lock().await; + Arc::make_mut(&mut state.session_configuration.step_settings).approval_policy = + Constrained::allow_any(AskForApproval::OnRequest); + } + let prepared = PreparedTurnInputSettings::prepare( + &session, + ThreadSettingsOverrides { + approval_policy: Some(AskForApproval::Never), + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + }, + TurnStartOptions::default(), + ) + .await + .expect("approval-policy edit should initially be valid"); + + let approval_policy = Constrained::allow_only(AskForApproval::OnRequest); + let expected_message = CodexErr::InvalidRequest( + approval_policy + .can_set(&AskForApproval::Never) + .expect_err("new constraint must reject the prepared edit") + .to_string(), + ) + .to_string(); + { + let mut state = session.state.lock().await; + Arc::make_mut(&mut state.session_configuration.step_settings).approval_policy = + approval_policy; + } + let desired_settings = session.thread_settings_snapshot().await; + let submission_id = "constraints-after-preview"; + let result = prepared + .apply_started(&session, submission_id.to_string(), kind) + .await; + let Err(error) = result else { + panic!("commit-time constraint failure must return InvalidRequest"); + }; + let CodexErrorDetails::InvalidRequest(message) = error.details() else { + panic!("unexpected commit-time error: {error}"); + }; + assert_eq!(message, &expected_message); + let errors: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()) + .filter(|event| event.id == submission_id) + .map(|event| match event.msg { + EventMsg::Error(error) => error, + other => panic!("unexpected rejected-turn event: {other:?}"), + }) + .collect(); + assert_eq!( + errors, + vec![ErrorEvent { + message: expected_message, + codex_error_info: Some(CodexErrorInfo::BadRequest), + }] + ); + assert_eq!(session.thread_settings_snapshot().await, desired_settings); + assert!(session.active_turn.lock().await.is_none()); + assert_eq!( + session + .input_queue + .get_pending_input(&session.active_turn) + .await, + (Vec::::new(), None, None) + ); +} + #[tokio::test] async fn start_only_accepts_user_input_in_plan_mode() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; @@ -334,6 +717,12 @@ async fn start_only_rejects_empty_user_input_in_plan_mode() { #[tokio::test] async fn start_only_rejects_pending_trigger_turn_without_injecting() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; + { + let mut state = session.state.lock().await; + Arc::make_mut(&mut state.session_configuration.step_settings) + .collaboration_mode + .mode = ModeKind::Plan; + } session .input_queue .enqueue_mailbox_communication( @@ -363,6 +752,7 @@ async fn start_only_rejects_pending_trigger_turn_without_injecting() { ); assert!(session.active_turn.lock().await.is_none()); assert!(session.input_queue.has_trigger_turn_mailbox_items().await); + assert_eq!(session.collaboration_mode().await.mode, ModeKind::Plan); } #[tokio::test] diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index 09d211f4b1..f333ba84f3 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -20,6 +20,7 @@ use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; use core_test_support::streaming_sse::StreamingSseChunk; use core_test_support::streaming_sse::start_streaming_sse_server; +use core_test_support::submit_thread_settings; use core_test_support::test_codex::local; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; @@ -27,6 +28,7 @@ use pretty_assertions::assert_eq; use serde_json::Value; use std::sync::Arc; use std::time::Duration; +use test_case::test_case; use tokio::sync::Barrier; use tokio::sync::oneshot; use tokio::time::timeout; @@ -45,36 +47,45 @@ async fn submit_user_message( codex.start_or_steer_turn(user_message_request(text)).await } +#[test_case(ModeKind::Default, ModeKind::Plan; "automatic input cannot enter Plan")] +#[test_case(ModeKind::Plan, ModeKind::Default; "automatic input cannot leave Plan")] #[tokio::test] -async fn start_turn_if_idle_rejects_non_user_input_that_requests_plan_mode() { +async fn start_turn_if_idle_keeps_automatic_plan_rejections_atomic( + current_mode: ModeKind, + proposed_mode: ModeKind, +) { let server = responses::start_mock_server().await; let test = test_codex() .build_with_auto_env(&server) .await .expect("build turn-input submission session"); - let original_collaboration_mode = test.codex.config_snapshot().await.collaboration_mode; - + let mut collaboration_mode = test.codex.config_snapshot().await.collaboration_mode; + collaboration_mode.mode = current_mode; + submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + collaboration_mode: Some(collaboration_mode.clone()), + ..Default::default() + }, + ) + .await + .expect("set the current collaboration mode"); + let current_settings = test.codex.thread_settings_snapshot().await; + collaboration_mode.mode = proposed_mode; + let overrides = ThreadSettingsOverrides { + collaboration_mode: Some(collaboration_mode.clone()), + ..Default::default() + }; let submission = test .codex .start_turn_if_idle( TurnInputRequest::new(TurnInput::ResponseItem(responses::user_message_item( - "automatic input", + "rejected automatic input", ))) - .with_thread_settings(ThreadSettingsOverrides { - collaboration_mode: Some(CollaborationMode { - mode: ModeKind::Plan, - settings: Settings { - model: test.session_configured.model.clone(), - reasoning_effort: None, - developer_instructions: None, - }, - }), - ..Default::default() - }), + .with_thread_settings(overrides.clone()), ) .await - .expect("idle turn submission should return a typed rejection"); - + .expect("automatic Plan admission should return a typed rejection"); assert_eq!( submission, StartIfIdleSubmission::NotSubmitted { @@ -82,9 +93,36 @@ async fn start_turn_if_idle_rejects_non_user_input_that_requests_plan_mode() { } ); assert_eq!( - test.codex.config_snapshot().await.collaboration_mode, - original_collaboration_mode + test.codex.thread_settings_snapshot().await, + current_settings ); + + // Rejection releases the idle reservation, and an explicit user can make + // either transition without receiving the rejected automatic input. + let response_mock = responses::mount_sse_once( + &server, + responses::sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + ) + .await; + let started = test + .codex + .start_turn_if_idle( + user_message_request("explicit user input").with_thread_settings(overrides), + ) + .await + .expect("rejection must release the idle reservation for explicit user input"); + assert!(matches!(started, StartIfIdleSubmission::Started { .. })); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!( + test.codex.config_snapshot().await.collaboration_mode, + collaboration_mode + ); + let request = response_mock.single_request(); + assert!(request.body_contains_text("explicit user input")); + assert!(!request.body_contains_text("rejected automatic input")); } #[tokio::test]