From a46e9baa4be7ee142b99e9691e6a9d4ed83f05cc Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Fri, 15 May 2026 18:03:20 -0700 Subject: [PATCH] Add queued turn context core op --- codex-rs/core/src/session/handlers.rs | 130 ++++++++++++-- codex-rs/core/src/session/tests.rs | 33 +--- codex-rs/core/src/session/turn.rs | 1 + codex-rs/protocol/src/protocol.rs | 169 ++++++++++++------- codex-rs/rollout-trace/src/protocol_event.rs | 2 + codex-rs/rollout/src/policy.rs | 1 + 6 files changed, 236 insertions(+), 100 deletions(-) diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 653d1311d0..d9c5f76c9b 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -43,6 +43,9 @@ use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::ThreadMemoryMode; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::TurnAbortReason; +use codex_protocol::protocol::TurnContextAppliedEvent; +use codex_protocol::protocol::TurnContextOverrides; +use codex_protocol::protocol::TurnContextSnapshot; use codex_protocol::protocol::WarningEvent; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::request_user_input::RequestUserInputResponse; @@ -93,13 +96,95 @@ pub async fn user_input_or_turn(sess: &Arc, sub_id: String, op: Op) { .await; } +pub async fn update_turn_context(sess: &Arc, sub_id: String, op: Op) { + let Op::TurnContext { + turn_context: + TurnContextOverrides { + cwd, + approval_policy, + approvals_reviewer, + sandbox_policy, + permission_profile, + active_permission_profile, + windows_sandbox_level, + model, + effort, + summary, + service_tier, + collaboration_mode, + personality, + }, + } = op + else { + unreachable!(); + }; + let collaboration_mode = if let Some(collaboration_mode) = collaboration_mode { + Some(collaboration_mode) + } else { + let state = sess.state.lock().await; + Some( + state + .session_configuration + .collaboration_mode + .with_updates(model, effort, /*developer_instructions*/ None), + ) + }; + let msg = match sess + .update_settings(SessionSettingsUpdate { + cwd, + approval_policy, + approvals_reviewer, + sandbox_policy, + permission_profile, + active_permission_profile, + windows_sandbox_level, + collaboration_mode, + reasoning_summary: summary, + service_tier, + personality, + ..Default::default() + }) + .await + { + Ok(()) => turn_context_applied_event(sess).await, + Err(err) => EventMsg::Error(ErrorEvent { + message: format!("invalid turn context override: {err}"), + codex_error_info: Some(CodexErrorInfo::Other), + }), + }; + sess.send_event_raw(Event { id: sub_id, msg }).await; +} + +async fn turn_context_applied_event(sess: &Session) -> EventMsg { + let snapshot = { + let state = sess.state.lock().await; + state.session_configuration.thread_config_snapshot() + }; + EventMsg::TurnContextApplied(TurnContextAppliedEvent { + turn_context: TurnContextSnapshot { + model: snapshot.model, + model_provider_id: snapshot.model_provider_id, + service_tier: snapshot.service_tier, + approval_policy: snapshot.approval_policy, + approvals_reviewer: snapshot.approvals_reviewer, + permission_profile: snapshot.permission_profile, + active_permission_profile: snapshot.active_permission_profile, + cwd: snapshot.cwd, + reasoning_effort: snapshot.reasoning_effort, + reasoning_summary: snapshot.reasoning_summary, + personality: snapshot.personality, + collaboration_mode: snapshot.collaboration_mode, + }, + }) +} + pub(super) async fn user_input_or_turn_inner( sess: &Arc, sub_id: String, op: Op, mirror_user_text_to_realtime: Option<()>, ) { - let (items, updates, responsesapi_client_metadata) = match op { + let (items, updates, responsesapi_client_metadata, emit_turn_context_applied) = match op { Op::UserTurn { cwd, approval_policy, @@ -146,26 +231,30 @@ pub(super) async fn user_input_or_turn_inner( app_server_client_version: None, }, None, + false, ) } Op::UserInputWithTurnContext { - cwd, - approval_policy, - approvals_reviewer, - sandbox_policy, - permission_profile, - active_permission_profile, - windows_sandbox_level, - model, - effort, - summary, - service_tier, final_output_json_schema, items, responsesapi_client_metadata, - collaboration_mode, - personality, environments, + turn_context: + TurnContextOverrides { + cwd, + approval_policy, + approvals_reviewer, + sandbox_policy, + permission_profile, + active_permission_profile, + windows_sandbox_level, + model, + effort, + summary, + service_tier, + collaboration_mode, + personality, + }, } => { let collaboration_mode = if let Some(collab_mode) = collaboration_mode { Some(collab_mode) @@ -198,6 +287,7 @@ pub(super) async fn user_input_or_turn_inner( app_server_client_version: None, }, responsesapi_client_metadata, + true, ) } Op::UserInput { @@ -213,6 +303,7 @@ pub(super) async fn user_input_or_turn_inner( ..Default::default() }, responsesapi_client_metadata, + false, ), _ => unreachable!(), }; @@ -221,6 +312,13 @@ pub(super) async fn user_input_or_turn_inner( // new_turn_with_sub_id already emits the error event. return; }; + if emit_turn_context_applied { + sess.send_event_raw(Event { + id: sub_id.clone(), + msg: turn_context_applied_event(sess).await, + }) + .await; + } sess.maybe_emit_unknown_model_warning_for_turn(current_context.as_ref()) .await; let accepted_items = match sess @@ -768,6 +866,10 @@ pub(super) async fn submission_loop( user_input_or_turn(&sess, sub.id.clone(), sub.op).await; false } + Op::TurnContext { .. } => { + update_turn_context(&sess, sub.id.clone(), sub.op).await; + false + } Op::InterAgentCommunication { communication } => { inter_agent_communication(&sess, sub.id.clone(), communication).await; false diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d20184bac5..c48024486a 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -121,6 +121,7 @@ use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnContextOverrides; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::W3cTraceContext; @@ -2229,19 +2230,11 @@ async fn fork_startup_context_then_first_turn_diff_snapshot() -> anyhow::Result< }], final_output_json_schema: None, responsesapi_client_metadata: None, - cwd: None, - approval_policy: Some(AskForApproval::Never), - approvals_reviewer: None, - sandbox_policy: None, - permission_profile: None, - active_permission_profile: None, - windows_sandbox_level: None, - model: None, - effort: None, - summary: None, - service_tier: None, - collaboration_mode: Some(collaboration_mode), - personality: None, + turn_context: TurnContextOverrides { + approval_policy: Some(AskForApproval::Never), + collaboration_mode: Some(collaboration_mode), + ..Default::default() + }, }) .await?; wait_for_event(&forked.thread, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; @@ -5067,19 +5060,7 @@ fn op_kind_distinguishes_turn_ops() { items: vec![], final_output_json_schema: None, responsesapi_client_metadata: None, - cwd: None, - approval_policy: None, - approvals_reviewer: None, - sandbox_policy: None, - permission_profile: None, - active_permission_profile: None, - windows_sandbox_level: None, - model: None, - effort: None, - summary: None, - service_tier: None, - collaboration_mode: None, - personality: None, + turn_context: TurnContextOverrides::default(), } .kind(), "user_input_with_turn_context" diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 86710e4871..5d9b3b0c08 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1463,6 +1463,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { | EventMsg::ContextCompacted(_) | EventMsg::ThreadRolledBack(_) | EventMsg::TurnStarted(_) + | EventMsg::TurnContextApplied(_) | EventMsg::TurnComplete(_) | EventMsg::TokenCount(_) | EventMsg::UserMessage(_) diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index f55a60b387..196c6d7d61 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -395,6 +395,71 @@ pub struct ConversationTextParams { pub text: String, } +/// Persistent turn-context overrides that can be applied before user input or +/// on their own. +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, JsonSchema)] +pub struct TurnContextOverrides { + /// Updated `cwd` for sandbox/tool calls. + #[serde(skip_serializing_if = "Option::is_none")] + pub cwd: Option, + + /// Updated command approval policy. + #[serde(skip_serializing_if = "Option::is_none")] + pub approval_policy: Option, + + /// Updated approval reviewer for future approval prompts. + #[serde(skip_serializing_if = "Option::is_none")] + pub approvals_reviewer: Option, + + /// Updated sandbox policy for tool calls. + #[serde(skip_serializing_if = "Option::is_none")] + pub sandbox_policy: Option, + + /// Updated permissions profile for tool calls. + #[serde(skip_serializing_if = "Option::is_none")] + pub permission_profile: Option, + + /// Named or built-in profile that produced `permission_profile`, if the + /// update selected a profile rather than supplying raw permissions. + #[serde(skip_serializing_if = "Option::is_none")] + pub active_permission_profile: Option, + + /// Updated Windows sandbox mode for tool execution. + #[serde(skip_serializing_if = "Option::is_none")] + pub windows_sandbox_level: Option, + + /// Updated model slug. When set, the model info is derived automatically. + #[serde(skip_serializing_if = "Option::is_none")] + pub model: Option, + + /// Updated reasoning effort (honored only for reasoning-capable models). + /// + /// Use `Some(Some(_))` to set a specific effort, `Some(None)` to clear the + /// effort, or `None` to leave the existing value unchanged. + #[serde(skip_serializing_if = "Option::is_none")] + pub effort: Option>, + + /// Updated reasoning summary preference (honored only for reasoning-capable models). + #[serde(skip_serializing_if = "Option::is_none")] + pub summary: Option, + + /// Updated service tier preference for future turns. + /// + /// Use `Some(Some(_))` to set a specific tier, `Some(None)` to clear the + /// preference, or `None` to leave the existing value unchanged. + #[serde(skip_serializing_if = "Option::is_none")] + pub service_tier: Option>, + + /// EXPERIMENTAL - set a pre-set collaboration mode. + /// Takes precedence over model, effort, and developer instructions if set. + #[serde(skip_serializing_if = "Option::is_none")] + pub collaboration_mode: Option, + + /// Updated personality preference. + #[serde(skip_serializing_if = "Option::is_none")] + pub personality: Option, +} + /// Submission operation #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema)] #[serde(tag = "type", rename_all = "snake_case")] @@ -458,67 +523,19 @@ pub enum Op { #[serde(default, skip_serializing_if = "Option::is_none")] responsesapi_client_metadata: Option>, - /// Updated `cwd` for sandbox/tool calls. - #[serde(skip_serializing_if = "Option::is_none")] - cwd: Option, + /// Persistent turn-context overrides to apply before the input. + #[serde(flatten)] + turn_context: TurnContextOverrides, + }, - /// Updated command approval policy. - #[serde(skip_serializing_if = "Option::is_none")] - approval_policy: Option, - - /// Updated approval reviewer for future approval prompts. - #[serde(skip_serializing_if = "Option::is_none")] - approvals_reviewer: Option, - - /// Updated sandbox policy for tool calls. - #[serde(skip_serializing_if = "Option::is_none")] - sandbox_policy: Option, - - /// Updated permissions profile for tool calls. - #[serde(skip_serializing_if = "Option::is_none")] - permission_profile: Option, - - /// Named or built-in profile that produced `permission_profile`, if - /// the update selected a profile rather than supplying raw - /// permissions. - #[serde(skip_serializing_if = "Option::is_none")] - active_permission_profile: Option, - - /// Updated Windows sandbox mode for tool execution. - #[serde(skip_serializing_if = "Option::is_none")] - windows_sandbox_level: Option, - - /// Updated model slug. When set, the model info is derived - /// automatically. - #[serde(skip_serializing_if = "Option::is_none")] - model: Option, - - /// Updated reasoning effort (honored only for reasoning-capable models). - /// - /// Use `Some(Some(_))` to set a specific effort, `Some(None)` to clear - /// the effort, or `None` to leave the existing value unchanged. - #[serde(skip_serializing_if = "Option::is_none")] - effort: Option>, - - /// Updated reasoning summary preference (honored only for reasoning-capable models). - #[serde(skip_serializing_if = "Option::is_none")] - summary: Option, - - /// Updated service tier preference for future turns. - /// - /// Use `Some(Some(_))` to set a specific tier, `Some(None)` to clear the - /// preference, or `None` to leave the existing value unchanged. - #[serde(skip_serializing_if = "Option::is_none")] - service_tier: Option>, - - /// EXPERIMENTAL - set a pre-set collaboration mode. - /// Takes precedence over model, effort, and developer instructions if set. - #[serde(skip_serializing_if = "Option::is_none")] - collaboration_mode: Option, - - /// Updated personality preference. - #[serde(skip_serializing_if = "Option::is_none")] - personality: Option, + /// Apply persistent turn-context overrides without starting a turn. + /// + /// This uses the same submission queue as turn starts so app-server can + /// preserve caller order between both kinds of mutation. + TurnContext { + /// Persistent turn-context overrides to apply. + #[serde(flatten)] + turn_context: TurnContextOverrides, }, /// Similar to [`Op::UserInput`], but contains additional context required @@ -782,6 +799,7 @@ impl Op { Self::RealtimeConversationListVoices => "realtime_conversation_list_voices", Self::UserInput { .. } => "user_input", Self::UserInputWithTurnContext { .. } => "user_input_with_turn_context", + Self::TurnContext { .. } => "turn_context", Self::UserTurn { .. } => "user_turn", Self::InterAgentCommunication { .. } => "inter_agent_communication", Self::ExecApproval { .. } => "exec_approval", @@ -1234,6 +1252,10 @@ pub enum EventMsg { #[serde(rename = "task_started", alias = "turn_started")] TurnStarted(TurnStartedEvent), + /// Persistent turn-context overrides from the correlated submission have + /// been applied to the session configuration. + TurnContextApplied(TurnContextAppliedEvent), + /// Agent has completed all actions. /// v1 wire format uses `task_complete`; accept `turn_complete` for v2 interop. #[serde(rename = "task_complete", alias = "turn_complete")] @@ -1914,6 +1936,33 @@ pub struct TurnStartedEvent { pub collaboration_mode_kind: ModeKind, } +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +pub struct TurnContextAppliedEvent { + pub turn_context: TurnContextSnapshot, +} + +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +pub struct TurnContextSnapshot { + pub model: String, + pub model_provider_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub service_tier: Option, + pub approval_policy: AskForApproval, + pub approvals_reviewer: ApprovalsReviewer, + pub permission_profile: PermissionProfile, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub active_permission_profile: Option, + pub cwd: AbsolutePathBuf, + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_effort: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_summary: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub personality: Option, + pub collaboration_mode: CollaborationMode, +} + #[derive(Debug, Clone, Deserialize, Serialize, Default, PartialEq, Eq, JsonSchema, TS)] pub struct TokenUsage { #[ts(type = "number")] diff --git a/codex-rs/rollout-trace/src/protocol_event.rs b/codex-rs/rollout-trace/src/protocol_event.rs index 1e49c82be2..9a7f59842a 100644 --- a/codex-rs/rollout-trace/src/protocol_event.rs +++ b/codex-rs/rollout-trace/src/protocol_event.rs @@ -228,6 +228,7 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option Option<&'static s | EventMsg::ModelReroute(_) | EventMsg::ModelVerification(_) | EventMsg::ContextCompacted(_) + | EventMsg::TurnContextApplied(_) | EventMsg::TokenCount(_) | EventMsg::AgentMessage(_) | EventMsg::UserMessage(_) diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 9c42f8db61..3b6b05a0fb 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -181,6 +181,7 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option { | EventMsg::AgentReasoningSectionBreak(_) | EventMsg::RawResponseItem(_) | EventMsg::SessionConfigured(_) + | EventMsg::TurnContextApplied(_) | EventMsg::McpToolCallBegin(_) | EventMsg::ExecCommandBegin(_) | EventMsg::TerminalInteraction(_)