feat: enable manual flushing in elevenlabs stt

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

@@ -31,6 +31,7 @@
     DEFAULT_API_CONNECT_OPTIONS,
     APIConnectionError,
     APIConnectOptions,
+    APIError,
     APIStatusError,
     APITimeoutError,
     LanguageCode,
@@ -116,7 +117,7 @@ def __init__(
         use_realtime: NotGivenOr[bool] = NOT_GIVEN,  # Deprecated
         sample_rate: STTRealtimeSampleRates = 16000,
         audio_chunk_duration_ms: int = 50,
-        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,
@@ -152,7 +153,9 @@ def __init__(
                 milliseconds. Must be a positive integer. Defaults to 50. Larger chunks reduce
                 message frequency but increase buffering latency. Flushes send any shorter
                 remaining chunk before committing. Only used for Scribe v2 realtime.
-            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.
@@ -230,6 +233,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),
             )
         )
 
@@ -393,16 +397,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
@@ -474,11 +486,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()
@@ -635,6 +654,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")
 
@@ -855,16 +876,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",

tests/test_plugin_elevenlabs_stt.py

@@ -6,6 +6,7 @@
 import base64
 import dataclasses
 import json
+import logging
 import time
 from collections.abc import Callable
 from types import SimpleNamespace
@@ -19,7 +20,13 @@
 from yarl import URL
 
 from livekit import rtc
-from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, APIStatusError, LanguageCode, stt
+from livekit.agents import (
+    DEFAULT_API_CONNECT_OPTIONS,
+    APIConnectionError,
+    APIStatusError,
+    LanguageCode,
+    stt,
+)
 from livekit.agents.types import NOT_GIVEN
 from livekit.plugins.elevenlabs import stt as elevenlabs_stt
 from livekit.plugins.elevenlabs._utils import trace_id_from_headers
@@ -551,13 +558,15 @@ class _FakeWS:
     def __init__(self) -> None:
         self.sent: list[dict] = []
         self._closed = asyncio.Event()
+        self.received: asyncio.Queue[dict] = asyncio.Queue()
 
     async def send_str(self, data: str) -> None:
         self.sent.append(json.loads(data))
 
     async def receive(self) -> Any:
-        await self._closed.wait()
-        raise AssertionError("the test should never let recv_task resume")
+        return SimpleNamespace(
+            type=aiohttp.WSMsgType.TEXT, data=json.dumps(await self.received.get())
+        )
 
     async def close(self) -> None:
         self._closed.set()
@@ -682,6 +691,81 @@ def test_committed_transcript_sets_confidence() -> None:
     assert final.alternatives[0].confidence > 0.9
 
 
+@pytest.mark.parametrize(
+    ("model", "server_vad", "manual_flush"),
+    [
+        ("scribe_v2_realtime", NOT_GIVEN, True),
+        ("scribe_v2_realtime", None, True),
+        ("scribe_v2_realtime", {}, False),
+        ("scribe_v2", NOT_GIVEN, False),
+    ],
+)
+def test_manual_flush_capability(model, server_vad, manual_flush) -> None:
+    instance = elevenlabs_stt.STT(api_key="test-key", model=model, server_vad=server_vad)
+    assert instance.capabilities.manual_flush is manual_flush
+
+
+def test_manual_flush_capability_tracks_server_vad_updates() -> None:
+    instance = elevenlabs_stt.STT(api_key="test-key", model="scribe_v2_realtime")
+    instance.update_options(server_vad={})
+    assert instance.capabilities.manual_flush is False
+    instance.update_options(server_vad=None)
+    assert instance.capabilities.manual_flush is True
+
+
+@pytest.mark.parametrize(
+    "message_type", ["auth_error", "quota_exceeded", "transcriber_error", "input_error", "error"]
+)
+def test_provider_error_content_stays_in_pii_attributes(
+    message_type: str, caplog: pytest.LogCaptureFixture
+) -> None:
+    payload = {
+        "message_type": message_type,
+        "message": "customer transcript",
+        "details": "provider credential",
+    }
+    stream = _new_stream()
+    with caplog.at_level(logging.ERROR), pytest.raises(APIConnectionError) as exc:
+        stream._process_stream_event(payload)
+
+    assert str(exc.value) == f"ElevenLabs STT error [{message_type}]"
+    assert exc.value.__cause__ is None
+    assert exc.value.__suppress_context__
+    assert len(caplog.records) == 1
+    record = caplog.records[0]
+    assert record.getMessage() == f"ElevenLabs STT error [{message_type}]"
+    assert record.__dict__["lk.pii.data"] == payload
+    assert record.exc_info is None
+
+
+@pytest.mark.parametrize(
+    ("message_type", "retryable"),
+    [
+        ("auth_error", False),
+        ("quota_exceeded", False),
+        ("input_error", False),
+        ("transcriber_error", True),
+        ("error", True),
+    ],
+)
+async def test_provider_error_reaches_stream_consumer(message_type: str, retryable: bool) -> None:
+    ws = _FakeWS()
+    stream = _live_stream(ws)
+    if retryable:
+        stream._conn_options = dataclasses.replace(DEFAULT_API_CONNECT_OPTIONS, max_retry=0)
+    try:
+        ws.received.put_nowait({"message_type": message_type, "message": "customer content"})
+        with pytest.raises(
+            APIConnectionError, match=rf"ElevenLabs STT error \[{message_type}\]"
+        ) as exc:
+            await asyncio.wait_for(stream.__anext__(), timeout=1)
+        assert exc.value.retryable is retryable
+        assert stream._num_retries == 0
+        assert ws._closed.is_set()
+    finally:
+        await stream.aclose()
+
+
 def test_audio_chunk_duration_defaults_to_50ms() -> None:
     assert _stt()._opts.audio_chunk_duration_ms == 50
 

tests/test_plugin_elevenlabs_stt_live.py

@@ -0,0 +1,101 @@
+"""Live ElevenLabs transcription with manual and automatic turn commits."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+import wave
+from pathlib import Path
+
+import aiohttp
+import pytest
+from dotenv import load_dotenv
+
+from livekit import rtc
+from livekit.agents import Agent, AgentSession
+from livekit.plugins import elevenlabs, silero
+
+pytestmark = pytest.mark.plugin("elevenlabs")
+
+
+class _FixedTurnDetector:
+    model = "fixed-test-prediction"
+    provider = "test"
+
+    def __init__(self, probability: float) -> None:
+        self.probability = probability
+
+    async def supports_language(self, language):
+        return True
+
+    async def unlikely_threshold(self, language):
+        return 0.5
+
+    async def predict_end_of_turn(self, chat_ctx, *, timeout=None):
+        return self.probability
+
+
+@pytest.mark.parametrize("turn_detection", ["vad", "manual", 0.1, 0.9])
+async def test_live_flush_preserves_each_turn(turn_detection) -> None:
+    load_dotenv(Path(__file__).parents[1] / ".env")
+    if not os.environ.get("ELEVEN_API_KEY"):
+        pytest.skip("ELEVEN_API_KEY is required for live ElevenLabs validation")
+
+    with wave.open(str(Path(__file__).parent / "test_realtime/weather_question.wav")) as wav:
+        sample_rate = wav.getframerate()
+        channels = wav.getnchannels()
+        audio = wav.readframes(wav.getnframes())
+
+    async with aiohttp.ClientSession() as http:
+        stt = elevenlabs.STT(model="scribe_v2_realtime", language_code="en", http_session=http)
+        session = AgentSession(
+            stt=stt,
+            vad=silero.VAD.load(),
+            turn_handling={
+                "turn_detection": _FixedTurnDetector(turn_detection)
+                if isinstance(turn_detection, float)
+                else turn_detection,
+            },
+            session_close_transcript_timeout=0,
+        )
+        finals: asyncio.Queue[str] = asyncio.Queue()
+
+        @session.on("user_input_transcribed")
+        def on_transcript(event):
+            if event.is_final:
+                finals.put_nowait(event.transcript)
+
+        async def feed(data: bytes) -> None:
+            deadline = time.monotonic()
+            chunk_bytes = sample_rate // 50 * channels * 2
+            for offset in range(0, len(data), chunk_bytes):
+                chunk = data[offset : offset + chunk_bytes]
+                assert session._activity is not None
+                session._activity.push_audio(
+                    rtc.AudioFrame(chunk, sample_rate, channels, len(chunk) // (2 * channels))
+                )
+                deadline += len(chunk) / (2 * channels * sample_rate)
+                await asyncio.sleep(max(0, deadline - time.monotonic()))
+
+        try:
+            await session.start(Agent(instructions="Transcription test."))
+            silence = bytes(sample_rate * channels * 2 * 2)
+            await feed(silence)
+            assert finals.empty()
+            for _ in range(2):
+                await feed(audio)
+                if turn_detection == "manual":
+                    committed = await session.commit_user_turn(
+                        transcript_timeout=10, skip_reply=True
+                    )
+                    assert "weather" in committed.lower()
+                else:
+                    await feed(silence)
+                transcript = await asyncio.wait_for(finals.get(), timeout=20)
+                assert "weather" in transcript.lower()
+                assert "paris" in transcript.lower()
+                assert finals.empty()
+        finally:
+            await session.aclose()
+            await stt.aclose()