From 79792eea85bb057c5ab89362a2ade732f397c39c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 1 Nov 2026 01:35:50 +0100 Subject: [PATCH] usage-meter-poll: the usage poll, the QuotaCell's second sensor usage_poll asks the subscription's usage endpoint when the seven_day reading is 15 min old or older, maps five_hour and seven_day onto PlanQuota with status polled, and honours Retry-After; failures wait, log once per change of state, and never touch the cell. Co-Authored-By: Claude Sonnet 5.5 --- Cargo.lock | 378 +++++++++++ crates/claude-connector/Cargo.toml | 2 + crates/claude-connector/src/lib.rs | 1 + crates/claude-connector/src/quota.rs | 10 +- crates/claude-connector/src/usage_poll.rs | 729 ++++++++++++++++++++++ 5 files changed, 1116 insertions(+), 4 deletions(-) create mode 100644 crates/claude-connector/src/usage_poll.rs diff --git a/Cargo.lock b/Cargo.lock index efa3d84..a1da85a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -139,6 +139,7 @@ name = "claude-connector" version = "0.1.0" dependencies = [ "anyhow", + "chrono", "cultcache-rs", "hmac", "libc", @@ -149,6 +150,7 @@ dependencies = [ "serde_json", "sha2", "thiserror", + "ureq", "zeroize", ] @@ -296,6 +298,17 @@ dependencies = [ "subtle", ] +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "dyn-clone" version = "1.0.20" @@ -338,6 +351,15 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + [[package]] name = "fs2" version = "0.4.3" @@ -459,6 +481,110 @@ dependencies = [ "cc", ] +[[package]] +name = "icu_collections" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe090a42a0d3140fed8c63263ed1fecc02422cb2f4bfe55cd8de2de1b8e105dd" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25d29b57c17c6528d0dc4c2141cfa2d08200465acb537a3909c8b9b8a4a984c0" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "404a3b8c486f205c87fe7f23ab6894d84a4ad76d6e4671ae1981ba7def87c599" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20f0088bfb7441f7a37fa0c9e20cb1e9cba1cac6b9ce8cd2a6dc44294f8e8cbc" + +[[package]] +name = "icu_properties" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ea83d991c56158f7cf898bf86436c1dfe7c07ee82fe88951f2695c9f37a379f" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b98f29af82d02b163fe50161388350e743bb3107cc8e8c2f7314cb505faddfcf" + +[[package]] +name = "icu_provider" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413099315636c588ee8f011190b329b971a09a8fae9d157e01dcae8379468514" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + [[package]] name = "inout" version = "0.1.4" @@ -491,6 +617,12 @@ version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + [[package]] name = "log" version = "0.4.34" @@ -554,6 +686,12 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -582,6 +720,15 @@ dependencies = [ "universal-hash", ] +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -698,6 +845,20 @@ dependencies = [ "syn 3.0.4", ] +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rmp" version = "0.8.15" @@ -737,6 +898,41 @@ dependencies = [ "semver", ] +[[package]] +name = "rustls" +version = "0.23.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +dependencies = [ + "log", + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" @@ -881,6 +1077,12 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "smallvec" +version = "1.16.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8" + [[package]] name = "socket2" version = "0.5.10" @@ -901,6 +1103,12 @@ dependencies = [ "der", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "stacker" version = "0.1.25" @@ -942,6 +1150,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "synstructure" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "thiserror" version = "2.0.20" @@ -962,6 +1181,16 @@ dependencies = [ "syn 3.0.4", ] +[[package]] +name = "tinystr" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abc840c238154d58f9e86b2792261edfde565a17696e1cda1a449de27979801a" +dependencies = [ + "displaydoc", + "zerovec", +] + [[package]] name = "typenum" version = "1.20.1" @@ -984,6 +1213,54 @@ dependencies = [ "subtle", ] +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "ureq" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d1a66277ed75f640d608235660df48c8e3c19f3b4edb6a263315626cc3c01d" +dependencies = [ + "base64", + "log", + "once_cell", + "rustls", + "rustls-pki-types", + "url", + "webpki-roots 0.26.11", +] + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af3bd9d3e571c26fcce9d033b87ea5ab061b8d2ffe9b7a5675d810d68f03edf0" +dependencies = [ + "utf_types", +] + +[[package]] +name = "utf_types" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a3b425286980a367278453f3b6d28cdcb134b574e5a35824a6ff5eebad50bb6" + [[package]] name = "uuid" version = "1.25.0" @@ -1061,6 +1338,24 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.11", +] + +[[package]] +name = "webpki-roots" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5f280bbb2b287bc8e7de7e3852945a0ff50d1bbcb5ed42b6c7a568c546df171" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi" version = "0.3.9" @@ -1239,6 +1534,35 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "417ef3fcf6c300b7dbd25c85531a746931c02ad536f3cf4786ed1e88223340b3" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.63" @@ -1259,12 +1583,66 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", + "synstructure", +] + [[package]] name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f58a2799cfe9a1670c7ea8edbd52ef8edb93cb37a7f83bf31ba9e35d9dae6a40" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3743ddbf47738d929cd1ae2e199d40d0162979dfe31aef8ae646d2bf00bf97a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/crates/claude-connector/Cargo.toml b/crates/claude-connector/Cargo.toml index 7b310e5..bc28e9f 100644 --- a/crates/claude-connector/Cargo.toml +++ b/crates/claude-connector/Cargo.toml @@ -13,6 +13,7 @@ path = "src/main.rs" [dependencies] anyhow = "1" +chrono = { version = "0.4", default-features = false } hmac = "0.12" cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a4eae9056152335662f1115f97bf35fb00ab6d23", package = "cultcache-rs" } model-connector = { path = "../model-connector", features = ["daemon"] } @@ -21,6 +22,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" thiserror = "2" +ureq = { version = "2", default-features = false, features = ["tls"] } zeroize = "1" [target.'cfg(unix)'.dependencies] diff --git a/crates/claude-connector/src/lib.rs b/crates/claude-connector/src/lib.rs index 70a2e53..1b01d30 100644 --- a/crates/claude-connector/src/lib.rs +++ b/crates/claude-connector/src/lib.rs @@ -15,6 +15,7 @@ mod quota; mod stream; #[cfg(target_os = "linux")] mod unit; +mod usage_poll; pub use backend::{ClaudeCodeBackend, CredentialTokenSource, StderrSink, TokenSource}; pub use config::{ClaudeCodeConfig, ClaudeCodeLimits}; diff --git a/crates/claude-connector/src/quota.rs b/crates/claude-connector/src/quota.rs index 450d3e2..1b3d27e 100644 --- a/crates/claude-connector/src/quota.rs +++ b/crates/claude-connector/src/quota.rs @@ -1,7 +1,9 @@ -//! The plan's own usage windows, as the CLI reports them. The subscription -//! decides how much a plan may burn; the CLI's `rate_limit_event` is the only -//! place that number appears, so the latest observation of each window is kept -//! here and published as the usage readout and lent with each lease. +//! The plan's own usage windows, as the subscription reports them. The +//! subscription decides how much a plan may burn, and two sensors observe it: +//! the CLI's `rate_limit_event` on the connector's own calls, and the usage +//! poll of `usage_poll` while that stream is quiet. The latest observation of +//! each window, from either, is kept here and published as the usage readout +//! and lent with each lease. use std::collections::BTreeMap; use std::sync::Mutex; diff --git a/crates/claude-connector/src/usage_poll.rs b/crates/claude-connector/src/usage_poll.rs new file mode 100644 index 0000000..469e234 --- /dev/null +++ b/crates/claude-connector/src/usage_poll.rs @@ -0,0 +1,729 @@ +//! The second sensor of the plan's usage windows. The CLI's `rate_limit_event` +//! only arrives when the connector's own calls run, so while the swarm is +//! quiet the weekly reading goes stale; this thread asks the subscription's +//! usage endpoint when the `seven_day` reading is `STALE` or older, and +//! records the answer in the same [`QuotaCell`] the stream feeds. Neither +//! sensor owns the reading: the newest observation of a window wins. A failed +//! poll never touches the cell; only the cell's own reset rule retires a +//! window. +//! +//! The endpoint is undocumented (the shape is the one Claude Code 2.1.312 +//! parses), so every answer it can give is handled and none is trusted to +//! change the schedule beyond the rules below. + +use std::io::Read; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use model_connector::PlanQuota; +use serde_json::Value; +use zeroize::Zeroizing; + +use crate::backend::TokenSource; +use crate::quota::QuotaCell; + +/// The usage endpoint. +const ENDPOINT: &str = "https://api.anthropic.com/api/oauth/usage"; +/// A reading this old, or older, is stale; also the least gap between polls. +const STALE: Duration = Duration::from_secs(15 * 60); +/// The longest any backoff may wait. +const CEILING: Duration = Duration::from_secs(6 * 60 * 60); +const TIMEOUT: Duration = Duration::from_secs(10); +/// A usage body is a few hundred bytes; this is a ceiling, not an expectation. +const MAX_BODY: u64 = 1 << 20; +/// The status every polled window carries. +const POLLED: &str = "polled"; + +/// What the endpoint said, reduced to what the schedule needs. It never holds +/// the token. +#[derive(Debug, PartialEq, Eq)] +enum Answer { + Body(String), + Status(u16, Option), + Unreachable, +} + +/// Why the poller is waiting; logged once per change. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum State { + Polled, + NoToken, + RateLimited, + Refused, + NotFound, + Unreadable, + Unexpected, + Unreachable, +} + +impl State { + fn line(self) -> &'static str { + match self { + Self::Polled => "usage poll: reading recorded", + Self::NoToken => "usage poll: no token to ask with", + Self::RateLimited => "usage poll: rate limited, waiting", + Self::Refused => { + "usage poll: refused (401/403), the token likely lacks the user:profile scope; polling stopped until restart" + } + Self::NotFound => "usage poll: endpoint not found, backing off", + Self::Unreadable => "usage poll: answer not readable, backing off", + Self::Unexpected => "usage poll: unexpected status, retrying later", + Self::Unreachable => "usage poll: endpoint unreachable, retrying later", + } + } +} + +#[derive(Debug, PartialEq, Eq)] +enum Next { + Wait(Duration), + Stop, +} + +#[derive(Debug, PartialEq)] +struct Verdict { + windows: Vec, + next: Next, + state: State, +} + +/// What one answer means: the windows to record (only a readable body has +/// any), how long to wait, and the state to report. +fn judge(answer: &Answer) -> Verdict { + let (windows, next, state) = match answer { + Answer::Body(body) => match parse_usage(body) { + Some(windows) => (windows, Next::Wait(STALE), State::Polled), + None => (Vec::new(), Next::Wait(CEILING), State::Unreadable), + }, + Answer::Status(429, retry_after) => ( + Vec::new(), + Next::Wait(retry_after.unwrap_or_default().clamp(STALE, CEILING)), + State::RateLimited, + ), + Answer::Status(401 | 403, _) => (Vec::new(), Next::Stop, State::Refused), + Answer::Status(404, _) => (Vec::new(), Next::Wait(CEILING), State::NotFound), + Answer::Status(..) => (Vec::new(), Next::Wait(STALE), State::Unexpected), + Answer::Unreachable => (Vec::new(), Next::Wait(STALE), State::Unreachable), + }; + Verdict { + windows, + next, + state, + } +} + +/// The plan windows a usage body reports: `five_hour` and `seven_day`, each +/// `{utilization: percent 0-100, resets_at: ISO 8601}`. A window missing +/// either field, or reporting nonsense, is skipped, and an unknown reading is +/// never zero. Other keys are ignored. `None` when the body is not a JSON +/// object at all. +fn parse_usage(body: &str) -> Option> { + let usage: Value = serde_json::from_str(body).ok()?; + let usage = usage.as_object()?; + Some( + ["five_hour", "seven_day"] + .into_iter() + .filter_map(|window| { + let reading = usage.get(window)?; + let percent = reading.get("utilization")?.as_f64()?; + let resets_at = reading.get("resets_at")?.as_str()?; + if !percent.is_finite() || percent < 0.0 { + return None; + } + let resets_at = chrono::DateTime::parse_from_rfc3339(resets_at).ok()?; + Some(PlanQuota { + window: window.to_string(), + // `as` saturates a float at the integer's bounds. + utilization_permille: Some((percent * 10.0).round() as u32), + resets_at_unix_ms: u64::try_from(resets_at.timestamp_millis()).ok()?, + status: POLLED.to_string(), + // Stamped by the cell that keeps it. + observed_at_unix_ms: 0, + }) + }) + .collect(), + ) +} + +/// The blocking HTTPS client: 10 s end to end, no redirects (the token must +/// not follow one anywhere). +struct UsageClient { + agent: ureq::Agent, + url: String, +} + +impl UsageClient { + fn new(url: impl Into) -> Self { + Self { + agent: ureq::AgentBuilder::new() + .timeout(TIMEOUT) + .redirects(0) + .build(), + url: url.into(), + } + } + + fn ask(&self, token: &Zeroizing) -> Answer { + let bearer = Zeroizing::new(format!("Bearer {}", token.as_str())); + let sent = self + .agent + .get(&self.url) + .set("Authorization", bearer.as_str()) + .set("anthropic-beta", "oauth-2025-04-20") + .call(); + match sent { + Ok(response) if response.status() == 200 => { + let mut body = String::new(); + match response + .into_reader() + .take(MAX_BODY) + .read_to_string(&mut body) + { + Ok(_) => Answer::Body(body), + Err(_) => Answer::Unreachable, + } + } + Ok(response) => Answer::Status(response.status(), None), + Err(ureq::Error::Status(code, response)) => { + // Whole seconds only; the HTTP-date form counts as absent. + let retry_after = response + .header("retry-after") + .and_then(|seconds| seconds.trim().parse::().ok()) + .map(Duration::from_secs); + Answer::Status(code, retry_after) + } + Err(ureq::Error::Transport(_)) => Answer::Unreachable, + } + } +} + +/// The schedule and the thing it drives. Time comes in as an argument so the +/// schedule is testable without waiting. +struct Poller { + cell: Arc, + tokens: Box, + client: UsageClient, + log: Box, + /// No poll before this instant, unix ms: the last answer's own wait. + not_before_ms: u64, + started: bool, + stopped: bool, + last_state: Option, +} + +impl Poller { + fn new( + cell: Arc, + tokens: Box, + client: UsageClient, + log: Box, + ) -> Self { + Self { + cell, + tokens, + client, + log, + not_before_ms: 0, + started: false, + stopped: false, + last_state: None, + } + } + + /// Polls if a poll is due at `now_ms`; returns how long to wait before + /// asking again, or `None` once polling has stopped for good. + /// + /// The first call polls. After that a poll is due only when the cell's + /// `seven_day` observation is `STALE` old or older (or absent), and never + /// before the last answer's own wait has passed. + fn tick(&mut self, now_ms: u64) -> Option { + if self.stopped { + return None; + } + let due_at = self.due_at_ms(); + if now_ms < due_at { + return Some(Duration::from_millis(due_at - now_ms)); + } + self.started = true; + let verdict = match self.tokens.read() { + Some(token) => judge(&self.client.ask(&token)), + None => Verdict { + windows: Vec::new(), + next: Next::Wait(STALE), + state: State::NoToken, + }, + }; + self.cell.observe(verdict.windows); + if self.last_state != Some(verdict.state) { + (self.log)(verdict.state.line()); + self.last_state = Some(verdict.state); + } + match verdict.next { + Next::Stop => { + self.stopped = true; + None + } + Next::Wait(wait) => { + self.not_before_ms = now_ms.saturating_add(duration_ms(wait)); + Some(Duration::from_millis( + self.due_at_ms().saturating_sub(now_ms), + )) + } + } + } + + /// The earliest instant the next poll may happen. + fn due_at_ms(&self) -> u64 { + if !self.started { + return 0; + } + let stale_at = self + .cell + .snapshot() + .windows + .iter() + .find(|quota| quota.window == "seven_day") + .map_or(0, |quota| { + quota.observed_at_unix_ms.saturating_add(duration_ms(STALE)) + }); + stale_at.max(self.not_before_ms) + } + + fn run(mut self) { + while let Some(wait) = self.tick(unix_ms()) { + std::thread::sleep(wait); + } + } +} + +fn duration_ms(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, duration_ms) +} + +/// Starts the poll thread. It reads the token afresh at each poll, writes +/// only to `cell`, and logs one line to stderr when its state changes. +pub fn start( + cell: Arc, + tokens: Box, +) -> std::io::Result> { + let poller = Poller::new( + cell, + tokens, + UsageClient::new(ENDPOINT), + Box::new(|line| eprintln!("claude-connector: {line}")), + ); + std::thread::Builder::new() + .name("claude-usage-poll".to_string()) + .spawn(move || poller.run()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::net::TcpListener; + use std::sync::Mutex; + + const MIN: u64 = 60_000; + const CANARY: &str = "sk-ant-oat01-CANARY-must-never-surface"; + const FAR: &str = "2100-01-01T00:00:00+00:00"; + const FAR_MS: u64 = 4_102_444_800_000; + + fn body(five: &str, seven: &str) -> String { + format!( + r#"{{"five_hour":{five},"seven_day":{seven},"seven_day_opus":null,"extra_usage":{{"x":1}}}}"# + ) + } + + fn reading(percent: f64) -> String { + format!(r#"{{"utilization":{percent},"resets_at":"{FAR}"}}"#) + } + + fn secs(s: u64) -> Duration { + Duration::from_secs(s) + } + + // ---- mapping ---- + + #[test] + fn usage_body_maps_percent_to_permille() { + let windows = parse_usage(&body(&reading(6.0), &reading(35.0))).unwrap(); + let expect = |window: &str, permille| PlanQuota { + window: window.to_string(), + utilization_permille: Some(permille), + resets_at_unix_ms: FAR_MS, + status: "polled".to_string(), + observed_at_unix_ms: 0, + }; + assert_eq!( + windows, + vec![expect("five_hour", 60), expect("seven_day", 350)] + ); + } + + #[test] + fn usage_percent_rounds_to_the_nearest_permille() { + let windows = parse_usage(&body(&reading(12.34), &reading(0.06))).unwrap(); + assert_eq!(windows[0].utilization_permille, Some(123)); + assert_eq!(windows[1].utilization_permille, Some(1)); + let zero = parse_usage(&body(&reading(0.0), "null")).unwrap(); + assert_eq!(zero[0].utilization_permille, Some(0), "a real zero stays"); + } + + #[test] + fn resets_at_is_parsed_with_its_offset_to_unix_ms() { + let seven = r#"{"utilization":1.0,"resets_at":"2026-11-07T12:30:15.250+02:00"}"#; + let windows = parse_usage(&body("null", seven)).unwrap(); + // 2026-11-07T10:30:15.250Z + assert_eq!(windows[0].resets_at_unix_ms, 1_794_047_415_250); + } + + #[test] + fn a_window_missing_a_field_or_reading_nonsense_is_skipped_not_zero() { + let no_percent = format!(r#"{{"utilization":null,"resets_at":"{FAR}"}}"#); + let no_reset = r#"{"utilization":50.0,"resets_at":null}"#; + assert_eq!(parse_usage(&body(&no_percent, no_reset)).unwrap(), vec![]); + let junk = r#"{"utilization":5.0,"resets_at":"soon"}"#; + assert_eq!(parse_usage(&body(&reading(-1.0), junk)).unwrap(), vec![]); + let before_epoch = r#"{"utilization":5.0,"resets_at":"1969-01-01T00:00:00Z"}"#; + assert_eq!(parse_usage(&body(before_epoch, "null")).unwrap(), vec![]); + // One good window survives its neighbour's null. + let one = parse_usage(&body(&no_percent, &reading(35.0))).unwrap(); + assert_eq!(one.len(), 1); + assert_eq!(one[0].window, "seven_day"); + } + + #[test] + fn a_body_that_is_not_an_object_is_unreadable() { + for bad in ["", "not json", "[1,2]", "42", "null"] { + assert_eq!(parse_usage(bad), None, "{bad:?}"); + } + assert_eq!(parse_usage("{}"), Some(vec![])); + } + + // ---- judgement ---- + + fn wait_of(answer: &Answer) -> Next { + judge(answer).next + } + + #[test] + fn rate_limit_waits_the_longer_of_retry_after_and_the_floor_capped() { + let limited = |after| wait_of(&Answer::Status(429, after)); + assert_eq!(limited(Some(secs(3600))), Next::Wait(secs(3600))); + assert_eq!(limited(Some(secs(20 * 60))), Next::Wait(secs(20 * 60))); + assert_eq!(limited(Some(secs(10))), Next::Wait(STALE)); + assert_eq!(limited(None), Next::Wait(STALE)); + assert_eq!(limited(Some(secs(100_000))), Next::Wait(CEILING)); + assert_eq!(limited(Some(secs(6 * 3600))), Next::Wait(CEILING)); + assert_eq!(judge(&Answer::Status(429, None)).state, State::RateLimited); + } + + #[test] + fn refusal_stops_polling_and_the_rest_back_off() { + assert_eq!(wait_of(&Answer::Status(401, None)), Next::Stop); + assert_eq!(wait_of(&Answer::Status(403, Some(secs(5)))), Next::Stop); + assert_eq!(judge(&Answer::Status(403, None)).state, State::Refused); + assert_eq!(wait_of(&Answer::Status(404, None)), Next::Wait(CEILING)); + assert_eq!( + wait_of(&Answer::Body("".to_string())), + Next::Wait(CEILING) + ); + assert_eq!(wait_of(&Answer::Unreachable), Next::Wait(STALE)); + assert_eq!(wait_of(&Answer::Status(503, None)), Next::Wait(STALE)); + assert_eq!( + wait_of(&Answer::Body(body(&reading(1.0), "null"))), + Next::Wait(STALE) + ); + } + + // ---- a fake server, a poller, and the cell ---- + + /// Serves each canned reply to one connection in turn and remembers the + /// requests. + fn serve(replies: Vec) -> (String, Arc>>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}/api/oauth/usage", listener.local_addr().unwrap()); + let seen = Arc::new(Mutex::new(Vec::new())); + let record = seen.clone(); + std::thread::spawn(move || { + for reply in replies { + let Ok((mut stream, _)) = listener.accept() else { + return; + }; + let mut request = vec![0u8; 4096]; + let n = stream.read(&mut request).unwrap_or(0); + record + .lock() + .unwrap() + .push(String::from_utf8_lossy(&request[..n]).into_owned()); + let _ = stream.write_all(reply.as_bytes()); + } + }); + (url, seen) + } + + fn http(status: &str, headers: &str, body: &str) -> String { + format!( + "HTTP/1.1 {status}\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + } + + struct Fixed(Option<&'static str>); + impl TokenSource for Fixed { + fn read(&self) -> Option> { + self.0.map(|t| Zeroizing::new(t.to_string())) + } + } + + type Lines = Arc>>; + + fn poller(cell: &Arc, url: &str, token: Option<&'static str>) -> (Poller, Lines) { + let lines = Arc::new(Mutex::new(Vec::new())); + let sink = lines.clone(); + let poller = Poller::new( + cell.clone(), + Box::new(Fixed(token)), + UsageClient::new(url), + Box::new(move |line| sink.lock().unwrap().push(line.to_string())), + ); + (poller, lines) + } + + fn window(cell: &QuotaCell, name: &str) -> Option { + cell.snapshot() + .windows + .into_iter() + .find(|q| q.window == name) + } + + fn stream_reading(window: &str, permille: u32) -> PlanQuota { + PlanQuota { + window: window.to_string(), + utilization_permille: Some(permille), + resets_at_unix_ms: FAR_MS, + status: "allowed".to_string(), + observed_at_unix_ms: 0, + } + } + + fn requests(seen: &Arc>>) -> usize { + seen.lock().unwrap().len() + } + + #[test] + fn the_first_tick_polls_and_records_what_the_endpoint_said() { + let (url, seen) = serve(vec![http( + "200 OK", + "Content-Type: application/json\r\n", + &body(&reading(6.0), &reading(35.0)), + )]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, lines) = poller(&cell, &url, Some(CANARY)); + let wait = poller.tick(unix_ms()).unwrap(); + let windows = cell.snapshot().windows; + assert_eq!(windows.len(), 2); + assert_eq!(windows[0].status, "polled"); + assert_eq!(windows[1].utilization_permille, Some(350)); + assert!( + (STALE..STALE + secs(5)).contains(&wait), + "a fresh reading waits until it is stale: {wait:?}" + ); + let request = seen.lock().unwrap()[0].to_lowercase(); + assert!(request.contains(&format!("authorization: bearer {}", CANARY.to_lowercase()))); + assert!(request.contains("anthropic-beta: oauth-2025-04-20")); + assert_eq!(lines.lock().unwrap().len(), 1); + } + + #[test] + fn stale_only_polls() { + let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0))); 3]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, _) = poller(&cell, &url, Some(CANARY)); + poller.tick(unix_ms()); + assert_eq!(requests(&seen), 1, "the startup poll"); + let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms; + + // 5 min old: no request, and the wait is the time left. + let wait = poller.tick(observed + 5 * MIN).unwrap(); + assert_eq!(requests(&seen), 1); + assert_eq!(wait, Duration::from_millis(10 * MIN)); + // One millisecond short of 15 min: still none. + poller.tick(observed + 15 * MIN - 1); + assert_eq!(requests(&seen), 1); + // 16 min old: one request. + poller.tick(observed + 16 * MIN); + assert_eq!(requests(&seen), 2); + // That poll's answer holds its own 15 min, whatever the stamp says. + poller.tick(observed + 16 * MIN + 1); + assert_eq!(requests(&seen), 2); + } + + #[test] + fn a_reading_is_stale_at_exactly_fifteen_minutes() { + let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0))); 2]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, _) = poller(&cell, &url, Some(CANARY)); + poller.tick(unix_ms()); + let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms; + poller.tick(observed + 15 * MIN); + assert_eq!(requests(&seen), 2); + } + + #[test] + fn a_stream_observation_keeps_the_poll_quiet() { + let (url, seen) = serve(vec![http("200 OK", "", &body("null", &reading(35.0)))]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, _) = poller(&cell, &url, Some(CANARY)); + poller.tick(unix_ms()); + std::thread::sleep(Duration::from_millis(20)); + // The stream reports seven_day afresh, after the poll. + cell.observe([stream_reading("seven_day", 400)]); + let observed = window(&cell, "seven_day").unwrap().observed_at_unix_ms; + let wait = poller.tick(observed + 14 * MIN).unwrap(); + assert_eq!(requests(&seen), 1); + assert_eq!(wait, Duration::from_millis(MIN)); + } + + #[test] + fn failures_never_touch_the_cell() { + let limited = |headers: &str| http("429 Too Many Requests", headers, "{}"); + let cases = [ + (limited("Retry-After: 3600\r\n"), Some(60)), + (limited(""), Some(15)), + ( + limited("Retry-After: Wed, 21 Oct 2026 07:28:00 GMT\r\n"), + Some(15), + ), + (http("403 Forbidden", "", "{}"), None), + (http("401 Unauthorized", "", "{}"), None), + (http("404 Not Found", "", "{}"), Some(360)), + (http("200 OK", "", "not json"), Some(360)), + ( + http("302 Found", "Location: http://127.0.0.1:9/\r\n", ""), + Some(15), + ), + ]; + for (reply, minutes) in cases { + let (url, _) = serve(vec![reply.clone()]); + let cell = Arc::new(QuotaCell::new()); + cell.observe([ + stream_reading("five_hour", 120), + stream_reading("seven_day", 300), + ]); + let before = cell.snapshot(); + let (mut poller, lines) = poller(&cell, &url, Some(CANARY)); + let wait = poller.tick(unix_ms()); + assert_eq!(cell.snapshot(), before, "{reply}"); + assert_eq!(wait.map(|w| w.as_secs() / 60), minutes, "{reply}"); + assert_eq!(lines.lock().unwrap().len(), 1, "{reply}"); + assert_eq!(poller.tick(unix_ms()).is_none(), minutes.is_none()); + } + } + + #[test] + fn an_unreachable_endpoint_waits_the_floor_and_a_missing_token_never_asks() { + // Nothing listens here: bind, learn the port, drop. + let port = TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, lines) = + poller(&cell, &format!("http://127.0.0.1:{port}/u"), Some(CANARY)); + assert_eq!(poller.tick(unix_ms()), Some(STALE)); + assert!(cell.snapshot().windows.is_empty()); + assert_eq!(lines.lock().unwrap().len(), 1); + + let (url, seen) = serve(vec![http("200 OK", "", "{}")]); + let (mut poller, lines) = poller(&cell, &url, None); + assert_eq!(poller.tick(unix_ms()), Some(STALE)); + assert_eq!(requests(&seen), 0); + assert_eq!(lines.lock().unwrap()[0], State::NoToken.line()); + } + + #[test] + fn a_state_is_logged_once_until_it_changes() { + let limited = http("429 Too Many Requests", "", "{}"); + let (url, _) = serve(vec![ + limited.clone(), + limited, + http("200 OK", "", &body("null", &reading(35.0))), + ]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, lines) = poller(&cell, &url, Some(CANARY)); + let t = unix_ms(); + poller.tick(t); + poller.tick(t + 16 * MIN); + assert_eq!(lines.lock().unwrap().len(), 1, "same state, one line"); + poller.tick(t + 32 * MIN); + let lines = lines.lock().unwrap(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[1], State::Polled.line()); + } + + #[test] + fn nothing_logged_or_stored_holds_the_token() { + let echo = format!(r#"{{"echo":"{CANARY}"}}"#); + let (url, _) = serve(vec![ + http("429 Too Many Requests", "Retry-After: 60\r\n", &echo), + http( + "200 OK", + "", + &format!(r#"{{"five_hour":{},"note":"{CANARY}"}}"#, reading(1.0)), + ), + http("403 Forbidden", "", CANARY), + ]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, lines) = poller(&cell, &url, Some(CANARY)); + let t = unix_ms(); + poller.tick(t); + poller.tick(t + 16 * MIN); + assert!(poller.tick(t + 32 * MIN).is_none()); + let lines = lines.lock().unwrap(); + assert_eq!(lines.len(), 3, "one line per change of state"); + let seen = format!("{lines:?} {:?}", cell.snapshot()); + assert!(!seen.contains("CANARY"), "{seen}"); + } + + #[test] + fn stream_and_poll_share_one_cell() { + let (url, _) = serve(vec![http( + "200 OK", + "", + &body(&reading(6.0), &reading(35.0)), + )]); + let cell = Arc::new(QuotaCell::new()); + let (mut poller, _) = poller(&cell, &url, Some(CANARY)); + poller.tick(unix_ms()); + assert_eq!(window(&cell, "five_hour").unwrap().status, "polled"); + // A later stream reading of one window replaces the poll's; the other + // window keeps the poll's. + cell.observe([stream_reading("five_hour", 90)]); + let five = window(&cell, "five_hour").unwrap(); + assert_eq!( + (five.status.as_str(), five.utilization_permille), + ("allowed", Some(90)) + ); + assert_eq!(window(&cell, "seven_day").unwrap().status, "polled"); + // And a later poll replaces the stream's. + let (url, _) = serve(vec![http("200 OK", "", &body(&reading(7.0), "null"))]); + let (mut again, _) = poller_at(&cell, &url); + again.tick(unix_ms()); + assert_eq!( + window(&cell, "five_hour").unwrap().utilization_permille, + Some(70) + ); + } + + fn poller_at(cell: &Arc, url: &str) -> (Poller, Lines) { + poller(cell, url, Some(CANARY)) + } +}