Surface Flex capacity failures as a distinct terminal error (#47967)

## What changed

- Recognize `flex_unavailable` in HTTP 429 responses and streamed `error` and `response.failed` events. End normal turns without retries and report `Flex capacity unavailable.`
- Expose `flexUnavailable` through the core and app-server protocols and generated schemas and SDK types.
- Keep Flex capacity failures eligible for Guardian review and sampler retries, with a distinct telemetry classification.
- Map unexpected HTTP responses to `HttpConnectionFailed`, preserving their status codes.

## Testing

Add regression coverage for terminal Flex failures over HTTP, SSE, and WebSocket, HTTP 504 status preservation after a stream retry, and protocol serialization. Update Guardian retry and sampler recovery tests to cover Flex failures.

GitOrigin-RevId: f2e03a6d2d36a6d9748c35952e40397f8670d121
This commit is contained in:
Steve Coffey
2026-09-24 23:33:38 +00:00
committed by copyberry
parent 8f103417ef
commit dafb133c5b
39 changed files with 259 additions and 11 deletions
+8
View File
@@ -1,5 +1,6 @@
use crate::TransportError;
use crate::error::ApiError;
use crate::error::parse_flex_unavailable;
use crate::rate_limits::parse_promo_message;
use crate::rate_limits::parse_rate_limit_for_limit;
use crate::rate_limits::parse_rate_limit_reached_type;
@@ -30,6 +31,7 @@ pub fn map_api_error(err: ApiError) -> CodexErr {
| ApiError::ContextWindowExceeded
| ApiError::QuotaExceeded
| ApiError::UsageNotIncluded
| ApiError::FlexUnavailable
| ApiError::RateLimit(_)
| ApiError::InvalidRequest { .. }
| ApiError::InvalidPrompt { .. }
@@ -55,6 +57,7 @@ fn map_api_error_details(err: ApiError) -> CodexErr {
}
ApiError::Stream(msg) => CodexErr::Stream(msg),
ApiError::ServerOverloaded { .. } => CodexErr::ServerOverloaded,
ApiError::FlexUnavailable => CodexErr::new(CodexErrorDetails::FlexUnavailable),
ApiError::Api { status, message } => {
let user_message = api_error_user_message(status, &message);
CodexErr::UnexpectedStatus(UnexpectedResponseError {
@@ -174,6 +177,11 @@ fn map_api_error_details(err: ApiError) -> CodexErr {
} else if status == http::StatusCode::INTERNAL_SERVER_ERROR {
CodexErr::InternalServerError
} else if status == http::StatusCode::TOO_MANY_REQUESTS {
if let Ok(body) = serde_json::from_str::<Value>(&body_text)
&& let Some(error) = body.get("error").and_then(parse_flex_unavailable)
{
return map_api_error(error);
}
if let Ok(err) = serde_json::from_str::<UsageErrorResponse>(&body_text) {
if err.error.error_type.as_deref() == Some("usage_limit_reached") {
let limit_id = extract_header(headers.as_ref(), ACTIVE_LIMIT_HEADER);
+7 -1
View File
@@ -64,7 +64,13 @@ fn map_api_error_distinguishes_capacity_from_slow_down() {
false,
),
("slow_down", CodexErrorInfo::RateLimitExceeded, true),
("unknown_error", CodexErrorInfo::Other, true),
(
"unknown_error",
CodexErrorInfo::HttpConnectionFailed {
http_status_code: Some(503),
},
true,
),
] {
let err = map_api_error(ApiError::Transport(TransportError::Http {
retry_after: None,
+8
View File
@@ -3,6 +3,7 @@ use codex_client::TransportError;
use codex_http_client::RetryAfter;
use codex_protocol::protocol::MisalignmentErrorDetails;
use http::StatusCode;
use serde_json::Value;
use thiserror::Error;
#[derive(Debug, Error)]
@@ -44,6 +45,8 @@ pub enum ApiError {
message: String,
misalignment: Option<MisalignmentErrorDetails>,
},
#[error("Flex capacity unavailable.")]
FlexUnavailable,
#[error("server overloaded")]
ServerOverloaded { retry_after: Option<RetryAfter> },
}
@@ -53,3 +56,8 @@ impl From<RateLimitError> for ApiError {
Self::RateLimit(err.to_string())
}
}
pub(crate) fn parse_flex_unavailable(error: &Value) -> Option<ApiError> {
(error.get("code").and_then(Value::as_str) == Some("flex_unavailable"))
.then_some(ApiError::FlexUnavailable)
}
+16 -1
View File
@@ -3,6 +3,7 @@ use crate::common::ResponseStream;
use crate::common::SafetyBuffering;
use crate::common::SafetyBufferingTreatment;
use crate::error::ApiError;
use crate::error::parse_flex_unavailable;
use crate::rate_limits::parse_all_rate_limits;
use crate::safety_buffering::treatment_from_headers;
use crate::telemetry::SseTelemetry;
@@ -173,6 +174,7 @@ pub struct ResponsesStreamEvent {
pub(crate) headers: Option<Value>,
metadata: Option<Value>,
response: Option<Value>,
error: Option<Value>,
item: Option<Value>,
item_id: Option<String>,
call_id: Option<String>,
@@ -355,6 +357,11 @@ pub fn process_responses_event(
event: ResponsesStreamEvent,
) -> std::result::Result<Option<ResponseEvent>, ResponsesEventError> {
match event.kind.as_str() {
"error" => {
if let Some(error) = event.error.as_ref().and_then(parse_flex_unavailable) {
return Err(ResponsesEventError::Api(error));
}
}
"response.output_item.done" => {
if let Some(item_val) = event.item {
if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
@@ -418,6 +425,9 @@ 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>(error.clone())
{
@@ -697,7 +707,12 @@ async fn process_sse_with_treatment(
}
Ok(None) => {}
Err(error) => {
response_error = Some(error.into_api_error());
let error = error.into_api_error();
if matches!(error, ApiError::FlexUnavailable) {
let _ = tx_event.send(Err(error)).await;
return;
}
response_error = Some(error);
}
};
}