Skip to content

fix(google): finalize the realtime generation once per message so a combined turn_complete + tool_call isn't lost #8376

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
himanshumehta wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(google): finalize the realtime generation once per message so a combined turn_complete + tool_call isn't lost #8376

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
884212f
fix(google): finalize the Gemini realtime generation once in the recv…
himanshumehta Nov 4, 2026
776dbb6
fix(google): record tool calls in chat_ctx after the turn they end
himanshumehta Nov 4, 2026
b1f2501
refactor(google): return llm.FunctionCall from _handle_tool_calls
himanshumehta Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 84 additions & 84 deletions livekit-plugins/livekit-plugins-google/livekit/plugins/google/realtime/realtime_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1283,6 +1283,82 @@ async def _send_task(self, session: AsyncSession) -> None:
finally:
logger.debug("send task finished.")

def _handle_live_server_message(self, response: types.LiveServerMessage) -> None:
if lk_google_debug:
Comment on lines +1286 to +1287

@devin-ai-integration devin-ai-integration Bot Nov 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 New message handler lacks required documentation

_handle_live_server_message has no docstring, despite the repository's requirement to document every new method.

Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

@himanshumehta himanshumehta Nov 4, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm leaving this one as it is. AGENTS.md asks for Google-style docstrings where a docstring is written, and the private per-message handlers in this file don't have one: _handle_server_content, _handle_tool_calls, _handle_usage_metadata and _handle_go_away. _handle_live_server_message is the same kind of method, so it follows that convention. Its behaviour is pinned by test_combined_turn_complete_and_tool_call_finalizes_once, test_lone_tool_call_finalizes_generation and test_tool_call_is_recorded_after_the_turn_it_ends.

resp_copy = response.model_dump(exclude_defaults=True)
# remove audio from debugging logs
if (
(sc := resp_copy.get("server_content"))
and (mt := sc.get("model_turn"))
and (parts := mt.get("parts"))
):
for part in parts:
if part and part.get("inline_data"):
part["inline_data"] = "<audio>"
logger.debug("<<< received response", extra={"lk.pii.response": resp_copy})

if response.tool_call and self._opts.tool_choice == "none":
# reject without opening a generation, so the pending generate_reply
# stays bound to the model's eventual reply and tools stay suppressed
# for the whole turn.
self._reject_tool_calls(response.tool_call.function_calls or [])
return

if not self._current_generation or self._current_generation._done:
if (sc := response.server_content) and sc.interrupted:
# two cases an interrupted event is sent without an active generation
# 1) the generation is done but playout is not finished (turn_complete -> interrupted)
# 2) the generation is not started (interrupted -> turn_complete)
# for both cases, we interrupt the agent if there is no pending generation from `generate_reply`
# for the second case, the pending generation will be stopped by `turn_complete` event coming later
if not self._pending_generation_fut:
self._handle_input_speech_started()

sc.interrupted = None
sc_copy = sc.model_dump(exclude_none=True)
if not sc_copy:
# ignore empty server content
response.server_content = None
if lk_google_debug:
logger.debug("ignoring empty server content")

if self._is_new_generation(response):
self._start_new_generation()
if lk_google_debug:
logger.debug(f"new generation started: {self._current_generation}")

if response.session_resumption_update:
if (
response.session_resumption_update.resumable
and response.session_resumption_update.new_handle
):
self._session_resumption_handle = response.session_resumption_update.new_handle
self._resumption_chat_ctx = llm.ChatContext(
[item for item in self._chat_ctx.items if item.id not in self._unsent_item_ids]
)

calls: list[llm.FunctionCall] = []
if response.server_content:
self._handle_server_content(response.server_content)
if response.tool_call:
calls = self._handle_tool_calls(response.tool_call)

@longcw longcw Nov 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: return the list of llm.FunctionCall instead of tuple of str?

@himanshumehta himanshumehta Nov 5, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in b1f2501. _handle_tool_calls now returns list[llm.FunctionCall].

One thing came up while doing it. ChatContext.insert orders items by created_at, and a FunctionCall built inside _handle_tool_calls is created before _mark_current_generation_done adds the assistant message. Inserting it as-is put the call ahead of the message, and test_tool_call_is_recorded_after_the_turn_it_ends went red:

FAILED tests/test_plugin_google_realtime.py::test_tool_call_is_recorded_after_the_turn_it_ends
1 failed, 77 passed

So the loop inserts each call with the insertion time:

for call in calls:
    self._chat_ctx.insert(call.model_copy(update={"created_at": time.time()}))

The call sent on function_ch is still a separate instance, as on main.

Tests:

  • New test_handle_tool_calls_returns_the_function_calls. It failed on 776dbb6 (assert all(isinstance(call, llm.FunctionCall) ...)) and passes now.
  • uv run pytest tests/test_plugin_google_realtime.py tests/test_realtime_*.py: 255 passed.
  • ruff check, ruff format --check and mypy -p livekit.plugins.google: clean.

if ((sc := response.server_content) and sc.turn_complete) or response.tool_call:
self._mark_current_generation_done()
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved

# mirror the server history, after the turn's messages added above, so the
# tool results the framework appends to `chat_ctx` keep their calls
for call in calls:
self._chat_ctx.insert(call.model_copy(update={"created_at": time.time()}))
if response.tool_call_cancellation:
self._handle_tool_call_cancellation(response.tool_call_cancellation)
if response.usage_metadata:
self._handle_usage_metadata(response.usage_metadata)
if response.go_away:
self._handle_go_away(response.go_away)

if self._num_retries > 0:
self._num_retries = 0 # reset the retry counter

async def _recv_task(self, session: AsyncSession) -> None:
try:
while True:
Expand All @@ -1294,78 +1370,7 @@ async def _recv_task(self, session: AsyncSession) -> None:
break

async for response in session.receive():
if lk_google_debug:
resp_copy = response.model_dump(exclude_defaults=True)
# remove audio from debugging logs
if (
(sc := resp_copy.get("server_content"))
and (mt := sc.get("model_turn"))
and (parts := mt.get("parts"))
):
for part in parts:
if part and part.get("inline_data"):
part["inline_data"] = "<audio>"
logger.debug("<<< received response", extra={"lk.pii.response": resp_copy})

if response.tool_call and self._opts.tool_choice == "none":
# reject without opening a generation, so the pending generate_reply
# stays bound to the model's eventual reply and tools stay suppressed
# for the whole turn.
self._reject_tool_calls(response.tool_call.function_calls or [])
continue

if not self._current_generation or self._current_generation._done:
if (sc := response.server_content) and sc.interrupted:
# two cases an interrupted event is sent without an active generation
# 1) the generation is done but playout is not finished (turn_complete -> interrupted)
# 2) the generation is not started (interrupted -> turn_complete)
# for both cases, we interrupt the agent if there is no pending generation from `generate_reply`
# for the second case, the pending generation will be stopped by `turn_complete` event coming later
if not self._pending_generation_fut:
self._handle_input_speech_started()

sc.interrupted = None
sc_copy = sc.model_dump(exclude_none=True)
if not sc_copy:
# ignore empty server content
response.server_content = None
if lk_google_debug:
logger.debug("ignoring empty server content")

if self._is_new_generation(response):
self._start_new_generation()
if lk_google_debug:
logger.debug(f"new generation started: {self._current_generation}")

if response.session_resumption_update:
if (
response.session_resumption_update.resumable
and response.session_resumption_update.new_handle
):
self._session_resumption_handle = (
response.session_resumption_update.new_handle
)
self._resumption_chat_ctx = llm.ChatContext(
[
item
for item in self._chat_ctx.items
if item.id not in self._unsent_item_ids
]
)

if response.server_content:
self._handle_server_content(response.server_content)
if response.tool_call:
self._handle_tool_calls(response.tool_call)
if response.tool_call_cancellation:
self._handle_tool_call_cancellation(response.tool_call_cancellation)
if response.usage_metadata:
self._handle_usage_metadata(response.usage_metadata)
if response.go_away:
self._handle_go_away(response.go_away)

if self._num_retries > 0:
self._num_retries = 0 # reset the retry counter
self._handle_live_server_message(response)

# TODO(dz): a server-side turn is complete
except Exception as e:
Expand Down Expand Up @@ -1577,9 +1582,6 @@ def _handle_server_content(self, server_content: types.LiveServerContent) -> Non
# 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 _mark_current_generation_done(self) -> None:
if not self._current_generation or self._current_generation._done:
return
Expand Down Expand Up @@ -1678,21 +1680,23 @@ def _reject_tool_calls(self, function_calls: list[types.FunctionCall]) -> None:
]
self._send_client_event(types.LiveClientToolResponse(function_responses=responses))

def _handle_tool_calls(self, tool_call: types.LiveServerToolCall) -> None:
def _handle_tool_calls(self, tool_call: types.LiveServerToolCall) -> list[llm.FunctionCall]:
if not self._current_generation:
logger.warning("received tool call but no active generation.")
return
return []

gen = self._current_generation
calls: list[tuple[str, str, str]] = []
calls: list[llm.FunctionCall] = []
for fnc_call in tool_call.function_calls or []:
arguments = json.dumps(fnc_call.args)
call_id = fnc_call.id
if not call_id:
call_id = utils.shortuuid("fnc-call-")
self._synthetic_call_ids.add(call_id)
self._open_call_ids.add(call_id)
calls.append((call_id, fnc_call.name or "", arguments))
calls.append(
llm.FunctionCall(call_id=call_id, name=fnc_call.name or "", arguments=arguments)
)

gen.function_ch.send_nowait(
llm.FunctionCall(
Expand All @@ -1701,12 +1705,8 @@ def _handle_tool_calls(self, tool_call: types.LiveServerToolCall) -> None:
arguments=arguments,
)
)
self._mark_current_generation_done()

# mirror the server history, after the turn's messages added above, so the
# tool results the framework appends to `chat_ctx` keep their calls
for call_id, name, arguments in calls:
self._chat_ctx.insert(llm.FunctionCall(call_id=call_id, name=name, arguments=arguments))
return calls

def _handle_tool_call_cancellation(
self, tool_call_cancellation: types.LiveServerToolCallCancellation
Expand Down
Loading
Loading