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

Keep settings operations bound to their committed snapshot (#40645)

Browse Source 
## Why

Post-commit work can overlap with another settings write. Reading the session
again afterward could make a settings operation report or use the later state
instead of the state that it committed.

## What changed

- Return the committed configuration and `ThreadSettingsSnapshot` from session
  settings updates.
- Use that commit result when creating a turn and emitting
  `ThreadSettingsApplied`, while retaining current-state snapshots for
  synthesized fork history.
- Route turn-start updates through the shared settings update path.

## Testing

Add coverage for standalone and turn-start updates that pause after committing,
restore different settings concurrently, and verify that each operation keeps
its original configuration and notification snapshot.

GitOrigin-RevId: cee5be9536d2840170893e6eb0c64cd039fb0a96
This commit is contained in:
joeflorencio-openai
2026-08-25 17:20:37 +00:00
committed by copyberry
parent a95391160a
commit 4e883ce36e
10 changed files with 279 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex_thread.rs
+1 -1
View File
@@ -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(
codex-rs/core/src/session/mod.rs
+11 -3
View File
@@ -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<SessionSettingsCommit> {
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(
codex-rs/core/src/session/session.rs
+6
View File
@@ -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<TurnEnvironmentSelections>,
codex-rs/core/src/session/tests.rs
+56 -5
View File
@@ -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 {
codex-rs/core/src/session/thread_settings.rs
+13 -7
View File
@@ -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,
codex-rs/core/src/session/turn_context.rs
+16 -71
View File
@@ -743,59 +743,15 @@ impl Session {
&self,
sub_id: String,
updates: SessionSettingsUpdate,
) -> CodexResult<Arc<TurnContext>> {
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<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();
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(
codex-rs/core/src/session/turn_input.rs
+2 -2
View File
@@ -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
codex-rs/core/src/thread_manager_tests.rs
+2 -2
View File
@@ -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())
codex-rs/core/tests/suite/mod.rs
+1
View File
@@ -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;
codex-rs/core/tests/suite/settings_commits.rs
+171
View File
@@ -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<Option<(oneshot::Sender<()>, mpsc::Receiver<()>)>>,
}
impl ConfigContributor<Config> 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::<Config>::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(())
}