"""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 from hermes_cli.config import get_hermes_home # Same logger the code used before extraction (record parity). _log = logging.getLogger("hermes_cli.web_server") 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("/")) 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: return True, json.loads(resp.read()) 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/.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]] = {} 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 `` action. The dashboard loads its own profile's ``.env`` into process-global ``os.environ``. Copying that mapping verbatim into ``hermes -p ...`` 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 `` 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) log_file.write(f"\n=== {name} started {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n".encode()) 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 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) detach = {"creationflags": windows_detach_flags()} if sys.platform == "win32" else {"start_new_session": True} proc = subprocess.Popen( cmd, cwd=str(PROJECT_ROOT), stdin=subprocess.DEVNULL, stdout=log_file, stderr=subprocess.STDOUT, env=action_env, **detach, ) 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_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 _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 `` 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 `` 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 _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.""" 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 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