From 15e79f3c2696672d292c62fa7823e98cd155462a Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Mon, 11 May 2026 19:16:49 +0300 Subject: [PATCH] [codex] Harden overflow auto-compaction recovery (#22141) ## Why Dogfooder feedback exposed two correctness gaps in normal-loop overflow recovery: 1. a sampling request that hit `ContextWindowExceeded` could keep re-entering auto-compaction indefinitely if the compacted retry still did not fit, and 2. local compact-history rebuilds flattened user messages down to text, so an overflowing `[image, "what is this?"]` turn could be retried without the image after compaction. That means recovery could either fail to terminate cleanly or proceed with a materially weakened version of the user request. ## What changed - Move normal-loop `ContextWindowExceeded` handling into the sampling retry loop, so successful rescue compaction consumes the provider retry budget instead of creating an unbounded outer-turn loop. - Keep compacted user-history rebuilds structured: `collect_user_messages` now carries user `UserInput` content rather than flattened strings, and `build_compacted_history` reconstructs full user messages from that structured representation. - Preserve image inputs while retaining the existing text-budget truncation behavior for compacted user history. - Preserve existing compaction-task failure handling and client-session reset behavior while bounding repeated overflow retries. - Add focused regression coverage for: - recovery after a normal-loop overflow, - retry-budget exhaustion after repeated overflow, - local recovery preserving image + text input, - remote recovery preserving image + text input, - remote compaction v2 preserving image + text input, and - compaction failure still terminating cleanly. The main behavior changes are in `codex-rs/core/src/session/turn.rs` and `codex-rs/core/src/compact.rs`. ## Verification - Not run locally; relying on PR CI for this update. --------- Co-authored-by: Codex --- codex-rs/core/src/compact.rs | 62 +++- codex-rs/core/src/compact_tests.rs | 90 +++++- codex-rs/core/src/session/turn.rs | 33 +- codex-rs/core/tests/suite/client.rs | 74 +++-- codex-rs/core/tests/suite/compact.rs | 88 ++++++ codex-rs/core/tests/suite/compact_remote.rs | 329 ++++++++++++++++++++ 6 files changed, 619 insertions(+), 57 deletions(-) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 9d1c82eb6..d0fac9cd1 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -386,7 +386,7 @@ pub fn content_items_to_text(content: &[ContentItem]) -> Option { } } -pub(crate) fn collect_user_messages(items: &[ResponseItem]) -> Vec { +pub(crate) fn collect_user_messages(items: &[ResponseItem]) -> Vec> { items .iter() .filter_map(|item| match crate::event_mapping::parse_turn_item(item) { @@ -394,7 +394,7 @@ pub(crate) fn collect_user_messages(items: &[ResponseItem]) -> Vec { if is_summary_message(&user.message()) { None } else { - Some(user.message()) + Some(user.content) } } _ => None, @@ -465,7 +465,7 @@ pub(crate) fn insert_initial_context_before_last_real_user_or_summary( pub(crate) fn build_compacted_history( initial_context: Vec, - user_messages: &[String], + user_messages: &[Vec], summary_text: &str, ) -> Vec { build_compacted_history_with_limit( @@ -478,23 +478,24 @@ pub(crate) fn build_compacted_history( fn build_compacted_history_with_limit( mut history: Vec, - user_messages: &[String], + user_messages: &[Vec], summary_text: &str, max_tokens: usize, ) -> Vec { - let mut selected_messages: Vec = Vec::new(); + let mut selected_messages: Vec> = Vec::new(); if max_tokens > 0 { let mut remaining = max_tokens; for message in user_messages.iter().rev() { if remaining == 0 { break; } - let tokens = approx_token_count(message); + let message_text = user_message_text(message); + let tokens = approx_token_count(&message_text); if tokens <= remaining { selected_messages.push(message.clone()); remaining = remaining.saturating_sub(tokens); } else { - let truncated = truncate_text(message, TruncationPolicy::Tokens(remaining)); + let truncated = truncate_user_message(message, remaining); selected_messages.push(truncated); break; } @@ -502,15 +503,8 @@ fn build_compacted_history_with_limit( selected_messages.reverse(); } - for message in &selected_messages { - history.push(ResponseItem::Message { - id: None, - role: "user".to_string(), - content: vec![ContentItem::InputText { - text: message.clone(), - }], - phase: None, - }); + for message in selected_messages { + history.push(ResponseItem::from(ResponseInputItem::from(message))); } let summary_text = if summary_text.is_empty() { @@ -529,6 +523,42 @@ fn build_compacted_history_with_limit( history } +pub(crate) fn user_message_text(message: &[UserInput]) -> String { + message + .iter() + .filter_map(|item| match item { + UserInput::Text { text, .. } => Some(text.as_str()), + _ => None, + }) + .collect::>() + .join("") +} + +fn truncate_user_message(message: &[UserInput], remaining_tokens: usize) -> Vec { + let mut remaining_tokens = remaining_tokens; + let mut truncated = Vec::with_capacity(message.len()); + for item in message { + match item { + UserInput::Text { text, .. } if remaining_tokens > 0 => { + let token_count = approx_token_count(text); + if token_count <= remaining_tokens { + truncated.push(item.clone()); + remaining_tokens = remaining_tokens.saturating_sub(token_count); + } else { + truncated.push(UserInput::Text { + text: truncate_text(text, TruncationPolicy::Tokens(remaining_tokens)), + text_elements: Vec::new(), + }); + remaining_tokens = 0; + } + } + UserInput::Text { .. } => {} + _ => truncated.push(item.clone()), + } + } + truncated +} + async fn drain_to_completed( sess: &Session, turn_context: &TurnContext, diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index def82b129..786fb412a 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/codex-rs/core/src/compact_tests.rs @@ -55,7 +55,7 @@ fn content_items_to_text_ignores_image_only_content() { } #[test] -fn collect_user_messages_extracts_user_text_only() { +fn collect_user_messages_preserves_user_content() { let items = vec![ ResponseItem::Message { id: Some("assistant".to_string()), @@ -68,9 +68,15 @@ fn collect_user_messages_extracts_user_text_only() { ResponseItem::Message { id: Some("user".to_string()), role: "user".to_string(), - content: vec![ContentItem::InputText { - text: "first".to_string(), - }], + content: vec![ + ContentItem::InputImage { + image_url: "file://image.png".to_string(), + detail: Some(DEFAULT_IMAGE_DETAIL), + }, + ContentItem::InputText { + text: "first".to_string(), + }, + ], phase: None, }, ResponseItem::Other, @@ -78,7 +84,18 @@ fn collect_user_messages_extracts_user_text_only() { let collected = collect_user_messages(&items); - assert_eq!(vec!["first".to_string()], collected); + assert_eq!( + vec![vec![ + UserInput::Image { + image_url: "file://image.png".to_string(), + }, + UserInput::Text { + text: "first".to_string(), + text_elements: Vec::new(), + }, + ]], + collected + ); } #[test] @@ -117,7 +134,13 @@ do things let collected = collect_user_messages(&items); - assert_eq!(vec!["real user message".to_string()], collected); + assert_eq!( + vec![vec![UserInput::Text { + text: "real user message".to_string(), + text_elements: Vec::new(), + }]], + collected + ); } #[test] @@ -126,9 +149,13 @@ fn build_token_limited_compacted_history_truncates_overlong_user_messages() { // that oversized user content is truncated. let max_tokens = 16; let big = "word ".repeat(200); + let user_message = vec![UserInput::Text { + text: big.clone(), + text_elements: Vec::new(), + }]; let history = super::build_compacted_history_with_limit( Vec::new(), - std::slice::from_ref(&big), + std::slice::from_ref(&user_message), "SUMMARY", max_tokens, ); @@ -162,10 +189,57 @@ fn build_token_limited_compacted_history_truncates_overlong_user_messages() { assert_eq!(summary_text, "SUMMARY"); } +#[test] +fn truncate_user_message_preserves_text_segment_order_around_images() { + let before_image = "before ".repeat(8); + let after_image = "after ".repeat(200); + let before_image_tokens = approx_token_count(&before_image); + let after_image_token_budget = 16; + let remaining_tokens = before_image_tokens + after_image_token_budget; + let user_message = vec![ + UserInput::Text { + text: before_image.clone(), + text_elements: Vec::new(), + }, + UserInput::Image { + image_url: "file://image.png".to_string(), + }, + UserInput::Text { + text: after_image.clone(), + text_elements: Vec::new(), + }, + ]; + + let truncated = super::truncate_user_message(&user_message, remaining_tokens); + + assert_eq!( + vec![ + UserInput::Text { + text: before_image, + text_elements: Vec::new(), + }, + UserInput::Image { + image_url: "file://image.png".to_string(), + }, + UserInput::Text { + text: truncate_text( + &after_image, + TruncationPolicy::Tokens(after_image_token_budget) + ), + text_elements: Vec::new(), + }, + ], + truncated + ); +} + #[test] fn build_token_limited_compacted_history_appends_summary_message() { let initial_context: Vec = Vec::new(); - let user_messages = vec!["first user message".to_string()]; + let user_messages = vec![vec![UserInput::Text { + text: "first user message".to_string(), + text_elements: Vec::new(), + }]]; let summary_text = "summary text"; let history = build_compacted_history(initial_context, &user_messages, summary_text); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 4a75ed063..deb4e2a32 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -15,6 +15,7 @@ use crate::compact::InitialContextInjection; use crate::compact::collect_user_messages; use crate::compact::run_inline_auto_compact_task; use crate::compact::should_use_remote_compact_task; +use crate::compact::user_message_text; use crate::compact_remote::run_inline_remote_auto_compact_task; use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_remote_auto_compact_task_v2; use crate::connectors; @@ -310,7 +311,7 @@ pub(crate) async fn run_turn( Vec::new() } else { let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input.clone()); - let response_item: ResponseItem = initial_input_for_turn.clone().into(); + let response_item: ResponseItem = initial_input_for_turn.into(); let user_prompt_submit_outcome = run_user_prompt_submit_hooks( &sess, &turn_context, @@ -918,7 +919,11 @@ pub(super) fn filter_connectors_for_input( return Vec::new(); } - let mentions = collect_tool_mentions_from_messages(&user_messages); + let user_message_texts = user_messages + .iter() + .map(|message| user_message_text(message)) + .collect::>(); + let mentions = collect_tool_mentions_from_messages(&user_message_texts); let mention_names_lower = mentions .plain_names .iter() @@ -1041,6 +1046,7 @@ async fn run_sampling_request( Arc::clone(&turn_diff_tracker), ) .await; + let max_retries = turn_context.provider.info().stream_max_retries(); let mut retries = 0; let mut initial_input = Some(input); loop { @@ -1074,7 +1080,27 @@ async fn run_sampling_request( } Err(CodexErr::ContextWindowExceeded) => { sess.set_total_tokens_full(&turn_context).await; - return Err(CodexErr::ContextWindowExceeded); + if retries >= max_retries { + return Err(CodexErr::ContextWindowExceeded); + } + retries += 1; + let reset_client_session = match run_auto_compact( + &sess, + &turn_context, + client_session, + InitialContextInjection::BeforeLastUserMessage, + CompactionReason::ContextLimit, + CompactionPhase::MidTurn, + ) + .await + { + Ok(reset_client_session) => reset_client_session, + Err(_) => return Err(CodexErr::TurnAborted), + }; + if reset_client_session { + client_session.reset_websocket_session(); + } + continue; } Err(CodexErr::UsageLimitReached(e)) => { let rate_limits = e.rate_limits.clone(); @@ -1091,7 +1117,6 @@ async fn run_sampling_request( } // Use the configured provider-specific stream retry budget. - let max_retries = turn_context.provider.info().stream_max_retries(); if retries >= max_retries && client_session.try_switch_fallback_transport( &turn_context.session_telemetry, diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 510de6f1b..247575352 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -25,7 +25,6 @@ use codex_protocol::config_types::ModelProviderAuthInfo; use codex_protocol::config_types::ReasoningSummary; use codex_protocol::config_types::Settings; use codex_protocol::config_types::Verbosity; -use codex_protocol::error::CodexErr; use codex_protocol::models::ContentItem; use codex_protocol::models::DEFAULT_IMAGE_DETAIL; use codex_protocol::models::FunctionCallOutputContentItem; @@ -57,6 +56,7 @@ use core_test_support::responses::ev_completed_with_tokens; use core_test_support::responses::ev_message_item_added; use core_test_support::responses::ev_output_text_delta; use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_compact_user_history_with_summary_once; use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_once_match; use core_test_support::responses::mount_sse_sequence; @@ -79,7 +79,6 @@ use uuid::Uuid; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; -use wiremock::matchers::body_string_contains; use wiremock::matchers::header; use wiremock::matchers::header_regex; use wiremock::matchers::method; @@ -2695,32 +2694,34 @@ async fn usage_limit_error_emits_rate_limit_event() -> anyhow::Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn context_window_error_sets_total_tokens_to_model_window() -> anyhow::Result<()> { +async fn context_window_error_sets_total_tokens_to_model_window_before_auto_compact_recovery() +-> anyhow::Result<()> { skip_if_no_network!(Ok(())); let server = MockServer::start().await; const EFFECTIVE_CONTEXT_WINDOW: i64 = (272_000 * 95) / 100; - mount_sse_once_match( + let responses_mock = mount_sse_sequence( &server, - body_string_contains("trigger context window"), - sse_failed( - "resp_context_window", - "context_length_exceeded", - "Your input exceeds the context window of this model. Please adjust your input and try again.", - ), - ) - .await; - - mount_sse_once_match( - &server, - body_string_contains("seed turn"), - sse(vec![ - ev_response_created("resp_seed"), - ev_completed("resp_seed"), - ]), + vec![ + sse(vec![ + ev_response_created("resp_seed"), + ev_completed("resp_seed"), + ]), + sse_failed( + "resp_context_window", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + sse(vec![ + ev_response_created("resp_retry"), + ev_completed("resp_retry"), + ]), + ], ) .await; + let compact_mock = + mount_compact_user_history_with_summary_once(&server, "AUTO_RECOVERY_SUMMARY").await; let TestCodex { codex, .. } = test_codex() .with_config(|config| { @@ -2782,17 +2783,32 @@ async fn context_window_error_sets_total_tokens_to_model_window() -> anyhow::Res EFFECTIVE_CONTEXT_WINDOW ); - let error_event = wait_for_event(&codex, |ev| matches!(ev, EventMsg::Error(_))).await; - let expected_context_window_message = CodexErr::ContextWindowExceeded.to_string(); - assert!( - matches!( - error_event, - EventMsg::Error(ref err) if err.message == expected_context_window_message - ), - "expected context window error; got {error_event:?}" + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + + assert_eq!( + compact_mock.requests().len(), + 1, + "expected overflow recovery to issue one remote compaction request" ); - wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + let requests = responses_mock.requests(); + assert_eq!( + requests.len(), + 3, + "expected seed, overflowing, and recovered sampling requests" + ); + let recovered_request = requests + .last() + .expect("recovered sampling request should be captured"); + let recovered_user_messages = recovered_request.message_input_texts("user"); + assert_eq!( + recovered_user_messages + .iter() + .filter(|message| message.as_str() == "trigger context window") + .count(), + 1, + "recovered sampling request should preserve incoming user text exactly once" + ); Ok(()) } diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index b1620ee36..416279acb 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -2443,6 +2443,94 @@ async fn manual_compact_retries_after_context_window_error() { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_loop_context_window_error_auto_compacts_and_resumes_turn() { + skip_if_no_network!(); + + let server = start_mock_server().await; + let user_message = "turn rescued after sampling overflow"; + let image_url = "data:image/png;base64,local-overflow-preserve-image"; + + let request_log = mount_sse_sequence( + &server, + vec![ + sse_failed( + "resp-overflow", + "context_length_exceeded", + CONTEXT_LIMIT_MESSAGE, + ), + sse(vec![ + ev_assistant_message("compact-summary", &auto_summary(AUTO_SUMMARY_TEXT)), + ev_completed_with_tokens("compact-response", /*total_tokens*/ 10), + ]), + sse(vec![ + ev_assistant_message("recovered-assistant", FINAL_REPLY), + ev_completed_with_tokens("recovered-response", /*total_tokens*/ 10), + ]), + ], + ) + .await; + + let model_provider = non_openai_model_provider(&server); + let codex = test_codex() + .with_config(move |config| { + config.model_provider = model_provider; + set_test_compact_prompt(config); + config.model_auto_compact_token_limit = Some(200_000); + }) + .build(&server) + .await + .expect("build codex") + .codex; + + codex + .submit(Op::UserInput { + environments: None, + items: vec![ + UserInput::Image { + image_url: image_url.to_string(), + }, + UserInput::Text { + text: user_message.to_string(), + text_elements: Vec::new(), + }, + ], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await + .expect("submit overflowing user turn"); + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + let requests = request_log.requests(); + assert_eq!( + requests.len(), + 3, + "expected overflow, local compaction, and recovered sampling requests" + ); + assert!( + requests[1].body_contains_text(SUMMARIZATION_PROMPT), + "overflow recovery should reuse the automatic local compaction prompt" + ); + + let recovered_user_messages = requests[2].message_input_texts("user"); + let recovered_user_images = requests[2].message_input_image_urls("user"); + assert_eq!( + ( + recovered_user_messages + .iter() + .filter(|message| message.as_str() == user_message) + .count(), + recovered_user_images + .iter() + .filter(|url| url.as_str() == image_url) + .count(), + ), + (1, 1), + "recovered sampling request should preserve incoming user text and image exactly once" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] // TODO(ccunningham): Re-enable after the follow-up compaction behavior PR lands. // Current main behavior around non-context manual /compact failures is known-incorrect. diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index bec0ea436..9e755b497 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -1327,6 +1327,335 @@ async fn auto_remote_compact_trims_function_call_history_to_fit_context_window() Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_loop_context_window_error_auto_remote_compacts_and_resumes_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + let user_message = "turn rescued by remote compaction"; + let image_url = "data:image/png;base64,remote-overflow-preserve-image"; + let harness = TestCodexHarness::with_builder( + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + config.model_auto_compact_token_limit = Some(200_000); + }), + ) + .await?; + let codex = harness.test().codex.clone(); + + let responses_mock = responses::mount_sse_sequence( + harness.server(), + vec![ + responses::sse_failed( + "remote-overflow", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + responses::sse(vec![ + responses::ev_assistant_message("recovered-assistant", "REMOTE_RECOVERED"), + responses::ev_completed("recovered-response"), + ]), + ], + ) + .await; + let compact_mock = responses::mount_compact_user_history_with_summary_once( + harness.server(), + "REMOTE_AUTO_RECOVERY_SUMMARY", + ) + .await; + + codex + .submit(Op::UserInput { + environments: None, + items: vec![ + UserInput::Image { + image_url: image_url.to_string(), + }, + UserInput::Text { + text: user_message.to_string(), + text_elements: Vec::new(), + }, + ], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await?; + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + assert_eq!( + compact_mock.requests().len(), + 1, + "expected normal-loop overflow to issue one remote compact request" + ); + + let responses = responses_mock.requests(); + assert_eq!( + responses.len(), + 2, + "expected overflowing and recovered sampling requests" + ); + let recovered_user_messages = responses[1].message_input_texts("user"); + let recovered_user_images = responses[1].message_input_image_urls("user"); + assert_eq!( + ( + recovered_user_messages + .iter() + .filter(|message| message.as_str() == user_message) + .count(), + recovered_user_images + .iter() + .filter(|url| url.as_str() == image_url) + .count(), + ), + (1, 1), + "recovered sampling request should preserve incoming user text and image exactly once" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_loop_context_window_error_auto_remote_v2_compacts_and_preserves_user_turn() +-> Result<()> { + skip_if_no_network!(Ok(())); + + let user_message = "turn rescued by remote compaction v2"; + let image_url = "data:image/png;base64,remote-v2-overflow-preserve-image"; + let harness = TestCodexHarness::with_builder( + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + config.model_auto_compact_token_limit = Some(200_000); + let _ = config.features.enable(Feature::RemoteCompactionV2); + }), + ) + .await?; + let codex = harness.test().codex.clone(); + + let responses_mock = responses::mount_sse_sequence( + harness.server(), + vec![ + responses::sse_failed( + "remote-v2-overflow", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + responses::sse(vec![ + serde_json::json!({ + "type": "response.output_item.done", + "item": { + "type": "context_compaction", + "encrypted_content": "REMOTE_V2_AUTO_RECOVERY_SUMMARY", + } + }), + responses::ev_completed("remote-v2-compact"), + ]), + responses::sse(vec![ + responses::ev_assistant_message("recovered-assistant", "REMOTE_V2_RECOVERED"), + responses::ev_completed("recovered-response"), + ]), + ], + ) + .await; + + codex + .submit(Op::UserInput { + environments: None, + items: vec![ + UserInput::Image { + image_url: image_url.to_string(), + }, + UserInput::Text { + text: user_message.to_string(), + text_elements: Vec::new(), + }, + ], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await?; + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + let responses = responses_mock.requests(); + assert_eq!( + responses.len(), + 3, + "expected overflowing sampling, remote v2 compaction, and recovered sampling requests" + ); + assert!( + responses[1] + .body_json() + .to_string() + .contains("\"type\":\"context_compaction\""), + "expected the v2 rescue request to include the context compaction trigger" + ); + assert_eq!( + ( + responses[2] + .message_input_texts("user") + .iter() + .filter(|message| message.as_str() == user_message) + .count(), + responses[2] + .message_input_image_urls("user") + .iter() + .filter(|url| url.as_str() == image_url) + .count(), + ), + (1, 1), + "recovered v2 sampling request should preserve incoming user text and image exactly once" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_loop_context_window_error_stops_after_sample_retry_budget() -> Result<()> { + skip_if_no_network!(Ok(())); + + let harness = TestCodexHarness::with_builder( + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + config.model_auto_compact_token_limit = Some(200_000); + config.model_provider.stream_max_retries = Some(1); + }), + ) + .await?; + let codex = harness.test().codex.clone(); + + let responses_mock = responses::mount_sse_sequence( + harness.server(), + vec![ + responses::sse_failed( + "remote-overflow-before-compact", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + responses::sse_failed( + "remote-overflow-after-compact", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + ], + ) + .await; + let compact_mock = responses::mount_compact_user_history_with_summary_once( + harness.server(), + "REMOTE_AUTO_RECOVERY_SUMMARY", + ) + .await; + + codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: "turn whose compacted retry still overflows".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await?; + + let error_message = wait_for_event_match(&codex, |event| match event { + EventMsg::Error(err) => Some(err.message.clone()), + _ => None, + }) + .await; + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + assert_eq!( + ( + error_message.to_lowercase().contains("context window"), + compact_mock.requests().len(), + responses_mock.requests().len(), + ), + (true, 1, 2), + "expected the overflow error after one recovery compaction and one compacted retry, got {error_message}" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn normal_loop_context_window_error_stops_after_remote_compaction_failure() -> Result<()> { + skip_if_no_network!(Ok(())); + + let harness = TestCodexHarness::with_builder( + test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + config.model_auto_compact_token_limit = Some(200_000); + }), + ) + .await?; + let codex = harness.test().codex.clone(); + + let overflow_mock = mount_sse_once( + harness.server(), + responses::sse_failed( + "remote-overflow", + "context_length_exceeded", + "Your input exceeds the context window of this model. Please adjust your input and try again.", + ), + ) + .await; + let post_compact_turn_mock = mount_sse_once( + harness.server(), + responses::sse(vec![ + responses::ev_assistant_message("should-not-run", "SHOULD_NOT_RUN"), + responses::ev_completed("should-not-run-response"), + ]), + ) + .await; + let compact_mock = responses::mount_compact_json_once( + harness.server(), + serde_json::json!({ "output": "invalid compact payload shape" }), + ) + .await; + + codex + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: "turn whose overflow rescue fails".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }) + .await?; + + let error_message = wait_for_event_match(&codex, |event| match event { + EventMsg::Error(err) => Some(err.message.clone()), + _ => None, + }) + .await; + wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + assert!( + error_message.contains("Error running remote compact task"), + "expected remote compact task error prefix, got {error_message}" + ); + assert_eq!( + compact_mock.requests().len(), + 1, + "expected exactly one failed remote compact request" + ); + assert_eq!( + overflow_mock.requests().len(), + 1, + "expected exactly one overflowing sampling request" + ); + assert!( + post_compact_turn_mock.requests().is_empty(), + "expected agent loop to stop before retrying sampling after compact failure" + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn auto_remote_compact_failure_stops_agent_loop() -> Result<()> { skip_if_no_network!(Ok(()));