diff --git a/gateway/run.py b/gateway/run.py index f05d89540b5e..37c4718d80b9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -8698,6 +8698,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew for platform, platform_config in profile_cfg.platforms.items(): if not platform_config.enabled: continue + # Relay is shared process-level ingress in multiplex mode. The + # active profile owns the one connection; connector-stamped + # source.profile routes inbound turns to secondary profiles. + if ( + getattr(self.config, "multiplex_profiles", False) + and platform is Platform.RELAY + ): + continue # A secondary profile must NOT enable a port-binding platform: the # default profile's listener already serves every profile via the # /p// prefix, so a second bind can only collide. This is a diff --git a/tests/gateway/test_multiplex_adapter_registry.py b/tests/gateway/test_multiplex_adapter_registry.py index 43f26b0ff3e3..9a0f49bfb04e 100644 --- a/tests/gateway/test_multiplex_adapter_registry.py +++ b/tests/gateway/test_multiplex_adapter_registry.py @@ -191,6 +191,147 @@ class TestPortBindingHardError: connected = await runner._start_one_profile_adapters("reviewer", "/tmp/x", {}) assert connected == 0 # nothing connected, but no MultiplexConfigError + @pytest.mark.asyncio + async def test_multiplex_secondary_skips_relay_but_starts_direct_adapter( + self, monkeypatch + ): + """Relay is process-shared; direct adapters remain per-profile.""" + from gateway.config import GatewayConfig, Platform, PlatformConfig + + class _DirectAdapter: + platform = Platform.TELEGRAM + + def set_message_handler(self, handler): + self.message_handler = handler + + def set_fatal_error_handler(self, handler): + self.fatal_error_handler = handler + + def set_session_store(self, store): + self.session_store = store + + def set_busy_session_handler(self, handler): + self.busy_session_handler = handler + + def set_topic_recovery_fn(self, handler): + self.topic_recovery_fn = handler + + def set_authorization_check(self, handler): + self.authorization_check = handler + + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig(multiplex_profiles=True) + runner._profile_adapters = {} + runner.session_store = object() + runner._handle_adapter_fatal_error = object() + runner._handle_active_session_busy_message = object() + runner._recover_telegram_topic_thread_id = object() + runner._busy_text_mode = "queue" + runner._make_adapter_auth_check = lambda platform: object() + + reviewer_cfg = GatewayConfig(multiplex_profiles=True) + reviewer_cfg.platforms = { + Platform.RELAY: PlatformConfig(enabled=True), + Platform.TELEGRAM: PlatformConfig(enabled=True, token="reviewer-token"), + } + monkeypatch.setattr( + "gateway.config.load_gateway_config", lambda: reviewer_cfg + ) + + direct = _DirectAdapter() + factory_calls = [] + + def _create_adapter(platform, config): + factory_calls.append(platform) + if platform is Platform.RELAY: + raise AssertionError("secondary Relay factory must not be invoked") + return direct + + connect_calls = [] + + async def _connect(adapter, platform): + connect_calls.append((adapter, platform)) + return True + + monkeypatch.setattr(runner, "_create_adapter", _create_adapter) + monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect) + + connected = await runner._start_one_profile_adapters( + "reviewer", "/tmp/x", {} + ) + + assert connected == 1 + assert factory_calls == [Platform.TELEGRAM] + assert connect_calls == [(direct, Platform.TELEGRAM)] + assert runner._profile_adapters["reviewer"] == { + Platform.TELEGRAM: direct, + } + + @pytest.mark.asyncio + async def test_non_multiplex_profile_adapter_start_keeps_relay(self, monkeypatch): + """The Relay skip is gated to multiplex mode.""" + from gateway.config import GatewayConfig, Platform, PlatformConfig + + class _RelayAdapter: + platform = Platform.RELAY + + def set_message_handler(self, handler): + pass + + def set_fatal_error_handler(self, handler): + pass + + def set_session_store(self, store): + pass + + def set_busy_session_handler(self, handler): + pass + + def set_topic_recovery_fn(self, handler): + pass + + def set_authorization_check(self, handler): + pass + + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig(multiplex_profiles=False) + runner._profile_adapters = {} + runner.session_store = object() + runner._handle_adapter_fatal_error = object() + runner._handle_active_session_busy_message = object() + runner._recover_telegram_topic_thread_id = object() + runner._busy_text_mode = "queue" + runner._make_adapter_auth_check = lambda platform: object() + + profile_cfg = GatewayConfig(multiplex_profiles=False) + profile_cfg.platforms = { + Platform.RELAY: PlatformConfig(enabled=True), + } + monkeypatch.setattr("gateway.config.load_gateway_config", lambda: profile_cfg) + + relay = _RelayAdapter() + factory_calls = [] + connect_calls = [] + + def _create_adapter(platform, config): + factory_calls.append(platform) + return relay + + async def _connect(adapter, platform): + connect_calls.append((adapter, platform)) + return True + + monkeypatch.setattr(runner, "_create_adapter", _create_adapter) + monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect) + + connected = await runner._start_one_profile_adapters( + "reviewer", "/tmp/x", {} + ) + + assert connected == 1 + assert factory_calls == [Platform.RELAY] + assert connect_calls == [(relay, Platform.RELAY)] + @pytest.mark.asyncio async def test_secondary_same_config_token_is_refused(self, monkeypatch): """Adapters that keep their token on config still trip the mux guard."""