fix(xai): match id-less delete acks to the oldest pending delete

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

@@ -424,8 +424,13 @@ def _handle_conversion_item_added(self, event: ConversationItemAdded) -> None:
         super()._handle_conversion_item_added(event)
 
     def _handle_conversion_item_deleted(self, event: ConversationItemDeletedEvent) -> None:
-        if event.item_id == "" and self._item_delete_future:
-            event.item_id = list(self._item_delete_future.keys())[0]
+        if event.item_id == "":
+            # xAI omits the id, so the ack answers the oldest delete still waiting; one the
+            # server already rejected by event id is settled and cannot be the one acked
+            for item_id, fut in self._item_delete_future.items():
+                if not fut.done():
+                    event.item_id = item_id
+                    break
         super()._handle_conversion_item_deleted(event)
 
     def _handle_conversion_item_input_audio_transcription_completed(

scripts/xai-realtime-repro/README.md

@@ -0,0 +1,107 @@
+# xAI realtime local repro harness
+
+Local probes for three xAI realtime situations that show up next to the
+id-less delete-ack hang fix. Run them from the monorepo root after
+`make install`.
+
+## Setup
+
+```bash
+make install
+export XAI_API_KEY=...          # required for live probes
+# optional:
+# export XAI_REALTIME_MODEL=grok-voice-latest
+# export XAI_RECYCLE_SECONDS=45
+```
+
+Live probes talk to `wss://api.x.ai/v1/realtime`. They do not need LiveKit
+room credentials for the default path.
+
+## Commands
+
+Entrypoint:
+
+```bash
+./scripts/xai-realtime-repro/run.sh <unit|ref-probe|recycle|context|all>
+```
+
+### 1. Mid-session hang / delete-ack (unit) + `$ref` probe
+
+Hermetic coverage for the id-less delete-ack fix:
+
+```bash
+./scripts/xai-realtime-repro/run.sh unit
+# same as:
+uv run pytest tests/test_realtime/test_xai_realtime_model.py --unit -q
+```
+
+Expect every test to pass (including
+`test_delete_ack_without_id_answers_the_oldest_pending_delete`).
+
+Live mid-session nested `$ref` tools update:
+
+```bash
+./scripts/xai-realtime-repro/run.sh ref-probe
+```
+
+Sends a tools `session.update` whose schema is named `NESTED_REF_RAW_SCHEMA`
+(nested `$ref` / `$defs`), then `response.create`.
+
+- **PASS:** tools update accepted and a response completes without a schema error
+- **FAIL:** connection error, timeout, or server `error` after the tools update
+
+### 2. Websocket recycle / long-lived session
+
+```bash
+# print construction notes only (no network)
+./scripts/xai-realtime-repro/run.sh recycle --dry-log
+
+# live demo with a short recycle window (default 45s)
+./scripts/xai-realtime-repro/run.sh recycle
+```
+
+The plugin default is `max_session_duration=None` (no recycle). OpenAI
+realtime defaults to 20 minutes. Callers can opt in today:
+
+```python
+from livekit.plugins.xai.realtime import RealtimeModel
+
+RealtimeModel()  # no recycle
+RealtimeModel(max_session_duration=20 * 60)  # production-like
+RealtimeModel(max_session_duration=45)  # local demo
+```
+
+What to watch on a live run:
+
+- log lines about reconnecting / reconnected
+- the `session_reconnected` event (this script treats that as **PASS**)
+
+### 3. Context loss across turns
+
+```bash
+./scripts/xai-realtime-repro/run.sh context
+```
+
+Seeds a secret code, runs a few filler turns, then asks for the code again.
+Only `XAI_API_KEY` is required. If `LIVEKIT_*` is set, the script notes that
+and still uses the websocket path.
+
+- **PASS:** the recall reply contains `BLUE-ORBIT-7`
+- **FAIL:** the code is missing from the recall reply, or the socket errors
+
+The probe also prints `WARNING` lines and a final count when conversation
+item events omit `item_id` / `previous_item_id` (or empty `item.id`).
+
+## All
+
+```bash
+./scripts/xai-realtime-repro/run.sh all
+```
+
+Runs `unit`, then the three live probes. Live steps fail fast if
+`XAI_API_KEY` is unset.
+
+## Notes
+
+- No API keys belong in this directory.
+- These scripts are for local diagnosis. They are not part of the pytest gate.

scripts/xai-realtime-repro/common.py

@@ -0,0 +1,118 @@
+"""Shared helpers for local xAI realtime repro probes."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import aiohttp
+
+XAI_REALTIME_URL = "wss://api.x.ai/v1/realtime"
+DEFAULT_MODEL = os.environ.get("XAI_REALTIME_MODEL", "grok-voice-latest")
+
+
+def require_xai_api_key() -> str:
+    key = os.environ.get("XAI_API_KEY")
+    if not key:
+        print("FAIL: set XAI_API_KEY in the environment", file=sys.stderr)
+        raise SystemExit(1)
+    return key
+
+
+def pass_fail(ok: bool, message: str) -> int:
+    label = "PASS" if ok else "FAIL"
+    print(f"{label}: {message}")
+    return 0 if ok else 1
+
+
+async def open_realtime_ws(
+    api_key: str,
+) -> tuple[aiohttp.ClientSession, aiohttp.ClientWebSocketResponse]:
+    session = aiohttp.ClientSession()
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "User-Agent": "LiveKit Agents xAI repro",
+    }
+    try:
+        ws = await session.ws_connect(XAI_REALTIME_URL, headers=headers)
+    except Exception:
+        await session.close()
+        raise
+    return session, ws
+
+
+async def send_event(ws: aiohttp.ClientWebSocketResponse, event: dict[str, Any]) -> None:
+    await ws.send_str(json.dumps(event))
+
+
+async def recv_json(
+    ws: aiohttp.ClientWebSocketResponse,
+    *,
+    timeout: float = 30.0,
+) -> dict[str, Any]:
+    msg = await asyncio.wait_for(ws.receive(), timeout=timeout)
+    if msg.type == aiohttp.WSMsgType.TEXT:
+        data = json.loads(msg.data)
+        if not isinstance(data, dict):
+            raise TypeError(f"expected JSON object, got {type(data).__name__}")
+        return data
+    if msg.type == aiohttp.WSMsgType.ERROR:
+        raise RuntimeError(f"websocket error: {ws.exception()}")
+    if msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED):
+        raise RuntimeError("websocket closed")
+    raise RuntimeError(f"unexpected websocket message type: {msg.type}")
+
+
+async def wait_for_event(
+    ws: aiohttp.ClientWebSocketResponse,
+    predicate: Callable[[dict[str, Any]], bool],
+    *,
+    timeout: float = 45.0,
+    on_event: Callable[[dict[str, Any]], None] | None = None,
+) -> dict[str, Any]:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while True:
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            raise TimeoutError("timed out waiting for matching event")
+        event = await recv_json(ws, timeout=remaining)
+        if on_event is not None:
+            on_event(event)
+        if predicate(event):
+            return event
+
+
+async def drain_until(
+    ws: aiohttp.ClientWebSocketResponse,
+    predicate: Callable[[dict[str, Any]], bool],
+    *,
+    timeout: float = 45.0,
+    on_event: Callable[[dict[str, Any]], None] | None = None,
+) -> list[dict[str, Any]]:
+    seen: list[dict[str, Any]] = []
+
+    def _capture(event: dict[str, Any]) -> None:
+        seen.append(event)
+        if on_event is not None:
+            on_event(event)
+
+    await wait_for_event(ws, predicate, timeout=timeout, on_event=_capture)
+    return seen
+
+
+def event_error_message(event: dict[str, Any]) -> str | None:
+    if event.get("type") != "error":
+        return None
+    err = event.get("error") or {}
+    if isinstance(err, dict):
+        return str(err.get("message") or err.get("code") or err)
+    return str(err)
+
+
+def run_async(main: Callable[[], Awaitable[int]]) -> None:
+    raise SystemExit(asyncio.run(main()))