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

Merge branch 'simp/r3-24-D' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:58:17 -07:00
parent 2a96c4c8ae 456377f704
commit dcd33ce2b2
15 changed files with 822 additions and 1323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/setup_summary.py
+90 -126
View File
@@ -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.<name>`` take effect.
Names originating in setup.py are imported lazily inside function bodies so test patches on
``hermes_cli.setup.<name>`` 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 <key> <value>", ""),
)
("hermes config set <key> <value>", ""))
_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.<registry>`` 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 ' <green cmd><desc>'."""
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)
hermes_cli/setup_terminal.py
+133 -173
View File
@@ -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.<name>`` 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)
hermes_cli/setup_tts.py
+101 -144
View File
@@ -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.<name>`` 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)
hermes_cli/setup_whatsapp_cloud.py
+205 -306
View File
@@ -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://<random>.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: <tunnel-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://<random>.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: <tunnel-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
hermes_cli/sizefmt.py
+4 -14
View File
@@ -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):
hermes_cli/sqlite_runtime.py
+12 -39
View File
@@ -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
hermes_cli/sqlite_safe_read.py
+65 -141
View File
@@ -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
hermes_cli/sqlite_util.py
+4 -12
View File
@@ -7,12 +7,8 @@ import sqlite3
def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl: str) -> bool:
"""``ALTER TABLE <table> ADD COLUMN <ddl>``, 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 <table> ADD COLUMN <ddl>``, 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
hermes_cli/status.py
+76 -77
View File
@@ -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):
hermes_cli/status_auth.py
+53 -69
View File
@@ -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)
hermes_cli/stderr_timestamp.py
+3 -13
View File
@@ -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=<job label>`` 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)
hermes_cli/stdio.py
+36 -111
View File
@@ -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])
hermes_cli/terminal_breadcrumbs.py
+26 -56
View File
@@ -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
hermes_cli/terminal_notify.py
+10 -25
View File
@@ -1,8 +1,8 @@
"""Terminal-native desktop notifications: OSC 9 and Warp's OSC 777 CLI-agent protocol.
- **OSC 9** (``ESC ] 9 ; <body> 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 ; <json> BEL``): Warp's structured CLI-agent protocol (tab status + notification mailbox).
OSC 9 (``ESC ] 9 ; <body> BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification;
others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; <json> 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"
hermes_cli/timeouts.py
+4 -17
View File
@@ -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.<id>.models.<model>.<model_key>`` wins over ``providers.<id>.<provider_key>``."""
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