Handle budget_reached as a terminal session budget error (#59228)

## What changed

Map provider `budget_reached` errors to `SessionBudgetExceeded` for HTTP 400 responses and streaming failures. End SSE streams immediately on this error, ignoring retry headers and later completion events, and display `budget reached` in the TUI.

## Testing

Add coverage for HTTP error classification, immediate SSE termination without retries, WebSocket error delivery and subsequent request recovery, and TUI error rendering and task-state cleanup.

GitOrigin-RevId: 6a62d699cdb9cb446214cacfacd0c9ed68bf21a6

codex-rs/codex-api/src/api_bridge.rs

@@ -34,6 +34,7 @@ pub fn map_api_error(err: ApiError) -> CodexErr {
         | ApiError::ContextWindowExceeded
         | ApiError::QuotaExceeded
         | ApiError::UsageNotIncluded
+        | ApiError::SessionBudgetExceeded
         | ApiError::FlexUnavailable
         | ApiError::RateLimit(_)
         | ApiError::InvalidRequest { .. }
@@ -56,6 +57,7 @@ fn map_api_error_details(err: ApiError) -> CodexErr {
         ApiError::ContextWindowExceeded => CodexErr::ContextWindowExceeded,
         ApiError::QuotaExceeded => CodexErr::QuotaExceeded,
         ApiError::UsageNotIncluded => CodexErr::UsageNotIncluded,
+        ApiError::SessionBudgetExceeded => CodexErr::SessionBudgetExceeded,
         ApiError::Retryable { message, .. } => CodexErr::Stream(message),
         ApiError::RateLimitExceeded { message, .. } => {
             CodexErr::new(CodexErrorDetails::RateLimitExceeded(message))

codex-rs/codex-api/src/api_bridge_tests.rs

@@ -184,6 +184,12 @@ fn map_http_error_classifies_provider_codes() {
             true,
         ),
         (400, "model_not_found", ModelNotFound, false),
+        (
+            400,
+            "budget_reached",
+            CodexErrorInfo::SessionBudgetExceeded,
+            false,
+        ),
         (403, "model_not_found", ModelNotFound, false),
         (404, "model_not_found", ModelNotFound, false),
         (

codex-rs/codex-api/src/error.rs

@@ -25,6 +25,8 @@ pub enum ApiError {
     QuotaExceeded,
     #[error("usage not included")]
     UsageNotIncluded,
+    #[error("session budget exceeded")]
+    SessionBudgetExceeded,
     #[error("retryable error: {message}")]
     Retryable {
         message: String,
@@ -79,6 +81,9 @@ pub(crate) fn parse_terminal_provider_error(
     let error = error?;
     let code = error.get("code")?.as_str()?;
     match (code, status) {
+        ("budget_reached", None | Some(StatusCode::BAD_REQUEST)) => {
+            Some(ApiError::SessionBudgetExceeded)
+        }
         (
             "audio_too_large"
             | "audio_too_short"

codex-rs/codex-api/src/sse/responses.rs

@@ -679,7 +679,9 @@ async fn process_sse_with_treatment(
                 let error = error.into_api_error();
                 if matches!(
                     error,
-                    ApiError::FlexUnavailable | ApiError::ResponseIncomplete(_)
+                    ApiError::FlexUnavailable
+                        | ApiError::SessionBudgetExceeded
+                        | ApiError::ResponseIncomplete(_)
                 ) {
                     let _ = tx_event.send(Err(error)).await;
                     return;

codex-rs/codex-api/src/sse/responses_error_tests.rs

@@ -109,48 +109,53 @@ async fn policy_capacity_classification_preserves_overload_retry_behavior() {
     }
 }
 
-/// Flex failures remain terminal even with retry headers and a later completed event.
+/// Terminal provider failures ignore retry headers and stop before a later completed event.
 #[tokio::test(start_paused = true)]
-async fn flex_failure_with_retry_header_ends_stream_immediately() {
-    let error = json!({
-        "code": "flex_unavailable",
-        "plan_type": 42,
-        "headers": {"retry-after": "300"}
-    });
-    for failure in [
-        json!({"type": "error", "error": error}),
-        json!({"type": "response.failed", "response": {"error": error}}),
+async fn terminal_failure_with_retry_header_ends_stream_immediately() {
+    for (code, expected) in [
+        ("flex_unavailable", CodexErrorInfo::FlexUnavailable),
+        ("budget_reached", CodexErrorInfo::SessionBudgetExceeded),
     ] {
-        let completed = json!({"type": "response.completed", "response": {"id": "later"}});
-        let bytes = stream::iter([Ok(Bytes::from(format!(
-            "data: {failure}\n\ndata: {completed}\n\n"
-        )))])
-        .chain(stream::pending());
-        let mut stream = spawn_response_stream(
-            StreamResponse {
-                status: StatusCode::OK,
-                headers: HeaderMap::new(),
-                bytes: Box::pin(bytes),
-            },
-            Duration::from_secs(60),
-            /*telemetry*/ None,
-            /*turn_state*/ None,
-        );
-        let error = loop {
-            match stream.next().await {
-                Some(Ok(ResponseEvent::RateLimits(_))) => {}
-                Some(Err(error)) => break map_api_error(error),
-                event => panic!("expected a Flex error, got {event:?}"),
-            }
-        };
-        assert_eq!(
-            (
-                error.to_codex_protocol_error(),
-                error.retry_delay(/*retry_count*/ 1)
-            ),
-            (CodexErrorInfo::FlexUnavailable, None)
-        );
-        assert!(stream.next().await.is_none());
+        let error = json!({
+            "code": code,
+            "plan_type": 42,
+            "headers": {"retry-after": "300"}
+        });
+        for failure in [
+            json!({"type": "error", "error": error}),
+            json!({"type": "response.failed", "response": {"error": error}}),
+        ] {
+            let completed = json!({"type": "response.completed", "response": {"id": "later"}});
+            let bytes = stream::iter([Ok(Bytes::from(format!(
+                "data: {failure}\n\ndata: {completed}\n\n"
+            )))])
+            .chain(stream::pending());
+            let mut stream = spawn_response_stream(
+                StreamResponse {
+                    status: StatusCode::OK,
+                    headers: HeaderMap::new(),
+                    bytes: Box::pin(bytes),
+                },
+                Duration::from_secs(60),
+                /*telemetry*/ None,
+                /*turn_state*/ None,
+            );
+            let error = loop {
+                match stream.next().await {
+                    Some(Ok(ResponseEvent::RateLimits(_))) => {}
+                    Some(Err(error)) => break map_api_error(error),
+                    event => panic!("expected a terminal error for {code}, got {event:?}"),
+                }
+            };
+            assert_eq!(
+                (
+                    error.to_codex_protocol_error(),
+                    error.retry_delay(/*retry_count*/ 1)
+                ),
+                (expected.clone(), None)
+            );
+            assert!(stream.next().await.is_none());
+        }
     }
 }