forge_gateway/
anthropic.rs

1//! The Anthropic backend. The account's credential is its setup
2//! token, which probes a minimal billed `/v1/messages` call - the
3//! usage endpoint refuses setup tokens, while a `/v1/messages`
4//! response carries the `anthropic-ratelimit-unified-*` windows as
5//! headers, a 429 most of all. An account with no token has no
6//! credential at all.
7
8use std::collections::HashMap;
9
10use async_trait::async_trait;
11use reqwest::header::{AUTHORIZATION, CONTENT_TYPE, HeaderMap, HeaderValue, USER_AGENT};
12use serde_json::json;
13
14use forge_primitives::usage::oauth::OauthUsageError;
15use forge_primitives::usage::{UsageSnapshot, UsageSourceKind, UsageWindow};
16
17use crate::helpers::{OAUTH_TIMEOUT, parse_retry_after, redacted_body, system_time_from_epoch};
18use crate::{AccountEnv, BillingModel, ProbeError, Provider, ProviderBackend, ProviderHost};
19
20/// The variable the flat `token` key is stamped onto for an Anthropic
21/// account (minted by `claude setup-token`). Its presence makes the
22/// account token-mode: the token is the account's only credential.
23/// The value is trimmed once where it enters forge's config, so every
24/// reader sees the same credential.
25pub const CLAUDE_CODE_OAUTH_TOKEN_ENV: &str = "CLAUDE_CODE_OAUTH_TOKEN";
26
27/// The setup token `env` carries, when non-empty. An empty value stays
28/// None: whitespace is the config loader's to strip (once, at load),
29/// and this reads the credential as the child gets it.
30pub fn token_bearer<S: std::hash::BuildHasher>(env: &HashMap<String, String, S>) -> Option<&str> {
31    env.get(CLAUDE_CODE_OAUTH_TOKEN_ENV).map(String::as_str).filter(|token| !token.is_empty())
32}
33
34/// The Anthropic `[[accounts]] provider` token.
35pub struct Anthropic;
36
37#[async_trait]
38impl ProviderBackend for Anthropic {
39    fn token(&self) -> Provider {
40        Provider::Anthropic
41    }
42
43    fn billing(&self) -> BillingModel {
44        BillingModel::Windows
45    }
46
47    fn source(&self) -> UsageSourceKind {
48        UsageSourceKind::Oauth
49    }
50
51    async fn probe(
52        &self,
53        account: &AccountEnv<'_>,
54        host: &dyn ProviderHost,
55    ) -> Result<UsageSnapshot, ProbeError> {
56        let Some(bearer) = token_bearer(account.env) else {
57            return Err(ProbeError::NoCredentials);
58        };
59        let ua = host.user_agent().await.map_err(OauthUsageError::UaProbe)?;
60        let client = host.http_client(OAUTH_TIMEOUT).map_err(OauthUsageError::Network)?;
61        let headers =
62            messages_probe(&client, &ua, None, bearer).await.map_err(ProbeError::Fetch)?;
63        tracing::info!(
64            target: "forge_gateway::anthropic",
65            event_name = "unified_usage_probe_settled",
66            outcome = "ok",
67            "token account unified usage probe settled",
68        );
69        Ok(snapshot_from_unified_headers(&headers))
70    }
71}
72
73/// The token probe's model: the cheapest haiku-class ID; its retired
74/// predecessor 404s, and 404s carry no headers.
75const PROBE_MODEL: &str = "claude-haiku-4-5";
76
77const MESSAGES_URL: &str = "https://api.anthropic.com/v1/messages";
78const UNIFIED_PREFIX: &str = "anthropic-ratelimit-unified-";
79
80/// `/v1/messages` on the official host, or an override with any
81/// trailing slash trimmed. Production always passes `None`: the
82/// unified headers only ride the official base.
83fn messages_url(base_url: Option<&str>) -> String {
84    match base_url {
85        Some(base) => format!("{}/v1/messages", base.trim_end_matches('/')),
86        None => MESSAGES_URL.to_owned(),
87    }
88}
89
90fn messages_headers(user_agent: &str, access_token: &str) -> Result<HeaderMap, OauthUsageError> {
91    let mut headers = HeaderMap::new();
92    headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
93    headers.insert("anthropic-version", HeaderValue::from_static("2023-06-01"));
94    let ua = HeaderValue::from_str(user_agent)
95        .map_err(|error| OauthUsageError::Network(format!("bad UA header: {error}")))?;
96    headers.insert(USER_AGENT, ua);
97    let bearer = HeaderValue::from_str(&format!("Bearer {access_token}"))
98        .map_err(|error| OauthUsageError::Network(format!("bad bearer header: {error}")))?;
99    headers.insert(AUTHORIZATION, bearer);
100    Ok(headers)
101}
102
103/// One round-trip against `/v1/messages` with the minimal probe call:
104/// the cheapest haiku model, `max_tokens` 1, a tiny input - about
105/// nine billed tokens. On a 200 the response headers carry the
106/// unified windows, so the HeaderMap is the payload; a 429 keeps the
107/// windows it carried on the error; every other status classifies
108/// exactly like the windowed probe so the loader and poller treat
109/// both probes alike.
110async fn messages_probe(
111    client: &reqwest::Client,
112    user_agent: &str,
113    base_url: Option<&str>,
114    access_token: &str,
115) -> Result<HeaderMap, OauthUsageError> {
116    let payload = json!({
117        "model": PROBE_MODEL,
118        "max_tokens": 1,
119        "messages": [{ "role": "user", "content": "hi" }],
120    });
121    let response = client
122        .post(messages_url(base_url))
123        .headers(messages_headers(user_agent, access_token)?)
124        .json(&payload)
125        .send()
126        .await
127        .map_err(|error| OauthUsageError::Network(error.to_string()))?;
128
129    let status = response.status().as_u16();
130    // Parse Retry-After BEFORE consuming the response body - once we
131    // call .bytes() the response object is moved.
132    let retry_after = if status == 429 {
133        response
134            .headers()
135            .get(reqwest::header::RETRY_AFTER)
136            .and_then(|value| value.to_str().ok())
137            .and_then(parse_retry_after)
138    } else {
139        None
140    };
141    let headers = response.headers().clone();
142    let body = response
143        .bytes()
144        .await
145        .map_err(|error| OauthUsageError::Network(format!("body read: {error}")))?;
146
147    if status == 200 {
148        tracing::debug!(
149            target: "forge_gateway::anthropic",
150            event_name = "unified_usage_response",
151            status,
152            outcome = "ok",
153            body_bytes = body.len(),
154            five_hour_status = %unified_header_value(&headers, "5h-status"),
155            seven_day_status = %unified_header_value(&headers, "7d-status"),
156            unified_status = %unified_header_value(&headers, "status"),
157            unified_fallback = %unified_header_value(&headers, "fallback"),
158            overage_status = %unified_header_value(&headers, "overage-status"),
159        );
160    } else {
161        // The unified words are logged on every status, not only a 200:
162        // on a 429 the account-level one is the verdict and the window
163        // ones say which window is refusing.
164        tracing::warn!(
165            target: "forge_gateway::anthropic",
166            event_name = "unified_usage_response",
167            status,
168            outcome = "non_ok",
169            retry_after_secs = ?retry_after.map(|duration| duration.as_secs()),
170            five_hour_status = %unified_header_value(&headers, "5h-status"),
171            seven_day_status = %unified_header_value(&headers, "7d-status"),
172            unified_status = %unified_header_value(&headers, "status"),
173            body_suffix = %redacted_body(&body, access_token),
174        );
175    }
176
177    match status {
178        200 => Ok(headers),
179        401 | 403 => Err(OauthUsageError::Unauthorized(status)),
180        429 => Err(OauthUsageError::RateLimited {
181            retry_after,
182            snapshot: rate_limited_snapshot(&headers).map(Box::new),
183        }),
184        _ => Err(OauthUsageError::HttpStatus(status, redacted_body(&body, access_token))),
185    }
186}
187
188/// The windows a 429 carried. A 429 with no unified headers - a
189/// reverse proxy's limit, or a plan without unified limits - has no
190/// reading and keeps the bare rate-limited verdict.
191fn rate_limited_snapshot(headers: &HeaderMap) -> Option<UsageSnapshot> {
192    let snapshot = snapshot_from_unified_headers(headers);
193    (snapshot.five_hour.is_some() || snapshot.seven_day.is_some()).then_some(snapshot)
194}
195
196fn unified_header_value(headers: &HeaderMap, suffix: &str) -> String {
197    headers
198        .get(format!("{UNIFIED_PREFIX}{suffix}"))
199        .and_then(|value| value.to_str().ok())
200        .unwrap_or_default()
201        .to_owned()
202}
203
204/// One unified window off a response's headers: utilization arrives on a
205/// 0..1 scale and maps into the snapshot's percentage scale, the reset
206/// arrives as an epoch, and the status is the provider's own verdict on
207/// the window. Each header is independently optional.
208fn unified_window(headers: &HeaderMap, window: &str) -> Option<UsageWindow> {
209    let utilization = headers
210        .get(format!("{UNIFIED_PREFIX}{window}-utilization"))
211        .and_then(|value| value.to_str().ok())
212        .and_then(|value| value.trim().parse::<f64>().ok())
213        // A NaN survives the clamp, so a header carrying one would map to a
214        // window holding something that is not a reading. An infinity clamps
215        // into the range and is left one.
216        .filter(|raw| !raw.is_nan())
217        .map(|raw| (raw * 100.0).clamp(0.0, 100.0))?;
218    let resets_at = headers
219        .get(format!("{UNIFIED_PREFIX}{window}-reset"))
220        .and_then(|value| value.to_str().ok())
221        .and_then(|value| value.trim().parse::<i64>().ok())
222        .and_then(system_time_from_epoch);
223    let status = headers
224        .get(format!("{UNIFIED_PREFIX}{window}-status"))
225        .and_then(|value| value.to_str().ok())
226        .map(str::trim)
227        .filter(|value| !value.is_empty())
228        .map(str::to_owned);
229    Some(UsageWindow { utilization, resets_at, reset_description: None, status })
230}
231
232/// Map a 200's unified headers into a snapshot. Lenient by design: a
233/// 200 without the unified headers (an API-key bearer rather than a
234/// plan credential, or a plan without unified limits) is the barless
235/// row the scope refusal used to settle to, not a fetch error.
236fn snapshot_from_unified_headers(headers: &HeaderMap) -> UsageSnapshot {
237    UsageSnapshot {
238        source: UsageSourceKind::Oauth,
239        fetched_at: std::time::SystemTime::now(),
240        five_hour: unified_window(headers, "5h"),
241        seven_day: unified_window(headers, "7d"),
242        seven_day_opus: None,
243        seven_day_sonnet: None,
244        extra_usage: None,
245        spend: None,
246        balance: None,
247    }
248}
249
250#[cfg(test)]
251mod tests {
252    use std::time::{Duration, SystemTime};
253
254    use async_trait::async_trait;
255
256    use super::*;
257
258    #[test]
259    fn token_bearer_takes_a_non_empty_env_token() {
260        let mut env = HashMap::new();
261        env.insert("CLAUDE_CODE_OAUTH_TOKEN".to_owned(), "setup-token".to_owned());
262        assert_eq!(token_bearer(&env), Some("setup-token"));
263    }
264
265    /// An empty CLAUDE_CODE_OAUTH_TOKEN is not a credential: the token
266    /// is trimmed once at config load, so whitespace arrives as empty
267    /// and the account reads as token-less rather than probe-less.
268    #[test]
269    fn token_bearer_rejects_blank_tokens() {
270        let mut env = HashMap::new();
271        env.insert("CLAUDE_CODE_OAUTH_TOKEN".to_owned(), String::new());
272        assert_eq!(token_bearer(&env), None);
273        assert_eq!(token_bearer(&HashMap::new()), None);
274    }
275
276    fn unified_header_map() -> HeaderMap {
277        let mut headers = HeaderMap::new();
278        for (name, value) in [
279            ("anthropic-ratelimit-unified-5h-status", "allowed"),
280            ("anthropic-ratelimit-unified-5h-utilization", "0.53"),
281            ("anthropic-ratelimit-unified-5h-reset", "1766664000"),
282            ("anthropic-ratelimit-unified-7d-status", "allowed"),
283            ("anthropic-ratelimit-unified-7d-utilization", "0.06"),
284            ("anthropic-ratelimit-unified-7d-reset", "1767268800"),
285        ] {
286            headers.insert(name, HeaderValue::from_static(value));
287        }
288        headers
289    }
290
291    /// The unified utilization rides the headers on a 0..1 scale; the
292    /// snapshot's windows are percentages, so the mapping scales it
293    /// and the reset epoch lands on SystemTime.
294    #[test]
295    fn unified_headers_map_both_windows() {
296        let snapshot = snapshot_from_unified_headers(&unified_header_map());
297        let five_hour = snapshot.five_hour.expect("five hour window");
298        assert!((five_hour.utilization - 53.0).abs() < 1e-9, "got {}", five_hour.utilization);
299        assert_eq!(
300            five_hour.resets_at,
301            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_766_664_000)),
302        );
303        assert_eq!(five_hour.status.as_deref(), Some("allowed"), "the window's own verdict");
304        let seven_day = snapshot.seven_day.expect("seven day window");
305        assert!((seven_day.utilization - 6.0).abs() < 1e-9, "got {}", seven_day.utilization);
306        assert_eq!(
307            seven_day.resets_at,
308            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_268_800)),
309        );
310        assert_eq!(seven_day.status.as_deref(), Some("allowed"), "the window's own verdict");
311        assert!(snapshot.seven_day_opus.is_none());
312        assert!(snapshot.seven_day_sonnet.is_none());
313        assert_eq!(snapshot.source, UsageSourceKind::Oauth);
314    }
315
316    /// **A utilisation that is not a number is not a window at zero.** `NaN`
317    /// passes a plain float parse and survives the clamp, so the window landed
318    /// as `Some` holding a value the panel draws as `0%` beside an empty track
319    /// - the one picture it refuses to draw for a window nobody has reported.
320    #[test]
321    fn a_nan_utilization_leaves_the_window_unreported() {
322        let mut headers = HeaderMap::new();
323        headers
324            .insert("anthropic-ratelimit-unified-5h-utilization", HeaderValue::from_static("nan"));
325
326        assert!(unified_window(&headers, "5h").is_none(), "a NaN is not a reading");
327    }
328
329    /// The other half of the same door, and the reason the check rejects `NaN`
330    /// alone: an infinity clamps into the range and is still a reading, so
331    /// widening this to every non-finite value would throw one away.
332    #[test]
333    fn an_infinite_utilization_still_clamps_to_the_range() {
334        let mut headers = HeaderMap::new();
335        headers
336            .insert("anthropic-ratelimit-unified-5h-utilization", HeaderValue::from_static("inf"));
337
338        let window = unified_window(&headers, "5h").expect("an infinity is a reading, clamped");
339        assert!((window.utilization - 100.0).abs() < f64::EPSILON, "got {}", window.utilization);
340    }
341
342    #[test]
343    fn unified_utilization_clamps_to_the_percentage_range() {
344        let mut headers = HeaderMap::new();
345        headers
346            .insert("anthropic-ratelimit-unified-5h-utilization", HeaderValue::from_static("1.4"));
347        let window = unified_window(&headers, "5h").expect("window");
348        assert!((window.utilization - 100.0).abs() < f64::EPSILON, "got {}", window.utilization);
349    }
350
351    /// A 200 without the unified headers - an API-key bearer rather
352    /// than a plan credential, or a plan without unified limits -
353    /// maps to the barless snapshot the scope refusal used to settle
354    /// to, not a fetch error.
355    #[test]
356    fn a_200_without_unified_headers_maps_to_a_barless_snapshot() {
357        let snapshot = snapshot_from_unified_headers(&HeaderMap::new());
358        assert!(snapshot.five_hour.is_none());
359        assert!(snapshot.seven_day.is_none());
360        assert_eq!(snapshot.source, UsageSourceKind::Oauth);
361    }
362
363    const PROBE_RESPONSE_200: &str = concat!(
364        "HTTP/1.1 200 OK\r\n",
365        "content-type: application/json\r\n",
366        "anthropic-ratelimit-unified-5h-status: allowed\r\n",
367        "anthropic-ratelimit-unified-5h-utilization: 0.21\r\n",
368        "anthropic-ratelimit-unified-5h-reset: 1766664000\r\n",
369        "anthropic-ratelimit-unified-7d-utilization: 0.15\r\n",
370        "anthropic-ratelimit-unified-7d-reset: 1767268800\r\n",
371        "content-length: 2\r\n\r\n{}",
372    );
373
374    const PROBE_RESPONSE_429: &str =
375        "HTTP/1.1 429 Too Many Requests\r\nretry-after: 30\r\ncontent-length: 0\r\n\r\n";
376
377    /// A 429 as Anthropic sends one: the refuse and the windows that
378    /// explain it on the same response.
379    const PROBE_RESPONSE_429_WITH_WINDOWS: &str = concat!(
380        "HTTP/1.1 429 Too Many Requests\r\n",
381        "retry-after: 30\r\n",
382        "anthropic-ratelimit-unified-status: rejected\r\n",
383        "anthropic-ratelimit-unified-5h-status: rejected\r\n",
384        "anthropic-ratelimit-unified-5h-utilization: 1.0\r\n",
385        "anthropic-ratelimit-unified-5h-reset: 1766664000\r\n",
386        "anthropic-ratelimit-unified-7d-status: allowed\r\n",
387        "anthropic-ratelimit-unified-7d-utilization: 0.42\r\n",
388        "anthropic-ratelimit-unified-7d-reset: 1767268800\r\n",
389        "content-length: 0\r\n\r\n",
390    );
391
392    const PROBE_RESPONSE_401: &str =
393        "HTTP/1.1 401 Unauthorized\r\ncontent-type: application/json\r\ncontent-length: 0\r\n\r\n";
394
395    /// Answer one request, then hand the raw request text back so a
396    /// test can pin the probe's request shape. The request body is
397    /// drained before answering: closing with unread bytes pending
398    /// sends an RST that can destroy the in-flight response.
399    fn serve_one(
400        listener: std::net::TcpListener,
401        response: &'static str,
402    ) -> std::sync::mpsc::Receiver<String> {
403        let (sender, receiver) = std::sync::mpsc::channel();
404        std::thread::spawn(move || {
405            use std::io::{Read, Write as _};
406            let Ok((mut sock, _)) = listener.accept() else { return };
407            let mut request = Vec::new();
408            let mut byte = [0u8; 1];
409            while !request.ends_with(b"\r\n\r\n") {
410                match sock.read(&mut byte) {
411                    Ok(1) => request.push(byte[0]),
412                    _ => return,
413                }
414            }
415            let length: usize = String::from_utf8_lossy(&request)
416                .lines()
417                .find_map(|line| line.strip_prefix("content-length: "))
418                .and_then(|value| value.trim().parse().ok())
419                .unwrap_or(0);
420            for _ in 0..length {
421                match sock.read(&mut byte) {
422                    Ok(1) => request.push(byte[0]),
423                    _ => return,
424                }
425            }
426            let _ = sock.write_all(response.as_bytes());
427            let _ = sock.shutdown(std::net::Shutdown::Both);
428            let _ = sender.send(String::from_utf8_lossy(&request).to_string());
429        });
430        receiver
431    }
432
433    /// The offline round-trip: the probe posts the minimal call to
434    /// /v1/messages and the 200's headers map into windows. The
435    /// bearer here is a fake; no credential material is printed.
436    #[tokio::test]
437    async fn the_messages_probe_posts_the_minimal_call_and_maps_the_headers() {
438        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
439        let addr = listener.local_addr().expect("addr");
440        let requests = serve_one(listener, PROBE_RESPONSE_200);
441        let client = reqwest::Client::builder().no_proxy().build().expect("client");
442        let headers = tokio::time::timeout(
443            Duration::from_secs(5),
444            messages_probe(
445                &client,
446                "claude-code/1.0.0",
447                Some(&format!("http://{addr}")),
448                "probe-token",
449            ),
450        )
451        .await
452        .expect("the probe returns")
453        .expect("probe");
454        let request =
455            requests.recv_timeout(Duration::from_secs(5)).expect("the server saw a request");
456        assert!(request.starts_with("POST /v1/messages "), "got {request:?}");
457        assert!(request.contains("anthropic-version: 2023-06-01"));
458        assert!(request.contains("authorization: Bearer probe-token"));
459        assert!(request.contains("user-agent: claude-code/1.0.0"));
460        assert!(request.contains("\"max_tokens\":1"));
461        assert!(request.contains("\"model\":\"claude-haiku-4-5\""));
462
463        let snapshot = snapshot_from_unified_headers(&headers);
464        let five_hour = snapshot.five_hour.expect("five hour window");
465        assert!((five_hour.utilization - 21.0).abs() < 1e-9, "got {}", five_hour.utilization);
466        let seven_day = snapshot.seven_day.expect("seven day window");
467        assert!((seven_day.utilization - 15.0).abs() < 1e-9, "got {}", seven_day.utilization);
468    }
469
470    /// A 429 classifies as RateLimited carrying the server Retry-After,
471    /// and a response with no unified headers has no windows to keep.
472    #[tokio::test]
473    async fn a_429_is_a_rate_limited_error_with_the_retry_after() {
474        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
475        let addr = listener.local_addr().expect("addr");
476        let _requests = serve_one(listener, PROBE_RESPONSE_429);
477        let client = reqwest::Client::builder().no_proxy().build().expect("client");
478        let result = tokio::time::timeout(
479            Duration::from_secs(5),
480            messages_probe(&client, "claude-code/1.0.0", Some(&format!("http://{addr}")), "tok"),
481        )
482        .await
483        .expect("the probe returns");
484        assert!(
485            matches!(
486                result,
487                Err(OauthUsageError::RateLimited {
488                    retry_after: Some(retry_after),
489                    snapshot: None,
490                }) if retry_after == Duration::from_secs(30)
491            ),
492            "got {result:?}",
493        );
494    }
495
496    /// **A 429 is the response that says which window is spent, so its
497    /// windows ride the error rather than being thrown away with it.** The
498    /// refuse and the reading arrive on one response; keeping only the
499    /// first is why a cooling account read as `not reported`.
500    #[tokio::test]
501    async fn a_429_carrying_unified_headers_keeps_the_windows() {
502        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
503        let addr = listener.local_addr().expect("addr");
504        let _requests = serve_one(listener, PROBE_RESPONSE_429_WITH_WINDOWS);
505        let client = reqwest::Client::builder().no_proxy().build().expect("client");
506        let result = tokio::time::timeout(
507            Duration::from_secs(5),
508            messages_probe(&client, "claude-code/1.0.0", Some(&format!("http://{addr}")), "tok"),
509        )
510        .await
511        .expect("the probe returns");
512        let Err(OauthUsageError::RateLimited { retry_after, snapshot }) = result else {
513            panic!("a 429 must stay a rate-limited verdict, got {result:?}");
514        };
515        assert_eq!(
516            retry_after,
517            Some(Duration::from_secs(30)),
518            "the server hold-down survives the windows",
519        );
520        let snapshot = snapshot.expect("the 429's windows were dropped");
521        let five_hour = snapshot.five_hour.expect("five hour window");
522        assert!((five_hour.utilization - 100.0).abs() < f64::EPSILON);
523        assert_eq!(
524            five_hour.resets_at,
525            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_766_664_000)),
526            "which window is spent comes with when it lifts",
527        );
528        assert_eq!(five_hour.status.as_deref(), Some("rejected"), "the spent window's own verdict");
529        let seven_day = snapshot.seven_day.expect("seven day window");
530        assert!((seven_day.utilization - 42.0).abs() < 1e-9, "got {}", seven_day.utilization);
531        assert_eq!(
532            seven_day.resets_at,
533            Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_268_800))
534        );
535        assert_eq!(seven_day.status.as_deref(), Some("allowed"));
536    }
537
538    /// A rejected token reaches the loader as Unauthorized - the
539    /// re-mint repair hinges on the classification.
540    #[tokio::test]
541    async fn a_401_is_an_unauthorized_error() {
542        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
543        let addr = listener.local_addr().expect("addr");
544        let _requests = serve_one(listener, PROBE_RESPONSE_401);
545        let client = reqwest::Client::builder().no_proxy().build().expect("client");
546        let result = tokio::time::timeout(
547            Duration::from_secs(5),
548            messages_probe(&client, "claude-code/1.0.0", Some(&format!("http://{addr}")), "tok"),
549        )
550        .await
551        .expect("the probe returns");
552        assert!(matches!(result, Err(OauthUsageError::Unauthorized(401))), "got {result:?}");
553    }
554
555    /// The full unified family the CLI binary enumerates; only the
556    /// two windows map out of it, the rest are ignored.
557    #[test]
558    fn the_complete_unified_family_maps_only_the_two_windows() {
559        let mut headers = unified_header_map();
560        for (name, value) in [
561            ("anthropic-ratelimit-unified-5h-status", "allowed"),
562            ("anthropic-ratelimit-unified-7d-status", "allowed"),
563            ("anthropic-ratelimit-unified-status", "allowed"),
564            ("anthropic-ratelimit-unified-reset", "1767268800"),
565            ("anthropic-ratelimit-unified-representative-claim", "5h"),
566            ("anthropic-ratelimit-unified-fallback", "0"),
567            ("anthropic-ratelimit-unified-overage-status", "active"),
568            ("anthropic-ratelimit-unified-overage-disabled-reason", ""),
569            ("anthropic-ratelimit-unified-grace-status", "none"),
570            ("anthropic-ratelimit-unified-grace-5h-utilization", "0"),
571            ("anthropic-ratelimit-unified-grace-7d-utilization", "0"),
572            ("anthropic-ratelimit-unified-upgrade-paths", "x"),
573        ] {
574            headers.insert(name, HeaderValue::from_static(value));
575        }
576        let snapshot = snapshot_from_unified_headers(&headers);
577        let five_hour = snapshot.five_hour.expect("five hour window");
578        assert!((five_hour.utilization - 53.0).abs() < 1e-9, "got {}", five_hour.utilization);
579        let seven_day = snapshot.seven_day.expect("seven day window");
580        assert!((seven_day.utilization - 6.0).abs() < 1e-9, "got {}", seven_day.utilization);
581        assert!(snapshot.seven_day_opus.is_none());
582        assert!(snapshot.seven_day_sonnet.is_none());
583        assert!(snapshot.extra_usage.is_none());
584        assert!(snapshot.spend.is_none());
585    }
586
587    /// Either header alone is a legal shape: utilization without a
588    /// reset maps a window with no reset instant, and a reset without
589    /// utilization maps no window at all.
590    #[test]
591    fn a_partial_window_maps_what_is_present() {
592        let mut utilization_only = HeaderMap::new();
593        utilization_only
594            .insert("anthropic-ratelimit-unified-5h-utilization", HeaderValue::from_static("0.2"));
595        let snapshot = snapshot_from_unified_headers(&utilization_only);
596        let five_hour = snapshot.five_hour.expect("window from utilization alone");
597        assert!((five_hour.utilization - 20.0).abs() < 1e-9, "got {}", five_hour.utilization);
598        assert_eq!(five_hour.resets_at, None);
599
600        let mut reset_only = HeaderMap::new();
601        reset_only
602            .insert("anthropic-ratelimit-unified-7d-reset", HeaderValue::from_static("1767268800"));
603        let snapshot = snapshot_from_unified_headers(&reset_only);
604        assert!(snapshot.seven_day.is_none(), "a reset without utilization maps no window");
605    }
606
607    /// A host that cannot resolve the UA surfaces the UaProbe class -
608    /// a local exec problem, not a verdict about the endpoint - so the
609    /// callers' retry path engages.
610    #[tokio::test]
611    async fn a_host_ua_failure_is_a_ua_probe_error_not_a_network_failure() {
612        let backend = Anthropic;
613        let mut env = HashMap::new();
614        env.insert("CLAUDE_CODE_OAUTH_TOKEN".to_owned(), "setup-token".to_owned());
615        let account = AccountEnv { env: &env };
616        let result = backend.probe(&account, &FailingUaHost).await;
617        assert!(
618            matches!(result, Err(ProbeError::Fetch(OauthUsageError::UaProbe(_)))),
619            "got {result:?}",
620        );
621    }
622
623    /// An account with no token has no credential at all: the probe's
624    /// NoCredentials, and it must not reach the network for it.
625    #[tokio::test]
626    async fn a_tokenless_account_is_no_credentials_without_probing() {
627        let backend = Anthropic;
628        let account = AccountEnv { env: &HashMap::new() };
629        let result = backend.probe(&account, &EmptyHost).await;
630        assert!(matches!(result, Err(ProbeError::NoCredentials)), "got {result:?}");
631    }
632
633    struct FailingUaHost;
634
635    #[async_trait]
636    impl ProviderHost for FailingUaHost {
637        fn http_client(&self, _timeout: Duration) -> Result<reqwest::Client, String> {
638            reqwest::Client::builder().build().map_err(|e| e.to_string())
639        }
640
641        fn streaming_http_client(
642            &self,
643            _connect_timeout: Duration,
644            _idle_timeout: Duration,
645        ) -> Result<reqwest::Client, String> {
646            unreachable!("no probe streams a response")
647        }
648
649        async fn user_agent(&self) -> Result<String, String> {
650            Err("claude missing from PATH".to_owned())
651        }
652    }
653
654    struct EmptyHost;
655
656    #[async_trait]
657    impl ProviderHost for EmptyHost {
658        fn http_client(&self, _timeout: Duration) -> Result<reqwest::Client, String> {
659            unreachable!("the probe must not build a client for a missing credential")
660        }
661
662        fn streaming_http_client(
663            &self,
664            _connect_timeout: Duration,
665            _idle_timeout: Duration,
666        ) -> Result<reqwest::Client, String> {
667            unreachable!("the probe must not build a client for a missing credential")
668        }
669
670        async fn user_agent(&self) -> Result<String, String> {
671            unreachable!("the probe must not resolve a UA for a missing credential")
672        }
673    }
674}
forge_gateway/anthropic.rs

forge_gateway/
anthropic.rs
