fix(gateway): bind api_server directly instead of pre-probing 127.0.0.1 (#65621)

The single-family pre-probe (_port_is_available) raced the real bind and
reported a lingering TIME_WAIT socket as 'in use', failing gateway
restarts for up to ~60s (#10297). Port the webhook adapter's bind
mechanics (#63711/#65482): delete the probe, bind directly with clean
OSError handling and runner teardown, and scope reuse_address=False to
macOS only so Linux restarts rebind past TIME_WAIT instantly.

Credit to @lrawnsley (#10297) for identifying the TIME_WAIT restart
failure.

gateway/platforms/api_server.py

@@ -48,9 +48,9 @@ from contextvars import ContextVar
 from functools import wraps
 import logging
 import os
-import socket as _socket
 import re
 import sqlite3
+import sys
 import time
 import uuid
 from pathlib import Path
@@ -5115,21 +5115,6 @@ class APIServerAdapter(BasePlatformAdapter):
             pass
         return True
 
-    def _port_is_available(self) -> bool:
-        """Return True when the configured listen port is free."""
-        try:
-            with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as _s:
-                _s.settimeout(1)
-                _s.connect(('127.0.0.1', self._port))
-            logger.error(
-                "[%s] Port %d already in use. Set a different port in config.yaml: "
-                "platforms.api_server.port",
-                self.name, self._port,
-            )
-            return False
-        except (ConnectionRefusedError, OSError):
-            return True
-
     async def connect(self, *, is_reconnect: bool = False) -> bool:
         """Start the aiohttp web server."""
         if not AIOHTTP_AVAILABLE:
@@ -5139,9 +5124,6 @@ class APIServerAdapter(BasePlatformAdapter):
         if not self._api_key_passes_startup_guard():
             return False
 
-        if not self._port_is_available():
-            return False
-
         try:
             mws = [
                 mw
@@ -5206,8 +5188,36 @@ class APIServerAdapter(BasePlatformAdapter):
 
             self._runner = web.AppRunner(self._app)
             await self._runner.setup()
-            self._site = web.TCPSite(self._runner, self._host, self._port)
+            # Bind directly instead of probing 127.0.0.1 first — the old
-            await self._site.start()
+            # single-family pre-probe raced the real bind and reported a
+            # TIME_WAIT socket as "in use" (#10297), failing gateway
+            # restarts for up to ~60s.
+            #
+            # SO_REUSEADDR is platform-dependent (same rationale as the
+            # webhook adapter, #65482):
+            #   - macOS (BSD semantics): two sockets with SO_REUSEADDR can
+            #     silently split traffic while both report success — disable.
+            #   - Linux: SO_REUSEADDR only permits rebinding past TIME_WAIT
+            #     (a second live listener needs SO_REUSEPORT, never set), so
+            #     keep the default (enabled) for instant restart rebinds.
+            self._site = web.TCPSite(
+                self._runner,
+                self._host,
+                self._port,
+                reuse_address=False if sys.platform == "darwin" else None,
+            )
+            try:
+                await self._site.start()
+            except OSError as exc:
+                await self._runner.cleanup()
+                self._runner = None
+                self._site = None
+                logger.error(
+                    "[%s] Could not bind %s:%d: %s. Set a different port in "
+                    "config.yaml: platforms.api_server.port",
+                    self.name, self._host, self._port, exc,
+                )
+                return False
 
             self._mark_connected()
             logger.info(

tests/gateway/test_api_server_bind_guard.py

@@ -144,3 +144,76 @@ class TestConnectBindGuard:
         assert adapter._api_key == "sk-test"
         assert is_network_accessible("0.0.0.0") is True
         # Combined: the guard condition is False (key is set), so it passes
+
+
+# ---------------------------------------------------------------------------
+# Integration tests: bind mechanics (direct bind, no pre-probe — #10297)
+# ---------------------------------------------------------------------------
+
+
+class TestBindMechanics:
+    """connect() binds directly instead of pre-probing 127.0.0.1.
+
+    The old ``_port_is_available()`` probe connected to 127.0.0.1 only and
+    reported a lingering TIME_WAIT socket as "in use", failing gateway
+    restarts for up to ~60s (#10297). The fix removes the probe: bind
+    directly, keep SO_REUSEADDR default semantics on Linux (rebind past
+    TIME_WAIT), and surface a real bind conflict as a clean ``False`` with
+    the runner torn down.
+    """
+
+    _KEY = "sk-test-strong-key-0123456789"
+
+    def _make_adapter(self, port: int) -> APIServerAdapter:
+        return APIServerAdapter(
+            PlatformConfig(
+                enabled=True,
+                extra={"host": "127.0.0.1", "port": port, "key": self._KEY},
+            )
+        )
+
+    @staticmethod
+    def _free_port() -> int:
+        with socket.socket() as s:
+            s.bind(("", 0))
+            return s.getsockname()[1]
+
+    @pytest.mark.asyncio
+    async def test_immediate_rebind_after_disconnect(self):
+        """A restarted adapter can rebind the same port immediately.
+
+        This is the #10297 symptom: the old pre-probe (and disabled address
+        reuse) made a quick gateway restart fail while the previous socket
+        sat in TIME_WAIT.
+        """
+        port = self._free_port()
+        first = self._make_adapter(port)
+        assert await first.connect() is True
+        await first.disconnect()
+
+        second = self._make_adapter(port)
+        try:
+            assert await second.connect() is True
+        finally:
+            await second.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_live_listener_conflict_returns_false_and_cleans_up(self):
+        """A second adapter on an occupied port fails cleanly, not with a raise."""
+        port = self._free_port()
+        first = self._make_adapter(port)
+        assert await first.connect() is True
+        second = self._make_adapter(port)
+        try:
+            result = await second.connect()
+            assert result is False
+            assert second._runner is None
+            assert second._site is None
+            assert second.is_connected is False
+        finally:
+            await first.disconnect()
+            await second.disconnect()
+
+    def test_pre_probe_helper_removed(self):
+        """The racy single-family pre-probe must not come back."""
+        assert not hasattr(APIServerAdapter, "_port_is_available")