Skip to content

fix(google): skip context replay on resumed realtime sessions #6000

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

Merged
longcw merged 6 commits into from
Sep 9, 2026
Merged

fix(google): skip context replay on resumed realtime sessions #6000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b85fff7
fix(google): skip context replay on resumed realtime sessions
he-yufeng Jun 7, 2026
7ab8e4f
Merge branch 'main' into fix/google-realtime-resume-context
longcw Sep 9, 2026
9f75dfd
fix(google): sync a disconnected chat ctx update as a diff on resume
longcw Sep 9, 2026
8603f0f
fix(google): diff against the chat ctx the resumption handle stands for
longcw Sep 9, 2026
3eeeffc
fix(google): keep queued chat ctx items out of the handle snapshot
longcw Sep 9, 2026
659d49b
test(google): failed send with a queued update replays each item once
longcw Sep 9, 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
127 changes: 95 additions & 32 deletions livekit-plugins/livekit-plugins-google/livekit/plugins/google/realtime/realtime_api.py
View file
Open in desktop
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
from dataclasses import dataclass, field
from typing import Literal

from pydantic import Field

import google.auth.credentials
from google.auth._default_async import default_async
from google.genai import Client as GenAIClient, types
Expand Down Expand Up @@ -44,6 +46,19 @@

lk_google_debug = int(os.getenv("LK_GOOGLE_DEBUG", 0))


class _ChatCtxContent(types.LiveClientContent):
"""Client content built from chat ctx items; the ids let the send task mark them sent."""

item_ids: set[str] = Field(default_factory=set, exclude=True)


class _ChatCtxToolResponse(types.LiveClientToolResponse):
"""Tool responses built from chat ctx items; the ids let the send task mark them sent."""

item_ids: set[str] = Field(default_factory=set, exclude=True)


# stop rejecting tool calls after this many in a row to avoid a loop (tool_choice="none")
MAX_TOOL_CALL_REJECTIONS = 3

Expand Down Expand Up @@ -523,6 +538,12 @@ def __init__(self, realtime_model: RealtimeModel) -> None:
if is_given(self._opts.session_resumption)
else None
)
# chat ctx the handle stands for; None until the first handle arrives
self._resumption_chat_ctx: llm.ChatContext | None = None
# chat ctx received while no session is active, synced on the next connect
self._pending_chat_ctx: llm.ChatContext | None = 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()

self._in_user_activity = False
self._session_lock = asyncio.Lock()
Expand Down Expand Up @@ -655,10 +676,18 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
)
async with self._session_lock:
if not self._active_session:
self._chat_ctx = chat_ctx
self._pending_chat_ctx = chat_ctx
return

diff_ops = llm.utils.compute_chat_ctx_diff(self._chat_ctx, chat_ctx)
self._sync_chat_ctx(chat_ctx)

def _sync_chat_ctx(
self, chat_ctx: llm.ChatContext, *, known: llm.ChatContext | None = None
) -> None:
"""Queue the items missing from `known` and adopt `chat_ctx` as the known state."""
diff_ops = llm.utils.compute_chat_ctx_diff(
known if known is not None else self._chat_ctx, chat_ctx
)

if diff_ops.to_remove:
logger.warning("Gemini Live does not support removing messages")
Expand Down Expand Up @@ -694,17 +723,30 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
tool_response_scheduling=self._opts.tool_response_scheduling,
supports_silent_scheduling=supports_silent_scheduling,
)
turns: list[types.Content] = []
if self._realtime_model.capabilities.mutable_chat_context:
turns_dict, _ = append_ctx.copy(exclude_function_call=True).to_provider_format(
format="google", inject_dummy_user_message=False
)
turns = [types.Content.model_validate(turn) for turn in turns_dict]
if turns:
item_ids = {
item.id for item in append_ctx.items if item.type != "function_call_output"
}
self._unsent_item_ids |= item_ids
self._send_client_event(
types.LiveClientContent(turns=turns, turn_complete=False)
_ChatCtxContent(turns=turns, turn_complete=False, item_ids=item_ids)
)
if tool_results:
self._send_client_event(tool_results)
item_ids = {
item.id for item in append_ctx.items if item.type == "function_call_output"
}
self._unsent_item_ids |= item_ids
self._send_client_event(
_ChatCtxToolResponse(
function_responses=tool_results.function_responses, item_ids=item_ids
)
)

# since we don't have a view of the history on the server side, we'll assume
# the current state is accurate. this isn't perfect because removals aren't done.
Expand All @@ -720,7 +762,7 @@ async def update_tools(self, tools: list[llm.Tool]) -> None:

@property
def chat_ctx(self) -> llm.ChatContext:
return self._chat_ctx.copy()
return (self._pending_chat_ctx or self._chat_ctx).copy()

@property
def tools(self) -> llm.ToolContext:
Expand Down Expand Up @@ -920,34 +962,46 @@ async def _main_task(self) -> None:
async with self._session_lock:
self._active_session = session

# Check for system/developer messages in initial chat context
system_msg_count = sum(
1
for msg in self._chat_ctx.messages()
if msg.role in ("system", "developer")
)
if system_msg_count > 0:
logger.warning(
f"Gemini Realtime model '{self._opts.model}' does not support 'system' or "
f"'developer' roles in chat history. Dropping {system_msg_count} system "
f"message(s) from initial chat context during session initialization. "
f"Gemini Realtime only supports 'user' and 'model' roles. Use "
f"update_instructions() to set system-level context instead."
)

turns_dict, _ = self._chat_ctx.copy(
exclude_function_call=True,
exclude_handoff=True,
exclude_instructions=True,
exclude_empty_message=True,
exclude_config_update=True,
).to_provider_format(format="google", inject_dummy_user_message=False)
turns = [types.Content.model_validate(turn) for turn in turns_dict]
if turns:
await session.send_client_content(
turns=turns, # type: ignore
turn_complete=False,
pending_ctx, self._pending_chat_ctx = self._pending_chat_ctx, None
if self._session_resumption_handle is not None:
# the handle restores the conversation; send only what came after it
target = pending_ctx if pending_ctx is not None else self._chat_ctx
if self._resumption_chat_ctx is None:
self._chat_ctx = target
else:
self._sync_chat_ctx(target, known=self._resumption_chat_ctx)
else:
if pending_ctx is not None:
self._chat_ctx = pending_ctx

system_msg_count = sum(
1
for msg in self._chat_ctx.messages()
if msg.role in ("system", "developer")
)
if system_msg_count > 0:
logger.warning(
f"Gemini Realtime model '{self._opts.model}' does not support 'system' or "
f"'developer' roles in chat history. Dropping {system_msg_count} system "
f"message(s) from initial chat context during session initialization. "
f"Gemini Realtime only supports 'user' and 'model' roles. Use "
f"update_instructions() to set system-level context instead."
)

turns_dict, _ = self._chat_ctx.copy(
exclude_function_call=True,
exclude_handoff=True,
exclude_instructions=True,
exclude_empty_message=True,
exclude_config_update=True,
).to_provider_format(format="google", inject_dummy_user_message=False)
turns = [types.Content.model_validate(turn) for turn in turns_dict]
if turns:
await session.send_client_content(
turns=turns, # type: ignore
turn_complete=False,
)
self._unsent_item_ids.clear()

# queue up existing chat context
send_task = asyncio.create_task(
Expand Down Expand Up @@ -1068,6 +1122,8 @@ async def _send_task(self, session: AsyncSession) -> None:
else:
logger.warning(f"Warning: Received unhandled message type: {type(msg)}")

if isinstance(msg, _ChatCtxContent | _ChatCtxToolResponse):
self._unsent_item_ids -= msg.item_ids
if lk_google_debug and isinstance(
msg,
(
Expand Down Expand Up @@ -1154,6 +1210,13 @@ async def _recv_task(self, session: AsyncSession) -> None:
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
]
)
Comment on lines +1213 to +1219

@devin-ai-integration devin-ai-integration Bot Sep 9, 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.

🔴 Queued updates replay twice after failures

When one context send fails with later events queued, _resumption_chat_ctx excludes every unsent item. The reconnect requeues them while _mark_restart_needed preserves their original events. Gemini receives duplicate messages or tool results after recovery.

Prompt for agents 
Track failed in-flight context events separately from context events still waiting in `_msg_ch`. On an error restart, the reconnect diff must restore the failed event without duplicating queued events. One approach is to drain/reset tagged chat-context events before rebuilding the diff, while preserving the existing retry semantics for unrelated realtime input. Add a test with at least two queued chat-context updates where the first send raises and verify each message or tool result reaches the replacement socket exactly once.
Devin Review

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

@longcw longcw Sep 9, 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.

an error restart discards the queue as well, and the resume diff is the only thing that re-sends.


if response.server_content:
self._handle_server_content(response.server_content)
Expand Down
Loading