publish the Codex quota reading clients warn on

src/providers/codex/events.rs

@@ -267,6 +267,122 @@ pub(crate) fn classify_event_failure(payload: &Value) -> Option<CodexEventFailur
     })
 }
 
+/// Which upstream quota window ran out. Codex reports a 300 minute primary
+/// window and a 10080 minute secondary one, which line up with the session and
+/// weekly windows the Anthropic rate limit headers describe.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum CodexLimitWindow {
+    FiveHour,
+    SevenDay,
+}
+
+impl CodexLimitWindow {
+    pub(crate) fn claim(self) -> &'static str {
+        match self {
+            CodexLimitWindow::FiveHour => "five_hour",
+            CodexLimitWindow::SevenDay => "seven_day",
+        }
+    }
+}
+
+/// Quota exhaustion reported by Codex, together with the reset clock upstream
+/// sends alongside it. Unlike a transient rate limit this does not clear on a
+/// backoff, so the reset time is the only useful thing to report.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct CodexUsageLimit {
+    pub message: String,
+    pub resets_at: Option<u64>,
+    pub window: Option<CodexLimitWindow>,
+}
+
+/// Recognise the `usage_limit_reached` error Codex emits when a subscription
+/// window is spent. Upstream puts the clock both in the error body
+/// (`resets_at`, `resets_in_seconds`) and in `X-Codex-*` headers mirrored into
+/// the event payload.
+pub(crate) fn usage_limit_from_event(payload: &Value) -> Option<CodexUsageLimit> {
+    if !matches!(
+        payload.get("type").and_then(Value::as_str),
+        Some("response.failed" | "response.error" | "error")
+    ) {
+        return None;
+    }
+    let error = event_error(payload)?;
+
+    let resets_at = numeric_value(error.get("resets_at"));
+    let resets_in_seconds = numeric_value(error.get("resets_in_seconds"));
+    let is_usage_limit = error.get("type").and_then(Value::as_str) == Some("usage_limit_reached")
+        || (numeric_status(payload) == Some(429)
+            && (resets_at.is_some() || resets_in_seconds.is_some()));
+    if !is_usage_limit {
+        return None;
+    }
+
+    let limiting = limiting_window(payload, resets_in_seconds);
+    let resets_at = resets_at.or_else(|| {
+        let (prefix, _) = limiting?;
+        header_number(payload, &format!("X-Codex-{prefix}-Reset-At"))
+    });
+
+    Some(CodexUsageLimit {
+        message: error
+            .get("message")
+            .and_then(Value::as_str)
+            .unwrap_or("Usage limit reached")
+            .to_string(),
+        resets_at,
+        window: limiting.map(|(_, window)| window),
+    })
+}
+
+/// Codex sends the clock for both windows on every limit error, so the one that
+/// actually ran out is the one whose countdown matches the error's own. Returns
+/// the header prefix naming that window along with the window itself.
+fn limiting_window(
+    payload: &Value,
+    resets_in_seconds: Option<u64>,
+) -> Option<(&'static str, CodexLimitWindow)> {
+    let primary = header_number(payload, "X-Codex-Primary-Reset-After-Seconds");
+    let secondary = header_number(payload, "X-Codex-Secondary-Reset-After-Seconds");
+    let secondary_is_limiting = match (resets_in_seconds, primary, secondary) {
+        (Some(actual), Some(primary), Some(secondary)) => {
+            actual.abs_diff(secondary) < actual.abs_diff(primary)
+        }
+        (_, None, Some(_)) => true,
+        _ => false,
+    };
+    let prefix = if secondary_is_limiting {
+        "Secondary"
+    } else {
+        "Primary"
+    };
+
+    // A 300 minute window is the five hour one; anything longer is the weekly.
+    let minutes = header_number(payload, &format!("X-Codex-{prefix}-Window-Minutes"))?;
+    let window = if minutes <= 360 {
+        CodexLimitWindow::FiveHour
+    } else {
+        CodexLimitWindow::SevenDay
+    };
+    Some((prefix, window))
+}
+
+fn header_number(payload: &Value, name: &str) -> Option<u64> {
+    payload
+        .get("headers")?
+        .as_object()?
+        .iter()
+        .find(|(key, _)| key.eq_ignore_ascii_case(name))
+        .and_then(|(_, value)| numeric_value(Some(value)))
+}
+
+fn numeric_value(value: Option<&Value>) -> Option<u64> {
+    match value? {
+        Value::Number(number) => number.as_u64(),
+        Value::String(raw) => raw.parse().ok(),
+        _ => None,
+    }
+}
+
 pub(crate) fn first_retryable_failure(body: &[u8]) -> Option<CodexEventFailure> {
     first_event_failure(body).filter(CodexEventFailure::retryable)
 }
@@ -336,6 +452,81 @@ fn retryable_message(message: &str) -> bool {
 mod tests {
     use super::*;
 
+    /// Shape recorded from a live turn that exhausted the five hour window: the
+    /// clock arrives both in the error body and in the mirrored `X-Codex-*`
+    /// headers, and the primary window is the one that ran out.
+    fn spent_five_hour_window() -> Value {
+        serde_json::json!({
+            "type": "error",
+            "status_code": 429,
+            "error": {
+                "type": "usage_limit_reached",
+                "message": "The usage limit has been reached",
+                "plan_type": "plus",
+                "resets_at": 1788879437u64,
+                "resets_in_seconds": 9568u64
+            },
+            "headers": {
+                "X-Codex-Primary-Used-Percent": "100",
+                "X-Codex-Primary-Window-Minutes": "300",
+                "X-Codex-Primary-Reset-After-Seconds": "9569",
+                "X-Codex-Primary-Reset-At": "1788879438",
+                "X-Codex-Secondary-Used-Percent": "16",
+                "X-Codex-Secondary-Window-Minutes": "10080",
+                "X-Codex-Secondary-Reset-After-Seconds": "596369",
+                "X-Codex-Secondary-Reset-At": "1789466238"
+            }
+        })
+    }
+
+    #[test]
+    fn reads_usage_limit_reset_clock() {
+        let limit = usage_limit_from_event(&spent_five_hour_window()).expect("usage limit");
+        assert_eq!(limit.message, "The usage limit has been reached");
+        assert_eq!(limit.resets_at, Some(1788879437));
+        assert_eq!(limit.window, Some(CodexLimitWindow::FiveHour));
+        assert_eq!(limit.window.unwrap().claim(), "five_hour");
+    }
+
+    #[test]
+    fn attributes_the_window_whose_clock_matches() {
+        let mut payload = spent_five_hour_window();
+        // Same error, but it is the weekly window that ran out.
+        payload["error"]["resets_in_seconds"] = serde_json::json!(596_368u64);
+        let limit = usage_limit_from_event(&payload).expect("usage limit");
+        assert_eq!(limit.window, Some(CodexLimitWindow::SevenDay));
+    }
+
+    #[test]
+    fn falls_back_to_header_clock_when_body_omits_it() {
+        let mut payload = spent_five_hour_window();
+        payload["error"]
+            .as_object_mut()
+            .unwrap()
+            .remove("resets_at");
+        let limit = usage_limit_from_event(&payload).expect("usage limit");
+        assert_eq!(limit.resets_at, Some(1788879438));
+    }
+
+    #[test]
+    fn ignores_errors_that_are_not_usage_limits() {
+        assert!(
+            usage_limit_from_event(&serde_json::json!({
+                "type": "error",
+                "status_code": 429,
+                "error": {"type": "rate_limit_exceeded", "message": "slow down"}
+            }))
+            .is_none()
+        );
+        assert!(
+            usage_limit_from_event(&serde_json::json!({
+                "type": "response.output_text.delta",
+                "delta": "hello"
+            }))
+            .is_none()
+        );
+    }
+
     #[test]
     fn classifies_retryable_failure_kinds() {
         let overload = classify_event_failure(&serde_json::json!({

src/providers/codex/mod.rs

@@ -7,6 +7,7 @@ pub mod count_tokens;
 pub(crate) mod events;
 pub mod images;
 pub mod native;
+pub(crate) mod rate_limits;
 pub mod request_summary;
 pub mod search;
 pub mod transcription;
@@ -18,7 +19,7 @@ use axum::Json;
 use axum::body::Body;
 use axum::response::{IntoResponse, Response};
 use bytes::Bytes;
-use http::StatusCode;
+use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -791,7 +792,7 @@ async fn live_stream_response(
         {
             LiveStreamStart::Response(response) => {
                 cleanup.disarm();
-                return response;
+                return with_rate_limit_headers(response);
             }
             LiveStreamStart::Retry {
                 error,
@@ -885,11 +886,23 @@ async fn live_stream_response_once(
             }
             generation_started = true;
         }
+        rate_limits::observe_event(&payload);
         append_upstream_sse_payload(&mut upstream_sse_body, &payload);
         let (chunk, terminal) = match translate_live_stream_payload(&mut translator, &payload, None)
         {
             Ok(result) => result,
             Err(message) => {
+                // A spent subscription window reopens hours from now, so the
+                // retry budget can only burn the request down to the same 429.
+                // Report it once, with the reset clock upstream supplied.
+                if let Some(limit) = events::usage_limit_from_event(&payload) {
+                    abort_request_state(
+                        ctx.session_id.as_deref(),
+                        &request_continuation,
+                        compaction.attempt,
+                    );
+                    return LiveStreamStart::Response(usage_limit_response(&limit));
+                }
                 if let Some(failure) = events::classify_event_failure(&payload) {
                     if failure.retryable() {
                         return provider_retry(
@@ -1106,6 +1119,7 @@ fn remaining_live_stream_response(
             };
             match item {
                 Ok(payload) => {
+                    rate_limits::observe_event(&payload);
                     append_upstream_sse_payload(&mut upstream_sse_body, &payload);
                     let (chunk, terminal) = match translate_live_stream_payload(
                         &mut translator,
@@ -1405,6 +1419,76 @@ fn update_continuation_from_upstream(
 // Error mapping
 // ---------------------------------------------------------------------------
 
+/// Attach the newest Codex quota reading to a response, so a client can warn
+/// its user before the allowance runs out rather than only when it has.
+///
+/// A response that already states a rate limit status is left alone: a refusal
+/// carries the exact state of the window that refused it, which is better than
+/// a reading taken earlier in the turn.
+fn with_rate_limit_headers(mut response: Response) -> Response {
+    if response
+        .headers()
+        .contains_key("anthropic-ratelimit-unified-status")
+    {
+        return response;
+    }
+    let Some(snapshot) = rate_limits::latest() else {
+        return response;
+    };
+
+    let headers = response.headers_mut();
+    for (name, value) in snapshot.headers() {
+        if let Ok(name) = HeaderName::from_bytes(name.as_bytes())
+            && let Ok(value) = HeaderValue::from_str(&value)
+        {
+            headers.insert(name, value);
+        }
+    }
+    response
+}
+
+/// Answer a spent subscription window with the rate limit headers the client
+/// reads, so it can name the exhausted window and show when it reopens.
+///
+/// `Retry-After` is deliberately absent: clients sleep for its full value, and
+/// here that is hours. `x-should-retry: false` stops the retry loop instead,
+/// which is the honest signal — a spent window does not reopen on a backoff.
+fn usage_limit_response(limit: &events::CodexUsageLimit) -> Response {
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        HeaderName::from_static("x-should-retry"),
+        HeaderValue::from_static("false"),
+    );
+    headers.insert(
+        HeaderName::from_static("anthropic-ratelimit-unified-status"),
+        HeaderValue::from_static("rejected"),
+    );
+    if let Some(resets_at) = limit.resets_at
+        && let Ok(value) = HeaderValue::from_str(&resets_at.to_string())
+    {
+        headers.insert(
+            HeaderName::from_static("anthropic-ratelimit-unified-reset"),
+            value,
+        );
+    }
+    if let Some(window) = limit.window {
+        headers.insert(
+            HeaderName::from_static("anthropic-ratelimit-unified-representative-claim"),
+            HeaderValue::from_static(window.claim()),
+        );
+    }
+
+    (
+        headers,
+        json_error(
+            StatusCode::TOO_MANY_REQUESTS,
+            "rate_limit_error",
+            &limit.message,
+        ),
+    )
+        .into_response()
+}
+
 fn map_codex_error_to_response(err: &client::CodexError) -> Response {
     let message = codex_error_message(err);
     if is_context_window_overflow(message) {
@@ -1598,6 +1682,48 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn a_refusal_keeps_the_state_of_the_window_that_refused_it() {
+        // A reading from earlier in the turn says the allowance was fine, and
+        // both windows are still running so it survives to the response.
+        rate_limits::observe_event(&serde_json::json!({
+            "type": "codex.rate_limits",
+            "rate_limits": {
+                "primary": {
+                    "used_percent": 5.0,
+                    "window_minutes": 300,
+                    "reset_at": 4_000_000_000u64
+                },
+                "secondary": {
+                    "used_percent": 5.0,
+                    "window_minutes": 10080,
+                    "reset_at": 4_000_000_001u64
+                }
+            }
+        }));
+        assert!(rate_limits::latest().is_some(), "reading must be live");
+
+        let refusal = with_rate_limit_headers(usage_limit_response(&events::CodexUsageLimit {
+            message: "The usage limit has been reached".to_string(),
+            resets_at: Some(1788879437),
+            window: Some(events::CodexLimitWindow::FiveHour),
+        }));
+
+        let headers = refusal.headers();
+        assert_eq!(
+            headers
+                .get("anthropic-ratelimit-unified-status")
+                .and_then(|value| value.to_str().ok()),
+            Some("rejected"),
+        );
+        assert_eq!(
+            headers
+                .get("anthropic-ratelimit-unified-reset")
+                .and_then(|value| value.to_str().ok()),
+            Some("1788879437"),
+        );
+    }
+
     fn live_test_request(text: &str) -> translate::request::ResponsesRequest {
         translate::request::ResponsesRequest {
             model: "gpt-5.6-sol".to_string(),