From fbd0c0d96ca0fbc13360b6fd844e45053af7d1ee Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Thu, 24 Sep 2026 20:45:55 +0000 Subject: [PATCH] Add direct replies for thread settings updates (#47937) ## Why Callers need to await acceptance or rejection of thread settings without consuming the thread event stream. Acceptance must precede event delivery because the caller may hold a lock needed by the event consumer. ## What changed - Add `CodexThread::update_thread_settings` and re-export `ThreadSettingsOverrides` from `core-api`. - Add an optional reply channel to `Op::ThreadSettings`. Return validation failures as `InvalidRequest` through that channel, retaining error events for callers without a reply channel. - Reply to successful updates before emitting `ThreadSettingsApplied`, while holding the settings persistence lock through notification. ## Testing Extend settings constraint coverage to direct replies, including rejection without an error event and successful notification. Verify that acceptance and continuation invalidation precede blocked event delivery, while checkpointing waits for notification. GitOrigin-RevId: e09b0d3cc4b6750709c173371632f19619e5c250 --- .../src/request_processors/turn_processor.rs | 5 ++- codex-rs/core-api/src/lib.rs | 1 + codex-rs/core/src/codex_thread.rs | 16 ++++++++ codex-rs/core/src/session/handlers.rs | 33 ++++++++++++++- codex-rs/core/src/session/tests.rs | 40 +++++++++++++++---- codex-rs/core/src/session/thread_settings.rs | 35 ++++------------ codex-rs/core/tests/common/lib.rs | 7 +++- .../core/tests/suite/agent_eviction_tests.rs | 1 + codex-rs/core/tests/suite/remote_env.rs | 1 + .../scenarios_indirect_namespace_prefixes.rs | 1 + codex-rs/core/tests/suite/settings_commits.rs | 7 +++- .../core/tests/suite/settings_constraints.rs | 38 +++++++++++++++++- codex-rs/core/tests/suite/step_settings.rs | 4 ++ codex-rs/protocol/src/protocol.rs | 3 ++ 14 files changed, 150 insertions(+), 42 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 0594e8114e..e42e53de1f 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -959,7 +959,10 @@ impl TurnRequestProcessor { self.submit_core_op( request_id, thread.as_ref(), - Op::ThreadSettings { thread_settings }, + Op::ThreadSettings { + thread_settings, + reply: None, + }, ) .await .map_err(|err| internal_error(format!("failed to update thread settings: {err}")))?; diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index a8facc3ae1..4a4047dd4e 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -130,6 +130,7 @@ pub use codex_protocol::protocol::EventMsg; pub use codex_protocol::protocol::Op; pub use codex_protocol::protocol::SessionConfiguredEvent; pub use codex_protocol::protocol::SessionSource; +pub use codex_protocol::protocol::ThreadSettingsOverrides; pub use codex_protocol::protocol::TurnEnvironmentSelection; pub use codex_protocol::protocol::W3cTraceContext; pub use codex_protocol::user_input::UserInput; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index ccbb5d54c8..9d2c3adb43 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -45,6 +45,7 @@ use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; +use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TokenUsageInfo; @@ -586,6 +587,21 @@ impl CodexThread { self.session.preview_settings(&updates).await } + /// Queues settings for future turns and waits for Core to accept or reject them. + /// Rejections are returned to the caller instead of emitted as thread errors. + pub async fn update_thread_settings( + &self, + thread_settings: ThreadSettingsOverrides, + ) -> CodexResult<()> { + let (reply, result) = oneshot::channel(); + self.submit(Op::ThreadSettings { + thread_settings, + reply: Some(reply), + }) + .await?; + result.await.unwrap_or(Err(CodexErr::InternalAgentDied)) + } + /// Restores thread-owned mutable settings captured from another loaded runtime. /// /// Runtime replacement uses this after resume so clients keep their current thread settings diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index c6648e3df6..cc8413a653 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -23,6 +23,7 @@ use crate::tasks::CompactTask; use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; +use codex_protocol::error::CodexErr; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; @@ -516,8 +517,36 @@ pub(super) async fn submission_loop( let _ = reply.send(result); should_exit } - Op::ThreadSettings { thread_settings } => { - thread_settings::update(&sess, sub.id.clone(), thread_settings).await; + Op::ThreadSettings { + thread_settings, + reply, + } => { + let _settings_guard = thread_settings::acquire_persistence_lock(&sess).await; + match thread_settings::update(&sess, thread_settings).await { + Ok(snapshot) => { + // Reply first: the caller may hold a lock its event consumer needs. + if let Some(reply) = reply { + let _ = reply.send(Ok(())); + } + thread_settings::emit_applied(&sess, sub.id.clone(), snapshot).await; + } + Err(error) => { + let message = format!("invalid thread settings override: {error}"); + if let Some(reply) = reply { + let _ = reply.send(Err(CodexErr::InvalidRequest(message))); + } else { + sess.send_event_raw(Event { + id: sub.id.clone(), + msg: EventMsg::Error(ErrorEvent { + misalignment: None, + message, + codex_error_info: Some(CodexErrorInfo::BadRequest), + }), + }) + .await; + } + } + } false } Op::TurnSettings { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 42e14804ee..1b43d3932c 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5718,7 +5718,7 @@ async fn mcp_attribution_checkpoints_cover_batch_prefixes_compaction_and_restore #[tokio::test] async fn standalone_settings_invalidate_continuation_before_delivering_acceptance() { - let (mut session, _) = make_session_and_context().await; + let (mut session, turn_context) = make_session_and_context().await; let (tx, rx) = async_channel::bounded(1); session.tx_event = tx; session.state.lock().await.last_started_turn_id = Some("superseded-turn".into()); @@ -5736,18 +5736,44 @@ async fn standalone_settings_invalidate_continuation_before_delivering_acceptanc .await .expect("fill event channel"); let session = Arc::new(session); - let mut update = Box::pin(tokio::task::unconstrained(thread_settings::update( - &session, - "settings".into(), - codex_protocol::protocol::ThreadSettingsOverrides::default(), + let (reply, mut accepted) = tokio::sync::oneshot::channel(); + let (tx_sub, rx_sub) = async_channel::bounded(1); + tx_sub + .send(Submission { + id: "settings".into(), + op: Op::ThreadSettings { + thread_settings: codex_protocol::protocol::ThreadSettingsOverrides::default(), + reply: Some(reply), + }, + trace: None, + parent_turn_id: None, + root_turn_id: None, + residency_guard: None, + }) + .await + .expect("submit settings"); + let mut submissions = Box::pin(tokio::task::unconstrained(submission_loop( + Arc::clone(&session), + turn_context.config, + rx_sub, ))); - assert!(futures::poll!(update.as_mut()).is_pending()); + assert!(futures::poll!(submissions.as_mut()).is_pending()); assert_eq!(session.state.lock().await.last_started_turn_id, None); + accepted + .try_recv() + .expect("receive acceptance before the event is delivered") + .expect("settings accepted"); let mut checkpoint = Box::pin(session.checkpoint_thread_settings()); assert!(futures::poll!(checkpoint.as_mut()).is_pending()); rx.recv().await.expect("release event delivery"); - update.await; + assert!(futures::poll!(submissions.as_mut()).is_pending()); checkpoint.await.expect("checkpoint after settings update"); + assert!(matches!( + rx.recv().await.expect("receive settings event").msg, + EventMsg::ThreadSettingsApplied(_) + )); + drop(tx_sub); + submissions.await; } #[tokio::test] diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index 36d6b86e9d..21059048fb 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -6,15 +6,12 @@ use super::session::SessionSettingsUpdate; use super::step_settings::StepSettingsUpdate; use crate::config::ConstraintResult; use codex_history::RolloutItem; -use codex_protocol::protocol::CodexErrorInfo; -use codex_protocol::protocol::ErrorEvent; 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 codex_thread_store::ThreadStoreResult; -use std::sync::Arc; use tokio::sync::SemaphorePermit; impl Session { @@ -31,34 +28,16 @@ impl Session { } } -/// Applies standalone thread settings and reports invalid overrides through the -/// normal event stream. +/// Applies standalone thread settings. The caller holds the persistence permit through notification. pub(super) async fn update( - session: &Arc, - submission_id: String, + session: &Session, overrides: ThreadSettingsOverrides, -) { +) -> ConstraintResult { let updates = prepare_update(overrides); - let _settings_guard = acquire_persistence_lock(session).await; - match session.update_settings(updates).await { - Ok(commit) => { - // Standalone settings changes supersede a pending automatic continuation. - session.state.lock().await.last_started_turn_id = None; - emit_applied(session, submission_id, commit.snapshot).await; - } - Err(error) => { - session - .send_event_raw(Event { - id: submission_id, - msg: EventMsg::Error(ErrorEvent { - misalignment: None, - message: format!("invalid thread settings override: {error}"), - codex_error_info: Some(CodexErrorInfo::BadRequest), - }), - }) - .await; - } - } + let commit = session.update_settings(updates).await?; + // Standalone settings changes supersede a pending automatic continuation. + session.state.lock().await.last_started_turn_id = None; + Ok(commit.snapshot) } /// Converts protocol overrides into the internal settings update shape. diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index 424cdb0f9c..0b05675541 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -313,7 +313,12 @@ pub async fn submit_thread_settings( use tokio::time::Duration; use tokio::time::timeout; - let submission_id = codex.submit(Op::ThreadSettings { thread_settings }).await?; + let submission_id = codex + .submit(Op::ThreadSettings { + thread_settings, + reply: None, + }) + .await?; loop { let ev = timeout(Duration::from_secs(10), codex.next_event()) .await diff --git a/codex-rs/core/tests/suite/agent_eviction_tests.rs b/codex-rs/core/tests/suite/agent_eviction_tests.rs index 8e2d12ced9..ae7c6e282d 100644 --- a/codex-rs/core/tests/suite/agent_eviction_tests.rs +++ b/codex-rs/core/tests/suite/agent_eviction_tests.rs @@ -129,6 +129,7 @@ async fn queued_mail_and_cancelled_eviction_keep_worker_ownership() -> Result<() model: Some(COMMITTED_MODEL.to_string()), ..Default::default() }, + reply: None, }) .await?; timeout(Duration::from_secs(10), entered_rx).await??; diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index 6ad3d66c3e..1b21ec0e1b 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -624,6 +624,7 @@ async fn environment_permissions_follow_configuration_ownership() -> Result<()> permission_profile: Some(PermissionProfile::workspace_write()), ..Default::default() }, + reply: None, }) .await?; let persisted_settings = wait_for_event_match(&test.codex, |event| match event { diff --git a/codex-rs/core/tests/suite/scenarios_indirect_namespace_prefixes.rs b/codex-rs/core/tests/suite/scenarios_indirect_namespace_prefixes.rs index f1b96caa31..aeb1b0c197 100644 --- a/codex-rs/core/tests/suite/scenarios_indirect_namespace_prefixes.rs +++ b/codex-rs/core/tests/suite/scenarios_indirect_namespace_prefixes.rs @@ -145,6 +145,7 @@ text({descriptions, echo: report.structuredContent.echo, agent: agents.agents[0] model: Some("prefixes-cleared".to_string()), ..Default::default() }, + reply: None, }) .await?; test.submit_text_turn("Find echo again, call it, and list the agents.") diff --git a/codex-rs/core/tests/suite/settings_commits.rs b/codex-rs/core/tests/suite/settings_commits.rs index 9c8a9db591..ea368e2c03 100644 --- a/codex-rs/core/tests/suite/settings_commits.rs +++ b/codex-rs/core/tests/suite/settings_commits.rs @@ -226,7 +226,12 @@ async fn settings_notifications_keep_their_commit_across_postcommit_work( Ok(turn_id) } SettingsOperation::Standalone => { - codex.submit(Op::ThreadSettings { thread_settings }).await + codex + .submit(Op::ThreadSettings { + thread_settings, + reply: None, + }) + .await } } } diff --git a/codex-rs/core/tests/suite/settings_constraints.rs b/codex-rs/core/tests/suite/settings_constraints.rs index ecdca6461c..e85d9a2351 100644 --- a/codex-rs/core/tests/suite/settings_constraints.rs +++ b/codex-rs/core/tests/suite/settings_constraints.rs @@ -7,6 +7,7 @@ use codex_core::config::Constrained; use codex_features::Feature; use codex_models_manager::model_info::model_info_from_slug; use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::error::CodexErrorDetails; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ApprovalMessages; use codex_protocol::openai_models::ModelsResponse; @@ -35,6 +36,7 @@ const PERMISSION_INSTRUCTIONS: &str = "Protected-model read-only instructions."; #[derive(Clone, Copy)] enum SettingsOperation { Standalone, + StandaloneWithReply, TurnStart, } @@ -45,7 +47,35 @@ impl SettingsOperation { thread_settings: ThreadSettingsOverrides, ) -> Result { let id = match self { - Self::Standalone => codex.submit(Op::ThreadSettings { thread_settings }).await?, + Self::Standalone => { + codex + .submit(Op::ThreadSettings { + thread_settings, + reply: None, + }) + .await? + } + Self::StandaloneWithReply => { + if let Err(error) = codex.update_thread_settings(thread_settings).await { + assert!(matches!( + error.details(), + CodexErrorDetails::InvalidRequest(_) + )); + return Err(error.into()); + } + // A successful update still notifies subscribers. A prior direct rejection must not. + return match wait_for_event(codex, |event| { + matches!( + event, + EventMsg::ThreadSettingsApplied(_) | EventMsg::Error(_) + ) + }) + .await + { + EventMsg::ThreadSettingsApplied(applied) => Ok(applied.thread_settings), + event => anyhow::bail!("unexpected settings event: {event:?}"), + }; + } Self::TurnStart => { let result = codex .start_or_steer_turn( @@ -83,6 +113,7 @@ impl SettingsOperation { } #[test_case(SettingsOperation::Standalone; "standalone settings")] +#[test_case(SettingsOperation::StandaloneWithReply; "standalone settings with reply")] #[test_case(SettingsOperation::TurnStart; "turn-start settings")] #[tokio::test] async fn protected_model_settings_use_the_proposed_permissions( @@ -202,7 +233,10 @@ async fn protected_model_settings_use_the_proposed_permissions( ); assert_eq!(test.codex.thread_settings_snapshot().await, expected); - if let SettingsOperation::Standalone = operation { + if matches!( + operation, + SettingsOperation::Standalone | SettingsOperation::StandaloneWithReply + ) { test.submit_text_turn("use the committed settings").await?; } let request = response.single_request(); diff --git a/codex-rs/core/tests/suite/step_settings.rs b/codex-rs/core/tests/suite/step_settings.rs index 9006687a29..c532200b1c 100644 --- a/codex-rs/core/tests/suite/step_settings.rs +++ b/codex-rs/core/tests/suite/step_settings.rs @@ -491,6 +491,7 @@ async fn tool_result_history_keeps_originating_model_across_switch_and_replay() model: Some(model.to_string()), ..Default::default() }, + reply: None, }) .await?; test.submit_text_turn("review previous diagnostics").await?; @@ -601,6 +602,7 @@ async fn tool_result_history_keeps_originating_model_across_switch_and_replay() model: Some(model.to_string()), ..Default::default() }, + reply: None, }) .await?; thread @@ -781,6 +783,7 @@ async fn settings_updates_preserve_turn_identity_and_target(target: SettingsTarg service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), ..Default::default() }, + reply: None, }) .await?; } @@ -1979,6 +1982,7 @@ async fn sparse_updates_preserve_divergent_active_and_future_models() -> Result< service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())), ..Default::default() }, + reply: None, }) .await?; apply_turn_settings( diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index a025b0bd29..3c2f05ac2b 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -647,6 +647,9 @@ pub enum Op { ThreadSettings { /// Sparse thread-settings overrides to apply. thread_settings: ThreadSettingsOverrides, + /// When present, report validation errors here instead of emitting an error event. + /// Successful updates still emit `ThreadSettingsApplied` for all callers. + reply: Option>>, }, /// Update only the named running turn, without changing future settings.