Add live turn settings updates (#40653)

## What changed

- Add a feature-gated `TurnSettings` operation that can update the model,
  reasoning effort, reasoning summary, and service tier for subsequent steps of
  a named running turn without changing future thread settings.
- Report whether an update was applied, rejected, or lost its live target.
- Revalidate managed constraints and preserve the turn's admitted approval and
  Guardian safety properties before publishing an updated settings snapshot.

## Testing

- Cover sparse updates, step capture and ordering, target replacement, managed
  policy changes, model metadata safety, and separation from future settings.

GitOrigin-RevId: ecaaaa95b4fd4ec9d62265bef551582dddb36d78
This commit is contained in:
joeflorencio-openai
2026-08-25 17:44:50 +00:00
committed by copyberry
parent 68301fa45f
commit a10c8127f7
12 changed files with 2554 additions and 5 deletions
+6
View File
@@ -907,6 +907,9 @@
"steer": {
"type": "boolean"
},
"step_model_switching": {
"type": "boolean"
},
"telepathy": {
"type": "boolean"
},
@@ -5916,6 +5919,9 @@
"steer": {
"type": "boolean"
},
"step_model_switching": {
"type": "boolean"
},
"telepathy": {
"type": "boolean"
},
+1
View File
@@ -35,6 +35,7 @@ pub(crate) use approval_request::GuardianNetworkAccessTrigger;
#[cfg(test)]
pub(crate) use approval_request::guardian_approval_request_to_json;
pub(crate) use prompt::BUNDLED_GUARDIAN_POLICY;
pub(crate) use prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE;
pub(crate) use prompt::guardian_truncate_text;
pub(crate) use review::GuardianReviewOptions;
pub(crate) use review::guardian_timeout_message;
+1 -1
View File
@@ -817,7 +817,7 @@ For anything else, use this JSON schema:
}
pub(crate) const BUNDLED_GUARDIAN_POLICY: &str = include_str!("policy.md");
pub(super) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str = include_str!("policy_template.md");
pub(crate) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str = include_str!("policy_template.md");
const TENANT_POLICY_CONFIG_PLACEHOLDER: &str = "{{ tenant_policy_config }}";
/// Guardian policy prompt.
+9
View File
@@ -596,6 +596,15 @@ pub(super) async fn submission_loop(
thread_settings::update(&sess, sub.id.clone(), thread_settings).await;
false
}
Op::TurnSettings {
turn_id,
update,
reply,
} => {
let outcome = sess.apply_turn_settings(&turn_id, update).await;
let _ = reply.send(outcome);
false
}
Op::InterAgentCommunication { communication } => {
inter_agent_communication(
&sess,
+1
View File
@@ -229,6 +229,7 @@ mod rollout_budget;
mod rollout_reconstruction;
#[allow(clippy::module_inception)]
pub(crate) mod session;
mod step_activation;
pub(crate) mod step_context;
pub(crate) mod step_settings;
mod thread_settings;
@@ -0,0 +1,385 @@
//! Restricted updates to a running turn's immutable settings snapshots.
use super::session::Session;
use super::session::SessionConfiguration;
use super::step_settings::ResolvedStepSettings;
use super::step_settings::StepSettingsConstraints;
use super::step_settings::StepSettingsUpdate;
use super::turn_context::TurnContext;
use crate::config::Config;
use crate::config::ConstraintResult;
use crate::exec_policy::AllowPrefixRules;
use crate::guardian::BUNDLED_GUARDIAN_POLICY_TEMPLATE;
use codex_features::Feature;
use codex_protocol::openai_models::GuardianV2ModelConfig;
use codex_protocol::openai_models::GuardianV2TranscriptModelConfig;
use codex_protocol::openai_models::MODEL_SPECIALTY_CYBER;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::TurnSettingsUpdate;
use codex_protocol::protocol::TurnSettingsUpdateOutcome;
use std::sync::Arc;
/// Temporary restrictions while approvals and Guardian still read the admitted
/// `TurnContext`. Ordinary live authorization is validated separately. Remove
/// these restrictions as their consumers migrate to captured step settings.
fn check_legacy_turn_safety(
turn_context: &TurnContext,
current: &ResolvedStepSettings,
destination: &ResolvedStepSettings,
live_config: &Config,
) -> Result<(), String> {
let stack = &live_config.config_layer_stack;
let requirements = stack.requirements();
let required_review = requirements
.auto_review_required_for_model(destination.selected_collaboration_mode().model());
let admitted_required_review = turn_context
.config
.config_layer_stack
.requirements()
.auto_review_required_for_model(&turn_context.model_info().slug);
let ignored_models = stack
.requirements_toml()
.auto_review
.as_ref()
.and_then(|review| review.ignore_rules.as_ref());
let ignores_prefix_rules = |model: &ModelInfo| {
model.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER)
|| ignored_models.is_some_and(|models| models.contains(&model.slug))
};
// tools::approvals and guardian::review still route approvals using the
// originating turn's policy, reviewer, and required-model classification.
if destination.constrained_approval_policy() != current.constrained_approval_policy()
|| destination.approval_policy() != turn_context.approval_policy()
{
return Err("the destination changes the admitted approval policy".to_string());
}
if destination.approvals_reviewer() != current.approvals_reviewer()
|| destination.approvals_reviewer() != turn_context.config.approvals_reviewer
{
return Err("the destination changes the admitted approvals reviewer".to_string());
}
if required_review
!= requirements
.auto_review_required_for_model(current.selected_collaboration_mode().model())
|| required_review != admitted_required_review
{
return Err("the destination changes model-required approval authority".to_string());
}
// Command approval continues to use TurnContext::allow_prefix_rules.
if ignores_prefix_rules(&destination.model_info) != ignores_prefix_rules(&current.model_info)
|| ignores_prefix_rules(&destination.model_info)
!= (turn_context.allow_prefix_rules() == AllowPrefixRules::IgnoreForCyberModel)
{
return Err("the destination changes the admitted prefix-rule policy".to_string());
}
check_legacy_model_safety(
turn_context.model_info(),
&current.model_info,
&destination.model_info,
&turn_context.config,
live_config,
)
}
/// Model-owned portion of the temporary legacy-turn safety check. Ordinary
/// model metadata may differ so diagnostics can expose unmigrated consumers.
fn check_legacy_model_safety(
admitted: &ModelInfo,
current: &ModelInfo,
destination: &ModelInfo,
admitted_config: &Config,
live_config: &Config,
) -> Result<(), String> {
if admitted.used_fallback_model_metadata || current.used_fallback_model_metadata {
return Err("the active model has only fallback metadata".to_string());
}
if destination.used_fallback_model_metadata {
return Err("the destination model has only fallback metadata".to_string());
}
let retained_models = [admitted, current];
// guardian::review::record_guardian_denial still selects its circuit-breaker
// policy from the admitted model's Cyber classification.
let destination_is_cyber =
destination.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER);
if retained_models.iter().any(|model| {
(model.model_specialty.as_deref() == Some(MODEL_SPECIALTY_CYBER)) != destination_is_cyber
}) {
return Err("the destination changes the admitted Guardian rejection policy".to_string());
}
// TurnMetadataState pins both node REPL flags. Guardian prompt/evidence
// construction also reads node_repl_auto_review_required from the turn.
if retained_models.iter().any(|model| {
model.node_repl_auto_review_required != destination.node_repl_auto_review_required
}) {
return Err(
"the destination changes the admitted node REPL review requirement".to_string(),
);
}
if retained_models
.iter()
.any(|model| model.node_repl_disabled != destination.node_repl_disabled)
{
return Err(
"the destination changes the admitted node REPL availability restriction".to_string(),
);
}
// guardian::review::guardian_review_session_config and Guardian V2 still
// select the reviewer from the retained parent metadata.
if retained_models
.iter()
.any(|model| model.auto_review_model_override != destination.auto_review_model_override)
{
return Err("the destination changes the explicit Guardian reviewer model".to_string());
}
if admitted_config.features.enabled(Feature::GuardianV2)
&& admitted_config.features.enabled(Feature::GuardianApproval)
{
// GuardianV2Extension::on_tool_start reads the parent ModelInfo from
// thread_store. Its classifier settings are independent of the reviewer
// override. Local overrides may mask differences, but resolving those
// overrides remains the extension's responsibility.
let classification_settings =
|model: &ModelInfo| -> GuardianV2ModelConfig {
let mut settings = model
.model_messages
.as_ref()
.and_then(|messages| messages.guardian_v2.clone())
.unwrap_or_default();
// Missing and empty transcript records supply the same defaults.
if settings.transcript.as_ref().is_some_and(|transcript| {
*transcript == GuardianV2TranscriptModelConfig::default()
}) {
settings.transcript = None;
}
settings
};
let destination_settings = classification_settings(destination);
if retained_models
.iter()
.any(|model| classification_settings(model) != destination_settings)
{
return Err(
"the destination changes the admitted Guardian V2 classification settings"
.to_string(),
);
}
}
// guardian_review_session_config and Guardian V2 can fall back to parent
// metadata if their preferred reviewer is unavailable, including after a
// catalog refresh. V1 uses the admitted config; V2 can use the live config.
// An unchanged explicit reviewer override prevents both fallback paths.
if destination.auto_review_model_override.is_none() {
for config in [admitted_config, live_config] {
let destination_policy =
config.resolve_guardian_policy(destination.model_messages.as_ref());
if retained_models.iter().any(|model| {
config.resolve_guardian_policy(model.model_messages.as_ref()) != destination_policy
}) {
return Err(
"the destination changes the Guardian parent-fallback policy".to_string(),
);
}
}
let destination_template = destination
.model_messages
.as_ref()
.and_then(|messages| messages.auto_review.as_ref())
.and_then(|messages| messages.policy_template.as_deref())
.unwrap_or(BUNDLED_GUARDIAN_POLICY_TEMPLATE)
.trim_end();
if retained_models.iter().any(|model| {
model
.model_messages
.as_ref()
.and_then(|messages| messages.auto_review.as_ref())
.and_then(|messages| messages.policy_template.as_deref())
.unwrap_or(BUNDLED_GUARDIAN_POLICY_TEMPLATE)
.trim_end()
!= destination_template
}) {
return Err(
"the destination changes the Guardian parent-fallback policy template".to_string(),
);
}
}
Ok(())
}
impl Session {
/// Publishes settings to the named, originally captured live task, regardless
/// of task kind. Publication does not propagate to child sessions or require
/// the task to sample; consumers using initial settings remain unchanged.
///
/// Callers must serialize updates through completion, including model
/// resolution, so each sparse patch sees the preceding publication.
#[expect(
clippy::await_holding_invalid_type,
reason = "the final managed-policy check and active settings publication must remain atomic"
)]
pub(super) async fn apply_turn_settings(
&self,
turn_id: &str,
update: TurnSettingsUpdate,
) -> TurnSettingsUpdateOutcome {
if !self.features.enabled(Feature::StepModelSwitching) {
return TurnSettingsUpdateOutcome::Rejected {
reason: "turn settings updates require the step_model_switching feature"
.to_string(),
};
}
// Capture the exact live task and its settings, then release the
// lock. A task that starts during preparation is never a new target.
let target = {
let active = self.active_turn.lock().await;
active.as_ref().and_then(|active| {
active.task.as_ref().and_then(|task| {
(task.turn_context.sub_id == turn_id && !task.cancellation_token.is_cancelled())
.then(|| {
(
Arc::clone(&task.turn_context),
Arc::clone(&task.done),
task.turn_context.current_settings.load_full(),
)
})
})
})
};
let Some((turn_context, task_done, current)) = target else {
return TurnSettingsUpdateOutcome::TargetUnavailable;
};
let TurnSettingsUpdate {
model,
effort,
summary,
service_tier,
} = update;
let update = StepSettingsUpdate {
model,
effort,
reasoning_summary: summary,
service_tier,
..Default::default()
};
// Apply the sparse patch to the captured active base using the shared
// settings rules. The task can progress, finish, or be cancelled while
// preparation awaits; no publication locks are held here.
let prepared = self
.prepare_step_settings_activation(&turn_context, &current, &update)
.await;
let active = self.active_turn.lock().await;
let Some(task) = active.as_ref().and_then(|active| active.task.as_ref()) else {
return TurnSettingsUpdateOutcome::TargetUnavailable;
};
// A later task may reuse the same context and turn ID. `done` is
// allocated per task, so matching only the ID/context is insufficient.
// A mismatch abandons the update without retrying or retargeting.
if !Arc::ptr_eq(&task.done, &task_done)
|| !Arc::ptr_eq(&task.turn_context, &turn_context)
|| !Arc::ptr_eq(&task.turn_context.current_settings.load_full(), &current)
|| task.cancellation_token.is_cancelled()
{
return TurnSettingsUpdateOutcome::TargetUnavailable;
}
let destination = match prepared {
Ok(destination) => destination,
Err(reason) => return TurnSettingsUpdateOutcome::Rejected { reason },
};
// Managed requirements can change during resolution. Keep the live
// authorization and safety checks atomic with publication under state
// and active_turn; no asynchronous preparation runs under these locks.
let state = self.state.lock().await;
if let Err(reason) = self
.validate_active_step_settings(
&turn_context,
&destination,
&state.session_configuration,
)
.map_err(|error| error.to_string())
.and_then(|()| {
check_legacy_turn_safety(
&turn_context,
&current,
&destination,
&state.session_configuration.original_config_do_not_use,
)
})
{
return TurnSettingsUpdateOutcome::Rejected { reason };
}
// Publish the immutable snapshot. Frozen initial settings, existing step
// captures, and future thread settings are not changed.
task.turn_context
.current_settings
.store(Arc::new(destination));
TurnSettingsUpdateOutcome::Applied
}
async fn prepare_step_settings_activation(
&self,
turn_context: &TurnContext,
current: &ResolvedStepSettings,
update: &StepSettingsUpdate,
) -> Result<ResolvedStepSettings, String> {
let (requirements, overrides, trusted_guardian_reviewer) = {
let state = self.state.lock().await;
let configuration = &state.session_configuration;
let stack = &configuration.original_config_do_not_use.config_layer_stack;
(
stack.requirements().clone(),
configuration.model_info_overrides.clone(),
configuration.trusted_guardian_reviewer,
)
};
let constraints = StepSettingsConstraints {
requirements: &requirements,
guardian_approval_enabled: self.features.enabled(Feature::GuardianApproval),
trusted_guardian_reviewer,
has_full_disk_write_access: turn_context
.file_system_sandbox_policy()
.has_full_disk_write_access(),
};
current
.apply_update(
update,
&constraints,
self.services.models_manager.as_ref(),
&overrides,
self.features.enabled(Feature::Personality),
self.features.enabled(Feature::FastMode),
)
.await
.map_err(|error| error.to_string())
}
/// Rechecks ordinary managed authorization after asynchronous resolution.
/// Unlike the temporary legacy-turn check, these requirements also apply
/// once all execution consumers read their captured `StepContext`.
fn validate_active_step_settings(
&self,
turn_context: &TurnContext,
settings: &ResolvedStepSettings,
configuration: &SessionConfiguration,
) -> ConstraintResult<()> {
let requirements = configuration
.original_config_do_not_use
.config_layer_stack
.requirements();
settings.revalidate(&StepSettingsConstraints {
requirements,
guardian_approval_enabled: self.features.enabled(Feature::GuardianApproval),
trusted_guardian_reviewer: configuration.trusted_guardian_reviewer,
has_full_disk_write_access: turn_context
.file_system_sandbox_policy()
.has_full_disk_write_access(),
})
}
}
#[cfg(test)]
#[path = "step_activation_tests.rs"]
mod tests;
File diff suppressed because it is too large Load Diff
@@ -101,6 +101,57 @@ impl ResolvedStepSettings {
self.selected.personality
}
pub(super) fn constrained_approval_policy(&self) -> &Constrained<AskForApproval> {
&self.selected.approval_policy
}
/// Applies sparse edits to the retained selection, preserving pinned metadata
/// unless model or personality selection changes, then resolves request values.
pub(super) async fn apply_update(
&self,
update: &StepSettingsUpdate,
constraints: &StepSettingsConstraints<'_>,
models_manager: &dyn ModelsManager,
overrides: &ModelInfoOverrides,
personality_enabled: bool,
fast_mode_enabled: bool,
) -> ConstraintResult<Self> {
let selected = self.selected.apply(update, constraints)?;
let model_info = if selected.collaboration_mode.model()
== self.selected.collaboration_mode.model()
&& selected.personality == self.selected.personality
{
Arc::clone(&self.model_info)
} else {
Arc::new(
selected
.resolve_model_info(models_manager, overrides, personality_enabled)
.await,
)
};
Ok(Self::new(Arc::new(selected), model_info, fast_mode_enabled))
}
/// Rechecks the retained selection against current managed constraints.
pub(super) fn revalidate(
&self,
constraints: &StepSettingsConstraints<'_>,
) -> ConstraintResult<()> {
// TODO: Revisit the split with StepSettings::validate, which does not
// check inherited approval values against the supplied managed allow-lists.
// Sharing these checks would also make ordinary settings updates stricter
// after a requirements refresh.
self.selected.validate(constraints)?;
constraints
.requirements
.approval_policy
.can_set(&self.approval_policy())?;
constraints
.requirements
.approvals_reviewer
.can_set(&self.approvals_reviewer())
}
pub(super) fn telemetry(&self, base: &SessionTelemetry) -> SessionTelemetry {
base.clone().with_model(
self.selected.collaboration_mode.model(),
+1
View File
@@ -153,6 +153,7 @@ mod skills;
mod skills_extension;
mod spawn_agent_description;
mod sqlite_state;
mod step_settings;
mod step_settings_snapshots;
mod stream_error_allows_next_turn;
mod stream_no_completed;
+911
View File
@@ -0,0 +1,911 @@
use anyhow::Result;
use codex_core::CodexThread;
use codex_core::TurnInputRequest;
use codex_core::config::Constrained;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_models_manager::bundled_models_response;
use codex_models_manager::manager::RefreshStrategy;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SafetyBufferingEvent;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::TurnSettingsUpdate;
use codex_protocol::protocol::TurnSettingsUpdateOutcome;
use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputEvent;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_protocol::user_input::UserInput;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_models_once;
use core_test_support::responses::mount_response_sequence;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::responses::sse_completed;
use core_test_support::responses::sse_response;
use core_test_support::responses::start_mock_server;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::TestCodexBuilder;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_match;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::collections::HashMap;
use test_case::test_case;
const MODEL_A: &str = "step-settings-a";
const MODEL_B: &str = "step-settings-b";
const MODEL_C: &str = "step-settings-c";
const TURN_STATE_HEADER: &str = "x-codex-turn-state";
fn step_settings_models() -> Vec<ModelInfo> {
let model = bundled_models_response()
.expect("bundled models should parse")
.models
.into_iter()
.find(|model| model.slug == "gpt-5.4")
.expect("bundled gpt-5.4 model");
[MODEL_A, MODEL_B, MODEL_C]
.into_iter()
.map(|slug| {
let mut model = model.clone();
// Tests add model-owned differences when they need to exercise
// activation or an explicit safety restriction.
model.slug = slug.to_string();
model
})
.collect()
}
fn step_settings_test() -> TestCodexBuilder {
test_codex().with_model(MODEL_A).with_config(move |config| {
for feature in [
Feature::StepModelSwitching,
Feature::DefaultModeRequestUserInput,
Feature::FastMode,
] {
config
.features
.enable(feature)
.expect("test config should allow feature update");
}
config.model_catalog = Some(ModelsResponse {
models: step_settings_models(),
});
config.model_reasoning_effort = Some(ReasoningEffort::Low);
config.model_reasoning_summary = Some(ReasoningSummary::Concise);
config.service_tier = None;
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
config.approvals_reviewer = ApprovalsReviewer::User;
})
}
fn paused_response(response_id: &str, call_id: &str) -> String {
sse(vec![
ev_response_created(response_id),
ev_function_call(
call_id,
"request_user_input",
&json!({
"questions": [{
"id": "continue",
"header": "Continue",
"question": "Continue after the settings update?",
"options": [{
"label": "Yes (Recommended)",
"description": "Continue the current turn."
}, {
"label": "No",
"description": "Stop the current turn."
}]
}]
})
.to_string(),
),
ev_completed(response_id),
])
}
async fn start_paused_turn(thread: &CodexThread) -> Result<RequestUserInputEvent> {
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "pause before continuing".into(),
text_elements: Vec::new(),
}]))
.await?;
Ok(wait_for_event_match(thread, |event| match event {
EventMsg::RequestUserInput(request) => Some(request.clone()),
_ => None,
})
.await)
}
async fn answer_paused_turn(thread: &CodexThread, turn_id: &str) -> Result<()> {
thread
.submit(Op::UserInputAnswer {
id: turn_id.to_string(),
response: RequestUserInputResponse {
answers: HashMap::from([(
"continue".to_string(),
RequestUserInputAnswer {
answers: vec!["Yes (Recommended)".to_string()],
},
)]),
},
})
.await?;
Ok(())
}
async fn submit_turn_settings(
thread: &CodexThread,
turn_id: &str,
update: TurnSettingsUpdate,
) -> Result<TurnSettingsUpdateOutcome> {
let (reply, outcome) = tokio::sync::oneshot::channel();
thread
.submit(Op::TurnSettings {
turn_id: turn_id.to_string(),
update,
reply,
})
.await?;
Ok(tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), outcome).await??)
}
fn request_settings(request: &ResponsesRequest) -> Value {
let body = request.body_json();
json!({
"model": body["model"],
"reasoning": body["reasoning"],
"service_tier": body.get("service_tier"),
})
}
fn request_turn_id(request: &ResponsesRequest) -> String {
let metadata: Value = serde_json::from_str(
&request
.header("x-codex-turn-metadata")
.expect("request should include turn metadata"),
)
.expect("valid turn metadata");
metadata["turn_id"]
.as_str()
.expect("request should include turn_id")
.to_string()
}
#[derive(Clone, Copy)]
enum SettingsTarget {
Thread,
Turn,
}
#[test_case(SettingsTarget::Thread; "thread updates stay next-turn-only with the feature enabled")]
#[test_case(SettingsTarget::Turn; "turn updates leave future settings unchanged")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn settings_updates_preserve_turn_identity_and_target(target: SettingsTarget) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response_mock = mount_response_sequence(
&server,
vec![
sse_response(paused_response("resp-1", "pause-turn"))
.insert_header(TURN_STATE_HEADER, "original-turn-state"),
sse_response(sse_completed("resp-2")),
sse_response(sse_completed("resp-3")),
],
)
.await;
let test = step_settings_test().build_with_auto_env(&server).await?;
let original_settings = test.codex.thread_settings_snapshot().await;
let request = start_paused_turn(&test.codex).await?;
match target {
SettingsTarget::Thread => {
test.codex
.submit(Op::ThreadSettings {
thread_settings: ThreadSettingsOverrides {
model: Some(MODEL_B.to_string()),
effort: Some(Some(ReasoningEffort::High)),
summary: Some(ReasoningSummary::Detailed),
service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())),
..Default::default()
},
})
.await?;
}
SettingsTarget::Turn => {
let update = TurnSettingsUpdate {
model: Some(MODEL_B.to_string()),
effort: Some(Some(ReasoningEffort::High)),
summary: Some(ReasoningSummary::Detailed),
service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())),
};
assert_eq!(
submit_turn_settings(&test.codex, "different-turn", update.clone()).await?,
TurnSettingsUpdateOutcome::TargetUnavailable
);
assert_eq!(
submit_turn_settings(&test.codex, &request.turn_id, update).await?,
TurnSettingsUpdateOutcome::Applied
);
}
}
answer_paused_turn(&test.codex, &request.turn_id).await?;
let mut settings_events = Vec::new();
let mut new_turns = Vec::new();
let completion = wait_for_event(&test.codex, |event| match event {
EventMsg::ThreadSettingsApplied(event) => {
settings_events.push(event.thread_settings.clone());
false
}
EventMsg::TurnStarted(event) => {
new_turns.push(event.turn_id.clone());
false
}
EventMsg::Error(error) => panic!("settings activation failed: {}", error.message),
EventMsg::TurnComplete(_) => true,
_ => false,
})
.await;
let EventMsg::TurnComplete(completion) = completion else {
unreachable!("waited for turn completion")
};
assert_eq!(completion.turn_id, request.turn_id);
assert_eq!(new_turns, Vec::<String>::new());
let mut changed_settings = original_settings.clone();
changed_settings.model = MODEL_B.to_string();
changed_settings.reasoning_effort = Some(ReasoningEffort::High);
changed_settings.reasoning_summary = Some(ReasoningSummary::Detailed);
changed_settings.service_tier = Some(ServiceTier::Fast.request_value().to_string());
changed_settings.collaboration_mode = changed_settings.collaboration_mode.with_updates(
Some(MODEL_B.to_string()),
Some(Some(ReasoningEffort::High)),
/*developer_instructions*/ None,
);
let expected_future_settings = match target {
SettingsTarget::Turn => {
assert_eq!(settings_events, Vec::new());
original_settings
}
SettingsTarget::Thread => {
assert_eq!(settings_events, vec![changed_settings.clone()]);
changed_settings
}
};
assert_eq!(
test.codex.thread_settings_snapshot().await,
expected_future_settings
);
test.submit_text_turn("start the next turn").await?;
let original_request_settings = json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
});
let changed_request_settings = json!({
"model": MODEL_B,
"reasoning": { "effort": "high", "summary": "detailed" },
"service_tier": "priority",
});
let (continued_settings, next_turn_settings) = match target {
SettingsTarget::Thread => (original_request_settings.clone(), changed_request_settings),
SettingsTarget::Turn => (changed_request_settings, original_request_settings.clone()),
};
let requests = response_mock.requests();
assert_eq!(
requests.iter().map(request_settings).collect::<Vec<_>>(),
vec![
original_request_settings,
continued_settings,
next_turn_settings,
]
);
assert_eq!(request_turn_id(&requests[0]), request.turn_id);
assert_eq!(request_turn_id(&requests[1]), request.turn_id);
assert_ne!(request_turn_id(&requests[2]), request.turn_id);
let session_id = requests[0]
.header("session-id")
.expect("initial request session id");
assert_eq!(requests[1].header("session-id"), Some(session_id));
assert_eq!(requests[0].header(TURN_STATE_HEADER), None);
assert_eq!(
requests[1].header(TURN_STATE_HEADER),
Some("original-turn-state".to_string())
);
assert_eq!(requests[2].header(TURN_STATE_HEADER), None);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn response_metadata_uses_the_captured_model_after_a_turn_update() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let mut created = ev_response_created("resp-2");
created["safety_buffering"] = json!({
"use_cases": ["cyber"],
"reasons": ["policy-check"],
"retry_model": MODEL_C,
});
let response_mock = mount_response_sequence(
&server,
vec![
sse_response(paused_response("resp-1", "pause-turn")),
sse_response(sse(vec![created, ev_completed("resp-2")]))
.insert_header("OpenAI-Model", MODEL_B),
],
)
.await;
let test = step_settings_test().build_with_auto_env(&server).await?;
let request = start_paused_turn(&test.codex).await?;
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
model: Some(MODEL_B.to_string()),
..Default::default()
},
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
answer_paused_turn(&test.codex, &request.turn_id).await?;
let mut reroutes = Vec::new();
let mut buffering_events = Vec::new();
wait_for_event(&test.codex, |event| match event {
EventMsg::ModelReroute(event) => {
reroutes.push(event.clone());
false
}
EventMsg::SafetyBuffering(event) => {
buffering_events.push(event.clone());
false
}
EventMsg::Error(error) => panic!("sampling failed: {}", error.message),
EventMsg::TurnComplete(_) => true,
_ => false,
})
.await;
assert_eq!(
response_mock
.requests()
.iter()
.map(|request| request.body_json()["model"].clone())
.collect::<Vec<_>>(),
vec![json!(MODEL_A), json!(MODEL_B)]
);
// B's matching response header is not a reroute from the turn's initial A.
// Buffering metadata likewise belongs to the captured B step.
assert_eq!(
(reroutes, buffering_events),
(
vec![],
vec![SafetyBufferingEvent {
model: MODEL_B.to_string(),
use_cases: vec!["cyber".to_string()],
reasons: vec!["policy-check".to_string()],
show_buffering_ui: true,
faster_model: Some(MODEL_C.to_string()),
}],
)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sparse_updates_preserve_divergent_active_and_future_models() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response_mock = mount_sse_sequence(
&server,
vec![
paused_response("resp-1", "pause-first-step"),
paused_response("resp-2", "pause-second-step"),
sse_completed("resp-3"),
sse_completed("resp-4"),
],
)
.await;
let test = step_settings_test().build_with_auto_env(&server).await?;
let request = start_paused_turn(&test.codex).await?;
core_test_support::submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
model: Some(MODEL_B.to_string()),
..Default::default()
},
)
.await?;
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
model: Some(MODEL_C.to_string()),
effort: Some(Some(ReasoningEffort::High)),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
answer_paused_turn(&test.codex, &request.turn_id).await?;
let second_request = wait_for_event_match(&test.codex, |event| match event {
EventMsg::RequestUserInput(request) => Some(request.clone()),
_ => None,
})
.await;
assert_eq!(second_request.turn_id, request.turn_id);
assert_eq!(second_request.call_id, "pause-second-step");
test.codex
.submit(Op::ThreadSettings {
thread_settings: ThreadSettingsOverrides {
service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())),
..Default::default()
},
})
.await?;
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
let durable_settings = wait_for_event_match(&test.codex, |event| match event {
EventMsg::ThreadSettingsApplied(event) => Some(event.thread_settings.clone()),
_ => None,
})
.await;
assert_eq!(durable_settings.model, MODEL_B);
assert_eq!(
durable_settings.reasoning_effort,
Some(ReasoningEffort::Low)
);
assert_eq!(
durable_settings.service_tier,
Some(ServiceTier::Fast.request_value().to_string())
);
answer_paused_turn(&test.codex, &second_request.turn_id).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.submit_text_turn("start the next turn").await?;
let requests = response_mock.requests();
assert_eq!(
requests.iter().map(request_settings).collect::<Vec<_>>(),
vec![
json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
}),
json!({
"model": MODEL_C,
"reasoning": { "effort": "high", "summary": "concise" },
"service_tier": null,
}),
json!({
"model": MODEL_C,
"reasoning": { "effort": "high", "summary": "concise" },
"service_tier": "priority",
}),
json!({
"model": MODEL_B,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": "priority",
}),
]
);
assert_eq!(
requests[..3]
.iter()
.map(request_turn_id)
.collect::<Vec<_>>(),
vec![request.turn_id.clone(); 3]
);
assert_ne!(request_turn_id(&requests[3]), request.turn_id);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn turn_settings_do_not_target_idle_or_finished_turns() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response_mock = mount_sse_sequence(
&server,
vec![sse_completed("resp-1"), sse_completed("resp-2")],
)
.await;
let test = step_settings_test().build_with_auto_env(&server).await?;
for (discarded_model, model) in [(MODEL_C, MODEL_B), (MODEL_A, MODEL_C)] {
let turn_id = response_mock
.requests()
.last()
.map(request_turn_id)
.unwrap_or_else(|| "never-started".to_string());
let before = test.codex.thread_settings_snapshot().await;
assert_eq!(
submit_turn_settings(
&test.codex,
&turn_id,
TurnSettingsUpdate {
model: Some(discarded_model.to_string()),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::TargetUnavailable
);
assert_eq!(test.codex.thread_settings_snapshot().await, before);
core_test_support::submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
model: Some(model.to_string()),
..Default::default()
},
)
.await?;
test.submit_text_turn("start the next turn").await?;
}
assert_eq!(
response_mock
.requests()
.iter()
.map(request_settings)
.collect::<Vec<_>>(),
vec![
json!({
"model": MODEL_B,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
}),
json!({
"model": MODEL_C,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
}),
]
);
Ok(())
}
#[test_case(None, "detailed"; "unset summary follows the destination model")]
#[test_case(Some(ReasoningSummary::Concise), "concise"; "explicit summary is preserved")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn model_activation_uses_destination_metadata_defaults(
configured_summary: Option<ReasoningSummary>,
expected_summary: &str,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response_mock = mount_sse_sequence(
&server,
vec![
paused_response("resp-1", "pause-before-model-change"),
paused_response("resp-2", "pause-before-restoring-model"),
sse_completed("resp-3"),
],
)
.await;
let test = step_settings_test()
.with_config(move |config| {
config.model_context_window = None;
config.model_auto_compact_token_limit = None;
config.model_reasoning_summary = configured_summary;
config.service_tier = Some(ServiceTier::Fast.request_value().to_string());
config.base_instructions = None;
for model in &mut config
.model_catalog
.as_mut()
.expect("controlled model catalog")
.models
{
model.context_window = Some(256_000);
model.auto_compact_token_limit = Some(200_000);
model.default_reasoning_summary = ReasoningSummary::Concise;
if model.slug == MODEL_B {
model.context_window = Some(128_000);
model.auto_compact_token_limit = Some(100_000);
model.default_reasoning_summary = ReasoningSummary::Detailed;
model
.service_tiers
.retain(|tier| tier.id != ServiceTier::Fast.request_value());
model
.model_messages
.as_mut()
.expect("model instruction metadata")
.instructions_template =
Some("Destination model instructions.".to_string());
}
}
})
.build_with_auto_env(&server)
.await?;
let request = start_paused_turn(&test.codex).await?;
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
model: Some(MODEL_B.to_string()),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
answer_paused_turn(&test.codex, &request.turn_id).await?;
let paused = wait_for_event_match(&test.codex, |event| match event {
EventMsg::RequestUserInput(request) => Some(request.clone()),
_ => None,
})
.await;
// B cannot use the requested tier. Switching back must recover that
// selection and preserve an unset summary, not reuse B's effective values.
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
model: Some(MODEL_A.to_string()),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
answer_paused_turn(&test.codex, &paused.turn_id).await?;
wait_for_event(&test.codex, |event| match event {
EventMsg::Error(error) => panic!("model activation failed: {}", error.message),
EventMsg::TurnComplete(_) => true,
_ => false,
})
.await;
let requests = response_mock.requests();
assert_eq!(
requests.iter().map(request_settings).collect::<Vec<_>>(),
vec![
json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": "priority",
}),
json!({
"model": MODEL_B,
"reasoning": { "effort": "low", "summary": expected_summary },
"service_tier": null,
}),
json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": "priority",
}),
]
);
assert_eq!(
requests.iter().map(request_turn_id).collect::<Vec<_>>(),
vec![request.turn_id; 3],
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn turn_settings_rejection_preserves_independent_future_settings() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response_mock = mount_sse_sequence(
&server,
vec![
paused_response("resp-1", "pause-before-node-repl-restriction"),
sse_completed("resp-2"),
sse_completed("resp-3"),
],
)
.await;
let test = step_settings_test()
.with_config(|config| {
for model in &mut config
.model_catalog
.as_mut()
.expect("controlled model catalog")
.models
{
model.node_repl_disabled = model.slug == MODEL_B;
}
})
.build_with_auto_env(&server)
.await?;
let request = start_paused_turn(&test.codex).await?;
core_test_support::submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
model: Some(MODEL_B.to_string()),
..Default::default()
},
)
.await?;
let future = test.codex.thread_settings_snapshot().await;
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
model: Some(MODEL_B.to_string()),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Rejected {
reason: "the destination changes the admitted node REPL availability restriction"
.to_string(),
}
);
assert_eq!(test.codex.thread_settings_snapshot().await, future);
answer_paused_turn(&test.codex, &request.turn_id).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.submit_text_turn("admit the next turn").await?;
let requests = response_mock.requests();
assert_eq!(
requests.iter().map(request_settings).collect::<Vec<_>>(),
[MODEL_A, MODEL_A, MODEL_B]
.into_iter()
.map(|model| {
json!({
"model": model,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
})
})
.collect::<Vec<_>>()
);
assert_eq!(request_turn_id(&requests[1]), request.turn_id);
assert_ne!(request_turn_id(&requests[2]), request.turn_id);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn request_preference_activation_keeps_admitted_model_metadata() -> Result<()> {
skip_if_no_network!(Ok(()));
// This test owns both catalog responses. The generic mock-server helper
// installs an extra one-shot /models response that would shift the sequence.
let server = wiremock::MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![
paused_response("resp-1", "pause-before-models-refresh"),
sse_completed("resp-2"),
sse_completed("resp-3"),
],
)
.await;
let mut models = step_settings_models();
for model in &mut models {
model.default_reasoning_summary = ReasoningSummary::Concise;
}
let initial_catalog = mount_models_once(
&server,
ModelsResponse {
models: models.clone(),
},
)
.await;
let test = step_settings_test()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(|config| {
// Use the real refreshable models manager, and retain an unset
// preference so the selected model's default is observable.
config.model_catalog = None;
config.model_reasoning_summary = None;
})
.build_with_auto_env(&server)
.await?;
let request = start_paused_turn(&test.codex).await?;
assert_eq!(initial_catalog.requests().len(), 1);
for model in &mut models {
model.default_reasoning_summary = ReasoningSummary::Detailed;
}
let refresh = mount_models_once(&server, ModelsResponse { models }).await;
test.thread_manager
.get_models_manager()
.list_models(
RefreshStrategy::Online,
codex_core::test_support::default_http_client_factory(),
)
.await;
assert_eq!(refresh.requests().len(), 1);
assert_eq!(
submit_turn_settings(
&test.codex,
&request.turn_id,
TurnSettingsUpdate {
effort: Some(Some(ReasoningEffort::High)),
..Default::default()
}
)
.await?,
TurnSettingsUpdateOutcome::Applied
);
answer_paused_turn(&test.codex, &request.turn_id).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.submit_text_turn("admit the next turn").await?;
assert_eq!(
response_mock
.requests()
.iter()
.map(request_settings)
.collect::<Vec<_>>(),
vec![
json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "concise" },
"service_tier": null,
}),
json!({
"model": MODEL_A,
"reasoning": { "effort": "high", "summary": "concise" },
"service_tier": null,
}),
json!({
"model": MODEL_A,
"reasoning": { "effort": "low", "summary": "detailed" },
"service_tier": null,
}),
]
);
Ok(())
}
+8
View File
@@ -309,6 +309,8 @@ pub enum Feature {
Artifact,
/// Enable Fast mode selection in the TUI and request layer.
FastMode,
/// Enable explicitly requested model changes for later step captures.
StepModelSwitching,
/// Enable experimental realtime voice conversation mode in the TUI.
RealtimeConversation,
/// Prevent idle system sleep while a turn is actively running.
@@ -1515,6 +1517,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::Stable,
default_enabled: true,
},
FeatureSpec {
id: Feature::StepModelSwitching,
key: "step_model_switching",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::RealtimeConversation,
key: "realtime_conversation",
+38 -4
View File
@@ -468,8 +468,33 @@ pub struct ConversationSpeechParams {
pub text: String,
}
/// Persistent thread-settings overrides that can be applied before user input or
/// on their own.
/// Supported sparse changes to one live task's current settings, regardless of
/// task kind. Child sessions and consumers of frozen initial settings are unchanged.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TurnSettingsUpdate {
pub model: Option<String>,
/// `None` preserves the selection; `Some(None)` clears it.
pub effort: Option<Option<ReasoningEffortConfig>>,
pub summary: Option<ReasoningSummaryConfig>,
/// `None` preserves the requested tier; `Some(None)` clears it.
pub service_tier: Option<Option<String>>,
}
/// The result of processing a turn-settings update, not merely queueing it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnSettingsUpdateOutcome {
/// Published for subsequent captures; already captured steps are unchanged.
/// The task need not sample or consume every selected preference.
Applied,
/// The named live task was absent or lost before publication.
TargetUnavailable,
Rejected {
reason: String,
},
}
/// Thread-settings overrides that can be applied before user input or on their
/// own. Standalone updates change the settings inherited by future turns.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ThreadSettingsOverrides {
/// Updated fallback `cwd` and environments supplied together as a complete pair.
@@ -587,15 +612,23 @@ pub enum Op {
reply: oneshot::Sender<CodexResult<SuspendTurnOutcome>>,
},
/// Apply persistent thread-settings overrides without starting a turn.
/// Apply thread-settings 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.
ThreadSettings {
/// Persistent thread-settings overrides to apply.
/// Sparse thread-settings overrides to apply.
thread_settings: ThreadSettingsOverrides,
},
/// Update only the named running turn, without changing future settings.
/// The reply reports the actual publication or why it did not occur.
TurnSettings {
turn_id: String,
update: TurnSettingsUpdate,
reply: oneshot::Sender<TurnSettingsUpdateOutcome>,
},
/// Inter-agent communication that should be recorded as agent-message history
/// while still using the normal thread submission lifecycle.
InterAgentCommunication {
@@ -883,6 +916,7 @@ impl Op {
Self::RecoverTurn { .. } => "recover_turn",
Self::SuspendTurnAndShutdown { .. } => "suspend_turn_and_shutdown",
Self::ThreadSettings { .. } => "thread_settings",
Self::TurnSettings { .. } => "turn_settings",
Self::InterAgentCommunication { .. } => "inter_agent_communication",
Self::ExecApproval { .. } => "exec_approval",
Self::PatchApproval { .. } => "patch_approval",