diff --git a/hermes_cli/setup_summary.py b/hermes_cli/setup_summary.py index 552696b10c..934597d52d 100644 --- a/hermes_cli/setup_summary.py +++ b/hermes_cli/setup_summary.py @@ -1,7 +1,7 @@ """Setup-completion summary (tool availability + "Setup Complete!" banner). -Extracted from hermes_cli/setup.py. Names originating in setup.py are imported lazily -inside function bodies so test patches on ``hermes_cli.setup.`` take effect. +Names originating in setup.py are imported lazily inside function bodies so test patches on +``hermes_cli.setup.`` take effect. """ import logging @@ -17,16 +17,14 @@ _TTS_SUMMARY_ROWS = { "mistral": ("Mistral Voxtral", ("MISTRAL_API_KEY",)), "gemini": ("Google Gemini", ("GEMINI_API_KEY", "GOOGLE_API_KEY")), "neutts": ("NeuTTS", "neutts", "run 'hermes setup tts'"), - "kittentts": ("KittenTTS", "kittentts", "run 'hermes setup tts'"), -} + "kittentts": ("KittenTTS", "kittentts", "run 'hermes setup tts'")} _TTS_SUMMARY_DEFAULT = ("Edge TTS", ()) _STT_SUMMARY_ROWS = { "openai": ("OpenAI", ("VOICE_TOOLS_OPENAI_KEY", "OPENAI_API_KEY")), "groq": ("Groq Whisper", ("GROQ_API_KEY",)), "elevenlabs": ("ElevenLabs Scribe", ("ELEVENLABS_API_KEY",)), "xai": ("xAI", ()), - "deepinfra": ("DeepInfra", ("DEEPINFRA_API_KEY",)), -} + "deepinfra": ("DeepInfra", ("DEEPINFRA_API_KEY",))} _STT_SUMMARY_DEFAULT = ("Local Whisper", "faster_whisper", "run 'hermes tools' → Speech-to-Text") # Browser "missing" hint keyed by the configured provider; anything else gets the generic hint. @@ -34,50 +32,66 @@ _BROWSER_MISSING_HINTS = { "Browserbase": "npm install -g agent-browser and set BROWSERBASE_API_KEY/BROWSERBASE_PROJECT_ID", "Browser Use": "npm install -g agent-browser and set BROWSER_USE_API_KEY", "Camofox": "CAMOFOX_URL", - "Local browser": "npm install -g agent-browser && agent-browser install --with-deps", -} + "Local browser": "npm install -g agent-browser && agent-browser install --with-deps"} _BROWSER_MISSING_DEFAULT = "npm install -g agent-browser, set CAMOFOX_URL, or configure Browser Use or Browserbase" +_WEB_MISSING = ("EXA_API_KEY, PARALLEL_API_KEY, FIRECRAWL_API_KEY/FIRECRAWL_API_URL, TAVILY_API_KEY, " + "KEENABLE_API_KEY, or SEARXNG_URL") _DONE_BANNER = ( "┌─────────────────────────────────────────────────────────┐", "│ ✓ Setup Complete! │", - "└─────────────────────────────────────────────────────────┘", -) + "└─────────────────────────────────────────────────────────┘") # (command, description) rows; the description carries its own alignment padding. _EDIT_WIZARD_ROWS = ( ("hermes setup", " Re-run the full wizard"), ("hermes setup model", " Change model/provider"), ("hermes setup terminal", " Change terminal backend"), ("hermes setup gateway", " Configure messaging"), - ("hermes setup tools", " Configure tool providers"), -) + ("hermes setup tools", " Configure tool providers")) _EDIT_CONFIG_ROWS = ( ("hermes config", " View current settings"), ("hermes config edit", " Open config in your editor"), - ("hermes config set ", ""), -) + ("hermes config set ", "")) _READY_ROWS = ( - ("hermes", " Start chatting"), - ("hermes gateway", " Start messaging gateway"), - ("hermes doctor", " Check for issues"), -) + ("hermes", " Start chatting"), ("hermes gateway", " Start messaging gateway"), + ("hermes doctor", " Check for issues")) def _voice_provider_status(kind: str, provider: str, rows: dict, default: tuple) -> tuple: """Summary row for a TTS/STT provider. A keyed provider whose key is missing falls through to the default row, matching the runtime fallback.""" - from hermes_cli.setup import get_env_value, _module_installed row = rows.get(provider, default) - if isinstance(row[1], tuple) and row[1] and not any(get_env_value(v) for v in row[1]): + if isinstance(row[1], tuple) and row[1] and not any(_setup.get_env_value(v) for v in row[1]): row = default if isinstance(row[1], tuple): return (f"{kind} ({row[0]})", True, None) label, module, hint = row - if _module_installed(module): + if _setup._module_installed(module): return (f"{kind} ({label}{' local' if kind == 'Text-to-Speech' else ''})", True, None) return (f"{kind} ({label} — not installed)", False, hint) +def _first_available_plugin_provider(registry: str, skip: str = None): + """display_name of the first plugin-registered provider in ``agent.`` that reports + available (fail-soft: any error means none), skipping ``skip``.""" + try: + import importlib + from hermes_cli.plugins import _ensure_plugins_discovered + + _ensure_plugins_discovered() + for provider in importlib.import_module(f"agent.{registry}").list_providers(): + if provider.name == skip: + continue + try: + if provider.is_available(): + return provider.display_name + except Exception: + continue + except Exception: + pass + return None + + # ---- tool_status row builders: each takes (config, subscription_features) and returns # a (name, available, hint) row or None (row omitted). Evaluated in _TOOL_ROW_BUILDERS order. @@ -86,34 +100,30 @@ def _vision_row(config, feats): try: from agent.auxiliary_client import get_available_vision_backends - _vision_backends = get_available_vision_backends() + ok = bool(get_available_vision_backends()) except Exception: - _vision_backends = [] - if _vision_backends: - return ("Vision (image analysis)", True, None) - return ("Vision (image analysis)", False, "run 'hermes setup' to configure") + ok = False + return ("Vision (image analysis)", ok, None if ok else "run 'hermes setup' to configure") + + +def _managed_or_provider_row(feature, name: str, managed_label: str, missing_hint: str): + """Row for a Nous-manageable feature: managed > available (with provider) > missing hint.""" + if feature.managed_by_nous: + return (f"{name} ({managed_label})", True, None) + if feature.available: + return (f"{name} ({feature.current_provider})" if feature.current_provider else name, True, None) + return (name, False, missing_hint) def _web_row(config, feats): # Web tools (Exa, Parallel, Firecrawl, Tavily, or Keenable) - web = feats.web - if web.managed_by_nous: - return ("Web Search & Extract (Nous subscription)", True, None) - if web.available: - label = f"Web Search & Extract ({web.current_provider})" if web.current_provider else "Web Search & Extract" - return (label, True, None) - return ("Web Search & Extract", False, "EXA_API_KEY, PARALLEL_API_KEY, FIRECRAWL_API_KEY/FIRECRAWL_API_URL, TAVILY_API_KEY, KEENABLE_API_KEY, or SEARXNG_URL") + return _managed_or_provider_row(feats.web, "Web Search & Extract", "Nous subscription", _WEB_MISSING) def _browser_row(config, feats): # Browser tools (local Chromium, Camofox, Browserbase, Browser Use, or Firecrawl) - browser_provider = feats.browser.current_provider - if feats.browser.managed_by_nous: - return ("Browser Automation (Nous Browser Use)", True, None) - if feats.browser.available: - label = f"Browser Automation ({browser_provider})" if browser_provider else "Browser Automation" - return (label, True, None) - return ("Browser Automation", False, _BROWSER_MISSING_HINTS.get(browser_provider, _BROWSER_MISSING_DEFAULT)) + hint = _BROWSER_MISSING_HINTS.get(feats.browser.current_provider, _BROWSER_MISSING_DEFAULT) + return _managed_or_provider_row(feats.browser, "Browser Automation", "Nous Browser Use", hint) def _image_gen_row(config, feats): @@ -123,25 +133,9 @@ def _image_gen_row(config, feats): if feats.image_gen.available: return ("Image Generation", True, None) # Probe plugin-registered providers so OpenAI-only setups don't show as "missing FAL_KEY". - _img_backend = None - try: - from agent.image_gen_registry import list_providers - from hermes_cli.plugins import _ensure_plugins_discovered - - _ensure_plugins_discovered() - for _p in list_providers(): - if _p.name == "fal": - continue - try: - if _p.is_available(): - _img_backend = _p.display_name - break - except Exception: - continue - except Exception: - pass - if _img_backend: - return (f"Image Generation ({_img_backend})", True, None) + backend = _first_available_plugin_provider("image_gen_registry", skip="fal") + if backend: + return (f"Image Generation ({backend})", True, None) return ("Image Generation", False, "FAL_KEY or OPENAI_API_KEY") @@ -150,35 +144,20 @@ def _video_gen_row(config, feats): # available so we don't badger users who don't care about video gen with a "missing" line. if feats.video_gen.managed_by_nous: return ("Video Generation (FAL via Nous subscription)", True, None) - try: - from agent.video_gen_registry import list_providers as _list_video_providers - from hermes_cli.plugins import _ensure_plugins_discovered as _ensure_plugins - _ensure_plugins() - _video_backend = None - for _vp in _list_video_providers(): - try: - if _vp.is_available(): - _video_backend = _vp.display_name - break - except Exception: - continue - except Exception: - _video_backend = None - return (f"Video Generation ({_video_backend})", True, None) if _video_backend else None + backend = _first_available_plugin_provider("video_gen_registry") + return (f"Video Generation ({backend})", True, None) if backend else None def _tts_row(config, feats): # Configured provider, gated on its key (or local install) - from hermes_cli.setup import cfg_get - tts_provider = cfg_get(config, "tts", "provider", default="edge") + tts_provider = _setup.cfg_get(config, "tts", "provider", default="edge") if feats.tts.managed_by_nous: return ("Text-to-Speech (OpenAI via Nous subscription)", True, None) return _voice_provider_status("Text-to-Speech", tts_provider, _TTS_SUMMARY_ROWS, _TTS_SUMMARY_DEFAULT) def _stt_row(config, feats): - from hermes_cli.setup import cfg_get - stt_provider = cfg_get(config, "stt", "provider", default="local") or "local" + stt_provider = _setup.cfg_get(config, "stt", "provider", default="local") or "local" _stt_feature = feats.features.get("stt") if _stt_feature is not None and _stt_feature.managed_by_nous: return ("Speech-to-Text (OpenAI via Nous subscription)", True, None) @@ -186,21 +165,19 @@ def _stt_row(config, feats): def _modal_row(config, feats): - from hermes_cli.setup import cfg_get, managed_nous_tools_enabled if feats.modal.managed_by_nous: return ("Modal Execution (Nous subscription)", True, None) - if cfg_get(config, "terminal", "backend") == "modal": + if _setup.cfg_get(config, "terminal", "backend") == "modal": if feats.modal.direct_override: return ("Modal Execution (direct Modal)", True, None) return ("Modal Execution", False, "run 'hermes setup terminal'") - if managed_nous_tools_enabled() and feats.nous_auth_present: + if _setup.managed_nous_tools_enabled() and feats.nous_auth_present: return ("Modal Execution (optional via Nous subscription)", True, None) return None def _home_assistant_row(config, feats): - from hermes_cli.setup import get_env_value - return ("Smart Home (Home Assistant)", True, None) if get_env_value("HASS_TOKEN") else None + return ("Smart Home (Home Assistant)", True, None) if _setup.get_env_value("HASS_TOKEN") else None def _spotify_row(config, feats): @@ -216,99 +193,83 @@ def _spotify_row(config, feats): def _skills_hub_row(config, feats): - from hermes_cli.setup import get_env_value - if get_env_value("GITHUB_TOKEN"): - return ("Skills Hub (GitHub)", True, None) - return ("Skills Hub (GitHub)", False, "GITHUB_TOKEN") + ok = bool(_setup.get_env_value("GITHUB_TOKEN")) + return ("Skills Hub (GitHub)", ok, None if ok else "GITHUB_TOKEN") def _always_on_rows(config, feats): # Terminal (system deps met), task planning (in-memory), skills (bundled + user-created). - return [ - ("Terminal/Commands", True, None), - ("Task Planning (todo)", True, None), - ("Skills (view, create, edit)", True, None), - ] + return [("Terminal/Commands", True, None), ("Task Planning (todo)", True, None), + ("Skills (view, create, edit)", True, None)] _TOOL_ROW_BUILDERS = ( _vision_row, _web_row, _browser_row, _image_gen_row, _video_gen_row, _tts_row, _stt_row, - _modal_row, _home_assistant_row, _spotify_row, _skills_hub_row, _always_on_rows, -) + _modal_row, _home_assistant_row, _spotify_row, _skills_hub_row, _always_on_rows) def _print_cmd_rows(rows): """Print (command, description) rows as ' '.""" - from hermes_cli.setup import color, Colors for cmd, desc in rows: - print(f" {color(cmd, Colors.GREEN)}{desc}") + print(f" {_setup.color(cmd, _setup.Colors.GREEN)}{desc}") def _print_section_header(title): - from hermes_cli.setup import color, Colors - print(color("─" * 60, Colors.DIM)) + print(_setup.color("─" * 60, _setup.Colors.DIM)) print() - print(color(title, Colors.CYAN, Colors.BOLD)) + print(_setup.color(title, _setup.Colors.CYAN, _setup.Colors.BOLD)) print() def _print_setup_summary(config: dict, hermes_home): """Print the setup completion summary.""" - from hermes_cli.setup import ( - color, Colors, get_config_path, get_env_path, get_nous_subscription_features, _info, print_header, - print_warning, - ) + from hermes_constants import display_hermes_home as _dhh # Provider readiness — the one thing setup must produce. A user who cancelled the API-key # prompt mid-wizard used to exit "successfully" with NO working model; say so loudly. try: from hermes_cli.auth import resolve_provider resolve_provider() - _provider_ready = True except Exception: - _provider_ready = False - if not _provider_ready: print() - print_warning("No inference provider is configured — Hermes cannot chat yet.") - _info(" Finish this one step with either of:", + _setup.print_warning("No inference provider is configured — Hermes cannot chat yet.") + _setup._info(" Finish this one step with either of:", " hermes model (pick any provider/model)", " hermes setup --portal (Nous Portal OAuth, no API key)") print() - print_header("Tool Availability Summary") + _setup.print_header("Tool Availability Summary") tool_status = [] - subscription_features = get_nous_subscription_features(config) + subscription_features = _setup.get_nous_subscription_features(config) for build in _TOOL_ROW_BUILDERS: row = build(config, subscription_features) tool_status.extend(row if isinstance(row, list) else [row] if row is not None else []) available_count = sum(1 for _, avail, _ in tool_status if avail) - _info(f"{available_count}/{len(tool_status)} tool categories available:", None) + _setup._info(f"{available_count}/{len(tool_status)} tool categories available:", None) for name, available, missing_var in tool_status: if available: - print(f" {color('✓', Colors.GREEN)} {name}") + print(f" {_setup.color('✓', _setup.Colors.GREEN)} {name}") else: - print(f" {color('✗', Colors.RED)} {name} {color(f'(missing {missing_var})', Colors.DIM)}") + print(f" {_setup.color('✗', _setup.Colors.RED)} {name} " + f"{_setup.color(f'(missing {missing_var})', _setup.Colors.DIM)}") print() - if any(not avail for _, avail, _ in tool_status): - print_warning("Some tools are disabled. Run 'hermes setup tools' to configure them,") - from hermes_constants import display_hermes_home as _dhh - print_warning(f"or edit {_dhh()}/.env directly to add the missing API keys.") + if available_count < len(tool_status): + _setup.print_warning("Some tools are disabled. Run 'hermes setup tools' to configure them,") + _setup.print_warning(f"or edit {_dhh()}/.env directly to add the missing API keys.") print() print() for line in _DONE_BANNER: - print(color(line, Colors.GREEN)) + print(_setup.color(line, _setup.Colors.GREEN)) print() - - from hermes_constants import display_hermes_home as _dhh - print(color(f"📁 All your files are in {_dhh()}/:", Colors.CYAN, Colors.BOLD)) + print(_setup.color(f"📁 All your files are in {_dhh()}/:", _setup.Colors.CYAN, _setup.Colors.BOLD)) print() - print(f" {color('Settings:', Colors.YELLOW)} {get_config_path()}") - print(f" {color('API Keys:', Colors.YELLOW)} {get_env_path()}") - print(f" {color('Data:', Colors.YELLOW)} {hermes_home}/cron/, sessions/, logs/") + print(f" {_setup.color('Settings:', _setup.Colors.YELLOW)} {_setup.get_config_path()}") + print(f" {_setup.color('API Keys:', _setup.Colors.YELLOW)} {_setup.get_env_path()}") + print(f" {_setup.color('Data:', _setup.Colors.YELLOW)} {hermes_home}/cron/, sessions/, logs/") print() _print_section_header("📝 To edit your configuration:") @@ -318,10 +279,13 @@ def _print_setup_summary(config: dict, hermes_home): print(" Set a specific value") print() print(" Or edit the files directly:") - print(f" {color(f'nano {get_config_path()}', Colors.DIM)}") - print(f" {color(f'nano {get_env_path()}', Colors.DIM)}") + print(f" {_setup.color(f'nano {_setup.get_config_path()}', _setup.Colors.DIM)}") + print(f" {_setup.color(f'nano {_setup.get_env_path()}', _setup.Colors.DIM)}") print() _print_section_header("🚀 Ready to go!") _print_cmd_rows(_READY_ROWS) print() + + +import hermes_cli.setup as _setup # noqa: E402 (bottom: hermes_cli.setup imports this module) diff --git a/hermes_cli/setup_terminal.py b/hermes_cli/setup_terminal.py index 06883779b8..676f6f3218 100644 --- a/hermes_cli/setup_terminal.py +++ b/hermes_cli/setup_terminal.py @@ -1,6 +1,7 @@ """Terminal-backend setup wizard (local/docker/singularity/modal/daytona/vercel/ssh/plugin). -Extracted from hermes_cli/setup.py; setup.py re-exports the public entry points. +Names from setup.py are imported lazily inside bodies so test patches on +``hermes_cli.setup.`` take effect; setup.py re-exports the public entry points. """ import json @@ -13,63 +14,55 @@ from pathlib import Path logger = logging.getLogger("hermes_cli.setup") _SANDBOX_IMAGE = "nikolaik/python-nodejs:python3.11-nodejs20" +_RUN_KW = dict(capture_output=True, text=True, encoding="utf-8", errors="replace") def _prompt_vercel_sandbox_settings(config: dict): """Prompt for Vercel Sandbox settings without exposing unsupported disk sizing.""" - from hermes_cli.setup import ( - get_env_value, _info, print_info, print_warning, prompt, remove_env_value, save_env_value, - ) terminal = config.setdefault("terminal", {}) - _info(None, "Vercel Sandbox settings:", " Filesystem persistence uses Vercel snapshots.", + _setup._info(None, "Vercel Sandbox settings:", " Filesystem persistence uses Vercel snapshots.", " Snapshots restore files only; live processes do not continue after sandbox recreation.") from tools.terminal_tool import _SUPPORTED_VERCEL_RUNTIMES - current_runtime = terminal.get("vercel_runtime") or "node24" supported_label = ", ".join(_SUPPORTED_VERCEL_RUNTIMES) - runtime = prompt(f" Runtime ({supported_label})", current_runtime).strip() or current_runtime + runtime = _setup.prompt(f" Runtime ({supported_label})", current_runtime).strip() or current_runtime if runtime not in _SUPPORTED_VERCEL_RUNTIMES: - print_warning(f"Unsupported Vercel runtime '{runtime}', keeping {current_runtime}.") + _setup.print_warning(f"Unsupported Vercel runtime '{runtime}', keeping {current_runtime}.") runtime = current_runtime if current_runtime in _SUPPORTED_VERCEL_RUNTIMES else "node24" terminal["vercel_runtime"] = runtime - save_env_value("TERMINAL_VERCEL_RUNTIME", runtime) + _setup.save_env_value("TERMINAL_VERCEL_RUNTIME", runtime) persist_label = "yes" if terminal.get("container_persistent", True) else "no" - terminal["container_persistent"] = ( - prompt(" Persist filesystem with snapshots? (yes/no)", persist_label).lower() in {"yes", "true", "y", "1"} - ) - + persist = _setup.prompt(" Persist filesystem with snapshots? (yes/no)", persist_label).lower() + terminal["container_persistent"] = persist in {"yes", "true", "y", "1"} # (key, prompt label, default, parser) — unparseable input leaves the value untouched. for key, label, default, parse in ( ("container_cpu", " CPU cores", 1, float), - ("container_memory", " Memory in MB (5120 = 5GB)", 5120, int), - ): + ("container_memory", " Memory in MB (5120 = 5GB)", 5120, int)): try: - terminal[key] = parse(prompt(label, str(terminal.get(key, default)))) + terminal[key] = parse(_setup.prompt(label, str(terminal.get(key, default)))) except ValueError: pass - if terminal.get("container_disk", 51200) not in {0, 51200}: - print_warning("Vercel Sandbox does not support custom disk sizing; resetting container_disk to 51200.") + _setup.print_warning( + "Vercel Sandbox does not support custom disk sizing; resetting container_disk to 51200.") terminal["container_disk"] = 51200 - _info(None, "Vercel authentication:", " Use a long-lived Vercel access token plus project/team IDs.") + _setup._info(None, "Vercel authentication:", " Use a long-lived Vercel access token plus project/team IDs.") linked = _read_nearest_vercel_project() if linked: - print_info(" Found defaults in nearest .vercel/project.json.") - - remove_env_value("VERCEL_OIDC_TOKEN") + _setup.print_info(" Found defaults in nearest .vercel/project.json.") + _setup.remove_env_value("VERCEL_OIDC_TOKEN") # (label, env var, linked-project fallback key, secret) — prompted in order, saved when non-empty. for label, env_var, linked_key, secret in ( (" Vercel access token", "VERCEL_TOKEN", None, True), (" Vercel project ID", "VERCEL_PROJECT_ID", "projectId", False), - (" Vercel team ID", "VERCEL_TEAM_ID", "orgId", False), - ): - default = get_env_value(env_var) or (linked.get(linked_key, "") if linked_key else "") - value = prompt(label, default, password=secret) + (" Vercel team ID", "VERCEL_TEAM_ID", "orgId", False)): + default = _setup.get_env_value(env_var) or (linked.get(linked_key, "") if linked_key else "") + value = _setup.prompt(label, default, password=secret) if value: - save_env_value(env_var, value) + _setup.save_env_value(env_var, value) def _read_nearest_vercel_project(start: Path | None = None) -> dict[str, str]: @@ -88,205 +81,175 @@ def _read_nearest_vercel_project(start: Path | None = None) -> dict[str, str]: if not isinstance(data, dict): return {} return { - key: value - for key, value in {"projectId": data.get("projectId"), "orgId": data.get("orgId")}.items() - if isinstance(value, str) and value.strip() - } + key: value for key, value in {"projectId": data.get("projectId"), "orgId": data.get("orgId")}.items() + if isinstance(value, str) and value.strip()} return {} def _prompt_secret_env(label: str, env_var: str, *, confirm_msg: str = "") -> None: """Prompt for a secret and persist it to .env when non-empty.""" - from hermes_cli.setup import print_success, prompt, save_env_value - value = prompt(label, password=True) + value = _setup.prompt(label, password=True) if value: - save_env_value(env_var, value) + _setup.save_env_value(env_var, value) if confirm_msg: - print_success(confirm_msg) + _setup.print_success(confirm_msg) -def _ensure_sdk(package: str, manual_hint: str, *, show_stderr: bool = False) -> None: - """Import *package*; if missing, install it via the venv pip helper.""" - from hermes_cli.setup import print_info, print_success, print_warning +def _pip_install_vercel(package): + """uv when Hermes has one ($HERMES_HOME/bin is never on PATH, so which() misses it and + bootstrapping mid-wizard is fine), else pip — a `uv venv` venv may not even have pip.""" + import subprocess + from hermes_cli.managed_uv import ensure_uv + uv_bin = ensure_uv() + cmd = ([uv_bin, "pip", "install", "--python", sys.executable, package] if uv_bin + else [sys.executable, "-m", "pip", "install", package]) + return subprocess.run(cmd, **_RUN_KW) + + +def _ensure_sdk(package: str, manual_hint: str, *, show_stderr: bool = False, install=None) -> None: + """Import *package*; if missing, install it (default: the venv pip ladder).""" try: __import__(package) except ImportError: - print_info(f"Installing {package} SDK...") - from hermes_cli.tools_config import _pip_install - - result = _pip_install([package]) + _setup.print_info(f"Installing {package} SDK...") + if install is None: + from hermes_cli.tools_config import _pip_install + install = lambda pkg: _pip_install([pkg]) # noqa: E731 + result = install(package) if result.returncode == 0: - print_success(f"{package} SDK installed") + _setup.print_success(f"{package} SDK installed") else: - print_warning(f"Install failed — run manually: {manual_hint}") + _setup.print_warning(f"Install failed — run manually: {manual_hint}") if show_stderr and result.stderr: - print_info(f" Error: {result.stderr.strip().splitlines()[-1]}") + _setup.print_info(f" Error: {result.stderr.strip().splitlines()[-1]}") + + +def _report_binary(found: str | None, missing: str, install_hint: str, found_prefix: str = "Found: ") -> None: + if found: + _setup.print_info(f"{found_prefix}{found}") + else: + _setup.print_warning(missing) + _setup.print_info(install_hint) def _setup_backend_local(config: dict) -> None: - from hermes_cli.setup import print_info, print_success - print_success("Terminal backend: Local") - print_info("Commands run directly on this machine.") - # Gateway working directory defaults to home; sudo stays off. Both are - # configurable later via `hermes setup terminal` / config.yaml. + _setup.print_success("Terminal backend: Local") + _setup.print_info("Commands run directly on this machine.") + # Gateway cwd defaults to home; sudo stays off. Both configurable via `hermes setup terminal`. config["terminal"].setdefault("cwd", str(Path.home())) def _setup_backend_docker(config: dict) -> None: - from hermes_cli.setup import _info, print_info, print_success, print_warning, prompt_yes_no - print_success("Terminal backend: Docker") - docker_bin = shutil.which("docker") - if docker_bin: - print_info(f"Docker found: {docker_bin}") - else: - print_warning("Docker not found in PATH!") - print_info("Install Docker: https://docs.docker.com/get-docker/") - + _setup.print_success("Terminal backend: Docker") + _report_binary(shutil.which("docker"), "Docker not found in PATH!", + "Install Docker: https://docs.docker.com/get-docker/", "Docker found: ") # Image and resource limits use defaults; tune via `hermes setup terminal`. config["terminal"].setdefault("docker_image", _SANDBOX_IMAGE) - _info(None, "Docker sandboxes can be protected with the egress credential firewall.", + _setup._info(None, "Docker sandboxes can be protected with the egress credential firewall.", "It routes sandbox traffic through iron-proxy so containers receive " - "proxy tokens instead of real API keys.", + "proxy tokens instead of real API keys.", " Docker only for now; Modal, SSH, Daytona, and Singularity are not wired yet.") - if prompt_yes_no(" Enable egress firewall for Docker sandboxes?", False): + if _setup.prompt_yes_no(" Enable egress firewall for Docker sandboxes?", False): proxy_cfg = config.setdefault("proxy", {}) proxy_cfg["enabled"] = True proxy_cfg.setdefault("enforce_on_docker", True) - print_success("Egress firewall enabled in config") - print_info("Run `hermes egress setup` then `hermes egress start` to mint tokens and launch the proxy.") + _setup.print_success("Egress firewall enabled in config") + _setup.print_info( + "Run `hermes egress setup` then `hermes egress start` to mint tokens and launch the proxy.") else: - print_info("Skipping egress firewall. You can enable it later with `hermes egress setup`.") + _setup.print_info("Skipping egress firewall. You can enable it later with `hermes egress setup`.") def _setup_backend_singularity(config: dict) -> None: - from hermes_cli.setup import print_info, print_success, print_warning - print_success("Terminal backend: Singularity/Apptainer") - sing_bin = shutil.which("apptainer") or shutil.which("singularity") - if sing_bin: - print_info(f"Found: {sing_bin}") - else: - print_warning("Singularity/Apptainer not found in PATH!") - print_info("Install: https://apptainer.org/docs/admin/main/installation.html") - # Image and resource limits use defaults; tune via `hermes setup terminal`. - config["terminal"].setdefault("singularity_image", "docker://nikolaik/python-nodejs:python3.11-nodejs20") + _setup.print_success("Terminal backend: Singularity/Apptainer") + _report_binary(shutil.which("apptainer") or shutil.which("singularity"), + "Singularity/Apptainer not found in PATH!", + "Install: https://apptainer.org/docs/admin/main/installation.html") + config["terminal"].setdefault("singularity_image", f"docker://{_SANDBOX_IMAGE}") def _setup_backend_modal(config: dict) -> None: - from hermes_cli.setup import ( - cfg_get, get_env_value, get_nous_subscription_features, _info, managed_nous_tools_enabled, - print_info, print_success, prompt_choice, prompt_yes_no, - ) - print_success("Terminal backend: Modal") - print_info("Serverless cloud sandboxes. Each session gets its own container.") + _setup.print_success("Terminal backend: Modal") + _setup.print_info("Serverless cloud sandboxes. Each session gets its own container.") from tools.managed_tool_gateway import is_managed_tool_gateway_ready from tools.tool_backend_helpers import normalize_modal_mode - managed_modal_available = bool( - managed_nous_tools_enabled() - and get_nous_subscription_features(config).nous_auth_present - and is_managed_tool_gateway_ready("modal") - ) - modal_mode = normalize_modal_mode(cfg_get(config, "terminal", "modal_mode")) + _setup.managed_nous_tools_enabled() + and _setup.get_nous_subscription_features(config).nous_auth_present + and is_managed_tool_gateway_ready("modal")) + modal_mode = normalize_modal_mode(_setup.cfg_get(config, "terminal", "modal_mode")) use_managed_modal = False if managed_modal_available: - if modal_mode in ("managed", "direct"): - default_modal_idx = 0 if modal_mode == "managed" else 1 - else: - default_modal_idx = 1 if get_env_value("MODAL_TOKEN_ID") else 0 - use_managed_modal = prompt_choice( + # Default to the configured mode; when unset, to "direct" only if Modal creds exist. + default_idx = {"managed": 0, "direct": 1}.get(modal_mode, 1 if _setup.get_env_value("MODAL_TOKEN_ID") else 0) + use_managed_modal = _setup.prompt_choice( "Select how Modal execution should be billed:", - ["Use my Nous subscription", "Use my own Modal account"], default_modal_idx, - ) == 0 + ["Use my Nous subscription", "Use my own Modal account"], default_idx) == 0 if use_managed_modal: config["terminal"]["modal_mode"] = "managed" - print_info("Modal execution will use the managed Nous gateway and bill to your subscription.") - if get_env_value("MODAL_TOKEN_ID") or get_env_value("MODAL_TOKEN_SECRET"): - print_info("Direct Modal credentials are still configured, but this backend is pinned to managed mode.") + _setup.print_info("Modal execution will use the managed Nous gateway and bill to your subscription.") + if _setup.get_env_value("MODAL_TOKEN_ID") or _setup.get_env_value("MODAL_TOKEN_SECRET"): + _setup.print_info( + "Direct Modal credentials are still configured, but this backend is pinned to managed mode.") return config["terminal"]["modal_mode"] = "direct" - print_info("Requires a Modal account: https://modal.com") + _setup.print_info("Requires a Modal account: https://modal.com") _ensure_sdk("modal", "uv pip install modal") - _info(None, "Modal authentication:", " Get your token at: https://modal.com/settings") - if get_env_value("MODAL_TOKEN_ID"): - print_info(" Modal token: already configured") - if not prompt_yes_no(" Update Modal credentials?", False): + _setup._info(None, "Modal authentication:", " Get your token at: https://modal.com/settings") + if _setup.get_env_value("MODAL_TOKEN_ID"): + _setup.print_info(" Modal token: already configured") + if not _setup.prompt_yes_no(" Update Modal credentials?", False): return _prompt_secret_env(" Modal Token ID", "MODAL_TOKEN_ID") _prompt_secret_env(" Modal Token Secret", "MODAL_TOKEN_SECRET") def _setup_backend_daytona(config: dict) -> None: - from hermes_cli.setup import get_env_value, _info, print_info, print_success, prompt_yes_no - print_success("Terminal backend: Daytona") - _info("Persistent cloud development environments.", + _setup.print_success("Terminal backend: Daytona") + _setup._info("Persistent cloud development environments.", "Each session gets a dedicated sandbox with filesystem persistence.", "Sign up at: https://daytona.io") _ensure_sdk("daytona", "uv pip install daytona", show_stderr=True) print() - if get_env_value("DAYTONA_API_KEY"): - print_info(" Daytona API key: already configured") - if prompt_yes_no(" Update API key?", False): + if _setup.get_env_value("DAYTONA_API_KEY"): + _setup.print_info(" Daytona API key: already configured") + if _setup.prompt_yes_no(" Update API key?", False): _prompt_secret_env(" Daytona API key", "DAYTONA_API_KEY", confirm_msg=" Updated") else: _prompt_secret_env(" Daytona API key", "DAYTONA_API_KEY", confirm_msg=" Configured") - # Image and resource limits use defaults; tune via `hermes setup terminal`. config["terminal"].setdefault("daytona_image", _SANDBOX_IMAGE) def _setup_backend_vercel(config: dict) -> None: - from hermes_cli.setup import _info, print_info, print_success, print_warning - print_success("Terminal backend: Vercel Sandbox") - _info("Cloud microVM sandboxes with snapshot-backed filesystem persistence.", + _setup.print_success("Terminal backend: Vercel Sandbox") + _setup._info("Cloud microVM sandboxes with snapshot-backed filesystem persistence.", "Requires the optional SDK: pip install 'hermes-agent[vercel]'") - try: - __import__("vercel") - except ImportError: - print_info("Installing vercel SDK...") - import subprocess - - # Managed uv first: $HERMES_HOME/bin is never on PATH so which() misses Hermes' uv; - # bootstrapping one mid-wizard is fine, and a `uv venv` venv may not even have pip. - from hermes_cli.managed_uv import ensure_uv - - uv_bin = ensure_uv() - cmd = [uv_bin, "pip", "install", "--python", sys.executable, "vercel"] if uv_bin else [ - sys.executable, "-m", "pip", "install", "vercel"] - result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace") - if result.returncode == 0: - print_success("vercel SDK installed") - else: - print_warning("Install failed — run manually: pip install 'hermes-agent[vercel]'") - if result.stderr: - print_info(f" Error: {result.stderr.strip().splitlines()[-1]}") + _ensure_sdk("vercel", "pip install 'hermes-agent[vercel]'", show_stderr=True, install=_pip_install_vercel) _prompt_vercel_sandbox_settings(config) def _setup_backend_ssh(config: dict) -> None: - from hermes_cli.setup import ( - get_env_value, print_info, print_success, print_warning, prompt, prompt_yes_no, save_env_value, - ) - print_success("Terminal backend: SSH") - print_info("Run commands on a remote machine via SSH.") + _setup.print_success("Terminal backend: SSH") + _setup.print_info("Run commands on a remote machine via SSH.") # (label, env var, fallback default when .env is empty); the port is only saved when not 22. fields = ( (" SSH host (hostname or IP)", "TERMINAL_SSH_HOST", ""), (" SSH user", "TERMINAL_SSH_USER", os.getenv("USER", "")), (" SSH port", "TERMINAL_SSH_PORT", "22"), - (" SSH private key path", "TERMINAL_SSH_KEY", str(Path.home() / ".ssh" / "id_rsa")), - ) + (" SSH private key path", "TERMINAL_SSH_KEY", str(Path.home() / ".ssh" / "id_rsa"))) values = [] for label, env_var, default in fields: - value = prompt(label, (get_env_value(env_var) or "") or default) + value = _setup.prompt(label, _setup.get_env_value(env_var) or default) values.append(value) if value and (env_var != "TERMINAL_SSH_PORT" or value != "22"): - save_env_value(env_var, value) + _setup.save_env_value(env_var, value) host, user, port, ssh_key = values - if host and prompt_yes_no(" Test SSH connection?", True): - print_info(" Testing connection...") + if host and _setup.prompt_yes_no(" Test SSH connection?", True): + _setup.print_info(" Testing connection...") import subprocess ssh_cmd = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5"] @@ -295,26 +258,25 @@ def _setup_backend_ssh(config: dict) -> None: if port and port != "22": ssh_cmd.extend(["-p", port]) ssh_cmd += [f"{user}@{host}" if user else host, "echo ok"] - result = subprocess.run(ssh_cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10) + result = subprocess.run(ssh_cmd, timeout=10, **_RUN_KW) if result.returncode == 0: - print_success(" SSH connection successful!") + _setup.print_success(" SSH connection successful!") else: - print_warning(f" SSH connection failed: {result.stderr.strip()}") - print_info(" Check your SSH key and host settings.") + _setup.print_warning(f" SSH connection failed: {result.stderr.strip()}") + _setup.print_info(" Check your SSH key and host settings.") def _setup_backend_plugin(config: dict, backend: str) -> None: - from hermes_cli.setup import print_info, print_success, print_warning try: from agent.terminal_env_registry import get_provider provider = get_provider(backend) - print_success(f"Terminal backend: {provider.display_name}") + _setup.print_success(f"Terminal backend: {provider.display_name}") for line in provider.setup_instructions(): - print_info(line) + _setup.print_info(line) provider.post_setup() except Exception as exc: - print_warning(f"Backend plugin setup hook failed: {exc}") + _setup.print_warning(f"Backend plugin setup hook failed: {exc}") _BUILTIN_TERMINAL_BACKENDS = [ @@ -323,28 +285,26 @@ _BUILTIN_TERMINAL_BACKENDS = [ ("modal", "Modal - serverless cloud sandbox"), ("ssh", "SSH - run on a remote machine"), ("daytona", "Daytona - persistent cloud development environment"), - ("vercel_sandbox", "Vercel Sandbox - cloud microVM with snapshot filesystem persistence"), -] + ("vercel_sandbox", "Vercel Sandbox - cloud microVM with snapshot filesystem persistence")] _TERMINAL_BACKEND_SETUP = { "local": _setup_backend_local, "docker": _setup_backend_docker, "singularity": _setup_backend_singularity, "modal": _setup_backend_modal, "daytona": _setup_backend_daytona, "vercel_sandbox": _setup_backend_vercel, - "ssh": _setup_backend_ssh, -} + "ssh": _setup_backend_ssh} +# Backend -> env var mirrored from config after setup (config.yaml is the source of truth, but +# terminal_tool reads these from .env). +_BACKEND_ENV_MIRROR = {"modal": ("TERMINAL_MODAL_MODE", "modal_mode", "auto"), + "vercel_sandbox": ("TERMINAL_VERCEL_RUNTIME", "vercel_runtime", "node24")} def setup_terminal_backend(config: dict): """Configure the terminal execution backend.""" - from hermes_cli.setup import ( - cfg_get, _DOCS_BASE, _info, print_header, print_info, print_success, prompt_choice, save_config, - save_env_value, - ) import platform as _platform - print_header("Terminal Backend") - _info("Choose where Hermes runs shell commands and code.", + _setup.print_header("Terminal Backend") + _setup._info("Choose where Hermes runs shell commands and code.", "This affects tool execution, file access, and isolation.", - f" Guide: {_DOCS_BASE}/user-guide/configuration#terminal-backend-configuration", None) + f" Guide: {_setup._DOCS_BASE}/user-guide/configuration#terminal-backend-configuration", None) - current_backend = cfg_get(config, "terminal", "backend", default="local") + current_backend = _setup.cfg_get(config, "terminal", "backend", default="local") backends = list(_BUILTIN_TERMINAL_BACKENDS) if _platform.system() == "Linux": backends.append(("singularity", "Singularity/Apptainer - HPC-friendly container")) @@ -367,9 +327,9 @@ def setup_terminal_backend(config: dict): keep_current_idx = len(backends) terminal_choices = [label for _, label in backends] + [f"Keep current ({current_backend})"] - terminal_idx = prompt_choice("Select terminal backend:", terminal_choices, keep_current_idx) + terminal_idx = _setup.prompt_choice("Select terminal backend:", terminal_choices, keep_current_idx) if terminal_idx == keep_current_idx: - print_info(f"Keeping current backend: {current_backend}") + _setup.print_info(f"Keeping current backend: {current_backend}") return selected_backend = backends[terminal_idx][0] if 0 <= terminal_idx < len(backends) else None @@ -381,13 +341,13 @@ def setup_terminal_backend(config: dict): elif selected_backend in plugin_backend_names: _setup_backend_plugin(config, selected_backend) - # config.yaml is the source of truth, but terminal_tool reads TERMINAL_ENV from .env — sync. - save_env_value("TERMINAL_ENV", selected_backend) - if selected_backend == "modal": - save_env_value("TERMINAL_MODAL_MODE", config["terminal"].get("modal_mode", "auto")) - if selected_backend == "vercel_sandbox": - save_env_value("TERMINAL_VERCEL_RUNTIME", config["terminal"].get("vercel_runtime", "node24")) - save_config(config) + _setup.save_env_value("TERMINAL_ENV", selected_backend) + if selected_backend in _BACKEND_ENV_MIRROR: + env_var, key, default = _BACKEND_ENV_MIRROR[selected_backend] + _setup.save_env_value(env_var, config["terminal"].get(key, default)) + _setup.save_config(config) print() - print_success(f"Terminal backend set to: {selected_backend}") + _setup.print_success(f"Terminal backend set to: {selected_backend}") + +import hermes_cli.setup as _setup # noqa: E402 (bottom: hermes_cli.setup imports this module) diff --git a/hermes_cli/setup_tts.py b/hermes_cli/setup_tts.py index 010559978f..eda2ba48d7 100644 --- a/hermes_cli/setup_tts.py +++ b/hermes_cli/setup_tts.py @@ -1,6 +1,7 @@ """Text-to-speech provider setup (provider picker, API-key prompts, local engine installs, xAI OAuth). -Extracted from hermes_cli/setup.py; setup.py re-exports the public entry points. +Names from setup.py are imported lazily inside bodies so test patches on +``hermes_cli.setup.`` take effect; setup.py re-exports the public entry points. """ import logging @@ -11,28 +12,21 @@ import sys logger = logging.getLogger("hermes_cli.setup") -def _check_espeak_ng() -> bool: - """Check if espeak-ng is installed.""" - return shutil.which("espeak-ng") is not None or shutil.which("espeak") is not None - - def _pip_install_tts_package(name: str, pip_args: list, manual_cmd: str) -> bool: """Install a local TTS engine through the canonical uv → pip → ensurepip ladder so pip-less venvs (Ubuntu 25.10 ``python -m venv``, ``uv venv``) work.""" - from hermes_cli.setup import print_error, print_info, print_success from hermes_cli.tools_config import _pip_install try: result = _pip_install(pip_args, timeout=300) + if result.returncode == 0: + _setup.print_success(f"{name} installed successfully") + return True + err = (result.stderr or "").strip() + reason = err[:300] if err else "install failed" except Exception as e: - print_error(f"Failed to install {name}: {e}") - print_info(f"Try manually: {manual_cmd}") - return False - if result.returncode == 0: - print_success(f"{name} installed successfully") - return True - err = (result.stderr or "").strip() - print_error(f"Failed to install {name}: {err[:300] if err else 'install failed'}") - print_info(f"Try manually: {manual_cmd}") + reason = e + _setup.print_error(f"Failed to install {name}: {reason}") + _setup.print_info(f"Try manually: {manual_cmd}") return False @@ -46,37 +40,32 @@ _ESPEAK_INSTALL = { def _install_neutts_deps() -> bool: """Install NeuTTS dependencies with user approval. Returns True on success.""" - from hermes_cli.setup import _info, print_info, print_success, print_warning, prompt_yes_no - if not _check_espeak_ng(): + if not (shutil.which("espeak-ng") or shutil.which("espeak")): hint, install_cmd = _ESPEAK_INSTALL.get(sys.platform, _ESPEAK_INSTALL["linux"]) print() - print_warning("NeuTTS requires espeak-ng for phonemization.") - print_info(hint) + _setup.print_warning("NeuTTS requires espeak-ng for phonemization.") + _setup.print_info(hint) print() - if prompt_yes_no("Install espeak-ng now?", True): + if _setup.prompt_yes_no("Install espeak-ng now?", True): try: subprocess.run(install_cmd, check=True) - print_success("espeak-ng installed") + _setup.print_success("espeak-ng installed") except (subprocess.CalledProcessError, FileNotFoundError) as e: - print_warning(f"Could not install espeak-ng automatically: {e}") - print_info("Please install it manually and re-run setup.") + _setup.print_warning(f"Could not install espeak-ng automatically: {e}") + _setup.print_info("Please install it manually and re-run setup.") return False else: - print_warning("espeak-ng is required for NeuTTS. Install it manually before using NeuTTS.") + _setup.print_warning("espeak-ng is required for NeuTTS. Install it manually before using NeuTTS.") - _info(None, "Installing neutts Python package...", + _setup._info(None, "Installing neutts Python package...", "This will also download the TTS model (~300MB) on first use.", None) return _pip_install_tts_package("neutts", ["-U", "neutts[all]", "--quiet"], "uv pip install -U 'neutts[all]'") def _install_kittentts_deps() -> bool: """Install KittenTTS dependencies with user approval. Returns True on success.""" - from hermes_cli.setup import _info - wheel_url = ( - "https://github.com/KittenML/KittenTTS/releases/download/" - "0.8.1/kittentts-0.8.1-py3-none-any.whl" - ) - _info(None, "Installing kittentts Python package (~25-80MB model downloaded on first use)...", None) + wheel_url = "https://github.com/KittenML/KittenTTS/releases/download/0.8.1/kittentts-0.8.1-py3-none-any.whl" + _setup._info(None, "Installing kittentts Python package (~25-80MB model downloaded on first use)...", None) return _pip_install_tts_package( "kittentts", ["-U", wheel_url, "soundfile", "--quiet"], f"uv pip install -U '{wheel_url}' soundfile", ) @@ -96,42 +85,33 @@ def _xai_oauth_logged_in_for_setup() -> bool: def _run_xai_oauth_login_from_setup() -> bool: """Run the xAI Grok OAuth device-code login from inside the setup wizard. - Saves OAuth tokens only — does **not** switch the active provider or rewrite - ``model.provider`` (callers only need credentials for side tools). Returns True on success, - False on any failure (the caller falls back, e.g. to Edge TTS). + Saves OAuth tokens only — does **not** switch the active provider or rewrite ``model.provider`` + (callers only need credentials for side tools). False on any failure (caller falls back). """ - from hermes_cli.setup import _info, print_warning try: from hermes_cli.auth import ( _is_remote_session, _save_xai_oauth_tokens, _xai_oauth_device_code_login, - unsuppress_credential_source, - ) + unsuppress_credential_source) except Exception as exc: - print_warning(f"xAI Grok OAuth helpers unavailable: {exc}") + _setup.print_warning(f"xAI Grok OAuth helpers unavailable: {exc}") return False open_browser = not _is_remote_session() - _info(None, "Signing in to xAI Grok OAuth (SuperGrok / Premium+)...") + _setup._info(None, "Signing in to xAI Grok OAuth (SuperGrok / Premium+)...") try: creds = _xai_oauth_device_code_login(open_browser=open_browser) _save_xai_oauth_tokens( creds["tokens"], discovery=creds.get("discovery"), redirect_uri=creds.get("redirect_uri", ""), - last_refresh=creds.get("last_refresh"), auth_mode="oauth_device_code", set_active=False, - ) + last_refresh=creds.get("last_refresh"), auth_mode="oauth_device_code", set_active=False) # Mirror model/dashboard re-login: clear device_code suppression so the pool can seed # from the singleton after a prior `auth remove`. unsuppress_credential_source("xai-oauth", "device_code") return True except Exception as exc: - print_warning(f"xAI Grok OAuth login failed: {exc}") + _setup.print_warning(f"xAI Grok OAuth login failed: {exc}") return False -_TTS_PROVIDER_LABELS = { - "edge": "Edge TTS", "elevenlabs": "ElevenLabs", "openai": "OpenAI TTS", "xai": "xAI TTS", - "minimax": "MiniMax TTS", "mistral": "Mistral Voxtral TTS", "gemini": "Google Gemini TTS", - "neutts": "NeuTTS", "kittentts": "KittenTTS", -} _TTS_PROVIDER_CHOICES = [ ("edge", "Edge TTS (free, cloud-based, no setup needed)"), ("elevenlabs", "ElevenLabs (premium quality, needs API key)"), @@ -141,75 +121,66 @@ _TTS_PROVIDER_CHOICES = [ ("mistral", "Mistral Voxtral TTS (multilingual, native Opus, needs API key)"), ("gemini", "Google Gemini TTS (30 prebuilt voices, prompt-controllable, needs API key)"), ("neutts", "NeuTTS (local on-device, free, ~300MB model download)"), - ("kittentts", "KittenTTS (local on-device, free, lightweight ~25-80MB ONNX)"), -] + ("kittentts", "KittenTTS (local on-device, free, lightweight ~25-80MB ONNX)")] +# Short label = menu label minus its parenthetical ("Edge TTS", "Mistral Voxtral TTS", ...). +_TTS_PROVIDER_LABELS = {key: label.split(" (")[0] for key, label in _TTS_PROVIDER_CHOICES} # provider -> (env vars that satisfy it, env var to save, prompt, success line, pre-prompt hint) _TTS_API_KEY_PROVIDERS = { - "elevenlabs": (("ELEVENLABS_API_KEY",), "ELEVENLABS_API_KEY", "ElevenLabs API key", "ElevenLabs API key saved", ""), - "openai": ( - ("VOICE_TOOLS_OPENAI_KEY", "OPENAI_API_KEY"), "VOICE_TOOLS_OPENAI_KEY", - "OpenAI API key for TTS", "OpenAI TTS API key saved", "", - ), - "minimax": (("MINIMAX_API_KEY",), "MINIMAX_API_KEY", "MiniMax API key for TTS", "MiniMax TTS API key saved", ""), - "mistral": (("MISTRAL_API_KEY",), "MISTRAL_API_KEY", "Mistral API key for TTS", "Mistral TTS API key saved", ""), - "gemini": ( - ("GEMINI_API_KEY", "GOOGLE_API_KEY"), "GEMINI_API_KEY", "Gemini API key for TTS", "Gemini TTS API key saved", - "Get a free API key at https://aistudio.google.com/app/apikey", - ), + "elevenlabs": (("ELEVENLABS_API_KEY",), "ELEVENLABS_API_KEY", "ElevenLabs API key", + "ElevenLabs API key saved", ""), + "openai": (("VOICE_TOOLS_OPENAI_KEY", "OPENAI_API_KEY"), "VOICE_TOOLS_OPENAI_KEY", + "OpenAI API key for TTS", "OpenAI TTS API key saved", ""), + "minimax": (("MINIMAX_API_KEY",), "MINIMAX_API_KEY", "MiniMax API key for TTS", + "MiniMax TTS API key saved", ""), + "mistral": (("MISTRAL_API_KEY",), "MISTRAL_API_KEY", "Mistral API key for TTS", + "Mistral TTS API key saved", ""), + "gemini": (("GEMINI_API_KEY", "GOOGLE_API_KEY"), "GEMINI_API_KEY", "Gemini API key for TTS", + "Gemini TTS API key saved", "Get a free API key at https://aistudio.google.com/app/apikey"), } # provider -> (module, display name, requirement lines, install question, installer) _TTS_LOCAL_PROVIDERS = { - "neutts": ( - "neutts", "NeuTTS", - ("NeuTTS requires:", " • Python package: neutts (~50MB install + ~300MB model on first use)", - " • System package: espeak-ng (phonemizer)"), - "Install NeuTTS dependencies now?", _install_neutts_deps, - ), - "kittentts": ( - "kittentts", "KittenTTS", - ("KittenTTS is lightweight (~25-80MB, CPU-only, no API key required).", - "Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo"), - "Install KittenTTS now?", _install_kittentts_deps, - ), -} + "neutts": ("neutts", "NeuTTS", + ("NeuTTS requires:", " • Python package: neutts (~50MB install + ~300MB model on first use)", + " • System package: espeak-ng (phonemizer)"), + "Install NeuTTS dependencies now?", _install_neutts_deps), + "kittentts": ("kittentts", "KittenTTS", + ("KittenTTS is lightweight (~25-80MB, CPU-only, no API key required).", + "Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo"), + "Install KittenTTS now?", _install_kittentts_deps)} def _tts_api_key_step(selected: str) -> str: """Ensure the key for an API-key TTS provider exists; fall back to edge otherwise.""" - from hermes_cli.setup import ( - get_env_value, print_info, print_success, print_warning, prompt, save_env_value, - ) env_vars, save_var, prompt_label, saved_msg, hint = _TTS_API_KEY_PROVIDERS[selected] - if any(get_env_value(v) for v in env_vars): + if any(_setup.get_env_value(v) for v in env_vars): return selected print() if hint: - print_info(hint) - api_key = prompt(prompt_label, password=True) + _setup.print_info(hint) + api_key = _setup.prompt(prompt_label, password=True) if api_key: - save_env_value(save_var, api_key) - print_success(saved_msg) + _setup.save_env_value(save_var, api_key) + _setup.print_success(saved_msg) return selected - print_warning("No API key provided. Falling back to Edge TTS.") + _setup.print_warning("No API key provided. Falling back to Edge TTS.") return "edge" def _tts_local_install_step(selected: str) -> str: """Offer to install a local TTS engine; fall back to edge if declined/failed.""" - from hermes_cli.setup import _module_installed, print_info, print_success, print_warning, prompt_yes_no module, name, lines, question, installer = _TTS_LOCAL_PROVIDERS[selected] - if _module_installed(module): - print_success(f"{name} is already installed") + if _setup._module_installed(module): + _setup.print_success(f"{name} is already installed") return selected print() for line in lines: - print_info(line) + _setup.print_info(line) print() - if not prompt_yes_no(question, True): - print_info(f"Skipping install. Set tts.provider to '{selected}' after installing manually.") + if not _setup.prompt_yes_no(question, True): + _setup.print_info(f"Skipping install. Set tts.provider to '{selected}' after installing manually.") return "edge" if not installer(): - print_warning(f"{name} installation incomplete. Falling back to Edge TTS.") + _setup.print_warning(f"{name} installation incomplete. Falling back to Edge TTS.") return "edge" return selected @@ -217,76 +188,62 @@ def _tts_local_install_step(selected: str) -> str: def _tts_xai_step(config: dict) -> str: """xAI TTS auth. Order: existing OAuth tokens (free for SuperGrok) > existing XAI_API_KEY > offer both paths — xAI TTS works with OAuth bearer tokens too.""" - from hermes_cli.setup import ( - get_env_value, print_success, print_warning, prompt, prompt_choice, _run_xai_oauth_login_from_setup, - save_env_value, _xai_oauth_logged_in_for_setup, - ) - selected = "xai" - if _xai_oauth_logged_in_for_setup(): - print_success("xAI TTS will use your xAI Grok OAuth (SuperGrok / Premium+) credentials") - elif get_env_value("XAI_API_KEY"): - print_success("xAI TTS will use your existing XAI_API_KEY") + if _setup._xai_oauth_logged_in_for_setup(): + _setup.print_success("xAI TTS will use your xAI Grok OAuth (SuperGrok / Premium+) credentials") + elif _setup.get_env_value("XAI_API_KEY"): + _setup.print_success("xAI TTS will use your existing XAI_API_KEY") else: print() - choice_idx = prompt_choice( + choice_idx = _setup.prompt_choice( "How do you want xAI TTS to authenticate?", - choices=[ - "Sign in with xAI Grok OAuth (SuperGrok / Premium+) — browser login", - "Paste an xAI API key (console.x.ai)", "Skip → fallback to Edge TTS", - ], - default=0, - ) + choices=["Sign in with xAI Grok OAuth (SuperGrok / Premium+) — browser login", + "Paste an xAI API key (console.x.ai)", "Skip → fallback to Edge TTS"], + default=0) + fallback = None # warning printed when xAI auth did not happen; result is then "edge" if choice_idx == 0: - if _run_xai_oauth_login_from_setup(): - print_success("Logged in — xAI TTS will use these OAuth credentials") + if _setup._run_xai_oauth_login_from_setup(): + _setup.print_success("Logged in — xAI TTS will use these OAuth credentials") else: - print_warning("xAI Grok OAuth login did not complete. Falling back to Edge TTS.") - selected = "edge" + fallback = "xAI Grok OAuth login did not complete. Falling back to Edge TTS." elif choice_idx == 1: - api_key = prompt("xAI API key for TTS", password=True) + api_key = _setup.prompt("xAI API key for TTS", password=True) if api_key: - save_env_value("XAI_API_KEY", api_key) - print_success("xAI TTS API key saved") + _setup.save_env_value("XAI_API_KEY", api_key) + _setup.print_success("xAI TTS API key saved") else: from hermes_constants import display_hermes_home as _dhh - print_warning( - "No xAI API key provided for TTS. Configure XAI_API_KEY " - f"via hermes setup model or {_dhh()}/.env to use xAI TTS. " - "Falling back to Edge TTS." - ) - selected = "edge" + fallback = ("No xAI API key provided for TTS. Configure XAI_API_KEY via hermes setup model " + f"or {_dhh()}/.env to use xAI TTS. Falling back to Edge TTS.") else: - print_warning("xAI TTS skipped. Falling back to Edge TTS.") - selected = "edge" + fallback = "xAI TTS skipped. Falling back to Edge TTS." + if fallback: + _setup.print_warning(fallback) + return "edge" - if selected == "xai": - print() - voice_id = prompt("xAI voice_id (Enter for 'eve', or paste a custom voice ID)") - if voice_id and voice_id.strip(): - config.setdefault("tts", {}).setdefault("xai", {})["voice_id"] = voice_id.strip() - print_success(f"xAI voice_id set to: {voice_id.strip()}") - return selected + print() + voice_id = (_setup.prompt("xAI voice_id (Enter for 'eve', or paste a custom voice ID)") or "").strip() + if voice_id: + config.setdefault("tts", {}).setdefault("xai", {})["voice_id"] = voice_id + _setup.print_success(f"xAI voice_id set to: {voice_id}") + return "xai" def _setup_tts_provider(config: dict): """Interactive TTS provider selection with install flow for local engines.""" - from hermes_cli.setup import ( - get_env_value, get_nous_subscription_features, _info, managed_nous_tools_enabled, print_header, - print_info, print_success, print_warning, prompt_choice, save_config, - ) current_provider = config.get("tts", {}).get("provider", "edge") current_label = _TTS_PROVIDER_LABELS.get(current_provider, current_provider) print() - print_header("Text-to-Speech Provider (optional)") - _info(f"Current: {current_label}", None) + _setup.print_header("Text-to-Speech Provider (optional)") + _setup._info(f"Current: {current_label}", None) options = list(_TTS_PROVIDER_CHOICES) - if managed_nous_tools_enabled() and get_nous_subscription_features(config).nous_auth_present: - options.insert(0, ("nous-openai", "Nous Subscription (managed OpenAI TTS, billed to your subscription)")) + if _setup.managed_nous_tools_enabled() and _setup.get_nous_subscription_features(config).nous_auth_present: + options.insert(0, ("nous-openai", + "Nous Subscription (managed OpenAI TTS, billed to your subscription)")) choices = [label for _, label in options] + [f"Keep current ({current_label})"] keep_current_idx = len(choices) - 1 - idx = prompt_choice("Select TTS provider:", choices, keep_current_idx) + idx = _setup.prompt_choice("Select TTS provider:", choices, keep_current_idx) if idx == keep_current_idx: return @@ -294,11 +251,10 @@ def _setup_tts_provider(config: dict): selected_via_nous = selected == "nous-openai" if selected_via_nous: selected = "openai" - print_info("OpenAI TTS will use the managed Nous gateway and bill to your subscription.") - if get_env_value("VOICE_TOOLS_OPENAI_KEY") or get_env_value("OPENAI_API_KEY"): - print_warning( - "Direct OpenAI credentials are still configured and may take precedence until removed from ~/.hermes/.env." - ) + _setup.print_info("OpenAI TTS will use the managed Nous gateway and bill to your subscription.") + if _setup.get_env_value("VOICE_TOOLS_OPENAI_KEY") or _setup.get_env_value("OPENAI_API_KEY"): + _setup.print_warning("Direct OpenAI credentials are still configured and may take precedence " + "until removed from ~/.hermes/.env.") if selected in _TTS_LOCAL_PROVIDERS: selected = _tts_local_install_step(selected) @@ -308,8 +264,8 @@ def _setup_tts_provider(config: dict): selected = _tts_xai_step(config) config.setdefault("tts", {})["provider"] = selected - save_config(config) - print_success(f"TTS provider set to: {_TTS_PROVIDER_LABELS.get(selected, selected)}") + _setup.save_config(config) + _setup.print_success(f"TTS provider set to: {_TTS_PROVIDER_LABELS.get(selected, selected)}") def setup_tts(config: dict): @@ -317,3 +273,4 @@ def setup_tts(config: dict): _setup_tts_provider(config) +import hermes_cli.setup as _setup # noqa: E402 (bottom: hermes_cli.setup imports this module) diff --git a/hermes_cli/setup_whatsapp_cloud.py b/hermes_cli/setup_whatsapp_cloud.py index af26fc6f3b..814cce7236 100644 --- a/hermes_cli/setup_whatsapp_cloud.py +++ b/hermes_cli/setup_whatsapp_cloud.py @@ -2,12 +2,10 @@ Walks the user through the 6 credentials Meta requires + recipient allowlist, auto-generates the verify token, and prints exact follow-up instructions for the parts that can't happen inside the -wizard process (starting cloudflared, starting the gateway, configuring Meta's webhook dashboard, -adding their phone to the recipient list). +wizard process (cloudflared, gateway, Meta's webhook dashboard, recipient list). -The wizard intentionally does NOT smoke-test the webhook itself — the Hermes gateway and the -cloudflared tunnel both run in separate processes the user starts AFTER this wizard exits, so any -in-wizard probe would fail by design. +The wizard intentionally does NOT smoke-test the webhook: the gateway and the tunnel both run in +separate processes the user starts AFTER this wizard exits, so any in-wizard probe would fail. """ from __future__ import annotations @@ -19,38 +17,26 @@ import sys from typing import Optional -# --------------------------------------------------------------------------- -# Field-shape validators -# --------------------------------------------------------------------------- -# -# Each validator returns (ok, reason_if_not_ok). The wizard uses them to -# reject obviously-malformed input before saving — saves users a round -# trip with Meta's 401 / 400 errors. - +# --- Field-shape validators: each returns (ok, reason_if_not_ok) so obviously-malformed input is +# rejected before saving, sparing a round trip with Meta's 401 / 400 errors. def _validate_phone_number_id(value: str) -> tuple[bool, Optional[str]]: - """Phone Number ID is a 15-17 digit numeric ID assigned by Meta. + """Phone Number ID is a 15-17 digit numeric ID assigned by Meta — NOT a phone number. - It's NOT a phone number. The #1 setup mistake is pasting the actual phone number (e.g. - ``15556422442``) into this field — that's only 10-11 digits and gets rejected by Graph as - "Object with ID does not exist." + The #1 setup mistake is pasting the actual phone number (10-11 digits), which Graph rejects + with "Object with ID does not exist." """ if not value: return False, "Phone Number ID is required" s = value.strip() if not s.isdigit(): return False, "Phone Number ID must be numeric (no '+', spaces, or dashes)" - # Real phone numbers are 10-11 digits (US/CA country code + area code - # + 7 digits). Meta's internal IDs are 15-17 digits. If we see a - # phone-number-sized value, the user almost certainly pasted the - # phone number by mistake. - if 10 <= len(s) <= 12: + if 10 <= len(s) <= 12: # phone-number-sized: almost certainly the number itself return False, ( "That looks like a phone number — but this field needs the " "Phone Number ID (Meta's internal ID, 15-17 digits, e.g. " "'7794189252778687'). Look just BELOW the 'From' dropdown in " - "API Setup → it's labelled 'Phone number ID'." - ) + "API Setup → it's labelled 'Phone number ID'.") if len(s) < 13: return False, "Phone Number ID looks too short (expected 13-18 digits)" if len(s) > 20: @@ -83,8 +69,7 @@ _FOREIGN_TOKEN_PREFIXES = ( (("xoxb-", "xoxp-"), "That's a Slack token, not a Meta WhatsApp access token. " "Meta tokens start with 'EAA'."), (("ghp_", "gho_"), "That's a GitHub token, not a Meta WhatsApp access " - "token. Meta tokens start with 'EAA'."), -) + "token. Meta tokens start with 'EAA'.")) def _validate_app_secret(value: str) -> tuple[bool, Optional[str]]: @@ -96,24 +81,18 @@ def _validate_app_secret(value: str) -> tuple[bool, Optional[str]]: return False, ( "App Secret should be a hex string (only digits 0-9 and " "letters a-f). Make sure you copied the 'App secret' from " - "Settings → Basic, not some other token." - ) + "Settings → Basic, not some other token.") if len(s) != 32: return False, f"App Secret should be exactly 32 hex characters (got {len(s)})" return True, None def _validate_access_token(value: str) -> tuple[bool, Optional[str]]: - """Meta access tokens start with ``EAA`` and are 100-300+ characters. - - Both temp tokens (24h) and System User permanent tokens share this prefix. We don't try to - distinguish them. - """ + """Meta access tokens (temp and System User alike) start with ``EAA``, 100-300+ chars.""" if not value: return False, "Access token is required" s = value.strip() if not s.startswith("EAA"): - # Diagnose common paste mistakes for prefixes, reason in _FOREIGN_TOKEN_PREFIXES: if s.startswith(prefixes): return False, reason @@ -121,52 +100,34 @@ def _validate_access_token(value: str) -> tuple[bool, Optional[str]]: "Meta WhatsApp access tokens start with 'EAA'. Check that " "you're copying from the right place (API Setup → 'Generate " "access token', or Business Settings → System Users → " - "'Generate token' for a permanent one)." - ) + "'Generate token' for a permanent one).") if len(s) < 100: return False, f"Access token looks too short ({len(s)} chars, expected 100+)" return True, None -# --------------------------------------------------------------------------- -# Prompt helpers -# --------------------------------------------------------------------------- - +# --- Prompt helpers def _prompt(message: str, default: Optional[str] = None, secret: bool = False) -> str: - """Read one line of input. Returns "" on EOF / Ctrl+C / empty input. - - ``default`` is shown but NOT auto-applied on empty input: callers handle "kept existing" - explicitly so a real value is distinguishable from a display preview (masked secrets). - ``secret=True`` reads via ``getpass`` so credentials are not echoed or left in scrollback. - """ + """Read one line; "" on EOF / Ctrl+C / empty. ``default`` is shown but NOT auto-applied so a + real value stays distinguishable from a masked preview; ``secret`` reads via ``getpass``.""" try: suffix = f" [{default}]" if default else "" if secret and sys.stdin.isatty(): import getpass - raw = getpass.getpass(f"{message}{suffix} (input hidden): ").strip() - else: - raw = line_input(f"{message}{suffix}: ").strip() + return getpass.getpass(f"{message}{suffix} (input hidden): ").strip() + return line_input(f"{message}{suffix}: ").strip() except (EOFError, KeyboardInterrupt): print() return "" - return raw def _prompt_validated( - message: str, - validator, - *, - current: Optional[str] = None, - help_text: Optional[str] = None, - secret: bool = False, -) -> Optional[str]: - """Repeat the prompt until the user enters a valid value or aborts. - - Returns the validated value, or None if the user gave up (empty response after an error, or - Ctrl+C). ``current`` is shown as a default for re-runs of the wizard with existing config. - """ + message: str, validator, *, current: Optional[str] = None, help_text: Optional[str] = None, + secret: bool = False) -> Optional[str]: + """Repeat the prompt until a valid value or the user gives up (None: empty answer, Ctrl+C). + ``current`` is shown as the default on wizard re-runs.""" if help_text: for line in help_text.strip().splitlines(): print(f" {line}") @@ -190,10 +151,7 @@ def _prompt_validated( attempts = 0 -# --------------------------------------------------------------------------- -# Wizard -# --------------------------------------------------------------------------- - +# --- Wizard def _header(title: str) -> None: print("─" * 50) @@ -201,9 +159,14 @@ def _header(title: str) -> None: print("─" * 50) +def _lines(*lines: str) -> None: + """print() each line; ``""`` yields a blank line.""" + for line in lines: + print(line) + + def _save_optional(key: str, value: Optional[str], current: Optional[str]) -> None: from hermes_cli.config import save_env_value - if value: save_env_value(key, value) print(f" ✓ Saved: {value}") @@ -211,36 +174,103 @@ def _save_optional(key: str, value: Optional[str], current: Optional[str]) -> No print(f" ✓ Keeping existing: {current}") -def run_whatsapp_cloud_setup() -> int: - """Interactive wizard for the WhatsApp Cloud API adapter. +# Credential steps 1-3: (step title, env var, prompt label, validator, secret, preview chars of +# the existing value shown as default (0 = full), "saved" line, "kept" line, lines printed when +# nothing is configured, abort-when-missing). ``{v}`` in the saved/kept lines is the value. +_CREDENTIAL_STEPS = ( + ("STEP 1 — Phone Number ID", "WHATSAPP_CLOUD_PHONE_NUMBER_ID", "Phone Number ID", + _validate_phone_number_id, False, 0, " ✓ Saved: {v}", " ✓ Keeping existing: {v}", + ("\n✗ Phone Number ID is required. Aborting.",), True, + "Found in: App Dashboard → WhatsApp → API Setup, in the\n" + "'Send and receive messages' section.\n" + "Look BELOW the 'From' dropdown — there's a 'Phone number ID'\n" + "line with the value (15-17 digits, e.g. '7794189252778687').\n" + "It is NOT the phone number itself (+1 555-...). That's the\n" + "single most common setup mistake."), + ("STEP 2 — Access Token", "WHATSAPP_CLOUD_ACCESS_TOKEN", "Access Token", + _validate_access_token, True, 15, " ✓ Saved (token hidden)", " ✓ Keeping existing token", + ("\n✗ Access Token is required. Aborting.",), True, + "Two options for getting one:\n\n" + " (a) TEMP — App Dashboard → WhatsApp → API Setup →\n" + " 'Generate access token' button. Lasts 24 hours.\n" + " Fine for testing today; you'll have to regenerate\n" + " tomorrow.\n\n" + " (b) PERMANENT (production) — System User token. One-time\n" + " setup, never expires:\n" + " • business.facebook.com → Settings → System users →\n" + " Add → Admin role\n" + " • Assign Assets → your app (Manage app), your\n" + " WhatsApp account (Manage WABAs)\n" + " • Generate token → expiration: Never → permissions:\n" + " business_management, whatsapp_business_messaging,\n" + " whatsapp_business_management\n\n" + "Tokens start with 'EAA'."), + ("STEP 3 — App Secret (required for webhook signature verification)", "WHATSAPP_CLOUD_APP_SECRET", + "App Secret", _validate_app_secret, True, 8, " ✓ Saved (secret hidden)", + " ✓ Keeping existing App Secret", + ("\n⚠ Skipping App Secret — inbound webhooks will be refused", + " until you set WHATSAPP_CLOUD_APP_SECRET manually."), False, + "Found in: App Dashboard → Settings → Basic →\n" + "'App secret' field (click 'Show', enter your Facebook password).\n\n" + "If 'Show' doesn't appear, you may need Admin role on the app.\n" + "It's a 32-character lowercase hex string.\n\n" + "Without the App Secret, inbound webhook POSTs are refused\n" + "with HTTP 503 (we can't verify they actually came from Meta).")) - Returns 0 on full success, 1 on user abort, 2 on partial completion (some fields written but the - user bailed before finishing). - """ +# Optional step-4 IDs: (prompt label, env var, validator, help text). +_OPTIONAL_ID_STEPS = ( + ("App ID (optional, press Enter to skip)", "WHATSAPP_CLOUD_APP_ID", _validate_app_id, + "Found in: App Dashboard → Settings → Basic → 'App ID' at the\n" + "top of the page. Numeric, ~15-16 digits.\n" + "Not required for messaging — useful only for analytics later."), + ("WABA ID (optional, press Enter to skip)", "WHATSAPP_CLOUD_WABA_ID", _validate_waba_id, + "WhatsApp Business Account ID. Found in: App Dashboard →\n" + "WhatsApp → API Setup, near the top — 'WhatsApp Business\n" + "Account ID'. Numeric, ~15+ digits.\n" + "Not required for messaging — useful for analytics.")) + + +def _credential_step(step) -> tuple[Optional[str], bool]: + """Run one _CREDENTIAL_STEPS entry. Returns (effective value, abort).""" from hermes_cli.config import get_env_value, save_env_value + title, env_var, label, validator, secret, preview, saved, kept, missing, required, help_text = step + _header(title) + current = get_env_value(env_var) or None + shown = (current[:preview] + "...") if (preview and current) else current + value = _prompt_validated(label, validator, current=shown, secret=secret, help_text=help_text) + if value: + save_env_value(env_var, value) + print(saved.format(v=value)) + elif current: + value = current + print(kept.format(v=value)) + else: + _lines(*missing) + if required: + return None, True + print() + return value, False - print() - print("⚕ WhatsApp Business Cloud API Setup") - print("=" * 50) - print() - print("This wizard configures Hermes to talk to WhatsApp via Meta's") - print("official Cloud API. It's the production-grade path:") - print() - print(" • No QR codes, no Node.js bridge subprocess") - print(" • Stable connection — no account-ban risk") - print(" • Business account required (not personal WhatsApp)") - print(" • Public webhook URL required (Cloudflare Tunnel, ngrok,") - print(" or your own reverse proxy with TLS)") - print() - print("If you don't have a Meta app set up yet, follow these steps") - print("FIRST, then come back and re-run this wizard:") - print() - print(" 1. https://developers.facebook.com/apps → Create App") - print(" → 'Connect with customers through WhatsApp'") - print(" 2. App Dashboard → WhatsApp → API Setup") - print(" 3. Click 'Generate access token' (temp 24h token is fine to") - print(" start; switch to a System User permanent token later)") - print() + +def run_whatsapp_cloud_setup() -> int: + """Interactive wizard for the WhatsApp Cloud API adapter. Returns 0 on success, 1 on abort.""" + from hermes_cli.config import get_env_value, save_env_value + _lines( + "", "⚕ WhatsApp Business Cloud API Setup", "=" * 50, "", + "This wizard configures Hermes to talk to WhatsApp via Meta's", + "official Cloud API. It's the production-grade path:", "", + " • No QR codes, no Node.js bridge subprocess", + " • Stable connection — no account-ban risk", + " • Business account required (not personal WhatsApp)", + " • Public webhook URL required (Cloudflare Tunnel, ngrok,", + " or your own reverse proxy with TLS)", "", + "If you don't have a Meta app set up yet, follow these steps", + "FIRST, then come back and re-run this wizard:", "", + " 1. https://developers.facebook.com/apps → Create App", + " → 'Connect with customers through WhatsApp'", + " 2. App Dashboard → WhatsApp → API Setup", + " 3. Click 'Generate access token' (temp 24h token is fine to", + " start; switch to a System User permanent token later)", "") try: input("Press Enter to continue, or Ctrl+C to abort... ") except (EOFError, KeyboardInterrupt): @@ -248,161 +278,47 @@ def run_whatsapp_cloud_setup() -> int: return 1 print() - _header("STEP 1 — Phone Number ID") - current_phone_id = get_env_value("WHATSAPP_CLOUD_PHONE_NUMBER_ID") or None - phone_id = _prompt_validated( - "Phone Number ID", - _validate_phone_number_id, - current=current_phone_id, - help_text=( - "Found in: App Dashboard → WhatsApp → API Setup, in the\n" - "'Send and receive messages' section.\n" - "Look BELOW the 'From' dropdown — there's a 'Phone number ID'\n" - "line with the value (15-17 digits, e.g. '7794189252778687').\n" - "It is NOT the phone number itself (+1 555-...). That's the\n" - "single most common setup mistake." - ), - ) - if not phone_id: - if current_phone_id: - phone_id = current_phone_id - print(f" ✓ Keeping existing: {phone_id}") - else: - print("\n✗ Phone Number ID is required. Aborting.") + for step in _CREDENTIAL_STEPS: + _, abort = _credential_step(step) + if abort: return 1 - else: - save_env_value("WHATSAPP_CLOUD_PHONE_NUMBER_ID", phone_id) - print(f" ✓ Saved: {phone_id}") - print() - - _header("STEP 2 — Access Token") - current_token = get_env_value("WHATSAPP_CLOUD_ACCESS_TOKEN") or None - current_display = (current_token[:15] + "...") if current_token else None - token = _prompt_validated( - "Access Token", - _validate_access_token, - current=current_display, - secret=True, - help_text=( - "Two options for getting one:\n\n" - " (a) TEMP — App Dashboard → WhatsApp → API Setup →\n" - " 'Generate access token' button. Lasts 24 hours.\n" - " Fine for testing today; you'll have to regenerate\n" - " tomorrow.\n\n" - " (b) PERMANENT (production) — System User token. One-time\n" - " setup, never expires:\n" - " • business.facebook.com → Settings → System users →\n" - " Add → Admin role\n" - " • Assign Assets → your app (Manage app), your\n" - " WhatsApp account (Manage WABAs)\n" - " • Generate token → expiration: Never → permissions:\n" - " business_management, whatsapp_business_messaging,\n" - " whatsapp_business_management\n\n" - "Tokens start with 'EAA'." - ), - ) - # If they had a current token and just hit Enter, keep it. - if not token: - if current_token: - token = current_token - print(" ✓ Keeping existing token") - else: - print("\n✗ Access Token is required. Aborting.") - return 1 - else: - save_env_value("WHATSAPP_CLOUD_ACCESS_TOKEN", token) - print(" ✓ Saved (token hidden)") - print() - - _header("STEP 3 — App Secret (required for webhook signature verification)") - current_secret = get_env_value("WHATSAPP_CLOUD_APP_SECRET") or None - current_secret_display = (current_secret[:8] + "...") if current_secret else None - app_secret = _prompt_validated( - "App Secret", - _validate_app_secret, - current=current_secret_display, - secret=True, - help_text=( - "Found in: App Dashboard → Settings → Basic →\n" - "'App secret' field (click 'Show', enter your Facebook password).\n\n" - "If 'Show' doesn't appear, you may need Admin role on the app.\n" - "It's a 32-character lowercase hex string.\n\n" - "Without the App Secret, inbound webhook POSTs are refused\n" - "with HTTP 503 (we can't verify they actually came from Meta)." - ), - ) - if not app_secret: - if current_secret: - app_secret = current_secret - print(" ✓ Keeping existing App Secret") - else: - print("\n⚠ Skipping App Secret — inbound webhooks will be refused") - print(" until you set WHATSAPP_CLOUD_APP_SECRET manually.") - else: - save_env_value("WHATSAPP_CLOUD_APP_SECRET", app_secret) - print(" ✓ Saved (secret hidden)") - print() _header("STEP 4 — App ID & WABA ID (optional, for analytics)") - current_app_id = get_env_value("WHATSAPP_CLOUD_APP_ID") or None - app_id = _prompt_validated( - "App ID (optional, press Enter to skip)", - lambda v: (True, None) if not v else _validate_app_id(v), - current=current_app_id, - help_text=( - "Found in: App Dashboard → Settings → Basic → 'App ID' at the\n" - "top of the page. Numeric, ~15-16 digits.\n" - "Not required for messaging — useful only for analytics later." - ), - ) - _save_optional("WHATSAPP_CLOUD_APP_ID", app_id, current_app_id) - - current_waba_id = get_env_value("WHATSAPP_CLOUD_WABA_ID") or None - waba_id = _prompt_validated( - "WABA ID (optional, press Enter to skip)", - lambda v: (True, None) if not v else _validate_waba_id(v), - current=current_waba_id, - help_text=( - "WhatsApp Business Account ID. Found in: App Dashboard →\n" - "WhatsApp → API Setup, near the top — 'WhatsApp Business\n" - "Account ID'. Numeric, ~15+ digits.\n" - "Not required for messaging — useful for analytics." - ), - ) - _save_optional("WHATSAPP_CLOUD_WABA_ID", waba_id, current_waba_id) + ids = {} + for label, env_var, validator, help_text in _OPTIONAL_ID_STEPS: + current = get_env_value(env_var) or None + value = _prompt_validated( + label, lambda v, _val=validator: (True, None) if not v else _val(v), + current=current, help_text=help_text) + _save_optional(env_var, value, current) + ids[env_var] = value or current print() _header("STEP 5 — Verify Token (auto-generated)") - current_verify = get_env_value("WHATSAPP_CLOUD_VERIFY_TOKEN") or None - if current_verify: - print(f" An existing verify token is already set ({current_verify[:8]}...).") + verify_token = get_env_value("WHATSAPP_CLOUD_VERIFY_TOKEN") or None + regen = "y" + if verify_token: + print(f" An existing verify token is already set ({verify_token[:8]}...).") try: regen = input(" Generate a new one? [y/N]: ").strip().lower() except (EOFError, KeyboardInterrupt): regen = "n" - if regen in {"y", "yes"}: - verify_token = secrets.token_urlsafe(32) - save_env_value("WHATSAPP_CLOUD_VERIFY_TOKEN", verify_token) - print(f" ✓ New verify token: {verify_token}") - else: - verify_token = current_verify - print(" ✓ Keeping existing verify token") - else: + if regen in {"y", "yes"}: + label = "New verify token" if verify_token else "Generated" verify_token = secrets.token_urlsafe(32) save_env_value("WHATSAPP_CLOUD_VERIFY_TOKEN", verify_token) - print(f" ✓ Generated: {verify_token}") - print() - print(" → COPY THIS TOKEN NOW. You'll paste it into Meta's webhook") - print(" configuration dialog (next step).") - print() + print(f" ✓ {label}: {verify_token}") + else: + print(" ✓ Keeping existing verify token") + _lines("", " → COPY THIS TOKEN NOW. You'll paste it into Meta's webhook", + " configuration dialog (next step).", "") _header("STEP 6 — Recipient Allowlist") - print() - print(" Who is allowed to message the bot? (Comma-separated phone") - print(" numbers with country code, no '+' / spaces / dashes. Use '*'") - print(" to allow anyone — only safe if you've also configured Meta's") - print(" recipient whitelist for app-development mode.)") - print() + _lines( + "", " Who is allowed to message the bot? (Comma-separated phone", + " numbers with country code, no '+' / spaces / dashes. Use '*'", + " to allow anyone — only safe if you've also configured Meta's", + " recipient whitelist for app-development mode.)", "") allow_default = get_env_value("WHATSAPP_CLOUD_ALLOWED_USERS") or None try: allowed = line_input( @@ -412,84 +328,67 @@ def run_whatsapp_cloud_setup() -> int: allowed = "" if allowed: # Light normalization — strip spaces and dashes from each entry. - allowed = ",".join( - re.sub(r"[\s\-+]", "", part) for part in allowed.split(",") if part.strip() - ) + allowed = ",".join(re.sub(r"[\s\-+]", "", part) for part in allowed.split(",") if part.strip()) save_env_value("WHATSAPP_CLOUD_ALLOWED_USERS", allowed) print(f" ✓ Saved: {allowed}") else: - print(" ⚠ No allowlist — every inbound message will be denied.") - print(" Re-run this wizard or set WHATSAPP_CLOUD_ALLOWED_USERS manually.") + _lines(" ⚠ No allowlist — every inbound message will be denied.", + " Re-run this wizard or set WHATSAPP_CLOUD_ALLOWED_USERS manually.") print() _header("SETUP COMPLETE — Next steps") - print() - print(" Hermes needs a public HTTPS URL to receive WhatsApp messages.") - print(" The recommended path is Cloudflare Tunnel (free, no port") - print(" forwarding, no DNS setup).") - print() - print(" 1. Install cloudflared (one-time, if you don't have it):") - print(" Windows: winget install Cloudflare.cloudflared") - print(" macOS: brew install cloudflared") - print(" Linux: https://github.com/cloudflare/cloudflared/releases") - print() - print(" Alternatives: ngrok, or your own domain + reverse proxy") - print(" with TLS.") - print() - print(" 2. Start the tunnel in a separate terminal:") - print(" cloudflared tunnel --url http://localhost:8090") - print(" Note the printed https://.trycloudflare.com URL.") - print() - print(" 3. Start the Hermes gateway in another terminal:") - print(" hermes gateway") - print() - print(" 4. Verify your local config is reachable. From a third") - print(" terminal, with the tunnel URL substituted:") - print() - print(" curl 'https://YOUR-TUNNEL.trycloudflare.com/whatsapp/webhook?\\") - print(f" hub.mode=subscribe&hub.verify_token={verify_token}&\\") - print(" hub.challenge=hello'") - print() - print(" Expected: HTTP 200 with body 'hello'.") - print(" Also try: curl https://YOUR-TUNNEL.trycloudflare.com/health") - print(" (should return JSON with verify_token_configured: true).") - print() - print(" 5. Configure Meta to point at your tunnel:") - print(" App Dashboard → WhatsApp → Configuration → Edit webhook") - print(" Callback URL: /whatsapp/webhook") - print(f" Verify Token: {verify_token}") - print(" → Click 'Verify and save'") - print(" → Then 'Manage' webhook fields → subscribe to 'messages'") - print() - print(" 6. Add your phone to Meta's recipient list:") - print(" App Dashboard → WhatsApp → API Setup → 'To' →") - print(" 'Manage phone number list'") - print() - print(" 7. DM the bot's test number from your phone.") - print() + _lines( + "", " Hermes needs a public HTTPS URL to receive WhatsApp messages.", + " The recommended path is Cloudflare Tunnel (free, no port", + " forwarding, no DNS setup).", "", + " 1. Install cloudflared (one-time, if you don't have it):", + " Windows: winget install Cloudflare.cloudflared", + " macOS: brew install cloudflared", + " Linux: https://github.com/cloudflare/cloudflared/releases", "", + " Alternatives: ngrok, or your own domain + reverse proxy", + " with TLS.", "", + " 2. Start the tunnel in a separate terminal:", + " cloudflared tunnel --url http://localhost:8090", + " Note the printed https://.trycloudflare.com URL.", "", + " 3. Start the Hermes gateway in another terminal:", + " hermes gateway", "", + " 4. Verify your local config is reachable. From a third", + " terminal, with the tunnel URL substituted:", "", + " curl 'https://YOUR-TUNNEL.trycloudflare.com/whatsapp/webhook?\\", + f" hub.mode=subscribe&hub.verify_token={verify_token}&\\", + " hub.challenge=hello'", "", + " Expected: HTTP 200 with body 'hello'.", + " Also try: curl https://YOUR-TUNNEL.trycloudflare.com/health", + " (should return JSON with verify_token_configured: true).", "", + " 5. Configure Meta to point at your tunnel:", + " App Dashboard → WhatsApp → Configuration → Edit webhook", + " Callback URL: /whatsapp/webhook", + f" Verify Token: {verify_token}", + " → Click 'Verify and save'", + " → Then 'Manage' webhook fields → subscribe to 'messages'", "", + " 6. Add your phone to Meta's recipient list:", + " App Dashboard → WhatsApp → API Setup → 'To' →", + " 'Manage phone number list'", "", + " 7. DM the bot's test number from your phone.", "") _header("Optional: polish your bot's WhatsApp profile") - print() - print(" WhatsApp shows a display name and profile picture for your bot") - print(" in every chat header and contact list. These are set in Meta's") - print(" Business Manager, not via this wizard — but here's where to do") - print(" it once you're up and running:") - print() - effective_waba = waba_id or current_waba_id - print(" • Display name + profile picture:") - print(" https://business.facebook.com/wa/manage/phone-numbers/" - + (f"?waba_id={effective_waba}" if effective_waba else "")) + effective_waba = ids["WHATSAPP_CLOUD_WABA_ID"] + _lines( + "", " WhatsApp shows a display name and profile picture for your bot", + " in every chat header and contact list. These are set in Meta's", + " Business Manager, not via this wizard — but here's where to do", + " it once you're up and running:", "", + " • Display name + profile picture:", + " https://business.facebook.com/wa/manage/phone-numbers/" + + (f"?waba_id={effective_waba}" if effective_waba else "")) if not effective_waba: print(" (select your WhatsApp Business Account on that page)") - print(" Display-name changes go through a ~24-48h Meta review.") - print() - print(" • About, description, website, hours, business category:") - print(" Same page → click your phone number → 'Edit profile'.") - print() - print(" • Verified badge (the green check):") - print(" Requires Meta's business verification process —") - print(" Business Manager → Security Center → Start Verification.") - print() - print(" Docs: https://hermes-agent.nousresearch.com/docs/user-guide/") - print(" messaging/whatsapp-cloud") - print() + _lines( + " Display-name changes go through a ~24-48h Meta review.", "", + " • About, description, website, hours, business category:", + " Same page → click your phone number → 'Edit profile'.", "", + " • Verified badge (the green check):", + " Requires Meta's business verification process —", + " Business Manager → Security Center → Start Verification.", "", + " Docs: https://hermes-agent.nousresearch.com/docs/user-guide/", + " messaging/whatsapp-cloud", "") return 0 diff --git a/hermes_cli/sizefmt.py b/hermes_cli/sizefmt.py index 0448da095e..29dddcfa85 100644 --- a/hermes_cli/sizefmt.py +++ b/hermes_cli/sizefmt.py @@ -1,22 +1,12 @@ -"""Small shared size-formatting helpers for CLI/agent output. - -Sibling of ``hermes_cli.timefmt`` (same extraction rationale: a tiny purpose-named module -lightweight consumers can import without dragging in the CLI surface). Replaces six near-identical -private byte formatters. - -Two in-repo formatters intentionally do NOT delegate here: -""" +"""Small shared size-formatting helpers for CLI/agent output (sibling of ``hermes_cli.timefmt``: +a tiny purpose-named module lightweight consumers can import without the CLI surface).""" from __future__ import annotations def format_bytes(n) -> str: - """1234567 -> '1.2 MB' (B/KB/MB/GB/TB; integer bytes, one decimal above). - - Accepts anything ``float()`` accepts; returns ``"?"`` for None or unparseable input so display - call sites never raise (contract inherited from doctor's original copy — its stats dict - tolerates None fields). - """ + """1234567 -> '1.2 MB' (B/KB/MB/GB/TB; integer bytes, one decimal above). Returns ``"?"`` for + None or unparseable input so display call sites never raise.""" try: size = float(n) except (TypeError, ValueError): diff --git a/hermes_cli/sqlite_runtime.py b/hermes_cli/sqlite_runtime.py index ba9f0ca23a..d46ef77fb3 100644 --- a/hermes_cli/sqlite_runtime.py +++ b/hermes_cli/sqlite_runtime.py @@ -20,17 +20,14 @@ def _version_tuple(parts: Iterable[object]) -> tuple[int, int, int]: return tuple(values[:3]) -def is_sqlite_wal_reset_vulnerable( - version_info: tuple[int, ...], -) -> bool: +def is_sqlite_wal_reset_vulnerable(version_info: tuple[int, ...]) -> bool: """Return whether *version_info* contains SQLite's WAL-reset bug.""" info = _version_tuple(version_info) return not ( info < (3, 7, 0) or info >= (3, 51, 3) or (3, 50, 7) <= info < (3, 51, 0) - or (3, 44, 6) <= info < (3, 45, 0) - ) + or (3, 44, 6) <= info < (3, 45, 0)) @dataclass(frozen=True) @@ -50,21 +47,15 @@ class SQLiteRuntimeInfo: _PROBE_SCRIPT = """ -import json -import sqlite3 -import sys - +import json, sqlite3, sys conn = sqlite3.connect(":memory:") try: row = conn.execute("SELECT sqlite_source_id()").fetchone() finally: conn.close() - print(json.dumps({ - "base_prefix": sys.base_prefix, - "executable": sys.executable, - "python_version": list(sys.version_info[:3]), - "sqlite_version": list(sqlite3.sqlite_version_info), + "base_prefix": sys.base_prefix, "executable": sys.executable, + "python_version": list(sys.version_info[:3]), "sqlite_version": list(sqlite3.sqlite_version_info), "sqlite_version_string": sqlite3.sqlite_version, "sqlite_source_id": str(row[0]) if row and row[0] is not None else "", })) @@ -75,34 +66,18 @@ def isolated_interpreter_env() -> dict[str, str]: """Copy of ``os.environ`` with conda/uv/venv/PYTHON* overrides stripped, so a child interpreter reports its *own* runtime rather than the caller's.""" env = dict(os.environ) - for key in ( - "CONDA_DEFAULT_ENV", - "CONDA_PREFIX", - "PYTHONHOME", - "PYTHONPATH", - "UV_PROJECT_ENVIRONMENT", - "UV_PYTHON", - "VIRTUAL_ENV", - ): + for key in ("CONDA_DEFAULT_ENV", "CONDA_PREFIX", "PYTHONHOME", "PYTHONPATH", "UV_PROJECT_ENVIRONMENT", + "UV_PYTHON", "VIRTUAL_ENV"): env.pop(key, None) return env -def probe_sqlite_runtime( - python: str | Path, - *, - timeout: float = 30.0, -) -> SQLiteRuntimeInfo | None: +def probe_sqlite_runtime(python: str | Path, *, timeout: float = 30.0) -> SQLiteRuntimeInfo | None: """Probe SQLite in *python*, never the caller's linked SQLite.""" try: result = subprocess.run( - [str(python), "-I", "-c", _PROBE_SCRIPT], - capture_output=True, - text=True, - timeout=timeout, - check=False, - env=isolated_interpreter_env(), - ) + [str(python), "-I", "-c", _PROBE_SCRIPT], capture_output=True, text=True, timeout=timeout, + check=False, env=isolated_interpreter_env()) except (OSError, subprocess.TimeoutExpired): return None if result.returncode != 0: @@ -110,12 +85,10 @@ def probe_sqlite_runtime( try: payload = json.loads(result.stdout) return SQLiteRuntimeInfo( - executable=Path(str(payload["executable"])), - base_prefix=Path(str(payload["base_prefix"])), + executable=Path(str(payload["executable"])), base_prefix=Path(str(payload["base_prefix"])), python_version=_version_tuple(payload["python_version"]), sqlite_version=_version_tuple(payload["sqlite_version"]), sqlite_version_string=str(payload["sqlite_version_string"]), - sqlite_source_id=str(payload.get("sqlite_source_id", "")), - ) + sqlite_source_id=str(payload.get("sqlite_source_id", ""))) except (KeyError, TypeError, ValueError, json.JSONDecodeError): return None diff --git a/hermes_cli/sqlite_safe_read.py b/hermes_cli/sqlite_safe_read.py index e759f0437c..a8a6d12aba 100644 --- a/hermes_cli/sqlite_safe_read.py +++ b/hermes_cli/sqlite_safe_read.py @@ -1,11 +1,9 @@ """Lock-safe inspection of SQLite database files. -Why this module exists ---------------------- POSIX advisory locks are cancelled **process-wide** by -``close()`` on *any* file descriptor for that file:: - -So a bare ``open(db_path, "rb") ... close()`` on a **live** database silently drops every lock -SQLite holds on it from this process -- including the EXCLUSIVE lock a ``VACUUM`` is holding while -it rewrites the whole file, and the RESERVED lock an in-flight ``BEGIN IMMEDIATE`` is holding. +POSIX advisory locks are cancelled **process-wide** by ``close()`` on *any* fd for that file, so a +bare ``open(db_path, "rb") ... close()`` on a live database drops every lock SQLite holds from this +process (a VACUUM's EXCLUSIVE lock, an in-flight BEGIN IMMEDIATE's RESERVED lock). This module +tracks live connections so raw reads happen only when none exist. """ from __future__ import annotations @@ -28,11 +26,12 @@ _live_connections: dict[str, int] = {} class UntrackableConnectionError(RuntimeError): - """A connection to a probe-able database could not be tracked. + """A connection to a probe-able database could not be tracked. Raised rather than returning an + untracked connection: on these paths tracking is part of the correctness contract.""" - Raised rather than silently returning an untracked connection: on these paths tracking is part - of the correctness contract, not an optimisation. - """ + +class LiveConnectionError(RuntimeError): + """A raw file operation was attempted on a database with live connections.""" def _key(path: Path | str) -> str: @@ -44,11 +43,8 @@ def _key(path: Path | str) -> str: def _canonical_db_path(conn: sqlite3.Connection) -> Optional[str]: - """The on-disk path of ``main``, as SQLite itself reports it. - - Immune to the caller's spelling (``file:`` URIs, relative paths, symlinks). Returns ``None`` for - in-memory or unnamed databases, which cannot be byte-probed and therefore need no tracking. - """ + """The on-disk path of ``main`` as SQLite reports it (immune to ``file:`` URIs, relative paths, + symlinks). ``None`` for in-memory/unnamed databases, which cannot be byte-probed.""" try: row = conn.execute("PRAGMA database_list").fetchone() except sqlite3.Error: @@ -59,29 +55,25 @@ def _canonical_db_path(conn: sqlite3.Connection) -> Optional[str]: def track_connection(path: Path | str) -> None: - """Record that this process now holds a connection to *path*. - - Prefer :func:`connect_tracked`; this exists for callers that manage their own connection - objects, and for tests. - """ + """Record that this process holds a connection to *path* (prefer :func:`connect_tracked`; this + is for callers managing their own connection objects, and for tests).""" with _live_lock: _track_key(_key(path)) -def _track_key(key: str) -> None: - """Bump the live count for an already-canonical key (caller holds ``_live_lock``).""" - _live_connections[key] = _live_connections.get(key, 0) + 1 +def _track_key(key: str, delta: int = 1) -> None: + """Adjust the live count for an already-canonical key (caller holds ``_live_lock``).""" + remaining = _live_connections.get(key, 0) + delta + if remaining > 0: + _live_connections[key] = remaining + else: + _live_connections.pop(key, None) def untrack_connection(path: Path | str) -> None: """Record that one connection to *path* has been closed.""" - key = _key(path) with _live_lock: - remaining = _live_connections.get(key, 0) - 1 - if remaining > 0: - _live_connections[key] = remaining - else: - _live_connections.pop(key, None) + _track_key(_key(path), -1) def has_live_connection(path: Path | str) -> bool: @@ -91,17 +83,19 @@ def has_live_connection(path: Path | str) -> bool: class _TrackingMixin: - """Untrack-on-close behaviour, mixable into any Connection subclass.""" + """Untrack-on-close behaviour, mixable into any Connection subclass. + + The real ``close()`` and the unregister happen together under ``_live_lock`` so a concurrent + probe can never observe "no live connection" while this descriptor is still open. Unregister + runs only after ``close()`` succeeds: untracking before a failing close (e.g. cross-thread + ProgrammingError) would leave the FD open while the byte-probe guard thinks nothing is live. + """ _hermes_tracked_path: str | None = None def close(self) -> None: # type: ignore[misc] with _live_lock: path = getattr(self, "_hermes_tracked_path", None) - # Close first; untrack only once the descriptor is actually gone. - # Untracking before a failing close (e.g. cross-thread - # ProgrammingError) leaves the FD open while the byte-probe - # guard thinks nothing is live — see #75629. super().close() # type: ignore[misc] if path is not None: self._hermes_tracked_path = None @@ -109,28 +103,17 @@ class _TrackingMixin: class TrackedConnection(_TrackingMixin, sqlite3.Connection): - """A ``sqlite3.Connection`` that untracks its path exactly once on close. - - Counting opens is easy; counting closes reliably is not, because callers close connections in - many places (and some hand them to ``contextlib.closing``). - - The real ``close()`` and the unregister happen together under ``_live_lock`` so a concurrent - probe can never observe "no live connection" while this descriptor is still open. Unregister - runs only after ``close()`` succeeds; a raising close leaves the connection tracked so the byte- - probe guard keeps refusing. - """ + """A ``sqlite3.Connection`` that untracks its path exactly once on close (callers close in many + places, some via ``contextlib.closing``, so counting closes by hand is unreliable).""" _tracked_factory_cache: dict[type, type] = {} def _tracking_factory(factory: type) -> type: - """Return *factory* augmented with untrack-on-close. - - Callers legitimately pass their own ``Connection`` subclasses (tests simulate FTS5-less or - pragma-failing runtimes); refusing them or leaving them untracked would quietly unguard the - database, so the tracking ``close()`` is mixed into the caller's class instead. - """ + """Return *factory* augmented with untrack-on-close. Callers legitimately pass their own + ``Connection`` subclasses (tests simulate FTS5-less or pragma-failing runtimes); leaving them + untracked would quietly unguard the database, so the tracking ``close()`` is mixed in.""" if factory is sqlite3.Connection: return TrackedConnection if issubclass(factory, _TrackingMixin): @@ -143,48 +126,33 @@ def _tracking_factory(factory: type) -> type: def connect_tracked( - path: Path | str, - *, - tracking_path: Path | str | None = None, - connect_fn=None, - **kwargs, + path: Path | str, *, tracking_path: Path | str | None = None, connect_fn=None, **kwargs, ) -> sqlite3.Connection: - """``sqlite3.connect`` that registers the connection for the lifetime of the fd. - - Use for any connection to a database whose file might otherwise be byte-probed (``state.db``, - ``kanban.db``). The registration is released automatically on ``close()``. - - The open and the registration happen together under ``_live_lock``, so a concurrent - :func:`read_header_bytes_preopen` cannot slip between them and cancel this connection's locks. - """ + """``sqlite3.connect`` that registers the connection for the lifetime of the fd (released on + ``close()``). Use for any database that might be byte-probed (``state.db``, ``kanban.db``). + Open and registration happen together under ``_live_lock`` so a concurrent + :func:`read_header_bytes_preopen` cannot slip between them and cancel this connection's locks.""" opener = connect_fn if connect_fn is not None else sqlite3.connect kwargs["factory"] = _tracking_factory(kwargs.get("factory", sqlite3.Connection)) with _live_lock: conn = opener(str(path), **kwargs) try: - resolved = ( - _key(tracking_path) - if tracking_path is not None - else _canonical_db_path(conn) - ) + resolved = _key(tracking_path) if tracking_path is not None else _canonical_db_path(conn) if resolved is None: - # In-memory / unnamed: nothing on disk to byte-probe. - return conn + return conn # in-memory / unnamed: nothing on disk to byte-probe if not isinstance(conn, _TrackingMixin): - # The opener substituted its own factory and discarded ours - # (test doubles simulating FTS5-less runtimes do this). Retag - # the instance's class with the tracking mixin so close() still - # releases the registry entry, rather than handing back a - # connection whose database has silently lost probe safety. + # The opener substituted its own factory and discarded ours (test doubles + # simulating FTS5-less runtimes do this). Retag the instance's class so close() + # still releases the registry entry rather than silently losing probe safety. conn = _retrofit_tracking(conn, resolved) conn._hermes_tracked_path = resolved _track_key(resolved) return conn except Exception: try: - # Close via sqlite3 directly: the tracking entry was either - # never made or is being unwound here. + # Close via sqlite3 directly: the tracking entry was either never made or is + # being unwound here. sqlite3.Connection.close(conn) except Exception: pass @@ -192,11 +160,8 @@ def connect_tracked( def _retrofit_tracking(conn: sqlite3.Connection, resolved: str) -> sqlite3.Connection: - """Give an already-open connection untrack-on-close semantics. - - ``sqlite3.Connection`` subclasses are ordinary classes, so ``__class__`` can be swapped for - one mixing in the tracking ``close()``. Used when an opener ignored the factory we asked for. - """ + """Give an already-open connection untrack-on-close semantics by swapping ``__class__`` for + one mixing in the tracking ``close()`` (used when an opener ignored the factory we asked for).""" cls = type(conn) try: conn.__class__ = _tracking_factory(cls) # type: ignore[assignment] @@ -210,74 +175,44 @@ def _retrofit_tracking(conn: sqlite3.Connection, resolved: str) -> sqlite3.Conne def page_count_bytes(conn: sqlite3.Connection) -> Optional[int]: - """Logical database size in bytes, read through *conn*. - - ``page_count * page_size`` is the same quantity the 4-byte header field at offset 28 carries, - but reading it via ``PRAGMA`` opens no new file descriptor and therefore cannot cancel this - process's POSIX locks. - - Returns ``None`` when the pragmas cannot be read. - """ + """Logical database size in bytes (``page_count * page_size``, the header field at offset 28) + read via PRAGMA over *conn* so no new fd is opened. ``None`` when the pragmas cannot be read.""" try: page_count = conn.execute("PRAGMA page_count").fetchone()[0] page_size = conn.execute("PRAGMA page_size").fetchone()[0] - except (sqlite3.Error, TypeError, IndexError) as exc: - logger.debug("page_count/page_size unavailable: %s", exc) - return None - try: return int(page_count) * int(page_size) - except (TypeError, ValueError): + except (sqlite3.Error, TypeError, IndexError, ValueError) as exc: + logger.debug("page_count/page_size unavailable: %s", exc) return None def file_length_matches_header(conn: sqlite3.Connection) -> Optional[bool]: - """Whether the file on disk is at least as long as the header claims. - - Detects the "torn extend" shape (file shorter than its own page count) without ever opening the - database file: the header side comes from ``PRAGMA page_count`` over *conn*, and the on-disk - side from ``stat()``, which takes no descriptor and cannot break locks. - - Note: in WAL mode a freshly committed page may still live in the ``-wal`` file, so the main file - legitimately lags. Callers must treat this as advisory unless the database is in a rollback - journal mode. - """ + """Whether the file on disk is at least as long as the header claims ("torn extend" check), + without opening the file (PRAGMA over *conn* + ``stat()``). Advisory in WAL mode: a freshly + committed page may still live in ``-wal`` so the main file legitimately lags.""" path_str = _canonical_db_path(conn) if path_str is None: return None - logical = page_count_bytes(conn) if not logical: return None try: - actual = os.path.getsize(path_str) + return os.path.getsize(path_str) >= logical except OSError: return None - return actual >= logical -def read_header_bytes_preopen( - path: Path | str, - *, - length: int = 100, - force: bool = False, -) -> Optional[bytes]: - """Read the first *length* bytes of *path* -- only when no connection is live. - - This is the ONLY sanctioned byte-level read of a database file, and it is restricted to first- - open validation (is this file a real SQLite database, is it zeroed, has it been overwritten by - something else). - - The registry check and the ``open``/``read``/``close`` are performed together under - ``_live_lock``, so a connection cannot be opened in the window between deciding "nothing is - live" and closing this descriptor. - """ +def read_header_bytes_preopen(path: Path | str, *, length: int = 100, force: bool = False) -> Optional[bytes]: + """Read the first *length* bytes of *path* -- only when no connection is live. The ONLY + sanctioned byte-level read of a database file, for first-open validation (real SQLite? zeroed? + overwritten?). Check and open/read/close run together under ``_live_lock`` so a connection + cannot be opened between deciding "nothing is live" and closing this descriptor.""" with _live_lock: if not force and _key(path) in _live_connections: logger.debug( "refusing byte-level read of %s: a live connection exists in " "this process and close() would cancel its POSIX locks", - path, - ) + path) return None try: with open(path, "rb") as handle: @@ -286,27 +221,16 @@ def read_header_bytes_preopen( return None -class LiveConnectionError(RuntimeError): - """A raw file operation was attempted on a database with live connections.""" - - @contextlib.contextmanager def offline_file_access(path: Path | str, *, what: str = "read"): - """Hold the connection-lifecycle lock across a raw read of a database file. - - Checking :func:`has_live_connection` and *then* doing the raw I/O is a check/use race: a - connection can be opened in the window between the two, and the raw ``close()`` will cancel its - POSIX advisory locks — the exact failure class the registry exists to prevent. - - The lock is only held for the duration of the raw I/O; it never spans caller work on an open - connection, so it does not serialise database use. - """ + """Hold the connection-lifecycle lock across a raw read of a database file: checking + :func:`has_live_connection` and *then* doing raw I/O is a check/use race (a connection opened + in between loses its POSIX locks to the raw ``close()``). Held only for the raw I/O.""" with _live_lock: if _key(path) in _live_connections: raise LiveConnectionError( f"Refusing to {what} {path}: a connection to it is still open " "in this process, and raw file access would cancel that " "connection's POSIX advisory locks. Close all database " - "handles (stop the gateway/dashboard) and retry." - ) + "handles (stop the gateway/dashboard) and retry.") yield diff --git a/hermes_cli/sqlite_util.py b/hermes_cli/sqlite_util.py index 920fad1ff0..bd5e7779b4 100644 --- a/hermes_cli/sqlite_util.py +++ b/hermes_cli/sqlite_util.py @@ -7,12 +7,8 @@ import sqlite3 def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl: str) -> bool: - """``ALTER TABLE ADD COLUMN ``, idempotent across races. - - Returns True when this call added the column; swallows the ``duplicate column name`` error a - concurrent migrator may have caused. ``column`` is the human-readable name, ``ddl`` the - actual definition. - """ + """``ALTER TABLE
ADD COLUMN ``, idempotent across races: True when this call added + it, False on the ``duplicate column name`` a concurrent migrator caused.""" try: conn.execute(f"ALTER TABLE {table} ADD COLUMN {ddl}") return True @@ -24,12 +20,8 @@ def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl @contextlib.contextmanager def write_txn(conn: sqlite3.Connection): - """An IMMEDIATE write transaction: at most one concurrent writer wins. - - The explicit ROLLBACK is guarded so a SQLite auto-rollback (no active transaction left under EIO - / lock contention / corruption) cannot shadow the original exception with a spurious rollback - error. - """ + """An IMMEDIATE write transaction. The explicit ROLLBACK is guarded so a SQLite auto-rollback + (no transaction left under EIO / contention / corruption) cannot shadow the original error.""" conn.execute("BEGIN IMMEDIATE") try: yield conn diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 26e7d9ab00..5e06920225 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -17,11 +17,7 @@ from hermes_cli.models import provider_label 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, -) + _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 @@ -46,6 +42,16 @@ def _detail(label: str, value) -> None: print(f" {label:<12}{value}") +def _kv(label: str, value) -> None: + """Print a `` Label: value`` line (label padded to the 14-col status layout).""" + print(f" {label:<14}{value}") + + +def _kv_flag(label: str, ok, on: str, off: str) -> None: + """``_kv`` with a ✓/✗ mark followed by ``on`` or ``off`` text.""" + _kv(label, f"{check_mark(bool(ok))} {on if ok else off}") + + def _first_env_value(names) -> str: """Return the first non-empty env value among ``names`` (a str or tuple of names).""" return next((v for v in (get_env_value(n) or "" for n in ((names,) if isinstance(names, str) else names)) if v), "") @@ -68,9 +74,8 @@ def _effective_provider_label() -> str: effective = requested or "auto" if effective == "openrouter": - # 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. + # 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 "" @@ -96,7 +101,6 @@ def _estop_status_line(): # --- Data tables driving the per-section renderers ------------------------- - # Simple env-driven terminal backends: (label, env var, default, empty-counts-as-unset). _TERMINAL_ENV_ROWS = { "ssh": (("SSH Host:", "TERMINAL_SSH_HOST", "(not set)", True), ("SSH User:", "TERMINAL_SSH_USER", "(not set)", True)), @@ -119,16 +123,15 @@ _PLATFORMS = { # name -> (token env var, home-channel env var or None) "Weixin": ("WEIXIN_ACCOUNT_ID", "WEIXIN_HOME_CHANNEL"), "BlueBubbles": ("BLUEBUBBLES_SERVER_URL", "BLUEBUBBLES_HOME_CHANNEL"), "QQBot": ("QQ_APP_ID", "QQ_HOME_CHANNEL"), - "Yuanbao": ("YUANBAO_APP_ID", "YUANBAO_HOME_CHANNEL"), -} + "Yuanbao": ("YUANBAO_APP_ID", "YUANBAO_HOME_CHANNEL")} # Gateway manager label when the runtime snapshot is unavailable, keyed by platform. _GATEWAY_FALLBACK = {"linux": ("unknown", "systemd/manual"), "darwin": ("unknown", "launchd")} class _StatusContext: - """State shared across section renderers: config, --deep, and the Nous login facts - the Auth Providers section derives that the Nous Tool Gateway section needs later.""" + """Shared by section renderers: config, --deep, and the Nous login facts Auth Providers + derives for the later Nous Tool Gateway section.""" def __init__(self, deep: bool): self.deep, self.config = deep, {} @@ -138,9 +141,10 @@ class _StatusContext: def _render_header(ctx): print() - print(color("┌─────────────────────────────────────────────────────────┐", Colors.CYAN)) - print(color("│ ⚕ Hermes Agent Status │", Colors.CYAN)) - print(color("└─────────────────────────────────────────────────────────┘", Colors.CYAN)) + for line in ("┌─────────────────────────────────────────────────────────┐", + "│ ⚕ Hermes Agent Status │", + "└─────────────────────────────────────────────────────────┘"): + print(color(line, Colors.CYAN)) paused = _estop_status_line() if paused: print() @@ -149,23 +153,22 @@ def _render_header(ctx): def _render_environment(ctx): _section("Environment") - print(f" Project: {PROJECT_ROOT}") - print(f" Python: {sys.version.split()[0]}") - env_exists = get_env_path().exists() - print(f" .env file: {check_mark(env_exists)} {'exists' if env_exists else 'not found'}") + _kv("Project:", PROJECT_ROOT) + _kv("Python:", sys.version.split()[0]) + _kv_flag(".env file:", get_env_path().exists(), "exists", "not found") try: ctx.config = load_config() except Exception: ctx.config = {} - print(f" Model: {_configured_model_label(ctx.config)}") - print(f" Provider: {_effective_provider_label()}") + _kv("Model:", _configured_model_label(ctx.config)) + _kv("Provider:", _effective_provider_label()) def _render_terminal(ctx): _section("Terminal Backend") terminal_cfg = ctx.config.get("terminal", {}) if isinstance(ctx.config.get("terminal"), dict) else {} terminal_env = os.getenv("TERMINAL_ENV", "") or terminal_cfg.get("backend", "local") - print(f" Backend: {terminal_env}") + _kv("Backend:", terminal_env) if terminal_env in _TERMINAL_ENV_ROWS: for label, var, default, empty_is_unset in _TERMINAL_ENV_ROWS[terminal_env]: @@ -176,18 +179,17 @@ def _render_terminal(ctx): 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: {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}") + _kv("Runtime:", os.getenv('TERMINAL_VERCEL_RUNTIME') or terminal_cfg.get('vercel_runtime') or 'node24') + _kv_flag("SDK:", importlib.util.find_spec("vercel") is not None, "installed", + "missing (install: pip install 'hermes-agent[vercel]')") + _kv("Auth:", f"{check_mark(auth_status.ok)} {auth_status.label}") for line in auth_status.detail_lines: - print(f" Auth detail: {line}") - print(f" Persistence: {'snapshot filesystem' if persist_enabled else 'ephemeral filesystem'}") - print(" Processes: live processes do not survive cleanup, snapshots, or sandbox recreation") + _kv("Auth detail:", line) + _kv("Persistence:", 'snapshot filesystem' if persist_enabled else 'ephemeral filesystem') + _kv("Processes:", "live processes do not survive cleanup, snapshots, or sandbox recreation") else: - # Plugin-registered terminal backends: show availability via the - # provider's doctor rows (fail-soft — never break `hermes status`). + # Plugin-registered terminal backends: show availability via the provider's doctor rows + # (fail-soft — never break `hermes status`). try: from hermes_cli.plugins import discover_plugins discover_plugins() @@ -199,8 +201,7 @@ def _render_terminal(ctx): except Exception: pass - sudo_password = os.getenv("SUDO_PASSWORD", "") - print(f" Sudo: {check_mark(bool(sudo_password))} {'enabled' if sudo_password else 'disabled'}") + _kv_flag("Sudo:", os.getenv("SUDO_PASSWORD", ""), "enabled", "disabled") def _render_platforms(ctx): @@ -213,8 +214,8 @@ def _render_platforms(ctx): 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 check_fn 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: @@ -230,62 +231,64 @@ def _render_gateway(ctx): from hermes_cli.gateway import get_gateway_runtime_snapshot, _format_gateway_pids snapshot = get_gateway_runtime_snapshot() - print(f" Status: {check_mark(snapshot.running)} {'running' if snapshot.running else 'stopped'}") - print(f" Manager: {snapshot.manager}") + _kv_flag("Status:", snapshot.running, "running", "stopped") + _kv("Manager:", snapshot.manager) if snapshot.gateway_pids: - print(f" PID(s): {_format_gateway_pids(snapshot.gateway_pids)}") + _kv("PID(s):", _format_gateway_pids(snapshot.gateway_pids)) if snapshot.has_process_service_mismatch: - print(" Service: installed but not managing the current running gateway") + _kv("Service:", "installed but not managing the current running gateway") elif _is_termux() and not snapshot.gateway_pids: - print(" Start with: hermes gateway") - print(" Note: Android may stop background jobs when Termux is suspended") + _kv("Start with:", "hermes gateway") + _kv("Note:", "Android may stop background jobs when Termux is suspended") elif snapshot.service_installed and not snapshot.service_running: - print(" Service: installed but stopped") + _kv("Service:", "installed but stopped") except Exception: if _is_termux(): status_text, manager = "unknown", "Termux / manual process" else: platform = "linux" if sys.platform.startswith("linux") else sys.platform status_text, manager = _GATEWAY_FALLBACK.get(platform, ("N/A", "(not supported on this platform)")) - print(f" Status: {color(status_text, Colors.DIM)}") - print(f" Manager: {manager}") + _kv("Status:", color(status_text, Colors.DIM)) + _kv("Manager:", manager) + + +def _load_json(path: Path, encoding: str = "utf-8"): + with open(path, encoding=encoding) as f: + return json.load(f) def _render_cron(ctx): _section("Scheduled Jobs") jobs_file = get_hermes_home() / "cron" / "jobs.json" if not jobs_file.exists(): - print(" Jobs: 0") + _kv("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") + # 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. + jobs = _load_json(jobs_file, "utf-8-sig").get("jobs", []) + _kv("Jobs:", f"{sum(1 for j in jobs if j.get('enabled', True))} active, {len(jobs)} total") except Exception: - print(" Jobs: (error reading jobs file)") + _kv("Jobs:", "(error reading jobs file)") def _render_sessions(ctx): _section("Sessions") - # Gateway session count: state.db is the source of truth; fall back to - # sessions.json for pre-migration installs. + # 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() try: - lister = getattr(db, "list_gateway_sessions", None) - if callable(lister): - gateway_rows = lister(active_only=True) or [] + gateway_rows = db.list_gateway_sessions(active_only=True) or [] finally: db.close() except Exception: gateway_rows = [] if gateway_rows: - print(f" Active: {len(gateway_rows)} session(s)") + _kv("Active:", f"{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 @@ -294,23 +297,21 @@ def _render_sessions(ctx): sessions_file = get_hermes_home() / "sessions" / "sessions.json" if sessions_file.exists(): try: - with open(sessions_file, encoding="utf-8") as f: - data = json.load(f) + data = _load_json(sessions_file) entries = [k for k in data if not str(k).startswith("_")] if isinstance(data, dict) else [] - print(f" Active: {len(entries)} session(s)") + _kv("Active:", f"{len(entries)} session(s)") except Exception: - print(" Active: (error reading sessions file)") + _kv("Active:", "(error reading sessions file)") else: - print(" Active: 0") + _kv("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 - # gets rejected is rarely the one holding the slots — without this the only - # way to find out is reading runtime/active_sessions.json by hand. + # 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 gets rejected is rarely the one + # holding the slots — without this the only way to find out is reading + # runtime/active_sessions.json by hand. try: from hermes_cli.active_sessions import ( - active_session_registry_snapshot, format_age, resolve_max_concurrent_sessions, - ) + active_session_registry_snapshot, format_age, resolve_max_concurrent_sessions) cap = resolve_max_concurrent_sessions(ctx.config) except Exception: cap = None @@ -319,7 +320,7 @@ def _render_sessions(ctx): held = active_session_registry_snapshot() except Exception: held = [] - print(" Slots: " + color(f"{len(held)}/{cap} in use", Colors.YELLOW if len(held) >= cap else Colors.GREEN)) + _kv("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)) @@ -335,17 +336,16 @@ def _render_deep(ctx): try: import httpx response = httpx.get(OPENROUTER_MODELS_URL, headers={"Authorization": f"Bearer {openrouter_key}"}, timeout=10) - ok = response.status_code == 200 - print(f" OpenRouter: {check_mark(ok)} {'reachable' if ok else f'error ({response.status_code})'}") + _kv_flag("OpenRouter:", response.status_code == 200, "reachable", f"error ({response.status_code})") except Exception as e: - print(f" OpenRouter: {check_mark(False)} error: {e}") + _kv("OpenRouter:", f"{check_mark(False)} error: {e}") 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 sock.close() - print(f" Port 18789: {'in use' if port_in_use else 'available'}") + _kv("Port 18789:", 'in use' if port_in_use else 'available') except OSError: pass @@ -361,8 +361,7 @@ def _render_footer(ctx): _SECTIONS = ( _render_header, _render_environment, _render_api_keys, _render_auth_providers, _render_nous_gateway, _render_apikey_providers, _render_terminal, _render_platforms, _render_gateway, _render_cron, - _render_sessions, _render_deep, _render_footer, -) + _render_sessions, _render_deep, _render_footer) def show_status(args): diff --git a/hermes_cli/status_auth.py b/hermes_cli/status_auth.py index effba64a61..de23c0fcad 100644 --- a/hermes_cli/status_auth.py +++ b/hermes_cli/status_auth.py @@ -1,16 +1,11 @@ -"""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. -""" +"""Credential sections of `hermes status` (API keys, OAuth providers, Nous Tool Gateway, API-key +providers), 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, -) + 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 @@ -25,7 +20,9 @@ def _format_iso_timestamp(value) -> str: 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") + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") def _qwen_expiry(expires_at_ms) -> str: @@ -39,13 +36,12 @@ def _oauth_block(name: str, status: dict, hint: str, rows) -> None: ``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})") + _status._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) + _status._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). @@ -69,17 +65,12 @@ _API_KEYS: dict[str, str | tuple[str, ...]] = { "Browserbase": "BROWSERBASE_API_KEY", # Optional — direct credentials only "FAL": "FAL_KEY", "ElevenLabs": "ELEVENLABS_API_KEY", - "GitHub": "GITHUB_TOKEN", -} - + "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), -) - + ("Refreshed:", "last_refresh", _format_iso_timestamp, None), ("Error:", "error", None, False)) _OAUTH_BLOCKS = ( # (row name, auth getter, login hint, detail rows) @@ -87,16 +78,12 @@ _OAUTH_BLOCKS = ( ("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), - )), + ("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), -) - + ("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"), @@ -104,35 +91,31 @@ _APIKEY_PROVIDERS = { "StepFun Step Plan": ("STEPFUN_API_KEY",), "MiniMax": ("MINIMAX_API_KEY",), "MiniMax (China)": ("MINIMAX_CN_API_KEY",), - "DeepInfra": ("DEEPINFRA_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") + _status._section("API Keys") for name, env_ref in _API_KEYS.items(): - value = _first_env_value(env_ref) - _row(name, bool(value), redact_key(value)) + value = _status._first_env_value(env_ref) + _status._row(name, bool(value), _status.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)) + _status._row("Anthropic", bool(anthropic_value), _status.redact_key(anthropic_value)) def _render_auth_providers(ctx): - from hermes_cli.status import _detail, _row, _section - _section("Auth Providers") + _status._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. + # Read-only display: 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. + # xAI OAuth is guarded separately so an import failure there cannot disrupt the other rows. try: statuses["get_xai_oauth_auth_status"] = auth.get_xai_oauth_auth_status() or {} except Exception: @@ -144,19 +127,16 @@ def _render_auth_providers(ctx): try: info = get_nous_portal_account_info() except Exception: - info = None + pass 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)", - ) + _status._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 ( @@ -168,20 +148,19 @@ def _render_auth_providers(ctx): 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), - ): + ("Error:", nous_error, nous_error)): if show: - _detail(label, value) + _status._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") + _status._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" @@ -193,11 +172,12 @@ def _render_nous_gateway(ctx): 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}") + mark = _status.check_mark(f.available or f.active or f.managed_by_nous) + print(f" {f.label:<15} {mark} {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") + # Nous OAuth without entitlement, or an opaque inference key without Portal account + # information, cannot enable the Tool Gateway. + _status._section("Nous Tool Gateway") message = format_nous_portal_entitlement_message( ctx.nous_account_info, capability="managed web, image, TTS, STT, browser, and Modal tools" ) @@ -206,23 +186,27 @@ def _render_nous_gateway(ctx): 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") + _status._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)'}") + configured = bool(_status._first_env_value(env_vars)) + state = "configured" if configured else "not configured (run: hermes model)" + print(f" {pname:<16} {_status.check_mark(configured)} {state}") - # 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": + # 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 common support case. + if _status._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" + base = ((model_cfg.get("base_url") if isinstance(model_cfg, dict) else None) + or _status.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) + models = probe_lmstudio_models(api_key=_status.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}") + print(f" {'LM Studio':<16} {_status.check_mark(ok)} {msg}") + + +import hermes_cli.status as _status # noqa: E402 (bottom: hermes_cli.status imports this module) diff --git a/hermes_cli/stderr_timestamp.py b/hermes_cli/stderr_timestamp.py index c7263590d2..5ed7fe4b60 100644 --- a/hermes_cli/stderr_timestamp.py +++ b/hermes_cli/stderr_timestamp.py @@ -15,7 +15,6 @@ from typing import BinaryIO, Sequence, TextIO EXTERNAL_SUPERVISOR_FLAG = "--external-supervisor" - _TIMESTAMP_PREFIX = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}(?:\s|$)") @@ -73,10 +72,7 @@ def _is_hermes_gateway_run_argv(command: Sequence[str]) -> bool: return bool(looks_like_gateway_command_line(" ".join(str(part) for part in command))) -def _prepare_child_command( - command: Sequence[str], - environ: Mapping[str, str] | None = None, -) -> list[str]: +def _prepare_child_command(command: Sequence[str], environ: Mapping[str, str] | None = None) -> list[str]: """Return the argv to exec, upgrading stale launchd-wrapped gateway commands. launchd stamps ``XPC_SERVICE_NAME=`` only on this wrapper (its direct child; an @@ -87,19 +83,13 @@ def _prepare_child_command( argv = [str(part) for part in command] env = os.environ if environ is None else environ xpc_service = str(env.get("XPC_SERVICE_NAME", "")).strip() - if ( - EXTERNAL_SUPERVISOR_FLAG not in argv - and xpc_service and xpc_service != "0" - and _is_hermes_gateway_run_argv(argv) - ): + if EXTERNAL_SUPERVISOR_FLAG not in argv and xpc_service and xpc_service != "0" and _is_hermes_gateway_run_argv(argv): argv.append(EXTERNAL_SUPERVISOR_FLAG) return argv def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Run a command and timestamp each stderr line into a log file." - ) + parser = argparse.ArgumentParser(description="Run a command and timestamp each stderr line into a log file.") parser.add_argument("--error-log", required=True, type=Path) parser.add_argument("command", nargs=argparse.REMAINDER) args = parser.parse_args(argv) diff --git a/hermes_cli/stdio.py b/hermes_cli/stdio.py index b927404e67..4396bfb9c4 100644 --- a/hermes_cli/stdio.py +++ b/hermes_cli/stdio.py @@ -1,7 +1,7 @@ """Windows-safe stdio configuration. -The fix is to force UTF-8 on the Python side and also flip the console's code page to UTF-8 (65001). -Both matter: Python-level only helps when Python's stdout is a real TTY; code-page flipping lets +Forces UTF-8 on the Python side and also flips the console's code page to UTF-8 (65001). Both +matter: Python-level only helps when Python's stdout is a real TTY; code-page flipping lets subprocesses and child Python ``print()`` calls agree on encoding. """ @@ -12,7 +12,6 @@ import sys __all__ = ["configure_windows_stdio", "is_windows"] - _CONFIGURED = False @@ -22,174 +21,100 @@ def is_windows() -> bool: def _flip_console_code_page_to_utf8() -> None: - """Set the attached console's input and output code pages to UTF-8. - - Uses ``SetConsoleCP`` / ``SetConsoleOutputCP`` (CP_UTF8 = 65001). Failure is silent: without an - attached console (redirected stdout, service, PTY-less CI) the calls return 0 and we move on. - """ + """``SetConsoleCP``/``SetConsoleOutputCP`` to CP_UTF8 (65001). Silent on failure: without an + attached console (redirected stdout, service, PTY-less CI) the calls return 0 and we move on.""" try: import ctypes kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] - # Best-effort; if there's no console attached these just fail silently. kernel32.SetConsoleCP(65001) kernel32.SetConsoleOutputCP(65001) except Exception: - # ctypes import, missing kernel32, or non-Windows — any failure here - # is non-fatal. We've still reconfigured Python's own streams below. pass def _reconfigure_stream(stream, *, encoding: str = "utf-8", errors: str = "replace") -> None: - """Reconfigure a text stream to UTF-8 in place. - - Skips rather than raising when the stream isn't a ``TextIOWrapper`` (e.g. redirected to an - ``io.StringIO`` during tests). - """ + """Reconfigure a text stream to UTF-8 in place; skips streams without ``reconfigure`` (e.g. an + ``io.StringIO`` substituted during tests).""" try: reconfigure = getattr(stream, "reconfigure", None) - if reconfigure is None: - return - reconfigure(encoding=encoding, errors=errors) + if reconfigure is not None: + reconfigure(encoding=encoding, errors=errors) except Exception: pass def configure_windows_stdio() -> bool: - """Force UTF-8 stdio on Windows. No-op elsewhere. + """Force UTF-8 stdio on Windows. No-op elsewhere. Idempotent; returns ``True`` only when something actually changed. Set ``HERMES_DISABLE_WINDOWS_UTF8=1`` to opt out (forces the old cp1252 path for diagnosing - encoding bugs). Also sets a default ``EDITOR`` on Windows if none is set (see - ``_default_windows_editor``). + encoding bugs). Also sets a default ``EDITOR`` on Windows if none is set. """ global _CONFIGURED if _CONFIGURED: return False - if not is_windows(): - # Mark configured so repeated calls on POSIX are true no-ops. - _CONFIGURED = True + if not is_windows() or os.environ.get("HERMES_DISABLE_WINDOWS_UTF8") in {"1", "true", "True", "yes"}: + _CONFIGURED = True # repeated calls on POSIX / opted-out are true no-ops return False - if os.environ.get("HERMES_DISABLE_WINDOWS_UTF8") in {"1", "true", "True", "yes"}: - _CONFIGURED = True - return False - - # Encourage every child Python process spawned by the agent to also use - # UTF-8 for its stdio. PYTHONIOENCODING wins over the locale-based - # default in subprocesses. Don't override an explicit user setting. + # Make child Python processes use UTF-8 stdio too (PYTHONIOENCODING wins over the locale + # default; PYTHONUTF8=1 enables UTF-8 Mode, PEP 540). Never override an explicit user setting. os.environ.setdefault("PYTHONIOENCODING", "utf-8") - # PYTHONUTF8 = 1 enables UTF-8 Mode globally for any Python subprocess - # (PEP 540). Again, don't override an explicit setting. os.environ.setdefault("PYTHONUTF8", "1") - # Set EDITOR to a working Windows default if neither EDITOR nor VISUAL - # is set. prompt_toolkit's ``open_in_editor`` falls back to POSIX-only - # paths (``/usr/bin/nano``, ``/usr/bin/vi``) that don't exist on - # Windows — Ctrl+X Ctrl+E and ``/edit`` silently do nothing there - # otherwise. This happens even with full Git for Windows installed, - # so it's not a MinGit-specific issue. + # prompt_toolkit's ``open_in_editor`` falls back to POSIX-only paths (/usr/bin/nano, /usr/bin/vi) + # that don't exist on Windows — Ctrl+X Ctrl+E and ``/edit`` silently do nothing there + # otherwise, even with full Git for Windows installed. _default_editor = _default_windows_editor() if _default_editor and not os.environ.get("EDITOR") and not os.environ.get("VISUAL"): os.environ["EDITOR"] = _default_editor - # Augment PATH with the Hermes-managed Git install directories so - # subprocess calls (bash, rg, grep, etc.) resolve even in sessions - # that started before the User PATH broadcast reached them. When - # install.ps1 adds these to User PATH via SetEnvironmentVariable, - # already-running shells don't see the change — which means hermes - # launched from the install session won't find rg / bash / grep - # even though they're "installed". Prepending the known paths here - # closes that gap. No-op when the paths don't exist (e.g. system-Git - # install without Hermes-managed PortableGit). _augment_path_with_known_tools() - - # Flip the console code page first so that any subprocess that - # inherits the console (e.g. a launched shell) also sees CP_UTF8. + # Flip the console code page first so any subprocess inheriting the console also sees CP_UTF8. _flip_console_code_page_to_utf8() - - # Reconfigure Python's own stdio wrappers so ``print()`` calls from - # this process round-trip emoji / box-drawing / non-Latin text. - # ``errors="replace"`` means a genuinely unencodable byte sequence - # gets a ``?`` rather than crashing the interpreter — we prefer - # degraded output over a stack trace. - _reconfigure_stream(sys.stdout) - _reconfigure_stream(sys.stderr) - # stdin is re-configured for completeness; Hermes's interactive - # input path uses prompt_toolkit which manages its own encoding, - # but batch/pipe input benefits from UTF-8 decoding on stdin too. - _reconfigure_stream(sys.stdin) - + # ``errors="replace"``: a genuinely unencodable sequence prints ``?`` rather than crashing the + # interpreter. stdin is included for batch/pipe input (prompt_toolkit manages its own encoding). + for stream in (sys.stdout, sys.stderr, sys.stdin): + _reconfigure_stream(stream) _CONFIGURED = True return True def _default_windows_editor() -> str: - """Return a Windows-appropriate default for ``$EDITOR``. - - Priority order, first match wins: - - 1. ``notepad`` — ships with every Windows install, no deps, works as a blocking editor - (``subprocess.call(["notepad", file])`` blocks until the user closes the window). This is the - "always-works" default. - """ + """Windows default for ``$EDITOR``: ``notepad`` (ships with every install, blocks until the + window closes). The bare name keeps prompt_toolkit's shlex split away from paths with spaces; + "" when even notepad is missing (WinPE, Nano Server) so prompt_toolkit's no-op applies.""" import shutil - - # notepad.exe is always in %SystemRoot%\System32 on Windows, so shutil.which - # will reliably find it. Return the bare name so prompt_toolkit's shlex - # split doesn't trip over a path containing spaces. - if shutil.which("notepad"): - return "notepad" - # On the extreme off-chance notepad is missing (WinPE, Nano Server), fall - # back to nothing and let prompt_toolkit's silent no-op do its thing. - return "" + return "notepad" if shutil.which("notepad") else "" def _augment_path_with_known_tools() -> None: - r"""Prepend well-known Hermes-managed tool directories to os.environ['PATH']. + r"""Prepend Hermes-managed tool directories to ``PATH`` (no-op on POSIX / missing dirs). - Fixes the "User PATH was just updated but my process can't see it" gap on Windows. When - install.ps1 runs, it adds entries like ``%LOCALAPPDATA%\hermes\git\bin`` to the User PATH via - ``SetEnvironmentVariable(..., "User")``. - - Patch-up strategy: add the known Hermes-managed tool directories to our PATH at startup so - subprocess calls resolve correctly. No-op on POSIX and when the directories don't exist. The - User PATH broadcast still happens in the background for future shells; this just smooths over - the first-launch gap. + install.ps1 adds entries like ``%LOCALAPPDATA%\hermes\git\bin`` to the User PATH via + ``SetEnvironmentVariable``, but already-running shells never see that broadcast, so a hermes + launched from the install session would not find rg / bash / grep. Prepending the known dirs + at startup closes that first-launch gap. """ if not is_windows(): return - local_appdata = os.environ.get("LOCALAPPDATA", "") if not local_appdata: return - # Known tool dirs installed by scripts/install.ps1. Kept in sync with - # the PATH entries that installer adds to User scope — the two lists - # should match so this prefill fully mirrors what a fresh shell would - # see on next launch. + # Kept in sync with the PATH entries scripts/install.ps1 adds to User scope. The venv Scripts + # dir hosts hermes.exe + pip console scripts; WinGet\Links is where ``winget install`` drops + # CLI shims (ripgrep lands there as rg.exe). candidate_dirs = [ os.path.join(local_appdata, "hermes", "git", "cmd"), os.path.join(local_appdata, "hermes", "git", "bin"), os.path.join(local_appdata, "hermes", "git", "usr", "bin"), - # Hermes venv Scripts directory — host of the hermes.exe shim itself, - # also where any pip-installed console scripts land. Usually already - # on PATH when the user invokes hermes, but harmless to include. os.path.join(local_appdata, "hermes", "hermes-agent", "venv", "Scripts"), - # WinGet packages directory — where ``winget install`` drops CLI - # shims by default (ripgrep lands here as rg.exe). Covers the case - # of a system-Git install + ripgrep-via-winget that isn't yet on - # the spawning shell's PATH. - os.path.join(local_appdata, "Microsoft", "WinGet", "Links"), - ] - + os.path.join(local_appdata, "Microsoft", "WinGet", "Links")] existing = os.environ.get("PATH", "") existing_lower = {p.lower() for p in existing.split(os.pathsep) if p} - prepend = [] - for d in candidate_dirs: - if os.path.isdir(d) and d.lower() not in existing_lower: - prepend.append(d) - + prepend = [d for d in candidate_dirs if os.path.isdir(d) and d.lower() not in existing_lower] if prepend: os.environ["PATH"] = os.pathsep.join([*prepend, existing]) diff --git a/hermes_cli/terminal_breadcrumbs.py b/hermes_cli/terminal_breadcrumbs.py index 67defe1481..ed05eb4dda 100644 --- a/hermes_cli/terminal_breadcrumbs.py +++ b/hermes_cli/terminal_breadcrumbs.py @@ -1,10 +1,6 @@ -"""Per-terminal session breadcrumbs for ``hermes -c`` / ``--continue``. - -Everything here is strictly best-effort: no function raises, and when no stable terminal identity -can be derived (no tty and no known multiplexer env var) breadcrumbs are skipped entirely and ``-c`` -falls back to the existing latest-session behavior. Gated by ``session.terminal_continue`` in -config.yaml (default true). -""" +"""Per-terminal session breadcrumbs for ``hermes -c`` / ``--continue``. Strictly best-effort: no +function raises; without a stable terminal identity (no tty, no known multiplexer env var) ``-c`` +falls back to latest-session. Gated by ``session.terminal_continue`` (default true).""" from __future__ import annotations @@ -16,20 +12,12 @@ import time from pathlib import Path from typing import Optional -# Multiplexer / terminal-emulator identity env vars, checked in order when -# no real tty path is available (e.g. stdin piped but stdout still a pty -# owned by a known terminal). -_TERMINAL_ENV_VARS = ( - "ZELLIJ_PANE_ID", - "TMUX_PANE", - "KITTY_WINDOW_ID", - "WEZTERM_PANE", - "TERM_SESSION_ID", - "WT_SESSION", -) +# Multiplexer / terminal-emulator identity env vars, checked in order when no real tty path is +# available (e.g. stdin piped but stdout still a pty owned by a known terminal). +_TERMINAL_ENV_VARS = ("ZELLIJ_PANE_ID", "TMUX_PANE", "KITTY_WINDOW_ID", "WEZTERM_PANE", "TERM_SESSION_ID", "WT_SESSION") -# Breadcrumbs older than this are pruned opportunistically on each write — -# a pane id from a tmux server restarted last month means nothing today. +# Breadcrumbs older than this are pruned opportunistically on each write — a pane id from a tmux +# server restarted last month means nothing today. _STALE_AFTER_SECONDS = 30 * 24 * 60 * 60 _SANITIZE_RE = re.compile(r"[^A-Za-z0-9._-]") @@ -37,7 +25,6 @@ _SANITIZE_RE = re.compile(r"[^A-Za-z0-9._-]") def _breadcrumbs_dir() -> Path: from hermes_constants import get_hermes_home - return get_hermes_home() / "terminal-sessions" @@ -47,12 +34,8 @@ def _sanitize(raw: str) -> str: def get_terminal_id() -> Optional[str]: - """Derive a stable identity for the terminal this process runs in. - - Prefers the real tty device path (stdin, then stdout), else the first present - multiplexer/emulator env var. Returns ``None`` when neither is available — callers must then - skip breadcrumbs entirely. - """ + """Stable identity for this terminal: the tty device path (stdin, then stdout), else the first + present multiplexer/emulator env var; ``None`` when neither exists (callers skip breadcrumbs).""" for fd in (sys.stdin, sys.stdout): try: name = os.ttyname(fd.fileno()) @@ -91,11 +74,8 @@ def _prune_stale(directory: Path, now: float) -> None: def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None: - """Record that this terminal's live session is ``session_id``. - - Synchronous, best-effort, never raises. No-op when the feature is disabled, the session id is - empty, or no terminal identity exists. - """ + """Record that this terminal's live session is ``session_id``. Never raises; no-op when the + feature is disabled, the session id is empty, or no terminal identity exists.""" try: if not session_id or not is_enabled(): return @@ -105,11 +85,7 @@ def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None: directory = _breadcrumbs_dir() directory.mkdir(parents=True, exist_ok=True) now = time.time() - payload = { - "session_id": session_id, - "cwd": cwd or os.getcwd(), - "ts": now, - } + payload = {"session_id": session_id, "cwd": cwd or os.getcwd(), "ts": now} tmp = directory / f".{terminal_id}.tmp" tmp.write_text(json.dumps(payload), encoding="utf-8") os.replace(tmp, directory / terminal_id) @@ -119,10 +95,7 @@ def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None: def read_breadcrumb() -> Optional[dict]: - """Return this terminal's breadcrumb payload, or ``None``. - - Ignores breadcrumbs older than the staleness window. Never raises. - """ + """This terminal's breadcrumb payload, or ``None`` (missing, corrupt, or stale). Never raises.""" try: terminal_id = get_terminal_id() if not terminal_id: @@ -140,12 +113,9 @@ def read_breadcrumb() -> Optional[dict]: def resolve_breadcrumb_session() -> Optional[str]: - """Resolve a bare ``-c`` for this terminal, or ``None`` to fall back. - - Returns the breadcrumb's session id only when it still exists in the session DB, projected - forward through the compression chain so the resume lands on the live tip rather than a dead - compressed parent (same projection as ``main._resolve_session_by_name_or_id``). - """ + """Resolve a bare ``-c`` for this terminal, or ``None`` to fall back. The breadcrumb's session + id counts only if it still exists in the DB, projected through the compression chain so the + resume lands on the live tip (same projection as ``main._resolve_session_by_name_or_id``).""" if not is_enabled(): return None crumb = read_breadcrumb() @@ -154,23 +124,23 @@ def resolve_breadcrumb_session() -> Optional[str]: session_id = str(crumb.get("session_id") or "").strip() if not session_id: return None - db = None try: from hermes_state import SessionDB db = SessionDB() + except Exception: + return None + try: if not db.get_session(session_id): return None # session was deleted — fall back to latest try: - session_id = db.get_compression_tip(session_id) or session_id + return db.get_compression_tip(session_id) or session_id except Exception: - pass - return session_id + return session_id except Exception: return None finally: - if db is not None: - try: - db.close() - except Exception: - pass + try: + db.close() + except Exception: + pass diff --git a/hermes_cli/terminal_notify.py b/hermes_cli/terminal_notify.py index 47ff111761..4b370ef8d3 100644 --- a/hermes_cli/terminal_notify.py +++ b/hermes_cli/terminal_notify.py @@ -1,8 +1,8 @@ """Terminal-native desktop notifications: OSC 9 and Warp's OSC 777 CLI-agent protocol. -- **OSC 9** (``ESC ] 9 ; BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification; -terminals that don't know the sequence drop it. - **OSC 777** (``ESC ] 777 ; notify ; warp://cli- -agent ; BEL``): Warp's structured CLI-agent protocol (tab status + notification mailbox). +OSC 9 (``ESC ] 9 ; BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification; +others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; BEL``): Warp's +structured CLI-agent protocol (tab status + notification mailbox). Sequences are written to ``/dev/tty`` because prompt_toolkit's stdout wrapper can buffer or strip raw escapes; when ``/dev/tty`` can't be opened (Windows, no controlling terminal) they fall back to @@ -18,13 +18,9 @@ import sys _C0_AND_DEL = re.compile(r"[\x00-\x1f\x7f]") _WARP_PROTOCOL_VERSION = 1 -# Last Warp release per channel that set WARP_CLI_AGENT_PROTOCOL_VERSION but -# could not render structured payloads (Warp's reference agent plugin, -# should-use-structured.sh). Bash compares these lexicographically; so do we. -_WARP_LAST_BROKEN = { - "stable": "v0.2026.03.25.08.24.stable_05", - "preview": "v0.2026.03.25.08.24.preview_05", -} +# Last Warp release per channel that set WARP_CLI_AGENT_PROTOCOL_VERSION but could not render +# structured payloads (Warp's should-use-structured.sh). Bash compares lexicographically; so do we. +_WARP_LAST_BROKEN = {"stable": "v0.2026.03.25.08.24.stable_05", "preview": "v0.2026.03.25.08.24.preview_05"} def _write_tty(seq: str) -> None: @@ -50,15 +46,10 @@ def osc9(body: str) -> str: def warp_supported(env=None) -> bool: """True when running in a Warp build that can render OSC 777 agent payloads.""" env = os.environ if env is None else env - if env.get("TERM_PROGRAM") != "WarpTerminal" or not env.get("WARP_CLI_AGENT_PROTOCOL_VERSION"): - return False client = env.get("WARP_CLIENT_VERSION", "") - if not client: + if env.get("TERM_PROGRAM") != "WarpTerminal" or not env.get("WARP_CLI_AGENT_PROTOCOL_VERSION") or not client: return False - for channel, last_broken in _WARP_LAST_BROKEN.items(): - if channel in client and client <= last_broken: - return False - return True + return not any(channel in client and client <= last_broken for channel, last_broken in _WARP_LAST_BROKEN.items()) def warp_osc777(event: str, detail: str, session_id: str = "") -> str: @@ -68,14 +59,8 @@ def warp_osc777(event: str, detail: str, session_id: str = "") -> str: except ValueError: advertised = 1 cwd = os.getcwd() - payload = { - "v": min(advertised, _WARP_PROTOCOL_VERSION), - "agent": "hermes", - "event": event, - "session_id": session_id, - "cwd": cwd, - "project": os.path.basename(cwd), - } + payload = {"v": min(advertised, _WARP_PROTOCOL_VERSION), "agent": "hermes", "event": event, + "session_id": session_id, "cwd": cwd, "project": os.path.basename(cwd)} payload["summary" if event == "permission_request" else "response"] = detail[:200] return f"\x1b]777;notify;warp://cli-agent;{json.dumps(payload, separators=(',', ':'))}\x07" diff --git a/hermes_cli/timeouts.py b/hermes_cli/timeouts.py index 7480634319..71b1880c17 100644 --- a/hermes_cli/timeouts.py +++ b/hermes_cli/timeouts.py @@ -9,53 +9,40 @@ def _coerce_timeout(raw: object) -> float | None: return timeout if timeout > 0 else None -def _configured_timeout( - provider_id: str, model: str | None, model_key: str, provider_key: str -) -> float | None: +def _configured_timeout(provider_id: str, model: str | None, model_key: str, provider_key: str) -> float | None: """Per-model ``providers..models..`` wins over ``providers..``.""" if not provider_id: return None - try: from hermes_cli.config import load_config_readonly config = load_config_readonly() except Exception: return None - providers = config.get("providers", {}) if isinstance(config, dict) else {} provider_config = providers.get(provider_id, {}) if isinstance(providers, dict) else {} if not isinstance(provider_config, dict): return None - model_config = _get_model_config(provider_config, model) if model_config is not None: timeout = _coerce_timeout(model_config.get(model_key)) if timeout is not None: return timeout - return _coerce_timeout(provider_config.get(provider_key)) -def get_provider_request_timeout( - provider_id: str, model: str | None = None -) -> float | None: +def get_provider_request_timeout(provider_id: str, model: str | None = None) -> float | None: """Return a configured provider request timeout in seconds, if any.""" return _configured_timeout(provider_id, model, "timeout_seconds", "request_timeout_seconds") -def get_provider_stale_timeout( - provider_id: str, model: str | None = None -) -> float | None: +def get_provider_stale_timeout(provider_id: str, model: str | None = None) -> float | None: """Return a configured non-stream stale timeout in seconds, if any.""" return _configured_timeout(provider_id, model, "stale_timeout_seconds", "stale_timeout_seconds") -def _get_model_config( - provider_config: dict[str, object], model: str | None -) -> dict[str, object] | None: +def _get_model_config(provider_config: dict[str, object], model: str | None) -> dict[str, object] | None: if not model: return None - models = provider_config.get("models", {}) model_config = models.get(model, {}) if isinstance(models, dict) else {} return model_config if isinstance(model_config, dict) else None