From c4eb872690badff93bb0da254c55917d31664771 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 24 Aug 2026 14:13:35 -0300 Subject: [PATCH] monitoring(ai): show Claude Fable weekly quota --- scripts/render/dashboards_render_atlas.py | 40 +- services/hermes/agent-deployment.yaml | 4 +- services/hermes/scripts/ai_usage_claude.py | 296 +++++++++++++- services/hermes/scripts/ai_usage_exporter.py | 1 + services/monitoring/dashboards/atlas-ai.json | 11 +- services/monitoring/grafana-dashboard-ai.yaml | 11 +- testing/tests/test_atlas_ai_dashboard.py | 19 + .../test_hermes_ai_usage_claude_scoped.py | 365 ++++++++++++++++++ .../tests/test_hermes_ai_usage_exporter.py | 19 +- .../test_hermes_ai_usage_exporter_coverage.py | 4 +- 10 files changed, 748 insertions(+), 22 deletions(-) create mode 100644 testing/tests/test_hermes_ai_usage_claude_scoped.py diff --git a/scripts/render/dashboards_render_atlas.py b/scripts/render/dashboards_render_atlas.py index 0b22498f..6541f7c5 100644 --- a/scripts/render/dashboards_render_atlas.py +++ b/scripts/render/dashboards_render_atlas.py @@ -5199,6 +5199,8 @@ def build_ai_dashboard(): unit="percent", authenticated_expr=None, description=None, + targets=None, + text_mode="value", ): thresholds = remaining_thresholds if unit != "percent": @@ -5209,8 +5211,20 @@ def build_ai_dashboard(): {"color": "blue", "value": 0}, ], } - if authenticated_expr: - expr = f"({expr}) or on() (({authenticated_expr}) * 0 - 2)" + def bounded_expression(target_expr): + """Keep each stat series explicit when access or freshness is absent.""" + if authenticated_expr: + target_expr = ( + f"({target_expr}) or on() (({authenticated_expr}) * 0 - 2)" + ) + return f"({target_expr}) or on() vector(-1)" + + raw_targets = targets or [{"expr": expr, "refId": "A"}] + bounded_targets = [] + for target in raw_targets: + bounded = dict(target) + bounded["expr"] = bounded_expression(target["expr"]) + bounded_targets.append(bounded) quota_description = ( "A last-good snapshot remains visible for up to 20 minutes through " "a transient fetch miss. Not exposed means access is healthy but the " @@ -5220,12 +5234,14 @@ def build_ai_dashboard(): panel = stat_panel( panel_id, title, - f"({expr}) or on() vector(-1)", + "", grid, unit=unit, decimals=1, thresholds=thresholds, instant=True, + targets=bounded_targets, + text_mode=text_mode, description=( f"{description} {quota_description}" if description @@ -5370,7 +5386,23 @@ def build_ai_dashboard(): f'{provider_quota(quota, "anthropic", "seven_day")} and on(provider) ({anthropic_snapshot})', {"h": 4, "w": 4, "x": 8, "y": 4}, authenticated_expr=anthropic_authenticated, - description=remaining_description, + description=( + "Overall weekly and Fable 5 weekly remaining percentages from " + "Claude Code's first-party scoped usage ledger." + ), + targets=[ + { + "expr": f'{provider_quota(quota, "anthropic", "seven_day")} and on(provider) ({anthropic_snapshot})', + "refId": "A", + "legendFormat": "Overall weekly", + }, + { + "expr": f'{provider_quota(quota, "anthropic", "seven_day_fable")} and on(provider) ({anthropic_snapshot})', + "refId": "B", + "legendFormat": "Fable 5 weekly", + }, + ], + text_mode="value_and_name", ), quota_stat( 10, diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 1bee1c0a..2e17c842 100644 --- a/services/hermes/agent-deployment.yaml +++ b/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: titan-08 rpi5; storage-backbone nodes excluded - ai.bstein.dev/config-rev: "20260823-image-release-automation" + ai.bstein.dev/config-rev: "20260824-claude-fable-quota" prometheus.io/scrape: "true" prometheus.io/path: /metrics prometheus.io/port: "9010" @@ -1168,9 +1168,11 @@ 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_CLAUDE_USAGE_URL, value: 'https://api.anthropic.com/api/oauth/usage'} - {name: ATLAS_AI_AUTHENTICATION_GRACE_SECONDS, value: "1200"} - {name: ATLAS_AI_PROVIDER_HEALTH_ROOT, value: /provider-health} - {name: ATLAS_AI_USAGE_INTERVAL_SECONDS, value: "300"} diff --git a/services/hermes/scripts/ai_usage_claude.py b/services/hermes/scripts/ai_usage_claude.py index 4346a059..bfe00bd7 100644 --- a/services/hermes/scripts/ai_usage_claude.py +++ b/services/hermes/scripts/ai_usage_claude.py @@ -1,12 +1,16 @@ #!/usr/bin/env python3 -"""Read Claude subscription quotas from first-party response headers.""" +"""Read Claude subscription quotas without exposing account credentials.""" from __future__ import annotations import json import os +import stat +import time +import uuid from pathlib import Path from typing import Any +from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @@ -16,10 +20,25 @@ OAUTH_TOKEN_FILE = Path( "/claude-oauth-access/token", ) ) +CREDENTIALS_FILE = Path( + os.environ.get( + "ATLAS_AI_CLAUDE_CREDENTIALS", + "/runtime-access/claude/.credentials.json", + ) +) +USAGE_URL = os.environ.get( + "ATLAS_AI_CLAUDE_USAGE_URL", + "https://api.anthropic.com/api/oauth/usage", +) +TOKEN_URL = os.environ.get( + "ATLAS_AI_CLAUDE_TOKEN_URL", + "https://platform.claude.com/v1/oauth/token", +) MESSAGES_URL = os.environ.get( "ATLAS_AI_CLAUDE_MESSAGES_URL", "https://api.anthropic.com/v1/messages", ) +OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" QUOTA_MODEL = os.environ.get( "ATLAS_AI_CLAUDE_QUOTA_MODEL", "claude-haiku-4-5-20251001", @@ -32,13 +51,29 @@ WINDOW_HEADERS = { "five_hour": "5h", "seven_day": "7d", } +SCOPED_WINDOWS = ( + "five_hour", + "seven_day", + "seven_day_opus", + "seven_day_sonnet", +) +FABLE_DISPLAY_NAMES = frozenset(("fable", "fable 5")) +DEFAULT_SCOPES = ( + "user:profile", + "user:inference", + "user:sessions:claude_code", + "user:mcp_servers", + "user:file_upload", +) +MAX_DOCUMENT_BYTES = 1 << 20 +REFRESH_SKEW_MILLISECONDS = 5 * 60 * 1000 class QuotaNotExposed(RuntimeError): """The active provider credential cannot expose account quota.""" -def _token() -> str: +def _setup_token() -> str: """Read the Vault-rendered setup token from its memory-backed file.""" try: token = OAUTH_TOKEN_FILE.read_text(encoding="utf-8").strip() @@ -60,6 +95,16 @@ def _percentage(value: str | None) -> float | None: return utilization * 100 +def _used_percentage(value: object) -> float | None: + """Validate one percentage returned by the scoped usage endpoint.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + percentage = float(value) + if not 0 <= percentage <= 100: + return None + return percentage + + def _timestamp(value: str | None) -> float | None: """Validate one Unix reset timestamp from a response header.""" try: @@ -71,8 +116,8 @@ def _timestamp(value: str | None) -> float | None: return timestamp -def _payload(headers: Any) -> dict[str, Any]: - """Translate bounded unified-rate-limit headers into the usage document.""" +def _header_payload(headers: Any) -> dict[str, Any]: + """Translate bounded unified-rate-limit headers into a usage document.""" payload: dict[str, Any] = {} for window, abbreviation in WINDOW_HEADERS.items(): prefix = f"anthropic-ratelimit-unified-{abbreviation}" @@ -89,9 +134,222 @@ def _payload(headers: Any) -> dict[str, Any]: return payload -def query_claude() -> dict[str, Any]: - """Make one minimal subscription request and return exact quota windows.""" - token = _token() +def _bounded_window(value: object) -> dict[str, Any] | None: + """Copy only the numeric utilization and bounded reset field.""" + if not isinstance(value, dict): + return None + utilization = _used_percentage(value.get("utilization")) + if utilization is None: + return None + result: dict[str, Any] = {"utilization": utilization} + reset = value.get("resets_at") + if isinstance(reset, (int, float, str)) and not isinstance(reset, bool): + result["resets_at"] = reset + return result + + +def _scoped_payload(document: object) -> dict[str, Any]: + """Normalize overall and exact Fable weekly limits from Claude Code usage.""" + if not isinstance(document, dict): + raise QuotaNotExposed("Claude usage response is not an object") + payload: dict[str, Any] = {} + for window in SCOPED_WINDOWS: + bounded = _bounded_window(document.get(window)) + if bounded is not None: + payload[window] = bounded + limits = document.get("limits") + if isinstance(limits, list): + for limit in limits[:100]: + if not isinstance(limit, dict) or limit.get("kind") != "weekly_scoped": + continue + scope = limit.get("scope") + model = scope.get("model") if isinstance(scope, dict) else None + display_name = model.get("display_name") if isinstance(model, dict) else None + if not isinstance(display_name, str): + continue + if display_name.strip().casefold() not in FABLE_DISPLAY_NAMES: + continue + utilization = _used_percentage(limit.get("percent")) + if utilization is None: + continue + fable: dict[str, Any] = {"utilization": utilization} + reset = limit.get("resets_at") + if isinstance(reset, (int, float, str)) and not isinstance(reset, bool): + fable["resets_at"] = reset + payload["seven_day_fable"] = fable + break + if not payload: + raise QuotaNotExposed("Claude usage response omitted supported quota windows") + return payload + + +def _read_credentials() -> tuple[dict[str, Any], dict[str, Any]]: + """Read one private Claude OAuth document without following a final symlink.""" + try: + descriptor = os.open( + CREDENTIALS_FILE, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), + ) + except OSError as error: + raise QuotaNotExposed("Claude scoped quota credential is unavailable") from error + try: + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_mode & 0o077 + or not 0 < info.st_size <= MAX_DOCUMENT_BYTES + ): + raise QuotaNotExposed("Claude scoped quota credential is unsafe") + raw = os.read(descriptor, info.st_size + 1) + finally: + os.close(descriptor) + try: + document = json.loads(raw) + except (UnicodeError, json.JSONDecodeError) as error: + raise QuotaNotExposed("Claude scoped quota credential is invalid") from error + credentials = document.get("claudeAiOauth") if isinstance(document, dict) else None + if not isinstance(credentials, dict): + raise QuotaNotExposed("Claude scoped quota credential is invalid") + return document, credentials + + +def _write_credentials(document: dict[str, Any]) -> None: + """Atomically persist a provider-rotated OAuth document for Vault sync.""" + encoded = (json.dumps(document, separators=(",", ":")) + "\n").encode("utf-8") + if len(encoded) > MAX_DOCUMENT_BYTES: + raise QuotaNotExposed("Claude scoped quota credential is too large") + temporary = CREDENTIALS_FILE.with_name( + f".{CREDENTIALS_FILE.name}.{uuid.uuid4().hex}.tmp" + ) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary, flags, 0o600) + try: + remaining = memoryview(encoded) + while remaining: + remaining = remaining[os.write(descriptor, remaining) :] + os.fsync(descriptor) + os.fchmod(descriptor, 0o600) + finally: + os.close(descriptor) + try: + os.replace(temporary, CREDENTIALS_FILE) + finally: + temporary.unlink(missing_ok=True) + + +def _refresh_access_token( + document: dict[str, Any], credentials: dict[str, Any] +) -> str: + """Refresh the scoped access token and retain any refresh-token rotation.""" + refresh_token = credentials.get("refreshToken") + if not isinstance(refresh_token, str) or not refresh_token: + raise QuotaNotExposed("Claude scoped quota refresh credential is unavailable") + scopes = credentials.get("scopes") + if not ( + isinstance(scopes, list) + and scopes + and all(isinstance(scope, str) and scope in DEFAULT_SCOPES for scope in scopes) + ): + scopes = list(DEFAULT_SCOPES) + body = json.dumps( + { + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": OAUTH_CLIENT_ID, + "scope": " ".join(scopes), + }, + separators=(",", ":"), + ).encode("utf-8") + request = Request( + TOKEN_URL, + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": "atlas-ai-usage-exporter/3.0", + }, + ) + with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + response_document = json.loads(response.read(MAX_DOCUMENT_BYTES)) + if not isinstance(response_document, dict): + raise QuotaNotExposed("Claude token refresh response is invalid") + access_token = response_document.get("access_token") + expires_in = response_document.get("expires_in") + rotated_refresh = response_document.get("refresh_token", refresh_token) + if ( + not isinstance(access_token, str) + or not access_token + or isinstance(expires_in, bool) + or not isinstance(expires_in, (int, float)) + or expires_in <= 0 + or not isinstance(rotated_refresh, str) + or not rotated_refresh + ): + raise QuotaNotExposed("Claude token refresh response is invalid") + credentials["accessToken"] = access_token + credentials["refreshToken"] = rotated_refresh + credentials["expiresAt"] = int(time.time() * 1000 + expires_in * 1000) + response_scopes = response_document.get("scope") + if isinstance(response_scopes, str) and response_scopes: + credentials["scopes"] = response_scopes.split() + refresh_expires_in = response_document.get("refresh_token_expires_in") + if ( + not isinstance(refresh_expires_in, bool) + and isinstance(refresh_expires_in, (int, float)) + and refresh_expires_in > 0 + ): + credentials["refreshTokenExpiresAt"] = int( + time.time() * 1000 + refresh_expires_in * 1000 + ) + _write_credentials(document) + return access_token + + +def _access_token(*, force_refresh: bool = False) -> str: + """Return a current scoped token, refreshing it before provider expiry.""" + document, credentials = _read_credentials() + access_token = credentials.get("accessToken") + expires_at = credentials.get("expiresAt") + current = time.time() * 1000 + if ( + not force_refresh + and isinstance(access_token, str) + and access_token + and not isinstance(expires_at, bool) + and isinstance(expires_at, (int, float)) + and expires_at - current > REFRESH_SKEW_MILLISECONDS + ): + return access_token + return _refresh_access_token(document, credentials) + + +def _query_scoped_usage() -> dict[str, Any]: + """Fetch Claude Code's authenticated overall and model-scoped quotas.""" + token = _access_token() + for attempt in range(2): + request = Request( + USAGE_URL, + headers={ + "Authorization": f"Bearer {token}", + "Accept": "application/json", + "anthropic-beta": "oauth-2025-04-20", + "User-Agent": "atlas-ai-usage-exporter/3.0", + }, + ) + try: + with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + return _scoped_payload(json.loads(response.read(MAX_DOCUMENT_BYTES))) + except HTTPError as error: + if error.code != 401 or attempt: + raise + token = _access_token(force_refresh=True) + raise QuotaNotExposed("Claude scoped quota request failed") + + +def _query_header_usage() -> dict[str, Any]: + """Fetch unified quotas through the inference-scoped setup-token fallback.""" + token = _setup_token() body = json.dumps( { "model": QUOTA_MODEL, @@ -110,10 +368,28 @@ def query_claude() -> dict[str, Any]: "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", + "User-Agent": "atlas-ai-usage-exporter/3.0", }, ) with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: - payload = _payload(response.headers) - response.read(1 << 20) + payload = _header_payload(response.headers) + response.read(MAX_DOCUMENT_BYTES) return payload + + +def query_claude() -> dict[str, Any]: + """Prefer exact scoped quotas and preserve unified metrics as a fallback.""" + try: + return _query_scoped_usage() + except ( + HTTPError, + URLError, + TimeoutError, + OSError, + UnicodeError, + json.JSONDecodeError, + QuotaNotExposed, + TypeError, + ValueError, + ): + return _query_header_usage() diff --git a/services/hermes/scripts/ai_usage_exporter.py b/services/hermes/scripts/ai_usage_exporter.py index 9f03de71..bb271551 100644 --- a/services/hermes/scripts/ai_usage_exporter.py +++ b/services/hermes/scripts/ai_usage_exporter.py @@ -28,6 +28,7 @@ AUTHENTICATION_GRACE_SECONDS = max( CLAUDE_WINDOWS = ( "five_hour", "seven_day", + "seven_day_fable", "seven_day_opus", "seven_day_sonnet", ) diff --git a/services/monitoring/dashboards/atlas-ai.json b/services/monitoring/dashboards/atlas-ai.json index d25d81c3..862e0b65 100644 --- a/services/monitoring/dashboards/atlas-ai.json +++ b/services/monitoring/dashboards/atlas-ai.json @@ -646,6 +646,13 @@ { "expr": "((min by (provider) (atlas_ai_quota_remaining_percent{provider=\"anthropic\",window=\"seven_day\"}) and on(provider) ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) and on(provider) (time() - atlas_ai_quota_last_success_timestamp_seconds{provider=\"anthropic\"} < 1200))) or on() ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) * 0 - 2)) or on() vector(-1)", "refId": "A", + "legendFormat": "Overall weekly", + "instant": true + }, + { + "expr": "((min by (provider) (atlas_ai_quota_remaining_percent{provider=\"anthropic\",window=\"seven_day_fable\"}) and on(provider) ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) and on(provider) (time() - atlas_ai_quota_last_success_timestamp_seconds{provider=\"anthropic\"} < 1200))) or on() ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) * 0 - 2)) or on() vector(-1)", + "refId": "B", + "legendFormat": "Fable 5 weekly", "instant": true } ], @@ -713,9 +720,9 @@ "fields": "", "values": false }, - "textMode": "value" + "textMode": "value_and_name" }, - "description": "Lowest remaining percentage across the provider's active first-party quota buckets for this window. This conservative value shows the limit most likely to constrain new work. A last-good snapshot remains visible for up to 20 minutes through a transient fetch miss. Not exposed means access is healthy but the credential cannot read this quota; unavailable means access is unhealthy or the snapshot is stale." + "description": "Overall weekly and Fable 5 weekly remaining percentages from Claude Code's first-party scoped usage ledger. A last-good snapshot remains visible for up to 20 minutes through a transient fetch miss. Not exposed means access is healthy but the credential cannot read this quota; unavailable means access is unhealthy or the snapshot is stale." }, { "id": 10, diff --git a/services/monitoring/grafana-dashboard-ai.yaml b/services/monitoring/grafana-dashboard-ai.yaml index 81ae304a..b96f13d4 100644 --- a/services/monitoring/grafana-dashboard-ai.yaml +++ b/services/monitoring/grafana-dashboard-ai.yaml @@ -655,6 +655,13 @@ data: { "expr": "((min by (provider) (atlas_ai_quota_remaining_percent{provider=\"anthropic\",window=\"seven_day\"}) and on(provider) ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) and on(provider) (time() - atlas_ai_quota_last_success_timestamp_seconds{provider=\"anthropic\"} < 1200))) or on() ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) * 0 - 2)) or on() vector(-1)", "refId": "A", + "legendFormat": "Overall weekly", + "instant": true + }, + { + "expr": "((min by (provider) (atlas_ai_quota_remaining_percent{provider=\"anthropic\",window=\"seven_day_fable\"}) and on(provider) ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) and on(provider) (time() - atlas_ai_quota_last_success_timestamp_seconds{provider=\"anthropic\"} < 1200))) or on() ((atlas_ai_provider_authenticated{provider=\"anthropic\"} == 1) * 0 - 2)) or on() vector(-1)", + "refId": "B", + "legendFormat": "Fable 5 weekly", "instant": true } ], @@ -722,9 +729,9 @@ data: "fields": "", "values": false }, - "textMode": "value" + "textMode": "value_and_name" }, - "description": "Lowest remaining percentage across the provider's active first-party quota buckets for this window. This conservative value shows the limit most likely to constrain new work. A last-good snapshot remains visible for up to 20 minutes through a transient fetch miss. Not exposed means access is healthy but the credential cannot read this quota; unavailable means access is unhealthy or the snapshot is stale." + "description": "Overall weekly and Fable 5 weekly remaining percentages from Claude Code's first-party scoped usage ledger. A last-good snapshot remains visible for up to 20 minutes through a transient fetch miss. Not exposed means access is healthy but the credential cannot read this quota; unavailable means access is unhealthy or the snapshot is stale." }, { "id": 10, diff --git a/testing/tests/test_atlas_ai_dashboard.py b/testing/tests/test_atlas_ai_dashboard.py index cf3c67e0..e20fcfb5 100644 --- a/testing/tests/test_atlas_ai_dashboard.py +++ b/testing/tests/test_atlas_ai_dashboard.py @@ -67,6 +67,25 @@ def test_ai_dashboard_is_internal_and_uses_real_quota_and_switchyard_metrics(): ) +def test_claude_weekly_panel_compares_overall_and_fable_remaining(): + """The Claude weekly KPI must show both first-party quota ledgers.""" + mod = load_module() + panels = {panel["title"]: panel for panel in mod.build_ai_dashboard()["panels"]} + targets = panels["Claude 7d Remaining"]["targets"] + + assert panels["Claude 7d Remaining"]["options"]["textMode"] == "value_and_name" + assert [target["legendFormat"] for target in targets] == [ + "Overall weekly", + "Fable 5 weekly", + ] + assert 'window="seven_day"' in targets[0]["expr"] + assert 'window="seven_day_fable"' in targets[1]["expr"] + assert all( + "atlas_ai_quota_last_success_timestamp_seconds" in target["expr"] + for target in targets + ) + + def test_ai_quota_panels_retain_a_bounded_last_good_snapshot(): """One transient collector miss must not blank every account panel.""" mod = load_module() diff --git a/testing/tests/test_hermes_ai_usage_claude_scoped.py b/testing/tests/test_hermes_ai_usage_claude_scoped.py new file mode 100644 index 00000000..d8519df7 --- /dev/null +++ b/testing/tests/test_hermes_ai_usage_claude_scoped.py @@ -0,0 +1,365 @@ +"""Scoped Claude Code quota collection and credential-safety contracts.""" + +from __future__ import annotations + +import importlib.util +import json +import stat +import sys +from pathlib import Path +from urllib.error import HTTPError + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = ROOT / "services/hermes/scripts/ai_usage_claude.py" + + +def load_module(): + """Load a fresh Claude usage helper for environment-independent tests.""" + name = "ai_usage_claude_scoped" + spec = importlib.util.spec_from_file_location(name, SCRIPT) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +class Response: + """Small bounded urllib response fixture.""" + + def __init__(self, body, *, headers=None): + self.body = body if isinstance(body, bytes) else json.dumps(body).encode() + self.headers = headers or {} + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self, maximum): + assert maximum == 1 << 20 + return self.body + + +def write_credentials(path, **overrides): + """Write a private synthetic Claude OAuth document.""" + credentials = { + "accessToken": "scoped-access", + "refreshToken": "scoped-refresh", + "expiresAt": 2_000_000, + "scopes": [ + "user:profile", + "user:inference", + "user:sessions:claude_code", + ], + } + credentials.update(overrides) + path.write_text( + json.dumps({"claudeAiOauth": credentials}) + "\n", + encoding="utf-8", + ) + path.chmod(0o600) + + +def test_scoped_payload_selects_only_overall_and_exact_fable_limits(): + mod = load_module() + secret = "provider-document-detail-must-not-leak" + + payload = mod._scoped_payload( + { + "five_hour": {"utilization": 58, "resets_at": "2026-08-24T17:20:00Z"}, + "seven_day": {"utilization": 59, "resets_at": "2026-08-27T18:00:00Z"}, + "limits": [ + { + "kind": "weekly_scoped", + "percent": 81, + "resets_at": "2026-08-27T18:00:00Z", + "scope": {"model": {"display_name": "Fable"}}, + "private": secret, + }, + { + "kind": "weekly_scoped", + "percent": 90, + "scope": {"model": {"display_name": "Unknown"}}, + }, + ], + "account": secret, + } + ) + + assert payload == { + "five_hour": { + "utilization": 58.0, + "resets_at": "2026-08-24T17:20:00Z", + }, + "seven_day": { + "utilization": 59.0, + "resets_at": "2026-08-27T18:00:00Z", + }, + "seven_day_fable": { + "utilization": 81.0, + "resets_at": "2026-08-27T18:00:00Z", + }, + } + assert secret not in repr(payload) + + +def test_scoped_query_uses_current_private_access_token(tmp_path, monkeypatch): + mod = load_module() + credentials = tmp_path / ".credentials.json" + write_credentials(credentials) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + monkeypatch.setattr(mod.time, "time", lambda: 1_000) + requests = [] + + def open_request(request, timeout): + requests.append(request) + assert timeout == 30 + return Response( + { + "seven_day": {"utilization": 42}, + "limits": [ + { + "kind": "weekly_scoped", + "percent": 73, + "scope": {"model": {"display_name": "Fable 5"}}, + } + ], + } + ) + + monkeypatch.setattr(mod, "urlopen", open_request) + + assert mod._query_scoped_usage() == { + "seven_day": {"utilization": 42.0}, + "seven_day_fable": {"utilization": 73.0}, + } + assert len(requests) == 1 + assert requests[0].full_url == mod.USAGE_URL + assert requests[0].data is None + assert requests[0].get_header("Authorization") == "Bearer scoped-access" + + +def test_expired_scoped_token_refreshes_and_persists_rotation( + tmp_path, monkeypatch +): + mod = load_module() + credentials = tmp_path / ".credentials.json" + write_credentials(credentials, expiresAt=900_000) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + monkeypatch.setattr(mod.time, "time", lambda: 1_000) + requests = [] + + def open_request(request, timeout): + requests.append(request) + assert timeout == 30 + if request.full_url == mod.TOKEN_URL: + return Response( + { + "access_token": "rotated-access", + "refresh_token": "rotated-refresh", + "expires_in": 3600, + "refresh_token_expires_in": 7200, + "scope": "user:profile user:inference", + } + ) + return Response( + { + "seven_day": {"utilization": 50}, + "limits": [ + { + "kind": "weekly_scoped", + "percent": 75, + "scope": {"model": {"display_name": "Fable"}}, + } + ], + } + ) + + monkeypatch.setattr(mod, "urlopen", open_request) + + assert mod._query_scoped_usage()["seven_day_fable"]["utilization"] == 75 + refresh_request = requests[0] + assert refresh_request.full_url == mod.TOKEN_URL + refresh_body = json.loads(refresh_request.data) + assert refresh_body == { + "grant_type": "refresh_token", + "refresh_token": "scoped-refresh", + "client_id": mod.OAUTH_CLIENT_ID, + "scope": "user:profile user:inference user:sessions:claude_code", + } + persisted = json.loads(credentials.read_text())["claudeAiOauth"] + assert persisted["accessToken"] == "rotated-access" + assert persisted["refreshToken"] == "rotated-refresh" + assert persisted["expiresAt"] == 4_600_000 + assert persisted["refreshTokenExpiresAt"] == 8_200_000 + assert stat.S_IMODE(credentials.stat().st_mode) == 0o600 + + +def test_query_falls_back_to_setup_token_headers_when_scope_is_unavailable( + tmp_path, monkeypatch +): + mod = load_module() + missing_credentials = tmp_path / "missing-credentials" + setup_token = tmp_path / "setup-token" + setup_token.write_text("setup-only-token\n", encoding="utf-8") + monkeypatch.setattr(mod, "CREDENTIALS_FILE", missing_credentials) + monkeypatch.setattr(mod, "OAUTH_TOKEN_FILE", setup_token) + requests = [] + + def open_request(request, timeout): + requests.append(request) + assert timeout == 30 + return Response( + {}, + headers={ + "anthropic-ratelimit-unified-5h-utilization": "0.25", + "anthropic-ratelimit-unified-7d-utilization": "0.40", + }, + ) + + monkeypatch.setattr(mod, "urlopen", open_request) + + assert mod.query_claude() == { + "five_hour": {"utilization": 25.0}, + "seven_day": {"utilization": 40.0}, + } + assert len(requests) == 1 + assert requests[0].full_url == mod.MESSAGES_URL + assert requests[0].get_header("Authorization") == "Bearer setup-only-token" + + +def test_scoped_validation_and_unauthorized_retry_fail_closed( + tmp_path, monkeypatch +): + mod = load_module() + credentials = tmp_path / ".credentials.json" + write_credentials(credentials) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + monkeypatch.setattr(mod.time, "time", lambda: 1_000) + + assert mod._used_percentage(True) is None + assert mod._used_percentage(-1) is None + assert mod._used_percentage(101) is None + with pytest.raises(RuntimeError, match="supported quota windows"): + mod._scoped_payload({"limits": [{"kind": "weekly_scoped"}]}) + + attempts = [] + + def open_request(request, timeout): + attempts.append(request.full_url) + if len(attempts) == 1: + raise HTTPError(request.full_url, 401, "secret-body", {}, None) + if request.full_url == mod.TOKEN_URL: + return Response( + { + "access_token": "retried-access", + "refresh_token": "retried-refresh", + "expires_in": 3600, + } + ) + return Response({"seven_day": {"utilization": 12}}) + + monkeypatch.setattr(mod, "urlopen", open_request) + assert mod._query_scoped_usage() == {"seven_day": {"utilization": 12.0}} + assert attempts == [mod.USAGE_URL, mod.TOKEN_URL, mod.USAGE_URL] + + +def test_scoped_payload_ignores_malformed_and_unrelated_limit_entries(): + mod = load_module() + + assert mod._scoped_payload( + { + "five_hour": {"utilization": "invalid"}, + "seven_day": {"utilization": 10}, + "limits": [ + None, + {"kind": "session", "percent": 20}, + { + "kind": "weekly_scoped", + "percent": 30, + "scope": {"model": {"display_name": "Sonnet"}}, + }, + { + "kind": "weekly_scoped", + "percent": "invalid", + "scope": {"model": {"display_name": "Fable"}}, + }, + ], + } + ) == {"seven_day": {"utilization": 10.0}} + with pytest.raises(RuntimeError, match="not an object"): + mod._scoped_payload(None) + + +@pytest.mark.parametrize( + ("content", "mode", "message"), + [ + ('{"claudeAiOauth":{}}\n', 0o644, "unsafe"), + ("not-json\n", 0o600, "invalid"), + ("[]\n", 0o600, "invalid"), + ], +) +def test_private_credential_reader_rejects_unsafe_documents( + tmp_path, monkeypatch, content, mode, message +): + mod = load_module() + credentials = tmp_path / ".credentials.json" + credentials.write_text(content, encoding="utf-8") + credentials.chmod(mode) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + + with pytest.raises(RuntimeError, match=message): + mod._read_credentials() + + +def test_credential_writer_and_refresh_validation_are_bounded(tmp_path, monkeypatch): + mod = load_module() + credentials = tmp_path / ".credentials.json" + write_credentials(credentials) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + monkeypatch.setattr(mod, "MAX_DOCUMENT_BYTES", 1) + with pytest.raises(RuntimeError, match="too large"): + mod._write_credentials({"claudeAiOauth": {"accessToken": "large"}}) + + monkeypatch.setattr(mod, "MAX_DOCUMENT_BYTES", 1 << 20) + with pytest.raises(RuntimeError, match="refresh credential is unavailable"): + mod._refresh_access_token({"claudeAiOauth": {}}, {}) + + document = { + "claudeAiOauth": { + "accessToken": "old-access", + "refreshToken": "old-refresh", + "scopes": ["unapproved:scope"], + } + } + for response_document in ([], {"access_token": "", "expires_in": True}): + monkeypatch.setattr( + mod, + "urlopen", + lambda request, timeout, value=response_document: Response(value), + ) + with pytest.raises(RuntimeError, match="refresh response is invalid"): + mod._refresh_access_token(document, document["claudeAiOauth"]) + + +def test_non_authentication_usage_error_is_not_retried(tmp_path, monkeypatch): + mod = load_module() + credentials = tmp_path / ".credentials.json" + write_credentials(credentials) + monkeypatch.setattr(mod, "CREDENTIALS_FILE", credentials) + monkeypatch.setattr(mod.time, "time", lambda: 1_000) + attempts = [] + + def reject(request, timeout): + attempts.append(request.full_url) + raise HTTPError(request.full_url, 503, "provider-detail", {}, None) + + monkeypatch.setattr(mod, "urlopen", reject) + with pytest.raises(HTTPError): + mod._query_scoped_usage() + assert attempts == [mod.USAGE_URL] diff --git a/testing/tests/test_hermes_ai_usage_exporter.py b/testing/tests/test_hermes_ai_usage_exporter.py index 8b5f61b5..8b41438e 100644 --- a/testing/tests/test_hermes_ai_usage_exporter.py +++ b/testing/tests/test_hermes_ai_usage_exporter.py @@ -110,6 +110,10 @@ def test_claude_payload_uses_only_supported_quota_fields(): "utilization": 26.0, "resets_at": "2026-08-20T17:59:59+00:00", }, + "seven_day_fable": { + "utilization": 81.0, + "resets_at": "2026-08-20T17:59:59+00:00", + }, "internal_bucket": {"utilization": 99, "resets_at": secret}, "extra_usage": {"is_enabled": False, "used_credits": secret}, } @@ -118,6 +122,7 @@ def test_claude_payload_uses_only_supported_quota_fields(): assert {item.labels["window"] for item in samples if "window" in item.labels} == { "five_hour", "seven_day", + "seven_day_fable", } assert any( item.name == "atlas_ai_quota_remaining_percent" @@ -125,6 +130,12 @@ def test_claude_payload_uses_only_supported_quota_fields(): and item.value == 74 for item in samples ) + assert any( + item.name == "atlas_ai_quota_remaining_percent" + and item.labels["window"] == "seven_day_fable" + and item.value == 19 + for item in samples + ) assert secret not in repr(samples) @@ -263,13 +274,19 @@ def test_manifest_rolls_out_the_bounded_codex_deadline_and_poller_module(): environment = {item["name"]: item["value"] for item in exporter["env"]} assert annotations["ai.bstein.dev/config-rev"] == ( - "20260823-dual-provider-quota-health" + "20260824-claude-fable-quota" ) assert environment["ATLAS_AI_CODEX_QUERY_TIMEOUT_SECONDS"] == "45" assert environment["ATLAS_AI_AUTHENTICATION_GRACE_SECONDS"] == "1200" assert environment["ATLAS_AI_CLAUDE_OAUTH_TOKEN_FILE"] == ( "/claude-oauth-access/token" ) + assert environment["ATLAS_AI_CLAUDE_CREDENTIALS"] == ( + "/runtime-access/claude/.credentials.json" + ) + assert environment["ATLAS_AI_CLAUDE_USAGE_URL"] == ( + "https://api.anthropic.com/api/oauth/usage" + ) assert environment["ATLAS_AI_CLAUDE_QUOTA_MODEL"] == ( "claude-haiku-4-5-20251001" ) diff --git a/testing/tests/test_hermes_ai_usage_exporter_coverage.py b/testing/tests/test_hermes_ai_usage_exporter_coverage.py index 122b8877..5903bdaf 100644 --- a/testing/tests/test_hermes_ai_usage_exporter_coverage.py +++ b/testing/tests/test_hermes_ai_usage_exporter_coverage.py @@ -98,7 +98,7 @@ def test_claude_header_probe_rejects_malformed_optional_values( monkeypatch.setattr(mod.claude_query, "OAUTH_TOKEN_FILE", missing) with pytest.raises(RuntimeError, match="token is unavailable"): - mod.claude_query._token() + mod.claude_query._setup_token() assert mod.claude_query._percentage("invalid") is None assert mod.claude_query._percentage(None) is None @@ -107,7 +107,7 @@ def test_claude_header_probe_rejects_malformed_optional_values( 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( + assert mod.claude_query._header_payload( {"anthropic-ratelimit-unified-5h-utilization": "0.5"} ) == {"five_hour": {"utilization": 50}}