github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 00:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: retry rejected previous-model compaction with selected model (#30319)

Browse Source 
## Why

Pre-sampling compaction intentionally uses the previous turn's model
when the compaction compatibility hash changes or when switching to a
model with a smaller context window. This keeps compaction aligned with
the settings that produced the history, but it can block the next turn
when a resumed ChatGPT thread still references a model slug that has
since been retired. The Codex backend rejects that compaction request
before the user's currently selected model gets a chance to sample.

This PR lets those threads recover without changing previous-model
compaction behavior for API-key authentication or custom providers. It
is stacked on #31316, which is a behavior-preserving extraction of the
individual remote compaction attempts; this PR contains the fallback
behavior.

## What changed

- For automatic previous-model compaction, capture the selected model's
request context when using ChatGPT authentication with the OpenAI
provider and the selected model differs from the previous model.
- If the previous-model attempt returns an `InvalidRequest`, retry
compaction once with the selected model for both `/responses/compact`
and Responses Compaction V2.
- Complete history processing, lifecycle events, and token accounting
with the context of the model that successfully compacted the thread.
- If the fallback also fails, return the original previous-model error
so the retry does not change the user-visible failure.
- Record fallback attempts with reason, implementation, and outcome
telemetry.
- Leave API-key authentication, custom providers, same-model turns, and
non-`InvalidRequest` failures on their existing paths.

## Testing

- `just test -p codex-core -E 'test(pre_sampling_compact) |
test(model_unavailable_error)'` (10 tests)
- Added integration coverage for a resumed thread whose model was
renamed, a model downshift using Responses Compaction V2, and API-key
authentication with a custom provider.
This commit is contained in:
Celia Chen
2026-07-07 14:26:03 -07:00
committed by GitHub
parent 6b4882528e
commit 172ab264bd
9 changed files with 797 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/compact_model_fallback.rs
+49
View File
@@ -0,0 +1,49 @@
use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionReason;
use codex_otel::SessionTelemetry;
use codex_protocol::error::CodexErr;
use tracing::warn;
pub(crate) fn record_model_fallback(
session_telemetry: &SessionTelemetry,
previous_model: &str,
current_model: &str,
reason: CompactionReason,
implementation: CompactionImplementation,
fallback_error: Option<&CodexErr>,
) {
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 fallback_error.is_none() {
"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,
?fallback_error,
"previous-model compaction failed; retried with current model"
);
}
codex-rs/core/src/compact_remote.rs
+81 -28
View File
@@ -7,6 +7,7 @@ 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::record_model_fallback;
use crate::context::world_state::WorldState;
use crate::context_manager::ContextManager;
use crate::hook_runtime::PostCompactHookOutcome;
@@ -45,19 +46,25 @@ const CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE: &str =
pub(crate) async fn run_inline_remote_auto_compact_task(
sess: Arc<Session>,
step_context: Arc<StepContext>,
fallback_step_context: Option<Arc<StepContext>>,
turn_state: Arc<OnceLock<String>>,
initial_context_injection: InitialContextInjection,
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(())
@@ -78,14 +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(())
@@ -94,19 +106,16 @@ pub(crate) async fn run_remote_compact_task(
async fn run_remote_compact_task_inner(
sess: &Arc<Session>,
step_context: &Arc<StepContext>,
fallback_step_context: Option<&Arc<StepContext>>,
turn_state: Option<Arc<OnceLock<String>>>,
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()
@@ -116,7 +125,7 @@ async fn run_remote_compact_task_inner(
turn_context.as_ref(),
trigger,
reason,
CompactionImplementation::ResponsesCompact,
implementation,
phase,
)
.await;
@@ -139,6 +148,7 @@ 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,
@@ -173,6 +183,7 @@ async fn run_remote_compact_task_inner(
async fn run_remote_compact_task_inner_impl(
sess: &Arc<Session>,
step_context: &Arc<StepContext>,
fallback_step_context: Option<&Arc<StepContext>>,
turn_state: Option<Arc<OnceLock<String>>>,
initial_context_injection: InitialContextInjection,
compaction_metadata: CompactionTurnMetadata,
@@ -180,33 +191,75 @@ 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(),
);
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 !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,
&fallback_compaction_trace,
compaction_metadata,
analytics_details,
)
.await;
record_model_fallback(
&sess.services.session_telemetry,
turn_context.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_turn_context),
Err(_) => return Err(error),
}
}
};
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 +268,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 +287,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(())
}
codex-rs/core/src/compact_remote_v2.rs
+85 -32
View File
@@ -8,6 +8,7 @@ use crate::compact::CompactionAnalyticsAttempt;
use crate::compact::CompactionAnalyticsDetails;
use crate::compact::InitialContextInjection;
use crate::compact::compaction_status_from_result;
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,19 +58,25 @@ const MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES: u64 = 2;
pub(crate) async fn run_inline_remote_auto_compact_task(
sess: Arc<Session>,
step_context: Arc<StepContext>,
fallback_step_context: Option<Arc<StepContext>>,
client_session: &mut ModelClientSession,
initial_context_injection: InitialContextInjection,
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
}
@@ -89,14 +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
}
@@ -104,19 +116,16 @@ pub(crate) async fn run_remote_compact_task(
async fn run_remote_compact_task_inner(
sess: &Arc<Session>,
step_context: &Arc<StepContext>,
fallback_step_context: Option<&Arc<StepContext>>,
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()
@@ -126,7 +135,7 @@ async fn run_remote_compact_task_inner(
turn_context.as_ref(),
trigger,
reason,
CompactionImplementation::ResponsesCompactionV2,
implementation,
phase,
)
.await;
@@ -149,6 +158,7 @@ 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,
@@ -186,16 +196,18 @@ async fn run_remote_compact_task_inner(
async fn run_remote_compact_task_inner_impl(
sess: &Arc<Session>,
step_context: &Arc<StepContext>,
client_session: Option<&mut ModelClientSession>,
fallback_step_context: Option<&Arc<StepContext>>,
mut client_session: Option<&mut ModelClientSession>,
initial_context_injection: InitialContextInjection,
compaction_metadata: CompactionTurnMetadata,
analytics_details: &mut CompactionAnalyticsDetails,
) -> 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(),
);
@@ -203,21 +215,62 @@ async fn run_remote_compact_task_inner_impl(
sess.emit_turn_item_started(turn_context, &compaction_item)
.await;
let attempt = run_remote_compact_v2_attempt(
sess,
step_context,
client_session.as_deref_mut(),
&compaction_trace,
compaction_metadata,
analytics_details,
)
.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 !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,
&fallback_compaction_trace,
compaction_metadata,
analytics_details,
)
.await;
record_model_fallback(
&sess.services.session_telemetry,
turn_context.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_turn_context),
Err(_) => return Err(error),
}
}
};
let RemoteCompactV2Attempt {
trace_input_history,
prompt_input,
compaction_output,
token_usage,
owned_client_session: _owned_client_session,
} = run_remote_compact_v2_attempt(
sess,
step_context,
client_session,
&compaction_trace,
compaction_metadata,
analytics_details,
)
.await?;
} = 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 +283,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 +292,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 +308,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(())
}
codex-rs/core/src/lib.rs
+1
View File
@@ -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;
codex-rs/core/src/responses_metadata.rs
+16
View File
@@ -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)]
codex-rs/core/src/session/turn.rs
+43 -3
View File
@@ -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,28 @@ fn comp_hash_changed(previous: Option<&str>, current: Option<&str>) -> bool {
.is_some_and(|(previous, current)| previous != current)
}
/// 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<Session>,
turn_context: &Arc<TurnContext>,
previous_model: &str,
) -> Option<Arc<StepContext>> {
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 +868,27 @@ 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 = capture_current_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 +923,19 @@ 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 = capture_current_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 +954,7 @@ async fn maybe_run_previous_model_inline_compact(
async fn run_auto_compact(
sess: &Arc<Session>,
step_context: Arc<StepContext>,
fallback_step_context: Option<Arc<StepContext>>,
client_session: &mut ModelClientSession,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
@@ -949,6 +987,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 +1004,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,
codex-rs/core/tests/common/responses.rs
+39
View File
@@ -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<ResponseTemplate>,
) -> ResponseMock {
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
struct SeqResponder {
num_calls: AtomicUsize,
responses: Vec<ResponseTemplate>,
}
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`.
codex-rs/core/tests/suite/compact.rs
+449
View File
@@ -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;
@@ -274,6 +275,20 @@ 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 invalid_request_response(message: impl Into<String>) -> wiremock::ResponseTemplate {
wiremock::ResponseTemplate::new(/*status*/ 400).set_body_json(json!({
"detail": message.into(),
}))
}
fn write_global_file(
home: &TempDir,
filename: &str,
@@ -421,6 +436,7 @@ async fn assert_compaction_uses_turn_lifecycle_id(codex: &std::sync::Arc<codex_c
item: TurnItem::ContextCompaction(_),
..
}) => 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 +2292,439 @@ 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.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"));
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),
])),
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",
"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_after_previous_model_invalid_request_on_downshift() {
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_response_sequence(
&server,
vec![
sse_response(sse(vec![
ev_assistant_message("m1", "before switch"),
ev_completed_with_tokens("r1", /*total_tokens*/ 120_000),
])),
invalid_request_response("previous-model compaction was rejected"),
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_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!();
let server = MockServer::start().await;
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"));
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!();
codex-rs/rollout-trace/src/thread_tests.rs
+34
View File
@@ -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::<Vec<_>>();
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()?;