hermes: expose Claude remaining quota

services/hermes/agent-deployment.yaml

@@ -25,7 +25,7 @@ spec:
         ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers
         ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback
         ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available
-        ai.bstein.dev/config-rev: "20260823-ai-quota-stability"
+        ai.bstein.dev/config-rev: "20260823-claude-quota-headers"
         prometheus.io/scrape: "true"
         prometheus.io/path: /metrics
         prometheus.io/port: "9010"
@@ -1034,7 +1034,9 @@ spec:
             - {name: CODEX_HOME, value: /runtime-access/codex}
             - {name: ATLAS_AI_CODEX_BIN, value: /opt/data/tools/bin/codex}
             - {name: ATLAS_AI_CODEX_QUERY_TIMEOUT_SECONDS, value: "45"}
-            - {name: ATLAS_AI_CLAUDE_CREDENTIALS, value: /runtime-access/claude/.credentials.json}
+            - {name: ATLAS_AI_CLAUDE_OAUTH_TOKEN_FILE, value: /claude-oauth-access/token}
+            - {name: ATLAS_AI_CLAUDE_QUERY_TIMEOUT_SECONDS, value: "30"}
+            - {name: ATLAS_AI_CLAUDE_QUOTA_MODEL, value: claude-haiku-4-5-20251001}
             - {name: ATLAS_AI_PROVIDER_HEALTH_ROOT, value: /provider-health}
             - {name: ATLAS_AI_USAGE_INTERVAL_SECONDS, value: "300"}
             - {name: ATLAS_AI_USAGE_PORT, value: "9010"}
@@ -1057,6 +1059,7 @@ spec:
             seccompProfile:
               type: RuntimeDefault
           volumeMounts:
+            - {name: claude-oauth-access, mountPath: /claude-oauth-access, readOnly: true}
             - {name: home, mountPath: /opt/data/tools, subPath: tools, readOnly: true}
             - {name: home, mountPath: /provider-health, subPath: provider-health, readOnly: true}
             - {name: runtime-access, mountPath: /runtime-access/claude, subPath: claude}

services/hermes/kustomization.yaml

@@ -102,6 +102,7 @@ configMapGenerator:
   - name: hermes-coordinator
     namespace: hermes
     files:
+      - ai_usage_claude.py=scripts/ai_usage_claude.py
       - ai_usage_codex.py=scripts/ai_usage_codex.py
       - ai_usage_exporter.py=scripts/ai_usage_exporter.py
       - ai_usage_http.py=scripts/ai_usage_http.py

services/hermes/scripts/ai_usage_claude.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Read Claude subscription quotas from first-party response headers."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+from urllib.request import Request, urlopen
+
+
+OAUTH_TOKEN_FILE = Path(
+    os.environ.get(
+        "ATLAS_AI_CLAUDE_OAUTH_TOKEN_FILE",
+        "/claude-oauth-access/token",
+    )
+)
+MESSAGES_URL = os.environ.get(
+    "ATLAS_AI_CLAUDE_MESSAGES_URL",
+    "https://api.anthropic.com/v1/messages",
+)
+QUOTA_MODEL = os.environ.get(
+    "ATLAS_AI_CLAUDE_QUOTA_MODEL",
+    "claude-haiku-4-5-20251001",
+)
+REQUEST_TIMEOUT_SECONDS = max(
+    5.0,
+    float(os.environ.get("ATLAS_AI_CLAUDE_QUERY_TIMEOUT_SECONDS", "30")),
+)
+WINDOW_HEADERS = {
+    "five_hour": "5h",
+    "seven_day": "7d",
+}
+
+
+class QuotaNotExposed(RuntimeError):
+    """The active provider credential cannot expose account quota."""
+
+
+def _token() -> str:
+    """Read the Vault-rendered setup token from its memory-backed file."""
+    try:
+        token = OAUTH_TOKEN_FILE.read_text(encoding="utf-8").strip()
+    except (OSError, UnicodeError) as error:
+        raise QuotaNotExposed("Claude quota token is unavailable") from error
+    if not token:
+        raise QuotaNotExposed("Claude quota token is unavailable")
+    return token
+
+
+def _percentage(value: str | None) -> float | None:
+    """Convert Anthropic's fractional utilization header to a percentage."""
+    try:
+        utilization = float(value) if value is not None else None
+    except ValueError:
+        return None
+    if utilization is None or not 0 <= utilization <= 1:
+        return None
+    return utilization * 100
+
+
+def _timestamp(value: str | None) -> float | None:
+    """Validate one Unix reset timestamp from a response header."""
+    try:
+        timestamp = float(value) if value is not None else None
+    except ValueError:
+        return None
+    if timestamp is None or timestamp <= 0:
+        return None
+    return timestamp
+
+
+def _payload(headers: Any) -> dict[str, Any]:
+    """Translate bounded unified-rate-limit headers into the usage document."""
+    payload: dict[str, Any] = {}
+    for window, abbreviation in WINDOW_HEADERS.items():
+        prefix = f"anthropic-ratelimit-unified-{abbreviation}"
+        utilization = _percentage(headers.get(f"{prefix}-utilization"))
+        if utilization is None:
+            continue
+        result: dict[str, Any] = {"utilization": utilization}
+        reset = _timestamp(headers.get(f"{prefix}-reset"))
+        if reset is not None:
+            result["resets_at"] = reset
+        payload[window] = result
+    if not payload:
+        raise QuotaNotExposed("Claude response omitted subscription quota headers")
+    return payload
+
+
+def query_claude() -> dict[str, Any]:
+    """Make one minimal subscription request and return exact quota windows."""
+    token = _token()
+    body = json.dumps(
+        {
+            "model": QUOTA_MODEL,
+            "max_tokens": 1,
+            "messages": [{"role": "user", "content": "quota"}],
+        },
+        separators=(",", ":"),
+    ).encode("utf-8")
+    request = Request(
+        MESSAGES_URL,
+        data=body,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "anthropic-version": "2023-06-01",
+            "anthropic-beta": "oauth-2025-04-20,claude-code-20250219",
+            "User-Agent": "atlas-ai-usage-exporter/2.0",
+        },
+    )
+    with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+        payload = _payload(response.headers)
+        response.read(1 << 20)
+    return payload

services/hermes/scripts/ai_usage_exporter.py

@@ -11,26 +11,16 @@ from dataclasses import dataclass, field
 from datetime import UTC, date, datetime, timedelta
 from pathlib import Path
 from typing import Any
-from urllib.request import Request, urlopen
 
+import ai_usage_claude as claude_query
 import ai_usage_codex as codex_query
 import ai_usage_http as http_engine
 import ai_usage_polling as polling_engine
 
 
-CLAUDE_CREDENTIALS = Path(
-    os.environ.get(
-        "ATLAS_AI_CLAUDE_CREDENTIALS",
-        "/runtime-access/claude/.credentials.json",
-    )
-)
 PROVIDER_HEALTH_ROOT = Path(
     os.environ.get("ATLAS_AI_PROVIDER_HEALTH_ROOT", "/provider-health")
 )
-CLAUDE_USAGE_URL = os.environ.get(
-    "ATLAS_AI_CLAUDE_USAGE_URL",
-    "https://api.anthropic.com/api/oauth/usage",
-)
 CLAUDE_WINDOWS = (
     "five_hour",
     "seven_day",
@@ -51,8 +41,8 @@ METRIC_HELP = {
     "atlas_ai_quota_used_percent": "Used percentage in a first-party coding CLI quota window.",
     "atlas_ai_quota_window_duration_seconds": "Nominal duration of a coding CLI quota window.",
 }
-class QuotaNotExposed(RuntimeError):
+
-    """The active provider credential cannot expose account quota."""
+QuotaNotExposed = claude_query.QuotaNotExposed
 
 
 @dataclass(frozen=True)
@@ -258,6 +248,7 @@ def parse_claude_payload(payload: dict[str, Any]) -> list[Sample]:
 
 
 query_codex = codex_query.query_codex
+query_claude = claude_query.query_claude
 
 
 def _provider_authenticated(provider: str) -> bool:
@@ -272,38 +263,6 @@ def _provider_authenticated(provider: str) -> bool:
     return isinstance(document, dict) and document.get("authenticated") is True
 
 
-def query_claude() -> dict[str, Any]:
-    """Read Claude account quota with the runtime OAuth token held only in memory."""
-    try:
-        document = json.loads(CLAUDE_CREDENTIALS.read_text(encoding="utf-8"))
-    except (OSError, UnicodeError, json.JSONDecodeError) as error:
-        raise QuotaNotExposed("Claude runtime credentials are incomplete") from error
-    oauth = document.get("claudeAiOauth", {})
-    token = oauth.get("accessToken") if isinstance(oauth, dict) else None
-    if not isinstance(token, str) or not token:
-        raise QuotaNotExposed("Claude runtime credentials are incomplete")
-    expires_at = _number(oauth.get("expiresAt"))
-    if expires_at is not None:
-        if expires_at > 10_000_000_000:
-            expires_at /= 1000
-        if expires_at <= time.time():
-            raise QuotaNotExposed("Claude quota credential has expired")
-    request = Request(
-        CLAUDE_USAGE_URL,
-        headers={
-            "Authorization": f"Bearer {token}",
-            "Accept": "application/json",
-            "anthropic-version": "2023-06-01",
-            "User-Agent": "atlas-ai-usage-exporter/1.0",
-        },
-    )
-    with urlopen(request, timeout=15) as response:
-        payload = json.loads(response.read(1 << 20))
-    if not isinstance(payload, dict):
-        raise RuntimeError("Claude usage response is not an object")
-    return payload
-
-
 def _escape_label(value: str) -> str:
     """Escape one Prometheus label value."""
     return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')

testing/tests/test_hermes_ai_usage_exporter.py

@@ -262,11 +262,21 @@ def test_manifest_rolls_out_the_bounded_codex_deadline_and_poller_module():
     exporter = next(item for item in containers if item["name"] == "ai-usage-exporter")
     environment = {item["name"]: item["value"] for item in exporter["env"]}
 
-    assert annotations["ai.bstein.dev/config-rev"] == "20260823-ai-quota-stability"
+    assert annotations["ai.bstein.dev/config-rev"] == (
+        "20260823-claude-quota-headers"
+    )
     assert environment["ATLAS_AI_CODEX_QUERY_TIMEOUT_SECONDS"] == "45"
-    assert "ai_usage_polling.py=scripts/ai_usage_polling.py" in (
+    assert environment["ATLAS_AI_CLAUDE_OAUTH_TOKEN_FILE"] == (
-        HERMES / "kustomization.yaml"
+        "/claude-oauth-access/token"
-    ).read_text()
+    )
+    assert environment["ATLAS_AI_CLAUDE_QUOTA_MODEL"] == (
+        "claude-haiku-4-5-20251001"
+    )
+    mounts = {item["name"]: item for item in exporter["volumeMounts"]}
+    assert mounts["claude-oauth-access"]["readOnly"] is True
+    kustomization = (HERMES / "kustomization.yaml").read_text()
+    assert "ai_usage_claude.py=scripts/ai_usage_claude.py" in kustomization
+    assert "ai_usage_polling.py=scripts/ai_usage_polling.py" in kustomization
 
 
 def test_health_endpoint_uses_poller_state_and_ignores_provider_failure():

testing/tests/test_hermes_ai_usage_exporter_coverage.py

@@ -90,25 +90,57 @@ def test_claude_parser_ignores_unbounded_fields_and_invalid_extra_usage():
     assert mod.parse_claude_payload({"extra_usage": "invalid"}) == []
 
 
+def test_claude_header_probe_rejects_malformed_optional_values(
+    tmp_path, monkeypatch
+):
+    mod = load_module()
+    missing = tmp_path / "missing-token"
+    monkeypatch.setattr(mod.claude_query, "OAUTH_TOKEN_FILE", missing)
+
+    with pytest.raises(RuntimeError, match="token is unavailable"):
+        mod.claude_query._token()
+
+    assert mod.claude_query._percentage("invalid") is None
+    assert mod.claude_query._percentage(None) is None
+    assert mod.claude_query._percentage("-0.1") is None
+    assert mod.claude_query._percentage("1.1") is None
+    assert mod.claude_query._timestamp("invalid") is None
+    assert mod.claude_query._timestamp(None) is None
+    assert mod.claude_query._timestamp("0") is None
+    assert mod.claude_query._payload(
+        {"anthropic-ratelimit-unified-5h-utilization": "0.5"}
+    ) == {"five_hour": {"utilization": 50}}
+
+
 def test_provider_queries_validate_credentials_and_response_shape(
     tmp_path, monkeypatch
 ):
     mod = load_module()
-    credentials = tmp_path / "credentials.json"
+    token_file = tmp_path / "claude-oauth-token"
-    credentials.write_text("{}", encoding="utf-8")
+    token_file.write_text("", encoding="utf-8")
-    monkeypatch.setattr(mod, "CLAUDE_CREDENTIALS", credentials)
+    monkeypatch.setattr(mod.claude_query, "OAUTH_TOKEN_FILE", token_file)
 
-    with pytest.raises(RuntimeError, match="credentials are incomplete"):
+    with pytest.raises(RuntimeError, match="token is unavailable"):
         mod.query_claude()
 
-    credentials.write_text(
+    token_file.write_text("runtime-only-token\n", encoding="utf-8")
-        json.dumps({"claudeAiOauth": {"accessToken": "runtime-only-token"}}),
+    response_headers = iter(
-        encoding="utf-8",
+        [
+            {
+                "anthropic-ratelimit-unified-5h-utilization": "0.16",
+                "anthropic-ratelimit-unified-5h-reset": "1800000000",
+                "anthropic-ratelimit-unified-7d-utilization": "0.26",
+                "anthropic-ratelimit-unified-7d-reset": "1800100000",
+            },
+            {},
+        ]
     )
-    bodies = iter([{"five_hour": {}}, []])
     requests = []
 
     class Response:
+        def __init__(self):
+            self.headers = next(response_headers)
+
         def __enter__(self):
             return self
 
@@ -117,44 +149,46 @@ def test_provider_queries_validate_credentials_and_response_shape(
 
         def read(self, maximum):
             assert maximum == 1 << 20
-            return json.dumps(next(bodies)).encode()
+            return b"{}"
 
     def open_request(request, timeout):
         requests.append(request)
-        assert timeout == 15
+        assert timeout == 30
         return Response()
 
-    monkeypatch.setattr(mod, "urlopen", open_request)
+    monkeypatch.setattr(mod.claude_query, "urlopen", open_request)
-    assert mod.query_claude() == {"five_hour": {}}
+    assert mod.query_claude() == {
+        "five_hour": {"utilization": 16, "resets_at": 1_800_000_000},
+        "seven_day": {"utilization": 26, "resets_at": 1_800_100_000},
+    }
     assert requests[0].get_header("Authorization") == "Bearer runtime-only-token"
-    with pytest.raises(RuntimeError, match="response is not an object"):
+    assert requests[0].get_header("Anthropic-beta") == (
+        "oauth-2025-04-20,claude-code-20250219"
+    )
+    assert json.loads(requests[0].data) == {
+        "model": "claude-haiku-4-5-20251001",
+        "max_tokens": 1,
+        "messages": [{"role": "user", "content": "quota"}],
+    }
+    with pytest.raises(RuntimeError, match="omitted subscription quota headers"):
         mod.query_claude()
 
     assert mod.query_codex is mod.codex_query.query_codex
 
 
-def test_expired_claude_quota_credential_is_not_sent_or_logged(
+def test_claude_quota_failure_does_not_log_the_vault_token(
     tmp_path, monkeypatch, capsys
 ):
     mod = load_module()
-    credentials = tmp_path / "credentials.json"
+    token_file = tmp_path / "claude-oauth-token"
-    credentials.write_text(
+    token_file.write_text("revoked-runtime-token", encoding="utf-8")
-        json.dumps(
+    monkeypatch.setattr(mod.claude_query, "OAUTH_TOKEN_FILE", token_file)
-            {
-                "claudeAiOauth": {
-                    "accessToken": "revoked-runtime-token",
-                    "expiresAt": 1_700_000_000_000,
-                }
-            }
-        ),
-        encoding="utf-8",
-    )
-    monkeypatch.setattr(mod, "CLAUDE_CREDENTIALS", credentials)
-    monkeypatch.setattr(mod.time, "time", lambda: 1_800_000_000)
     monkeypatch.setattr(
-        mod,
+        mod.claude_query,
         "urlopen",
-        lambda *_args, **_kwargs: pytest.fail("expired credentials must not be sent"),
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            TimeoutError("revoked-runtime-token")
+        ),
     )
 
     collector = mod.Collector()

testing/tests/test_hermes_runtime_access.py

@@ -467,6 +467,7 @@ def test_manifests_never_seed_access_material_into_persistent_env():
         "cli-lane-runner",
         "model-steward",
         "claude-broker",
+        "ai-usage-exporter",
     }
     token_init_containers = {
         item["name"]