"""``gateway.multiplex_profiles`` defaults to ON, but an UNSET flag is settled at boot by ``hermes_cli.gateway_multiplex_mode.resolve_multiplex_mode`` — the same preflight ``hermes gateway migrate --multiplex`` runs — so flipping the default can never make a default gateway double-bind a fleet that still runs per-profile gateways. The service layer is faked through ``gateway_migrate``'s ``_installed_services`` / ``_live_gateway_pid`` seams (the shape ``test_gateway_migrate_multiplex.py`` uses). """ from __future__ import annotations import json import os import shutil from pathlib import Path import pytest import hermes_constants from gateway.config import GatewayConfig, load_gateway_config from hermes_cli import gateway_migrate as gm from hermes_cli import gateway_multiplex_mode as mode @pytest.fixture def fleet(tmp_path, monkeypatch): """default + coder + ops with distinct bot tokens; nothing runs a gateway unless a test says so.""" root = tmp_path / "hermes" for sub in ("profiles/coder", "profiles/ops"): (root / sub).mkdir(parents=True) (root / "config.yaml").write_text("model:\n default: x\n", encoding="utf-8") (root / ".env").write_text("TELEGRAM_BOT_TOKEN=111111:default-token\n", encoding="utf-8") (root / "profiles/coder/.env").write_text("TELEGRAM_BOT_TOKEN=222222:coder-token\n", encoding="utf-8") (root / "profiles/ops/.env").write_text("DISCORD_BOT_TOKEN=ops-discord-333333\n", encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(root)) monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False) for name in ("TELEGRAM_BOT_TOKEN", "DISCORD_BOT_TOKEN", "API_SERVER_KEY", "WEBHOOK_ENABLED"): monkeypatch.delenv(name, raising=False) monkeypatch.setattr(hermes_constants, "_default_hermes_root_memo", None) services: dict[str, list] = {} pids: dict[str, int] = {} monkeypatch.setattr(gm, "_installed_services", lambda home: services.get(_name(home), [])) monkeypatch.setattr(gm, "_live_gateway_pid", lambda home: pids.get(_name(home))) monkeypatch.setattr(gm, "_host_supports_migration", lambda: None) return root, services, pids def _name(home: Path) -> str: return hermes_constants.profile_name_for_home(home) or "default" def test_default_config_is_on_but_the_loader_leaves_an_unset_flag_undecided(): """DEFAULT_CONFIG says on; GatewayConfig keeps "unset" distinguishable from "chosen" so the boot guard can tell them apart (an explicit value must survive verbatim).""" assert GatewayConfig.from_dict({}).multiplex_profiles is None assert not GatewayConfig.from_dict({}).multiplex_profiles # readers treat undecided as off assert GatewayConfig.from_dict({"gateway": {"multiplex_profiles": False}}).multiplex_profiles is False assert GatewayConfig.from_dict({"multiplex_profiles": "true"}).multiplex_profiles is True def test_unset_flag_multiplexes_a_quiet_fleet_and_stays_standalone_beside_a_live_secondary(fleet): root, services, pids = fleet decision = mode.resolve_multiplex_mode(cfg := load_gateway_config()) assert decision == mode.MultiplexDecision(True, "default", decision.reason) assert cfg.multiplex_profiles is True pids["coder"] = 4101 # coder still runs its own gateway: a multiplexer would double-poll its bot decision = mode.resolve_multiplex_mode(cfg := load_gateway_config()) assert not decision.enabled and decision.source == "guard" assert "'coder'" in decision.reason and gm.MIGRATE_COMMAND in decision.reason assert cfg.multiplex_profiles is False # The CLI side agrees: no live multiplexer record and no explicit opt-in means coder is NOT served. from hermes_cli.gateway import named_profile_served_by_running_multiplexer assert named_profile_served_by_running_multiplexer("coder") is False assert mode.default_gateway_multiplexes(root) is False pids.clear() services["ops"] = [("systemd", False)] # an installed unit counts the same as a live pid assert mode.resolve_multiplex_mode(load_gateway_config()).source == "guard" def test_preflight_blocker_and_single_profile_keep_the_unset_default_standalone(fleet): root, _services, _pids = fleet # coder reuses the default's bot token -> the migration preflight's duplicate-credential blocker. (root / "profiles/coder/.env").write_text("TELEGRAM_BOT_TOKEN=111111:default-token\n", encoding="utf-8") decision = mode.resolve_multiplex_mode(load_gateway_config()) assert decision.source == "guard" and "profile_routes" in decision.reason for sub in ("profiles/coder", "profiles/ops"): shutil.rmtree(root / sub) decision = mode.resolve_multiplex_mode(load_gateway_config()) assert decision == mode.MultiplexDecision(False, "guard", mode.SINGLE_PROFILE_REASON) def test_standalone_named_profile_leaves_nothing_to_multiplex(fleet): """A sole named profile that opts itself out via `gateway.standalone: true` is not served, so the host still has a single profile to serve: guard, not multiplex.""" root, _services, _pids = fleet shutil.rmtree(root / "profiles/ops") # exactly one named profile remains (root / "profiles/coder/config.yaml").write_text("gateway:\n standalone: true\n") decision = mode.resolve_multiplex_mode(load_gateway_config()) assert decision == mode.MultiplexDecision(False, "guard", mode.SINGLE_PROFILE_REASON) @pytest.mark.parametrize("flag", ["", " multiplex_profiles: true\n", " multiplex_profiles: false\n"]) def test_standalone_launcher_never_multiplexes_other_profiles(fleet, flag): from gateway.run import _profile_runtime_scope root, _services, _pids = fleet solo = root / "profiles/coder" (solo / "config.yaml").write_text("gateway:\n standalone: true\n" + flag) # Real loader + runtime scopes: A -> B -> A, without borrowing the launch home's config. for home in (solo, root, solo): with _profile_runtime_scope(home): cfg = load_gateway_config() decision = mode.resolve_multiplex_mode(cfg) if home == solo: assert decision.enabled is False assert cfg.multiplex_profiles is False assert "this profile is standalone" in decision.reason else: assert decision.enabled is True def test_explicit_true_is_never_second_guessed_and_explicit_false_is_retired(fleet, monkeypatch): root, _services, pids = fleet pids["coder"] = 4101 (root / "config.yaml").write_text("gateway:\n multiplex_profiles: true\n", encoding="utf-8") cfg = load_gateway_config() assert mode.resolve_multiplex_mode(cfg) == mode.MultiplexDecision(True, "config") assert cfg.multiplex_profiles is True assert mode.explicit_multiplex_flag(root) is True # RETIRED opt-out: `false` still parses, but it can no longer pin a second gateway onto this # host — with nothing blocking, the process multiplexes and says the key was ignored. monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "false") pids.clear() cfg = load_gateway_config() decision = mode.resolve_multiplex_mode(cfg) assert decision.enabled is True and decision.source == "retired-opt-out" assert cfg.multiplex_profiles is True assert mode.explicit_multiplex_flag(root) is False # ...and it still cannot force a fold that would double-bind: a secondary that owns a live # gateway keeps this process standalone, with the blocker named (never a silent second bind). pids["coder"] = 4101 cfg = load_gateway_config() guarded = mode.resolve_multiplex_mode(cfg) assert guarded.enabled is False and guarded.source == "guard" and "coder" in guarded.reason def test_migration_plan_treats_the_unset_default_as_not_yet_multiplexed(fleet): """The fleet the boot guard refuses is exactly the one ``hermes gateway migrate --multiplex`` folds: an unset flag must not read as "already multiplexed" or the migration would short-circuit.""" root, services, pids = fleet pids.update({"coder": 4101, "ops": 4102}) services.update({"coder": [("systemd", False)], "ops": [("systemd", False)]}) plan = gm.build_migration_plan() assert plan.multiplex_flag_on is False and not plan.already_multiplexed assert plan.eligible_for_migration() def test_live_record_outranks_the_raw_flag_for_other_processes(fleet, monkeypatch): """A CLI process asks the LIVE default gateway (which settled the unset default itself) before reading config; a gateway that stayed standalone recorded an empty served set.""" root, _services, _pids = fleet import gateway.status as status monkeypatch.setattr(status, "_read_process_cmdline", lambda pid: "hermes gateway run") (root / "gateway.pid").write_text(json.dumps({"pid": os.getpid(), "hermes_home": str(root)})) record = {"pid": os.getpid(), "hermes_home": str(root), "gateway_state": "running", "served_profiles": []} (root / "gateway_state.json").write_text(json.dumps(record)) assert mode.default_gateway_multiplexes(root) is False record["served_profiles"] = ["default", "coder", "ops"] (root / "gateway_state.json").write_text(json.dumps(record)) assert mode.default_gateway_multiplexes(root) is True def test_guard_refusal_is_recorded_in_runtime_status_and_cleared_on_default(tmp_path, monkeypatch): """A guard refusal must be visible to `hermes gateway status`, not only in the boot log; a later boot that multiplexes clears it (a stale reason would misdescribe the live gateway).""" from gateway import status as gw_status from hermes_cli.gateway_multiplex_mode import MultiplexDecision, record_multiplex_decision monkeypatch.setattr(gw_status, "_get_runtime_status_path", lambda: tmp_path / "gateway_state.json") record_multiplex_decision(MultiplexDecision(False, "guard", "profile(s) 'coder' still run their own gateway")) assert "coder" in gw_status.read_runtime_status(tmp_path / "gateway_state.json")["multiplex_standalone_reason"] record_multiplex_decision(MultiplexDecision(True, "default", "unset; default applies")) assert gw_status.read_runtime_status(tmp_path / "gateway_state.json")["multiplex_standalone_reason"] is None def test_recorded_standalone_warning_lines_suppressed_for_dead_or_stale_record(tmp_path, monkeypatch): """Dead or stale gateway_state.json must not emit standalone warnings (#120991).""" import gateway.status as gw_status from datetime import datetime, timezone state_file = tmp_path / "gateway_state.json" monkeypatch.setattr(gw_status, "_get_runtime_status_path", lambda: state_file) # A multi-profile host: without unserved profiles the box is empty regardless of liveness, # so the dead/stale cases below would pass vacuously. monkeypatch.setattr(mode, "unserved_profiles", lambda: ["other_profile"]) # 1. Stopped gateway state_file.write_text(json.dumps({ "gateway_state": "stopped", "pid": os.getpid(), "updated_at": datetime.now(timezone.utc).isoformat(), "multiplex_standalone_reason": "orphan reason", }), encoding="utf-8") assert mode.recorded_standalone_warning_lines() == [] # 2. Dead PID state_file.write_text(json.dumps({ "gateway_state": "running", "pid": 999999999, "updated_at": datetime.now(timezone.utc).isoformat(), "multiplex_standalone_reason": "orphan reason", }), encoding="utf-8") monkeypatch.setattr(gw_status, "runtime_status_pid_is_live", lambda r: False) assert mode.recorded_standalone_warning_lines() == [] # 3. Stale heartbeat but LIVE PID: a paused/wedged standalone gateway still warns (the # heartbeat is a health signal, not liveness). state_file.write_text(json.dumps({ "gateway_state": "running", "pid": os.getpid(), "updated_at": "2020-01-01T00:00:00Z", "multiplex_standalone_reason": "wedged reason", }), encoding="utf-8") monkeypatch.setattr(gw_status, "runtime_status_pid_is_live", lambda r: True) assert any("wedged reason" in line for line in mode.recorded_standalone_warning_lines()) # 4. Live and fresh record emits warning state_file.write_text(json.dumps({ "gateway_state": "running", "pid": os.getpid(), "updated_at": datetime.now(timezone.utc).isoformat(), "multiplex_standalone_reason": "real standalone reason", }), encoding="utf-8") monkeypatch.setattr(gw_status, "runtime_status_pid_is_live", lambda r: True) monkeypatch.setattr(gw_status, "runtime_status_is_stale", lambda r: False) lines = mode.recorded_standalone_warning_lines() assert any("STANDALONE" in line for line in lines) assert any("real standalone reason" in line for line in lines)