Skip to content

fix(realtime): sync remote items to local chat_ctx with placeholders to prevent in-flight deletion #5114

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 3 commits into from
Mar 18, 2026
Merged

fix(realtime): sync remote items to local chat_ctx with placeholders to prevent in-flight deletion #5114

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
c113557
fix(realtime): sync remote items to local chat ctx with placeholder t…
longcw Mar 16, 2026
6009a7b
add rt_session.off
longcw Mar 16, 2026
f83f176
use copy as the placeholder
longcw Mar 16, 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
2 changes: 2 additions & 0 deletions livekit-agents/livekit/agents/llm/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
RealtimeModelError,
RealtimeSession,
RealtimeSessionReconnectedEvent,
RemoteItemAddedEvent,
)
from .tool_context import (
FunctionTool,
Expand Down Expand Up @@ -106,6 +107,7 @@
"RealtimeSessionReconnectedEvent",
"RealtimeSessionRestoredEvent",
"LLMError",
"RemoteItemAddedEvent",
]

# Cleanup docs of unexported modules
Expand Down
10 changes: 10 additions & 0 deletions livekit-agents/livekit/agents/llm/chat_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -672,6 +672,16 @@ def find_insertion_index(self, *, created_at: float) -> int:

return 0

def _upsert_item(self, item: ChatItem, *, allow_type_mismatch: bool = False) -> None:
"""Update an item with the same ID if it exists, otherwise append it."""
idx = self.index_by_id(item.id)
if idx is not None:
if not allow_type_mismatch and item.type != self._items[idx].type:
raise ValueError(f"Item type mismatch: {item.type} != {self._items[idx].type}")
self._items[idx] = item
else:
self._items.append(item)

async def _summarize(
self,
llm_v: LLM,
Expand Down
9 changes: 8 additions & 1 deletion livekit-agents/livekit/agents/llm/realtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from livekit import rtc

from ..types import NOT_GIVEN, NotGivenOr
from .chat_context import ChatContext, FunctionCall
from .chat_context import ChatContext, ChatItem, FunctionCall
from .tool_context import Tool, ToolChoice, ToolContext


Expand Down Expand Up @@ -114,6 +114,7 @@ async def __aexit__(
"generation_created",
"session_reconnected",
"metrics_collected",
"remote_item_added",
"error",
]

Expand All @@ -136,6 +137,12 @@ class RealtimeSessionReconnectedEvent:
pass


@dataclass
class RemoteItemAddedEvent:
previous_item_id: str | None
item: ChatItem


class RealtimeSession(ABC, rtc.EventEmitter[EventTypes | TEvent], Generic[TEvent]):
def __init__(self, realtime_model: RealtimeModel) -> None:
super().__init__()
Expand Down
26 changes: 21 additions & 5 deletions livekit-agents/livekit/agents/voice/agent_activity.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -564,6 +564,7 @@ async def _list_mcp_tools_task(
self._on_input_audio_transcription_completed,
)
self._rt_session.on("metrics_collected", self._on_metrics_collected)
self._rt_session.on("remote_item_added", self._on_remote_item_added)
self._rt_session.on("error", self._on_error)

remove_instructions(self._agent._chat_ctx)
Expand Down Expand Up @@ -736,6 +737,7 @@ async def _close_session(self) -> None:
self._on_input_audio_transcription_completed,
)
self._rt_session.off("metrics_collected", self._on_metrics_collected)
self._rt_session.off("remote_item_added", self._on_remote_item_added)
self._rt_session.off("error", self._on_error)

if isinstance(self.stt, stt.STT):
Expand Down Expand Up @@ -1161,6 +1163,20 @@ def _on_metrics_collected(
trace_utils.record_realtime_metrics(realtime_span, ev)
self._session.emit("metrics_collected", MetricsCollectedEvent(metrics=ev))

def _on_remote_item_added(self, ev: llm.RemoteItemAddedEvent) -> None:
# add the remote item to the local chat context as a placeholder
local_chat_ctx = self._agent._chat_ctx
if local_chat_ctx.get_by_id(ev.item.id) is not None:
return

# only add placeholders for server-initiated items (responses, function calls),
# which always append at the end of the conversation. client-initiated items
# (from update_chat_ctx) already exist in _agent._chat_ctx and go local→remote,
# so they don't need placeholders.
last_item_id = local_chat_ctx.items[-1].id if local_chat_ctx.items else None
if ev.previous_item_id is None or ev.previous_item_id == last_item_id:
Copy link
Copy Markdown
Member

@chenghao-mou chenghao-mou Mar 16, 2026

Choose a reason for hiding this comment

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

QQ: would parallel tool calls have the same previous_item_id? I wonder if we should check for ev.previous_item_id in item_ids instead.

Copy link
Copy Markdown
Contributor

@StianHanssen StianHanssen Mar 16, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

gpt-realtime can do "parallel" tool calls by doing multiple tool calls concurrently on one turn, but they have a sequential order on arrival from what I understand.

When I saw this line, I have to admit it stood out as a risk, but I can't think of a realistic scenario that causes an issue.

I thought we could be on the safe side, and log a warning in the else condition, just to catch if there is anything we have missed.

Copy link
Copy Markdown
Contributor Author

@longcw longcw Mar 16, 2026

Choose a reason for hiding this comment

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

yeah, I think the function calls will come in sequential with different previous item id, each ConversationItemAdded event carries a ConversationItem (either a message or a function call) and a previous_item_id.

class ConversationItemAdded(BaseModel):
    """Sent by the server when an Item is added to the default Conversation.

    This can happen in several cases:
    - When the client sends a `conversation.item.create` event.
    - When the input audio buffer is committed. In this case the item will be a user message containing the audio from the buffer.
    - When the model is generating a Response. In this case the `conversation.item.added` event will be sent when the model starts generating a specific Item, and thus it will not yet have any content (and `status` will be `in_progress`).

    The event will include the full content of the Item (except when model is generating a Response) except for audio data, which can be retrieved separately with a `conversation.item.retrieve` event if necessary.
    """

    event_id: str
    """The unique ID of the server event."""

    item: ConversationItem
    """A single item within a Realtime conversation."""

    type: Literal["conversation.item.added"]
    """The event type, must be `conversation.item.added`."""

    previous_item_id: Optional[str] = None
    """The ID of the item that precedes this one, if any.

    This is used to maintain ordering when items are inserted.
    """



ConversationItem: TypeAlias = Annotated[
    Union[
        RealtimeConversationItemSystemMessage,
        RealtimeConversationItemUserMessage,
        RealtimeConversationItemAssistantMessage,
        RealtimeConversationItemFunctionCall,
        RealtimeConversationItemFunctionCallOutput,
        RealtimeMcpApprovalResponse,
        RealtimeMcpListTools,
        RealtimeMcpToolCall,
        RealtimeMcpApprovalRequest,
    ],
    PropertyInfo(discriminator="type"),
]

local_chat_ctx.items.append(ev.item.model_copy())
Comment on lines +1166 to +1178
Copy link
Copy Markdown
Contributor

@devin-ai-integration devin-ai-integration bot Mar 16, 2026

Choose a reason for hiding this comment

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

🔴 _on_remote_item_added leaves stale empty placeholder messages in local chat context

When the server creates an assistant message item, conversation.item.created fires immediately (before any content is streamed), so _on_remote_item_added appends a ChatMessage(role="assistant", content=[]) placeholder to the local context. Later, at agent_activity.py:2702, the placeholder is only replaced if msg_gen and forwarded_text is truthy. If the response is interrupted before any text is forwarded, or the response produces no text (e.g., audio-only without transcript), forwarded_text is "" and _upsert_item is never called—leaving a permanent empty assistant message in _agent._chat_ctx. Before this PR, no message was added in that case. This empty message could confuse downstream consumers of the chat context (e.g., summarization, provider format conversion, or the next LLM turn).

Prompt for agents 
In livekit-agents/livekit/agents/voice/agent_activity.py, the _on_remote_item_added method (lines 1166-1178) unconditionally appends the placeholder item. However, in _realtime_generation_task_impl around line 2702, if msg_gen exists but forwarded_text is empty (interrupted or audio-only), the placeholder is never replaced or removed, leaving a stale empty assistant message in the local chat context.

To fix this, either:
1. In _realtime_generation_task_impl, after the interrupted handling and after the forwarded_text check, add cleanup logic: if msg_gen exists but forwarded_text is empty and the placeholder exists in _agent._chat_ctx, remove it (or update it with interrupted=True and empty content).
2. Or, change _on_remote_item_added to only add placeholders for item types that will always be replaced (e.g., only user messages and function calls, not assistant messages which are conditionally added).

Option 1 is likely cleaner. After line 2711, add an else branch like:
  elif msg_gen:
      # Remove stale placeholder if no text was forwarded
      idx = self._agent._chat_ctx.index_by_id(msg_gen.message_id)
      if idx is not None:
          del self._agent._chat_ctx.items[idx]
Open in Devin Review

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

Copy link
Copy Markdown
Contributor Author

@longcw longcw Mar 16, 2026

Choose a reason for hiding this comment

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

The server still has that item, the next server-initiated item would have previous_item_id pointing to this message, which wouldn't match last_item_id in _on_remote_item_added, so the next placeholder would be silently dropped.

so maybe we should allow empty item for realtime session.

Copy link
Copy Markdown
Member

@chenghao-mou chenghao-mou Mar 16, 2026

Choose a reason for hiding this comment

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

I think empty message is fine? If it is interrupted w/o any text forwarded or is an audio-only message, the remote message should be "empty" as well, right?


def _on_error(
self, error: llm.LLMError | stt.STTError | tts.TTSError | llm.RealtimeModelError
) -> None:
Expand Down Expand Up @@ -1210,7 +1226,7 @@ def _on_input_audio_transcription_completed(self, ev: llm.InputTranscriptionComp
# TODO: for realtime models, the created_at field is off. it should be set to when the user started speaking.
# but we don't have that information here.
msg = llm.ChatMessage(role="user", content=[ev.transcript], id=ev.item_id)
self._agent._chat_ctx.items.append(msg)
self._agent._chat_ctx._upsert_item(msg)
self._session._conversation_item_added(msg)

def _on_generation_created(self, ev: llm.GenerationCreatedEvent) -> None:
Expand Down Expand Up @@ -2328,7 +2344,7 @@ async def _realtime_reply_task(
chat_ctx = self._rt_session.chat_ctx.copy()
msg = chat_ctx.add_message(role="user", content=user_input)
await self._rt_session.update_chat_ctx(chat_ctx)
self._agent._chat_ctx.items.append(msg)
self._agent._chat_ctx._upsert_item(msg)
self._session._conversation_item_added(msg)

ori_tool_choice = self._tool_choice
Expand Down Expand Up @@ -2587,7 +2603,7 @@ async def _read_fnc_stream() -> None:

def _tool_execution_started_cb(fnc_call: llm.FunctionCall) -> None:
speech_handle._item_added([fnc_call])
self._agent._chat_ctx.items.append(fnc_call)
self._agent._chat_ctx._upsert_item(fnc_call)
self._session._tool_items_added([fnc_call])

def _tool_execution_completed_cb(out: ToolExecutionOutput) -> None:
Expand Down Expand Up @@ -2689,7 +2705,7 @@ def _create_assistant_message(
forwarded_text=forwarded_text,
interrupted=speech_handle.interrupted,
)
self._agent._chat_ctx.items.append(msg)
self._agent._chat_ctx._upsert_item(msg)
speech_handle._item_added([msg])
self._session._conversation_item_added(msg)
current_span.set_attribute(trace_types.ATTR_RESPONSE_TEXT, forwarded_text)
Expand Down Expand Up @@ -2741,7 +2757,7 @@ def _create_assistant_message(
fnc_executed_ev._reply_required = True

# add tool output to the chat context
self._agent._chat_ctx.items.append(sanitized_out.fnc_call_out)
self._agent._chat_ctx._upsert_item(sanitized_out.fnc_call_out)
self._session._tool_items_added([sanitized_out.fnc_call_out])

if new_agent_task is not None and sanitized_out.agent_task is not None:
Expand Down
7 changes: 5 additions & 2 deletions livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1491,8 +1491,11 @@ def _handle_conversion_item_added(self, event: ConversationItemAdded) -> None:
assert event.item.id is not None, "item.id is None"

try:
self._remote_chat_ctx.insert(
event.previous_item_id, openai_item_to_livekit_item(event.item)
lk_item = openai_item_to_livekit_item(event.item)
self._remote_chat_ctx.insert(event.previous_item_id, lk_item)
self.emit(
"remote_item_added",
llm.RemoteItemAddedEvent(previous_item_id=event.previous_item_id, item=lk_item),
)
except ValueError as e:
logger.warning(
Expand Down
7 changes: 5 additions & 2 deletions ...kit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model_beta.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1304,8 +1304,11 @@ def _handle_conversion_item_created(self, event: ConversationItemCreatedEvent) -
assert event.item.id is not None, "item.id is None"

try:
self._remote_chat_ctx.insert(
event.previous_item_id, _openai_item_to_livekit_item(event.item)
lk_item = _openai_item_to_livekit_item(event.item)
self._remote_chat_ctx.insert(event.previous_item_id, lk_item)
self.emit(
"remote_item_added",
llm.RemoteItemAddedEvent(previous_item_id=event.previous_item_id, item=lk_item),
)
except ValueError as e:
logger.warning(
Expand Down
Loading