fix(openai): re-generate when gpt-5.4 leaks internal tool-call format into assistant text

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/responses/llm.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import os
+import re
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 from typing import Any, Literal, cast
@@ -52,6 +53,45 @@
 OPENAI_RESPONSES_WS_URL = "wss://api.openai.com/v1/responses"
 
 
+# --- internal tool-call scaffolding leak detection -----------------------------
+# The gpt-5.4 series intermittently emits its INTERNAL tool-call wire format as
+# assistant TEXT instead of as structured function calls, e.g.
+#   to=multi_tool_use.parallel {"tool_uses":[{"recipient_name":"functions.foo",
+#   "parameters":{...}}, ...]}
+# frequently interleaved with memorized training-data spam. `multi_tool_use.parallel`
+# / `recipient_name` / `tool_uses` are old ChatGPT plugin-era internal constructs and
+# must never reach the caller as content.
+#
+# We do NOT try to interpret the corrupt payload (reconstructing tool calls from
+# garbage tokens risks acting on a misread). Instead we detect the signature at the
+# very start of the assistant text -- before any of it is streamed -- and raise a
+# retryable error, so the LLMStream retry loop re-issues the request for a fresh,
+# clean generation. Because we raise before emitting anything, `response.completed`
+# never runs, so no corrupt state is committed and previous_response_id chaining is
+# untouched.
+_SCAFFOLD_MARKERS = re.compile(
+    r"multi_tool_use|recipient_name|\"tool_uses\"|to\s*=\s*functions\.",
+)
+# Leading assistant-text chars buffered per output item before it is declared clean.
+# The signature appears at the very start of a corrupted item, so a small head catches
+# it before any is streamed; the latency impact is negligible and touches only the
+# first tokens of each message item -- and only for the affected model.
+_SCAFFOLD_HEAD_CHARS = 40
+
+# Scope the guard to the affected family ONLY. Every other model the Responses plugin
+# serves keeps the original, verbatim streaming path: no buffering, no detection, no
+# behavior change.
+_SCAFFOLD_AFFECTED_RE = re.compile(r"gpt-5\.4")
+
+
+def _model_may_leak_scaffolding(model: str) -> bool:
+    return bool(model) and _SCAFFOLD_AFFECTED_RE.search(model) is not None
+
+
+def _looks_like_tool_scaffolding(text: str) -> bool:
+    return bool(text) and _SCAFFOLD_MARKERS.search(text) is not None
+
+
 class _ResponsesWebsocket:
     def __init__(
         self, api_key: str | None, timeout: float | None, model: str, base_url: str | None = None
@@ -408,6 +448,17 @@ def __init__(
         self._response_id: str = ""
         self._response_completed: bool = False
         self._pending_tool_calls = set[str]()
+        # Per-output-item scaffolding-leak guard (see _handle_response_output_text_delta):
+        # item_id -> buffered leading text, and item_id -> verdict ("clean").
+        # _emitted_content records whether any assistant text has already been streamed
+        # this response -- if so a detected leak can't be retried safely (would
+        # double-emit), so it's raised non-retryable instead.
+        self._text_head: dict[str, str] = {}
+        self._text_verdict: dict[str, str] = {}
+        self._emitted_content: bool = False
+        # Guard is active ONLY for the affected model family; for every other model the
+        # text stream below takes the original verbatim path.
+        self._scaffold_guard: bool = _model_may_leak_scaffolding(str(model))
 
         self._client = client
         self._llm: LLM = llm
@@ -429,6 +480,10 @@ async def _run(self) -> None:
 
     async def _run_impl(self) -> None:
         self._response_completed = False
+        # Fresh scaffolding-guard state per attempt (this method re-runs on retry).
+        self._text_head.clear()
+        self._text_verdict.clear()
+        self._emitted_content = False
         chat_ctx, _ = self._chat_ctx.to_provider_format(format="openai.responses")
         self._tool_ctx = llm.ToolContext(self.tools)
         tool_schemas = cast(
@@ -631,23 +686,108 @@ def _handle_output_items_done(self, event: ResponseOutputItemDoneEvent) -> llm.C
                 ),
             )
             self._pending_tool_calls.add(event.item.call_id)
-        elif isinstance(event.item, ResponseOutputMessage) and event.item.phase is not None:
-            # Models like gpt-5.3-codex label assistant messages as intermediate
-            # `commentary` or the `final_answer`
-            chunk = llm.ChatChunk(
+        elif isinstance(event.item, ResponseOutputMessage):
+            if self._scaffold_guard:
+                # Affected model: finalize through the leak handler (flush clean
+                # buffered head; raise retryable if the item text is corrupt).
+                chunk = self._finalize_message_item(event.item)
+            elif event.item.phase is not None:
+                # Unaffected model -- original behavior: models like gpt-5.3-codex
+                # label assistant messages `commentary` / `final_answer`.
+                chunk = llm.ChatChunk(
+                    id=self._response_id,
+                    delta=llm.ChoiceDelta(
+                        role="assistant",
+                        content=None,
+                        extra={"openai": {"phase": event.item.phase}},
+                    ),
+                )
+        return chunk
+
+    def _raise_scaffolding_leak(self) -> None:
+        """Detected the internal tool-call format in assistant text. Raise so the
+        retry loop re-issues a fresh generation. Retryable ONLY if no content has been
+        streamed yet this response (otherwise a retry would double-emit what the caller
+        already received, so fail the turn instead)."""
+        retryable = not self._emitted_content
+        logger.warning(
+            "internal tool-call scaffolding leaked into assistant text (model=%s); %s",
+            self._model,
+            "raising retryable error to re-generate"
+            if retryable
+            else "content already streamed -- failing turn (cannot retry safely)",
+        )
+        raise APIStatusError(
+            "internal tool-call format leaked into assistant text",
+            status_code=502,
+            retryable=retryable,
+        )
+
+    def _finalize_message_item(self, item: ResponseOutputMessage) -> llm.ChatChunk | None:
+        item_id = getattr(item, "id", "") or ""
+        phase = item.phase
+        extra = {"openai": {"phase": phase}} if phase is not None else None
+        full_text = "".join(
+            getattr(c, "text", "") or ""
+            for c in (item.content or [])
+            if getattr(c, "type", "") == "output_text"
+        )
+
+        # Belt for coalesced deltas: if the complete item text trips the signature
+        # (and the streaming head somehow didn't), raise here.
+        if _looks_like_tool_scaffolding(full_text):
+            self._raise_scaffolding_leak()
+
+        # Clean, but short enough that the head was still buffered and never streamed --
+        # flush it now (carrying the phase marker if present).
+        if item_id in self._text_head:
+            head = self._text_head.pop(item_id, "")
+            self._text_verdict[item_id] = "clean"
+            if head:
+                self._emitted_content = True
+                return llm.ChatChunk(
+                    id=self._response_id,
+                    delta=llm.ChoiceDelta(role="assistant", content=head, extra=extra),
+                )
+
+        # Clean and already fully streamed -- forward the phase marker (the original
+        # behavior) if there is one.
+        if extra is not None:
+            return llm.ChatChunk(
                 id=self._response_id,
-                delta=llm.ChoiceDelta(
-                    role="assistant",
-                    content=None,
-                    extra={"openai": {"phase": event.item.phase}},
-                ),
+                delta=llm.ChoiceDelta(role="assistant", content=None, extra=extra),
             )
-        return chunk
+        return None
 
     def _handle_response_output_text_delta(
         self, event: ResponseTextDeltaEvent
     ) -> llm.ChatChunk | None:
-        return llm.ChatChunk(
-            id=self._response_id,
-            delta=llm.ChoiceDelta(content=event.delta, role="assistant"),
-        )
+        if not self._scaffold_guard:
+            # Unaffected model -- original verbatim behavior.
+            return llm.ChatChunk(
+                id=self._response_id,
+                delta=llm.ChoiceDelta(content=event.delta, role="assistant"),
+            )
+        item_id = event.item_id
+        if self._text_verdict.get(item_id) == "clean":
+            self._emitted_content = True
+            return llm.ChatChunk(
+                id=self._response_id,
+                delta=llm.ChoiceDelta(content=event.delta, role="assistant"),
+            )
+        # Deciding: buffer a bounded head, emitting nothing, until the item is ruled
+        # clean or trips the signature.
+        head = self._text_head.get(item_id, "") + (event.delta or "")
+        self._text_head[item_id] = head
+        if _looks_like_tool_scaffolding(head):
+            # Leak caught before any of it was streamed -- re-generate (retryable).
+            self._raise_scaffolding_leak()
+        if len(head) >= _SCAFFOLD_HEAD_CHARS:
+            self._text_verdict[item_id] = "clean"
+            self._text_head.pop(item_id, None)
+            self._emitted_content = True
+            return llm.ChatChunk(
+                id=self._response_id,
+                delta=llm.ChoiceDelta(content=head, role="assistant"),
+            )
+        return None  # keep buffering the head

tests/test_plugin_openai_responses.py

@@ -5,8 +5,20 @@
 import aiohttp
 import pytest
 from openai.types import Reasoning
-
-from livekit.plugins.openai.responses.llm import LLMStream, _ResponsesWebsocket
+from openai.types.responses import (
+    ResponseOutputItemDoneEvent,
+    ResponseOutputMessage,
+    ResponseOutputText,
+    ResponseTextDeltaEvent,
+)
+
+from livekit.agents import APIStatusError
+from livekit.plugins.openai.responses.llm import (
+    LLMStream,
+    _looks_like_tool_scaffolding,
+    _model_may_leak_scaffolding,
+    _ResponsesWebsocket,
+)
 
 pytestmark = pytest.mark.plugin("openai")
 
@@ -107,3 +119,134 @@ def test_error_event_missing_sequence_number_parses_cleanly() -> None:
     assert parsed.type == "error"
     assert parsed.message == frame["message"]
     assert parsed.param == "reasoning.mode"
+
+
+# --- internal tool-call scaffolding leak guard --------------------------------
+# The gpt-5.4 series intermittently emits its internal `multi_tool_use.parallel`
+# tool-call format as assistant TEXT (interleaved with training-data spam) instead
+# of as structured function calls. The guard detects that signature at the start of
+# the assistant text -- before any is streamed -- and raises a retryable error so
+# the stream is re-generated, rather than surfacing corrupt text to the caller.
+_TOOL_USES = {
+    "tool_uses": [
+        {
+            "recipient_name": "functions.do_thing",
+            "parameters": {"id": "aaaa0000bbbb1111", "action": "skipped"},
+        },
+        {"recipient_name": "functions.read_thing", "parameters": {"id": "cccc2222dddd3333"}},
+    ]
+}
+_LEAK_TEXT = "to=multi_tool_use.parallel ,最新高清无码专区" + json.dumps(_TOOL_USES) + "Two. Text."
+_CLEAN_LONG = "This is an ordinary assistant reply that is comfortably longer than the head window."
+_CLEAN_SHORT = "Yes."
+_ITEM_ID = "msg_abc"
+
+
+class _CollectingCh:
+    def __init__(self) -> None:
+        self.chunks: list = []
+
+    def send_nowait(self, chunk) -> None:
+        self.chunks.append(chunk)
+
+
+def _make_stream(model: str = "gpt-5.4") -> LLMStream:
+    s = LLMStream.__new__(LLMStream)
+    s._model = model
+    s._response_id = "resp_123"
+    s._text_head = {}
+    s._text_verdict = {}
+    s._emitted_content = False
+    s._scaffold_guard = _model_may_leak_scaffolding(str(model))
+    s._pending_tool_calls = set()
+    s._event_ch = _CollectingCh()
+    return s
+
+
+def _delta(piece: str, seq: int) -> ResponseTextDeltaEvent:
+    return ResponseTextDeltaEvent.model_construct(
+        item_id=_ITEM_ID,
+        output_index=0,
+        content_index=0,
+        delta=piece,
+        sequence_number=seq,
+        type="response.output_text.delta",
+        logprobs=[],
+    )
+
+
+def _item_done(full_text: str, phase: str | None = "final_answer") -> ResponseOutputItemDoneEvent:
+    msg = ResponseOutputMessage.model_construct(
+        id=_ITEM_ID,
+        role="assistant",
+        status="completed",
+        type="message",
+        phase=phase,
+        content=[
+            ResponseOutputText.model_construct(text=full_text, type="output_text", annotations=[])
+        ],
+    )
+    return ResponseOutputItemDoneEvent.model_construct(
+        item=msg, output_index=0, sequence_number=98, type="response.output_item.done"
+    )
+
+
+def _drive(stream: LLMStream, text: str) -> str:
+    """Feed `text` as token-sized deltas + the item.done event through the real
+    `_process_event` dispatch; return the content that reached the event channel.
+    Propagates the guard's error if it fires."""
+    for i in range(0, len(text), 6):
+        stream._process_event(_delta(text[i : i + 6], i))
+    stream._process_event(_item_done(text))
+    return "".join(
+        c.delta.content for c in stream._event_ch.chunks if c.delta is not None and c.delta.content
+    )
+
+
+def test_scaffolding_helpers() -> None:
+    assert _looks_like_tool_scaffolding(_LEAK_TEXT)
+    assert not _looks_like_tool_scaffolding(_CLEAN_LONG)
+    # scope: only the gpt-5.4 family is guarded
+    assert _model_may_leak_scaffolding("gpt-5.4")
+    assert _model_may_leak_scaffolding("gpt-5.4-codex")
+    assert not _model_may_leak_scaffolding("gpt-4.1")
+    assert not _model_may_leak_scaffolding("gpt-5.0")
+    assert not _model_may_leak_scaffolding("o3")
+
+
+def test_scaffolding_leak_raises_retryable_before_emitting() -> None:
+    stream = _make_stream()
+    with pytest.raises(APIStatusError) as ei:
+        _drive(stream, _LEAK_TEXT)
+    assert ei.value.retryable is True  # nothing streamed yet -> safe to re-generate
+    # none of the corrupt text reached the event channel
+    for chunk in stream._event_ch.chunks:
+        assert not (chunk.delta and chunk.delta.content)
+    assert stream._emitted_content is False
+
+
+def test_scaffolding_leak_after_content_raises_non_retryable() -> None:
+    """If assistant text was already streamed this response, a later leak can't be
+    retried without double-emitting, so it fails the turn instead."""
+    stream = _make_stream()
+    stream._emitted_content = True  # simulate a clean chunk already delivered
+    with pytest.raises(APIStatusError) as ei:
+        stream._handle_response_output_text_delta(_delta("to=multi_tool_use.parallel {", 0))
+    assert ei.value.retryable is False
+
+
+def test_clean_long_text_streams_unchanged() -> None:
+    stream = _make_stream()
+    assert _drive(stream, _CLEAN_LONG) == _CLEAN_LONG
+
+
+def test_clean_short_text_flushed_at_item_done() -> None:
+    stream = _make_stream()
+    assert _drive(stream, _CLEAN_SHORT) == _CLEAN_SHORT
+
+
+def test_unaffected_model_streams_leak_verbatim() -> None:
+    """On a non-5.4 model the text path is byte-for-byte the original: no buffering,
+    no detection, no raise -- even for the leak input."""
+    stream = _make_stream(model="gpt-4.1")
+    assert _drive(stream, _LEAK_TEXT) == _LEAK_TEXT