fix(gateway): share relay adapter in multiplex mode (#65366)

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/<profile>/ prefix, so a second bind can only collide. This is a

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."""