fix(google): fail generate_reply when the server ends the turn without a reply

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

@@ -1260,9 +1260,28 @@ def _start_new_generation(self) -> None:
 
         self.emit("generation_created", generation_event)
 
+    def _fail_pending_generation(self, reason: types.TurnCompleteReason | None) -> None:
+        """Fail a generate_reply the server ended without creating a generation."""
+        fut = self._pending_generation_fut
+        if fut is None or fut.done():
+            return
+
+        detail = reason.value if reason is not None else "no reason given"
+        logger.warning("Gemini ended the turn without generating a reply: %s", detail)
+        fut.set_exception(
+            llm.RealtimeError(f"the server ended the turn without generating a reply: {detail}")
+        )
+
     def _handle_server_content(self, server_content: types.LiveServerContent) -> None:
         current_gen = self._current_generation
         if not current_gen:
+            # a turn the server ends without generating anything (a malformed function
+            # call, a rejected response) never reaches _handle_generation_created, so a
+            # pending generate_reply would sit on its timeout with the outcome already
+            # known - fail it now with the reason the server gave
+            if server_content.turn_complete:
+                self._fail_pending_generation(server_content.turn_complete_reason)
+
             if self._rejected_tool_calls:
                 logger.debug(
                     "ignoring server content from a rejected tool call turn",

tests/test_plugin_google_realtime.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
 import pytest
 from google.genai import types
 
-from livekit.agents import utils
+from livekit.agents import llm, utils
 from livekit.plugins.google.realtime.realtime_api import RealtimeModel, RealtimeSession
 
 pytestmark = pytest.mark.unit
@@ -130,3 +131,56 @@ async def _spy() -> None:
         monkeypatch.setattr(session._client.aio, "aclose", _spy)
 
     assert closed
+
+
+async def test_aborted_turn_fails_generate_reply_at_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A turn the server rejects never creates a generation (#6708).
+
+    ``generate_reply``'s future is only ever resolved by ``generation_created``, so an
+    abort left the caller waiting out the full 5s timeout for an outcome the server had
+    already reported in ~250ms.
+    """
+    async with _make_session(monkeypatch) as session:
+        fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.Future()
+        session._pending_generation_fut = fut
+
+        session._handle_server_content(
+            types.LiveServerContent(
+                turn_complete=True,
+                turn_complete_reason=types.TurnCompleteReason.MALFORMED_FUNCTION_CALL,
+            )
+        )
+
+        assert fut.done(), "the caller is still waiting on a turn the server already ended"
+        with pytest.raises(llm.RealtimeError, match="MALFORMED_FUNCTION_CALL"):
+            fut.result()
+
+
+async def test_turn_without_a_reason_still_fails_the_caller(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async with _make_session(monkeypatch) as session:
+        fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.Future()
+        session._pending_generation_fut = fut
+
+        session._handle_server_content(types.LiveServerContent(turn_complete=True))
+
+        assert fut.done()
+        with pytest.raises(llm.RealtimeError):
+            fut.result()
+
+
+async def test_content_without_turn_complete_leaves_the_caller_waiting(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # only the end of a turn settles it; a stray frame must not fail a live request
+    async with _make_session(monkeypatch) as session:
+        fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.Future()
+        session._pending_generation_fut = fut
+
+        session._handle_server_content(types.LiveServerContent(interrupted=True))
+
+        assert not fut.done()
+        fut.cancel()