diff --git a/codex-rs/cli/src/doctor.rs b/codex-rs/cli/src/doctor.rs index 859fad9f69..533cf0d4e8 100644 --- a/codex-rs/cli/src/doctor.rs +++ b/codex-rs/cli/src/doctor.rs @@ -2438,6 +2438,7 @@ fn websocket_error_detail(err: &ApiError) -> String { } ApiError::Stream(message) => format!("handshake stream error: {message}"), ApiError::ContentFilter + | ApiError::ResponseIncomplete(_) | ApiError::ContextWindowExceeded | ApiError::QuotaExceeded | ApiError::UsageNotIncluded diff --git a/codex-rs/codex-api/src/api_bridge.rs b/codex-rs/codex-api/src/api_bridge.rs index abf0bf0d70..98d8d17ef6 100644 --- a/codex-rs/codex-api/src/api_bridge.rs +++ b/codex-rs/codex-api/src/api_bridge.rs @@ -29,6 +29,7 @@ pub fn map_api_error(err: ApiError) -> CodexErr { | ApiError::Api { .. } | ApiError::Stream(_) | ApiError::ContentFilter + | ApiError::ResponseIncomplete(_) | ApiError::ContextWindowExceeded | ApiError::QuotaExceeded | ApiError::UsageNotIncluded @@ -58,6 +59,9 @@ fn map_api_error_details(err: ApiError) -> CodexErr { } ApiError::Stream(msg) => CodexErr::Stream(msg), ApiError::ContentFilter => CodexErr::new(CodexErrorDetails::ContentFilter), + ApiError::ResponseIncomplete(incomplete) => { + CodexErr::new(CodexErrorDetails::ResponseIncomplete(incomplete)) + } ApiError::ServerOverloaded { .. } => CodexErr::ServerOverloaded, ApiError::FlexUnavailable => CodexErr::new(CodexErrorDetails::FlexUnavailable), ApiError::Api { status, message } => { diff --git a/codex-rs/codex-api/src/error.rs b/codex-rs/codex-api/src/error.rs index 2834137129..631d892d20 100644 --- a/codex-rs/codex-api/src/error.rs +++ b/codex-rs/codex-api/src/error.rs @@ -16,6 +16,8 @@ pub enum ApiError { Stream(String), #[error("stream error: Incomplete response returned, reason: content_filter")] ContentFilter, + #[error("Incomplete response returned, reason: {0}")] + ResponseIncomplete(String), #[error("context window exceeded")] ContextWindowExceeded, #[error("quota exceeded")] diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index e68d4bb7b7..0ea2e668ce 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -429,8 +429,9 @@ pub fn process_responses_event( return Err(ResponsesEventError::Api(ApiError::ContentFilter)); } if reason != "interrupted" { - let message = format!("Incomplete response returned, reason: {reason}"); - return Err(ResponsesEventError::Api(ApiError::Stream(message))); + return Err(ResponsesEventError::Api(ApiError::ResponseIncomplete( + reason.to_string(), + ))); } } if let Some(resp_val) = event.response { @@ -639,7 +640,10 @@ async fn process_sse_with_treatment( Ok(None) => {} Err(error) => { let error = error.into_api_error(); - if matches!(error, ApiError::FlexUnavailable) { + if matches!( + error, + ApiError::FlexUnavailable | ApiError::ResponseIncomplete(_) + ) { let _ = tx_event.send(Err(error)).await; return; } diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index e60ca02243..a3c3165fdb 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -332,7 +332,13 @@ async fn run_compact_task_inner_impl( { return Err(err); } - Err(e) if matches!(e.details(), CodexErrorDetails::SessionBudgetExceeded) => { + Err(e) + if matches!( + e.details(), + CodexErrorDetails::SessionBudgetExceeded + | CodexErrorDetails::ResponseIncomplete(_) + ) => + { return Err(e); } Err(e) if matches!(e.details(), CodexErrorDetails::ContextWindowExceeded) => { diff --git a/codex-rs/core/src/compact_model_fallback.rs b/codex-rs/core/src/compact_model_fallback.rs index 9e9a6213d5..7d4c0e25cb 100644 --- a/codex-rs/core/src/compact_model_fallback.rs +++ b/codex-rs/core/src/compact_model_fallback.rs @@ -17,6 +17,7 @@ pub(crate) fn should_retry_with_current_model( CodexErrorDetails::TurnAborted | CodexErrorDetails::Interrupted | CodexErrorDetails::SessionBudgetExceeded + | CodexErrorDetails::ResponseIncomplete(_) ) || (previous.model_info().slug == current.model_info().slug && previous.cyber_access_program == current.cyber_access_program) { diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index b79a2b0d17..35591b2e69 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -3799,6 +3799,83 @@ async fn incomplete_response_emits_content_filter_error_message() -> anyhow::Res Ok(()) } +#[test_case::test_case(false; "sse")] +#[test_case::test_case(true; "websocket")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn incomplete_response_is_terminal(use_websocket: bool) -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + let incomplete = vec![ + ev_response_created("resp-incomplete"), + ev_assistant_message("partial", "Partial reply."), + json!({ + "type": "response.incomplete", + "response": { + "id": "resp-incomplete", + "incomplete_details": {"reason": "max_output_tokens"} + } + }), + ]; + let server = MockServer::start().await; + let response_mock = if use_websocket { + None + } else { + Some(mount_sse_once(&server, sse(incomplete.clone())).await) + }; + let websocket_server = if use_websocket { + Some( + core_test_support::responses::start_websocket_server(vec![vec![ + vec![ev_response_created("prewarm"), ev_completed("prewarm")], + incomplete, + ]]) + .await, + ) + } else { + None + }; + let mut builder = test_codex().with_config(|config| { + config.model_provider.stream_max_retries = Some(2); + config.model_provider.request_max_retries = Some(0); + }); + let test = if let Some(server) = &websocket_server { + builder.build_with_websocket_server(server).await? + } else { + builder.build_with_auto_env(&server).await? + }; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "trigger output limit".into(), + text_elements: Vec::new(), + }])) + .await?; + let mut errors = Vec::new(); + wait_for_event(&test.codex, |event| { + match event { + EventMsg::Error(error) => errors.push(error.message.clone()), + EventMsg::StreamError(_) | EventMsg::Warning(_) | EventMsg::RawResponseCompleted(_) => { + panic!("unexpected incomplete-response event: {event:?}") + } + _ => {} + } + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + insta::assert_snapshot!(errors.join("\n"), @"Incomplete response returned, reason: max_output_tokens"); + if let Some(server) = websocket_server { + assert_eq!( + server + .connections() + .iter() + .map(Vec::len) + .collect::>(), + vec![2], + ); + server.shutdown().await; + } else { + response_mock.expect("SSE response mock").single_request(); + } + Ok(()) +} + /// We try to avoid setting env vars in tests because std::env::set_var() is /// process-wide and unsafe. Though for this test, we want to simulate the /// presence of an environment variable that the provider will read for auth, so diff --git a/codex-rs/core/tests/suite/compact.rs b/codex-rs/core/tests/suite/compact.rs index b46bc107e9..414fbbc9af 100644 --- a/codex-rs/core/tests/suite/compact.rs +++ b/codex-rs/core/tests/suite/compact.rs @@ -1151,6 +1151,110 @@ async fn reasoning_effort_override_remote_v2_compaction_resets_pinned_effort( Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn incomplete_response_during_compaction_is_terminal() -> Result<()> { + skip_if_no_network!(Ok(())); + for (provider, expected_error) in [ + ( + non_openai_model_provider as fn(&MockServer) -> ModelProviderInfo, + "Incomplete response returned, reason: max_output_tokens", + ), + ( + openai_model_provider, + "Error running remote compact task: Incomplete response returned, reason: max_output_tokens", + ), + ] { + let server = start_mock_server().await; + let previous_model = "gpt-5.4"; + let selected_model = "gpt-5.2"; + let response_mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_assistant_message("initial", "Initial reply."), + ev_completed_with_tokens("initial", /*total_tokens*/ 120_000), + ]), + sse(vec![json!({ + "type": "response.incomplete", + "response": { + "id": "incomplete-compaction", + "incomplete_details": {"reason": "max_output_tokens"} + } + })]), + ], + ) + .await; + let model_provider = provider(&server); + let test = test_codex() + .with_model(previous_model) + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(move |config| { + config.model_provider = model_provider; + config.model_catalog = Some(ModelsResponse { + models: vec![ + model_info_with_context_window( + previous_model, + /*context_window*/ 273_000, + ), + model_info_with_context_window( + selected_model, + /*context_window*/ 125_000, + ), + ], + }); + config.model_provider.stream_max_retries = Some(2); + set_test_compact_prompt(config); + }) + .build_with_auto_env(&server) + .await?; + test.submit_turn("hello").await?; + // A smaller selected model forces compaction with the previous model. + // An incomplete result must stop both retries and fallback to the selected model. + test.codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "continue with the smaller model".into(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + collaboration_mode: Some(CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model: selected_model.to_string(), + reasoning_effort: None, + developer_instructions: None, + }, + }), + ..Default::default() + }), + ) + .await?; + let mut errors = Vec::new(); + wait_for_event(&test.codex, |event| { + match event { + EventMsg::Error(error) => errors.push(error.message.clone()), + EventMsg::StreamError(value) => panic!("unexpected compaction retry: {value:?}"), + EventMsg::RawResponseCompleted(value) => { + panic!("incomplete response marked complete: {value:?}") + } + _ => {} + } + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert_eq!(errors, vec![expected_error]); + assert_eq!( + response_mock + .requests() + .iter() + .map(|request| request.body_json()["model"].clone()) + .collect::>(), + vec![json!(previous_model), json!(previous_model)], + ); + } + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn manual_compact_records_durable_and_local_token_usage() { skip_if_no_network!(); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs b/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs index de96c9ed9b..d863f16944 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/metrics.rs @@ -47,6 +47,7 @@ pub(super) fn sampler_failure_reason(error: &LunaSamplerError) -> &'static str { ApiError::Transport(TransportError::ResponseTooLarge { .. }) => "response_too_large", ApiError::Transport(TransportError::Policy(_)) => "network_policy_denied", ApiError::Stream(_) | ApiError::ContentFilter => "stream_error", + ApiError::ResponseIncomplete(_) => "incomplete_response", ApiError::ContextWindowExceeded => "context_window_exceeded", ApiError::QuotaExceeded => "quota_exceeded", ApiError::UsageNotIncluded => "usage_not_included", diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs index 04c9ec75da..89021097e3 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs @@ -172,6 +172,7 @@ impl SamplingExecution { | TransportError::Policy(_), ) | ApiError::ContextWindowExceeded + | ApiError::ResponseIncomplete(_) | ApiError::QuotaExceeded | ApiError::UsageNotIncluded | ApiError::RateLimit(_) diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index 9e44430d6f..571182b452 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -105,6 +105,8 @@ pub enum CodexErrorDetails { "stream disconnected before completion: Incomplete response returned, reason: content_filter" )] ContentFilter, + #[error("Incomplete response returned, reason: {0}")] + ResponseIncomplete(String), /// A retryable upstream rate limit received inside the response stream. #[error("rate limit exceeded: {0}")] RateLimitExceeded(String), @@ -405,6 +407,7 @@ impl CodexErr { | CodexErrorDetails::QuotaExceeded | CodexErrorDetails::InvalidImageRequest() | CodexErrorDetails::InvalidRequest(_) + | CodexErrorDetails::ResponseIncomplete(_) | CodexErrorDetails::InvalidPrompt { .. } | CodexErrorDetails::ToolCollision(_) | CodexErrorDetails::RefreshTokenFailed(_) diff --git a/codex-rs/response-debug-context/src/lib.rs b/codex-rs/response-debug-context/src/lib.rs index e02ec6bc93..97f4024271 100644 --- a/codex-rs/response-debug-context/src/lib.rs +++ b/codex-rs/response-debug-context/src/lib.rs @@ -78,6 +78,7 @@ pub fn telemetry_api_error_message(error: &ApiError) -> String { ApiError::ContentFilter => { "Incomplete response returned, reason: content_filter".to_string() } + ApiError::ResponseIncomplete(_) => "incomplete response".to_string(), ApiError::ContextWindowExceeded => "context window exceeded".to_string(), ApiError::QuotaExceeded => "quota exceeded".to_string(), ApiError::UsageNotIncluded => "usage not included".to_string(),