From 318a6791b91be976649053d741216f4f9462e4ec Mon Sep 17 00:00:00 2001 From: Omar Sobh Date: Wed, 23 Sep 2026 07:55:10 -0500 Subject: [PATCH] feat(judge): watch the judge providers' plan usage; switch to the fallback before the wall MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GLM's weekly window ran out while the judge had spent ~0.1% of it: other consumers of the shared key starve the judge, and ClawMates learned only from failed phases. A poller now reads z.ai's quota API and Kimi's usages API every 10 minutes, warns once per window per reset at 80%, and the evaluator skips a judge whose plan is at 95% in any window for the (equally independent) fallback — only on a real reading, never on a missing one. GET /api/judge/quota shows the readings. Parsers pinned to the shapes both APIs returned on 2026-09-23. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/bins/clawmates-server/src/main.rs | 4 + crates/cm-api/src/evaluator.rs | 26 +- crates/cm-api/src/judge_quota.rs | 293 +++++++++++++++++++++++ crates/cm-api/src/lib.rs | 2 + crates/cm-api/src/routes/nodes.rs | 11 + 5 files changed, 335 insertions(+), 1 deletion(-) create mode 100644 crates/cm-api/src/judge_quota.rs diff --git a/crates/bins/clawmates-server/src/main.rs b/crates/bins/clawmates-server/src/main.rs index daf9d65..7e37a87 100644 --- a/crates/bins/clawmates-server/src/main.rs +++ b/crates/bins/clawmates-server/src/main.rs @@ -448,6 +448,10 @@ async fn run() -> Result<(), String> { cm_api::beszel::spawn_poller(pool.clone(), std::time::Duration::from_secs(15)); // Fleet automation: evaluate metric-threshold rules → drain/undrain/alert. cm_api::node_rules::spawn_evaluator(pool.clone(), std::time::Duration::from_secs(20)); + // Judge providers' plan usage (z.ai, Kimi): warn at 80%, and let the + // evaluator skip a judge at 95% for the fallback. Ten minutes: the windows + // are hours and days long, and each poll is one tiny GET per provider. + cm_api::judge_quota::spawn_poller(std::time::Duration::from_secs(600)); // Nightly: check upstream for newer dev-tool releases (claude/kimi/ollama). cm_api::tool_versions::spawn_latest_checker( pool.clone(), diff --git a/crates/cm-api/src/evaluator.rs b/crates/cm-api/src/evaluator.rs index ebdfd77..4fa02b0 100644 --- a/crates/cm-api/src/evaluator.rs +++ b/crates/cm-api/src/evaluator.rs @@ -679,7 +679,31 @@ pub async fn evaluate( // to accept it — and the tool loop is what makes the check evidence rather // than opinion, so an independent judge must have it too. let implementer = mission_implementer_family(runtime, mission_id).await; - if let Some((provider, model)) = cross_provider_judge(runtime, mission_id, implementer).await { + // Skip a judge whose plan is about to run out, BEFORE spending a call + // that would fail with a 429. Only on a real reading + // (`judge_quota::near_limit` is None without one), and only when a + // fallback that passes the same independence checks exists. + let chosen = match cross_provider_judge(runtime, mission_id, implementer).await { + Some((p, m)) => { + let family = provider_family(&m); + match crate::judge_quota::near_limit(&family) { + Some(w) => match fallback_judge(runtime, implementer, &family).await { + Some((fp, fm)) => { + eprintln!( + "evaluator: {m}'s plan is at {:.0}% of its {} window — judging \ + with {fm} instead of spending a call that would fail", + w.used_pct, w.name + ); + Some((fp, fm)) + } + None => Some((p, m)), + }, + None => Some((p, m)), + } + } + None => None, + }; + if let Some((provider, model)) = chosen { let system = match &sandbox { Some(_) => format!("{EVAL_SYSTEM_VERIFYING}\n\n{VERDICT_CONTRACT}"), None => format!("{EVAL_SYSTEM_EVIDENCE_ONLY}\n\n{VERDICT_CONTRACT}"), diff --git a/crates/cm-api/src/judge_quota.rs b/crates/cm-api/src/judge_quota.rs new file mode 100644 index 0000000..c9bef2c --- /dev/null +++ b/crates/cm-api/src/judge_quota.rs @@ -0,0 +1,293 @@ +//! How much of each judge provider's plan is left, read from the providers' +//! own usage APIs. +//! +//! Built 2026-09-23 after GLM's weekly limit ran out for the second time in a +//! month. Measured then: the judge spent about 0.1% of what the shared z.ai key +//! used that week (275 requests of 6,959; ~0.4M of ~470M tokens). Other +//! consumers of the same key starve it, and nothing in ClawMates could see it +//! coming: the first sign was every conditioned phase failing on a 1310. +//! +//! Two jobs: +//! +//! - **Warn** once per window when it crosses [`WARN_AT`] percent, so the +//! operator hears about it days early, not from a failed mission. +//! - **Switch** the judge before the wall: once the primary judge's provider +//! crosses [`SWITCH_AT`] in any window, the evaluator goes straight to the +//! fallback judge (`evaluator::fallback_judge`) instead of spending a call +//! that is about to fail with a 429. +//! +//! Best-effort throughout: a usage API that is down or changes shape leaves +//! the readings empty, and an empty reading NEVER switches anything. The 429 +//! fallback in the evaluator is still there behind this. + +use serde_json::Value; +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; +use std::time::Duration; + +/// Percent of a window at which the operator is warned. +pub const WARN_AT: f64 = 80.0; +/// Percent of a window at which the primary judge is skipped for the fallback. +pub const SWITCH_AT: f64 = 95.0; + +/// One quota window as a provider reported it. +#[derive(Debug, Clone, PartialEq, serde::Serialize)] +pub struct Window { + /// `5h`, `7d`, … — the window's length, as the provider describes it. + pub name: String, + /// 0–100. + pub used_pct: f64, + /// RFC 3339, when the provider said. + pub resets_at: Option, +} + +/// Latest readings per provider family (`glm`, `kimi`), with when they were +/// taken. +#[derive(Debug, Clone, Default, serde::Serialize)] +pub struct Snapshot { + pub providers: HashMap, +} + +#[derive(Debug, Clone, serde::Serialize)] +pub struct Reading { + pub windows: Vec, + pub read_at: String, +} + +fn state() -> &'static Mutex { + static S: OnceLock> = OnceLock::new(); + S.get_or_init(|| Mutex::new(Snapshot::default())) +} + +/// Windows already warned about, keyed `family/window/resets_at`, so a window +/// warns once per reset cycle, not on every poll. +fn warned() -> &'static Mutex> { + static W: OnceLock>> = OnceLock::new(); + W.get_or_init(|| Mutex::new(Default::default())) +} + +/// The current readings. +pub fn snapshot() -> Snapshot { + state().lock().map(|s| s.clone()).unwrap_or_default() +} + +/// Should the evaluator skip a judge of this family for the fallback? True +/// only on a REAL reading at or past [`SWITCH_AT`]; no reading means no. +pub fn near_limit(family: &str) -> Option { + let snap = snapshot(); + snap.providers + .get(family)? + .windows + .iter() + .find(|w| w.used_pct >= SWITCH_AT) + .cloned() +} + +/// Parse z.ai's `GET /api/monitor/usage/quota/limit`. +/// +/// Its `TOKENS_LIMIT` entries carry `unit` + `number` for the window and a +/// `percentage`. Measured on the Pro plan 2026-09-23: `unit 3, number 5` is the +/// 5-hour window and `unit 6, number 1` the weekly one (its reset matched the +/// 1310 error's own "will reset at"). `nextResetTime` is epoch milliseconds. +pub fn parse_zai(v: &Value) -> Vec { + let Some(limits) = v.pointer("/data/limits").and_then(Value::as_array) else { + return Vec::new(); + }; + limits + .iter() + .filter(|l| l.get("type").and_then(Value::as_str) == Some("TOKENS_LIMIT")) + .filter_map(|l| { + let pct = l.get("percentage").and_then(Value::as_f64)?; + let number = l.get("number").and_then(Value::as_i64).unwrap_or(1); + let name = match l.get("unit").and_then(Value::as_i64) { + Some(3) => format!("{number}h"), + Some(6) => format!("{}d", number * 7), + Some(u) => format!("unit{u}x{number}"), + None => "unknown".to_string(), + }; + let resets_at = l + .get("nextResetTime") + .and_then(Value::as_i64) + .and_then(|ms| { + time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(ms) * 1_000_000).ok() + }) + .and_then(|t| t.format(&time::format_description::well_known::Rfc3339).ok()); + Some(Window { name, used_pct: pct, resets_at }) + }) + .collect() +} + +/// Parse Kimi's `GET https://api.kimi.com/coding/v1/usages`. +/// +/// `usages.limit_5h` / `usages.limit_7d` carry `used_ratio` (0–1) and +/// `reset_time`. A ratio above 1 is taken as already a percentage, so a unit +/// change on their side reads as "very used", which fails toward warning. +pub fn parse_kimi(v: &Value) -> Vec { + let Some(usages) = v.get("usages").and_then(Value::as_object) else { + return Vec::new(); + }; + let mut out: Vec = usages + .iter() + .filter_map(|(k, u)| { + let ratio = u.get("used_ratio").and_then(Value::as_f64)?; + let pct = if ratio <= 1.0 { ratio * 100.0 } else { ratio }; + Some(Window { + name: k.trim_start_matches("limit_").to_string(), + used_pct: pct, + resets_at: u.get("reset_time").and_then(Value::as_str).map(str::to_string), + }) + }) + .collect(); + out.sort_by(|a, b| a.name.cmp(&b.name)); + out +} + +async fn fetch(client: &reqwest::Client, url: &str, auth: &str) -> Option { + let resp = client + .get(url) + .header("Authorization", auth) + .header("Accept-Language", "en-US,en") + .timeout(Duration::from_secs(20)) + .send() + .await + .ok()?; + if !resp.status().is_success() { + eprintln!("judge_quota: {url} answered {}", resp.status()); + return None; + } + resp.json().await.ok() +} + +/// One poll of both providers. Keys come from the same env vars the provider +/// registry uses; a provider whose key is unset is simply not read. +pub async fn poll_once(client: &reqwest::Client) { + let mut readings: Vec<(&str, Vec)> = Vec::new(); + if let Some(key) = std::env::var("ZAI_API_KEY").ok().filter(|k| !k.is_empty()) { + // z.ai takes the bare key, no `Bearer` (measured). + if let Some(v) = fetch(client, "https://api.z.ai/api/monitor/usage/quota/limit", &key).await { + readings.push(("glm", parse_zai(&v))); + } + } + if let Some(key) = std::env::var("KIMI_API_KEY").ok().filter(|k| !k.is_empty()) { + if let Some(v) = + fetch(client, "https://api.kimi.com/coding/v1/usages", &format!("Bearer {key}")).await + { + readings.push(("kimi", parse_kimi(&v))); + } + } + let now = time::OffsetDateTime::now_utc() + .format(&time::format_description::well_known::Rfc3339) + .unwrap_or_default(); + for (family, windows) in readings { + if windows.is_empty() { + eprintln!("judge_quota: {family} usage API answered but no window parsed — shape changed?"); + continue; + } + for w in &windows { + if w.used_pct >= WARN_AT { + let key = format!("{family}/{}/{}", w.name, w.resets_at.as_deref().unwrap_or("")); + let first = warned().lock().map(|mut s| s.insert(key)).unwrap_or(false); + if first { + eprintln!( + "judge_quota: WARNING {family} {} window at {:.0}% (resets {}){}", + w.name, + w.used_pct, + w.resets_at.as_deref().unwrap_or("?"), + if w.used_pct >= SWITCH_AT { + " — the evaluator now skips this judge for the fallback" + } else { + "" + } + ); + } + } + } + if let Ok(mut s) = state().lock() { + s.providers + .insert(family.to_string(), Reading { windows, read_at: now.clone() }); + } + } +} + +/// Poll forever. +pub fn spawn_poller(interval: Duration) { + tokio::spawn(async move { + let client = reqwest::Client::new(); + let mut tick = tokio::time::interval(interval); + loop { + tick.tick().await; + poll_once(&client).await; + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + /// The shape z.ai actually returned on 2026-09-23, on the day the weekly + /// window was exhausted. + #[test] + fn zai_reading_names_both_windows() { + let v = json!({"code":200,"data":{"limits":[ + {"type":"TIME_LIMIT","unit":5,"number":1,"usage":1000,"percentage":0}, + {"type":"TOKENS_LIMIT","unit":3,"number":5,"percentage":23,"nextResetTime":1790167210082i64}, + {"type":"TOKENS_LIMIT","unit":6,"number":1,"percentage":100,"nextResetTime":1790301693982i64} + ],"level":"pro"}}); + let w = parse_zai(&v); + assert_eq!(w.len(), 2, "TIME_LIMIT (tool calls) is not a token window: {w:?}"); + assert_eq!(w[0].name, "5h"); + assert_eq!(w[0].used_pct, 23.0); + assert_eq!(w[1].name, "7d"); + assert_eq!(w[1].used_pct, 100.0); + assert!(w[1].resets_at.as_deref().unwrap().starts_with("2026-09-25T02:01"), "{:?}", w[1]); + } + + /// Kimi's measured shape; ratios become percentages. + #[test] + fn kimi_reading_converts_ratios() { + let v = json!({"usages":{ + "limit_5h":{"used_ratio":0.01,"reset_time":"2026-09-23T15:49:20Z"}, + "limit_7d":{"used_ratio":0.97,"reset_time":"2026-09-28T19:49:20Z"}}}); + let w = parse_kimi(&v); + assert_eq!(w.iter().map(|w| w.name.as_str()).collect::>(), ["5h", "7d"]); + assert!((w[0].used_pct - 1.0).abs() < 1e-9); + assert!((w[1].used_pct - 97.0).abs() < 1e-9); + } + + /// A changed or empty shape yields no windows — and no windows never + /// switches the judge. + #[test] + fn an_unreadable_answer_switches_nothing() { + assert!(parse_zai(&json!({"data":{}})).is_empty()); + assert!(parse_kimi(&json!({"error":"x"})).is_empty()); + assert!(near_limit("some-family-never-read").is_none()); + } + + #[test] + fn near_limit_fires_only_at_the_switch_threshold() { + { + let mut s = state().lock().unwrap(); + s.providers.insert( + "test-fam-a".into(), + Reading { + windows: vec![Window { name: "7d".into(), used_pct: SWITCH_AT - 0.5, resets_at: None }], + read_at: String::new(), + }, + ); + s.providers.insert( + "test-fam-b".into(), + Reading { + windows: vec![ + Window { name: "5h".into(), used_pct: 10.0, resets_at: None }, + Window { name: "7d".into(), used_pct: SWITCH_AT, resets_at: None }, + ], + read_at: String::new(), + }, + ); + } + assert!(near_limit("test-fam-a").is_none()); + assert_eq!(near_limit("test-fam-b").unwrap().name, "7d"); + } +} diff --git a/crates/cm-api/src/lib.rs b/crates/cm-api/src/lib.rs index 88ccca6..fa8fe7f 100644 --- a/crates/cm-api/src/lib.rs +++ b/crates/cm-api/src/lib.rs @@ -12,6 +12,7 @@ pub mod corpus; mod error; pub mod evaluator; pub mod evaluator_tools; +pub mod judge_quota; mod extract; pub mod fleet; pub mod fleet_herdr; @@ -190,6 +191,7 @@ pub fn router(state: AppState) -> Router { .route("/api/nodes", get(routes::nodes::list)) .route("/api/fleet/capacity", get(routes::nodes::capacity)) .route("/api/fleet/backends", get(routes::nodes::backends)) + .route("/api/judge/quota", get(routes::nodes::judge_quota)) .route("/api/nodes/pair", post(routes::nodes::pair)) .route("/api/nodes/live", get(routes::nodes::live)) .route("/api/nodes/agent", get(routes::nodes::agent_ws)) diff --git a/crates/cm-api/src/routes/nodes.rs b/crates/cm-api/src/routes/nodes.rs index e4b6c03..c2eebae 100644 --- a/crates/cm-api/src/routes/nodes.rs +++ b/crates/cm-api/src/routes/nodes.rs @@ -514,3 +514,14 @@ fn backend_label(id: &str) -> String { other => other.to_string(), } } + +/// `GET /api/judge/quota` — the judge providers' plan usage as last polled by +/// `judge_quota`, and the thresholds that act on it. Read-only; no workspace +/// data. The readings are per deployment, since the keys are. +pub async fn judge_quota(Authed(_user): Authed) -> Json { + Json(serde_json::json!({ + "warn_at_pct": crate::judge_quota::WARN_AT, + "switch_at_pct": crate::judge_quota::SWITCH_AT, + "readings": crate::judge_quota::snapshot(), + })) +}