From ec8e85965d87addc3d6281d04f6f8237f7454b07 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 16:29:58 -0700 Subject: [PATCH] refactor(tools_config): compact extracted modules (join statements, tighten comments) --- hermes_cli/tools_config_cua.py | 401 +++++++++----------------- hermes_cli/tools_config_mcp.py | 85 ++---- hermes_cli/tools_config_post_setup.py | 240 +++++---------- hermes_cli/tools_config_providers.py | 399 +++++++++---------------- 4 files changed, 361 insertions(+), 764 deletions(-) diff --git a/hermes_cli/tools_config_cua.py b/hermes_cli/tools_config_cua.py index e6cd4ee73c..94ea7b0277 100644 --- a/hermes_cli/tools_config_cua.py +++ b/hermes_cli/tools_config_cua.py @@ -11,9 +11,7 @@ from pathlib import Path from typing import List, Optional from hermes_cli.cli_output import ( - print_info as _print_info, - print_success as _print_success, - print_warning as _print_warning, + print_info as _print_info, print_success as _print_success, print_warning as _print_warning, ) logger = logging.getLogger("hermes_cli.tools_config") @@ -22,11 +20,9 @@ logger = logging.getLogger("hermes_cli.tools_config") def _post_setup_no_window_flags(*, streams_to_console: bool = False) -> int: """Win32 creationflags that stop post-setup children flashing a console. - The GUI runs post-setup hooks via a console-less child; on Windows every console grandchild - (npm, pip, powershell) would flash a new window. CREATE_NO_WINDOW suppresses that while keeping - stdio inheritable (unlike DETACHED_PROCESS). Returns 0 on POSIX. ``streams_to_console`` children - are only hidden when our own stdout is not a console, so live installer output is never - swallowed. + CREATE_NO_WINDOW hides console grandchildren (npm, pip, powershell) while keeping stdio inheritable + (unlike DETACHED_PROCESS). Returns 0 on POSIX. ``streams_to_console`` children are only hidden when + our own stdout is not a console, so live installer output is never swallowed. """ from hermes_cli._subprocess_compat import windows_hide_flags @@ -48,11 +44,7 @@ def _cua_driver_cmd() -> str: def _cua_version_summary(raw: str, *, limit: int = 120) -> str: - """Reduce a driver's ``--version`` output to one short status line. - - A ``HERMES_CUA_DRIVER_CMD`` override may print a multi-line banner (e.g. cmd.exe); interpolating - it verbatim shattered the one-line status summary, so keep the first non-empty line, bounded. - """ + """First non-empty line of ``--version`` output, bounded (an override may print a multi-line banner).""" for line in (raw or "").splitlines(): text = line.strip() if text: @@ -68,11 +60,9 @@ def _resolved_cua_driver_cmd() -> Optional[str]: def _cua_driver_env() -> dict: - """cua-driver child env with the Hermes telemetry policy applied. + """cua-driver child env with the Hermes telemetry policy applied (``cua_backend.cua_driver_child_env``). - Delegates to ``cua_backend.cua_driver_child_env`` (telemetry disabled by default; user opt-in - via ``computer_use.cua_telemetry``). Falls back to the current environment if the helper can't - be imported, so install/status never break on a telemetry-helper error. + Falls back to the current environment if the helper can't be imported, so install/status never break. """ try: from tools.computer_use.cua_backend import cua_driver_child_env @@ -108,11 +98,7 @@ def _cua_driver_contract_status(binary: Optional[str] = None) -> dict: return dict(_CUA_DRIVER_CONTRACT_CACHE["state"]) state = cua_driver_runtime_contract_status(resolved) - _CUA_DRIVER_CONTRACT_CACHE.update( - fingerprint=fingerprint, - checked_at=now, - state=dict(state), - ) + _CUA_DRIVER_CONTRACT_CACHE.update(fingerprint=fingerprint, checked_at=now, state=dict(state)) return state @@ -123,28 +109,18 @@ def _cua_driver_install_ready() -> bool: return sys.platform != "win32" or _cua_driver_autostart_registered_windows() -def _pip_install( - args: List[str], - *, - timeout: int = 300, - capture_output: bool = True, -): +def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool = True): """Install Python packages from a post-setup hook. - Strategy (in order): 1. ``uv pip install`` if uv is on PATH — fast, doesn't need pip in the - venv. 2. ``python -m pip install`` — works on stdlib venvs. 3. ``python -m ensurepip --upgrade`` - then retry pip — covers ``uv venv`` which creates a venv WITHOUT pip. - - Why this exists: the Windows installer creates the venv via ``uv venv``, which doesn't seed pip. - Post-setup hooks that shelled out to ``[sys.executable, '-m', 'pip', 'install', ...]`` failed - with ``No module named pip`` on every fresh install. uv-first sidesteps that. + Order: ``uv pip install`` (fast, needs no pip in the venv), then ``python -m pip install``, then + ``python -m ensurepip --upgrade`` and retry pip. The last tier exists because the Windows installer + creates the venv via ``uv venv``, which does NOT seed pip, so bare ``-m pip`` failed on fresh installs. """ venv_root = Path(sys.executable).parent.parent uv_env = {**os.environ, "VIRTUAL_ENV": str(venv_root)} - # Managed uv first: $HERMES_HOME/bin is never on PATH, so a bare which() misses the uv Hermes - # installed. ensure_uv() (not a pure lookup) because installing uv is in scope during setup, and - # tier 2 is a pip that the Windows installer's `uv venv` does not seed. + # Managed uv first: $HERMES_HOME/bin is never on PATH, so a bare which() misses the uv Hermes installed; + # ensure_uv() (not a pure lookup) because installing uv is in scope during setup. from hermes_cli.managed_uv import ensure_uv uv_bin = ensure_uv() @@ -154,9 +130,7 @@ def _pip_install( [uv_bin, "pip", "install", *args], capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout, env=uv_env, - creationflags=_post_setup_no_window_flags( - streams_to_console=not capture_output - ), + creationflags=_post_setup_no_window_flags(streams_to_console=not capture_output), ) if result.returncode == 0: return result @@ -184,24 +158,20 @@ def _pip_install( except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: # Synthesize a result so callers see a clean failure path. return subprocess.CompletedProcess( - pip_cmd, returncode=1, stdout="", - stderr=f"pip not available and ensurepip failed: {e}", + pip_cmd, returncode=1, stdout="", stderr=f"pip not available and ensurepip failed: {e}", ) return subprocess.run( pip_cmd + ["install", *args], capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout, - creationflags=_post_setup_no_window_flags( - streams_to_console=not capture_output - ), + creationflags=_post_setup_no_window_flags(streams_to_console=not capture_output), ) # No pre-install release/asset probe: cua-driver-rs releases are all prereleases, which GitHub's -# `/releases/latest` skips (it returned a package with zero binary assets and made every non-arm64 -# host skip the install), and re-implementing upstream's tag resolution here would drift. Trust the -# installer: fresh installs run install.sh directly (it errors clean on a missing-arch asset); -# upgrades ask the binary via `cua_driver_update_check()`. +# `/releases/latest` skips (zero binary assets → every non-arm64 host skipped the install), and +# re-implementing upstream's tag resolution here would drift. Fresh installs run install.sh directly (it +# errors clean on a missing-arch asset); upgrades ask the binary via `cua_driver_update_check()`. def _cua_install_target_writable() -> bool: @@ -227,15 +197,13 @@ def _cua_driver_version(binary: str) -> Optional[str]: def _confirmed_update_check(driver_cmd: str, require_confirmed_update: bool) -> tuple: - """Ask the installed driver whether a newer release exists. + """Ask the installed driver whether a newer release exists; returns ``(proceed, pin_version)``. - Returns ``(proceed, pin_version)``. ``proceed=False`` means the caller should stop with - success (already latest, or indeterminate under ``require_confirmed_update``). Best-effort: an - older driver (no check-update verb) or an offline check yields None; then `hermes update` - keeps the installed version — an indeterminate check must never cost a multi-minute silent - reinstall on every update — while an explicit `computer-use install --upgrade` falls through. + ``proceed=False`` means stop with success (already latest, or indeterminate under + ``require_confirmed_update``). An old driver (no check-update verb) or offline check yields None: + `hermes update` then keeps the installed version — an indeterminate check must never cost a + multi-minute silent reinstall on every update — while explicit `install --upgrade` falls through. """ - _state = None try: from tools.computer_use.cua_backend import cua_driver_update_check _state = cua_driver_update_check() @@ -253,22 +221,16 @@ def _confirmed_update_check(driver_cmd: str, require_confirmed_update: bool) -> "(offline, rate-limited, or driver too old to check); " "keeping the installed version." ) - _print_info( - " Force a refresh with: hermes computer-use install --upgrade" - ) + _print_info(" Force a refresh with: hermes computer-use install --upgrade") return False, None confirmed_version = None if _state is not None and _state.get("update_available"): - # Windows routine upgrades run unattended-safe rather than deferring: stdin is closed (a - # consent Read-Host can't block), the version is pinned, the ceiling is - # _CUA_BACKGROUND_UPDATE_TIMEOUT, and the installer preflights skip in seconds when the - # lock is held or GitHub is unreachable. Only contract repairs and fresh installs stay - # interactive-only — there upstream legitimately needs a human (autostart elevation, - # SmartScreen). - # Pin to the release check-update just confirmed: `latest_version` comes from the GitHub - # Releases API so its assets exist, unlike the installer's baked version on `main`, which - # is bumped before the assets are published and 404s when installed unpinned. Malformed - # values are ignored → unpinned fallback. + # Windows routine upgrades run unattended-safe (stdin closed, version pinned, ceiling + # _CUA_BACKGROUND_UPDATE_TIMEOUT, preflights skip in seconds); only contract repairs and fresh + # installs stay interactive-only, where upstream needs a human (autostart elevation, SmartScreen). + # Pin to the release check-update confirmed: `latest_version` comes from the GitHub Releases API so + # its assets exist, unlike the installer's baked version on `main`, which is bumped before assets + # are published and 404s unpinned. Malformed values are ignored → unpinned fallback. import re as _re _latest = str(_state.get("latest_version") or "").strip().lstrip("vV") @@ -284,9 +246,9 @@ def install_cua_driver( ) -> bool: """Install or refresh the cua-driver binary used by Computer Use. - The upstream installer always pulls the latest release tag, so re-running it is the canonical - way to upgrade. ``upgrade=False`` (toolset enable flow) keeps a compatible installation, repairs - an old/incomplete one and installs when missing; ``upgrade=True`` always refreshes. + The upstream installer always pulls the latest release tag, so re-running it is the canonical way to + upgrade. ``upgrade=False`` (toolset enable flow) keeps a compatible installation, repairs an + old/incomplete one and installs when missing; ``upgrade=True`` always refreshes. """ import platform as _plat @@ -310,24 +272,15 @@ def install_cua_driver( # cannot repair the configured path and would mutate an unrelated installation. override = os.environ.get("HERMES_CUA_DRIVER_CMD", "").strip() if override and not binary: - _print_warning( - " HERMES_CUA_DRIVER_CMD does not resolve to an executable: " - f"{override}" - ) - _print_info( - " Fix or unset the override before running computer-use install." - ) + _print_warning(f" HERMES_CUA_DRIVER_CMD does not resolve to an executable: {override}") + _print_info(" Fix or unset the override before running computer-use install.") return False # Not installed → fresh install path (only when caller asked for it). if not binary and not upgrade: if not _cua_install_target_writable(): - _print_info( - " /Applications is not writable; skipping cua-driver install." - ) - _print_info( - " Run from an admin account or install cua-driver manually." - ) + _print_info(" /Applications is not writable; skipping cua-driver install.") + _print_info(" Run from an admin account or install cua-driver manually.") return False if not shutil.which(fetch_tool): _print_warning(f" {fetch_tool} not found — install manually:") @@ -336,9 +289,9 @@ def install_cua_driver( return _run_cua_driver_installer(label="Installing") # A driver failing Hermes' runtime contract (version floor, missing manifest verbs) is repaired - # regardless of mode. Hermes' minimum requirement IS the confirmation an upgrade is needed, so - # this path must not defer to the driver's `check-update` verb — a cached/indeterminate "no - # update" answer would pin users on an unusable driver forever. + # regardless of mode. Hermes' minimum requirement IS the confirmation an upgrade is needed, so this + # path must not defer to the driver's `check-update` verb — a cached/indeterminate "no update" + # answer would pin users on an unusable driver forever. contract = _cua_driver_contract_status(binary) if binary else None repair_existing = bool(binary and contract and not contract.get("ready")) @@ -350,9 +303,7 @@ def install_cua_driver( else: _print_success(f" {driver_cmd} already installed: {version or 'unknown version'}") if is_windows and not _repair_cua_driver_autostart_windows(binary, verbose=False): - _print_warning( - " cua-driver is compatible, but Windows autostart repair failed." - ) + _print_warning(" cua-driver is compatible, but Windows autostart repair failed.") return False _print_cua_platform_notes(is_windows, is_linux, fresh_install=False) return True @@ -361,8 +312,7 @@ def install_cua_driver( version = contract.get("version") or "unknown version" reason = contract.get("reason") or "required runtime features are missing" _print_warning( - f" Found cua-driver {version}, but Hermes cannot use its current " - f"runtime contract: {reason}." + f" Found cua-driver {version}, but Hermes cannot use its current runtime contract: {reason}." ) if os.environ.get("HERMES_CUA_DRIVER_CMD", "").strip(): _print_info( @@ -372,24 +322,18 @@ def install_cua_driver( return False if is_windows and require_confirmed_update: _print_info( - " Automatic Windows updates cannot safely run cua-driver's " - "interactive repair installer." + " Automatic Windows updates cannot safely run cua-driver's interactive repair installer." ) _print_info( - " Repair it from an interactive terminal with: " - "hermes computer-use install --upgrade" + " Repair it from an interactive terminal with: hermes computer-use install --upgrade" ) return False _print_info(" Repairing it with the current upstream installer.") # upgrade=True path — refresh to the latest upstream release. if not _cua_install_target_writable(): - _print_info( - " /Applications is not writable; skipping cua-driver refresh." - ) - _print_info( - " Run `hermes computer-use install --upgrade` from an admin account to update it." - ) + _print_info(" /Applications is not writable; skipping cua-driver refresh.") + _print_info(" Run `hermes computer-use install --upgrade` from an admin account to update it.") return bool(binary) if not shutil.which(fetch_tool): @@ -403,17 +347,14 @@ def install_cua_driver( return True if is_windows and require_confirmed_update and not binary: - # Missing binary (enabled but never installed, or wiped by a failed install): an automatic - # Windows update must never launch install.ps1, which can demand console/UAC consent the - # hidden updater cannot provide. + # Missing binary (enabled but never installed, or wiped by a failed install): an automatic Windows + # update must never launch install.ps1, which can demand console/UAC consent the hidden updater + # cannot provide. _print_info( " cua-driver is not installed; automatic Windows updates " "cannot safely run its interactive installer." ) - _print_info( - " Install it from an interactive terminal with: " - "hermes computer-use install --upgrade" - ) + _print_info(" Install it from an interactive terminal with: hermes computer-use install --upgrade") return False # Best-effort before/after version display. @@ -444,22 +385,19 @@ def install_cua_driver( return ok -# Ceiling for one upstream-installer run. Must exceed the installer's own stale-lock recovery -# window: _install-rust.sh serializes installs with ~/.cua-driver/packages/.install.lock.d and only -# force-releases a dead holder's lock after LOCK_STALE_AFTER_SECONDS=600. A shorter Python-side -# timeout kills every run before that recovery fires — a permanent "always times out" wedge. -# 660s = 600s lock window + 60s headroom for the download/swap. +# Ceiling for one upstream-installer run: must exceed the installer's own stale-lock recovery window +# (_install-rust.sh force-releases a dead holder's lock only after LOCK_STALE_AFTER_SECONDS=600; a shorter +# timeout kills every run before that fires — a permanent wedge). 660s = 600s + 60s headroom. _CUA_INSTALLER_TIMEOUT = 660 -# Grace for draining the installer's pipes after a timeout kill. The kill is best-effort (see -# _reap_after_timeout), so the drain must be bounded: a surviving descendant still holds the -# inherited stdout handle and an unbounded read waits on an EOF that never comes, turning the -# ceiling above into no ceiling. A successful kill closes the pipe immediately, so this is free. +# Grace for draining the installer's pipes after a timeout kill. The kill is best-effort (_reap_after_timeout) +# so the drain must be bounded: a surviving descendant holding the inherited stdout handle would otherwise +# make the read wait on an EOF that never comes. A successful kill closes the pipe at once, so this is free. _CUA_INSTALLER_DRAIN_GRACE = 15 -# Quiet unattended refreshes from ``hermes update``: bounded even when upstream waits on Read-Host -# or a consent prompt. Explicit ``computer-use install --upgrade`` keeps the full ceiling. (The -# lock/network preflights make a legitimate long wait impossible here, so a short ceiling is safe.) +# Quiet unattended refreshes from ``hermes update``: bounded even when upstream waits on Read-Host or a +# consent prompt; explicit ``computer-use install --upgrade`` keeps the full ceiling. Safe because the +# lock/network preflights make a legitimate long wait impossible here. _CUA_BACKGROUND_UPDATE_TIMEOUT = 120 # Upstream installer's stale-lock threshold (LOCK_STALE_AFTER_SECONDS in _install-rust.sh), so the @@ -469,10 +407,7 @@ _CUA_LOCK_STALE_AFTER = 600 def _cua_install_home() -> "Path": """Package home shared by the upstream POSIX and Windows installers.""" - return Path( - os.environ.get("CUA_DRIVER_RS_HOME") - or str(Path.home() / ".cua-driver") - ) + return Path(os.environ.get("CUA_DRIVER_RS_HOME") or str(Path.home() / ".cua-driver")) def _cua_install_lock_dir() -> "Path": @@ -488,9 +423,9 @@ def _cua_windows_install_lock_file() -> "Path": def _clear_stale_windows_cua_install_lock() -> None: """Delete install.ps1's lock file only when no process still holds it. - install.ps1 locks with ``FileShare::None``; mirror it with a zero-share ``CreateFileW`` probe - and ``FILE_FLAG_DELETE_ON_CLOSE`` so an unlocked leftover is removed atomically, with no window - in which a new installer could acquire the file between probe and delete. + install.ps1 locks with ``FileShare::None``; mirror it with a zero-share ``CreateFileW`` probe and + ``FILE_FLAG_DELETE_ON_CLOSE`` so an unlocked leftover is removed atomically, with no window in which + a new installer could acquire the file between probe and delete. """ lock_file = _cua_windows_install_lock_file() try: @@ -511,13 +446,8 @@ def _clear_stale_windows_cua_install_lock() -> None: kernel32 = _ctypes.WinDLL("kernel32", use_last_error=True) create_file = kernel32.CreateFileW create_file.argtypes = [ - _wintypes.LPCWSTR, - _wintypes.DWORD, - _wintypes.DWORD, - _wintypes.LPVOID, - _wintypes.DWORD, - _wintypes.DWORD, - _wintypes.HANDLE, + _wintypes.LPCWSTR, _wintypes.DWORD, _wintypes.DWORD, _wintypes.LPVOID, + _wintypes.DWORD, _wintypes.DWORD, _wintypes.HANDLE, ] create_file.restype = _wintypes.HANDLE close_handle = kernel32.CloseHandle @@ -536,26 +466,19 @@ def _clear_stale_windows_cua_install_lock() -> None: invalid_handle = _wintypes.HANDLE(-1).value if handle == invalid_handle: logger.debug( - "Windows cua install lock at %s is still held or cannot be " - "removed (winerror %s)", - lock_file, - _ctypes.get_last_error(), + "Windows cua install lock at %s is still held or cannot be removed (winerror %s)", + lock_file, _ctypes.get_last_error(), ) return if not close_handle(handle): logger.debug( - "could not close Windows cua install lock probe at %s " - "(winerror %s)", - lock_file, - _ctypes.get_last_error(), + "could not close Windows cua install lock probe at %s (winerror %s)", + lock_file, _ctypes.get_last_error(), ) return if lock_file.exists(): - logger.debug( - "Windows cua install lock probe succeeded but %s remains", - lock_file, - ) + logger.debug("Windows cua install lock probe succeeded but %s remains", lock_file) return logger.info("Cleared stale Windows cua-driver install lock at %s", lock_file) @@ -567,10 +490,9 @@ def _clear_stale_windows_cua_install_lock() -> None: def _clear_stale_cua_install_lock() -> None: """Best-effort: remove a stale installer lock left by a dead holder. - The POSIX installer stamps its holder pid into ``~/.cua-driver/packages/.install.lock.d/info``. - The Windows installer instead holds ``~/.cua-driver/install.lock`` open with - ``FileShare::None``. Clear either artifact up front only when its platform-specific liveness - check proves that no install still holds it. + POSIX stamps the holder pid into ``~/.cua-driver/packages/.install.lock.d/info``; Windows holds + ``~/.cua-driver/install.lock`` open with ``FileShare::None``. Clear either artifact up front only + when its platform-specific liveness check proves that no install still holds it. """ if sys.platform == "win32": _clear_stale_windows_cua_install_lock() @@ -625,16 +547,14 @@ def _cua_install_lock_held() -> bool: lock_file = _cua_windows_install_lock_file() if not lock_file.is_file(): return False - # install.ps1 holds the file with FileShare::None — any open fails with a sharing - # violation while held. The stale-clear already deleted an unheld file, so surviving - # = held; confirm with an open probe. + # install.ps1 holds the file with FileShare::None — any open fails with a sharing violation + # while held. Surviving the stale-clear = held; confirm with an open probe. try: with open(lock_file, "r+b"): return False # opened fine → not held (racy leftover) except OSError: # sharing violation surfaces as PermissionError return True - lock_dir = _cua_install_lock_dir() - return lock_dir.is_dir() + return _cua_install_lock_dir().is_dir() except Exception as e: logger.debug("cua install lock probe failed: %s", e) return False @@ -643,17 +563,15 @@ def _cua_install_lock_held() -> bool: def _cua_release_endpoint_reachable(timeout: float = 5.0) -> bool: """Fast probe: can we reach GitHub's release download host at all? - When github.com is down the installer dies slowly inside its own retries and eats the whole - unattended ceiling; a 5s HEAD decides in seconds. Only a connection-level failure counts as - unreachable — any HTTP response (even 4xx/5xx) proves the path works. + When github.com is down the installer dies slowly inside its own retries and eats the whole unattended + ceiling; a 5s HEAD decides in seconds. Only a connection-level failure counts as unreachable — any + HTTP response (even 4xx/5xx) proves the path works. """ import urllib.error import urllib.request try: - req = urllib.request.Request( - "https://github.com/trycua/cua/releases", method="HEAD" - ) + req = urllib.request.Request("https://github.com/trycua/cua/releases", method="HEAD") with urllib.request.urlopen(req, timeout=timeout): return True except urllib.error.HTTPError: @@ -675,9 +593,7 @@ def _cua_driver_autostart_registered_windows() -> bool: try: result = subprocess.run( ["schtasks.exe", "/Query", "/TN", "cua-driver-serve"], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - timeout=10, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10, ) except Exception: return False @@ -687,9 +603,9 @@ def _cua_driver_autostart_registered_windows() -> bool: def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> bool: """Best-effort repair for Windows installer autostart quoting failures. - Older install.ps1 builds interpolated the binary path into a PowerShell command string, which - split at the first space. If the scheduled task is missing, retry via Start-Process's structured - ``-FilePath`` / ``-ArgumentList`` parameters instead. + Older install.ps1 builds interpolated the binary path into a PowerShell command string, which split at + the first space. If the scheduled task is missing, retry via Start-Process's structured ``-FilePath`` / + ``-ArgumentList`` parameters instead. """ if sys.platform != "win32": return True @@ -717,11 +633,7 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b try: result = subprocess.run( [ps, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_cmd], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=300, + capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=300, env=_cua_driver_env(), ) except subprocess.TimeoutExpired: @@ -777,9 +689,8 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None: if not is_windows: os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) # windows-footgun: ok — POSIX branch only else: - # PowerShell may leave download/install helpers alive after its direct process is - # killed. They inherit stdout and can keep both communicate() and install.lock wedged, - # so collect the tree first and kill it leaf-up. + # PowerShell may leave download/install helpers alive after its direct process is killed. They + # inherit stdout and can keep communicate() and install.lock wedged, so kill the tree leaf-up. import psutil as _psutil try: @@ -789,9 +700,7 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None: return except _psutil.Error as e: logger.debug( - "could not enumerate cua-driver installer tree for pid %s: %s", - proc.pid, - e, + "could not enumerate cua-driver installer tree for pid %s: %s", proc.pid, e ) proc.kill() return @@ -802,21 +711,13 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None: except _psutil.NoSuchProcess: pass except _psutil.Error as e: - logger.debug( - "could not kill cua-driver installer child pid %s: %s", - child.pid, - e, - ) + logger.debug("could not kill cua-driver installer child pid %s: %s", child.pid, e) try: parent.kill() except _psutil.NoSuchProcess: pass except _psutil.Error as e: - logger.debug( - "could not kill cua-driver installer parent pid %s: %s", - proc.pid, - e, - ) + logger.debug("could not kill cua-driver installer parent pid %s: %s", proc.pid, e) proc.kill() except (OSError, ProcessLookupError): proc.kill() @@ -825,27 +726,23 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None: def _reap_after_timeout(proc, *, is_windows: bool) -> None: """Kill the installer tree, then drain its pipes under a deadline. - An unbounded drain blocks on an EOF that only arrives when someone kills a surviving - descendant by hand, so ``_CUA_INSTALLER_TIMEOUT`` would stop bounding anything. + An unbounded drain blocks on an EOF that only arrives when someone kills a surviving descendant by + hand, so ``_CUA_INSTALLER_TIMEOUT`` would stop bounding anything. """ _kill_installer_tree(proc, is_windows=is_windows) try: drained_out, _ = proc.communicate(timeout=_CUA_INSTALLER_DRAIN_GRACE) - # The partial output names WHERE the installer was stuck (lock wait, consent prompt, - # download); without it the timeout line is unactionable. + # The partial output names WHERE the installer was stuck (lock wait, consent prompt, download). if drained_out: logger.warning( - "cua-driver installer timed out; last output before " - "kill:\n%s", - drained_out[-2000:], + "cua-driver installer timed out; last output before kill:\n%s", drained_out[-2000:], ) except subprocess.TimeoutExpired: - # Deliberately not closing proc.stdout: communicate()'s reader threads are still blocked - # on that handle and closing it underneath them races; they are daemon threads. + # Deliberately not closing proc.stdout: communicate()'s reader threads are still blocked on that + # handle and closing it underneath them races; they are daemon threads. logger.debug( "cua-driver installer pipes still open %ss after the kill — " - "abandoning the drain, a surviving descendant holds the " - "inherited handle", + "abandoning the drain, a surviving descendant holds the inherited handle", _CUA_INSTALLER_DRAIN_GRACE, ) except (OSError, ValueError) as e: @@ -857,26 +754,16 @@ def _cua_installer_command(is_windows: bool): if is_windows: # Mirror the one-liner printed by cua_driver_install_hint(). ps_oneliner = f"irm {_CUA_INSTALL_PS1_URL} | iex" - install_cmd = [ - "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", - "-Command", ps_oneliner, - ] - manual_hint = ( - 'powershell -NoProfile -ExecutionPolicy Bypass -Command ' - f'"{ps_oneliner}"' - ) + install_cmd = ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_oneliner] + manual_hint = f'powershell -NoProfile -ExecutionPolicy Bypass -Command "{ps_oneliner}"' return install_cmd, manual_hint, None - # Download-then-exec instead of `bash -c "$(curl …)"`: no shell=True, no command substitution, - # and the script lands in a mkstemp file (unpredictable name, 0600) rather than a fixed /tmp - # path — avoiding both shell injection and a symlink/TOCTOU race on multi-user machines. The - # manual hint stays the upstream one-liner the docs teach. + # Download-then-exec instead of `bash -c "$(curl …)"`: no shell=True, no command substitution, and the + # script lands in a mkstemp file (unpredictable name, 0600) rather than a fixed /tmp path — avoiding + # both shell injection and a symlink/TOCTOU race. The manual hint stays the upstream one-liner. import tempfile as _tempfile - install_url = ( - "https://raw.githubusercontent.com/trycua/cua/main/" - "libs/cua-driver/scripts/install.sh" - ) + install_url = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.sh" manual_hint = f'/bin/bash -c "$(curl -fsSL {install_url})"' fd, script_path = _tempfile.mkstemp(prefix="cua-driver-install-", suffix=".sh") os.close(fd) @@ -898,11 +785,9 @@ def _cua_installer_command(is_windows: bool): def _unattended_installer_preflight(install_cmd: list, is_windows: bool): """Fail FAST on the two conditions that otherwise consume the whole unattended ceiling. - 1. Install lock held by a live process — upstream would poll it for up to - LOCK_STALE_AFTER_SECONDS=600 before probing the holder (the 11-minute silent hang class). - 2. Release host unreachable — the installer would die slowly inside its own retries; a 5s HEAD - answers now. - + 1. Install lock held by a live process — upstream would poll it for up to LOCK_STALE_AFTER_SECONDS=600 + before probing the holder (the 11-minute silent hang class). + 2. Release host unreachable — the installer would die slowly inside its own retries; a 5s HEAD answers. Returns the (possibly rewritten) install command, or None to skip this refresh. Explicit `computer-use install --upgrade` runs never come here and keep upstream's full lock-recovery. """ @@ -911,27 +796,20 @@ def _unattended_installer_preflight(install_cmd: list, is_windows: bool): " Another cua-driver install is in progress (upstream " "install lock is held) — skipping this refresh." ) - _print_info( - " If no install is really running, retry with: " - "hermes computer-use install --upgrade" - ) + _print_info(" If no install is really running, retry with: hermes computer-use install --upgrade") return None if not _cua_release_endpoint_reachable(): _print_info( - " github.com is unreachable — skipping cua-driver " - "refresh (will retry on the next update)." + " github.com is unreachable — skipping cua-driver refresh (will retry on the next update)." ) return None if is_windows: - # -NoAutoStart skips Register-CuaDriverAutostart — the ONLY branch of install.ps1 that - # self-elevates (UAC). Cost: an existing cua-driver-serve task keeps pointing at the - # previous binary until the next interactive upgrade. Scriptblock invocation (instead of - # `| iex`) is what lets us pass the parameter to a piped script. + # -NoAutoStart skips Register-CuaDriverAutostart — the ONLY branch of install.ps1 that self-elevates + # (UAC). Cost: an existing cua-driver-serve task keeps pointing at the previous binary until the + # next interactive upgrade. Scriptblock invocation (not `| iex`) is what lets us pass the parameter. install_cmd = [ "powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", - "-Command", - f"$sc = irm {_CUA_INSTALL_PS1_URL}; " - "& ([scriptblock]::Create($sc)) -NoAutoStart", + "-Command", f"$sc = irm {_CUA_INSTALL_PS1_URL}; & ([scriptblock]::Create($sc)) -NoAutoStart", ] return install_cmd @@ -946,8 +824,8 @@ def _run_cua_driver_installer( """Run the upstream cua-driver installer for this platform. The scripts are idempotent: they always download the latest release, so re-running on an - already-installed system performs an upgrade. ``installer_timeout`` lets quiet callers use a - shorter ceiling without weakening the explicit install path's stale-lock recovery window. + already-installed system performs an upgrade. ``installer_timeout`` lets quiet callers use a shorter + ceiling without weakening the explicit install path's stale-lock recovery window. """ import platform as _plat @@ -973,8 +851,8 @@ def _run_cua_driver_installer( # Both upstream installers honour CUA_DRIVER_RS_VERSION over their baked default. installer_env["CUA_DRIVER_RS_VERSION"] = pin_version - # A previous timed-out install can leave upstream's concurrent-install lock behind; clear it - # when provably stale so the refresh doesn't wedge waiting on a dead holder. + # A previous timed-out install can leave upstream's concurrent-install lock behind; clear it when + # provably stale so the refresh doesn't wedge waiting on a dead holder. _clear_stale_cua_install_lock() # Unattended refreshes (installer_timeout set by `hermes update`) preflight and may skip. @@ -983,16 +861,16 @@ def _run_cua_driver_installer( if install_cmd is None: return False - # POSIX: own process group so a timeout kill takes out the whole `curl | bash` pipeline (and - # the exec'd _install-rust.sh), not just the outer shell — surviving grandchildren would keep - # holding the install lock and wedge every later run. + # POSIX: own process group so a timeout kill takes out the whole `curl | bash` pipeline (and the exec'd + # _install-rust.sh), not just the outer shell — surviving grandchildren would keep holding the install + # lock and wedge every later run. popen_kwargs = {} if not is_windows: popen_kwargs["start_new_session"] = True try: - # Non-verbose (`hermes update` refresh): capture the installer's chatty "Next steps" wall - # and log it so a failure stays debuggable. Verbose interactive installs stream live. + # Non-verbose (`hermes update` refresh): capture the installer's chatty "Next steps" wall and log it + # so a failure stays debuggable. Verbose interactive installs stream live. if verbose: proc = subprocess.Popen( install_cmd, shell=False, env=installer_env, @@ -1004,9 +882,7 @@ def _run_cua_driver_installer( except subprocess.TimeoutExpired: _reap_after_timeout(proc, is_windows=is_windows) raise - result = subprocess.CompletedProcess( - install_cmd, proc.returncode, stdout=None, stderr=None - ) + result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=None, stderr=None) else: proc = subprocess.Popen( install_cmd, shell=False, env=installer_env, @@ -1021,21 +897,15 @@ def _run_cua_driver_installer( except subprocess.TimeoutExpired: _reap_after_timeout(proc, is_windows=is_windows) raise - result = subprocess.CompletedProcess( - install_cmd, proc.returncode, stdout=out, stderr=None - ) - # During `hermes update`, sys.stdout is the mirroring _UpdateOutputStream whose `_log` - # handle is ~/.hermes/logs/update.log — write straight to it so the full installer - # output is kept (success AND failure) without echoing it to the terminal. + result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=out, stderr=None) + # During `hermes update`, sys.stdout is the mirroring _UpdateOutputStream whose `_log` handle is + # ~/.hermes/logs/update.log — write straight to it so the full installer output is kept + # (success AND failure) without echoing it to the terminal. if result.stdout: _update_log = getattr(sys.stdout, "_log", None) if _update_log is not None: try: - _update_log.write( - "\n--- cua-driver installer output ---\n" - + result.stdout - + "\n" - ) + _update_log.write("\n--- cua-driver installer output ---\n" + result.stdout + "\n") _update_log.flush() except Exception: pass @@ -1043,12 +913,8 @@ def _run_cua_driver_installer( logger.debug("cua-driver installer output:\n%s", result.stdout) installed_binary = _resolved_cua_driver_cmd() if result.returncode == 0 and installed_binary: - if is_windows and not _repair_cua_driver_autostart_windows( - installed_binary, verbose=verbose - ): - _print_warning( - " cua-driver installed, but auto-start was not registered." - ) + if is_windows and not _repair_cua_driver_autostart_windows(installed_binary, verbose=verbose): + _print_warning(" cua-driver installed, but auto-start was not registered.") if verbose: _print_success(f" {driver_cmd} installed.") _print_cua_platform_notes(is_windows, is_linux, fresh_install=True) @@ -1057,10 +923,7 @@ def _run_cua_driver_installer( _print_info(f" {manual_hint}") return False except subprocess.TimeoutExpired: - _print_warning( - f" cua-driver {label.lower()} timed out after " - f"{timeout}s." - ) + _print_warning(f" cua-driver {label.lower()} timed out after {timeout}s.") if not is_windows: _print_info( " If this repeats, a stale installer lock may be present — " diff --git a/hermes_cli/tools_config_mcp.py b/hermes_cli/tools_config_mcp.py index 3b1274b6a8..69260a5091 100644 --- a/hermes_cli/tools_config_mcp.py +++ b/hermes_cli/tools_config_mcp.py @@ -17,9 +17,6 @@ from hermes_cli.toolset_scope import ( ) -# ─── MCP Tools Interactive Configuration ───────────────────────────────────── - - def _mcp_match_filter(): """Runtime name-filter matcher (exact names or fnmatch globs), with a literal fallback. @@ -29,7 +26,6 @@ def _mcp_match_filter(): """ try: from tools.mcp_tool import matches_name_filter - return matches_name_filter except ImportError: # pragma: no cover — defensive fallback return lambda tool_name, patterns: tool_name in patterns @@ -50,30 +46,22 @@ def _apply_mcp_checklist(server_name: str, tools_cfg: dict, tool_names: List[str exclude_mode = bool(exclude_set) and not include_set if len(chosen) == len(tool_names) and not exclude_mode: - # All tools enabled — clear filters (cleanest config shape; the - # server's native tool set is the active set, and any tools the - # server adds later are auto-enabled). + # All tools enabled — clear filters so tools the server adds later are auto-enabled. tools_cfg.pop("exclude", None) tools_cfg.pop("include", None) elif exclude_mode: - # Exclude-mode server (catalog default_excluded / hand-written - # tools.exclude): stay in exclude mode — do NOT demote the - # dynamic filter to a frozen include list. Unchecked tools are - # added as literal excludes; re-checked literals are dropped; - # glob patterns are preserved (they intentionally keep matching - # tools the vendor ships later). + # Exclude-mode server (catalog default_excluded / hand-written tools.exclude): stay in exclude + # mode — do NOT demote the dynamic filter to a frozen include list. Unchecked tools become literal + # excludes; re-checked literals are dropped; glob patterns are preserved (they intentionally keep + # matching tools the vendor ships later). old_exclude = sorted(exclude_set or set()) glob_entries = [p for p in old_exclude if "*" in p or "?" in p or "[" in p] literal_entries = {p for p in old_exclude if p not in glob_entries} unchecked = {tn for i, tn in enumerate(tool_names) if i not in chosen} checked = {tool_names[i] for i in chosen} - new_literals = (literal_entries - checked) | { - tn for tn in unchecked if not match(tn, set(old_exclude)) - } + new_literals = (literal_entries - checked) | {tn for tn in unchecked if not match(tn, set(old_exclude))} new_exclude = glob_entries + sorted(new_literals) - glob_shadowed = sorted( - tn for tn in checked if glob_entries and match(tn, set(glob_entries)) - ) + glob_shadowed = sorted(tn for tn in checked if glob_entries and match(tn, set(glob_entries))) if glob_shadowed: _print_warning( f" {server_name}: {len(glob_shadowed)} re-enabled " @@ -89,16 +77,12 @@ def _apply_mcp_checklist(server_name: str, tools_cfg: dict, tool_names: List[str tools_cfg.pop("include", None) else: tools_cfg["include"] = [tool_names[i] for i in sorted(chosen)] - # Drop any legacy exclude block — we're include-mode now. - tools_cfg.pop("exclude", None) + tools_cfg.pop("exclude", None) # include-mode now; drop any legacy exclude block def _configure_mcp_tools_interactive(config: dict): - """Probe MCP servers for available tools and let user toggle them on/off. - - Connects to each server, discovers tools, shows a per-server curses checklist, and writes the - result back as ``tools.exclude`` entries in config.yaml. - """ + """Probe each MCP server for its tools, show a per-server curses checklist, and write the result + back as ``tools.exclude`` / ``tools.include`` entries in config.yaml.""" from hermes_cli.tools_config import save_config from hermes_cli.curses_ui import curses_checklist @@ -108,10 +92,7 @@ def _configure_mcp_tools_interactive(config: dict): _print_info("No MCP servers configured.") return - enabled_names = [ - k for k, v in mcp_servers.items() - if v.get("enabled", True) not in {False, "false", "0", "no", "off"} - ] + enabled_names = [k for k, v in mcp_servers.items() if v.get("enabled", True) not in {False, "false", "0", "no", "off"}] if not enabled_names: _print_info("All MCP servers are disabled.") return @@ -162,10 +143,7 @@ def _configure_mcp_tools_interactive(config: dict): pre_selected = _mcp_preselected(tool_names, include_set, exclude_set, match) chosen = curses_checklist( - f"MCP Server: {server_name} ({len(tools)} tools)", - labels, - pre_selected, - cancel_returns=pre_selected, + f"MCP Server: {server_name} ({len(tools)} tools)", labels, pre_selected, cancel_returns=pre_selected, ) if chosen == pre_selected: @@ -175,9 +153,7 @@ def _configure_mcp_tools_interactive(config: dict): tools_cfg = mcp_servers.setdefault(server_name, {}).setdefault("tools", {}) _apply_mcp_checklist(server_name, tools_cfg, tool_names, chosen, include_set, exclude_set, match) - _print_success( - f" {server_name}: {len(chosen)} enabled, {len(tools) - len(chosen)} disabled" - ) + _print_success(f" {server_name}: {len(chosen)} enabled, {len(tools) - len(chosen)} disabled") any_changes = True if any_changes: @@ -188,18 +164,12 @@ def _configure_mcp_tools_interactive(config: dict): print(color(" No changes to MCP tools", Colors.DIM)) -# ─── Non-interactive disable/enable ────────────────────────────────────────── - - def _apply_toolset_change(config: dict, platform: str, toolset_names: List[str], action: str): """Add or remove built-in toolsets for a platform.""" from hermes_cli.tools_config import _get_platform_tools, _save_platform_tools enabled = _get_platform_tools(config, platform, include_default_mcp_servers=False) - if action == "disable": - updated = enabled - set(toolset_names) - else: - updated = enabled | set(toolset_names) + updated = enabled - set(toolset_names) if action == "disable" else enabled | set(toolset_names) _save_platform_tools(config, platform, updated) @@ -230,10 +200,7 @@ def _print_tools_list(enabled_toolsets: set, mcp_servers: dict, platform: str = from hermes_cli.tools_config import CONFIGURABLE_TOOLSETS, _get_effective_configurable_toolsets effective_all = _get_effective_configurable_toolsets() - effective = [ - (k, l, d) for (k, l, d) in effective_all - if _toolset_allowed_for_platform(k, platform) - ] + effective = [(k, l, d) for (k, l, d) in effective_all if _toolset_allowed_for_platform(k, platform)] builtin_keys = {ts_key for ts_key, _, _ in CONFIGURABLE_TOOLSETS} def _print_rows(entries): @@ -267,12 +234,9 @@ def _print_tools_list(enabled_toolsets: set, mcp_servers: dict, platform: str = def _known_tool_platforms() -> set[str]: - """Return built-in plus discovered plugin platform names. - - Plugin platforms are registered at runtime rather than in the static CLI display registry. Tool - introspection/configuration must recognize those names too, otherwise an active plugin platform - cannot audit its authority. - """ + """Return built-in plus discovered plugin platform names. Plugin platforms register at runtime, not + in the static CLI display registry, and must be recognized so an active plugin platform can audit + its authority.""" from hermes_cli.tools_config import PLATFORMS known = set(PLATFORMS) @@ -283,8 +247,7 @@ def _known_tool_platforms() -> set[str]: discover_plugins() # idempotent known.update(platform_registry.registered_names()) except Exception: - # Plugin discovery is optional. Preserve the built-in CLI path when a - # third-party plugin is malformed or its dependencies are unavailable. + # Plugin discovery is optional: keep the built-in path when a plugin is malformed or deps are missing. pass return known @@ -319,17 +282,11 @@ def tools_disable_enable_command(args): toolset_targets = [t for t in toolset_targets if t in valid_toolsets] # Reject platform-scoped toolsets on platforms that don't allow them. - restricted_targets = [ - t for t in toolset_targets - if not _toolset_allowed_for_platform(t, platform) - ] + restricted_targets = [t for t in toolset_targets if not _toolset_allowed_for_platform(t, platform)] if restricted_targets: for name in restricted_targets: allowed = sorted(_TOOLSET_PLATFORM_RESTRICTIONS.get(name) or set()) - _print_error( - f"Toolset '{name}' is not available on platform '{platform}' " - f"(only: {', '.join(allowed)})" - ) + _print_error(f"Toolset '{name}' is not available on platform '{platform}' (only: {', '.join(allowed)})") toolset_targets = [t for t in toolset_targets if t not in restricted_targets] if toolset_targets: diff --git a/hermes_cli/tools_config_post_setup.py b/hermes_cli/tools_config_post_setup.py index 6305595ec5..d86c7fa7ed 100644 --- a/hermes_cli/tools_config_post_setup.py +++ b/hermes_cli/tools_config_post_setup.py @@ -29,23 +29,16 @@ logger = logging.getLogger("hermes_cli.tools_config") PROJECT_ROOT = Path(__file__).parent.parent.resolve() -# ─── Post-Setup Hooks ───────────────────────────────────────────────────────── - - def _ensure_browser_use_cli(*, verbose_hints: bool = False) -> None: """Install the Browser Use CLI if it isn't already runnable. - The Browser Use CLI 3.0 is the primary driver engine for EVERY browser backend except Camofox - (which is Firefox-based with no CDP surface, so the CDP-only browser-use harness cannot drive - it). - - MANAGED-FIRST: a browser-use on the user's PATH does NOT satisfy this check — only the Hermes- - managed ``$HERMES_HOME/bin`` copy does. + Primary driver engine for EVERY browser backend except Camofox (Firefox-based, no CDP surface). + MANAGED-FIRST: a browser-use on the user's PATH does NOT satisfy this check — only the + Hermes-managed ``$HERMES_HOME/bin`` copy does. """ _print_info(" Ensuring browser-use CLI (managed install)...") try: from tools.browser_use_cli import install_cli - ok, message = install_cli() except Exception as exc: # pragma: no cover — defensive ok, message = False, f"install failed: {exc}" @@ -64,22 +57,15 @@ def _ensure_browser_use_cli(*, verbose_hints: bool = False) -> None: def _post_setup_lightpanda() -> None: - # Browser Use mode drives Lightpanda directly (Hermes spawns - # ``lightpanda serve``); the built-in tools go through agent-browser. - # Neither needs a Chromium build. + # Browser Use mode spawns ``lightpanda serve``; built-in tools go through agent-browser. No Chromium needed. _ensure_browser_use_cli() - from tools.browser_lightpanda import ( - LIGHTPANDA_INSTALL_HINT, - find_lightpanda_binary, - ) + from tools.browser_lightpanda import LIGHTPANDA_INSTALL_HINT, find_lightpanda_binary lightpanda_bin = find_lightpanda_binary() if lightpanda_bin: _print_success(f" Lightpanda found: {lightpanda_bin}") else: - _print_warning( - " lightpanda binary not found on PATH, ~/.lightpanda or ~/.local/bin" - ) + _print_warning(" lightpanda binary not found on PATH, ~/.lightpanda or ~/.local/bin") _print_info(f" {LIGHTPANDA_INSTALL_HINT}") if os.name == "nt": _print_info(" Lightpanda has no native Windows build; run Hermes under WSL2.") @@ -119,8 +105,7 @@ def _post_setup_agent_browser(post_setup_key: str) -> None: agent-browser is not a root package.json dependency — it resolves lazily via npx (or a global/Hermes-managed install), so there is no ``npm install`` step here. """ - # Every non-Camofox backend drives through the Browser Use CLI when runnable — install it here - # too, not only on the explicit "Browser Use" picker row. + # Every non-Camofox backend drives through the Browser Use CLI — install it here too. _ensure_browser_use_cli() try: # Lazy import so the tools_config UI doesn't pull in browser_tool at import time. @@ -141,9 +126,7 @@ def _post_setup_agent_browser(post_setup_key: str) -> None: try: browser_cmd = _find_agent_browser(validate=False) except FileNotFoundError: - _print_warning( - " npx not found - browser tools require Node.js: https://nodejs.org" - ) + _print_warning(" npx not found - browser tools require Node.js: https://nodejs.org") return # Only the local provider needs Chromium on disk; cloud providers host their own. @@ -157,15 +140,9 @@ def _post_setup_agent_browser(post_setup_key: str) -> None: return if _running_in_docker(): - _print_warning( - " Chromium is missing but you're running in Docker." - ) - _print_info( - " Pull the latest image to get the bundled Chromium:" - ) - _print_info( - " docker pull ghcr.io/nousresearch/hermes-agent:latest" - ) + _print_warning(" Chromium is missing but you're running in Docker.") + _print_info(" Pull the latest image to get the bundled Chromium:") + _print_info(" docker pull ghcr.io/nousresearch/hermes-agent:latest") return if _is_npx_agent_browser_sentinel(browser_cmd): @@ -173,9 +150,7 @@ def _post_setup_agent_browser(post_setup_key: str) -> None: # would silently diverge and hand subprocess.run a None argument. npx_bin = _resolve_npx_bin() if not npx_bin: - _print_warning( - " npx not found - install Chromium manually: npx agent-browser install --with-deps" - ) + _print_warning(" npx not found - install Chromium manually: npx agent-browser install --with-deps") return install_cmd = [npx_bin, "--ignore-scripts", "-y", AGENT_BROWSER_NPX_SPEC, "install", "--with-deps"] else: @@ -192,8 +167,7 @@ def _post_setup_camofox() -> None: _print_success(" Camofox already installed, nothing to do") elif _npm_bin: _print_info(" Installing Camofox browser server...") - # Absolute npm path so the .cmd shim executes on Windows; --workspaces=false avoids - # resolving apps/desktop. + # Absolute npm path so the .cmd shim executes on Windows; --workspaces=false avoids resolving apps/desktop. result = subprocess.run( [_npm_bin, "install", "--silent", "--workspaces=false"], capture_output=True, text=True, encoding="utf-8", errors="replace", cwd=str(PROJECT_ROOT), @@ -213,69 +187,46 @@ def _post_setup_camofox() -> None: _print_info(" docker run -p 9377:9377 -e CAMOFOX_PORT=9377 jo-inc/camofox-browser") -_KITTENTTS_WHEEL_URL = ( - "https://github.com/KittenML/KittenTTS/releases/download/" - "0.8.1/kittentts-0.8.1-py3-none-any.whl" -) +_KITTENTTS_WHEEL_URL = "https://github.com/KittenML/KittenTTS/releases/download/0.8.1/kittentts-0.8.1-py3-none-any.whl" -# Post-setup hooks that only pip-install a Python package. Fields: -# module import probe (already installed → skip the install) -# label package name used in status lines -# installing progress line printed before the install -# args _pip_install arguments (keep in sync with -# _RESTORABLE_PYTHON_TOOL_DEPENDENCIES) -# manual the "Run manually:" command shown on failure/timeout -# on_install info lines printed only after a fresh successful install -# always info lines printed whenever the package ends up present +# Post-setup hooks that only pip-install a Python package. Fields: module (import probe; present → skip +# install), label (status lines), installing (progress line), args (_pip_install args — keep in sync with +# _RESTORABLE_PYTHON_TOOL_DEPENDENCIES), manual ("Run manually:" command on failure/timeout), +# on_install (info lines only after a fresh install), always (info lines whenever the package is present). _PIP_POST_SETUP_HOOKS: dict = { "faster_whisper": { - "module": "faster_whisper", - "label": "faster-whisper", + "module": "faster_whisper", "label": "faster-whisper", "installing": "Installing faster-whisper (model ~150MB downloads on first use)...", - "args": ["-U", "faster-whisper", "--quiet"], - "manual": "uv pip install -U faster-whisper", - "on_install": ( - "Model sizes: tiny, base (default), small, medium, large-v3", - "Change via stt.local.model in ~/.hermes/config.yaml", - ), + "args": ["-U", "faster-whisper", "--quiet"], "manual": "uv pip install -U faster-whisper", + "on_install": ("Model sizes: tiny, base (default), small, medium, large-v3", + "Change via stt.local.model in ~/.hermes/config.yaml"), "always": (), }, "kittentts": { - "module": "kittentts", - "label": "kittentts", + "module": "kittentts", "label": "kittentts", "installing": "Installing kittentts (~25-80MB model, CPU-only)...", "args": ["-U", _KITTENTTS_WHEEL_URL, "soundfile", "--quiet"], "manual": f"uv pip install -U '{_KITTENTTS_WHEEL_URL}' soundfile", - "on_install": ( - "Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo", - "Models: KittenML/kitten-tts-nano-0.8-int8 (25MB), micro (41MB), mini (80MB)", - ), + "on_install": ("Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo", + "Models: KittenML/kitten-tts-nano-0.8-int8 (25MB), micro (41MB), mini (80MB)"), "always": (), }, "piper": { - "module": "piper", - "label": "piper-tts", + "module": "piper", "label": "piper-tts", "installing": "Installing piper-tts (~14MB wheel, voices downloaded on first use)...", - "args": ["-U", "piper-tts", "--quiet"], - "manual": "uv pip install -U piper-tts", + "args": ["-U", "piper-tts", "--quiet"], "manual": "uv pip install -U piper-tts", "on_install": (), - "always": ( - "Default voice: en_US-lessac-medium (downloaded on first TTS call)", - "Full voice list: https://github.com/OHF-Voice/piper1-gpl/blob/main/docs/VOICES.md", - "Switch voices by setting tts.piper.voice in ~/.hermes/config.yaml", - ), + "always": ("Default voice: en_US-lessac-medium (downloaded on first TTS call)", + "Full voice list: https://github.com/OHF-Voice/piper1-gpl/blob/main/docs/VOICES.md", + "Switch voices by setting tts.piper.voice in ~/.hermes/config.yaml"), }, "ddgs": { - "module": "ddgs", - "label": "ddgs", + "module": "ddgs", "label": "ddgs", "installing": "Installing ddgs (DuckDuckGo search package)...", - "args": ["-U", "ddgs", "--quiet"], - "manual": "uv pip install -U ddgs", + "args": ["-U", "ddgs", "--quiet"], "manual": "uv pip install -U ddgs", "on_install": (), - "always": ( - "No API key required. DuckDuckGo enforces server-side rate limits.", - "Pair with an extract provider if you also need web_extract.", - ), + "always": ("No API key required. DuckDuckGo enforces server-side rate limits.", + "Pair with an extract provider if you also need web_extract."), }, } @@ -311,11 +262,8 @@ def _post_setup_pip(spec: dict) -> None: def _post_setup_spotify() -> None: - # Run the full `hermes auth spotify` flow — if the user has no - # client_id yet, this drops them into the interactive wizard - # (opens the Spotify dashboard, prompts for client_id, persists - # to ~/.hermes/.env), then continues straight into PKCE. If they - # already have an app, it skips the wizard and just does OAuth. + # Full `hermes auth spotify` flow: no client_id yet → interactive wizard (persists to ~/.hermes/.env) + # then PKCE; existing app → OAuth only. from types import SimpleNamespace try: from hermes_cli.auth import login_spotify_command @@ -326,13 +274,11 @@ def _post_setup_spotify() -> None: _print_info(" Starting Spotify login...") try: login_spotify_command(SimpleNamespace( - client_id=None, redirect_uri=None, scope=None, - no_browser=False, timeout=None, + client_id=None, redirect_uri=None, scope=None, no_browser=False, timeout=None, )) _print_success(" Spotify authenticated") except SystemExit as exc: - # User aborted the wizard, or OAuth failed — don't fail the - # toolset enable; they can retry with `hermes auth spotify`. + # User aborted the wizard or OAuth failed — don't fail the toolset enable. _print_warning(f" Spotify login did not complete: {exc}") _print_info(" Run later: hermes auth spotify") except Exception as exc: @@ -341,7 +287,6 @@ def _post_setup_spotify() -> None: def _post_setup_langfuse() -> None: - # Install the langfuse SDK. try: __import__("langfuse") _print_success(" langfuse SDK already installed") @@ -352,9 +297,7 @@ def _post_setup_langfuse() -> None: _print_success(" langfuse SDK installed") else: _print_warning(" langfuse SDK install failed — run manually: uv pip install langfuse") - # Opt the bundled observability/langfuse plugin into plugins.enabled. - # The plugin ships in the repo but doesn't load until the user enables - # it (standalone plugins are opt-in). + # The bundled observability/langfuse plugin is opt-in (standalone plugins don't load until enabled). try: from hermes_cli.plugins_cmd import _get_enabled_set, _save_enabled_set enabled = _get_enabled_set() @@ -384,9 +327,7 @@ def _post_setup_xai_grok() -> None: existing_api_key = get_env_value("XAI_API_KEY") if oauth_logged_in: - _print_success( - " xAI will use your xAI Grok OAuth (SuperGrok / Premium+) credentials" - ) + _print_success(" xAI will use your xAI Grok OAuth (SuperGrok / Premium+) credentials") return if existing_api_key: _print_success(" xAI will use your existing XAI_API_KEY") @@ -394,11 +335,7 @@ def _post_setup_xai_grok() -> None: _print_info(" xAI needs credentials. Choose one:") try: - from hermes_cli.setup import ( - _run_xai_oauth_login_from_setup, - prompt_choice, - prompt as _setup_prompt, - ) + from hermes_cli.setup import _run_xai_oauth_login_from_setup, prompt_choice, prompt as _setup_prompt from hermes_cli.config import save_env_value except Exception as exc: _print_warning(f" Could not load setup helpers: {exc}") @@ -416,29 +353,21 @@ def _post_setup_xai_grok() -> None: ) if idx == 0: if _run_xai_oauth_login_from_setup(): - _print_success( - " Logged in — xAI will use these OAuth credentials" - ) + _print_success(" Logged in — xAI will use these OAuth credentials") else: - _print_warning( - " xAI Grok OAuth login did not complete. " - "Run later: hermes auth add xai-oauth" - ) + _print_warning(" xAI Grok OAuth login did not complete. Run later: hermes auth add xai-oauth") elif idx == 1: api_key = _setup_prompt(" xAI API key", password=True) if api_key: save_env_value("XAI_API_KEY", api_key) _print_success(" XAI_API_KEY saved") else: - _print_warning( - " No API key provided. Run later: hermes auth add xai-oauth" - ) + _print_warning(" No API key provided. Run later: hermes auth add xai-oauth") else: _print_info(" xAI will remain inactive until credentials are configured.") -# post_setup key -> hook. Unknown keys are a silent no-op (callers validate -# against valid_post_setup_keys()). +# post_setup key -> hook. Unknown keys are a silent no-op (callers validate against valid_post_setup_keys()). _POST_SETUP_HOOKS: dict = { "lightpanda": _post_setup_lightpanda, "agent_browser": lambda: _post_setup_agent_browser("agent_browser"), @@ -461,11 +390,9 @@ def _run_post_setup(post_setup_key: str): def valid_post_setup_keys() -> Set[str]: - """Return the set of post-setup keys declared by any visible provider. - - Collected from ``TOOL_CATEGORIES`` plus plugin-registered web/image/video/browser providers. - This is the allowlist the ``post-setup`` command and dashboard endpoint validate against, so a - caller cannot drive ``_run_post_setup`` with an arbitrary key. + """Return the set of post-setup keys declared by any visible provider (``TOOL_CATEGORIES`` plus + plugin-registered providers). This is the allowlist ``post-setup`` and the dashboard endpoint + validate against, so a caller cannot drive ``_run_post_setup`` with an arbitrary key. """ from hermes_cli.tools_config import ( TOOL_CATEGORIES, @@ -481,12 +408,9 @@ def valid_post_setup_keys() -> Set[str]: ps = prov.get("post_setup") if ps: keys.add(ps) - # Plugin-registered providers can declare their own post_setup hooks. for builder in ( - _plugin_web_search_providers, - _plugin_image_gen_providers, - _plugin_video_gen_providers, - _plugin_browser_providers, + _plugin_web_search_providers, _plugin_image_gen_providers, + _plugin_video_gen_providers, _plugin_browser_providers, ): try: for prov in builder(): @@ -499,10 +423,8 @@ def valid_post_setup_keys() -> Set[str]: def run_post_setup_command(args) -> int: - """``hermes tools post-setup `` — non-interactive post-setup runner. - - Stable, scriptable target the dashboard spawns so the GUI can drive backend setup without - re-implementing install logic. Returns a process exit code (0 ok, 2 unknown key). + """``hermes tools post-setup `` — non-interactive runner the dashboard spawns so the GUI can + drive backend setup without re-implementing install logic. Exit code: 0 ok, 2 unknown key. """ key = getattr(args, "post_setup_key", None) if not key: @@ -510,10 +432,7 @@ def run_post_setup_command(args) -> int: return 2 valid = valid_post_setup_keys() if key not in valid: - _print_error( - f"Unknown post-setup key: {key!r}. " - f"Valid keys: {', '.join(sorted(valid)) or '(none)'}" - ) + _print_error(f"Unknown post-setup key: {key!r}. Valid keys: {', '.join(sorted(valid)) or '(none)'}") return 2 _print_info(f"Running post-setup hook: {key}") try: @@ -527,20 +446,18 @@ def run_post_setup_command(args) -> int: # post_setup_key -> predicate(): True when the install side-effect is already satisfied. Used by # `_toolset_needs_configuration_prompt` to force provider setup when a no-key provider still needs a -# binary/dependency install (otherwise toggling the toolset on is a silent no-op that skips the hook). -# Only add an entry when the post_setup is the ONLY install side-effect for a no-key provider and the -# check is local, bounded, and import-light; other hooks keep their existing behaviour. +# binary/dependency install (otherwise toggling the toolset on silently skips the hook). Only add an +# entry when the post_setup is the ONLY install side-effect for a no-key provider and the check is +# local, bounded, and import-light. _POST_SETUP_INSTALLED: dict = { "cua_driver": lambda: _cua_driver_install_ready(), } def _post_setup_already_installed(post_setup_key: str) -> bool: - """Return True when the post_setup install side-effect is satisfied.""" + """Return True when the post_setup install side-effect is satisfied (or no check is registered).""" predicate = _POST_SETUP_INSTALLED.get(post_setup_key) if predicate is None: - # No install-state check registered → assume satisfied (don't - # change behaviour for hooks we haven't explicitly opted in). return True try: return bool(predicate()) @@ -558,11 +475,9 @@ def _module_installed(module_name: str) -> bool: return False -# Python dependencies installed explicitly through ``hermes tools`` are not -# part of the managed runtime's locked ``all`` sync. A runtime replacement -# therefore needs a small, static allowlist that can be snapshotted before the -# old site-packages disappears and restored afterward. Keep these install -# arguments in sync with the corresponding ``_run_post_setup`` branches. +# Python deps installed via ``hermes tools`` aren't in the managed runtime's locked ``all`` sync, so a +# runtime replacement snapshots this static allowlist before the old site-packages disappears and +# restores it afterward. Keep install args in sync with the ``_run_post_setup`` hooks. _RESTORABLE_PYTHON_TOOL_DEPENDENCIES: dict[str, tuple[str, tuple[str, ...]]] = { "faster_whisper": ("faster_whisper", ("-U", "faster-whisper")), "kittentts": ("kittentts", ("-U", _KITTENTTS_WHEEL_URL, "soundfile")), @@ -575,44 +490,33 @@ _RESTORABLE_PYTHON_TOOL_DEPENDENCIES: dict[str, tuple[str, tuple[str, ...]]] = { def active_restorable_python_tool_dependencies() -> list[str]: """Return ``hermes tools`` Python dependencies present in this runtime.""" return [ - name - for name, (module_name, _install_args) in ( - _RESTORABLE_PYTHON_TOOL_DEPENDENCIES.items() - ) + name for name, (module_name, _install_args) in _RESTORABLE_PYTHON_TOOL_DEPENDENCIES.items() if _module_installed(module_name) ] -def restorable_python_tool_dependency( - name: str, -) -> tuple[str, tuple[str, ...]] | None: +def restorable_python_tool_dependency(name: str) -> tuple[str, tuple[str, ...]] | None: """Return the import probe and pip arguments for an allowlisted tool.""" return _RESTORABLE_PYTHON_TOOL_DEPENDENCIES.get(name) def _agent_browser_installed() -> bool: """True when everything ``_run_post_setup("agent_browser")`` installs is present: the agent-browser - CLI *and* the Chromium build it drives (or the Lightpanda engine, which needs no Chromium). - Mirrors the hook so "Run setup" flips to an installed state only when re-running it would be a - no-op. + CLI *and* the Chromium build it drives (or the Lightpanda engine, which needs no Chromium), so + "Run setup" flips to installed only when re-running it would be a no-op. """ from hermes_cli.nous_subscription import _local_browser_runnable - # The install hook runs in a spawned ``hermes tools post-setup`` process, - # but this probe runs in the long-lived web-server/CLI process, whose - # browser_tool module may have cached a stale "Chromium missing" result - # from before the install. Drop the cache (when the module is loaded) so - # the readiness pill flips to Ready right after a successful setup run. + # The hook runs in a spawned process; this probe runs in the long-lived web-server/CLI process whose + # browser_tool may have cached a stale "Chromium missing" result. Drop the cache so the pill flips to Ready. bt = sys.modules.get("tools.browser_tool") if bt is not None: bt._cached_chromium_installed = None - return _local_browser_runnable() def _camofox_installed() -> bool: - """True when the Camofox npm package ``_run_post_setup("camofox")`` - installs is already in node_modules.""" + """True when the Camofox npm package ``_run_post_setup("camofox")`` installs is in node_modules.""" return (PROJECT_ROOT / "node_modules" / "@askjo" / "camofox-browser").exists() @@ -620,20 +524,15 @@ def _lightpanda_installed() -> bool: """True when a lightpanda binary is on PATH or in a known install dir.""" try: from tools.browser_lightpanda import find_lightpanda_binary - return find_lightpanda_binary() is not None except Exception: return False def _cloud_agent_browser_installed() -> bool: - """Installed-check for the ``browserbase`` hook (cloud provider rows). - - Cloud providers host their own Chromium, so their hook only installs the agent-browser npm - package — presence of the CLI is the whole contract. - """ + """Installed-check for the ``browserbase`` hook: cloud providers host their own Chromium, so + presence of the agent-browser CLI is the whole contract.""" from hermes_cli.nous_subscription import _has_agent_browser - return _has_agent_browser() @@ -649,4 +548,3 @@ _POST_SETUP_READY: dict = { "lightpanda": lambda: _lightpanda_installed(), "cua_driver": lambda: _cua_driver_install_ready(), } - diff --git a/hermes_cli/tools_config_providers.py b/hermes_cli/tools_config_providers.py index eb73eae5f4..0c0bf99082 100644 --- a/hermes_cli/tools_config_providers.py +++ b/hermes_cli/tools_config_providers.py @@ -21,10 +21,8 @@ from utils import base_url_hostname, is_truthy_value logger = logging.getLogger("hermes_cli.tools_config") -# NOTE: tools_config-internal names (TOOL_CATEGORIES, _cfg_section, _prompt_choice, -# get_nous_subscription_features, post-setup hooks, ...) are imported lazily inside the -# functions that need them: tools_config re-imports this module, and tests patch those -# names on ``hermes_cli.tools_config``. +# tools_config-internal names (TOOL_CATEGORIES, _cfg_section, _prompt_choice, ...) are imported lazily +# inside the functions that need them: tools_config re-imports this module, and tests patch those names there. def _plugin_registry(module: str): @@ -47,14 +45,11 @@ def _plugin_provider_rows( skip_builtin: bool = False, flatten_variants: bool = False, ) -> list[dict]: - """Build picker-row dicts from a plugin registry's providers. + """Picker-row dicts (TOOL_CATEGORIES-shaped) for a plugin registry's providers. - Each row looks like a hardcoded ``TOOL_CATEGORIES`` provider row plus the ``marker_keys`` - (all set to the provider's registry name) that route downstream config-writing / model-picker - code through the plugin registry. ``skip_builtin`` drops providers whose name shadows the - registry's ``_BUILTIN_NAMES`` (defence in depth). ``flatten_variants`` expands a schema's tier - ``variants`` (e.g. Exa/Parallel free keyless vs paid SDK) into separate rows sharing one - backend name, distinguished by ``web_tier``. + ``marker_keys`` are all set to the registry name so downstream config/model-picker code routes through + the registry. ``skip_builtin`` drops names shadowing ``_BUILTIN_NAMES``; ``flatten_variants`` expands a + schema's tier ``variants`` into separate rows sharing one backend name, distinguished by ``web_tier``. """ registry = _plugin_registry(registry_module) if registry is None: @@ -69,9 +64,7 @@ def _plugin_provider_rows( for provider in providers: if require_name: name = getattr(provider, "name", None) - if not name: - continue - if skip_builtin and name.lower().strip() in builtin: + if not name or (skip_builtin and name.lower().strip() in builtin): continue try: schema = provider.get_setup_schema() @@ -85,14 +78,9 @@ def _plugin_provider_rows( if flatten_variants: entries += [v for v in (schema.get("variants") or []) if isinstance(v, dict)] for entry in entries: - row = { - "name": entry.get("name", provider.display_name), - "badge": entry.get("badge", ""), - "tag": entry.get("tag", ""), - "env_vars": entry.get("env_vars", []), - } - for key in marker_keys: - row[key] = name + row = {"name": entry.get("name", provider.display_name), "badge": entry.get("badge", ""), + "tag": entry.get("tag", ""), "env_vars": entry.get("env_vars", [])} + row.update({key: name for key in marker_keys}) if flatten_variants and entry.get("web_tier"): row["web_tier"] = entry["web_tier"] if entry.get("post_setup"): @@ -101,12 +89,10 @@ def _plugin_provider_rows( return rows -# Category -> (registry module, marker keys, _plugin_provider_rows kwargs). Marker semantics: -# image/video ``*_plugin_name`` route config writes + model pickers through the registry (video -# has no in-tree backend, so this is the only row source); web ``web_backend`` + ``web_search_plugin_name`` -# (all bundled web providers are plugins; only the firecrawl setup-flow rows stay hardcoded); -# browser ``browser_provider`` is the legacy key written to ``browser.cloud_provider``; TTS rows -# render below the built-in rows and write ``tts.provider: `` like them. +# Category -> (registry module, marker keys, _plugin_provider_rows kwargs). image/video ``*_plugin_name`` route +# config writes + model pickers through the registry (video has no in-tree backend); all bundled web providers +# are plugins (only firecrawl setup-flow rows stay hardcoded); ``browser_provider`` is the legacy key written to +# ``browser.cloud_provider``; TTS rows render below the built-in rows and write ``tts.provider: ``. _PLUGIN_PROVIDER_ROW_SPECS = { "image_gen": ("agent.image_gen_registry", ("image_gen_plugin_name",), {"require_name": False}), "video_gen": ("agent.video_gen_registry", ("video_gen_plugin_name",), {"require_name": False}), @@ -147,11 +133,10 @@ def _plugin_tts_providers() -> list[dict]: def web_provider_capabilities(backend: str) -> list: - """Return the capabilities (``search`` / ``extract``) a web backend supports. + """Capabilities (``search`` / ``extract``) a web backend supports, per the registry provider instance. - Consults the plugin registry's provider instance (``supports_search`` / ``supports_extract``) so - the Capabilities GUI can offer per-capability selection (``web.search_backend`` / - ``web.extract_backend``) only where it makes sense — e.g. ddgs and brave-free are search-only. + Lets the Capabilities GUI offer ``web.search_backend`` / ``web.extract_backend`` only where it makes sense + (ddgs and brave-free are search-only). Unknown backend or registry failure -> both. """ try: from agent.web_search_registry import get_provider @@ -186,34 +171,24 @@ def _visible_providers( force_fresh: bool = False, features: Optional[NousSubscriptionFeatures] = None, ) -> list[dict]: - """Return provider entries visible for the current auth/config state. + """Provider entries visible for the current auth/config state. - Nous-managed Tool Gateway rows (``managed_nous_feature``) are always shown — even to logged-out - / unentitled users — so the picker advertises that the capability exists. + Nous-managed rows (``managed_nous_feature``) are always shown, even logged-out/unentitled, to advertise + the capability. """ from hermes_cli.tools_config import get_nous_subscription_features if features is None: features = get_nous_subscription_features(config, force_fresh=force_fresh) acct = features.account_info - # Pool-only users (free tool pool, no paid access) get image gen but NOT video gen — the pool - # doesn't fund `fal-video`, so hide the managed video row rather than advertise a denial. - # Logged-out users still see it (advertising) and paid users are entitled to it. - pool_only = bool( - acct - and acct.logged_in - and acct.paid_service_access is not True - and acct.tool_gateway_entitled - ) + # Pool-only users (free tool pool, no paid access) get image gen but NOT video gen — the pool doesn't + # fund `fal-video`, so hide the managed video row rather than advertise a denial. + pool_only = bool(acct and acct.logged_in and acct.paid_service_access is not True and acct.tool_gateway_entitled) visible = [] for provider in cat.get("providers", []): - # Managed rows stay visible regardless of auth (selecting one drives an inline Portal - # login); a `requires_nous_auth` row without a managed feature hides until logged in. - if ( - provider.get("requires_nous_auth") - and not provider.get("managed_nous_feature") - and not features.nous_auth_present - ): + # Managed rows stay visible regardless of auth (selecting one drives an inline Portal login); a + # `requires_nous_auth` row without a managed feature hides until logged in. + if provider.get("requires_nous_auth") and not provider.get("managed_nous_feature") and not features.nous_auth_present: continue if ( pool_only @@ -223,12 +198,10 @@ def _visible_providers( continue visible.append(provider) - # Plugin-registered rows render BELOW the hardcoded rows. For web and browser they are the - # only real provider rows — the hardcoded leftovers are non-provider UX setup flows. + # Plugin-registered rows render BELOW the hardcoded rows (for web/browser they are the only real provider rows). builder = _PLUGIN_ROW_BUILDERS.get(cat.get("name")) if builder is not None: visible.extend(builder()) - return visible @@ -239,11 +212,10 @@ def provider_readiness_status( features=None, is_active: Optional[bool] = None, ) -> str: - """Compute an honest readiness state for a provider picker row. + """Honest readiness state for a provider picker row. - ``features`` (a ``NousSubscriptionFeatures``) can be passed to avoid re-fetching portal state - per row. ``is_active`` is the completed-setup fallback signal for post_setup hooks with no - registered installed-check (selecting a row runs its hook, so the active row has been set up). + ``features`` avoids re-fetching portal state per row. ``is_active`` is the completed-setup fallback for + post_setup hooks with no registered installed-check (selecting a row runs its hook). """ from hermes_cli.tools_config import ( _POST_SETUP_READY, @@ -265,19 +237,11 @@ def provider_readiness_status( # Same per-category entitlement gate the CLI applies at selection time. acct = features.account_info category = MANAGED_FEATURE_COVERAGE_CATEGORY.get(managed_feature) - entitled = bool( - acct - and acct.logged_in - and ( - acct.tool_gateway_entitled_for(category) - if category - else acct.tool_gateway_entitled - ) - ) + entitled = bool(acct and acct.logged_in and ( + acct.tool_gateway_entitled_for(category) if category else acct.tool_gateway_entitled)) if not entitled: return "needs_auth" - # Signed in and entitled — fall through: a managed row may still carry a local install - # hook (e.g. the managed browser row needs the agent-browser CLI on this machine). + # Signed in and entitled — fall through: a managed row may still carry a local install hook. post_setup = provider.get("post_setup") if post_setup: @@ -288,8 +252,7 @@ def provider_readiness_status( try: return "ready" if predicate() else "needs_setup" except Exception: - # Flaky detection must not manufacture a warning state. - return "ready" + return "ready" # flaky detection must not manufacture a warning state # No installed-check registered → the active-provider signal means "setup completed". if is_active is None: is_active = _is_provider_active(provider, config) @@ -323,11 +286,9 @@ def _toolset_needs_configuration_prompt( if selection_key: section = config.get(ts_key, {}) return not isinstance(section, dict) or selection_key not in section - if ts_key == "image_gen": - # Satisfied by the in-tree FAL backend OR any available plugin image gen provider. + if ts_key == "image_gen": # in-tree FAL backend OR any available plugin image gen provider satisfies return not fal_key_is_configured() and not _any_plugin_provider_available("agent.image_gen_registry") - if ts_key == "video_gen": - # No in-tree fallback — every video backend is a plugin. + if ts_key == "video_gen": # no in-tree fallback — every video backend is a plugin return not _any_plugin_provider_available("agent.video_gen_registry") return not _toolset_has_keys(ts_key, config, force_fresh=force_fresh) @@ -360,9 +321,8 @@ def _configure_tool_category( ): """Provider selection for a tool category, then API-key setup for the chosen row. - ``reconfigure`` is the "Reconfigure an existing tool" flow: no setup note / skip row / Nous - marker, and the chosen provider goes through the key-update prompts instead of the new-enable - prompts. + ``reconfigure`` ("Reconfigure an existing tool"): no setup note / skip row / Nous marker, and the chosen + provider goes through the key-update prompts instead of the new-enable prompts. """ from hermes_cli.tools_config import _prompt_choice, _provider_env_ready, get_nous_subscription_features @@ -389,8 +349,7 @@ def _configure_tool_category( _print_info(f" {cat['setup_note']}") print() - # Logged-in Nous users get a marker on rows included in their subscription so it is obvious - # which options cost extra vs. nothing on top of Nous. + # Logged-in Nous users get a marker on rows included in their subscription (cost-extra vs. included). _nous_logged_in = False if not reconfigure: try: @@ -398,8 +357,7 @@ def _configure_tool_category( except Exception: _nous_logged_in = False - # Plain text labels only (no ANSI codes in menu items) - provider_choices = [] + provider_choices = [] # plain text labels only (no ANSI codes in menu items) for p in providers: badge = f" [{p['badge']}]" if p.get("badge") else "" tag = f" — {p['tag']}" if p.get("tag") else "" @@ -409,46 +367,35 @@ def _configure_tool_category( configured = " [active]" elif p.get("env_vars", []): configured = " [configured]" - # Logged-in subscribers get the "included" star; everyone else a "via Nous Portal" hint - # so it is clear selecting the row triggers a Portal login. + # Subscribers get the "included" star; everyone else a hint that selecting triggers a Portal login. sub_marker = "" if not reconfigure and p.get("managed_nous_feature"): - if _nous_logged_in: - sub_marker = " ★ Included with your Nous subscription" - else: - sub_marker = " ★ via Nous Portal (login on select)" + sub_marker = " ★ Included with your Nous subscription" if _nous_logged_in else " ★ via Nous Portal (login on select)" provider_choices.append(f"{p['name']}{badge}{tag}{configured}{sub_marker}") if not reconfigure: provider_choices.append("Skip — keep defaults / configure later") default_idx = _detect_active_provider_index(providers, config, force_fresh=force_fresh) - question = " Select provider:" if reconfigure else f" {title}:" provider_idx = _prompt_choice(question, provider_choices, default_idx) - if provider_idx >= len(providers): _print_info(f" Skipped {name}") return - _configure_provider(providers[provider_idx], config, force_fresh=force_fresh, reconfigure=reconfigure) def _web_tier_matches(provider: dict, config: dict) -> bool: - """Return True when a web picker row's tier matches the configured tier. + """True when a web picker row's tier matches the configured tier (``web.provider_tier.``). - Tiered rows (Exa/Parallel Free vs Paid) share one ``web_backend`` name and differ only in - ``web_tier``; the configured tier lives at ``web.provider_tier.`` (set on selection). - Rules: no ``web_tier`` on the row → tier-agnostic, matches; configured tier set → must equal the - row's tier; unset → "auto": paid when the row's key is present, free otherwise (highlight the - row the runtime would actually use). + Tiered rows (Exa/Parallel Free vs Paid) share one ``web_backend`` and differ only in ``web_tier``. No + ``web_tier`` on the row → matches; configured tier set → must equal the row's tier; unset → "auto": + paid when the row's key is present, free otherwise (highlight the row the runtime would actually use). """ row_tier = provider.get("web_tier") if not row_tier: return True - web_cfg = config.get("web") - if not isinstance(web_cfg, dict): - web_cfg = {} + web_cfg = config.get("web") if isinstance(config.get("web"), dict) else {} tiers = web_cfg.get("provider_tier") if not isinstance(tiers, dict): tiers = {} @@ -495,8 +442,8 @@ def _managed_provider_active(provider: dict, config: dict, managed_feature: str, current = cfg_get(config, "stt", "provider") return feature.managed_by_nous and current in {provider["stt_provider"], NOUS_MANAGED_PROVIDER} if "browser_provider" in provider: - # Browser Use mode is a driver on top of the provider (attaches to its CDP endpoint), so - # the provider row stays active alongside the Browser Use row. + # Browser Use mode is a driver on top of the provider (attaches to its CDP endpoint), so the + # provider row stays active alongside the Browser Use row. current = cfg_get(config, "browser", "cloud_provider") return feature.managed_by_nous and current in {provider["browser_provider"], NOUS_MANAGED_PROVIDER} if provider.get("web_backend"): @@ -510,8 +457,8 @@ def _managed_provider_active(provider: dict, config: dict, managed_feature: str, def _browser_use_default_active(config: dict) -> bool: - """``browser.backend`` unset: Browser Use mode is the default, so the row is active whenever - the effective mode resolves on (legacy direct-API cloud config, or CLI runnable and no Camofox).""" + """``browser.backend`` unset: Browser Use mode is the default, so the row is active whenever the + effective mode resolves on (legacy direct-API cloud config, or CLI runnable and no Camofox).""" browser_cfg = config.get("browser") if isinstance(config, dict) else None try: from tools.browser_use_cli import _find_cli, is_legacy_browser_use_cloud_config @@ -540,12 +487,11 @@ def _stt_active(provider: dict, config: dict) -> bool: def _browser_provider_active(provider: dict, config: dict) -> bool: - # Browser Use mode composes with the provider (driver over its CDP endpoint) — don't - # deactivate the provider row. + # Browser Use mode composes with the provider (driver over its CDP endpoint) — don't deactivate the row. if provider["browser_provider"] != cfg_get(config, "browser", "cloud_provider"): return False - # Two local rows differ only by engine ("Local Browser" vs "Lightpanda"): config.yaml is the - # picker's source of truth here, the AGENT_BROWSER_ENGINE env var is not consulted. + # Two local rows differ only by engine ("Local Browser" vs "Lightpanda"): config.yaml is the picker's + # source of truth here, the AGENT_BROWSER_ENGINE env var is not consulted. if provider.get("browser_engine"): engine = str(cfg_get(config, "browser", "engine") or "auto").strip().lower() return engine == provider["browser_engine"] @@ -566,9 +512,7 @@ def _browser_backend_active(provider: dict, config: dict) -> bool: def _web_backend_active(provider: dict, config: dict) -> bool: - if cfg_get(config, "web", "backend") != provider["web_backend"]: - return False - return _web_tier_matches(provider, config) + return cfg_get(config, "web", "backend") == provider["web_backend"] and _web_tier_matches(provider, config) def _computer_use_active(provider: dict, config: dict) -> bool: @@ -577,18 +521,16 @@ def _computer_use_active(provider: dict, config: dict) -> bool: def _imagegen_backend_active(provider: dict, config: dict) -> bool: image_cfg = config.get("image_gen", {}) - if not isinstance(image_cfg, dict): - return False return ( - provider["imagegen_backend"] == "fal" + isinstance(image_cfg, dict) + and provider["imagegen_backend"] == "fal" and image_cfg.get("provider") in {None, "", "fal"} and not is_truthy_value(image_cfg.get("use_gateway"), default=False) ) -# Non-managed active checks, evaluated in order; the first marker the row carries decides. -# ``browser_provider`` is a membership test (a local row carries ``browser_provider: ""``), -# every other marker is a truthiness test. +# Non-managed active checks, evaluated in order; the first marker the row carries decides. ``browser_provider`` +# is a membership test (a local row carries ``browser_provider: ""``), every other marker is a truthiness test. _ACTIVE_CHECKS: tuple[tuple[str, Callable[[dict, dict], bool]], ...] = ( ("tts_provider", _tts_active), ("stt_provider", _stt_active), @@ -611,8 +553,8 @@ def _is_provider_active( managed_feature = provider.get("managed_nous_feature") plugin_name = provider.get("image_gen_plugin_name") if plugin_name and not managed_feature: - # Managed entries fall through to the managed branch, which also checks use_gateway — - # otherwise a managed FAL pick and a direct-key FAL pick would both report active. + # Managed entries fall through to the managed branch, which also checks use_gateway — otherwise a + # managed FAL pick and a direct-key FAL pick would both report active. image_cfg = config.get("image_gen", {}) if not (isinstance(image_cfg, dict) and image_cfg.get("provider") == plugin_name): return False @@ -646,40 +588,27 @@ def _detect_active_provider_index( for i, p in enumerate(providers): if _is_provider_active(p, config, force_fresh=force_fresh): return i - # Fallback: env vars present → likely configured - if p.get("env_vars", []) and _provider_env_ready(p): + if p.get("env_vars", []) and _provider_env_ready(p): # fallback: env vars present → likely configured return i return 0 -# ─── Image Generation Model Pickers ─────────────────────────────────────────── -# -# IMAGEGEN_BACKENDS: per-backend catalog (config_key = top-level config.yaml section, catalog_fn -> -# ({model_id: metadata}, default_model)). A TOOL_CATEGORIES row tagged `imagegen_backend: ""` -# selects the catalog at picker time. - - def _fal_model_catalog(): """Lazy-load the FAL model catalog from the tool module.""" from tools.image_generation_tool import FAL_MODELS, DEFAULT_MODEL return FAL_MODELS, DEFAULT_MODEL +# Per-backend model catalog (config_key = top-level config.yaml section, catalog_fn -> ({model_id: metadata}, +# default_model)); a TOOL_CATEGORIES row tagged `imagegen_backend: ""` selects the catalog at picker time. IMAGEGEN_BACKENDS = { - "fal": { - "display": "FAL.ai", - "config_key": "image_gen", - "catalog_fn": _fal_model_catalog, - }, + "fal": {"display": "FAL.ai", "config_key": "image_gen", "catalog_fn": _fal_model_catalog}, } def _plugin_model_catalog(registry_module: str, plugin_name: str): - """Return ``(catalog_dict, default_model_id)`` for a plugin provider. - - ``catalog_dict`` is shaped like the legacy ``FAL_MODELS`` table so the picker code path is - shared. Returns ``({}, None)`` if the provider is unregistered or has no models. - """ + """``(catalog_dict, default_model_id)`` for a plugin provider; ``catalog_dict`` is shaped like the legacy + ``FAL_MODELS`` table so the picker path is shared. ``({}, None)`` if unregistered or no models.""" registry = _plugin_registry(registry_module) if registry is None: return {}, None @@ -719,10 +648,9 @@ def _pick_model_from_catalog( ) -> None: """Column-aligned model picker shared by the FAL, plugin image gen and video gen flows. - Writes the choice to ``config[cfg_key]["model"]``. The current model is listed first so the - cursor lands on it; a saved model that belongs to another provider (shared config key) or a - drifted catalog default never indexes the catalog. Safe when stdin is not a TTY — - curses_radiolist keeps the current selection. + Writes the choice to ``config[cfg_key]["model"]``. The current model is listed first so the cursor lands + on it; a saved model belonging to another provider (shared config key) or a drifted catalog default never + indexes the catalog. Safe when stdin is not a TTY — curses_radiolist keeps the current selection. """ from hermes_cli.tools_config import _cfg_section, _prompt_choice @@ -735,7 +663,6 @@ def _pick_model_from_catalog( model_ids = list(catalog.keys()) ordered = [current_model] + [m for m in model_ids if m != current_model] - widths = { "model": max(len(m) for m in model_ids), "speed": max((len(catalog[m].get("speed", "")) for m in model_ids), default=6), @@ -743,29 +670,20 @@ def _pick_model_from_catalog( } print() - header = ( - f" {'Model':<{widths['model']}} " - f"{'Speed':<{widths['speed']}} " - f"{'Strengths':<{widths['strengths']}} " - f"Price" - ) + header = (f" {'Model':<{widths['model']}} {'Speed':<{widths['speed']}} " + f"{'Strengths':<{widths['strengths']}} Price") print(color(header, Colors.CYAN)) rows = [] for mid in ordered: meta = catalog[mid] - row = ( - f"{row_indent}{mid:<{widths['model']}} " - f"{meta.get('speed', ''):<{widths['speed']}} " - f"{meta.get('strengths', ''):<{widths['strengths']}} " - f"{meta.get('price', '')}" - ) + row = (f"{row_indent}{mid:<{widths['model']}} {meta.get('speed', ''):<{widths['speed']}} " + f"{meta.get('strengths', ''):<{widths['strengths']}} {meta.get('price', '')}") if mid == current_model: row += " ← currently in use" rows.append(row) idx = _prompt_choice(f" Choose {display} model:", rows, default=0) - chosen = ordered[idx] cur_cfg["model"] = chosen _print_success(f" Model set to: {chosen}") @@ -805,15 +723,9 @@ def _configure_xai_imagine_storage(section_name: str, config: dict) -> None: " xAI Imagine can store generated media and create reusable public URLs. " "xAI may bill for stored files and public URL hosting." ) - idx = _prompt_choice( - " Stored public URLs:", - [ - "Enable public URLs without automatic expiry (recommended)", - "Disable stored public URLs", - "Enable public URLs for 2 days", - ], - default=0, - ) + choices = ["Enable public URLs without automatic expiry (recommended)", "Disable stored public URLs", + "Enable public URLs for 2 days"] + idx = _prompt_choice(" Stored public URLs:", choices, default=0) if idx == 1: storage_cfg["enabled"] = False _print_success(" xAI stored public URLs disabled") @@ -830,11 +742,10 @@ def _configure_xai_imagine_storage(section_name: str, config: dict) -> None: def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *, use_gateway: bool) -> None: - """Persist a plugin-backed image/video generation provider selection and run its model picker. + """Persist a plugin-backed image/video gen provider selection and run its model picker. - ``use_gateway=True`` (Nous-managed pick) stores ``
.provider: nous``; BYOK picks store - the plugin name. Any legacy ``use_gateway`` key is removed so old read-time shims cannot - override the fresh selection. + ``use_gateway=True`` (Nous-managed pick) stores ``
.provider: nous``; BYOK picks store the plugin + name. Any legacy ``use_gateway`` key is removed so old read-time shims cannot override the selection. """ from hermes_cli.tools_config import _cfg_section @@ -858,9 +769,9 @@ def _select_plugin_video_gen_provider(plugin_name: str, config: dict, *, use_gat _select_plugin_gen_provider("video_gen", plugin_name, config, use_gateway=use_gateway) -# Per-provider STT model catalogs for the picker; keys are ``stt.`` sections, first entry -# is the default. Kept in sync with the dashboard selects (web_server _CONFIG_FIELD_META) and the -# desktop settings enums (apps/desktop/src/app/settings/constants.ts). +# Per-provider STT model catalogs for the picker; keys are ``stt.`` sections, first entry is the +# default. Kept in sync with the dashboard selects (web_server _CONFIG_FIELD_META) and the desktop settings +# enums (apps/desktop/src/app/settings/constants.ts). STT_MODEL_CATALOG = { "local": ["base", "tiny", "small", "medium", "large-v3"], "groq": ["whisper-large-v3-turbo", "whisper-large-v3", "distil-whisper-large-v3-en"], @@ -906,11 +817,10 @@ _PROVIDER_MARKER_SECTIONS = { def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> None: """Persist the provider/backend config keys for a selected provider. - Pure, non-interactive core of :func:`_configure_provider`: no env prompts, post-setup hooks, - Nous auth gating or model pickers. Both the CLI and the GUI ``PUT .../provider`` endpoint call - through here so there is one code path. Each pick writes exactly ONE provider string per - category (``nous`` for managed rows) and removes any legacy ``use_gateway`` key so the read-time - shim cannot override the new choice. + Pure, non-interactive core of :func:`_configure_provider` (no env prompts, post-setup hooks, Nous auth + gating or model pickers) shared by the CLI and the GUI ``PUT .../provider`` endpoint. Each pick writes + exactly ONE provider string per category (``nous`` for managed rows) and removes any legacy + ``use_gateway`` key so the read-time shim cannot override the new choice. """ from hermes_cli.tools_config import TOOL_CATEGORIES, _cfg_section @@ -932,15 +842,13 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> bp = provider["browser_provider"] browser_cfg = config.setdefault("browser", {}) if bp or managed_feature: - # Browser Use mode (browser.backend) composes with the provider — switching providers - # keeps the driver choice intact. + # Browser Use mode (browser.backend) composes with the provider — keep the driver choice intact. _set_selection("browser", "cloud_provider", bp) else: browser_cfg.pop("use_gateway", None) if provider.get("browser_backend"): config.setdefault("browser", {})["backend"] = provider["browser_backend"] - # Local engine rows ("Local Browser" resets to auto, "Lightpanda" sets lightpanda). Composes - # with browser.backend like the provider does. + # Local engine rows ("Local Browser" resets to auto, "Lightpanda" sets lightpanda); composes with browser.backend. if provider.get("browser_engine"): config.setdefault("browser", {})["engine"] = provider["browser_engine"] @@ -948,30 +856,28 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> _set_selection("web", "backend", provider["web_backend"]) web_cfg = config.get("web") if isinstance(web_cfg, dict): - if provider.get("web_tier"): - tiers = web_cfg.setdefault("provider_tier", {}) - if isinstance(tiers, dict): - tiers[provider["web_backend"]] = provider["web_tier"] - else: - stale_tiers = web_cfg.get("provider_tier") - if isinstance(stale_tiers, dict): - stale_tiers.pop(provider["web_backend"], None) + tier = provider.get("web_tier") + tiers = web_cfg.setdefault("provider_tier", {}) if tier else web_cfg.get("provider_tier") + if isinstance(tiers, dict): + if tier: + tiers[provider["web_backend"]] = tier + else: + tiers.pop(provider["web_backend"], None) if provider.get("computer_use_backend"): config.setdefault("computer_use", {})["backend"] = provider["computer_use_backend"] if managed_feature and managed_feature not in {"web", "tts", "stt", "browser"}: - # Managed rows for categories without a marker above (image_gen/video_gen "Nous - # Subscription" rows carry only managed_nous_feature) still persist the "nous" selection. + # Managed rows without a marker above (image_gen/video_gen "Nous Subscription" rows carry only + # managed_nous_feature) still persist the "nous" selection. section = config.setdefault(managed_feature, {}) if isinstance(section, dict): section["provider"] = NOUS_MANAGED_PROVIDER _drop_use_gateway(section) elif not managed_feature: - # Non-gateway pick — clear any stale legacy use_gateway key on the category so the - # read-time shim cannot override it. Resolve the category from the row's own markers first - # (plugin-injected rows are NOT in TOOL_CATEGORIES' hardcoded lists), then fall back to the - # category-membership walk. + # Non-gateway pick — clear any stale legacy use_gateway key on the category. Resolve the category from + # the row's own markers first (plugin-injected rows are NOT in TOOL_CATEGORIES' hardcoded lists), then + # fall back to the category-membership walk. sections = [section_key for marker, section_key in _PROVIDER_MARKER_SECTIONS.items() if marker in provider] if not sections: sections = [cat_key for cat_key, cat in TOOL_CATEGORIES.items() if provider in cat.get("providers", [])][:1] @@ -980,12 +886,9 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> None: - """Non-interactively persist a provider selection for a toolset. - - Resolves ``provider_name`` among the rows the picker shows (:func:`_visible_providers`) and - writes the config keys only — API keys, post-setup hooks, auth gating and model pickers are - handled by separate GUI endpoints. Raises ``KeyError`` for an unknown toolset or provider. - """ + """Non-interactively persist a provider selection for a toolset (config keys only — API keys, post-setup + hooks, auth gating and model pickers are separate GUI endpoints). ``provider_name`` is resolved among + :func:`_visible_providers` rows; raises ``KeyError`` for an unknown toolset or provider.""" from hermes_cli.tools_config import TOOL_CATEGORIES, _cfg_section cat = TOOL_CATEGORIES.get(ts_key) @@ -1000,10 +903,9 @@ def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> N managed_feature = provider.get("managed_nous_feature") _write_provider_config(provider, config, managed_feature=managed_feature) - # Plugin image/video gen backends record the provider name in their own section (model choice - # is a separate GUI flow); managed picks store "nous". The in-tree FAL BYOK row always persists - # an explicit ``image_gen.provider: fal`` so a deliberate pick is distinguishable from a - # never-configured install. + # Plugin image/video gen backends record the provider name in their own section (model choice is a separate + # GUI flow); managed picks store "nous". The in-tree FAL BYOK row always persists an explicit + # ``image_gen.provider: fal`` so a deliberate pick is distinguishable from a never-configured install. selections = [ ("image_gen", provider.get("image_gen_plugin_name")), ("video_gen", provider.get("video_gen_plugin_name")), @@ -1019,10 +921,9 @@ def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> N def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_fresh: bool) -> bool: """Return False (after printing why) when a Nous-gated row cannot be selected. - Managed Tool Gateway rows are always listed (see ``_visible_providers``) but only *activate* - with paid Nous Portal access — selecting one runs an inline Portal login (auth + entitlement - only, no inference-provider switch). Pure pre-auth UX rows (``requires_nous_auth`` without a - managed feature) keep the older logged-in + entitled gate. + Managed Tool Gateway rows are always listed but only *activate* with paid Nous Portal access — selecting + one runs an inline Portal login (auth + entitlement only, no inference-provider switch). Pure pre-auth UX + rows (``requires_nous_auth`` without a managed feature) keep the older logged-in + entitled gate. """ from hermes_cli.tools_config import get_nous_subscription_features @@ -1042,8 +943,7 @@ def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_ entitled = bool(features.account_info and features.account_info.paid_service_access is True) if not features.nous_auth_present or not entitled: message = format_nous_portal_entitlement_message( - features.account_info, - capability=f"{provider.get('name', 'Nous Subscription')}", + features.account_info, capability=f"{provider.get('name', 'Nous Subscription')}" ) _print_warning(f" {message or 'Nous Subscription is only available after logging into Nous Portal.'}") return False @@ -1065,8 +965,8 @@ def _finish_provider_selection(provider: dict, config: dict, managed_feature) -> backend = provider.get("imagegen_backend") if backend: _configure_imagegen_model(backend, config) - # In-tree FAL is the only non-plugin backend. Persist "nous" for a managed row, "fal" for - # BYOK, and drop legacy use_gateway — never clobber a managed pick back onto direct keys. + # In-tree FAL is the only non-plugin backend: "nous" for a managed row, "fal" for BYOK, drop legacy + # use_gateway — never clobber a managed pick back onto direct keys. img_cfg = _cfg_section(config, "image_gen") img_cfg["provider"] = NOUS_MANAGED_PROVIDER if managed_feature else "fal" img_cfg.pop("use_gateway", None) @@ -1103,8 +1003,8 @@ def _print_provider_selection(provider: dict, managed_feature, *, reconfigure: b def _show_portal_hint(provider: dict, config: dict, managed_feature, force_fresh: bool) -> bool: - """True when a BYOK row shares its category with a Nous-managed sibling and the user is not - authed to Nous — a single dim hint tells them the key is avoidable via a Portal subscription.""" + """True when a BYOK row shares its category with a Nous-managed sibling and the user is not authed to + Nous — a single dim hint tells them the key is avoidable via a Portal subscription.""" from hermes_cli.tools_config import TOOL_CATEGORIES, get_nous_subscription_features if managed_feature or provider.get("requires_nous_auth"): @@ -1120,10 +1020,10 @@ def _show_portal_hint(provider: dict, config: dict, managed_feature, force_fresh def _prompt_env_vars(env_vars: list, *, reconfigure: bool) -> bool: - """Prompt for a provider's env vars; returns True when every key ended up configured. + """Prompt for a provider's env vars; True when every key ended up configured. - Reconfigure mode re-prompts every key ("Enter to keep current") and always returns True; the - new-enable flow keeps already-set keys without asking and reports False on any skipped key. + Reconfigure mode re-prompts every key ("Enter to keep current") and always returns True; the new-enable + flow keeps already-set keys without asking and reports False on any skipped key. """ all_configured = True for var in env_vars: @@ -1171,9 +1071,8 @@ def _configure_provider( ): """Configure a single provider - prompt for API keys and set config. - ``reconfigure=False`` is the new-enable flow: already-set keys are kept without asking, and - the post-setup hook only runs when every key was provided. ``reconfigure=True`` re-prompts - every key ("Enter to keep current") and always runs the post-setup hook. + ``reconfigure=False`` (new-enable): already-set keys are kept without asking and the post-setup hook only + runs when every key was provided. ``reconfigure=True`` re-prompts every key and always runs the hook. """ from hermes_cli.tools_config import _run_post_setup @@ -1184,8 +1083,7 @@ def _configure_provider( return _print_provider_selection(provider, managed_feature, reconfigure=reconfigure) - # Shared with the GUI provider-select endpoint (apply_provider_selection): single source of - # truth for the config-key writes. + # Shared with the GUI provider-select endpoint (apply_provider_selection): one source of truth for config writes. _write_provider_config(provider, config, managed_feature=managed_feature) if not env_vars: @@ -1201,7 +1099,6 @@ def _configure_provider( _print_info(" Available through Nous Portal subscription.") all_configured = _prompt_env_vars(env_vars, reconfigure=reconfigure) - if provider.get("post_setup") and all_configured: _run_post_setup(provider["post_setup"]) if all_configured: @@ -1216,12 +1113,10 @@ def _reconfigure_provider(provider: dict, config: dict, *, force_fresh: bool = T def _configure_vision_backend() -> None: - """Interactive vision-backend configuration. + """Interactive vision-backend configuration (``auxiliary.vision.{provider,model,base_url}``). - Vision resolves from ``auxiliary.vision.{provider,model,base_url}``. Rather than forcing - OpenRouter, offer any authenticated provider + model (same surface as ``hermes model``) or a - custom endpoint. "Auto" leaves the keys empty so the resolver uses the main-model fallback - chain. + Offers any authenticated provider + model (same surface as ``hermes model``) or a custom endpoint rather + than forcing OpenRouter. "Auto" leaves the keys empty so the resolver uses the main-model fallback chain. """ from hermes_cli.tools_config import _cfg_section, _prompt_choice @@ -1263,9 +1158,9 @@ def _configure_vision_backend() -> None: default_model = "gpt-4o-mini" if is_native_openai else "" model = _prompt(f" Vision model{f' (blank for {default_model})' if default_model else ''}").strip() or default_model save_env_value("OPENAI_API_KEY", api_key.strip()) - # Only base_url + model go to config.yaml; the key is the secret. Pin provider="custom" so - # the resolver routes through this endpoint — at the "auto" default - # _resolve_task_provider_model ignores base_url unless paired with a config api_key. + # Only base_url + model go to config.yaml; the key is the secret. Pin provider="custom" so the resolver + # routes through this endpoint — at the "auto" default _resolve_task_provider_model ignores base_url + # unless paired with a config api_key. vision_cfg["provider"] = "custom" vision_cfg["base_url"] = base_url if model: @@ -1276,16 +1171,15 @@ def _configure_vision_backend() -> None: _print_success(f" Vision set to custom endpoint{f' ({model})' if model else ''}") return - # Skip _print_info(" Skipped vision configuration") def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None: """Provider + model picker for vision, mirroring the ``/model`` surface. - Rows come from ``build_aux_picker_rows()`` so this lists exactly what the ``hermes model`` - aux-task picker lists, including user-defined ``providers:`` / ``custom_providers:`` endpoints. - Persists ``auxiliary.vision.provider`` + ``.model``. + Rows come from ``build_aux_picker_rows()`` so this lists exactly what the ``hermes model`` aux-task picker + lists, including user-defined ``providers:`` / ``custom_providers:`` endpoints. Persists + ``auxiliary.vision.provider`` + ``.model``. """ from hermes_cli.tools_config import _prompt_choice @@ -1300,12 +1194,8 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None: current_base_url = str(vision_cfg.get("base_url") or "").strip() try: - providers = build_aux_picker_rows( - current_provider=current_provider, - current_model=current_model, - current_base_url=current_base_url, - max_models=40, - ) + providers = build_aux_picker_rows(current_provider=current_provider, current_model=current_model, + current_base_url=current_base_url, max_models=40) except Exception as exc: _print_warning(f" Could not detect providers: {exc}") providers = [] @@ -1317,14 +1207,8 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None: ) return - provider_labels = [ - label - for _slug, label, _models in format_aux_picker_entries( - providers, - current_provider=current_provider, - current_base_url=current_base_url, - ) - ] + provider_labels = [label for _slug, label, _models in format_aux_picker_entries( + providers, current_provider=current_provider, current_base_url=current_base_url)] provider_labels.append("Cancel") pidx = _prompt_choice(" Choose vision provider:", provider_labels, 0) @@ -1335,7 +1219,6 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None: chosen = providers[pidx] slug = chosen.get("slug") models = list(chosen.get("models", [])) - model_choices = list(models) + ["Type a custom model id…"] midx = _prompt_choice(f" Choose vision model for {chosen.get('name') or slug}:", model_choices, 0) if midx < len(models): @@ -1358,9 +1241,8 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None: def _configure_simple_requirements(ts_key: str, *, reconfigure: bool = False): """Fallback for toolsets that just need env vars (no provider selection). - Vision has its own provider/model picker (any provider, like ``hermes model``) — run it - directly so neither flow falls back to the generic single-key prompt (which would re-ask for - OPENROUTER_API_KEY). + Vision has its own provider/model picker — run it directly so neither flow falls back to the generic + single-key prompt (which would re-ask for OPENROUTER_API_KEY). """ from hermes_cli.tools_config import TOOLSET_ENV_REQUIREMENTS, _toolset_has_keys, _toolset_label @@ -1394,10 +1276,7 @@ def _configure_simple_requirements(ts_key: str, *, reconfigure: bool = False): if value and value.strip(): save_env_value(var, value.strip()) _print_success(" Updated" if reconfigure else " Saved") + elif reconfigure: + _print_info(" Kept current") else: - if reconfigure: - _print_info(" Kept current") - else: - _print_warning(" Skipped") - - + _print_warning(" Skipped")