hermes-agent/hermes_cli/web_server_gateway.py

"""Gateway/process helpers for the dashboard: per-profile gateway topology (+cache), action
subprocess spawning, gateway restart plumbing, system platform display.
"""

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, windows_detach_flags_without_breakaway
from hermes_cli.config import get_hermes_home

# Same logger the code used before extraction (record parity).
_log = logging.getLogger("hermes_cli.web_server")


# Defense against a hostile/compromised upstream serving an unbounded body to
# the dashboard's JSON readers: refuse to buffer past this size.
_DASHBOARD_JSON_RESPONSE_BODY_MAX_BYTES = 1024 * 1024


def _read_dashboard_json_response(response) -> Any:
    """Read a JSON HTTP response body with a hard size cap (issue #54808)."""
    raw = response.read(_DASHBOARD_JSON_RESPONSE_BODY_MAX_BYTES + 1)
    if len(raw) > _DASHBOARD_JSON_RESPONSE_BODY_MAX_BYTES:
        raise ValueError(
            "dashboard HTTP JSON response exceeded "
            f"{_DASHBOARD_JSON_RESPONSE_BODY_MAX_BYTES} bytes"
        )
    return json.loads(raw.decode("utf-8"))


def _probe_gateway_health() -> tuple[bool, dict | None]:
    """Probe the gateway's HTTP health endpoint (cross-container). Blocking — run in an executor.

    DEPRECATED: driven by the ``GATEWAY_HEALTH_URL`` / ``GATEWAY_HEALTH_TIMEOUT`` env vars,
    to be replaced by a dashboard config key; do not add callers. Accepts a base URL or an
    explicit ``/health`` / ``/health/detailed`` path; tries ``/health/detailed`` first.
    """
    from hermes_cli.web_server import _GATEWAY_HEALTH_TIMEOUT, _GATEWAY_HEALTH_URL
    if not _GATEWAY_HEALTH_URL:
        return False, None
    base = re.sub(r"/health(/detailed)?$", "", _GATEWAY_HEALTH_URL.rstrip("/"))
    # Bearer credential for an auth-protected gateway (issue #76051): sent to the
    # authenticated /health/detailed probe ONLY — never to the public /health fallback,
    # which is unauthenticated by design.
    api_key = os.environ.get("API_SERVER_KEY", "").strip()
    for path in (f"{base}/health/detailed", f"{base}/health"):
        try:
            req = urllib.request.Request(path, method="GET")
            if api_key and path.endswith("/health/detailed"):
                req.add_header("Authorization", f"Bearer {api_key}")
            with urllib.request.urlopen(req, timeout=_GATEWAY_HEALTH_TIMEOUT) as resp:
                if resp.status == 200:
                    return True, _read_dashboard_json_response(resp)
        except Exception:
            continue
    return False, None


# ``platform-name -> (config port key, adapter default)`` for port-binding gateway platforms.
# Mirrors PORT_BINDING_PLATFORM_VALUES (gateway/config.py) and each adapter's DEFAULT_PORT /
# DEFAULT_WEBHOOK_PORT. Display-only data for the topology readout, 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 ``platform -> host TCP port`` for one profile's live gateway.

    Ports come from the profile's own config.yaml (``gateway.platforms`` then top-level
    ``platforms`` — later wins, matching load_gateway_config precedence), falling back to the
    adapter default. Env-var overrides (e.g. WEBHOOK_PORT in that profile's .env) are not resolved.
    """
    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:
        # load_config() targets the ACTIVE profile's home; read the probed profile's file raw.
        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 {}
        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)`` of the profile's LIVE gateway, or None.

    Uses the same validated-liveness helper and the same ``_get_process_start_time`` that stamped
    the record, so equality is exact (no unit/clock-source mismatch).
    """
    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)
        return None if start_time is None else (pid, start_time)
    except Exception:
        return None


def _owned_profile_platforms(writer_identity: Optional[tuple], platforms: dict) -> dict:
    """Keep only platform entries stamped by the profile's CURRENT gateway process.

    Gateway startup preserves plain platform entries in gateway_state.json across restarts, so the
    raw map can carry fatal state for platforms since disabled/removed. Cross-profile aggregation
    has no config context to filter against, so it demands exact ``(pid, start_time)`` writer
    identity instead. Fail closed: legacy entries without identity, or no live process, yield {} —
    a false "degraded forever" is the worse failure mode.
    """
    if writer_identity is None:
        return {}
    live_pid, live_start = writer_identity
    return {
        key: value for key, value in platforms.items()
        if isinstance(value, dict)
        and value.get("writer_pid") == live_pid
        and value.get("writer_start_time") == live_start}


def _collect_profile_gateway_topology() -> dict[str, Any]:
    """Enumerate profiles and the gateways serving them for ``/api/status``.

    Returns ``profiles`` (all profile names via the cheap ``profiles_to_serve(True)`` chokepoint),
    ``gateways`` (one ``{"profile", "ports", "served_profiles"?}`` per LIVE gateway; liveness via
    ``_check_gateway_running`` so it agrees with the sidebar), ``gateway_mode``
    (multiplex / single / multiple / none) and ``profile_platforms`` — ownership-filtered runtime
    platform maps per live gateway, an internal aggregation input never exposed directly.
    """
    try:
        from hermes_cli.profiles import _check_gateway_running, profiles_to_serve, profile_is_parked
        from gateway.status import read_runtime_status
        homes = profiles_to_serve(True, include_standalone=True, include_parked=True)
    except Exception:
        _log.debug("profile/gateway topology enumeration failed", exc_info=True)
        return {"profiles": [], "gateway_mode": "unknown", "gateways": [], "profile_platforms": {}}

    gateways: list[dict[str, Any]] = []
    profile_platforms: dict[str, dict] = {}
    multiplex = False
    standalone_reason: Optional[str] = None
    for name, home in homes:
        try:
            # A served profile's liveness is the multiplexer's: listing it here showed one phantom
            # gateway per served profile beside the host.
            if not (_check_gateway_running(home) if name == "default" else _has_own_gateway(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
        if (runtime or {}).get("multiplex_standalone_reason"):
            standalone_reason = str(runtime["multiplex_standalone_reason"])
        plats = (runtime or {}).get("platforms")
        owned: dict = {}
        if isinstance(plats, dict) and plats:
            owned = _owned_profile_platforms(_profile_gateway_writer_identity(home, runtime), plats)
            if owned:
                profile_platforms[name] = owned
        # Ports from the OWNED entries too: a platform entry a previous process left "connected"
        # reported a port the live gateway does not bind.
        entry: dict[str, Any] = {"profile": name, "ports": _profile_platform_ports(home, {"platforms": owned})}
        if served:
            entry["served_profiles"] = served
        gateways.append(entry)

    if multiplex:
        mode = "multiplex"
    else:
        mode = {0: "none", 1: "single"}.get(len(gateways), "multiple")
    # A guard refusal on a multi-profile host is what the dashboard banner shows; a single-profile
    # install has nothing unserved and gets no banner.
    from hermes_cli.gateway_multiplex_mode import SINGLE_PROFILE_REASON
    if standalone_reason == SINGLE_PROFILE_REASON or len(homes) < 2:
        standalone_reason = None
    return {
        "profiles": [name for name, _home in homes],
        "parked_profiles": [name for name, home in homes if name != "default" and profile_is_parked(home)],
        "gateway_mode": mode,
        "multiplex_standalone_reason": standalone_reason,
        "gateways": gateways,
        "profile_platforms": profile_platforms}


# /api/status is polled ~1/s by the desktop app while it waits for the backend. Each uncached
# collect walks 7+ profile homes (pure-Python yaml + psutil + realpath) in the default executor;
# concurrent polls pile up and hold the GIL for 14-16s, starving the loop so the desktop WS never
# gets gateway.ready. A short TTL cache with a collapse lock keeps the scan to one per window.
# The cache remembers which collector produced the entry: tests monkeypatch
# _collect_profile_gateway_topology per case, and a swapped collector is a miss (no reset hook).
_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]]:
    c = _TOPOLOGY_CACHE
    fresh = c["fn"] is fn and time.monotonic() - c["ts"] < _TOPOLOGY_CACHE_TTL
    return c["data"] if fresh and c["data"] is not None else 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.update(data=data, fn=fn, ts=time.monotonic())
        return data


def _load_configured_gateway_platforms() -> set[str]:
    """Connected platform names; synchronous by design — the first ``load_gateway_config()`` does
    platform discovery and can outlast Desktop's WS connect timeout on Windows, so ``get_status``
    runs this in Starlette's worker pool."""
    from gateway.config import load_gateway_config
    return {platform.value for platform in load_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 match:
            return int(match.group(1))
    return None


def _display_system_platform(*, system: str, release: str, version: str, platform_label: str) -> dict[str, str]:
    """Host OS fields for display; Windows 10 builds >= 22000 are relabelled Windows 11."""
    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). Spawned detached so the request
# returns immediately; stdin is DEVNULL so stray input() fails fast; stdout/stderr stream to
# ~/.hermes/logs/<action>.log which the dashboard tails.

_ACTION_LOG_DIR: Path = get_hermes_home() / "logs"

# Short ``name`` (from the URL) → log file name under _ACTION_LOG_DIR.
_ACTION_LOG_FILES: dict[str, str] = {
    "gateway-restart": "gateway-restart.log",
    "gateway-start": "gateway-start.log",
    "gateway-stop": "gateway-stop.log",
    "gateway-migrate": "gateway-migrate.log",
    "hermes-update": "hermes-update.log",
    **{name: f"action-{name}.log" for name in (
        "doctor", "security-audit", "backup", "import", "checkpoints-prune", "skills-install",
        "skills-uninstall", "skills-update", "curator-run", "prompt-size", "dump", "config-migrate",
        "tools-post-setup",
    )},
}

# ``name`` → most recent Popen handle / argv / action id, so ``status`` needs no ``ps``.
_ACTION_PROCS: dict[str, subprocess.Popen] = {}
_ACTION_COMMANDS: dict[str, tuple[str, ...]] = {}
_ACTION_IDS: dict[str, str] = {}
# ``name`` → synthetic result for actions handled without a subprocess (e.g. unsupported Docker updates).
_ACTION_RESULTS: dict[str, dict[str, Any]] = {}
# ``name`` → (pid, byte offset of that child's section in its log): a later spawn that fails after
# writing its own header must not lend its text to the child that is still on record.
_ACTION_LOG_STARTS: dict[str, tuple[int, int]] = {}


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:
        pass  # exited between poll() and terminate()


def _named_profile_from_action(subcommand: list[str]) -> Optional[str]:
    """Return the named-profile selector that :func:`_profile_cli_args` puts in front of an action.

    Deliberately inspects only the leading selector: values after the real subcommand may
    legitimately contain ``-p`` / ``--profile`` for a nested process (``mcp add --args ...``).
    """
    if len(subcommand) >= 2 and subcommand[0] in {"-p", "--profile"}:
        return str(subcommand[1]).strip() or None
    if subcommand and str(subcommand[0]).startswith("--profile="):
        return str(subcommand[0]).split("=", 1)[1].strip() or None
    return None


def _is_host_gateway_spawn(subcommand: list[str]) -> bool:
    """True when *subcommand* starts the host multiplexer, not a named profile's own gateway.

    ``hermes gateway restart`` and ``hermes -p default gateway restart`` are the host.
    ``hermes -p coder gateway stop`` is not.
    """
    profile = _named_profile_from_action(subcommand)
    if profile not in (None, "default"):
        return False
    args = list(subcommand)
    if profile is not None:
        if args and args[0] in {"-p", "--profile"}:
            args = args[2:]
        elif args and str(args[0]).startswith("--profile="):
            args = args[1:]
    return bool(args) and args[0] == "gateway"


def _profile_action_environment(
    subcommand: list[str], env_overrides: Optional[dict[str, str]] = None,
) -> dict[str, str]:
    """Environment for a detached ``hermes <subcommand>`` action.

    The dashboard loads its own profile's ``.env`` into process-global ``os.environ``. Copying
    that mapping verbatim into ``hermes -p <other> ...`` lets the named child see the dashboard
    profile's platform credentials and ports *before* its own dotenv loads (``load_hermes_dotenv``
    does not override keys already present): a supposedly A2A-only profile then claims the default
    Discord token and binds the default API/BlueBubbles ports.

    Named-profile actions therefore start from Hermes' standard scrubbed subprocess env, then drop
    the profile-managed keys plus every key declared by the dashboard/default profile dotenv files
    and their hydrated secret sources, and pin ``HERMES_HOME`` to the target profile. The child's
    normal startup then loads that profile's own ``.env``. A host-gateway verb (bare ``gateway``
    or ``-p default gateway``) starts from ``host_gateway_child_env`` so a named-profile dashboard
    cannot donate its dotenv to the multiplexer. Other actions without a profile selector keep
    the historical environment exactly.
    """
    profile = _named_profile_from_action(subcommand)
    if _is_host_gateway_spawn(subcommand):
        from tools.environments.local import host_gateway_child_env
        action_env = host_gateway_child_env()
    elif profile is None:
        action_env = dict(os.environ)
    else:
        from hermes_cli.env_loader import (
            _PROFILE_MANAGED_ENV_KEYS, _env_keys_defined_in_dotenv, get_secret_source_values,
        )
        from hermes_cli.web_server_profiles import _resolve_profile_dir
        from hermes_constants import apply_subprocess_home_env, get_default_hermes_root
        from tools.environments.local import build_subprocess_env, strip_launch_profile_env

        target_home = _resolve_profile_dir(profile)
        action_env = build_subprocess_env(base=os.environ, scrub_secrets=True)

        profile_keys = set(_PROFILE_MANAGED_ENV_KEYS)
        try:
            source_homes = {str(get_default_hermes_root()), str(get_hermes_home())}
        except Exception:
            source_homes = set()
        for source_home in source_homes:
            profile_keys.update(_env_keys_defined_in_dotenv(Path(source_home) / ".env"))
            # Secret managers contribute locally named credentials that never appear in .env;
            # the dashboard already hydrated its own sources, so their key names are a boundary too.
            profile_keys.update(get_secret_source_values(source_home).keys())
        for key in profile_keys:
            action_env.pop(key, None)
        # Authorization gates that reached this process outside any dotenv (unit-file
        # ``Environment=``, an operator export) are not in ``profile_keys``; the target
        # profile's ``.env`` rarely defines them, so they would survive into the child (#113270).
        strip_launch_profile_env(action_env, target_home)

        # Pin the child before import-time startup runs; the explicit -p flag stays authoritative
        # and resolves to the same validated directory.
        action_env["HERMES_HOME"] = str(target_home)
        apply_subprocess_home_env(action_env)

    action_env["HERMES_NONINTERACTIVE"] = "1"
    # A config.yaml allow_all_users grant bridged into os.environ must not outlive the config that
    # produced it: drop it so the restarted child re-derives the posture from its own config.yaml.
    from gateway.config_loader import drop_bridged_env
    drop_bridged_env(action_env)
    # The dashboard runs inside the gateway process, so os.environ carries _HERMES_GATEWAY=1;
    # inheriting it trips the child's in-process restart-loop guard (exit 1). Drop it, like
    # the gateway's own restart watcher does (gateway/run.py, #52470).
    action_env.pop("_HERMES_GATEWAY", None)
    if env_overrides:
        action_env.update(env_overrides)
    return action_env


# Gateway lifecycle verbs the CLI refuses below root on a system-scope install
# (``gateway.py::_require_root_for_system_service``).
_ROOT_REQUIRING_GATEWAY_VERBS = frozenset({"restart", "start", "stop"})


def _action_targets_system_gateway(subcommand: list[str]) -> bool:
    """True when *subcommand* is a gateway lifecycle verb that resolves to the SYSTEM unit.

    Scope is decided by the CLI's own picker (``_select_systemd_scope``) evaluated for the profile
    the action addresses, not by "a system unit exists": a host carrying both units resolves to the
    user unit, which the dashboard user operates unelevated. Same root/sudo posture as the
    ``hermes update`` fleet restart (``update_cmd_fleet._needs_sudo`` / ``_sudo_noninteractive_ok``).
    """
    from hermes_cli.update_cmd_fleet import _needs_sudo

    if not _needs_sudo("system"):
        return False
    try:
        verb = subcommand[subcommand.index("gateway") + 1]
    except (ValueError, IndexError):
        return False
    if verb not in _ROOT_REQUIRING_GATEWAY_VERBS:
        return False

    from hermes_cli.gateway import _select_systemd_scope
    from hermes_constants import reset_hermes_home_override, set_hermes_home_override

    profile = _named_profile_from_action(subcommand)
    if profile is None:
        return _select_systemd_scope(False)
    # Unit names are derived from HERMES_HOME, so a selector-bearing action must be resolved
    # against the TARGET profile's home (``-p default gateway restart`` from a pooled named
    # dashboard asks about the default unit, not about its own).
    from hermes_cli.web_server_profiles import _resolve_profile_dir
    token = set_hermes_home_override(_resolve_profile_dir(profile))
    try:
        return _select_systemd_scope(False)
    finally:
        reset_hermes_home_override(token)


def _spawn_hermes_action(
    subcommand: list[str], name: str, *, env_overrides: Optional[dict[str, str]] = None
) -> subprocess.Popen:
    """Spawn ``hermes <subcommand>`` detached (via ``hermes_cli.main``) and record the handle."""
    from hermes_cli.web_server import PROJECT_ROOT
    _ACTION_LOG_DIR.mkdir(parents=True, exist_ok=True)
    log_file = open(_ACTION_LOG_DIR / _ACTION_LOG_FILES[name], "ab", buffering=0)
    section_start = os.fstat(log_file.fileno()).st_size
    # The header carries the action id durably: the update's completion marker lands in the ROOT
    # update.log every profile shares, and only this id ties one to the action THIS log started.
    action_id = (env_overrides or {}).get("HERMES_ACTION_ID")
    stamp = time.strftime('%Y-%m-%d %H:%M:%S') + (f" {action_id}" if action_id else "")
    log_file.write(f"\n=== {name} started {stamp} ===\n".encode())
    if action_id:
        # ...and a sidecar keeps it once build output pushes the header past the status route's
        # bounded tail, or the log rotates: a restarted dashboard still knows which action it ran.
        from hermes_cli.runtime_state import _atomic_bytes
        _atomic_bytes(_ACTION_LOG_DIR / f"{name}.action_id", action_id.encode("ascii"))

    from hermes_cli._launchers import runtime_command
    cmd = runtime_command(PROJECT_ROOT, subcommand)
    if _action_targets_system_gateway(subcommand):
        # A system-scope lifecycle verb spawned as the dashboard's own user can only ever write
        # "System gateway <verb> requires root" into this log, so the button never worked on a
        # system install (#110820). Elevate — the CLI is sudo-aware: it adopts the unit's
        # HERMES_HOME past sudo's env_reset and reads SUDO_USER for the service identity.
        # ``-n`` never prompts (stdin is DEVNULL anyway); without a passwordless path the
        # REQUEST fails instead of reporting a started action whose child refuses. Same
        # two-step gate as the ``hermes update`` fleet restart: a refused blanket probe falls
        # back to ``sudo -l`` on the exact argv, so a command-scoped NOPASSWD entry qualifies.
        from hermes_cli.update_cmd_fleet import _sudo_noninteractive_ok

        if not _sudo_noninteractive_ok(["-l", "--", *cmd]):
            message = (
                f"{name} targets the system-scope gateway service, which requires root, and "
                "passwordless sudo is unavailable for the dashboard user. Run "
                f"'sudo hermes {' '.join(subcommand)}' on the host, or grant that user NOPASSWD sudo."
            )
            log_file.write(f"{message}\n".encode())
            log_file.close()
            raise RuntimeError(message)
        cmd = ["sudo", "-n", *cmd]
    # Named-profile actions get a scrubbed, pinned environment so the child cannot inherit the
    # dashboard profile's credentials; see _profile_action_environment (also drops _HERMES_GATEWAY).
    action_env = _profile_action_environment(subcommand, env_overrides)

    def _popen(**detach: Any) -> subprocess.Popen:
        return subprocess.Popen(
            cmd, cwd=str(PROJECT_ROOT), stdin=subprocess.DEVNULL, stdout=log_file, stderr=subprocess.STDOUT,
            env=action_env, **detach,
        )

    try:
        if sys.platform == "win32":
            try:
                proc = _popen(creationflags=windows_detach_flags())
            except OSError as exc:
                # CREATE_BREAKAWAY_FROM_JOB is refused ("Access is denied") when the parent's job
                # object forbids breakaway — an Electron/Tauri-launched dashboard, some Windows
                # Terminal configs. The flag is best-effort by contract (windows_detach_flags);
                # retry without it so the action still spawns, and record why for the status route.
                if getattr(exc, "winerror", None) != 5:
                    raise
                log_file.write(
                    f"{name}: breakaway spawn refused ({exc}); retrying without CREATE_BREAKAWAY_FROM_JOB\n".encode()
                )
                proc = _popen(creationflags=windows_detach_flags_without_breakaway())
        else:
            # detached-spawn: ok — dashboard gateway start/restart actions, which must outlive the dashboard
            proc = _popen(start_new_session=True)
    except OSError as exc:
        # Without this the action log would hold only the started-header for a spawn that never
        # happened; close the handle the child would have owned and let the endpoint report it.
        log_file.write(f"{name}: spawn failed: {exc}\n".encode())
        log_file.close()
        raise
    log_file.close()  # child holds its own dup'd fd; keeping ours leaks one per action
    _ACTION_RESULTS.pop(name, None)
    _ACTION_COMMANDS[name] = tuple(subcommand)
    _ACTION_PROCS[name] = proc
    _ACTION_LOG_STARTS[name] = (proc.pid, section_start)
    if action_id:
        _ACTION_IDS[name] = action_id
    else:
        _ACTION_IDS.pop(name, None)
    return proc


def _own_profile_selector(profile: Optional[str]) -> Optional[str]:
    """The profile a lifecycle verb addresses: the explicit selector, else the process's own
    profile (a pooled Desktop ``hermes --profile X serve`` answers ``/api/gateway/*`` without
    ``?profile=``; an unscoped verb there is about X). The default home resolves to the literal
    ``"default"`` selector, never to a bare argv: a selector-less child re-reads the sticky
    ``active_profile`` and would act on another profile's gateway (and skips the named-target
    env scrub). ``None`` is reserved for a home that is not a profile home at all."""
    requested = (profile or "").strip()
    if requested:
        return requested
    from hermes_constants import get_process_hermes_home, profile_name_for_home
    return profile_name_for_home(get_process_hermes_home()) or None


def _gateway_subcommand(profile: Optional[str], verb: str) -> list[str]:
    """``hermes [-p X] gateway <verb>`` argv for a dashboard lifecycle action. A profile served by the
    live default multiplexer has no gateway of its own: ``restart`` targets the multiplexer (the process
    that actually serves X — a ``-p X gateway restart`` child only exits 78 into the action log while the
    UI reports "restarted"); ``start``/``stop`` are refused by the caller (``multiplexed_profile_refusal``).
    The multiplexer is addressed as ``-p default`` explicitly: a bare ``gateway restart`` spawned from a
    pooled ``--profile X serve`` would inherit X's ``HERMES_HOME`` and hit the same exit-78 refusal. The
    selector is explicit for the default home too: a bare child re-reads the sticky ``active_profile``
    and would restart another profile's gateway."""
    from hermes_cli.web_server_profiles import _profile_cli_args
    profile = _own_profile_selector(profile)
    args = _profile_cli_args(profile)
    if profile and verb == "restart" and multiplexed_profile_refusal(profile, verb) is not None:
        # A served profile's restart targets the multiplexer, addressed as ``-p default``.
        args = ["-p", "default"]
    return args + ["gateway", verb]


def _profile_is_multiplexed(profile: str) -> bool:
    from hermes_cli.gateway import named_profile_served_by_running_multiplexer
    return named_profile_served_by_running_multiplexer(profile)


def _has_own_gateway(profile_dir: Path) -> bool:
    """A live gateway of the profile's OWN (a ``--force``-started separate one), not the multiplexer that
    serves it. Gateway liveness reports a served profile as running on the multiplexer's PID (#97120),
    so reading liveness alone made every served profile look self-hosted and the refusal below never
    fired while a multiplexer was live, which is the only time it is needed."""
    from gateway.status import get_running_pid, multiplexer_liveness_for_profile, resolve_gateway_liveness
    from hermes_cli.profiles import _check_gateway_running
    if not _check_gateway_running(profile_dir):
        return False
    served = multiplexer_liveness_for_profile(profile_dir)
    if served is None:
        return True
    liveness = resolve_gateway_liveness(
        profile_dir=profile_dir, use_cache=False,
        pid_probe=lambda path: get_running_pid(path, cleanup_stale=False))
    return liveness.running and liveness.pid != served[0]


def multiplexed_profile_refusal(profile: Optional[str], verb: str) -> Optional[str]:
    """Refusal text for ``gateway start``/``stop`` on a named profile with no gateway of its own (a
    ``--force``-started separate one is managed normally), else None. A profile the live host
    multiplexer serves is parked by ``stop`` and a parked one is unparked by ``start`` (the spawned
    ``hermes -p X gateway <verb>`` runs ``gateway_profile_lifecycle``), so neither is refused;
    ``start`` on an unparked named profile is — one host gateway serves every profile, so a new
    per-profile gateway is never the answer (the CLI twin ``_named_profile_refused_under_multiplexer``
    exits 78 into an action log nobody reads while the UI shows the verb as done)."""
    requested = _own_profile_selector(profile) or ""
    if not requested or requested.lower() in {"current", "default"}:
        return None
    served = _profile_is_multiplexed(requested)
    from hermes_cli.profiles import profile_is_parked, profile_is_standalone
    from hermes_cli.web_server_profiles import _resolve_profile_dir
    profile_dir = _resolve_profile_dir(requested)
    standalone = profile_is_standalone(profile_dir)
    if standalone:
        # The profile opted out of the host multiplexer, so its own gateway is the answer now: only a
        # host record that still lists it (the host started before the key was set) is refused.
        if not served:
            return None
        from gateway.host_attach import standalone_rescan_message
        return standalone_rescan_message(requested)
    if verb == "start" and profile_is_parked(profile_dir):
        from gateway.host_attach import host_gateway
        if host_gateway() is not None:
            return None  # a live host unparks it; with no host the refusal below still applies
    if not served and verb != "start":
        return None
    if _has_own_gateway(profile_dir):
        return None
    if served:
        if verb == "stop":
            return None  # parks the profile inside the host
        return (f"The default gateway already serves profile '{requested}' as a multiplexer; "
                f"{verb} it from the default profile instead of a separate gateway for this profile.")
    from hermes_cli.gateway_migrate import _installed_services
    if _installed_services(profile_dir):
        return None  # a --force-installed fleet member is not NEW; its own service is started normally
    return (f"Profile '{requested}' does not get a gateway of its own: one host gateway serves every "
            f"profile. Install or start it from the default profile (hermes gateway install), or fold an "
            f"existing per-profile fleet with `hermes gateway migrate --multiplex`; "
            f"`hermes -p {requested} gateway install --force` starts a separate one anyway.")


def live_session_refusal(profile: Optional[str], verb: str) -> Optional[str]:
    """Refusal text when the gateway ``verb`` would act on has live sessions, else None. Asked before
    spawning: the child refuses the same way, but into an action log the UI never reads."""
    from hermes_cli.gateway_lifecycle_guards import refusal_lines
    from hermes_cli.web_server_profiles import _resolve_profile_dir
    homes = [_resolve_profile_dir(_own_profile_selector(profile) or "default")]
    from gateway.host_attach import host_gateway
    owner = host_gateway()
    if owner is not None:
        homes.append(Path(owner.home))
    lines = refusal_lines(verb, homes, flag_hint=f"POST /api/gateway/{verb}?interrupt=true")
    return "\n".join(lines) or None


def _restart_gateway_after(profile: Optional[str], *, what: str, label: str) -> dict[str, Any]:
    """Best-effort gateway restart after a config change. The save stays authoritative: a failed
    spawn is reported (``restart_started: False`` + ``restart_error``) so the UI can fall back to
    its manual restart banner instead of failing the request. Live sessions refuse it the same way,
    with ``restart_busy`` set so a UI can say so in its own words."""
    from hermes_cli.web_server import _spawn_gateway_restart
    refusal = live_session_refusal(profile, "restart")
    if refusal:
        return {"restart_started": False, "restart_error": refusal, "restart_busy": True}
    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}


_ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
_RESTART_REASON_MAX = 300


def _restart_log_last_line(pid: int) -> str:
    """The last line the ``gateway-restart`` child *pid* wrote to its log, or ``""``: its section
    runs from the offset ``_spawn_hermes_action`` recorded to the next spawn's header."""
    start = _ACTION_LOG_STARTS.get("gateway-restart")
    if start is None or start[0] != pid:
        return ""
    log_path = _ACTION_LOG_DIR / _ACTION_LOG_FILES["gateway-restart"]
    try:
        with log_path.open("rb") as log:
            log.seek(start[1])
            section = log.read(256 * 1024).decode("utf-8", errors="replace")
    except OSError:
        return ""
    header, _, body = section.lstrip("\n").partition("\n")
    if not header.startswith("=== gateway-restart started"):
        return ""  # the log was rotated or truncated since
    body = body.split("\n=== gateway-restart started", 1)[0]
    lines = [_ANSI_ESCAPE.sub("", line).strip() for line in body.splitlines()]
    last = next((line for line in reversed(lines) if line), "")
    return last[:_RESTART_REASON_MAX]


def gateway_restart_failure(profile: Optional[str]) -> Optional[dict[str, Any]]:
    """Why the newest restart of the *profile* gateway failed, or ``None`` while it runs, after it
    succeeded, or when this process started none. A child that exits nonzero leaves the old
    gateway serving the old config, so a UI waiting for the new one learns here that none is coming.
    ``pid`` is the ``restart_pid`` the restart answered, so the UI ignores an older restart."""
    from hermes_cli import web_server
    recent = web_server._LAST_GATEWAY_RESTART
    if recent is None:
        return None
    _spawned_at, proc, command = recent
    if tuple(arg for arg in command if arg != "--interrupt") != tuple(_gateway_subcommand(profile, "restart")):
        return None
    exit_code = proc.poll()
    if not exit_code:
        return None
    return {"pid": proc.pid, "exit_code": exit_code, "reason": _restart_log_last_line(proc.pid) or f"exit code {exit_code}"}


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
    reflows whitespace (sentences re-joined with single spaces) and has no fence semantics.
    """
    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 no session LIST consumer reads but that dominate the payload (``system_prompt``
# is the fully rendered prompt, tens of KB per row — 96% of a 528KB /api/sessions response).
# Detail reads stay complete; list callers that need full rows 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