From 2bc58007b96c6138cd4d6fd2f6e563f8ff1cb738 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Mon, 6 Jul 2026 16:25:32 -0700 Subject: [PATCH 1/8] Retry retired model compaction with current model --- codex-rs/core/src/compact_model_fallback.rs | 70 ++++ .../core/src/compact_model_fallback_tests.rs | 37 ++ codex-rs/core/src/compact_remote.rs | 62 +++- codex-rs/core/src/compact_remote_v2.rs | 70 +++- codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/turn.rs | 34 +- codex-rs/core/tests/suite/compact.rs | 345 ++++++++++++++++++ 7 files changed, 590 insertions(+), 29 deletions(-) create mode 100644 codex-rs/core/src/compact_model_fallback.rs create mode 100644 codex-rs/core/src/compact_model_fallback_tests.rs diff --git a/codex-rs/core/src/compact_model_fallback.rs b/codex-rs/core/src/compact_model_fallback.rs new file mode 100644 index 000000000000..6646fb4e98cc --- /dev/null +++ b/codex-rs/core/src/compact_model_fallback.rs @@ -0,0 +1,70 @@ +use codex_analytics::CompactionImplementation; +use codex_analytics::CompactionReason; +use codex_otel::SessionTelemetry; +use codex_protocol::error::CodexErr; +use serde_json::Value; +use tracing::warn; + +#[cfg(test)] +#[path = "compact_model_fallback_tests.rs"] +mod tests; + +pub(crate) fn is_model_unavailable_error(error: &CodexErr, model: &str) -> bool { + let CodexErr::InvalidRequest(message) = error else { + return false; + }; + let expected = + format!("The '{model}' model is not supported when using Codex with a ChatGPT account."); + if message == &expected { + return true; + } + serde_json::from_str::(message) + .ok() + .and_then(|value| { + value + .get("error") + .and_then(|error| error.get("message")) + .and_then(Value::as_str) + .map(str::to_string) + }) + .is_some_and(|message| message == expected) +} + +pub(crate) fn record_model_fallback( + session_telemetry: &SessionTelemetry, + previous_model: &str, + current_model: &str, + reason: CompactionReason, + implementation: CompactionImplementation, + succeeded: bool, +) { + let reason_tag = match reason { + CompactionReason::UserRequested => "user_requested", + CompactionReason::ContextLimit => "context_limit", + CompactionReason::ModelDownshift => "model_downshift", + CompactionReason::CompHashChanged => "comp_hash_changed", + }; + let implementation_tag = match implementation { + CompactionImplementation::Responses => "responses", + CompactionImplementation::ResponsesCompactionV2 => "responses_compaction_v2", + CompactionImplementation::ResponsesCompact => "responses_compact", + }; + let outcome = if succeeded { "succeeded" } else { "failed" }; + session_telemetry.counter( + "codex.compaction.model_fallback", + /*inc*/ 1, + &[ + ("reason", reason_tag), + ("implementation", implementation_tag), + ("outcome", outcome), + ], + ); + warn!( + previous_model, + current_model, + ?reason, + ?implementation, + outcome, + "previous model was unavailable during compaction; retried with current model" + ); +} diff --git a/codex-rs/core/src/compact_model_fallback_tests.rs b/codex-rs/core/src/compact_model_fallback_tests.rs new file mode 100644 index 000000000000..9fc3691dff48 --- /dev/null +++ b/codex-rs/core/src/compact_model_fallback_tests.rs @@ -0,0 +1,37 @@ +use super::is_model_unavailable_error; +use codex_protocol::error::CodexErr; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn model_unavailable_error_requires_the_expected_model_rejection() { + let model = "gpt-5.6-oai"; + let expected_message = + "The 'gpt-5.6-oai' model is not supported when using Codex with a ChatGPT account."; + let errors = [ + CodexErr::InvalidRequest(expected_message.to_string()), + CodexErr::InvalidRequest( + json!({ + "error": { + "type": "invalid_request_error", + "message": expected_message, + } + }) + .to_string(), + ), + CodexErr::InvalidRequest( + "The 'gpt-5.7-oai' model is not supported when using Codex with a ChatGPT account." + .to_string(), + ), + CodexErr::InvalidRequest("generic invalid request".to_string()), + CodexErr::RequestTimeout, + ]; + + assert_eq!( + errors + .iter() + .map(|error| is_model_unavailable_error(error, model)) + .collect::>(), + vec![true, true, false, false, false] + ); +} diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index 8964b9aa425d..d276cecbc588 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -7,6 +7,8 @@ use crate::compact::InitialContextInjection; use crate::compact::build_compaction_initial_context; use crate::compact::compaction_status_from_result; use crate::compact::insert_initial_context_before_last_real_user_or_summary; +use crate::compact_model_fallback::is_model_unavailable_error; +use crate::compact_model_fallback::record_model_fallback; use crate::context::world_state::WorldState; use crate::context_manager::ContextManager; use crate::hook_runtime::PostCompactHookOutcome; @@ -45,6 +47,7 @@ const CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE: &str = pub(crate) async fn run_inline_remote_auto_compact_task( sess: Arc, step_context: Arc, + fallback_step_context: Option>, turn_state: Arc>, initial_context_injection: InitialContextInjection, reason: CompactionReason, @@ -53,6 +56,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task( run_remote_compact_task_inner( &sess, &step_context, + fallback_step_context.as_ref(), Some(turn_state), initial_context_injection, CompactionTrigger::Auto, @@ -81,6 +85,7 @@ pub(crate) async fn run_remote_compact_task( run_remote_compact_task_inner( &sess, &step_context, + /*fallback_step_context*/ None, /*turn_state*/ None, InitialContextInjection::DoNotInject, CompactionTrigger::Manual, @@ -94,6 +99,7 @@ pub(crate) async fn run_remote_compact_task( async fn run_remote_compact_task_inner( sess: &Arc, step_context: &Arc, + fallback_step_context: Option<&Arc>, turn_state: Option>>, initial_context_injection: InitialContextInjection, trigger: CompactionTrigger, @@ -139,9 +145,11 @@ async fn run_remote_compact_task_inner( let result = run_remote_compact_task_inner_impl( sess, step_context, + fallback_step_context, turn_state, initial_context_injection, compaction_metadata, + reason, &mut analytics_details, ) .await; @@ -173,9 +181,11 @@ async fn run_remote_compact_task_inner( async fn run_remote_compact_task_inner_impl( sess: &Arc, step_context: &Arc, + fallback_step_context: Option<&Arc>, turn_state: Option>>, initial_context_injection: InitialContextInjection, compaction_metadata: CompactionTurnMetadata, + reason: CompactionReason, analytics_details: &mut CompactionAnalyticsDetails, ) -> CodexResult<()> { let turn_context = &step_context.turn; @@ -191,22 +201,52 @@ async fn run_remote_compact_task_inner_impl( let compaction_item = TurnItem::ContextCompaction(context_compaction_item); sess.emit_turn_item_started(turn_context, &compaction_item) .await; - let RemoteCompactAttempt { - new_history, - trace_input_history, - } = run_remote_compact_attempt( + let attempt = run_remote_compact_attempt( sess, step_context, - turn_state, + turn_state.clone(), &compaction_trace, compaction_metadata, analytics_details, ) - .await?; + .await; + let (attempt, compaction_turn_context) = match attempt { + Ok(attempt) => (attempt, turn_context), + Err(error) => { + let Some(fallback_step_context) = fallback_step_context else { + return Err(error); + }; + if !is_model_unavailable_error(&error, turn_context.model_info.slug.as_str()) { + return Err(error); + } + let fallback_result = run_remote_compact_attempt( + sess, + fallback_step_context, + turn_state, + &compaction_trace, + compaction_metadata, + analytics_details, + ) + .await; + record_model_fallback( + &sess.services.session_telemetry, + turn_context.model_info.slug.as_str(), + fallback_step_context.turn.model_info.slug.as_str(), + reason, + CompactionImplementation::ResponsesCompact, + fallback_result.is_ok(), + ); + (fallback_result?, &fallback_step_context.turn) + } + }; + let RemoteCompactAttempt { + new_history, + trace_input_history, + } = attempt; let (new_window_number, new_window_ids) = sess.advance_auto_compact_window().await; let (new_history, world_state_baseline) = process_compacted_history( sess.as_ref(), - turn_context.as_ref(), + compaction_turn_context.as_ref(), new_history, &initial_context_injection, ) @@ -215,7 +255,7 @@ async fn run_remote_compact_task_inner_impl( let reference_context_item = match initial_context_injection { InitialContextInjection::DoNotInject => None, InitialContextInjection::BeforeLastUserMessage(_) => { - Some(turn_context.to_turn_context_item()) + Some(compaction_turn_context.to_turn_context_item()) } }; let compacted_item = CompactedItem { @@ -234,16 +274,16 @@ async fn run_remote_compact_task_inner_impl( replacement_history: &new_history, }); sess.replace_compacted_history( - turn_context.as_ref(), + compaction_turn_context.as_ref(), new_history, reference_context_item, world_state_baseline, compacted_item, ) .await; - sess.recompute_token_usage(turn_context).await; + sess.recompute_token_usage(compaction_turn_context).await; - sess.emit_turn_item_completed(turn_context, compaction_item) + sess.emit_turn_item_completed(compaction_turn_context, compaction_item) .await; Ok(()) } diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 8105f63b7afa..98ead9c56778 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -8,6 +8,8 @@ use crate::compact::CompactionAnalyticsAttempt; use crate::compact::CompactionAnalyticsDetails; use crate::compact::InitialContextInjection; use crate::compact::compaction_status_from_result; +use crate::compact_model_fallback::is_model_unavailable_error; +use crate::compact_model_fallback::record_model_fallback; use crate::compact_remote::process_compacted_history; use crate::compact_remote::should_keep_compacted_history_item; use crate::hook_runtime::PostCompactHookOutcome; @@ -57,6 +59,7 @@ const MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES: u64 = 2; pub(crate) async fn run_inline_remote_auto_compact_task( sess: Arc, step_context: Arc, + fallback_step_context: Option>, client_session: &mut ModelClientSession, initial_context_injection: InitialContextInjection, reason: CompactionReason, @@ -65,6 +68,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task( run_remote_compact_task_inner( &sess, &step_context, + fallback_step_context.as_ref(), Some(client_session), initial_context_injection, CompactionTrigger::Auto, @@ -92,6 +96,7 @@ pub(crate) async fn run_remote_compact_task( run_remote_compact_task_inner( &sess, &step_context, + /*fallback_step_context*/ None, /*client_session*/ None, InitialContextInjection::DoNotInject, CompactionTrigger::Manual, @@ -104,6 +109,7 @@ pub(crate) async fn run_remote_compact_task( async fn run_remote_compact_task_inner( sess: &Arc, step_context: &Arc, + fallback_step_context: Option<&Arc>, client_session: Option<&mut ModelClientSession>, initial_context_injection: InitialContextInjection, trigger: CompactionTrigger, @@ -149,9 +155,11 @@ async fn run_remote_compact_task_inner( let result = run_remote_compact_task_inner_impl( sess, step_context, + fallback_step_context, client_session, initial_context_injection, compaction_metadata, + reason, &mut analytics_details, ) .await; @@ -186,9 +194,11 @@ async fn run_remote_compact_task_inner( async fn run_remote_compact_task_inner_impl( sess: &Arc, step_context: &Arc, - client_session: Option<&mut ModelClientSession>, + fallback_step_context: Option<&Arc>, + mut client_session: Option<&mut ModelClientSession>, initial_context_injection: InitialContextInjection, compaction_metadata: CompactionTurnMetadata, + reason: CompactionReason, analytics_details: &mut CompactionAnalyticsDetails, ) -> CodexResult<()> { let turn_context = &step_context.turn; @@ -203,21 +213,51 @@ async fn run_remote_compact_task_inner_impl( sess.emit_turn_item_started(turn_context, &compaction_item) .await; - let RemoteCompactV2Attempt { - trace_input_history, - prompt_input, - compaction_output, - token_usage, - owned_client_session: _owned_client_session, - } = run_remote_compact_v2_attempt( + let attempt = run_remote_compact_v2_attempt( sess, step_context, - client_session, + client_session.as_deref_mut(), &compaction_trace, compaction_metadata, analytics_details, ) - .await?; + .await; + let (attempt, compaction_turn_context) = match attempt { + Ok(attempt) => (attempt, turn_context), + Err(error) => { + let Some(fallback_step_context) = fallback_step_context else { + return Err(error); + }; + if !is_model_unavailable_error(&error, turn_context.model_info.slug.as_str()) { + return Err(error); + } + let fallback_result = run_remote_compact_v2_attempt( + sess, + fallback_step_context, + client_session.as_deref_mut(), + &compaction_trace, + compaction_metadata, + analytics_details, + ) + .await; + record_model_fallback( + &sess.services.session_telemetry, + turn_context.model_info.slug.as_str(), + fallback_step_context.turn.model_info.slug.as_str(), + reason, + CompactionImplementation::ResponsesCompactionV2, + fallback_result.is_ok(), + ); + (fallback_result?, &fallback_step_context.turn) + } + }; + let RemoteCompactV2Attempt { + trace_input_history, + prompt_input, + compaction_output, + token_usage, + owned_client_session: _owned_client_session, + } = attempt; if let Some(token_usage) = token_usage { sess.record_rollout_budget_usage(&token_usage)?; analytics_details.active_context_tokens_before = Some(token_usage.input_tokens); @@ -230,7 +270,7 @@ async fn run_remote_compact_task_inner_impl( let (new_window_number, new_window_ids) = sess.advance_auto_compact_window().await; let (new_history, world_state_baseline) = process_compacted_history( sess.as_ref(), - turn_context.as_ref(), + compaction_turn_context.as_ref(), compacted_history, &initial_context_injection, ) @@ -239,7 +279,7 @@ async fn run_remote_compact_task_inner_impl( let reference_context_item = match initial_context_injection { InitialContextInjection::DoNotInject => None, InitialContextInjection::BeforeLastUserMessage(_) => { - Some(turn_context.to_turn_context_item()) + Some(compaction_turn_context.to_turn_context_item()) } }; let compacted_item = CompactedItem { @@ -255,16 +295,16 @@ async fn run_remote_compact_task_inner_impl( replacement_history: &new_history, }); sess.replace_compacted_history( - turn_context.as_ref(), + compaction_turn_context.as_ref(), new_history, reference_context_item, world_state_baseline, compacted_item, ) .await; - sess.recompute_token_usage(turn_context).await; + sess.recompute_token_usage(compaction_turn_context).await; - sess.emit_turn_item_completed(turn_context, compaction_item) + sess.emit_turn_item_completed(compaction_turn_context, compaction_item) .await; Ok(()) } diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 91ccbd6e63ff..e57288f05dfe 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -19,6 +19,7 @@ pub use responses_metadata::CodexResponsesMetadata; pub use session::SteerInputError; pub use turn_metadata::detached_memory_responses_metadata; mod codex_thread; +mod compact_model_fallback; mod compact_remote; mod compact_remote_v2; mod compact_token_budget; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 708f42f0ffd7..e3d07c913e3c 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -349,6 +349,7 @@ pub(crate) async fn run_turn( if let Err(err) = run_auto_compact( &sess, Arc::clone(&step_context), + /*fallback_step_context*/ None, &mut client_session, InitialContextInjection::BeforeLastUserMessage(Arc::clone(&world_state)), CompactionReason::ContextLimit, @@ -810,6 +811,7 @@ async fn run_pre_sampling_compact( run_auto_compact( sess, step_context, + /*fallback_step_context*/ None, client_session, InitialContextInjection::DoNotInject, CompactionReason::ContextLimit, @@ -828,6 +830,24 @@ fn comp_hash_changed(previous: Option<&str>, current: Option<&str>) -> bool { .is_some_and(|(previous, current)| previous != current) } +async fn previous_model_fallback_step_context( + sess: &Arc, + turn_context: &Arc, + previous_model: &str, +) -> Option> { + let uses_codex_backend = turn_context + .auth_manager + .as_deref() + .is_some_and(codex_login::AuthManager::current_auth_uses_codex_backend); + if !uses_codex_backend + || !turn_context.provider.info().is_openai() + || previous_model == turn_context.model_info.slug + { + return None; + } + Some(sess.capture_step_context(Arc::clone(turn_context)).await) +} + /// Runs pre-sampling compaction against the previous model when its compaction compatibility /// hash changed or when switching to a smaller context-window model. /// @@ -844,20 +864,23 @@ async fn maybe_run_previous_model_inline_compact( previous_turn_settings.comp_hash.as_deref(), turn_context.model_info.comp_hash.as_deref(), ); + let previous_model = previous_turn_settings.model; let previous_model_turn_context = Arc::new( turn_context - .with_model(previous_turn_settings.model, &sess.services.models_manager) + .with_model(previous_model.clone(), &sess.services.models_manager) .await, ); if should_compact_for_comp_hash_change { - // This pre-turn request needs a step built from the previous model's turn context. let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; + let fallback_step_context = + previous_model_fallback_step_context(sess, turn_context, previous_model.as_str()).await; run_auto_compact( sess, step_context, + fallback_step_context, client_session, InitialContextInjection::DoNotInject, CompactionReason::CompHashChanged, @@ -892,13 +915,15 @@ async fn maybe_run_previous_model_inline_compact( && previous_model_turn_context.model_info.slug != turn_context.model_info.slug && old_context_window > new_context_window; if should_run { - // This pre-turn request needs a step built from the previous model's turn context. let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; + let fallback_step_context = + previous_model_fallback_step_context(sess, turn_context, previous_model.as_str()).await; run_auto_compact( sess, step_context, + fallback_step_context, client_session, InitialContextInjection::DoNotInject, CompactionReason::ModelDownshift, @@ -917,6 +942,7 @@ async fn maybe_run_previous_model_inline_compact( async fn run_auto_compact( sess: &Arc, step_context: Arc, + fallback_step_context: Option>, client_session: &mut ModelClientSession, initial_context_injection: InitialContextInjection, reason: CompactionReason, @@ -949,6 +975,7 @@ async fn run_auto_compact( run_inline_remote_auto_compact_task_v2( Arc::clone(sess), step_context, + fallback_step_context, client_session, initial_context_injection, reason, @@ -965,6 +992,7 @@ async fn run_auto_compact( run_inline_remote_auto_compact_task( Arc::clone(sess), step_context, + fallback_step_context, client_session.turn_state(), initial_context_injection, reason, diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 74c28664f7f6..074bdd079f5f 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -274,6 +274,25 @@ fn non_openai_model_provider(server: &MockServer) -> ModelProviderInfo { provider } +fn openai_model_provider(server: &MockServer) -> ModelProviderInfo { + let mut provider = + built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["openai"].clone(); + provider.base_url = Some(format!("{}/v1", server.uri())); + provider.supports_websockets = false; + provider +} + +fn unsupported_chatgpt_model_response(model: &str) -> wiremock::ResponseTemplate { + wiremock::ResponseTemplate::new(/*status*/ 400).set_body_json(json!({ + "error": { + "type": "invalid_request_error", + "message": format!( + "The '{model}' model is not supported when using Codex with a ChatGPT account." + ), + } + })) +} + fn write_global_file( home: &TempDir, filename: &str, @@ -421,6 +440,7 @@ async fn assert_compaction_uses_turn_lifecycle_id(codex: &std::sync::Arc compact_completed_id = Some(event.id.clone()), + EventMsg::Error(error) => panic!("unexpected compaction error: {error:?}"), EventMsg::TurnComplete(_) => turn_completed_id = Some(event.id.clone()), _ => {} } @@ -2276,6 +2296,331 @@ async fn pre_sampling_compact_runs_when_comp_hash_changes() { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pre_sampling_compact_falls_back_from_retired_previous_model_after_rename() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let retired_model = "gpt-5.6-oai"; + let previous_model_family = "gpt-5.6"; + let renamed_model = "gpt-5.6-sol-oai"; + let mut previous_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-a")); + previous_model_info.slug = previous_model_family.to_string(); + let mut renamed_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-b")); + renamed_model_info.slug = renamed_model.to_string(); + + let models_mock = mount_models_once( + &server, + ModelsResponse { + models: vec![previous_model_info, renamed_model_info], + }, + ) + .await; + + let request_log = mount_response_sequence( + &server, + vec![ + sse_response(sse(vec![ + ev_assistant_message("m1", "before switch"), + ev_completed_with_tokens("r1", /*total_tokens*/ 100), + ])), + unsupported_chatgpt_model_response(retired_model), + sse_response(sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { + "type": "compaction", + "encrypted_content": "RETIRED_MODEL_SUMMARY", + } + }), + ev_completed_with_tokens("r2", /*total_tokens*/ 10), + ])), + sse_response(sse(vec![ + ev_assistant_message("m3", "after switch"), + ev_completed_with_tokens("r3", /*total_tokens*/ 100), + ])), + ], + ) + .await; + + let model_provider = openai_model_provider(&server); + let mut initial_builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(retired_model) + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + }); + let initial = initial_builder + .build(&server) + .await + .expect("build initial test codex"); + let home = initial.home.clone(); + let rollout_path = initial + .session_configured + .rollout_path + .clone() + .expect("rollout path"); + + initial + .codex + .submit(disabled_permission_user_turn( + "before switch", + initial.cwd.path().to_path_buf(), + retired_model.to_string(), + )) + .await + .expect("submit first user turn"); + wait_for_event(&initial.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + initial + .codex + .submit(Op::Shutdown) + .await + .expect("shutdown initial session"); + wait_for_event(&initial.codex, |event| { + matches!(event, EventMsg::ShutdownComplete) + }) + .await; + + let model_provider = openai_model_provider(&server); + let mut resumed_builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(retired_model) + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + }); + let resumed = resumed_builder + .resume(&server, home, rollout_path) + .await + .expect("resume codex"); + + resumed + .codex + .submit(disabled_permission_user_turn( + "after switch", + resumed.cwd.path().to_path_buf(), + renamed_model.to_string(), + )) + .await + .expect("submit renamed-model turn"); + assert_compaction_uses_turn_lifecycle_id(&resumed.codex).await; + + let requests = request_log.requests(); + assert_eq!(models_mock.requests().len(), 1); + assert_eq!( + requests.len(), + 4, + "the renamed-model turn should retry compaction and then sample" + ); + assert_eq!( + requests[0].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!( + requests[1].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!( + requests[2].body_json()["model"].as_str(), + Some(renamed_model) + ); + assert_eq!( + requests[3].body_json()["model"].as_str(), + Some(renamed_model) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pre_sampling_compact_falls_back_from_retired_previous_model_on_downshift() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let retired_model = "gpt-5.6-oai"; + let previous_model_family = "gpt-5.6"; + let next_model = "gpt-5.4-ultrafast"; + let mut previous_model_info = + model_info_with_context_window("gpt-5.4", /*context_window*/ 273_000); + previous_model_info.slug = previous_model_family.to_string(); + let mut next_model_info = + model_info_with_context_window("gpt-5.4", /*context_window*/ 125_000); + next_model_info.slug = next_model.to_string(); + + let models_mock = mount_models_once( + &server, + ModelsResponse { + models: vec![previous_model_info, next_model_info], + }, + ) + .await; + + let request_log = mount_response_sequence( + &server, + vec![ + sse_response(sse(vec![ + ev_assistant_message("m1", "before switch"), + ev_completed_with_tokens("r1", /*total_tokens*/ 120_000), + ])), + unsupported_chatgpt_model_response(retired_model), + sse_response(sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { + "type": "compaction", + "encrypted_content": "DOWNSHIFT_SUMMARY", + } + }), + ev_completed_with_tokens("r2", /*total_tokens*/ 10), + ])), + sse_response(sse(vec![ + ev_assistant_message("m3", "after switch"), + ev_completed_with_tokens("r3", /*total_tokens*/ 100), + ])), + ], + ) + .await; + + let model_provider = openai_model_provider(&server); + let mut builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(retired_model) + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + let _ = config.features.enable(Feature::RemoteCompactionV2); + }); + let test = builder.build(&server).await.expect("build test codex"); + + test.codex + .submit(disabled_permission_user_turn( + "before switch", + test.cwd.path().to_path_buf(), + retired_model.to_string(), + )) + .await + .expect("submit first user turn"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + test.codex + .submit(disabled_permission_user_turn( + "after switch", + test.cwd.path().to_path_buf(), + next_model.to_string(), + )) + .await + .expect("submit smaller-model turn"); + assert_compaction_uses_turn_lifecycle_id(&test.codex).await; + + let requests = request_log.requests(); + assert_eq!(models_mock.requests().len(), 1); + assert_eq!( + requests.len(), + 4, + "the smaller-model turn should retry compaction and then sample" + ); + assert_eq!( + requests[0].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!( + requests[1].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!(requests[2].body_json()["model"].as_str(), Some(next_model)); + assert_eq!(requests[3].body_json()["model"].as_str(), Some(next_model)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pre_sampling_compact_keeps_unknown_previous_model_for_api_key_auth_and_custom_provider() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let previous_model = "custom/gpt-5.6-oai"; + let previous_model_family = "gpt-5.6"; + let next_model = "gpt-5.6-sol-oai"; + let mut previous_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-a")); + previous_model_info.slug = previous_model_family.to_string(); + let mut next_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-b")); + next_model_info.slug = next_model.to_string(); + + let model_catalog = ModelsResponse { + models: vec![previous_model_info, next_model_info], + }; + + let request_log = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_assistant_message("m1", "before switch"), + ev_completed_with_tokens("r1", /*total_tokens*/ 100), + ]), + sse(vec![ + ev_assistant_message("m2", "CUSTOM_MODEL_SUMMARY"), + ev_completed_with_tokens("r2", /*total_tokens*/ 10), + ]), + sse(vec![ + ev_assistant_message("m3", "after switch"), + ev_completed_with_tokens("r3", /*total_tokens*/ 100), + ]), + ], + ) + .await; + + let model_provider = non_openai_model_provider(&server); + let mut builder = test_codex() + .with_auth(CodexAuth::from_api_key("Test API Key")) + .with_model(previous_model) + .with_config(move |config| { + config.model_provider = model_provider; + config.model_catalog = Some(model_catalog); + set_test_compact_prompt(config); + }); + let test = builder.build(&server).await.expect("build test codex"); + + test.codex + .submit(disabled_permission_user_turn( + "before switch", + test.cwd.path().to_path_buf(), + previous_model.to_string(), + )) + .await + .expect("submit first user turn"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + test.codex + .submit(disabled_permission_user_turn( + "after switch", + test.cwd.path().to_path_buf(), + next_model.to_string(), + )) + .await + .expect("submit next-model turn"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let requests = request_log.requests(); + assert_eq!(requests.len(), 3); + assert_pre_sampling_switch_compaction_requests( + &requests[0].body_json(), + &requests[1].body_json(), + &requests[2].body_json(), + previous_model, + next_model, + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn pre_sampling_compact_skips_when_either_comp_hash_is_missing() { skip_if_no_network!(); From b65b5decbe826b0452035ef854d44c35967a3538 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Mon, 6 Jul 2026 17:58:52 -0700 Subject: [PATCH 2/8] Clarify current-model compaction fallback --- codex-rs/core/src/session/turn.rs | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index e3d07c913e3c..e231f6af8671 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -830,7 +830,11 @@ fn comp_hash_changed(previous: Option<&str>, current: Option<&str>) -> bool { .is_some_and(|(previous, current)| previous != current) } -async fn previous_model_fallback_step_context( +/// Captures the current model's request-scoped state for retrying previous-model compaction. +/// +/// Returns `None` when the active authentication does not use the Codex backend, the provider is +/// not OpenAI, or the previous and current model are the same. +async fn capture_current_model_fallback_step_context( sess: &Arc, turn_context: &Arc, previous_model: &str, @@ -875,8 +879,12 @@ async fn maybe_run_previous_model_inline_compact( let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; - let fallback_step_context = - previous_model_fallback_step_context(sess, turn_context, previous_model.as_str()).await; + let fallback_step_context = capture_current_model_fallback_step_context( + sess, + turn_context, + previous_model.as_str(), + ) + .await; run_auto_compact( sess, step_context, @@ -918,8 +926,12 @@ async fn maybe_run_previous_model_inline_compact( let step_context = sess .capture_step_context(Arc::clone(&previous_model_turn_context)) .await; - let fallback_step_context = - previous_model_fallback_step_context(sess, turn_context, previous_model.as_str()).await; + let fallback_step_context = capture_current_model_fallback_step_context( + sess, + turn_context, + previous_model.as_str(), + ) + .await; run_auto_compact( sess, step_context, From 64fed666878d1604ed6996514d1849c9c89fdca3 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Mon, 6 Jul 2026 22:38:59 -0700 Subject: [PATCH 3/8] Use invalid request for compaction fallback --- codex-rs/core/src/compact_model_fallback.rs | 37 ++++--------------- .../core/src/compact_model_fallback_tests.rs | 37 ------------------- codex-rs/core/src/compact_remote.rs | 10 +++-- codex-rs/core/src/compact_remote_v2.rs | 10 +++-- codex-rs/core/tests/suite/compact.rs | 33 ++++++++--------- 5 files changed, 35 insertions(+), 92 deletions(-) delete mode 100644 codex-rs/core/src/compact_model_fallback_tests.rs diff --git a/codex-rs/core/src/compact_model_fallback.rs b/codex-rs/core/src/compact_model_fallback.rs index 6646fb4e98cc..10d1ce743aaf 100644 --- a/codex-rs/core/src/compact_model_fallback.rs +++ b/codex-rs/core/src/compact_model_fallback.rs @@ -2,41 +2,15 @@ use codex_analytics::CompactionImplementation; use codex_analytics::CompactionReason; use codex_otel::SessionTelemetry; use codex_protocol::error::CodexErr; -use serde_json::Value; use tracing::warn; -#[cfg(test)] -#[path = "compact_model_fallback_tests.rs"] -mod tests; - -pub(crate) fn is_model_unavailable_error(error: &CodexErr, model: &str) -> bool { - let CodexErr::InvalidRequest(message) = error else { - return false; - }; - let expected = - format!("The '{model}' model is not supported when using Codex with a ChatGPT account."); - if message == &expected { - return true; - } - serde_json::from_str::(message) - .ok() - .and_then(|value| { - value - .get("error") - .and_then(|error| error.get("message")) - .and_then(Value::as_str) - .map(str::to_string) - }) - .is_some_and(|message| message == expected) -} - pub(crate) fn record_model_fallback( session_telemetry: &SessionTelemetry, previous_model: &str, current_model: &str, reason: CompactionReason, implementation: CompactionImplementation, - succeeded: bool, + fallback_error: Option<&CodexErr>, ) { let reason_tag = match reason { CompactionReason::UserRequested => "user_requested", @@ -49,7 +23,11 @@ pub(crate) fn record_model_fallback( CompactionImplementation::ResponsesCompactionV2 => "responses_compaction_v2", CompactionImplementation::ResponsesCompact => "responses_compact", }; - let outcome = if succeeded { "succeeded" } else { "failed" }; + let outcome = if fallback_error.is_none() { + "succeeded" + } else { + "failed" + }; session_telemetry.counter( "codex.compaction.model_fallback", /*inc*/ 1, @@ -65,6 +43,7 @@ pub(crate) fn record_model_fallback( ?reason, ?implementation, outcome, - "previous model was unavailable during compaction; retried with current model" + ?fallback_error, + "previous-model compaction failed; retried with current model" ); } diff --git a/codex-rs/core/src/compact_model_fallback_tests.rs b/codex-rs/core/src/compact_model_fallback_tests.rs deleted file mode 100644 index 9fc3691dff48..000000000000 --- a/codex-rs/core/src/compact_model_fallback_tests.rs +++ /dev/null @@ -1,37 +0,0 @@ -use super::is_model_unavailable_error; -use codex_protocol::error::CodexErr; -use pretty_assertions::assert_eq; -use serde_json::json; - -#[test] -fn model_unavailable_error_requires_the_expected_model_rejection() { - let model = "gpt-5.6-oai"; - let expected_message = - "The 'gpt-5.6-oai' model is not supported when using Codex with a ChatGPT account."; - let errors = [ - CodexErr::InvalidRequest(expected_message.to_string()), - CodexErr::InvalidRequest( - json!({ - "error": { - "type": "invalid_request_error", - "message": expected_message, - } - }) - .to_string(), - ), - CodexErr::InvalidRequest( - "The 'gpt-5.7-oai' model is not supported when using Codex with a ChatGPT account." - .to_string(), - ), - CodexErr::InvalidRequest("generic invalid request".to_string()), - CodexErr::RequestTimeout, - ]; - - assert_eq!( - errors - .iter() - .map(|error| is_model_unavailable_error(error, model)) - .collect::>(), - vec![true, true, false, false, false] - ); -} diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index d276cecbc588..bbb29ad6cfcc 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -7,7 +7,6 @@ use crate::compact::InitialContextInjection; use crate::compact::build_compaction_initial_context; use crate::compact::compaction_status_from_result; use crate::compact::insert_initial_context_before_last_real_user_or_summary; -use crate::compact_model_fallback::is_model_unavailable_error; use crate::compact_model_fallback::record_model_fallback; use crate::context::world_state::WorldState; use crate::context_manager::ContextManager; @@ -216,7 +215,7 @@ async fn run_remote_compact_task_inner_impl( let Some(fallback_step_context) = fallback_step_context else { return Err(error); }; - if !is_model_unavailable_error(&error, turn_context.model_info.slug.as_str()) { + if !matches!(&error, CodexErr::InvalidRequest(_)) { return Err(error); } let fallback_result = run_remote_compact_attempt( @@ -234,9 +233,12 @@ async fn run_remote_compact_task_inner_impl( fallback_step_context.turn.model_info.slug.as_str(), reason, CompactionImplementation::ResponsesCompact, - fallback_result.is_ok(), + fallback_result.as_ref().err(), ); - (fallback_result?, &fallback_step_context.turn) + match fallback_result { + Ok(attempt) => (attempt, &fallback_step_context.turn), + Err(_) => return Err(error), + } } }; let RemoteCompactAttempt { diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 98ead9c56778..6d65c20c02b0 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -8,7 +8,6 @@ use crate::compact::CompactionAnalyticsAttempt; use crate::compact::CompactionAnalyticsDetails; use crate::compact::InitialContextInjection; use crate::compact::compaction_status_from_result; -use crate::compact_model_fallback::is_model_unavailable_error; use crate::compact_model_fallback::record_model_fallback; use crate::compact_remote::process_compacted_history; use crate::compact_remote::should_keep_compacted_history_item; @@ -228,7 +227,7 @@ async fn run_remote_compact_task_inner_impl( let Some(fallback_step_context) = fallback_step_context else { return Err(error); }; - if !is_model_unavailable_error(&error, turn_context.model_info.slug.as_str()) { + if !matches!(&error, CodexErr::InvalidRequest(_)) { return Err(error); } let fallback_result = run_remote_compact_v2_attempt( @@ -246,9 +245,12 @@ async fn run_remote_compact_task_inner_impl( fallback_step_context.turn.model_info.slug.as_str(), reason, CompactionImplementation::ResponsesCompactionV2, - fallback_result.is_ok(), + fallback_result.as_ref().err(), ); - (fallback_result?, &fallback_step_context.turn) + match fallback_result { + Ok(attempt) => (attempt, &fallback_step_context.turn), + Err(_) => return Err(error), + } } }; let RemoteCompactV2Attempt { diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 074bdd079f5f..3a81f6c1b1a2 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -282,14 +282,9 @@ fn openai_model_provider(server: &MockServer) -> ModelProviderInfo { provider } -fn unsupported_chatgpt_model_response(model: &str) -> wiremock::ResponseTemplate { +fn invalid_request_response(message: impl Into) -> wiremock::ResponseTemplate { wiremock::ResponseTemplate::new(/*status*/ 400).set_body_json(json!({ - "error": { - "type": "invalid_request_error", - "message": format!( - "The '{model}' model is not supported when using Codex with a ChatGPT account." - ), - } + "detail": message.into(), })) } @@ -2301,9 +2296,9 @@ async fn pre_sampling_compact_falls_back_from_retired_previous_model_after_renam skip_if_no_network!(); let server = MockServer::start().await; - let retired_model = "gpt-5.6-oai"; - let previous_model_family = "gpt-5.6"; - let renamed_model = "gpt-5.6-sol-oai"; + let retired_model = "gpt-5.5"; + let previous_model_family = "gpt-5.5"; + let renamed_model = "gpt-5.6"; let mut previous_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-a")); previous_model_info.slug = previous_model_family.to_string(); let mut renamed_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-b")); @@ -2324,7 +2319,9 @@ async fn pre_sampling_compact_falls_back_from_retired_previous_model_after_renam ev_assistant_message("m1", "before switch"), ev_completed_with_tokens("r1", /*total_tokens*/ 100), ])), - unsupported_chatgpt_model_response(retired_model), + invalid_request_response(format!( + "The '{retired_model}' model is not supported when using Codex with a ChatGPT account." + )), sse_response(sse(vec![ json!({ "type": "response.output_item.done", @@ -2436,13 +2433,13 @@ async fn pre_sampling_compact_falls_back_from_retired_previous_model_after_renam } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn pre_sampling_compact_falls_back_from_retired_previous_model_on_downshift() { +async fn pre_sampling_compact_falls_back_after_previous_model_invalid_request_on_downshift() { skip_if_no_network!(); let server = MockServer::start().await; - let retired_model = "gpt-5.6-oai"; + let retired_model = "gpt-5.6"; let previous_model_family = "gpt-5.6"; - let next_model = "gpt-5.4-ultrafast"; + let next_model = "gpt-5.5"; let mut previous_model_info = model_info_with_context_window("gpt-5.4", /*context_window*/ 273_000); previous_model_info.slug = previous_model_family.to_string(); @@ -2465,7 +2462,7 @@ async fn pre_sampling_compact_falls_back_from_retired_previous_model_on_downshif ev_assistant_message("m1", "before switch"), ev_completed_with_tokens("r1", /*total_tokens*/ 120_000), ])), - unsupported_chatgpt_model_response(retired_model), + invalid_request_response("previous-model compaction was rejected"), sse_response(sse(vec![ json!({ "type": "response.output_item.done", @@ -2542,9 +2539,9 @@ async fn pre_sampling_compact_keeps_unknown_previous_model_for_api_key_auth_and_ skip_if_no_network!(); let server = MockServer::start().await; - let previous_model = "custom/gpt-5.6-oai"; - let previous_model_family = "gpt-5.6"; - let next_model = "gpt-5.6-sol-oai"; + let previous_model = "custom/gpt-5.5"; + let previous_model_family = "gpt-5.5"; + let next_model = "gpt-5.6"; let mut previous_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-a")); previous_model_info.slug = previous_model_family.to_string(); let mut next_model_info = model_info_with_optional_comp_hash("gpt-5.4", Some("hash-b")); From 5e2acb572e73eb0c587c75e86345fb4f4c1d076e Mon Sep 17 00:00:00 2001 From: celia-oai Date: Mon, 6 Jul 2026 23:35:00 -0700 Subject: [PATCH 4/8] Fix remote compaction clippy failures --- codex-rs/core/src/compact_remote.rs | 42 ++++++++++++----------- codex-rs/core/src/compact_remote_v2.rs | 44 +++++++++++++------------ codex-rs/core/src/responses_metadata.rs | 16 +++++++++ 3 files changed, 61 insertions(+), 41 deletions(-) diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index bbb29ad6cfcc..af36009d4002 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -52,15 +52,19 @@ pub(crate) async fn run_inline_remote_auto_compact_task( reason: CompactionReason, phase: CompactionPhase, ) -> CodexResult<()> { + let compaction_metadata = CompactionTurnMetadata::new( + CompactionTrigger::Auto, + reason, + CompactionImplementation::ResponsesCompact, + phase, + ); run_remote_compact_task_inner( &sess, &step_context, fallback_step_context.as_ref(), Some(turn_state), initial_context_injection, - CompactionTrigger::Auto, - reason, - phase, + compaction_metadata, ) .await?; Ok(()) @@ -81,15 +85,19 @@ pub(crate) async fn run_remote_compact_task( }); sess.send_event(&turn_context, start_event).await; + let compaction_metadata = CompactionTurnMetadata::new( + CompactionTrigger::Manual, + CompactionReason::UserRequested, + CompactionImplementation::ResponsesCompact, + CompactionPhase::StandaloneTurn, + ); run_remote_compact_task_inner( &sess, &step_context, /*fallback_step_context*/ None, /*turn_state*/ None, InitialContextInjection::DoNotInject, - CompactionTrigger::Manual, - CompactionReason::UserRequested, - CompactionPhase::StandaloneTurn, + compaction_metadata, ) .await?; Ok(()) @@ -101,17 +109,13 @@ async fn run_remote_compact_task_inner( fallback_step_context: Option<&Arc>, turn_state: Option>>, initial_context_injection: InitialContextInjection, - trigger: CompactionTrigger, - reason: CompactionReason, - phase: CompactionPhase, + compaction_metadata: CompactionTurnMetadata, ) -> CodexResult<()> { let turn_context = &step_context.turn; - let compaction_metadata = CompactionTurnMetadata::new( - trigger, - reason, - CompactionImplementation::ResponsesCompact, - phase, - ); + let trigger = compaction_metadata.trigger(); + let reason = compaction_metadata.reason(); + let implementation = compaction_metadata.implementation(); + let phase = compaction_metadata.phase(); let mut analytics_details = CompactionAnalyticsDetails { active_context_tokens_before: Some(sess.get_total_token_usage().await), ..Default::default() @@ -121,7 +125,7 @@ async fn run_remote_compact_task_inner( turn_context.as_ref(), trigger, reason, - CompactionImplementation::ResponsesCompact, + implementation, phase, ) .await; @@ -148,7 +152,6 @@ async fn run_remote_compact_task_inner( turn_state, initial_context_injection, compaction_metadata, - reason, &mut analytics_details, ) .await; @@ -184,7 +187,6 @@ async fn run_remote_compact_task_inner_impl( turn_state: Option>>, initial_context_injection: InitialContextInjection, compaction_metadata: CompactionTurnMetadata, - reason: CompactionReason, analytics_details: &mut CompactionAnalyticsDetails, ) -> CodexResult<()> { let turn_context = &step_context.turn; @@ -231,8 +233,8 @@ async fn run_remote_compact_task_inner_impl( &sess.services.session_telemetry, turn_context.model_info.slug.as_str(), fallback_step_context.turn.model_info.slug.as_str(), - reason, - CompactionImplementation::ResponsesCompact, + compaction_metadata.reason(), + compaction_metadata.implementation(), fallback_result.as_ref().err(), ); match fallback_result { diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 6d65c20c02b0..7991440b8542 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -64,15 +64,19 @@ pub(crate) async fn run_inline_remote_auto_compact_task( reason: CompactionReason, phase: CompactionPhase, ) -> CodexResult<()> { + let compaction_metadata = CompactionTurnMetadata::new( + CompactionTrigger::Auto, + reason, + CompactionImplementation::ResponsesCompactionV2, + phase, + ); run_remote_compact_task_inner( &sess, &step_context, fallback_step_context.as_ref(), Some(client_session), initial_context_injection, - CompactionTrigger::Auto, - reason, - phase, + compaction_metadata, ) .await } @@ -92,15 +96,19 @@ pub(crate) async fn run_remote_compact_task( }); sess.send_event(&turn_context, start_event).await; + let compaction_metadata = CompactionTurnMetadata::new( + CompactionTrigger::Manual, + CompactionReason::UserRequested, + CompactionImplementation::ResponsesCompactionV2, + CompactionPhase::StandaloneTurn, + ); run_remote_compact_task_inner( &sess, &step_context, /*fallback_step_context*/ None, /*client_session*/ None, InitialContextInjection::DoNotInject, - CompactionTrigger::Manual, - CompactionReason::UserRequested, - CompactionPhase::StandaloneTurn, + compaction_metadata, ) .await } @@ -111,17 +119,13 @@ async fn run_remote_compact_task_inner( fallback_step_context: Option<&Arc>, client_session: Option<&mut ModelClientSession>, initial_context_injection: InitialContextInjection, - trigger: CompactionTrigger, - reason: CompactionReason, - phase: CompactionPhase, + compaction_metadata: CompactionTurnMetadata, ) -> CodexResult<()> { let turn_context = &step_context.turn; - let compaction_metadata = CompactionTurnMetadata::new( - trigger, - reason, - CompactionImplementation::ResponsesCompactionV2, - phase, - ); + let trigger = compaction_metadata.trigger(); + let reason = compaction_metadata.reason(); + let implementation = compaction_metadata.implementation(); + let phase = compaction_metadata.phase(); let mut analytics_details = CompactionAnalyticsDetails { active_context_tokens_before: Some(sess.get_total_token_usage().await), ..Default::default() @@ -131,7 +135,7 @@ async fn run_remote_compact_task_inner( turn_context.as_ref(), trigger, reason, - CompactionImplementation::ResponsesCompactionV2, + implementation, phase, ) .await; @@ -158,7 +162,6 @@ async fn run_remote_compact_task_inner( client_session, initial_context_injection, compaction_metadata, - reason, &mut analytics_details, ) .await; @@ -197,7 +200,6 @@ async fn run_remote_compact_task_inner_impl( mut client_session: Option<&mut ModelClientSession>, initial_context_injection: InitialContextInjection, compaction_metadata: CompactionTurnMetadata, - reason: CompactionReason, analytics_details: &mut CompactionAnalyticsDetails, ) -> CodexResult<()> { let turn_context = &step_context.turn; @@ -233,7 +235,7 @@ async fn run_remote_compact_task_inner_impl( let fallback_result = run_remote_compact_v2_attempt( sess, fallback_step_context, - client_session.as_deref_mut(), + client_session, &compaction_trace, compaction_metadata, analytics_details, @@ -243,8 +245,8 @@ async fn run_remote_compact_task_inner_impl( &sess.services.session_telemetry, turn_context.model_info.slug.as_str(), fallback_step_context.turn.model_info.slug.as_str(), - reason, - CompactionImplementation::ResponsesCompactionV2, + compaction_metadata.reason(), + compaction_metadata.implementation(), fallback_result.as_ref().err(), ); match fallback_result { diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 1852c4bea680..c76d3078cde4 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -93,6 +93,22 @@ impl CompactionTurnMetadata { strategy: CompactionStrategy::Memento, } } + + pub(crate) fn trigger(self) -> CompactionTrigger { + self.trigger + } + + pub(crate) fn reason(self) -> CompactionReason { + self.reason + } + + pub(crate) fn implementation(self) -> CompactionImplementation { + self.implementation + } + + pub(crate) fn phase(self) -> CompactionPhase { + self.phase + } } #[derive(Clone, Copy, Debug)] From 01c3ab57a9604e09476e503593442b01bad5d4a9 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Tue, 7 Jul 2026 13:17:19 -0700 Subject: [PATCH 5/8] Test legacy remote compaction model fallback --- codex-rs/core/tests/common/responses.rs | 39 +++++++++ codex-rs/core/tests/suite/compact.rs | 107 ++++++++++++++++++++++++ 2 files changed, 146 insertions(+) diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 9569df1c90ec..1ab3a70cbb9e 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -1569,6 +1569,45 @@ pub async fn mount_response_sequence( response_mock } +/// Mounts a sequence of responses for each POST to `/v1/responses/compact`. +/// Panics if more requests are received than responses provided. +pub async fn mount_compact_response_sequence( + server: &MockServer, + responses: Vec, +) -> ResponseMock { + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + + struct SeqResponder { + num_calls: AtomicUsize, + responses: Vec, + } + + impl Respond for SeqResponder { + fn respond(&self, _: &wiremock::Request) -> ResponseTemplate { + let call_num = self.num_calls.fetch_add(1, Ordering::SeqCst); + self.responses + .get(call_num) + .expect("missing response for compact call") + .clone() + } + } + + let num_calls = responses.len(); + let responder = SeqResponder { + num_calls: AtomicUsize::new(0), + responses, + }; + + let (mock, response_mock) = compact_mock(); + mock.respond_with(responder) + .up_to_n_times(num_calls as u64) + .expect(num_calls as u64) + .mount(server) + .await; + response_mock +} + /// Validate invariants on the request body sent to `/v1/responses`. /// /// - No `function_call_output`/`custom_tool_call_output` with missing/empty `call_id`. diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index 3a81f6c1b1a2..fdcd4c05657c 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -48,6 +48,7 @@ use core_test_support::responses::ev_completed; use core_test_support::responses::ev_completed_with_tokens; use core_test_support::responses::ev_function_call; use core_test_support::responses::mount_compact_json_once; +use core_test_support::responses::mount_compact_response_sequence; use core_test_support::responses::mount_response_sequence; use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_once_match; @@ -2534,6 +2535,112 @@ async fn pre_sampling_compact_falls_back_after_previous_model_invalid_request_on assert_eq!(requests[3].body_json()["model"].as_str(), Some(next_model)); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pre_sampling_legacy_remote_compact_falls_back_after_previous_model_invalid_request() { + skip_if_no_network!(); + + let server = MockServer::start().await; + let retired_model = "gpt-5.6"; + let previous_model_family = "gpt-5.6"; + let next_model = "gpt-5.5"; + let mut previous_model_info = + model_info_with_context_window("gpt-5.4", /*context_window*/ 273_000); + previous_model_info.slug = previous_model_family.to_string(); + let mut next_model_info = + model_info_with_context_window("gpt-5.4", /*context_window*/ 125_000); + next_model_info.slug = next_model.to_string(); + + let models_mock = mount_models_once( + &server, + ModelsResponse { + models: vec![previous_model_info, next_model_info], + }, + ) + .await; + let request_log = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_assistant_message("m1", "before switch"), + ev_completed_with_tokens("r1", /*total_tokens*/ 120_000), + ]), + sse(vec![ + ev_assistant_message("m3", "after switch"), + ev_completed_with_tokens("r3", /*total_tokens*/ 100), + ]), + ], + ) + .await; + let compact_request_log = mount_compact_response_sequence( + &server, + vec![ + invalid_request_response("previous-model compaction was rejected"), + wiremock::ResponseTemplate::new(/*status*/ 200) + .insert_header("content-type", "application/json") + .set_body_json(json!({ + "output": [{ + "type": "compaction", + "encrypted_content": "DOWNSHIFT_SUMMARY", + }], + })), + ], + ) + .await; + + let model_provider = openai_model_provider(&server); + let mut builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(retired_model) + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + let _ = config.features.disable(Feature::RemoteCompactionV2); + }); + let test = builder.build(&server).await.expect("build test codex"); + + test.codex + .submit(disabled_permission_user_turn( + "before switch", + test.cwd.path().to_path_buf(), + retired_model.to_string(), + )) + .await + .expect("submit first user turn"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + test.codex + .submit(disabled_permission_user_turn( + "after switch", + test.cwd.path().to_path_buf(), + next_model.to_string(), + )) + .await + .expect("submit smaller-model turn"); + assert_compaction_uses_turn_lifecycle_id(&test.codex).await; + + let requests = request_log.requests(); + let compact_requests = compact_request_log.requests(); + assert_eq!(models_mock.requests().len(), 1); + assert_eq!(requests.len(), 2); + assert_eq!(compact_requests.len(), 2); + assert_eq!( + requests[0].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!( + compact_requests[0].body_json()["model"].as_str(), + Some(retired_model) + ); + assert_eq!( + compact_requests[1].body_json()["model"].as_str(), + Some(next_model) + ); + assert_eq!(requests[1].body_json()["model"].as_str(), Some(next_model)); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn pre_sampling_compact_keeps_unknown_previous_model_for_api_key_auth_and_custom_provider() { skip_if_no_network!(); From 34aec86b38789d0536138bbd53576aafb1772fb0 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Tue, 7 Jul 2026 13:23:11 -0700 Subject: [PATCH 6/8] Record compaction fallback model in traces --- codex-rs/core/src/client.rs | 6 +- codex-rs/core/src/compact_remote.rs | 2 - codex-rs/core/src/compact_remote_v2.rs | 2 - .../core/src/compact_remote_v2_attempt.rs | 16 +++-- codex-rs/rollout-trace/src/compaction.rs | 21 +++---- codex-rs/rollout-trace/src/thread.rs | 4 -- codex-rs/rollout-trace/src/thread_tests.rs | 62 +++++++++++++++++-- 7 files changed, 82 insertions(+), 31 deletions(-) diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 0bf33b39933c..1209e1283411 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -620,7 +620,11 @@ impl ModelClient { let client = ApiCompactClient::new(transport, client_setup.api_provider, client_setup.api_auth) .with_telemetry(Some(request_telemetry)); - let trace_attempt = compaction_trace.start_attempt(&payload); + let trace_attempt = compaction_trace.start_attempt( + model.as_str(), + self.state.provider.info().name.as_str(), + &payload, + ); let result = client .compact_input( &payload, diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index af36009d4002..edd4fc5d94eb 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -196,8 +196,6 @@ async fn run_remote_compact_task_inner_impl( let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), context_compaction_item.id.as_str(), - turn_context.model_info.slug.as_str(), - turn_context.provider.info().name.as_str(), ); let compaction_item = TurnItem::ContextCompaction(context_compaction_item); sess.emit_turn_item_started(turn_context, &compaction_item) diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 7991440b8542..009daae39cb1 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -207,8 +207,6 @@ async fn run_remote_compact_task_inner_impl( let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), context_compaction_item.id.as_str(), - turn_context.model_info.slug.as_str(), - turn_context.provider.info().name.as_str(), ); let compaction_item = TurnItem::ContextCompaction(context_compaction_item); sess.emit_turn_item_started(turn_context, &compaction_item) diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index a145d00ae668..01d1e8efaf5e 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -88,12 +88,16 @@ pub(super) async fn run_remote_compact_v2_attempt( window_id, CodexResponsesRequestKind::Compaction(compaction_metadata), ); - let trace_attempt = compaction_trace.start_attempt(&serde_json::json!({ - "model": turn_context.model_info.slug.as_str(), - "instructions": prompt.base_instructions.text.as_str(), - "input": &prompt.input, - "parallel_tool_calls": prompt.parallel_tool_calls, - })); + let trace_attempt = compaction_trace.start_attempt( + turn_context.model_info.slug.as_str(), + turn_context.provider.info().name.as_str(), + &serde_json::json!({ + "model": turn_context.model_info.slug.as_str(), + "instructions": prompt.base_instructions.text.as_str(), + "input": &prompt.input, + "parallel_tool_calls": prompt.parallel_tool_calls, + }), + ); let mut owned_client_session = None; let client_session = match client_session { Some(client_session) => client_session, diff --git a/codex-rs/rollout-trace/src/compaction.rs b/codex-rs/rollout-trace/src/compaction.rs index 5dffad2462d7..8dd72c5096f2 100644 --- a/codex-rs/rollout-trace/src/compaction.rs +++ b/codex-rs/rollout-trace/src/compaction.rs @@ -48,8 +48,6 @@ struct EnabledCompactionTraceContext { thread_id: AgentThreadId, codex_turn_id: CodexTurnId, compaction_id: CompactionId, - model: String, - provider_name: String, } /// One upstream request attempt made while computing a compaction checkpoint. @@ -100,8 +98,6 @@ impl CompactionTraceContext { thread_id: AgentThreadId, codex_turn_id: CodexTurnId, compaction_id: CompactionId, - model: String, - provider_name: String, ) -> Self { Self { state: CompactionTraceContextState::Enabled(EnabledCompactionTraceContext { @@ -109,14 +105,17 @@ impl CompactionTraceContext { thread_id, codex_turn_id, compaction_id, - model, - provider_name, }), } } /// Starts a new upstream attempt and records the exact compact endpoint request. - pub fn start_attempt(&self, request: &impl Serialize) -> CompactionTraceAttempt { + pub fn start_attempt( + &self, + model: &str, + provider_name: &str, + request: &impl Serialize, + ) -> CompactionTraceAttempt { let CompactionTraceContextState::Enabled(context) = &self.state else { return CompactionTraceAttempt::disabled(); }; @@ -127,7 +126,7 @@ impl CompactionTraceContext { compaction_request_id: next_compaction_request_id(), }), }; - attempt.record_started(request); + attempt.record_started(model, provider_name, request); attempt } @@ -174,7 +173,7 @@ impl CompactionTraceAttempt { } } - fn record_started(&self, request: &impl Serialize) { + fn record_started(&self, model: &str, provider_name: &str, request: &impl Serialize) { let CompactionTraceAttemptState::Enabled(attempt) = &self.state else { return; }; @@ -193,8 +192,8 @@ impl CompactionTraceAttempt { compaction_request_id: attempt.compaction_request_id.clone(), thread_id: attempt.context.thread_id.clone(), codex_turn_id: attempt.context.codex_turn_id.clone(), - model: attempt.context.model.clone(), - provider_name: attempt.context.provider_name.clone(), + model: model.to_string(), + provider_name: provider_name.to_string(), request_payload, }, ); diff --git a/codex-rs/rollout-trace/src/thread.rs b/codex-rs/rollout-trace/src/thread.rs index 514b8d382bc0..c874c043844a 100644 --- a/codex-rs/rollout-trace/src/thread.rs +++ b/codex-rs/rollout-trace/src/thread.rs @@ -382,8 +382,6 @@ impl ThreadTraceContext { &self, codex_turn_id: impl Into, compaction_id: impl Into, - model: impl Into, - provider_name: impl Into, ) -> CompactionTraceContext { let ThreadTraceContextState::Enabled(context) = &self.state else { return CompactionTraceContext::disabled(); @@ -393,8 +391,6 @@ impl ThreadTraceContext { context.thread_id.clone(), codex_turn_id.into(), compaction_id.into(), - model.into(), - provider_name.into(), ) } diff --git a/codex-rs/rollout-trace/src/thread_tests.rs b/codex-rs/rollout-trace/src/thread_tests.rs index 6421b29af3cb..33f2784d9f82 100644 --- a/codex-rs/rollout-trace/src/thread_tests.rs +++ b/codex-rs/rollout-trace/src/thread_tests.rs @@ -136,14 +136,12 @@ fn disabled_thread_context_accepts_trace_calls_without_writing() -> anyhow::Resu inference_attempt.record_completed("response-1", Some("req-1"), &token_usage, &[]); inference_attempt.record_failed("inference failed", /*upstream_request_id*/ None, &[]); - let compaction_trace = thread_trace.compaction_trace_context( - "turn-1", - "compaction-1", + let compaction_trace = thread_trace.compaction_trace_context("turn-1", "compaction-1"); + let compaction_attempt = compaction_trace.start_attempt( "gpt-test", "test-provider", + &serde_json::json!({ "kind": "compaction" }), ); - let compaction_attempt = - compaction_trace.start_attempt(&serde_json::json!({ "kind": "compaction" })); compaction_attempt.record_completed(&[]); compaction_attempt.record_failed("compaction failed"); compaction_trace.record_installed(&CompactionCheckpointTracePayload { @@ -164,6 +162,60 @@ fn disabled_thread_context_accepts_trace_calls_without_writing() -> anyhow::Resu Ok(()) } +#[test] +fn compaction_attempts_record_request_scoped_models() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let thread_id = ThreadId::new(); + let thread_trace = + ThreadTraceContext::start_root_in_root_for_test(temp.path(), minimal_metadata(thread_id))?; + thread_trace.record_codex_turn_started("turn-1"); + + let compaction_trace = thread_trace.compaction_trace_context("turn-1", "compaction-1"); + let previous_model_attempt = compaction_trace.start_attempt( + "gpt-previous", + "test-provider", + &serde_json::json!({ "model": "gpt-previous" }), + ); + previous_model_attempt.record_failed("previous model unavailable"); + let selected_model_attempt = compaction_trace.start_attempt( + "gpt-selected", + "test-provider", + &serde_json::json!({ "model": "gpt-selected" }), + ); + selected_model_attempt.record_completed(&[]); + + let replayed = replay_bundle(&single_bundle_dir(temp.path())?)?; + let mut attempts = replayed + .compaction_requests + .values() + .map(|attempt| { + ( + attempt.model.clone(), + attempt.provider_name.clone(), + attempt.compaction_id.clone(), + ) + }) + .collect::>(); + attempts.sort(); + assert_eq!( + attempts, + vec![ + ( + "gpt-previous".to_string(), + "test-provider".to_string(), + "compaction-1".to_string(), + ), + ( + "gpt-selected".to_string(), + "test-provider".to_string(), + "compaction-1".to_string(), + ), + ] + ); + + Ok(()) +} + #[test] fn protocol_wrapper_records_selected_events_as_raw_payloads() -> anyhow::Result<()> { let temp = TempDir::new()?; From d2b96df5dc08a80f6c169a9f6bdd164e5f742965 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Tue, 7 Jul 2026 13:55:48 -0700 Subject: [PATCH 7/8] Revert "Record compaction fallback model in traces" This reverts commit 34aec86b38789d0536138bbd53576aafb1772fb0. --- codex-rs/core/src/client.rs | 6 +- codex-rs/core/src/compact_remote.rs | 2 + codex-rs/core/src/compact_remote_v2.rs | 2 + .../core/src/compact_remote_v2_attempt.rs | 16 ++--- codex-rs/rollout-trace/src/compaction.rs | 21 ++++--- codex-rs/rollout-trace/src/thread.rs | 4 ++ codex-rs/rollout-trace/src/thread_tests.rs | 62 ++----------------- 7 files changed, 31 insertions(+), 82 deletions(-) diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 1209e1283411..0bf33b39933c 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -620,11 +620,7 @@ impl ModelClient { let client = ApiCompactClient::new(transport, client_setup.api_provider, client_setup.api_auth) .with_telemetry(Some(request_telemetry)); - let trace_attempt = compaction_trace.start_attempt( - model.as_str(), - self.state.provider.info().name.as_str(), - &payload, - ); + let trace_attempt = compaction_trace.start_attempt(&payload); let result = client .compact_input( &payload, diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index edd4fc5d94eb..af36009d4002 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -196,6 +196,8 @@ async fn run_remote_compact_task_inner_impl( let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), context_compaction_item.id.as_str(), + turn_context.model_info.slug.as_str(), + turn_context.provider.info().name.as_str(), ); let compaction_item = TurnItem::ContextCompaction(context_compaction_item); sess.emit_turn_item_started(turn_context, &compaction_item) diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 009daae39cb1..7991440b8542 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -207,6 +207,8 @@ async fn run_remote_compact_task_inner_impl( let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), context_compaction_item.id.as_str(), + turn_context.model_info.slug.as_str(), + turn_context.provider.info().name.as_str(), ); let compaction_item = TurnItem::ContextCompaction(context_compaction_item); sess.emit_turn_item_started(turn_context, &compaction_item) diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 01d1e8efaf5e..a145d00ae668 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -88,16 +88,12 @@ pub(super) async fn run_remote_compact_v2_attempt( window_id, CodexResponsesRequestKind::Compaction(compaction_metadata), ); - let trace_attempt = compaction_trace.start_attempt( - turn_context.model_info.slug.as_str(), - turn_context.provider.info().name.as_str(), - &serde_json::json!({ - "model": turn_context.model_info.slug.as_str(), - "instructions": prompt.base_instructions.text.as_str(), - "input": &prompt.input, - "parallel_tool_calls": prompt.parallel_tool_calls, - }), - ); + let trace_attempt = compaction_trace.start_attempt(&serde_json::json!({ + "model": turn_context.model_info.slug.as_str(), + "instructions": prompt.base_instructions.text.as_str(), + "input": &prompt.input, + "parallel_tool_calls": prompt.parallel_tool_calls, + })); let mut owned_client_session = None; let client_session = match client_session { Some(client_session) => client_session, diff --git a/codex-rs/rollout-trace/src/compaction.rs b/codex-rs/rollout-trace/src/compaction.rs index 8dd72c5096f2..5dffad2462d7 100644 --- a/codex-rs/rollout-trace/src/compaction.rs +++ b/codex-rs/rollout-trace/src/compaction.rs @@ -48,6 +48,8 @@ struct EnabledCompactionTraceContext { thread_id: AgentThreadId, codex_turn_id: CodexTurnId, compaction_id: CompactionId, + model: String, + provider_name: String, } /// One upstream request attempt made while computing a compaction checkpoint. @@ -98,6 +100,8 @@ impl CompactionTraceContext { thread_id: AgentThreadId, codex_turn_id: CodexTurnId, compaction_id: CompactionId, + model: String, + provider_name: String, ) -> Self { Self { state: CompactionTraceContextState::Enabled(EnabledCompactionTraceContext { @@ -105,17 +109,14 @@ impl CompactionTraceContext { thread_id, codex_turn_id, compaction_id, + model, + provider_name, }), } } /// Starts a new upstream attempt and records the exact compact endpoint request. - pub fn start_attempt( - &self, - model: &str, - provider_name: &str, - request: &impl Serialize, - ) -> CompactionTraceAttempt { + pub fn start_attempt(&self, request: &impl Serialize) -> CompactionTraceAttempt { let CompactionTraceContextState::Enabled(context) = &self.state else { return CompactionTraceAttempt::disabled(); }; @@ -126,7 +127,7 @@ impl CompactionTraceContext { compaction_request_id: next_compaction_request_id(), }), }; - attempt.record_started(model, provider_name, request); + attempt.record_started(request); attempt } @@ -173,7 +174,7 @@ impl CompactionTraceAttempt { } } - fn record_started(&self, model: &str, provider_name: &str, request: &impl Serialize) { + fn record_started(&self, request: &impl Serialize) { let CompactionTraceAttemptState::Enabled(attempt) = &self.state else { return; }; @@ -192,8 +193,8 @@ impl CompactionTraceAttempt { compaction_request_id: attempt.compaction_request_id.clone(), thread_id: attempt.context.thread_id.clone(), codex_turn_id: attempt.context.codex_turn_id.clone(), - model: model.to_string(), - provider_name: provider_name.to_string(), + model: attempt.context.model.clone(), + provider_name: attempt.context.provider_name.clone(), request_payload, }, ); diff --git a/codex-rs/rollout-trace/src/thread.rs b/codex-rs/rollout-trace/src/thread.rs index c874c043844a..514b8d382bc0 100644 --- a/codex-rs/rollout-trace/src/thread.rs +++ b/codex-rs/rollout-trace/src/thread.rs @@ -382,6 +382,8 @@ impl ThreadTraceContext { &self, codex_turn_id: impl Into, compaction_id: impl Into, + model: impl Into, + provider_name: impl Into, ) -> CompactionTraceContext { let ThreadTraceContextState::Enabled(context) = &self.state else { return CompactionTraceContext::disabled(); @@ -391,6 +393,8 @@ impl ThreadTraceContext { context.thread_id.clone(), codex_turn_id.into(), compaction_id.into(), + model.into(), + provider_name.into(), ) } diff --git a/codex-rs/rollout-trace/src/thread_tests.rs b/codex-rs/rollout-trace/src/thread_tests.rs index 33f2784d9f82..6421b29af3cb 100644 --- a/codex-rs/rollout-trace/src/thread_tests.rs +++ b/codex-rs/rollout-trace/src/thread_tests.rs @@ -136,12 +136,14 @@ fn disabled_thread_context_accepts_trace_calls_without_writing() -> anyhow::Resu inference_attempt.record_completed("response-1", Some("req-1"), &token_usage, &[]); inference_attempt.record_failed("inference failed", /*upstream_request_id*/ None, &[]); - let compaction_trace = thread_trace.compaction_trace_context("turn-1", "compaction-1"); - let compaction_attempt = compaction_trace.start_attempt( + let compaction_trace = thread_trace.compaction_trace_context( + "turn-1", + "compaction-1", "gpt-test", "test-provider", - &serde_json::json!({ "kind": "compaction" }), ); + let compaction_attempt = + compaction_trace.start_attempt(&serde_json::json!({ "kind": "compaction" })); compaction_attempt.record_completed(&[]); compaction_attempt.record_failed("compaction failed"); compaction_trace.record_installed(&CompactionCheckpointTracePayload { @@ -162,60 +164,6 @@ fn disabled_thread_context_accepts_trace_calls_without_writing() -> anyhow::Resu Ok(()) } -#[test] -fn compaction_attempts_record_request_scoped_models() -> anyhow::Result<()> { - let temp = TempDir::new()?; - let thread_id = ThreadId::new(); - let thread_trace = - ThreadTraceContext::start_root_in_root_for_test(temp.path(), minimal_metadata(thread_id))?; - thread_trace.record_codex_turn_started("turn-1"); - - let compaction_trace = thread_trace.compaction_trace_context("turn-1", "compaction-1"); - let previous_model_attempt = compaction_trace.start_attempt( - "gpt-previous", - "test-provider", - &serde_json::json!({ "model": "gpt-previous" }), - ); - previous_model_attempt.record_failed("previous model unavailable"); - let selected_model_attempt = compaction_trace.start_attempt( - "gpt-selected", - "test-provider", - &serde_json::json!({ "model": "gpt-selected" }), - ); - selected_model_attempt.record_completed(&[]); - - let replayed = replay_bundle(&single_bundle_dir(temp.path())?)?; - let mut attempts = replayed - .compaction_requests - .values() - .map(|attempt| { - ( - attempt.model.clone(), - attempt.provider_name.clone(), - attempt.compaction_id.clone(), - ) - }) - .collect::>(); - attempts.sort(); - assert_eq!( - attempts, - vec![ - ( - "gpt-previous".to_string(), - "test-provider".to_string(), - "compaction-1".to_string(), - ), - ( - "gpt-selected".to_string(), - "test-provider".to_string(), - "compaction-1".to_string(), - ), - ] - ); - - Ok(()) -} - #[test] fn protocol_wrapper_records_selected_events_as_raw_payloads() -> anyhow::Result<()> { let temp = TempDir::new()?; From 818642630e3164f8bb86d9e29bef428368a374a4 Mon Sep 17 00:00:00 2001 From: celia-oai Date: Tue, 7 Jul 2026 13:57:47 -0700 Subject: [PATCH 8/8] Trace compaction fallback with selected model --- codex-rs/core/src/compact_remote.rs | 17 ++++++++--- codex-rs/core/src/compact_remote_v2.rs | 17 ++++++++--- codex-rs/rollout-trace/src/thread_tests.rs | 34 ++++++++++++++++++++++ 3 files changed, 60 insertions(+), 8 deletions(-) diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index af36009d4002..4b1db9355544 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -191,11 +191,12 @@ async fn run_remote_compact_task_inner_impl( ) -> CodexResult<()> { let turn_context = &step_context.turn; let context_compaction_item = ContextCompactionItem::new(); + let compaction_id = context_compaction_item.id.clone(); // Use the UI compaction item ID as the trace compaction ID so protocol lifecycle events, // endpoint attempts, and the installed history checkpoint all have one join key. let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), - context_compaction_item.id.as_str(), + compaction_id.as_str(), turn_context.model_info.slug.as_str(), turn_context.provider.info().name.as_str(), ); @@ -220,11 +221,19 @@ async fn run_remote_compact_task_inner_impl( if !matches!(&error, CodexErr::InvalidRequest(_)) { return Err(error); } + let fallback_turn_context = &fallback_step_context.turn; + let fallback_compaction_trace = + sess.services.rollout_thread_trace.compaction_trace_context( + fallback_turn_context.sub_id.as_str(), + compaction_id.as_str(), + fallback_turn_context.model_info.slug.as_str(), + fallback_turn_context.provider.info().name.as_str(), + ); let fallback_result = run_remote_compact_attempt( sess, fallback_step_context, turn_state, - &compaction_trace, + &fallback_compaction_trace, compaction_metadata, analytics_details, ) @@ -232,13 +241,13 @@ async fn run_remote_compact_task_inner_impl( record_model_fallback( &sess.services.session_telemetry, turn_context.model_info.slug.as_str(), - fallback_step_context.turn.model_info.slug.as_str(), + fallback_turn_context.model_info.slug.as_str(), compaction_metadata.reason(), compaction_metadata.implementation(), fallback_result.as_ref().err(), ); match fallback_result { - Ok(attempt) => (attempt, &fallback_step_context.turn), + Ok(attempt) => (attempt, fallback_turn_context), Err(_) => return Err(error), } } diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 7991440b8542..1571602cc140 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -204,9 +204,10 @@ async fn run_remote_compact_task_inner_impl( ) -> CodexResult<()> { let turn_context = &step_context.turn; let context_compaction_item = ContextCompactionItem::new(); + let compaction_id = context_compaction_item.id.clone(); let compaction_trace = sess.services.rollout_thread_trace.compaction_trace_context( turn_context.sub_id.as_str(), - context_compaction_item.id.as_str(), + compaction_id.as_str(), turn_context.model_info.slug.as_str(), turn_context.provider.info().name.as_str(), ); @@ -232,11 +233,19 @@ async fn run_remote_compact_task_inner_impl( if !matches!(&error, CodexErr::InvalidRequest(_)) { return Err(error); } + let fallback_turn_context = &fallback_step_context.turn; + let fallback_compaction_trace = + sess.services.rollout_thread_trace.compaction_trace_context( + fallback_turn_context.sub_id.as_str(), + compaction_id.as_str(), + fallback_turn_context.model_info.slug.as_str(), + fallback_turn_context.provider.info().name.as_str(), + ); let fallback_result = run_remote_compact_v2_attempt( sess, fallback_step_context, client_session, - &compaction_trace, + &fallback_compaction_trace, compaction_metadata, analytics_details, ) @@ -244,13 +253,13 @@ async fn run_remote_compact_task_inner_impl( record_model_fallback( &sess.services.session_telemetry, turn_context.model_info.slug.as_str(), - fallback_step_context.turn.model_info.slug.as_str(), + fallback_turn_context.model_info.slug.as_str(), compaction_metadata.reason(), compaction_metadata.implementation(), fallback_result.as_ref().err(), ); match fallback_result { - Ok(attempt) => (attempt, &fallback_step_context.turn), + Ok(attempt) => (attempt, fallback_turn_context), Err(_) => return Err(error), } } diff --git a/codex-rs/rollout-trace/src/thread_tests.rs b/codex-rs/rollout-trace/src/thread_tests.rs index 6421b29af3cb..6a41e0beebc5 100644 --- a/codex-rs/rollout-trace/src/thread_tests.rs +++ b/codex-rs/rollout-trace/src/thread_tests.rs @@ -164,6 +164,40 @@ fn disabled_thread_context_accepts_trace_calls_without_writing() -> anyhow::Resu Ok(()) } +#[test] +fn compaction_contexts_share_identity_across_models() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let thread_id = ThreadId::new(); + let thread_trace = + ThreadTraceContext::start_root_in_root_for_test(temp.path(), minimal_metadata(thread_id))?; + thread_trace.record_codex_turn_started("turn-1"); + + for model in ["gpt-previous", "gpt-selected"] { + let compaction_trace = + thread_trace.compaction_trace_context("turn-1", "compaction-1", model, "test-provider"); + compaction_trace + .start_attempt(&serde_json::json!({ "model": model })) + .record_failed("test failure"); + } + + let replayed = replay_bundle(&single_bundle_dir(temp.path())?)?; + let mut attempts = replayed + .compaction_requests + .values() + .map(|attempt| (attempt.model.clone(), attempt.compaction_id.clone())) + .collect::>(); + attempts.sort(); + assert_eq!( + attempts, + vec![ + ("gpt-previous".to_string(), "compaction-1".to_string()), + ("gpt-selected".to_string(), "compaction-1".to_string()), + ] + ); + + Ok(()) +} + #[test] fn protocol_wrapper_records_selected_events_as_raw_payloads() -> anyhow::Result<()> { let temp = TempDir::new()?;