fix(openai): pass response_id in ResponseCancelEvent on interrupt

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/inference_realtime_model.py

@@ -137,6 +137,9 @@ def __init__(
         self._opts.api_version = None
         if is_xai:
             self._capabilities.can_disable_turn_detection = can_disable_turn_detection
+            self._supports_targeted_cancellation = False
+        else:
+            self._supports_targeted_cancellation = True
         self._inference_opts = _InferenceOptions(
             provider=provider,
             api_key=resolved_api_key,

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

@@ -208,14 +208,17 @@ def _normalize_azure_client_event(event: dict[str, Any]) -> None:
     """In-place normalization of client event dicts for legacy Azure compatibility.
 
     The legacy Azure Realtime API uses "text" for assistant content parts,
-    while the newer OpenAI API uses "output_text".
+    while the newer OpenAI API uses "output_text". Additionally, the legacy
+    Azure beta response.cancel schema does not support response_id.
     """
     item = event.get("item")
-    if item is None:
-        return
-    for content_part in item.get("content", ()):
-        if content_part.get("type") == "output_text":
-            content_part["type"] = "text"
+    if item is not None:
+        for content_part in item.get("content", ()):
+            if content_part.get("type") == "output_text":
+                content_part["type"] = "text"
+
+    if event.get("type") == "response.cancel":
+        event.pop("response_id", None)
 
 
 @dataclass
@@ -264,6 +267,7 @@ class _ResponseGeneration:
     """timestamp when the response was created"""
     _first_token_timestamp: float | None = None
     """timestamp when the first token was received"""
+    response_id: str | None = None
 
     def _close(self) -> None:
         for msg in self.messages.values():
@@ -546,6 +550,7 @@ def __init__(
         self._http_session_owned = False
         self._sessions = weakref.WeakSet[RealtimeSession]()
         self._provider_label = "OpenAI Realtime API"
+        self._supports_targeted_cancellation: bool = not (is_azure and api_version is not None)
 
     @property
     def model(self) -> str:
@@ -1747,10 +1752,45 @@ def _on_fut_done(f: asyncio.Future[llm.GenerationCreatedEvent]) -> None:
     def has_active_generation(self) -> bool:
         return self._current_generation is not None or len(self._response_created_futures) > 0
 
+    @property
+    def _supports_targeted_cancellation(self) -> bool:
+        """Whether this session's provider supports response.cancel with response_id.
+
+        Only OpenAI Realtime API (including non-legacy Azure and LiveKit Inference
+        OpenAI routes) supports targeted cancellation with response_id. Legacy
+        Azure Realtime (with api_version) and subclasses such as xAI Realtime API
+        use the bare response.cancel schema.
+        """
+        if (
+            getattr(self._opts, "is_azure", False)
+            and getattr(self._opts, "api_version", None) is not None
+        ):
+            return False
+        if hasattr(self, "_xai_model"):
+            return False
+        model = getattr(self, "_realtime_model", None)
+        if model is not None:
+            if getattr(model, "_provider_label", None) == "xAI Realtime API":
+                return False
+            return getattr(model, "_supports_targeted_cancellation", True)
+        return True
+
     def interrupt(self) -> None:
         if not self.has_active_generation:
             return
-        self.send_event(ResponseCancelEvent(type="response.cancel"))
+        if (
+            isinstance(self._current_generation, _ResponseGeneration)
+            and self._current_generation.response_id
+            and self._supports_targeted_cancellation
+        ):
+            self.send_event(
+                ResponseCancelEvent(
+                    type="response.cancel",
+                    response_id=self._current_generation.response_id,
+                )
+            )
+        else:
+            self.send_event(ResponseCancelEvent(type="response.cancel"))
 
     def truncate(
         self,
@@ -1877,9 +1917,12 @@ def _handle_response_created(self, event: ResponseCreatedEvent) -> None:
             # interrupted or timed out before the server created it: cancel by id and mark it
             # discarded so its trailing events are skipped, instead of surfacing it
             self._discarded_event_ids.discard(client_event_id)
-            self.send_event(
-                ResponseCancelEvent(type="response.cancel", response_id=event.response.id)
-            )
+            if self._supports_targeted_cancellation:
+                self.send_event(
+                    ResponseCancelEvent(type="response.cancel", response_id=event.response.id)
+                )
+            else:
+                self.send_event(ResponseCancelEvent(type="response.cancel"))
             self._current_generation = _DiscardedGeneration()
             logger.warning("discarding response that arrived after it was timed out or interrupted")
             return
@@ -1890,6 +1933,7 @@ def _handle_response_created(self, event: ResponseCreatedEvent) -> None:
             messages={},
             _created_timestamp=time.time(),
             _done_fut=asyncio.Future(),
+            response_id=event.response.id,
         )
 
         generation_ev = llm.GenerationCreatedEvent(

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

@@ -126,6 +126,7 @@ def __init__(
         # xAI force_message drives scripted TTS without a follow-up response.create
         self._capabilities.supports_say = True
         self._provider_label = "xAI Realtime API"
+        self._supports_targeted_cancellation = False
 
     def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
         # manual turn-taking is unsupported (can_disable_turn_detection=False)

tests/test_realtime/test_openai_inference_realtime_model.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import inspect
 import json
 from types import SimpleNamespace
@@ -368,3 +369,77 @@ def test_xai_explicit_turn_detection_is_preserved() -> None:
     assert model._opts.turn_detection == turn_detection
     assert model.capabilities.turn_detection is False
     assert model.capabilities.can_disable_turn_detection is False
+
+
+async def test_inference_openai_models_support_targeted_cancellation(
+    paused_realtime_main: None,
+) -> None:
+    from livekit.agents import utils
+    from livekit.plugins.openai.realtime.realtime_model import (
+        ResponseCancelEvent,
+        _ResponseGeneration,
+    )
+
+    model = InferenceRealtimeModel(
+        "openai/gpt-realtime",
+        api_key="key",
+        api_secret="secret" * 8,
+    )
+    assert model._supports_targeted_cancellation is True
+
+    session = model.session()
+    sent: list[object] = []
+    session.send_event = lambda ev: sent.append(ev)  # type: ignore
+
+    gen = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id="resp_123",
+    )
+    session._current_generation = gen
+    session.interrupt()
+    assert len(sent) == 1
+    assert isinstance(sent[0], ResponseCancelEvent)
+    assert sent[0].response_id == "resp_123"
+
+    await session.aclose()
+
+
+async def test_inference_xai_models_omit_response_id_on_cancel(
+    paused_realtime_main: None,
+) -> None:
+    from livekit.agents import utils
+    from livekit.plugins.openai.realtime.realtime_model import (
+        ResponseCancelEvent,
+        _ResponseGeneration,
+    )
+
+    model = InferenceRealtimeModel(
+        "xai/grok-voice-latest",
+        api_key="key",
+        api_secret="secret" * 8,
+    )
+    assert model._supports_targeted_cancellation is False
+
+    session = model.session()
+    sent: list[object] = []
+    session.send_event = lambda ev: sent.append(ev)  # type: ignore
+
+    gen = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id="resp_123",
+    )
+    session._current_generation = gen
+    session.interrupt()
+    assert len(sent) == 1
+    assert isinstance(sent[0], ResponseCancelEvent)
+    assert sent[0].response_id is None
+
+    await session.aclose()

tests/test_realtime/test_openai_realtime_model.py

@@ -4,7 +4,7 @@
 import logging
 from dataclasses import replace
 from types import SimpleNamespace
-from typing import cast
+from typing import Any, cast
 
 import pytest
 from openai.types.beta.realtime.session import TurnDetection as BetaTurnDetection
@@ -517,3 +517,121 @@ def test_error_with_unknown_event_id_leaves_generate_reply_futures_untouched() -
     assert session._response_created_futures == {"response_create_1": fut}
     # still reported down the ordinary path
     assert captured["recoverable"] is True
+
+
+async def test_interrupt_includes_response_id_from_current_generation() -> None:
+    # interrupt() must specify response_id when available so concurrent responses are not a no-op (#5564)
+    from livekit.agents import utils
+    from livekit.plugins.openai.realtime.realtime_model import (
+        ResponseCancelEvent,
+        _ResponseGeneration,
+    )
+
+    sent: list[object] = []
+    session = RealtimeModel(api_key="fake").session()
+    session.send_event = lambda ev: sent.append(ev)  # type: ignore
+
+    session.interrupt()
+    assert len(sent) == 0
+
+    gen = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id="resp_123",
+    )
+    session._current_generation = gen
+    session.interrupt()
+    assert len(sent) == 1
+    assert isinstance(sent[0], ResponseCancelEvent)
+    assert sent[0].response_id == "resp_123"
+
+    gen_no_id = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id=None,
+    )
+    session._current_generation = gen_no_id
+    session.interrupt()
+    assert len(sent) == 2
+    assert isinstance(sent[1], ResponseCancelEvent)
+    assert sent[1].response_id is None
+
+    await session.aclose()
+
+
+async def test_interrupt_omits_response_id_for_legacy_azure() -> None:
+    # Legacy Azure API (with api_version) beta schema does not support response_id in response.cancel
+    from livekit.agents import utils
+    from livekit.plugins.openai.realtime.realtime_model import (
+        ResponseCancelEvent,
+        _normalize_azure_client_event,
+        _ResponseGeneration,
+    )
+
+    # Test _normalize_azure_client_event strips response_id from response.cancel
+    event: dict[str, Any] = {"type": "response.cancel", "response_id": "resp_123"}
+    _normalize_azure_client_event(event)
+    assert "response_id" not in event
+
+    # Test RealtimeSession.interrupt() omits response_id for legacy Azure
+    sent: list[object] = []
+    session = RealtimeModel.with_azure(
+        azure_deployment="dep",
+        api_key="fake",
+        base_url="https://example.com/openai",
+        api_version="2024-10-01-preview",
+    ).session()
+    session.send_event = lambda ev: sent.append(ev)  # type: ignore
+
+    gen = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id="resp_123",
+    )
+    session._current_generation = gen
+    session.interrupt()
+    assert len(sent) == 1
+    assert isinstance(sent[0], ResponseCancelEvent)
+    assert sent[0].response_id is None
+
+    await session.aclose()
+
+
+async def test_interrupt_omits_response_id_for_xai() -> None:
+    # xAI Realtime API v1 uses legacy bare cancellation event
+    from livekit.agents import utils
+    from livekit.plugins.openai.realtime.realtime_model import (
+        ResponseCancelEvent,
+        _ResponseGeneration,
+    )
+
+    sent: list[object] = []
+    session = RealtimeModel(api_key="fake").session()
+    session._realtime_model._provider_label = "xAI Realtime API"
+    session._realtime_model._supports_targeted_cancellation = False
+    session.send_event = lambda ev: sent.append(ev)  # type: ignore
+
+    gen = _ResponseGeneration(
+        message_ch=utils.aio.Chan(),
+        function_ch=utils.aio.Chan(),
+        messages={},
+        _created_timestamp=0.0,
+        _done_fut=asyncio.Future(),
+        response_id="resp_123",
+    )
+    session._current_generation = gen
+    session.interrupt()
+    assert len(sent) == 1
+    assert isinstance(sent[0], ResponseCancelEvent)
+    assert sent[0].response_id is None
+
+    await session.aclose()

tests/test_realtime/test_xai_realtime_model.py

@@ -666,3 +666,14 @@ def test_pending_say_ids_are_consumed_fifo() -> None:
 
     assert tagged == ["say_first", "say_second"]
     assert list(session._pending_say_event_ids) == []
+
+
+def test_xai_realtime_model_disables_targeted_cancellation() -> None:
+    model = RealtimeModel(api_key="fake")
+    assert model._supports_targeted_cancellation is False
+
+    session = RealtimeSession.__new__(RealtimeSession)
+    session._opts = SimpleNamespace(is_azure=False, api_version=None)
+    session._xai_model = model
+    session._realtime_model = model
+    assert session._supports_targeted_cancellation is False