fix(gateway): guard Photon sidecar listener collisions

gateway/run.py

@@ -10220,14 +10220,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         active = get_active_profile_name() or "default"
         connected = 0
-        # (platform, token-fingerprint) -> profile that claimed it. Detects two
-        # profiles trying to poll the same bot credential (impossible to do
-        # concurrently). Seed with the active profile's adapters.
+        # Resource claim -> profile that owns it. Credential claims prevent two
+        # profiles polling the same account; listener claims prevent sidecars
+        # with distinct credentials from binding the same endpoint.
         claimed: Dict[tuple, str] = {}
         for _plat, _ad in self.adapters.items():
             fp = self._adapter_credential_fingerprint(_ad)
             if fp is not None:
                 claimed[(_plat, fp)] = active
+            listener_claim = self._adapter_listener_claim(_plat, _ad)
+            if listener_claim is not None:
+                claimed[listener_claim] = active
 
         for profile_name, profile_home in profiles_to_serve(multiplex=True):
             if profile_name == active:
@@ -10354,6 +10357,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     continue
                 claimed[(platform, fp)] = profile_name
 
+            listener_claim = self._adapter_listener_claim(platform, adapter)
+            if listener_claim is not None:
+                owner = claimed.get(listener_claim)
+                if owner is not None:
+                    bind, port = listener_claim[-2:]
+                    logger.error(
+                        "Profile '%s' and '%s' both configure %s sidecars on "
+                        "%s:%s — refusing to start the duplicate listener. "
+                        "Set platforms.%s.extra.sidecar_port to a distinct port "
+                        "for profile '%s'.",
+                        owner,
+                        profile_name,
+                        platform.value,
+                        bind,
+                        port,
+                        platform.value,
+                        profile_name,
+                    )
+                    # Like credential conflicts, this adapter never connected
+                    # and owns no resources that should be disconnected.
+                    continue
+                claimed[listener_claim] = profile_name
+
             self._configure_profile_adapter(adapter, profile_name, platform)
 
             try:
@@ -10590,6 +10616,28 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         return _handler
 
+    @staticmethod
+    def _adapter_listener_claim(platform: Platform, adapter: Any) -> Optional[tuple]:
+        """Return the exclusive listener resource claimed by an adapter.
+
+        Photon sidecars are per-profile processes. Even when two profiles use
+        different project credentials, their sidecars cannot share a bind and
+        port. Represent that endpoint as a claim so multiplex startup rejects
+        the later adapter before either ``connect()`` or ``disconnect()`` can
+        disturb the first profile.
+        """
+        if getattr(platform, "value", None) != "photon":
+            return None
+        bind = getattr(adapter, "_sidecar_bind", None)
+        port = getattr(adapter, "_sidecar_port", None)
+        if not isinstance(bind, str) or not bind.strip():
+            return None
+        try:
+            port = int(port)
+        except (TypeError, ValueError):
+            return None
+        return ("listener", "photon", bind.strip().lower(), port)
+
     @staticmethod
     def _adapter_credential_fingerprint(adapter: Any) -> Optional[str]:
         """Return a stable, log-safe fingerprint of an adapter's credential.

tests/gateway/test_multiplex_adapter_registry.py

@@ -783,6 +783,112 @@ class TestSecondaryProfileConfigHandling:
         assert duplicate.disconnected is False
         assert runner._profile_adapters["reviewer"] == {}
 
+    @pytest.mark.asyncio
+    async def test_secondary_distinct_photon_credentials_same_port_are_refused(
+        self, monkeypatch
+    ):
+        """The sidecar listener is exclusive even when credentials differ."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        class _PhotonAdapter:
+            def __init__(self, secret, port=8789):
+                self._project_secret = secret
+                self._sidecar_bind = "127.0.0.1"
+                self._sidecar_port = port
+                self.platform = Platform("photon")
+                self.connected = False
+                self.disconnected = False
+
+            async def connect(self):
+                self.connected = True
+                raise AssertionError("conflicting sidecar must not connect")
+
+            async def disconnect(self):
+                self.disconnected = True
+
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = GatewayConfig(multiplex_profiles=True)
+        runner._profile_adapters = {}
+
+        photon = Platform("photon")
+        reviewer_cfg = GatewayConfig(multiplex_profiles=True)
+        reviewer_cfg.platforms = {photon: PlatformConfig(enabled=True)}
+        primary = _PhotonAdapter("primary-secret")
+        duplicate = _PhotonAdapter("different-secret")
+        claimed = {
+            GatewayRunner._adapter_listener_claim(photon, primary): "default"
+        }
+
+        monkeypatch.setattr("gateway.config.load_gateway_config", lambda: reviewer_cfg)
+        monkeypatch.setattr(runner, "_create_adapter", lambda p, c: duplicate)
+
+        connected = await runner._start_one_profile_adapters(
+            "reviewer", "/tmp/x", claimed
+        )
+
+        assert connected == 0
+        assert duplicate.connected is False
+        assert duplicate.disconnected is False
+        assert runner._profile_adapters["reviewer"] == {}
+
+    @pytest.mark.asyncio
+    async def test_secondary_distinct_photon_credentials_distinct_ports_connect(
+        self, monkeypatch
+    ):
+        """Multiplexing remains supported when Photon sidecars cannot collide."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        class _PhotonAdapter:
+            def __init__(self, secret, port):
+                self._project_secret = secret
+                self._sidecar_bind = "127.0.0.1"
+                self._sidecar_port = port
+                self.platform = Platform("photon")
+                self.connected = False
+                self.disconnected = False
+                self.config = PlatformConfig(enabled=True)
+
+            def __getattr__(self, name):
+                if name.startswith("set_"):
+                    return lambda *args, **kwargs: None
+                raise AttributeError(name)
+
+            async def disconnect(self):
+                self.disconnected = True
+
+        runner = GatewayRunner.__new__(GatewayRunner)
+        runner.config = GatewayConfig(multiplex_profiles=True)
+        runner._profile_adapters = {}
+        runner.session_store = None
+        runner._busy_text_mode = "queue"
+
+        photon = Platform("photon")
+        reviewer_cfg = GatewayConfig(multiplex_profiles=True)
+        reviewer_cfg.platforms = {photon: PlatformConfig(enabled=True)}
+        primary = _PhotonAdapter("primary-secret", 8789)
+        secondary = _PhotonAdapter("different-secret", 8790)
+        claimed = {
+            GatewayRunner._adapter_listener_claim(photon, primary): "default"
+        }
+
+        async def _connect(adapter, platform):
+            adapter.connected = True
+            return True
+
+        monkeypatch.setattr("gateway.config.load_gateway_config", lambda: reviewer_cfg)
+        monkeypatch.setattr(runner, "_create_adapter", lambda p, c: secondary)
+        monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect)
+        monkeypatch.setattr(runner, "_make_adapter_auth_check", lambda p: None)
+
+        connected = await runner._start_one_profile_adapters(
+            "reviewer", "/tmp/x", claimed
+        )
+
+        assert connected == 1
+        assert secondary.connected is True
+        assert secondary.disconnected is False
+        assert runner._profile_adapters["reviewer"][photon] is secondary
+
     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.