From be6474745cc193f13e24baa1030e50359512dcdb Mon Sep 17 00:00:00 2001 From: LeonLewis <52706516+null-topology@users.noreply.github.com> Date: Tue, 8 Sep 2026 21:30:04 +0200 Subject: [PATCH 1/2] surface Codex usage limits instead of retrying them A spent subscription window arrives as an `error` event carrying `usage_limit_reached` and the moment the window reopens. Nothing read it: the failure classified as a retryable rate limit, and the delay lookup covers `retry_after`, `retry_after_seconds` and `headers.retry-after`, none of which this event carries. With no delay the live stream spent its whole budget -- eleven upstream attempts over 2m54s in a capture of the failure, each answered 429 in under a second -- and then returned 429 with no reset information, so clients retried immediately against a window that reopens hours later. Recognise the event and answer it once, with the headers clients already read for rate limit state: which window ran out and when it reopens. Retry-After is deliberately absent, since clients sleep for its full value and here that is hours; x-should-retry stops the retry loop instead. --- src/providers/codex/events.rs | 191 ++++++++++++++++++++++++++++++++++ src/providers/codex/mod.rs | 55 +++++++++- 2 files changed, 245 insertions(+), 1 deletion(-) diff --git a/src/providers/codex/events.rs b/src/providers/codex/events.rs index 56b220ee..f400845f 100644 --- a/src/providers/codex/events.rs +++ b/src/providers/codex/events.rs @@ -267,6 +267,122 @@ pub(crate) fn classify_event_failure(payload: &Value) -> Option &'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, + pub window: Option, +} + +/// 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 { + 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, +) -> 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 { + 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 { + match value? { + Value::Number(number) => number.as_u64(), + Value::String(raw) => raw.parse().ok(), + _ => None, + } +} + pub(crate) fn first_retryable_failure(body: &[u8]) -> Option { 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!({ diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 450de58d..435757fe 100644 --- a/src/providers/codex/mod.rs +++ b/src/providers/codex/mod.rs @@ -18,7 +18,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}; @@ -890,6 +890,17 @@ async fn live_stream_response_once( { 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( @@ -1405,6 +1416,48 @@ fn update_continuation_from_upstream( // Error mapping // --------------------------------------------------------------------------- +/// 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) { From 446ee7a15fbfeaed0d02494f54d87fdbfb80efe9 Mon Sep 17 00:00:00 2001 From: LeonLewis <52706516+null-topology@users.noreply.github.com> Date: Tue, 8 Sep 2026 21:45:05 +0200 Subject: [PATCH 2/2] publish the Codex quota reading clients warn on Codex reports how much of each subscription window is spent as a `codex.rate_limits` event inside the stream, and nothing forwarded it. The only quota signal reaching a client was the refusal itself, so work stopped with no warning that the allowance was running low. Keep the newest reading and put it on the response, in the headers clients already parse for rate limit state: the spent fraction and reset time of each window, and which one is closest to running out. A window is identified by the length it reports rather than the slot it arrived in, since which rank carries which length has changed between payload versions, and both spellings of the reset field are read for the same reason. The spent fraction only reaches a caller once a threshold is declared surpassed, so a window past its line says which line it crossed. The defaults -- 0.9 for the session window, 0.75 for the weekly -- are the points Anthropic's own windows warn at; CCP_CODEX_QUOTA_WARN_AT lowers both for a consumer that wants the figure earlier. A reading whose window has since reopened is dropped rather than published, so the turn after a reset does not announce an allowance as nearly gone at the moment it came back. A refusal keeps its own headers: it carries the exact state of the window that refused it. --- src/providers/codex/mod.rs | 75 +++++- src/providers/codex/rate_limits.rs | 392 +++++++++++++++++++++++++++++ 2 files changed, 466 insertions(+), 1 deletion(-) create mode 100644 src/providers/codex/rate_limits.rs diff --git a/src/providers/codex/mod.rs b/src/providers/codex/mod.rs index 435757fe..66af93b8 100644 --- a/src/providers/codex/mod.rs +++ b/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; @@ -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,6 +886,7 @@ 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) { @@ -1117,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, @@ -1416,6 +1419,34 @@ 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. /// @@ -1651,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(), diff --git a/src/providers/codex/rate_limits.rs b/src/providers/codex/rate_limits.rs new file mode 100644 index 00000000..c5131247 --- /dev/null +++ b/src/providers/codex/rate_limits.rs @@ -0,0 +1,392 @@ +//! Publishing Codex quota state in the vocabulary Anthropic clients read. +//! +//! Codex reports how much of each subscription window is spent inside the +//! stream, as a `codex.rate_limits` event, while Anthropic clients expect that +//! state in response headers. The event arrives early — ahead of the first +//! content event — so the reading is usually ready in time for the response it +//! came from; when it is not, it travels with the next one, a gauge a turn +//! stale. +//! +//! Without this the only quota signal a client ever sees is the refusal, and a +//! limit reached with no warning beforehand is exactly the case where nobody +//! knows why the assistant stopped answering. + +use std::sync::{Mutex, OnceLock}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde_json::Value; + +/// Codex reports `used_percent` on a 0..100 scale; the headers carry a ratio. +const PERCENT: f64 = 100.0; + +/// A window is the five hour one when it is no longer than this. Codex reports +/// 300 minutes for it (299 in older payloads), against 10080 for the weekly. +const FIVE_HOUR_MAX_MINUTES: u64 = 360; + +/// Where each window starts being worth announcing. These mirror the thresholds +/// clients apply to Anthropic's own windows, so a Codex session warns at the +/// same points a Claude session does. +const FIVE_HOUR_WARN_AT: f64 = 0.9; +const SEVEN_DAY_WARN_AT: f64 = 0.75; + +/// Lowers both thresholds, for a consumer that wants the spent fraction earlier +/// than the defaults publish it — the fraction only reaches a caller once a +/// threshold is declared surpassed. Interfaces that draw their own warning have +/// a floor of their own well above zero, so a low setting here feeds a watcher +/// without turning the terminal noisy. +const WARN_AT_ENV: &str = "CCP_CODEX_QUOTA_WARN_AT"; + +fn warn_at(default: f64) -> f64 { + static OVERRIDE: OnceLock> = OnceLock::new(); + OVERRIDE + .get_or_init(|| { + std::env::var(WARN_AT_ENV) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|value| (0.0..=1.0).contains(value)) + }) + .unwrap_or(default) +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) struct WindowState { + /// Spent fraction of the window. Can exceed 1.0: usage legitimately runs + /// past a cap before the refusal lands. + pub utilization: f64, + pub resets_at: u64, +} + +impl WindowState { + fn surpassed(&self, threshold: f64) -> bool { + self.utilization >= threshold + } +} + +#[derive(Debug, Clone, Default, PartialEq)] +pub(crate) struct Snapshot { + pub five_hour: Option, + pub seven_day: Option, +} + +impl Snapshot { + /// The window a client should name when it mentions one: whichever is + /// closest to running out, since that is the one that will stop the work. + fn representative(&self) -> Option<(&'static str, WindowState)> { + match (self.five_hour, self.seven_day) { + (Some(five), Some(seven)) if seven.utilization > five.utilization => { + Some(("seven_day", seven)) + } + (Some(five), _) => Some(("five_hour", five)), + (None, Some(seven)) => Some(("seven_day", seven)), + (None, None) => None, + } + } + + /// The headers that carry this snapshot, as name/value pairs. + pub(crate) fn headers(&self) -> Vec<(&'static str, String)> { + let mut out = Vec::new(); + let Some((claim, representative)) = self.representative() else { + return out; + }; + + out.push(("anthropic-ratelimit-unified-status", "allowed".to_string())); + out.push(( + "anthropic-ratelimit-unified-reset", + representative.resets_at.to_string(), + )); + out.push(( + "anthropic-ratelimit-unified-representative-claim", + claim.to_string(), + )); + + for (window, threshold, utilization_header, reset_header, threshold_header) in [ + ( + self.five_hour, + FIVE_HOUR_WARN_AT, + "anthropic-ratelimit-unified-5h-utilization", + "anthropic-ratelimit-unified-5h-reset", + "anthropic-ratelimit-unified-5h-surpassed-threshold", + ), + ( + self.seven_day, + SEVEN_DAY_WARN_AT, + "anthropic-ratelimit-unified-7d-utilization", + "anthropic-ratelimit-unified-7d-reset", + "anthropic-ratelimit-unified-7d-surpassed-threshold", + ), + ] { + let Some(state) = window else { continue }; + let threshold = warn_at(threshold); + out.push((utilization_header, format!("{:.4}", state.utilization))); + out.push((reset_header, state.resets_at.to_string())); + // Clients report the spent fraction to their caller only once a + // threshold is declared surpassed, so a window worth warning about + // has to say which line it crossed. + if state.surpassed(threshold) { + out.push((threshold_header, format!("{threshold:.2}"))); + } + } + + out + } +} + +static LATEST: OnceLock>> = OnceLock::new(); + +fn cell() -> &'static Mutex> { + LATEST.get_or_init(|| Mutex::new(None)) +} + +/// Record the quota state carried by a `codex.rate_limits` event. Any other +/// event is ignored, so this can be handed every frame off the stream. +pub(crate) fn observe_event(payload: &Value) { + if payload.get("type").and_then(Value::as_str) != Some("codex.rate_limits") { + return; + } + let Some(snapshot) = snapshot_from_event(payload) else { + return; + }; + if let Ok(mut latest) = cell().lock() { + *latest = Some(snapshot); + } +} + +/// The most recent snapshot, with any window that has since rolled over left +/// out. +/// +/// A reading can outlive the window it describes — nothing new arrives while a +/// process sits idle — and the turn after a window reopens would then carry the +/// spent figure from before the reset, announcing an allowance as nearly gone +/// at the moment it came back. A window whose reset time has passed says +/// nothing about the one now running. +pub(crate) fn latest() -> Option { + let snapshot = cell().lock().ok().and_then(|latest| latest.clone())?; + still_running(snapshot, now()) +} + +fn still_running(snapshot: Snapshot, now: u64) -> Option { + let fresh = |window: Option| window.filter(|state| state.resets_at > now); + let snapshot = Snapshot { + five_hour: fresh(snapshot.five_hour), + seven_day: fresh(snapshot.seven_day), + }; + + (snapshot != Snapshot::default()).then_some(snapshot) +} + +fn snapshot_from_event(payload: &Value) -> Option { + let limits = payload.get("rate_limits")?; + let mut snapshot = Snapshot::default(); + // Codex names the windows by rank rather than by length, and which rank + // holds which length has changed between payload versions — so the window + // is decided by the length it reports, not by the slot it arrived in. + for slot in ["primary", "secondary"] { + let Some(window) = limits.get(slot) else { + continue; + }; + let Some((minutes, state)) = window_state(window) else { + continue; + }; + if minutes <= FIVE_HOUR_MAX_MINUTES { + snapshot.five_hour = Some(state); + } else { + snapshot.seven_day = Some(state); + } + } + + (snapshot != Snapshot::default()).then_some(snapshot) +} + +fn window_state(window: &Value) -> Option<(u64, WindowState)> { + let minutes = number(window.get("window_minutes"))? as u64; + let used_percent = number(window.get("used_percent"))?; + // The stream names the moment `reset_at`; the same window reserialised into + // a Codex CLI session log calls it `resets_at`. Both spellings are read so + // a payload from either side parses. + let resets_at = match first(window, &["reset_at", "resets_at"]) { + Some(at) => at as u64, + // Some payloads count down instead of naming the moment. + None => now() + first(window, &["reset_after_seconds", "resets_in_seconds"])? as u64, + }; + + Some(( + minutes, + WindowState { + utilization: used_percent / PERCENT, + resets_at, + }, + )) +} + +fn first(window: &Value, names: &[&str]) -> Option { + names.iter().find_map(|name| number(window.get(*name))) +} + +fn number(value: Option<&Value>) -> Option { + match value? { + Value::Number(number) => number.as_f64(), + Value::String(raw) => raw.parse().ok(), + _ => None, + } +} + +fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_secs()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Shape recorded off a live stream, down to the field names. + fn event(primary_percent: f64, secondary_percent: f64) -> Value { + serde_json::json!({ + "type": "codex.rate_limits", + "plan_type": "plus", + "credits": {"balance": "0", "has_credits": false, "unlimited": false}, + "rate_limits": { + "allowed": true, + "limit_reached": false, + "primary": { + "used_percent": primary_percent, + "window_minutes": 300, + "reset_after_seconds": 16394, + "reset_at": 1788879437u64 + }, + "secondary": { + "used_percent": secondary_percent, + "window_minutes": 10080, + "reset_after_seconds": 584857, + "reset_at": 1789466238u64 + } + } + }) + } + + fn headers_of(payload: &Value) -> std::collections::HashMap<&'static str, String> { + snapshot_from_event(payload) + .expect("snapshot") + .headers() + .into_iter() + .collect() + } + + #[test] + fn maps_windows_by_their_length() { + let headers = headers_of(&event(42.0, 16.0)); + assert_eq!( + headers["anthropic-ratelimit-unified-5h-utilization"], + "0.4200" + ); + assert_eq!( + headers["anthropic-ratelimit-unified-5h-reset"], + "1788879437" + ); + assert_eq!( + headers["anthropic-ratelimit-unified-7d-utilization"], + "0.1600" + ); + assert_eq!(headers["anthropic-ratelimit-unified-status"], "allowed"); + } + + #[test] + fn stays_quiet_below_the_thresholds() { + let headers = headers_of(&event(42.0, 16.0)); + assert!(!headers.contains_key("anthropic-ratelimit-unified-5h-surpassed-threshold")); + assert!(!headers.contains_key("anthropic-ratelimit-unified-7d-surpassed-threshold")); + } + + #[test] + fn declares_the_threshold_a_window_crossed() { + let headers = headers_of(&event(93.0, 16.0)); + assert_eq!( + headers["anthropic-ratelimit-unified-5h-surpassed-threshold"], + "0.90" + ); + assert!(!headers.contains_key("anthropic-ratelimit-unified-7d-surpassed-threshold")); + + let weekly = headers_of(&event(10.0, 80.0)); + assert_eq!( + weekly["anthropic-ratelimit-unified-7d-surpassed-threshold"], + "0.75" + ); + } + + #[test] + fn names_the_window_closest_to_running_out() { + let headers = headers_of(&event(10.0, 80.0)); + assert_eq!( + headers["anthropic-ratelimit-unified-representative-claim"], + "seven_day" + ); + assert_eq!(headers["anthropic-ratelimit-unified-reset"], "1789466238"); + + let five = headers_of(&event(80.0, 10.0)); + assert_eq!( + five["anthropic-ratelimit-unified-representative-claim"], + "five_hour" + ); + } + + #[test] + fn accepts_the_spelling_a_session_log_uses() { + // The same window, reserialised by the Codex CLI into its rollout file. + let payload = serde_json::json!({ + "type": "codex.rate_limits", + "rate_limits": { + "primary": { + "used_percent": 5.0, + "window_minutes": 299, + "resets_at": 1788879437u64 + } + } + }); + let snapshot = snapshot_from_event(&payload).expect("snapshot"); + assert_eq!(snapshot.five_hour.expect("five hour").resets_at, 1788879437); + } + + #[test] + fn accepts_a_countdown_instead_of_a_moment() { + let payload = serde_json::json!({ + "type": "codex.rate_limits", + "rate_limits": { + "primary": { + "used_percent": 5.0, + "window_minutes": 299, + "resets_in_seconds": 17940 + } + } + }); + let snapshot = snapshot_from_event(&payload).expect("snapshot"); + let five = snapshot.five_hour.expect("five hour window"); + assert!(five.resets_at > now()); + assert!(five.resets_at <= now() + 17940); + } + + #[test] + fn drops_a_window_that_has_since_rolled_over() { + let spent = snapshot_from_event(&event(100.0, 80.0)).expect("snapshot"); + // Both windows reset long before this moment. + assert_eq!(still_running(spent.clone(), 1_800_000_000), None); + + // The weekly window is still running; the five hour one has rolled over. + let surviving = still_running(spent, 1_789_000_000).expect("weekly window survives"); + assert!(surviving.five_hour.is_none()); + assert!(surviving.seven_day.is_some()); + let headers: std::collections::HashMap<_, _> = surviving.headers().into_iter().collect(); + assert!(!headers.contains_key("anthropic-ratelimit-unified-5h-utilization")); + assert_eq!( + headers["anthropic-ratelimit-unified-representative-claim"], + "seven_day" + ); + } + + #[test] + fn ignores_frames_that_are_not_quota_telemetry() { + observe_event(&serde_json::json!({"type": "response.completed"})); + assert!(snapshot_from_event(&serde_json::json!({"type": "codex.rate_limits"})).is_none()); + } +}