From f5c5d9b2b093583a1251c5f5cff0bf52addef914 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Wed, 9 Sep 2026 18:29:35 +0000 Subject: [PATCH] Avoid duplicate reasoning effort updates during turn recovery (#44276) ## Why Recovering a suspended turn can append a duplicate reasoning effort update when history already ends with a matching trusted update but the runtime has not established its effort pin. ## What changed Reuse a matching harness-authored `ConfigurationUpdate` at the end of history even when the runtime effort pin is unset. ## Testing Add a regression test that suspends and recovers a turn without WebSocket prewarming, then verifies the recovered request contains exactly one effort update and retains `high` reasoning effort. GitOrigin-RevId: 2e49073eb5e8d4fae39a8dc1d9d2b07f77802adb --- codex-rs/core/src/session/reasoning_effort.rs | 53 ++++----- .../tests/suite/reasoning_effort_override.rs | 102 ++++++++++++++++++ 2 files changed, 131 insertions(+), 24 deletions(-) diff --git a/codex-rs/core/src/session/reasoning_effort.rs b/codex-rs/core/src/session/reasoning_effort.rs index ec945f94a3..fa585a51e8 100644 --- a/codex-rs/core/src/session/reasoning_effort.rs +++ b/codex-rs/core/src/session/reasoning_effort.rs @@ -37,31 +37,36 @@ impl Session { .reasoning_effort_pin .pin(&settings.model_info.slug, effort.clone()); } - let established_effort = - state - .history - .annotated_items() - .iter() - .rev() - .find_map(|envelope| { - if !envelope - .metadata - .as_ref() - .is_some_and(|metadata| metadata.harness_authored_configuration) - { - return None; + let latest_override = state + .history + .annotated_items() + .iter() + .rev() + .enumerate() + .find_map(|(index, envelope)| { + if !envelope + .metadata + .as_ref() + .is_some_and(|metadata| metadata.harness_authored_configuration) + { + return None; + } + match &envelope.item { + ResponseItem::ConfigurationUpdate { reasoning } => { + Some((index, &reasoning.effort)) } - match &envelope.item { - ResponseItem::ConfigurationUpdate { reasoning } => { - Some(&reasoning.effort) - } - _ => None, - } - }); - state - .reasoning_effort_pin - .get(&settings.model_info.slug) - .is_some_and(|pinned| established_effort.unwrap_or(&pinned) == &effort) + _ => None, + } + }); + // Recovery adds no user message. Reuse a matching trusted tail update even + // before this runtime establishes its pin, rather than append adjacent updates. + latest_override.is_some_and(|(index, established)| index == 0 && established == &effort) + || state + .reasoning_effort_pin + .get(&settings.model_info.slug) + .is_some_and(|pinned| { + latest_override.map_or(&pinned, |(_, established)| established) == &effort + }) }; if should_skip { return; diff --git a/codex-rs/core/tests/suite/reasoning_effort_override.rs b/codex-rs/core/tests/suite/reasoning_effort_override.rs index 3d0a966e9f..8a895c7130 100644 --- a/codex-rs/core/tests/suite/reasoning_effort_override.rs +++ b/codex-rs/core/tests/suite/reasoning_effort_override.rs @@ -1,5 +1,10 @@ //! Trusted reasoning-effort updates follow surviving history and the next turn's selected settings. +use codex_core::RecoverTurnRequest; +use codex_core::StartIfIdleSubmission; +use codex_core::SuspendTurnOutcome; +use codex_core::TurnInputRequest; +use codex_core::TurnInputSubmission; use codex_core::config::Config; use codex_extension_api::ExtensionFuture; use codex_extension_api::ExtensionRegistryBuilder; @@ -7,11 +12,13 @@ use codex_extension_api::ThreadIdleInput; use codex_extension_api::ThreadLifecycleContributor; use codex_features::Feature; use codex_login::CodexAuth; +use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::openai_models::ReasoningEffortPreset; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::user_input::UserInput; use core_test_support::responses; use core_test_support::responses::ResponsesRequest; use core_test_support::skip_if_no_network; @@ -65,6 +72,101 @@ fn message(role: &str, text: &str) -> Value { }) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reasoning_effort_override_recovery_reuses_trusted_tail_update() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + // Keep the turn active without allowing any model output after the update. + responses::mount_response_once( + &server, + responses::sse_response(responses::sse(vec![responses::ev_completed("suspended")])) + .set_delay(Duration::from_secs(/*secs*/ 60)), + ) + .await; + let builder = || { + override_builder().with_config(|config| { + config.model_reasoning_effort = Some(ReasoningEffort::High); + // Recovery must work without a prewarm establishing a runtime pin. + config.model_provider.supports_websockets = false; + }) + }; + let test = builder().build_with_auto_env(&server).await?; + let submission = test + .codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "recover this turn".to_string(), + text_elements: Vec::new(), + }])) + .await?; + let TurnInputSubmission::Started { turn_id } = submission else { + panic!("expected a new turn"); + }; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::RawResponseItem(raw) + if matches!(&raw.item, ResponseItem::ConfigurationUpdate { .. })) + }) + .await; + let thread_settings = test.codex.restorable_thread_settings().await; + assert_eq!( + test.codex.suspend_turn_and_shutdown().await?, + SuspendTurnOutcome::Suspended { + turn_id: turn_id.clone(), + }, + ); + let rollout_path = test.codex.rollout_path().expect("rollout path"); + test.thread_manager + .remove_thread(&test.session_configured.thread_id) + .await + .expect("unload suspended thread"); + + let recovery_server = responses::start_mock_server().await; + let mock = responses::mount_sse_once( + &recovery_server, + responses::sse(vec![responses::ev_completed("recovered")]), + ) + .await; + let cwd = test.config.cwd.clone(); + let mut resume_builder = builder().with_config(move |config| config.cwd = cwd); + if let Some(url) = test.executor_environment().exec_server_url() { + resume_builder = resume_builder.with_exec_server_url(url); + } + let resumed = resume_builder + .resume(&recovery_server, Arc::clone(&test.home), rollout_path) + .await?; + resumed + .codex + .restore_thread_settings(thread_settings) + .await?; + assert_eq!( + resumed + .codex + .recover_turn_if_idle(RecoverTurnRequest { + turn_id: turn_id.clone(), + thread_settings: Default::default(), + trace: None, + cyber_access_program: None, + }) + .await?, + StartIfIdleSubmission::Started { turn_id }, + ); + wait_for_event(&resumed.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let request = mock.single_request(); + assert_eq!( + effort_updates(&request), + vec![effort_update(ReasoningEffort::High)] + ); + assert_eq!( + request.input().last(), + Some(&effort_update(ReasoningEffort::High)) + ); + assert_eq!(request.body_json()["reasoning"]["effort"], "high"); + Ok(()) +} + #[test_case(ReasoningEffort::High; "high to persistent and back")] #[test_case(ReasoningEffort::Persistent; "persistent to high and back")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]