fix(xai): self-heal the realtime chat context when an item's anchor is missing

livekit-plugins/livekit-plugins-xai/livekit/plugins/xai/realtime/realtime_model.py

@@ -180,12 +180,20 @@ def _handle_function_call(self, item: RealtimeConversationItemFunctionCall) -> N
         super()._handle_function_call(item)
 
     def _handle_conversion_item_added(self, event: ConversationItemAdded) -> None:
-        # xAI's `conversation.item.added` event always has the previous_item_id as None
-        # replace it with the last item in the remote chat context
-        if event.previous_item_id is None:
-            event.previous_item_id = (
-                self._remote_chat_ctx._tail.item.id if self._remote_chat_ctx._tail else None
-            )
+        # xAI's `conversation.item.added` event always has the previous_item_id as None;
+        # anchor it to the last item in the remote chat context.
+        #
+        # The event can also reference an item our local copy no longer has, after an
+        # interruption's truncation/delete traffic desyncs us from the server (the
+        # `conversation.item.deleted` event carries no item id, so the delete is guessed
+        # and can drop the wrong item). Anchoring to a missing item makes the base handler
+        # raise `previous_item_id not found` and drop this item, which strands every later
+        # item that chains off it and corrupts the whole context. Fall back to the current
+        # tail so a single desync self-heals instead of cascading.
+        prev_id = event.previous_item_id
+        if prev_id is None or self._remote_chat_ctx.get(prev_id) is None:
+            prev_id = self._remote_chat_ctx._tail.item.id if self._remote_chat_ctx._tail else None
+        event.previous_item_id = prev_id
 
         super()._handle_conversion_item_added(event)
 

tests/test_realtime/test_xai_realtime_model.py

@@ -1,7 +1,11 @@
 from __future__ import annotations
 
 import pytest
-from openai.types.realtime import ConversationItemInputAudioTranscriptionCompletedEvent
+from openai.types.realtime import (
+    ConversationItemAdded,
+    ConversationItemInputAudioTranscriptionCompletedEvent,
+    RealtimeConversationItemUserMessage,
+)
 
 from livekit.agents import llm
 from livekit.agents.llm.remote_chat_context import RemoteChatContext
@@ -15,12 +19,28 @@ def _make_session() -> tuple[RealtimeSession, list[llm.InputTranscriptionComplet
     session = RealtimeSession.__new__(RealtimeSession)
     session._remote_chat_ctx = RemoteChatContext()  # type: ignore[attr-defined]
     session._input_transcript_accumulators = {}  # type: ignore[attr-defined]
+    session._item_create_future = {}  # type: ignore[attr-defined]
 
     emitted: list[llm.InputTranscriptionCompleted] = []
     session.emit = lambda name, ev: emitted.append(ev)  # type: ignore[method-assign,assignment]
     return session, emitted
 
 
+def _item_added_event(*, item_id: str, previous_item_id: str | None) -> ConversationItemAdded:
+    item = RealtimeConversationItemUserMessage.construct(
+        id=item_id,
+        type="message",
+        role="user",
+        content=[{"type": "input_text", "text": item_id}],
+    )
+    return ConversationItemAdded.construct(
+        event_id="evt",
+        type="conversation.item.added",
+        item=item,
+        previous_item_id=previous_item_id,
+    )
+
+
 def _completed_event(
     *, item_id: str, transcript: str, status: str | None
 ) -> ConversationItemInputAudioTranscriptionCompletedEvent:
@@ -89,3 +109,32 @@ def test_missing_status_defaults_to_final() -> None:
 
     assert len(emitted) == 1
     assert emitted[0].is_final is True
+
+
+def test_item_added_with_missing_anchor_self_heals_to_tail() -> None:
+    # After an interruption's delete/truncate traffic desyncs the local chat context, a
+    # server `conversation.item.added` can reference an item we no longer have. Anchoring
+    # to a missing item made the base handler drop the item; instead we append at the tail.
+    session, _ = _make_session()
+    session._remote_chat_ctx.insert(None, llm.ChatMessage(id="a", role="user", content=["hi"]))
+    session._remote_chat_ctx.insert("a", llm.ChatMessage(id="b", role="user", content=["there"]))
+
+    session._handle_conversion_item_added(_item_added_event(item_id="c", previous_item_id="ghost"))
+
+    # The item is kept (not dropped) and lands at the tail.
+    assert session._remote_chat_ctx.get("c") is not None
+    assert session._remote_chat_ctx._tail.item.id == "c"
+
+
+def test_item_added_missing_anchor_does_not_cascade() -> None:
+    # The real damage was the cascade: once one item is dropped, every later item that
+    # anchors to it is dropped too. Self-healing keeps the chain intact.
+    session, _ = _make_session()
+    session._remote_chat_ctx.insert(None, llm.ChatMessage(id="a", role="user", content=["hi"]))
+
+    session._handle_conversion_item_added(_item_added_event(item_id="c", previous_item_id="ghost"))
+    # `d` anchors to `c`, which only exists because `c` self-healed above.
+    session._handle_conversion_item_added(_item_added_event(item_id="d", previous_item_id="c"))
+
+    assert session._remote_chat_ctx.get("c") is not None
+    assert session._remote_chat_ctx.get("d") is not None