[Bugfix][Core] Apply Realtime item truncation once per command (#7800)

Signed-off-by: Nolen Liang <nliang@nvidia.com>

docs/serving/realtime_duplex_api.md

@@ -948,6 +948,10 @@ deferred commit during an active response (`event.response_create_deferred`)
 {"type": "conversation.item.truncated", "item_id": "item_resp_01", "content_index": 0, "audio_end_ms": 1850, "event": {"…": "…"}}
 ```
 
+After this acknowledgement, `conversation.item.retrieve` returns the truncated
+transcript. Without text/audio alignment marks, its prefix is estimated from
+the requested position and audio duration; this is not exact word alignment.
+
 #### Response lifecycle
 
 `C→S response.create`

tests/engine/duplex/test_duplex_events.py

@@ -11,6 +11,7 @@ import inspect
 import pytest
 
 from vllm_omni.engine.duplex import events as events_module
+from vllm_omni.engine.duplex.commands import TruncateItem
 from vllm_omni.engine.duplex.events import (
     REALTIME_ERROR_TYPES_BY_CODE,
     AudioDelta,
@@ -40,6 +41,8 @@ from vllm_omni.engine.duplex.events import (
 from vllm_omni.engine.duplex.realtime_events import (
     RealtimeProjectionState,
     project_internal_event,
+    resolve_truncate_item,
+    retrieve_item_events,
 )
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
@@ -373,6 +376,30 @@ def test_projection_of_output_audio_buffer_clear_emits_cleared_before_terminals(
     assert state.item_truncation_cursors["item_resp_1"] == (0, 250)
 
 
+def test_one_truncate_command_truncates_retrieved_transcript_once():
+    state = RealtimeProjectionState(session_id="duplex-truncate")
+    state.conversation_items["item_resp_1"] = {
+        "id": "item_resp_1",
+        "type": "message",
+        "role": "assistant",
+        "status": "completed",
+        "content": [{"type": "output_audio", "transcript": "abcdefghij", "audio_duration_ms": 10_000}],
+    }
+
+    control = resolve_truncate_item(state, TruncateItem(item_id="item_resp_1", audio_end_ms=4_000))
+    # The runner executes this signal and projects its acknowledgement.
+    signal = control.payloads[0]
+    payload = signal["payload"]
+    assert isinstance(payload, dict)
+    assert signal["event"] == "conversation.item.truncate"
+    project_internal_event(state, {"type": "conversation.item.truncated", **payload})
+
+    retrieved = retrieve_item_events(state, {"item_id": "item_resp_1"})[0].to_realtime()
+    assert retrieved["item"]["content"][0]["transcript"] == "abcd"
+    # Keep the cursor: later output must not restore the untruncated item.
+    assert state.item_truncation_cursors["item_resp_1"] == (0, 4_000)
+
+
 def test_projection_leaves_error_to_typed_emit_sites():
     # Errors are constructed typed (``error_event``) at the emit site; the projector
     # has no ``error`` branch and falls back to the raw wrapper for such a dict.

vllm_omni/engine/duplex/realtime_events.py

@@ -1362,7 +1362,6 @@ def resolve_truncate_item(state: RealtimeProjectionState, command: TruncateItem)
             payloads=[], events=[error_event("bad_event", truncate_error, event_id=command.event_id)]
         )
     state.item_truncation_cursors[command.item_id] = (command.content_index, command.audio_end_ms)
-    truncate_realtime_item_content(item, content_index=command.content_index, audio_end_ms=command.audio_end_ms)
     ack_payload: dict[str, object] = {
         "type": "playback.ack",
         "item_id": command.item_id,