fix(gateway): detect config token credential collisions (#59321)

Co-authored-by: markoub <2418548+markoub@users.noreply.github.com>

gateway/run.py

@@ -8452,6 +8452,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             if isinstance(val, str) and val.strip():
                 token = val.strip()
                 break
+        if not token:
+            config = getattr(adapter, "config", None)
+            val = getattr(config, "token", None)
+            if isinstance(val, str) and val.strip():
+                token = val.strip()
         if not token:
             return None
         import hashlib

tests/gateway/test_multiplex_adapter_registry.py

@@ -5,8 +5,9 @@ from gateway.run import GatewayRunner
 
 
 class _FakeAdapter:
-    def __init__(self, token=None):
+    def __init__(self, token=None, config=None):
         self.token = token
+        self.config = config
 
 
 class TestCredentialFingerprint:
@@ -32,6 +33,17 @@ class TestCredentialFingerprint:
                 self.bot_token = "alt-token"
         assert GatewayRunner._adapter_credential_fingerprint(_AltAdapter()) is not None
 
+    def test_reads_platform_config_token(self):
+        class _Config:
+            token = "config-token"
+
+        fp = GatewayRunner._adapter_credential_fingerprint(
+            _FakeAdapter(token=None, config=_Config())
+        )
+
+        assert fp is not None
+        assert "config-token" not in fp
+
 
 class TestProfileMessageHandler:
     @pytest.mark.asyncio
@@ -127,10 +139,57 @@ 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_secondary_same_config_token_is_refused(self, monkeypatch):
+        """Adapters that keep their token on config still trip the mux guard."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        class _ConfigTokenAdapter:
+            def __init__(self, token):
+                self.config = PlatformConfig(enabled=True, token=token)
+                self.disconnected = False
+
+            async def connect(self):
+                raise AssertionError("duplicate adapter must not connect")
+
+            async def disconnect(self):
+                self.disconnected = True
+
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = GatewayConfig(multiplex_profiles=True)
+        runner._profile_adapters = {}
+
+        reviewer_cfg = GatewayConfig(multiplex_profiles=True)
+        reviewer_cfg.platforms = {
+            Platform.TELEGRAM: PlatformConfig(enabled=True, token="same-token"),
+        }
+        duplicate = _ConfigTokenAdapter("same-token")
+        claimed = {
+            (
+                Platform.TELEGRAM,
+                GatewayRunner._adapter_credential_fingerprint(
+                    _ConfigTokenAdapter("same-token")
+                ),
+            ): "default"
+        }
+
+        monkeypatch.setattr(
+            "gateway.config.load_gateway_config", lambda: reviewer_cfg
+        )
+        monkeypatch.setattr(runner, "_create_adapter", lambda p, c: duplicate)
+        monkeypatch.setattr(runner, "_adapter_disconnect_timeout_secs", lambda: 0)
+
+        connected = await runner._start_one_profile_adapters(
+            "reviewer", "/tmp/x", claimed
+        )
+
+        assert connected == 0
+        assert duplicate.disconnected is True
+        assert runner._profile_adapters["reviewer"] == {}
+
     def test_port_binding_set_covers_known_listeners(self):
         from gateway.run import _PORT_BINDING_PLATFORM_VALUES
         # Every adapter that binds a TCP port must be in the guard set.
         for p in ("webhook", "api_server", "msgraph_webhook", "feishu",
                   "wecom_callback", "bluebubbles", "sms"):
             assert p in _PORT_BINDING_PLATFORM_VALUES
-