From 3a056dd46641afa613200700716a1c311af3ade5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:59:08 -0700 Subject: [PATCH] refactor(web_server): extract gateway topology/actions/process helpers to web_server_gateway; restore late-bound config re-exports --- hermes_cli/web_server.py | 644 ++---------------------------- hermes_cli/web_server_gateway.py | 646 +++++++++++++++++++++++++++++++ 2 files changed, 678 insertions(+), 612 deletions(-) create mode 100644 hermes_cli/web_server_gateway.py diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index ad177adff0..eca07772a9 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -32,7 +32,6 @@ import threading import time import urllib.parse -from hermes_cli._subprocess_compat import windows_detach_flags from hermes_cli.install_identity import get_install_id as _shared_get_install_id from hermes_cli.pty_session import run_reaper import urllib.request @@ -47,16 +46,18 @@ if str(PROJECT_ROOT) not in sys.path: from hermes_cli import __version__ from hermes_cli.config import ( DEFAULT_CONFIG, - get_hermes_home, get_process_hermes_home, load_config, - # Late-bound by extracted routers (tests monkeypatch web_server.). - check_config_version, # noqa: F401 - remove_env_value, # noqa: F401 save_config, - save_env_value, # noqa: F401 — late-bound by extracted routers - detect_install_method, redact_key, + # Late-bound by extracted routers/modules; tests monkeypatch web_server.. + cfg_get, # noqa: F401 + check_config_version, # noqa: F401 + detect_install_method, # noqa: F401 + get_hermes_home, # noqa: F401 + load_env, # noqa: F401 + remove_env_value, # noqa: F401 + save_env_value, # noqa: F401 ) from gateway.status import ( # noqa: F401 — late-bound by web_routers/status + tests monkeypatch web_server. get_running_pid, @@ -1336,54 +1337,30 @@ elif _GATEWAY_HEALTH_TIMEOUT > _GATEWAY_HEALTH_TIMEOUT_MAX: _GATEWAY_HEALTH_TIMEOUT = _GATEWAY_HEALTH_TIMEOUT_MAX -# DEPRECATED (scheduled for removal): GATEWAY_HEALTH_URL / GATEWAY_HEALTH_TIMEOUT. -# Cross-container / cross-host gateway liveness detection will be folded into a -# first-class dashboard config key so it's no longer Docker-adjacent lore buried -# in env vars. The env vars still work for now so existing Compose deployments -# don't break. Do not add new callers — wire new uses through the planned -# config surface. - - -def _probe_gateway_health() -> tuple[bool, dict | None]: - """Probe the gateway via its HTTP health endpoint (cross-container). - - .. deprecated:: - Driven by the deprecated ``GATEWAY_HEALTH_URL`` / - ``GATEWAY_HEALTH_TIMEOUT`` env vars. Scheduled for removal alongside - a move to a first-class dashboard config key. See - :data:`_GATEWAY_HEALTH_URL` for context. - - Uses ``/health/detailed`` first (returns full state), falling back to - the simpler ``/health`` endpoint. Returns ``(is_alive, body_dict)``. - - Accepts any of these as ``GATEWAY_HEALTH_URL``: - - ``http://gateway:8642`` (base URL — recommended) - - ``http://gateway:8642/health`` (explicit health path) - - ``http://gateway:8642/health/detailed`` (explicit detailed path) - - This is a **blocking** call — run via ``run_in_executor`` from async code. - """ - if not _GATEWAY_HEALTH_URL: - return False, None - - # Normalise to base URL so we always probe the right paths regardless of - # whether the user included /health or /health/detailed in the env var. - base = _GATEWAY_HEALTH_URL.rstrip("/") - if base.endswith("/health/detailed"): - base = base[: -len("/health/detailed")] - elif base.endswith("/health"): - base = base[: -len("/health")] - - for path in (f"{base}/health/detailed", f"{base}/health"): - try: - req = urllib.request.Request(path, method="GET") - with urllib.request.urlopen(req, timeout=_GATEWAY_HEALTH_TIMEOUT) as resp: - if resp.status == 200: - body = json.loads(resp.read()) - return True, body - except Exception: - continue - return False, None +from hermes_cli.web_server_gateway import ( # noqa: E402,F401 — re-exported; routers/tests reach these via web_server. + _ACTION_COMMANDS, + _ACTION_IDS, + _ACTION_LOG_DIR, + _ACTION_LOG_FILES, + _ACTION_PROCS, + _ACTION_RESULTS, + _TOPOLOGY_CACHE, + _TOPOLOGY_CACHE_TTL, + _collect_profile_gateway_topology, + _collect_profile_gateway_topology_cached, + _dashboard_spawn_executable, + _display_system_platform, + _gateway_subcommand, + _load_configured_gateway_platforms, + _probe_gateway_health, + _profile_gateway_writer_identity, + _profile_platform_ports, + _restart_gateway_after, + _spawn_hermes_action, + _split_text_for_speak_stream, + _strip_session_list_rows, + _terminate_desktop_managed_gateway, +) _MANAGED_FILES_ROOT_ENV = "HERMES_DASHBOARD_FILES_ROOT" @@ -1646,254 +1623,6 @@ from hermes_cli.web_routers.git import ( # noqa: E402,F401 — legacy re-export ) -# Host TCP ports each port-binding gateway platform listens on, as -# ``platform-name -> (config port key, adapter default)``. Mirrors -# ``PORT_BINDING_PLATFORM_VALUES`` in gateway/config.py and each adapter's -# DEFAULT_PORT / DEFAULT_WEBHOOK_PORT constant. Used only for the dashboard's -# gateway-topology readout — best-effort display data, not a bind source. -_PORT_BINDING_PLATFORM_PORTS: Dict[str, Tuple[str, int]] = { - "webhook": ("port", 8644), - "api_server": ("port", 8642), - "msgraph_webhook": ("port", 8646), - "feishu": ("webhook_port", 8765), - "wecom_callback": ("port", 8645), - "bluebubbles": ("webhook_port", 8645), - "sms": ("webhook_port", 8080), - "whatsapp_cloud": ("webhook_port", 8090), - "line": ("port", 8646), - "teams": ("port", 3978), -} - -# Platform states that mean the adapter is NOT serving its port right now. -_PLATFORM_DEAD_STATES = frozenset({"fatal", "disconnected", "stopped"}) - - -def _profile_platform_ports(profile_home: Path, runtime: Optional[dict]) -> Dict[str, int]: - """Best-effort map of ``platform -> host TCP port`` for one profile's gateway. - - Reads the platforms the running gateway reported in its - ``gateway_state.json`` and resolves each port-binding platform's port from - the profile's ``config.yaml`` (top-level ``platforms:`` wins over - ``gateway.platforms:``, matching ``load_gateway_config`` precedence), - falling back to the adapter default. Display-only: env-var port overrides - (e.g. ``WEBHOOK_PORT`` in that profile's .env) are not resolved here. - """ - platforms = (runtime or {}).get("platforms") or {} - active = [ - name for name, state in platforms.items() - if name in _PORT_BINDING_PLATFORM_PORTS - and isinstance(state, dict) - and state.get("state") not in _PLATFORM_DEAD_STATES - ] - if not active: - return {} - - blocks: Dict[str, dict] = {} - try: - # Multi-profile probe: load_config() targets the ACTIVE profile's - # home, so read the probed profile's file via the raw primitive. - from hermes_cli.config import read_user_config_raw - cfg = read_user_config_raw(profile_home / "config.yaml") - gateway_cfg = cfg.get("gateway") if isinstance(cfg.get("gateway"), dict) else {} - # gateway.platforms first, top-level platforms second — later wins, - # matching the precedence in gateway.config.load_gateway_config(). - for src in ((gateway_cfg or {}).get("platforms"), cfg.get("platforms")): - if not isinstance(src, dict): - continue - for plat_name, plat_block in src.items(): - if isinstance(plat_block, dict): - blocks.setdefault(plat_name, {}).update(plat_block) - except Exception: - blocks = {} - - ports: Dict[str, int] = {} - for name in active: - port_key, default_port = _PORT_BINDING_PLATFORM_PORTS[name] - block = blocks.get(name) or {} - extra = block.get("extra") if isinstance(block.get("extra"), dict) else {} - raw = block.get(port_key, (extra or {}).get(port_key, default_port)) - try: - ports[name] = int(raw) - except (TypeError, ValueError): - ports[name] = default_port - return ports - - -def _profile_gateway_writer_identity( - profile_home: Path, runtime: Optional[dict] -) -> Optional[tuple]: - """``(pid, start_time)`` identity of the profile's LIVE gateway, or None. - - Reuses the validated-liveness helper — recorded PID checked against the - live process table, the start-time PID-reuse fingerprint, and the - profile's home — then reads the live process's fingerprint via the same - ``_get_process_start_time`` that stamped it, so equality is exact (no - unit or clock-source mismatch). None when the record doesn't belong to - a live gateway; nothing in it is current by definition then. - """ - try: - from gateway.status import ( - _get_process_start_time, - get_runtime_status_running_pid, - ) - - pid = get_runtime_status_running_pid(runtime, expected_home=profile_home) - if pid is None: - return None - start_time = _get_process_start_time(pid) - if start_time is None: - return None - return (pid, start_time) - except Exception: - return None - - -def _owned_profile_platforms( - writer_identity: Optional[tuple], platforms: dict -) -> dict: - """Keep only platform entries the profile's CURRENT process wrote. - - Gateway startup deliberately preserves plain platform entries in - ``gateway_state.json`` across restarts (the dashboard keeps showing - last-known state while adapters reconnect), and the active-profile - endpoint compensates by filtering them against the current - configuration. The cross-profile aggregation has no equivalent config - context (a profile's platform set depends on tokens in that profile's - ``.env`` behind its secret scope), so it demands strict process - ownership instead: ``write_runtime_status`` stamps every platform write - with the writer's ``(pid, start_time)`` identity, and an entry is - aggregatable only when that identity equals the profile's live gateway - process — exact match, no clock heuristics, so an entry written moments - before a fast restart can never masquerade as current. A fatal entry - left behind by a platform the operator has since disabled/removed thus - stops degrading fleet health as soon as that profile's gateway restarts - (a config change requires that restart to take effect anyway). Fail - closed: entries without a writer identity (legacy records) or records - with no live process are excluded — aggregation is a supplement, and a - false "degraded forever" is the worse failure mode. - """ - if writer_identity is None: - return {} - live_pid, live_start = writer_identity - owned: Dict[str, dict] = {} - for key, value in platforms.items(): - if not isinstance(value, dict): - continue - if ( - value.get("writer_pid") == live_pid - and value.get("writer_start_time") == live_start - ): - owned[key] = value - return owned - - -def _collect_profile_gateway_topology() -> Dict[str, Any]: - """Enumerate profiles and the gateways serving them for ``/api/status``. - - Returns ``{"profiles": [...], "gateway_mode": ..., "gateways": [...]}``: - - * ``profiles`` — every profile on the host (default + named), from - ``profiles_to_serve(True)`` (the cheap enumeration chokepoint — no - per-profile config reads or skill counts). - * ``gateways`` — one entry per profile with a LIVE gateway process: - ``{"profile", "ports", "served_profiles"?}``. Liveness reuses - ``_check_gateway_running`` so this agrees with the profiles sidebar. - * ``gateway_mode`` — ``"multiplex"`` when the default gateway serves - multiple profiles (gateway.multiplex_profiles), ``"single"`` for one - live gateway, ``"multiple"`` for independent per-profile gateways, - ``"none"`` when nothing is running. - * ``profile_platforms`` — ``{profile: platforms}`` runtime platform maps - for each LIVE gateway, ownership-filtered to entries stamped by that - profile's current process (stale preserved entries for since-removed - platforms are excluded — see ``_owned_profile_platforms``). Internal - aggregation input for ``/api/status`` (independent per-profile gateways - write failures to their own ``gateway_state.json``, which the - unparameterized endpoint would otherwise never see). Never exposed - directly. - """ - try: - from hermes_cli.profiles import _check_gateway_running, profiles_to_serve - from gateway.status import read_runtime_status - homes = profiles_to_serve(True) - except Exception: - _log.debug("profile/gateway topology enumeration failed", exc_info=True) - return { - "profiles": [], - "gateway_mode": "unknown", - "gateways": [], - "profile_platforms": {}, - } - - profile_names = [name for name, _home in homes] - gateways: List[Dict[str, Any]] = [] - profile_platforms: Dict[str, dict] = {} - multiplex = False - for name, home in homes: - try: - if not _check_gateway_running(home): - continue - except Exception: - continue - try: - runtime = read_runtime_status(home / "gateway_state.json") - except Exception: - runtime = None - served = [str(p) for p in ((runtime or {}).get("served_profiles") or [])] - if name == "default" and len(served) > 1: - multiplex = True - plats = (runtime or {}).get("platforms") - if isinstance(plats, dict) and plats: - # Ownership filter: gateway startup preserves plain platform - # entries across restarts, so the raw map can carry fatal state - # for platforms the operator has since disabled/removed. Only - # entries stamped with the profile's current live process's - # writer identity are aggregation candidates (see - # _owned_profile_platforms). - owned = _owned_profile_platforms( - _profile_gateway_writer_identity(home, runtime), plats - ) - if owned: - profile_platforms[name] = owned - entry: Dict[str, Any] = { - "profile": name, - "ports": _profile_platform_ports(home, runtime), - } - if served: - entry["served_profiles"] = served - gateways.append(entry) - - if multiplex: - mode = "multiplex" - elif len(gateways) > 1: - mode = "multiple" - elif len(gateways) == 1: - mode = "single" - else: - mode = "none" - - return { - "profiles": profile_names, - "gateway_mode": mode, - "gateways": gateways, - "profile_platforms": profile_platforms, - } - - -# /api/status is polled ~1/s by the desktop app while it waits for the backend -# (and again by the dashboard badge). Each uncached call above walks 7+ profile -# homes (yaml.safe_load with the pure-Python loader + psutil process-table -# probes + realpath walks) inside the default executor; concurrent polls pile -# up and hold the GIL for 14-16s, starving the event loop — the desktop WS -# never receives gateway.ready and boot fails ("event loop stalled ... GIL -# pressure suspected"). Topology changes on gateway start/stop, so a short TTL -# cache with a collapse lock keeps the scan to one per window. The cache also -# remembers which collector produced the entry: tests monkeypatch -# _collect_profile_gateway_topology per case, and the identity check keeps -# them hermetic without needing a reset hook (a swapped collector is a miss). -_TOPOLOGY_CACHE: Dict[str, Any] = {"ts": 0.0, "data": None, "fn": None} -_TOPOLOGY_CACHE_LOCK = threading.Lock() -_TOPOLOGY_CACHE_TTL = 10.0 - # Stable install identity for /api/status. One random opaque id per physical # install, minted on first read and persisted under the ROOT Hermes home # (get_default_hermes_root()) — NOT the profile-scoped HERMES_HOME — so every @@ -1923,46 +1652,6 @@ def get_install_id() -> Optional[str]: _CONFIG_MUTATION_LOCK = threading.RLock() -def _topology_cache_get(fn: Any) -> Optional[Dict[str, Any]]: - if ( - _TOPOLOGY_CACHE["data"] is not None - and _TOPOLOGY_CACHE["fn"] is fn - and time.monotonic() - _TOPOLOGY_CACHE["ts"] < _TOPOLOGY_CACHE_TTL - ): - return _TOPOLOGY_CACHE["data"] - return None - - -def _collect_profile_gateway_topology_cached() -> Dict[str, Any]: - fn = _collect_profile_gateway_topology - cached = _topology_cache_get(fn) - if cached is not None: - return cached - with _TOPOLOGY_CACHE_LOCK: - cached = _topology_cache_get(fn) - if cached is not None: - return cached - data = fn() - _TOPOLOGY_CACHE["data"] = data - _TOPOLOGY_CACHE["fn"] = fn - _TOPOLOGY_CACHE["ts"] = time.monotonic() - return data - - -def _load_configured_gateway_platforms() -> set[str]: - """Load connected platform names away from the asyncio event loop. - - The first ``load_gateway_config()`` call performs platform discovery and - can take longer than Desktop's WebSocket connect timeout on Windows. This - helper is synchronous by design; ``get_status`` runs it in Starlette's - worker pool so a concurrent ``/api/ws`` handshake can still complete. - """ - from gateway.config import load_gateway_config - - gateway_config = load_gateway_config() - return {platform.value for platform in gateway_config.get_connected_platforms()} - - from hermes_cli.web_routers import status as _status_routes # noqa: E402 app.include_router(_status_routes.router) @@ -1972,88 +1661,6 @@ from hermes_cli.web_routers.status import ( # noqa: E402,F401 — legacy re-exp ) -_WINDOWS_11_MIN_BUILD = 22000 - - -def _windows_build_number(version: str, platform_label: str) -> Optional[int]: - """Extract the Windows NT build number from stdlib platform strings.""" - for value in (version or "", platform_label or ""): - match = re.search(r"(?:^|[^\d])10\.0\.(\d{5,})(?:[^\d]|$)", value) - if not match: - continue - try: - return int(match.group(1)) - except ValueError: - continue - return None - - -def _display_system_platform( - *, - system: str, - release: str, - version: str, - platform_label: str, -) -> Dict[str, str]: - """Return host OS fields for display while preserving stdlib detail.""" - if system == "Windows" and release == "10": - build = _windows_build_number(version, platform_label) - if build is not None and build >= _WINDOWS_11_MIN_BUILD: - platform_label = re.sub( - r"^Windows-10(?=-)", - "Windows-11", - platform_label, - count=1, - ) - release = "11" - - return { - "os": system, - "os_release": release, - "os_version": version, - "platform": platform_label, - } - - -# --------------------------------------------------------------------------- -# Gateway + update actions (invoked from the Status page). -# -# Both commands are spawned as detached subprocesses so the HTTP request -# returns immediately. stdin is closed (``DEVNULL``) so any stray ``input()`` -# calls fail fast with EOF rather than hanging forever. stdout/stderr are -# streamed to a per-action log file under ``~/.hermes/logs/.log`` so -# the dashboard can tail them back to the user. -# --------------------------------------------------------------------------- - -_ACTION_LOG_DIR: Path = get_hermes_home() / "logs" - -# Short ``name`` (from the URL) → absolute log file path. -_ACTION_LOG_FILES: Dict[str, str] = { - "gateway-restart": "gateway-restart.log", - "gateway-start": "gateway-start.log", - "gateway-stop": "gateway-stop.log", - "hermes-update": "hermes-update.log", - "doctor": "action-doctor.log", - "security-audit": "action-security-audit.log", - "backup": "action-backup.log", - "import": "action-import.log", - "checkpoints-prune": "action-checkpoints-prune.log", - "skills-install": "action-skills-install.log", - "skills-uninstall": "action-skills-uninstall.log", - "skills-update": "action-skills-update.log", - "curator-run": "action-curator-run.log", - "prompt-size": "action-prompt-size.log", - "dump": "action-dump.log", - "config-migrate": "action-config-migrate.log", - "tools-post-setup": "action-tools-post-setup.log", -} - -# ``name`` → most recently spawned Popen handle. Used so ``status`` can -# report liveness and exit code without shelling out to ``ps``. -_ACTION_PROCS: Dict[str, subprocess.Popen] = {} -_ACTION_COMMANDS: Dict[str, Tuple[str, ...]] = {} -_ACTION_IDS: Dict[str, str] = {} - # A finished ``gateway-restart`` child does not mean the gateway is back: the # child exits as soon as it has handed the restart to the supervisor (or to the # running gateway), while the gateway itself is still stopping and coming up. @@ -2079,133 +1686,6 @@ GATEWAY_RESTART_COOLDOWN_SECONDS = 10.0 _LAST_GATEWAY_RESTART: Optional[Tuple[float, subprocess.Popen, Tuple[str, ...]]] = None -# ``name`` → completed synthetic action result for actions the server handled -# without spawning a subprocess (for example, unsupported Docker updates). -_ACTION_RESULTS: Dict[str, Dict[str, Any]] = {} - - -def _terminate_desktop_managed_gateway() -> None: - """Stop a live gateway restart child when its Desktop backend shuts down.""" - proc = _ACTION_PROCS.get("gateway-restart") - if proc is None: - return - try: - if proc.poll() is None: - proc.terminate() - except OSError: - # The child may have exited between poll() and terminate(). - pass - - -def _dashboard_spawn_executable() -> str: - """Interpreter for detached dashboard actions. - - Prefers the install's own venv interpreter over ``sys.executable`` when - they differ. Under an SSH remote backend the web server is launched by - running the **uv base interpreter** with the venv's site-packages - injected into ``sys.path`` at startup (``-c "sys.path[:0]=[...]; - runpy.run_module('hermes_cli.main', ...)"``) — so ``sys.executable`` is - the dependency-less base python and a detached action spawned from it - dies on the first third-party import (``ModuleNotFoundError: yaml``), - because the injected path is a startup artifact of the parent and is - not inherited (#90026). The venv launcher resolves the same dependency - set on its own. - - Falls back to ``sys.executable`` when no venv interpreter exists next - to the install (in-process dev runs, exotic layouts). On Windows the - spawn below carries ``windows_detach_flags()`` (CREATE_NO_WINDOW), so - the console python owns a single hidden console that its own subprocess - spawns inherit — the action stays invisible without resorting to - console-less pythonw.exe, which would make every console-subsystem - descendant flash its own conhost (#54220/#56747). - """ - exe = Path(sys.executable) - try: - for rel in ("venv/bin/python", "venv/Scripts/python.exe"): - candidate = PROJECT_ROOT / rel - if candidate.is_file(): - # Same interpreter → keep sys.executable (preserves the - # docstring's console-ownership behavior verbatim). Compare - # UNRESOLVED normalized paths: a venv's bin/python is - # typically a SYMLINK to the base interpreter, so resolving - # both sides makes the venv python and the dependency-less - # base compare equal — exactly the SSH-runtime case this - # function exists to fix. The unresolved path IS the venv's - # identity (pyvenv.cfg discovery keys off argv0's location). - if os.path.normcase(os.path.normpath(str(candidate))) == ( - os.path.normcase(os.path.normpath(str(exe))) - ): - return sys.executable - # Return the candidate UNRESOLVED for the same reason: - # invoking the resolved target would bypass pyvenv.cfg and - # run the bare base interpreter again. - return str(candidate) - except OSError: - pass - return sys.executable - - -def _spawn_hermes_action( - subcommand: List[str], - name: str, - *, - env_overrides: Optional[Dict[str, str]] = None, -) -> subprocess.Popen: - """Spawn ``hermes `` detached and record the Popen handle. - - Uses the running interpreter's ``hermes_cli.main`` module so the action - inherits the same venv/PYTHONPATH the web server is using. - """ - log_file_name = _ACTION_LOG_FILES[name] - _ACTION_LOG_DIR.mkdir(parents=True, exist_ok=True) - log_path = _ACTION_LOG_DIR / log_file_name - log_file = open(log_path, "ab", buffering=0) - log_file.write( - f"\n=== {name} started {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode() - ) - - cmd = [_dashboard_spawn_executable(), "-m", "hermes_cli.main", *subcommand] - - # The dashboard runs *inside* the gateway process, so os.environ carries - # _HERMES_GATEWAY=1. Inheriting it makes a spawned `hermes gateway restart` - # trip the in-process restart-loop guard and exit 1 — silently failing the - # dashboard's auto-restart paths. The gateway's own restart watcher already - # drops it (gateway/run.py); mirror that here (#52470). - action_env = {**os.environ, "HERMES_NONINTERACTIVE": "1"} - action_env.pop("_HERMES_GATEWAY", None) - - popen_kwargs: Dict[str, Any] = { - "cwd": str(PROJECT_ROOT), - "stdin": subprocess.DEVNULL, - "stdout": log_file, - "stderr": subprocess.STDOUT, - "env": {**action_env, **(env_overrides or {})}, - } - if sys.platform == "win32": - popen_kwargs["creationflags"] = windows_detach_flags() - else: - popen_kwargs["start_new_session"] = True - - proc = subprocess.Popen(cmd, **popen_kwargs) - # The child inherits its own duplicated fd for stdout/stderr, so the - # parent's handle can be released immediately — otherwise we leak one - # fd per spawned action. - log_file.close() - _ACTION_RESULTS.pop(name, None) - _ACTION_COMMANDS[name] = tuple(subcommand) - _ACTION_PROCS[name] = proc - action_id = (env_overrides or {}).get("HERMES_ACTION_ID") - if action_id: - _ACTION_IDS[name] = action_id - else: - _ACTION_IDS.pop(name, None) - return proc - - -def _gateway_subcommand(profile: Optional[str], verb: str) -> List[str]: - return _profile_cli_args(profile) + ["gateway", verb] - - def _spawn_gateway_restart(profile: Optional[str] = None) -> Tuple[subprocess.Popen, bool]: """Spawn ``hermes gateway restart``, reusing an in-flight restart. @@ -2266,23 +1746,6 @@ def _spawn_gateway_restart(profile: Optional[str] = None) -> Tuple[subprocess.Po return proc, False -def _restart_gateway_after(profile: Optional[str], *, what: str, label: str) -> dict[str, Any]: - """Best-effort gateway restart after a config change (webhooks, onboarding). - - The config save stays authoritative; a failed spawn is reported in the - result (``restart_started: False`` + ``restart_error``) so the UI can fall - back to its manual restart banner instead of failing the request. - """ - try: - proc, reused = _spawn_gateway_restart(profile) - except Exception as exc: - _log.exception("Failed to auto-restart gateway after %s", what) - return {"restart_started": False, "restart_error": str(exc)} - if reused: - _log.info("%s: reusing in-flight gateway restart (pid %s)", label, proc.pid) - return {"restart_started": True, "restart_action": "gateway-restart", "restart_pid": proc.pid} - - from hermes_cli.web_routers import actions as _actions_routes # noqa: E402 app.include_router(_actions_routes.router) @@ -2318,52 +1781,9 @@ def _voice_list_error_logged_once(signature: Optional[str]) -> bool: return True -def _split_text_for_speak_stream(text: str, cap: int) -> list: - """Split *text* into provider-cap-sized pieces on sentence boundaries. - - Deliberately NOT unified with gateway.platforms.helpers' - split_text_fence_aware: this splitter reflows whitespace (sentences are - re-joined with single spaces) and has no fence/markdown semantics, so - expressing it as knobs on the fence-aware core would change behavior. - """ - from tools.tts_streaming import SENTENCE_BOUNDARY_RE as _SENTENCE_BOUNDARY_RE - - cap = cap if cap and cap > 0 else 4000 - pieces, buf = [], "" - for sentence in filter(str.strip, _SENTENCE_BOUNDARY_RE.split(text)): - while len(sentence) > cap: - pieces.append(sentence[:cap]) - sentence = sentence[cap:] - if buf and len(buf) + len(sentence) + 1 > cap: - pieces.append(buf) - buf = sentence - else: - buf = f"{buf} {sentence}" if buf else sentence - if buf: - pieces.append(buf) - return pieces - - app.include_router(_actions_routes.status_router) -# Per-row fields that no session LIST consumer reads but that dominate the -# payload. ``system_prompt`` is the fully rendered prompt — tens of KB per -# row — and made a 21-row /api/sessions response 528KB (96% dead weight), -# re-fetched by the desktop sidebar on every refresh. The desktop's -# SessionInfo type doesn't declare either field and the web UI never touches -# them; ``GET /api/sessions/{id}`` detail reads stay complete. List callers -# that genuinely need the full rows can pass ``?full=1``. -_SESSION_LIST_HEAVY_FIELDS = ("system_prompt", "model_config") - - -def _strip_session_list_rows(sessions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - for s in sessions: - for key in _SESSION_LIST_HEAVY_FIELDS: - s.pop(key, None) - return sessions - - from hermes_cli.web_routers import sessions as _sessions_routes # noqa: E402 app.include_router(_sessions_routes.list_router) diff --git a/hermes_cli/web_server_gateway.py b/hermes_cli/web_server_gateway.py new file mode 100644 index 0000000000..af87beb40e --- /dev/null +++ b/hermes_cli/web_server_gateway.py @@ -0,0 +1,646 @@ +"""Gateway/process helpers for the dashboard: per-profile gateway topology (+cache), action subprocess spawning, gateway restart plumbing, system platform display. + +Split out of ``hermes_cli.web_server``; every externally used name is re-imported +there, so ``web_server.`` keeps resolving (and monkeypatching) as before. +Helpers that tests patch on ``web_server`` are reached lazily through it. +""" + +import logging +import json +import os +import re +import subprocess +import sys +import threading +import time +import urllib.request +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple +from hermes_cli._subprocess_compat import windows_detach_flags +from hermes_cli.config import get_hermes_home + +# Same logger the code used before extraction (record parity). +_log = logging.getLogger("hermes_cli.web_server") + + +# DEPRECATED (scheduled for removal): GATEWAY_HEALTH_URL / GATEWAY_HEALTH_TIMEOUT. +# Cross-container / cross-host gateway liveness detection will be folded into a +# first-class dashboard config key so it's no longer Docker-adjacent lore buried +# in env vars. The env vars still work for now so existing Compose deployments +# don't break. Do not add new callers — wire new uses through the planned +# config surface. + + +def _probe_gateway_health() -> tuple[bool, dict | None]: + """Probe the gateway via its HTTP health endpoint (cross-container). + + .. deprecated:: + Driven by the deprecated ``GATEWAY_HEALTH_URL`` / + ``GATEWAY_HEALTH_TIMEOUT`` env vars. Scheduled for removal alongside + a move to a first-class dashboard config key. See + :data:`_GATEWAY_HEALTH_URL` for context. + + Uses ``/health/detailed`` first (returns full state), falling back to + the simpler ``/health`` endpoint. Returns ``(is_alive, body_dict)``. + + Accepts any of these as ``GATEWAY_HEALTH_URL``: + - ``http://gateway:8642`` (base URL — recommended) + - ``http://gateway:8642/health`` (explicit health path) + - ``http://gateway:8642/health/detailed`` (explicit detailed path) + + This is a **blocking** call — run via ``run_in_executor`` from async code. + """ + from hermes_cli.web_server import _GATEWAY_HEALTH_TIMEOUT, _GATEWAY_HEALTH_URL + if not _GATEWAY_HEALTH_URL: + return False, None + + # Normalise to base URL so we always probe the right paths regardless of + # whether the user included /health or /health/detailed in the env var. + base = _GATEWAY_HEALTH_URL.rstrip("/") + if base.endswith("/health/detailed"): + base = base[: -len("/health/detailed")] + elif base.endswith("/health"): + base = base[: -len("/health")] + + for path in (f"{base}/health/detailed", f"{base}/health"): + try: + req = urllib.request.Request(path, method="GET") + with urllib.request.urlopen(req, timeout=_GATEWAY_HEALTH_TIMEOUT) as resp: + if resp.status == 200: + body = json.loads(resp.read()) + return True, body + except Exception: + continue + return False, None + + +# Host TCP ports each port-binding gateway platform listens on, as +# ``platform-name -> (config port key, adapter default)``. Mirrors +# ``PORT_BINDING_PLATFORM_VALUES`` in gateway/config.py and each adapter's +# DEFAULT_PORT / DEFAULT_WEBHOOK_PORT constant. Used only for the dashboard's +# gateway-topology readout — best-effort display data, not a bind source. +_PORT_BINDING_PLATFORM_PORTS: Dict[str, Tuple[str, int]] = { + "webhook": ("port", 8644), + "api_server": ("port", 8642), + "msgraph_webhook": ("port", 8646), + "feishu": ("webhook_port", 8765), + "wecom_callback": ("port", 8645), + "bluebubbles": ("webhook_port", 8645), + "sms": ("webhook_port", 8080), + "whatsapp_cloud": ("webhook_port", 8090), + "line": ("port", 8646), + "teams": ("port", 3978), +} + +# Platform states that mean the adapter is NOT serving its port right now. +_PLATFORM_DEAD_STATES = frozenset({"fatal", "disconnected", "stopped"}) + + +def _profile_platform_ports(profile_home: Path, runtime: Optional[dict]) -> Dict[str, int]: + """Best-effort map of ``platform -> host TCP port`` for one profile's gateway. + + Reads the platforms the running gateway reported in its + ``gateway_state.json`` and resolves each port-binding platform's port from + the profile's ``config.yaml`` (top-level ``platforms:`` wins over + ``gateway.platforms:``, matching ``load_gateway_config`` precedence), + falling back to the adapter default. Display-only: env-var port overrides + (e.g. ``WEBHOOK_PORT`` in that profile's .env) are not resolved here. + """ + platforms = (runtime or {}).get("platforms") or {} + active = [ + name for name, state in platforms.items() + if name in _PORT_BINDING_PLATFORM_PORTS + and isinstance(state, dict) + and state.get("state") not in _PLATFORM_DEAD_STATES + ] + if not active: + return {} + + blocks: Dict[str, dict] = {} + try: + # Multi-profile probe: load_config() targets the ACTIVE profile's + # home, so read the probed profile's file via the raw primitive. + from hermes_cli.config import read_user_config_raw + cfg = read_user_config_raw(profile_home / "config.yaml") + gateway_cfg = cfg.get("gateway") if isinstance(cfg.get("gateway"), dict) else {} + # gateway.platforms first, top-level platforms second — later wins, + # matching the precedence in gateway.config.load_gateway_config(). + for src in ((gateway_cfg or {}).get("platforms"), cfg.get("platforms")): + if not isinstance(src, dict): + continue + for plat_name, plat_block in src.items(): + if isinstance(plat_block, dict): + blocks.setdefault(plat_name, {}).update(plat_block) + except Exception: + blocks = {} + + ports: Dict[str, int] = {} + for name in active: + port_key, default_port = _PORT_BINDING_PLATFORM_PORTS[name] + block = blocks.get(name) or {} + extra = block.get("extra") if isinstance(block.get("extra"), dict) else {} + raw = block.get(port_key, (extra or {}).get(port_key, default_port)) + try: + ports[name] = int(raw) + except (TypeError, ValueError): + ports[name] = default_port + return ports + + +def _profile_gateway_writer_identity( + profile_home: Path, runtime: Optional[dict] +) -> Optional[tuple]: + """``(pid, start_time)`` identity of the profile's LIVE gateway, or None. + + Reuses the validated-liveness helper — recorded PID checked against the + live process table, the start-time PID-reuse fingerprint, and the + profile's home — then reads the live process's fingerprint via the same + ``_get_process_start_time`` that stamped it, so equality is exact (no + unit or clock-source mismatch). None when the record doesn't belong to + a live gateway; nothing in it is current by definition then. + """ + try: + from gateway.status import ( + _get_process_start_time, + get_runtime_status_running_pid, + ) + + pid = get_runtime_status_running_pid(runtime, expected_home=profile_home) + if pid is None: + return None + start_time = _get_process_start_time(pid) + if start_time is None: + return None + return (pid, start_time) + except Exception: + return None + + +def _owned_profile_platforms( + writer_identity: Optional[tuple], platforms: dict +) -> dict: + """Keep only platform entries the profile's CURRENT process wrote. + + Gateway startup deliberately preserves plain platform entries in + ``gateway_state.json`` across restarts (the dashboard keeps showing + last-known state while adapters reconnect), and the active-profile + endpoint compensates by filtering them against the current + configuration. The cross-profile aggregation has no equivalent config + context (a profile's platform set depends on tokens in that profile's + ``.env`` behind its secret scope), so it demands strict process + ownership instead: ``write_runtime_status`` stamps every platform write + with the writer's ``(pid, start_time)`` identity, and an entry is + aggregatable only when that identity equals the profile's live gateway + process — exact match, no clock heuristics, so an entry written moments + before a fast restart can never masquerade as current. A fatal entry + left behind by a platform the operator has since disabled/removed thus + stops degrading fleet health as soon as that profile's gateway restarts + (a config change requires that restart to take effect anyway). Fail + closed: entries without a writer identity (legacy records) or records + with no live process are excluded — aggregation is a supplement, and a + false "degraded forever" is the worse failure mode. + """ + if writer_identity is None: + return {} + live_pid, live_start = writer_identity + owned: Dict[str, dict] = {} + for key, value in platforms.items(): + if not isinstance(value, dict): + continue + if ( + value.get("writer_pid") == live_pid + and value.get("writer_start_time") == live_start + ): + owned[key] = value + return owned + + +def _collect_profile_gateway_topology() -> Dict[str, Any]: + """Enumerate profiles and the gateways serving them for ``/api/status``. + + Returns ``{"profiles": [...], "gateway_mode": ..., "gateways": [...]}``: + + * ``profiles`` — every profile on the host (default + named), from + ``profiles_to_serve(True)`` (the cheap enumeration chokepoint — no + per-profile config reads or skill counts). + * ``gateways`` — one entry per profile with a LIVE gateway process: + ``{"profile", "ports", "served_profiles"?}``. Liveness reuses + ``_check_gateway_running`` so this agrees with the profiles sidebar. + * ``gateway_mode`` — ``"multiplex"`` when the default gateway serves + multiple profiles (gateway.multiplex_profiles), ``"single"`` for one + live gateway, ``"multiple"`` for independent per-profile gateways, + ``"none"`` when nothing is running. + * ``profile_platforms`` — ``{profile: platforms}`` runtime platform maps + for each LIVE gateway, ownership-filtered to entries stamped by that + profile's current process (stale preserved entries for since-removed + platforms are excluded — see ``_owned_profile_platforms``). Internal + aggregation input for ``/api/status`` (independent per-profile gateways + write failures to their own ``gateway_state.json``, which the + unparameterized endpoint would otherwise never see). Never exposed + directly. + """ + from hermes_cli.web_server import _profile_gateway_writer_identity + try: + from hermes_cli.profiles import _check_gateway_running, profiles_to_serve + from gateway.status import read_runtime_status + homes = profiles_to_serve(True) + except Exception: + _log.debug("profile/gateway topology enumeration failed", exc_info=True) + return { + "profiles": [], + "gateway_mode": "unknown", + "gateways": [], + "profile_platforms": {}, + } + + profile_names = [name for name, _home in homes] + gateways: List[Dict[str, Any]] = [] + profile_platforms: Dict[str, dict] = {} + multiplex = False + for name, home in homes: + try: + if not _check_gateway_running(home): + continue + except Exception: + continue + try: + runtime = read_runtime_status(home / "gateway_state.json") + except Exception: + runtime = None + served = [str(p) for p in ((runtime or {}).get("served_profiles") or [])] + if name == "default" and len(served) > 1: + multiplex = True + plats = (runtime or {}).get("platforms") + if isinstance(plats, dict) and plats: + # Ownership filter: gateway startup preserves plain platform + # entries across restarts, so the raw map can carry fatal state + # for platforms the operator has since disabled/removed. Only + # entries stamped with the profile's current live process's + # writer identity are aggregation candidates (see + # _owned_profile_platforms). + owned = _owned_profile_platforms( + _profile_gateway_writer_identity(home, runtime), plats + ) + if owned: + profile_platforms[name] = owned + entry: Dict[str, Any] = { + "profile": name, + "ports": _profile_platform_ports(home, runtime), + } + if served: + entry["served_profiles"] = served + gateways.append(entry) + + if multiplex: + mode = "multiplex" + elif len(gateways) > 1: + mode = "multiple" + elif len(gateways) == 1: + mode = "single" + else: + mode = "none" + + return { + "profiles": profile_names, + "gateway_mode": mode, + "gateways": gateways, + "profile_platforms": profile_platforms, + } + + +# /api/status is polled ~1/s by the desktop app while it waits for the backend +# (and again by the dashboard badge). Each uncached call above walks 7+ profile +# homes (yaml.safe_load with the pure-Python loader + psutil process-table +# probes + realpath walks) inside the default executor; concurrent polls pile +# up and hold the GIL for 14-16s, starving the event loop — the desktop WS +# never receives gateway.ready and boot fails ("event loop stalled ... GIL +# pressure suspected"). Topology changes on gateway start/stop, so a short TTL +# cache with a collapse lock keeps the scan to one per window. The cache also +# remembers which collector produced the entry: tests monkeypatch +# _collect_profile_gateway_topology per case, and the identity check keeps +# them hermetic without needing a reset hook (a swapped collector is a miss). +_TOPOLOGY_CACHE: Dict[str, Any] = {"ts": 0.0, "data": None, "fn": None} +_TOPOLOGY_CACHE_LOCK = threading.Lock() +_TOPOLOGY_CACHE_TTL = 10.0 + + +def _topology_cache_get(fn: Any) -> Optional[Dict[str, Any]]: + if ( + _TOPOLOGY_CACHE["data"] is not None + and _TOPOLOGY_CACHE["fn"] is fn + and time.monotonic() - _TOPOLOGY_CACHE["ts"] < _TOPOLOGY_CACHE_TTL + ): + return _TOPOLOGY_CACHE["data"] + return None + + +def _collect_profile_gateway_topology_cached() -> Dict[str, Any]: + from hermes_cli.web_server import _collect_profile_gateway_topology + fn = _collect_profile_gateway_topology + cached = _topology_cache_get(fn) + if cached is not None: + return cached + with _TOPOLOGY_CACHE_LOCK: + cached = _topology_cache_get(fn) + if cached is not None: + return cached + data = fn() + _TOPOLOGY_CACHE["data"] = data + _TOPOLOGY_CACHE["fn"] = fn + _TOPOLOGY_CACHE["ts"] = time.monotonic() + return data + + +def _load_configured_gateway_platforms() -> set[str]: + """Load connected platform names away from the asyncio event loop. + + The first ``load_gateway_config()`` call performs platform discovery and + can take longer than Desktop's WebSocket connect timeout on Windows. This + helper is synchronous by design; ``get_status`` runs it in Starlette's + worker pool so a concurrent ``/api/ws`` handshake can still complete. + """ + from gateway.config import load_gateway_config + + gateway_config = load_gateway_config() + return {platform.value for platform in gateway_config.get_connected_platforms()} + + +_WINDOWS_11_MIN_BUILD = 22000 + + +def _windows_build_number(version: str, platform_label: str) -> Optional[int]: + """Extract the Windows NT build number from stdlib platform strings.""" + for value in (version or "", platform_label or ""): + match = re.search(r"(?:^|[^\d])10\.0\.(\d{5,})(?:[^\d]|$)", value) + if not match: + continue + try: + return int(match.group(1)) + except ValueError: + continue + return None + + +def _display_system_platform( + *, + system: str, + release: str, + version: str, + platform_label: str, +) -> Dict[str, str]: + """Return host OS fields for display while preserving stdlib detail.""" + if system == "Windows" and release == "10": + build = _windows_build_number(version, platform_label) + if build is not None and build >= _WINDOWS_11_MIN_BUILD: + platform_label = re.sub( + r"^Windows-10(?=-)", + "Windows-11", + platform_label, + count=1, + ) + release = "11" + + return { + "os": system, + "os_release": release, + "os_version": version, + "platform": platform_label, + } + + +# --------------------------------------------------------------------------- +# Gateway + update actions (invoked from the Status page). +# +# Both commands are spawned as detached subprocesses so the HTTP request +# returns immediately. stdin is closed (``DEVNULL``) so any stray ``input()`` +# calls fail fast with EOF rather than hanging forever. stdout/stderr are +# streamed to a per-action log file under ``~/.hermes/logs/.log`` so +# the dashboard can tail them back to the user. +# --------------------------------------------------------------------------- + +_ACTION_LOG_DIR: Path = get_hermes_home() / "logs" + +# Short ``name`` (from the URL) → absolute log file path. +_ACTION_LOG_FILES: Dict[str, str] = { + "gateway-restart": "gateway-restart.log", + "gateway-start": "gateway-start.log", + "gateway-stop": "gateway-stop.log", + "hermes-update": "hermes-update.log", + "doctor": "action-doctor.log", + "security-audit": "action-security-audit.log", + "backup": "action-backup.log", + "import": "action-import.log", + "checkpoints-prune": "action-checkpoints-prune.log", + "skills-install": "action-skills-install.log", + "skills-uninstall": "action-skills-uninstall.log", + "skills-update": "action-skills-update.log", + "curator-run": "action-curator-run.log", + "prompt-size": "action-prompt-size.log", + "dump": "action-dump.log", + "config-migrate": "action-config-migrate.log", + "tools-post-setup": "action-tools-post-setup.log", +} + +# ``name`` → most recently spawned Popen handle. Used so ``status`` can +# report liveness and exit code without shelling out to ``ps``. +_ACTION_PROCS: Dict[str, subprocess.Popen] = {} +_ACTION_COMMANDS: Dict[str, Tuple[str, ...]] = {} +_ACTION_IDS: Dict[str, str] = {} + +# ``name`` → completed synthetic action result for actions the server handled +# without spawning a subprocess (for example, unsupported Docker updates). +_ACTION_RESULTS: Dict[str, Dict[str, Any]] = {} + + +def _terminate_desktop_managed_gateway() -> None: + """Stop a live gateway restart child when its Desktop backend shuts down.""" + from hermes_cli.web_server import _ACTION_PROCS + proc = _ACTION_PROCS.get("gateway-restart") + if proc is None: + return + try: + if proc.poll() is None: + proc.terminate() + except OSError: + # The child may have exited between poll() and terminate(). + pass + + +def _dashboard_spawn_executable() -> str: + """Interpreter for detached dashboard actions. + + Prefers the install's own venv interpreter over ``sys.executable`` when + they differ. Under an SSH remote backend the web server is launched by + running the **uv base interpreter** with the venv's site-packages + injected into ``sys.path`` at startup (``-c "sys.path[:0]=[...]; + runpy.run_module('hermes_cli.main', ...)"``) — so ``sys.executable`` is + the dependency-less base python and a detached action spawned from it + dies on the first third-party import (``ModuleNotFoundError: yaml``), + because the injected path is a startup artifact of the parent and is + not inherited (#90026). The venv launcher resolves the same dependency + set on its own. + + Falls back to ``sys.executable`` when no venv interpreter exists next + to the install (in-process dev runs, exotic layouts). On Windows the + spawn below carries ``windows_detach_flags()`` (CREATE_NO_WINDOW), so + the console python owns a single hidden console that its own subprocess + spawns inherit — the action stays invisible without resorting to + console-less pythonw.exe, which would make every console-subsystem + descendant flash its own conhost (#54220/#56747). + """ + from hermes_cli.web_server import PROJECT_ROOT + exe = Path(sys.executable) + try: + for rel in ("venv/bin/python", "venv/Scripts/python.exe"): + candidate = PROJECT_ROOT / rel + if candidate.is_file(): + # Same interpreter → keep sys.executable (preserves the + # docstring's console-ownership behavior verbatim). Compare + # UNRESOLVED normalized paths: a venv's bin/python is + # typically a SYMLINK to the base interpreter, so resolving + # both sides makes the venv python and the dependency-less + # base compare equal — exactly the SSH-runtime case this + # function exists to fix. The unresolved path IS the venv's + # identity (pyvenv.cfg discovery keys off argv0's location). + if os.path.normcase(os.path.normpath(str(candidate))) == ( + os.path.normcase(os.path.normpath(str(exe))) + ): + return sys.executable + # Return the candidate UNRESOLVED for the same reason: + # invoking the resolved target would bypass pyvenv.cfg and + # run the bare base interpreter again. + return str(candidate) + except OSError: + pass + return sys.executable + + +def _spawn_hermes_action( + subcommand: List[str], + name: str, + *, + env_overrides: Optional[Dict[str, str]] = None, +) -> subprocess.Popen: + """Spawn ``hermes `` detached and record the Popen handle. + + Uses the running interpreter's ``hermes_cli.main`` module so the action + inherits the same venv/PYTHONPATH the web server is using. + """ + from hermes_cli.web_server import ( + PROJECT_ROOT, + _ACTION_COMMANDS, + _ACTION_IDS, + _ACTION_LOG_DIR, + _ACTION_PROCS, + _ACTION_RESULTS, + ) + log_file_name = _ACTION_LOG_FILES[name] + _ACTION_LOG_DIR.mkdir(parents=True, exist_ok=True) + log_path = _ACTION_LOG_DIR / log_file_name + log_file = open(log_path, "ab", buffering=0) + log_file.write( + f"\n=== {name} started {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode() + ) + + cmd = [_dashboard_spawn_executable(), "-m", "hermes_cli.main", *subcommand] + + # The dashboard runs *inside* the gateway process, so os.environ carries + # _HERMES_GATEWAY=1. Inheriting it makes a spawned `hermes gateway restart` + # trip the in-process restart-loop guard and exit 1 — silently failing the + # dashboard's auto-restart paths. The gateway's own restart watcher already + # drops it (gateway/run.py); mirror that here (#52470). + action_env = {**os.environ, "HERMES_NONINTERACTIVE": "1"} + action_env.pop("_HERMES_GATEWAY", None) + + popen_kwargs: Dict[str, Any] = { + "cwd": str(PROJECT_ROOT), + "stdin": subprocess.DEVNULL, + "stdout": log_file, + "stderr": subprocess.STDOUT, + "env": {**action_env, **(env_overrides or {})}, + } + if sys.platform == "win32": + popen_kwargs["creationflags"] = windows_detach_flags() + else: + popen_kwargs["start_new_session"] = True + + proc = subprocess.Popen(cmd, **popen_kwargs) + # The child inherits its own duplicated fd for stdout/stderr, so the + # parent's handle can be released immediately — otherwise we leak one + # fd per spawned action. + log_file.close() + _ACTION_RESULTS.pop(name, None) + _ACTION_COMMANDS[name] = tuple(subcommand) + _ACTION_PROCS[name] = proc + action_id = (env_overrides or {}).get("HERMES_ACTION_ID") + if action_id: + _ACTION_IDS[name] = action_id + else: + _ACTION_IDS.pop(name, None) + return proc + + +def _gateway_subcommand(profile: Optional[str], verb: str) -> List[str]: + from hermes_cli.web_server import _profile_cli_args + return _profile_cli_args(profile) + ["gateway", verb] + + +def _restart_gateway_after(profile: Optional[str], *, what: str, label: str) -> dict[str, Any]: + """Best-effort gateway restart after a config change (webhooks, onboarding). + + The config save stays authoritative; a failed spawn is reported in the + result (``restart_started: False`` + ``restart_error``) so the UI can fall + back to its manual restart banner instead of failing the request. + """ + from hermes_cli.web_server import _spawn_gateway_restart + try: + proc, reused = _spawn_gateway_restart(profile) + except Exception as exc: + _log.exception("Failed to auto-restart gateway after %s", what) + return {"restart_started": False, "restart_error": str(exc)} + if reused: + _log.info("%s: reusing in-flight gateway restart (pid %s)", label, proc.pid) + return {"restart_started": True, "restart_action": "gateway-restart", "restart_pid": proc.pid} + + +def _split_text_for_speak_stream(text: str, cap: int) -> list: + """Split *text* into provider-cap-sized pieces on sentence boundaries. + + Deliberately NOT unified with gateway.platforms.helpers' + split_text_fence_aware: this splitter reflows whitespace (sentences are + re-joined with single spaces) and has no fence/markdown semantics, so + expressing it as knobs on the fence-aware core would change behavior. + """ + from tools.tts_streaming import SENTENCE_BOUNDARY_RE as _SENTENCE_BOUNDARY_RE + + cap = cap if cap and cap > 0 else 4000 + pieces, buf = [], "" + for sentence in filter(str.strip, _SENTENCE_BOUNDARY_RE.split(text)): + while len(sentence) > cap: + pieces.append(sentence[:cap]) + sentence = sentence[cap:] + if buf and len(buf) + len(sentence) + 1 > cap: + pieces.append(buf) + buf = sentence + else: + buf = f"{buf} {sentence}" if buf else sentence + if buf: + pieces.append(buf) + return pieces + + +# Per-row fields that no session LIST consumer reads but that dominate the +# payload. ``system_prompt`` is the fully rendered prompt — tens of KB per +# row — and made a 21-row /api/sessions response 528KB (96% dead weight), +# re-fetched by the desktop sidebar on every refresh. The desktop's +# SessionInfo type doesn't declare either field and the web UI never touches +# them; ``GET /api/sessions/{id}`` detail reads stay complete. List callers +# that genuinely need the full rows can pass ``?full=1``. +_SESSION_LIST_HEAVY_FIELDS = ("system_prompt", "model_config") + + +def _strip_session_list_rows(sessions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + for s in sessions: + for key in _SESSION_LIST_HEAVY_FIELDS: + s.pop(key, None) + return sessions