github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 05:22:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make automatic turn admission atomic with settings updates (#40648)

Browse Source 
## Why

Turn settings can change after an input request previews its overrides. Automatic idle work could therefore make a Plan-mode admission decision from stale settings or apply a sparse override built from an outdated collaboration mode.

## What changed

- Evaluate automatic turn admission against both the current and proposed configuration under the settings publication lock.
- Reject automatic work that would enter or leave Plan mode without committing settings or triggering runtime configuration effects.
- Keep model and reasoning-effort overrides sparse until they are merged with the configuration being committed.
- Continue allowing explicit user and recovery turns to make valid mode transitions.

## Testing

Add coverage for concurrent settings changes, sparse model and effort updates, commit-time constraint validation, atomic Plan-mode rejection, and subsequent explicit user submission.

GitOrigin-RevId: 234f06de116bdce7c3a3fb5cac3d15583304a253
This commit is contained in:
joeflorencio-openai
2026-08-25 17:30:45 +00:00
committed by copyberry
parent 1d64085e67
commit ba6cf9c692
9 changed files with 756 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex_thread.rs
+6 -16
View File
@@ -522,7 +522,7 @@ impl CodexThread {
&self,
overrides: CodexThreadSettingsOverrides,
) -> ConstraintResult<ThreadConfigSnapshot> {
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,
codex-rs/core/src/session/mod.rs
+23 -1
View File
@@ -1602,6 +1602,24 @@ impl Session {
&self,
updates: SessionSettingsUpdate,
) -> ConstraintResult<SessionSettingsCommit> {
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<Option<SessionSettingsCommit>> {
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(
codex-rs/core/src/session/step_settings.rs
+15 -4
View File
@@ -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<String>,
pub(crate) effort: Option<Option<ReasoningEffort>>,
/// A complete collaboration mode takes precedence over model and effort edits.
pub(crate) collaboration_mode: Option<CollaborationMode>,
pub(crate) reasoning_summary: Option<ReasoningSummary>,
pub(crate) service_tier: Option<Option<String>>,
@@ -92,9 +99,13 @@ impl StepSettings {
constraints: &StepSettingsConstraints<'_>,
) -> ConstraintResult<Self> {
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);
}
codex-rs/core/src/session/step_settings_tests.rs
+113
View File
@@ -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
}
);
}
codex-rs/core/src/session/thread_settings.rs
+5 -19
View File
@@ -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,
codex-rs/core/src/session/turn_context.rs
+29 -10
View File
@@ -758,18 +758,36 @@ impl Session {
sub_id: String,
updates: SessionSettingsUpdate,
) -> CodexResult<(Arc<TurnContext>, 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<Option<(Arc<TurnContext>, 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,
codex-rs/core/src/session/turn_input.rs
+107 -57
View File
@@ -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<Session>,
submission_id: String,
) -> CodexResult<Arc<TurnContext>> {
kind: TurnStartKind,
) -> CodexResult<Option<Arc<TurnContext>>> {
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<TurnInputSubmission> {
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<Session>,
request: TurnInputRequest,
submission_id: String,
is_recovery: bool,
kind: TurnStartKind,
) -> CodexResult<TurnInputSubmission> {
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<String, AdditionalContextEntry>,
codex-rs/core/src/session/turn_input_tests.rs
+401 -11
View File
@@ -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::<TurnInput>::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<std::sync::Mutex<Vec<(AskForApproval, ApprovalsReviewer)>>>);
impl codex_extension_api::ConfigContributor<crate::config::Config> 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::<crate::config::Config>::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::<TurnInput>::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::<TurnInput>::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]
codex-rs/core/tests/suite/turn_input_submission.rs
+57 -19
View File
@@ -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]