[devjoinedthechat]

livekit-plugins-assemblyai opens its STT socket without heartbeat, so the read side has no liveness bound on a half-open connection (no FIN, no RST). Same class as #7206 (Deepgram v1), #7357 (Soniox), #7359 (Telnyx), #7364 (xAI) and #7369 (Simplismart). The send side of this plugin was already covered by #6467, so this is only the read side.

The gap

_connect_ws passes only headers, and aiohttp defaults heartbeat to None, so nothing probes the connection.

Unlike the plugins above, recv_task here does not park — it reads with asyncio.wait_for(ws.receive(), timeout=5). But the asyncio.TimeoutError branch leaves the loop only when closing_ws marks a graceful shutdown; otherwise it increments a counter, logs every third timeout, and continues. There is no other exit, and the retry in SpeechStream._main_task only runs when _run() raises an APIError (stt/stt.py:471-477).

So on a dead socket the only thing that can end the stream is send_task's write failing, which waits on the kernel's TCP timeout — and nothing writes at all while no audio is being pushed. Meanwhile the plugin logs no messages received every 15s and the session goes on believing STT is live. The stall is visible in the log but never acted on.

Why a heartbeat and not a read-inactivity bound

The counter is already there, so giving up after N timeouts looks like the smaller change. It isn't safe: the plugin opts into neither session_heartbeat nor acknowledge_silence, so the server is quiet while nobody speaks, read inactivity is indistinguishable from a quiet caller, and a plugin-side give-up would recycle healthy connections. A ping is the only liveness probe silence does not suppress. heartbeat=30.0 is parity with deepgram, soniox, telnyx, xai, simplismart and sarvam.

Turning on session_heartbeat and bounding read inactivity would also work, but it changes the session parameters sent to AssemblyAI and adds a message type for the plugin to handle. The transport ping needs neither.

The v3 endpoint already answers pings in normal use: AssemblyAI's own Python SDK connects with the websockets library's default keepalive, a ping every 20s and a close if no pong arrives within 20s (AsyncRealTimeTranscriber passes only headers; the threading client has had the same keepalive since websockets 15.0; keepalive defaults). Both connect to the same wss://streaming.assemblyai.com/v3/ws as this plugin. I have no AssemblyAI key, so this rests on the SDK's behaviour rather than a live run.

Second, smaller change: keep the reason

WSMsgType.ERROR — the shape a heartbeat timeout arrives in — fell through to the msg.type != aiohttp.WSMsgType.TEXT branch, which logs an unexpected message type and continues.

I want to be precise that this half is diagnostics, not recovery. aiohttp feeds exactly one ERROR and then reports CLOSED for every later read (_handle_ping_pong_exception → _set_closed, then receive() returns WS_CLOSED_MESSAGE), so the stream did still end — through the close, raised as a generic AssemblyAI connection closed unexpectedly with status_code=1006 and the reason on ws.exception() discarded. Running the new test against the parent commit shows exactly that:

APIStatusError: message='AssemblyAI connection closed unexpectedly', status_code=1006, retryable=True

Raising from the ERROR frame keeps ws.exception() as the cause and stays retryable, so _main_task still reconnects and the error says why. Same gap #7206 fixed for Deepgram v1.

raise rather than break is deliberate: _run awaits asyncio.gather(send_task, recv_task) and send_task sits on self._input_ch, so a recv_task that returned normally would leave gather waiting. It also matches what this file already does on the neighbouring paths — APIStatusError for an unexpected close, APIConnectionError from send_task.

The existing timeout warnings stay; they still separate "no audio reaching us" from "nothing coming back", which a heartbeat does not answer.

Tests

Two regression tests in tests/test_plugin_assemblyai_stt.py, both failing on the parent commit:

• test_socket_is_opened_with_a_heartbeat — records the kwargs _connect_ws passes; on the parent they are ['headers'] and heartbeat is None.
• test_error_frame_surfaces_the_transport_cause — drives _run against a socket that mirrors aiohttp's real sequence, one ERROR then CLOSED. On the parent it raises the bare-1006 APIStatusError above; with this change the ServerTimeoutError is the raised error's __cause__.

uv run pytest tests/test_plugin_assemblyai_stt.py --plugin assemblyai → 122 passed. make check (ruff format, ruff, mypy strict over 685 files) clean. uv run pytest --unit --audio_eot gives the same result on this branch and on main: 3609 passed, 5 skipped, 9 errors.

Per REVIEW.md, the new log line keeps sensitive values out: session= is a generated correlation id, and the %r is a transport exception on an already-established connection — recv_task only runs after _connect_ws returned, so it cannot be a handshake error carrying the Authorization header.

[CLAassistant]

All committers have signed the CLA.

[davidzhao]

Thanks @devjoinedthechat for this! #8123 added a built-in liveness monitor to AssemblyAI STT, which covers this, so closing.