diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 9645e1f4ee..11f0555325 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -1246,8 +1246,10 @@ pub(crate) async fn apply_bespoke_event_handling( } EventMsg::ThreadRolledBack(_) | EventMsg::ThreadQueueChanged(_) => {} EventMsg::ThreadSettingsApplied(_) => { - let thread_settings = - thread_settings_from_config_snapshot(&conversation.config_snapshot().await); + let (mut config_snapshot, selected_service_tier) = + conversation.resume_config_snapshot().await; + config_snapshot.service_tier = selected_service_tier; + let thread_settings = thread_settings_from_config_snapshot(&config_snapshot); let changed = { let mut state = thread_state.lock().await; state.note_thread_settings(thread_settings.clone()) diff --git a/codex-rs/app-server/src/request_processors/persisted_resume_settings.rs b/codex-rs/app-server/src/request_processors/persisted_resume_settings.rs index 5e408b02c9..9c4c3f18ad 100644 --- a/codex-rs/app-server/src/request_processors/persisted_resume_settings.rs +++ b/codex-rs/app-server/src/request_processors/persisted_resume_settings.rs @@ -2,6 +2,8 @@ use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::models::ActivePermissionProfile; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ThreadSettingsSnapshot; +use codex_rollout::InitialHistory; use codex_rollout::RolloutItem; #[derive(Debug, PartialEq, Eq)] @@ -46,6 +48,29 @@ pub(super) fn latest_persisted_resume_settings( }) } +pub(super) fn latest_persisted_model_settings( + history: &InitialHistory, +) -> Option<&ThreadSettingsSnapshot> { + let InitialHistory::Resumed(history) = history else { + return None; + }; + let mut crossed_foreign_snapshot = false; + for item in history.history.iter().rev() { + let RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(event)) = item else { + continue; + }; + match event.thread_id { + Some(thread_id) if thread_id == history.conversation_id => { + return Some(&event.thread_settings); + } + Some(_) => crossed_foreign_snapshot = true, + None if !crossed_foreign_snapshot => return Some(&event.thread_settings), + None => {} + } + } + None +} + #[cfg(test)] #[path = "persisted_resume_settings_tests.rs"] mod tests; diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 7c67ec17d4..4d104e2d5e 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -5,6 +5,7 @@ mod daemon_continuation; mod daemon_snapshot; use super::persisted_resume_settings::PersistedResumeSettings; +use super::persisted_resume_settings::latest_persisted_model_settings; use super::persisted_resume_settings::latest_persisted_resume_settings; use super::thread_enrichment::enrich_loaded_threads; use super::thread_fork_goal::inherit_thread_goal_snapshot; @@ -120,6 +121,7 @@ struct ThreadRevertRuntimeSnapshot { fn collect_resume_override_mismatches( request: &ThreadResumeParams, config_snapshot: &ThreadConfigSnapshot, + selected_service_tier: Option<&str>, ) -> Vec { let mut mismatch_details = Vec::new(); @@ -140,7 +142,7 @@ fn collect_resume_override_mismatches( )); } if let Some(requested_service_tier) = request.service_tier.as_ref() - && requested_service_tier != &config_snapshot.service_tier + && Some(requested_service_tier.as_deref().unwrap_or("default")) != selected_service_tier { mismatch_details.push(format!( "service_tier requested={requested_service_tier:?} active={:?}", @@ -2405,10 +2407,17 @@ impl ThreadRequestProcessor { app_server_client_version: Option, ) -> Result { let ThreadRevertRuntimeSnapshot { - config, + mut config, settings, client_mcp_extensions, } = runtime_snapshot; + if let Some(collaboration_mode) = settings.collaboration_mode.as_ref() { + config.model = Some(collaboration_mode.model().to_string()); + } + if let Some(service_tier) = settings.service_tier.as_ref() { + config.service_tier.clone_from(service_tier); + } + config.personality = settings.personality; let thread_id_string = thread_id.to_string(); let stored_thread = self .read_stored_thread_for_resume( @@ -4186,6 +4195,17 @@ impl ThreadRequestProcessor { } let has_explicit_model_resume_override = has_model_resume_override(request_overrides.as_ref(), &typesafe_overrides); + let persisted_model_settings = latest_persisted_model_settings(&thread_history); + let has_override = |key| { + request_overrides + .as_ref() + .is_some_and(|overrides| overrides.contains_key(key)) + }; + let restore_service_tier = + typesafe_overrides.service_tier.is_none() && !has_override("service_tier"); + let restore_summary = !has_override("model_reasoning_summary"); + let restore_personality = + typesafe_overrides.personality.is_none() && !has_override("personality"); typesafe_overrides.persisted_permission_profile = history_settings .filter(|settings| { settings.active_permission_profile.is_none() @@ -4249,7 +4269,17 @@ impl ThreadRequestProcessor { if clear_reasoning_effort { config.model_reasoning_effort = None; } - + if let Some(settings) = persisted_model_settings { + if restore_service_tier { + config.service_tier.clone_from(&settings.service_tier); + } + if restore_summary { + config.model_reasoning_summary = settings.reasoning_summary; + } + if restore_personality { + config.personality = settings.personality; + } + } let response_history = thread_history.clone(); match self @@ -4616,8 +4646,13 @@ impl ThreadRequestProcessor { active_path.display() ))); } - let config_snapshot = existing_thread.config_snapshot().await; - let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot); + let (config_snapshot, selected_service_tier) = + existing_thread.resume_config_snapshot().await; + let mismatch_details = collect_resume_override_mismatches( + params, + &config_snapshot, + selected_service_tier.as_deref(), + ); if !mismatch_details.is_empty() { let has_subscribers = !self .thread_state_manager diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 166142ffdf..dab5a6062f 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -759,7 +759,11 @@ mod thread_processor_behavior_tests { }; assert_eq!( - collect_resume_override_mismatches(&request, &config_snapshot), + collect_resume_override_mismatches( + &request, + &config_snapshot, + /* selected_service_tier */ Some("flex"), + ), vec!["service_tier requested=Some(\"priority\") active=Some(\"flex\")".to_string()] ); } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index e183b147fb..b872aab23f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -505,7 +505,9 @@ async fn thread_resume_rejects_unmaterialized_thread( async fn thread_resume_with_empty_path_uses_running_thread_id() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; - mock_responses_config(&server.uri()).write(codex_home.path())?; + mock_responses_config(&server.uri()) + .enable_feature(Feature::FastMode) + .write(codex_home.path())?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -514,7 +516,8 @@ async fn thread_resume_with_empty_path_uses_running_thread_id() -> Result<()> { let start_id = mcp .send_thread_start_request_with_auto_env(ThreadStartParams { - model: Some("gpt-5.4".to_string()), + model: Some("gpt-daybreak-blue-latest".to_string()), + service_tier: Some(Some("priority".to_string())), ..Default::default() }) .await?; @@ -547,6 +550,7 @@ async fn thread_resume_with_empty_path_uses_running_thread_id() -> Result<()> { .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), path: Some(PathBuf::new()), + service_tier: Some(Some("priority".to_string())), exclude_turns: true, ..Default::default() }) @@ -584,6 +588,27 @@ async fn thread_resume_with_empty_path_uses_running_thread_id() -> Result<()> { serde_json::Value::Array(initialization), json!(["new", "new"]) ); + + timeout(DEFAULT_READ_TIMEOUT, mcp.shutdown_gracefully()).await??; + mock_responses_config(&server.uri()) + .enable_feature(Feature::FastMode) + .with_root_config("service_tier = \"flex\"") + .write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let resumed: ThreadResumeResponse = mcp + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread.id, + ..Default::default() + }, + }) + .await?; + // The startup selection still wins over the new global flex tier, but remains ineffective. + assert_eq!(resumed.service_tier, None); Ok(()) } @@ -818,7 +843,9 @@ async fn thread_resume_restores_startup_and_updated_runtime_workspace_roots() -> #[tokio::test] async fn cold_resume_workspace_roots_honor_owned_snapshot_and_caller_overrides() -> Result<()> { - let server = create_mock_responses_server_repeating_assistant("Done").await; + let server = responses::start_mock_server().await; + let response_mock = + responses::mount_sse_sequence(&server, vec![responses::sse_completed("done"); 3]).await; let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) .with_sandbox_mode("workspace-write") @@ -849,7 +876,7 @@ sandbox = "unelevated" let owned: ThreadSettingsAppliedEvent = serde_json::from_value(json!({ "thread_id": thread_id, "thread_settings": { - "model": "gpt-5.4", + "model": "gpt-5.5", "model_provider_id": "mock_provider", "cwd": saved_cwd, "runtime_workspace_roots": [saved_cwd, new_cwd, extra_root], @@ -859,14 +886,21 @@ sandbox = "unelevated" .materialize_project_roots_with_workspace_roots( &[saved_cwd.clone(), new_cwd.clone(), extra_root.clone()] ), - "collaboration_mode": { "mode": "default", "settings": { "model": "gpt-5.4" } }, + "service_tier": "flex", + "reasoning_summary": "concise", + "personality": "friendly", + "collaboration_mode": { "mode": "default", "settings": { "model": "gpt-5.5" } }, }, }))?; let mut foreign = owned.clone(); foreign.thread_id = Some(ThreadId::new()); foreign.thread_settings.runtime_workspace_roots = Some(vec![extra_root.clone()]); foreign.thread_settings.permission_profile = PermissionProfile::Disabled; - for settings in [owned, foreign] { + foreign.thread_settings.service_tier = Some("default".to_string()); + let mut ownerless = owned.clone(); + ownerless.thread_id = None; + ownerless.thread_settings.service_tier = Some("priority".to_string()); + for settings in [owned, ownerless.clone(), foreign] { append_rollout_item_to_path( &path, &RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(settings)), @@ -874,7 +908,7 @@ sandbox = "unelevated" .await?; } // Every iteration is a cold resume; the final one must retain the preceding explicit clear. - for (cwd_override, roots_override, expected_cwd, expected_roots) in [ + for (index, (cwd_override, roots_override, expected_cwd, expected_roots)) in [ ( Some(&new_cwd), None, @@ -889,7 +923,10 @@ sandbox = "unelevated" ), (None, Some(Vec::new()), &saved_cwd, Vec::new()), (None, None, &saved_cwd, Vec::new()), - ] { + ] + .into_iter() + .enumerate() + { let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) .without_auto_env() @@ -901,6 +938,14 @@ sandbox = "unelevated" thread_id: thread_id.clone(), cwd: cwd_override.map(|cwd| cwd.to_string_lossy().into_owned()), runtime_workspace_roots: roots_override, + service_tier: (index == 2).then_some(Some("priority".to_string())), + personality: (index == 2).then_some(Personality::None), + config: (index == 1).then(|| { + std::collections::HashMap::from([ + ("model_reasoning_summary".to_string(), json!("detailed")), + ("personality".to_string(), json!("none")), + ]) + }), ..Default::default() }) .await?; @@ -910,6 +955,26 @@ sandbox = "unelevated" (resumed.cwd, resumed.runtime_workspace_roots), (expected_cwd.clone(), expected_roots.clone()) ); + if index < 3 { + mcp.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread_id.clone(), + ..Default::default() + }) + .await?; + let request = response_mock.last_request().expect("response request"); + assert_eq!( + ( + request.body_json()["service_tier"].as_str(), + request.body_json()["reasoning"]["summary"].as_str(), + request.body_contains_text("# Personality"), + ), + [ + (Some("flex"), Some("concise"), true), + (Some("priority"), Some("detailed"), false), + (Some("priority"), Some("detailed"), false), + ][index] + ); + } // Inspect persistence before shutdown or another resume can hide a missing checkpoint. let expected_thread_id = ThreadId::from_string(&thread_id)?; @@ -941,6 +1006,13 @@ sandbox = "unelevated" ); } timeout(DEFAULT_READ_TIMEOUT, mcp.shutdown_gracefully()).await??; + if index == 0 { + append_rollout_item_to_path( + &path, + &RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(ownerless.clone())), + ) + .await?; + } } Ok(()) } @@ -2617,6 +2689,7 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { "approval_policy": "never", "approvals_reviewer": "user", "permission_profile": PermissionProfile::read_only(), + "service_tier": "flex", "collaboration_mode": { "mode": "default", "settings": { "model": "gpt-5.4" } }, }, }))?; @@ -2637,8 +2710,12 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { ..Default::default() }) .await?; - let ThreadResumeResponse { thread, cwd, .. } = - timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??; + let ThreadResumeResponse { + thread, + cwd, + service_tier, + .. + } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??; assert_eq!(thread.id, conversation_id); assert_eq!(thread.preview, preview); @@ -2650,7 +2727,7 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); assert_eq!(thread.status, ThreadStatus::Idle); - + assert_eq!(service_tier.as_deref(), Some("flex")); assert_eq!( thread.turns.len(), 1, diff --git a/codex-rs/app-server/tests/suite/v2/thread_revert.rs b/codex-rs/app-server/tests/suite/v2/thread_revert.rs index 67b4be96de..ffbdbdef42 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_revert.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_revert.rs @@ -37,6 +37,7 @@ use codex_app_server_protocol::UserInput; use codex_features::Feature; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Personality; use codex_protocol::config_types::Settings; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::EventMsg; @@ -45,6 +46,7 @@ use codex_rollout::RolloutItem; use codex_rollout::read_session_meta_line; use codex_utils_absolute_path::AbsolutePathBuf; use core_test_support::load_default_config_for_test; +use core_test_support::responses; use pretty_assertions::assert_eq; use serde_json::Value; use tempfile::TempDir; @@ -418,9 +420,14 @@ async fn thread_revert_preserves_fork_cutoff_after_cold_resume() -> Result<()> { #[tokio::test] async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { - let server = create_mock_responses_server_repeating_assistant("Done").await; + let server = responses::start_mock_server().await; + let response_mock = + responses::mount_sse_sequence(&server, vec![responses::sse_completed("done"); 4]).await; let codex_home = TempDir::new()?; - MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + MockResponsesConfig::new(&server.uri()) + .with_model("gpt-5.5") + .enable_feature(Feature::FastMode) + .write(codex_home.path())?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) .build() @@ -430,6 +437,8 @@ async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { let ThreadStartResponse { thread, .. } = mcp .start_thread(ThreadStartParams { history_mode: Some(ThreadHistoryMode::Paginated), + service_tier: Some(Some("priority".to_string())), + personality: Some(Personality::Friendly), ..Default::default() }) .await?; @@ -443,6 +452,8 @@ async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { text: text.to_string(), text_elements: Vec::new(), }], + model: (text == "second").then_some("gpt-daybreak-blue-latest".to_string()), + personality: (text == "second").then_some(Personality::None), ..Default::default() }) .await?; @@ -565,21 +576,25 @@ async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> { text: "third".to_string(), text_elements: Vec::new(), }], + model: Some("gpt-5.5".to_string()), ..Default::default() }) .await?; - let requests = server.received_requests().await.expect("response requests"); - let model_input = requests - .iter() - .rev() - .find(|request| request.url.path().ends_with("/responses")) - .expect("third turn response request") - .body_json::()?["input"] - .clone(); + let body = response_mock + .last_request() + .expect("third turn request") + .body_json(); + assert_eq!(body["service_tier"], "priority"); + assert!( + !body["instructions"] + .as_str() + .is_some_and(|instructions| { instructions.contains("# Personality") }) + ); + let model_input = body["input"].clone(); let model_input = serde_json::to_string(&model_input)?; - assert!(model_input.contains("first")); - assert!(!model_input.contains("second")); - assert!(model_input.contains("third")); + assert!(model_input.contains(r#""text":"first""#)); + assert!(!model_input.contains(r#""text":"second""#)); + assert!(model_input.contains(r#""text":"third""#)); assert_eq!( turn_ids_from_cursor( &mut mcp, diff --git a/codex-rs/app-server/tests/suite/v2/thread_settings_update.rs b/codex-rs/app-server/tests/suite/v2/thread_settings_update.rs index 261f9e42da..ab688b12e9 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_settings_update.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_settings_update.rs @@ -235,7 +235,13 @@ async fn thread_settings_update_emits_notification_and_updates_future_turns() -> let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; write_models_cache(codex_home.path()).await?; - let (model_id, service_tier_id) = service_tier_model_and_tier_id()?; + let (_, service_tier_id) = service_tier_model_and_tier_id()?; + let model_id = all_model_presets() + .iter() + .find(|preset| preset.service_tiers.is_empty()) + .context("bundled model catalog should include a model without service tiers")? + .id + .clone(); let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -265,8 +271,6 @@ async fn thread_settings_update_emits_notification_and_updates_future_turns() -> "settings-only update should not start a model request" ); - start_text_turn(&mut mcp, thread.id.clone()).await?; - let updated = read_thread_settings_updated(&mut mcp).await?; assert_eq!(updated.thread_id, thread.id); assert_eq!(updated.thread_settings.model, model_id); @@ -275,6 +279,8 @@ async fn thread_settings_update_emits_notification_and_updates_future_turns() -> Some(service_tier_id.as_str()) ); + start_text_turn(&mut mcp, thread.id.clone()).await?; + timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), @@ -345,10 +351,9 @@ async fn thread_settings_update_emits_notification_and_updates_future_turns() -> assert!( request_bodies.iter().any(|body| { body.get("model").and_then(Value::as_str) == Some(model_id.as_str()) - && body.get("service_tier").and_then(Value::as_str) - == Some(service_tier_id.as_str()) + && body.get("service_tier").is_none() }), - "future turn did not use updated model/service tier: {request_bodies:#?}" + "future turn did not omit the unsupported service tier: {request_bodies:#?}" ); Ok(()) } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 8652ebde0c..e016026865 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -961,6 +961,10 @@ impl CodexThread { self.session.thread_config_snapshot().await } + pub async fn resume_config_snapshot(&self) -> (ThreadConfigSnapshot, Option) { + self.session.resume_config_snapshot().await + } + /// Returns the active turn's reviewer, including live updates, or the thread default. pub async fn approvals_reviewer_for_turn(&self, turn_id: &str) -> ApprovalsReviewer { if let Some((turn, settings, _, _)) = self diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index dbac6896c0..e08463dc59 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -909,8 +909,6 @@ impl Session { &config.features, &model_info, ); - let service_tier = - get_service_tier(config.service_tier.clone(), &config.features, &model_info); let storage_originator = AuthStorageOriginator::from_client_name(&originator); let session_configuration = SessionConfiguration { turn_extension_init, @@ -921,7 +919,7 @@ impl Session { step_settings: Arc::new(StepSettings { collaboration_mode, reasoning_summary: config.model_reasoning_summary, - service_tier, + service_tier: config.service_tier.clone(), personality: config.personality, approval_policy: config.permissions.approval_policy.clone(), approvals_reviewer: config.approvals_reviewer, @@ -2116,16 +2114,26 @@ impl Session { &self, updates: &SessionSettingsUpdate, ) -> ConstraintResult { - let state = self.state.lock().await; - let configuration = self.apply_session_settings(&state.session_configuration, updates)?; - Ok(configuration.thread_config_snapshot(configuration.environments.clone())) + let configuration = { + let state = self.state.lock().await; + self.apply_session_settings(&state.session_configuration, updates)? + }; + Ok(self + .effective_thread_config_snapshot(&configuration, configuration.environments.clone()) + .await) } pub(crate) async fn thread_config_snapshot(&self) -> ThreadConfigSnapshot { - let state = self.state.lock().await; - state - .session_configuration - .thread_config_snapshot(state.session_configuration.environments.clone()) + self.resume_config_snapshot().await.0 + } + + pub(crate) async fn resume_config_snapshot(&self) -> (ThreadConfigSnapshot, Option) { + let configuration = self.state.lock().await.session_configuration.clone(); + let selected_service_tier = configuration.step_settings.service_tier.clone(); + let snapshot = self + .effective_thread_config_snapshot(&configuration, configuration.environments.clone()) + .await; + (snapshot, selected_service_tier) } /// Captures configured child bindings, or the active turn's bindings for legacy inheritance. @@ -2240,7 +2248,8 @@ impl Session { } ( config, - configuration.thread_config_snapshot(environments.all_selections()), + self.effective_thread_config_snapshot(&configuration, environments.all_selections()) + .await, configuration.windows_sandbox_level, environments, reasoning_effort_pin, @@ -3766,11 +3775,28 @@ impl Session { metadata: image, }); } - let rollout_items: Vec = - items.into_iter().map(RolloutItem::ResponseItem).collect(); - if self.persist_rollout_items(&rollout_items).await - && let Some(revision) = mcp_revision + let settings_guard = + if force_mcp_checkpoint && self.reference_context_item().await.is_none() { + Some(thread_settings::acquire_persistence_lock(self).await) + } else { + None + }; + let initial_settings = if settings_guard.is_some() + && self.reference_context_item().await.is_none() + && self.state.lock().await.take_initial_settings_pending() { + Some(thread_settings::applied_event(self).await) + } else { + None + }; + let rollout_items = initial_settings + .into_iter() + .map(RolloutItem::EventMsg) + .chain(items.into_iter().map(RolloutItem::ResponseItem)) + .collect::>(); + let persisted = self.persist_rollout_items(&rollout_items).await; + drop(settings_guard); + if persisted && let Some(revision) = mcp_revision { self.services .executed_tool_calls .mark_mcp_attribution_persisted(revision); diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 0e5c4080d5..e3bcb2befb 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -663,6 +663,25 @@ async fn warm_plugins_and_skills_for_session_init( } impl Session { + pub(super) async fn effective_thread_config_snapshot( + &self, + configuration: &SessionConfiguration, + environments: Vec, + ) -> ThreadConfigSnapshot { + let mut snapshot = configuration.thread_config_snapshot(environments); + let model_info = Box::pin(configuration.step_settings.resolve_model_info( + self.services.models_manager.as_ref(), + &configuration.model_info_overrides, + )) + .await; + snapshot.service_tier = get_service_tier( + snapshot.service_tier.take(), + &configuration.original_config_do_not_use.features, + &model_info, + ); + snapshot + } + /// Returns the concrete identity for this thread. pub(crate) fn thread_id(&self) -> ThreadId { self.thread_id @@ -1879,8 +1898,12 @@ impl Session { *guard = Arc::downgrade(&sess); } // Dispatch the SessionConfiguredEvent first and then report any errors. - let thread_config = - session_configuration.thread_config_snapshot(turn_environments.selections()); + let thread_config = sess + .effective_thread_config_snapshot( + &session_configuration, + turn_environments.selections(), + ) + .await; let events = std::iter::once(Event { id: INITIAL_SUBMIT_ID.to_owned(), msg: EventMsg::SessionConfigured(SessionConfiguredEvent { diff --git a/codex-rs/core/src/session/step_settings.rs b/codex-rs/core/src/session/step_settings.rs index 83b1edd04c..b66fb1931d 100644 --- a/codex-rs/core/src/session/step_settings.rs +++ b/codex-rs/core/src/session/step_settings.rs @@ -26,8 +26,7 @@ pub(crate) struct StepSettings { pub(crate) collaboration_mode: CollaborationMode, /// `None` follows the pinned model's default. pub(crate) reasoning_summary: Option, - /// Normalized requested tier. Startup retains initial-model filtering; - /// resolution filters the effective tier against the pinned model. + /// Normalized requested tier; resolution filters it against the pinned model. pub(crate) service_tier: Option, pub(crate) personality: Option, pub(crate) approval_policy: Constrained, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index f30130e641..bba5b0e12c 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -394,7 +394,6 @@ pub(crate) async fn run_turn( { return Ok(None); } - // Only speculate after hooks accept the turn, using its finalized tools and permissions. { let mut state = sess.state.lock().await; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index fff1bf667f..d122db22d8 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -104,6 +104,7 @@ pub(crate) struct SessionState { pub(crate) active_connector_selection: HashSet, pub(crate) pending_session_start_sources: VecDeque, next_turn_is_first: bool, + initial_settings_pending: bool, } impl SessionState { @@ -146,6 +147,7 @@ impl SessionState { active_connector_selection: HashSet::new(), pending_session_start_sources: VecDeque::new(), next_turn_is_first: true, + initial_settings_pending: true, } } @@ -161,6 +163,7 @@ impl SessionState { pub(crate) fn set_next_turn_is_first(&mut self, value: bool) { self.next_turn_is_first = value; + self.initial_settings_pending = value; } pub(crate) fn take_next_turn_is_first(&mut self) -> bool { @@ -169,6 +172,12 @@ impl SessionState { is_first_turn } + pub(crate) fn take_initial_settings_pending(&mut self) -> bool { + let pending = self.initial_settings_pending; + self.initial_settings_pending = false; + pending + } + pub(crate) fn clone_history(&self) -> ContextManager { self.history.clone() } diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index ead690898c..006c2415e8 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -2101,6 +2101,15 @@ async fn interrupt_during_mcp_startup_preserves_user_input_in_history( ) }) .expect("an interrupted turn should retain its submitted user prompt"); + assert!( + history.items[..user_prompt_index] + .iter() + .any(|item| matches!( + item, + RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(applied)) + if applied.thread_id == Some(fixture.session_configured.thread_id) + )) + ); let RolloutItem::ResponseItem(envelope) = &history.items[user_prompt_index] else { unreachable!("found the user prompt"); }; diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 5ccd4bf751..ae4532818a 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -9486,7 +9486,7 @@ async fn override_turn_context_sends_thread_settings_update() { app.enqueue_primary_thread_session(started.session, started.turns) .await .expect("primary thread should be registered"); - let service_tier = ServiceTier::Fast.request_value().to_string(); + let service_tier = ServiceTier::Fast.request_value(); let collaboration_mode = CollaborationMode { mode: ModeKind::Plan, settings: Settings { @@ -9506,7 +9506,7 @@ async fn override_turn_context_sends_thread_settings_update() { Some("gpt-5.4".to_string()), Some(Some(ReasoningEffortConfig::High)), /*summary*/ None, - Some(Some(service_tier.clone())), + Some(Some(service_tier.to_string())), Some(collaboration_mode.clone()), ); @@ -9532,8 +9532,8 @@ async fn override_turn_context_sends_thread_settings_update() { Some(ReasoningEffortConfig::High) ); assert_eq!( - notification.thread_settings.service_tier, - Some(service_tier.clone()) + notification.thread_settings.service_tier.as_deref(), + Some(service_tier) ); assert_eq!( notification.thread_settings.approval_policy, @@ -9580,7 +9580,7 @@ async fn override_turn_context_sends_thread_settings_update() { updated_mode.settings.reasoning_effort, collaboration_mode.settings.reasoning_effort ); - assert_eq!(updated_session.service_tier, Some(service_tier)); + assert_eq!(updated_session.service_tier.as_deref(), Some(service_tier)); assert_eq!(updated_session.approval_policy, AskForApproval::OnRequest); assert_eq!( updated_session.approvals_reviewer, diff --git a/codex-rs/tui/src/app/tests/permission_selection_tests.rs b/codex-rs/tui/src/app/tests/permission_selection_tests.rs index d9f43a5b17..99ff64e4f2 100644 --- a/codex-rs/tui/src/app/tests/permission_selection_tests.rs +++ b/codex-rs/tui/src/app/tests/permission_selection_tests.rs @@ -244,6 +244,10 @@ async fn inline_server_permission_selection_holds_queued_prompt_until_confirmati #[tokio::test] async fn custom_permission_selection_uses_server_definition_and_preserves_state_on_rejection() -> Result<()> { + boxed_future!(custom_permission_selection_scenario()).await +} + +async fn custom_permission_selection_scenario() -> Result<()> { let (mut app, mut events, _ops) = make_test_app_with_channels().await; app.config.ephemeral = false; let home = tempdir()?;