Skip to content

fix(stt): reconnect on send-side socket drops in remaining plugins #6474

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
flasun wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(stt): reconnect on send-side socket drops in remaining plugins #6474

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dc5f491
fix(stt): reconnect on send-side socket drops in remaining plugins
flasun Jul 18, 2026
c32c19d
fix(slng): raise APIStatusError on send-side drop to match recv failover
flasun Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 18 additions & 12 deletions livekit-plugins/livekit-plugins-fireworksai/livekit/plugins/fireworksai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@

from livekit.agents import (
DEFAULT_API_CONNECT_OPTIONS,
APIConnectionError,
APIConnectOptions,
APIStatusError,
LanguageCode,
Expand Down Expand Up @@ -335,18 +336,23 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_per_buffer,
)

async for data in self._input_ch:
if isinstance(data, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(data.data.tobytes())

for frame in frames:
await self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())

closing_ws = True
await ws.send_str(self._CLOSE_MSG)
try:
async for data in self._input_ch:
if isinstance(data, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(data.data.tobytes())

for frame in frames:
await self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())

closing_ws = True
await ws.send_str(self._CLOSE_MSG)
except (aiohttp.ClientError, ConnectionError) as e:
if closing_ws:
return
raise APIConnectionError("Fireworks AI connection closed unexpectedly") from e

async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
nonlocal closing_ws
Expand Down
38 changes: 22 additions & 16 deletions livekit-plugins/livekit-plugins-gradium/livekit/plugins/gradium/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

from livekit.agents import (
DEFAULT_API_CONNECT_OPTIONS,
APIConnectionError,
APIConnectOptions,
APIStatusError,
LanguageCode,
Expand Down Expand Up @@ -256,22 +257,27 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_per_buffer,
)

async for data in self._input_ch:
if isinstance(data, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(data.data.tobytes())

for frame in frames:
if len(frame.data) % 2 != 0:
logger.warning("Frame data size not aligned to int16 (multiple of 2)")

audio_data = base64.b64encode(frame.data.tobytes()).decode("utf-8")
audio_msg = {
"type": "audio",
"audio": audio_data,
}
await ws.send_str(json.dumps(audio_msg))
try:
async for data in self._input_ch:
if isinstance(data, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(data.data.tobytes())

for frame in frames:
if len(frame.data) % 2 != 0:
logger.warning("Frame data size not aligned to int16 (multiple of 2)")

audio_data = base64.b64encode(frame.data.tobytes()).decode("utf-8")
audio_msg = {
"type": "audio",
"audio": audio_data,
}
await ws.send_str(json.dumps(audio_msg))
except (aiohttp.ClientError, ConnectionError) as e:
if self._session.closed:
return
raise APIConnectionError("Gradium connection closed unexpectedly") from e

async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
nonlocal closing_ws
Expand Down
26 changes: 17 additions & 9 deletions livekit-plugins/livekit-plugins-simplismart/livekit/plugins/simplismart/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,15 +339,23 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_50ms,
)

async for data in self._input_ch:
frames: list[rtc.AudioFrame] = []
if isinstance(data, rtc.AudioFrame):
frames.extend(audio_bstream.write(data.data.tobytes()))
elif isinstance(data, self._FlushSentinel):
frames.extend(audio_bstream.flush())

for frame in frames:
await ws.send_bytes(frame.data.tobytes())
try:
async for data in self._input_ch:
frames: list[rtc.AudioFrame] = []
if isinstance(data, rtc.AudioFrame):
frames.extend(audio_bstream.write(data.data.tobytes()))
elif isinstance(data, self._FlushSentinel):
frames.extend(audio_bstream.flush())

for frame in frames:
await ws.send_bytes(frame.data.tobytes())
except (aiohttp.ClientError, ConnectionError):
# mirror recv_task: a mid-send socket drop should trigger an internal
# reconnect (see the _run loop), not crash the session.
if self._session.closed:
return
self._reconnect_event.set()
return

@utils.log_exceptions(logger=logger)
async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
Expand Down
46 changes: 27 additions & 19 deletions livekit-plugins/livekit-plugins-slng/livekit/plugins/slng/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -523,25 +523,33 @@ async def send_task(
samples_per_channel=samples_per_buffer,
)

for frame in pending_frames:
frames = audio_bstream.write(frame.data.tobytes())
for out in frames:
if len(out.data) % bytes_per_sample != 0:
continue
await ws.send_bytes(bytes(out.data))
self._speech_duration += out.duration

async for item in self._input_ch:
if isinstance(item, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(item.data.tobytes())

for frame in frames:
if len(frame.data) % bytes_per_sample != 0:
continue
await ws.send_bytes(bytes(frame.data))
self._speech_duration += frame.duration
try:
for frame in pending_frames:
frames = audio_bstream.write(frame.data.tobytes())
for out in frames:
if len(out.data) % bytes_per_sample != 0:
continue
await ws.send_bytes(bytes(out.data))
self._speech_duration += out.duration

async for item in self._input_ch:
if isinstance(item, self._FlushSentinel):
frames = audio_bstream.flush()
else:
frames = audio_bstream.write(item.data.tobytes())

for frame in frames:
if len(frame.data) % bytes_per_sample != 0:
continue
await ws.send_bytes(bytes(frame.data))
self._speech_duration += frame.duration
except (aiohttp.ClientError, ConnectionError) as e:
if self._session.closed:
return
# match recv_task: raise APIStatusError (not APIConnectionError) so a
# transient send-side drop gets slng's same-endpoint immediate retry
# rather than a permanent failover to a lower-priority endpoint.
raise APIStatusError("SLNG connection closed unexpectedly") from e

async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
speech_started = False
Expand Down
41 changes: 23 additions & 18 deletions livekit-plugins/livekit-plugins-smallestai/livekit/plugins/smallestai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,24 +360,29 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_per_chunk,
)

async for data in self._input_ch:
if isinstance(data, rtc.AudioFrame):
for frame in audio_bstream.write(data.data.tobytes()):
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())
elif isinstance(data, self._FlushSentinel):
# User paused: drain the accumulator so the server gets all buffered
# audio. The server's eou_timeout_ms will then detect the silence and
# emit a final transcript — no explicit flush message is needed.
for frame in audio_bstream.flush():
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())
self._audio_duration_collector.flush()

# Input channel closed: close the stream so the server flushes remaining
# audio, emits final transcripts, and sends is_last=True.
closing_ws = True
await ws.send_str(SpeechStream._CLOSE_STREAM_MSG)
try:
async for data in self._input_ch:
if isinstance(data, rtc.AudioFrame):
for frame in audio_bstream.write(data.data.tobytes()):
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())
elif isinstance(data, self._FlushSentinel):
# User paused: drain the accumulator so the server gets all buffered
# audio. The server's eou_timeout_ms will then detect the silence and
# emit a final transcript — no explicit flush message is needed.
for frame in audio_bstream.flush():
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())
self._audio_duration_collector.flush()

# Input channel closed: close the stream so the server flushes remaining
# audio, emits final transcripts, and sends is_last=True.
closing_ws = True
await ws.send_str(SpeechStream._CLOSE_STREAM_MSG)
except (aiohttp.ClientError, ConnectionError) as e:
if closing_ws or self._session.closed:
return
raise APIConnectionError("Smallest AI connection closed unexpectedly") from e

@utils.log_exceptions(logger=logger)
async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
Expand Down
36 changes: 21 additions & 15 deletions livekit-plugins/livekit-plugins-telnyx/livekit/plugins/telnyx/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,6 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
nonlocal closing_ws

wav_header = _create_streaming_wav_header(self._stt._opts.sample_rate, NUM_CHANNELS)
await ws.send_bytes(wav_header)

samples_per_chunk = self._stt._opts.sample_rate // 20
audio_bstream = utils.audio.AudioByteStream(
Expand All @@ -193,20 +192,27 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_per_chunk,
)

async for data in self._input_ch:
if isinstance(data, rtc.AudioFrame):
for frame in audio_bstream.write(data.data.tobytes()):
await ws.send_bytes(frame.data.tobytes())
elif isinstance(data, self._FlushSentinel):
for frame in audio_bstream.flush():
await ws.send_bytes(frame.data.tobytes())

for frame in audio_bstream.flush():
await ws.send_bytes(frame.data.tobytes())

# Don't close the WS here — let recv_task read the final
# transcript before the server closes the connection.
closing_ws = True
try:
await ws.send_bytes(wav_header)

async for data in self._input_ch:
if isinstance(data, rtc.AudioFrame):
for frame in audio_bstream.write(data.data.tobytes()):
await ws.send_bytes(frame.data.tobytes())
elif isinstance(data, self._FlushSentinel):
for frame in audio_bstream.flush():
await ws.send_bytes(frame.data.tobytes())

for frame in audio_bstream.flush():
await ws.send_bytes(frame.data.tobytes())

# Don't close the WS here — let recv_task read the final
# transcript before the server closes the connection.
closing_ws = True
except (aiohttp.ClientError, ConnectionError) as e:
if closing_ws:
return
raise APIConnectionError("Telnyx STT connection closed unexpectedly") from e

@utils.log_exceptions(logger=logger)
async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
Expand Down
33 changes: 19 additions & 14 deletions livekit-plugins/livekit-plugins-xai/livekit/plugins/xai/stt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,20 +291,25 @@ async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
samples_per_channel=samples_50ms,
)

async for data in self._input_ch:
frames: list[rtc.AudioFrame] = []
if isinstance(data, rtc.AudioFrame):
frames.extend(audio_bstream.write(data.data.tobytes()))
elif isinstance(data, self._FlushSentinel):
frames.extend(audio_bstream.flush())

for frame in frames:
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())

self._audio_duration_collector.flush()
await ws.send_str(json.dumps({"type": "audio.done"}))
closing_ws = True
try:
async for data in self._input_ch:
frames: list[rtc.AudioFrame] = []
if isinstance(data, rtc.AudioFrame):
frames.extend(audio_bstream.write(data.data.tobytes()))
elif isinstance(data, self._FlushSentinel):
frames.extend(audio_bstream.flush())

for frame in frames:
self._audio_duration_collector.push(frame.duration)
await ws.send_bytes(frame.data.tobytes())

self._audio_duration_collector.flush()
closing_ws = True
await ws.send_str(json.dumps({"type": "audio.done"}))
except (aiohttp.ClientError, ConnectionError) as e:
if closing_ws or self._session.closed:
return
raise APIConnectionError("xAI connection closed unexpectedly") from e

@utils.log_exceptions(logger=logger)
async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
Expand Down
Loading