meter: the weekly reading is kept fresh by one real claude call, not an endpoint the token cannot reach

The usage endpoint answers 403 to a setup token (inference scope only), so the poll
could never work. The connector's own claude -p call already carries the week in its
rate_limit_event (captured 2026-11-03: seven_day utilization 0.06 and its reset).
The meter makes that call (haiku, no tools, trivial prompt) when the seven_day
reading is 15 minutes stale, only while the swarm brake store reads released, and
backs off per consecutive miss. Deleted usage_poll, the ureq dependency, and the
quota rule that only a polled reading needed (POLLED, rejection-stands).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-03 12:16:40 +01:00
commit 65665225bf
11 changed files with 651 additions and 1187 deletions

1
.gitattributes vendored
View file

@ -1,2 +1,3 @@
schemas/*.json text eol=lf
crates/model-connector/tests/fixtures/wire_v4/* -text
crates/claude-connector/tests/fixtures/*.cc -text

View file

@ -13,7 +13,7 @@ path = "src/main.rs"
[dependencies]
anyhow = "1"
chrono = { version = "0.4", default-features = false }
chrono = { version = "0.4", default-features = false, features = ["serde"] }
hmac = "0.12"
cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a4eae9056152335662f1115f97bf35fb00ab6d23", package = "cultcache-rs" }
model-connector = { path = "../model-connector", features = ["daemon"] }
@ -22,7 +22,6 @@ 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]

View file

@ -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,6 +162,27 @@ 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<Usage, String> {
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).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 {

View file

@ -2746,3 +2746,43 @@ 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_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());
}

View file

@ -23,7 +23,7 @@ use crate::lease::{LeaseCaller, LeaseLimits, serve_leases, user_uid};
#[cfg(target_os = "linux")]
use crate::unit::SystemdUnits;
#[cfg(unix)]
use crate::usage_poll::UsageClient;
use crate::meter::{SWARM_BRAKE_STORE, SwarmBrakeStore};
use crate::{
ClaudeCodeBackend, ClaudeCodeConfig, ClaudeCodeLimits, CredentialTokenSource, StderrSink,
};
@ -537,13 +537,14 @@ 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, UsageClient::production())
serve_config_with(path, Path::new(SWARM_BRAKE_STORE))
}
/// [`serve_config`] with the usage poll asking `usage`: the endpoint is not a
/// setting of the document, so a test reaches it only through here.
/// [`serve_config`] with the meter reading the swarm brake at `brake_store`:
/// the brake's place is not a setting of the document, so a test reaches it
/// only through here.
#[cfg(unix)]
fn serve_config_with(path: &Path, usage: UsageClient) -> Result<(), ConfigError> {
fn serve_config_with(path: &Path, brake_store: &Path) -> Result<(), ConfigError> {
let config = load_config(path)?;
let admissions = readiness_of(&config)?;
let listen = config.validate()?;
@ -565,14 +566,12 @@ fn serve_config_with(path: &Path, usage: UsageClient) -> Result<(), ConfigError>
if let Some(socket) = &config.lease_socket {
start_leases(&config, socket, service.clone(), backend.quota())?;
}
crate::usage_poll::start(
crate::meter::start(
backend.quota(),
Box::new(CredentialTokenSource::from_environment(
config.token_credential.clone(),
)),
usage,
Box::new(backend.clone()),
Box::new(SwarmBrakeStore::new(brake_store)),
)
.map_err(|_| ConfigError::Serving("usage poll"))?;
.map_err(|_| ConfigError::Serving("meter"))?;
serve_service(listener, config.serve_limits(), service, backend)?;
Ok(())
}
@ -599,16 +598,16 @@ mod tests {
}
}
/// `serve` starts the usage poll beside the listener, and the poll asks
/// where it was told: a fake on loopback receives its request, bearer
/// token and all, and nothing reaches the provider.
/// `serve` starts the meter beside the listener: with a released brake and
/// no reading of the week, the meter runs the claude binary once and serve
/// stays up. The binary is a script that replays a real captured call.
#[cfg(unix)]
#[test]
fn serve_config_starts_the_usage_poll_and_stays_up() {
use std::io::{Read, Write};
fn serve_config_starts_the_meter_and_stays_up() {
use std::os::unix::fs::PermissionsExt;
const TOKEN: &str = "sk-ant-oat01-fake-token-for-the-serve-test";
const CALL: &str = include_str!("../tests/fixtures/claude-2.1.312-meter-call.jsonl");
const BRAKE: &[u8] = include_bytes!("../tests/fixtures/swarm-brake-live-2026-11-03.cc");
let dir =
std::env::temp_dir().join(format!("claude-connector-serve-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
@ -617,15 +616,23 @@ mod tests {
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("claude"), b"not a real claude binary").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"), TOKEN).unwrap();
std::fs::write(at("swarm.cc"), BRAKE).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(b"not a real claude binary").into(),
claude_binary_sha256: sha2::Sha256::digest(script.as_bytes()).into(),
work_dir: at("work"),
config_dir: at("home"),
prompt_dir: at("prompts"),
@ -658,34 +665,19 @@ mod tests {
// serve thread it starts.
unsafe { std::env::set_var("CREDENTIALS_DIRECTORY", at("credentials")) };
let fake = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}/api/oauth/usage", fake.local_addr().unwrap());
let (asked, request) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let (mut stream, _) = fake.accept().unwrap();
let mut text = vec![0u8; 4096];
let n = stream.read(&mut text).unwrap();
let _ = asked.send(String::from_utf8_lossy(&text[..n]).into_owned());
let _ = stream.write_all(
b"HTTP/1.1 429 Too Many Requests
Content-Length: 0
Connection: close
",
);
});
let path = at("connector.cc");
let served = std::thread::spawn(move || serve_config_with(&path, UsageClient::new(url)));
let brake = at("swarm.cc");
let served = std::thread::spawn(move || serve_config_with(&path, &brake));
let request = request
.recv_timeout(Duration::from_secs(30))
.expect("the usage poll never asked the fake endpoint");
assert!(request.starts_with("GET /api/oauth/usage "), "{request}");
assert!(request.contains(&format!("Bearer {TOKEN}")), "{request}");
assert!(
!served.is_finished(),
"serve returned after starting the poll"
);
let deadline = std::time::Instant::now() + Duration::from_secs(30);
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);
}

View file

@ -10,12 +10,12 @@ mod handle;
mod launch;
#[cfg(target_os = "linux")]
mod lease;
mod meter;
mod process;
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};

View file

@ -0,0 +1,523 @@
//! The meter: keeps the plan's weekly reading fresh. 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.
//!
//! A held swarm brake spends nothing: the meter reads the swarm's brake store
//! before every call and, like every other organ that acts, treats anything
//! but a decodable released brake as held. 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::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use chrono::{DateTime, Utc};
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use model_connector::Usage;
use crate::backend::ClaudeCodeBackend;
use crate::quota::{EXPIRED, QuotaCell};
/// Where the swarm's brake is kept: the operator's root-owned, world-readable
/// store, the same file the Minds' organs read before they act.
pub(crate) const SWARM_BRAKE_STORE: &str = "/etc/gamecult/minds/swarm.cc";
/// The one key the brake document has in its store.
const BRAKE_KEY: &str = "brake";
/// The window the meter keeps fresh.
const WINDOW: &str = "seven_day";
/// 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.
const CEILING: Duration = Duration::from_secs(6 * 60 * 60);
/// How often a held brake is looked at again.
const BRAKE_RECHECK: Duration = Duration::from_secs(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<Usage, String>;
}
impl Probe for Arc<ClaudeCodeBackend> {
fn call(&self) -> Result<Usage, String> {
self.read_meter()
}
}
/// Whether the operator's swarm brake holds.
pub(crate) trait Brake: Send {
fn held(&self) -> bool;
}
/// The brake document of the operator's control stores. Declared here with the
/// type name and field order the stores are written with; the connector reads
/// it and never writes it.
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "eureka.control.brake.v1")]
struct SwarmBrake {
#[cultcache(key = 0)]
released: bool,
#[cultcache(key = 1)]
set_at: DateTime<Utc>,
#[cultcache(key = 2)]
set_by: String,
}
/// The swarm brake read from its store, read-only and without locks. Absent,
/// unreadable, undecodable or unreleased is held.
pub(crate) struct SwarmBrakeStore(PathBuf);
impl SwarmBrakeStore {
pub(crate) fn new(path: impl Into<PathBuf>) -> Self {
Self(path.into())
}
fn released(&self) -> Option<bool> {
let mut cache = CultCache::new();
cache.register_entry_type::<SwarmBrake>().ok()?;
for envelope in SingleFileMessagePackBackingStore::new(&self.0)
.pull_all_read_only_snapshot()
.ok()?
{
cache.put_raw_envelope(envelope).ok()?;
}
Some(cache.get::<SwarmBrake>(BRAKE_KEY).ok()??.released)
}
}
impl Brake for SwarmBrakeStore {
fn held(&self) -> bool {
self.released() != Some(true)
}
}
/// Why the meter is waiting; logged once per change.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum State {
Read,
Braked,
Failed,
Unread,
}
impl State {
fn line(self) -> &'static str {
match self {
Self::Read => "meter: weekly reading refreshed",
Self::Braked => "meter: swarm brake held, making no call",
Self::Failed => "meter: the call failed, backing off",
Self::Unread => "meter: the call completed without a weekly reading, backing off",
}
}
}
struct Meter {
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
log: Box<dyn FnMut(&str) + Send>,
/// No call before this instant, unix ms.
not_before_ms: u64,
/// Consecutive calls that left the reading unrefreshed.
misses: u32,
last_state: Option<State>,
}
impl Meter {
fn new(
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
log: Box<dyn FnMut(&str) + Send>,
) -> Self {
Self {
cell,
probe,
brake,
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 the brake is held, 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.brake.held() {
self.report(State::Braked);
return BRAKE_RECHECK;
}
let due_at = self.due_at_ms();
if now_ms < due_at {
return Duration::from_millis(due_at - now_ms);
}
let called = self.probe.call();
let state = if called.is_ok() && self.refreshed_since(now_ms) {
self.misses = 0;
State::Read
} else {
self.misses = self.misses.saturating_add(1);
if called.is_ok() {
State::Unread
} else {
State::Failed
}
};
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 == WINDOW)
.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 == WINDOW && quota.observed_at_unix_ms >= since_ms)
}
fn report(&mut self, state: State) {
if self.last_state != Some(state) {
(self.log)(state.line());
self.last_state = Some(state);
}
}
fn run(mut self) {
loop {
let wait = self.tick(unix_ms());
std::thread::sleep(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.
pub(crate) fn start(
cell: Arc<QuotaCell>,
probe: Box<dyn Probe>,
brake: Box<dyn Brake>,
) -> std::io::Result<std::thread::JoinHandle<()>> {
let meter = Meter::new(
cell,
probe,
brake,
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 model_connector::PlanQuota;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, 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() -> Vec<PlanQuota> {
vec![PlanQuota {
window: WINDOW.to_string(),
utilization_permille: Some(60),
resets_at_unix_ms: RESET,
status: "allowed".to_string(),
observed_at_unix_ms: 0,
}]
}
/// A call that answers as scripted and, when it succeeds with a reading,
/// feeds the cell the way the stream does.
struct Calls {
cell: Arc<QuotaCell>,
made: Arc<AtomicUsize>,
outcome: Arc<Mutex<Result<Option<u64>, String>>>,
}
impl Probe for Calls {
fn call(&self) -> Result<Usage, String> {
self.made.fetch_add(1, Ordering::SeqCst);
match &*self.outcome.lock().unwrap() {
Ok(Some(at)) => self.cell.observe_at(*at, reading()),
Ok(None) => {}
Err(kind) => return Err(kind.clone()),
}
Ok(Usage {
input_tokens: 385,
cache_read_tokens: 0,
cache_write_tokens: 0,
output_tokens: 4,
reasoning_tokens: None,
})
}
}
struct Held(Arc<AtomicBool>);
impl Brake for Held {
fn held(&self) -> bool {
self.0.load(Ordering::SeqCst)
}
}
struct Rig {
meter: Meter,
cell: Arc<QuotaCell>,
made: Arc<AtomicUsize>,
outcome: Arc<Mutex<Result<Option<u64>, String>>>,
braked: Arc<AtomicBool>,
log: Arc<Mutex<Vec<String>>>,
}
fn rig() -> Rig {
let cell = Arc::new(QuotaCell::new());
let made = Arc::new(AtomicUsize::new(0));
let outcome = Arc::new(Mutex::new(Ok(None)));
let braked = Arc::new(AtomicBool::new(false));
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(),
outcome: outcome.clone(),
}),
Box::new(Held(braked.clone())),
Box::new(move |line| sink.lock().unwrap().push(line.to_string())),
);
Rig {
meter,
cell,
made,
outcome,
braked,
log,
}
}
impl Rig {
fn answers(&self, outcome: Result<Option<u64>, &str>) {
*self.outcome.lock().unwrap() = outcome.map_err(str::to_string);
}
fn calls(&self) -> usize {
self.made.load(Ordering::SeqCst)
}
}
#[test]
fn an_unread_week_triggers_one_call_and_the_fresh_reading_ends_the_asking() {
let mut rig = rig();
rig.answers(Ok(Some(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());
rig.answers(Ok(Some(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().remove(0);
over.resets_at_unix_ms = 1;
rig.cell.observe_at(now, vec![over]);
rig.answers(Ok(Some(now)));
rig.meter.tick(now);
assert_eq!(rig.calls(), 1);
}
#[test]
fn a_held_brake_makes_no_call_however_stale_the_reading() {
let mut rig = rig();
rig.braked.store(true, Ordering::SeqCst);
for step in 0..10 {
assert_eq!(rig.meter.tick(NOW + step * 60 * MIN), BRAKE_RECHECK);
}
assert_eq!(rig.calls(), 0);
assert_eq!(*rig.log.lock().unwrap(), vec![State::Braked.line()]);
rig.braked.store(false, Ordering::SeqCst);
rig.answers(Ok(Some(NOW + 600 * MIN)));
rig.meter.tick(NOW + 600 * MIN);
assert_eq!(rig.calls(), 1, "released, the stale week is read at once");
}
#[test]
fn a_failed_call_leaves_the_cell_and_backs_off_instead_of_hammering() {
let mut rig = rig();
rig.cell.observe_at(NOW, reading());
let before = rig.cell.snapshot();
rig.answers(Err("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.log.lock().unwrap(), vec![State::Failed.line()]);
}
#[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.answers(Err("Provider"));
rig.meter.tick(NOW);
rig.meter.tick(NOW + 15 * MIN);
assert_eq!(rig.meter.misses, 2);
rig.answers(Ok(Some(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.answers(Ok(None));
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.log.lock().unwrap(), vec![State::Unread.line()]);
}
#[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.answers(Ok(Some(NOW - 1)));
rig.meter.tick(NOW);
assert_eq!(rig.meter.misses, 1);
}
// ---- the brake store, read as the operator's tool writes it ----
/// The swarm store as it stood on Yggdrasil on 2026-11-03 (released),
/// copied byte for byte.
const LIVE_STORE: &[u8] = include_bytes!("../tests/fixtures/swarm-brake-live-2026-11-03.cc");
fn store_with(bytes: &[u8]) -> (PathBuf, SwarmBrakeStore) {
static NEXT: AtomicUsize = AtomicUsize::new(0);
let dir = std::env::temp_dir().join(format!(
"claude-meter-brake-{}-{}",
std::process::id(),
NEXT.fetch_add(1, Ordering::SeqCst)
));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("swarm.cc");
std::fs::write(&path, bytes).unwrap();
(dir, SwarmBrakeStore::new(path))
}
/// The live store with `released` flipped to false: the msgpack `true`
/// byte that precedes the timestamp string.
fn held_store() -> Vec<u8> {
let mut bytes = LIVE_STORE.to_vec();
let at = bytes
.windows(5)
.position(|w| w == b"\xc3\xbe202")
.expect("the live store holds released=true before its timestamp");
bytes[at] = 0xc2;
bytes
}
#[test]
fn the_live_released_store_reads_as_released() {
let (dir, store) = store_with(LIVE_STORE);
assert_eq!(store.released(), Some(true));
assert!(!store.held());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn the_same_store_with_the_brake_held_reads_as_held() {
let (dir, store) = store_with(&held_store());
assert_eq!(store.released(), Some(false));
assert!(store.held());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_missing_garbled_or_truncated_store_is_held() {
assert!(SwarmBrakeStore::new("/nonexistent-claude-meter/swarm.cc").held());
for bytes in [&b""[..], b"not a store at all", &LIVE_STORE[..40]] {
let (dir, store) = store_with(bytes);
assert!(store.held(), "{bytes:?}");
let _ = std::fs::remove_dir_all(dir);
}
}
}

View file

@ -1,9 +1,9 @@
//! 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.
//! 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;
@ -29,26 +29,17 @@ impl QuotaCell {
self.observe_at(unix_ms(), windows);
}
/// Records readings taken at `at` (unix ms): the moment the sensor asked,
/// not the moment its answer arrived. The cell alone decides how a reading
/// combines with the one it holds for that window, and every sensor goes
/// through this decision:
///
/// - a reading older than the held one is dropped, so a slow answer cannot
/// replace what a faster sensor saw meanwhile;
/// - the provider's rejection stands until its window resets, or until a
/// reading that itself carries a verdict from the provider replaces it:
/// a [`POLLED`] reading reports utilization only, and says nothing about
/// whether the provider accepts calls.
///
/// Windows the readings did not mention stay as they were, stamp included.
pub fn observe_at(&self, at: u64, windows: impl IntoIterator<Item = PlanQuota>) {
/// 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.
(&self, at: u64, windows: impl IntoIterator<Item = PlanQuota>) {
let mut held = self.0.lock().unwrap_or_else(|e| e.into_inner());
for mut quota in windows {
quota.observed_at_unix_ms = at;
if held
.get(&quota.window)
.is_none_or(|current| supersedes(current, &quota))
.is_none_or(|current| quota.observed_at_unix_ms >= current.observed_at_unix_ms)
{
held.insert(quota.window.clone(), quota);
}
@ -81,24 +72,7 @@ impl QuotaCell {
}
/// The status of a window whose reset has passed since it was observed.
const EXPIRED: &str = "expired";
/// The status of a window the provider refuses calls on.
pub(crate) const REJECTED: &str = "rejected";
/// The status of a window read from the usage endpoint: utilization and reset
/// time, with no verdict on whether calls are accepted.
pub(crate) const POLLED: &str = "polled";
/// Whether `incoming` (already stamped with its own reading time) replaces
/// `current` for the same window.
fn supersedes(current: &PlanQuota, incoming: &PlanQuota) -> bool {
if incoming.observed_at_unix_ms < current.observed_at_unix_ms {
return false;
}
let rejection_stands = current.status == REJECTED
&& incoming.status == POLLED
&& incoming.observed_at_unix_ms < current.resets_at_unix_ms;
!rejection_stands
}
pub(crate) const EXPIRED: &str = "expired";
fn unix_ms() -> u64 {
SystemTime::now()
@ -194,33 +168,13 @@ mod tests {
}
#[test]
fn a_rejection_stands_against_a_poll_until_the_window_resets() {
fn a_newer_reading_lifts_a_rejection() {
let cell = QuotaCell::new();
let polled = with_status(quota("five_hour", 500, 0), POLLED);
cell.observe_at(100, [with_status(quota("five_hour", 1000, 0), REJECTED)]);
cell.observe_at(200, [polled.clone()]);
assert_eq!(held(&cell), (REJECTED.to_string(), Some(1000), 100));
// The window the rejection described is over: a poll of the next one lands.
cell.observe_at(RESET, [polled]);
assert_eq!(cell.snapshot_at(RESET - 1).windows[0].status, POLLED);
}
#[test]
fn a_newer_reading_that_carries_its_own_verdict_lifts_a_rejection() {
let cell = QuotaCell::new();
cell.observe_at(100, [with_status(quota("five_hour", 1000, 0), REJECTED)]);
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 a_poll_replaces_what_is_not_a_rejection() {
let cell = QuotaCell::new();
cell.observe_at(100, [quota("five_hour", 300, 0)]);
cell.observe_at(200, [with_status(quota("five_hour", 400, 0), POLLED)]);
assert_eq!(held(&cell), (POLLED.to_string(), Some(400), 200));
}
#[test]
fn the_live_clock_stamps_and_expires() {
let cell = QuotaCell::new();

File diff suppressed because it is too large Load diff

View file

@ -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"}}}

View file

@ -0,0 +1 @@
“²cultcache.store.v1‘—·eureka.control.brake.v1·eureka.control.brake.v1ºeureka.control.brake.v1.v1·eureka.control.brake.v1Ùb{"schemaName":"eureka.control.brake.v1","schemaVersion":"eureka.control.brake.v1.v1","members":[]}‘·eureka.control.brake.v1<EFBFBD>‘”¥brake·eureka.control.brake.v1´2026-11-03T08:45:01ZÄ/“þ2026-11-03T08:45:01.468996363Z­gamecultadmin