diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index d5305974ee..647c0e7899 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -534,7 +534,7 @@ impl CodexThread { settings: CodexThreadSettingsOverrides, ) -> ConstraintResult<()> { let updates = self.thread_settings_update(settings).await; - self.session.update_settings(updates).await + self.session.update_settings(updates).await.map(|_| ()) } async fn thread_settings_update( diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d14cadcad9..f0250debe3 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -249,6 +249,7 @@ use self::review::spawn_review_thread; use self::session::AppServerClientMetadata; use self::session::Session; use self::session::SessionConfiguration; +use self::session::SessionSettingsCommit; pub(crate) use self::session::SessionSettingsUpdate; #[cfg(test)] use self::turn::AssistantMessageStreamParsers; @@ -1596,9 +1597,9 @@ impl Session { pub(crate) async fn update_settings( &self, updates: SessionSettingsUpdate, - ) -> ConstraintResult<()> { + ) -> ConstraintResult { let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); - let (previous_config, new_config, permission_profile_changed, mcp_inputs_changed) = { + let (commit, previous_config, new_config, permission_profile_changed, mcp_inputs_changed) = { let mut state = self.state.lock().await; let updated = match self.apply_session_settings(&state.session_configuration, &updates) { @@ -1635,7 +1636,14 @@ impl Session { state.session_configuration = updated; let new_config = notify_config_contributors .then(|| self.build_effective_session_config(&state.session_configuration)); + let commit = SessionSettingsCommit { + configuration: state.session_configuration.clone(), + snapshot: state + .session_configuration + .thread_settings_snapshot(&self.services.turn_environments.selections()), + }; ( + commit, previous_config, new_config, permission_profile_changed, @@ -1650,7 +1658,7 @@ impl Session { if mcp_inputs_changed { self.schedule_mcp_prewarm(); } - Ok(()) + Ok(commit) } pub(crate) async fn preview_settings( diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index fe4d9ba5c7..11afc57169 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -589,6 +589,12 @@ impl SessionConfiguration { } } +/// The configuration and public snapshot published by one settings commit. +pub(crate) struct SessionSettingsCommit { + pub(crate) configuration: SessionConfiguration, + pub(crate) snapshot: ThreadSettingsSnapshot, +} + #[derive(Default, Clone)] pub(crate) struct SessionSettingsUpdate { pub(crate) environments: Option, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 30b27e78d0..c0299fdaf6 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5569,6 +5569,56 @@ async fn session_configuration_apply_preserves_absolute_cwd_write_root_on_cwd_up ); } +#[tokio::test] +async fn session_settings_commit_keeps_snapshot_across_postcommit_wait() { + let (session, _turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| { + config + .permissions + .set_permission_profile(PermissionProfile::workspace_write()) + .expect("set initial permission profile"); + }, + ) + .await; + let refresh_guard = session + .managed_network_proxy_refresh_lock + .acquire() + .await + .expect("network refresh lock"); + let mut first_update = Box::pin(tokio::task::unconstrained(session.update_settings( + SessionSettingsUpdate { + permission_profile: Some(PermissionProfile::read_only()), + service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), + ..Default::default() + }, + ))); + + // Pause after the commit, while its managed-network refresh is blocked. + { + let mut context = std::task::Context::from_waker(futures::task::noop_waker_ref()); + assert!(std::future::Future::poll(first_update.as_mut(), &mut context).is_pending()); + } + let expected = session.thread_settings_snapshot().await; + let later_commit = session + .update_settings(SessionSettingsUpdate { + service_tier: Some(None), + ..Default::default() + }) + .await + .expect("later settings update"); + drop(refresh_guard); + + let commit = first_update.await.expect("first settings update"); + let configuration_snapshot = commit + .configuration + .thread_settings_snapshot(&session.services.turn_environments.selections()); + assert_eq!(commit.snapshot, expected); + assert_eq!(configuration_snapshot, expected); + assert_ne!(later_commit.snapshot, expected); +} + #[tokio::test] async fn session_update_settings_does_not_rewrite_sticky_environment_cwds() { let (session, turn_context) = make_session_and_context().await; @@ -5628,10 +5678,11 @@ async fn permission_profile_updates_apply_to_next_turn_environment() { }; let next_turn = if apply_on_turn_start { - session + let (next_turn, _) = session .new_turn_with_sub_id("permission-profile-update".to_string(), updates) .await - .expect("turn permission profile update should succeed") + .expect("turn permission profile update should succeed"); + next_turn } else { session .update_settings(updates) @@ -5733,7 +5784,7 @@ async fn absolute_cwd_update_with_turn_environment_is_allowed() { }; std::fs::create_dir_all(absolute_cwd.as_path()).expect("create absolute turn dir"); - let turn_context = session + let (turn_context, _) = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { @@ -7284,7 +7335,7 @@ async fn turn_environments_set_primary_environment() { AbsolutePathBuf::try_from(session.get_config().await.cwd.as_path().join("selected")) .expect("absolute path"); - let turn_context = session + let (turn_context, _) = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { @@ -7478,7 +7529,7 @@ async fn primary_environment_uses_first_turn_environment() { async fn empty_turn_environments_clear_primary_environment() { let (session, _turn_context, _rx) = make_session_and_context_with_rx().await; - let turn_context = session + let (turn_context, _) = session .new_turn_with_sub_id( "sub-1".to_string(), SessionSettingsUpdate { diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index 286cf78e06..66f2c3e424 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -10,6 +10,7 @@ use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::ThreadSettingsSnapshot; use std::sync::Arc; /// Applies standalone thread settings and reports invalid overrides through the @@ -89,14 +90,20 @@ pub(super) async fn apply_update( submission_id: String, updates: SessionSettingsUpdate, ) -> ConstraintResult<()> { - session.update_settings(updates).await?; - emit_applied(session, submission_id).await; + let commit = session.update_settings(updates).await?; + emit_applied(session, submission_id, commit.snapshot).await; Ok(()) } -/// Emits the thread-owned settings after a successful update. -pub(super) async fn emit_applied(session: &Session, submission_id: String) { - let msg = applied_event(session).await; +/// Emits the snapshot published by one successful settings update. +pub(super) async fn emit_applied( + session: &Session, + submission_id: String, + snapshot: ThreadSettingsSnapshot, +) { + let msg = EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent { + thread_settings: snapshot, + }); session .send_event_raw_without_materializing_rollout(Event { id: submission_id, @@ -105,8 +112,7 @@ pub(super) async fn emit_applied(session: &Session, submission_id: String) { .await; } -/// Builds the thread-owned settings event used by live updates and -/// synthesized fork history. +/// Builds the current thread-settings event for synthesized fork history. pub(super) async fn applied_event(session: &Session) -> EventMsg { EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent { thread_settings: session.thread_settings_snapshot().await, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 8a4ae62f51..1b815875a4 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -743,59 +743,15 @@ impl Session { &self, sub_id: String, updates: SessionSettingsUpdate, - ) -> CodexResult> { - let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); - let update_result: CodexResult<_> = { - let mut state = self.state.lock().await; - match self.apply_session_settings(&state.session_configuration, &updates) { - Ok(next) => { - let mcp_inputs_changed = - self.mcp_inputs_differ(&state.session_configuration, &next, &updates); - if mcp_inputs_changed { - self.mark_mcp_runtime_dirty(); - } - let previous_permission_profile = - state.session_configuration.permission_profile(); - let next_permission_profile = next.permission_profile(); - let permission_profile_changed = - previous_permission_profile != next_permission_profile; - let previous_config = notify_config_contributors - .then(|| self.build_effective_session_config(&state.session_configuration)); - let environment_config = next.inferred_environment_config(); - if let Some(environments) = &updates.environments { - self.services - .turn_environments - .update_selections(&environments.environments, &environment_config); - } else if state.session_configuration.inferred_environment_config() - != environment_config - { - self.services - .turn_environments - .update_thread_config(&environment_config); - } - state.session_configuration = next.clone(); - let new_config = notify_config_contributors - .then(|| self.build_effective_session_config(&state.session_configuration)); - Ok(( - next, - mcp_inputs_changed, - permission_profile_changed, - previous_config, - new_config, - )) - } - Err(err) => Err(CodexErr::InvalidRequest(err.to_string())), - } - }; - - let ( - mut session_configuration, - mcp_inputs_changed, - permission_profile_changed, - previous_config, - new_config, - ) = match update_result { - Ok(update) => update, + ) -> CodexResult<(Arc, 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(); self.send_event_raw(Event { @@ -809,26 +765,15 @@ impl Session { return Err(CodexErr::InvalidRequest(message)); } }; - self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref()); - if mcp_inputs_changed { - self.schedule_mcp_prewarm(); - } - - if permission_profile_changed { - self.refresh_managed_network_proxy_for_current_permission_profile() - .await; - } + let mut configuration = commit.configuration; // Apply the override only to the turn's copy, after persisting thread settings. - if let Some(service_tier) = updates.service_tier_for_turn { - session_configuration.service_tier = Some(service_tier); + if let Some(service_tier) = service_tier_for_turn { + configuration.service_tier = Some(service_tier); } - Ok(self - .new_turn_from_configuration( - sub_id, - session_configuration, - updates.final_output_json_schema, - ) - .await) + let turn_context = self + .new_turn_from_configuration(sub_id, configuration, final_output_json_schema) + .await; + Ok((turn_context, commit.snapshot)) } async fn new_turn_from_configuration( diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 1e8bc70340..3e49c20b4a 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -108,11 +108,11 @@ impl PreparedTurnInputSettings { updates.service_tier_for_turn = service_tier; // new_turn_with_sub_id already emits an error event when settings are invalid. - let turn_context = session + let (turn_context, settings_snapshot) = session .new_turn_with_sub_id(submission_id.clone(), updates) .await?; if emit_thread_settings_applied { - thread_settings::emit_applied(session, submission_id).await; + thread_settings::emit_applied(session, submission_id, settings_snapshot).await; } if let Some(parent_turn_id) = parent_turn_id { turn_context diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index d64b5c7439..41728a6684 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -1561,7 +1561,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { ) .await .expect("resume source thread"); - let resumed_turn = resumed + let (resumed_turn, _) = resumed .thread .session .new_turn_with_sub_id("resume-turn".to_string(), SessionSettingsUpdate::default()) @@ -1595,7 +1595,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { ) .await .expect("fork source thread"); - let forked_turn = forked + let (forked_turn, _) = forked .thread .session .new_turn_with_sub_id("fork-turn".to_string(), SessionSettingsUpdate::default()) diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 47c1b598cf..fec77eaef7 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -145,6 +145,7 @@ mod safety_buffering; mod safety_check_downgrade; mod search_tool; mod send_user_message_async; +mod settings_commits; mod shell_snapshot; mod skill_approval; mod skills; diff --git a/codex-rs/core/tests/suite/settings_commits.rs b/codex-rs/core/tests/suite/settings_commits.rs new file mode 100644 index 0000000000..f9a6fe70c6 --- /dev/null +++ b/codex-rs/core/tests/suite/settings_commits.rs @@ -0,0 +1,171 @@ +use anyhow::Result; +use codex_core::TurnInputRequest; +use codex_core::TurnInputSubmission; +use codex_core::config::Config; +use codex_extension_api::ConfigContributor; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::user_input::UserInput; +use core_test_support::responses; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use pretty_assertions::assert_eq; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::mpsc; +use std::time::Duration; +use test_case::test_case; +use tokio::sync::oneshot; +use tokio::time::timeout; + +const INITIAL_MODEL: &str = "gpt-5.4"; +const COMMITTED_MODEL: &str = "gpt-5.2"; +const TIMEOUT: Duration = Duration::from_secs(10); + +struct PauseAfterCommit { + gate: Mutex, mpsc::Receiver<()>)>>, +} + +impl ConfigContributor for PauseAfterCommit { + fn on_config_changed( + &self, + _session_store: &ExtensionData, + _thread_store: &ExtensionData, + _previous_config: &Config, + new_config: &Config, + ) { + if new_config.model.as_deref() != Some(COMMITTED_MODEL) { + return; + } + let Some((entered, release)) = self.gate.lock().expect("commit gate lock").take() else { + return; + }; + entered.send(()).expect("test is waiting for the commit"); + // The callback is synchronous, so this test uses a second runtime worker. + release + .recv_timeout(TIMEOUT) + .expect("test releases the committed update"); + } +} + +#[derive(Clone, Copy)] +enum SettingsOperation { + TurnStart, + Standalone, +} + +#[test_case(SettingsOperation::TurnStart; "turn start retains its committed settings and notification")] +#[test_case(SettingsOperation::Standalone; "standalone notification retains its committed settings")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn settings_notifications_keep_their_commit_across_postcommit_work( + operation: SettingsOperation, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let response = responses::mount_sse_once( + &server, + responses::sse(vec![ + responses::ev_response_created("response"), + responses::ev_completed("response"), + ]), + ) + .await; + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let mut extensions = ExtensionRegistryBuilder::::new(); + extensions.config_contributor(Arc::new(PauseAfterCommit { + gate: Mutex::new(Some((entered_tx, release_rx))), + })); + let test = test_codex() + .with_model(INITIAL_MODEL) + .with_extensions(Arc::new(extensions.build())) + .build_with_auto_env(&server) + .await?; + let initial = test.codex.restorable_thread_settings().await; + let thread_settings = ThreadSettingsOverrides { + model: Some(COMMITTED_MODEL.to_string()), + ..Default::default() + }; + let submission = tokio::spawn({ + let codex = Arc::clone(&test.codex); + async move { + match operation { + SettingsOperation::TurnStart => { + let result = codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "use the committed model".to_string(), + text_elements: Vec::new(), + }]) + .with_thread_settings(thread_settings), + ) + .await?; + let TurnInputSubmission::Started { turn_id } = result else { + panic!("expected a new turn, got {result:?}"); + }; + Ok(turn_id) + } + SettingsOperation::Standalone => { + codex.submit(Op::ThreadSettings { thread_settings }).await + } + } + } + }); + + timeout(TIMEOUT, entered_rx).await??; + let expected = test.codex.thread_settings_snapshot().await; + assert_eq!(expected.model, COMMITTED_MODEL); + // Submitted operations are serialized. Runtime restoration is an existing + // direct writer, so it can overlap the first operation's post-commit work. + timeout(TIMEOUT, test.codex.restore_thread_settings(initial)).await??; + let restored = test.codex.thread_settings_snapshot().await; + assert_eq!(restored.model, INITIAL_MODEL); + release_tx.send(())?; + let submission_id = timeout(TIMEOUT, submission).await???; + + let applied = timeout(TIMEOUT, async { + loop { + let event = test.codex.next_event().await?; + match event.msg { + EventMsg::ThreadSettingsApplied(applied) if event.id == submission_id => { + return Ok::<_, anyhow::Error>(applied.thread_settings); + } + EventMsg::Error(error) => { + anyhow::bail!("settings update failed: {}", error.message) + } + _ => {} + } + } + }) + .await??; + assert_eq!(applied, expected); + + let expected_request_model = match operation { + SettingsOperation::TurnStart => COMMITTED_MODEL, + SettingsOperation::Standalone => { + assert!(response.requests().is_empty()); + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "use the restored settings".to_string(), + text_elements: Vec::new(), + }])) + .await?; + INITIAL_MODEL + } + }; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!( + response.single_request().body_json()["model"], + expected_request_model + ); + assert_eq!(test.codex.thread_settings_snapshot().await, restored); + Ok(()) +}