From 10fd75bba78c7bbc59826ef1fa69df9c67615bd0 Mon Sep 17 00:00:00 2001 From: "Adam Perry @ OpenAI" Date: Fri, 25 Sep 2026 22:29:48 +0000 Subject: [PATCH] Extract Responses failure parsing into a dedicated module (#48229) ## What changed Move `response.failed` classification and rate-limit retry-delay parsing into `responses_error.rs`, preserving existing error mappings, fallback messages, and malformed-payload behavior. Replace the inline classification logic in the SSE event handler with `parse_failed_response`. ## Testing Add regression coverage for missing or malformed failed-response payloads, including invalid values in unused error fields. Move the existing retry-delay parsing tests alongside the extracted parser. GitOrigin-RevId: 6faa09b3fb312f1990509eef7088c8a086724615 --- codex-rs/codex-api/src/sse/mod.rs | 1 + codex-rs/codex-api/src/sse/responses.rs | 228 +++--------------- codex-rs/codex-api/src/sse/responses_error.rs | 138 +++++++++++ .../src/sse/responses_error_parser_tests.rs | 47 ++++ 4 files changed, 213 insertions(+), 201 deletions(-) create mode 100644 codex-rs/codex-api/src/sse/responses_error.rs create mode 100644 codex-rs/codex-api/src/sse/responses_error_parser_tests.rs diff --git a/codex-rs/codex-api/src/sse/mod.rs b/codex-rs/codex-api/src/sse/mod.rs index 441078bdcb..b4872ba6ac 100644 --- a/codex-rs/codex-api/src/sse/mod.rs +++ b/codex-rs/codex-api/src/sse/mod.rs @@ -1,4 +1,5 @@ pub(crate) mod responses; +mod responses_error; pub(crate) use responses::ResponsesStreamEvent; pub(crate) use responses::process_responses_event; diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 24a81bcc14..b01b8ad46b 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -1,3 +1,4 @@ +use super::responses_error::parse_failed_response; use crate::common::ResponseEvent; use crate::common::ResponseStream; use crate::common::SafetyBuffering; @@ -9,10 +10,8 @@ use crate::safety_buffering::treatment_from_headers; use crate::telemetry::SseTelemetry; use codex_client::ByteStream; use codex_client::StreamResponse; -use codex_http_client::RetryAfter; use codex_protocol::ResponseUsageMetadata; use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::MisalignmentErrorDetails; use codex_protocol::protocol::ModelVerification; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TurnModerationMetadataEvent; @@ -103,18 +102,6 @@ pub fn spawn_response_stream( } } -#[derive(Debug, Deserialize)] -#[allow(dead_code)] -struct Error { - r#type: Option, - code: Option, - message: Option, - plan_type: Option, - resets_at: Option, - #[serde(default)] - misalignment: Option, -} - #[derive(Debug, Deserialize)] #[allow(dead_code)] struct ResponseCompleted { @@ -423,75 +410,8 @@ pub fn process_responses_event( } } "response.failed" => { - if let Some(resp_val) = event.response { - let mut response_error = ApiError::Stream("response.failed event received".into()); - if let Some(error) = resp_val.get("error").and_then(parse_flex_unavailable) { - return Err(ResponsesEventError::Api(error)); - } - if let Some(error) = resp_val.get("error") - && let Ok(error) = serde_json::from_value::(error.clone()) - { - if is_context_window_error(&error) { - response_error = ApiError::ContextWindowExceeded; - } else if is_quota_exceeded_error(&error) { - response_error = ApiError::QuotaExceeded; - } else if is_usage_not_included(&error) { - response_error = ApiError::UsageNotIncluded; - } else if is_cyber_policy_error(&error) { - let message = cyber_policy_message(error.message); - response_error = ApiError::CyberPolicy { message }; - } else if error.code.as_deref() == Some("bio_policy") { - let message = error - .message - .filter(|message| !message.trim().is_empty()) - .unwrap_or_else(|| { - "This content was flagged for possible biological risk.".to_string() - }); - response_error = ApiError::BioPolicy { message }; - } else if error.code.as_deref() == Some("misalignment_policy_violation") { - let message = error - .message - .filter(|message| !message.trim().is_empty()) - .unwrap_or_else(|| { - "This request was blocked due to a misalignment policy violation." - .to_string() - }); - response_error = ApiError::MisalignmentPolicyViolation { - message, - misalignment: error.misalignment.and_then(|details| { - serde_json::from_value::(details).ok() - }), - }; - } else if error.code.as_deref() == Some("invalid_prompt") { - let message = error - .message - .unwrap_or_else(|| "Invalid request.".to_string()); - response_error = ApiError::InvalidPrompt { message }; - } else if is_server_overloaded_error(&error) { - response_error = ApiError::ServerOverloaded { retry_after: None }; - } else { - let retry_after = - try_parse_retry_delay(&error).and_then(RetryAfter::from_delay); - let message = error.message.unwrap_or_default(); - response_error = match error.code.as_deref() { - Some("rate_limit_exceeded" | "slow_down") => { - ApiError::RateLimitExceeded { - message, - retry_after, - } - } - _ => ApiError::Retryable { - message, - retry_after, - }, - }; - } - } - return Err(ResponsesEventError::Api(response_error)); - } - - return Err(ResponsesEventError::Api(ApiError::Stream( - "response.failed event received".into(), + return Err(ResponsesEventError::Api(parse_failed_response( + event.response, ))); } "response.incomplete" => { @@ -718,81 +638,6 @@ async fn process_sse_with_treatment( } } -fn try_parse_retry_delay(err: &Error) -> Option { - if !matches!( - err.code.as_deref(), - Some("rate_limit_exceeded" | "slow_down") - ) { - return None; - } - - let re = rate_limit_regex(); - if let Some(message) = &err.message - && let Some(captures) = re.captures(message) - { - let seconds = captures.get(1); - let unit = captures.get(2); - - if let (Some(value), Some(unit)) = (seconds, unit) { - let value = value.as_str().parse::().ok()?; - let unit = unit.as_str().to_ascii_lowercase(); - - if unit == "s" || unit.starts_with("second") { - return Some(Duration::from_secs_f64(value)); - } else if unit == "ms" { - return Some(Duration::from_millis(value as u64)); - } - } - } - None -} - -fn is_context_window_error(error: &Error) -> bool { - error.code.as_deref() == Some("context_length_exceeded") -} - -fn is_quota_exceeded_error(error: &Error) -> bool { - matches!( - error.code.as_deref(), - Some( - "insufficient_quota" - | "credit_balance_exhausted" - | "organization_spend_limit_exceeded" - | "project_spend_limit_exceeded" - ) - ) -} - -fn is_usage_not_included(error: &Error) -> bool { - error.code.as_deref() == Some("usage_not_included") -} - -fn is_cyber_policy_error(error: &Error) -> bool { - error.code.as_deref() == Some("cyber_policy") -} - -fn is_server_overloaded_error(error: &Error) -> bool { - error.code.as_deref() == Some("server_is_overloaded") -} - -fn cyber_policy_fallback_message() -> String { - "This request has been flagged for possible cybersecurity risk.".to_string() -} - -fn cyber_policy_message(message: Option) -> String { - message - .filter(|message| !message.trim().is_empty()) - .unwrap_or_else(cyber_policy_fallback_message) -} - -fn rate_limit_regex() -> &'static regex_lite::Regex { - static RE: std::sync::OnceLock = std::sync::OnceLock::new(); - #[expect(clippy::unwrap_used)] - RE.get_or_init(|| { - regex_lite::Regex::new(r"(?i)try again in\s*(\d+(?:\.\d+)?)\s*(s|ms|seconds?)").unwrap() - }) -} - #[cfg(test)] mod tests { use super::*; @@ -800,8 +645,10 @@ mod tests { use bytes::Bytes; use codex_client::StreamResponse; use codex_client::TransportError; + use codex_http_client::RetryAfter; use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; + use codex_protocol::protocol::MisalignmentErrorDetails; use futures::TryStreamExt; use futures::stream; use http::HeaderMap; @@ -1208,6 +1055,28 @@ mod tests { } } + #[tokio::test] + async fn malformed_failed_responses_use_stream_error() { + for event in [ + json!({"type": "response.failed"}), + json!({"type": "response.failed", "response": null}), + json!({"type": "response.failed", "response": []}), + json!({"type": "response.failed", "response": {}}), + json!({"type": "response.failed", "response": {"error": null}}), + json!({"type": "response.failed", "response": {"error": {"code": 42}}}), + json!({"type": "response.failed", "response": {"error": {"code": "server_is_overloaded", "plan_type": 42}}}), + ] { + let sse = format!("event: response.failed\ndata: {event}\n\n"); + let events = collect_events(&[sse.as_bytes()]).await; + match events.as_slice() { + [Err(ApiError::Stream(message))] => { + assert_eq!(message, "response.failed event received"); + } + _ => panic!("unexpected events: {events:?}"), + } + } + } + #[tokio::test] async fn context_window_error_is_fatal() { let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_5c66275b97b9baef1ed95550adb3b7ec13b17aafd1d2f11b","object":"response","created_at":1759510079,"status":"failed","background":false,"error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."},"usage":null,"user":null,"metadata":{}}}"#; @@ -2106,48 +1975,5 @@ mod tests { assert_eq!(event.model_verifications(), None); } - #[test] - fn test_try_parse_retry_delay() { - let err = Error { - r#type: None, - message: Some("Rate limit reached for gpt-5.1 in organization org- on tokens per min (TPM): Limit 1, Used 1, Requested 19304. Please try again in 28ms. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), - code: Some("rate_limit_exceeded".to_string()), - plan_type: None, - resets_at: None, - misalignment: None, - }; - - let delay = try_parse_retry_delay(&err); - assert_eq!(delay, Some(Duration::from_millis(28))); - } - - #[test] - fn test_try_parse_retry_delay_no_delay() { - let err = Error { - r#type: None, - message: Some("Rate limit reached for gpt-5.1 in organization on tokens per min (TPM): Limit 30000, Used 6899, Requested 24050. Please try again in 1.898s. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), - code: Some("rate_limit_exceeded".to_string()), - plan_type: None, - resets_at: None, - misalignment: None, - }; - let delay = try_parse_retry_delay(&err); - assert_eq!(delay, Some(Duration::from_secs_f64(1.898))); - } - - #[test] - fn test_try_parse_retry_delay_azure() { - let err = Error { - r#type: None, - message: Some("Rate limit exceeded. Try again in 35 seconds.".to_string()), - code: Some("rate_limit_exceeded".to_string()), - plan_type: None, - resets_at: None, - misalignment: None, - }; - let delay = try_parse_retry_delay(&err); - assert_eq!(delay, Some(Duration::from_secs(35))); - } - const CYBER_RESTRICTED_MODEL_FOR_TESTS: &str = "gpt-5.3-codex"; } diff --git a/codex-rs/codex-api/src/sse/responses_error.rs b/codex-rs/codex-api/src/sse/responses_error.rs new file mode 100644 index 0000000000..872769c3b1 --- /dev/null +++ b/codex-rs/codex-api/src/sse/responses_error.rs @@ -0,0 +1,138 @@ +//! Classifies failed Responses events and their rate-limit retry delays. +//! Only rate-limit codes derive retry delays from plaintext messages. + +use crate::error::ApiError; +use crate::error::parse_flex_unavailable; +use codex_http_client::RetryAfter; +use codex_protocol::protocol::MisalignmentErrorDetails; +use serde::Deserialize; +use serde_json::Value; +use std::time::Duration; + +// Keep decoding unused fields: malformed values currently produce a stream error. +#[derive(Debug, Deserialize)] +#[allow(dead_code)] +struct Error { + r#type: Option, + code: Option, + message: Option, + plan_type: Option, + resets_at: Option, + #[serde(default)] + misalignment: Option, +} + +pub(super) fn parse_failed_response(response: Option) -> ApiError { + if let Some(error) = response + .as_ref() + .and_then(|response| response.get("error")) + .and_then(parse_flex_unavailable) + { + return error; + } + let Some(error) = response + .as_ref() + .and_then(|response| response.get("error")) + .and_then(|error| serde_json::from_value::(error.clone()).ok()) + else { + return ApiError::Stream("response.failed event received".into()); + }; + + match error.code.as_deref() { + Some("context_length_exceeded") => ApiError::ContextWindowExceeded, + Some( + "insufficient_quota" + | "credit_balance_exhausted" + | "organization_spend_limit_exceeded" + | "project_spend_limit_exceeded", + ) => ApiError::QuotaExceeded, + Some("usage_not_included") => ApiError::UsageNotIncluded, + Some("cyber_policy") => ApiError::CyberPolicy { + message: cyber_policy_message(error.message), + }, + Some("bio_policy") => { + let message = error + .message + .filter(|message| !message.trim().is_empty()) + .unwrap_or_else(|| { + "This content was flagged for possible biological risk.".to_string() + }); + ApiError::BioPolicy { message } + } + Some("misalignment_policy_violation") => { + let message = error + .message + .filter(|message| !message.trim().is_empty()) + .unwrap_or_else(|| { + "This request was blocked due to a misalignment policy violation.".to_string() + }); + ApiError::MisalignmentPolicyViolation { + message, + misalignment: error.misalignment.and_then(|details| { + serde_json::from_value::(details).ok() + }), + } + } + Some("invalid_prompt") => ApiError::InvalidPrompt { + message: error + .message + .unwrap_or_else(|| "Invalid request.".to_string()), + }, + Some("server_is_overloaded") => ApiError::ServerOverloaded { retry_after: None }, + Some("rate_limit_exceeded" | "slow_down") => { + let retry_after = try_parse_retry_delay(&error).and_then(RetryAfter::from_delay); + ApiError::RateLimitExceeded { + message: error.message.unwrap_or_default(), + retry_after, + } + } + _ => ApiError::Retryable { + message: error.message.unwrap_or_default(), + retry_after: None, + }, + } +} + +fn try_parse_retry_delay(err: &Error) -> Option { + let re = rate_limit_regex(); + if let Some(message) = &err.message + && let Some(captures) = re.captures(message) + { + let seconds = captures.get(1); + let unit = captures.get(2); + + if let (Some(value), Some(unit)) = (seconds, unit) { + let value = value.as_str().parse::().ok()?; + let unit = unit.as_str().to_ascii_lowercase(); + + if unit == "s" || unit.starts_with("second") { + return Some(Duration::from_secs_f64(value)); + } else if unit == "ms" { + return Some(Duration::from_millis(value as u64)); + } + } + } + None +} + +fn cyber_policy_fallback_message() -> String { + "This request has been flagged for possible cybersecurity risk.".to_string() +} + +fn cyber_policy_message(message: Option) -> String { + message + .filter(|message| !message.trim().is_empty()) + .unwrap_or_else(cyber_policy_fallback_message) +} + +fn rate_limit_regex() -> &'static regex_lite::Regex { + static RE: std::sync::OnceLock = std::sync::OnceLock::new(); + #[expect(clippy::unwrap_used)] + RE.get_or_init(|| { + regex_lite::Regex::new(r"(?i)try again in\s*(\d+(?:\.\d+)?)\s*(s|ms|seconds?)").unwrap() + }) +} + +#[cfg(test)] +#[path = "responses_error_parser_tests.rs"] +mod tests; diff --git a/codex-rs/codex-api/src/sse/responses_error_parser_tests.rs b/codex-rs/codex-api/src/sse/responses_error_parser_tests.rs new file mode 100644 index 0000000000..feb4098e07 --- /dev/null +++ b/codex-rs/codex-api/src/sse/responses_error_parser_tests.rs @@ -0,0 +1,47 @@ +//! Covers retry-delay parsing for provider error messages. + +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn test_try_parse_retry_delay() { + let err = Error { + r#type: None, + message: Some("Rate limit reached for gpt-5.1 in organization org- on tokens per min (TPM): Limit 1, Used 1, Requested 19304. Please try again in 28ms. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), + code: Some("rate_limit_exceeded".to_string()), + plan_type: None, + resets_at: None, + misalignment: None, + }; + + let delay = try_parse_retry_delay(&err); + assert_eq!(delay, Some(Duration::from_millis(28))); +} + +#[test] +fn test_try_parse_retry_delay_no_delay() { + let err = Error { + r#type: None, + message: Some("Rate limit reached for gpt-5.1 in organization on tokens per min (TPM): Limit 30000, Used 6899, Requested 24050. Please try again in 1.898s. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), + code: Some("rate_limit_exceeded".to_string()), + plan_type: None, + resets_at: None, + misalignment: None, + }; + let delay = try_parse_retry_delay(&err); + assert_eq!(delay, Some(Duration::from_secs_f64(1.898))); +} + +#[test] +fn test_try_parse_retry_delay_azure() { + let err = Error { + r#type: None, + message: Some("Rate limit exceeded. Try again in 35 seconds.".to_string()), + code: Some("rate_limit_exceeded".to_string()), + plan_type: None, + resets_at: None, + misalignment: None, + }; + let delay = try_parse_retry_delay(&err); + assert_eq!(delay, Some(Duration::from_secs(35))); +}