fix(google): finalize realtime generation when turn_complete is missing

Author: 3DRX

livekit-plugins/livekit-plugins-google/livekit/plugins/google/realtime/realtime_api.py

@@ -34,6 +34,7 @@
 INPUT_AUDIO_CHANNELS = 1
 OUTPUT_AUDIO_SAMPLE_RATE = 24000
 OUTPUT_AUDIO_CHANNELS = 1
+TURN_COMPLETE_FALLBACK_SECONDS = 1.0
 
 DEFAULT_IMAGE_ENCODE_OPTIONS = images.EncodeOptions(
     format="JPEG",
@@ -463,6 +464,7 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
         self._session_should_close = asyncio.Event()
         self._response_created_futures: dict[str, asyncio.Future[llm.GenerationCreatedEvent]] = {}
         self._pending_generation_fut: asyncio.Future[llm.GenerationCreatedEvent] | None = None
+        self._turn_complete_fallback_task: asyncio.Task[None] | None = None
 
         self._session_resumption_handle: str | None = (
             self._opts.session_resumption.handle
@@ -728,6 +730,7 @@ def truncate(
         pass
 
     async def aclose(self) -> None:
+        self._cancel_turn_complete_fallback()
         self._msg_ch.close()
         self._session_should_close.set()
 
@@ -1037,6 +1040,8 @@ def _build_connect_config(self) -> types.LiveConnectConfig:
         return conf
 
     def _start_new_generation(self) -> None:
+        self._cancel_turn_complete_fallback()
+
         if self._current_generation and not self._current_generation._done:
             logger.warning("starting new generation while another is active. Finalizing previous.")
             self._mark_current_generation_done()
@@ -1140,14 +1145,50 @@ def _handle_server_content(self, server_content: types.LiveServerContent) -> Non
         if server_content.generation_complete or server_content.turn_complete:
             current_gen._completed_timestamp = time.time()
 
+        if server_content.generation_complete and not server_content.turn_complete:
+            self._schedule_turn_complete_fallback(current_gen.response_id)
+
         if server_content.interrupted and not self._pending_generation_fut:
             # interrupt agent if there is no pending user initiated generation
             self._handle_input_speech_started()
 
         if server_content.turn_complete:
             self._mark_current_generation_done()
 
+    def _cancel_turn_complete_fallback(self) -> None:
+        if self._turn_complete_fallback_task and not self._turn_complete_fallback_task.done():
+            self._turn_complete_fallback_task.cancel()
+        self._turn_complete_fallback_task = None
+
+    def _schedule_turn_complete_fallback(self, response_id: str) -> None:
+        self._cancel_turn_complete_fallback()
+        self._turn_complete_fallback_task = asyncio.create_task(
+            self._wait_for_turn_complete_fallback(
+                response_id=response_id,
+                timeout=TURN_COMPLETE_FALLBACK_SECONDS,
+            ),
+            name=f"gemini_turn_complete_fallback_{response_id}",
+        )
+
+    async def _wait_for_turn_complete_fallback(self, *, response_id: str, timeout: float) -> None:
+        try:
+            await asyncio.sleep(timeout)
+        except asyncio.CancelledError:
+            return
+
+        current_gen = self._current_generation
+        if not current_gen or current_gen._done or current_gen.response_id != response_id:
+            return
+
+        logger.warning(
+            "Gemini Realtime did not emit turn_complete after generation_complete; "
+            f"finalizing generation (response_id={response_id}, timeout={timeout:.2f}s)"
+        )
+        self._mark_current_generation_done()
+
     def _mark_current_generation_done(self) -> None:
+        self._cancel_turn_complete_fallback()
+
         if not self._current_generation or self._current_generation._done:
             return
 

tests/test_plugin_google_realtime.py

@@ -0,0 +1,79 @@
+from types import SimpleNamespace
+
+from livekit import rtc
+from livekit.agents import llm, utils
+from livekit.plugins.google.realtime.realtime_api import (
+    RealtimeSession,
+    _ResponseGeneration,  # pyright: ignore[reportPrivateUsage]
+)
+
+
+def _make_session() -> RealtimeSession:
+    session = RealtimeSession.__new__(RealtimeSession)
+    session._current_generation = None
+    session._turn_complete_fallback_task = None
+    session._opts = SimpleNamespace(output_audio_transcription=object())
+    session._chat_ctx = llm.ChatContext.empty()
+    session._pending_generation_fut = None
+    session.emit = lambda *args, **kwargs: None
+    return session
+
+
+def _make_generation(response_id: str = "GR_test") -> _ResponseGeneration:
+    return _ResponseGeneration(
+        message_ch=utils.aio.Chan[llm.MessageGeneration](),
+        function_ch=utils.aio.Chan[llm.FunctionCall](),
+        input_id="GI_test",
+        response_id=response_id,
+        text_ch=utils.aio.Chan[str](),
+        audio_ch=utils.aio.Chan[rtc.AudioFrame](),
+    )
+
+
+def _server_content(
+    *, generation_complete: bool = False, turn_complete: bool = False
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        model_turn=None,
+        input_transcription=None,
+        output_transcription=None,
+        generation_complete=generation_complete,
+        turn_complete=turn_complete,
+        interrupted=False,
+    )
+
+
+async def test_generation_complete_schedules_turn_complete_fallback() -> None:
+    sess = _make_session()
+    gen = _make_generation("GR_schedule")
+    sess._current_generation = gen
+
+    scheduled: list[str] = []
+    sess._schedule_turn_complete_fallback = scheduled.append
+
+    sess._handle_server_content(_server_content(generation_complete=True))
+
+    assert scheduled == [gen.response_id]
+    assert not gen._done
+
+
+async def test_turn_complete_still_finalizes_generation() -> None:
+    sess = _make_session()
+    gen = _make_generation("GR_turn")
+    sess._current_generation = gen
+
+    sess._handle_server_content(_server_content(turn_complete=True))
+
+    assert gen._done
+
+
+async def test_turn_complete_fallback_finalizes_generation() -> None:
+    sess = _make_session()
+    gen = _make_generation("GR_fallback")
+    sess._current_generation = gen
+
+    await sess._wait_for_turn_complete_fallback(response_id=gen.response_id, timeout=0.0)
+
+    assert gen._done
+    assert gen.audio_ch.closed
+    assert gen.message_ch.closed