refactor(hermes_cli): tools_config_cua — shared _run_text/_UPGRADE_CMD, single Popen path, folded ladders; tools_config header + small collapses

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 ``<emoji>`` 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-<platform>``)."""
     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:

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]:
     """``<binary> --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: