/ hermes_cli / status.py
status.py
1 """ 2 Status command for hermes CLI. 3 4 Shows the status of all Hermes Agent components. 5 """ 6 7 import os 8 import sys 9 import subprocess # noqa: F401 — re-exported for tests that monkeypatch status.subprocess to guard against regressions 10 import importlib.util 11 from pathlib import Path 12 13 PROJECT_ROOT = Path(__file__).parent.parent.resolve() 14 15 from hermes_cli.auth import AuthError, resolve_provider 16 from hermes_cli.colors import Colors, color 17 from hermes_cli.config import get_env_path, get_env_value, get_hermes_home, load_config 18 from hermes_cli.models import provider_label 19 from hermes_cli.nous_subscription import get_nous_subscription_features 20 from hermes_cli.runtime_provider import resolve_requested_provider 21 from hermes_cli.vercel_auth import describe_vercel_auth 22 from hermes_constants import OPENROUTER_MODELS_URL 23 from tools.tool_backend_helpers import managed_nous_tools_enabled 24 25 def check_mark(ok: bool) -> str: 26 if ok: 27 return color("✓", Colors.GREEN) 28 return color("✗", Colors.RED) 29 30 def redact_key(key: str) -> str: 31 """Redact an API key for display. 32 33 Thin wrapper over :func:`agent.redact.mask_secret`. Preserves the 34 "(not set)" placeholder in dim color to match ``hermes config``'s 35 output (previously this variant was missing the DIM color — 36 consolidated via PR that also introduced ``mask_secret``). 37 """ 38 from agent.redact import mask_secret 39 return mask_secret(key, empty=color("(not set)", Colors.DIM)) 40 41 42 def _format_iso_timestamp(value) -> str: 43 """Format ISO timestamps for status output, converting to local timezone.""" 44 if not value or not isinstance(value, str): 45 return "(unknown)" 46 from datetime import datetime, timezone 47 text = value.strip() 48 if not text: 49 return "(unknown)" 50 if text.endswith("Z"): 51 text = text[:-1] + "+00:00" 52 try: 53 parsed = datetime.fromisoformat(text) 54 if parsed.tzinfo is None: 55 parsed = parsed.replace(tzinfo=timezone.utc) 56 except Exception: 57 return value 58 return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") 59 60 61 def _configured_model_label(config: dict) -> str: 62 """Return the configured default model from config.yaml.""" 63 model_cfg = config.get("model") 64 if isinstance(model_cfg, dict): 65 model = (model_cfg.get("default") or model_cfg.get("name") or "").strip() 66 elif isinstance(model_cfg, str): 67 model = model_cfg.strip() 68 else: 69 model = "" 70 return model or "(not set)" 71 72 73 def _effective_provider_label() -> str: 74 """Return the provider label matching current CLI runtime resolution.""" 75 requested = resolve_requested_provider() 76 try: 77 effective = resolve_provider(requested) 78 except AuthError: 79 effective = requested or "auto" 80 81 if effective == "openrouter" and get_env_value("OPENAI_BASE_URL"): 82 effective = "custom" 83 84 return provider_label(effective) 85 86 87 from hermes_constants import is_termux as _is_termux 88 89 90 def show_status(args): 91 """Show status of all Hermes Agent components.""" 92 show_all = getattr(args, 'all', False) 93 deep = getattr(args, 'deep', False) 94 95 print() 96 print(color("┌─────────────────────────────────────────────────────────┐", Colors.CYAN)) 97 print(color("│ ⚕ Hermes Agent Status │", Colors.CYAN)) 98 print(color("└─────────────────────────────────────────────────────────┘", Colors.CYAN)) 99 100 # ========================================================================= 101 # Environment 102 # ========================================================================= 103 print() 104 print(color("◆ Environment", Colors.CYAN, Colors.BOLD)) 105 print(f" Project: {PROJECT_ROOT}") 106 print(f" Python: {sys.version.split()[0]}") 107 108 env_path = get_env_path() 109 print(f" .env file: {check_mark(env_path.exists())} {'exists' if env_path.exists() else 'not found'}") 110 111 try: 112 config = load_config() 113 except Exception: 114 config = {} 115 116 print(f" Model: {_configured_model_label(config)}") 117 print(f" Provider: {_effective_provider_label()}") 118 119 # ========================================================================= 120 # API Keys 121 # ========================================================================= 122 print() 123 print(color("◆ API Keys", Colors.CYAN, Colors.BOLD)) 124 125 # Values may be a single env var name (str) or a tuple of alternates (first found wins). 126 keys: dict[str, str | tuple[str, ...]] = { 127 "OpenRouter": "OPENROUTER_API_KEY", 128 "OpenAI": "OPENAI_API_KEY", 129 "Anthropic": ("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN"), 130 "Google / Gemini": ("GOOGLE_API_KEY", "GEMINI_API_KEY"), 131 "DeepSeek": "DEEPSEEK_API_KEY", 132 "xAI / Grok": "XAI_API_KEY", 133 "NVIDIA NIM": "NVIDIA_API_KEY", 134 "Z.AI / GLM": "GLM_API_KEY", 135 "Kimi": "KIMI_API_KEY", 136 "StepFun Step Plan": "STEPFUN_API_KEY", 137 "MiniMax": "MINIMAX_API_KEY", 138 "MiniMax-CN": "MINIMAX_CN_API_KEY", 139 "Firecrawl": "FIRECRAWL_API_KEY", 140 "Tavily": "TAVILY_API_KEY", 141 "Browser Use": "BROWSER_USE_API_KEY", # Optional — local browser works without this 142 "Browserbase": "BROWSERBASE_API_KEY", # Optional — direct credentials only 143 "FAL": "FAL_KEY", 144 "Tinker": "TINKER_API_KEY", 145 "WandB": "WANDB_API_KEY", 146 "ElevenLabs": "ELEVENLABS_API_KEY", 147 "GitHub": "GITHUB_TOKEN", 148 } 149 150 def _resolve_env(env_ref) -> str: 151 """Return first non-empty env var value from a str or tuple of names.""" 152 if isinstance(env_ref, tuple): 153 for candidate in env_ref: 154 v = get_env_value(candidate) or "" 155 if v: 156 return v 157 return "" 158 return get_env_value(env_ref) or "" 159 160 for name, env_ref in keys.items(): 161 # Anthropic already has a dedicated lookup below; keep that as the 162 # single source of truth (it also resolves OAuth tokens), skip here 163 # so we don't print two "Anthropic" rows. 164 if name == "Anthropic": 165 continue 166 value = _resolve_env(env_ref) 167 has_key = bool(value) 168 display = redact_key(value) if not show_all else value 169 print(f" {name:<12} {check_mark(has_key)} {display}") 170 171 from hermes_cli.auth import get_anthropic_key 172 anthropic_value = get_anthropic_key() 173 anthropic_display = redact_key(anthropic_value) if not show_all else anthropic_value 174 print(f" {'Anthropic':<12} {check_mark(bool(anthropic_value))} {anthropic_display}") 175 176 # ========================================================================= 177 # Auth Providers (OAuth) 178 # ========================================================================= 179 print() 180 print(color("◆ Auth Providers", Colors.CYAN, Colors.BOLD)) 181 182 try: 183 from hermes_cli.auth import ( 184 get_nous_auth_status, 185 get_codex_auth_status, 186 get_qwen_auth_status, 187 get_minimax_oauth_auth_status, 188 ) 189 nous_status = get_nous_auth_status() 190 codex_status = get_codex_auth_status() 191 qwen_status = get_qwen_auth_status() 192 minimax_status = get_minimax_oauth_auth_status() 193 except Exception: 194 nous_status = {} 195 codex_status = {} 196 qwen_status = {} 197 minimax_status = {} 198 199 nous_logged_in = bool(nous_status.get("logged_in")) 200 nous_error = nous_status.get("error") 201 nous_label = "logged in" if nous_logged_in else "not logged in (run: hermes auth add nous --type oauth)" 202 print( 203 f" {'Nous Portal':<12} {check_mark(nous_logged_in)} " 204 f"{nous_label}" 205 ) 206 portal_url = nous_status.get("portal_base_url") or "(unknown)" 207 access_exp = _format_iso_timestamp(nous_status.get("access_expires_at")) 208 key_exp = _format_iso_timestamp(nous_status.get("agent_key_expires_at")) 209 refresh_label = "yes" if nous_status.get("has_refresh_token") else "no" 210 if nous_logged_in or portal_url != "(unknown)" or nous_error: 211 print(f" Portal URL: {portal_url}") 212 if nous_logged_in or nous_status.get("access_expires_at"): 213 print(f" Access exp: {access_exp}") 214 if nous_logged_in or nous_status.get("agent_key_expires_at"): 215 print(f" Key exp: {key_exp}") 216 if nous_logged_in or nous_status.get("has_refresh_token"): 217 print(f" Refresh: {refresh_label}") 218 if nous_error and not nous_logged_in: 219 print(f" Error: {nous_error}") 220 221 codex_logged_in = bool(codex_status.get("logged_in")) 222 print( 223 f" {'OpenAI Codex':<12} {check_mark(codex_logged_in)} " 224 f"{'logged in' if codex_logged_in else 'not logged in (run: hermes model)'}" 225 ) 226 codex_auth_file = codex_status.get("auth_store") 227 if codex_auth_file: 228 print(f" Auth file: {codex_auth_file}") 229 codex_last_refresh = _format_iso_timestamp(codex_status.get("last_refresh")) 230 if codex_status.get("last_refresh"): 231 print(f" Refreshed: {codex_last_refresh}") 232 if codex_status.get("error") and not codex_logged_in: 233 print(f" Error: {codex_status.get('error')}") 234 235 qwen_logged_in = bool(qwen_status.get("logged_in")) 236 print( 237 f" {'Qwen OAuth':<12} {check_mark(qwen_logged_in)} " 238 f"{'logged in' if qwen_logged_in else 'not logged in (run: qwen auth qwen-oauth)'}" 239 ) 240 qwen_auth_file = qwen_status.get("auth_file") 241 if qwen_auth_file: 242 print(f" Auth file: {qwen_auth_file}") 243 qwen_exp = qwen_status.get("expires_at_ms") 244 if qwen_exp: 245 from datetime import datetime, timezone 246 print(f" Access exp: {datetime.fromtimestamp(int(qwen_exp) / 1000, tz=timezone.utc).isoformat()}") 247 if qwen_status.get("error") and not qwen_logged_in: 248 print(f" Error: {qwen_status.get('error')}") 249 250 minimax_logged_in = bool(minimax_status.get("logged_in")) 251 print( 252 f" {'MiniMax OAuth':<12} {check_mark(minimax_logged_in)} " 253 f"{'logged in' if minimax_logged_in else 'not logged in (run: hermes auth add minimax-oauth)'}" 254 ) 255 minimax_region = minimax_status.get("region") 256 if minimax_logged_in and minimax_region: 257 print(f" Region: {minimax_region}") 258 minimax_exp = minimax_status.get("expires_at") 259 if minimax_exp: 260 print(f" Access exp: {minimax_exp}") 261 if minimax_status.get("error") and not minimax_logged_in: 262 print(f" Error: {minimax_status.get('error')}") 263 264 # ========================================================================= 265 # Nous Subscription Features 266 # ========================================================================= 267 if managed_nous_tools_enabled(): 268 features = get_nous_subscription_features(config) 269 print() 270 print(color("◆ Nous Tool Gateway", Colors.CYAN, Colors.BOLD)) 271 if not features.nous_auth_present: 272 print(" Nous Portal ✗ not logged in") 273 else: 274 print(" Nous Portal ✓ managed tools available") 275 for feature in features.items(): 276 if feature.managed_by_nous: 277 state = "active via Nous subscription" 278 elif feature.active: 279 current = feature.current_provider or "configured provider" 280 state = f"active via {current}" 281 elif feature.included_by_default and features.nous_auth_present: 282 state = "included by subscription, not currently selected" 283 elif feature.key == "modal" and features.nous_auth_present: 284 state = "available via subscription (optional)" 285 else: 286 state = "not configured" 287 print(f" {feature.label:<15} {check_mark(feature.available or feature.active or feature.managed_by_nous)} {state}") 288 elif nous_logged_in: 289 # Logged into Nous but on the free tier — show upgrade nudge 290 print() 291 print(color("◆ Nous Tool Gateway", Colors.CYAN, Colors.BOLD)) 292 print(" Your free-tier Nous account does not include Tool Gateway access.") 293 print(" Upgrade your subscription to unlock managed web, image, TTS, and browser tools.") 294 try: 295 portal_url = nous_status.get("portal_base_url", "").rstrip("/") 296 if portal_url: 297 print(f" Upgrade: {portal_url}") 298 except Exception: 299 pass 300 301 # ========================================================================= 302 # API-Key Providers 303 # ========================================================================= 304 print() 305 print(color("◆ API-Key Providers", Colors.CYAN, Colors.BOLD)) 306 307 apikey_providers = { 308 "Z.AI / GLM": ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), 309 "Kimi / Moonshot": ("KIMI_API_KEY",), 310 "StepFun Step Plan": ("STEPFUN_API_KEY",), 311 "MiniMax": ("MINIMAX_API_KEY",), 312 "MiniMax (China)": ("MINIMAX_CN_API_KEY",), 313 } 314 for pname, env_vars in apikey_providers.items(): 315 key_val = "" 316 for ev in env_vars: 317 key_val = get_env_value(ev) or "" 318 if key_val: 319 break 320 configured = bool(key_val) 321 label = "configured" if configured else "not configured (run: hermes model)" 322 print(f" {pname:<16} {check_mark(configured)} {label}") 323 324 # LM Studio reachability — only probe when it's the active provider so 325 # users with foreign configs don't see noise. Auth rejection vs. silent 326 # empty list is the most common LM Studio support case. 327 if _effective_provider_label() == "LM Studio": 328 from hermes_cli.models import probe_lmstudio_models 329 model_cfg = config.get("model") 330 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" 331 try: 332 models = probe_lmstudio_models(api_key=get_env_value("LM_API_KEY") or "", base_url=base, timeout=1.5) 333 if models is None: 334 ok, msg = False, f"unreachable at {base}" 335 else: 336 ok, msg = True, f"reachable ({len(models)} model(s)) at {base}" 337 except AuthError: 338 ok, msg = False, "auth rejected — set LM_API_KEY" 339 print(f" {'LM Studio':<16} {check_mark(ok)} {msg}") 340 341 # ========================================================================= 342 # Terminal Configuration 343 # ========================================================================= 344 print() 345 print(color("◆ Terminal Backend", Colors.CYAN, Colors.BOLD)) 346 347 terminal_cfg = config.get("terminal", {}) if isinstance(config.get("terminal"), dict) else {} 348 terminal_env = os.getenv("TERMINAL_ENV", "") 349 if not terminal_env: 350 terminal_env = terminal_cfg.get("backend", "local") 351 print(f" Backend: {terminal_env}") 352 353 if terminal_env == "ssh": 354 ssh_host = os.getenv("TERMINAL_SSH_HOST", "") 355 ssh_user = os.getenv("TERMINAL_SSH_USER", "") 356 print(f" SSH Host: {ssh_host or '(not set)'}") 357 print(f" SSH User: {ssh_user or '(not set)'}") 358 elif terminal_env == "docker": 359 docker_image = os.getenv("TERMINAL_DOCKER_IMAGE", "python:3.11-slim") 360 print(f" Docker Image: {docker_image}") 361 elif terminal_env == "daytona": 362 daytona_image = os.getenv("TERMINAL_DAYTONA_IMAGE", "nikolaik/python-nodejs:python3.11-nodejs20") 363 print(f" Daytona Image: {daytona_image}") 364 elif terminal_env == "vercel_sandbox": 365 runtime = os.getenv("TERMINAL_VERCEL_RUNTIME") or terminal_cfg.get("vercel_runtime") or "node24" 366 persist = os.getenv("TERMINAL_CONTAINER_PERSISTENT") 367 if persist is None: 368 persist_enabled = bool(terminal_cfg.get("container_persistent", True)) 369 else: 370 persist_enabled = persist.lower() in ("1", "true", "yes", "on") 371 auth_status = describe_vercel_auth() 372 sdk_ok = importlib.util.find_spec("vercel") is not None 373 sdk_label = "installed" if sdk_ok else "missing (install: pip install 'hermes-agent[vercel]')" 374 print(f" Runtime: {runtime}") 375 print(f" SDK: {check_mark(sdk_ok)} {sdk_label}") 376 print(f" Auth: {check_mark(auth_status.ok)} {auth_status.label}") 377 for line in auth_status.detail_lines: 378 print(f" Auth detail: {line}") 379 print(f" Persistence: {'snapshot filesystem' if persist_enabled else 'ephemeral filesystem'}") 380 print(" Processes: live processes do not survive cleanup, snapshots, or sandbox recreation") 381 382 sudo_password = os.getenv("SUDO_PASSWORD", "") 383 print(f" Sudo: {check_mark(bool(sudo_password))} {'enabled' if sudo_password else 'disabled'}") 384 385 # ========================================================================= 386 # Messaging Platforms 387 # ========================================================================= 388 print() 389 print(color("◆ Messaging Platforms", Colors.CYAN, Colors.BOLD)) 390 391 platforms = { 392 "Telegram": ("TELEGRAM_BOT_TOKEN", "TELEGRAM_HOME_CHANNEL"), 393 "Discord": ("DISCORD_BOT_TOKEN", "DISCORD_HOME_CHANNEL"), 394 "WhatsApp": ("WHATSAPP_ENABLED", None), 395 "Signal": ("SIGNAL_HTTP_URL", "SIGNAL_HOME_CHANNEL"), 396 "Slack": ("SLACK_BOT_TOKEN", None), 397 "Email": ("EMAIL_ADDRESS", "EMAIL_HOME_ADDRESS"), 398 "SMS": ("TWILIO_ACCOUNT_SID", "SMS_HOME_CHANNEL"), 399 "DingTalk": ("DINGTALK_CLIENT_ID", None), 400 "Feishu": ("FEISHU_APP_ID", "FEISHU_HOME_CHANNEL"), 401 "WeCom": ("WECOM_BOT_ID", "WECOM_HOME_CHANNEL"), 402 "WeCom Callback": ("WECOM_CALLBACK_CORP_ID", None), 403 "Weixin": ("WEIXIN_ACCOUNT_ID", "WEIXIN_HOME_CHANNEL"), 404 "BlueBubbles": ("BLUEBUBBLES_SERVER_URL", "BLUEBUBBLES_HOME_CHANNEL"), 405 "QQBot": ("QQ_APP_ID", "QQ_HOME_CHANNEL"), 406 "Yuanbao": ("YUANBAO_APP_ID", "YUANBAO_HOME_CHANNEL"), 407 } 408 409 for name, (token_var, home_var) in platforms.items(): 410 token = os.getenv(token_var, "") 411 has_token = bool(token) 412 413 home_channel = "" 414 if home_var: 415 home_channel = os.getenv(home_var, "") 416 # Back-compat: QQBot home channel was renamed from QQ_HOME_CHANNEL to QQBOT_HOME_CHANNEL 417 if not home_channel and home_var == "QQBOT_HOME_CHANNEL": 418 home_channel = os.getenv("QQ_HOME_CHANNEL", "") 419 420 status = "configured" if has_token else "not configured" 421 if home_channel: 422 status += f" (home: {home_channel})" 423 424 print(f" {name:<12} {check_mark(has_token)} {status}") 425 426 # Plugin-registered platforms 427 try: 428 from gateway.platform_registry import platform_registry 429 for entry in platform_registry.plugin_entries(): 430 configured = entry.check_fn() 431 status_str = "configured" if configured else "not configured" 432 label = entry.label 433 print(f" {label:<12} {check_mark(configured)} {status_str} (plugin)") 434 except Exception: 435 pass 436 437 # ========================================================================= 438 # Gateway Status 439 # ========================================================================= 440 print() 441 print(color("◆ Gateway Service", Colors.CYAN, Colors.BOLD)) 442 443 try: 444 from hermes_cli.gateway import get_gateway_runtime_snapshot, _format_gateway_pids 445 446 snapshot = get_gateway_runtime_snapshot() 447 is_running = snapshot.running 448 print(f" Status: {check_mark(is_running)} {'running' if is_running else 'stopped'}") 449 print(f" Manager: {snapshot.manager}") 450 if snapshot.gateway_pids: 451 print(f" PID(s): {_format_gateway_pids(snapshot.gateway_pids)}") 452 if snapshot.has_process_service_mismatch: 453 print(" Service: installed but not managing the current running gateway") 454 elif _is_termux() and not snapshot.gateway_pids: 455 print(" Start with: hermes gateway") 456 print(" Note: Android may stop background jobs when Termux is suspended") 457 elif snapshot.service_installed and not snapshot.service_running: 458 print(" Service: installed but stopped") 459 except Exception: 460 if _is_termux(): 461 print(f" Status: {color('unknown', Colors.DIM)}") 462 print(" Manager: Termux / manual process") 463 elif sys.platform.startswith('linux'): 464 print(f" Status: {color('unknown', Colors.DIM)}") 465 print(" Manager: systemd/manual") 466 elif sys.platform == 'darwin': 467 print(f" Status: {color('unknown', Colors.DIM)}") 468 print(" Manager: launchd") 469 else: 470 print(f" Status: {color('N/A', Colors.DIM)}") 471 print(" Manager: (not supported on this platform)") 472 473 # ========================================================================= 474 # Cron Jobs 475 # ========================================================================= 476 print() 477 print(color("◆ Scheduled Jobs", Colors.CYAN, Colors.BOLD)) 478 479 jobs_file = get_hermes_home() / "cron" / "jobs.json" 480 if jobs_file.exists(): 481 import json 482 try: 483 with open(jobs_file, encoding="utf-8") as f: 484 data = json.load(f) 485 jobs = data.get("jobs", []) 486 enabled_jobs = [j for j in jobs if j.get("enabled", True)] 487 print(f" Jobs: {len(enabled_jobs)} active, {len(jobs)} total") 488 except Exception: 489 print(" Jobs: (error reading jobs file)") 490 else: 491 print(" Jobs: 0") 492 493 # ========================================================================= 494 # Sessions 495 # ========================================================================= 496 print() 497 print(color("◆ Sessions", Colors.CYAN, Colors.BOLD)) 498 499 sessions_file = get_hermes_home() / "sessions" / "sessions.json" 500 if sessions_file.exists(): 501 import json 502 try: 503 with open(sessions_file, encoding="utf-8") as f: 504 data = json.load(f) 505 print(f" Active: {len(data)} session(s)") 506 except Exception: 507 print(" Active: (error reading sessions file)") 508 else: 509 print(" Active: 0") 510 511 # ========================================================================= 512 # Deep checks 513 # ========================================================================= 514 if deep: 515 print() 516 print(color("◆ Deep Checks", Colors.CYAN, Colors.BOLD)) 517 518 # Check OpenRouter connectivity 519 openrouter_key = os.getenv("OPENROUTER_API_KEY", "") 520 if openrouter_key: 521 try: 522 import httpx 523 response = httpx.get( 524 OPENROUTER_MODELS_URL, 525 headers={"Authorization": f"Bearer {openrouter_key}"}, 526 timeout=10 527 ) 528 ok = response.status_code == 200 529 print(f" OpenRouter: {check_mark(ok)} {'reachable' if ok else f'error ({response.status_code})'}") 530 except Exception as e: 531 print(f" OpenRouter: {check_mark(False)} error: {e}") 532 533 # Check gateway port 534 try: 535 import socket 536 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 537 sock.settimeout(1) 538 result = sock.connect_ex(('127.0.0.1', 18789)) 539 sock.close() 540 # Port in use = gateway likely running 541 port_in_use = result == 0 542 # This is informational, not necessarily bad 543 print(f" Port 18789: {'in use' if port_in_use else 'available'}") 544 except OSError: 545 pass 546 547 print() 548 print(color("─" * 60, Colors.DIM)) 549 print(color(" Run 'hermes doctor' for detailed diagnostics", Colors.DIM)) 550 print(color(" Run 'hermes setup' to configure", Colors.DIM)) 551 print()