hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(agent/creds): unify Anthropic credential I/O and rate-limit header helpers (-30% LOC)

Browse Source 
- 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.
This commit is contained in:
Teknium
2026-09-02 13:07:08 -07:00
parent a3d33fe22f
commit dd522d0f77
3 changed files with 425 additions and 948 deletions
Download Patch File Download Diff File Expand all files Collapse all files

997
agent/anthropic_credentials.py
View File

File diff suppressed because it is too large Load Diff

230
agent/nous_rate_guard.py
View File

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

146
agent/rate_limit_tracker.py
View File

@@ -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: