fix: flush supported stt streams on vad end of speech

livekit-agents/livekit/agents/stt/fallback_adapter.py

@@ -95,6 +95,7 @@ def __init__(
                 aligned_transcript=aligned_transcript,
                 keyterms=any(t.capabilities.keyterms for t in stt),
                 chat_context=any(t.capabilities.chat_context for t in stt),
+                manual_flush=all(t.capabilities.manual_flush for t in stt),
             )
         )
 
@@ -130,6 +131,14 @@ def _next_instance(self) -> STT:
                 return instance
         return self._stt_instances[0]
 
+    @property
+    def capabilities(self) -> STTCapabilities:
+        # update manual_flush as instance might be updated
+        self._capabilities.manual_flush = all(
+            instance.capabilities.manual_flush for instance in self._stt_instances
+        )
+        return self._capabilities
+
     @property
     def model(self) -> str:
         """The model of the instance that serves next (see :meth:`_next_instance`). Spans and

livekit-agents/livekit/agents/stt/stt.py

@@ -140,6 +140,12 @@ class STTCapabilities:
     """Whether the STT supports keyterm prompting"""
     chat_context: bool = False
     """Whether the STT can natively consume conversation context (see STT._push_conversation_item)"""
+    manual_flush: bool = False
+    """Whether the STT supports flushing the stream to finalize a segment.
+
+    When enabled, VAD-based turn detection automatically flushes on VAD end-of-speech.
+    Disable this for custom STT nodes that buffer audio and manage flushing themselves.
+    """
 
 
 class STTError(BaseModel):

livekit-agents/livekit/agents/voice/agent.py

@@ -388,6 +388,11 @@ def stt_node(
         You can override this node with your own implementation for more flexibility (e.g.,
         custom pre-processing of audio, additional buffering, or alternative STT strategies).
 
+        If your override buffers audio passed to Agent.default.stt_node, automatic VAD
+        flushing can occur before that audio reaches STT. Set
+        stt.capabilities.manual_flush = False on the underlying STT providers and manage
+        flushing in your implementation.
+
         Args:
             audio (AsyncIterable[rtc.AudioFrame]): An asynchronous stream of audio frames.
             model_settings (ModelSettings): Configuration and parameters for model execution.
@@ -506,6 +511,8 @@ async def stt_node(
             agent: Agent, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings
         ) -> AsyncGenerator[stt.SpeechEvent, None]:
             """Default implementation for `Agent.stt_node`"""
+            from .audio_recognition import _STTPipelineContextVar
+
             activity = agent._get_activity_or_raise()
             assert activity.stt is not None, "stt_node called but no STT node is available"
 
@@ -540,6 +547,14 @@ async def stt_node(
                     )
                     stream.start_time_offset = time.time() - _audio_input_started_at
 
+                    def _flush() -> None:
+                        if wrapped_stt.capabilities.manual_flush:
+                            stream.flush()
+
+                    pipeline = _STTPipelineContextVar.get(None)
+                    if pipeline:
+                        pipeline._flush_callback = _flush
+
                     @utils.log_exceptions(logger=logger)
                     async def _forward_input() -> None:
                         async for frame in audio:
@@ -550,6 +565,8 @@ async def _forward_input() -> None:
                         async for event in stream:
                             yield event
                     finally:
+                        if pipeline:
+                            pipeline._flush_callback = None
                         await utils.aio.cancel_and_wait(forward_task)
             finally:
                 if temporary_adapter is not None:

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -2,11 +2,12 @@
 
 import asyncio
 import contextlib
+import contextvars
 import json
 import math
 import time
 from collections import deque
-from collections.abc import AsyncIterable, Callable, Iterator
+from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterator
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, Any, Literal, Protocol
 
@@ -160,6 +161,9 @@ def on_user_turn_exceeded(self, ev: UserTurnExceededEvent) -> None: ...
     def retrieve_chat_ctx(self) -> llm.ChatContext: ...
 
 
+_STTPipelineContextVar = contextvars.ContextVar["_STTPipeline"]("stt_pipeline")
+
+
 class _STTPipeline:
     """Transferable STT pipeline that survives agent handoff.
 
@@ -173,17 +177,30 @@ def __init__(
         self._stt_node = stt_node
         # don't recreate the stream while the session is closing
         self._is_closing = is_closing or (lambda: False)
-        self._audio_ch = aio.Chan[rtc.AudioFrame]()
+        self._audio_ch = aio.Chan[rtc.AudioFrame | stt.RecognizeStream._FlushSentinel]()
+        self._flush_callback: Callable[[], None] | None = None
         self._event_ch = aio.Chan[stt.SpeechEvent]()
         self._pump_task = asyncio.create_task(self._stt_pump())
         self._pump_task.add_done_callback(lambda _: self._event_ch.close())
         # wall-clock anchor for stream-based (STT and barge-in) timestamps
         self.input_started_at: float | None = None
 
     @property
-    def audio_ch(self) -> aio.Chan[rtc.AudioFrame]:
+    def audio_ch(self) -> aio.Chan[rtc.AudioFrame | stt.RecognizeStream._FlushSentinel]:
         return self._audio_ch
 
+    def flush(self) -> None:
+        self._audio_ch.send_nowait(stt.RecognizeStream._FlushSentinel())
+
+    async def _audio_input(self) -> AsyncIterator[rtc.AudioFrame]:
+        """Apply flushes in audio order without exposing sentinels to custom STT nodes."""
+        async for data in self._audio_ch:
+            if isinstance(data, stt.RecognizeStream._FlushSentinel):
+                if self._flush_callback:
+                    self._flush_callback()
+            else:
+                yield data
+
     @property
     def event_ch(self) -> aio.Chan[stt.SpeechEvent]:
         return self._event_ch
@@ -198,9 +215,10 @@ async def _stt_pump(self) -> None:
         """
         from .agent import ModelSettings
 
+        _STTPipelineContextVar.set(self)
         while True:
             try:
-                node = self._stt_node(self._audio_ch, ModelSettings())
+                node = self._stt_node(self._audio_input(), ModelSettings())
                 if asyncio.iscoroutine(node):
                     node = await node
 
@@ -1439,6 +1457,8 @@ async def _on_vad_event(self, ev: vad.VADEvent) -> None:
 
             # A committed turn clears _vad_speech_started before its late VAD EOS arrives.
             if self._stt_pipeline is not None and vad_speech_started:
+                if self._vad_base_turn_detection:
+                    self._stt_pipeline.flush()
                 self._arm_transcription_timeout(
                     ev.speech_duration,
                     delay=ev.silence_duration + ev.inference_duration,

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

@@ -31,6 +31,7 @@
     DEFAULT_API_CONNECT_OPTIONS,
     APIConnectionError,
     APIConnectOptions,
+    APIError,
     APIStatusError,
     APITimeoutError,
     LanguageCode,
@@ -114,7 +115,7 @@ def __init__(
         tag_audio_events: bool = True,
         use_realtime: NotGivenOr[bool] = NOT_GIVEN,  # Deprecated
         sample_rate: STTRealtimeSampleRates = 16000,
-        server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
+        server_vad: NotGivenOr[VADOptions | None] = NOT_GIVEN,
         include_timestamps: bool = False,
         http_session: aiohttp.ClientSession | None = None,
         model: NotGivenOr[ElevenLabsSTTModels | str] = NOT_GIVEN,
@@ -146,7 +147,9 @@ def __init__(
             use_realtime (bool): Whether to use "scribe_v2_realtime" model for streaming mode. Default is NOT_GIVEN.
                 Note that this flag is deprecated in favour of explicitly specifying the model id.
             sample_rate (STTRealtimeSampleRates): Audio sample rate in Hz. Default is 16000.
-            server_vad (NotGivenOr[VADOptions]): Server-side VAD options, only supported for Scribe v2 realtime model.
+            server_vad (NotGivenOr[VADOptions | None]): Server-side VAD options, only supported for Scribe v2 realtime model.
+                At construction, omit or set to None to use manual commits. Pass {} to enable server VAD defaults.
+                In update_options(), omission keeps the current setting; None disables server VAD and restores manual commits.
             http_session (aiohttp.ClientSession | None): Custom HTTP session for API requests. Optional.
             model (ElevenLabsSTTModels | str): ElevenLabs STT model to use. If not specified a default model will
                 be selected based on parameters provided.
@@ -217,6 +220,7 @@ def __init__(
                 streaming=use_realtime,
                 interim_results=True,
                 aligned_transcript="word" if include_timestamps and use_realtime else False,
+                manual_flush=use_realtime and (not is_given(server_vad) or server_vad is None),
             )
         )
 
@@ -368,16 +372,24 @@ def update_options(
         self,
         *,
         tag_audio_events: NotGivenOr[bool] = NOT_GIVEN,
-        server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
+        server_vad: NotGivenOr[VADOptions | None] = NOT_GIVEN,
         keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
         secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN,
         no_verbatim: NotGivenOr[bool] = NOT_GIVEN,
     ) -> None:
+        """Update STT options. Omitted options keep their current values.
+
+        Set server_vad to None to disable server VAD and use manual commits.
+        Pass {} to enable server VAD with defaults, or a VADOptions dict to configure it.
+        Changing server_vad reconnects active streams and can discard unfinished transcripts.
+        Wait for the current turn's final transcript before changing the commit strategy.
+        """
         if is_given(tag_audio_events):
             self._opts.tag_audio_events = tag_audio_events
 
         if is_given(server_vad):
             self._opts.server_vad = server_vad
+            self._capabilities.manual_flush = self._capabilities.streaming and server_vad is None
 
         if is_given(keyterms):
             self._opts.keyterms = keyterms
@@ -449,11 +461,18 @@ def __init__(
     def update_options(
         self,
         *,
-        server_vad: NotGivenOr[VADOptions] = NOT_GIVEN,
+        server_vad: NotGivenOr[VADOptions | None] = NOT_GIVEN,
         no_verbatim: NotGivenOr[bool] = NOT_GIVEN,
         keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
         secondary_languages: NotGivenOr[list[str]] = NOT_GIVEN,
     ) -> None:
+        """Update stream options. Omitted options keep their current values.
+
+        Set server_vad to None to disable server VAD and use manual commits.
+        Pass {} to enable server VAD with defaults, or a VADOptions dict to configure it.
+        Changes reconnect the stream and can discard unfinished transcripts.
+        Wait for the current turn's final transcript before changing these options.
+        """
         if is_given(server_vad):
             self._opts.server_vad = server_vad
             self._reconnect_event.set()
@@ -610,6 +629,8 @@ async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
                 try:
                     parsed = json.loads(msg.data)
                     self._process_stream_event(parsed)
+                except APIError:
+                    raise
                 except Exception:
                     logger.exception("failed to process ElevenLabs STT message")
 
@@ -830,16 +851,15 @@ def _process_stream_event(self, data: dict) -> None:
             "input_error",
             "error",
         ):
-            error_msg = data.get("message", "Unknown error")
-            error_details = data.get("details", "")
-            details_suffix = " - " + error_details if error_details else ""
             logger.error(
-                "ElevenLabs STT error [%s]: %s%s",
+                "ElevenLabs STT error [%s]",
                 message_type,
-                error_msg,
-                details_suffix,
+                extra={"lk.pii.data": data},
             )
-            raise APIConnectionError(f"{message_type}: {error_msg}{details_suffix}")
+            raise APIConnectionError(
+                f"ElevenLabs STT error [{message_type}]",
+                retryable=message_type not in ("auth_error", "quota_exceeded", "input_error"),
+            ) from None
         else:
             logger.warning(
                 "ElevenLabs STT unknown message type: %s",