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.