fix(openai realtime): keep derived capabilities in sync with update_options (#6943)

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py

@@ -299,6 +299,28 @@ def _is_fatal_error(error: object | None) -> bool:
     return isinstance(code, str) and code in _FATAL_ERROR_CODES
 
 
+def _server_turn_taking_enabled(
+    turn_detection: RealtimeAudioInputTurnDetection | None,
+) -> bool:
+    """Whether the server both detects turns and answers them."""
+    return turn_detection is not None and turn_detection.create_response is not False
+
+
+def _warn_on_half_disabled_turn_taking(
+    turn_detection: RealtimeAudioInputTurnDetection | None,
+) -> None:
+    """Warn when the caller hands turn taking to the client but leaves interruption on the server."""
+    if (
+        turn_detection is not None
+        and turn_detection.create_response is False
+        and turn_detection.interrupt_response is not False
+    ):
+        logger.warning(
+            "create_response=False hands turn taking to the client, but the server still "
+            "cancels its response on user speech, pass interrupt_response=False as well"
+        )
+
+
 class RealtimeModel(llm.RealtimeModel):
     @overload
     def __init__(
@@ -454,22 +476,11 @@ class RealtimeModel(llm.RealtimeModel):
 
         modalities = modalities if is_given(modalities) else ["text", "audio"]
         resolved_turn_detection = to_turn_detection(turn_detection)
-        if (
-            resolved_turn_detection is not None
-            and resolved_turn_detection.create_response is False
-            and resolved_turn_detection.interrupt_response is not False
-        ):
-            logger.warning(
-                "create_response=False hands turn taking to the client, but the server still "
-                "cancels its response on user speech, pass interrupt_response=False as well"
-            )
-
+        _warn_on_half_disabled_turn_taking(resolved_turn_detection)
         super().__init__(
             capabilities=llm.RealtimeCapabilities(
                 message_truncation=True,
-                # create_response=False leaves the reply to the client: client-side turn taking
-                turn_detection=resolved_turn_detection is not None
-                and resolved_turn_detection.create_response is not False,
+                turn_detection=_server_turn_taking_enabled(resolved_turn_detection),
                 can_disable_turn_detection=not is_given(turn_detection),
                 user_transcription=input_audio_transcription is not None,
                 auto_tool_reply_generation=False,
@@ -736,13 +747,20 @@ class RealtimeModel(llm.RealtimeModel):
             self._opts.voice = voice
 
         if is_given(turn_detection):
+            # a derived capability has to follow the option it is derived from
             self._opts.turn_detection = to_turn_detection(turn_detection)
+            self._capabilities.turn_detection = _server_turn_taking_enabled(
+                self._opts.turn_detection
+            )
+            # only the model warns: it re-runs the update on every session it owns
+            _warn_on_half_disabled_turn_taking(self._opts.turn_detection)
 
         if is_given(tool_choice):
             self._opts.tool_choice = tool_choice
 
         if is_given(input_audio_transcription):
             self._opts.input_audio_transcription = to_audio_transcription(input_audio_transcription)
+            self._capabilities.user_transcription = self._opts.input_audio_transcription is not None
 
         if is_given(input_audio_noise_reduction):
             self._opts.input_audio_noise_reduction = to_noise_reduction(input_audio_noise_reduction)
@@ -872,6 +890,13 @@ class RealtimeSession(
             realtime_model._opts,
             turn_detection=None if turn_detection_disabled else realtime_model._opts.turn_detection,
         )
+        # this session's own copy: turn detection can be off here and on for the model
+        self._capabilities = replace(
+            realtime_model.capabilities,
+            turn_detection=False
+            if turn_detection_disabled
+            else realtime_model.capabilities.turn_detection,
+        )
         self._tools = llm.ToolContext.empty()
         self._msg_ch = utils.aio.Chan[RealtimeClientEvent | dict[str, Any]]()
         self._input_resampler: rtc.AudioResampler | None = None
@@ -1318,6 +1343,10 @@ class RealtimeSession(
             event_id=utils.shortuuid("session_update_"), session=session
         )
 
+    @property
+    def capabilities(self) -> llm.RealtimeCapabilities:
+        return self._capabilities
+
     @property
     def chat_ctx(self) -> llm.ChatContext:
         return self._remote_chat_ctx.to_chat_ctx()
@@ -1394,12 +1423,14 @@ class RealtimeSession(
                 audio_input.turn_detection = turn_detection
                 has_audio_config = True
             self._opts.turn_detection = turn_detection
+            self._capabilities.turn_detection = _server_turn_taking_enabled(turn_detection)
 
         if is_given(input_audio_transcription):
             if self._opts.input_audio_transcription != input_audio_transcription:
                 audio_input.transcription = input_audio_transcription
                 has_audio_config = True
             self._opts.input_audio_transcription = input_audio_transcription
+            self._capabilities.user_transcription = input_audio_transcription is not None
 
         if is_given(input_audio_noise_reduction):
             input_audio_noise_reduction = to_noise_reduction(input_audio_noise_reduction)

tests/test_realtime/test_openai_realtime_model.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from dataclasses import replace
 from types import SimpleNamespace
 from typing import cast
 
@@ -12,6 +13,7 @@ from openai.types.realtime import (
     ConversationItemDeletedEvent,
     RealtimeErrorEvent,
 )
+from openai.types.realtime.audio_transcription import AudioTranscription
 from openai.types.realtime.realtime_audio_input_turn_detection import ServerVad
 
 from livekit.agents import llm
@@ -112,6 +114,75 @@ def test_create_response_false_warns_when_the_server_still_interrupts(
     assert caplog.text == ""
 
 
+def test_update_options_keeps_derived_capabilities_in_sync() -> None:
+    # a stale capability reports the server owning the turn after the caller handed turn
+    # taking to the client, and AgentActivity then rejects allow_interruptions=False
+    model = RealtimeModel(api_key="fake", turn_detection=ServerVad(type="server_vad"))
+    assert model.capabilities.turn_detection is True
+    assert model.capabilities.user_transcription is True
+
+    model.update_options(
+        turn_detection=ServerVad(
+            type="server_vad", create_response=False, interrupt_response=False
+        ),
+        input_audio_transcription=None,
+    )
+    assert model.capabilities.turn_detection is False
+    assert model.capabilities.user_transcription is False
+
+    model.update_options(
+        turn_detection=ServerVad(type="server_vad"),
+        input_audio_transcription=AudioTranscription(model="whisper-1"),
+    )
+    assert model.capabilities.turn_detection is True
+    assert model.capabilities.user_transcription is True
+
+
+def test_update_options_leaves_derived_capabilities_alone_when_unset() -> None:
+    # an unrelated update must not resync a model that opted out of server-side turn taking
+    model = RealtimeModel(
+        api_key="fake",
+        turn_detection=ServerVad(
+            type="server_vad", create_response=False, interrupt_response=False
+        ),
+    )
+    model.update_options(voice="marin")
+    assert model.capabilities.turn_detection is False
+
+
+def _capabilities_session(model: RealtimeModel) -> RealtimeSession:
+    # only the state update_options touches; a real session would open a websocket
+    return cast(
+        RealtimeSession,
+        SimpleNamespace(
+            _opts=replace(model._opts),
+            _capabilities=replace(model.capabilities),
+            send_event=lambda event: None,
+            _wrap_session_update=lambda event_id, session: session,
+        ),
+    )
+
+
+def test_session_update_options_keeps_capabilities_on_the_session() -> None:
+    # turn detection is per session: one session handing turn taking to the client must not
+    # change what the model, or any other session on it, reports
+    model = RealtimeModel(api_key="fake", turn_detection=ServerVad(type="server_vad"))
+    session = _capabilities_session(model)
+
+    RealtimeSession.update_options(
+        session,
+        turn_detection=ServerVad(
+            type="server_vad", create_response=False, interrupt_response=False
+        ),
+        input_audio_transcription=None,
+    )
+
+    assert session._capabilities.turn_detection is False
+    assert session._capabilities.user_transcription is False
+    assert model.capabilities.turn_detection is True
+    assert model.capabilities.user_transcription is True
+
+
 def test_legacy_turn_detection_keeps_interrupt_response() -> None:
     # the deprecated session.TurnDetection carries interrupt_response for server_vad too;
     # dropping it silently re-enabled server-side interruption