diff --git a/Cargo.lock b/Cargo.lock index efa3d84..3709c14 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", diff --git a/crates/claude-connector/Cargo.toml b/crates/claude-connector/Cargo.toml index 7b310e5..6cb9e45 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"] } diff --git a/crates/claude-connector/src/backend.rs b/crates/claude-connector/src/backend.rs index 4833521..0bc2d77 100644 --- a/crates/claude-connector/src/backend.rs +++ b/crates/claude-connector/src/backend.rs @@ -20,6 +20,12 @@ use crate::process::{ProcessPort, RunningChild, StdoutRead}; use crate::quota::QuotaCell; use crate::stream::{Failure, Finished, Step, StreamState}; +/// The meter's call: the cheapest model, a trivial prompt, and an answer cap +/// that only bounds a runaway. +const METER_MODEL: &str = "claude-haiku-5-5"; +const METER_CALLER: &str = "meter"; +const METER_MAX_OUTPUT_TOKENS: u32 = 16; + /// Where the child's credential comes from, read afresh at every use. The /// connector's own environment is never a source (ruling /// claude-connector-token-credential). @@ -156,16 +162,40 @@ impl ClaudeCodeBackend { self.quota.clone() } + /// The meter's call: the smallest real one, through the same child, + /// stream and cell as any other. The stream's `rate_limit_event` is what + /// the caller wants; this returns what the call billed, or the failure's + /// kind (never its message). + pub(crate) fn read_meter(&self) -> Result { + let mut request = ProviderRequest::new( + METER_CALLER, + METER_CALLER, + METER_MODEL, + "Reply with the single word ok.", + METER_MAX_OUTPUT_TOKENS, + ); + request.input = vec![InputItem::UserText { + text: "ok?".to_string(), + }]; + match self.run(&request, METER_CALLER, true).1 { + TransportOutcome::Completed { usage, .. } => Ok(usage), + TransportOutcome::Failed { kind, .. } => Err(format!("{kind:?}")), + } + } + /// The one pin check: `ready` and every spawn use it, and nothing caches /// its answer. fn binary_matches_pin(&self) -> bool { binary_matches_pin(&self.config) } + /// `ephemeral` leaves no transcript behind: the meter keeps no handle and + /// resumes nothing, every other call is resumed by its session. fn run( &self, request: &ProviderRequest, caller: &str, + ephemeral: bool, ) -> (Vec, TransportOutcome) { let Some(secret) = usable_secret(self.handle_secret.as_ref()) else { return failed( @@ -209,7 +239,7 @@ impl ClaudeCodeBackend { "claude-code: the system prompt could not be staged", )); }; - let argv = argv(&self.config, request, &plan, prompt.path()); + let argv = argv(&self.config, request, &plan, prompt.path(), ephemeral); let env = child_env(&self.config, request); let deadline = Instant::now() + self.config.limits.deadline; @@ -323,7 +353,7 @@ impl ProviderBackend for ClaudeCodeBackend { } fn execute(&self, call: &ProviderCall) -> TransportResult { - let (events, outcome) = self.run(call.request(), call.caller_runtime_id()); + let (events, outcome) = self.run(call.request(), call.caller_runtime_id(), false); call.transported(events, outcome) } diff --git a/crates/claude-connector/src/backend/tests.rs b/crates/claude-connector/src/backend/tests.rs index 61ee59d..ecd98fa 100644 --- a/crates/claude-connector/src/backend/tests.rs +++ b/crates/claude-connector/src/backend/tests.rs @@ -2160,7 +2160,7 @@ fn rate_limit_event_updates_quota_cell() { #[test] fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself() { - let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":4102444800,"status":"rejected"},"seven_day":{"resetsAt":4102531200,"status":"allowed"},"idle":{"utilization":0,"resetsAt":4102444800,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#; + let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":4102444800,"status":"rejected"},"seven_day":{"utilization":0,"resetsAt":4102531200,"status":"allowed"},"seven_day_overage_included":{"utilization":1,"resetsAt":4102531200,"status":"rejected"},"idle":{"utilization":0,"resetsAt":4102444800,"status":"allowed"}}}}"#; let rig = rig(vec![Script::lines(vec![ init_line(MODEL, SESSION), event.to_string(), @@ -2170,12 +2170,12 @@ fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself( assert_eq!(kind, FailureKind::RateLimited); let mut rejected = window("five_hour", 1000, 4_102_444_800_000); rejected.status = "rejected".to_string(); - // Unknown is never zero: a window missing a field or of another type is - // skipped, not read as empty. - // A window at exactly zero use is a reading, and is kept. + // A window at exactly zero use is a reading, and is kept. Only the two + // windows pacing reads are taken: a per-model bucket that reads rejected + // must not close every lease, and no other name is a window of the plan. assert_eq!( readings(&rig), - vec![rejected, window("idle", 0, 4_102_444_800_000)] + vec![rejected, window("seven_day", 0, 4_102_531_200_000)] ); } @@ -2746,3 +2746,47 @@ fn a_turn_after_a_schema_turn_is_seeded_not_resumed() { assert!(!argv.iter().any(|a| a == "--resume"), "{argv:?}"); assert!(!argv.iter().any(|a| a == "--json-schema")); } + +// ---- the meter's call ---- + +/// One real call of the pinned CLI (2.1.312), captured on Yggdrasil on +/// 2026-11-03 with the connector's own argv: haiku, no tools, the trivial +/// prompt. Its `rate_limit_event` names only the week. +const METER_CALL: &str = include_str!("../../tests/fixtures/claude-2.1.312-meter-call.jsonl"); + +#[test] +fn the_meter_call_reads_the_week_from_a_real_captured_event() { + let rig = rig(vec![Script::lines( + METER_CALL.lines().map(str::to_string).collect(), + )]); + let usage = rig + .backend + .read_meter() + .expect("the captured call completes"); + assert_eq!((usage.input_tokens, usage.output_tokens), (384, 4)); + // 0.06 of the week used; it resets at 1794261600 (unix s). + assert_eq!( + readings(&rig), + vec![window("seven_day", 60, 1_794_261_600_000)] + ); + let argv = rig.seen().argv; + let after = |flag: &str| argv[argv.iter().position(|a| a == flag).unwrap() + 1].clone(); + assert_eq!(after("--model"), METER_MODEL); + assert_eq!(after("--tools"), "", "the call can use no tool"); + assert!( + argv.iter().any(|a| a == "--no-session-persistence"), + "the meter's transcript is nobody's to keep: {argv:?}" + ); + assert_eq!(rig.spawns(), 1, "one child, never run again"); +} + +#[test] +fn a_failed_meter_call_names_its_kind_and_leaves_the_cell_alone() { + let rig = rig(vec![Script::lines(vec![ + init_line(MODEL, SESSION), + error_result_line(r#","api_error_status":500"#), + ])]); + let failed = rig.backend.read_meter().expect_err("the call fails"); + assert!(!failed.is_empty() && !failed.contains(' '), "{failed}"); + assert!(quota_of(&rig).windows.is_empty()); +} diff --git a/crates/claude-connector/src/daemon.rs b/crates/claude-connector/src/daemon.rs index 1a1a80b..e64e473 100644 --- a/crates/claude-connector/src/daemon.rs +++ b/crates/claude-connector/src/daemon.rs @@ -22,6 +22,8 @@ use crate::backend::readiness; use crate::lease::{LeaseCaller, LeaseLimits, serve_leases, user_uid}; #[cfg(target_os = "linux")] use crate::unit::SystemdUnits; +#[cfg(unix)] +use crate::meter; use crate::{ ClaudeCodeBackend, ClaudeCodeConfig, ClaudeCodeLimits, CredentialTokenSource, StderrSink, }; @@ -535,6 +537,16 @@ pub fn check_ready(config: &ClaudeConnectorConfig) -> Result<(), ConfigError> { /// Loads the document and serves until the listener fails. #[cfg(unix)] pub fn serve_config(path: &Path) -> Result<(), ConfigError> { + serve_config_with(path, |_| ()) +} + +/// [`serve_config`] handing the quota cell to `started` once the meter runs: +/// a test stands in for the swarm's lease requests only through here. +#[cfg(unix)] +fn serve_config_with( + path: &Path, + started: impl FnOnce(&Arc), +) -> Result<(), ConfigError> { let config = load_config(path)?; let admissions = readiness_of(&config)?; let listen = config.validate()?; @@ -552,9 +564,15 @@ pub fn serve_config(path: &Path) -> Result<(), ConfigError> { ) .map_err(|error| ConfigError::from(ServeError::Service(error)))?, )); + // The meter is running before the first lease can ask, so that lease can + // wait for its first read. + let quota = backend.quota(); + meter::start(quota.clone(), Box::new(backend.clone())) + .map_err(|_| ConfigError::Serving("meter"))?; + started("a); #[cfg(target_os = "linux")] if let Some(socket) = &config.lease_socket { - start_leases(&config, socket, service.clone(), backend.quota())?; + start_leases(&config, socket, service.clone(), quota.clone())?; } serve_service(listener, config.serve_limits(), service, backend)?; Ok(()) @@ -564,6 +582,8 @@ pub fn serve_config(path: &Path) -> Result<(), ConfigError> { mod tests { use std::sync::{Arc, Mutex}; + use sha2::Digest; + use super::*; /// A log that a test can read back. @@ -580,6 +600,104 @@ mod tests { } } + /// `serve` starts the meter beside the listener, wired to the backend's own + /// cell: it makes no call while nobody asks, and once a lease is asked for + /// it runs the claude binary and serve stays up. The binary is a script + /// that replays a real captured call. + #[cfg(unix)] + #[test] + fn serve_config_starts_the_meter_and_stays_up() { + use std::os::unix::fs::PermissionsExt; + + const CALL: &str = include_str!("../tests/fixtures/claude-2.1.312-meter-call.jsonl"); + let dir = + std::env::temp_dir().join(format!("claude-connector-serve-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let at = |name: &str| dir.join(name); + for sub in ["work", "home", "prompts", "credentials"] { + std::fs::create_dir_all(at(sub)).unwrap(); + } + std::fs::set_permissions(at("prompts"), std::fs::Permissions::from_mode(0o700)).unwrap(); + std::fs::write(at("call.jsonl"), CALL).unwrap(); + let script = format!( + "#!/bin/sh\ncat > /dev/null\necho spawned >> '{}'\ncat '{}'\n", + at("spawned").display(), + at("call.jsonl").display() + ); + std::fs::write(at("claude"), &script).unwrap(); + std::fs::set_permissions(at("claude"), std::fs::Permissions::from_mode(0o755)).unwrap(); + std::fs::write(at("caller.key"), "k".repeat(48)).unwrap(); + std::fs::write(at("credentials/claude-token"), "sk-ant-oat01-fake-token").unwrap(); + std::fs::write(at("credentials/claude-handle-secret"), "s".repeat(48)).unwrap(); + let config = ClaudeConnectorConfig { + listen: "127.0.0.1:0".to_string(), + replay_store: at("replay.cc"), + claude_binary: at("claude"), + claude_binary_sha256: sha2::Sha256::digest(script.as_bytes()).into(), + work_dir: at("work"), + config_dir: at("home"), + prompt_dir: at("prompts"), + instance: "persona-a".to_string(), + token_credential: "claude-token".to_string(), + handle_secret_credential: "claude-handle-secret".to_string(), + deadline_ms: 600_000, + max_line_bytes: 4 << 20, + max_total_bytes: 64 << 20, + stderr_tail_bytes: 8192, + max_frame_bytes: 2 << 20, + max_connections: 8, + socket_timeout_ms: 10_000, + max_expiry_skew_ms: 300_000, + callers: vec![CallerConfig { + caller_runtime_id: "persona-a-caller".to_string(), + connection_key_file: at("caller.key"), + connection_key_epoch: 1, + allowed_models: vec!["claude-haiku-5-5".to_string()], + max_concurrent_requests: 4, + max_payload_bytes: 1 << 20, + max_output_tokens: 32_768, + }], + lease_socket: None, + lease_callers: Vec::new(), + lease_max_ms: 3_600_000, + }; + write_config(&at("connector.cc"), &config).unwrap(); + // SAFETY: set once, to a directory only this test reads, before the + // serve thread it starts. + unsafe { std::env::set_var("CREDENTIALS_DIRECTORY", at("credentials")) }; + + let path = at("connector.cc"); + let lease = std::sync::Arc::new(std::sync::Mutex::new(None)); + let swarm = lease.clone(); + let served = std::thread::spawn(move || { + serve_config_with(&path, |cell| { + *swarm.lock().unwrap() = Some(cell.clone()); + }) + }); + + let deadline = std::time::Instant::now() + Duration::from_secs(30); + let cell = loop { + if let Some(cell) = lease.lock().unwrap().clone() { + break cell; + } + assert!(std::time::Instant::now() < deadline, "serve never started"); + std::thread::sleep(Duration::from_millis(20)); + }; + std::thread::sleep(Duration::from_millis(500)); + assert!(!at("spawned").exists(), "a quiet swarm spends nothing"); + let _lease = cell.asking(); + + while !at("spawned").exists() { + assert!( + std::time::Instant::now() < deadline, + "the meter never ran the claude binary" + ); + std::thread::sleep(Duration::from_millis(50)); + } + assert!(!served.is_finished(), "serve returned after starting the meter"); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn serve_errors_name_a_field_or_stage_and_never_a_stored_value() { const CANARY: &str = "canary-path-8e2a"; diff --git a/crates/claude-connector/src/launch.rs b/crates/claude-connector/src/launch.rs index b233c09..1f8dbd0 100644 --- a/crates/claude-connector/src/launch.rs +++ b/crates/claude-connector/src/launch.rs @@ -107,13 +107,15 @@ fn effort(effort: ReasoningEffort) -> &'static str { /// rather than an argument because one argument is capped near 128 KiB /// (E2BIG) and instructions are not, the child has /// no tools, no MCP servers, no settings, no slash commands and nobody to ask -/// for permission, and a resume always forks. A request's output schema goes -/// to the CLI as `--json-schema`; without one the flag is absent. +/// for permission, and a resume always forks. An ephemeral call (the +/// meter's) keeps no session on disk. A request's output schema goes to the +/// CLI as `--json-schema`; without one the flag is absent. pub(crate) fn argv( config: &ClaudeCodeConfig, request: &ProviderRequest, plan: &Plan, system_prompt_file: &Path, + ephemeral: bool, ) -> Vec { let mut argv: Vec = [ config.claude_binary.to_string_lossy().as_ref(), @@ -139,6 +141,9 @@ pub(crate) fn argv( .into_iter() .map(str::to_string) .collect(); + if ephemeral { + argv.push("--no-session-persistence".to_string()); + } if let Some(level) = request.reasoning_effort { argv.extend(["--effort".to_string(), effort(level).to_string()]); } diff --git a/crates/claude-connector/src/lease.rs b/crates/claude-connector/src/lease.rs index 0d4a582..7e503aa 100644 --- a/crates/claude-connector/src/lease.rs +++ b/crates/claude-connector/src/lease.rs @@ -468,8 +468,14 @@ impl Lender { ); } // The lease is claimed before the token is read, let alone lent. The - // plan the refusal reads is the one this process last observed. - let plan = self.quota.snapshot(); + // plan the refusal reads is the one this process last observed, but a + // cell with no weekly reading (a restart empties it) gets the meter's + // first read, bounded, before the lease is lent "unobserved" for its + // whole life. The request is the swarm's sign of life, however it + // ends: the meter reads while one is held or lately asked for. + let _asking = self.quota.asking(); + let plan = self.quota.snapshot_with_week(FIRST_READ_WAIT); + let now = unix_ms(); let admission = self .service .lock() @@ -603,6 +609,10 @@ impl Lender { } } +/// The longest a lease waits for the meter's first read of the week. A call +/// takes a few seconds; the holder's own wait for the grant is far longer. +const FIRST_READ_WAIT: Duration = Duration::from_secs(10); + /// How often a held lease looks again at its holder's unit invocation. const HOLD_POLL_MS: u64 = 500; diff --git a/crates/claude-connector/src/lib.rs b/crates/claude-connector/src/lib.rs index 70a2e53..671ee36 100644 --- a/crates/claude-connector/src/lib.rs +++ b/crates/claude-connector/src/lib.rs @@ -10,6 +10,7 @@ mod handle; mod launch; #[cfg(target_os = "linux")] mod lease; +mod meter; mod process; mod quota; mod stream; diff --git a/crates/claude-connector/src/meter.rs b/crates/claude-connector/src/meter.rs new file mode 100644 index 0000000..2f572a2 --- /dev/null +++ b/crates/claude-connector/src/meter.rs @@ -0,0 +1,543 @@ +//! The meter: keeps the plan's weekly reading fresh while the swarm is +//! working. The subscription reports its windows only inside the CLI's +//! `rate_limit_event`, on a call the CLI makes, and Personas make theirs under +//! leases that report nothing back. So when the `seven_day` reading in the +//! [`QuotaCell`] is `STALE` old, the meter makes the smallest real call +//! through the backend's own path (one `claude -p` on the cheapest model with +//! a trivial prompt) and the stream fills the cell exactly as it does for any +//! call. The meter holds no credential, speaks no protocol and writes nothing +//! itself. +//! +//! The meter is driven by the connector's own lease state. It reads only +//! while a lease is held or one was asked for less than `STALE` ago: a held +//! swarm brake stops the units, they ask for nothing, and a braked or quiet +//! swarm spends nothing. A call that fails, or completes without refreshing +//! the reading, is not retried at the staleness period: the wait doubles per +//! consecutive miss, up to `CEILING`. + +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use model_connector::Usage; + +use crate::backend::ClaudeCodeBackend; +use crate::quota::{EXPIRED, QuotaCell, WEEK}; + +/// A reading this old, or older, is stale; also the least gap between calls. +const STALE: Duration = Duration::from_secs(15 * 60); +/// The longest any backoff may wait; also how long an idle meter sleeps +/// before looking again, since a lease request wakes it. +const CEILING: Duration = Duration::from_secs(6 * 60 * 60); + +/// The smallest real call, as the backend makes it. +pub(crate) trait Probe: Send { + /// What the call billed, or the failure's kind. Never a message: a + /// provider's words do not enter the meter's log. + fn call(&self) -> Result; +} + +impl Probe for Arc { + fn call(&self) -> Result { + self.read_meter() + } +} + +/// Why the meter is waiting; logged once per change. +#[derive(Debug, Clone, PartialEq, Eq)] +enum State { + Read, + Idle, + /// The call failed; the kind is the failure's kind, never its message. + Failed(String), + Unread, +} + +impl State { + fn line(&self) -> String { + match self { + Self::Read => "meter: weekly reading refreshed".to_string(), + Self::Idle => "meter: no lease asked lately, making no call".to_string(), + Self::Failed(kind) => format!("meter: the call failed ({kind}), backing off"), + Self::Unread => { + "meter: the call completed without a weekly reading, backing off".to_string() + } + } + } +} + +struct Meter { + cell: Arc, + probe: Box, + log: Box, + /// No call before this instant, unix ms. + not_before_ms: u64, + /// Consecutive calls that left the reading unrefreshed. + misses: u32, + last_state: Option, +} + +impl Meter { + fn new(cell: Arc, probe: Box, log: Box) -> Self { + Self { + cell, + probe, + log, + not_before_ms: 0, + misses: 0, + last_state: None, + } + } + + /// Makes the call if one is due at `now_ms`; returns how long to wait + /// before looking again. Nothing is due while no lease is held or lately + /// asked for, nor before the last miss's backoff has passed, nor while the + /// weekly reading is younger than `STALE`. + fn tick(&mut self, now_ms: u64) -> Duration { + if !self.cell.wanted_at(now_ms, STALE) { + self.report(State::Idle); + return CEILING; + } + let due_at = self.due_at_ms(); + if now_ms < due_at { + return Duration::from_millis(due_at - now_ms); + } + // A failed call may still have refreshed the reading: a rejected turn + // reports its windows and then errors. Only the cell says whether the + // week was read. + let called = self.probe.call(); + let state = if self.refreshed_since(now_ms) { + self.misses = 0; + State::Read + } else { + self.misses = self.misses.saturating_add(1); + match called { + Ok(_) => State::Unread, + Err(kind) => State::Failed(kind), + } + }; + self.report(state); + self.not_before_ms = now_ms.saturating_add(duration_ms(backoff(self.misses))); + Duration::from_millis(self.due_at_ms().saturating_sub(now_ms)) + } + + /// The earliest instant the next call may happen. A window past its reset + /// described a week that is over, so it is due at once. + fn due_at_ms(&self) -> u64 { + let stale_at = self + .cell + .snapshot() + .windows + .iter() + .find(|quota| quota.window == WEEK) + .map_or(0, |quota| { + if quota.status == EXPIRED { + 0 + } else { + quota.observed_at_unix_ms.saturating_add(duration_ms(STALE)) + } + }); + stale_at.max(self.not_before_ms) + } + + fn refreshed_since(&self, since_ms: u64) -> bool { + self.cell + .snapshot() + .windows + .iter() + .any(|quota| quota.window == WEEK && quota.observed_at_unix_ms >= since_ms) + } + + fn report(&mut self, state: State) { + if self.last_state.as_ref() != Some(&state) { + (self.log)(&state.line()); + self.last_state = Some(state); + } + } + + fn run(mut self) { + loop { + let seen = self.cell.asks_so_far(); + let wait = self.tick(unix_ms()); + self.cell.sleep_until_asked(seen, wait); + } + } +} + +/// The wait after `misses` consecutive unrefreshed calls: the staleness +/// period, doubling per miss beyond the first, up to the ceiling. +fn backoff(misses: u32) -> Duration { + let doublings = misses.saturating_sub(1).min(16); + STALE.saturating_mul(1 << doublings).min(CEILING) +} + +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 meter thread. It writes only through the calls it makes, and +/// logs one line to stderr when its state changes. From here a lease asked of +/// a cell with no weekly reading waits for the meter's first read. +pub(crate) fn start( + cell: Arc, + probe: Box, +) -> std::io::Result> { + cell.attach_meter(); + let meter = Meter::new( + cell, + probe, + Box::new(|line| eprintln!("claude-connector: {line}")), + ); + std::thread::Builder::new() + .name("claude-meter".to_string()) + .spawn(move || meter.run()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::quota::Asking; + use model_connector::PlanQuota; + use std::sync::Mutex; + use std::sync::atomic::{AtomicUsize, Ordering}; + + const MIN: u64 = 60_000; + const NOW: u64 = 1_800_000_000_000; + const RESET: u64 = 4_102_444_800_000; + + fn reading(status: &str) -> Vec { + vec![PlanQuota { + window: WEEK.to_string(), + utilization_permille: Some(60), + resets_at_unix_ms: RESET, + status: status.to_string(), + observed_at_unix_ms: 0, + }] + } + + /// What the next call does: the instant its stream stamps a weekly + /// reading (if any) and the kind it then fails with (if any). + #[derive(Clone, Default)] + struct Script { + reading_at: Option, + status: String, + fails_with: Option, + } + + /// A call that answers as scripted and feeds the cell the way the stream + /// does, before it fails, as a rejected turn does. + struct Calls { + cell: Arc, + made: Arc, + script: Arc>, + } + + impl Probe for Calls { + fn call(&self) -> Result { + self.made.fetch_add(1, Ordering::SeqCst); + let script = self.script.lock().unwrap().clone(); + if let Some(at) = script.reading_at { + self.cell.observe_at(at, reading(&script.status)); + } + if let Some(kind) = script.fails_with { + return Err(kind); + } + Ok(Usage { + input_tokens: 385, + cache_read_tokens: 0, + cache_write_tokens: 0, + output_tokens: 4, + reasoning_tokens: None, + }) + } + } + + struct Rig { + meter: Meter, + cell: Arc, + made: Arc, + script: Arc>, + log: Arc>>, + /// The lease the swarm holds until a test lets go of it. + lease: Option, + } + + fn rig() -> Rig { + let cell = Arc::new(QuotaCell::new()); + let made = Arc::new(AtomicUsize::new(0)); + let script = Arc::new(Mutex::new(Script::default())); + let log = Arc::new(Mutex::new(Vec::new())); + let sink = log.clone(); + let meter = Meter::new( + cell.clone(), + Box::new(Calls { + cell: cell.clone(), + made: made.clone(), + script: script.clone(), + }), + Box::new(move |line| sink.lock().unwrap().push(line.to_string())), + ); + let lease = Some(cell.asking_at(NOW)); + Rig { + meter, + cell, + made, + script, + log, + lease, + } + } + + impl Rig { + fn reads(&self, at: u64) { + *self.script.lock().unwrap() = Script { + reading_at: Some(at), + status: "allowed".to_string(), + fails_with: None, + }; + } + fn reads_nothing(&self) { + *self.script.lock().unwrap() = Script::default(); + } + fn fails(&self, kind: &str) { + *self.script.lock().unwrap() = Script { + fails_with: Some(kind.to_string()), + ..Script::default() + }; + } + fn calls(&self) -> usize { + self.made.load(Ordering::SeqCst) + } + fn lines(&self) -> Vec { + self.log.lock().unwrap().clone() + } + } + + #[test] + fn an_unread_week_triggers_one_call_and_the_fresh_reading_ends_the_asking() { + let mut rig = rig(); + rig.reads(NOW); + assert_eq!(rig.meter.tick(NOW), STALE, "fresh for the staleness period"); + assert_eq!(rig.calls(), 1); + assert_eq!( + rig.meter.tick(NOW + 5 * MIN), + Duration::from_millis(10 * MIN) + ); + assert_eq!(rig.calls(), 1, "a fresh reading makes no call"); + } + + #[test] + fn a_stale_reading_triggers_a_call_at_the_period_and_not_before() { + let mut rig = rig(); + rig.cell.observe_at(NOW, reading("allowed")); + rig.reads(NOW + 15 * MIN); + rig.meter.tick(NOW + 15 * MIN - 1); + assert_eq!(rig.calls(), 0, "one millisecond short of stale"); + rig.meter.tick(NOW + 15 * MIN); + assert_eq!(rig.calls(), 1); + } + + #[test] + fn a_window_past_its_reset_is_read_at_once_not_after_the_period() { + let mut rig = rig(); + let now = unix_ms(); + let mut over = reading("allowed").remove(0); + over.resets_at_unix_ms = 1; + rig.cell.observe_at(now, vec![over]); + rig.reads(now); + rig.meter.tick(now); + assert_eq!(rig.calls(), 1); + } + + #[test] + fn a_swarm_that_asks_for_nothing_spends_nothing_however_stale_the_reading() { + let mut rig = rig(); + drop(rig.lease.take()); + for step in 0..10 { + let at = NOW + 15 * MIN + step * 60 * MIN; + assert_eq!(rig.meter.tick(at), CEILING); + } + assert_eq!(rig.calls(), 0); + assert_eq!(rig.lines(), vec![State::Idle.line()]); + } + + #[test] + fn a_lease_asked_for_wakes_the_meter_and_the_empty_cell_is_read() { + let mut rig = rig(); + drop(rig.lease.take()); + rig.reads(NOW + 180 * MIN); + rig.meter.tick(NOW + 180 * MIN); + assert_eq!(rig.calls(), 0, "asked three hours ago and let go"); + let _later = rig.cell.asking_at(NOW + 180 * MIN); + rig.meter.tick(NOW + 180 * MIN); + assert_eq!(rig.calls(), 1); + } + + #[test] + fn a_lease_let_go_is_still_wanted_for_exactly_one_staleness_period() { + let mut rig = rig(); + drop(rig.lease.take()); + rig.reads(NOW + 15 * MIN - 1); + rig.meter.tick(NOW + 15 * MIN - 1); + assert_eq!(rig.calls(), 1, "one millisecond inside the period"); + rig.reads(NOW + 30 * MIN); + rig.meter.tick(NOW + 30 * MIN); + assert_eq!(rig.calls(), 1, "a period after the last ask: unwanted"); + } + + #[test] + fn a_lease_held_for_four_hours_keeps_the_week_read_throughout() { + let mut rig = rig(); + for step in 0..=16 { + let at = NOW + step * 15 * MIN; + rig.reads(at); + rig.meter.tick(at); + } + assert_eq!(rig.calls(), 17, "one read per period, the lease never lapsing"); + } + + #[test] + fn a_failed_call_leaves_the_cell_and_backs_off_instead_of_hammering() { + let mut rig = rig(); + rig.cell.observe_at(NOW, reading("allowed")); + let before = rig.cell.snapshot(); + rig.fails("Provider"); + let mut at = NOW + 15 * MIN; + let mut waits = Vec::new(); + for _ in 0..4 { + waits.push(rig.meter.tick(at)); + at += duration_ms(*waits.last().unwrap()); + } + assert_eq!(rig.calls(), 4); + assert_eq!(rig.cell.snapshot(), before, "the cell is untouched"); + assert_eq!( + waits, + [15, 30, 60, 120].map(|m| Duration::from_secs(m * 60)), + "the wait doubles per consecutive miss" + ); + rig.meter.tick(at - MIN); + assert_eq!(rig.calls(), 4, "nothing is asked before the backoff ends"); + assert_eq!( + rig.lines(), + vec!["meter: the call failed (Provider), backing off"] + ); + } + + #[test] + fn the_failure_kind_is_logged_again_when_it_changes_and_not_when_it_repeats() { + let mut rig = rig(); + rig.fails("Authentication"); + rig.meter.tick(NOW); + rig.meter.tick(NOW + 15 * MIN); + rig.fails("InvalidRequest"); + rig.meter.tick(NOW + 45 * MIN); + assert_eq!( + rig.lines(), + vec![ + "meter: the call failed (Authentication), backing off", + "meter: the call failed (InvalidRequest), backing off", + ] + ); + } + + #[test] + fn the_backoff_stops_at_the_ceiling_and_one_good_reading_resets_it() { + assert_eq!(backoff(0), STALE); + assert_eq!(backoff(1), STALE); + assert_eq!(backoff(u32::MAX), CEILING); + let mut rig = rig(); + rig.fails("Provider"); + rig.meter.tick(NOW); + rig.meter.tick(NOW + 15 * MIN); + assert_eq!(rig.meter.misses, 2); + rig.reads(NOW + 45 * MIN); + rig.meter.tick(NOW + 45 * MIN); + assert_eq!(rig.meter.misses, 0); + } + + #[test] + fn a_call_that_completes_without_a_weekly_reading_is_a_miss_too() { + let mut rig = rig(); + rig.reads_nothing(); + assert_eq!(rig.meter.tick(NOW), STALE); + assert_eq!(rig.meter.tick(NOW + 15 * MIN), 2 * STALE); + assert_eq!(rig.calls(), 2); + assert_eq!(rig.lines(), vec![State::Unread.line()]); + } + + #[test] + fn a_rejected_turn_that_refreshed_the_reading_and_then_errored_is_no_miss() { + let mut rig = rig(); + *rig.script.lock().unwrap() = Script { + reading_at: Some(NOW), + status: "rejected".to_string(), + fails_with: Some("RateLimited".to_string()), + }; + assert_eq!(rig.meter.tick(NOW), STALE, "the next read is a period away"); + assert_eq!(rig.meter.misses, 0); + assert_eq!(rig.lines(), vec![State::Read.line()]); + rig.reads(NOW + 15 * MIN); + rig.meter.tick(NOW + 15 * MIN); + assert_eq!(rig.calls(), 2, "a rejected reading is kept fresh"); + } + + #[test] + fn a_reading_from_before_the_call_does_not_count_as_its_answer() { + let mut rig = rig(); + // A slow concurrent sensor stamped earlier than this call began. + rig.reads(NOW - 1); + rig.meter.tick(NOW); + assert_eq!(rig.meter.misses, 1); + } + + #[test] + fn the_log_says_what_the_meter_is_doing_in_words() { + assert_eq!( + State::Idle.line(), + "meter: no lease asked lately, making no call" + ); + assert_eq!(State::Read.line(), "meter: weekly reading refreshed"); + assert_eq!( + State::Unread.line(), + "meter: the call completed without a weekly reading, backing off" + ); + } + + #[test] + fn the_clock_is_the_system_clock_in_unix_milliseconds() { + let before = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); + let read = unix_ms(); + let after = SystemTime::now().duration_since(UNIX_EPOCH).unwrap(); + assert!((before.as_millis() as u64..=after.as_millis() as u64).contains(&read)); + } + + #[test] + fn a_started_meter_sleeps_until_a_lease_asks_and_then_reads() { + let cell = Arc::new(QuotaCell::new()); + let made = Arc::new(AtomicUsize::new(0)); + let probe = Calls { + cell: cell.clone(), + made: made.clone(), + script: Arc::new(Mutex::new(Script { + reading_at: Some(unix_ms() + 60 * MIN), + status: "allowed".to_string(), + fails_with: None, + })), + }; + start(cell.clone(), Box::new(probe)).unwrap(); + std::thread::sleep(Duration::from_millis(300)); + assert_eq!(made.load(Ordering::SeqCst), 0, "no lease, no call"); + let _lease = cell.asking(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while made.load(Ordering::SeqCst) == 0 { + assert!(std::time::Instant::now() < deadline, "the ask woke nobody"); + std::thread::sleep(Duration::from_millis(20)); + } + assert_eq!(cell.snapshot_with_week(Duration::ZERO).windows.len(), 1); + } +} diff --git a/crates/claude-connector/src/quota.rs b/crates/claude-connector/src/quota.rs index 450d3e2..fe71383 100644 --- a/crates/claude-connector/src/quota.rs +++ b/crates/claude-connector/src/quota.rs @@ -1,38 +1,150 @@ -//! 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 one sensor observes it: +//! the CLI's `rate_limit_event` on the connector's own calls, which the +//! `meter` makes when the week's reading goes stale. The latest observation of +//! each window is kept here and published as the usage readout and lent with +//! each lease. use std::collections::BTreeMap; -use std::sync::Mutex; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use model_connector::{PlanQuota, PlanQuotaSnapshot}; +/// The window the weekly pacing reads, and the one the meter keeps fresh. +pub(crate) const WEEK: &str = "seven_day"; +/// The windows the pacing reads, and so the only ones the cell keeps. +pub(crate) const PACED: [&str; 2] = ["five_hour", WEEK]; + +/// How often a lease waiting for its first reading looks at the cell again. +const FIRST_READ_POLL: Duration = Duration::from_millis(50); + /// The latest observation of each plan window, shared by the backend that /// feeds it and whatever reads it out. Each window carries the time of its /// own observation: an event that names one window says nothing about the /// others. +/// +/// It also keeps who is asking: a lease request is the swarm's only sign of +/// life the connector sees, and the meter reads the plan only while someone +/// asks. #[derive(Default)] -pub struct QuotaCell(Mutex>); +pub struct QuotaCell { + windows: Mutex>, + asks: Mutex, + asked: Condvar, + metered: AtomicBool, +} + +#[derive(Default)] +struct Asks { + /// Lease requests in flight or held. + live: usize, + /// Requests ever made; the meter sleeps until it changes. + count: u64, + /// When the latest request came, unix ms. + last_ms: Option, +} + +/// A lease request while it is in flight or held. Dropping it is the lease +/// ending, by whatever road. +pub(crate) struct Asking(Arc); + +impl Drop for Asking { + fn drop(&mut self) { + let mut asks = self.0.asks.lock().unwrap_or_else(|e| e.into_inner()); + asks.live = asks.live.saturating_sub(1); + } +} impl QuotaCell { pub fn new() -> Self { Self::default() } - /// Records the windows one event reported, stamping each with now. Each - /// replaces that window's earlier observation, and windows the event did - /// not mention stay as they were, stamp included. + /// Notes a lease request, now, and wakes the meter. + pub(crate) fn asking(self: &Arc) -> Asking { + self.asking_at(unix_ms()) + } + + pub(crate) fn asking_at(self: &Arc, now: u64) -> Asking { + { + let mut asks = self.asks.lock().unwrap_or_else(|e| e.into_inner()); + asks.live += 1; + asks.count += 1; + asks.last_ms = Some(now); + } + self.asked.notify_all(); + Asking(self.clone()) + } + + /// Whether a lease is held, or one was asked for less than `within` ago. + pub(crate) fn wanted_at(&self, now: u64, within: Duration) -> bool { + let asks = self.asks.lock().unwrap_or_else(|e| e.into_inner()); + asks.live > 0 + || asks.last_ms.is_some_and(|last| { + now < last.saturating_add(u64::try_from(within.as_millis()).unwrap_or(u64::MAX)) + }) + } + + pub(crate) fn asks_so_far(&self) -> u64 { + self.asks.lock().unwrap_or_else(|e| e.into_inner()).count + } + + /// Sleeps up to `limit`, or until a request beyond the first `seen` comes. + pub(crate) fn sleep_until_asked(&self, seen: u64, limit: Duration) { + let asks = self.asks.lock().unwrap_or_else(|e| e.into_inner()); + drop( + self.asked + .wait_timeout_while(asks, limit, |asks| asks.count == seen) + .unwrap_or_else(|e| e.into_inner()), + ); + } + + /// Declares that a meter reads this cell, so a first lease may wait for it. + pub(crate) fn attach_meter(&self) { + self.metered.store(true, Ordering::SeqCst); + } + + /// The windows as they stand, but with no weekly reading yet (a restart + /// leaves the cell empty) a metered cell is given up to `within` to get + /// one, so a lease is not lent "unobserved" for want of a few seconds. + pub(crate) fn snapshot_with_week(&self, within: Duration) -> PlanQuotaSnapshot { + let deadline = Instant::now() + within; + loop { + let snapshot = self.snapshot(); + let read = snapshot + .windows + .iter() + .any(|quota| quota.window == WEEK && quota.status != EXPIRED); + let now = Instant::now(); + if read || !self.metered.load(Ordering::SeqCst) || now >= deadline { + return snapshot; + } + std::thread::sleep(FIRST_READ_POLL.min(deadline - now)); + } + } + + /// Records the windows one event reported, stamping each with now. See + /// [`QuotaCell::observe_at`] for how a reading combines with what is held. pub fn observe(&self, windows: impl IntoIterator) { self.observe_at(unix_ms(), windows); } - fn observe_at(&self, now: u64, windows: impl IntoIterator) { - let mut held = self.0.lock().unwrap_or_else(|e| e.into_inner()); + /// Records readings taken at `at` (unix ms). A reading older than the + /// held one is dropped, so a slow call cannot replace what a faster one + /// saw meanwhile. Windows the readings did not mention stay as they were, + /// stamp included. + pub fn observe_at(&self, at: u64, windows: impl IntoIterator) { + let mut held = self.windows.lock().unwrap_or_else(|e| e.into_inner()); for mut quota in windows { - quota.observed_at_unix_ms = now; - held.insert(quota.window.clone(), quota); + quota.observed_at_unix_ms = at; + if held + .get("a.window) + .is_none_or(|current| quota.observed_at_unix_ms >= current.observed_at_unix_ms) + { + held.insert(quota.window.clone(), quota); + } } } @@ -44,7 +156,7 @@ impl QuotaCell { /// `expired` with no utilization: the reading described a window that no /// longer exists. fn snapshot_at(&self, now: u64) -> PlanQuotaSnapshot { - let held = self.0.lock().unwrap_or_else(|e| e.into_inner()); + let held = self.windows.lock().unwrap_or_else(|e| e.into_inner()); PlanQuotaSnapshot { windows: held .values() @@ -62,7 +174,7 @@ impl QuotaCell { } /// The status of a window whose reset has passed since it was observed. -const EXPIRED: &str = "expired"; +pub(crate) const EXPIRED: &str = "expired"; fn unix_ms() -> u64 { SystemTime::now() @@ -133,6 +245,38 @@ mod tests { assert_eq!(expired[0].resets_at_unix_ms, RESET); } + fn with_status(mut quota: PlanQuota, status: &str) -> PlanQuota { + quota.status = status.to_string(); + quota + } + + fn held(cell: &QuotaCell) -> (String, Option, u64) { + let window = cell.snapshot_at(250).windows.remove(0); + ( + window.status, + window.utilization_permille, + window.observed_at_unix_ms, + ) + } + + #[test] + fn a_reading_older_than_the_held_one_is_dropped_and_an_equal_one_replaces() { + let cell = QuotaCell::new(); + cell.observe_at(200, [quota("five_hour", 80, 0)]); + cell.observe_at(100, [quota("five_hour", 10, 0)]); + assert_eq!(held(&cell), ("allowed".to_string(), Some(80), 200)); + cell.observe_at(200, [quota("five_hour", 90, 0)]); + assert_eq!(held(&cell), ("allowed".to_string(), Some(90), 200)); + } + + #[test] + fn a_newer_reading_lifts_a_rejection() { + let cell = QuotaCell::new(); + cell.observe_at(100, [with_status(quota("five_hour", 1000, 0), "rejected")]); + cell.observe_at(200, [quota("five_hour", 300, 0)]); + assert_eq!(held(&cell), ("allowed".to_string(), Some(300), 200)); + } + #[test] fn the_live_clock_stamps_and_expires() { let cell = QuotaCell::new(); @@ -148,4 +292,114 @@ mod tests { assert_eq!(windows[0].utilization_permille, Some(80)); assert_eq!(windows[1].status, "expired"); } + + const HOUR: u64 = 3_600_000; + + #[test] + fn nobody_wants_the_plan_before_anyone_asks() { + let cell = Arc::new(QuotaCell::new()); + assert!(!cell.wanted_at(0, Duration::from_secs(900))); + } + + #[test] + fn a_lease_in_flight_or_held_is_wanted_for_as_long_as_it_lives() { + let cell = Arc::new(QuotaCell::new()); + let lease = cell.asking_at(1_000); + let other = cell.asking_at(2_000); + let day = Duration::from_secs(86_400); + assert!(cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1))); + drop(lease); + assert!(cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1)), "one lease still lives"); + drop(other); + assert!(!cell.wanted_at(1_000 + 4 * HOUR, Duration::from_secs(1))); + assert!(cell.wanted_at(2_000 + 4 * HOUR, day), "within the window of the last ask"); + } + + #[test] + fn a_lease_let_go_is_wanted_until_exactly_the_period_after_its_ask() { + let cell = Arc::new(QuotaCell::new()); + drop(cell.asking_at(10_000)); + let period = Duration::from_millis(900); + assert!(cell.wanted_at(10_899, period)); + assert!(!cell.wanted_at(10_900, period)); + } + + #[test] + fn each_ask_is_counted_and_wakes_a_sleeper_at_once() { + let cell = Arc::new(QuotaCell::new()); + assert_eq!(cell.asks_so_far(), 0); + let sleeper = { + let cell = cell.clone(); + std::thread::spawn(move || { + let started = Instant::now(); + cell.sleep_until_asked(0, Duration::from_secs(30)); + started.elapsed() + }) + }; + std::thread::sleep(Duration::from_millis(100)); + drop(cell.asking()); + assert!(sleeper.join().unwrap() < Duration::from_secs(10), "woken, not timed out"); + assert_eq!(cell.asks_so_far(), 1); + // An ask that came after `seen` was read ends the sleep before it starts. + let started = Instant::now(); + cell.sleep_until_asked(0, Duration::from_secs(30)); + assert!(started.elapsed() < Duration::from_secs(10)); + // Nothing new: the sleep is the whole limit. + let started = Instant::now(); + cell.sleep_until_asked(1, Duration::from_millis(200)); + assert!(started.elapsed() >= Duration::from_millis(200)); + } + + #[test] + fn a_metered_cell_with_no_week_gives_a_first_lease_the_reading_when_it_lands() { + let cell = Arc::new(QuotaCell::new()); + cell.attach_meter(); + let feeder = { + let cell = cell.clone(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(200)); + cell.observe([quota(WEEK, 60, 0)]); + }) + }; + let plan = cell.snapshot_with_week(Duration::from_secs(20)); + feeder.join().unwrap(); + assert_eq!(plan.windows.len(), 1, "the lease waited for the week"); + assert_eq!(plan.windows[0].window, WEEK); + } + + #[test] + fn the_wait_for_the_first_reading_is_bounded() { + let cell = Arc::new(QuotaCell::new()); + cell.attach_meter(); + let started = Instant::now(); + let plan = cell.snapshot_with_week(Duration::from_millis(300)); + let waited = started.elapsed(); + assert!(plan.windows.is_empty(), "no reading came: unobserved, not hung"); + assert!(waited >= Duration::from_millis(300), "{waited:?}"); + assert!(waited < Duration::from_secs(10), "{waited:?}"); + } + + #[test] + fn a_cell_nobody_meters_or_that_holds_the_week_is_never_waited_on() { + let bare = QuotaCell::new(); + let started = Instant::now(); + assert!(bare.snapshot_with_week(Duration::from_secs(20)).windows.is_empty()); + let held = QuotaCell::new(); + held.attach_meter(); + held.observe([quota(WEEK, 60, 0)]); + assert_eq!(held.snapshot_with_week(Duration::from_secs(20)).windows.len(), 1); + assert!(started.elapsed() < Duration::from_secs(5)); + } + + #[test] + fn a_week_past_its_reset_is_waited_on_like_no_week() { + let cell = QuotaCell::new(); + cell.attach_meter(); + let mut over = quota(WEEK, 60, 0); + over.resets_at_unix_ms = 1; + cell.observe([over]); + let started = Instant::now(); + cell.snapshot_with_week(Duration::from_millis(300)); + assert!(started.elapsed() >= Duration::from_millis(300)); + } } diff --git a/crates/claude-connector/src/stream.rs b/crates/claude-connector/src/stream.rs index c5e2e41..039ff75 100644 --- a/crates/claude-connector/src/stream.rs +++ b/crates/claude-connector/src/stream.rs @@ -7,7 +7,7 @@ use model_connector::{FailureKind, PlanQuota, StopReason, Usage}; use serde::Deserialize; use serde_json::Value; -use crate::quota::QuotaCell; +use crate::quota::{PACED, QuotaCell}; /// The tool the CLI gives the model to deliver an answer that must match a /// `--json-schema`. Its call is the structured output, not a tool exchange. @@ -320,12 +320,16 @@ impl<'a> StreamState<'a> { /// `rate_limit_info.unifiedWindows`: utilization as a fraction of the window, /// reset as unix seconds. A window missing any of its three fields is skipped; /// an unknown reading is never zero. +/// Only the windows pacing reads are taken ([`PACED`]): the event also names +/// per-model buckets, and one of those reading `rejected` must not close every +/// lease. fn plan_windows(info: &Value) -> Vec { let Some(windows) = info.get("unifiedWindows").and_then(Value::as_object) else { return Vec::new(); }; windows .iter() + .filter(|(window, _)| PACED.contains(&window.as_str())) .filter_map(|(window, reading)| { let utilization = reading.get("utilization")?.as_f64()?; let resets_at = reading.get("resetsAt")?.as_u64()?; @@ -392,3 +396,50 @@ fn sum_usage(models: impl Iterator) -> Usage { } usage } + +#[cfg(test)] +mod tests { + use super::*; + + fn windows(body: &str) -> Vec<(String, Option, u64)> { + let info: Value = serde_json::from_str(&format!(r#"{{"unifiedWindows":{body}}}"#)).unwrap(); + plan_windows(&info) + .into_iter() + .map(|q| (q.window, q.utilization_permille, q.resets_at_unix_ms)) + .collect() + } + + #[test] + fn only_the_five_hour_and_seven_day_windows_are_taken() { + let taken = windows( + r#"{"five_hour":{"utilization":0.5,"resetsAt":2,"status":"allowed"}, + "seven_day":{"utilization":0.25,"resetsAt":3,"status":"allowed"}, + "seven_day_opus":{"utilization":0.1,"resetsAt":4,"status":"allowed"}, + "seven_day_overage_included":{"utilization":1,"resetsAt":5,"status":"rejected"}}"#, + ); + assert_eq!( + taken, + vec![ + ("five_hour".to_string(), Some(500), 2000), + ("seven_day".to_string(), Some(250), 3000), + ] + ); + } + + #[test] + fn a_window_missing_a_field_or_of_the_wrong_type_costs_only_itself() { + // Unknown is never zero: such a window is skipped, not read as empty. + for broken in [ + r#"{"resetsAt":2,"status":"allowed"}"#, + r#"{"utilization":0.5,"status":"allowed"}"#, + r#"{"utilization":0.5,"resetsAt":2}"#, + r#"{"utilization":"high","resetsAt":2,"status":"allowed"}"#, + r#"{"utilization":-0.5,"resetsAt":2,"status":"allowed"}"#, + ] { + let both = windows(&format!( + r#"{{"five_hour":{broken},"seven_day":{{"utilization":0,"resetsAt":3,"status":"allowed"}}}}"# + )); + assert_eq!(both, vec![("seven_day".to_string(), Some(0), 3000)], "{broken}"); + } + } +} diff --git a/crates/claude-connector/tests/config_document.rs b/crates/claude-connector/tests/config_document.rs index d2af150..65c7b5d 100644 --- a/crates/claude-connector/tests/config_document.rs +++ b/crates/claude-connector/tests/config_document.rs @@ -286,3 +286,12 @@ fn the_dollar_flags_are_not_flags() { ); } } + +#[test] +fn serve_refuses_a_document_that_is_not_there() { + let scratch = Scratch::new(); + let missing = scratch.path("missing.cc"); + let served = run(&["serve", "--config", missing.to_str().unwrap()]); + assert!(!served.status.success(), "{served:?}"); + assert!(!String::from_utf8_lossy(&served.stderr).is_empty()); +} diff --git a/crates/claude-connector/tests/fixtures/claude-2.1.312-meter-call.jsonl b/crates/claude-connector/tests/fixtures/claude-2.1.312-meter-call.jsonl new file mode 100644 index 0000000..f7238e1 --- /dev/null +++ b/crates/claude-connector/tests/fixtures/claude-2.1.312-meter-call.jsonl @@ -0,0 +1,4 @@ +{"type":"system","subtype":"init","cwd":"/tmp/meter-probe.Ye68uq/work","session_id":"1c3d29e9-8277-4ee0-86df-efc86d994f01","tools":[],"mcp_servers":[],"model":"claude-haiku-5-5","permissionMode":"bypassPermissions","slash_commands":[],"apiKeySource":"none","claude_code_version":"2.1.312","output_style":"default","agents":["claude","Explore","general-purpose","Plan","statusline-setup"],"skills":[],"plugins":[{"name":"cc-plugin-sec-default","path":"builtin","source":"cc-plugin-sec-default@builtin"},{"name":"cc-plugin-agents-md","path":"builtin","source":"cc-plugin-agents-md@builtin"},{"name":"cc-plugin-telemetry","path":"builtin","source":"cc-plugin-telemetry@builtin"},{"name":"cc-plugin-plugin-authoring","path":"builtin","source":"cc-plugin-plugin-authoring@builtin"}],"capabilities":["interrupt_receipt_v1","interrupt_cancel_queued_v1","interrupt_send_now_v1","interrupt_requeue_taken_v1","msg_lifecycle_v1","request_marker_lists_v1","ask_user_question_unanswered_v1","lifecycle_wakeup_source_v1","sdk_mcp_tools_list_changed","sdk_mcp_manifests","mcp_read_resource_v1","mcp_tool_ui_meta_v1","ui_surface_v1","mcp_call_result_not_for_model_v1"],"analytics_disabled":false,"product_feedback_disabled":false,"uuid":"8ad7f8d7-2831-43f2-a221-8af6323dcbd9","memory_paths":{"auto":"/tmp/meter-probe.Ye68uq/home/projects/-tmp-meter-probe-Ye68uq-work/memory/"},"original_cwd":"/tmp/meter-probe.Ye68uq/work","additional_directories":[],"messaging_socket_path":"/tmp/cc-socks/2838104.sock","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required","per_turn_effort_active":true,"view_mode":"default"} +{"type":"assistant","message":{"model":"claude-haiku-5-5","id":"msg_011Cgfu1sQvRAVmjwWNHc58s","type":"message","role":"assistant","content":[{"type":"text","text":"ok"}],"container":null,"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":384,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"output_tokens":4,"service_tier":"standard","inference_geo":"global","speed":"standard"},"input_transformations":[],"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"1c3d29e9-8277-4ee0-86df-efc86d994f01","uuid":"f80db687-b098-415c-ac44-51c40a6cb63a","timestamp":"2026-11-03T11:14:35.024Z","request_id":"req_011Cgfu1sCXBsDrfz8a1LoMd"} +{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","resetsAt":1794261600,"rateLimitType":"seven_day","overageStatus":"rejected","overageDisabledReason":"org_level_disabled","isUsingOverage":false,"unifiedWindows":{"seven_day":{"utilization":0.06,"resetsAt":1794261600,"status":"allowed"}}},"uuid":"acb3b364-a218-4965-b967-f73028254f8b","session_id":"1c3d29e9-8277-4ee0-86df-efc86d994f01"} +{"duration_api_ms":995,"stop_reason":"end_turn","session_id":"1c3d29e9-8277-4ee0-86df-efc86d994f01","total_cost_usd":0.000040400000000000006,"usage":{"input_tokens":384,"cache_creation_input_tokens":0,"cache_read_input_tokens":0,"output_tokens":4,"output_tokens_details":{"thinking_tokens":0},"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":0,"ephemeral_5m_input_tokens":0},"inference_geo":"global","iterations":[{"input_tokens":384,"output_tokens":4,"cache_read_input_tokens":0,"cache_creation_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":0},"type":"message"}],"speed":"standard","fallback_credit":null},"modelUsage":{"claude-haiku-5-5":{"inputTokens":384,"outputTokens":4,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"webSearchRequests":0,"costUSD":0.000040400000000000006,"contextWindow":1000000,"maxOutputTokens":128000,"thinkingTokens":0,"cacheCreationEphemeral5mInputTokens":0,"cacheCreationEphemeral1hInputTokens":0,"canonicalModel":"claude-haiku-5-5","provider":"firstParty","costBasis":"list"}},"permission_denials":[],"terminal_reason":"completed","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required","subagent_stats":{"spawned":0,"requested":{"background":0,"foreground":0,"unset":0},"started_in_background":0,"max_depth":0,"spawned_by_subagents":0,"completed":0,"failed":0,"killed":{"parent":0,"user":0,"system":0},"refused":{"depth_limit":0,"concurrency_limit":0,"budget":0},"by_type":{}},"safety_stops":0,"is_error":false,"num_turns":1,"subtype":"success","api_error_status":null,"result":"ok","ttft_ms":1210,"type":"result","duration_ms":1228,"uuid":"d0b1b31b-1428-45aa-b90e-d35903c9b91b","ttft_stream_ms":1209,"time_to_request_ms":232,"first_content_frame_ms":1209,"queued_turn_count":0,"result_index":0,"model_usage_since_last_result":{"claude-haiku-5-5":{"inputTokens":384,"outputTokens":4,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"webSearchRequests":0,"costUSD":0.000040400000000000006,"contextWindow":1000000,"maxOutputTokens":128000,"thinkingTokens":0,"cacheCreationEphemeral5mInputTokens":0,"cacheCreationEphemeral1hInputTokens":0,"canonicalModel":"claude-haiku-5-5","provider":"firstParty","costBasis":"list"}}}