mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-22 16:25:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(auth): detect upstream Codex quota resets and lift stale pool cooldowns

Browse source 
When Codex returns 429 usage_limit_reached, Hermes persists the provider's
reset_at on the pool entry and freezes the credential until it elapses --
which can be days out for weekly windows. But the upstream window can
reopen EARLY: the user redeems a banked rate-limit reset (Codex CLI /
ChatGPT UI), upgrades their plan, or OpenAI resets the window. Hermes
never re-checked, so it kept erroring with 'Codex provider quota
exhausted (429); retry after Ns' until a manual re-auth rewrote the
tokens (issue #43747, externally-reset variant).

- hermes_cli/auth.py: add _probe_codex_quota_restored() -- a throttled
  (5 min/token) GET of the Codex /usage endpoint; quota counts as
  restored when every reported window is <100% used. Add
  clear_codex_pool_quota_cooldowns() to lift 429/quota-shaped cooldowns
  from persisted pool entries (DEAD and auth-shaped entries untouched).
- resolve_codex_runtime_credentials(): before surfacing a pool-only
  cooldown as 'quota exhausted', probe upstream; on a positive probe
  clear the cooldown and return the pool credential.
- agent/credential_pool.py: _available_entries() probes frozen
  openai-codex entries (clear_expired path only) and unfreezes them when
  upstream confirms the reset.
- agent/account_usage.py: a successful /usage reset redemption now
  clears persisted pool cooldowns immediately.

Negative paths preserved: probe 429/exhausted/indeterminate keeps the
cooldown; read-only enumeration never probes; non-JWT tokens never
probe (no network in hermetic tests).
This commit is contained in:
Teknium 2026-07-22 09:00:48 -07:00
parent cbc1054e23
commit 3af00425a0
No known key found for this signature in database
4 changed files with 783 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

12
agent/account_usage.py
View file

@ -701,6 +701,18 @@ def redeem_codex_reset_credit(
remaining = max(0, available - 1)
plural = "s" if remaining != 1 else ""
if code == "reset":
# The redeemed reset restores the account's quota upstream — lift any
# persisted pool cooldowns so Hermes doesn't keep the credential
# frozen behind the now-stale ``last_error_reset_at`` (issue #43747).
try:
from hermes_cli.auth import clear_codex_pool_quota_cooldowns
clear_codex_pool_quota_cooldowns()
except Exception:
logger.debug(
"Failed to clear Codex pool cooldowns after reset redemption",
exc_info=True,
)
return CodexResetRedeemResult(
status="reset",
message=(

50
agent/credential_pool.py
View file

@ -1484,6 +1484,43 @@ class CredentialPool:
self._sync_device_code_entry_to_auth_store(updated)
return updated
def _codex_quota_restored_upstream(self, entry: PooledCredential) -> bool:
"""Live-check whether an exhausted Codex entry's quota reset early.
A Codex 429 persists a ``last_error_reset_at`` that can be days in
the future (weekly windows), but the upstream window can reopen
before then — the user redeems a banked rate-limit reset via the
Codex CLI / ChatGPT UI, upgrades their plan, or OpenAI resets the
window. Without this check the pool keeps the credential frozen
until the stale timestamp elapses even though the account is
usable (issue #43747).
Only fires for openai-codex entries frozen by a 429/quota-shaped
error. The underlying probe is throttled per token (5 min) so this
is safe on the hot selection path.
"""
if self.provider != "openai-codex" or entry.last_status != STATUS_EXHAUSTED:
return False
if not auth_mod._is_codex_rate_limit_shaped(
entry.last_error_code,
entry.last_error_reason,
entry.last_error_message,
):
return False
token = entry.access_token or ""
if not token:
return False
try:
return bool(
auth_mod._probe_codex_quota_restored(
token,
base_url=entry.base_url,
)
)
except Exception:
logger.debug("Codex quota-restored probe failed", exc_info=True)
return False
def _entry_needs_refresh(self, entry: PooledCredential) -> bool:
if entry.auth_type != AUTH_TYPE_OAUTH:
return False
@ -1605,7 +1642,18 @@ class CredentialPool:
if entry.last_status == STATUS_EXHAUSTED:
exhausted_until = _exhausted_until(entry)
if exhausted_until is not None and now < exhausted_until:
continue
# Codex quota windows can reopen EARLY: the user redeems a
# banked rate-limit reset (Codex CLI / ChatGPT UI), upgrades
# their plan, or OpenAI resets the window. The persisted
# ``last_error_reset_at`` can then be days in the future
# while the account is already usable again — a throttled
# live probe of the Codex usage endpoint detects that and
# lifts the stale cooldown (issue #43747).
if not (
clear_expired
and self._codex_quota_restored_upstream(entry)
):
continue
if clear_expired:
cleared = replace(
entry,

214
hermes_cli/auth.py
View file

@ -3774,6 +3774,34 @@ def resolve_codex_runtime_credentials(
}
pool_rate_limit = _codex_pool_rate_limit_status()
if pool_rate_limit:
# Before surfacing the persisted cooldown, ask the Codex usage
# endpoint whether the quota actually reset early (banked reset
# redeemed, plan upgraded, window reset upstream). The persisted
# ``last_error_reset_at`` can be days in the future while the
# account is already usable again — see issue #43747.
stale_token = str(pool_rate_limit.get("access_token") or "").strip()
if stale_token and _probe_codex_quota_restored(
stale_token,
base_url=pool_rate_limit.get("base_url"),
):
logger.info(
"Codex quota restored upstream — clearing stale pool cooldown(s)."
)
clear_codex_pool_quota_cooldowns()
pool_token = _pool_codex_access_token()
if pool_token:
base_url = (
os.getenv("HERMES_CODEX_BASE_URL", "").strip().rstrip("/")
or DEFAULT_CODEX_BASE_URL
)
return {
"provider": "openai-codex",
"base_url": base_url,
"api_key": pool_token,
"source": "credential_pool",
"last_refresh": None,
"auth_mode": "chatgpt",
}
reset_at = pool_rate_limit.get("reset_at")
if isinstance(reset_at, (int, float)) and reset_at > time.time():
remaining = int(reset_at - time.time())
@ -3838,6 +3866,190 @@ def resolve_codex_runtime_credentials(
}
def _is_codex_rate_limit_shaped(
code: Any,
reason: Any,
message: Any,
) -> bool:
"""True when persisted pool-entry error metadata describes a 429/quota stop."""
reason_l = str(reason or "").lower()
message_l = str(message or "").lower()
return (
code == 429
or "rate_limit" in reason_l
or "usage_limit" in reason_l
or "quota" in reason_l
or "rate limit" in message_l
or "usage limit" in message_l
or "quota" in message_l
)
# Throttle for the live Codex quota probe below. The probe runs on the hot
# credential-selection path while the pool is exhausted, so without a floor a
# busy gateway would hammer the usage endpoint on every model/auxiliary call.
CODEX_QUOTA_PROBE_MIN_INTERVAL_SECONDS = 300 # 5 minutes
_codex_quota_probe_cache: Dict[str, Tuple[float, Optional[bool]]] = {}
_codex_quota_probe_lock = threading.Lock()
def _codex_usage_probe_url(base_url: Optional[str]) -> str:
"""Resolve the Codex usage endpoint for a probe.
Mirrors the Codex CLI's PathStyle split (codex-rs backend-client, same
logic as ``agent.account_usage._codex_backend_urls``): base URLs
containing ``/backend-api`` use the ChatGPT ``/wham/usage`` path;
everything else uses ``/api/codex/usage``. Kept local so this low-level
auth module doesn't import the auxiliary account-usage module.
"""
normalized = str(base_url or "").strip().rstrip("/")
if not normalized:
normalized = (
os.getenv("HERMES_CODEX_BASE_URL", "").strip().rstrip("/")
or DEFAULT_CODEX_BASE_URL
)
if normalized.endswith("/codex"):
normalized = normalized[: -len("/codex")]
prefix = normalized + ("/wham" if "/backend-api" in normalized else "/api/codex")
return prefix + "/usage"
def _probe_codex_quota_restored(
access_token: Any,
*,
base_url: Optional[str] = None,
min_interval_seconds: float = CODEX_QUOTA_PROBE_MIN_INTERVAL_SECONDS,
) -> Optional[bool]:
"""Ask the Codex usage endpoint whether this account's quota is usable again.
Hermes persists a Codex 429's ``reset_at`` locally and freezes the
credential until it elapses — but the upstream window can reopen EARLY
(the user redeems a banked rate-limit reset via the Codex CLI/ChatGPT UI,
upgrades their plan, or OpenAI resets the window). This probe detects
that: it GETs the same ``/usage`` endpoint the Codex CLI uses and checks
the reported windows.
Returns:
* ``True`` — every reported rate-limit window is below 100% used;
the account can serve requests again and stale local cooldowns
should be lifted.
* ``False`` — a window is still fully used (or the probe itself 429'd);
keep the cooldown.
* ``None`` — indeterminate (no token, network error, unexpected
payload/status); keep the cooldown.
Probes are throttled per access token (module-local cache) so the hot
selection path can fire this freely.
"""
token = str(access_token or "").strip()
if not token:
return None
# Real Codex access tokens are JWTs. Refusing to probe non-JWT tokens
# avoids pointless network calls for corrupt/placeholder entries (and
# keeps hermetic test fixtures with dummy tokens offline).
if not _decode_jwt_claims(token):
return None
cache_key = hashlib.sha256(token.encode("utf-8")).hexdigest()[:16]
now = time.monotonic()
with _codex_quota_probe_lock:
cached = _codex_quota_probe_cache.get(cache_key)
if cached is not None and (now - cached[0]) < min_interval_seconds:
return cached[1]
# Reserve the slot immediately so concurrent selectors don't stampede
# the endpoint while this probe is in flight.
_codex_quota_probe_cache[cache_key] = (now, None)
result: Optional[bool] = None
try:
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/json",
"User-Agent": "codex-cli",
}
# Best-effort ChatGPT-Account-Id from the JWT (the backend requires it
# for some account shapes; harmless to omit for others).
claims = _decode_jwt_claims(token)
account_id = (
claims.get("https://api.openai.com/auth", {}).get("chatgpt_account_id")
if isinstance(claims.get("https://api.openai.com/auth"), dict)
else None
)
if isinstance(account_id, str) and account_id.strip():
headers["ChatGPT-Account-Id"] = account_id.strip()
with httpx.Client(timeout=10.0) as client:
response = client.get(_codex_usage_probe_url(base_url), headers=headers)
if response.status_code == 200:
payload = response.json() or {}
rate_limit = payload.get("rate_limit") or {}
worst_used: Optional[float] = None
for key in ("primary_window", "secondary_window"):
used = (rate_limit.get(key) or {}).get("used_percent")
if isinstance(used, (int, float)):
worst_used = max(worst_used or 0.0, float(used))
if worst_used is not None:
result = worst_used < 100.0
elif response.status_code == 429:
result = False
except Exception:
logger.debug("Codex quota probe failed", exc_info=True)
result = None
with _codex_quota_probe_lock:
_codex_quota_probe_cache[cache_key] = (now, result)
return result
def clear_codex_pool_quota_cooldowns(access_token: Optional[str] = None) -> int:
"""Clear rate-limit cooldowns on persisted openai-codex pool entries.
Called after the upstream quota is KNOWN to be restored (a successful
``/usage reset`` redemption, or a positive live probe) so auth.json stops
freezing credentials behind a stale ``last_error_reset_at``. Only lifts
``exhausted`` entries whose error metadata is 429/quota-shaped — DEAD
(terminal auth) entries and non-rate-limit failures are untouched.
When *access_token* is given, only the matching entry is cleared;
otherwise every rate-limited entry clears (a redeemed banked reset
restores the whole account, and any entry that is genuinely still
exhausted just re-freezes with fresh metadata on its next 429).
Returns the number of entries cleared.
"""
cleared = 0
try:
with _auth_store_lock():
auth_store = _load_auth_store()
pool = auth_store.get("credential_pool")
entries = pool.get("openai-codex") if isinstance(pool, dict) else None
if not isinstance(entries, list):
return 0
for entry in entries:
if not isinstance(entry, dict):
continue
if entry.get("last_status") != "exhausted":
continue
if access_token and str(entry.get("access_token") or "") != access_token:
continue
if not _is_codex_rate_limit_shaped(
entry.get("last_error_code"),
entry.get("last_error_reason"),
entry.get("last_error_message"),
):
continue
entry["last_status"] = None
entry["last_status_at"] = None
entry["last_error_code"] = None
entry["last_error_reason"] = None
entry["last_error_message"] = None
entry["last_error_reset_at"] = None
cleared += 1
if cleared:
_save_auth_store(auth_store)
except Exception:
logger.debug("Failed to clear Codex pool quota cooldowns", exc_info=True)
return cleared
def _codex_pool_rate_limit_status() -> Optional[Dict[str, Any]]:
"""Return metadata for a pool-only Codex credential in quota cooldown."""
def _parse_reset_at(value: Any) -> Optional[float]:
@ -3905,6 +4117,8 @@ def _codex_pool_rate_limit_status() -> Optional[Dict[str, Any]]:
"reset_at": reset_at,
"reason": entry.get("last_error_reason"),
"message": entry.get("last_error_message"),
"access_token": token.strip(),
"base_url": entry.get("base_url"),
}
except Exception:
logger.debug("Codex pool rate-limit lookup failed", exc_info=True)

508
tests/hermes_cli/test_auth_codex_quota_probe.py Normal file
View file

@ -0,0 +1,508 @@
"""Tests for the Codex upstream-quota-restored probe and cooldown clearing.
Covers issue #43747 (externally-reset variant): Codex 429s persist a
``last_error_reset_at`` that can be days in the future, but the upstream
window can reopen early (banked reset redeemed, plan upgrade, upstream
reset). Hermes must detect that and lift the stale local cooldown instead
of refusing requests until re-auth.
"""
import base64
import json
import time
from types import SimpleNamespace
import pytest
import hermes_cli.auth as auth_mod
from hermes_cli.auth import (
AuthError,
_codex_usage_probe_url,
_is_codex_rate_limit_shaped,
_probe_codex_quota_restored,
clear_codex_pool_quota_cooldowns,
resolve_codex_runtime_credentials,
)
@pytest.fixture(autouse=True)
def _clear_probe_cache():
auth_mod._codex_quota_probe_cache.clear()
yield
auth_mod._codex_quota_probe_cache.clear()
def _jwt(claims: dict) -> str:
def _part(payload: dict) -> str:
raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
return f"{_part({'alg': 'none'})}.{_part(claims)}.sig"
class _StubResponse:
def __init__(self, status_code: int, payload: dict):
self.status_code = status_code
self._payload = payload
def json(self):
return self._payload
def raise_for_status(self):
if self.status_code >= 400:
import httpx
raise httpx.HTTPStatusError(
f"HTTP {self.status_code}", request=None, response=self # type: ignore[arg-type]
)
class _StubClient:
def __init__(self, calls, response):
self._calls = calls
self._response = response
def __enter__(self):
return self
def __exit__(self, *args):
return False
def get(self, url, headers=None):
self._calls.append({"url": url, "headers": dict(headers or {})})
return self._response
def _patch_httpx(monkeypatch, response, calls=None):
calls = calls if calls is not None else []
monkeypatch.setattr(
auth_mod.httpx, "Client", lambda **kwargs: _StubClient(calls, response)
)
return calls
def _usage_payload(primary_used: float, secondary_used: float) -> dict:
return {
"rate_limit": {
"primary_window": {"used_percent": primary_used},
"secondary_window": {"used_percent": secondary_used},
}
}
# ---------------------------------------------------------------------------
# _is_codex_rate_limit_shaped
# ---------------------------------------------------------------------------
def test_rate_limit_shaped_variants():
assert _is_codex_rate_limit_shaped(429, None, None)
assert _is_codex_rate_limit_shaped(None, "usage_limit_reached", None)
assert _is_codex_rate_limit_shaped(None, None, "The usage limit has been reached")
assert _is_codex_rate_limit_shaped(None, "quota_exceeded", None)
assert not _is_codex_rate_limit_shaped(401, "token_invalidated", "token revoked")
assert not _is_codex_rate_limit_shaped(None, None, None)
# ---------------------------------------------------------------------------
# _codex_usage_probe_url
# ---------------------------------------------------------------------------
def test_probe_url_backend_api_uses_wham():
assert (
_codex_usage_probe_url("https://chatgpt.com/backend-api/codex")
== "https://chatgpt.com/backend-api/wham/usage"
)
def test_probe_url_non_backend_uses_api_codex():
assert (
_codex_usage_probe_url("https://example.com/codex")
== "https://example.com/api/codex/usage"
)
# ---------------------------------------------------------------------------
# _probe_codex_quota_restored
# ---------------------------------------------------------------------------
def test_probe_returns_true_when_windows_below_100(monkeypatch):
calls = _patch_httpx(monkeypatch, _StubResponse(200, _usage_payload(12.0, 34.0)))
token = _jwt({"exp": time.time() + 3600})
assert _probe_codex_quota_restored(token) is True
assert calls and calls[0]["url"].endswith("/usage")
def test_probe_returns_false_when_window_still_exhausted(monkeypatch):
_patch_httpx(monkeypatch, _StubResponse(200, _usage_payload(100.0, 34.0)))
token = _jwt({"exp": time.time() + 3600})
assert _probe_codex_quota_restored(token) is False
def test_probe_returns_false_on_429(monkeypatch):
_patch_httpx(monkeypatch, _StubResponse(429, {}))
token = _jwt({"exp": time.time() + 3600})
assert _probe_codex_quota_restored(token) is False
def test_probe_indeterminate_on_unexpected_payload(monkeypatch):
_patch_httpx(monkeypatch, _StubResponse(200, {}))
token = _jwt({"exp": time.time() + 3600})
assert _probe_codex_quota_restored(token) is None
def test_probe_skips_non_jwt_tokens_without_network(monkeypatch):
calls = _patch_httpx(monkeypatch, _StubResponse(200, _usage_payload(0.0, 0.0)))
assert _probe_codex_quota_restored("not-a-jwt") is None
assert _probe_codex_quota_restored("") is None
assert calls == []
def test_probe_throttles_repeat_calls(monkeypatch):
calls = _patch_httpx(monkeypatch, _StubResponse(200, _usage_payload(12.0, 34.0)))
token = _jwt({"exp": time.time() + 3600})
assert _probe_codex_quota_restored(token) is True
assert _probe_codex_quota_restored(token) is True # cached
assert len(calls) == 1
def test_probe_sends_chatgpt_account_id_from_jwt(monkeypatch):
calls = _patch_httpx(monkeypatch, _StubResponse(200, _usage_payload(0.0, 0.0)))
token = _jwt(
{
"exp": time.time() + 3600,
"https://api.openai.com/auth": {"chatgpt_account_id": "acct-123"},
}
)
assert _probe_codex_quota_restored(token) is True
assert calls[0]["headers"].get("ChatGPT-Account-Id") == "acct-123"
# ---------------------------------------------------------------------------
# clear_codex_pool_quota_cooldowns
# ---------------------------------------------------------------------------
def _write_auth_store(hermes_home, payload):
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps(payload, indent=2))
def _exhausted_pool_store(now=None):
now = now or time.time()
return {
"version": 1,
"providers": {},
"credential_pool": {
"openai-codex": [
{
"id": "cred-quota",
"label": "quota-frozen",
"auth_type": "oauth",
"priority": 0,
"source": "device_code",
"access_token": "tok-quota",
"last_status": "exhausted",
"last_status_at": now,
"last_error_code": 429,
"last_error_reason": "usage_limit_reached",
"last_error_message": "The usage limit has been reached",
"last_error_reset_at": now + 6 * 24 * 3600,
},
{
"id": "cred-dead",
"label": "revoked",
"auth_type": "oauth",
"priority": 1,
"source": "device_code",
"access_token": "tok-dead",
"last_status": "dead",
"last_status_at": now,
"last_error_code": 401,
"last_error_reason": "token_invalidated",
},
{
"id": "cred-auth",
"label": "auth-failure",
"auth_type": "oauth",
"priority": 2,
"source": "device_code",
"access_token": "tok-auth",
"last_status": "exhausted",
"last_status_at": now,
"last_error_code": 401,
"last_error_reason": "token_expired",
},
]
},
}
def test_clear_cooldowns_only_touches_quota_shaped_entries(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, _exhausted_pool_store())
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
assert clear_codex_pool_quota_cooldowns() == 1
store = json.loads((hermes_home / "auth.json").read_text())
entries = {e["id"]: e for e in store["credential_pool"]["openai-codex"]}
assert entries["cred-quota"]["last_status"] is None
assert entries["cred-quota"]["last_error_reset_at"] is None
# DEAD (terminal auth) and non-quota exhausted entries stay untouched.
assert entries["cred-dead"]["last_status"] == "dead"
assert entries["cred-auth"]["last_status"] == "exhausted"
def test_clear_cooldowns_scoped_to_access_token(tmp_path, monkeypatch):
now = time.time()
store = _exhausted_pool_store(now)
store["credential_pool"]["openai-codex"].append(
{
"id": "cred-quota-2",
"label": "other-quota",
"auth_type": "oauth",
"priority": 3,
"source": "device_code",
"access_token": "tok-other",
"last_status": "exhausted",
"last_status_at": now,
"last_error_code": 429,
"last_error_reset_at": now + 3600,
}
)
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, store)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
assert clear_codex_pool_quota_cooldowns("tok-other") == 1
persisted = json.loads((hermes_home / "auth.json").read_text())
entries = {e["id"]: e for e in persisted["credential_pool"]["openai-codex"]}
assert entries["cred-quota-2"]["last_status"] is None
assert entries["cred-quota"]["last_status"] == "exhausted"
def test_clear_cooldowns_noop_without_pool(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, {"version": 1, "providers": {}})
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
assert clear_codex_pool_quota_cooldowns() == 0
# ---------------------------------------------------------------------------
# resolve_codex_runtime_credentials — stale cooldown lifted by live probe
# ---------------------------------------------------------------------------
def _pool_only_rate_limited_store(now=None):
now = now or time.time()
return {
"version": 1,
"providers": {},
"credential_pool": {
"openai-codex": [
{
"id": "cred-quota",
"label": "quota-frozen",
"auth_type": "oauth",
"priority": 0,
"source": "device_code",
"access_token": "tok-quota",
"last_status": "exhausted",
"last_status_at": now,
"last_error_code": 429,
"last_error_reason": "usage_limit_reached",
"last_error_message": "The usage limit has been reached",
"last_error_reset_at": now + 3 * 24 * 3600,
}
]
},
}
def test_resolver_recovers_when_probe_confirms_reset(tmp_path, monkeypatch):
"""The screenshot bug: pool-only cooldown raises `quota exhausted (429);
retry after Ns` even though the upstream window already reset. A positive
probe must clear the cooldown and return the pool credential."""
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, _pool_only_rate_limited_store())
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(
auth_mod, "_probe_codex_quota_restored", lambda token, **kw: True
)
resolved = resolve_codex_runtime_credentials()
assert resolved["api_key"] == "tok-quota"
assert resolved["source"] == "credential_pool"
store = json.loads((hermes_home / "auth.json").read_text())
entry = store["credential_pool"]["openai-codex"][0]
assert entry["last_status"] is None
assert entry["last_error_reset_at"] is None
def test_resolver_keeps_cooldown_when_probe_negative(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, _pool_only_rate_limited_store())
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(
auth_mod, "_probe_codex_quota_restored", lambda token, **kw: False
)
with pytest.raises(AuthError) as exc:
resolve_codex_runtime_credentials()
assert exc.value.code == auth_mod.CODEX_RATE_LIMITED_CODE
assert "retry after" in str(exc.value)
def test_resolver_keeps_cooldown_when_probe_indeterminate(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, _pool_only_rate_limited_store())
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(
auth_mod, "_probe_codex_quota_restored", lambda token, **kw: None
)
with pytest.raises(AuthError) as exc:
resolve_codex_runtime_credentials()
assert exc.value.code == auth_mod.CODEX_RATE_LIMITED_CODE
# ---------------------------------------------------------------------------
# CredentialPool._available_entries — frozen entry recovers via probe
# ---------------------------------------------------------------------------
def test_pool_entry_recovers_when_probe_confirms_reset(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
_write_auth_store(tmp_path / "hermes", _pool_only_rate_limited_store())
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
monkeypatch.setattr(
auth_mod, "_probe_codex_quota_restored", lambda token, **kw: True
)
available = pool._available_entries(clear_expired=True, refresh=False)
assert len(available) == 1
assert available[0].last_status == "ok"
assert available[0].last_error_reset_at is None
def test_pool_entry_stays_frozen_when_probe_negative(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
_write_auth_store(tmp_path / "hermes", _pool_only_rate_limited_store())
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
monkeypatch.setattr(
auth_mod, "_probe_codex_quota_restored", lambda token, **kw: False
)
assert pool._available_entries(clear_expired=True, refresh=False) == []
def test_pool_probe_not_fired_for_non_quota_exhaustion(tmp_path, monkeypatch):
"""Entries frozen by auth-shaped failures must not trigger the probe."""
now = time.time()
store = _pool_only_rate_limited_store(now)
entry = store["credential_pool"]["openai-codex"][0]
entry["last_error_code"] = 401
entry["last_error_reason"] = "token_expired"
entry["last_error_message"] = "expired"
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
_write_auth_store(tmp_path / "hermes", store)
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
probes = []
def _spy(token, **kw):
probes.append(token)
return True
monkeypatch.setattr(auth_mod, "_probe_codex_quota_restored", _spy)
pool._available_entries(clear_expired=True, refresh=False)
assert probes == []
def test_pool_readonly_enumeration_does_not_probe(tmp_path, monkeypatch):
"""clear_expired=False callers (read-only listing) must not fire probes
or mutate persisted state."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
_write_auth_store(tmp_path / "hermes", _pool_only_rate_limited_store())
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
probes = []
def _spy(token, **kw):
probes.append(token)
return True
monkeypatch.setattr(auth_mod, "_probe_codex_quota_restored", _spy)
assert pool._available_entries(clear_expired=False, refresh=False) == []
assert probes == []
# ---------------------------------------------------------------------------
# /usage reset redemption clears persisted pool cooldowns
# ---------------------------------------------------------------------------
def test_redeem_reset_clears_pool_cooldowns(tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
_write_auth_store(hermes_home, _pool_only_rate_limited_store())
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
from agent import account_usage
class _FakeResetClient:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
return False
def get(self, url, headers=None):
return _StubResponse(
200,
{
"rate_limit": {
"primary_window": {"used_percent": 100.0},
"secondary_window": {"used_percent": 40.0},
},
"rate_limit_reset_credits": {"available_count": 1},
},
)
def post(self, url, headers=None, json=None):
return _StubResponse(200, {"code": "reset", "windows_reset": 2})
monkeypatch.setattr(
account_usage.httpx, "Client", lambda **kwargs: _FakeResetClient(**kwargs)
)
result = account_usage.redeem_codex_reset_credit(
base_url="https://chatgpt.com/backend-api/codex",
api_key="live-agent-token",
)
assert result.redeemed
store = json.loads((hermes_home / "auth.json").read_text())
entry = store["credential_pool"]["openai-codex"][0]
assert entry["last_status"] is None
assert entry["last_error_reset_at"] is None