fix(openai realtime): treat create_response=False as client-side turn taking

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -1882,14 +1882,15 @@ def _on_input_speech_started(self, _: llm.InputSpeechStartedEvent) -> None:
                     user_speaking_span=self._session._user_speaking_span,
                 )
 
-        # self.interrupt() is going to raise when allow_interruptions is False, llm.InputSpeechStartedEvent is only fired by the server when the turn_detection is enabled.  # noqa: E501
-        # When using the server-side turn_detection, we don't allow allow_interruptions to be False.
         try:
             self.interrupt()  # input_speech_started is also interrupting on the serverside realtime session  # noqa: E501
         except RuntimeError:
-            logger.exception(
-                "RealtimeAPI input_speech_started, but current speech is not interruptable, this should never happen!"  # noqa: E501
-            )
+            # only out of sync when the server cancelled its own response, with client-side turn
+            # taking an uninterruptible speech is expected
+            if self._rt_turn_detection_enabled:
+                logger.exception(
+                    "RealtimeAPI input_speech_started, but current speech is not interruptable, this should never happen!"  # noqa: E501
+                )
 
     def _on_input_speech_stopped(self, ev: llm.InputSpeechStoppedEvent) -> None:
         if self.vad is None or self.using_default_vad:

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

@@ -453,10 +453,23 @@ def __init__(
             )
 
         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"
+            )
+
         super().__init__(
             capabilities=llm.RealtimeCapabilities(
                 message_truncation=True,
-                turn_detection=turn_detection is not None,
+                # 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,
                 can_disable_turn_detection=not is_given(turn_detection),
                 user_transcription=input_audio_transcription is not None,
                 auto_tool_reply_generation=False,
@@ -501,7 +514,7 @@ def __init__(
             modalities=modalities,
             input_audio_transcription=to_audio_transcription(input_audio_transcription),
             input_audio_noise_reduction=to_noise_reduction(input_audio_noise_reduction),
-            turn_detection=to_turn_detection(turn_detection),
+            turn_detection=resolved_turn_detection,
             api_key=api_key,
             base_url=base_url_val,
             is_azure=is_azure,
@@ -2246,6 +2259,12 @@ def _handle_error(self, event: RealtimeErrorEvent) -> None:
         if event.error.message.startswith("Cancellation failed"):
             return
 
+        if event.error.code == "input_audio_buffer_commit_empty" and (
+            self._opts.turn_detection is not None
+        ):
+            # the server VAD commits each segment itself, ours lands on an emptied buffer
+            return
+
         provider_label = self._realtime_model._provider_label
         logger.error(
             f"{provider_label} returned an error: {event.error}",

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

@@ -115,6 +115,8 @@ def to_turn_detection(
                 kwargs["silence_duration_ms"] = turn_detection.silence_duration_ms
             if turn_detection.create_response is not None:
                 kwargs["create_response"] = turn_detection.create_response
+            if turn_detection.interrupt_response is not None:
+                kwargs["interrupt_response"] = turn_detection.interrupt_response
             return realtime.realtime_audio_input_turn_detection.ServerVad(**kwargs)
         elif turn_detection.type == "semantic_vad":
             kwargs["type"] = "semantic_vad"

tests/test_realtime/test_openai_realtime_model.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+import logging
 from types import SimpleNamespace
 from typing import cast
 
 import pytest
+from openai.types.beta.realtime.session import TurnDetection as BetaTurnDetection
+from openai.types.realtime.realtime_audio_input_turn_detection import ServerVad
 
 from livekit.agents import llm
 from livekit.agents._exceptions import APIError
@@ -61,6 +64,62 @@ def test_with_azure_preserves_can_disable_turn_detection() -> None:
     assert explicit_off.capabilities.can_disable_turn_detection is False
 
 
+def test_create_response_false_reports_client_side_turn_taking() -> None:
+    # server VAD with create_response=False commits and transcribes the audio server-side but
+    # leaves the reply to the client, so it must not count as server-side turn detection —
+    # otherwise allow_interruptions=False is rejected (issue #6635)
+    manual_reply = RealtimeModel(
+        api_key="fake",
+        turn_detection=ServerVad(type="server_vad", create_response=False),
+    )
+    assert manual_reply.capabilities.turn_detection is False
+
+    auto_reply = RealtimeModel(
+        api_key="fake",
+        turn_detection=ServerVad(type="server_vad"),
+    )
+    assert auto_reply.capabilities.turn_detection is True
+
+    assert RealtimeModel(api_key="fake").capabilities.turn_detection is True
+
+
+def test_create_response_false_warns_when_the_server_still_interrupts(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # interrupt_response is a separate switch, left on the server keeps cancelling its response
+    # on user speech while the client believes it owns interruptions
+    with caplog.at_level(logging.WARNING, logger="livekit.plugins.openai"):
+        RealtimeModel(
+            api_key="fake",
+            turn_detection=ServerVad(type="server_vad", create_response=False),
+        )
+    assert "pass interrupt_response=False" in caplog.text
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING, logger="livekit.plugins.openai"):
+        RealtimeModel(
+            api_key="fake",
+            turn_detection=ServerVad(
+                type="server_vad", create_response=False, interrupt_response=False
+            ),
+        )
+    assert caplog.text == ""
+
+
+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
+    model = RealtimeModel(
+        api_key="fake",
+        turn_detection=BetaTurnDetection(
+            type="server_vad", create_response=False, interrupt_response=False
+        ),
+    )
+    assert model._opts.turn_detection == ServerVad(
+        type="server_vad", create_response=False, interrupt_response=False
+    )
+
+
 def test_update_chat_ctx_deletes_empty_remote_items() -> None:
     remote_ctx = RemoteChatContext()
     audio_item = llm.ChatMessage(id="audio_item", role="user", content=[])
@@ -96,11 +155,14 @@ def test_is_fatal_error_matches_known_codes() -> None:
     assert not _is_fatal_error(None)
 
 
-def _handle_error_session(capture: dict[str, object]) -> RealtimeSession:
+def _handle_error_session(
+    capture: dict[str, object], *, turn_detection: ServerVad | None = None
+) -> RealtimeSession:
     return cast(
         RealtimeSession,
         SimpleNamespace(
             _realtime_model=SimpleNamespace(_provider_label="openai"),
+            _opts=SimpleNamespace(turn_detection=turn_detection),
             _emit_error=lambda error, recoverable: capture.update(recoverable=recoverable),
         ),
     )
@@ -135,6 +197,30 @@ def test_handle_error_ignores_cancellation_failed() -> None:
     assert captured == {}  # early return, nothing emitted
 
 
+def _empty_commit_event() -> SimpleNamespace:
+    return SimpleNamespace(
+        error=SimpleNamespace(
+            message="Error committing input audio buffer: buffer too small.",
+            code="input_audio_buffer_commit_empty",
+        )
+    )
+
+
+def test_handle_error_ignores_empty_commit_with_server_vad() -> None:
+    # our commit raced the server VAD's own; the server owns the buffer, nothing to report
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured, turn_detection=ServerVad(type="server_vad"))
+    RealtimeSession._handle_error(session, _empty_commit_event())
+    assert captured == {}
+
+
+def test_handle_error_reports_empty_commit_without_server_vad() -> None:
+    # nothing else commits the buffer, so an empty commit is a real bug worth surfacing
+    captured: dict[str, object] = {}
+    RealtimeSession._handle_error(_handle_error_session(captured), _empty_commit_event())
+    assert captured["recoverable"] is True
+
+
 def test_response_done_failed_fatal_raises() -> None:
     captured: dict[str, object] = {}
     session = _handle_error_session(captured)

tests/test_realtime_input_speech_interrupt.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import logging
+from unittest.mock import MagicMock
+
+import pytest
+
+from livekit.agents import Agent, AgentSession, TurnHandlingOptions, llm
+from livekit.agents.voice.agent_activity import AgentActivity
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+from .fake_realtime import FakeRealtimeModel, fake_capabilities
+from .fake_vad import FakeVAD
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture(autouse=True)
+def _livekit_keys(monkeypatch: pytest.MonkeyPatch) -> None:
+    # the eager TurnDetector() default reads these; keep construction hermetic
+    monkeypatch.setenv("LIVEKIT_API_KEY", "k")
+    monkeypatch.setenv("LIVEKIT_API_SECRET", "s")
+
+
+def _activity(*, server_turn_detection: bool, allow_interruptions: bool = True) -> AgentActivity:
+    session = AgentSession(
+        llm=FakeRealtimeModel(
+            capabilities=fake_capabilities(
+                turn_detection=server_turn_detection, can_disable_turn_detection=False
+            )
+        ),
+        vad=FakeVAD(fake_user_speeches=[]),
+        turn_handling=TurnHandlingOptions(interruption={"enabled": allow_interruptions}),
+    )
+    return AgentActivity(Agent(instructions="test"), session)
+
+
+def _speech_started(activity: AgentActivity, *, allow_interruptions: bool) -> SpeechHandle:
+    handle = SpeechHandle.create(allow_interruptions=allow_interruptions)
+    activity._current_speech = handle
+    activity._rt_session = MagicMock()
+    activity._on_input_speech_started(llm.InputSpeechStartedEvent())
+    return handle
+
+
+def test_allow_interruptions_false_rejected_with_server_turn_detection() -> None:
+    # the server cancels its own response on user speech, so the local speech can't opt out
+    with pytest.raises(ValueError, match="allow_interruptions cannot be False"):
+        _activity(server_turn_detection=True, allow_interruptions=False)
+
+
+def test_allow_interruptions_false_allowed_with_client_turn_taking() -> None:
+    # server VAD with create_response=False reports no server-side turn detection, so a
+    # turn-taking agent can disable interruptions (issue #6635)
+    activity = _activity(server_turn_detection=False, allow_interruptions=False)
+
+    assert activity._rt_turn_detection_enabled is False
+
+
+def test_input_speech_started_keeps_uninterruptible_speech(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # the model still reports user speech (server VAD is on to commit and transcribe audio), but
+    # neither side interrupts: no response.cancel, no local interruption, and no error log since
+    # this is a valid configuration
+    activity = _activity(server_turn_detection=False, allow_interruptions=False)
+
+    with caplog.at_level(logging.ERROR, logger="livekit.agents"):
+        handle = _speech_started(activity, allow_interruptions=False)
+
+    assert handle.interrupted is False
+    activity._rt_session.interrupt.assert_not_called()
+    assert not caplog.records
+
+
+def test_input_speech_started_interrupts_interruptible_speech() -> None:
+    activity = _activity(server_turn_detection=True)
+
+    handle = _speech_started(activity, allow_interruptions=True)
+
+    assert handle.interrupted is True
+    activity._rt_session.interrupt.assert_called_once()