fix(gateway): release failed Photon listener claims

gateway/run.py

@@ -10340,8 +10340,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
             # Same-token conflict detection — refuse a duplicate poll.
             fp = self._adapter_credential_fingerprint(adapter)
-            if fp is not None:
-                owner = claimed.get((platform, fp))
+            credential_claim = (platform, fp) if fp is not None else None
+            if credential_claim is not None:
+                owner = claimed.get(credential_claim)
                 if owner is not None:
                     logger.error(
                         "Profile '%s' and '%s' both configure %s with the same "
@@ -10355,7 +10356,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     # the shared platform state and, for a same-credential Photon
                     # adapter, shut down the primary profile's live sidecar.
                     continue
-                claimed[(platform, fp)] = profile_name
 
             listener_claim = self._adapter_listener_claim(platform, adapter)
             if listener_claim is not None:
@@ -10378,7 +10378,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     # 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)
 
@@ -10389,6 +10388,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     )
                 if success:
                     profile_map[platform] = adapter
+                    if credential_claim is not None:
+                        claimed[credential_claim] = profile_name
+                    if listener_claim is not None:
+                        claimed[listener_claim] = profile_name
                     connected += 1
                     logger.info("✓ %s connected (profile: %s)", platform.value, profile_name)
                 else:

tests/gateway/test_multiplex_adapter_registry.py

@@ -889,6 +889,61 @@ class TestSecondaryProfileConfigHandling:
         assert secondary.disconnected is False
         assert runner._profile_adapters["reviewer"][photon] is secondary
 
+    @pytest.mark.asyncio
+    async def test_failed_photon_connect_releases_listener_for_later_profile(
+        self, monkeypatch
+    ):
+        """A failed sidecar must not reserve an endpoint it never owned."""
+        from gateway.config import GatewayConfig, Platform, PlatformConfig
+
+        class _PhotonAdapter:
+            def __init__(self, secret, should_connect):
+                self._project_secret = secret
+                self._sidecar_bind = "127.0.0.1"
+                self._sidecar_port = 8789
+                self.platform = Platform("photon")
+                self.should_connect = should_connect
+                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")
+        profile_cfg = GatewayConfig(multiplex_profiles=True)
+        profile_cfg.platforms = {photon: PlatformConfig(enabled=True)}
+        failed = _PhotonAdapter("failed-secret", False)
+        later = _PhotonAdapter("later-secret", True)
+        adapters = iter((failed, later))
+        claimed = {}
+
+        async def _connect(adapter, platform):
+            return adapter.should_connect
+
+        monkeypatch.setattr("gateway.config.load_gateway_config", lambda: profile_cfg)
+        monkeypatch.setattr(runner, "_create_adapter", lambda p, c: next(adapters))
+        monkeypatch.setattr(runner, "_connect_adapter_with_timeout", _connect)
+        monkeypatch.setattr(runner, "_make_adapter_auth_check", lambda p: None)
+
+        first = await runner._start_one_profile_adapters("broken", "/tmp/x", claimed)
+        second = await runner._start_one_profile_adapters("later", "/tmp/y", claimed)
+
+        assert first == 0
+        assert failed.disconnected is True
+        assert second == 1
+        assert runner._profile_adapters["later"][photon] is later
+
     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.