fix(gateway): multiplex must not apply default-profile creds to unconfigured profiles (#84079)

Secondary profile startup and reconnect now call the existing
`_platform_has_bot_credential` gate (the same one the primary loop and
primary reconnect use since #64674), so an enabled-in-YAML platform whose
credential is absent from that profile's secret scope is skipped instead
of built with an empty token and fanned out.

Independently reported and fixed in #72313 (@manny3), which added a
duplicate helper; the shared main helper is used here instead.

Co-authored-by: manny3 <16465310+manny3@users.noreply.github.com>

gateway/run.py

@@ -16995,6 +16995,25 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         for platform, platform_config in profile_cfg.platforms.items():
             if not platform_config.enabled:
                 continue
+            # A platform enabled in a secondary profile's config.yaml may
+            # have no credential in that profile's secret scope — the shared
+            # YAML enables it for the default profile only (#84079). Building
+            # an adapter here would treat every credential-less profile as
+            # configured for the platform and one inbound message would fan
+            # out across all of them. Mirror the primary startup loop's
+            # credential gate and skip instead; profiles with their own
+            # credential still connect below.
+            if (
+                getattr(self.config, "multiplex_profiles", False)
+                and not _platform_has_bot_credential(platform, platform_config)
+            ):
+                logger.info(
+                    "[MULTIPLEX] Profile '%s': skipping %s - no bot credential "
+                    "in this profile's secrets",
+                    profile_name,
+                    platform.value,
+                )
+                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.
@@ -17180,6 +17199,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                         profile_config = load_gateway_config().platforms.get(platform)
                         if profile_config is None or not profile_config.enabled:
                             return
+                        # Mirrors the startup credential gate (#84079): a
+                        # credential removed from this profile's scope must
+                        # not rebuild an adapter that would fan out turns.
+                        if not _platform_has_bot_credential(platform, profile_config):
+                            logger.info(
+                                "Secondary %s reconnect skipped: no bot credential "
+                                "(profile: %s)",
+                                platform.value,
+                                profile_name,
+                            )
+                            return
                         adapter = self._create_adapter(platform, profile_config)
                         if adapter is None:
                             logger.warning(

tests/gateway/test_multiplex_adapter_registry.py

@@ -897,3 +897,78 @@ class TestFeishuPortBindingConditional:
         assert connected == 0  # no error, just nothing connected
 
 
+class TestSecondarySkipsCredentiallessPlatforms:
+    """#84079 — multiplex must not build adapters for platforms a profile
+    has no credential for.
+
+    The shared config.yaml enables a platform once; under multiplex every
+    secondary profile reloads it inside its own secret scope, so a profile
+    whose scope lacks the platform credential resolves ``enabled=True`` with
+    an empty token. Constructing an adapter anyway treats every profile as
+    configured for the platform — one inbound message fans out across all of
+    them. These tests lock the credential gate on the secondary startup path
+    (the primary path got the same gate in #64674; the reconnect path shares
+    the helper). Also reported independently in #72313.
+    """
+
+    def _make_runner(self, monkeypatch, profile_cfg):
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = GatewayConfig(multiplex_profiles=True)
+        runner._profile_adapters = {}
+        runner.adapters = {}
+        created = []
+
+        def fake_create(platform, platform_config):
+            created.append((platform, platform_config))
+            return _FakeAdapter(token=platform_config.token or None)
+
+        monkeypatch.setattr("gateway.config.load_gateway_config", lambda: profile_cfg)
+        monkeypatch.setattr(runner, "_create_adapter", fake_create)
+        monkeypatch.setattr(runner, "_configure_profile_adapter", lambda *a, **k: None)
+        monkeypatch.setattr(
+            runner,
+            "_connect_initial_adapter_with_timeout",
+            AsyncMock(return_value=True),
+        )
+        return runner, created
+
+    @pytest.mark.asyncio
+    async def test_credentialless_platform_builds_no_adapter(self, monkeypatch, tmp_path):
+        """Enabled-in-YAML but no credential in the profile scope -> no adapter."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        profile_cfg = GatewayConfig(multiplex_profiles=True)
+        profile_cfg.platforms = {
+            # Shared config.yaml enables Slack; profile-b's .env has no
+            # SLACK_BOT_TOKEN, so its scoped load resolves token="" but
+            # keeps enabled=True (#84079).
+            Platform.SLACK: PlatformConfig(enabled=True, token=""),
+            Platform.TELEGRAM: PlatformConfig(enabled=True, token="telegram-token-b"),
+        }
+        runner, created = self._make_runner(monkeypatch, profile_cfg)
+
+        connected = await runner._start_one_profile_adapters("profile-b", tmp_path, {})
+
+        # Only Telegram (which profile-b has its own credential for) gets an
+        # adapter; Slack is skipped instead of fanning out a turn per profile.
+        assert [p for p, _ in created] == [Platform.TELEGRAM]
+        assert connected == 1
+        assert Platform.TELEGRAM in runner._profile_adapters["profile-b"]
+        assert Platform.SLACK not in runner._profile_adapters["profile-b"]
+
+    @pytest.mark.asyncio
+    async def test_profile_with_own_credential_still_connects(self, monkeypatch, tmp_path):
+        """A profile that defines its own credential keeps its adapter."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        profile_cfg = GatewayConfig(multiplex_profiles=True)
+        profile_cfg.platforms = {
+            Platform.SLACK: PlatformConfig(enabled=True, token="slack-token-b"),
+        }
+        runner, created = self._make_runner(monkeypatch, profile_cfg)
+
+        connected = await runner._start_one_profile_adapters("profile-b", tmp_path, {})
+
+        assert connected == 1
+        assert created == [(Platform.SLACK, profile_cfg.platforms[Platform.SLACK])]
+        assert Platform.SLACK in runner._profile_adapters["profile-b"]