From ebb9f312dc2372ee76e7c24b84eeef5552e3e39f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:54:55 -0700 Subject: [PATCH] refactor(status): extract credential sections into status_auth, unify redact_key - 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. --- hermes_cli/status.py | 390 +++++++------------------------------- hermes_cli/status_auth.py | 228 ++++++++++++++++++++++ 2 files changed, 297 insertions(+), 321 deletions(-) create mode 100644 hermes_cli/status_auth.py diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 4c683aef70..26e7d9ab00 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -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() diff --git a/hermes_cli/status_auth.py b/hermes_cli/status_auth.py new file mode 100644 index 0000000000..effba64a61 --- /dev/null +++ b/hermes_cli/status_auth.py @@ -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}")