diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index 9c3d22a429fa..a1de278f424a 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -108,6 +108,23 @@ def _is_audio_ext(ext: str) -> bool: return ext.lower() in {".mp3", ".wav", ".ogg", ".m4a", ".aac"} +def _handshake_status_code(exc: Exception) -> Optional[int]: + """Extract the HTTP status from a websockets handshake-rejection error. + + websockets >= 14 raises ``InvalidStatus`` carrying ``.response.status_code``; + older releases raised ``InvalidStatusCode`` with a flat ``.status_code``. + Return the status as an int, or ``None`` if it cannot be determined. + """ + response = getattr(exc, "response", None) + status = getattr(response, "status_code", None) + if status is None: + status = getattr(exc, "status_code", None) + try: + return int(status) if status is not None else None + except (TypeError, ValueError): + return None + + # --------------------------------------------------------------------------- # SimpleX Adapter # --------------------------------------------------------------------------- @@ -215,6 +232,17 @@ async def _ws_listener(self) -> None: """Maintain a persistent WebSocket connection to the daemon.""" import websockets as _wsclient import websockets as _wsexc + # websockets >= 14 raises InvalidStatus on a rejected handshake + # (carrying a .response with .status_code). Older releases used + # InvalidStatusCode (with a .status_code attribute) and still expose + # the name as a deprecated alias; fall back to it so the adapter works + # against either version. + try: + from websockets.exceptions import InvalidStatus as _WSInvalidStatus + except ImportError: # pragma: no cover - very old websockets + from websockets.exceptions import ( # type: ignore[attr-defined] + InvalidStatusCode as _WSInvalidStatus, + ) backoff = WS_RETRY_DELAY_INITIAL @@ -245,6 +273,25 @@ async def _ws_listener(self) -> None: except asyncio.CancelledError: break + except _WSInvalidStatus as e: + # The daemon rejected the WebSocket handshake with an HTTP + # status. A 4xx is a permanent misconfiguration (bad path, + # auth, wrong endpoint) that will never succeed on retry — + # escalate to a gateway-visible fatal state instead of + # busy-looping forever. 5xx may be transient, so keep + # retrying those via the generic path below. + status = _handshake_status_code(e) + if status is not None and 400 <= status < 500: + message = f"SimpleX daemon rejected WebSocket handshake (HTTP {status})" + logger.error("SimpleX WS: %s — stopping reconnect", message) + self._set_fatal_error("handshake_rejected", message, retryable=False) + break + if self._running: + logger.warning( + "SimpleX WS: handshake rejected (HTTP %s) " + "(reconnecting in %.0fs)", + status, backoff, + ) except _wsexc.WebSocketException as e: if self._running: logger.warning( diff --git a/tests/gateway/test_simplex_plugin.py b/tests/gateway/test_simplex_plugin.py index 1048168aa6e6..92f8a22a58ab 100644 --- a/tests/gateway/test_simplex_plugin.py +++ b/tests/gateway/test_simplex_plugin.py @@ -320,6 +320,116 @@ async def test_standalone_send_missing_url(monkeypatch): assert "error" in result +# --------------------------------------------------------------------------- +# 9b. Reconnect loop escalates a non-retryable handshake rejection +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_ws_listener_escalates_4xx_handshake_rejection(monkeypatch): + """A 4xx WebSocket-handshake rejection is fatal, not an endless retry. + + A permanently misconfigured daemon answers the handshake with an HTTP + 4xx. That can never succeed on retry, so the listener must flag a + non-retryable fatal error and stop looping instead of busy-reconnecting + forever. + """ + import asyncio + + import websockets + from gateway.config import PlatformConfig + + # Build a real handshake-rejection exception for the installed + # websockets version (InvalidStatus on >= 14, else InvalidStatusCode). + try: + from websockets.exceptions import InvalidStatus + from websockets.datastructures import Headers + from websockets.http11 import Response + + rejection = InvalidStatus(Response(403, "Forbidden", Headers())) + except ImportError: # pragma: no cover - very old websockets + from websockets.exceptions import InvalidStatusCode + + rejection = InvalidStatusCode(403, Headers()) # type: ignore[call-arg] + + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._running = True + + # Every connect attempt is rejected with the 4xx handshake error. + def _raise_rejection(*args, **kwargs): + raise rejection + + monkeypatch.setattr(websockets, "connect", _raise_rejection) + + fatal_calls = [] + monkeypatch.setattr( + adapter, + "_set_fatal_error", + lambda code, message, *, retryable: fatal_calls.append( + (code, message, retryable) + ), + ) + + # If the fix is missing the loop busy-retries forever — fail fast rather + # than hang the suite. + await asyncio.wait_for(adapter._ws_listener(), timeout=5.0) + + assert len(fatal_calls) == 1, "expected exactly one fatal-error escalation" + code, _message, retryable = fatal_calls[0] + assert retryable is False + assert code == "handshake_rejected" + + +@pytest.mark.asyncio +async def test_ws_listener_retries_5xx_handshake_rejection(monkeypatch): + """A 5xx handshake rejection is transient — keep retrying, don't escalate.""" + import asyncio + + import websockets + from gateway.config import PlatformConfig + + try: + from websockets.exceptions import InvalidStatus + from websockets.datastructures import Headers + from websockets.http11 import Response + + rejection = InvalidStatus(Response(503, "Unavailable", Headers())) + except ImportError: # pragma: no cover - very old websockets + from websockets.exceptions import InvalidStatusCode + + rejection = InvalidStatusCode(503, Headers()) # type: ignore[call-arg] + + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._running = True + + attempts = 0 + + def _raise_rejection(*args, **kwargs): + nonlocal attempts + attempts += 1 + # Stop the loop after it has demonstrated a retry rather than escalating. + if attempts >= 2: + adapter._running = False + raise rejection + + monkeypatch.setattr(websockets, "connect", _raise_rejection) + # Don't actually sleep through the backoff. + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + fatal_calls = [] + monkeypatch.setattr( + adapter, + "_set_fatal_error", + lambda code, message, *, retryable: fatal_calls.append(retryable), + ) + + await asyncio.wait_for(adapter._ws_listener(), timeout=5.0) + + assert attempts >= 2, "5xx should be retried, not escalated on first hit" + assert fatal_calls == [], "5xx must not trigger a fatal escalation" + + # --------------------------------------------------------------------------- # 10. register() — plugin-side metadata # ---------------------------------------------------------------------------