294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""Read-only provider-pool status derived from Switchyard and local auth metadata."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
SWITCHYARD_ROOT = os.environ.get(
|
|
"HERMES_SWITCHYARD_STATUS_URL",
|
|
"http://hermes-switchyard.hermes.svc.cluster.local:9005",
|
|
).rstrip("/")
|
|
CODEX_USAGE_URL = "https://chatgpt.com/codex/settings/usage"
|
|
CLAUDE_USAGE_URL = "https://claude.ai/settings/usage"
|
|
CODEX_AUTH_PATH = Path(
|
|
os.environ.get("CODEX_HOME", "/opt/data/home/.codex")
|
|
) / "auth.json"
|
|
CLAUDE_AUTH_PATH = Path(
|
|
os.environ.get("CLAUDE_CONFIG_DIR", "/opt/data/home/.claude")
|
|
) / ".credentials.json"
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any]:
|
|
"""Return one local JSON object without leaking parsing details."""
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, TypeError, ValueError, json.JSONDecodeError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _get_json(url: str, timeout: float = 3.0) -> dict[str, Any]:
|
|
"""Fetch a bounded JSON status response from the in-cluster router."""
|
|
request = Request(url, headers={"Accept": "application/json"})
|
|
try:
|
|
with urlopen(request, timeout=timeout) as response:
|
|
value = json.load(response)
|
|
except (HTTPError, URLError, OSError, TimeoutError, ValueError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _jwt_claims(token: Any) -> dict[str, Any]:
|
|
"""Decode trusted local token metadata for display; do not verify or return it."""
|
|
if not isinstance(token, str):
|
|
return {}
|
|
try:
|
|
encoded = token.split(".")[1]
|
|
encoded += "=" * (-len(encoded) % 4)
|
|
value = json.loads(base64.urlsafe_b64decode(encoded))
|
|
except (
|
|
IndexError,
|
|
TypeError,
|
|
ValueError,
|
|
json.JSONDecodeError,
|
|
base64.binascii.Error,
|
|
):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _timestamp(value: Any) -> tuple[str | None, bool | None]:
|
|
"""Normalize an epoch or ISO-8601 timestamp and report whether it is live."""
|
|
if isinstance(value, str):
|
|
rendered_value = value.strip()
|
|
if rendered_value and not rendered_value.replace(".", "", 1).isdigit():
|
|
try:
|
|
parsed = datetime.fromisoformat(rendered_value.replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return None, None
|
|
rendered = parsed.astimezone(timezone.utc).isoformat()
|
|
return rendered, parsed.timestamp() > time.time()
|
|
try:
|
|
raw = float(value)
|
|
except (TypeError, ValueError):
|
|
return None, None
|
|
if raw > 100_000_000_000:
|
|
raw /= 1000
|
|
try:
|
|
rendered = datetime.fromtimestamp(raw, tz=timezone.utc).isoformat()
|
|
except (OverflowError, OSError, ValueError):
|
|
return None, None
|
|
return rendered, raw > time.time()
|
|
|
|
|
|
def _codex_account() -> dict[str, Any]:
|
|
"""Return non-secret Codex plan and credential health metadata."""
|
|
auth = _read_json(CODEX_AUTH_PATH)
|
|
tokens = auth.get("tokens") if isinstance(auth.get("tokens"), dict) else {}
|
|
access = _jwt_claims(tokens.get("access_token"))
|
|
identity = _jwt_claims(tokens.get("id_token"))
|
|
claims = identity.get("https://api.openai.com/auth")
|
|
claims = claims if isinstance(claims, dict) else {}
|
|
expires_at, token_live = _timestamp(access.get("exp"))
|
|
subscription_until, subscription_live = _timestamp(
|
|
claims.get("chatgpt_subscription_active_until")
|
|
)
|
|
authenticated = bool(tokens.get("access_token")) and token_live is not False
|
|
return {
|
|
"authenticated": authenticated,
|
|
"auth_mode": auth.get("auth_mode") or "unknown",
|
|
"plan": claims.get("chatgpt_plan_type") or "unknown",
|
|
"token_expires_at": expires_at,
|
|
"subscription_active": subscription_live,
|
|
"subscription_until": subscription_until,
|
|
"last_refresh": auth.get("last_refresh"),
|
|
"quota_reported": False,
|
|
"usage_url": CODEX_USAGE_URL,
|
|
}
|
|
|
|
|
|
def _claude_account() -> dict[str, Any]:
|
|
"""Return non-secret Claude plan and credential health metadata."""
|
|
auth = _read_json(CLAUDE_AUTH_PATH)
|
|
oauth = auth.get("claudeAiOauth")
|
|
oauth = oauth if isinstance(oauth, dict) else {}
|
|
expires_at, token_live = _timestamp(oauth.get("expiresAt"))
|
|
refresh_expires_at, refresh_live = _timestamp(oauth.get("refreshTokenExpiresAt"))
|
|
refreshable = bool(oauth.get("refreshToken")) and refresh_live is not False
|
|
authenticated = (
|
|
bool(oauth.get("accessToken")) and token_live is not False
|
|
) or refreshable
|
|
return {
|
|
"authenticated": authenticated,
|
|
"access_token_live": token_live,
|
|
"refreshable": refreshable,
|
|
"plan": oauth.get("subscriptionType") or "unknown",
|
|
"rate_limit_tier": oauth.get("rateLimitTier") or "unknown",
|
|
"token_expires_at": expires_at,
|
|
"refresh_expires_at": refresh_expires_at,
|
|
"refresh_live": refresh_live,
|
|
"quota_reported": False,
|
|
"usage_url": CLAUDE_USAGE_URL,
|
|
}
|
|
|
|
|
|
def _provider_for_model(model_id: str) -> str:
|
|
"""Map Switchyard target identifiers to their operator-facing provider."""
|
|
value = model_id.lower()
|
|
if "/codex/" in value or value.startswith("openai"):
|
|
return "codex"
|
|
if "/claude/" in value or value.startswith("anthropic"):
|
|
return "claude"
|
|
if "/local/" in value or "qwen" in value or "ollama" in value:
|
|
return "local"
|
|
return "other"
|
|
|
|
|
|
def _number(value: Any) -> int:
|
|
"""Coerce non-negative counters from Switchyard's JSON safely."""
|
|
try:
|
|
return max(0, int(value or 0))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _provider_summary(name: str, models: dict[str, Any]) -> dict[str, Any]:
|
|
"""Aggregate successful calls, errors, tokens, and latency for one lane."""
|
|
selected: list[dict[str, Any]] = []
|
|
totals = {
|
|
"calls": 0,
|
|
"errors": 0,
|
|
"prompt_tokens": 0,
|
|
"completion_tokens": 0,
|
|
"reasoning_tokens": 0,
|
|
"total_tokens": 0,
|
|
}
|
|
latency_weight = 0.0
|
|
latency_calls = 0
|
|
for model_id, raw in sorted(models.items()):
|
|
if _provider_for_model(str(model_id)) != name or not isinstance(raw, dict):
|
|
continue
|
|
item = {"id": str(model_id)}
|
|
for key in totals:
|
|
item[key] = _number(raw.get(key))
|
|
totals[key] += item[key]
|
|
try:
|
|
item["avg_latency_ms"] = round(float(raw.get("avg_latency_ms") or 0), 1)
|
|
except (TypeError, ValueError):
|
|
item["avg_latency_ms"] = 0.0
|
|
latency_weight += item["avg_latency_ms"] * item["calls"]
|
|
latency_calls += item["calls"]
|
|
selected.append(item)
|
|
|
|
calls = totals["calls"]
|
|
errors = totals["errors"]
|
|
if calls and errors:
|
|
state = "degraded"
|
|
elif calls:
|
|
state = "available"
|
|
elif errors:
|
|
state = "unavailable"
|
|
else:
|
|
state = "unobserved"
|
|
return {
|
|
"state": state,
|
|
**totals,
|
|
"avg_latency_ms": round(latency_weight / latency_calls, 1)
|
|
if latency_calls
|
|
else 0.0,
|
|
"models": selected,
|
|
}
|
|
|
|
|
|
def provider_status_payload() -> dict[str, Any]:
|
|
"""Build the owner-safe status document shared by dashboard and TUI."""
|
|
health = _get_json(f"{SWITCHYARD_ROOT}/health")
|
|
stats = _get_json(f"{SWITCHYARD_ROOT}/v1/stats")
|
|
router_ok = health.get("status") == "ok" and bool(stats)
|
|
models = stats.get("models") if isinstance(stats.get("models"), dict) else {}
|
|
providers = {
|
|
name: _provider_summary(name, models)
|
|
for name in ("codex", "claude", "local")
|
|
}
|
|
providers["codex"]["account"] = _codex_account()
|
|
providers["claude"]["account"] = _claude_account()
|
|
classifier = stats.get("classifier")
|
|
classifier = classifier if isinstance(classifier, dict) else {}
|
|
fallbacks = stats.get("routing_fallbacks")
|
|
fallbacks = fallbacks if isinstance(fallbacks, dict) else {}
|
|
return {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"window": "Since the last Switchyard restart",
|
|
"quota_note": (
|
|
"Codex and Claude subscription balances are not exposed to this "
|
|
"router. Open the provider usage page for authoritative remaining "
|
|
"capacity; the counters here show actual work observed by Switchyard."
|
|
),
|
|
"router": {
|
|
"state": "available" if router_ok else "unavailable",
|
|
"total_requests": _number(stats.get("total_requests")),
|
|
"total_errors": _number(stats.get("total_errors")),
|
|
"total_tokens": _number(
|
|
(stats.get("total_tokens") or {}).get("total")
|
|
if isinstance(stats.get("total_tokens"), dict)
|
|
else 0
|
|
),
|
|
"fallbacks": {str(key): _number(value) for key, value in fallbacks.items()},
|
|
"classifier_calls": _number(classifier.get("total_requests")),
|
|
"classifier_errors": _number(classifier.get("total_errors")),
|
|
},
|
|
"providers": providers,
|
|
}
|
|
|
|
|
|
def provider_status_text() -> str:
|
|
"""Render the same status as a compact native Hermes command response."""
|
|
payload = provider_status_payload()
|
|
router = payload["router"]
|
|
labels = {"codex": "Codex", "claude": "Claude", "local": "Local"}
|
|
lines = [
|
|
f"Provider pool: Switchyard {router['state']}",
|
|
str(payload["window"]),
|
|
]
|
|
for key in ("codex", "claude", "local"):
|
|
item = payload["providers"][key]
|
|
account = item.get("account") or {}
|
|
plan = f" · plan {account.get('plan', 'unknown')}" if account else ""
|
|
auth = ""
|
|
if account:
|
|
if account.get("access_token_live") is False and account.get("refreshable"):
|
|
auth = " · auth refreshable"
|
|
else:
|
|
auth = (
|
|
" · auth ready"
|
|
if account.get("authenticated")
|
|
else " · auth unavailable"
|
|
)
|
|
lines.append(
|
|
f"{labels[key]}: {item['state']} · {item['calls']} completed · "
|
|
f"{item['errors']} errors · {item['total_tokens']} tokens{plan}{auth}"
|
|
)
|
|
lines.extend(
|
|
[
|
|
f"Fallbacks: {sum(router['fallbacks'].values())} · "
|
|
f"classifier: {router['classifier_calls']} calls / "
|
|
f"{router['classifier_errors']} errors",
|
|
str(payload["quota_note"]),
|
|
f"Codex usage: {CODEX_USAGE_URL}",
|
|
f"Claude usage: {CLAUDE_USAGE_URL}",
|
|
"Dashboard: Provider Status",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|