diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index 0468b96916..c8414f1af4 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -6,24 +6,18 @@ import os from pathlib import Path from typing import Dict, List, Optional, Set -from hermes_cli.config import cfg_get, load_config, save_config, get_env_value +from hermes_cli.cli_output import print_info as _print_info from hermes_cli.colors import Colors, color +from hermes_cli.config import cfg_get, load_config, save_config, get_env_value from hermes_cli.nous_subscription import ( NousSubscriptionFeatures, apply_nous_managed_defaults, get_nous_subscription_features, ) +from hermes_cli.platforms import PLATFORMS as _PLATFORMS_REGISTRY from hermes_cli.toolset_scope import ( _TOOLSET_PLATFORM_RESTRICTIONS, toolset_allowed_for_platform as _toolset_allowed_for_platform, ) - -logger = logging.getLogger(__name__) - -# Platforms already warned about an all-invalid platform_toolsets list (warn once, not per resolution). -_warned_invalid_platform_toolsets: Set[str] = set() - -PROJECT_ROOT = Path(__file__).parent.parent.resolve() - -from hermes_cli.cli_output import print_info as _print_info # noqa: E402 — late import block -from hermes_cli.tools_config_cua import ( # noqa: F401 — re-exported for hermes_cli.tools_config.X callers and test patches +# Re-exports: keep ``hermes_cli.tools_config.X`` callers and test patch targets resolving. +from hermes_cli.tools_config_cua import ( # noqa: F401 _post_setup_no_window_flags, _cua_driver_cmd, _cua_version_summary, _resolved_cua_driver_cmd, _cua_driver_env, _CUA_DRIVER_CONTRACT_CACHE, _cua_driver_contract_status, _cua_driver_install_ready, _pip_install, _cua_install_target_writable, _cua_driver_version, install_cua_driver, @@ -34,7 +28,7 @@ from hermes_cli.tools_config_cua import ( # noqa: F401 — re-exported for herm _repair_cua_driver_autostart_windows, _remove_quietly, _print_cua_platform_notes, _run_cua_driver_installer, ) -from hermes_cli.tools_config_post_setup import ( # noqa: F401 — re-exported for hermes_cli.tools_config.X callers and test patches +from hermes_cli.tools_config_post_setup import ( # noqa: F401 _ensure_browser_use_cli, _post_setup_lightpanda, _post_setup_agent_browser, _post_setup_camofox, _KITTENTTS_WHEEL_URL, _PIP_POST_SETUP_HOOKS, _post_setup_pip, _post_setup_spotify, _post_setup_langfuse, _post_setup_xai_grok, _POST_SETUP_HOOKS, _run_post_setup, valid_post_setup_keys, run_post_setup_command, @@ -43,7 +37,7 @@ from hermes_cli.tools_config_post_setup import ( # noqa: F401 — re-exported f restorable_python_tool_dependency, _agent_browser_installed, _camofox_installed, _lightpanda_installed, _cloud_agent_browser_installed, _POST_SETUP_READY, ) -from hermes_cli.tools_config_providers import ( # noqa: F401 — re-exported for hermes_cli.tools_config.X callers and test patches +from hermes_cli.tools_config_providers import ( # noqa: F401 _plugin_provider_rows, _plugin_image_gen_providers, _plugin_video_gen_providers, _plugin_web_search_providers, _plugin_browser_providers, _plugin_tts_providers, web_provider_capabilities, _PLUGIN_ROW_BUILDERS, _visible_providers, provider_readiness_status, _toolset_needs_configuration_prompt, @@ -57,12 +51,22 @@ from hermes_cli.tools_config_providers import ( # noqa: F401 — re-exported fo _finish_provider_selection, _print_provider_selection, _configure_provider, _reconfigure_provider, _configure_vision_backend, _configure_vision_provider_model, _configure_simple_requirements, ) -from hermes_cli.tools_config_mcp import ( # noqa: F401 — re-exported for hermes_cli.tools_config.X callers and test patches +from hermes_cli.tools_config_mcp import ( # noqa: F401 _mcp_match_filter, _mcp_preselected, _apply_mcp_checklist, _configure_mcp_tools_interactive, _apply_toolset_change, _apply_mcp_change, _print_tools_list, _known_tool_platforms, tools_disable_enable_command, ) +logger = logging.getLogger(__name__) + +# Platforms already warned about an all-invalid platform_toolsets list (warn once, not per resolution). +_warned_invalid_platform_toolsets: Set[str] = set() + +PROJECT_ROOT = Path(__file__).parent.parent.resolve() + +# Platform display config derived from the canonical registry (dict-of-dicts for ``PLATFORMS[key]["label"]``). +PLATFORMS = {k: {"label": info.label, "default_toolset": info.default_toolset} for k, info in _PLATFORMS_REGISTRY.items()} + # --- Toolset Registry --- # Toolsets shown in the configurator: (toolset key in toolsets.py TOOLSETS, label, description). CONFIGURABLE_TOOLSETS = [ @@ -99,10 +103,8 @@ def gui_toolset_label(label: str) -> str: """Strip the leading ```` from a toolset title for GUI surfaces (plugins prefix ``🔌``). CLI/TUI keeps the raw label — only HTTP APIs call this.""" text = (label or "").strip() - if not text: - return text parts = text.split(None, 1) - if len(parts) == 2 and parts[0] and not any(ch.isascii() and ch.isalnum() for ch in parts[0]): + if len(parts) == 2 and not any(ch.isascii() and ch.isalnum() for ch in parts[0]): return parts[1].strip() return text @@ -136,7 +138,7 @@ def _xai_credentials_present() -> bool: try: from agent.secret_scope import get_secret except ImportError: # pragma: no cover — secret_scope is in-repo - return bool(str(os.environ.get("XAI_API_KEY") or "").strip()) + get_secret = os.environ.get return bool(str(get_secret("XAI_API_KEY") or "").strip()) @@ -153,9 +155,7 @@ def _toolset_configuration_platform(ts_key: str, default: str = "cli") -> str: """Platform a platform-less configuration UI should target: a toolset restricted away from ``default`` must be configured on a supported platform, else the save helper drops it and the UI reports a no-op.""" allowed = _TOOLSET_PLATFORM_RESTRICTIONS.get(ts_key) - if not allowed or default in allowed: - return default - return sorted(allowed)[0] + return default if not allowed or default in allowed else sorted(allowed)[0] def _get_effective_configurable_toolsets(): @@ -166,10 +166,9 @@ def _get_effective_configurable_toolsets(): from hermes_cli.plugins import discover_plugins, get_plugin_toolsets discover_plugins() # idempotent — ensures plugins are loaded for entry in get_plugin_toolsets(): - if entry[0] in seen: - continue - seen.add(entry[0]) - result.append(entry) + if entry[0] not in seen: + seen.add(entry[0]) + result.append(entry) except Exception: pass return result @@ -196,12 +195,6 @@ def _checklist_toolset_keys(platform: str) -> Set[str]: } -# Platform display config derived from the canonical registry (dict-of-dicts for ``PLATFORMS[key]["label"]``). -from hermes_cli.platforms import PLATFORMS as _PLATFORMS_REGISTRY - -PLATFORMS = {k: {"label": info.label, "default_toolset": info.default_toolset} for k, info in _PLATFORMS_REGISTRY.items()} - - def _platform_default_toolset(platform: str) -> str: """Composite toolset a platform falls back to (plugin platforms derive ``hermes-``).""" plat_info = PLATFORMS.get(platform) @@ -440,12 +433,8 @@ def _platform_toolset_summary(config: dict, platforms: Optional[List[str]] = Non def _parse_enabled_flag(value, default: bool = True) -> bool: """Parse bool-like config values used by tool/platform settings.""" - if value is None: - return default - if isinstance(value, bool): - return value - if isinstance(value, int): - return value != 0 + if isinstance(value, (bool, int)): + return bool(value) if isinstance(value, str): lowered = value.strip().lower() if lowered in {"true", "1", "yes", "on"}: @@ -614,9 +603,7 @@ def _enabled_plugin_toolsets(config: dict, platform: str, toolset_names: List[st def _context_engine_active(config: dict) -> bool: context_cfg = config.get("context") or {} - if not isinstance(context_cfg, dict): - context_cfg = {} - name = str(context_cfg.get("engine") or "compressor").strip().lower() + name = str(context_cfg.get("engine") or "compressor").strip().lower() if isinstance(context_cfg, dict) else "compressor" return bool(name) and name != "compressor" @@ -749,11 +736,9 @@ def _save_platform_tools(config: dict, platform: str, enabled_toolset_keys: Set[ # user's unchecked selections on the next read. platform_default_keys = _platform_default_keys() existing_toolsets = cfg_get(config, "platform_toolsets", platform, default=[]) - if not isinstance(existing_toolsets, list): - existing_toolsets = [] # Preserve only entries that are neither configurable nor platform defaults (MCP server names). preserved_entries = { - str(entry) for entry in existing_toolsets + str(entry) for entry in (existing_toolsets if isinstance(existing_toolsets, list) else []) if str(entry) not in configurable_keys and str(entry) not in platform_default_keys } # Saving from the picker is consent to clear the "no_mcp" sentinel: the picker has no checkbox for it, @@ -796,8 +781,7 @@ def _toolset_has_keys( if ts_key == "vision": try: from agent.auxiliary_client import resolve_vision_provider_client - _provider, client, _model = resolve_vision_provider_client() - return client is not None + return resolve_vision_provider_client()[1] is not None except Exception: return False if ts_key in {"web", "image_gen", "video_gen", "tts", "stt", "browser"}: @@ -850,11 +834,11 @@ def _estimate_tool_tokens() -> Dict[str, int]: except Exception: logger.debug("tiktoken unavailable; skipping tool token estimation") return _tool_token_cache.setdefault(cache_key, {}) - counts: Dict[str, int] = {} - for name in registry.get_all_tool_names(): - schema = registry.get_schema(name) - if schema: # mirror the wire shape sent to the API - counts[name] = len(enc.encode(_json.dumps({"type": "function", "function": schema}))) + # Mirror the wire shape sent to the API. + counts = { + name: len(enc.encode(_json.dumps({"type": "function", "function": schema}))) + for name in registry.get_all_tool_names() if (schema := registry.get_schema(name)) + } _tool_token_cache[cache_key] = counts return counts @@ -870,12 +854,11 @@ def _prompt_toolset_checklist(platform_label: str, enabled: Set[str], platform: (k, l, d) for (k, l, d) in _get_effective_configurable_toolsets() if _toolset_allowed_for_platform(k, platform) and k not in _CONFIG_ONLY_TOOLSETS ] - labels = [] - for ts_key, ts_label, ts_desc in effective: - suffix = "" - if not _toolset_has_keys(ts_key, force_fresh=force_fresh) and _is_configurable(ts_key): - suffix = " [no API key]" - labels.append(f"{ts_label} ({ts_desc}){suffix}") + labels = [ + f"{ts_label} ({ts_desc})" + + (" [no API key]" if not _toolset_has_keys(ts_key, force_fresh=force_fresh) and _is_configurable(ts_key) else "") + for ts_key, ts_label, ts_desc in effective + ] pre_selected = {i for i, (ts_key, _, _) in enumerate(effective) if ts_key in enabled} status_fn = None @@ -888,9 +871,7 @@ def _prompt_toolset_checklist(platform_label: str, enabled: Set[str], platform: for idx in chosen: all_tools.update(resolve_toolset(ts_keys[idx])) total = sum(tool_tokens.get(name, 0) for name in all_tools) - if total >= 1000: - return f"Est. tool context: ~{total / 1000:.1f}k tokens" - return f"Est. tool context: ~{total} tokens" + return f"Est. tool context: ~{total / 1000:.1f}k tokens" if total >= 1000 else f"Est. tool context: ~{total} tokens" chosen = curses_checklist( f"Tools for {platform_label}", labels, pre_selected, cancel_returns=pre_selected, status_fn=status_fn, @@ -936,11 +917,10 @@ def _toolset_enabled_for_reconfigure(ts_key: str, config: dict) -> bool: if not _toolset_allowed_for_platform(ts_key, platform): continue try: - enabled = _get_platform_tools(config, platform, include_default_mcp_servers=False) + if ts_key in _get_platform_tools(config, platform, include_default_mcp_servers=False): + return True except Exception: continue - if ts_key in enabled: - return True return False @@ -948,8 +928,7 @@ def _toolset_enabled_for_reconfigure(ts_key: str, config: dict) -> bool: def _shared_metrics_state(config: dict) -> tuple[bool, bool]: """Return (collection_enabled, send_enabled) from a config dict.""" telemetry = config.get("telemetry") - telemetry = telemetry if isinstance(telemetry, dict) else {} - shared = telemetry.get("shared_metrics") + shared = telemetry.get("shared_metrics") if isinstance(telemetry, dict) else None shared = shared if isinstance(shared, dict) else {} return shared.get("enabled") is True, shared.get("send") is True @@ -957,12 +936,7 @@ def _shared_metrics_state(config: dict) -> tuple[bool, bool]: def _shared_metrics_menu_label(config: dict) -> str: """Menu row for shared metrics, showing both consent states.""" enabled, send = _shared_metrics_state(config) - if not enabled: - state = "off" - elif send: - state = "collecting + sending to Nous" - else: - state = "collecting locally" + state = "off" if not enabled else ("collecting + sending to Nous" if send else "collecting locally") return f"Configure shared metrics ({state})" @@ -1140,21 +1114,20 @@ def tools_command(args=None, first_install: bool = False, config: dict = None): platform_keys = list(enabled_platforms) platform_choices = [_platform_menu_label(config, pkey) for pkey in platform_keys] has_global = len(platform_keys) > 1 + has_mcp = bool(config.get("mcp_servers")) + global_idx = len(platform_keys) if has_global else -1 if has_global: platform_choices.append("Configure all platforms (global)") + reconfig_idx = len(platform_choices) platform_choices.append("Reconfigure an existing tool's provider or API key") + metrics_idx = len(platform_choices) platform_choices.append(_shared_metrics_menu_label(config)) - has_mcp = bool(config.get("mcp_servers")) + mcp_idx = len(platform_choices) if has_mcp else -1 if has_mcp: platform_choices.append("Configure MCP server tools") + done_idx = len(platform_choices) platform_choices.append("Done") - global_idx = len(platform_keys) if has_global else -1 - reconfig_idx = len(platform_keys) + (1 if has_global else 0) - metrics_idx = reconfig_idx + 1 - mcp_idx = (metrics_idx + 1) if has_mcp else -1 - done_idx = metrics_idx + (2 if has_mcp else 1) - while True: idx = _prompt_choice("Select an option:", platform_choices, default=0) if idx == done_idx: diff --git a/hermes_cli/tools_config_cua.py b/hermes_cli/tools_config_cua.py index 94ea7b0277..d25f9e2f28 100644 --- a/hermes_cli/tools_config_cua.py +++ b/hermes_cli/tools_config_cua.py @@ -16,6 +16,37 @@ from hermes_cli.cli_output import ( logger = logging.getLogger("hermes_cli.tools_config") +# 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 (_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. 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 +# pre-clear never yanks a lock a live-but-slow install still holds. +_CUA_LOCK_STALE_AFTER = 600 + +_CUA_INSTALL_PS1_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.ps1" +_CUA_INSTALL_SH_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.sh" +_CUA_MANUAL_README = "https://github.com/trycua/cua/blob/main/libs/cua-driver/README.md" +_UPGRADE_CMD = "hermes computer-use install --upgrade" + + +def _run_text(cmd: list, *, timeout, capture_output: bool = True, **kwargs) -> subprocess.CompletedProcess: + """``subprocess.run`` with the utf-8/replace text decoding every helper here uses.""" + return subprocess.run( + cmd, capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout, **kwargs, + ) + def _post_setup_no_window_flags(*, streams_to_console: bool = False) -> int: """Win32 creationflags that stop post-setup children flashing a console. @@ -45,11 +76,7 @@ def _cua_driver_cmd() -> str: def _cua_version_summary(raw: str, *, limit: int = 120) -> str: """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: - return text[:limit] - return "" + return next((line.strip()[:limit] for line in (raw or "").splitlines() if line.strip()), "") def _resolved_cua_driver_cmd() -> Optional[str]: @@ -76,7 +103,8 @@ _CUA_DRIVER_CONTRACT_CACHE: dict = {} def _cua_driver_contract_status(binary: Optional[str] = None) -> dict: - """Inspect whether an installed driver supports Hermes' runtime contract.""" + """Inspect whether an installed driver supports Hermes' runtime contract (30s cache keyed on the binary's + path/mtime/size fingerprint).""" import time from tools.computer_use.cua_backend import cua_driver_runtime_contract_status @@ -91,14 +119,12 @@ def _cua_driver_contract_status(binary: Optional[str] = None) -> dict: return cua_driver_runtime_contract_status(resolved) now = time.monotonic() - if ( - _CUA_DRIVER_CONTRACT_CACHE.get("fingerprint") == fingerprint - and now - _CUA_DRIVER_CONTRACT_CACHE.get("checked_at", 0.0) < 30.0 - ): - return dict(_CUA_DRIVER_CONTRACT_CACHE["state"]) + cache = _CUA_DRIVER_CONTRACT_CACHE + if cache.get("fingerprint") == fingerprint and now - cache.get("checked_at", 0.0) < 30.0: + return dict(cache["state"]) state = cua_driver_runtime_contract_status(resolved) - _CUA_DRIVER_CONTRACT_CACHE.update(fingerprint=fingerprint, checked_at=now, state=dict(state)) + cache.update(fingerprint=fingerprint, checked_at=now, state=dict(state)) return state @@ -118,6 +144,7 @@ def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool = """ venv_root = Path(sys.executable).parent.parent uv_env = {**os.environ, "VIRTUAL_ENV": str(venv_root)} + install_flags = _post_setup_no_window_flags(streams_to_console=not capture_output) # 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. @@ -126,12 +153,8 @@ def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool = uv_bin = ensure_uv() if uv_bin: try: - result = subprocess.run( - [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), - ) + result = _run_text([uv_bin, "pip", "install", *args], capture_output=capture_output, timeout=timeout, + env=uv_env, creationflags=install_flags) if result.returncode == 0: return result # Fall through to pip — uv may have failed for a reason pip can handle. @@ -141,31 +164,21 @@ def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool = pip_cmd = [sys.executable, "-m", "pip"] try: # Probe for pip; bootstrap via ensurepip if missing (uv venv lacks it). - probe = subprocess.run( - pip_cmd + ["--version"], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15, - creationflags=_post_setup_no_window_flags(), - ) + probe = _run_text(pip_cmd + ["--version"], timeout=15, creationflags=_post_setup_no_window_flags()) if probe.returncode != 0: raise FileNotFoundError("pip not in venv") except (subprocess.TimeoutExpired, FileNotFoundError): try: - subprocess.run( - [sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120, check=True, - creationflags=_post_setup_no_window_flags(), - ) + _run_text([sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], timeout=120, check=True, + creationflags=_post_setup_no_window_flags()) 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}", ) - 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), - ) + return _run_text(pip_cmd + ["install", *args], capture_output=capture_output, timeout=timeout, + creationflags=install_flags) # No pre-install release/asset probe: cua-driver-rs releases are all prereleases, which GitHub's @@ -187,11 +200,8 @@ def _cua_install_target_writable() -> bool: def _cua_driver_version(binary: str) -> Optional[str]: """`` --version`` stdout (possibly ""), or None when the probe itself fails.""" try: - return subprocess.run( - [binary, "--version"], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, env=_cua_driver_env(), - creationflags=_post_setup_no_window_flags(), - ).stdout.strip() + return _run_text([binary, "--version"], timeout=5, env=_cua_driver_env(), + creationflags=_post_setup_no_window_flags()).stdout.strip() except Exception: return None @@ -209,40 +219,36 @@ def _confirmed_update_check(driver_cmd: str, require_confirmed_update: bool) -> _state = cua_driver_update_check() except Exception: _state = None - if _state is not None and not _state.get("update_available"): + if _state is None: + if require_confirmed_update: + _print_info( + f" Could not confirm a newer {driver_cmd} release " + "(offline, rate-limited, or driver too old to check); " + "keeping the installed version." + ) + _print_info(f" Force a refresh with: {_UPGRADE_CMD}") + return False, None + return True, None + if not _state.get("update_available"): _print_success( f" {driver_cmd} is already on the latest release " f"({_state.get('current_version') or 'unknown'})." ) return False, None - if _state is None and require_confirmed_update: - _print_info( - f" Could not confirm a newer {driver_cmd} release " - "(offline, rate-limited, or driver too old to check); " - "keeping the installed version." - ) - _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 (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 + # 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") - if _re.fullmatch(r"\d+(\.\d+)*", _latest): - confirmed_version = _latest - return True, confirmed_version + _latest = str(_state.get("latest_version") or "").strip().lstrip("vV") + return True, (_latest if _re.fullmatch(r"\d+(\.\d+)*", _latest) else None) def install_cua_driver( - upgrade: bool = False, - require_confirmed_update: bool = False, - show_installer_progress: bool = True, + upgrade: bool = False, require_confirmed_update: bool = False, show_installer_progress: bool = True, ) -> bool: """Install or refresh the cua-driver binary used by Computer Use. @@ -254,10 +260,8 @@ def install_cua_driver( system = _plat.system() if system not in ("Darwin", "Windows", "Linux"): - if upgrade: - # Silent: `hermes update` calls this for every user. - return False - _print_warning(" Computer Use (cua-driver) is unsupported on this platform; skipping.") + if not upgrade: # silent under `hermes update`, which calls this for every user + _print_warning(" Computer Use (cua-driver) is unsupported on this platform; skipping.") return False is_windows = system == "Windows" @@ -284,7 +288,7 @@ def install_cua_driver( return False if not shutil.which(fetch_tool): _print_warning(f" {fetch_tool} not found — install manually:") - _print_info(" https://github.com/trycua/cua/blob/main/libs/cua-driver/README.md") + _print_info(f" {_CUA_MANUAL_README}") return False return _run_cua_driver_installer(label="Installing") @@ -298,10 +302,8 @@ def install_cua_driver( # Compatible existing install: no download, just the host-specific setup upstream normally owns. if binary and not upgrade and not repair_existing: version = _cua_driver_version(binary) - if version is None: - _print_success(f" {driver_cmd} already installed.") - else: - _print_success(f" {driver_cmd} already installed: {version or 'unknown version'}") + suffix = "" if version is None else f": {version or 'unknown version'}" + _print_success(f" {driver_cmd} already installed{suffix}.") if is_windows and not _repair_cua_driver_autostart_windows(binary, verbose=False): _print_warning(" cua-driver is compatible, but Windows autostart repair failed.") return False @@ -311,29 +313,20 @@ def install_cua_driver( if repair_existing: 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 runtime contract: {reason}." - ) - if os.environ.get("HERMES_CUA_DRIVER_CMD", "").strip(): - _print_info( - " Update the binary selected by HERMES_CUA_DRIVER_CMD, or unset " - "the override and run: hermes computer-use install --upgrade" - ) + _print_warning(f" Found cua-driver {version}, but Hermes cannot use its current runtime contract: {reason}.") + if override: + _print_info(f" Update the binary selected by HERMES_CUA_DRIVER_CMD, or unset the override and run: {_UPGRADE_CMD}") return False if is_windows and require_confirmed_update: - _print_info( - " 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" - ) + _print_info(" Automatic Windows updates cannot safely run cua-driver's interactive repair installer.") + _print_info(f" Repair it from an interactive terminal with: {_UPGRADE_CMD}") 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(f" Run `{_UPGRADE_CMD}` from an admin account to update it.") return bool(binary) if not shutil.which(fetch_tool): @@ -350,20 +343,15 @@ def install_cua_driver( # 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(" cua-driver is not installed; automatic Windows updates cannot safely run its interactive installer.") + _print_info(f" Install it from an interactive terminal with: {_UPGRADE_CMD}") return False # Best-effort before/after version display. before = (_cua_driver_version(binary) or "") if binary else "" ok = _run_cua_driver_installer( - label="Repairing" if repair_existing else "Refreshing", - verbose=False, - pin_version=confirmed_version, + label="Repairing" if repair_existing else "Refreshing", verbose=False, pin_version=confirmed_version, show_progress=show_installer_progress, installer_timeout=_CUA_BACKGROUND_UPDATE_TIMEOUT if require_confirmed_update else None, ) @@ -385,26 +373,6 @@ 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 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 (_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. 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 -# pre-clear never yanks a lock a live-but-slow install still holds. -_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")) @@ -436,12 +404,8 @@ def _clear_stale_windows_cua_install_lock() -> None: from ctypes import wintypes as _wintypes # Win32 constants used by install.ps1's FileShare::None equivalent. - delete_access = 0x00010000 - generic_read = 0x80000000 - generic_write = 0x40000000 - open_existing = 3 - file_attribute_normal = 0x00000080 - file_flag_delete_on_close = 0x04000000 + delete_access, generic_read, generic_write = 0x00010000, 0x80000000, 0x40000000 + open_existing, file_attribute_normal, file_flag_delete_on_close = 3, 0x00000080, 0x04000000 kernel32 = _ctypes.WinDLL("kernel32", use_last_error=True) create_file = kernel32.CreateFileW @@ -455,22 +419,15 @@ def _clear_stale_windows_cua_install_lock() -> None: close_handle.restype = _wintypes.BOOL handle = create_file( - str(lock_file), - generic_read | generic_write | delete_access, - 0, # FileShare::None - None, - open_existing, - file_attribute_normal | file_flag_delete_on_close, - None, + str(lock_file), generic_read | generic_write | delete_access, 0, # 0 = FileShare::None + None, open_existing, file_attribute_normal | file_flag_delete_on_close, None, ) - invalid_handle = _wintypes.HANDLE(-1).value - if handle == invalid_handle: + if handle == _wintypes.HANDLE(-1).value: logger.debug( "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)", @@ -502,9 +459,8 @@ def _clear_stale_cua_install_lock() -> None: if not lock_dir.is_dir(): return holder_pid = None - info = lock_dir / "info" try: - for line in info.read_text(encoding="utf-8", errors="replace").splitlines(): + for line in (lock_dir / "info").read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith("pid="): holder_pid = int(line.split("=", 1)[1].strip()) break @@ -543,18 +499,18 @@ def _cua_install_lock_held() -> bool: surviving lock artifact means a concurrent (or orphaned-but-alive) install owns it. """ try: - if sys.platform == "win32": - 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. 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 - return _cua_install_lock_dir().is_dir() + if sys.platform != "win32": + return _cua_install_lock_dir().is_dir() + 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. 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 except Exception as e: logger.debug("cua install lock probe failed: %s", e) return False @@ -607,9 +563,7 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b 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 - if _cua_driver_autostart_registered_windows(): + if sys.platform != "win32" or _cua_driver_autostart_registered_windows(): return True binary = shutil.which(driver_cmd) @@ -624,18 +578,11 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b "-Verb RunAs -Wait -PassThru -ErrorAction Stop; " "exit $proc.ExitCode" ) - - if verbose: - _print_info(" Registering cua-driver auto-start...") - else: - _print_info(" Repairing cua-driver auto-start registration...") + _print_info(" Registering cua-driver auto-start..." if verbose else " Repairing cua-driver auto-start registration...") try: - result = subprocess.run( - [ps, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_cmd], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=300, - env=_cua_driver_env(), - ) + result = _run_text([ps, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_cmd], timeout=300, + env=_cua_driver_env()) except subprocess.TimeoutExpired: _print_warning(" cua-driver autostart registration timed out.") return False @@ -646,9 +593,8 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b if result.returncode == 0: return True - tail = (result.stderr or result.stdout or "").strip().splitlines()[-3:] _print_warning(" cua-driver autostart registration failed.") - for line in tail: + for line in (result.stderr or result.stdout or "").strip().splitlines()[-3:]: _print_info(f" {line[:200]}") _print_info(" From an elevated shell, run: cua-driver autostart enable") return False @@ -661,6 +607,12 @@ def _remove_quietly(path: str) -> None: pass +_MAC_PERMISSION_PATHS = ( + " System Settings > Privacy & Security > Accessibility", + " System Settings > Privacy & Security > Screen Recording", +) + + def _print_cua_platform_notes(is_windows: bool, is_linux: bool, *, fresh_install: bool) -> None: """Host-specific follow-up notes after an install or a compatible-install check.""" if is_windows: @@ -668,18 +620,13 @@ def _print_cua_platform_notes(is_windows: bool, is_linux: bool, *, fresh_install _print_info(" Windows/SmartScreen may prompt the first time it runs.") elif is_linux: _print_warning(" Linux support is alpha.") - elif fresh_install: - _print_info(" IMPORTANT — grant macOS permissions now:") - _print_info(" System Settings > Privacy & Security > Accessibility") - _print_info(" System Settings > Privacy & Security > Screen Recording") - _print_info(" Both must allow the terminal / Hermes process.") else: - _print_info(" Grant macOS permissions if not done yet:") - _print_info(" System Settings > Privacy & Security > Accessibility") - _print_info(" System Settings > Privacy & Security > Screen Recording") - - -_CUA_INSTALL_PS1_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.ps1" + _print_info(" IMPORTANT — grant macOS permissions now:" if fresh_install + else " Grant macOS permissions if not done yet:") + for line in _MAC_PERMISSION_PATHS: + _print_info(line) + if fresh_install: + _print_info(" Both must allow the terminal / Hermes process.") def _kill_installer_tree(proc, *, is_windows: bool) -> None: @@ -688,37 +635,31 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None: try: 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 communicate() and install.lock wedged, so kill the tree leaf-up. - import psutil as _psutil + return + # 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: - parent = _psutil.Process(proc.pid) - descendants = parent.children(recursive=True) - except _psutil.NoSuchProcess: - return - except _psutil.Error as e: - logger.debug( - "could not enumerate cua-driver installer tree for pid %s: %s", proc.pid, e - ) - proc.kill() - return + try: + parent = _psutil.Process(proc.pid) + descendants = parent.children(recursive=True) + except _psutil.NoSuchProcess: + return + except _psutil.Error as e: + logger.debug("could not enumerate cua-driver installer tree for pid %s: %s", proc.pid, e) + proc.kill() + return - for child in reversed(descendants): - try: - child.kill() - except _psutil.NoSuchProcess: - pass - except _psutil.Error as e: - logger.debug("could not kill cua-driver installer child pid %s: %s", child.pid, e) + for target, pid in [(child, child.pid) for child in reversed(descendants)] + [(parent, proc.pid)]: try: - parent.kill() + target.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) - proc.kill() + what = "parent" if target is parent else "child" + logger.debug("could not kill cua-driver installer %s pid %s: %s", what, pid, e) + if target is parent: + proc.kill() except (OSError, ProcessLookupError): proc.kill() @@ -734,9 +675,7 @@ def _reap_after_timeout(proc, *, is_windows: bool) -> None: drained_out, _ = proc.communicate(timeout=_CUA_INSTALLER_DRAIN_GRACE) # 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:], - ) + logger.warning("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. @@ -755,23 +694,18 @@ def _cua_installer_command(is_windows: bool): # 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 = f'powershell -NoProfile -ExecutionPolicy Bypass -Command "{ps_oneliner}"' - return install_cmd, manual_hint, None + return install_cmd, f'powershell -NoProfile -ExecutionPolicy Bypass -Command "{ps_oneliner}"', 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. 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" - manual_hint = f'/bin/bash -c "$(curl -fsSL {install_url})"' + manual_hint = f'/bin/bash -c "$(curl -fsSL {_CUA_INSTALL_SH_URL})"' fd, script_path = _tempfile.mkstemp(prefix="cua-driver-install-", suffix=".sh") os.close(fd) try: - dl = subprocess.run( - ["curl", "-fsSL", "-o", script_path, install_url], - capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120, - ) + dl = _run_text(["curl", "-fsSL", "-o", script_path, _CUA_INSTALL_SH_URL], timeout=120) failure = None if dl.returncode == 0 else (dl.stderr or "").strip()[:200] except (subprocess.TimeoutExpired, OSError) as e: failure = str(e) @@ -792,16 +726,11 @@ def _unattended_installer_preflight(install_cmd: list, is_windows: bool): `computer-use install --upgrade` runs never come here and keep upstream's full lock-recovery. """ if _cua_install_lock_held(): - _print_info( - " 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(" Another cua-driver install is in progress (upstream install lock is held) — skipping this refresh.") + _print_info(f" If no install is really running, retry with: {_UPGRADE_CMD}") return None if not _cua_release_endpoint_reachable(): - _print_info( - " github.com is unreachable — skipping cua-driver refresh (will retry on the next update)." - ) + _print_info(" 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 @@ -815,10 +744,7 @@ def _unattended_installer_preflight(install_cmd: list, is_windows: bool): def _run_cua_driver_installer( - label: str = "Installing", - verbose: bool = True, - pin_version: Optional[str] = None, - show_progress: bool = True, + label: str = "Installing", verbose: bool = True, pin_version: Optional[str] = None, show_progress: bool = True, installer_timeout: Optional[float] = None, ) -> bool: """Run the upstream cua-driver installer for this platform. @@ -839,10 +765,8 @@ def _run_cua_driver_installer( install_cmd, manual_hint, script_path = prepared if show_progress: - if verbose: - _print_info(f" {label} cua-driver (background computer-use)...") - else: - _print_info(f"→ {label} cua-driver (Computer Use)...") + _print_info(f" {label} cua-driver (background computer-use)..." if verbose + else f"→ {label} cua-driver (Computer Use)...") driver_cmd = _cua_driver_cmd() timeout = _CUA_INSTALLER_TIMEOUT if installer_timeout is None else installer_timeout @@ -864,53 +788,41 @@ def _run_cua_driver_installer( # 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 = {} + popen_kwargs: dict = {"shell": False, "env": installer_env} if not is_windows: popen_kwargs["start_new_session"] = True + # 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: + popen_kwargs["creationflags"] = _post_setup_no_window_flags(streams_to_console=True) + else: + popen_kwargs.update( + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, encoding="utf-8", errors="replace", creationflags=_post_setup_no_window_flags(), + ) 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. - if verbose: - proc = subprocess.Popen( - install_cmd, shell=False, env=installer_env, - creationflags=_post_setup_no_window_flags(streams_to_console=True), - **popen_kwargs - ) - try: - proc.communicate(timeout=timeout) - except subprocess.TimeoutExpired: - _reap_after_timeout(proc, is_windows=is_windows) - raise - result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=None, stderr=None) - else: - proc = subprocess.Popen( - install_cmd, shell=False, env=installer_env, - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, encoding="utf-8", errors="replace", - creationflags=_post_setup_no_window_flags(), - **popen_kwargs - ) - try: - out, _ = proc.communicate(timeout=timeout) - except subprocess.TimeoutExpired: - _reap_after_timeout(proc, is_windows=is_windows) - raise - result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=out, stderr=None) + proc = subprocess.Popen(install_cmd, **popen_kwargs) + try: + communicated = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + _reap_after_timeout(proc, is_windows=is_windows) + raise + out = None if verbose else communicated[0] + result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=out, stderr=None) + if not verbose and result.stdout: # 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.flush() - except Exception: - pass - if result.returncode != 0: - logger.debug("cua-driver installer output:\n%s", 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.flush() + except Exception: + pass + if result.returncode != 0: + 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): @@ -925,10 +837,7 @@ def _run_cua_driver_installer( except subprocess.TimeoutExpired: _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 — " - f"check {_cua_install_lock_dir()}" - ) + _print_info(f" If this repeats, a stale installer lock may be present — check {_cua_install_lock_dir()}") _print_info(f" Re-run manually: {manual_hint}") return False except Exception as e: