openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve thread model settings across resume and revert (#55923)

Browse Source 
## Why

Resuming a thread can replace saved model settings with global defaults. A selected service tier can also be lost when the current model does not support it.

## What changed

- Restore saved service tier, reasoning summary, and personality on resume unless the caller explicitly overrides them, respecting thread ownership of settings snapshots.
- Retain the selected service tier for persistence, settings notifications, and resume comparisons while filtering the effective tier against the current model's capabilities.
- Carry the runtime model, service tier, and personality into thread revert reconstruction.
- Persist initial thread settings before user input during MCP startup so an interrupted first turn retains its settings checkpoint.

## Testing

Extend regression coverage for cold resume and explicit overrides, unsupported service tiers, settings preservation during revert, and settings checkpoints before user input when MCP startup is interrupted.

GitOrigin-RevId: e5ec012a5678048c5eedd443e524f38c142bf9fd
This commit is contained in:
Eric Traut
2026-10-21 15:34:43 +00:00
committed by copyberry
parent a6839a4812
commit 430a16799d
16 changed files with 299 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+4 -2
View File
@@ -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())
codex-rs/app-server/src/request_processors/persisted_resume_settings.rs
+25
View File
@@ -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;
codex-rs/app-server/src/request_processors/thread_processor.rs
+40 -5
View File
@@ -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<String> {
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<String>,
) -> Result<ThreadRevertResponse, JSONRPCErrorError> {
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
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
+5 -1
View File
@@ -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()]
);
}
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+88 -11
View File
@@ -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,
codex-rs/app-server/tests/suite/v2/thread_revert.rs
+28 -13
View File
@@ -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::<serde_json::Value>()?["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,
codex-rs/app-server/tests/suite/v2/thread_settings_update.rs
+11 -6
View File
@@ -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(())
}
codex-rs/core/src/codex_thread.rs
+4
View File
@@ -961,6 +961,10 @@ impl CodexThread {
self.session.thread_config_snapshot().await
}
pub async fn resume_config_snapshot(&self) -> (ThreadConfigSnapshot, Option<String>) {
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
codex-rs/core/src/session/mod.rs
+41 -15
View File
@@ -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<ThreadConfigSnapshot> {
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<String>) {
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<RolloutItem> =
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::<Vec<_>>();
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);
codex-rs/core/src/session/session.rs
+25 -2
View File
@@ -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<TurnEnvironmentSelection>,
) -> 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 {
codex-rs/core/src/session/step_settings.rs
+1 -2
View File
@@ -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<ReasoningSummary>,
/// 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<String>,
pub(crate) personality: Option<Personality>,
pub(crate) approval_policy: Constrained<AskForApproval>,
codex-rs/core/src/session/turn.rs
-1
View File
@@ -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;
codex-rs/core/src/state/session.rs
+9
View File
@@ -104,6 +104,7 @@ pub(crate) struct SessionState {
pub(crate) active_connector_selection: HashSet<String>,
pub(crate) pending_session_start_sources: VecDeque<codex_hooks::SessionStartSource>,
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()
}
codex-rs/core/tests/suite/rmcp_client.rs
+9
View File
@@ -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");
};
codex-rs/tui/src/app/tests.rs
+5 -5
View File
@@ -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,
codex-rs/tui/src/app/tests/permission_selection_tests.rs
+4
View File
@@ -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()?;