diff --git a/gateway/config.py b/gateway/config.py index 0931d067b66da..99d46c6444f0e 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -1011,6 +1011,19 @@ class GatewayConfig: # raising this fleet-wide would only delay genuine-wedge recovery. loop_watchdog_max_strikes: int = DEFAULT_LOOP_WATCHDOG_MAX_STRIKES + # Routing-only multiplexing (opt-in refinement of multiplex_profiles; no + # effect unless multiplex_profiles is on). When True, the gateway does NOT + # start secondary-profile adapters — every profile shares the default + # profile's connections and credentials. Secondary profiles are still + # REGISTERED: ``_profile_adapters[profile]`` points at the shared adapter + # map and each profile gets its own PairingStore, so a stamped + # ``source.profile`` resolves to a live adapter for authorization and + # egress instead of failing closed. For deployments that route personas + # per-chat over ONE shared connection via ``gateway.profile_routes`` + # (e.g. one Matrix account serving a different profile per room), where + # per-profile adapters could only poll the same credential twice. + multiplex_routing_only: bool = False + # Unauthorized DM policy unauthorized_dm_behavior: str = "pair" # "pair" or "ignore" @@ -1153,6 +1166,7 @@ def to_dict(self) -> Dict[str, Any]: "loop_watchdog_probe_interval_s": self.loop_watchdog_probe_interval_s, "loop_watchdog_probe_timeout_s": self.loop_watchdog_probe_timeout_s, "loop_watchdog_max_strikes": self.loop_watchdog_max_strikes, + "multiplex_routing_only": self.multiplex_routing_only, "unauthorized_dm_behavior": self.unauthorized_dm_behavior, "streaming": self.streaming.to_dict(), "session_store_max_age_days": self.session_store_max_age_days, @@ -1284,6 +1298,11 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig": env_multiplex = _env_multiplex_profiles_override() if env_multiplex is not None: multiplex_profiles = env_multiplex + multiplex_routing_only = data.get("multiplex_routing_only") + if multiplex_routing_only is None and isinstance(nested_gateway, dict): + # Also honor gateway.multiplex_routing_only, mirroring + # gateway.multiplex_profiles above. + multiplex_routing_only = nested_gateway.get("multiplex_routing_only") if "max_concurrent_sessions" in data: max_concurrent_raw = data.get("max_concurrent_sessions") max_concurrent_key = "max_concurrent_sessions" @@ -1333,6 +1352,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig": loop_watchdog_probe_interval_s=loop_watchdog_probe_interval_s, loop_watchdog_probe_timeout_s=loop_watchdog_probe_timeout_s, loop_watchdog_max_strikes=loop_watchdog_max_strikes, + multiplex_routing_only=_coerce_bool(multiplex_routing_only, False), max_concurrent_sessions=max_concurrent_sessions, unauthorized_dm_behavior=unauthorized_dm_behavior, streaming=StreamingConfig.from_dict(data.get("streaming", {})), @@ -1485,6 +1505,10 @@ def load_gateway_config() -> GatewayConfig: gw_data["multiplex_profile_allowlist"] = gateway_section[ "multiplex_profile_allowlist" ] + # Routing-only refinement: same top-level + nested acceptance as + # multiplex_profiles. + if "multiplex_routing_only" in yaml_cfg: + gw_data["multiplex_routing_only"] = yaml_cfg["multiplex_routing_only"] # Profile-based routing rules: accept either top-level # ``profile_routes`` or the nested ``gateway.profile_routes`` form @@ -1499,6 +1523,8 @@ def load_gateway_config() -> GatewayConfig: if "multiplex_profiles" in gateway_section and "multiplex_profiles" not in gw_data: # gateway.multiplex_profiles written by `hermes config set gateway.multiplex_profiles true` gw_data["multiplex_profiles"] = gateway_section["multiplex_profiles"] + if "multiplex_routing_only" in gateway_section and "multiplex_routing_only" not in gw_data: + gw_data["multiplex_routing_only"] = gateway_section["multiplex_routing_only"] if "max_concurrent_sessions" in gateway_section: gw_data["max_concurrent_sessions"] = gateway_section["max_concurrent_sessions"] if "systemd_watchdog_seconds" in gateway_section: diff --git a/gateway/run.py b/gateway/run.py index e7b5167ff4957..5d9d85a619aa2 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -15376,6 +15376,16 @@ async def _start_secondary_profile_adapters(self) -> int: skills, and credentials. Same-platform credential collisions (two profiles polling the same bot token) are detected and refused here, the only point that sees every profile's resolved credentials together. + + With ``gateway.multiplex_routing_only`` on, no secondary adapter is + created or connected — every served profile SHARES the active + profile's adapters (single set of platform credentials, routed + per-chat via ``gateway.profile_routes``). The profiles are still + registered: ``_profile_adapters[profile]`` points at the shared + adapter map and the loop below gives each its own PairingStore, so a + stamped ``source.profile`` resolves to a live adapter for + authorization and egress (``_authorization_adapter`` fails closed on + unregistered profiles) and pairing state stays profile-isolated. """ if not getattr(self.config, "multiplex_profiles", False): return 0 @@ -15385,6 +15395,7 @@ async def _start_secondary_profile_adapters(self) -> int: except Exception: return 0 + routing_only = bool(getattr(self.config, "multiplex_routing_only", False)) active = get_active_profile_name() or "default" connected = 0 # Resource claim -> profile that owns it. Credential claims prevent two @@ -15411,6 +15422,20 @@ async def _start_secondary_profile_adapters(self) -> int: for profile_name, profile_home in profile_homes: if profile_name == active: continue # handled by the primary startup loop + if routing_only: + # Shared-adapter registration: nothing new is connected, but + # the registry entry makes authorization/egress for a stamped + # source resolve to the shared adapters instead of failing + # closed, and the loop below creates the profile's own + # PairingStore. + self._profile_adapters[profile_name] = dict(self.adapters) + logger.info( + "Multiplex: routing-only — profile '%s' registered on the " + "shared adapters (%s); no secondary connection started.", + profile_name, + ", ".join(sorted(p.value for p in self.adapters)) or "none", + ) + continue try: connected += await self._start_one_profile_adapters( profile_name, profile_home, claimed diff --git a/tests/gateway/test_multiplex_routing_only.py b/tests/gateway/test_multiplex_routing_only.py new file mode 100644 index 0000000000000..079716b059629 --- /dev/null +++ b/tests/gateway/test_multiplex_routing_only.py @@ -0,0 +1,255 @@ +"""Routing-only multiplexing: one shared connection, many personas. + +``gateway.multiplex_routing_only`` — with ``multiplex_profiles`` on, the +gateway starts no secondary-profile adapters (all profiles share the default +profile's credentials) but still REGISTERS every served profile: +``_profile_adapters[profile]`` points at the shared adapter map and each +profile gets its own PairingStore. That registration is what makes a source +stamped by ``gateway.profile_routes`` resolve to a live adapter for +authorization and egress — ``_authorization_adapter`` deliberately fails +closed on unregistered stamped profiles. +""" +import pytest +from unittest.mock import MagicMock + +from gateway.config import GatewayConfig, Platform +from gateway.run import GatewayRunner +from gateway.session import SessionSource + + +# --------------------------------------------------------------------------- +# Config: multiplex_routing_only parsing +# --------------------------------------------------------------------------- + +class TestRoutingOnlyConfig: + def test_defaults_off(self): + assert GatewayConfig().multiplex_routing_only is False + assert GatewayConfig.from_dict({}).multiplex_routing_only is False + + def test_from_dict_top_level(self): + cfg = GatewayConfig.from_dict( + {"multiplex_profiles": True, "multiplex_routing_only": True} + ) + assert cfg.multiplex_routing_only is True + + def test_from_dict_nested_gateway_section(self): + cfg = GatewayConfig.from_dict( + {"gateway": {"multiplex_profiles": True, "multiplex_routing_only": True}} + ) + assert cfg.multiplex_routing_only is True + + def test_to_dict_roundtrip(self): + cfg = GatewayConfig(multiplex_profiles=True, multiplex_routing_only=True) + restored = GatewayConfig.from_dict(cfg.to_dict()) + assert restored.multiplex_routing_only is True + + def test_load_gateway_config_nested_form(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + (hermes_home / "config.yaml").write_text( + "gateway:\n" + " multiplex_profiles: true\n" + " multiplex_routing_only: true\n", + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + from gateway.config import load_gateway_config + + cfg = load_gateway_config() + assert cfg.multiplex_profiles is True + assert cfg.multiplex_routing_only is True + + +# --------------------------------------------------------------------------- +# Gateway: routing-only shared-adapter registration +# --------------------------------------------------------------------------- + +def _bare_runner(routing_only=True, shared_adapter=None, profile_routes=None): + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig.from_dict( + { + "multiplex_profiles": True, + "multiplex_routing_only": routing_only, + "profile_routes": profile_routes or [], + } + ) + runner.adapters = {Platform.MATRIX: shared_adapter or MagicMock(name="shared-matrix")} + runner._profile_adapters = {} + runner.pairing_store = MagicMock(name="global-pairing-store") + runner.pairing_stores = {} + return runner + + +def _serve_default_and_milo(monkeypatch, tmp_path): + import hermes_cli.profiles as profiles_mod + + monkeypatch.setattr( + profiles_mod, + "profiles_to_serve", + lambda multiplex, **kw: [ + ("default", tmp_path / "default"), + ("milo", tmp_path / "profiles" / "milo"), + ], + ) + monkeypatch.setattr(profiles_mod, "get_active_profile_name", lambda: "default") + # PairingStore resolves its directory from HERMES_HOME at construction. + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + +class TestRoutingOnlyRegistration: + @pytest.mark.asyncio + async def test_registers_shared_adapters_and_pairing_store( + self, monkeypatch, tmp_path + ): + shared = MagicMock(name="shared-matrix") + runner = _bare_runner(shared_adapter=shared) + _serve_default_and_milo(monkeypatch, tmp_path) + + async def _boom(*a, **k): # noqa: ANN002, ANN003 + raise AssertionError( + "routing-only must not start secondary profile adapters" + ) + + monkeypatch.setattr(runner, "_start_one_profile_adapters", _boom) + + connected = await runner._start_secondary_profile_adapters() + + assert connected == 0 + assert runner._profile_adapters["milo"][Platform.MATRIX] is shared + milo_store = runner.pairing_stores.get("milo") + assert milo_store is not None + assert milo_store.profile == "milo" + + @pytest.mark.asyncio + async def test_stamped_source_resolves_to_shared_adapter( + self, monkeypatch, tmp_path + ): + """The failure mode routing-only exists to fix: a stamped profile with + no registry entry resolves to NO adapter (authorization fails closed + and egress has nothing to send through). Routing-only registration + must make the stamped source resolve to the shared adapter instead.""" + shared = MagicMock(name="shared-matrix") + runner = _bare_runner(shared_adapter=shared) + _serve_default_and_milo(monkeypatch, tmp_path) + + # Before registration: fail closed, exactly as authz_mixin documents. + assert runner._authorization_adapter(Platform.MATRIX, "milo") is None + + await runner._start_secondary_profile_adapters() + + assert runner._authorization_adapter(Platform.MATRIX, "milo") is shared + source = SessionSource( + platform=Platform.MATRIX, + chat_id="!milo:example.org", + user_id="@alice:example.org", + profile="milo", + ) + assert runner._adapter_for_source(source) is shared + + @pytest.mark.asyncio + async def test_profile_routes_stamp_resolves_shared_adapter( + self, monkeypatch, tmp_path + ): + """End-to-end with the merged route matcher: a gateway.profile_routes + rule keyed on a Matrix room resolves the profile for an inbound + source, and routing-only registration gives that profile a live + (shared) adapter.""" + shared = MagicMock(name="shared-matrix") + runner = _bare_runner( + shared_adapter=shared, + profile_routes=[ + { + "name": "milo-room", + "platform": "matrix", + "chat_id": "!milo:example.org", + "profile": "milo", + } + ], + ) + _serve_default_and_milo(monkeypatch, tmp_path) + await runner._start_secondary_profile_adapters() + + routed = SessionSource( + platform=Platform.MATRIX, + chat_id="!milo:example.org", + user_id="@alice:example.org", + ) + assert runner._profile_name_for_source(routed) == "milo" + routed.profile = "milo" + assert runner._adapter_for_source(routed) is shared + + unrouted = SessionSource( + platform=Platform.MATRIX, + chat_id="!other:example.org", + user_id="@alice:example.org", + ) + assert runner._profile_name_for_source(unrouted) is None + + @pytest.mark.asyncio + async def test_unserved_profile_still_fails_closed(self, monkeypatch, tmp_path): + runner = _bare_runner() + _serve_default_and_milo(monkeypatch, tmp_path) + + await runner._start_secondary_profile_adapters() + + assert runner._authorization_adapter(Platform.MATRIX, "ghost") is None + + @pytest.mark.asyncio + async def test_pairing_checks_use_per_profile_store(self, monkeypatch, tmp_path): + runner = _bare_runner() + _serve_default_and_milo(monkeypatch, tmp_path) + + await runner._start_secondary_profile_adapters() + + stamped = SessionSource( + platform=Platform.MATRIX, + chat_id="!milo:example.org", + user_id="@alice:example.org", + profile="milo", + ) + unstamped = SessionSource( + platform=Platform.MATRIX, + chat_id="!main:example.org", + user_id="@alice:example.org", + ) + assert runner._pairing_store_for(stamped) is runner.pairing_stores["milo"] + assert runner._pairing_store_for(unstamped) is runner.pairing_store + + @pytest.mark.asyncio + async def test_full_multiplex_path_unchanged_when_flag_off( + self, monkeypatch, tmp_path + ): + runner = _bare_runner(routing_only=False) + _serve_default_and_milo(monkeypatch, tmp_path) + + started = [] + + async def _fake_start(profile_name, profile_home, claimed): + started.append(profile_name) + return 1 + + monkeypatch.setattr(runner, "_start_one_profile_adapters", _fake_start) + + connected = await runner._start_secondary_profile_adapters() + + assert started == ["milo"] + assert connected == 1 + + @pytest.mark.asyncio + async def test_served_profiles_recorded(self, monkeypatch, tmp_path): + runner = _bare_runner() + _serve_default_and_milo(monkeypatch, tmp_path) + + recorded = {} + + def _fake_write(**kwargs): + recorded.update(kwargs) + + import gateway.status as status_mod + + monkeypatch.setattr(status_mod, "write_runtime_status", _fake_write) + + await runner._start_secondary_profile_adapters() + + assert recorded.get("served_profiles") == ["default", "milo"] diff --git a/website/docs/user-guide/multi-profile-gateways.md b/website/docs/user-guide/multi-profile-gateways.md index 7feddd069a90e..f19aa2b13404a 100644 --- a/website/docs/user-guide/multi-profile-gateways.md +++ b/website/docs/user-guide/multi-profile-gateways.md @@ -288,6 +288,48 @@ the gateway rejects that ingress and logs the route and target. It does not run the default profile. Traffic that matches no route keeps the historical default-profile behavior. +### Routing-only: one connection, many personas + +`profile_routes` stamps the profile, but under plain multiplexing each served +profile still tries to bring up its **own** adapters — and a secondary profile +that shares the default profile's credential is refused (a single bot token +cannot be polled twice), leaving its routed messages to fail closed in +authorization. Some deployments want exactly that shape: **one** bot account +whose personality depends on *where* it is talked to — e.g. a single Matrix +account serving a different profile per room. For those, add +`multiplex_routing_only` next to `multiplex_profiles`: + +```yaml +gateway: + multiplex_profiles: true + multiplex_routing_only: true + profile_routes: + - name: standup-room + platform: matrix + chat_id: "!standup:example.org" + profile: coder + - name: bookclub-room + platform: matrix + chat_id: "!bookclub:example.org" + profile: writer +``` + +With routing-only on, the gateway starts **no** secondary-profile adapters — +there is exactly one connection per platform, owned by the default profile, so +nothing ever polls the same credential twice. Every served profile is still +*registered*: it resolves to the shared adapters for authorization and +replies, and it gets its **own pairing whitelist** (approving a user for +`coder` does not approve them for `writer`). Inbound messages matching a route +are stamped with that profile, so sessions, skills, memory, and model routing +all resolve per-profile exactly as in full multiplexing; replies go out +through the shared connection. Chats matching no route behave as the default +profile. + +Routing-only is an all-or-nothing mode: with it on, **no** profile gets its +own adapters, so use it only when every persona rides the shared credential. +Mixed deployments (some profiles with their own bot tokens) should stay on +plain multiplexing. + ## Start, stop, or restart all gateways at once The CLI ships with single-profile lifecycle commands. To act across every