diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 07a5ef619b4..4ae8c3251d2 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -2252,6 +2252,136 @@ async def git_branch_switch_route(body: GitBranchSwitchBody): return await _git_op(_web_git.branch_switch, _git_path(body.path), body.branch) +# 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/run.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), +} + +# 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: + with open(profile_home / "config.yaml", encoding="utf-8") as f: + cfg = yaml.safe_load(f) or {} + 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 _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. + """ + 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_names = [name for name, _home in homes] + gateways: List[Dict[str, Any]] = [] + 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 + 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} + + @app.get("/api/status") async def get_status(profile: Optional[str] = None): status_scope = None @@ -2456,12 +2586,23 @@ async def get_status(profile: Optional[str] = None): # dashboard is local-only and the caller is already inside the trust # envelope — the same loopback/gated split ``should_require_auth`` draws. if not auth_required: + # Profile + gateway topology: which profiles exist, whether one + # multiplexed gateway or several per-profile gateways serve them, + # and which host ports the live gateways' port-binding platforms + # listen on. Enumerating profiles walks the filesystem and probes + # the process table, so keep it off the event loop. + topology = await asyncio.get_running_loop().run_in_executor( + None, _collect_profile_gateway_topology + ) status.update({ "hermes_home": str(get_hermes_home()), "config_path": str(get_config_path()), "env_path": str(get_env_path()), "gateway_pid": gateway_pid, "gateway_health_url": _GATEWAY_HEALTH_URL, + "profiles": topology["profiles"], + "gateway_mode": topology["gateway_mode"], + "gateways": topology["gateways"], }) return status diff --git a/tests/hermes_cli/test_web_server_gateway_topology.py b/tests/hermes_cli/test_web_server_gateway_topology.py new file mode 100644 index 00000000000..18de4aac72a --- /dev/null +++ b/tests/hermes_cli/test_web_server_gateway_topology.py @@ -0,0 +1,220 @@ +"""Tests for the /api/status profile + gateway topology readout. + +Covers the loopback-only ``profiles`` / ``gateway_mode`` / ``gateways`` fields +added to ``/api/status``: profile enumeration, single vs multiplex vs multiple +gateway detection, and per-platform port resolution. +""" + +import pytest + +from hermes_cli import web_server +from hermes_cli.web_server import ( + _collect_profile_gateway_topology, + _profile_platform_ports, +) + + +# --------------------------------------------------------------------------- +# _profile_platform_ports +# --------------------------------------------------------------------------- + +class TestProfilePlatformPorts: + def test_no_runtime_platforms_returns_empty(self, tmp_path): + assert _profile_platform_ports(tmp_path, None) == {} + assert _profile_platform_ports(tmp_path, {"platforms": {}}) == {} + + def test_non_port_binding_platform_ignored(self, tmp_path): + runtime = {"platforms": {"telegram": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {} + + def test_default_port_when_no_config(self, tmp_path): + runtime = {"platforms": {"webhook": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"webhook": 8644} + + def test_port_from_config_yaml_top_level(self, tmp_path): + (tmp_path / "config.yaml").write_text( + "platforms:\n webhook:\n port: 9001\n", encoding="utf-8" + ) + runtime = {"platforms": {"webhook": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"webhook": 9001} + + def test_port_from_gateway_platforms_block(self, tmp_path): + (tmp_path / "config.yaml").write_text( + "gateway:\n platforms:\n api_server:\n port: 9500\n", + encoding="utf-8", + ) + runtime = {"platforms": {"api_server": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"api_server": 9500} + + def test_top_level_platforms_wins_over_gateway_block(self, tmp_path): + (tmp_path / "config.yaml").write_text( + "gateway:\n platforms:\n webhook:\n port: 1111\n" + "platforms:\n webhook:\n port: 2222\n", + encoding="utf-8", + ) + runtime = {"platforms": {"webhook": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"webhook": 2222} + + def test_port_in_extra_block(self, tmp_path): + (tmp_path / "config.yaml").write_text( + "platforms:\n whatsapp_cloud:\n extra:\n webhook_port: 8095\n", + encoding="utf-8", + ) + runtime = {"platforms": {"whatsapp_cloud": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"whatsapp_cloud": 8095} + + def test_dead_platform_states_excluded(self, tmp_path): + runtime = { + "platforms": { + "webhook": {"state": "fatal"}, + "api_server": {"state": "disconnected"}, + "msgraph_webhook": {"state": "connected"}, + } + } + assert _profile_platform_ports(tmp_path, runtime) == {"msgraph_webhook": 8646} + + def test_invalid_port_value_falls_back_to_default(self, tmp_path): + (tmp_path / "config.yaml").write_text( + "platforms:\n webhook:\n port: notaport\n", encoding="utf-8" + ) + runtime = {"platforms": {"webhook": {"state": "connected"}}} + assert _profile_platform_ports(tmp_path, runtime) == {"webhook": 8644} + + +# --------------------------------------------------------------------------- +# _collect_profile_gateway_topology +# --------------------------------------------------------------------------- + +def _patch_topology(monkeypatch, homes, running, runtimes): + """Patch the topology collector's collaborators. + + ``homes``: list of (name, Path); ``running``: set of profile names with a + live gateway; ``runtimes``: {name: runtime dict}. + """ + import hermes_cli.profiles as profiles_mod + import gateway.status as status_mod + + monkeypatch.setattr(profiles_mod, "profiles_to_serve", lambda multiplex: homes) + monkeypatch.setattr( + profiles_mod, "_check_gateway_running", + lambda home: next(n for n, h in homes if h == home) in running, + ) + by_path = {home / "gateway_state.json": runtimes.get(name) for name, home in homes} + monkeypatch.setattr( + status_mod, "read_runtime_status", lambda path=None: by_path.get(path) + ) + + +class TestCollectProfileGatewayTopology: + def test_no_gateways_running(self, tmp_path, monkeypatch): + homes = [("default", tmp_path / "d"), ("coder", tmp_path / "c")] + _patch_topology(monkeypatch, homes, running=set(), runtimes={}) + topo = _collect_profile_gateway_topology() + assert topo["profiles"] == ["default", "coder"] + assert topo["gateway_mode"] == "none" + assert topo["gateways"] == [] + + def test_single_gateway(self, tmp_path, monkeypatch): + homes = [("default", tmp_path / "d"), ("coder", tmp_path / "c")] + _patch_topology( + monkeypatch, homes, running={"default"}, + runtimes={"default": {"platforms": {}}}, + ) + topo = _collect_profile_gateway_topology() + assert topo["gateway_mode"] == "single" + assert [g["profile"] for g in topo["gateways"]] == ["default"] + + def test_multiplex_gateway(self, tmp_path, monkeypatch): + homes = [("default", tmp_path / "d"), ("coder", tmp_path / "c")] + _patch_topology( + monkeypatch, homes, running={"default"}, + runtimes={"default": { + "platforms": {}, + "served_profiles": ["default", "coder"], + }}, + ) + topo = _collect_profile_gateway_topology() + assert topo["gateway_mode"] == "multiplex" + assert topo["gateways"][0]["served_profiles"] == ["default", "coder"] + + def test_multiple_independent_gateways_with_ports(self, tmp_path, monkeypatch): + d_home = tmp_path / "d" + c_home = tmp_path / "c" + d_home.mkdir() + c_home.mkdir() + (c_home / "config.yaml").write_text( + "platforms:\n webhook:\n port: 9644\n", encoding="utf-8" + ) + homes = [("default", d_home), ("coder", c_home)] + _patch_topology( + monkeypatch, homes, running={"default", "coder"}, + runtimes={ + "default": {"platforms": {"webhook": {"state": "connected"}}}, + "coder": {"platforms": {"webhook": {"state": "connected"}}}, + }, + ) + topo = _collect_profile_gateway_topology() + assert topo["gateway_mode"] == "multiple" + ports = {g["profile"]: g["ports"] for g in topo["gateways"]} + assert ports == {"default": {"webhook": 8644}, "coder": {"webhook": 9644}} + + def test_enumeration_failure_degrades_gracefully(self, monkeypatch): + import hermes_cli.profiles as profiles_mod + + def _boom(multiplex): + raise RuntimeError("no profiles root") + + monkeypatch.setattr(profiles_mod, "profiles_to_serve", _boom) + topo = _collect_profile_gateway_topology() + assert topo == {"profiles": [], "gateway_mode": "unknown", "gateways": []} + + +# --------------------------------------------------------------------------- +# /api/status wiring +# --------------------------------------------------------------------------- + +class TestStatusEndpointTopology: + @pytest.fixture(autouse=True) + def _setup_client(self, monkeypatch, _isolate_hermes_home): + try: + from starlette.testclient import TestClient + except ImportError: + pytest.skip("fastapi/starlette not installed") + + import hermes_state + from hermes_constants import get_hermes_home + from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN + + monkeypatch.setattr( + hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db" + ) + self.client = TestClient(app) + self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN + + def test_status_includes_topology_on_loopback(self, monkeypatch): + monkeypatch.setattr( + web_server, "_collect_profile_gateway_topology", + lambda: { + "profiles": ["default", "coder"], + "gateway_mode": "single", + "gateways": [{"profile": "default", "ports": {}}], + }, + ) + resp = self.client.get("/api/status") + assert resp.status_code == 200 + data = resp.json() + assert data["profiles"] == ["default", "coder"] + assert data["gateway_mode"] == "single" + assert data["gateways"] == [{"profile": "default", "ports": {}}] + + def test_status_omits_topology_when_auth_gated(self, monkeypatch): + monkeypatch.setattr(web_server.app.state, "auth_required", True, raising=False) + resp = self.client.get("/api/status") + assert resp.status_code == 200 + data = resp.json() + # Topology is deployment recon — hidden on gated binds, like + # hermes_home / gateway_pid. + assert "profiles" not in data + assert "gateway_mode" not in data + assert "gateways" not in data + monkeypatch.setattr(web_server.app.state, "auth_required", False, raising=False)