[BugFix] Optimize turn-based /v1/realtime auto truncation (#8566)

Signed-off-by: Nick Cao <ncao@redhat.com>
Co-authored-by: Codex <noreply@openai.com>

tests/entrypoints/openai_api/test_realtime_prompt_preflight.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import io
 import json
@@ -12,6 +13,8 @@ from typing import Any
 
 import numpy as np
 import pytest
+from openai.types import realtime as types
+from vllm.exceptions import VLLMValidationError
 from vllm.sampling_params import RequestOutputKind, SamplingParams
 
 from tests.helpers.serving_chat import build_serving_chat
@@ -20,7 +23,10 @@ from vllm_omni.entrypoints.openai.realtime.connection import (
     OpenAIFullDuplexConnection,
     _ResolvedResponse,
 )
-from vllm_omni.entrypoints.openai.realtime.session import ActiveResponse
+from vllm_omni.entrypoints.openai.realtime.session import (
+    ActiveResponse,
+    AudioFullDuplexSessionState,
+)
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -31,9 +37,13 @@ class _FakeRenderer:
 
     def __init__(self) -> None:
         self.raw_prompt_lengths: list[int] = []
+        self.message_counts: list[int] = []
+        self.skip_mm_cache: list[bool] = []
         self.tokenization_params: list[Any] = []
         self.engine_inputs: list[dict[str, Any]] = []
         self.conversations: list[list[dict[str, Any]]] = []
+        self.token_counts: dict[int, int] = {}
+        self.errors: dict[int, VLLMValidationError] = {}
 
     async def render_chat_async(
         self,
@@ -41,12 +51,24 @@ class _FakeRenderer:
         _chat_params: Any,
         tok_params: Any,
     ):
+        message_count = len(conversations[0])
+        self.message_counts.append(message_count)
         self.conversations.append(conversations[0])
         raw_prompt_length = 2 * len(conversations[0])
         self.raw_prompt_lengths.append(raw_prompt_length)
         self.tokenization_params.append(tok_params)
+        error = self.errors.get(message_count)
+        if error is not None:
+            raise error
         # Model-side expansion makes the rendered prompt much longer than its text tokens.
-        engine_input = {"prompt_token_ids": [0] * (10 + 20 * raw_prompt_length)}
+        token_count = self.token_counts.get(
+            message_count,
+            10 + 20 * raw_prompt_length,
+        )
+        engine_input = {
+            "prompt_token_ids": [0] * token_count,
+            "message_count": message_count,
+        }
         self.engine_inputs.append(engine_input)
         return conversations, [engine_input]
 
@@ -70,9 +92,8 @@ def _make_connection(
     model_config = SimpleNamespace(max_model_len=max_model_len, multimodal_config=None)
 
     async def preprocess_chat(request: Any, messages: list[dict[str, Any]], **kwargs: Any):
-        tok_params = kwargs.get("tok_params")
-        if tok_params is None:
-            tok_params = request.build_tok_params(model_config)
+        tok_params = request.build_tok_params(model_config)
+        renderer.skip_mm_cache.append(bool(kwargs.get("skip_mm_cache", False)))
         return await renderer.render_chat_async(
             [messages],
             kwargs.get("default_template_kwargs"),
@@ -97,22 +118,79 @@ def _make_connection(
     return connection, renderer
 
 
+def _message_item(
+    item_id: str,
+    text: str | None = None,
+) -> types.RealtimeConversationItemUserMessage:
+    return types.RealtimeConversationItemUserMessage(
+        id=item_id,
+        type="message",
+        role="user",
+        content=[{"type": "input_text", "text": text or item_id}],
+    )
+
+
+def _audio_item(item_id: str, audio: bytes) -> types.RealtimeConversationItemUserMessage:
+    return types.RealtimeConversationItemUserMessage(
+        id=item_id,
+        type="message",
+        role="user",
+        content=[
+            {
+                "type": "input_audio",
+                "audio": base64.b64encode(audio).decode("ascii"),
+            }
+        ],
+    )
+
+
+def _assistant_item(item_id: str, text: str) -> types.RealtimeConversationItemAssistantMessage:
+    return types.RealtimeConversationItemAssistantMessage(
+        id=item_id,
+        type="message",
+        role="assistant",
+        status="completed",
+        content=[{"type": "output_text", "text": text}],
+    )
+
+
+def _websocket_events(websocket: _FakeWebSocket) -> list[dict[str, Any]]:
+    return [json.loads(message) for message in websocket.messages]
+
+
+def _prompt_texts(conversation: list[dict[str, Any]]) -> list[str]:
+    texts: list[str] = []
+    for message in conversation:
+        content = message["content"]
+        if isinstance(content, str):
+            texts.append(content)
+            continue
+        texts.extend(part["text"] for part in content if part.get("type") == "text")
+    return texts
+
+
+async def _create_response(
+    connection: OpenAIFullDuplexConnection,
+) -> list[dict[str, Any]]:
+    submitted: list[dict[str, Any]] = []
+
+    async def run_response(_response_id: str, _response: Any, engine_input: dict[str, Any]) -> None:
+        submitted.append(engine_input)
+        connection.session.active_response = None
+
+    connection._run_response = run_response
+    await connection._handle_response_create(SimpleNamespace(event_id="evt_create", response=None))
+    if connection._response_task is not None:
+        await connection._response_task
+    return submitted
+
+
 @pytest.mark.asyncio
 async def test_preflight_truncates_by_rendered_token_count() -> None:
     connection, renderer = _make_connection()
 
-    first = SimpleNamespace(
-        id="first",
-        type="message",
-        role="user",
-        content=[SimpleNamespace(type="input_text", text="first")],
-    )
-    second = SimpleNamespace(
-        id="second",
-        type="message",
-        role="user",
-        content=[SimpleNamespace(type="input_text", text="second")],
-    )
+    first = _message_item("first")
+    second = _message_item("second")
     items = [first, second]
     response = _ResolvedResponse(
         input=items,
@@ -124,16 +202,405 @@ async def test_preflight_truncates_by_rendered_token_count() -> None:
         metadata=None,
     )
 
-    engine_input = await connection._prepare_engine_input_with_auto_truncation(response)
+    prompt_items = await connection._truncate_prompt_items(response)
+    engine_input = await connection._build_full_prompt(
+        tools=response.tools,
+        instructions=response.instructions,
+        items=prompt_items,
+    )
 
     assert renderer.raw_prompt_lengths == [4, 2, 2]
     assert [len(prompt["prompt_token_ids"]) for prompt in renderer.engine_inputs] == [90, 50, 50]
-    assert all(params.max_total_tokens is None for params in renderer.tokenization_params)
+    assert all(params.max_total_tokens == 100 for params in renderer.tokenization_params)
     assert all(params.max_output_tokens == 0 for params in renderer.tokenization_params)
-    assert items == [second]
+    assert items == [first, second]
+    assert prompt_items == [second]
     assert engine_input is renderer.engine_inputs[-1]
 
 
+@pytest.mark.asyncio
+async def test_persistent_model_context_cursor_reuses_truncated_window() -> None:
+    connection, renderer = _make_connection()
+    first = _message_item("first")
+    second = _message_item("second")
+    connection.session.insert_item(first)
+    connection.session.insert_item(second)
+    response = _ResolvedResponse(
+        input=None,
+        instructions=None,
+        modalities=["text"],
+        max_output_tokens="inf",
+        tools=None,
+        tool_choice="none",
+        metadata=None,
+    )
+
+    prompt_items = await connection._truncate_prompt_items(response)
+    await connection._build_full_prompt(
+        tools=response.tools,
+        instructions=response.instructions,
+        items=prompt_items,
+    )
+    connection.session.commit_model_context_items(prompt_items)
+    renderer.raw_prompt_lengths.clear()
+
+    prompt_items = await connection._truncate_prompt_items(response)
+
+    assert renderer.raw_prompt_lengths == [2]
+    assert prompt_items == [second]
+    assert connection.session.model_context_first_item_id == "second"
+    assert not connection.session.model_context_cursor_at_end
+
+
+@pytest.mark.asyncio
+async def test_empty_model_context_allows_a_later_appended_item() -> None:
+    connection, renderer = _make_connection(max_model_len=20)
+    response = _ResolvedResponse(
+        input=None,
+        instructions=None,
+        modalities=["text"],
+        max_output_tokens="inf",
+        tools=None,
+        tool_choice="none",
+        metadata=None,
+    )
+
+    prompt_items = await connection._truncate_prompt_items(response)
+    connection.session.commit_model_context_items(prompt_items)
+    connection.session.insert_item(_message_item("appended"))
+
+    assert renderer.raw_prompt_lengths == [0]
+    assert prompt_items == []
+    assert [item.id for item in connection.session.model_context_items()] == ["appended"]
+    assert connection.session.model_context_first_item_id == "appended"
+    assert not connection.session.model_context_cursor_at_end
+
+
+@pytest.mark.asyncio
+async def test_truncation_to_empty_selects_no_items_when_overhead_fits() -> None:
+    connection, renderer = _make_connection(max_model_len=20)
+    connection.session.insert_item(_message_item("item"))
+    response = _ResolvedResponse(
+        input=None,
+        instructions=None,
+        modalities=["text"],
+        max_output_tokens="inf",
+        tools=None,
+        tool_choice="none",
+        metadata=None,
+    )
+
+    prompt_items = await connection._truncate_prompt_items(response)
+    connection.session.commit_model_context_items(prompt_items)
+    connection.session.insert_item(_message_item("appended"))
+
+    assert renderer.raw_prompt_lengths == [2, 0]
+    assert prompt_items == []
+    assert [item.id for item in connection.session.model_context_items()] == ["appended"]
+
+
+def test_model_context_cursor_moves_only_forward() -> None:
+    session = AudioFullDuplexSessionState()
+
+    session.insert_item(_message_item("first"))
+    session.insert_item(_message_item("second"))
+    session.insert_item(_message_item("third"))
+    assert session.model_context_first_item_id == "first"
+    assert session.model_context_items() == session.items
+
+    session.commit_model_context_items(session.items[1:])
+    session.insert_item(_message_item("before-second"), previous_item_id="first")
+    assert session.model_context_first_item_id == "second"
+    assert [item.id for item in session.model_context_items()] == ["second", "third"]
+
+    session.insert_item(_message_item("after-second"), previous_item_id="second")
+    assert session.model_context_first_item_id == "second"
+    assert [item.id for item in session.model_context_items()] == [
+        "second",
+        "after-second",
+        "third",
+    ]
+
+    session.remove_item("second")
+    assert session.model_context_first_item_id == "after-second"
+    assert [item.id for item in session.model_context_items()] == ["after-second", "third"]
+
+    session.commit_model_context_items([])
+    session.insert_item(_message_item("fourth"))
+    session.insert_item(_message_item("root"), previous_item_id="root")
+    assert session.model_context_first_item_id == "fourth"
+    assert not session.model_context_cursor_at_end
+    assert [item.id for item in session.model_context_items()] == ["fourth"]
+
+
+@pytest.mark.asyncio
+async def test_prompt_truncation_bisects_the_model_context_cursor() -> None:
+    connection, renderer = _make_connection()
+    items = [
+        SimpleNamespace(
+            id=f"item_{index}",
+            type="message",
+            role="user",
+            content=[SimpleNamespace(type="input_text", text=str(index))],
+        )
+        for index in range(8)
+    ]
+    response = _ResolvedResponse(
+        input=items,
+        instructions=None,
+        modalities=["text"],
+        max_output_tokens="inf",
+        tools=None,
+        tool_choice="none",
+        metadata=None,
+    )
+
+    prompt_items = await connection._truncate_prompt_items(response)
+    await connection._build_full_prompt(
+        tools=response.tools,
+        instructions=response.instructions,
+        items=prompt_items,
+    )
+
+    assert renderer.raw_prompt_lengths == [16, 8, 4, 2, 2]
+    assert prompt_items == [items[-1]]
+
+
+@pytest.mark.asyncio
+async def test_realtime_auto_truncation_end_to_end() -> None:
+    websocket = _FakeWebSocket()
+    connection, renderer = _make_connection(websocket=websocket)
+    audio = b"\x01\x00\x02\x00\x03\x00\x04\x00"
+    items: list[Any] = [_message_item(f"item_{index}") for index in range(7)]
+    items.append(_audio_item("item_audio", audio))
+    for item in items:
+        connection.session.insert_item(item)
+
+    submitted = await _create_response(connection)
+
+    assert len(submitted) == 1
+    assert submitted[0]["message_count"] == 1
+    assert renderer.message_counts == [8, 4, 2, 1, 1]
+    assert renderer.skip_mm_cache == [True, True, True, True, False]
+    assert connection.session.model_context_first_item_id == "item_audio"
+    assert not connection.session.model_context_cursor_at_end
+    assert len(connection.session.items) == 8
+
+    events = _websocket_events(websocket)
+    assert not [event for event in events if event["type"] == "conversation.item.deleted"]
+    assert [event for event in events if event["type"] == "response.created"]
+
+    await connection._handle_item_retrieve(
+        types.ConversationItemRetrieveEvent(
+            type="conversation.item.retrieve",
+            event_id="evt_retrieve",
+            item_id="item_0",
+        )
+    )
+    retrieved = _websocket_events(websocket)[-1]
+    assert retrieved["type"] == "conversation.item.retrieved"
+    assert retrieved["item"]["id"] == "item_0"
+
+    final_conversation = renderer.conversations[-1]
+    assert len(final_conversation) == 1
+    audio_content = final_conversation[0]["content"][0]["input_audio"]
+    with wave.open(io.BytesIO(base64.b64decode(audio_content["data"]))) as wav:
+        assert wav.readframes(wav.getnframes()) == audio
+
+
+@pytest.mark.parametrize(
+    ("truncation", "max_output_tokens", "expected_texts"),
+    [
+        ("auto", "inf", ["item_1", "item_2", "item_3"]),
+        ("disabled", "inf", None),
+        (
+            types.RealtimeTruncationRetentionRatio(
+                type="retention_ratio",
+                retention_ratio=0.5,
+                token_limits={"post_instructions": 70},
+            ),
+            "inf",
+            ["item_3"],
+        ),
+        (
+            types.RealtimeTruncationRetentionRatio(
+                type="retention_ratio",
+                retention_ratio=0.3,
+            ),
+            "inf",
+            ["item_3"],
+        ),
+        ("auto", 25, ["item_3"]),
+    ],
+)
+@pytest.mark.asyncio
+async def test_realtime_truncation_configurations(
+    truncation: Any,
+    max_output_tokens: Any,
+    expected_texts: list[str] | None,
+) -> None:
+    websocket = _FakeWebSocket()
+    connection, renderer = _make_connection(websocket=websocket)
+    connection.session.config.truncation = truncation
+    connection.session.config.max_output_tokens = max_output_tokens
+    renderer.token_counts = {0: 10, 1: 30, 2: 45, 3: 50, 4: 110}
+    for index in range(4):
+        connection.session.insert_item(_message_item(f"item_{index}"))
+
+    submitted = await _create_response(connection)
+
+    if expected_texts is None:
+        assert submitted == []
+        assert connection._response_task is None
+        assert "exceeds the model's input token limit" in websocket.messages[-1]
+        return
+
+    assert len(submitted) == 1
+    assert _prompt_texts(renderer.conversations[-1]) == expected_texts
+    assert connection.session.model_context_first_item_id == f"item_{4 - len(expected_texts)}"
+
+
+@pytest.mark.asyncio
+async def test_realtime_truncation_render_errors() -> None:
+    websocket = _FakeWebSocket()
+    connection, renderer = _make_connection(websocket=websocket)
+    renderer.token_counts = {0: 10}
+    renderer.errors = {
+        2: VLLMValidationError("Prompt is too long", parameter="input_tokens"),
+        1: VLLMValidationError("Prompt is too long", parameter="input_tokens"),
+    }
+    connection.session.insert_item(_message_item("first"))
+    connection.session.insert_item(_message_item("second"))
+
+    submitted = await _create_response(connection)
+
+    assert len(submitted) == 1
+    assert submitted[0]["message_count"] == 0
+    assert renderer.message_counts == [2, 1, 0, 0]
+    assert renderer.skip_mm_cache == [True, True, True, False]
+    assert connection.session.model_context_cursor_at_end
+
+    websocket = _FakeWebSocket()
+    connection, renderer = _make_connection(websocket=websocket)
+    renderer.errors = {
+        2: VLLMValidationError("Tool rendering failed", parameter="tools"),
+    }
+    connection.session.insert_item(_message_item("first"))
+    connection.session.insert_item(_message_item("second"))
+
+    submitted = await _create_response(connection)
+
+    assert submitted == []
+    assert connection._response_task is None
+    assert renderer.message_counts == [2]
+    assert "Tool rendering failed" in websocket.messages[-1]
+
+
+@pytest.mark.asyncio
+async def test_realtime_truncation_survives_history_mutations() -> None:
+    websocket = _FakeWebSocket()
+    connection, renderer = _make_connection(websocket=websocket)
+    renderer.token_counts = {0: 10, 1: 30, 2: 45, 3: 50, 4: 110}
+
+    async def add_item(item: Any, previous_item_id: str | None = None) -> None:
+        await connection._handle_item_create(
+            types.ConversationItemCreateEvent(
+                type="conversation.item.create",
+                event_id=f"evt_create_{item.id}",
+                item=item,
+                previous_item_id=previous_item_id,
+            )
+        )
+
+    await add_item(_message_item("item_0"))
+    await add_item(_message_item("item_1"))
+    await add_item(_assistant_item("assistant", "assistant text"))
+    await add_item(_message_item("item_3"))
+
+    submitted = await _create_response(connection)
+    assert len(submitted) == 1
+    assert _prompt_texts(renderer.conversations[-1]) == [
+        "item_1",
+        "assistant text",
+        "item_3",
+    ]
+    assert connection.session.model_context_first_item_id == "item_1"
+
+    renderer.message_counts.clear()
+    renderer.skip_mm_cache.clear()
+    await add_item(_message_item("before"), previous_item_id="item_0")
+    await add_item(_message_item("after"), previous_item_id="item_3")
+    assert connection.session.model_context_first_item_id == "item_1"
+
+    await connection._handle_item_truncate(
+        types.ConversationItemTruncateEvent(
+            type="conversation.item.truncate",
+            event_id="evt_truncate",
+            item_id="assistant",
+            content_index=0,
+            audio_end_ms=0,
+        )
+    )
+    truncated_part = connection.session.find_item("assistant").content[0]
+    assert truncated_part.type == "output_audio"
+    assert truncated_part.transcript == ""
+
+    await connection._handle_item_delete(
+        types.ConversationItemDeleteEvent(
+            type="conversation.item.delete",
+            event_id="evt_delete",
+            item_id="item_1",
+        )
+    )
+    assert connection.session.model_context_first_item_id == "assistant"
+
+    active_item = _assistant_item("active", "active text")
+    connection.session.insert_item(active_item)
+    active = ActiveResponse(response_id="resp_active", request_id="req_active")
+    active.item_id = active_item.id
+    connection.session.active_response = active
+    cancellation_cleanup: list[str] = []
+
+    async def active_response_task() -> None:
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            connection.session.remove_item(active_item.id)
+            connection.session.active_response = None
+            cancellation_cleanup.append(active_item.id)
+            raise
+
+    connection._response_task = asyncio.create_task(active_response_task())
+    await asyncio.sleep(0)
+    aborted_requests: list[str] = []
+
+    async def abort(request_id: str) -> None:
+        aborted_requests.append(request_id)
+
+    connection.engine.abort = abort
+    renderer.message_counts.clear()
+    renderer.skip_mm_cache.clear()
+
+    submitted = await _create_response(connection)
+
+    assert aborted_requests == ["req_active"]
+    assert cancellation_cleanup == ["active"]
+    assert renderer.message_counts == [3, 2, 2]
+    assert renderer.skip_mm_cache == [True, True, False]
+    assert len(submitted) == 1
+    assert _prompt_texts(renderer.conversations[-1]) == ["item_3", "after"]
+    assert [item.id for item in connection.session.items] == [
+        "item_0",
+        "before",
+        "assistant",
+        "item_3",
+        "after",
+    ]
+    assert connection.session.model_context_first_item_id == "assistant"
+    assert connection.session.active_response is None
+
+
+@pytest.mark.asyncio
 @pytest.mark.asyncio
 async def test_audio_prompt_uses_standard_chat_audio_content() -> None:
     audio = b"\x00\x00\x00\x40"

vllm_omni/entrypoints/openai/realtime/connection.py

@@ -9,7 +9,7 @@ import io
 import json
 import warnings
 import wave
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
 import numpy as np
@@ -57,13 +57,17 @@ AUTO_TRUNCATION_TRIGGER_RATIO = 0.8
 AUTO_TRUNCATION_TARGET_RATIO = 0.5
 
 
+def _is_prompt_limit_error(exc: VLLMValidationError) -> bool:
+    return exc.parameter in {"input_text", "input_tokens"}
+
+
 class _UnsupportedAudioFormatError(ValueError):
     """The Realtime endpoint only accepts mono PCM16 audio at 24 kHz."""
 
 
 @dataclass(slots=True)
 class _ResolvedResponse:
-    input: list[Any] | None
+    input: list[types.ConversationItem] | None
     instructions: str | None
     modalities: list[str]
     max_output_tokens: int | str
@@ -434,24 +438,31 @@ class OpenAIFullDuplexConnection:
             metadata=metadata,
         )
 
-    async def _prepare_engine_input_with_auto_truncation(
-        self,
-        response: _ResolvedResponse,
-        *,
-        preflight_only: bool = False,
-    ) -> EngineInput | None:
+    async def _truncate_prompt_items(self, response: _ResolvedResponse) -> list[types.ConversationItem] | None:
         s = self.session
-        max_model_len = getattr(self.engine.model_config, "max_model_len", None)
-        persistent = response.input is None
-        items: list[Any] = s.items if response.input is None else response.input
-        staged_deletions: list[Any] = []
-        if max_model_len is None:
-            return await self._build_full_prompt(
-                tools=response.tools,
-                instructions=response.instructions,
-                items=items,
-                skip_mm_cache=preflight_only,
-            )
+        max_model_len = self.engine.model_config.max_model_len
+        if response.input is None:
+            items = s.model_context_items()
+        else:
+            items = response.input
+
+        async def probe_prompt(
+            current_items: list[types.ConversationItem],
+        ) -> tuple[EngineInput | None, int]:
+            # Budget checks are speculative and may render repeatedly while
+            # truncating, so keep them off the shared sender cache.
+            try:
+                prompt = await self._build_full_prompt(
+                    tools=response.tools,
+                    instructions=response.instructions,
+                    items=current_items,
+                    skip_mm_cache=True,
+                )
+            except VLLMValidationError as exc:
+                if not _is_prompt_limit_error(exc):
+                    raise
+                return None, max_model_len + 1
+            return prompt, len(prompt["prompt_token_ids"])
 
         truncation = s.config.truncation or "auto"
         ratio = 1.0
@@ -477,24 +488,9 @@ class OpenAIFullDuplexConnection:
             trigger = limit
             target = limit
 
-        # Budget checks are speculative and may render repeatedly while
-        # truncating, so keep them off the shared sender cache. Once accepted,
-        # render again through the normal cache path for engine submission.
-        engine_input = await self._build_full_prompt(
-            tools=response.tools,
-            instructions=response.instructions,
-            items=items,
-            skip_mm_cache=True,
-        )
-        total = len(engine_input["prompt_token_ids"])
+        engine_input, total = await probe_prompt(items)
         if total <= trigger:
-            if preflight_only:
-                return engine_input
-            return await self._build_full_prompt(
-                tools=response.tools,
-                instructions=response.instructions,
-                items=items,
-            )
+            return items
         if mode == "disabled":
             logger.warning(
                 "[realtime] token budget exceeded (%d/%d) and truncation is disabled -- rejecting response.create",
@@ -503,68 +499,30 @@ class OpenAIFullDuplexConnection:
             )
             return None
 
-        if persistent:
-            # Preflight truncation is speculative. Work on a snapshot so a
-            # rejected response.create does not delete items from session history.
-            items = list(items)
+        all_items = items
+        if not all_items:
+            return None if total > limit else all_items
 
-        idx = 0
-        while total > target and idx < len(items):
-            item = items[idx]
-            if getattr(item, "role", None) == "system":
-                idx += 1
-                continue
-            if persistent and item.id is not None and s.item_in_progress.get(item.id, False):
-                idx += 1
-                continue
+        low = 1
+        high = len(all_items)
+        feasible_index: int | None = None
+        while low < high:
+            middle = (low + high) // 2
+            items = all_items[middle:]
+            engine_input, total = await probe_prompt(items)
+            if total <= target:
+                high = middle
+                feasible_index = middle
+            else:
+                low = middle + 1
 
-            remove_indexes = [idx]
-            if item.type in ("function_call", "function_call_output"):
-                call_id = getattr(item, "call_id", None)
-                pair_idx = next(
-                    (
-                        other_idx
-                        for other_idx, other in enumerate(items)
-                        if other_idx != idx and getattr(other, "call_id", None) == call_id
-                    ),
-                    None,
-                )
-                if pair_idx is not None:
-                    remove_indexes.append(pair_idx)
-
-            for remove_idx in sorted(remove_indexes, reverse=True):
-                removed = items[remove_idx]
-                if persistent:
-                    staged_deletions.append(removed)
-                del items[remove_idx]
-            engine_input = await self._build_full_prompt(
-                tools=response.tools,
-                instructions=response.instructions,
-                items=items,
-                skip_mm_cache=True,
-            )
-            total = len(engine_input["prompt_token_ids"])
+        if feasible_index != low:
+            items = all_items[low:]
+            engine_input, total = await probe_prompt(items)
 
         if total > limit:
             return None
-        if preflight_only:
-            return engine_input
-        engine_input = await self._build_full_prompt(
-            tools=response.tools,
-            instructions=response.instructions,
-            items=items,
-        )
-        for removed in staged_deletions:
-            if removed.id is None or s.remove_item(removed.id) is None:
-                continue
-            await self._send_event(
-                types.ConversationItemDeletedEvent(
-                    event_id=_gen_id("evt"),
-                    type="conversation.item.deleted",
-                    item_id=removed.id,
-                )
-            )
-        return engine_input
+        return items
 
     async def _handle_response_create(self, event: types.ResponseCreateEvent):
         s = self.session
@@ -585,14 +543,11 @@ class OpenAIFullDuplexConnection:
 
         had_active_response = s.active_response is not None
         try:
-            engine_input = await self._prepare_engine_input_with_auto_truncation(
-                response,
-                preflight_only=had_active_response,
-            )
+            preflight_items = await self._truncate_prompt_items(response)
         except VLLMValidationError as exc:
             await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id)
             return
-        if engine_input is None:
+        if preflight_items is None:
             await self._send_error(
                 "The response input exceeds the model's input token limit",
                 "invalid_request_error",
@@ -604,11 +559,11 @@ class OpenAIFullDuplexConnection:
             if s.active_response is not None:
                 await self._cancel_active_response()
             try:
-                engine_input = await self._prepare_engine_input_with_auto_truncation(response)
+                preflight_items = await self._truncate_prompt_items(response)
             except VLLMValidationError as exc:
                 await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id)
                 return
-            if engine_input is None:
+            if preflight_items is None:
                 await self._send_error(
                     "The response input exceeds the model's input token limit",
                     "invalid_request_error",
@@ -616,6 +571,18 @@ class OpenAIFullDuplexConnection:
                 )
                 return
 
+        try:
+            engine_input = await self._build_full_prompt(
+                tools=response.tools,
+                instructions=response.instructions,
+                items=preflight_items,
+            )
+        except VLLMValidationError as exc:
+            await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id)
+            return
+        if response.input is None:
+            s.commit_model_context_items(preflight_items)
+
         response_id = _gen_id("resp")
         await self._send_event(
             types.ResponseCreatedEvent(
@@ -1609,8 +1576,8 @@ class OpenAIFullDuplexConnection:
                 return text
         return ""
 
-    def _resolve_response_input(self, items: list[Any]) -> list[Any]:
-        resolved = []
+    def _resolve_response_input(self, items: list[Any]) -> list[types.ConversationItem]:
+        resolved: list[types.ConversationItem] = []
         for item in items:
             if getattr(item, "type", None) != "item_reference":
                 self._validate_input_item(item)
@@ -1641,7 +1608,7 @@ class OpenAIFullDuplexConnection:
         tools: list | None = None,
         *,
         instructions: str | None = None,
-        items: list | None = None,
+        items: list[types.ConversationItem] | None = None,
         skip_mm_cache: bool = False,
     ) -> EngineInput:
         """Render the effective conversation through normal chat preprocessing."""
@@ -1706,10 +1673,6 @@ class OpenAIFullDuplexConnection:
                     messages.append({"role": "system", "content": text})
 
         request = ChatCompletionRequest(model=self.model_name, messages=messages)
-        tok_params = replace(
-            request.build_tok_params(self.engine.model_config),
-            max_total_tokens=None,
-        )
         _, (engine_input,) = await chat_handler._preprocess_chat(
             request,
             messages,
@@ -1717,7 +1680,6 @@ class OpenAIFullDuplexConnection:
             default_template_content_format=chat_handler.chat_template_content_format,
             default_template_kwargs=chat_handler._effective_chat_template_kwargs(request),
             tool_dicts=converted_tools,
-            tok_params=tok_params,
             skip_mm_cache=skip_mm_cache,
         )
         if engine_input.get("prompt_token_ids") is None:

vllm_omni/entrypoints/openai/realtime/session.py

@@ -87,7 +87,15 @@ class AudioFullDuplexSessionState:
     config: types.RealtimeSessionCreateRequest = field(default_factory=_default_config)
 
     conversation_id: str = field(default_factory=lambda: _gen_id("conv"))
-    items: list[Any] = field(default_factory=list)
+    items: list[types.ConversationItem] = field(default_factory=list)
+    # Valid model-context cursor states are:
+    # - (None, False): include from the first retained item.
+    # - (first item ID, False): include from that item onward.
+    # - (None, True): the cursor is past the end and the model context is
+    #   empty; a later append moves the cursor to the appended item.
+    # The combination (item ID, True) is invalid.
+    model_context_first_item_id: str | None = None
+    model_context_cursor_at_end: bool = False
 
     item_duration_ms: dict[str, float] = field(default_factory=dict)
     item_token_ids: dict[str, list[int]] = field(default_factory=dict)
@@ -103,17 +111,17 @@ class AudioFullDuplexSessionState:
                 return i
         return None
 
-    def find_item(self, item_id: str) -> Any | None:
+    def find_item(self, item_id: str) -> types.ConversationItem | None:
         idx = self.find_item_index(item_id)
         return self.items[idx] if idx is not None else None
 
     @staticmethod
-    def _jsonable_item(item: Any) -> Any:
+    def _jsonable_item(item: types.ConversationItem) -> Any:
         if hasattr(item, "model_dump"):
             return item.model_dump(mode="json", exclude_none=True)
         return item
 
-    def _history_size(self, items: list[Any] | None = None) -> int:
+    def _history_size(self, items: list[types.ConversationItem] | None = None) -> int:
         history = self.items if items is None else items
         serialized = json.dumps(
             [self._jsonable_item(item) for item in history],
@@ -122,10 +130,25 @@ class AudioFullDuplexSessionState:
         )
         return len(serialized.encode("utf-8"))
 
-    def _ensure_history_capacity(self, items: list[Any]) -> None:
+    def _ensure_history_capacity(self, items: list[types.ConversationItem]) -> None:
         if self._history_size(items) > MAX_HISTORY_BYTES:
             raise HistoryLimitError(f"Conversation history exceeds the {MAX_HISTORY_BYTES} byte limit")
 
+    def _model_context_start_index(self) -> int:
+        if self.model_context_cursor_at_end:
+            return len(self.items)
+        if self.model_context_first_item_id is None:
+            return 0
+        index = self.find_item_index(self.model_context_first_item_id)
+        return len(self.items) if index is None else index
+
+    def model_context_items(self) -> list[types.ConversationItem]:
+        return self.items[self._model_context_start_index() :]
+
+    def commit_model_context_items(self, items: list[types.ConversationItem]) -> None:
+        self.model_context_first_item_id = items[0].id if items else None
+        self.model_context_cursor_at_end = not items
+
     def insert_item(self, item: Any, previous_item_id: str | None = None) -> int:
         if item.id is not None:
             existing_idx = self.find_item_index(item.id)
@@ -153,6 +176,13 @@ class AudioFullDuplexSessionState:
         candidate.insert(pos, item)
         self._ensure_history_capacity(candidate)
         self.items.insert(pos, item)
+        # A past-the-end cursor only advances for an item appended after it.
+        # Inserting before the end leaves the cursor past every included item.
+        if self.model_context_cursor_at_end and pos == len(self.items) - 1:
+            self.model_context_first_item_id = item.id
+            self.model_context_cursor_at_end = False
+        elif not self.model_context_cursor_at_end and self.model_context_first_item_id is None:
+            self.model_context_first_item_id = item.id
         return pos
 
     def replace_item(self, item: Any) -> int:
@@ -177,11 +207,18 @@ class AudioFullDuplexSessionState:
         self.item_in_progress.pop(item_id, None)
         self.pending_truncations_ms.pop(item_id, None)
 
-    def remove_item(self, item_id: str) -> Any | None:
+    def remove_item(self, item_id: str) -> types.ConversationItem | None:
         idx = self.find_item_index(item_id)
         if idx is None:
             return None
+        cursor_index = self._model_context_start_index()
         item = self.items.pop(idx)
+        if idx == cursor_index and not self.model_context_cursor_at_end:
+            if idx < len(self.items):
+                self.model_context_first_item_id = self.items[idx].id
+            else:
+                self.model_context_first_item_id = None
+                self.model_context_cursor_at_end = True
         if item.id:
             self._clear_item_metadata(item.id)
         return item

vllm_omni/entrypoints/openai/serving_chat.py

@@ -105,7 +105,6 @@ from vllm.parser.mistral import MistralToolCall
 from vllm.reasoning import ReasoningParser
 from vllm.renderers import BaseRenderer, merge_kwargs
 from vllm.renderers.inputs import TokPrompt
-from vllm.renderers.params import TokenizeParams
 from vllm.sampling_params import RequestOutputKind, SamplingParams
 from vllm.tokenizers import TokenizerLike
 from vllm.tokenizers import TokenizerLike as AnyTokenizer
@@ -977,7 +976,6 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
         continue_final_message: bool = False,
         documents: list[dict[str, str]] | None = None,
         add_special_tokens: bool = False,
-        tok_params: TokenizeParams | None = None,
         skip_mm_cache: bool = False,
     ) -> tuple[list[ConversationMessage], list[TokPrompt]]:
         if renderer is None:
@@ -997,8 +995,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
             ),
         )
 
-        if tok_params is None:
-            tok_params = request.build_tok_params(self.model_config)
+        tok_params = request.build_tok_params(self.model_config)
         mm_config = self.model_config.multimodal_config
         chat_params = request.build_chat_params(
             default_template,