fix(stt): stop per-call language leaking in openai/elevenlabs, forward deepgram options to open streams

livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/stt.py

@@ -367,6 +367,7 @@ def update_options(
             self._opts.endpointing_ms = endpointing_ms
         if is_given(enable_diarization):
             self._opts.enable_diarization = enable_diarization
+            self._capabilities.diarization = enable_diarization
         if is_given(filler_words):
             self._opts.filler_words = filler_words
         if is_given(keywords):
@@ -413,6 +414,7 @@ def update_options(
                 sample_rate=sample_rate,
                 no_delay=no_delay,
                 endpointing_ms=endpointing_ms,
+                enable_diarization=enable_diarization,
                 filler_words=filler_words,
                 keywords=keywords,
                 keyterm=keyterm,
@@ -421,6 +423,7 @@ def update_options(
                 numerals=numerals,
                 mip_opt_out=mip_opt_out,
                 vad_events=vad_events,
+                tags=tags,
                 endpoint_url=endpoint_url,
                 utterance_end_ms=utterance_end_ms,
                 dictation=dictation,

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

@@ -283,16 +283,17 @@ async def _recognize_impl(
         language: NotGivenOr[str] = NOT_GIVEN,
         conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
     ) -> stt.SpeechEvent:
-        if is_given(language):
-            self._opts.language_code = LanguageCode(language)
+        request_language = (
+            LanguageCode(language) if is_given(language) else self._opts.language_code
+        )
 
         wav_bytes = rtc.combine_audio_frames(buffer).to_wav_bytes()
         form = aiohttp.FormData()
         form.add_field("file", wav_bytes, filename="audio.wav", content_type="audio/x-wav")
         form.add_field("model_id", self._opts.model_id)
         form.add_field("tag_audio_events", str(self._opts.tag_audio_events).lower())
-        if self._opts.language_code:
-            form.add_field("language_code", self._opts.language_code)
+        if request_language:
+            form.add_field("language_code", request_language)
         if is_given(self._opts.keyterms):
             for keyterm in self._opts.keyterms:
                 form.add_field("keyterms", keyterm)
@@ -346,7 +347,7 @@ async def _recognize_impl(
         except Exception as e:
             raise APIConnectionError() from e
 
-        normalized_language = LanguageCode(language_code or self._opts.language_code or "")
+        normalized_language = LanguageCode(language_code or request_language or "")
         return self._transcription_to_speech_event(
             language_code=normalized_language,
             text=extracted_text,

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

@@ -616,40 +616,39 @@ async def _recognize_impl(
         language: NotGivenOr[str | list[str]] = NOT_GIVEN,
         conn_options: APIConnectOptions,
     ) -> stt.SpeechEvent:
+        opts = self._opts
         if is_given(language):
             languages = _as_languages(language)
-            _validate_context(self._opts.model, languages, self._opts.keywords)
-            self._opts.languages = languages
+            _validate_context(opts.model, languages, opts.keywords)
+            opts = dataclasses.replace(opts, languages=languages)
 
         try:
             data = rtc.combine_audio_frames(buffer).to_wav_bytes()
 
             format = "json"
-            if self._opts.model == "whisper-1":
+            if opts.model == "whisper-1":
                 # verbose_json returns language and other details, only supported for whisper-1
                 format = "verbose_json"
 
-            transcription = _transcription(self._opts)
+            transcription = _transcription(opts)
             resp = await self._client.audio.transcriptions.create(
                 file=(
                     "file.wav",
                     data,
                     "audio/wav",
                 ),
-                model=self._opts.model,  # type: ignore
+                model=opts.model,  # type: ignore
                 language=transcription.language or openai.omit,
                 languages=transcription.languages or openai.omit,
                 keywords=transcription.keywords or openai.omit,
                 prompt=transcription.prompt or openai.omit,
                 response_format=format,
-                temperature=self._opts.temperature
-                if is_given(self._opts.temperature)
-                else openai.omit,
+                temperature=opts.temperature if is_given(opts.temperature) else openai.omit,
                 timeout=httpx.Timeout(30, connect=conn_options.timeout),
             )
 
             # the detected language beats the hint, dominant code first; an empty list keeps it
-            sd = stt.SpeechData(text=resp.text, language=_transcript_language(self._opts.languages))
+            sd = stt.SpeechData(text=resp.text, language=_transcript_language(opts.languages))
             if isinstance(resp, TranscriptionVerbose) and resp.language:
                 sd.language = LanguageCode(resp.language)
             elif isinstance(resp, Transcription) and resp.languages:

tests/test_plugin_deepgram_stt.py

@@ -391,6 +391,61 @@ async def ws_connect(self, url: str, **kwargs):
         return self._ws
 
 
+class _URLRecordingSession:
+    """Hands out a fresh socket per connect and keeps every URL it was asked to open."""
+
+    def __init__(self) -> None:
+        self.closed = False
+        self.urls: list[str] = []
+
+    async def ws_connect(self, url: str, **kwargs):
+        self.urls.append(url)
+        ws = _LiveWS()
+        ws._response = SimpleNamespace(headers={})  # read for the dg- response headers
+        return ws
+
+
+async def test_stt_update_options_reaches_open_streams_diarization_and_tags():
+    from urllib.parse import parse_qs, urlparse
+
+    from livekit.plugins.deepgram import STT
+
+    session = _URLRecordingSession()
+    instance = STT(api_key="test-key", language="en-US", http_session=session)
+    stream = instance.stream()
+    try:
+        await _wait_until(lambda: len(session.urls) == 1)
+        assert "diarize" not in parse_qs(urlparse(session.urls[0]).query)
+
+        instance.update_options(enable_diarization=True, tags=["prod"])
+
+        await _wait_until(lambda: len(session.urls) == 2)
+        query = parse_qs(urlparse(session.urls[1]).query)
+        assert query["diarize"] == ["true"]
+        assert query["tag"] == ["prod"]
+    finally:
+        await stream.aclose()
+        await instance.aclose()
+
+
+async def test_update_options_keeps_diarization_capability_in_sync():
+    from livekit.agents.stt import MultiSpeakerAdapter
+    from livekit.plugins.deepgram import STT
+
+    instance = STT(api_key="test-key")
+    try:
+        assert not instance.capabilities.diarization
+
+        instance.update_options(enable_diarization=True)
+        assert instance.capabilities.diarization
+        MultiSpeakerAdapter(stt=instance)
+
+        instance.update_options(enable_diarization=False)
+        assert not instance.capabilities.diarization
+    finally:
+        await instance.aclose()
+
+
 class _DeadSocket:
     """A socket that has silently gone away: writes fail, the read side never notices.
 

tests/test_plugin_elevenlabs_stt.py

@@ -763,3 +763,29 @@ 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_recognize_language_override_applies_to_one_call() -> None:
+    # StreamAdapter passes its stream's language to every recognize() call, so an
+    # override must not leak into the STT's own configuration
+    sent_languages: list[str] = []
+
+    async def handler(request: web.Request) -> web.Response:
+        form = await request.post()
+        sent_languages.append(str(form.get("language_code")))
+        return web.json_response({"text": "hello", "words": []})
+
+    app = web.Application()
+    app.router.add_post("/speech-to-text", handler)
+    async with TestServer(app) as server, aiohttp.ClientSession() as session:
+        url = str(server.make_url(""))
+        batch_stt = elevenlabs_stt.STT(
+            api_key="test-key", base_url=url, language_code="en", http_session=session
+        )
+        overridden = await batch_stt.recognize([_frame(100)], language="fr")
+        default = await batch_stt.recognize([_frame(100)])
+
+    assert sent_languages == ["fr", "en"]
+    assert overridden.alternatives[0].language == "fr"
+    assert default.alternatives[0].language == "en"
+    assert batch_stt._opts.language_code == "en"

tests/test_plugin_openai_stt_context.py

@@ -811,3 +811,22 @@ async def test_file_transcription_earlier_model_keeps_singular_language() -> Non
 
     assert 'name="language"\r\n\r\nen' in body
     assert 'name="languages[]"' not in body
+
+
+async def test_file_transcription_language_override_applies_to_one_call() -> None:
+    # StreamAdapter passes its stream's language to every recognize() call, so an
+    # override must not leak into the STT's own configuration
+    captured: list[httpx.Request] = []
+    instance = stt.STT(model="whisper-1", client=_mock_client(captured), language="en")
+
+    overridden = await instance._recognize_impl(
+        _silence(), language="fr", conn_options=DEFAULT_API_CONNECT_OPTIONS
+    )
+    default = await instance._recognize_impl(_silence(), conn_options=DEFAULT_API_CONNECT_OPTIONS)
+
+    first, second = (request.read().decode("utf-8", errors="replace") for request in captured)
+    assert 'name="language"\r\n\r\nfr' in first
+    assert overridden.alternatives[0].language == "fr"
+    assert 'name="language"\r\n\r\nen' in second
+    assert default.alternatives[0].language == "en"
+    assert instance._opts.languages == ["en"]

tests/test_transcription_delay_anchor.py

@@ -163,10 +163,12 @@ async def test_usable_provider_timestamp_replaces_the_anchor_only_in_stt_mode(mo
 
 
 @both_modes
-async def test_skewed_provider_timestamp(mode: str) -> None:
+async def test_skewed_provider_timestamp(mode: str, monkeypatch: pytest.MonkeyPatch) -> None:
     # the provider's audio clock runs ahead of wall clock: end_time maps into the
     # future and the estimate clamps to `now`, zeroing the delay
     now = time.time()
+    # the clamp reads the clock again, so a slow runner would drift past the tolerance
+    monkeypatch.setattr(audio_recognition.time, "time", lambda: now)
     ar = _make_recognition(vad=MagicMock(), input_started_at=now - 10.0, mode=mode)
     vad_anchor = now - 0.6
     ar._last_speaking_time = vad_anchor