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

refactor(status): extract credential sections into status_auth, unify redact_key

Browse Source 
- new hermes_cli/status_auth.py: _render_api_keys / _render_auth_providers /
  _render_nous_gateway / _render_apikey_providers + their data tables and the
  OAuth/timestamp helpers; origin helpers imported lazily so status monkeypatches hold
- status.redact_key was a verbatim copy of hermes_cli.config.redact_key -> imported
- sessions/cron/deep/footer renderers compacted; 624 -> 372 lines (base 639)

Parity: r2m_status_golden.py 33 fixtures byte-identical vs base; real
`hermes status`, `status --deep`, `status --help` byte-identical after normalizing
the per-run temp HERMES_HOME path; test_status*/test_estop/test_jobs_json_utf8_bom green.
This commit is contained in:
Teknium
2026-09-02 16:54:55 -07:00
parent ce54c8e04a
commit ebb9f312dc
2 changed files with 297 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

390
hermes_cli/status.py
View File

@@ -12,18 +12,18 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve()
from hermes_cli.auth import AuthError, resolve_provider
from hermes_cli.colors import Colors, color
from hermes_cli.config import get_env_path, get_env_value, get_hermes_home, load_config
from hermes_cli.config import get_env_path, get_env_value, get_hermes_home, load_config, redact_key # noqa: F401 (redact_key: status_auth resolves it here)
from hermes_cli.models import provider_label
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
from hermes_cli.nous_subscription import get_nous_subscription_features
from hermes_cli.runtime_provider import resolve_requested_provider
from hermes_cli.vercel_auth import describe_vercel_auth
from hermes_cli.status_auth import ( # renderers wired into _SECTIONS below
_render_api_keys,
_render_apikey_providers,
_render_auth_providers,
_render_nous_gateway,
)
from hermes_constants import OPENROUTER_MODELS_URL
from hermes_constants import is_termux as _is_termux
from tools.tool_backend_helpers import managed_nous_tools_enabled
def check_mark(ok: bool) -> str:
@@ -46,61 +46,9 @@ def _detail(label: str, value) -> None:
print(f" {label:<12}{value}")
def _oauth_block(name: str, status: dict, hint: str, rows) -> None:
"""Print an OAuth provider row plus its conditional detail lines.
``rows`` are ``(label, status_key, formatter, gate)``: a detail prints when the raw value is
truthy and ``gate`` is None or equals the logged-in state (False = only while logged out).
"""
logged_in = bool(status.get("logged_in"))
_row(name, logged_in, "logged in" if logged_in else f"not logged in (run: {hint})")
for label, key, fmt, gate in rows:
raw = status.get(key)
if raw and (gate is None or gate == logged_in):
_detail(label, fmt(raw) if fmt else raw)
def _first_env_value(names) -> str:
"""Return the first non-empty env value among ``names`` (a str or tuple of names)."""
if isinstance(names, str):
names = (names,)
for candidate in names:
v = get_env_value(candidate) or ""
if v:
return v
return ""
def redact_key(key: str) -> str:
"""Redact an API key for display.
Thin wrapper over :func:`agent.redact.mask_secret` that keeps the dim "(not set)" placeholder
consistent with ``hermes config`` output.
"""
from agent.redact import mask_secret
return mask_secret(key, empty=color("(not set)", Colors.DIM))
def _format_iso_timestamp(value) -> str:
"""Format ISO timestamps for status output, converting to local timezone."""
text = value.strip() if isinstance(value, str) else ""
if not text:
return "(unknown)"
from datetime import datetime, timezone
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
except Exception:
return value
return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
def _qwen_expiry(expires_at_ms) -> str:
from datetime import datetime, timezone
return datetime.fromtimestamp(int(expires_at_ms) / 1000, tz=timezone.utc).isoformat()
return next((v for v in (get_env_value(n) or "" for n in ((names,) if isinstance(names, str) else names)) if v), "")
def _configured_model_label(config: dict) -> str:
@@ -108,9 +56,7 @@ def _configured_model_label(config: dict) -> str:
model_cfg = config.get("model")
if isinstance(model_cfg, dict):
model_cfg = model_cfg.get("default") or model_cfg.get("name") or ""
elif not isinstance(model_cfg, str):
model_cfg = ""
return model_cfg.strip() or "(not set)"
return (model_cfg.strip() if isinstance(model_cfg, str) else "") or "(not set)"
def _effective_provider_label() -> str:
@@ -122,11 +68,9 @@ def _effective_provider_label() -> str:
effective = requested or "auto"
if effective == "openrouter":
# A custom endpoint may be configured either in config.yaml
# (model.base_url — the canonical location; the runtime treats
# config.yaml as the single source of truth) or via the legacy
# OPENAI_BASE_URL env var. Either way, labeling it "OpenRouter"
# is misleading (#3296).
# A custom endpoint may live in config.yaml (model.base_url, the canonical
# location) or the legacy OPENAI_BASE_URL env var; either way labeling it
# "OpenRouter" is misleading.
try:
model_cfg = load_config().get("model")
config_base_url = (model_cfg.get("base_url") or "").strip() if isinstance(model_cfg, dict) else ""
@@ -134,7 +78,6 @@ def _effective_provider_label() -> str:
config_base_url = ""
if config_base_url or get_env_value("OPENAI_BASE_URL"):
effective = "custom"
return provider_label(effective)
@@ -151,64 +94,8 @@ def _estop_status_line():
return f"⏸️ PAUSED (global emergency stop{f' — reason: {reason}' if reason else ''}; `hermes resume` to lift)"
# ---------------------------------------------------------------------------
# Data tables driving the per-section renderers below.
# ---------------------------------------------------------------------------
# --- Data tables driving the per-section renderers -------------------------
# Values may be a single env var name (str) or a tuple of alternates (first found wins).
_API_KEYS: dict[str, str | tuple[str, ...]] = {
"OpenRouter": "OPENROUTER_API_KEY",
"OpenAI": "OPENAI_API_KEY",
"Google / Gemini": ("GOOGLE_API_KEY", "GEMINI_API_KEY"),
"DeepSeek": "DEEPSEEK_API_KEY",
"xAI / Grok": "XAI_API_KEY",
"NVIDIA NIM": "NVIDIA_API_KEY",
"Z.AI / GLM": "GLM_API_KEY",
"Kimi": "KIMI_API_KEY",
"StepFun Step Plan": "STEPFUN_API_KEY",
"MiniMax": "MINIMAX_API_KEY",
"MiniMax-CN": "MINIMAX_CN_API_KEY",
"DeepInfra": "DEEPINFRA_API_KEY",
"Firecrawl": "FIRECRAWL_API_KEY",
"Tavily": "TAVILY_API_KEY",
"Keenable": "KEENABLE_API_KEY",
"Browser Use": "BROWSER_USE_API_KEY", # Optional — local browser works without this
"Browserbase": "BROWSERBASE_API_KEY", # Optional — direct credentials only
"FAL": "FAL_KEY",
"ElevenLabs": "ELEVENLABS_API_KEY",
"GitHub": "GITHUB_TOKEN",
}
# OAuth detail rows: (label, status key, formatter, gate) — see _oauth_block.
_FILE_REFRESH_ROWS = (
("Auth file:", "auth_store", None, None),
("Refreshed:", "last_refresh", _format_iso_timestamp, None),
("Error:", "error", None, False),
)
_OAUTH_BLOCKS = (
# (row name, auth getter, login hint, detail rows)
("OpenAI Codex", "get_codex_auth_status", "hermes model", _FILE_REFRESH_ROWS),
("Qwen OAuth", "get_qwen_auth_status", "qwen auth qwen-oauth", (
("Auth file:", "auth_file", None, None),
("Access exp:", "expires_at_ms", _qwen_expiry, None),
("Error:", "error", None, False),
)),
("MiniMax OAuth", "get_minimax_oauth_auth_status", "hermes auth add minimax-oauth", (
("Region:", "region", None, True),
("Access exp:", "expires_at", None, None),
("Error:", "error", None, False),
)),
("xAI OAuth", "get_xai_oauth_auth_status", "hermes auth add xai-oauth", _FILE_REFRESH_ROWS),
)
_APIKEY_PROVIDERS = {
"Z.AI / GLM": ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"),
"Kimi / Moonshot": ("KIMI_API_KEY",),
"StepFun Step Plan": ("STEPFUN_API_KEY",),
"MiniMax": ("MINIMAX_API_KEY",),
"MiniMax (China)": ("MINIMAX_CN_API_KEY",),
"DeepInfra": ("DEEPINFRA_API_KEY",),
}
# Simple env-driven terminal backends: (label, env var, default, empty-counts-as-unset).
_TERMINAL_ENV_ROWS = {
@@ -244,10 +131,8 @@ class _StatusContext:
the Auth Providers section derives that the Nous Tool Gateway section needs later."""
def __init__(self, deep: bool):
self.deep = deep
self.config: dict = {}
self.nous_logged_in = False
self.nous_inference_present = False
self.deep, self.config = deep, {}
self.nous_logged_in = self.nous_inference_present = False
self.nous_account_info = None
@@ -256,10 +141,10 @@ def _render_header(ctx):
print(color("┌─────────────────────────────────────────────────────────┐", Colors.CYAN))
print(color("│ ⚕ Hermes Agent Status │", Colors.CYAN))
print(color("└─────────────────────────────────────────────────────────┘", Colors.CYAN))
_paused_line = _estop_status_line()
if _paused_line:
paused = _estop_status_line()
if paused:
print()
print(color(_paused_line, Colors.YELLOW, Colors.BOLD))
print(color(paused, Colors.YELLOW, Colors.BOLD))
def _render_environment(ctx):
@@ -276,124 +161,6 @@ def _render_environment(ctx):
print(f" Provider: {_effective_provider_label()}")
def _render_api_keys(ctx):
_section("API Keys")
for name, env_ref in _API_KEYS.items():
value = _first_env_value(env_ref)
_row(name, bool(value), redact_key(value))
# Anthropic uses the dedicated lookup (it also resolves OAuth tokens).
from hermes_cli.auth import get_anthropic_key
anthropic_value = get_anthropic_key()
_row("Anthropic", bool(anthropic_value), redact_key(anthropic_value))
def _render_auth_providers(ctx):
_section("Auth Providers")
import hermes_cli.auth as auth
try:
# Read-only display: use the refresh-free snapshot so `hermes status`
# never performs an OAuth refresh or burns a single-use refresh token.
nous_status = auth.get_nous_auth_status_local()
statuses = {getter: getattr(auth, getter)() for _, getter, _, _ in _OAUTH_BLOCKS[:3]}
except Exception:
nous_status, statuses = {}, {}
# xAI OAuth — separate try/except so an import failure here cannot
# disrupt the Nous/Codex/Qwen/MiniMax rows.
try:
statuses["get_xai_oauth_auth_status"] = auth.get_xai_oauth_auth_status() or {}
except Exception:
statuses["get_xai_oauth_auth_status"] = {}
info = None
if any(nous_status.get(k) for k in (
"logged_in", "access_token", "portal_base_url", "inference_credential_present", "error_code"
)):
try:
info = get_nous_portal_account_info()
except Exception:
info = None
ctx.nous_account_info = info
ctx.nous_logged_in = logged_in = bool(nous_status.get("logged_in") or (info and info.logged_in))
ctx.nous_inference_present = inference = bool(
nous_status.get("inference_credential_present") or (info and info.inference_credential_present)
)
nous_error = nous_status.get("error")
_row(
"Nous Portal", logged_in,
"logged in" if logged_in
else "not logged in (Nous inference key configured)" if inference
else "not logged in (run: hermes portal)",
)
portal_url = nous_status.get("portal_base_url") or "(unknown)"
inference_url = nous_status.get("inference_base_url") or (info.inference_base_url if info else None)
for label, value, show in (
("Portal URL:", portal_url, logged_in or portal_url != "(unknown)" or nous_error),
("Inference:", inference_url, inference and inference_url),
("Access exp:", _format_iso_timestamp(nous_status.get("access_expires_at")),
logged_in or nous_status.get("access_expires_at")),
("Key exp:", _format_iso_timestamp(nous_status.get("agent_key_expires_at")),
logged_in or inference or nous_status.get("agent_key_expires_at")),
("Refresh:", "yes" if nous_status.get("has_refresh_token") else "no",
logged_in or nous_status.get("has_refresh_token")),
("Error:", nous_error, nous_error),
):
if show:
_detail(label, value)
for name, getter, hint, rows in _OAUTH_BLOCKS:
_oauth_block(name, statuses.get(getter, {}), hint, rows)
def _render_nous_gateway(ctx):
if managed_nous_tools_enabled():
features = get_nous_subscription_features(ctx.config)
_section("Nous Tool Gateway")
print(" Nous Portal ✓ managed tools available" if features.nous_auth_present else " Nous Portal ✗ not logged in")
for feature in features.items():
if feature.managed_by_nous:
state = "active via Nous subscription"
elif feature.active:
state = f"active via {feature.current_provider or 'configured provider'}"
elif feature.included_by_default and features.nous_auth_present:
state = "included by subscription, not currently selected"
elif feature.key == "modal" and features.nous_auth_present:
state = "available via subscription (optional)"
else:
state = "not configured"
print(f" {feature.label:<15} {check_mark(feature.available or feature.active or feature.managed_by_nous)} {state}")
elif ctx.nous_logged_in or ctx.nous_inference_present:
# Nous OAuth without entitlement, or an opaque inference key without
# Portal account information, cannot enable the Tool Gateway.
_section("Nous Tool Gateway")
message = format_nous_portal_entitlement_message(
ctx.nous_account_info, capability="managed web, image, TTS, STT, browser, and Modal tools"
)
for line in (message or "").splitlines():
print(f" {line}")
def _render_apikey_providers(ctx):
_section("API-Key Providers")
for pname, env_vars in _APIKEY_PROVIDERS.items():
configured = bool(_first_env_value(env_vars))
label = "configured" if configured else "not configured (run: hermes model)"
print(f" {pname:<16} {check_mark(configured)} {label}")
# LM Studio reachability — only probe when it's the active provider so
# users with foreign configs don't see noise. Auth rejection vs. silent
# empty list is the most common LM Studio support case.
if _effective_provider_label() == "LM Studio":
from hermes_cli.models import probe_lmstudio_models
model_cfg = ctx.config.get("model")
base = (model_cfg.get("base_url") if isinstance(model_cfg, dict) else None) or get_env_value("LM_BASE_URL") or "http://127.0.0.1:1234/v1"
try:
models = probe_lmstudio_models(api_key=get_env_value("LM_API_KEY") or "", base_url=base, timeout=1.5)
ok = models is not None
msg = f"reachable ({len(models)} model(s)) at {base}" if ok else f"unreachable at {base}"
except AuthError:
ok, msg = False, "auth rejected — set LM_API_KEY"
print(f" {'LM Studio':<16} {check_mark(ok)} {msg}")
def _render_terminal(ctx):
_section("Terminal Backend")
terminal_cfg = ctx.config.get("terminal", {}) if isinstance(ctx.config.get("terminal"), dict) else {}
@@ -405,17 +172,13 @@ def _render_terminal(ctx):
value = (os.getenv(var, "") or default) if empty_is_unset else os.getenv(var, default)
print(f" {label:<13} {value}")
elif terminal_env == "vercel_sandbox":
runtime = os.getenv("TERMINAL_VERCEL_RUNTIME") or terminal_cfg.get("vercel_runtime") or "node24"
persist = os.getenv("TERMINAL_CONTAINER_PERSISTENT")
persist_enabled = (
bool(terminal_cfg.get("container_persistent", True))
if persist is None
else persist.lower() in {"1", "true", "yes", "on"}
)
persist_enabled = (bool(terminal_cfg.get("container_persistent", True)) if persist is None
else persist.lower() in {"1", "true", "yes", "on"})
auth_status = describe_vercel_auth()
sdk_ok = importlib.util.find_spec("vercel") is not None
sdk_label = "installed" if sdk_ok else "missing (install: pip install 'hermes-agent[vercel]')"
print(f" Runtime: {runtime}")
print(f" Runtime: {os.getenv('TERMINAL_VERCEL_RUNTIME') or terminal_cfg.get('vercel_runtime') or 'node24'}")
print(f" SDK: {check_mark(sdk_ok)} {sdk_label}")
print(f" Auth: {check_mark(auth_status.ok)} {auth_status.label}")
for line in auth_status.detail_lines:
@@ -429,10 +192,10 @@ def _render_terminal(ctx):
from hermes_cli.plugins import discover_plugins
discover_plugins()
from agent.terminal_env_registry import get_provider
_provider = get_provider(terminal_env)
if _provider is not None:
for _ok, _label, _text in _provider.doctor_checks():
print(f" {_label}: {check_mark(bool(_ok))} {_text}")
provider = get_provider(terminal_env)
if provider is not None:
for ok, label, text in provider.doctor_checks():
print(f" {label}: {check_mark(bool(ok))} {text}")
except Exception:
pass
@@ -445,18 +208,13 @@ def _render_platforms(ctx):
for name, (token_var, home_var) in _PLATFORMS.items():
has_token = bool(os.getenv(token_var, ""))
home_channel = os.getenv(home_var, "") if home_var else ""
status = "configured" if has_token else "not configured"
if home_channel:
status += f" (home: {home_channel})"
_row(name, has_token, status)
_row(name, has_token, ("configured" if has_token else "not configured") + (f" (home: {home_channel})" if home_channel else ""))
# Plugin-registered platforms
try:
try: # Plugin-registered platforms
from gateway.platform_registry import platform_registry
for entry in platform_registry.plugin_entries():
# Per-entry guard: one raising probe must not abort the listing
# of every remaining plugin platform (matches the other three
# check_fn call sites).
# Per-entry guard: one raising probe must not abort the listing of
# every remaining plugin platform (matches the other check_fn sites).
try:
configured = bool(entry.check_fn())
except Exception:
@@ -472,8 +230,7 @@ def _render_gateway(ctx):
from hermes_cli.gateway import get_gateway_runtime_snapshot, _format_gateway_pids
snapshot = get_gateway_runtime_snapshot()
is_running = snapshot.running
print(f" Status: {check_mark(is_running)} {'running' if is_running else 'stopped'}")
print(f" Status: {check_mark(snapshot.running)} {'running' if snapshot.running else 'stopped'}")
print(f" Manager: {snapshot.manager}")
if snapshot.gateway_pids:
print(f" PID(s): {_format_gateway_pids(snapshot.gateway_pids)}")
@@ -497,41 +254,39 @@ def _render_gateway(ctx):
def _render_cron(ctx):
_section("Scheduled Jobs")
jobs_file = get_hermes_home() / "cron" / "jobs.json"
if jobs_file.exists():
try:
# utf-8-sig: same dialect as cron/jobs.load_jobs — Windows editors
# may leave a UTF-8 BOM that plain utf-8 json.load rejects.
with open(jobs_file, encoding="utf-8-sig") as f:
jobs = json.load(f).get("jobs", [])
enabled = sum(1 for j in jobs if j.get("enabled", True))
print(f" Jobs: {enabled} active, {len(jobs)} total")
except Exception:
print(" Jobs: (error reading jobs file)")
else:
if not jobs_file.exists():
print(" Jobs: 0")
return
try:
# utf-8-sig: same dialect as cron/jobs.load_jobs — Windows editors
# may leave a UTF-8 BOM that plain utf-8 json.load rejects.
with open(jobs_file, encoding="utf-8-sig") as f:
jobs = json.load(f).get("jobs", [])
print(f" Jobs: {sum(1 for j in jobs if j.get('enabled', True))} active, {len(jobs)} total")
except Exception:
print(" Jobs: (error reading jobs file)")
def _render_sessions(ctx):
_section("Sessions")
# Gateway session count: state.db is the source of truth (#9006);
# fall back to sessions.json for pre-migration installs.
_session_count, _gateway_rows = None, []
# Gateway session count: state.db is the source of truth; fall back to
# sessions.json for pre-migration installs.
gateway_rows = []
try:
from hermes_state import SessionDB
_db = SessionDB()
db = SessionDB()
try:
_lister = getattr(_db, "list_gateway_sessions", None)
if callable(_lister):
_gateway_rows = _lister(active_only=True) or []
_session_count = len(_gateway_rows)
lister = getattr(db, "list_gateway_sessions", None)
if callable(lister):
gateway_rows = lister(active_only=True) or []
finally:
_db.close()
db.close()
except Exception:
_session_count, _gateway_rows = None, []
gateway_rows = []
if _session_count:
print(f" Active: {_session_count} session(s)")
freshest = max((float(r.get("last_active") or 0) for r in _gateway_rows), default=0.0)
if gateway_rows:
print(f" Active: {len(gateway_rows)} session(s)")
freshest = max((float(r.get("last_active") or 0) for r in gateway_rows), default=0.0)
if freshest > 0:
from hermes_cli.timefmt import relative_time
print(f" Last activity:{relative_time(freshest):>13}")
@@ -541,12 +296,12 @@ def _render_sessions(ctx):
try:
with open(sessions_file, encoding="utf-8") as f:
data = json.load(f)
_entries = [k for k in data if not str(k).startswith("_")] if isinstance(data, dict) else []
print(f" Active: {len(_entries)} session(s)")
entries = [k for k in data if not str(k).startswith("_")] if isinstance(data, dict) else []
print(f" Active: {len(entries)} session(s)")
except Exception:
print(" Active: (error reading sessions file)")
else:
print(f" Active: {_session_count or 0}")
print(" Active: 0")
# Slot usage, only when max_concurrent_sessions is set. The cap is shared
# across CLI, desktop/TUI and the messaging gateway, so the surface that
@@ -556,30 +311,25 @@ def _render_sessions(ctx):
from hermes_cli.active_sessions import (
active_session_registry_snapshot, format_age, resolve_max_concurrent_sessions,
)
_cap = resolve_max_concurrent_sessions(ctx.config)
cap = resolve_max_concurrent_sessions(ctx.config)
except Exception:
_cap = None
if _cap:
cap = None
if cap:
try:
_held = active_session_registry_snapshot()
held = active_session_registry_snapshot()
except Exception:
_held = []
_full = len(_held) >= _cap
print(" Slots: " + color(f"{len(_held)}/{_cap} in use", Colors.YELLOW if _full else Colors.GREEN))
_now = time.time()
for _entry in sorted(_held, key=lambda e: e.get("started_at") or 0):
_age = format_age(_now - float(_entry.get("started_at") or _now))
print(
f" {_entry.get('surface') or 'unknown':<17} "
f"{_entry.get('session_id') or '?':<24} {_age}"
)
held = []
print(" Slots: " + color(f"{len(held)}/{cap} in use", Colors.YELLOW if len(held) >= cap else Colors.GREEN))
now = time.time()
for entry in sorted(held, key=lambda e: e.get("started_at") or 0):
age = format_age(now - float(entry.get("started_at") or now))
print(f" {entry.get('surface') or 'unknown':<17} {entry.get('session_id') or '?':<24} {age}")
def _render_deep(ctx):
if not ctx.deep:
return
_section("Deep Checks")
# Check OpenRouter connectivity
openrouter_key = os.getenv("OPENROUTER_API_KEY", "")
if openrouter_key:
try:
@@ -589,12 +339,11 @@ def _render_deep(ctx):
print(f" OpenRouter: {check_mark(ok)} {'reachable' if ok else f'error ({response.status_code})'}")
except Exception as e:
print(f" OpenRouter: {check_mark(False)} error: {e}")
# Check gateway port
try:
try: # gateway port, informational: in use == gateway likely running
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
port_in_use = sock.connect_ex(('127.0.0.1', 18789)) == 0 # informational: gateway likely running
port_in_use = sock.connect_ex(('127.0.0.1', 18789)) == 0
sock.close()
print(f" Port 18789: {'in use' if port_in_use else 'available'}")
except OSError:
@@ -603,9 +352,8 @@ def _render_deep(ctx):
def _render_footer(ctx):
print()
print(color("─" * 60, Colors.DIM))
print(color(" Run 'hermes doctor' for detailed diagnostics", Colors.DIM))
print(color(" Run 'hermes setup' to configure", Colors.DIM))
for line in ("─" * 60, " Run 'hermes doctor' for detailed diagnostics", " Run 'hermes setup' to configure"):
print(color(line, Colors.DIM))
print()

228
hermes_cli/status_auth.py Normal file
View File

@@ -0,0 +1,228 @@
"""Credential sections of `hermes status`: API keys, OAuth providers, Nous Tool Gateway,
API-key providers. Split out of ``hermes_cli/status.py``; the renderers are re-imported there
and run through ``status._SECTIONS`` with the shared ``_StatusContext``.
Origin helpers (``_row``, ``_first_env_value``, ...) are imported lazily from
``hermes_cli.status`` so tests that monkeypatch that module keep working.
"""
from hermes_cli.auth import AuthError
from hermes_cli.nous_account import (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
from hermes_cli.nous_subscription import get_nous_subscription_features
from tools.tool_backend_helpers import managed_nous_tools_enabled
def _format_iso_timestamp(value) -> str:
"""Format ISO timestamps for status output, converting to local timezone."""
text = value.strip() if isinstance(value, str) else ""
if not text:
return "(unknown)"
from datetime import datetime, timezone
try:
parsed = datetime.fromisoformat(text[:-1] + "+00:00" if text.endswith("Z") else text)
except Exception:
return value
return (parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
def _qwen_expiry(expires_at_ms) -> str:
from datetime import datetime, timezone
return datetime.fromtimestamp(int(expires_at_ms) / 1000, tz=timezone.utc).isoformat()
def _oauth_block(name: str, status: dict, hint: str, rows) -> None:
"""Print an OAuth provider row plus its conditional detail lines.
``rows`` are ``(label, status_key, formatter, gate)``: a detail prints when the raw value is
truthy and ``gate`` is None or equals the logged-in state (False = only while logged out).
"""
from hermes_cli.status import _detail, _row
logged_in = bool(status.get("logged_in"))
_row(name, logged_in, "logged in" if logged_in else f"not logged in (run: {hint})")
for label, key, fmt, gate in rows:
raw = status.get(key)
if raw and (gate is None or gate == logged_in):
_detail(label, fmt(raw) if fmt else raw)
# Values may be a single env var name (str) or a tuple of alternates (first found wins).
_API_KEYS: dict[str, str | tuple[str, ...]] = {
"OpenRouter": "OPENROUTER_API_KEY",
"OpenAI": "OPENAI_API_KEY",
"Google / Gemini": ("GOOGLE_API_KEY", "GEMINI_API_KEY"),
"DeepSeek": "DEEPSEEK_API_KEY",
"xAI / Grok": "XAI_API_KEY",
"NVIDIA NIM": "NVIDIA_API_KEY",
"Z.AI / GLM": "GLM_API_KEY",
"Kimi": "KIMI_API_KEY",
"StepFun Step Plan": "STEPFUN_API_KEY",
"MiniMax": "MINIMAX_API_KEY",
"MiniMax-CN": "MINIMAX_CN_API_KEY",
"DeepInfra": "DEEPINFRA_API_KEY",
"Firecrawl": "FIRECRAWL_API_KEY",
"Tavily": "TAVILY_API_KEY",
"Keenable": "KEENABLE_API_KEY",
"Browser Use": "BROWSER_USE_API_KEY", # Optional — local browser works without this
"Browserbase": "BROWSERBASE_API_KEY", # Optional — direct credentials only
"FAL": "FAL_KEY",
"ElevenLabs": "ELEVENLABS_API_KEY",
"GitHub": "GITHUB_TOKEN",
}
# OAuth detail rows: (label, status key, formatter, gate) — see _oauth_block.
_FILE_REFRESH_ROWS = (
("Auth file:", "auth_store", None, None),
("Refreshed:", "last_refresh", _format_iso_timestamp, None),
("Error:", "error", None, False),
)
_OAUTH_BLOCKS = (
# (row name, auth getter, login hint, detail rows)
("OpenAI Codex", "get_codex_auth_status", "hermes model", _FILE_REFRESH_ROWS),
("Qwen OAuth", "get_qwen_auth_status", "qwen auth qwen-oauth", (
("Auth file:", "auth_file", None, None),
("Access exp:", "expires_at_ms", _qwen_expiry, None),
("Error:", "error", None, False),
)),
("MiniMax OAuth", "get_minimax_oauth_auth_status", "hermes auth add minimax-oauth", (
("Region:", "region", None, True),
("Access exp:", "expires_at", None, None),
("Error:", "error", None, False),
)),
("xAI OAuth", "get_xai_oauth_auth_status", "hermes auth add xai-oauth", _FILE_REFRESH_ROWS),
)
_APIKEY_PROVIDERS = {
"Z.AI / GLM": ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"),
"Kimi / Moonshot": ("KIMI_API_KEY",),
"StepFun Step Plan": ("STEPFUN_API_KEY",),
"MiniMax": ("MINIMAX_API_KEY",),
"MiniMax (China)": ("MINIMAX_CN_API_KEY",),
"DeepInfra": ("DEEPINFRA_API_KEY",),
}
def _render_api_keys(ctx):
from hermes_cli.status import _first_env_value, _row, _section, redact_key
_section("API Keys")
for name, env_ref in _API_KEYS.items():
value = _first_env_value(env_ref)
_row(name, bool(value), redact_key(value))
# Anthropic uses the dedicated lookup (it also resolves OAuth tokens).
from hermes_cli.auth import get_anthropic_key
anthropic_value = get_anthropic_key()
_row("Anthropic", bool(anthropic_value), redact_key(anthropic_value))
def _render_auth_providers(ctx):
from hermes_cli.status import _detail, _row, _section
_section("Auth Providers")
import hermes_cli.auth as auth
try:
# Read-only display: use the refresh-free snapshot so `hermes status`
# never performs an OAuth refresh or burns a single-use refresh token.
nous_status = auth.get_nous_auth_status_local()
statuses = {getter: getattr(auth, getter)() for _, getter, _, _ in _OAUTH_BLOCKS[:3]}
except Exception:
nous_status, statuses = {}, {}
# xAI OAuth is guarded separately so an import failure there cannot disrupt
# the Nous/Codex/Qwen/MiniMax rows.
try:
statuses["get_xai_oauth_auth_status"] = auth.get_xai_oauth_auth_status() or {}
except Exception:
statuses["get_xai_oauth_auth_status"] = {}
info = None
if any(nous_status.get(k) for k in ("logged_in", "access_token", "portal_base_url",
"inference_credential_present", "error_code")):
try:
info = get_nous_portal_account_info()
except Exception:
info = None
ctx.nous_account_info = info
ctx.nous_logged_in = logged_in = bool(nous_status.get("logged_in") or (info and info.logged_in))
ctx.nous_inference_present = inference = bool(
nous_status.get("inference_credential_present") or (info and info.inference_credential_present)
)
nous_error = nous_status.get("error")
_row(
"Nous Portal", logged_in,
"logged in" if logged_in
else "not logged in (Nous inference key configured)" if inference
else "not logged in (run: hermes portal)",
)
portal_url = nous_status.get("portal_base_url") or "(unknown)"
inference_url = nous_status.get("inference_base_url") or (info.inference_base_url if info else None)
for label, value, show in (
("Portal URL:", portal_url, logged_in or portal_url != "(unknown)" or nous_error),
("Inference:", inference_url, inference and inference_url),
("Access exp:", _format_iso_timestamp(nous_status.get("access_expires_at")),
logged_in or nous_status.get("access_expires_at")),
("Key exp:", _format_iso_timestamp(nous_status.get("agent_key_expires_at")),
logged_in or inference or nous_status.get("agent_key_expires_at")),
("Refresh:", "yes" if nous_status.get("has_refresh_token") else "no",
logged_in or nous_status.get("has_refresh_token")),
("Error:", nous_error, nous_error),
):
if show:
_detail(label, value)
for name, getter, hint, rows in _OAUTH_BLOCKS:
_oauth_block(name, statuses.get(getter, {}), hint, rows)
def _render_nous_gateway(ctx):
from hermes_cli.status import _section, check_mark
if managed_nous_tools_enabled():
features = get_nous_subscription_features(ctx.config)
_section("Nous Tool Gateway")
print(" Nous Portal ✓ managed tools available" if features.nous_auth_present else " Nous Portal ✗ not logged in")
for f in features.items():
if f.managed_by_nous:
state = "active via Nous subscription"
elif f.active:
state = f"active via {f.current_provider or 'configured provider'}"
elif f.included_by_default and features.nous_auth_present:
state = "included by subscription, not currently selected"
elif f.key == "modal" and features.nous_auth_present:
state = "available via subscription (optional)"
else:
state = "not configured"
print(f" {f.label:<15} {check_mark(f.available or f.active or f.managed_by_nous)} {state}")
elif ctx.nous_logged_in or ctx.nous_inference_present:
# Nous OAuth without entitlement, or an opaque inference key without
# Portal account information, cannot enable the Tool Gateway.
_section("Nous Tool Gateway")
message = format_nous_portal_entitlement_message(
ctx.nous_account_info, capability="managed web, image, TTS, STT, browser, and Modal tools"
)
for line in (message or "").splitlines():
print(f" {line}")
def _render_apikey_providers(ctx):
from hermes_cli.status import _effective_provider_label, _first_env_value, _section, check_mark, get_env_value
_section("API-Key Providers")
for pname, env_vars in _APIKEY_PROVIDERS.items():
configured = bool(_first_env_value(env_vars))
print(f" {pname:<16} {check_mark(configured)} {'configured' if configured else 'not configured (run: hermes model)'}")
# LM Studio reachability: probe only when it is the active provider so users
# with foreign configs see no noise. Auth rejection vs. a silent empty list
# is the most common LM Studio support case.
if _effective_provider_label() == "LM Studio":
from hermes_cli.models import probe_lmstudio_models
model_cfg = ctx.config.get("model")
base = (model_cfg.get("base_url") if isinstance(model_cfg, dict) else None) or get_env_value("LM_BASE_URL") or "http://127.0.0.1:1234/v1"
try:
models = probe_lmstudio_models(api_key=get_env_value("LM_API_KEY") or "", base_url=base, timeout=1.5)
ok = models is not None
msg = f"reachable ({len(models)} model(s)) at {base}" if ok else f"unreachable at {base}"
except AuthError:
ok, msg = False, "auth rejected — set LM_API_KEY"
print(f" {'LM Studio':<16} {check_mark(ok)} {msg}")