fix(stt): notice a half-open websocket in the openai, elevenlabs and inworld streams

livekit-plugins/livekit-plugins-elevenlabs/livekit/plugins/elevenlabs/stt.py

@@ -628,6 +628,19 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
                         body=f"{msg.data=} {msg.extra=}",
                     )
 
+                # A heartbeat timeout arrives here as an ERROR frame rather than as a
+                # close frame. Letting it fall through to the unexpected-message-type
+                # branch stepped over it: the loop ended normally, `_run` broke out
+                # instead of reconnecting, and the stream died without the base class
+                # ever seeing a failure -- so FallbackAdapter had nothing to react to.
+                # The reason only survives on `ws.exception()`.
+                if msg.type == aiohttp.WSMsgType.ERROR:
+                    if closing_ws or self._session.closed:
+                        return
+                    raise APIConnectionError(
+                        message="ElevenLabs STT websocket failed"
+                    ) from ws.exception()
+
                 if msg.type != aiohttp.WSMsgType.TEXT:
                     logger.warning("unexpected ElevenLabs STT message type %s", msg.type)
                     continue
@@ -743,6 +756,12 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
                 self._session.ws_connect(
                     ws_url,
                     headers={AUTHORIZATION_HEADER: self._opts.api_key},
+                    # Without a heartbeat aiohttp never notices a half-open socket (one
+                    # that died without a FIN/RST): the receive side parks forever, the
+                    # stream never raises, and so neither the reconnect nor a
+                    # FallbackAdapter switch is ever triggered. An unanswered ping makes
+                    # aiohttp close the socket, which surfaces in the receive loop.
+                    heartbeat=30.0,
                 ),
                 self._conn_options.timeout,
             )

livekit-plugins/livekit-plugins-inworld/livekit/plugins/inworld/stt.py

@@ -275,6 +275,12 @@ async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
             self._ensure_session().ws_connect(
                 ws_url,
                 headers={"Authorization": self._stt._authorization},
+                # Without a heartbeat aiohttp never notices a half-open socket (one
+                # that died without a FIN/RST): the receive side parks forever, the
+                # stream never raises, and so neither the reconnect nor a
+                # FallbackAdapter switch is ever triggered. An unanswered ping makes
+                # aiohttp close the socket, which surfaces in the receive loop.
+                heartbeat=30.0,
             ),
             timeout=self._conn_options.timeout,
         )
@@ -394,8 +400,14 @@ async def _recv_messages_task(self) -> None:
                 ):
                     return
                 elif msg.type == aiohttp.WSMsgType.ERROR:
-                    logger.error(f"Inworld STT WebSocket error: {self._ws.exception()}")
-                    return
+                    # A heartbeat timeout arrives here as an ERROR frame rather than as
+                    # a close frame. Returning normally ended the stream silently:
+                    # `_run` breaks out instead of reconnecting, and the base class only
+                    # reports a failure when `_run` raises -- so FallbackAdapter had
+                    # nothing to react to.
+                    raise APIConnectionError(
+                        message="Inworld STT websocket failed"
+                    ) from self._ws.exception()
         except aiohttp.ClientError as e:
             logger.error(f"WebSocket error while receiving: {e}")
             raise

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/stt.py

@@ -590,7 +590,15 @@ async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
         session = self._ensure_session()
         # the config is sent once the stream acquires the connection, since the pool also
         # hands back sockets it opened earlier
-        return await asyncio.wait_for(session.ws_connect(url, headers=headers), timeout)
+        return await asyncio.wait_for(
+            # Without a heartbeat aiohttp never notices a half-open socket (one
+            # that died without a FIN/RST): the receive side parks forever, the
+            # stream never raises, and so neither the reconnect nor a
+            # FallbackAdapter switch is ever triggered. An unanswered ping makes
+            # aiohttp close the socket, which surfaces in the receive loop.
+            session.ws_connect(url, headers=headers, heartbeat=30.0),
+            timeout,
+        )
 
     async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None:
         await ws.close()
@@ -844,6 +852,19 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
                         body=f"{msg.data=} {msg.extra=}",
                     )
 
+                # A heartbeat timeout arrives here as an ERROR frame rather than as a
+                # close frame. Letting it fall through to the unexpected-message-type
+                # branch stepped over it: the loop ended normally, `_run` broke out
+                # instead of reconnecting, and the stream died without the base class
+                # ever seeing a failure -- so FallbackAdapter had nothing to react to.
+                # The reason only survives on `ws.exception()`.
+                if msg.type == aiohttp.WSMsgType.ERROR:
+                    if closing_ws:
+                        return
+                    raise APIConnectionError(
+                        message="OpenAI Realtime STT websocket failed"
+                    ) from ws.exception()
+
                 if msg.type != aiohttp.WSMsgType.TEXT:
                     logger.warning("unexpected OpenAI message type %s", msg.type)
                     continue

tests/test_plugin_elevenlabs_stt.py

@@ -763,3 +763,27 @@ async def handler(request: web.Request) -> web.Response:
     assert exc_info.value.message == message
     assert exc_info.value.body == parsed
     assert requests == 1
+
+
+async def test_connect_ws_opens_the_socket_with_a_heartbeat() -> None:
+    """aiohttp defaults `heartbeat` to None, so a socket that dies without a
+    FIN/RST leaves the receive side parked forever: the stream never raises, and
+    neither the reconnect nor a FallbackAdapter switch is ever triggered."""
+    stream = _new_stream()
+
+    class _ConnOptions:
+        timeout = 5.0
+
+    stream._conn_options = _ConnOptions()
+    captured: dict[str, object] = {}
+
+    class _FakeSession:
+        async def ws_connect(self, url: str, **kwargs: object) -> object:
+            captured.update(kwargs)
+            return object()
+
+    stream._session = _FakeSession()
+
+    await stream._connect_ws()
+
+    assert captured.get("heartbeat") == 30.0

tests/test_plugin_inworld_stt.py

@@ -0,0 +1,84 @@
+"""Unit tests for the Inworld STT websocket connection."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import aiohttp
+import pytest
+
+from livekit.agents import APIConnectionError
+
+pytestmark = pytest.mark.unit
+
+
+def _make_stream():
+    """Construct an Inworld `SpeechStream` without spawning its real `_main_task`."""
+    from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS
+    from livekit.plugins.inworld import STT
+    from livekit.plugins.inworld.stt import SpeechStream
+
+    def _fake_create_task(coro, *args, **kwargs):
+        coro.close()
+        return MagicMock()
+
+    stt_instance = STT(api_key="test-key")
+    with patch("livekit.agents.stt.stt.asyncio.create_task", side_effect=_fake_create_task):
+        return SpeechStream(stt=stt_instance, conn_options=DEFAULT_API_CONNECT_OPTIONS)
+
+
+async def test_socket_is_opened_with_a_heartbeat() -> None:
+    """aiohttp defaults `heartbeat` to None, so a socket that dies without a
+    FIN/RST leaves the receive side parked forever: the stream never raises, and
+    neither the reconnect nor a FallbackAdapter switch is ever triggered."""
+    stream = _make_stream()
+    captured: dict[str, Any] = {}
+
+    class _FakeWebSocket:
+        async def send_str(self, message: str) -> None:
+            pass
+
+    class _FakeSession:
+        async def ws_connect(self, url: str, **kwargs: Any) -> _FakeWebSocket:
+            captured.update(kwargs)
+            return _FakeWebSocket()
+
+    stream._ensure_session = lambda: _FakeSession()  # type: ignore[method-assign]
+
+    await stream._connect_ws()
+
+    assert captured.get("heartbeat") == 30.0
+
+
+async def test_an_error_frame_raises_so_the_stream_can_fail_over() -> None:
+    """A heartbeat timeout arrives as an ERROR frame, not as a close frame.
+
+    Returning normally from it ended the stream silently: `_run` breaks out of its
+    loop instead of reconnecting, and `stt.SpeechStream` only reports a failure
+    when `_run` raises -- so a FallbackAdapter had nothing to react to and the
+    agent simply stopped hearing the user.
+    """
+    stream = _make_stream()
+
+    class _ErrorWebSocket:
+        def __init__(self) -> None:
+            self._sent = False
+
+        def __aiter__(self) -> _ErrorWebSocket:
+            return self
+
+        async def __anext__(self) -> Any:
+            if self._sent:
+                raise StopAsyncIteration
+            self._sent = True
+            return SimpleNamespace(type=aiohttp.WSMsgType.ERROR, data=None, extra=None)
+
+        def exception(self) -> Exception:
+            return RuntimeError("no PONG received")
+
+    stream._ws = _ErrorWebSocket()  # type: ignore[assignment]
+
+    with pytest.raises(APIConnectionError):
+        await stream._recv_messages_task()

tests/test_plugin_openai_websocket_urls.py

@@ -17,9 +17,11 @@ async def send_json(self, _data: object) -> None:
 class _FakeSession:
     def __init__(self) -> None:
         self.url = ""
+        self.kwargs: dict[str, object] = {}
 
-    async def ws_connect(self, url: str, **_kwargs: object) -> _FakeWebSocket:
+    async def ws_connect(self, url: str, **kwargs: object) -> _FakeWebSocket:
         self.url = url
+        self.kwargs = kwargs
         return _FakeWebSocket()
 
 
@@ -111,3 +113,16 @@ async def test_native_responses_websocket_connect_url_omits_model() -> None:
     assert parsed_url.netloc == "api.openai.com"
     assert parsed_url.path == "/v1/responses"
     assert parse_qs(parsed_url.query) == {}
+
+
+async def test_realtime_stt_socket_is_opened_with_a_heartbeat() -> None:
+    """aiohttp defaults `heartbeat` to None, so a socket that dies without a
+    FIN/RST leaves the receive side parked forever: the stream never raises, and
+    neither the reconnect nor a FallbackAdapter switch is ever triggered."""
+    instance = stt.STT(api_key="test-key", base_url="https://gateway.example.com/v1")
+    session = _FakeSession()
+    instance._session = session
+
+    await instance._connect_ws(5)
+
+    assert session.kwargs.get("heartbeat") == 30.0