diff --git a/hermes_cli/secret_prompt.py b/hermes_cli/secret_prompt.py index 5f501d8b96..d97e1fe44c 100644 --- a/hermes_cli/secret_prompt.py +++ b/hermes_cli/secret_prompt.py @@ -10,7 +10,7 @@ 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( @@ -26,9 +26,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 +41,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 +55,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 +74,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,11 +89,7 @@ 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: @@ -111,16 +98,8 @@ def _masked_secret_prompt_posix(prompt: str, *, mask: str) -> str: 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..cf3bde7dc8 100644 --- a/hermes_cli/secrets_cli.py +++ b/hermes_cli/secrets_cli.py @@ -15,18 +15,14 @@ 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. +# The Bitwarden backend (``agent.secret_sources.bitwarden``) pulls in ``cryptography`` at import +# time; on Windows the mapped ``cryptography._rust.pyd`` makes the ``hermes update`` self-lock +# preflight defer. This module is registered parse-time from ``hermes_cli.main``, so the backend +# import stays lazy and nothing touches ``bw`` until a handler 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. +# ``_BWS_VERSION`` is duplicated here (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. _BWS_VERSION = "2.0.0" from hermes_cli._secrets_common import ( @@ -57,7 +53,7 @@ from hermes_cli.secret_prompt import masked_secret_prompt _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]" @@ -76,12 +72,8 @@ def _load_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: callers and tests monkeypatch ``hermes_cli.secrets_cli.bw`` + directly, and resolving it at import time would re-import ``cryptography`` eagerly.""" if name == "bw": return _load_bw() raise AttributeError(f"module {__name__!r} has no attribute {name!r}") @@ -125,21 +117,8 @@ 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 _setup_binary(bw, console: Console) -> Optional[Path]: + """Step 1: locate or download bws; None (after printing) on failure.""" console.print() console.print("[bold]Step 1[/bold] Install the bws CLI") try: @@ -147,25 +126,76 @@ def cmd_setup(args: argparse.Namespace) -> int: 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.""" + console.print() + console.print("[bold]Step 2[/bold] 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.""" + 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 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" @@ -178,74 +208,33 @@ def cmd_setup(args: argparse.Namespace) -> int: ) 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") 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") 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]") @@ -259,7 +248,6 @@ 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 @@ -270,15 +258,11 @@ def cmd_setup(args: argparse.Namespace) -> int: 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 +284,26 @@ 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 +311,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,22 +336,16 @@ 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( @@ -416,25 +382,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]") @@ -465,11 +425,9 @@ def cmd_sync(args: argparse.Namespace) -> int: 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 @@ -503,27 +461,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 +481,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,10 +506,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]") @@ -555,19 +516,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 +532,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"), @@ -596,60 +546,43 @@ def _resolve_server_url( ) -> 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 + Resolution order: ``--server-url`` flag, ``BWS_SERVER_URL`` env var (already set in the shell), + existing ``secrets.bitwarden.server_url`` (re-runs), 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..d7c202becd 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 @@ -57,28 +50,17 @@ def _read_message_body( 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 +71,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 +93,19 @@ 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 so `--list` never hides a working send + # target: the directory only holds platforms the gateway has discovered channels for, so a + # platform configured via env/config.yaml that never ran discovery (e.g. a fresh SimpleX + # setup used only for outbound `hermes send`) would otherwise be invisible. 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() @@ -155,9 +127,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 +150,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,11 +165,9 @@ 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 UTF-8 BOM (PowerShell 5.1 Set-Content -Encoding UTF8 / + # Notepad) and is a no-op otherwise; plain "utf-8" would keep U+FEFF on the first key + # name and silently drop it from os.environ under its canonical name. load_dotenv(str(env_path), override=True, encoding="utf-8-sig") except UnicodeDecodeError: try: @@ -219,17 +184,16 @@ 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 values already + # in the env) so gateway.config.load_gateway_config() sees them. 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. + # Presence-sensitive env bridge: raw read is deliberate — only keys the user actually + # wrote get bridged. Overlay + expansion below. from hermes_cli.config import read_user_config_raw raw = read_user_config_raw(config_path) except Exception: @@ -241,8 +205,8 @@ 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: overlay administrator-pinned values before bridging to env, so a managed + # top-level scalar wins 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 +223,13 @@ def _load_hermes_env() -> None: def cmd_send(args: argparse.Namespace) -> None: """Entry point wired into the top-level argparse dispatcher.""" - - # 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. + # The gateway config loader (used downstream by send_message_tool and the channel directory) + # needs platform credentials and home channels in os.environ. _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() @@ -292,21 +251,17 @@ def cmd_send(args: argparse.Namespace) -> None: _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. + # send_message_tool auto-loads gateway config + env and routes to the platform adapter + # (bot-token path for Telegram/Discord/Slack/Signal/SMS/WhatsApp; live-adapter path for + # plugin platforms). It 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))) @@ -325,7 +280,6 @@ _SEND_ARGUMENTS = ( ), )), (("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, diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index b236149c88..674200b6f9 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,21 +20,22 @@ 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" - - -# 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 +_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~") def print_header(title: str): @@ -44,16 +44,20 @@ def print_header(title: str): 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 +66,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 +93,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): @@ -187,14 +169,11 @@ 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 @@ -237,6 +216,11 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None: 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 +229,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 @@ -263,8 +246,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: @@ -283,8 +265,7 @@ def run_setup_action_with_navigation( try: _run_setup_steps([(label, action)]) except _SetupCancelled: - print() - print_info(cancelled_message) + _info(None, cancelled_message) # ── Prompt primitives ── @@ -304,14 +285,13 @@ def prompt_choice(question: str, choices: list, default: int = 0, description: s request to 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: @@ -319,9 +299,8 @@ def is_noninteractive() -> bool: 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. + prompt that aborts on EOF would kill the spawned action. Honour the flag so callers fall back + to their default. """ return os.environ.get("HERMES_NONINTERACTIVE", "").strip().lower() in {"1", "true", "yes", "on"} @@ -334,14 +313,12 @@ def prompt_yes_no(question: str, default: bool = True) -> bool: """ 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,7 +330,6 @@ 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"}: @@ -369,13 +345,9 @@ def prompt_checklist(title: str, items: list, pre_selected: list = None) -> list ``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 = [] - 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 +365,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,10 +394,6 @@ 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 ── @@ -449,8 +420,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,14 +449,19 @@ 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 = ( @@ -530,16 +506,15 @@ 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("") @@ -561,15 +536,10 @@ 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 ── @@ -640,15 +610,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"]: @@ -674,20 +636,15 @@ 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 +700,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,37 +747,24 @@ 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. 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..8a2ca0563c 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) @@ -255,6 +240,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 ``"