fix(providers): show pooled Codex quota status

This commit is contained in:
starship-s committed 2026-05-12 19:09:05 -06:00
1 parent 9268f411d8
commit b87548b68e
4 files changed
+875 -84

No files matched your search

+380 -42
View File
@@ -7,6 +7,8 @@ multi-provider support).
from __future__ import annotations
import base64
import hashlib
import json
import logging
import os
@@ -14,6 +16,7 @@ import signal
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
@@ -53,6 +56,8 @@ def _custom_provider_name_matches(provider_id: str, name: object) -> bool:
_OPENROUTER_KEY_URL = "https://openrouter.ai/api/v1/key"
_PROVIDER_QUOTA_TIMEOUT_SECONDS = 3.0
_ACCOUNT_USAGE_SUBPROCESS_TIMEOUT_SECONDS = 35.0
_ACCOUNT_USAGE_CACHE_TTL_SECONDS = 45.0
_ACCOUNT_USAGE_CACHE_MAX_ENTRIES = 64
_ACCOUNT_USAGE_PROVIDERS = frozenset({"openai-codex", "anthropic"})
# Upper bound on simultaneous profile-isolated quota probe subprocesses.
@@ -98,6 +103,14 @@ _ACCOUNT_USAGE_PARENT_DEATHSIG_BOOTSTRAP = (
# Lazily created so this module compiles even when threading isn't ready.
_account_usage_probe_semaphore: threading.BoundedSemaphore | None = None
# Short-lived account-usage cache. The Codex pooled probe may check multiple
# credentials, so cache the sanitized snapshot briefly to avoid re-querying the
# provider on every Settings repaint/profile-panel refresh. Pool composition
# changes can be stale for at most this TTL; that is preferred to hammering the
# provider usage API while the Settings panel is open.
_account_usage_status_cache: dict[tuple[str, str, str], tuple[float, Any]] = {}
_account_usage_status_cache_lock = threading.Lock()
def _get_account_usage_probe_semaphore() -> threading.BoundedSemaphore:
global _account_usage_probe_semaphore
@@ -136,6 +149,7 @@ from agent.account_usage import fetch_account_usage
_CODEX_DEFAULT_BASE_URL = "https://chatgpt.com/backend-api/codex"
_CODEX_POOL_USAGE_TIMEOUT_SECONDS = 4.0
def _iso(value):
@@ -159,7 +173,7 @@ def _snapshot_payload(snapshot):
"reset_at": _iso(getattr(window, "reset_at", None)),
"detail": getattr(window, "detail", None),
})
return {
payload = {
"provider": str(getattr(snapshot, "provider", "") or ""),
"source": str(getattr(snapshot, "source", "") or ""),
"title": str(getattr(snapshot, "title", "") or ""),
@@ -170,6 +184,10 @@ def _snapshot_payload(snapshot):
"unavailable_reason": getattr(snapshot, "unavailable_reason", None),
"fetched_at": _iso(getattr(snapshot, "fetched_at", None)),
}
pool = getattr(snapshot, "pool", None)
if isinstance(pool, dict):
payload["pool"] = pool
return payload
def _snapshot_available(snapshot):
@@ -317,25 +335,203 @@ def _codex_snapshot_from_usage_payload(payload):
)
def _snapshot_windows_payload(snapshot):
windows = []
for window in getattr(snapshot, "windows", ()) or ():
label = str(getattr(window, "label", "") or "").strip()
if not label:
continue
used_percent = _number(getattr(window, "used_percent", None))
remaining_percent = None
if used_percent is not None:
remaining_percent = max(0.0, min(100.0, 100.0 - float(used_percent)))
windows.append({
"label": label,
"used_percent": used_percent,
"remaining_percent": remaining_percent,
"reset_at": _iso(getattr(window, "reset_at", None)),
"detail": getattr(window, "detail", None),
})
return windows
def _snapshot_details_payload(snapshot):
return [
str(detail).strip()
for detail in (getattr(snapshot, "details", ()) or ())
if str(detail).strip()
]
def _safe_entry_label(entry, index):
label = _entry_value(entry, "label", "source") or ""
if not label:
label = "Credential " + str(index)
label = " ".join(str(label).split())
if len(label) > 64:
label = label[:61].rstrip() + "..."
return label
def _safe_unavailable_reason(reason):
text = " ".join(str(reason or "").split())
if not text:
return None
lowered = text.lower()
sensitive_terms = ("access_token", "refresh_token", "authorization", "bearer ", "jwt", "secret")
if any(term in lowered for term in sensitive_terms):
return "Usage unavailable for this credential."
return text[:180]
def _entry_is_pool_exhausted(entry):
return str(_entry_value(entry, "last_status") or "").strip().lower() == "exhausted"
def _fetch_codex_entry_snapshot(entry):
access_token = _entry_value(entry, "runtime_api_key", "access_token")
if not access_token:
return None, False, "No runtime token available."
base_url = _entry_value(entry, "runtime_base_url", "base_url") or _CODEX_DEFAULT_BASE_URL
request = urllib_request.Request(
_resolve_codex_usage_url(base_url),
headers=_codex_usage_headers(access_token),
)
with urllib_request.urlopen(request, timeout=_CODEX_POOL_USAGE_TIMEOUT_SECONDS) as response:
payload = json.loads(response.read().decode("utf-8") or "{}")
return _codex_snapshot_from_usage_payload(payload), True, None
def _best_remaining_by_window(rows):
best = {}
for row in rows:
if row.get("status") != "available":
continue
label = row.get("label") or "Credential"
for window in row.get("windows") or []:
if not isinstance(window, dict):
continue
window_label = str(window.get("label") or "").strip()
remaining = _number(window.get("remaining_percent"))
if not window_label or remaining is None:
continue
candidate = {
"label": window_label,
"remaining_percent": remaining,
"used_percent": window.get("used_percent"),
"reset_at": window.get("reset_at"),
"detail": window.get("detail"),
"credential_label": label,
}
current = best.get(window_label.lower())
if current is None or float(remaining) > float(current.get("remaining_percent") or -1):
best[window_label.lower()] = candidate
return list(best.values())
def _next_reset_at(rows):
best_dt = None
best_text = None
for row in rows:
for window in row.get("windows") or []:
if not isinstance(window, dict):
continue
reset_text = window.get("reset_at")
dt = _parse_dt(reset_text)
if dt is None:
continue
if best_dt is None or dt < best_dt:
best_dt = dt
best_text = _iso(dt)
return best_text
def _codex_pool_snapshot(entries, rows, queried):
available_rows = [row for row in rows if row.get("status") == "available"]
exhausted_rows = [row for row in rows if row.get("status") == "exhausted"]
failed_rows = [row for row in rows if row.get("status") not in {"available", "exhausted"}]
plans = []
for row in rows:
plan = row.get("plan")
if plan and plan not in plans:
plans.append(plan)
best_windows = _best_remaining_by_window(rows)
pool = {
"total_credentials": len(entries),
"queried_credentials": queried,
"available_credentials": len(available_rows),
"exhausted_credentials": len(exhausted_rows),
"failed_credentials": len(failed_rows),
"plans": plans,
"next_reset_at": _next_reset_at(rows),
"best_remaining_by_window": best_windows,
"credentials": rows,
}
details = [str(len(available_rows)) + "/" + str(len(entries)) + " credentials available"]
if exhausted_rows:
details.append(str(len(exhausted_rows)) + " exhausted")
if failed_rows:
details.append(str(len(failed_rows)) + " failed to load")
if plans:
details.append("Plans: " + ", ".join(plans))
plan = plans[0] if len(plans) == 1 else None
windows = tuple(
SimpleNamespace(
label=window.get("label"),
used_percent=window.get("used_percent"),
reset_at=window.get("reset_at"),
detail="Best of " + str(len(available_rows)) + " available credentials",
)
for window in best_windows
)
return SimpleNamespace(
provider="openai-codex",
source="usage_api_pool",
title="Account limits",
plan=plan,
windows=windows,
details=tuple(details),
available=bool(available_rows),
unavailable_reason=None if available_rows else "No Codex pool credentials returned available account limits.",
fetched_at=datetime.now(timezone.utc),
pool=pool,
)
def _fetch_codex_account_usage_from_pool():
try:
from agent.credential_pool import load_pool
pool = load_pool("openai-codex")
entry = pool.select() if pool is not None else None
if entry is None:
entries = pool.entries() if pool is not None and hasattr(pool, "entries") else []
if not entries:
return None
access_token = _entry_value(entry, "runtime_api_key", "access_token")
if not access_token:
return None
base_url = _entry_value(entry, "runtime_base_url", "base_url") or _CODEX_DEFAULT_BASE_URL
request = urllib_request.Request(
_resolve_codex_usage_url(base_url),
headers=_codex_usage_headers(access_token),
)
with urllib_request.urlopen(request, timeout=15.0) as response:
payload = json.loads(response.read().decode("utf-8") or "{}")
return _codex_snapshot_from_usage_payload(payload)
rows = []
queried = 0
for index, entry in enumerate(entries, start=1):
label = _safe_entry_label(entry, index)
pool_exhausted = _entry_is_pool_exhausted(entry)
try:
snapshot, did_query, reason = _fetch_codex_entry_snapshot(entry)
if did_query:
queried += 1
except Exception as exc:
snapshot = None
reason = str(exc)
windows = _snapshot_windows_payload(snapshot) if snapshot is not None else []
details = _snapshot_details_payload(snapshot) if snapshot is not None else []
snapshot_available = _snapshot_available(snapshot)
status = "available" if snapshot_available and not pool_exhausted else "exhausted" if pool_exhausted else "unavailable"
rows.append({
"label": label,
"status": status,
"plan": getattr(snapshot, "plan", None) if snapshot is not None else None,
"windows": windows,
"details": details,
"unavailable_reason": None if snapshot_available else _safe_unavailable_reason(reason or getattr(snapshot, "unavailable_reason", None)),
"fetched_at": _iso(getattr(snapshot, "fetched_at", None)) if snapshot is not None else None,
})
return _codex_pool_snapshot(entries, rows, queried)
except Exception:
return None
@@ -346,10 +542,12 @@ try:
snapshot = fetch_account_usage(provider, api_key=api_key)
except Exception:
snapshot = None
if str(provider or "").strip().lower() == "openai-codex" and not _snapshot_available(snapshot):
fallback_snapshot = _fetch_codex_account_usage_from_pool()
if _snapshot_available(fallback_snapshot) or snapshot is None:
snapshot = fallback_snapshot
if str(provider or "").strip().lower() == "openai-codex":
pool_snapshot = _fetch_codex_account_usage_from_pool()
if isinstance(getattr(pool_snapshot, "pool", None), dict):
snapshot = pool_snapshot
elif not _snapshot_available(snapshot) and _snapshot_available(pool_snapshot):
snapshot = pool_snapshot
print(json.dumps(_snapshot_payload(snapshot)))
"""
@@ -447,6 +645,85 @@ def _load_env_file(env_path: Path) -> dict[str, str]:
return values
def _decode_jwt_claims_unverified(token: str) -> dict[str, Any]:
if not isinstance(token, str) or token.count(".") != 2:
return {}
payload = token.split(".", 2)[1]
payload += "=" * ((4 - len(payload) % 4) % 4)
try:
claims = json.loads(base64.urlsafe_b64decode(payload.encode("utf-8")).decode("utf-8"))
except Exception:
return {}
return claims if isinstance(claims, dict) else {}
def _looks_like_codex_oauth_token(value: str) -> bool:
"""Return True when a value is a ChatGPT/Codex OAuth JWT, not an OpenAI API key."""
token = str(value or "").strip()
if not token or token.startswith("sk-"):
return False
claims = _decode_jwt_claims_unverified(token)
if not claims:
return False
auth_claim = claims.get("https://api.openai.com/auth")
if isinstance(auth_claim, dict) and auth_claim:
return True
return any(key in claims for key in ("chatgpt_account_id", "https://api.openai.com/profile"))
def _provider_value_counts_as_api_key(provider_id: str, value: object) -> bool:
text = str(value or "").strip()
if not text:
return False
if (provider_id or "").strip().lower() == "openai" and _looks_like_codex_oauth_token(text):
return False
return True
def _provider_has_shadowed_codex_oauth_value(provider_id: str) -> bool:
"""True when the bare OpenAI credential slot contains only a Codex OAuth JWT.
Users who authenticate Codex can end up with a ChatGPT/Codex JWT in a
legacy OPENAI_API_KEY-shaped location. That value should not make the bare
OpenAI API provider appear as configured, and the Providers tab should not
show an extra OpenAI card solely because of that Codex-only credential.
"""
if (provider_id or "").strip().lower() != "openai":
return False
values: list[object] = []
env_var = _PROVIDER_ENV_VAR.get(provider_id)
if env_var:
env_path = _get_hermes_home() / ".env"
env_values = _load_env_file(env_path)
values.append(env_values.get(env_var))
values.append(os.getenv(env_var))
for alias in _PROVIDER_ENV_VAR_ALIASES.get(provider_id, ()) or ():
values.append(env_values.get(alias))
values.append(os.getenv(alias))
cfg = get_config()
model_cfg = cfg.get("model", {})
if isinstance(model_cfg, dict):
active_provider = str(model_cfg.get("provider") or "").strip().lower()
if active_provider == provider_id:
values.append(model_cfg.get("api_key"))
providers_cfg = cfg.get("providers", {})
if isinstance(providers_cfg, dict):
provider_cfg = providers_cfg.get(provider_id, {})
if isinstance(provider_cfg, dict):
values.append(provider_cfg.get("api_key"))
custom_providers = cfg.get("custom_providers", [])
if isinstance(custom_providers, list):
for cp in custom_providers:
if isinstance(cp, dict) and _custom_provider_name_matches(provider_id, cp.get("name")):
cp_key = cp.get("api_key")
if isinstance(cp_key, str) and cp_key.startswith("${") and cp_key.endswith("}"):
values.append(os.getenv(cp_key[2:-1]))
else:
values.append(cp_key)
return any(_looks_like_codex_oauth_token(str(value or "")) for value in values)
def _write_env_file(env_path: Path, updates: dict[str, str | None]) -> None:
"""Write key=value pairs to the .env file.
@@ -559,17 +836,19 @@ def _provider_has_key(provider_id: str) -> bool:
if env_var:
env_path = _get_hermes_home() / ".env"
env_values = _load_env_file(env_path)
if env_values.get(env_var):
env_file_value = env_values.get(env_var)
if _provider_value_counts_as_api_key(provider_id, env_file_value):
return True
if os.getenv(env_var):
env_value = os.getenv(env_var)
if _provider_value_counts_as_api_key(provider_id, env_value):
return True
# Fall back to legacy env-var aliases (e.g. lmstudio's pre-#1500
# LMSTUDIO_API_KEY name) so existing users don't lose detection
# after an env-var rename. See _PROVIDER_ENV_VAR_ALIASES.
for alias in _PROVIDER_ENV_VAR_ALIASES.get(provider_id, ()) or ():
if env_values.get(alias):
if _provider_value_counts_as_api_key(provider_id, env_values.get(alias)):
return True
if os.getenv(alias):
if _provider_value_counts_as_api_key(provider_id, os.getenv(alias)):
return True
cfg = get_config()
@@ -580,20 +859,22 @@ def _provider_has_key(provider_id: str) -> bool:
if isinstance(model_cfg, dict) and str(model_cfg.get("api_key") or "").strip():
active_provider = model_cfg.get("provider")
if active_provider and str(active_provider).strip().lower() == provider_id.lower():
return True
if _provider_value_counts_as_api_key(provider_id, model_cfg.get("api_key")):
return True
# Check providers.<id>.api_key
providers_cfg = cfg.get("providers", {})
if isinstance(providers_cfg, dict):
provider_cfg = providers_cfg.get(provider_id, {})
if isinstance(provider_cfg, dict) and str(provider_cfg.get("api_key") or "").strip():
return True
if _provider_value_counts_as_api_key(provider_id, provider_cfg.get("api_key")):
return True
# Check custom_providers
custom_providers = cfg.get("custom_providers", [])
if isinstance(custom_providers, list):
for cp in custom_providers:
if isinstance(cp, dict):
if _custom_provider_name_matches(provider_id, cp.get("name")):
if str(cp.get("api_key") or "").strip():
if _provider_value_counts_as_api_key(provider_id, cp.get("api_key")):
return True
return False
@@ -605,22 +886,26 @@ def _get_provider_api_key(provider_id: str) -> str | None:
if env_var:
env_path = _get_hermes_home() / ".env"
env_values = _load_env_file(env_path)
if env_values.get(env_var):
return str(env_values[env_var]).strip() or None
if os.getenv(env_var):
return os.getenv(env_var, "").strip() or None
env_file_value = env_values.get(env_var)
if _provider_value_counts_as_api_key(provider_id, env_file_value):
return str(env_file_value).strip() or None
env_value = os.getenv(env_var)
if _provider_value_counts_as_api_key(provider_id, env_value):
return str(env_value).strip() or None
for alias in _PROVIDER_ENV_VAR_ALIASES.get(provider_id, ()) or ():
if env_values.get(alias):
return str(env_values[alias]).strip() or None
if os.getenv(alias):
return os.getenv(alias, "").strip() or None
alias_file_value = env_values.get(alias)
if _provider_value_counts_as_api_key(provider_id, alias_file_value):
return str(alias_file_value).strip() or None
alias_value = os.getenv(alias)
if _provider_value_counts_as_api_key(provider_id, alias_value):
return str(alias_value).strip() or None
cfg = get_config()
model_cfg = cfg.get("model", {})
if isinstance(model_cfg, dict):
active_provider = str(model_cfg.get("provider") or "").strip().lower()
model_key = str(model_cfg.get("api_key") or "").strip()
if model_key and active_provider == provider_id:
if model_key and active_provider == provider_id and _provider_value_counts_as_api_key(provider_id, model_key):
return model_key
providers_cfg = cfg.get("providers", {})
@@ -628,7 +913,7 @@ def _get_provider_api_key(provider_id: str) -> str | None:
provider_cfg = providers_cfg.get(provider_id, {})
if isinstance(provider_cfg, dict):
provider_key = str(provider_cfg.get("api_key") or "").strip()
if provider_key:
if _provider_value_counts_as_api_key(provider_id, provider_key):
return provider_key
custom_providers = cfg.get("custom_providers", [])
@@ -640,7 +925,7 @@ def _get_provider_api_key(provider_id: str) -> str | None:
cp_key = str(cp.get("api_key") or "").strip()
if cp_key.startswith("${") and cp_key.endswith("}"):
return os.getenv(cp_key[2:-1], "").strip() or None
if cp_key:
if _provider_value_counts_as_api_key(provider_id, cp_key):
return cp_key
return None
@@ -718,7 +1003,7 @@ def _serialize_account_usage_snapshot(snapshot: Any) -> dict[str, Any] | None:
]
plan = str(getattr(snapshot, "plan", "") or "").strip() or None
unavailable_reason = str(getattr(snapshot, "unavailable_reason", "") or "").strip() or None
return {
result = {
"provider": str(getattr(snapshot, "provider", "") or "").strip() or None,
"source": str(getattr(snapshot, "source", "") or "").strip() or None,
"title": str(getattr(snapshot, "title", "") or "").strip() or "Account limits",
@@ -729,6 +1014,10 @@ def _serialize_account_usage_snapshot(snapshot: Any) -> dict[str, Any] | None:
"unavailable_reason": unavailable_reason,
"fetched_at": _isoformat_utc(getattr(snapshot, "fetched_at", None)),
}
pool = getattr(snapshot, "pool", None)
if isinstance(pool, dict):
result["pool"] = pool
return result
def _agent_fetch_account_usage(provider: str, *, base_url: str | None = None, api_key: str | None = None) -> Any:
@@ -791,9 +1080,48 @@ def _account_usage_payload_to_snapshot(payload: Any) -> Any:
available=bool(payload.get("available")),
unavailable_reason=payload.get("unavailable_reason"),
fetched_at=payload.get("fetched_at"),
pool=payload.get("pool") if isinstance(payload.get("pool"), dict) else None,
)
def _account_usage_cache_key(provider: str, home: Path, api_key: str | None) -> tuple[str, str, str]:
key_fingerprint = ""
if api_key:
key_fingerprint = hashlib.sha256(api_key.encode("utf-8", "ignore")).hexdigest()
return ((provider or "").strip().lower(), str(Path(home)), key_fingerprint)
def _get_cached_account_usage(cache_key: tuple[str, str, str]) -> tuple[bool, Any]:
now = time.monotonic()
with _account_usage_status_cache_lock:
cached = _account_usage_status_cache.get(cache_key)
if cached is None:
return False, None
fetched_at, snapshot = cached
if now - fetched_at <= _ACCOUNT_USAGE_CACHE_TTL_SECONDS:
return True, snapshot
_account_usage_status_cache.pop(cache_key, None)
return False, None
def _set_cached_account_usage(cache_key: tuple[str, str, str], snapshot: Any) -> None:
now = time.monotonic()
with _account_usage_status_cache_lock:
_account_usage_status_cache[cache_key] = (now, snapshot)
expired = [
key for key, (fetched_at, _snapshot) in _account_usage_status_cache.items()
if now - fetched_at > _ACCOUNT_USAGE_CACHE_TTL_SECONDS
]
for key in expired:
_account_usage_status_cache.pop(key, None)
while len(_account_usage_status_cache) > _ACCOUNT_USAGE_CACHE_MAX_ENTRIES:
oldest_key = min(
_account_usage_status_cache,
key=lambda key: _account_usage_status_cache[key][0],
)
_account_usage_status_cache.pop(oldest_key, None)
def _agent_fetch_account_usage_for_home(provider: str, home: Path, *, api_key: str | None = None) -> Any:
try:
from api.config import PYTHON_EXE
@@ -856,16 +1184,23 @@ def _fetch_account_usage_with_profile_context(provider: str) -> Any:
"""
home = _get_hermes_home()
api_key = _get_provider_api_key(provider)
cache_key = _account_usage_cache_key(provider, home, api_key)
cache_hit, cached = _get_cached_account_usage(cache_key)
if cache_hit:
return cached
sem = _get_account_usage_probe_semaphore()
try:
with sem:
return _agent_fetch_account_usage_for_home(
snapshot = _agent_fetch_account_usage_for_home(
provider,
home,
api_key=api_key,
)
_set_cached_account_usage(cache_key, snapshot)
return snapshot
except Exception:
logger.debug("Failed to fetch account usage for %s", provider, exc_info=True)
_set_cached_account_usage(cache_key, None)
return None
@@ -1075,9 +1410,9 @@ def get_providers() -> dict[str, Any]:
if env_var:
env_path = _get_hermes_home() / ".env"
env_values = _load_env_file(env_path)
if env_values.get(env_var):
if _provider_value_counts_as_api_key(pid, env_values.get(env_var)):
key_source = "env_file"
elif os.getenv(env_var):
elif _provider_value_counts_as_api_key(pid, os.getenv(env_var)):
key_source = "env_var"
else:
# Canonical name not set; check legacy aliases (e.g. lmstudio's
@@ -1086,11 +1421,11 @@ def get_providers() -> dict[str, Any]:
# actually lives in .env under the old name.
aliased = False
for alias in _PROVIDER_ENV_VAR_ALIASES.get(pid, ()) or ():
if env_values.get(alias):
if _provider_value_counts_as_api_key(pid, env_values.get(alias)):
key_source = "env_file"
aliased = True
break
if os.getenv(alias):
if _provider_value_counts_as_api_key(pid, os.getenv(alias)):
key_source = "env_var"
aliased = True
break
@@ -1124,6 +1459,9 @@ def get_providers() -> dict[str, Any]:
except Exception:
pass
if pid == "openai" and not has_key and _provider_has_shadowed_codex_oauth_value(pid):
continue
models = list(_PROVIDER_MODELS.get(pid, []))
models_total = len(models)
# OpenAI Codex account catalogs drift independently from WebUI releases.
+79 -3
View File
@@ -5593,6 +5593,74 @@ function _formatProviderQuotaLastChecked(status){
try{return 'Last checked '+d.toLocaleString();}catch(e){return 'Last checked '+value;}
}
function _providerQuotaStateClass(value){
return String(value||'unavailable').replace(/[^a-z0-9_-]/gi,'').toLowerCase()||'unavailable';
}
function _providerQuotaPoolShouldDefaultOpen(pool){
try{
const saved=localStorage.getItem('hermes-provider-quota-pool-open');
if(saved==='1') return true;
if(saved==='0') return false;
}catch(e){}
const count=Array.isArray(pool&&pool.credentials)?pool.credentials.length:0;
return count>0&&count<=3;
}
function _buildProviderQuotaPoolBreakdown(accountLimits){
const pool=accountLimits&&accountLimits.pool;
if(!pool||!Array.isArray(pool.credentials)||pool.credentials.length===0) return '';
const defaultOpen=_providerQuotaPoolShouldDefaultOpen(pool);
const total=Number.isFinite(Number(pool.total_credentials))?Number(pool.total_credentials):pool.credentials.length;
const available=Number.isFinite(Number(pool.available_credentials))?Number(pool.available_credentials):pool.credentials.filter(c=>c&&c.status==='available').length;
const exhausted=Number.isFinite(Number(pool.exhausted_credentials))?Number(pool.exhausted_credentials):0;
const failed=Number.isFinite(Number(pool.failed_credentials))?Number(pool.failed_credentials):0;
const queried=Number.isFinite(Number(pool.queried_credentials))?Number(pool.queried_credentials):0;
const summaryParts=[`${available}/${total} available`];
if(exhausted>0) summaryParts.push(`${exhausted} exhausted`);
if(failed>0) summaryParts.push(`${failed} failed`);
if(queried>0) summaryParts.push(`${queried} checked`);
const planParts=Array.isArray(pool.plans)?pool.plans.filter(Boolean):[];
const rows=pool.credentials.map((credential,idx)=>{
const label=(credential&&credential.label)||`Credential ${idx+1}`;
const status=_providerQuotaStateClass(credential&&credential.status);
const statusText=status.replace(/_/g,' ');
const plan=credential&&credential.plan?` · ${credential.plan}`:'';
const windows=Array.isArray(credential&&credential.windows)?credential.windows:[];
const details=Array.isArray(credential&&credential.details)?credential.details.filter(Boolean):[];
const unavailableReason=(credential&&credential.unavailable_reason)||'';
const windowHtml=windows.length?windows.map(w=>{
const remaining=_formatProviderQuotaPercent(w&&w.remaining_percent);
const used=_formatProviderQuotaPercent(w&&w.used_percent);
const reset=_formatProviderQuotaReset(w&&w.reset_at);
const meta=[];
if(used!=='—') meta.push(`${used} used`);
if(reset) meta.push(`resets ${reset}`);
const detail=(w&&w.detail)?String(w.detail).trim():'';
return `<div class="provider-quota-pool-window"><span>${esc(_formatProviderQuotaWindowLabel(accountLimits,w))}</span><strong>${esc(remaining)}</strong>${meta.length?`<small>${esc(meta.join(' · '))}</small>`:''}${detail?`<small class="provider-quota-window-detail">${esc(detail)}</small>`:''}</div>`;
}).join(''):`<div class="provider-quota-pool-note">${esc(unavailableReason||'No account-limit windows reported.')}</div>`;
const detailHtml=details.length?`<div class="provider-quota-pool-details">${details.map(d=>`<span>${esc(d)}</span>`).join('')}</div>`:'';
return `
<div class="provider-quota-pool-row provider-quota-pool-row-${status}">
<div class="provider-quota-pool-row-head">
<span>${esc(label)}${esc(plan)}</span>
<strong>${esc(statusText)}</strong>
</div>
<div class="provider-quota-pool-windows">${windowHtml}</div>
${detailHtml}
</div>
`;
}).join('');
const planText=planParts.length?`<div class="provider-quota-pool-plans">Plans: ${esc(planParts.join(', '))}</div>`:'';
return `
<details class="provider-quota-pool"${defaultOpen?' open':''}>
<summary><span>Credential pool</span><strong>${esc(summaryParts.join(' · '))}</strong></summary>
${planText}
<div class="provider-quota-pool-rows">${rows}</div>
</details>
`;
}
function _buildProviderQuotaCard(status){
if(!status) return null;
const card=document.createElement('div');
@@ -5603,7 +5671,7 @@ function _buildProviderQuotaCard(status){
const provider=(accountLimits&&accountLimits.plan)?`${providerBase} · ${accountLimits.plan}`:providerBase;
const quota=status.quota||null;
let body='';
if(status.status==='available'&&accountLimits){
if(accountLimits&&(status.status==='available'||accountLimits.pool)){
const windows=Array.isArray(accountLimits.windows)?accountLimits.windows:[];
const details=Array.isArray(accountLimits.details)?accountLimits.details:[];
const windowHtml=windows.map(w=>{
@@ -5612,19 +5680,21 @@ function _buildProviderQuotaCard(status){
const meta=[];
if(used!=='—') meta.push(`${used} used`);
if(reset) meta.push(`resets ${reset}`);
if(w&&w.detail) meta.push(w.detail);
const detail=(w&&w.detail)?String(w.detail).trim():'';
return `
<div class="provider-quota-metric provider-quota-window">
<span>${esc(_formatProviderQuotaWindowLabel(accountLimits,w))}</span>
<strong>${esc(_formatProviderQuotaPercent(w&&w.remaining_percent))}</strong>
${meta.length?`<small>${esc(meta.join(' · '))}</small>`:''}
${detail?`<small class="provider-quota-window-detail">${esc(detail)}</small>`:''}
</div>
`;
}).join('');
const detailHtml=details.length
? `<div class="provider-quota-details">${details.map(d=>`<span>${esc(d)}</span>`).join('')}</div>`
: '';
body=windowHtml+detailHtml;
const poolHtml=_buildProviderQuotaPoolBreakdown(accountLimits);
body=windowHtml+detailHtml+poolHtml;
if(!body) body=`<div class="provider-quota-message">${esc(status.message||'Account limits loaded.')}</div>`;
}else if(status.status==='available'&&quota){
body=`
@@ -5651,6 +5721,12 @@ function _buildProviderQuotaCard(status){
`;
const refreshBtn=card.querySelector('[data-provider-quota-refresh]');
if(refreshBtn) refreshBtn.addEventListener('click',()=>_refreshProviderQuota(card,refreshBtn));
const poolDetails=card.querySelector('.provider-quota-pool');
if(poolDetails){
poolDetails.addEventListener('toggle',()=>{
try{localStorage.setItem('hermes-provider-quota-pool-open',poolDetails.open?'1':'0');}catch(e){}
});
}
return card;
}
+40 -1
View File
@@ -2490,6 +2490,7 @@ main.main.showing-logs > #mainLogs{display:flex;}
background:linear-gradient(180deg,var(--surface),var(--sidebar));
padding:12px 16px;
margin-bottom:12px;
min-width:0;
}
.provider-quota-header{display:flex;align-items:flex-start;justify-content:space-between;gap:12px;margin-bottom:10px;}
.provider-quota-title{font-size:13px;font-weight:650;color:var(--text);line-height:1.2;}
@@ -2500,19 +2501,57 @@ main.main.showing-logs > #mainLogs{display:flex;}
.provider-quota-refresh{appearance:none;border:1px solid var(--border);border-radius:999px;background:var(--sidebar);color:var(--text);font-size:11px;font-weight:650;line-height:1;padding:5px 9px;cursor:pointer;white-space:nowrap;}
.provider-quota-refresh:hover:not(:disabled){border-color:var(--accent);color:var(--accent);}
.provider-quota-refresh:disabled{opacity:.65;cursor:wait;}
.provider-quota-body{display:flex;flex-wrap:wrap;gap:8px;}
.provider-quota-body{display:flex;flex-wrap:wrap;gap:8px;min-width:0;}
.provider-quota-metric{flex:1;min-width:88px;border:1px solid var(--border);border-radius:8px;background:var(--sidebar);padding:8px 10px;}
.provider-quota-metric span{display:block;font-size:10.5px;color:var(--muted);margin-bottom:2px;}
.provider-quota-metric strong{display:block;font-size:14px;color:var(--text);font-weight:650;}
.provider-quota-metric small{display:block;font-size:10.5px;color:var(--muted);line-height:1.35;margin-top:3px;}
.provider-quota-window-detail{opacity:.9;}
.provider-quota-window{min-width:132px;}
.provider-quota-details{display:flex;flex-wrap:wrap;gap:6px;width:100%;}
.provider-quota-details span{font-size:11px;line-height:1.35;color:var(--muted);border:1px solid var(--border);border-radius:999px;background:var(--sidebar);padding:4px 8px;}
.provider-quota-message{font-size:12px;color:var(--muted);line-height:1.45;}
.provider-quota-pool{width:100%;min-width:0;border:1px solid var(--border);border-radius:10px;background:var(--sidebar);overflow:hidden;}
.provider-quota-pool summary{display:flex;align-items:center;justify-content:space-between;gap:10px;cursor:pointer;list-style:none;padding:9px 10px;font-size:12px;color:var(--text);font-weight:650;}
.provider-quota-pool summary::-webkit-details-marker{display:none;}
.provider-quota-pool summary strong{font-size:11px;color:var(--muted);font-weight:600;text-align:right;}
.provider-quota-pool-plans{padding:0 10px 8px;font-size:11px;color:var(--muted);}
.provider-quota-pool-rows{display:flex;flex-direction:column;gap:8px;padding:0 10px 10px;min-width:0;}
.provider-quota-pool-row{border:1px solid var(--border);border-radius:8px;background:var(--surface);padding:8px;min-width:0;}
.provider-quota-pool-row-head{display:flex;align-items:center;justify-content:space-between;gap:8px;margin-bottom:7px;font-size:12px;line-height:1.3;}
.provider-quota-pool-row-head span{min-width:0;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;color:var(--text);}
.provider-quota-pool-row-head strong{font-size:10px;text-transform:capitalize;border-radius:999px;padding:2px 7px;background:var(--accent-bg);color:var(--accent-text);white-space:nowrap;}
.provider-quota-pool-row-available .provider-quota-pool-row-head strong{background:rgba(34,197,94,.12);color:#16a34a;}
:root.dark .provider-quota-pool-row-available .provider-quota-pool-row-head strong{background:rgba(34,197,94,.16);color:#4ade80;}
.provider-quota-pool-row-exhausted .provider-quota-pool-row-head strong{background:rgba(234,179,8,.12);color:var(--warning);}
.provider-quota-pool-windows{display:grid;grid-template-columns:repeat(auto-fit,minmax(118px,1fr));gap:6px;}
.provider-quota-pool-window{min-width:0;border:1px solid var(--border);border-radius:7px;background:var(--sidebar);padding:7px 8px;}
.provider-quota-pool-window span{display:block;font-size:10.5px;color:var(--muted);margin-bottom:2px;}
.provider-quota-pool-window strong{display:block;font-size:13px;color:var(--text);font-weight:650;}
.provider-quota-pool-window small{display:block;font-size:10px;color:var(--muted);line-height:1.3;margin-top:2px;}
.provider-quota-pool-note{font-size:11px;color:var(--muted);line-height:1.4;}
.provider-quota-pool-details{display:flex;flex-wrap:wrap;gap:5px;margin-top:7px;}
.provider-quota-pool-details span{font-size:10.5px;line-height:1.3;color:var(--muted);border:1px solid var(--border);border-radius:999px;background:var(--sidebar);padding:3px 7px;}
.provider-quota-card-available .provider-quota-badge{background:rgba(34,197,94,.12);color:#16a34a;}
:root.dark .provider-quota-card-available .provider-quota-badge{background:rgba(34,197,94,.16);color:#4ade80;}
.provider-quota-card-no_key .provider-quota-badge,.provider-quota-card-unsupported .provider-quota-badge{background:rgba(234,179,8,.12);color:var(--warning);}
.provider-quota-card-invalid_key .provider-quota-badge{background:color-mix(in srgb,var(--error) 12%,transparent);color:var(--error);}
@media (max-width: 520px){
.provider-quota-card{padding:12px;}
.provider-quota-header{flex-wrap:wrap;}
.provider-quota-body{flex-direction:column;}
.provider-quota-metric{width:100%;min-width:0;flex:none;}
.provider-quota-window{min-width:0;}
.provider-quota-pool summary{align-items:flex-start;flex-direction:column;gap:4px;}
.provider-quota-pool summary strong{text-align:left;}
.provider-quota-pool-rows{padding:0 8px 8px;}
.provider-quota-pool-row{padding:8px;}
.provider-quota-pool-row-head{align-items:flex-start;flex-wrap:wrap;}
.provider-quota-metric small,.provider-quota-pool-window small{overflow-wrap:anywhere;}
.provider-quota-details span,.provider-quota-pool-details span{max-width:100%;white-space:normal;}
.provider-quota-pool-row-head span{white-space:normal;}
.provider-quota-pool-windows{grid-template-columns:1fr;}
}
.provider-card{
border:1px solid var(--border);
border-radius:12px;
+376 -38
View File
@@ -278,27 +278,31 @@ def test_codex_account_usage_unavailable_is_sanitized(monkeypatch, tmp_path):
assert "secret" not in repr(result).lower()
def test_codex_account_usage_subprocess_falls_back_to_credential_pool(monkeypatch, capsys):
"""Codex quota probes should use credential_pool credentials when legacy auth misses."""
def test_codex_account_usage_subprocess_reports_read_only_credential_pool(monkeypatch, capsys):
"""Codex quota probes should inspect pool entries without mutating selection order."""
import api.providers as providers
def b64url(payload: bytes) -> str:
return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii")
token = ".".join((
b64url(b'{"alg":"none","typ":"JWT"}'),
b64url(json.dumps({
"https://api.openai.com/auth": {
"chatgpt_account_id": "acct-test-123",
},
}).encode("utf-8")),
b64url(b"signature"),
))
def token_for(account_id: str) -> str:
return ".".join((
b64url(b'{"alg":"none","typ":"JWT"}'),
b64url(json.dumps({
"https://api.openai.com/auth": {
"chatgpt_account_id": account_id,
},
}).encode("utf-8")),
b64url(b"signature"),
))
primary_token = token_for("acct-primary")
exhausted_token = token_for("acct-exhausted")
fetch_calls = []
load_pool_calls = []
selected = []
seen = {}
entries_called = []
seen = []
agent_mod = types.ModuleType("agent")
agent_mod.__path__ = []
@@ -310,25 +314,41 @@ def test_codex_account_usage_subprocess_falls_back_to_credential_pool(monkeypatc
return None
class FakePool:
def entries(self):
entries_called.append(True)
return [
SimpleNamespace(
label="Team primary",
runtime_api_key=primary_token,
runtime_base_url="https://chatgpt.com/backend-api/codex",
last_status=None,
),
SimpleNamespace(
label="Plus backup",
runtime_api_key=exhausted_token,
runtime_base_url="https://chatgpt.com/backend-api/codex",
last_status="exhausted",
),
]
def select(self):
selected.append(True)
return SimpleNamespace(
runtime_api_key=token,
runtime_base_url="https://chatgpt.com/backend-api/codex",
)
raise AssertionError("quota display must not rotate credential_pool selection")
def fake_load_pool(provider):
load_pool_calls.append(provider)
return FakePool()
def fake_urlopen(req, timeout):
seen["url"] = req.full_url
seen["timeout"] = timeout
seen["headers"] = {key.lower(): value for key, value in req.header_items()}
headers = {key.lower(): value for key, value in req.header_items()}
seen.append({
"url": req.full_url,
"timeout": timeout,
"headers": headers,
})
payload = {
"plan_type": "pro",
"plan_type": "pro" if headers.get("chatgpt-account-id") == "acct-primary" else "plus",
"rate_limit": {
"primary_window": {"used_percent": 15, "reset_at": 1_900_000_000},
"primary_window": {"used_percent": 15 if headers.get("chatgpt-account-id") == "acct-primary" else 95, "reset_at": 1_900_000_000},
"secondary_window": {"used_percent": 40, "reset_at": "2030-03-24T12:30:00Z"},
},
"credits": {"has_credits": True, "balance": 12.5},
@@ -350,24 +370,156 @@ def test_codex_account_usage_subprocess_falls_back_to_credential_pool(monkeypatc
assert fetch_calls == [("openai-codex", None, None)]
assert load_pool_calls == ["openai-codex"]
assert selected == [True]
assert seen["url"] == "https://chatgpt.com/backend-api/wham/usage"
assert seen["timeout"] == 15.0
headers = seen["headers"]
assert headers["authorization"] == f"Bearer {token}"
assert headers["accept"] == "application/json"
assert headers["originator"] == "codex_cli_rs"
assert headers["user-agent"].startswith("codex_cli_rs/")
assert headers["chatgpt-account-id"] == "acct-test-123"
assert entries_called == [True]
assert [call["url"] for call in seen] == [
"https://chatgpt.com/backend-api/wham/usage",
"https://chatgpt.com/backend-api/wham/usage",
]
assert [call["timeout"] for call in seen] == [4.0, 4.0]
assert seen[0]["headers"]["authorization"] == f"Bearer {primary_token}"
assert seen[0]["headers"]["chatgpt-account-id"] == "acct-primary"
assert seen[1]["headers"]["chatgpt-account-id"] == "acct-exhausted"
assert snapshot["provider"] == "openai-codex"
assert snapshot["source"] == "usage_api"
assert snapshot["plan"] == "Pro"
assert snapshot["source"] == "usage_api_pool"
assert snapshot["windows"][0]["label"] == "Session"
assert snapshot["windows"][0]["used_percent"] == 15.0
assert snapshot["windows"][1]["label"] == "Weekly"
assert snapshot["details"] == ["Credits balance: $12.50"]
assert snapshot["windows"][0]["used_percent"] == 15
assert snapshot["details"] == ["1/2 credentials available", "1 exhausted", "Plans: Pro, Plus"]
assert snapshot["available"] is True
assert token not in output
assert snapshot["pool"] == {
"total_credentials": 2,
"queried_credentials": 2,
"available_credentials": 1,
"exhausted_credentials": 1,
"failed_credentials": 0,
"plans": ["Pro", "Plus"],
"next_reset_at": "2030-03-17T17:46:40Z",
"best_remaining_by_window": [
{
"label": "Session",
"remaining_percent": 85.0,
"used_percent": 15.0,
"reset_at": "2030-03-17T17:46:40Z",
"detail": None,
"credential_label": "Team primary",
},
{
"label": "Weekly",
"remaining_percent": 60.0,
"used_percent": 40.0,
"reset_at": "2030-03-24T12:30:00Z",
"detail": None,
"credential_label": "Team primary",
},
],
"credentials": [
{
"label": "Team primary",
"status": "available",
"plan": "Pro",
"windows": [
{
"label": "Session",
"used_percent": 15.0,
"remaining_percent": 85.0,
"reset_at": "2030-03-17T17:46:40Z",
"detail": None,
},
{
"label": "Weekly",
"used_percent": 40.0,
"remaining_percent": 60.0,
"reset_at": "2030-03-24T12:30:00Z",
"detail": None,
},
],
"details": ["Credits balance: $12.50"],
"unavailable_reason": None,
"fetched_at": snapshot["pool"]["credentials"][0]["fetched_at"],
},
{
"label": "Plus backup",
"status": "exhausted",
"plan": "Plus",
"windows": [
{
"label": "Session",
"used_percent": 95.0,
"remaining_percent": 5.0,
"reset_at": "2030-03-17T17:46:40Z",
"detail": None,
},
{
"label": "Weekly",
"used_percent": 40.0,
"remaining_percent": 60.0,
"reset_at": "2030-03-24T12:30:00Z",
"detail": None,
},
],
"details": ["Credits balance: $12.50"],
"unavailable_reason": None,
"fetched_at": snapshot["pool"]["credentials"][1]["fetched_at"],
},
],
}
assert primary_token not in output
assert exhausted_token not in output
def test_codex_account_usage_subprocess_sanitizes_pool_entry_errors(monkeypatch, capsys):
"""Pool per-entry failures must not leak bearer/JWT-like exception text."""
import api.providers as providers
fetch_calls = []
agent_mod = types.ModuleType("agent")
agent_mod.__path__ = []
account_usage_mod = types.ModuleType("agent.account_usage")
credential_pool_mod = types.ModuleType("agent.credential_pool")
def fake_fetch_account_usage(provider, *, base_url=None, api_key=None):
fetch_calls.append((provider, base_url, api_key))
return None
class FakePool:
def entries(self):
return [
SimpleNamespace(
label="Bad token",
runtime_api_key="header.payload.signature",
runtime_base_url="https://chatgpt.com/backend-api/codex",
last_status=None,
),
]
def select(self):
raise AssertionError("quota display must not rotate credential_pool selection")
def fake_load_pool(provider):
return FakePool()
def fake_urlopen(_req, timeout):
raise RuntimeError("Bearer eyJsecret-token-like-value should not leak")
account_usage_mod.fetch_account_usage = fake_fetch_account_usage
credential_pool_mod.load_pool = fake_load_pool
monkeypatch.setitem(sys.modules, "agent", agent_mod)
monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod)
monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod)
monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""])
exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"})
output = capsys.readouterr().out.strip()
snapshot = json.loads(output)
assert fetch_calls == [("openai-codex", None, None)]
assert snapshot["available"] is False
assert snapshot["pool"]["failed_credentials"] == 1
assert snapshot["pool"]["credentials"][0]["unavailable_reason"] == "Usage unavailable for this credential."
assert "eyJsecret" not in output
assert "Bearer" not in output
def test_codex_account_usage_subprocess_keeps_legacy_reason_when_pool_misses(monkeypatch, capsys):
@@ -426,6 +578,40 @@ def test_codex_account_usage_subprocess_keeps_legacy_reason_when_pool_misses(mon
assert snapshot["fetched_at"] == "2030-03-17T12:30:00Z"
def test_account_usage_pool_payload_round_trips_to_provider_quota_status():
"""Parent process serialization must preserve pooled credential summaries."""
import api.providers as providers
payload = {
"provider": "openai-codex",
"source": "usage_api_pool",
"title": "Account limits",
"plan": None,
"windows": [
{"label": "Session", "used_percent": 25, "reset_at": "2030-03-17T17:30:00Z", "detail": "Best of 2"},
],
"details": ["2/3 credentials available"],
"available": True,
"unavailable_reason": None,
"fetched_at": "2030-03-17T12:30:00Z",
"pool": {
"total_credentials": 3,
"available_credentials": 2,
"exhausted_credentials": 1,
"failed_credentials": 0,
"credentials": [
{"label": "Credential 1", "status": "available", "windows": []},
],
},
}
snapshot = providers._account_usage_payload_to_snapshot(payload)
serialized = providers._serialize_account_usage_snapshot(snapshot)
assert serialized["windows"][0]["remaining_percent"] == 75.0
assert serialized["pool"] == payload["pool"]
def test_anthropic_oauth_usage_unavailable_reason_is_reported(monkeypatch, tmp_path):
"""Hermes Agent can report why account limits are not available."""
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
@@ -486,6 +672,71 @@ def test_account_usage_profile_env_is_child_scoped(monkeypatch, tmp_path):
assert os.environ["ANTHROPIC_API_KEY"] == "process-key"
def test_account_usage_profile_fetch_uses_short_lived_cache(monkeypatch, tmp_path):
"""Repeated Settings refreshes should not re-query pooled account usage immediately."""
import api.providers as providers
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"})
providers._account_usage_status_cache.clear()
calls = []
snapshot = SimpleNamespace(
provider="openai-codex",
source="usage_api_pool",
title="Account limits",
plan=None,
windows=(),
details=(),
available=True,
unavailable_reason=None,
fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc),
pool={"total_credentials": 1, "credentials": []},
)
def fake_fetch(provider, home, api_key=None):
calls.append((provider, str(home), api_key))
return snapshot
monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch)
try:
first = providers._fetch_account_usage_with_profile_context("openai-codex")
second = providers._fetch_account_usage_with_profile_context("openai-codex")
finally:
providers._account_usage_status_cache.clear()
_restore_config(old_cfg, old_mtime)
assert first is snapshot
assert second is snapshot
assert calls == [("openai-codex", str(tmp_path), None)]
def test_account_usage_profile_fetch_caches_none_results(monkeypatch, tmp_path):
"""Known-empty/unavailable account usage should be cached, not re-probed each refresh."""
import api.providers as providers
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"})
providers._account_usage_status_cache.clear()
calls = []
def fake_fetch(provider, home, api_key=None):
calls.append((provider, str(home), api_key))
return None
monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch)
try:
first = providers._fetch_account_usage_with_profile_context("openai-codex")
second = providers._fetch_account_usage_with_profile_context("openai-codex")
finally:
providers._account_usage_status_cache.clear()
_restore_config(old_cfg, old_mtime)
assert first is None
assert second is None
assert calls == [("openai-codex", str(tmp_path), None)]
def test_account_usage_profile_fetches_can_overlap_for_different_homes(monkeypatch, tmp_path):
"""Different profile quota fetches should not serialize on cron's global lock."""
import api.providers as providers
@@ -531,6 +782,81 @@ def test_account_usage_profile_fetches_can_overlap_for_different_homes(monkeypat
assert [kind for kind, _home in events[:2]] == ["enter", "enter"]
def test_openai_api_key_detection_ignores_codex_oauth_jwt(monkeypatch, tmp_path):
"""A Codex OAuth JWT in OPENAI_API_KEY should not show a bare OpenAI card."""
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
def b64url(payload: bytes) -> str:
return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii")
codex_token = ".".join((
b64url(b'{"alg":"none","typ":"JWT"}'),
b64url(json.dumps({
"https://api.openai.com/auth": {"chatgpt_account_id": "acct-codex"},
}).encode("utf-8")),
b64url(b"signature"),
))
(tmp_path / ".env").write_text(f"OPENAI_API_KEY={codex_token}\n", encoding="utf-8")
old_cfg, old_mtime = _with_config(model={"provider": "openai"})
import api.providers as providers
try:
assert providers._provider_has_key("openai") is False
assert providers._get_provider_api_key("openai") is None
provider_ids = [p["id"] for p in providers.get_providers()["providers"]]
finally:
_restore_config(old_cfg, old_mtime)
assert "openai" not in provider_ids
assert "openai-codex" in provider_ids
assert codex_token not in repr(provider_ids)
def test_openai_api_key_detection_still_accepts_real_api_keys(monkeypatch, tmp_path):
"""Filtering Codex OAuth tokens must not hide real OpenAI API keys."""
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
(tmp_path / ".env").write_text("OPENAI_API_KEY=sk-test-real-openai-key\n", encoding="utf-8")
old_cfg, old_mtime = _with_config(model={"provider": "openai"})
import api.providers as providers
try:
assert providers._provider_has_key("openai") is True
assert providers._get_provider_api_key("openai") == "sk-test-real-openai-key"
finally:
_restore_config(old_cfg, old_mtime)
def test_openai_api_key_detection_falls_through_after_codex_jwt_config_value(monkeypatch, tmp_path):
"""A filtered OpenAI config value should not mask a later real API key source."""
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
def b64url(payload: bytes) -> str:
return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii")
codex_token = ".".join((
b64url(b'{"alg":"none","typ":"JWT"}'),
b64url(json.dumps({
"https://api.openai.com/auth": {"chatgpt_account_id": "acct-codex"},
}).encode("utf-8")),
b64url(b"signature"),
))
old_cfg, old_mtime = _with_config(
model={"provider": "openai", "api_key": codex_token},
providers={"openai": {"api_key": "sk-config-openai-key"}},
)
import api.providers as providers
try:
assert providers._provider_has_key("openai") is True
assert providers._get_provider_api_key("openai") == "sk-config-openai-key"
finally:
_restore_config(old_cfg, old_mtime)
def test_provider_quota_route_is_registered():
"""The backend must expose a route for the UI to poll quota status."""
routes = (ROOT / "api" / "routes.py").read_text(encoding="utf-8")
@@ -549,6 +875,14 @@ def test_provider_quota_card_is_rendered_in_providers_panel():
assert "account_limits" in panels
assert "remaining_percent" in panels
assert "provider-quota-details" in panels
assert "Credential pool" in panels
assert "provider-quota-pool-row" in panels
assert "_buildProviderQuotaPoolBreakdown" in panels
assert "_providerQuotaPoolShouldDefaultOpen" in panels
assert "hermes-provider-quota-pool-open" in panels
assert "count>0&&count<=3" in panels
assert "status.status==='available'||accountLimits.pool" in panels
assert "provider-quota-window-detail" in panels
assert "5-hour limit" in panels
@@ -581,6 +915,10 @@ def test_provider_quota_styles_exist():
".provider-quota-actions",
".provider-quota-refresh",
".provider-quota-checked",
".provider-quota-pool",
".provider-quota-pool-row",
".provider-quota-pool-window",
".provider-quota-window-detail",
):
assert token in css