diff --git a/hermes_cli/secret_prompt.py b/hermes_cli/secret_prompt.py index 5f501d8b96..b274c8f0b9 100644 --- a/hermes_cli/secret_prompt.py +++ b/hermes_cli/secret_prompt.py @@ -10,15 +10,11 @@ from collections.abc import Callable _BACKSPACE_CHARS = {"\b", "\x7f"} _ENTER_CHARS = {"\r", "\n"} -_EOF_CHARS = {"\x04", "\x1a"} +_EOF_CHARS = {"\x04", "\x1a", ""} # "" == stream closed def _collect_masked_input( - read_char: Callable[[], str], - write: Callable[[str], object], - prompt: str, - *, - mask: str = "*", + read_char: Callable[[], str], write: Callable[[str], object], prompt: str, *, mask: str = "*", ) -> str: """Read one secret line while writing a mask character per typed char.""" value: list[str] = [] @@ -26,9 +22,6 @@ def _collect_masked_input( while True: ch = read_char() - if ch == "": - write("\r\n") - raise EOFError if ch in _ENTER_CHARS: write("\r\n") return "".join(value) @@ -44,10 +37,9 @@ def _collect_masked_input( write("\b \b") continue if ch == "\x1b": - # Ignore escape itself. Terminals commonly send escape-prefixed - # navigation/delete sequences; they should not become secret text. + # Terminals send escape-prefixed navigation/delete sequences; they must not become + # secret text. continue - value.append(ch) if mask: write(mask) @@ -59,22 +51,12 @@ def masked_secret_prompt(prompt: str, *, mask: str = "*") -> str: Falls back to ``getpass.getpass`` when stdin/stdout are not interactive or when raw terminal handling is unavailable. """ - stdin = sys.stdin - stdout = sys.stdout - - if not _stream_is_tty(stdin) or not _stream_is_tty(stdout): + if not _stream_is_tty(sys.stdin) or not _stream_is_tty(sys.stdout): return getpass.getpass(prompt) - if os.name == "nt": - try: - return _masked_secret_prompt_windows(prompt, mask=mask) - except (KeyboardInterrupt, EOFError): - raise - except Exception: - return getpass.getpass(prompt) - + masked = _masked_secret_prompt_windows if os.name == "nt" else _masked_secret_prompt_posix try: - return _masked_secret_prompt_posix(prompt, mask=mask) + return masked(prompt, mask=mask) except (KeyboardInterrupt, EOFError): raise except Exception: @@ -88,6 +70,11 @@ def _stream_is_tty(stream) -> bool: return False +def _write(text: str) -> None: + sys.stdout.write(text) + sys.stdout.flush() + + def _masked_secret_prompt_windows(prompt: str, *, mask: str) -> str: import msvcrt @@ -98,29 +85,16 @@ def _masked_secret_prompt_windows(prompt: str, *, mask: str) -> str: return "\x1b" return ch - def write(text: str) -> None: - sys.stdout.write(text) - sys.stdout.flush() - - return _collect_masked_input(read_char, write, prompt, mask=mask) + return _collect_masked_input(read_char, _write, prompt, mask=mask) def _masked_secret_prompt_posix(prompt: str, *, mask: str) -> str: import termios import tty - fd = sys.stdin.fileno() old_attrs = termios.tcgetattr(fd) - - def read_char() -> str: - return sys.stdin.read(1) - - def write(text: str) -> None: - sys.stdout.write(text) - sys.stdout.flush() - try: tty.setraw(fd) - return _collect_masked_input(read_char, write, prompt, mask=mask) + return _collect_masked_input(lambda: sys.stdin.read(1), _write, prompt, mask=mask) finally: termios.tcsetattr(fd, termios.TCSADRAIN, old_attrs) diff --git a/hermes_cli/secrets_cli.py b/hermes_cli/secrets_cli.py index 9b5e7a0bf9..dc07681741 100644 --- a/hermes_cli/secrets_cli.py +++ b/hermes_cli/secrets_cli.py @@ -15,49 +15,26 @@ from rich.console import Console from rich.panel import Panel from rich.table import Table -# NOTE: the Bitwarden backend (``agent.secret_sources.bitwarden``) pulls in -# ``cryptography`` at module-import time. On Windows the resulting -# ``cryptography._rust.pyd`` is mapped into the running process — and when -# that process is ``hermes update``, the self-lock preflight detects the -# loaded native module and defers (#86781). Keep the backend import lazy: -# this module is registered parse-time from ``hermes_cli.main`` and must not -# touch ``bw`` until a handler actually runs. -# -# ``_BWS_VERSION`` is duplicated here (as a plain string) so ``register_cli`` -# can render the ``install --help`` text without importing the backend. -# ``agent.secret_sources.bitwarden._BWS_VERSION`` is the source of truth; -# bump both together when pinning a new bws release. +# The Bitwarden backend pulls in ``cryptography`` at import time; on Windows that mapped native +# module makes the ``hermes update`` self-lock preflight defer. This module is registered +# parse-time from ``hermes_cli.main``, so the backend import stays lazy (nothing touches ``bw`` +# until a handler runs) and ``_BWS_VERSION`` is duplicated here for the ``install --help`` text. +# ``agent.secret_sources.bitwarden._BWS_VERSION`` is the source of truth; bump both together. _BWS_VERSION = "2.0.0" from hermes_cli._secrets_common import ( - arg, - cfg_str, - cli_version, - disable_secret_source, - flag, - print_status_panel, - print_table, - prompt_index, - register_subcommands, - require_enabled, - rotate_token, - secret_cli_env, - section_cfg, + arg, cfg_str, cli_version, disable_secret_source, flag, print_status_panel, print_table, + prompt_index, register_subcommands, require_enabled, rotate_token, secret_cli_env, section_cfg, yn, ) -from hermes_cli.config import ( - get_env_path, - load_config, - save_config, - save_env_value, -) +from hermes_cli.config import get_env_path, load_config, save_config, save_env_value from hermes_cli.secret_prompt import masked_secret_prompt # Old names kept bound: tests monkeypatch ``secrets_cli._bws_version``. _bws_version = cli_version _yn = yn - +_DEFAULT_TOKEN_ENV = "BWS_ACCESS_TOKEN" _NOT_BSM_TOKEN_WARNING = ( "[yellow]Warning: token doesn't start with '0.' — usually that means " "you pasted something other than a BSM access token.[/yellow]" @@ -71,17 +48,12 @@ _NOT_BSM_TOKEN_WARNING_CONTINUING = ( def _load_bw(): """Import ``agent.secret_sources.bitwarden`` on first use (crypto payload).""" from agent.secret_sources import bitwarden as _bw - return _bw def __getattr__(name: str): - """PEP 562 module-level lazy resolver. - - Existing callers (and upstream tests) monkeypatch attributes on ``hermes_cli.secrets_cli.bw`` - directly. Resolving that attribute at module-import time would re-import ``cryptography`` - eagerly — the very self-lock we are preventing (#86781). - """ + """PEP 562 lazy ``bw`` attribute (tests monkeypatch ``secrets_cli.bw``); an eager binding + would re-import ``cryptography`` at import time.""" if name == "bw": return _load_bw() raise AttributeError(f"module {__name__!r} has no attribute {name!r}") @@ -125,47 +97,85 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None: # --------------------------------------------------------------------------- -def cmd_setup(args: argparse.Namespace) -> int: - bw = _load_bw() - console = Console() - console.print( - Panel.fit( - "[bold]Bitwarden Secrets Manager setup[/bold]\n\n" - "Need an access token? In the Bitwarden web app:\n" - " Secrets Manager → Machine accounts → [your account] →\n" - " Access tokens → Create access token\n\n" - "Copy the token (starts with [cyan]0.[/cyan]…) — it cannot be retrieved later.", - border_style="cyan", - ) - ) - - # ------------------------------------------------------------------ binary +def _step(console: Console, n: int, title: str) -> None: console.print() - console.print("[bold]Step 1[/bold] Install the bws CLI") + console.print(f"[bold]Step {n}[/bold] {title}") + + +def _setup_binary(bw, console: Console) -> Optional[Path]: + """Step 1: locate or download bws; None (after printing) on failure.""" + _step(console, 1, "Install the bws CLI") try: binary = bw.find_bws(install_if_missing=False) if binary is None: console.print(" No bws on PATH — downloading…") binary = bw.install_bws() - version = _bws_version(binary) - console.print(f" [green]✓[/green] {binary} ({version})") + console.print(f" [green]✓[/green] {binary} ({_bws_version(binary)})") + return binary except Exception as exc: # noqa: BLE001 console.print(f" [red]✗ Could not install bws: {exc}[/red]") - console.print( - " Manual install: " - "https://github.com/bitwarden/sdk-sm/releases" - ) + console.print(" Manual install: https://github.com/bitwarden/sdk-sm/releases") + return None + + +def _missing_noninteractive_flags(args: argparse.Namespace) -> list[str]: + """Setup flags a no-TTY run must supply (BWS_SERVER_URL env substitutes for --server-url).""" + provided = { + "--access-token": args.access_token, + "--server-url": (args.server_url or "").strip() or os.environ.get("BWS_SERVER_URL", ""), + "--project-id": args.project_id} + return [flag for flag, value in provided.items() if not (value and value.strip())] + + +def _setup_token(args: argparse.Namespace, console: Console, token_env: str) -> Optional[str]: + """Step 2: take the token from ``--access-token`` or a masked prompt and persist it.""" + _step(console, 2, "Provide your access token") + token = (args.access_token or "").strip() or masked_secret_prompt(f" Paste access token ({token_env}): ").strip() + if not token: + console.print(" [red]Empty token, aborting.[/red]") + return None + if not token.startswith("0."): + console.print(_NOT_BSM_TOKEN_WARNING_CONTINUING) + save_env_value(token_env, token) + os.environ[token_env] = token # so the test fetch below sees it + console.print(f" [green]✓[/green] stored in {get_env_path()} as {token_env}") + return token + + +def _setup_project(binary: Path, token: str, console: Console, server_url: str) -> Optional[str]: + """Step 4: list projects and let the user pick one; None (after printing) when none usable.""" + _step(console, 4, "Pick a project") + projects = _list_projects(binary, token, console, server_url=server_url) + if projects is None: + return None + if not projects: + console.print(" [yellow]No projects visible to this machine account.[/yellow]") + console.print(" In the Bitwarden web app, open the machine account → Projects tab " + "and grant it access to at least one project.") + return None + print_table(console, (("#", {"style": "cyan", "width": 4}), "Name", ("ID", {"style": "dim"})), + ((str(i), p.get("name", "?"), p.get("id", "?")) for i, p in enumerate(projects, 1))) + idx = prompt_index(console, f" Select project [1-{len(projects)}]: ", len(projects)) + return projects[idx - 1]["id"] + + +def cmd_setup(args: argparse.Namespace) -> int: + bw = _load_bw() + console = Console() + console.print(Panel.fit( + "[bold]Bitwarden Secrets Manager setup[/bold]\n\n" + "Need an access token? In the Bitwarden web app:\n" + " Secrets Manager → Machine accounts → [your account] →\n" + " Access tokens → Create access token\n\n" + "Copy the token (starts with [cyan]0.[/cyan]…) — it cannot be retrieved later.", + border_style="cyan")) + + binary = _setup_binary(bw, console) + if binary is None: return 1 - # -- non-interactive guard -- if not sys.stdin.isatty(): - # BWS_SERVER_URL env var is accepted as a non-interactive substitute for --server-url. - provided = { - "--access-token": args.access_token, - "--server-url": (args.server_url or "").strip() or os.environ.get("BWS_SERVER_URL", ""), - "--project-id": args.project_id, - } - missing = [flag for flag, value in provided.items() if not (value and value.strip())] + missing = _missing_noninteractive_flags(args) if missing: console.print( f" [red]Non-interactive mode (no TTY) requires all setup flags.[/red]\n" @@ -174,79 +184,34 @@ def cmd_setup(args: argparse.Namespace) -> int: " hermes secrets bitwarden setup \\\n" " --access-token '0.xxx' \\\n" " --server-url 'https://vault.bitwarden.com' \\\n" - " --project-id 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'" - ) + " --project-id 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'") return 1 - # ------------------------------------------------------------------- token - console.print() - console.print("[bold]Step 2[/bold] Provide your access token") cfg = load_config() secrets_cfg = cfg.setdefault("secrets", {}).setdefault("bitwarden", {}) - token_env = secrets_cfg.get("access_token_env", "BWS_ACCESS_TOKEN") - - token = (args.access_token or "").strip() - if not token: - token = masked_secret_prompt(f" Paste access token ({token_env}): ").strip() - if not token: - console.print(" [red]Empty token, aborting.[/red]") + token_env = secrets_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV) + token = _setup_token(args, console, token_env) + if token is None: return 1 - if not token.startswith("0."): - console.print(_NOT_BSM_TOKEN_WARNING_CONTINUING) - save_env_value(token_env, token) - os.environ[token_env] = token # so the test fetch below sees it - console.print(f" [green]✓[/green] stored in {get_env_path()} as {token_env}") - - # ------------------------------------------------------------------ region - console.print() - console.print("[bold]Step 3[/bold] Pick a Bitwarden region") + _step(console, 3, "Pick a Bitwarden region") server_url = _resolve_server_url(args, secrets_cfg, console) if server_url is None: return 1 - if server_url: - console.print(f" [green]✓[/green] using {server_url}") - else: - console.print( - " [green]✓[/green] using bws default " - "(US Cloud, https://vault.bitwarden.com)" - ) + console.print(f" [green]✓[/green] using {server_url}" if server_url + else " [green]✓[/green] using bws default (US Cloud, https://vault.bitwarden.com)") - # ------------------------------------------------------------------- project - project_given = bool(args.project_id and args.project_id.strip()) - if project_given: - project_id = args.project_id.strip() - else: - console.print() - console.print("[bold]Step 4[/bold] Pick a project") - projects = _list_projects(binary, token, console, server_url=server_url) - if projects is None: - return 1 - if not projects: - console.print(" [yellow]No projects visible to this machine account.[/yellow]") - console.print( - " In the Bitwarden web app, open the machine account → Projects tab " - "and grant it access to at least one project." - ) + project_id = (args.project_id or "").strip() + project_given = bool(project_id) + if not project_given: + project_id = _setup_project(binary, token, console, server_url) + if project_id is None: return 1 - print_table(console, (("#", {"style": "cyan", "width": 4}), "Name", ("ID", {"style": "dim"})), - ((str(i), p.get("name", "?"), p.get("id", "?")) for i, p in enumerate(projects, 1))) - - idx = prompt_index(console, f" Select project [1-{len(projects)}]: ", len(projects)) - project_id = projects[idx - 1]["id"] - - # ------------------------------------------------------------------- test - console.print() - console.print(f"[bold]Step {4 if project_given else 5}[/bold] Test fetch") + _step(console, 4 if project_given else 5, "Test fetch") try: secrets, warnings = bw.fetch_bitwarden_secrets( - access_token=token, - project_id=project_id, - binary=binary, - use_cache=False, - server_url=server_url, - ) + access_token=token, project_id=project_id, binary=binary, use_cache=False, server_url=server_url) except Exception as exc: # noqa: BLE001 console.print(f" [red]✗ Fetch failed: {exc}[/red]") return 1 @@ -259,26 +224,18 @@ def cmd_setup(args: argparse.Namespace) -> int: for w in warnings: console.print(f" [yellow]warning:[/yellow] {w}") - # ------------------------------------------------------------------- save - secrets_cfg["enabled"] = True - secrets_cfg["project_id"] = project_id - secrets_cfg["server_url"] = server_url - secrets_cfg.setdefault("access_token_env", token_env) - secrets_cfg.setdefault("cache_ttl_seconds", 300) - secrets_cfg.setdefault("override_existing", True) - secrets_cfg.setdefault("auto_install", True) + secrets_cfg.update(enabled=True, project_id=project_id, server_url=server_url) + for key, default in (("access_token_env", token_env), ("cache_ttl_seconds", 300), + ("override_existing", True), ("auto_install", True)): + secrets_cfg.setdefault(key, default) save_config(cfg) console.print() - console.print( - "[green]✓ Bitwarden Secrets Manager is enabled.[/green] " - "Secrets will be pulled at the start of every Hermes process." - ) - console.print( - " Status: [cyan]hermes secrets bitwarden status[/cyan]\n" - " Refresh: [cyan]hermes secrets bitwarden sync[/cyan]\n" - " Disable: [cyan]hermes secrets bitwarden disable[/cyan]" - ) + console.print("[green]✓ Bitwarden Secrets Manager is enabled.[/green] " + "Secrets will be pulled at the start of every Hermes process.") + console.print(" Status: [cyan]hermes secrets bitwarden status[/cyan]\n" + " Refresh: [cyan]hermes secrets bitwarden sync[/cyan]\n" + " Disable: [cyan]hermes secrets bitwarden disable[/cyan]") return 0 @@ -300,31 +257,25 @@ def cmd_status(args: argparse.Namespace) -> int: bw_cfg = _bw_cfg(load_config()) enabled = bool(bw_cfg.get("enabled")) - token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN") + token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV) project_id = bw_cfg.get("project_id", "") server_url = cfg_str(bw_cfg, "server_url") token = os.environ.get(token_env, "").strip() - token_set = bool(token) binary = bw.find_bws(install_if_missing=False) token_validation, validation_messages = _token_validation_status( - enabled=enabled, - binary=binary, - token=token, - server_url=server_url, - ) + enabled=enabled, binary=binary, token=token, server_url=server_url) print_status_panel(console, "Bitwarden Secrets Manager", ( ("Enabled", _yn(enabled)), ("Token env var", token_env), - ("Token in env", _yn(token_set)), + ("Token in env", _yn(bool(token))), ("Token validation", token_validation), ("Project ID", project_id or "[dim](unset)[/dim]"), ("Server URL", server_url or "[dim]default (US Cloud, https://vault.bitwarden.com)[/dim]"), ("Override existing", _yn(bool(bw_cfg.get("override_existing", False)))), ("Cache TTL (s)", str(bw_cfg.get("cache_ttl_seconds", 300))), ("Auto-install", _yn(bool(bw_cfg.get("auto_install", True)))), - ("bws binary", - f"{binary} ({_bws_version(binary)})" if binary else "[yellow]not installed[/yellow]"), + ("bws binary", f"{binary} ({_bws_version(binary)})" if binary else "[yellow]not installed[/yellow]"), )) for message in validation_messages: console.print(message) @@ -332,29 +283,22 @@ def cmd_status(args: argparse.Namespace) -> int: if not enabled: console.print("\n Run [cyan]hermes secrets bitwarden setup[/cyan] to enable.") return 0 - if not token_set: - console.print( - f"\n [yellow]Enabled but {token_env} is not set — Hermes will skip BSM " - "and warn on next startup.[/yellow]" - ) + if not token: + console.print(f"\n [yellow]Enabled but {token_env} is not set — Hermes will skip BSM " + "and warn on next startup.[/yellow]") if not project_id: - console.print( - "\n [yellow]Enabled but no project_id — nothing to fetch.[/yellow]" - ) + console.print("\n [yellow]Enabled but no project_id — nothing to fetch.[/yellow]") return 0 def cmd_token(args: argparse.Namespace) -> int: - """Rotate the BSM access token without re-running the whole setup wizard. - - Prompts for (or accepts via ``--access-token``) a new machine-account token, probes Bitwarden - with it (unless ``--no-verify``), and only then persists it to .env — so a bad paste never - bricks the working token. - """ + """Rotate the BSM access token without re-running the whole setup wizard: probe Bitwarden + with the new token (unless ``--no-verify``) and only then persist it, so a bad paste never + bricks the working token.""" bw = _load_bw() console = Console() bw_cfg = _bw_cfg(load_config()) - token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN") + token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV) server_url = cfg_str(bw_cfg, "server_url") def verify(token: str) -> bool: @@ -364,30 +308,22 @@ def cmd_token(args: argparse.Namespace) -> int: return True binary = bw.find_bws(install_if_missing=True) if binary is None: - console.print( - "[red]bws binary not available — cannot verify. " - "Re-run with --no-verify to store anyway.[/red]" - ) + console.print("[red]bws binary not available — cannot verify. " + "Re-run with --no-verify to store anyway.[/red]") return False console.print("Verifying against Bitwarden…") projects = _list_projects(binary, token, console, server_url=server_url) if projects is None: - console.print( - "[red]✗ New token was rejected — nothing was changed.[/red]" - ) + console.print("[red]✗ New token was rejected — nothing was changed.[/red]") return False - console.print( - f"[green]✓ Token accepted[/green] " - f"({len(projects)} project{'s' if len(projects) != 1 else ''} visible)." - ) + console.print(f"[green]✓ Token accepted[/green] " + f"({len(projects)} project{'s' if len(projects) != 1 else ''} visible).") project_id = str(bw_cfg.get("project_id", "") or "") if project_id and projects and project_id not in {p["id"] for p in projects}: console.print( f"[yellow]Warning: configured project {project_id} is not visible " "to this machine account. Grant it access in the Bitwarden web " - "app or re-run `hermes secrets bitwarden setup` to pick a " - "different project.[/yellow]" - ) + "app or re-run `hermes secrets bitwarden setup` to pick a different project.[/yellow]") return True return rotate_token( @@ -405,8 +341,7 @@ def cmd_token(args: argparse.Namespace) -> int: "[yellow]Note: the Bitwarden integration is currently disabled — " "run `hermes secrets bitwarden setup` (or set " "secrets.bitwarden.enabled: true) to turn it on.[/yellow]" - ), - ) + )) def cmd_sync(args: argparse.Namespace) -> int: @@ -416,25 +351,19 @@ def cmd_sync(args: argparse.Namespace) -> int: if not require_enabled(console, bw_cfg, "Bitwarden", "bitwarden"): return 1 - token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN") + token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV) token = os.environ.get(token_env, "").strip() if not token: console.print(f"[red]{token_env} is not set.[/red]") return 1 - project_id = bw_cfg.get("project_id", "") if not project_id: console.print("[red]No project_id configured.[/red]") return 1 - server_url = cfg_str(bw_cfg, "server_url") - try: secrets, warnings = bw.fetch_bitwarden_secrets( - access_token=token, - project_id=project_id, - use_cache=False, - server_url=server_url, + access_token=token, project_id=project_id, use_cache=False, server_url=cfg_str(bw_cfg, "server_url"), ) except Exception as exc: # noqa: BLE001 console.print(f"[red]Fetch failed: {exc}[/red]") @@ -448,28 +377,25 @@ def cmd_sync(args: argparse.Namespace) -> int: rows = [] applied = 0 for key in sorted(secrets): - if key == token_env: - rows.append((key, "[dim]skip (bootstrap token)[/dim]")) - continue already = bool(os.environ.get(key)) - if already and not override: - rows.append((key, "[dim]skip (already set)[/dim]")) - continue - if args.apply: + if key == token_env: + action = "[dim]skip (bootstrap token)[/dim]" + elif already and not override: + action = "[dim]skip (already set)[/dim]" + elif args.apply: os.environ[key] = secrets[key] applied += 1 - rows.append((key, "[green]exported[/green]" + (" (overrode)" if already else ""))) + action = "[green]exported[/green]" + (" (overrode)" if already else "") else: - rows.append((key, "[green]would export[/green]" + (" (overrides)" if already else ""))) + action = "[green]would export[/green]" + (" (overrides)" if already else "") + rows.append((key, action)) print_table(console, (("Name", {"style": "cyan"}), "Action"), rows, warnings) if not args.apply: - console.print( - "\n This was a dry-run — secrets are picked up automatically on the " - "next [cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] " - "to export into the current shell instead." - ) + console.print("\n This was a dry-run — secrets are picked up automatically on the " + "next [cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] " + "to export into the current shell instead.") else: console.print(f"\n [green]Exported {applied} secret(s) into current process.[/green]") return 0 @@ -481,8 +407,7 @@ def cmd_disable(args: argparse.Namespace) -> int: "[green]Disabled.[/green] Bitwarden secrets will NOT be pulled on the next " "Hermes invocation.\n" " Your access token is left in .env — remove it manually if you also want " - "to revoke the credential.", - ) + "to revoke the credential.") def cmd_install(args: argparse.Namespace) -> int: @@ -503,27 +428,19 @@ def cmd_install(args: argparse.Namespace) -> int: def _token_validation_status( - *, - enabled: bool, - binary: Optional[Path], - token: str, - server_url: str = "", + *, enabled: bool, binary: Optional[Path], token: str, server_url: str = "", ) -> tuple[str, list[str]]: - if not enabled: - return "[dim]not checked[/dim] (integration disabled)", [] - if not token: - return "[dim]not checked[/dim] (token missing)", [] - if binary is None: - return "[dim]not checked[/dim] (bws not installed)", [] + for skipped, reason in ((not enabled, "integration disabled"), (not token, "token missing"), + (binary is None, "bws not installed")): + if skipped: + return f"[dim]not checked[/dim] ({reason})", [] messages: list[str] = [] if not token.startswith("0."): messages.append(_NOT_BSM_TOKEN_WARNING_CONTINUING) - capture = io.StringIO() - probe_console = Console(file=capture, record=True, width=200) - projects = _list_projects(binary, token, probe_console, server_url=server_url) - if projects is None: + probe_console = Console(file=io.StringIO(), record=True, width=200) + if _list_projects(binary, token, probe_console, server_url=server_url) is None: details = probe_console.export_text(styles=False).strip() if details: messages.extend(line.rstrip() for line in details.splitlines()) @@ -531,6 +448,20 @@ def _token_validation_status( return "[green]passed[/green]", messages +# (substring of the lowercased bws error, follow-up hint) — first match wins. +_PROJECT_LIST_HINTS = ( + (("invalid_client", "400 bad request"), + " [yellow]'invalid_client' from the US identity endpoint usually " + "means the token is for a different Bitwarden region. Re-run " + "[cyan]hermes secrets bitwarden setup[/cyan] and pick EU or " + "self-hosted at the region prompt, or set [cyan]secrets.bitwarden." + "server_url[/cyan] in config.yaml.[/yellow]"), + (("authorization", "invalid"), + " [yellow]This usually means the access token is wrong or revoked. " + "Double-check it in the Bitwarden web app.[/yellow]"), +) + + def _list_projects( binary: Path, token: str, console: Console, *, server_url: str = "" ) -> Optional[List[dict]]: @@ -542,11 +473,7 @@ def _list_projects( try: res = subprocess.run( [str(binary), "project", "list", "--output", "json"], - env=env, - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=15, - ) + env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=15) except (OSError, subprocess.TimeoutExpired) as exc: console.print(f" [red]Couldn't list projects: {exc}[/red]") return None @@ -555,19 +482,10 @@ def _list_projects( err = (res.stderr or res.stdout).strip()[:300] console.print(f" [red]bws project list failed: {err}[/red]") lowered = err.lower() - if "invalid_client" in lowered or "400 bad request" in lowered: - console.print( - " [yellow]'invalid_client' from the US identity endpoint usually " - "means the token is for a different Bitwarden region. Re-run " - "[cyan]hermes secrets bitwarden setup[/cyan] and pick EU or " - "self-hosted at the region prompt, or set [cyan]secrets.bitwarden." - "server_url[/cyan] in config.yaml.[/yellow]" - ) - elif "authorization" in lowered or "invalid" in lowered: - console.print( - " [yellow]This usually means the access token is wrong or revoked. " - "Double-check it in the Bitwarden web app.[/yellow]" - ) + for needles, hint in _PROJECT_LIST_HINTS: + if any(n in lowered for n in needles): + console.print(hint) + break return None try: @@ -580,9 +498,7 @@ def _list_projects( return [p for p in data if isinstance(p, dict) and p.get("id")] -# Canonical Bitwarden region endpoints. Keep in sync with what Bitwarden -# publishes — these are stable but if a third region appears, add it here -# and to the prompt below. +# Canonical Bitwarden region endpoints; add a new region here and it appears in the prompt. _REGION_PRESETS = [ ("US Cloud (https://vault.bitwarden.com — bws default)", ""), ("EU Cloud (https://vault.bitwarden.eu)", "https://vault.bitwarden.eu"), @@ -590,66 +506,44 @@ _REGION_PRESETS = [ def _resolve_server_url( - args: argparse.Namespace, - secrets_cfg: dict, - console: Console, + args: argparse.Namespace, secrets_cfg: dict, console: Console, ) -> Optional[str]: - """Pick a Bitwarden server URL for setup. - - Resolution order: 1. ``--server-url`` CLI flag (non-interactive) 2. ``BWS_SERVER_URL`` env var - (so users running with that already set in their shell don't have to re-enter it) 3. Existing - ``secrets.bitwarden.server_url`` value (for re-runs) 4. Interactive menu: US / EU / self-hosted - """ + """Pick a Bitwarden server URL: ``--server-url``, then ``BWS_SERVER_URL``, then the existing + ``secrets.bitwarden.server_url``, then the interactive US / EU / self-hosted menu. None (after + printing) when a custom URL is left empty.""" if args.server_url and args.server_url.strip(): return args.server_url.strip() env_url = os.environ.get("BWS_SERVER_URL", "").strip() if env_url: - console.print( - f" Detected [cyan]BWS_SERVER_URL[/cyan]={env_url} in your shell — using it." - ) + console.print(f" Detected [cyan]BWS_SERVER_URL[/cyan]={env_url} in your shell — using it.") return env_url existing = cfg_str(secrets_cfg, "server_url") if existing: - console.print( - f" Existing config: [cyan]{existing}[/cyan]. " - "Press Enter to keep, or pick a different option below." - ) + console.print(f" Existing config: [cyan]{existing}[/cyan]. " + "Press Enter to keep, or pick a different option below.") table = Table(show_header=True, header_style="bold", box=None, padding=(0, 2)) table.add_column("#", style="cyan", width=4) table.add_column("Region / endpoint") for i, (label, _url) in enumerate(_REGION_PRESETS, 1): table.add_row(str(i), label) - table.add_row(str(len(_REGION_PRESETS) + 1), "Self-hosted / custom URL") + custom_idx = len(_REGION_PRESETS) + 1 + table.add_row(str(custom_idx), "Self-hosted / custom URL") console.print(table) - custom_idx = len(_REGION_PRESETS) + 1 - prompt = f" Select region [1-{custom_idx}]" - if existing: - prompt += " (Enter to keep current)" - idx = prompt_index( - console, - prompt + ": ", - custom_idx, - allow_empty=bool(existing), - empty_message=" [red]Enter a number.[/red]", - ) + prompt = f" Select region [1-{custom_idx}]" + (" (Enter to keep current)" if existing else "") + idx = prompt_index(console, prompt + ": ", custom_idx, allow_empty=bool(existing), + empty_message=" [red]Enter a number.[/red]") if idx == 0: return existing if idx <= len(_REGION_PRESETS): return _REGION_PRESETS[idx - 1][1] - custom = console.input( - " Enter your Bitwarden server URL " - "(e.g. https://vault.example.com): " - ).strip() + custom = console.input(" Enter your Bitwarden server URL (e.g. https://vault.example.com): ").strip() if not custom: console.print(" [red]Empty URL, aborting.[/red]") return None if not custom.startswith(("http://", "https://")): - console.print( - " [yellow]Warning: URL doesn't start with http:// or " - "https:// — bws may reject it.[/yellow]" - ) + console.print(" [yellow]Warning: URL doesn't start with http:// or https:// — bws may reject it.[/yellow]") return custom diff --git a/hermes_cli/send_cmd.py b/hermes_cli/send_cmd.py index 1d2a7e3fad..fbad9cc143 100644 --- a/hermes_cli/send_cmd.py +++ b/hermes_cli/send_cmd.py @@ -24,16 +24,9 @@ def _fail(msg: str, exit_code: int | None = None) -> int: return _FAILURE_EXIT -def _read_message_body( - positional: Optional[str], - file_path: Optional[str], -) -> Optional[str]: - """Resolve the message body from the positional arg, ``--file``, or piped stdin. - - Order: explicit positional argument, then ``--file PATH`` / ``--file -`` (stdin), then piped - stdin when not attached to a TTY. Returns ``None`` when nothing is available — callers must - treat that as a usage error. - """ +def _read_message_body(positional: Optional[str], file_path: Optional[str]) -> Optional[str]: + """Resolve the message body: positional arg, then ``--file PATH`` / ``--file -`` (stdin), then + piped stdin when not attached to a TTY. ``None`` when nothing is available (a usage error).""" if positional: return positional @@ -52,33 +45,21 @@ def _read_message_body( f' hermes send --to telegram "optional caption MEDIA:{file_path}"\n' "Add [[as_document]] to deliver an image as an uncompressed file:\n" f' hermes send --to telegram "[[as_document]] MEDIA:{file_path}"', - _USAGE_EXIT, - ) + _USAGE_EXIT) except OSError as exc: _fail(f"hermes send: cannot read {file_path}: {exc}", _USAGE_EXIT) - # Piped input: only consume stdin when it is not a TTY. Reading from a - # TTY would block the user in a half-broken "type your message" state, - # which is a poor default for an ops CLI. + # Reading from a TTY would block the user in a half-broken "type your message" state. return (sys.stdin.read() or None) if not sys.stdin.isatty() else None -def _emit_result( - result_json: str, - *, - json_mode: bool, - quiet: bool, -) -> int: - """Print the tool result in the requested format and return the exit code. - - The underlying ``send_message_tool`` always returns a JSON string. We parse it, decide - success/failure, and format accordingly. - """ +def _emit_result(result_json: str, *, json_mode: bool, quiet: bool) -> int: + """Print the ``send_message_tool`` JSON result in the requested format; return the exit code. + Unknown / unexpected shapes are failures so scripts notice.""" try: payload = json.loads(result_json) if result_json else {} except json.JSONDecodeError: - # Shouldn't happen with the shared tool, but be defensive — pass the - # raw string through so the user can still see what went wrong. + # Pass the raw string through so the user can still see what went wrong. payload = {"error": "invalid JSON from send_message_tool", "raw": result_json} if json_mode: @@ -89,21 +70,16 @@ def _emit_result( elif payload.get("success"): print(payload.get("note") or "sent") else: - # Unknown shape — dump it so nothing is silently dropped. - print(json.dumps(payload, indent=2)) + print(json.dumps(payload, indent=2)) # unknown shape — dump it, drop nothing - # Unknown / unexpected shapes are failures so scripts notice. if not payload.get("error") and (payload.get("skipped") or payload.get("success")): return _SUCCESS_EXIT return _FAILURE_EXIT def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int: - """Print the channel directory (all configured targets across platforms). - - Uses ``load_directory()`` for JSON and ``format_directory_for_display()`` for the human - rendering the send_message tool shows the model, keeping the two surfaces identical. - """ + """Print the channel directory (all configured targets across platforms), reusing the + ``format_directory_for_display`` rendering the send_message tool shows the model.""" try: from gateway.channel_directory import format_directory_for_display, load_directory except Exception as exc: @@ -116,24 +92,16 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int: platforms = dict(raw.get("platforms") or {}) - # Merge in configured-but-undiscovered platforms so `--list` never hides - # a working send target. The directory only contains platforms the - # gateway has discovered channels for; a platform configured via env / - # config.yaml that has never run channel discovery (e.g. a fresh SimpleX - # setup used only for outbound `hermes send`) would otherwise be - # invisible, leaving users guessing at platform names. + # Merge in configured-but-undiscovered platforms (e.g. a fresh SimpleX setup used only for + # outbound sends) so `--list` never hides a working send target. try: from gateway.config import load_gateway_config - - gw_config = load_gateway_config() - for plat in gw_config.get_connected_platforms(): + for plat in load_gateway_config().get_connected_platforms(): plat_name = getattr(plat, "value", str(plat)) if plat_name not in ("local", "api_server", "webhook"): platforms.setdefault(plat_name, []) except Exception: - # Directory contents alone are still useful; don't fail --list over - # a config parse problem. - pass + pass # directory contents alone are still useful; don't fail --list on a config problem if platform_filter: key = platform_filter.strip().lower() @@ -141,8 +109,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int: if not filtered: return _fail( f"hermes send: no targets found for platform '{platform_filter}'. " - f"Configured: {', '.join(sorted(platforms)) or '(none)'}" - ) + f"Configured: {', '.join(sorted(platforms)) or '(none)'}") platforms = filtered if json_mode: @@ -155,9 +122,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int: print("channel discovery can populate ~/.hermes/channel_directory.json.") return _SUCCESS_EXIT - # Human display — when unfiltered, reuse the shared formatter the agent - # already sees (passing the merged view so configured-but-undiscovered - # platforms are listed too). When filtered, build a minimal view ourselves. + # Unfiltered: the shared formatter over the merged view. Filtered: a minimal view of our own. if platform_filter is None: print(format_directory_for_display(platforms)) return _SUCCESS_EXIT @@ -180,10 +145,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int: def _load_hermes_env() -> None: """Populate ``os.environ`` from ``~/.hermes/.env`` AND bridge top-level ``config.yaml`` keys into - the environment so the underlying gateway config loader sees platform credentials and home - channel IDs. - """ - # Step 1: dotenv + the environment so the gateway config loader sees platform credentials and home channels.""" try: from dotenv import load_dotenv except Exception: @@ -198,18 +160,14 @@ def _load_hermes_env() -> None: env_path = home / ".env" if load_dotenv and env_path.exists(): try: - # utf-8-sig strips a leading UTF-8 BOM if present (PowerShell 5.1 - # Set-Content -Encoding UTF8 / Notepad) and is a no-op for - # BOM-less UTF-8. Plain "utf-8" would keep U+FEFF on the first - # key name and silently drop it from os.environ under its - # canonical name. + # utf-8-sig strips a leading BOM (PowerShell 5.1 / Notepad); plain "utf-8" would keep + # U+FEFF on the first key name and silently drop it from os.environ. load_dotenv(str(env_path), override=True, encoding="utf-8-sig") except UnicodeDecodeError: try: # utf-8-sig can't strip a BOM once we fall back to latin-1. import codecs import io - raw = env_path.read_bytes() if raw.startswith(codecs.BOM_UTF8): raw = raw[len(codecs.BOM_UTF8) :] @@ -219,17 +177,14 @@ def _load_hermes_env() -> None: except Exception: pass - # Step 2: bridge top-level config.yaml values into the environment so - # gateway.config.load_gateway_config() sees them. Scalars only; don't - # override values already in the env. + # Bridge top-level config.yaml scalars into the environment (never overriding existing values). import os config_path = home / "config.yaml" if not config_path.exists(): return try: - # Presence-sensitive env bridge: raw read is deliberate — only keys - # the user actually wrote get bridged. Overlay + expansion below. + # Raw read is deliberate — only keys the user actually wrote get bridged. from hermes_cli.config import read_user_config_raw raw = read_user_config_raw(config_path) except Exception: @@ -241,8 +196,7 @@ def _load_hermes_env() -> None: except Exception: pass - # Managed scope: overlay administrator-pinned values before bridging to env, - # so a managed top-level scalar wins here too. Fail-open via the helper. + # Managed scope: administrator-pinned values win here too (fail-open via the helper). try: from hermes_cli import managed_scope raw = managed_scope.apply_managed_overlay(raw if isinstance(raw, dict) else {}) @@ -259,18 +213,11 @@ def _load_hermes_env() -> None: def cmd_send(args: argparse.Namespace) -> None: """Entry point wired into the top-level argparse dispatcher.""" + _load_hermes_env() # the downstream gateway config loader reads credentials from os.environ - # Bridge ~/.hermes/.env and ~/.hermes/config.yaml into os.environ so the - # gateway config loader (invoked downstream by send_message_tool and by - # the channel directory) can see platform credentials and home channels. - _load_hermes_env() - - # --list short-circuits everything else. - if getattr(args, "list_targets", False): - # When `--list telegram` is used, argparse stores "telegram" in the - # `message` positional (since list_targets takes no argument). - platform_filter = getattr(args, "message", None) - exit_code = _list_targets(platform_filter, json_mode=getattr(args, "json", False)) + if getattr(args, "list_targets", False): # --list short-circuits everything else + # `hermes send --list telegram` lands "telegram" in the `message` positional. + exit_code = _list_targets(getattr(args, "message", None), json_mode=getattr(args, "json", False)) sys.exit(exit_code) target = (getattr(args, "to", None) or "").strip() @@ -281,32 +228,25 @@ def cmd_send(args: argparse.Namespace) -> None: " hermes send --to telegram \"hello\"\n" " hermes send --to discord:#ops --file report.md\n" " hermes send --list # list available targets", - _USAGE_EXIT, - ) + _USAGE_EXIT) message = _read_message_body(getattr(args, "message", None), getattr(args, "file", None)) if message is None or not message.strip(): _fail( "hermes send: no message provided. Pass text as a positional " "argument, use --file PATH, or pipe data via stdin.", - _USAGE_EXIT, - ) + _USAGE_EXIT) - # Optional: prepend a subject line. Useful for alerting scripts that - # want a consistent header without inlining it into every call. + # Optional subject line: a consistent header for alerting scripts. subject = getattr(args, "subject", None) if subject: message = f"{subject}\n\n{message.lstrip()}" - # Import lazily so `hermes send --help` stays fast and does not pull in - # the full tool registry / gateway config stack. + # Lazy import keeps `hermes send --help` fast (no tool registry / gateway config stack). from tools.send_message_tool import send_message_tool - # send_message_tool auto-loads gateway config + env and routes to the - # appropriate platform adapter (bot-token path for Telegram/Discord/Slack/ - # Signal/SMS/WhatsApp; live-adapter path for plugin platforms). - # - # It expects the standard tool-call dict and returns a JSON string. + # Routes to the platform adapter (bot-token path for built-ins, live-adapter path for plugin + # platforms); takes the standard tool-call dict and returns a JSON string. result = send_message_tool({"action": "send", "target": target, "message": message}) sys.exit(_emit_result(result, json_mode=getattr(args, "json", False), quiet=getattr(args, "quiet", False))) @@ -320,19 +260,16 @@ _SEND_ARGUMENTS = ( "Delivery target. Format: 'platform' (home channel), " "'platform:chat_id', 'platform:chat_id:thread_id', or " "'platform:#channel-name'. Examples: telegram, " - "telegram:-1001234567890:17585, discord:#ops, slack:C0123ABCD, " - "signal:+15551234567." + "telegram:-1001234567890:17585, discord:#ops, slack:C0123ABCD, signal:+15551234567." ), )), (("message",), dict(nargs="?", default=None, help="Message text. If omitted, read from --file or stdin.")), - # Legacy / convenience positional removed — use --to for clarity. (("-f", "--file"), dict( metavar="PATH", default=None, help=( "Read message body from PATH (text only). Use '-' to force stdin. " - "To send an image/document as an attachment, use MEDIA: in " - "the message text instead." + "To send an image/document as an attachment, use MEDIA: in the message text instead." ), )), (("-s", "--subject"), dict(metavar="LINE", default=None, help="Prepend a subject/header line before the message body.")), @@ -369,8 +306,7 @@ def register_send_subparser(subparsers) -> argparse.ArgumentParser: "\n" "Exit codes: 0 ok, 1 delivery/backend error, 2 usage error." ), - formatter_class=argparse.RawDescriptionHelpFormatter, - ) + formatter_class=argparse.RawDescriptionHelpFormatter) for flags, kwargs in _SEND_ARGUMENTS: parser.add_argument(*flags, **kwargs) diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index b236149c88..292250a2c8 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -1,10 +1,9 @@ """Interactive setup wizard for Hermes Agent (config lives in ~/.hermes/). -Independently-runnable sections: Model & Provider, Terminal Backend, Agent Settings, -Messaging Platforms, Tools (TTS, web search, image generation, ...). Section bodies live in -sibling modules (setup_tts, setup_terminal, setup_platforms, setup_summary, setup_migration, -setup_quick) and are re-exported here; they resolve shared prompt/config helpers lazily through -this module so test patches on ``hermes_cli.setup.`` keep working. +Independently-runnable sections: Model & Provider, Terminal Backend, Agent Settings, Messaging +Platforms, Tools. Section bodies live in sibling setup_* modules and are re-exported here; they +resolve shared prompt/config helpers lazily through this module so test patches on +``hermes_cli.setup.`` keep working. """ import importlib.util @@ -21,39 +20,46 @@ from typing import Callable from hermes_cli.curses_ui import MenuNavigationEvent, MenuNavigationStart from hermes_cli.nous_subscription import get_nous_subscription_features # noqa: F401 (re-export; patched by tests) from tools.tool_backend_helpers import managed_nous_tools_enabled # noqa: F401 (re-export; patched by tests) +# Config helpers are re-exported (tests patch them on this module). display_hermes_home is +# imported lazily at call sites (stale-module safety during hermes update). +from hermes_cli.config import ( + cfg_get, DEFAULT_CONFIG, get_hermes_home, get_config_path, get_env_path, load_config, save_config, + save_env_value, remove_env_value, get_env_value, ensure_hermes_home, +) +from hermes_cli.colors import Colors, color +from hermes_cli.cli_output import print_error, print_info, print_success, print_warning +from hermes_cli.secret_prompt import masked_secret_prompt logger = logging.getLogger(__name__) PROJECT_ROOT = Path(__file__).parent.parent.resolve() _DOCS_BASE = "https://hermes-agent.nousresearch.com/docs" +_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~") -# Config helpers (re-exported; tests patch them on this module). display_hermes_home is -# imported lazily at call sites (stale-module safety during hermes update). -from hermes_cli.config import ( # noqa: E402 - cfg_get, DEFAULT_CONFIG, get_hermes_home, get_config_path, get_env_path, load_config, save_config, - save_env_value, remove_env_value, get_env_value, ensure_hermes_home, -) -from hermes_cli.colors import Colors, color # noqa: E402 - - -def print_header(title: str): - """Print a section header.""" +def print_header(title: str, *, gap: bool = False): + """Print a section header (``gap`` adds an extra blank line before it).""" + if gap: + print() print() print(color(f"◆ {title}", Colors.CYAN, Colors.BOLD)) -from hermes_cli.cli_output import print_error, print_info, print_success, print_warning # noqa: E402 -from hermes_cli.secret_prompt import masked_secret_prompt # noqa: E402 - - def _info(*lines: str | None) -> None: """print_info each line in order; ``None`` emits a bare blank ``print()``.""" for line in lines: print() if line is None else print_info(line) +def _sub_dict(parent: dict, key: str) -> dict: + """``parent[key]`` as a dict, replacing a missing or non-dict value with ``{}``.""" + child = parent.get(key) + if not isinstance(child, dict): + child = parent[key] = {} + return child + + def _current_reasoning_effort(config: dict) -> str: agent_cfg = config.get("agent") if isinstance(agent_cfg, dict): @@ -62,20 +68,13 @@ def _current_reasoning_effort(config: dict) -> str: def _set_reasoning_effort(config: dict, effort: str) -> None: - agent_cfg = config.get("agent") - if not isinstance(agent_cfg, dict): - agent_cfg = {} - config["agent"] = agent_cfg - agent_cfg["reasoning_effort"] = effort + _sub_dict(config, "agent")["reasoning_effort"] = effort def is_interactive_stdin() -> bool: """Return True when stdin looks like a usable interactive TTY.""" - stdin = getattr(sys, "stdin", None) - if stdin is None: - return False try: - return bool(stdin.isatty()) + return bool(sys.stdin.isatty()) except Exception: return False @@ -96,48 +95,33 @@ def print_noninteractive_setup_guidance(reason: str | None = None) -> None: "Run 'hermes setup' in an interactive terminal to use the full wizard.", None) -_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~") - - def _sanitize_pasted_input(value: str) -> str: """Strip terminal bracketed-paste control markers from pasted text.""" - if not isinstance(value, str) or not value: - return value - return _BRACKETED_PASTE_PATTERN.sub("", value) + return _BRACKETED_PASTE_PATTERN.sub("", value) if isinstance(value, str) and value else value def prompt(question: str, default: str = None, password: bool = False) -> str: """Prompt for input with optional default.""" - display = f"{question} [{default}]: " if default else f"{question}: " - + display = color(f"{question} [{default}]: " if default else f"{question}: ", Colors.YELLOW) try: if password: - value = masked_secret_prompt(color(display, Colors.YELLOW)) + value = masked_secret_prompt(display) else: from hermes_cli.cli_output import line_input - - value = line_input(color(display, Colors.YELLOW)) - - cleaned = _sanitize_pasted_input(value) - return cleaned.strip() or default or "" + value = line_input(display) + return _sanitize_pasted_input(value).strip() or default or "" except (KeyboardInterrupt, EOFError): print() sys.exit(1) -# ============================================================================= -# Setup navigation (Escape cancels, Left arrow goes back) — a ContextVar state -# machine shared with the curses menus. -# ============================================================================= +# ── Setup navigation (Escape cancels, Left arrow goes back): a ContextVar state machine shared +# with the curses menus. ── class _SetupControlFlow(BaseException): - """Bypass provider error handlers that intentionally catch ``Exception``. - - Provider setup has broad compatibility boundaries around network, plugin and credential - integrations; navigation must cross them unchanged so the outer state machine can replay - the prior prompt. - """ + """Bypass provider error handlers that intentionally catch ``Exception`` so navigation reaches + the outer state machine unchanged and it can replay the prior prompt.""" class _SetupCancelled(_SetupControlFlow): @@ -156,29 +140,22 @@ class _SetupNavigationState: """Per-invocation navigation state for the synchronous setup wizard.""" def __init__(self, *, section_index: int = -1, prompt_index: int = 0): - self.section_index = section_index + self.reset(section_index) self.prompt_index = prompt_index - self.active_prompt_index = -1 - self.resolved_choices: list[object] = [] - self.replay_choices: list[object] = [] def reset(self, section_index: int = -1, replay: list | None = None) -> None: """Rewind per-section counters (entering a section, or leaving the wizard).""" self.section_index = section_index self.prompt_index = 0 self.active_prompt_index = -1 - self.resolved_choices = [] - self.replay_choices = copy.deepcopy(replay or []) + self.resolved_choices: list[object] = [] + self.replay_choices: list[object] = copy.deepcopy(replay or []) -_SETUP_NAVIGATION: ContextVar[_SetupNavigationState | None] = ContextVar( - "hermes_setup_navigation", default=None -) +_SETUP_NAVIGATION: ContextVar[_SetupNavigationState | None] = ContextVar("hermes_setup_navigation", default=None) -def _handle_setup_menu_navigation( - event: MenuNavigationEvent, value: object = None -) -> MenuNavigationStart | None: +def _handle_setup_menu_navigation(event: MenuNavigationEvent, value: object = None) -> MenuNavigationStart | None: """Translate shared curses menu events into setup control flow.""" state = _SETUP_NAVIGATION.get() if state is None: @@ -187,25 +164,16 @@ def _handle_setup_menu_navigation( if state.section_index < 0: state.active_prompt_index = -1 return MenuNavigationStart() - state.active_prompt_index = state.prompt_index + idx = state.active_prompt_index = state.prompt_index state.prompt_index += 1 - allow_back = state.section_index > 0 or state.active_prompt_index > 0 - if state.active_prompt_index < len(state.replay_choices): - return MenuNavigationStart( - allow_back=allow_back, - replay_value=copy.deepcopy(state.replay_choices[state.active_prompt_index]), - ) + allow_back = state.section_index > 0 or idx > 0 + if idx < len(state.replay_choices): + return MenuNavigationStart(allow_back=allow_back, replay_value=copy.deepcopy(state.replay_choices[idx])) return MenuNavigationStart(allow_back=allow_back) if event is MenuNavigationEvent.RESOLVE: prompt_index = state.active_prompt_index - if prompt_index < 0: - return None - resolved = copy.deepcopy(value) - if prompt_index < len(state.resolved_choices): - state.resolved_choices[prompt_index] = resolved - del state.resolved_choices[prompt_index + 1 :] - else: - state.resolved_choices.append(resolved) + if prompt_index >= 0: # replace this answer and drop every later one + state.resolved_choices[prompt_index:] = [copy.deepcopy(value)] return None if event is MenuNavigationEvent.CANCEL: raise _SetupCancelled() @@ -228,15 +196,18 @@ def _setup_navigation_scope(): def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: - """Run setup sections with left-arrow navigation between choices. - - Left arrow at a section's first choice returns to the previous section; from a later, nested - choice it replays earlier selections invisibly and reopens only the preceding prompt. - """ + """Run setup sections with left-arrow navigation: at a section's first choice it returns to + the previous section; from a later choice it replays earlier selections invisibly and reopens + only the preceding prompt.""" state = _SETUP_NAVIGATION.get() section_index = 0 answers_by_section: dict[int, list[object]] = {} replay_by_section: dict[int, list[object]] = {} + + def _record_answers() -> None: + if state is not None: + answers_by_section[section_index] = copy.deepcopy(state.resolved_choices) + try: while section_index < len(steps): label, action = steps[section_index] @@ -245,8 +216,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: try: action() except _SetupGoBack as navigation: - if state is not None: - answers_by_section[section_index] = copy.deepcopy(state.resolved_choices) + _record_answers() if navigation.prompt_index > 0: previous_index = section_index target_prompt = navigation.prompt_index - 1 @@ -254,8 +224,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: previous_index = max(0, section_index - 1) target_prompt = max(0, len(answers_by_section.get(previous_index, [])) - 1) replay_by_section[previous_index] = copy.deepcopy( - answers_by_section.get(previous_index, [])[:target_prompt] - ) + answers_by_section.get(previous_index, [])[:target_prompt]) print() if previous_index == section_index: print_info(f"Returning to the previous choice in {label}...") @@ -263,8 +232,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: print_info(f"Returning to {steps[previous_index][0]}...") section_index = previous_index continue - if state is not None: - answers_by_section[section_index] = copy.deepcopy(state.resolved_choices) + _record_answers() section_index += 1 finally: if state is not None: @@ -274,17 +242,13 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: def run_setup_action_with_navigation( label: str, action: Callable[[], None], *, cancelled_message: str = "Setup cancelled." ) -> None: - """Run a setup-style menu flow with Escape and nested Left navigation. - - Shared commands such as ``hermes model`` use the wizard's pickers outside ``run_setup_wizard``; - this installs the navigation context for them and reuses the prompt replay state machine. - """ + """Run a setup-style menu flow with Escape and nested Left navigation — for commands such as + ``hermes model`` that use the wizard's pickers outside ``run_setup_wizard``.""" with _setup_navigation_scope(): try: _run_setup_steps([(label, action)]) except _SetupCancelled: - print() - print_info(cancelled_message) + _info(None, cancelled_message) # ── Prompt primitives ── @@ -297,51 +261,37 @@ def _curses_prompt_choice(question: str, choices: list, default: int = 0, descri def prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int: - """Prompt for a choice from a list with arrow key navigation. - - Escape cancels an active setup wizard; outside setup it keeps the default. The curses - component owns its own numbered fallback, so a cancel result must never be mistaken for a - request to open another prompt. Ctrl+C exits the wizard. - """ + """Prompt for a choice from a list with arrow key navigation. Escape cancels an active setup + wizard; outside setup it keeps the default (the curses component owns its own numbered + fallback, so a cancel result must never open another prompt). Ctrl+C exits the wizard.""" idx = _curses_prompt_choice(question, choices, default, description=description) - if idx >= 0: - if idx == default: - _info(" Skipped (keeping current)", None) - return default - print() - return idx - - return default + if idx < 0: + return default + if idx == default: + _info(" Skipped (keeping current)", None) + return default + print() + return idx def is_noninteractive() -> bool: - """True when no human is available to answer a prompt. - - The dashboard/desktop spawn CLI actions with ``stdin=DEVNULL`` and ``HERMES_NONINTERACTIVE=1`` - (see ``hermes_cli/web_server.py``); there ``input()`` raises ``EOFError`` immediately, and a - prompt that aborts on EOF kills the spawned action (desktop "restart gateway" failed this way - when the Windows service was not installed yet). Honour the flag so callers fall back to their - default. - """ + """True when no human is available to answer a prompt: the dashboard/desktop spawn CLI actions + with ``stdin=DEVNULL`` and ``HERMES_NONINTERACTIVE=1`` (``hermes_cli/web_server.py``), where a + prompt that aborts on EOF would kill the spawned action — callers fall back to their default.""" return os.environ.get("HERMES_NONINTERACTIVE", "").strip().lower() in {"1", "true", "yes", "on"} def prompt_yes_no(question: str, default: bool = True) -> bool: - """Prompt for yes/no. Ctrl+C exits, empty input returns default. - - Non-interactive callers (``HERMES_NONINTERACTIVE=1`` or a closed/redirected stdin) have no - one to answer, so fall back to ``default`` instead of aborting the whole process. - """ + """Prompt for yes/no. Ctrl+C exits; empty input, ``HERMES_NONINTERACTIVE=1`` or a + closed/redirected stdin return ``default`` instead of aborting the whole process.""" if is_noninteractive(): return default - # Inside setup, route binary selections through the curses menu so ESC and left-arrow work # consistently; every other caller keeps the traditional line prompt. if _SETUP_NAVIGATION.get() is not None: return _curses_prompt_choice(question, ["Yes", "No"], 0 if default else 1) == 0 default_str = "Y/n" if default else "y/N" - while True: try: value = input(color(f"{question} [{default_str}]: ", Colors.YELLOW)).strip().lower() @@ -353,29 +303,18 @@ def prompt_yes_no(question: str, default: bool = True) -> bool: # proceeds unattended instead of failing the whole command. print() return default - - if not value: - return default - if value in {"y", "yes"}: - return True - if value in {"n", "no"}: - return False + answer = {"": default, "y": True, "yes": True, "n": False, "no": False}.get(value) + if answer is not None: + return answer print_error("Please enter 'y' or 'n'") def prompt_checklist(title: str, items: list, pre_selected: list = None) -> list: - """Multi-select checklist; returns the sorted indices of selected items. - - ``pre_selected`` indices start checked; Space toggles, Enter on the appended "Continue →" - confirms, cancel keeps the pre-selection. Numbered fallback when curses is unavailable. - """ - if pre_selected is None: - pre_selected = [] - + """Multi-select checklist; returns the sorted indices of selected items. ``pre_selected`` + start checked; Space toggles, Enter confirms, cancel keeps the pre-selection.""" from hermes_cli.curses_ui import curses_checklist - - chosen = curses_checklist(title, items, set(pre_selected), cancel_returns=set(pre_selected)) - return sorted(chosen) + pre = set(pre_selected or []) + return sorted(curses_checklist(title, items, pre, cancel_returns=pre)) def _prompt_api_key(var: dict): @@ -393,14 +332,17 @@ def _prompt_api_key(var: dict): if var.get("url"): print_info(f" Get your key at: {var['url']}") print() + _prompt_and_save_env_var(var, " ✓ Saved", " Skipped (configure later with 'hermes setup')") + +def _prompt_and_save_env_var(var: dict, saved_msg: str, skipped_msg: str) -> None: + """Prompt for one env-var value (masked when secret); persist and confirm, or report the skip.""" value = prompt(f" {var.get('prompt', var['name'])}", password=bool(var.get("password"))) - if value: save_env_value(var["name"], value) - print_success(" ✓ Saved") + print_success(saved_msg) else: - print_warning(" Skipped (configure later with 'hermes setup')") + print_warning(skipped_msg) def _module_installed(name: str) -> bool: @@ -419,20 +361,13 @@ def _print_banner(*lines: str) -> None: print(color("└─────────────────────────────────────────────────────────┘", Colors.MAGENTA)) -# Tool categories and provider config are in tools_config.py (shared -# between `hermes tools` and `hermes setup tools`). - - # ── Section 1: Model & Provider Configuration ── def setup_model_provider(config: dict, *, quick: bool = False): - """Configure the inference provider and default model. - - Delegates to the ``hermes model`` flow (provider picker, credential prompting, model pick, - persistence) so there is one code path — any provider added there is available here. - *quick* skips credential rotation, vision and TTS (first-time quick setup). - """ + """Configure the inference provider and default model via the ``hermes model`` flow (one code + path — any provider added there is available here). *quick* is accepted for the first-time + quick setup caller; rotation, vision and TTS keep safe defaults either way.""" from hermes_cli.config import load_config, save_config print_header("Inference Provider") _info("Choose how to connect to your main chat model.", @@ -449,8 +384,8 @@ def setup_model_provider(config: dict, *, quick: bool = False): print_info("You can try again later with: hermes model") # Re-sync from disk in place: cmd_model saved via its own load/save cycle and the wizard's - # final save_config(config) must not clobber it with stale values (#4172). Rotation, vision - # and TTS keep safe defaults (configure via `hermes auth add` / `hermes setup tts`). + # final save_config(config) must not clobber it with stale values. Rotation, vision and TTS + # keep safe defaults (configure via `hermes auth add` / `hermes setup tts`). config.clear() config.update(load_config()) save_config(config) @@ -478,19 +413,23 @@ def _apply_default_agent_settings(config: dict): " Run `hermes setup agent` later to customize.") +def _prompt_number(label: str, current, cast=int): + """Prompt for a number; ``None`` when the answer does not parse.""" + try: + return cast(prompt(label, str(current))) + except ValueError: + return None + + def _prompt_int_setting(section: dict, key: str, label: str, current, accept) -> None: """Prompt for an int; store it under *key* only when it parses and *accept* holds.""" - try: - value = int(prompt(label, str(current))) - if accept(value): - section[key] = value - except ValueError: - pass + value = _prompt_number(label, current) + if value is not None and accept(value): + section[key] = value _TOOL_PROGRESS_HELP = ( - "Tool Progress Display", - "Controls how much tool activity is shown (CLI and messaging).", + "Tool Progress Display", "Controls how much tool activity is shown (CLI and messaging).", " off — Silent, just the final response", " new — Show tool name only when it changes (less noise)", " all — Show every tool call with a short preview", @@ -499,14 +438,11 @@ _TOOL_PROGRESS_HELP = ( ) _SESSION_RESET_HELP = ( "Messaging sessions (Telegram, Discord, etc.) accumulate context over time.", - "Each message adds to the conversation history, which means growing API costs.", - "", + "Each message adds to the conversation history, which means growing API costs.", "", "To manage this, sessions can automatically reset after a period of inactivity", "or at a fixed time each day. When a reset happens, the agent saves important", - "things to its persistent memory first — but the conversation context is cleared.", - "", - "You can also manually reset anytime by typing /reset in chat.", - "", + "things to its persistent memory first — but the conversation context is cleared.", "", + "You can also manually reset anytime by typing /reset in chat.", "", ) _SESSION_RESET_CHOICES = [ "Inactivity + daily reset (reset whichever comes first)", @@ -530,21 +466,18 @@ def setup_agent_settings(config: dict): "Higher = more complex tasks, but costs more tokens.", f"Press Enter to keep {current_max}. Use 90 for most tasks or 150+ for open exploration.") - try: - max_iter = int(prompt("Max iterations", current_max)) - if max_iter > 0: - # config.yaml only; gateway/run.py derives HERMES_MAX_ITERATIONS from agent.max_turns. - config.setdefault("agent", {})["max_turns"] = max_iter - config.pop("max_turns", None) - remove_env_value("HERMES_MAX_ITERATIONS") - print_success(f"Max iterations set to {max_iter}") - except ValueError: + max_iter = _prompt_number("Max iterations", current_max) + if max_iter is None: print_warning("Invalid number, keeping current value") + elif max_iter > 0: + # config.yaml only; gateway/run.py derives HERMES_MAX_ITERATIONS from agent.max_turns. + config.setdefault("agent", {})["max_turns"] = max_iter + config.pop("max_turns", None) + remove_env_value("HERMES_MAX_ITERATIONS") + print_success(f"Max iterations set to {max_iter}") # ── Tool Progress Display ── - print_info("") - for line in _TOOL_PROGRESS_HELP: - print_info(line) + _info("", *_TOOL_PROGRESS_HELP) current_mode = cfg_get(config, "display", "tool_progress", default="all") mode = prompt("Tool progress mode", current_mode) @@ -561,34 +494,29 @@ def setup_agent_settings(config: dict): "Higher threshold = compress later (use more context). Lower = compress sooner.") config.setdefault("compression", {})["enabled"] = True - current_threshold = cfg_get(config, "compression", "threshold", default=0.50) - try: - threshold = float(prompt("Compression threshold (0.5-0.95)", str(current_threshold))) - if 0.5 <= threshold <= 0.95: - config["compression"]["threshold"] = threshold - except ValueError: - pass - + threshold = _prompt_number("Compression threshold (0.5-0.95)", current_threshold, float) + if threshold is not None and 0.5 <= threshold <= 0.95: + config["compression"]["threshold"] = threshold print_success(f"Context compression threshold set to {config['compression'].get('threshold', 0.50)}") # ── Session Reset Policy ── print_header("Session Reset Policy") - for line in _SESSION_RESET_HELP: - print_info(line) + _info(*_SESSION_RESET_HELP) + _prompt_session_reset(config.setdefault("session_reset", {})) + save_config(config) - current_policy = config.get("session_reset", {}) - current_mode = current_policy.get("mode", "none") - current_idle = current_policy.get("idle_minutes", 1440) - current_hour = current_policy.get("at_hour", 4) +def _prompt_session_reset(reset_cfg: dict) -> None: + """Pick the session reset mode and its idle/daily parameters in place.""" + current_mode = reset_cfg.get("mode", "none") + current_idle, current_hour = reset_cfg.get("idle_minutes", 1440), reset_cfg.get("at_hour", 4) default_reset = _SESSION_RESET_MODES.index(current_mode) if current_mode in _SESSION_RESET_MODES else 3 reset_idx = prompt_choice("Session reset mode:", _SESSION_RESET_CHOICES, default_reset) - - reset_cfg = config.setdefault("session_reset", {}) mode = _SESSION_RESET_MODES[reset_idx] if 0 <= reset_idx < len(_SESSION_RESET_MODES) else None - if mode is not None: - reset_cfg["mode"] = mode + if mode is None: # keep current settings + return + reset_cfg["mode"] = mode if mode in ("both", "idle"): _prompt_int_setting(reset_cfg, "idle_minutes", " Inactivity timeout (minutes)", current_idle, lambda v: v > 0) if mode in ("both", "daily"): @@ -597,15 +525,13 @@ def setup_agent_settings(config: dict): if mode == "none": print_info("Sessions will never auto-reset. Context is managed only by compression.") print_warning("Long conversations will grow in cost. Use /reset manually when needed.") - elif mode is not None: + else: print_success({ "both": f"Sessions reset after {idle_now} min idle or daily at {hour_now}:00", "idle": f"Sessions reset after {idle_now} min of inactivity", "daily": f"Sessions reset daily at {hour_now}:00", }[mode]) - save_config(config) - # ── Section 5: Tool Configuration (delegates to unified tools_config.py) ── @@ -621,16 +547,14 @@ def setup_tools(config: dict, first_install: bool = False): _SEND_CONSENT_EXPLAINER = ( - "", - "Sending uploads each daily package to the Nous telemetry", + "", "Sending uploads each daily package to the Nous telemetry", "service. Packages carry your profile-scoped install ID, a", "stable random UUID that identifies this profile across days", "(it contains no personal information and is reset by deleting", "the shared-metrics directory). Only packages whose entire", "collection period falls inside a recorded consent window are", "ever sent — data from before you opt in, or from any gap", - "while sending was off, stays on this machine. Sending can be", - "turned off again at any time.", + "while sending was off, stays on this machine. Sending can be", "turned off again at any time.", ) @@ -640,15 +564,7 @@ def setup_telemetry(config: dict): _info("Shared metrics contain only bounded counters and histograms.", "Collection is local. Sending them to Nous is a separate opt-in.") - telemetry = config.get("telemetry") - if not isinstance(telemetry, dict): - telemetry = {} - config["telemetry"] = telemetry - shared_metrics = telemetry.get("shared_metrics") - if not isinstance(shared_metrics, dict): - shared_metrics = {} - telemetry["shared_metrics"] = shared_metrics - + shared_metrics = _sub_dict(_sub_dict(config, "telemetry"), "shared_metrics") current = shared_metrics.get("enabled") is True shared_metrics["enabled"] = prompt_yes_no("Enable local shared metrics?", default=current) if not shared_metrics["enabled"]: @@ -663,8 +579,7 @@ def setup_telemetry(config: dict): return print_success("Local shared metrics enabled.") - for line in _SEND_CONSENT_EXPLAINER: - print_info(line) + _info(*_SEND_CONSENT_EXPLAINER) shared_metrics["send"] = prompt_yes_no("Send shared metrics to Nous?", default=shared_metrics.get("send") is True) _record_send_consent_change(enabled=shared_metrics["send"]) if shared_metrics["send"]: @@ -674,20 +589,14 @@ def setup_telemetry(config: dict): def _record_send_consent_change(*, enabled: bool) -> None: - """Reconcile consent windows at the moment the user decides. - - Same single writer as the relay and the sender, so wizard, relay and mid-pass callers cannot - disagree; the relay would reconcile on its next hook anyway — this makes the effect immediate. - """ + """Reconcile consent windows at the moment the user decides — same single writer as the relay + and the sender, so wizard, relay and mid-pass callers cannot disagree.""" try: from hermes_cli.observability.shared_metrics import SharedMetricsStore from hermes_cli.observability.shared_metrics_sender import reconcile_send_consent from hermes_cli.sqlite_util import write_txn - - store = SharedMetricsStore() - with store._connection() as connection: - with write_txn(connection): - reconcile_send_consent(connection, enabled) + with SharedMetricsStore()._connection() as connection, write_txn(connection): + reconcile_send_consent(connection, enabled) except Exception: # Never block the wizard on telemetry bookkeeping; the relay reconciles on the next hook. logger.debug("Unable to record shared-metrics consent change", exc_info=True) @@ -743,7 +652,7 @@ def run_setup_wizard(args): def _backup_config_file(config_path: Path) -> Path | None: - """Back up config.yaml before setup modifies it (#3522); None when absent or copy fails.""" + """Back up config.yaml before setup modifies it; None when absent or copy fails.""" if not config_path.exists(): return None from datetime import datetime as _dt @@ -790,46 +699,32 @@ def _run_full_setup(config: dict, hermes_home, *, is_existing: bool, migration_r def _skip(key: str, label: str) -> bool: return migration_ran and _skip_configured_section(config, key, label) - def _model_step() -> None: - if not _skip("model", "Model & Provider"): - setup_model_provider(config) - - def _terminal_step() -> None: - if not _skip("terminal", "Terminal Backend"): - setup_terminal_backend(config) - def _gateway_step() -> None: if not _skip("gateway", "Messaging Platforms"): setup_gateway(config) return - # A skipped (migrated) gateway section still needs its service so imported platforms # and cron jobs become active. from hermes_cli.gateway import ensure_gateway_service - ensure_gateway_service(context="setup") - def _tools_step() -> None: - if not _skip("tools", "Tools"): - setup_tools(config, first_install=not is_existing) + def _step(key: str, label: str, run) -> tuple: + return label, lambda: None if _skip(key, label) else run() - _run_setup_steps( - [ - ("Model & Provider", _model_step), - ("Terminal Backend", _terminal_step), - ("Messaging Platforms", _gateway_step), - ("Tools", _tools_step), - ] - ) + _run_setup_steps([ + _step("model", "Model & Provider", lambda: setup_model_provider(config)), + _step("terminal", "Terminal Backend", lambda: setup_terminal_backend(config)), + ("Messaging Platforms", _gateway_step), + _step("tools", "Tools", lambda: setup_tools(config, first_install=not is_existing))]) # First-time mode picker: (menu label, runner) — a None runner falls through to Full Setup. _FIRST_TIME_MODES = ( ("Quick Setup (Nous Portal) — free OAuth login, no API keys, model + tools (recommended)", - lambda config, home, is_existing: _run_first_time_quick_setup(config, home, is_existing)), + lambda *a: _run_first_time_quick_setup(*a)), ("Full setup — configure every provider, tool & option yourself (bring your own keys)", None), ("Blank Slate — everything off except the bare minimum; opt in to each capability", - lambda config, home, is_existing: _run_blank_slate_setup(config, home, is_existing)), + lambda *a: _run_blank_slate_setup(*a)), ) @@ -867,16 +762,12 @@ def _run_setup_wizard_impl(args): # Existing installation == a provider is configured from hermes_cli.auth import get_active_provider - - is_existing = bool( - get_env_value("OPENROUTER_API_KEY") or get_env_value("OPENAI_BASE_URL") or get_active_provider() is not None - ) - _print_banner( - "│ ⚕ Hermes Agent Setup Wizard │", - "├─────────────────────────────────────────────────────────┤", - "│ Let's configure your Hermes Agent installation. │", - "│ Press Ctrl+C at any time to exit. │", - ) + is_existing = bool(get_env_value("OPENROUTER_API_KEY") or get_env_value("OPENAI_BASE_URL") + or get_active_provider() is not None) + _print_banner("│ ⚕ Hermes Agent Setup Wizard │", + "├─────────────────────────────────────────────────────────┤", + "│ Let's configure your Hermes Agent installation. │", + "│ Press Ctrl+C at any time to exit. │") migration_ran = False if is_existing: @@ -886,8 +777,7 @@ def _run_setup_wizard_impl(args): if quick_requested: _run_setup_steps([("Quick Setup", lambda: _run_quick_setup(config, hermes_home))]) return - print() - print_header("Reconfigure") + print_header("Reconfigure", gap=True) print_success("You already have Hermes configured.") _info("Running the full wizard — each prompt shows your current value.", "Press Enter to keep it, or type a new value to change it.", "", diff --git a/hermes_cli/setup_hidden_env.py b/hermes_cli/setup_hidden_env.py index 885821029b..d722fd7044 100644 --- a/hermes_cli/setup_hidden_env.py +++ b/hermes_cli/setup_hidden_env.py @@ -1,25 +1,21 @@ """Which platform env vars the setup surfaces hide. -Hiding them is a *presentation* decision only. The env vars keep working through ``hermes config -set``, ``.env``, and ``config.yaml``; the gateway reads them exactly as before. This module just -says what a new user is asked during setup. +Hiding is a *presentation* decision only: the vars keep working through ``hermes config set``, +``.env`` and ``config.yaml``, and the gateway reads them exactly as before. """ -# Suffix match, so plugin adapters nobody enumerated (IRC, SimpleX, LINE, ntfy) -# get the same treatment without a code change here. +# Suffix match, so plugin adapters nobody enumerated (IRC, SimpleX, LINE, ntfy) get the same +# treatment without a code change here. # # *_HOME_CHANNEL* the bot offers /sethome on the first chat # *_ALLOW_ALL_USERS defaults off; enabling it is a security decision -# *_REPLY_TO_MODE cosmetic threading preference -# *_REPLY_MODE same, Mattermost's spelling -# *_REQUIRE_MENTION behavior toggle with a sane default -# *_AUTO_THREAD same -# *_FREE_RESPONSE_* per-channel tuning, done once the bot is in a server -# *_ALLOWED_CHANNELS same +# *_REPLY_TO_MODE / *_REPLY_MODE cosmetic threading preference (Mattermost spelling too) +# *_REQUIRE_MENTION / *_AUTO_THREAD behavior toggles with sane defaults +# *_FREE_RESPONSE_* / *_ALLOWED_CHANNELS per-channel tuning, done once the bot is in a server # *_PROXY only for networks that block the platform # -# Allowlists (*_ALLOWED_USERS) deliberately stay visible: that IS the decision -# a new user has to make, and the gateway denies everyone until it's set. +# Allowlists (*_ALLOWED_USERS) deliberately stay visible: that IS the decision a new user has to +# make, and the gateway denies everyone until it's set. SETUP_HIDDEN_ENV_SUFFIXES = ( "_HOME_CHANNEL", "_HOME_CHANNEL_NAME", @@ -38,9 +34,7 @@ SETUP_HIDDEN_ENV_SUFFIXES = ( def is_setup_hidden_env(name: str) -> bool: - """True when a var is self-configuring and shouldn't appear in setup forms. - - Callers must still keep any var a platform lists as *required* — hiding a required credential - would make that platform unconfigurable from the UI. - """ + """True when a var is self-configuring and shouldn't appear in setup forms. Callers must still + keep any var a platform lists as *required* — hiding a required credential would make that + platform unconfigurable from the UI.""" return name.endswith(SETUP_HIDDEN_ENV_SUFFIXES) diff --git a/hermes_cli/setup_migration.py b/hermes_cli/setup_migration.py index c941d5975f..3cf0870bbb 100644 --- a/hermes_cli/setup_migration.py +++ b/hermes_cli/setup_migration.py @@ -1,7 +1,5 @@ -"""Post-migration section-skip logic and the OpenClaw first-run migration flow. - -Extracted from hermes_cli/setup.py, which re-exports the names it still uses. -""" +"""Post-migration section-skip logic and the OpenClaw first-run migration flow (setup.py +re-exports the names it still uses; setup helpers are imported lazily so test patches apply).""" import importlib.util import logging @@ -33,7 +31,6 @@ def _model_section_has_credentials(config: dict) -> bool: return True except Exception: pass - try: from hermes_cli.auth import PROVIDER_REGISTRY except Exception: @@ -44,8 +41,7 @@ def _model_section_has_credentials(config: dict) -> bool: # mirrors is_provider_explicitly_configured in auth.py. return any(get_env_value(v) for v in pconfig.api_key_env_vars if v != "CLAUDE_CODE_OAUTH_TOKEN") - def _any_openrouter_key() -> bool: - return any(get_env_value(v) for v in _OPENROUTER_ENV_VARS) + any_openrouter_key = any(get_env_value(v) for v in _OPENROUTER_ENV_VARS) # Prefer the provider declared in config.yaml, avoids false positives from stray # env vars (GH_TOKEN, etc.) when the user has already picked a different provider. @@ -54,23 +50,17 @@ def _model_section_has_credentials(config: dict) -> bool: provider_id = (model_cfg.get("provider") or "").strip().lower() if provider_id in PROVIDER_REGISTRY and _has_key(PROVIDER_REGISTRY[provider_id]): return True - if provider_id == "openrouter" and _any_openrouter_key(): + if provider_id == "openrouter" and any_openrouter_key: return True # OpenRouter aggregator fallback (no provider declared in config). - if _any_openrouter_key(): + if any_openrouter_key: return True - - # Skip copilot in auto-detect: GH_TOKEN / GITHUB_TOKEN are commonly set for - # git tooling. Mirrors resolve_provider in auth.py. + # Skip copilot in auto-detect: GH_TOKEN / GITHUB_TOKEN are commonly set for git tooling. + # Mirrors resolve_provider in auth.py. return any(_has_key(pconfig) for pid, pconfig in PROVIDER_REGISTRY.items() if pid != "copilot") -def _gateway_platform_short_label(label: str) -> str: - """Strip trailing parenthetical qualifiers from a gateway platform label.""" - return label.split("(", 1)[0].strip() or label - - def _model_summary(config: dict) -> Optional[str]: if not _model_section_has_credentials(config): return None @@ -92,7 +82,8 @@ def _gateway_summary(config: dict) -> Optional[str]: # Any non-empty status other than "not configured" counts — WhatsApp ("enabled, not paired"), # Matrix ("configured + E2EE"), Signal ("partially configured") mean setup already started. configured = [ - _gateway_platform_short_label(plat["label"]) + # Trailing parenthetical qualifiers are stripped from the label. + plat["label"].split("(", 1)[0].strip() or plat["label"] for plat in _all_platforms() if _platform_status(plat) and _platform_status(plat) != "not configured" ] @@ -122,20 +113,15 @@ _SECTION_SUMMARIES = { def _get_section_config_summary(config: dict, section_key: str) -> Optional[str]: - """Return a short summary if a setup section is already configured, else None. - - Used after OpenClaw migration to detect which sections can be skipped. ``get_env_value`` is - reached through hermes_cli.setup so that test patches on ``setup_mod.get_env_value`` apply. - """ + """Short summary if a setup section is already configured (post-OpenClaw-migration skip + detection), else None. ``get_env_value`` is reached through hermes_cli.setup so test patches + on ``setup_mod.get_env_value`` apply.""" summarize = _SECTION_SUMMARIES.get(section_key) return summarize(config) if summarize else None def _skip_configured_section(config: dict, section_key: str, label: str) -> bool: - """Show an already-configured section summary and offer to skip. - - Returns True if the user chose to skip, False if the section should run. - """ + """Show an already-configured section summary and offer to skip; True when the user skips.""" from hermes_cli.setup import print_success, prompt_yes_no summary = _get_section_config_summary(config, section_key) if not summary: @@ -162,8 +148,7 @@ def _load_openclaw_migration_module(): if spec is None or spec.loader is None: return None mod = importlib.util.module_from_spec(spec) - # Register in sys.modules so @dataclass can resolve the module - # (Python 3.11+ requires this for dynamically loaded modules) + # Registered in sys.modules so @dataclass can resolve the module (Python 3.11+ requirement). sys.modules[spec.name] = mod try: spec.loader.exec_module(mod) @@ -232,8 +217,7 @@ def _print_migration_preview(report: dict): if status == "migrated": # collect warnings for high-impact items kind_lower, dest_lower = kind.lower(), str(item.get("destination", "")).lower() warnings_shown.update( - w for kw, w in _HIGH_IMPACT_KIND_KEYWORDS.items() if kw in kind_lower or kw in dest_lower - ) + w for kw, w in _HIGH_IMPACT_KIND_KEYWORDS.items() if kw in kind_lower or kw in dest_lower) print() if warnings_shown: @@ -255,6 +239,21 @@ def _run_migrator(mod, openclaw_dir: Path, hermes_home: Path, selected, *, execu ).migrate() +_FAILED = object() + + +def _migration_step(label: str, log_label: str, fn): + """Run one migration phase; on failure warn ``"