feat(judge): watch the judge providers' plan usage; switch to the fallback before the wall

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) <[email protected]>

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(),

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}"),

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<String>,
+}
+
+/// Latest readings per provider family (`glm`, `kimi`), with when they were
+/// taken.
+#[derive(Debug, Clone, Default, serde::Serialize)]
+pub struct Snapshot {
+    pub providers: HashMap<String, Reading>,
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct Reading {
+    pub windows: Vec<Window>,
+    pub read_at: String,
+}
+
+fn state() -> &'static Mutex<Snapshot> {
+    static S: OnceLock<Mutex<Snapshot>> = 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<std::collections::HashSet<String>> {
+    static W: OnceLock<Mutex<std::collections::HashSet<String>>> = 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<Window> {
+    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<Window> {
+    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<Window> {
+    let Some(usages) = v.get("usages").and_then(Value::as_object) else {
+        return Vec::new();
+    };
+    let mut out: Vec<Window> = 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<Value> {
+    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<Window>)> = 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::<Vec<_>>(), ["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");
+    }
+}

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))

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<Value> {
+    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(),
+    }))
+}