fix(google): don't start a generation from audio trailing a tool call

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

@@ -545,6 +545,11 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
         # ids of chat ctx items queued but not yet sent, so a handle does not claim them
         self._unsent_item_ids: set[str] = set()
 
+        # a tool call ends the turn, but the server can keep streaming audio that belongs
+        # to it. those frames must not open a generation for a turn that is already over.
+        # cleared by the events that end that turn or begin the next one.
+        self._turn_ended_by_tool_call = False
+
         self._in_user_activity = False
         self._session_lock = asyncio.Lock()
         self._num_retries = 0
@@ -1305,6 +1310,7 @@ def _build_connect_config(self) -> types.LiveConnectConfig:
 
     def _start_new_generation(self) -> None:
         self._rejected_tool_calls = 0
+        self._turn_ended_by_tool_call = False
         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()
@@ -1443,6 +1449,7 @@ def _handle_server_content(self, server_content: types.LiveServerContent) -> Non
             self._handle_input_speech_started()
 
         if server_content.turn_complete:
+            self._turn_ended_by_tool_call = False
             self._mark_current_generation_done()
 
     def _mark_current_generation_done(self) -> None:
@@ -1558,6 +1565,7 @@ def _handle_tool_calls(self, tool_call: types.LiveServerToolCall) -> None:
                     arguments=arguments,
                 )
             )
+        self._turn_ended_by_tool_call = True
         self._mark_current_generation_done()
 
     def _handle_tool_call_cancellation(
@@ -1688,7 +1696,8 @@ def _is_new_generation(self, resp: types.LiveServerMessage) -> bool:
             return True
 
         if (sc := resp.server_content) and (
-            sc.model_turn
+            # audio can trail a turn a tool call already ended; it belongs to that turn
+            (sc.model_turn and not self._turn_ended_by_tool_call)
             or (
                 sc.output_transcription and sc.output_transcription and sc.output_transcription.text
             )

tests/test_plugin_google_realtime.py

@@ -836,3 +836,75 @@ async def _connect(self: AsyncLive, **kwargs: object) -> AsyncIterator[_FakeLive
         assert session._unsent_item_ids == set()
     finally:
         await session.aclose()
+
+
+async def test_audio_trailing_a_tool_call_does_not_start_a_generation(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A tool call ends the turn, but the server keeps streaming audio that belongs to it.
+
+    Opening a generation for those frames interrupts the one still playing, so the reply
+    is committed truncated and the turn's usage lands on an empty generation (issue #7195).
+    """
+    async with _make_session(monkeypatch) as session:
+        generations: list[llm.GenerationCreatedEvent] = []
+        session.on("generation_created", generations.append)
+
+        session._start_new_generation()
+        session._handle_server_content(_audio_content())
+        session._handle_tool_calls(_tool_call())
+        assert len(generations) == 1
+
+        # the trailing frame, and the events that close the turn behind it
+        trailing = _audio_content()
+        assert session._is_new_generation(types.LiveServerMessage(server_content=trailing)) is False
+        session._handle_server_content(trailing)
+        session._handle_server_content(types.LiveServerContent(generation_complete=True))
+        session._handle_server_content(types.LiveServerContent(turn_complete=True))
+
+        assert len(generations) == 1, "the finished turn's audio must not open a new one"
+        assert await _drain_generation(generations[0]) == ("", 1, ["lookup"])
+
+
+async def test_the_next_real_turn_still_opens_a_generation(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The suppression lasts only until the turn actually ends."""
+    async with _make_session(monkeypatch) as session:
+        session._start_new_generation()
+        session._handle_tool_calls(_tool_call())
+        assert session._turn_ended_by_tool_call
+
+        session._handle_server_content(types.LiveServerContent(turn_complete=True))
+        assert not session._turn_ended_by_tool_call
+        assert session._is_new_generation(types.LiveServerMessage(server_content=_audio_content()))
+
+
+async def test_a_reply_requested_after_a_tool_call_ignores_the_finished_turns_audio(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The pending reply must not be resolved by audio left over from the tool call's turn.
+
+    `_start_new_generation` hands the pending `generate_reply` future whatever generation
+    it opens, so suppression has to outlive the call to `generate_reply` -- the request has
+    not even reached the socket yet.
+    """
+    async with _make_connected_session(monkeypatch) as session:
+        generations: list[llm.GenerationCreatedEvent] = []
+        session.on("generation_created", generations.append)
+
+        session._start_new_generation()
+        session._handle_tool_calls(_tool_call())
+        generations.clear()
+
+        fut = session.generate_reply()
+        assert session._turn_ended_by_tool_call, "the request is only queued, not sent"
+
+        # audio still arriving from the turn the tool call ended
+        session._handle_server_content(_audio_content())
+        assert not session._is_new_generation(
+            types.LiveServerMessage(server_content=_audio_content())
+        )
+        assert not generations, "no generation is opened for the finished turn"
+        assert not fut.done(), "the pending reply stays bound to the turn actually requested"
+        fut.cancel()