From dd522d0f77020482e90a2f0098cf28d632e73488 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:07:08 -0700 Subject: [PATCH] refactor(agent/creds): unify Anthropic credential I/O and rate-limit header helpers (-30% LOC) - anthropic_credentials: _load_json_if_exists / _claude_oauth_record / _atomic_write_private_json / _post_oauth_token / _oauth_token_state shared by the Claude Code + hermes_pkce read/write/refresh/exchange paths; _fingerprint delegates to credential_persistence.fingerprint_secret_value. - rate_limit_tracker: lower_headers / has_rate_limit_headers exported and reused by nous_rate_guard (header parsing, bucket exhaustion via _is_exhausted); format_remaining = _fmt_seconds alias (public name kept). - _safe_int stays tolerant of non-finite values (parity with origin). - Docstring/comment compaction; every WHY kept. - Parity: /tmp/rf/creds_c_parity_probe.py output identical to origin baseline (except wall-clock 'captured N ago' text); refresh_anthropic_oauth_pure old-vs-new on 6 token payloads identical. --- agent/anthropic_credentials.py | 997 +++++++++++---------------------- agent/nous_rate_guard.py | 230 ++------ agent/rate_limit_tracker.py | 146 ++--- 3 files changed, 425 insertions(+), 948 deletions(-) diff --git a/agent/anthropic_credentials.py b/agent/anthropic_credentials.py index 660a672211..38f9ca0a71 100644 --- a/agent/anthropic_credentials.py +++ b/agent/anthropic_credentials.py @@ -1,29 +1,23 @@ """Anthropic credential sources, OAuth flows, and token resolution. -Extracted from ``agent/anthropic_adapter.py``: the adapter is a message/HTTP -translation layer, while everything below owns *where an Anthropic credential -comes from* and *how a rotated one is committed*. Keeping the two apart means -the refresh transaction has a single home instead of being interleaved with -request building. - Sources, in the order ``resolve_anthropic_token()`` consults them: 1. ``ANTHROPIC_TOKEN`` / ``CLAUDE_CODE_OAUTH_TOKEN`` (explicit OAuth env) 2. ``ANTHROPIC_API_KEY`` (explicit API key) -3. ``~/.hermes/.anthropic_oauth.json`` (Hermes PKCE login) -4. ``~/.claude/.credentials.json`` / macOS Keychain (Claude Code) -5. the credential pool in ``auth.json`` +3. ``~/.claude/.credentials.json`` / macOS Keychain (Claude Code) +4. the credential pool in ``auth.json`` -Sources 3 and 4 are *singletons*: ``credential_pool._seed_from_singletons()`` -re-reads them on every ``load_pool()`` and writes what it finds over the pool -row, which is why a failed write here is a failed refresh (see -``CredentialPersistError``) rather than a best-effort cache miss. - -``agent.anthropic_adapter`` re-exports every public name below, so existing -``from agent.anthropic_adapter import resolve_anthropic_token`` imports keep -working. +``~/.hermes/.anthropic_oauth.json`` (Hermes PKCE login) and the Claude Code +file are *singletons*: ``credential_pool._seed_from_singletons()`` re-reads +them on every ``load_pool()``, so a failed write here is a failed refresh +(``CredentialPersistError``), not a cache miss. ``agent.anthropic_adapter`` +re-exports every public name below. """ +import base64 +import contextlib +import functools +import hashlib import json import logging import os @@ -32,6 +26,7 @@ import secrets import stat import subprocess import threading +import time from collections import OrderedDict from pathlib import Path from typing import Any, Dict, Optional @@ -41,61 +36,48 @@ from agent.secret_scope import get_secret as _get_secret logger = logging.getLogger(__name__) +_OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" +# platform.claude.com is the live token host; console.anthropic.com 404s but +# is kept as a fallback. _OAUTH_TOKEN_URL stays for backward-compatible imports. +_OAUTH_TOKEN_URLS = [ + "https://platform.claude.com/v1/oauth/token", + "https://console.anthropic.com/v1/oauth/token", +] +_OAUTH_TOKEN_URL = _OAUTH_TOKEN_URLS[0] +# Token-endpoint UA. Anthropic 429s any token-endpoint request whose UA starts +# with ``claude-code/`` (or Mozilla); the real CLI uses bare axios there. The +# inference path (build_anthropic_kwargs) still needs the claude-code/ UA. +_OAUTH_TOKEN_USER_AGENT = "axios/1.7.9" +_OAUTH_REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback" +_OAUTH_SCOPES = "org:create_api_key user:profile user:inference" + def _getenv(name: str, default: str = "") -> str: - """Profile-scoped replacement for os.getenv on credential reads. + """Profile-scoped os.getenv for credential reads (fail-closed on unscoped reads when multiplexing). - Routes through the secret scope (Workstream A): identical to os.getenv - when multiplexing is off, scope-aware (and fail-closed on an unscoped - read) when on. Mirrors the same wrapper in hermes_cli/runtime_provider.py. + Mirrors the same wrapper in hermes_cli/runtime_provider.py. """ val = _get_secret(name, default) return val if val is not None else default def _is_oauth_token(key: str) -> bool: - """Check if the key is an Anthropic OAuth/setup token. + """True for Anthropic OAuth/setup tokens; False for Console API keys and non-Anthropic keys. - Positively identifies Anthropic OAuth tokens by their key format: - - ``sk-ant-`` prefix (but NOT ``sk-ant-api``) → setup tokens, managed keys - - ``eyJ`` prefix → JWTs from the Anthropic OAuth flow - - ``cc-`` prefix → Claude Code OAuth access tokens (from CLAUDE_CODE_OAUTH_TOKEN) - - Non-Anthropic keys (MiniMax, Alibaba, etc.) don't match any pattern - and correctly return False. + sk-ant-api* = Console API key (x-api-key auth, never OAuth). sk-ant-* = setup + tokens / managed keys, eyJ = OAuth JWTs, cc- = CLAUDE_CODE_OAUTH_TOKEN access tokens. """ - if not key: + if not key or key.startswith("sk-ant-api"): return False - # Regular Anthropic Console API keys — x-api-key auth, never OAuth - if key.startswith("sk-ant-api"): - return False - # Anthropic-issued tokens (setup-tokens sk-ant-oat-*, managed keys) - if key.startswith("sk-ant-"): - return True - # JWTs from Anthropic OAuth flow - if key.startswith("eyJ"): - return True - # Claude Code OAuth access tokens (opaque, from CLAUDE_CODE_OAUTH_TOKEN) - if key.startswith("cc-"): - return True - return False - + return key.startswith(("sk-ant-", "eyJ", "cc-")) class CredentialPersistError(RuntimeError): """A rotated single-use credential could not be durably committed. - Anthropic OAuth refresh tokens are single-use: a successful refresh POST - consumes the old refresh token server-side and returns a replacement. The - replacement exists only in memory until it reaches its authoritative - on-disk store (``~/.claude/.credentials.json`` for ``claude_code``, - ``~/.hermes/.anthropic_oauth.json`` for ``hermes_pkce``). - - If that write fails and the caller reports success anyway, the on-disk - (already consumed) pair survives and is re-seeded on the next - ``load_pool()``, so the following refresh replays a spent token and fails - with ``invalid_grant`` / ``refresh_token_reused``. Callers must therefore - treat this as a failed refresh, not a successful one, and fail closed. + The refresh POST already spent the old refresh token, so a swallowed write + failure leaves a consumed pair on disk that later replays as invalid_grant. + Callers must treat this as a failed refresh. """ def __init__(self, path: Any, cause: BaseException) -> None: @@ -105,28 +87,11 @@ class CredentialPersistError(RuntimeError): self.path = path -# Fingerprints of Anthropic secrets whose refresh POST succeeded (so the -# server-side pair was rotated and the old refresh token is spent) but whose -# replacement never reached its authoritative store. The pre-rotation pair -# survives on disk and is re-seeded on the next ``load_pool()``, so without an -# explicit verdict the resolver happily hands that already-consumed credential -# back from a later source and the caller reads a silent success. -# -# Kept as non-reversible digests and bounded: a spent secret is spent forever, -# so entries never need clearing (a re-auth mints new tokens with new -# fingerprints). -# -# The registry has TWO scopes, because the credential it protects does: -# * process-local (this OrderedDict) — fast path, always recorded; -# * durable sidecar file next to the shared credential source — the -# authority boundary of ``claude_code``/``hermes_pkce`` is the shared -# singleton file, which other Hermes processes/profiles read with fresh -# interpreters. A process-local verdict only stops the process that -# lost the commit from lying to itself; the sidecar stops every OTHER -# process from leasing the stale pair or re-POSTing the spent refresh -# token. The sidecar stores only one-way fingerprints (never secrets) -# and is written under the same path-keyed cross-process lock that -# serializes refreshes of that source. +# ── Spent-rotation registry ───────────────────────────────────────────── +# Fingerprints of secrets whose refresh POST succeeded but whose replacement +# never reached its store. Two scopes: process-local (this OrderedDict) and a +# durable sidecar next to the shared singleton file so OTHER processes sharing +# it also fail closed. Non-reversible digests only; never cleared (spent is spent). _SPENT_ROTATION_LOCK = threading.Lock() _SPENT_ROTATION_FINGERPRINTS: "OrderedDict[str, None]" = OrderedDict() _SPENT_ROTATION_MAX_TRACKED = 64 @@ -134,31 +99,20 @@ _SPENT_ROTATION_SIDECAR_VERSION = 1 def _spent_rotation_sidecar_path(source_path: Path) -> Path: - """Sidecar registry path for a shared credential source file.""" return source_path.with_name(source_path.name + ".hermes-spent-rotations.json") def spent_rotation_source_path(source: Any) -> Optional[Path]: - """Map a pool-entry source to the shared singleton file it borrows from. - - Only singleton-backed sources have a cross-process authority boundary; - profile-owned rows are already protected by the process-local registry - plus the pool quarantine. - """ - if source == "claude_code": - return claude_code_credentials_path() - if source == "hermes_pkce": - return _get_hermes_oauth_file() - return None + """Map a pool-entry source to the shared singleton file it borrows from (or None).""" + getter = _SINGLETON_SOURCE_PATHS.get(source) if isinstance(source, str) else None + return getter() if getter else None def _read_spent_rotation_sidecar(source_path: Optional[Path]) -> set: if source_path is None: return set() try: - raw = json.loads( - _spent_rotation_sidecar_path(source_path).read_text(encoding="utf-8") - ) + raw = json.loads(_spent_rotation_sidecar_path(source_path).read_text(encoding="utf-8")) except (OSError, ValueError): return set() fingerprints = raw.get("fingerprints") if isinstance(raw, dict) else None @@ -168,12 +122,9 @@ def _read_spent_rotation_sidecar(source_path: Optional[Path]) -> set: def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> None: - """Merge fingerprints into the sidecar registry (atomic replace). + """Merge fingerprints into the sidecar (atomic replace; caller holds the path lock). - Callers on the refresh path already hold the path-keyed cross-process - lock for ``source_path``, so concurrent merge-writes are serialized. - Fail-soft: a sidecar write failure must never mask the fail-closed - verdict already recorded in the process-local registry. + Fail-soft: a sidecar write failure must never mask the process-local verdict. """ sidecar = _spent_rotation_sidecar_path(source_path) try: @@ -199,35 +150,25 @@ def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> Non tmp.write_text(payload, encoding="utf-8") os.replace(tmp, sidecar) except Exception: - logger.debug( - "Failed to persist spent-rotation fingerprints to %s", sidecar, - exc_info=True, - ) + logger.debug("Failed to persist spent-rotation fingerprints to %s", sidecar, exc_info=True) -def mark_rotation_consumed_uncommitted( - *secrets: Any, source_path: Optional[Path] = None -) -> None: - """Record secrets consumed by a refresh whose replacement never committed. - - Called from every commit-failure path (the direct resolver here and - ``CredentialPool._fail_closed_unpersisted_rotation``). Recording the - *pre-rotation* pair is what lets later resolution steps recognise the stale - copy they read back off disk as unusable rather than as a working token. - - When ``source_path`` names the shared singleton file the credential was - borrowed from, the verdict is additionally persisted to that source's - sidecar registry so other processes/profiles sharing the file adopt it too. - """ +def _fingerprint(secret: Any) -> Optional[str]: from agent.credential_persistence import fingerprint_secret_value + value = str(secret or "").strip() + return fingerprint_secret_value(value) if value else None + + +def mark_rotation_consumed_uncommitted(*secrets: Any, source_path: Optional[Path] = None) -> None: + """Record the pre-rotation pair of a refresh whose replacement never committed. + + With ``source_path`` the verdict is also persisted to that singleton's sidecar. + """ recorded: list = [] with _SPENT_ROTATION_LOCK: for secret in secrets: - value = str(secret or "").strip() - if not value: - continue - fingerprint = fingerprint_secret_value(value) + fingerprint = _fingerprint(secret) if not fingerprint: continue recorded.append(fingerprint) @@ -239,21 +180,9 @@ def mark_rotation_consumed_uncommitted( _append_spent_rotation_sidecar(source_path, recorded) -def is_rotation_consumed_uncommitted( - secret: Any, *, source_path: Optional[Path] = None -) -> bool: - """True when *secret* belongs to a rotation that was spent but not committed. - - Checks the process-local registry first, then (when ``source_path`` is - given) the durable sidecar registry of the shared credential source, so a - fresh interpreter in another process still sees the terminal verdict. - """ - from agent.credential_persistence import fingerprint_secret_value - - value = str(secret or "").strip() - if not value: - return False - fingerprint = fingerprint_secret_value(value) +def is_rotation_consumed_uncommitted(secret: Any, *, source_path: Optional[Path] = None) -> bool: + """True when *secret* belongs to a rotation that was spent but not committed.""" + fingerprint = _fingerprint(secret) if not fingerprint: return False with _SPENT_ROTATION_LOCK: @@ -262,90 +191,16 @@ def is_rotation_consumed_uncommitted( return fingerprint in _read_spent_rotation_sidecar(source_path) -def _read_claude_code_credentials_from_keychain() -> Optional[Dict[str, Any]]: - """Read Claude Code OAuth credentials from the macOS Keychain. - - Claude Code >=2.1.114 stores credentials in the macOS Keychain under the - service name "Claude Code-credentials" rather than (or in addition to) - the JSON file at ~/.claude/.credentials.json. - - The password field contains a JSON string with the same claudeAiOauth - structure as the JSON file. - - Returns dict with {accessToken, refreshToken?, expiresAt?} or None. - """ - if platform.system() != "Darwin": - return None - - try: - # Read the "Claude Code-credentials" generic password entry - result = subprocess.run( - ["security", "find-generic-password", - "-s", "Claude Code-credentials", - "-w"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=5, - stdin=subprocess.DEVNULL, - ) - except (OSError, subprocess.TimeoutExpired): - logger.debug("Keychain: security command not available or timed out") - return None - - if result.returncode != 0: - logger.debug("Keychain: no entry found for 'Claude Code-credentials'") - return None - - raw = result.stdout.strip() - if not raw: - return None - - try: - data = json.loads(raw) - except json.JSONDecodeError: - logger.debug("Keychain: credentials payload is not valid JSON") - return None - - oauth_data = data.get("claudeAiOauth") - if oauth_data and isinstance(oauth_data, dict): - access_token = oauth_data.get("accessToken", "") - if access_token: - return { - "accessToken": access_token, - "refreshToken": oauth_data.get("refreshToken", ""), - "expiresAt": oauth_data.get("expiresAt", 0), - "source": "macos_keychain", - } - - return None +# ── Claude Code credentials (Keychain / ~/.claude/.credentials.json) ──── +# Only singleton-backed pool sources have a cross-process authority boundary. +_SINGLETON_SOURCE_PATHS = { + "claude_code": lambda: claude_code_credentials_path(), + "hermes_pkce": lambda: _get_hermes_oauth_file(), +} -def claude_code_credentials_path() -> Path: - """Location Claude Code CLI writes its shared OAuth credentials file. - - This file is not profile-owned: every Hermes profile's credential pool - reads and writes the *same* path, so cross-profile refresh races on a - ``claude_code`` pool entry must be serialized against this exact path - (see ``CredentialPool._claude_code_credentials_lock`` in - ``agent/credential_pool.py``). - """ - return Path.home() / ".claude" / ".credentials.json" - - -def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]: - """Read Claude Code OAuth credentials from ~/.claude/.credentials.json. - - Returns dict with {accessToken, refreshToken?, expiresAt?, source} or None. - """ - cred_path = claude_code_credentials_path() - if not cred_path.exists(): - return None - try: - data = json.loads(cred_path.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError, IOError) as e: - logger.debug("Failed to read ~/.claude/.credentials.json: %s", e) - return None - +def _claude_oauth_record(data: Any, source: str) -> Optional[Dict[str, Any]]: + """Normalise a ``{"claudeAiOauth": {...}}`` payload into our credential dict.""" oauth_data = data.get("claudeAiOauth") if not (oauth_data and isinstance(oauth_data, dict)): return None @@ -356,192 +211,183 @@ def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]: "accessToken": access_token, "refreshToken": oauth_data.get("refreshToken", ""), "expiresAt": oauth_data.get("expiresAt", 0), - "source": "claude_code_credentials_file", + "source": source, } +def _read_claude_code_credentials_from_keychain() -> Optional[Dict[str, Any]]: + """Read the "Claude Code-credentials" macOS Keychain entry (Claude Code >=2.1.114).""" + if platform.system() != "Darwin": + return None + + try: + result = subprocess.run( + ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"], + capture_output=True, text=True, encoding='utf-8', errors='replace', + timeout=5, stdin=subprocess.DEVNULL, + ) + except (OSError, subprocess.TimeoutExpired): + logger.debug("Keychain: security command not available or timed out") + return None + if result.returncode != 0: + logger.debug("Keychain: no entry found for 'Claude Code-credentials'") + return None + raw = result.stdout.strip() + if not raw: + return None + try: + data = json.loads(raw) + except json.JSONDecodeError: + logger.debug("Keychain: credentials payload is not valid JSON") + return None + return _claude_oauth_record(data, "macos_keychain") + + +def claude_code_credentials_path() -> Path: + """Claude Code's shared OAuth file; every profile reads/writes this same path.""" + return Path.home() / ".claude" / ".credentials.json" + + +def _load_json_if_exists(path: Path, what: str) -> Optional[Any]: + """Parsed JSON from *path*, or None when missing/unreadable/corrupt (debug-logged).""" + if not path.exists(): + return None + try: + return json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as e: + logger.debug("Failed to read %s: %s", what, e) + return None + + +def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]: + data = _load_json_if_exists(claude_code_credentials_path(), "~/.claude/.credentials.json") + return _claude_oauth_record(data, "claude_code_credentials_file") if data is not None else None + + def read_claude_code_credentials() -> Optional[Dict[str, Any]]: - """Read refreshable Claude Code OAuth credentials. + """Read refreshable Claude Code OAuth credentials (Keychain and/or file). - Reads from two possible sources and reconciles them: - 1. macOS Keychain (Darwin only) — "Claude Code-credentials" entry - 2. ~/.claude/.credentials.json file - - Selection rules when both are present: - - If exactly one is non-expired, prefer that one. (Handles the case - where Claude Code refreshes one source but not the other — observed - in the wild on Claude Code 2.1.x.) - - Otherwise, prefer the source with the later ``expiresAt`` so that - any subsequent refresh uses the most recent ``refreshToken``. - - This intentionally excludes ~/.claude.json primaryApiKey. Opencode's - subscription flow is OAuth/setup-token based with refreshable credentials, - and native direct Anthropic provider usage should follow that path rather - than auto-detecting Claude's first-party managed key. - - Returns dict with {accessToken, refreshToken?, expiresAt?, source} or None. + When both exist: prefer the only non-expired one (Claude Code 2.1.x refreshes + one source but not the other), else the later ``expiresAt`` so a refresh uses + the freshest refreshToken. ~/.claude.json primaryApiKey is deliberately excluded. """ kc_creds = _read_claude_code_credentials_from_keychain() file_creds = _read_claude_code_credentials_from_file() - if kc_creds and file_creds: - kc_valid = is_claude_code_token_valid(kc_creds) - file_valid = is_claude_code_token_valid(file_creds) - if kc_valid and not file_valid: - return kc_creds - if file_valid and not kc_valid: - return file_creds - # Both valid or both expired: prefer the later expiresAt so the - # downstream refresh path uses the freshest refresh_token. - kc_exp = kc_creds.get("expiresAt", 0) or 0 - file_exp = file_creds.get("expiresAt", 0) or 0 - return kc_creds if kc_exp >= file_exp else file_creds - - return kc_creds or file_creds + if not (kc_creds and file_creds): + return kc_creds or file_creds + kc_valid = is_claude_code_token_valid(kc_creds) + file_valid = is_claude_code_token_valid(file_creds) + if kc_valid != file_valid: + return kc_creds if kc_valid else file_creds + kc_exp = kc_creds.get("expiresAt", 0) or 0 + file_exp = file_creds.get("expiresAt", 0) or 0 + return kc_creds if kc_exp >= file_exp else file_creds def is_claude_code_token_valid(creds: Dict[str, Any]) -> bool: - """Check if Claude Code credentials have a non-expired access token.""" - import time - + """Non-expired access token (60s buffer); no expiresAt means managed key → valid if present.""" expires_at = creds.get("expiresAt", 0) if not expires_at: - # No expiry set (managed keys) — valid if token is present return bool(creds.get("accessToken")) + return int(time.time() * 1000) < (expires_at - 60_000) - # expiresAt is in milliseconds since epoch - now_ms = int(time.time() * 1000) - # Allow 60 seconds of buffer - return now_ms < (expires_at - 60_000) + +# ── OAuth token endpoint ──────────────────────────────────────────────── + + +def _post_oauth_token( + data: bytes, *, content_type: str, timeout: int, what: str, user_agent: str = _OAUTH_TOKEN_USER_AGENT +) -> Dict[str, Any]: + """POST to the token endpoints in order; raise the last error if all fail.""" + import urllib.request + + last_error = None + for endpoint in _OAUTH_TOKEN_URLS: + req = urllib.request.Request( + endpoint, data=data, method="POST", + headers={"Content-Type": content_type, "User-Agent": user_agent}, + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + except Exception as exc: + last_error = exc + logger.debug("Anthropic token %s failed at %s: %s", what, endpoint, exc) + if last_error is not None: + raise last_error + raise ValueError(f"Anthropic token {what} failed") + + +def _oauth_token_state(result: Dict[str, Any], *, fallback_refresh_token: str = "") -> Dict[str, Any]: + """Token-endpoint JSON -> ``{access_token, refresh_token, expires_at_ms}`` (expires_in defaults to 3600s).""" + expires_in = result.get("expires_in", 3600) + return { + "access_token": result.get("access_token", ""), + "refresh_token": result.get("refresh_token", fallback_refresh_token), + "expires_at_ms": int(time.time() * 1000) + (expires_in * 1000), + } def refresh_anthropic_oauth_pure(refresh_token: str, *, use_json: bool = False) -> Dict[str, Any]: """Refresh an Anthropic OAuth token without mutating local credential files.""" - import time import urllib.parse - import urllib.request if not refresh_token: raise ValueError("refresh_token is required") - client_id = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" + payload = { + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": _OAUTH_CLIENT_ID, + } if use_json: - data = json.dumps({ - "grant_type": "refresh_token", - "refresh_token": refresh_token, - "client_id": client_id, - }).encode() - content_type = "application/json" + data, content_type = json.dumps(payload).encode(), "application/json" else: - data = urllib.parse.urlencode({ - "grant_type": "refresh_token", - "refresh_token": refresh_token, - "client_id": client_id, - }).encode() - content_type = "application/x-www-form-urlencoded" - - token_endpoints = [ - "https://platform.claude.com/v1/oauth/token", - "https://console.anthropic.com/v1/oauth/token", - ] - last_error = None - for endpoint in token_endpoints: - req = urllib.request.Request( - endpoint, - data=data, - headers={ - "Content-Type": content_type, - "User-Agent": _OAUTH_TOKEN_USER_AGENT, - }, - method="POST", - ) - try: - with urllib.request.urlopen(req, timeout=10) as resp: - result = json.loads(resp.read().decode()) - except Exception as exc: - last_error = exc - logger.debug("Anthropic token refresh failed at %s: %s", endpoint, exc) - continue - - access_token = result.get("access_token", "") - if not access_token: - raise ValueError("Anthropic refresh response was missing access_token") - next_refresh = result.get("refresh_token", refresh_token) - expires_in = result.get("expires_in", 3600) - return { - "access_token": access_token, - "refresh_token": next_refresh, - "expires_at_ms": int(time.time() * 1000) + (expires_in * 1000), - } - - if last_error is not None: - raise last_error - raise ValueError("Anthropic token refresh failed") + data, content_type = urllib.parse.urlencode(payload).encode(), "application/x-www-form-urlencoded" + result = _post_oauth_token( + data, content_type=content_type, timeout=10, what="refresh", user_agent=_OAUTH_TOKEN_USER_AGENT + ) + if not result.get("access_token"): + raise ValueError("Anthropic refresh response was missing access_token") + return _oauth_token_state(result, fallback_refresh_token=refresh_token) def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: - """Attempt to refresh an expired Claude Code OAuth token. + """Refresh an expired Claude Code OAuth token, returning the new access token. - Claude Code's OAuth refresh tokens are single-use: a successful refresh - rotates the pair and invalidates the old refresh token. Claude Code itself - also refreshes on its own schedule (IDE/CLI activity), so by the time - Hermes notices an expired token, Claude Code may have already rotated it. - POSTing our now-stale refresh token in that window races Claude Code and - fails with ``invalid_grant``. - - So before refreshing, re-read the live credential sources. If Claude Code - has already produced a valid token, adopt it and skip the POST entirely. - Only fall back to refreshing ourselves when no fresh credential is found. + Refresh tokens are single-use and Claude Code refreshes on its own schedule, + so we first re-read the live sources and adopt an already-rotated token + instead of racing it into ``invalid_grant``. Read, decision, POST and + write-back all run under the same path-keyed cross-process lock the pool + uses; without it two profiles can still spend one refresh token. """ - # Claude Code may have already refreshed — adopt its token rather than - # racing it with our (possibly already-rotated) refresh token. The read, - # decision, POST, and write-back all belong to the shared credentials - # source, so hold the same path-keyed cross-process lock used by the pool. - # Without this direct resolver path, two profiles can still spend one - # single-use refresh token even though CredentialPool is serialized. try: from hermes_cli.auth import AUTH_LOCK_TIMEOUT_SECONDS, _auth_store_lock, env_float - refresh_timeout_seconds = env_float( - "HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 20 - ) - lock_timeout_seconds = max( - float(AUTH_LOCK_TIMEOUT_SECONDS), - float(refresh_timeout_seconds) + 5.0, - ) - with _auth_store_lock( - timeout_seconds=lock_timeout_seconds, - target_path=claude_code_credentials_path(), - ): - # Only adopt when the live re-read produced a DIFFERENT token with - # a real future expiry: re-adopting the same credential we were - # just handed would be a no-op, and a 0/absent ``expiresAt`` means - # "managed key / unknown expiry" (see is_claude_code_token_valid). - current = read_claude_code_credentials() - if current: - current_token = current.get("accessToken", "") - current_exp = current.get("expiresAt", 0) or 0 - if ( - current_token - and current_token != creds.get("accessToken", "") - and current_exp > 0 - and is_claude_code_token_valid(current) - ): - logger.debug("Adopted Claude Code's already-refreshed OAuth token") - return current_token + refresh_timeout_seconds = env_float("HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 20) + lock_timeout_seconds = max(float(AUTH_LOCK_TIMEOUT_SECONDS), float(refresh_timeout_seconds) + 5.0) + with _auth_store_lock(timeout_seconds=lock_timeout_seconds, target_path=claude_code_credentials_path()): + # Adopt only a DIFFERENT token with a real future expiry (0/absent + # expiresAt means managed key / unknown expiry). + current = read_claude_code_credentials() or {} + current_token = current.get("accessToken", "") + if ( + current_token + and current_token != creds.get("accessToken", "") + and (current.get("expiresAt", 0) or 0) > 0 + and is_claude_code_token_valid(current) + ): + logger.debug("Adopted Claude Code's already-refreshed OAuth token") + return current_token - refresh_token = ( - (current or {}).get("refreshToken", "") - or creds.get("refreshToken", "") - ) + refresh_token = current.get("refreshToken", "") or creds.get("refreshToken", "") if not refresh_token: logger.debug("No refresh token available — cannot refresh") return None - # Another process may have spent this refresh token and lost the - # commit; its durable sidecar verdict is authoritative for the - # shared source. POSTing it again would just burn the family into - # ``invalid_grant``. + # Another process may have spent this token and lost the commit; + # its sidecar verdict is authoritative for the shared source. if is_rotation_consumed_uncommitted( refresh_token, source_path=claude_code_credentials_path() ): @@ -557,13 +403,8 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: logger.debug("Failed to refresh Claude Code token: %s", e) return None - # The POST above already consumed ``refresh_token`` server-side. - # Writing the replacement pair is the commit step of that - # transaction, not a cache update: if it fails, the rotation is - # unrecoverable and the pair still on disk is spent. Fail closed - # rather than handing back an access token whose refresh half was - # lost — reporting success here is what lets a later load replay - # the consumed token and produce ``invalid_grant``. + # The POST spent ``refresh_token``; this write is the commit step. + # On failure, fail closed and mark the pre-rotation pair as spent. try: _write_claude_code_credentials( refreshed["access_token"], @@ -578,15 +419,11 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: claude_code_credentials_path(), e, ) - # The POST already spent ``refresh_token`` server-side and the - # replacement is gone. The pre-rotation pair is still on disk, - # so mark it: without this, source 5 re-reads it through the - # pool and returns the consumed credential as a success. mark_rotation_consumed_uncommitted( refresh_token, creds.get("accessToken", ""), - (current or {}).get("accessToken", ""), - (current or {}).get("refreshToken", ""), + current.get("accessToken", ""), + current.get("refreshToken", ""), source_path=claude_code_credentials_path(), ) return None @@ -594,12 +431,34 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: logger.debug("Successfully refreshed Claude Code OAuth token") return refreshed["access_token"] except Exception as e: - # Lock acquisition/read failures should preserve the resolver's - # existing fail-soft contract rather than taking down agent startup. + # Lock/read failures keep the resolver's fail-soft contract. logger.debug("Failed to acquire Claude Code refresh lock: %s", e) return None +def _atomic_write_private_json(path: Path, payload: Any) -> None: + """Write *payload* to *path* via a 0o600 temp file + fsync + os.replace. + + The temp file is created O_EXCL at 0o600 so the token is never briefly + umask-readable (write_text + chmod had a TOCTOU window); the per-process + random suffix avoids collisions with concurrent writers and crashed leftovers. + The parent dir's mode is left alone (~/.claude/ is owned by Claude Code). + """ + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") + try: + fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp, path) + except OSError: + with contextlib.suppress(OSError): + tmp.unlink(missing_ok=True) + raise + + def _write_claude_code_credentials( access_token: str, refresh_token: str, @@ -607,25 +466,15 @@ def _write_claude_code_credentials( *, scopes: Optional[list] = None, ) -> None: - """Write refreshed credentials back to ~/.claude/.credentials.json. + """Commit refreshed credentials to ~/.claude/.credentials.json. - The optional *scopes* list (e.g. ``["user:inference", "user:profile", ...]``) - is persisted so that Claude Code's own auth check recognises the credential - as valid. Claude Code >=2.1.81 gates on the presence of ``"user:inference"`` - in the stored scopes before it will use the token. - - Raises ``CredentialPersistError`` when the rotated pair does not reach the - file. This write is the commit step of the refresh transaction, not a - best-effort cache update: a swallowed failure leaves the consumed - pre-rotation pair on disk to be re-seeded and replayed (see - ``CredentialPersistError``). + *scopes* (or the previously stored scopes) are persisted because Claude Code + >=2.1.81 gates on ``"user:inference"`` being present. Raises + ``CredentialPersistError`` on any failure, including a corrupt existing file. """ cred_path = claude_code_credentials_path() try: - # Read existing file to preserve other fields - existing = {} - if cred_path.exists(): - existing = json.loads(cred_path.read_text(encoding="utf-8")) + existing = json.loads(cred_path.read_text(encoding="utf-8")) if cred_path.exists() else {} oauth_data: Dict[str, Any] = { "accessToken": access_token, @@ -635,59 +484,25 @@ def _write_claude_code_credentials( if scopes is not None: oauth_data["scopes"] = scopes elif "claudeAiOauth" in existing and "scopes" in existing["claudeAiOauth"]: - # Preserve previously-stored scopes when the refresh response - # does not include a scope field. oauth_data["scopes"] = existing["claudeAiOauth"]["scopes"] existing["claudeAiOauth"] = oauth_data - - cred_path.parent.mkdir(parents=True, exist_ok=True) - # Per-process random suffix avoids collisions between concurrent - # writers and stale leftovers from a prior crashed write. - _tmp_cred = cred_path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") - try: - # Create the temp file atomically at 0o600. The previous - # write_text + post-replace chmod opened a TOCTOU window where - # both the temp file and the destination briefly inherited the - # process umask (commonly 0o644 = world-readable), exposing - # Claude Code OAuth tokens to other local users between create - # and chmod. Mirrors agent/google_oauth.py (#19673) and - # tools/mcp_oauth.py (#21148). Parent dir (~/.claude/) is - # owned by Claude Code itself, so we leave its mode alone. - fd = os.open( - str(_tmp_cred), - os.O_WRONLY | os.O_CREAT | os.O_EXCL, - stat.S_IRUSR | stat.S_IWUSR, - ) - with os.fdopen(fd, "w", encoding="utf-8") as fh: - json.dump(existing, fh, indent=2) - fh.flush() - os.fsync(fh.fileno()) - os.replace(_tmp_cred, cred_path) - except OSError: - try: - _tmp_cred.unlink(missing_ok=True) - except OSError: - pass - raise - except (OSError, IOError, ValueError) as e: - # ValueError covers a corrupt existing file (JSONDecodeError): the - # merge-read is part of the commit, so failing it means the rotated - # pair never landed either. + _atomic_write_private_json(cred_path, existing) + except (OSError, ValueError) as e: logger.error("Failed to write refreshed credentials to %s: %s", cred_path, e) raise CredentialPersistError(cred_path, e) from e +# ── Resolution ────────────────────────────────────────────────────────── + + def _resolve_claude_code_token_from_credentials(creds: Optional[Dict[str, Any]] = None) -> Optional[str]: """Resolve a token from Claude Code credential files, refreshing if needed.""" creds = creds or read_claude_code_credentials() if creds and is_rotation_consumed_uncommitted( creds.get("accessToken", ""), source_path=claude_code_credentials_path() ): - # This process already rotated this pair and failed to commit the - # replacement. The file still holds the spent copy; treating it as - # usable is exactly the silent success this transaction fails closed - # to prevent. + # The file still holds the spent pre-rotation copy of a failed commit. logger.debug( "Claude Code credentials hold a rotated-but-uncommitted token - refusing" ) @@ -705,16 +520,13 @@ def _resolve_claude_code_token_from_credentials(creds: Optional[Dict[str, Any]] def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[str, Any]]) -> Optional[str]: - """Prefer Claude Code creds when a persisted env OAuth token would shadow refresh. + """Prefer refreshable Claude Code creds over a static env OAuth token. - Hermes historically persisted setup tokens into ANTHROPIC_TOKEN. That makes - later refresh impossible because the static env token wins before we ever - inspect Claude Code's refreshable credential file. If we have a refreshable - Claude Code credential record, prefer it over the static env OAuth token. + Hermes historically persisted setup tokens into ANTHROPIC_TOKEN; that static + token wins before the refreshable credential file is ever inspected, making + refresh impossible. """ - if not env_token or not _is_oauth_token(env_token) or not isinstance(creds, dict): - return None - if not creds.get("refreshToken"): + if not (env_token and _is_oauth_token(env_token) and isinstance(creds, dict) and creds.get("refreshToken")): return None resolved = _resolve_claude_code_token_from_credentials(creds) @@ -727,13 +539,12 @@ def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[s def _resolve_anthropic_pool_token() -> Optional[str]: - """Return the first available Anthropic OAuth token from credential_pool. + """First available Anthropic OAuth token from credential_pool, read-only. - Read-only: enumerates with ``clear_expired=False, refresh=False`` so a bare - token *resolve* (which runs from diagnostic/read-only call sites such as - ``account_usage`` and ``hermes models``) never mutates ``~/.hermes/auth.json`` - or makes a network refresh call. Refresh-on-expiry is owned by the API call - path's pool recovery, not the resolver. + Enumerates with ``clear_expired=False, refresh=False`` (never ``select()``, + which does both) so diagnostic call sites (account_usage, ``hermes models``) + never mutate auth.json or hit the network; refresh-on-expiry belongs to the + API call path's pool recovery. """ try: from agent.credential_pool import AUTH_TYPE_OAUTH, load_pool @@ -742,39 +553,23 @@ def _resolve_anthropic_pool_token() -> Optional[str]: try: pool = load_pool("anthropic") - # Enumerate read-only (clear_expired=False, refresh=False): never persist - # to auth.json or trigger a network refresh from a bare resolve. select() - # is deliberately NOT used — it runs clear_expired=True, refresh=True, - # which would violate this read-only contract. entries, _pending = pool._available_entries(clear_expired=False, refresh=False) except Exception: logger.debug("Failed to read Anthropic credential_pool", exc_info=True) return None for entry in entries: - if getattr(entry, "auth_type", None) != AUTH_TYPE_OAUTH: - continue - # access_token is a declared field but a persisted entry can carry an - # explicit null (or a partially-written OAuth entry), so coerce before - # strip — a bare None.strip() here would escape the try/excepts above - # and crash the whole resolver, taking down the source #5 fallback too. - # Matches the aux-client analog (auxiliary_client.py: str(key or "")). + # access_token may be an explicit null on a persisted entry; a bare + # None.strip() would escape the try/excepts and crash the resolver. token = (getattr(entry, "access_token", None) or "").strip() - if not token: + if getattr(entry, "auth_type", None) != AUTH_TYPE_OAUTH or not token: continue - # ``load_pool()`` re-seeds pool rows from the singleton files, so a - # rotation that was consumed upstream but never committed comes back - # here looking healthy. Enumeration is deliberately read-only - # (refresh=False), which means nothing on this path would otherwise - # notice that the credential is spent. Singleton-backed sources also - # consult the durable sidecar registry: the failed commit may have - # happened in a DIFFERENT process, whose process-local verdict this - # interpreter never saw. + # load_pool() re-seeds rows from the singleton files, so a spent-but- + # uncommitted rotation (possibly from another process) looks healthy here. entry_source_path = spent_rotation_source_path(getattr(entry, "source", None)) - if is_rotation_consumed_uncommitted( - token, source_path=entry_source_path - ) or is_rotation_consumed_uncommitted( - getattr(entry, "refresh_token", None), source_path=entry_source_path + if any( + is_rotation_consumed_uncommitted(secret, source_path=entry_source_path) + for secret in (token, getattr(entry, "refresh_token", None)) ): logger.debug( "Skipping Anthropic pool entry %s: rotated-but-uncommitted credential", @@ -787,75 +582,33 @@ def _resolve_anthropic_pool_token() -> Optional[str]: def resolve_anthropic_token() -> Optional[str]: - """Resolve an Anthropic token from all available sources. + """Resolve an Anthropic token from all sources in priority order (see module docstring).""" + # Read the Claude Code file at most once per resolve. + _read_creds = functools.cache(read_claude_code_credentials) - Priority: - 1. ANTHROPIC_TOKEN env var (OAuth/setup token saved by Hermes) - 2. CLAUDE_CODE_OAUTH_TOKEN env var - 3. ANTHROPIC_API_KEY env var (explicit regular API key) - 4. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json) - — with automatic refresh if expired and a refresh token is available - 5. Anthropic credential_pool OAuth entry (~/.hermes/auth.json) + for env_var in ("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"): + token = _getenv(env_var).strip() + if token: + return _prefer_refreshable_claude_code_token(token, _read_creds()) or token - Returns the token string or None. - """ - creds: Optional[Dict[str, Any]] = None - creds_loaded = False - - def _read_creds() -> Optional[Dict[str, Any]]: - nonlocal creds, creds_loaded - if not creds_loaded: - creds = read_claude_code_credentials() - creds_loaded = True - return creds - - # 1. Hermes-managed OAuth/setup token env var - token = _getenv("ANTHROPIC_TOKEN").strip() - if token: - preferred = _prefer_refreshable_claude_code_token(token, _read_creds()) - if preferred: - return preferred - return token - - # 2. CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens) - cc_token = _getenv("CLAUDE_CODE_OAUTH_TOKEN").strip() - if cc_token: - preferred = _prefer_refreshable_claude_code_token(cc_token, _read_creds()) - if preferred: - return preferred - return cc_token - - # 3. Regular API key. An explicit user-configured key must not be shadowed - # by auto-discovered Claude Code or credential-pool OAuth credentials. + # An explicit API key must not be shadowed by auto-discovered OAuth creds. api_key = _getenv("ANTHROPIC_API_KEY").strip() if api_key: return api_key - # 4. Claude Code credential file - resolved_claude_token = _resolve_claude_code_token_from_credentials(_read_creds()) - if resolved_claude_token: - return resolved_claude_token - - # 5. Hermes credential_pool OAuth entry. - resolved_pool_token = _resolve_anthropic_pool_token() - if resolved_pool_token: - return resolved_pool_token - - return None + return ( + _resolve_claude_code_token_from_credentials(_read_creds()) + or _resolve_anthropic_pool_token() + or None + ) def run_oauth_setup_token() -> Optional[str]: - """Run 'claude setup-token' interactively and return the resulting token. + """Run 'claude setup-token' interactively; return the resulting token or None. - Checks multiple sources after the subprocess completes: - 1. Claude Code credential files (may be written by the subprocess) - 2. CLAUDE_CODE_OAUTH_TOKEN / ANTHROPIC_TOKEN env vars - - Returns the token string, or None if no credentials were obtained. Raises FileNotFoundError if the 'claude' CLI is not installed. """ import shutil - import subprocess claude_path = shutil.which("claude") if not claude_path: @@ -864,64 +617,35 @@ def run_oauth_setup_token() -> Optional[str]: "Install it with: npm install -g @anthropic-ai/claude-code" ) - # Run interactively — stdin/stdout/stderr inherited so the user can - # complete the OAuth login prompt. Must keep inherited stdin; the TUI-EOF - # concern does not apply to an interactive login the user explicitly - # invokes. noqa: subprocess-stdin + # Interactive: stdin/stdout/stderr inherited so the user can complete the + # OAuth prompt; the TUI-EOF stdin concern does not apply here. noqa: subprocess-stdin try: subprocess.run([claude_path, "setup-token"]) except (KeyboardInterrupt, EOFError): return None - # Check if credentials were saved to Claude Code's config files creds = read_claude_code_credentials() if creds and is_claude_code_token_valid(creds): return creds["accessToken"] - - # Check env vars that may have been set for env_var in ("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN"): val = _getenv(env_var).strip() if val: return val - return None -# ── Hermes-native PKCE OAuth flow ──────────────────────────────────────── +# ── Hermes-native PKCE OAuth flow (~/.hermes/.anthropic_oauth.json) ───── # Mirrors the flow used by Claude Code, pi-ai, and OpenCode. -# Stores credentials in ~/.hermes/.anthropic_oauth.json (our own file). -_OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" -# Anthropic migrated the OAuth token endpoint to platform.claude.com; -# console.anthropic.com now 404s. Callers should iterate _OAUTH_TOKEN_URLS -# (new host first, console fallback). _OAUTH_TOKEN_URL is kept as the primary -# for backward compatibility with existing imports and now points at the live host. -_OAUTH_TOKEN_URLS = [ - "https://platform.claude.com/v1/oauth/token", - "https://console.anthropic.com/v1/oauth/token", -] -_OAUTH_TOKEN_URL = _OAUTH_TOKEN_URLS[0] -# User-Agent sent on the OAuth *token endpoint* (login exchange + refresh). -# Anthropic rate-limits (HTTP 429) any token-endpoint request whose UA starts -# with ``claude-code/`` — verified empirically against platform.claude.com: -# ``claude-code/2.1.200`` and ``Mozilla/5.0`` -> 429; ``axios/*``, ``node``, -# and SDK-style UAs -> 400 (reached code validation). The real Claude Code CLI -# exchanges the auth code with a bare axios client (``axios/``), NOT its -# ``claude-code/`` inference UA. We mirror that here. NOTE: the *inference* path -# (build_anthropic_kwargs) still uses the ``claude-code/`` UA + ``x-app: cli`` — -# that fingerprint is required there and is NOT throttled on the messages API. -_OAUTH_TOKEN_USER_AGENT = "axios/1.7.9" -_OAUTH_REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback" -_OAUTH_SCOPES = "org:create_api_key user:profile user:inference" + def _get_hermes_oauth_file() -> Path: return get_hermes_home() / ".anthropic_oauth.json" def _root_hermes_oauth_file() -> Optional[Path]: - """Global-root ``.anthropic_oauth.json`` when running inside a named profile. + """Global-root ``.anthropic_oauth.json`` inside a named profile; None in classic mode. - ``None`` in classic mode (profile == root). Used to commit a rotation of a - grant the profile borrowed through the credential-pool root fallback. + Used to commit a rotation of a grant the profile borrowed via the pool's root fallback. """ try: from hermes_constants import get_default_hermes_root @@ -935,10 +659,6 @@ def _root_hermes_oauth_file() -> Optional[Path]: def _generate_pkce() -> tuple: """Generate PKCE code_verifier and code_challenge (S256).""" - import base64 - import hashlib - import secrets - verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode() challenge = base64.urlsafe_b64encode( hashlib.sha256(verifier.encode()).digest() @@ -948,9 +668,8 @@ def _generate_pkce() -> tuple: def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: """Run Hermes-native OAuth PKCE flow and return credential state.""" - import secrets - import time import webbrowser + from urllib.parse import urlencode verifier, challenge = _generate_pkce() oauth_state = secrets.token_urlsafe(32) @@ -965,20 +684,20 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: "code_challenge_method": "S256", "state": oauth_state, } - from urllib.parse import urlencode - auth_url = f"https://claude.ai/oauth/authorize?{urlencode(params)}" - print() - print("Authorize Hermes with your Claude Pro/Max subscription.") - print() - print("╭─ Claude Pro/Max Authorization ────────────────────╮") - print("│ │") - print("│ Open this link in your browser: │") - print("╰───────────────────────────────────────────────────╯") - print() - print(f" {auth_url}") - print() + print("\n".join([ + "", + "Authorize Hermes with your Claude Pro/Max subscription.", + "", + "╭─ Claude Pro/Max Authorization ────────────────────╮", + "│ │", + "│ Open this link in your browser: │", + "╰───────────────────────────────────────────────────╯", + "", + f" {auth_url}", + "", + ])) try: from hermes_cli.auth import _can_open_graphical_browser as _can_open_gui @@ -986,15 +705,11 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: _can_open_gui = lambda: True # noqa: E731 — degrade to prior behavior if _can_open_gui(): - try: + with contextlib.suppress(Exception): webbrowser.open(auth_url) print(" (Browser opened automatically)") - except Exception: - pass - print() - print("After authorizing, you'll see a code. Paste it below.") - print() + print("\nAfter authorizing, you'll see a code. Paste it below.\n") try: auth_code = input("Authorization code: ").strip() except (KeyboardInterrupt, EOFError): @@ -1005,17 +720,14 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: return None splits = auth_code.split("#") - code = splits[0] - received_state = splits[1] if len(splits) > 1 else "" + code, received_state = splits[0], (splits[1] if len(splits) > 1 else "") - # Validate state to prevent CSRF (RFC 6749 §10.12) + # CSRF guard (RFC 6749 §10.12) if received_state != oauth_state: logger.warning("OAuth state mismatch — possible CSRF, aborting") return None try: - import urllib.request - exchange_data = json.dumps({ "grant_type": "authorization_code", "client_id": _OAUTH_CLIENT_ID, @@ -1024,69 +736,21 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: "redirect_uri": _OAUTH_REDIRECT_URI, "code_verifier": verifier, }).encode() - - # Anthropic migrated the OAuth token endpoint to platform.claude.com; - # console.anthropic.com now 404s. Try the new host first, then fall - # back to console for older deployments (mirrors the refresh path). - # UA is _OAUTH_TOKEN_USER_AGENT (a non-claude-code UA) — see the - # constant's definition for why the token endpoint must not send - # claude-code/ (429 UA-prefix block). - result = None - last_error = None - for endpoint in _OAUTH_TOKEN_URLS: - req = urllib.request.Request( - endpoint, - data=exchange_data, - headers={ - "Content-Type": "application/json", - "User-Agent": _OAUTH_TOKEN_USER_AGENT, - }, - method="POST", - ) - try: - with urllib.request.urlopen(req, timeout=15) as resp: - result = json.loads(resp.read().decode()) - break - except Exception as exc: - last_error = exc - logger.debug("Anthropic token exchange failed at %s: %s", endpoint, exc) - continue - - if result is None: - raise last_error if last_error is not None else ValueError( - "Anthropic token exchange failed" - ) + result = _post_oauth_token(exchange_data, content_type="application/json", timeout=15, what="exchange") except Exception as e: print(f"Token exchange failed: {e}") return None - access_token = result.get("access_token", "") - refresh_token = result.get("refresh_token", "") - expires_in = result.get("expires_in", 3600) - - if not access_token: + if not result.get("access_token"): print("No access token in response.") return None - - expires_at_ms = int(time.time() * 1000) + (expires_in * 1000) - return { - "access_token": access_token, - "refresh_token": refresh_token, - "expires_at_ms": expires_at_ms, - } + return _oauth_token_state(result) def read_hermes_oauth_credentials() -> Optional[Dict[str, Any]]: """Read Hermes-managed OAuth credentials from ~/.hermes/.anthropic_oauth.json.""" - oauth_file = _get_hermes_oauth_file() - if oauth_file.exists(): - try: - data = json.loads(oauth_file.read_text(encoding="utf-8")) - if data.get("accessToken"): - return data - except (json.JSONDecodeError, OSError, IOError) as e: - logger.debug("Failed to read Hermes OAuth credentials: %s", e) - return None + data = _load_json_if_exists(_get_hermes_oauth_file(), "Hermes OAuth credentials") + return data if data is not None and data.get("accessToken") else None def _write_hermes_oauth_credentials( @@ -1096,52 +760,23 @@ def _write_hermes_oauth_credentials( *, target: Optional[Path] = None, ) -> None: - """Write refreshed hermes_pkce tokens back to ~/.hermes/.anthropic_oauth.json. + """Commit refreshed hermes_pkce tokens to ~/.hermes/.anthropic_oauth.json. - ``target`` overrides the destination: a named profile that rotated a grant - it BORROWED from the global root (credential-pool root fallback) must - commit the new pair to the ROOT singleton, not create a forked copy under - its own HERMES_HOME (#100339). - - Without this, a successful pool-level refresh of a ``hermes_pkce``-sourced - entry is invisible to this singleton file. The next ``load_pool()`` call - runs ``_seed_from_singletons()``, which reads the stale file and - overwrites the freshly-rotated pool entry with the pre-refresh (and, for - single-use Anthropic refresh tokens, already-consumed) token pair. - - Raises ``CredentialPersistError`` when the rotated pair does not reach the - file, for the same reason ``_write_claude_code_credentials`` does: this is - the commit step of the refresh transaction. + ``target`` lets a named profile commit a grant it BORROWED from the global + root back to the ROOT singleton instead of forking a copy under its own + HERMES_HOME. Without this write-through, the next ``load_pool()`` re-seeds + the stale (already-consumed) pair from the file over the rotated pool entry. + Raises ``CredentialPersistError`` when the rotated pair does not land. """ oauth_file = target if target is not None else _get_hermes_oauth_file() try: - oauth_data = { + _atomic_write_private_json(oauth_file, { "accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms, - } - oauth_file.parent.mkdir(parents=True, exist_ok=True) - _tmp_oauth = oauth_file.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") - try: - fd = os.open( - str(_tmp_oauth), - os.O_WRONLY | os.O_CREAT | os.O_EXCL, - stat.S_IRUSR | stat.S_IWUSR, - ) - with os.fdopen(fd, "w", encoding="utf-8") as fh: - json.dump(oauth_data, fh, indent=2) - fh.flush() - os.fsync(fh.fileno()) - os.replace(_tmp_oauth, oauth_file) - except OSError: - try: - _tmp_oauth.unlink(missing_ok=True) - except OSError: - pass - raise - except (OSError, IOError, ValueError) as e: + }) + except (OSError, ValueError) as e: logger.error( "Failed to write refreshed Hermes OAuth credentials to %s: %s", oauth_file, e ) raise CredentialPersistError(oauth_file, e) from e - diff --git a/agent/nous_rate_guard.py b/agent/nous_rate_guard.py index 0234eef2ea..b5acb048b3 100644 --- a/agent/nous_rate_guard.py +++ b/agent/nous_rate_guard.py @@ -2,16 +2,13 @@ Writes rate limit state to a shared file so all sessions (CLI, gateway, cron, auxiliary) can check whether Nous Portal is currently rate-limited -before making requests. Prevents retry amplification when RPH is tapped. - -Each 429 from Nous triggers up to 9 API calls per conversation turn -(3 SDK retries x 3 Hermes retries), and every one of those calls counts -against RPH. By recording the rate limit state on first 429 and checking -it before subsequent attempts, we eliminate the amplification effect. +before making requests. Without it each 429 fans out into up to 9 calls per +turn (3 SDK retries x 3 Hermes retries), all counted against RPH. """ from __future__ import annotations +import contextlib import json import logging import os @@ -19,11 +16,22 @@ import tempfile import time from typing import Any, Mapping, Optional from utils import atomic_replace +from agent.rate_limit_tracker import ( + _BUCKET_TAGS, + _fmt_seconds, + _safe_float, + _safe_int, + has_rate_limit_headers, + lower_headers, +) logger = logging.getLogger(__name__) -_STATE_SUBDIR = "rate_limits" -_STATE_FILENAME = "nous.json" +# Reset windows shorter than this are transient upstream jitter, not a quota +# exhaustion worth a cross-session breaker trip. +_MIN_RESET_FOR_BREAKER_SECONDS = 60.0 + +format_remaining = _fmt_seconds def _state_path() -> str: @@ -33,38 +41,20 @@ def _state_path() -> str: base = get_hermes_home() except ImportError: base = os.path.join(os.path.expanduser("~"), ".hermes") - return os.path.join(base, _STATE_SUBDIR, _STATE_FILENAME) + return os.path.join(base, "rate_limits", "nous.json") def _parse_reset_seconds(headers: Optional[Mapping[str, str]]) -> Optional[float]: - """Extract the best available reset-time estimate from response headers. - - Priority: - 1. x-ratelimit-reset-requests-1h (hourly RPH window — most useful) - 2. x-ratelimit-reset-requests (per-minute RPM window) - 3. retry-after (generic HTTP header) - - Returns seconds-from-now, or None if no usable header found. - """ - if not headers: - return None - - lowered = {k.lower(): v for k, v in headers.items()} - + """Best reset estimate (seconds from now) from hourly, per-minute, then retry-after headers.""" + lowered = lower_headers(headers) for key in ( "x-ratelimit-reset-requests-1h", "x-ratelimit-reset-requests", "retry-after", ): - raw = lowered.get(key) - if raw is not None: - try: - val = float(raw) - if val > 0: - return val - except (TypeError, ValueError): - pass - + val = _safe_float(lowered.get(key), 0.0) + if val > 0: + return val return None @@ -74,32 +64,23 @@ def record_nous_rate_limit( error_context: Optional[dict[str, Any]] = None, default_cooldown: float = 300.0, ) -> None: - """Record that Nous Portal is rate-limited. + """Record that Nous Portal is rate-limited in the shared state file. - Parses the reset time from response headers or error context. - Falls back to ``default_cooldown`` (5 minutes) if no reset info - is available. Writes to a shared file that all sessions can read. - - Args: - headers: HTTP response headers from the 429 error. - error_context: Structured error context from _extract_api_error_context(). - default_cooldown: Fallback cooldown in seconds when no header data. + Reset time comes from headers, then ``error_context["reset_at"]`` (body + parsing), then ``default_cooldown``. """ now = time.time() reset_at = None - # Try headers first (most accurate) header_seconds = _parse_reset_seconds(headers) if header_seconds is not None: reset_at = now + header_seconds - # Try error_context reset_at (from body parsing) if reset_at is None and isinstance(error_context, dict): ctx_reset = error_context.get("reset_at") if isinstance(ctx_reset, (int, float)) and ctx_reset > now: reset_at = float(ctx_reset) - # Default cooldown if reset_at is None: reset_at = now + default_cooldown @@ -114,18 +95,14 @@ def record_nous_rate_limit( "reset_seconds": reset_at - now, } - # Atomic write: write to temp file + rename fd, tmp_path = tempfile.mkstemp(dir=state_dir, suffix=".tmp") try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(state, f) atomic_replace(tmp_path, path) except Exception: - # Clean up temp file on failure - try: + with contextlib.suppress(OSError): os.unlink(tmp_path) - except OSError: - pass raise logger.info( @@ -137,24 +114,16 @@ def record_nous_rate_limit( def nous_rate_limit_remaining() -> Optional[float]: - """Check if Nous Portal is currently rate-limited. - - Returns: - Seconds remaining until reset, or None if not rate-limited. - """ + """Seconds remaining until reset, or None if not rate-limited (expired state is removed).""" path = _state_path() try: with open(path, encoding="utf-8") as f: state = json.load(f) - reset_at = state.get("reset_at", 0) - remaining = reset_at - time.time() + remaining = state.get("reset_at", 0) - time.time() if remaining > 0: return remaining - # Expired — clean up - try: + with contextlib.suppress(OSError): os.unlink(path) - except OSError: - pass return None except (FileNotFoundError, json.JSONDecodeError, KeyError, TypeError): return None @@ -170,23 +139,14 @@ def clear_nous_rate_limit() -> None: logger.debug("Failed to clear Nous rate limit state: %s", exc) -def format_remaining(seconds: float) -> str: - """Format seconds remaining into human-readable duration.""" - s = max(0, int(seconds)) - if s < 60: - return f"{s}s" - if s < 3600: - m, sec = divmod(s, 60) - return f"{m}m {sec}s" if sec else f"{m}m" - h, remainder = divmod(s, 3600) - m = remainder // 60 - return f"{h}h {m}m" if m else f"{h}h" - - -# Buckets with reset windows shorter than this are treated as transient -# (upstream jitter, secondary throttling) rather than a genuine quota -# exhaustion worth a cross-session breaker trip. -_MIN_RESET_FOR_BREAKER_SECONDS = 60.0 +def _is_exhausted(remaining: Optional[int], reset: Optional[float]) -> bool: + """remaining == 0 AND a reset window long enough to be a real quota exhaustion.""" + return ( + remaining is not None + and remaining <= 0 + and reset is not None + and reset >= _MIN_RESET_FOR_BREAKER_SECONDS + ) def is_genuine_nous_rate_limit( @@ -196,88 +156,29 @@ def is_genuine_nous_rate_limit( ) -> bool: """Decide whether a 429 from Nous Portal is a real account rate limit. - Nous Portal multiplexes multiple upstream providers (DeepSeek, Kimi, - MiMo, Hermes, ...) behind one endpoint. A 429 can mean either: - - (a) The caller's own RPM / RPH / TPM / TPH bucket on Nous is - exhausted — a genuine rate limit that will last until the - bucket resets. - (b) The upstream provider is out of capacity for a specific model - — transient, clears in seconds, and has nothing to do with - the caller's quota on Nous. - - Tripping the cross-session breaker on (b) blocks ALL Nous requests - (and all models, since Nous is one provider key) for minutes even - though the caller's account is healthy and a different model would - have worked. That's the bug users hit when DeepSeek V4 Pro 429s - trigger a breaker that then blocks Kimi 2.6 and MiMo V2.5 Pro. - - We tell the two apart by looking at: - - 1. The 429 response's own ``x-ratelimit-*`` headers. Nous emits - the full suite on every response including 429s. An exhausted - bucket (``remaining == 0`` with a reset window >= 60s) is - proof of (a). - 2. The last-known-good rate-limit state captured by - ``_capture_rate_limits()`` on the previous successful - response. If any bucket there was already near-exhausted with - a substantial reset window, the current 429 is almost - certainly (a) continuing from that condition. - - If neither signal fires, we treat the 429 as (b): fail the single - request, let the retry loop or model-switch proceed, and do NOT - write the cross-session breaker file. - - Returns True when the evidence points at (a). + Nous multiplexes upstream providers behind one key, so a 429 may be one + upstream model out of capacity (clears in seconds) rather than our quota; + tripping the breaker on that would block every Nous model for minutes. + Only an exhausted bucket (remaining == 0 with reset >= 60s) in the 429's + own headers, or in the last-known-good ``RateLimitState``, is genuine. """ - # Signal 1: current 429 response headers. - state = _parse_buckets_from_headers(headers) - if _has_exhausted_bucket(state): + if _has_exhausted_bucket(_parse_buckets_from_headers(headers)): return True - - # Signal 2: last-known-good state from a recent successful response. - # Accepts either a RateLimitState (dataclass from rate_limit_tracker) - # or a dict of bucket snapshots. - if last_known_state is not None and _has_exhausted_bucket_in_object(last_known_state): - return True - - return False + return last_known_state is not None and _has_exhausted_bucket_in_object(last_known_state) def _parse_buckets_from_headers( headers: Optional[Mapping[str, str]], ) -> dict[str, tuple[Optional[int], Optional[float]]]: - """Extract (remaining, reset_seconds) per bucket from x-ratelimit-* headers. - - Returns empty dict when no rate-limit headers are present. - """ - if not headers: + """Extract (remaining, reset_seconds) per bucket from x-ratelimit-* headers ({} if none).""" + lowered = lower_headers(headers) + if not has_rate_limit_headers(lowered): return {} - lowered = {k.lower(): v for k, v in headers.items()} - if not any(k.startswith("x-ratelimit-") for k in lowered): - return {} - - def _maybe_int(raw: Optional[str]) -> Optional[int]: - if raw is None: - return None - try: - return int(float(raw)) - except (TypeError, ValueError): - return None - - def _maybe_float(raw: Optional[str]) -> Optional[float]: - if raw is None: - return None - try: - return float(raw) - except (TypeError, ValueError): - return None - result: dict[str, tuple[Optional[int], Optional[float]]] = {} - for tag in ("requests", "requests-1h", "tokens", "tokens-1h"): - remaining = _maybe_int(lowered.get(f"x-ratelimit-remaining-{tag}")) - reset = _maybe_float(lowered.get(f"x-ratelimit-reset-{tag}")) + for _attr, tag in _BUCKET_TAGS: + remaining = _safe_int(lowered.get(f"x-ratelimit-remaining-{tag}"), None) + reset = _safe_float(lowered.get(f"x-ratelimit-reset-{tag}"), None) if remaining is not None or reset is not None: result[tag] = (remaining, reset) return result @@ -286,40 +187,19 @@ def _parse_buckets_from_headers( def _has_exhausted_bucket( buckets: Mapping[str, tuple[Optional[int], Optional[float]]], ) -> bool: - """Return True when any bucket has remaining == 0 AND a meaningful reset window.""" - for remaining, reset in buckets.values(): - if remaining is None or remaining > 0: - continue - if reset is None: - continue - if reset >= _MIN_RESET_FOR_BREAKER_SECONDS: - return True - return False + return any(_is_exhausted(remaining, reset) for remaining, reset in buckets.values()) def _has_exhausted_bucket_in_object(state: Any) -> bool: - """Check a RateLimitState-like object for an exhausted bucket. - - Accepts the dataclass from ``agent.rate_limit_tracker`` (buckets - exposed as attributes ``requests_min``, ``requests_hour``, - ``tokens_min``, ``tokens_hour``) and falls back gracefully for any - object missing those attributes. - """ - for attr in ("requests_min", "requests_hour", "tokens_min", "tokens_hour"): + """Check a RateLimitState-like object (duck-typed; missing attrs are skipped).""" + for attr, _tag in _BUCKET_TAGS: bucket = getattr(state, attr, None) - if bucket is None: + if bucket is None or (getattr(bucket, "limit", 0) or 0) <= 0: continue - limit = getattr(bucket, "limit", 0) or 0 remaining = getattr(bucket, "remaining", 0) or 0 - # Prefer the adjusted "remaining_seconds_now" property when present; - # fall back to raw reset_seconds. reset = getattr(bucket, "remaining_seconds_now", None) if reset is None: reset = getattr(bucket, "reset_seconds", 0.0) or 0.0 - if limit <= 0: - continue - if remaining > 0: - continue - if reset >= _MIN_RESET_FOR_BREAKER_SECONDS: + if _is_exhausted(remaining, reset): return True return False diff --git a/agent/rate_limit_tracker.py b/agent/rate_limit_tracker.py index e20c683341..b32e5968a1 100644 --- a/agent/rate_limit_tracker.py +++ b/agent/rate_limit_tracker.py @@ -1,23 +1,8 @@ """Rate limit tracking for inference API responses. -Captures x-ratelimit-* headers from provider responses and provides -formatted display for the /usage slash command. Currently supports -the Nous Portal header format (also used by OpenRouter and OpenAI-compatible -APIs that follow the same convention). - -Header schema (12 headers total): - x-ratelimit-limit-requests RPM cap - x-ratelimit-limit-requests-1h RPH cap - x-ratelimit-limit-tokens TPM cap - x-ratelimit-limit-tokens-1h TPH cap - x-ratelimit-remaining-requests requests left in minute window - x-ratelimit-remaining-requests-1h requests left in hour window - x-ratelimit-remaining-tokens tokens left in minute window - x-ratelimit-remaining-tokens-1h tokens left in hour window - x-ratelimit-reset-requests seconds until minute request window resets - x-ratelimit-reset-requests-1h seconds until hour request window resets - x-ratelimit-reset-tokens seconds until minute token window resets - x-ratelimit-reset-tokens-1h seconds until hour token window resets +Captures ``x-ratelimit-{limit,remaining,reset}-{requests,tokens}[-1h]`` +headers (Nous Portal format, also used by OpenRouter / OpenAI-compatible APIs) +and formats them for the /usage slash command. Reset values are seconds. """ from __future__ import annotations @@ -26,6 +11,14 @@ import time from dataclasses import dataclass, field from typing import Any, Mapping, Optional +# (state attribute, header tag) for the four windows. +_BUCKET_TAGS = ( + ("requests_min", "requests"), + ("requests_hour", "requests-1h"), + ("tokens_min", "tokens"), + ("tokens_hour", "tokens-1h"), +) + @dataclass class RateLimitBucket: @@ -42,15 +35,12 @@ class RateLimitBucket: @property def usage_pct(self) -> float: - if self.limit <= 0: - return 0.0 - return (self.used / self.limit) * 100.0 + return (self.used / self.limit) * 100.0 if self.limit > 0 else 0.0 @property def remaining_seconds_now(self) -> float: """Estimated seconds remaining until reset, adjusted for elapsed time.""" - elapsed = time.time() - self.captured_at - return max(0.0, self.reset_seconds - elapsed) + return max(0.0, self.reset_seconds - (time.time() - self.captured_at)) @dataclass @@ -70,63 +60,52 @@ class RateLimitState: @property def age_seconds(self) -> float: - if not self.has_data: - return float("inf") - return time.time() - self.captured_at + return time.time() - self.captured_at if self.has_data else float("inf") -def _safe_int(value: Any, default: int = 0) -> int: - try: - return int(float(value)) - except (TypeError, ValueError): - return default - - -def _safe_float(value: Any, default: float = 0.0) -> float: +def _safe_float(value: Any, default: Any = 0.0) -> Any: try: return float(value) except (TypeError, ValueError): return default +def _safe_int(value: Any, default: Any = 0) -> Any: + try: + return int(float(value)) + except (TypeError, ValueError): + return default + + +def lower_headers(headers: Optional[Mapping[str, str]]) -> dict[str, str]: + """Lowercase header names (HTTP header names are case-insensitive).""" + return {k.lower(): v for k, v in headers.items()} if headers else {} + + +def has_rate_limit_headers(lowered: Mapping[str, str]) -> bool: + return any(k.startswith("x-ratelimit-") for k in lowered) + + def parse_rate_limit_headers( headers: Mapping[str, str], provider: str = "", ) -> Optional[RateLimitState]: - """Parse x-ratelimit-* headers into a RateLimitState. - - Returns None if no rate limit headers are present. - """ - # Normalize to lowercase so lookups work regardless of how the server - # capitalises headers (HTTP header names are case-insensitive per RFC 7230). - lowered = {k.lower(): v for k, v in headers.items()} - - # Quick check: at least one rate limit header must exist - has_any = any(k.startswith("x-ratelimit-") for k in lowered) - if not has_any: + """Parse x-ratelimit-* headers into a RateLimitState (None if none present).""" + lowered = lower_headers(headers) + if not has_rate_limit_headers(lowered): return None now = time.time() - - def _bucket(resource: str, suffix: str = "") -> RateLimitBucket: - # e.g. resource="requests", suffix="" -> per-minute - # resource="tokens", suffix="-1h" -> per-hour - tag = f"{resource}{suffix}" - return RateLimitBucket( + buckets = { + attr: RateLimitBucket( limit=_safe_int(lowered.get(f"x-ratelimit-limit-{tag}")), remaining=_safe_int(lowered.get(f"x-ratelimit-remaining-{tag}")), reset_seconds=_safe_float(lowered.get(f"x-ratelimit-reset-{tag}")), captured_at=now, ) - - return RateLimitState( - requests_min=_bucket("requests"), - requests_hour=_bucket("requests", "-1h"), - tokens_min=_bucket("tokens"), - tokens_hour=_bucket("tokens", "-1h"), - captured_at=now, - provider=provider, - ) + for attr, tag in _BUCKET_TAGS + } + return RateLimitState(captured_at=now, provider=provider, **buckets) # ── Formatting ────────────────────────────────────────────────────────── @@ -136,8 +115,6 @@ def _fmt_count(n: int) -> str: """Human-friendly number: 7999856 -> '8.0M', 33599 -> '33.6K', 799 -> '799'.""" if n >= 1_000_000: return f"{n / 1_000_000:.1f}M" - if n >= 10_000: - return f"{n / 1_000:.1f}K" if n >= 1_000: return f"{n / 1_000:.1f}K" return str(n) @@ -158,10 +135,8 @@ def _fmt_seconds(seconds: float) -> str: def _bar(pct: float, width: int = 20) -> str: """ASCII progress bar: [████████░░░░░░░░░░░░] 40%.""" - filled = int(pct / 100.0 * width) - filled = max(0, min(width, filled)) - empty = width - filled - return f"[{'█' * filled}{'░' * empty}]" + filled = max(0, min(width, int(pct / 100.0 * width))) + return f"[{'█' * filled}{'░' * (width - filled)}]" def _bucket_line(label: str, bucket: RateLimitBucket, label_width: int = 14) -> str: @@ -170,13 +145,9 @@ def _bucket_line(label: str, bucket: RateLimitBucket, label_width: int = 14) -> return f" {label:<{label_width}} (no data)" pct = bucket.usage_pct - used = _fmt_count(bucket.used) - limit = _fmt_count(bucket.limit) - remaining = _fmt_count(bucket.remaining) + used, limit, remaining = map(_fmt_count, (bucket.used, bucket.limit, bucket.remaining)) reset = _fmt_seconds(bucket.remaining_seconds_now) - - bar = _bar(pct) - return f" {label:<{label_width}} {bar} {pct:5.1f}% {used}/{limit} used ({remaining} left, resets in {reset})" + return f" {label:<{label_width}} {_bar(pct)} {pct:5.1f}% {used}/{limit} used ({remaining} left, resets in {reset})" def format_rate_limit_display(state: RateLimitState) -> str: @@ -194,27 +165,21 @@ def format_rate_limit_display(state: RateLimitState) -> str: provider_label = state.provider.title() if state.provider else "Provider" - lines = [ - f"{provider_label} Rate Limits (captured {freshness}):", - "", - _bucket_line("Requests/min", state.requests_min), - _bucket_line("Requests/hr", state.requests_hour), - "", - _bucket_line("Tokens/min", state.tokens_min), - _bucket_line("Tokens/hr", state.tokens_hour), + labeled = [ + ("Requests/min", state.requests_min), + ("Requests/hr", state.requests_hour), + ("Tokens/min", state.tokens_min), + ("Tokens/hr", state.tokens_hour), ] + lines = [f"{provider_label} Rate Limits (captured {freshness}):", ""] + lines += [_bucket_line(label, bucket) for label, bucket in labeled[:2]] + lines += [""] + [_bucket_line(label, bucket) for label, bucket in labeled[2:]] - # Add warnings if any bucket is getting hot warnings = [] - for label, bucket in [ - ("requests/min", state.requests_min), - ("requests/hr", state.requests_hour), - ("tokens/min", state.tokens_min), - ("tokens/hr", state.tokens_hour), - ]: + for label, bucket in labeled: if bucket.limit > 0 and bucket.usage_pct >= 80: reset = _fmt_seconds(bucket.remaining_seconds_now) - warnings.append(f" ⚠ {label} at {bucket.usage_pct:.0f}% — resets in {reset}") + warnings.append(f" ⚠ {label.lower()} at {bucket.usage_pct:.0f}% — resets in {reset}") if warnings: lines.append("") @@ -228,10 +193,7 @@ def format_rate_limit_compact(state: RateLimitState) -> str: if not state.has_data: return "No rate limit data." - rm = state.requests_min - tm = state.tokens_min - rh = state.requests_hour - th = state.tokens_hour + rm, rh, tm, th = state.requests_min, state.requests_hour, state.tokens_min, state.tokens_hour parts = [] if rm.limit > 0: