[Bugfix][MiniCPM-o] Checkpoint playback before final input commit (#6821)

Signed-off-by: Gao Han <hgaoaf@connect.ust.hk>

examples/online_serving/minicpmo/realtime_duplex_demo.py

@@ -155,6 +155,23 @@ def _response_in_progress(events: list[dict[str, object]]) -> bool:
     )
 
 
+async def _commit_input_after_playback_checkpoint(client: RealtimeDuplexClient) -> float:
+    """Commit input only after preserving already-played assistant history.
+
+    Native duplex can finish several responses while the user is still
+    streaming.  Under the ``ack_only`` playback policy, acknowledging those
+    responses after the final input commit is intentionally rejected: it could
+    append an old assistant message after the newly committed user message.
+    Checkpoint playback first so the server commits each assistant item at its
+    original history position.  A second acknowledgement after the commit can
+    then safely advance response-local playback that drained in the meantime.
+    """
+    await client.acknowledge_playback()
+    commit_sent_at_s = time.monotonic()
+    await client.commit()
+    return commit_sent_at_s
+
+
 def _event_count_after(
     events: list[dict[str, object]],
     event_type: str,
@@ -448,8 +465,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
             chunk_period_ms=_chunk_period_ms(client.events.events),
         )
         wait_for_post_commit_decision = False
-        commit_sent_at_s = time.monotonic()
-        await client.commit()
+        commit_sent_at_s = await _commit_input_after_playback_checkpoint(client)
         wait_error: str | None = None
         committed_index: int | None = None
         post_commit_decision: str | None = None

tests/entrypoints/openai_api/test_duplex_handler.py

@@ -4425,6 +4425,167 @@ async def test_playback_ack_rejects_response_after_followup_user_commit():
     )
 
 
+@pytest.mark.asyncio
+async def test_precommit_zero_audio_ack_reserves_history_for_response_that_finishes_after_commit():
+    handler = OmniDuplexSessionHandler(
+        chat_service=FakeChatService(FakeEngineClient()),
+        config_timeout_s=0.1,
+        idle_timeout_s=1,
+    )
+    session = DuplexSession(
+        session_id="sid-zero-audio-checkpoint",
+        config=DuplexSessionConfig(playback_commit_policy="ack_only"),
+    )
+    session.mark_user_input_activity()
+    response_id = session.begin_response()
+    item_id = f"item_{response_id}"
+    ws = TimedWebSocket()
+
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 0,
+            "committed_ms": 0,
+        },
+        ws.send_json,
+    )
+    session.commit_native_audio_input(transcript="later user input")
+    session.append_assistant_text("hello world")
+    session.mark_audio_sent(1000, text_chars=len("hello world"))
+    committed_message = session.end_response(commit_text=True, preserve_request=True)
+    session.register_history_item(item_id, committed_message)
+
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 1000,
+            "committed_ms": 1000,
+        },
+        ws.send_json,
+    )
+
+    assert not any(message.get("type") == "error" for message in ws.sent)
+    assert session.history == (
+        {"role": "assistant", "content": "hello world"},
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "audio_url",
+                    "audio_url": {"url": "native-duplex:input-audio"},
+                    "transcript": "later user input",
+                }
+            ],
+            "transcript": "later user input",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_precommit_partial_ack_extends_history_after_response_finishes():
+    handler = OmniDuplexSessionHandler(
+        chat_service=FakeChatService(FakeEngineClient()),
+        config_timeout_s=0.1,
+        idle_timeout_s=1,
+    )
+    session = DuplexSession(
+        session_id="sid-partial-checkpoint",
+        config=DuplexSessionConfig(playback_commit_policy="ack_only"),
+    )
+    session.mark_user_input_activity()
+    response_id = session.begin_response()
+    item_id = f"item_{response_id}"
+    session.append_assistant_text("hello world")
+    session.mark_audio_sent(1000, text_chars=len("hello world"))
+    ws = TimedWebSocket()
+
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 500,
+            "committed_ms": 500,
+        },
+        ws.send_json,
+    )
+    session.commit_native_audio_input(transcript="later user input")
+    committed_message = session.end_response(commit_text=True, preserve_request=True)
+    session.register_history_item(item_id, committed_message)
+
+    assert session.history[0] == {"role": "assistant", "content": "hello"}
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 1000,
+            "committed_ms": 1000,
+        },
+        ws.send_json,
+    )
+
+    assert not any(message.get("type") == "error" for message in ws.sent)
+    assert session.history[0] == {"role": "assistant", "content": "hello world"}
+    assert session.history[1]["role"] == "user"
+
+
+@pytest.mark.asyncio
+async def test_explicit_truncate_seals_snapshot_against_later_playback_ack():
+    handler = OmniDuplexSessionHandler(
+        chat_service=FakeChatService(FakeEngineClient()),
+        config_timeout_s=0.1,
+        idle_timeout_s=1,
+    )
+    session = DuplexSession(
+        session_id="sid-hard-playback-truncate",
+        config=DuplexSessionConfig(playback_commit_policy="ack_only"),
+    )
+    response_id = session.begin_response()
+    item_id = f"item_{response_id}"
+    session.append_assistant_text("hello world")
+    session.mark_audio_sent(1000, text_chars=len("hello world"))
+    session.end_response(commit_text=False, preserve_request=True)
+    ws = TimedWebSocket()
+
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 500,
+            "committed_ms": 500,
+            "truncate": True,
+        },
+        ws.send_json,
+    )
+    assert session.history == ({"role": "assistant", "content": "hello"},)
+
+    await handler._handle_playback_ack(
+        session,
+        {
+            "type": "playback.ack",
+            "response_id": response_id,
+            "item_id": item_id,
+            "played_ms": 1000,
+            "committed_ms": 1000,
+        },
+        ws.send_json,
+    )
+
+    assert not any(message.get("type") == "error" for message in ws.sent)
+    assert session.history == ({"role": "assistant", "content": "hello"},)
+
+
 @pytest.mark.asyncio
 async def test_cancel_does_not_append_old_assistant_after_followup_user():
     handler = OmniDuplexSessionHandler(

tests/examples/test_minicpmo_realtime_duplex_simple_demo.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
+import asyncio
 import base64
 import importlib.util
 import sys
@@ -172,6 +173,29 @@ def test_open_streaming_response_requires_post_commit_drain():
     )
 
 
+def test_final_input_commit_checkpoints_playback_first():
+    demo = _load_demo_module()
+    client = demo.RealtimeDuplexClient("ws://unused")
+    client.events.add(
+        {
+            "type": "response.created",
+            "response": {"id": "resp-zero-audio"},
+        }
+    )
+    calls = []
+
+    async def send(event):
+        calls.append(event)
+
+    client.send = send
+    commit_sent_at_s = asyncio.run(demo._commit_input_after_playback_checkpoint(client))
+
+    assert [event["type"] for event in calls] == ["playback.ack", "input_audio_buffer.commit"]
+    assert calls[0]["response_id"] == "resp-zero-audio"
+    assert calls[0]["played_ms"] == 0
+    assert isinstance(commit_sent_at_s, float)
+
+
 def _asymmetric_image():
     from PIL import Image
 

vllm_omni/experimental/fullduplex/client.py

@@ -269,6 +269,11 @@ class RealtimeEventCollector:
             }
         )
 
+    def response_is_done(self, response_id: str) -> bool:
+        return any(
+            event.get("type") == "response.done" and self.response_id(event) == response_id for event in self.events
+        )
+
     def errors(self) -> list[dict[str, object]]:
         return [event for event in self.events if event.get("type") == "error"]
 
@@ -673,7 +678,7 @@ class RealtimeDuplexClient:
     async def acknowledge_playback(self) -> None:
         for response_id in self.events.response_ids:
             pcm16 = self.events.audio_bytes(response_id)
-            if not pcm16:
+            if not pcm16 and self.events.response_is_done(response_id):
                 continue
             played_ms = len(pcm16) * 1000 // (self.events.output_sample_rate_hz * PCM16_BYTES_PER_SAMPLE)
             await self.send_playback_ack(response_id, played_ms)

vllm_omni/experimental/fullduplex/openai/protocol.py

@@ -412,11 +412,13 @@ class PlaybackLedger:
 class ConversationHistory:
     messages: list[dict[str, object]] = field(default_factory=list)
     item_ids: dict[str, dict[str, object]] = field(default_factory=dict)
+    history_item_placeholders: dict[str, dict[str, object]] = field(default_factory=dict)
     item_audio_text_marks: dict[str, list[DuplexAssistantAudioTextMark]] = field(default_factory=dict)
     pending_item_ids: dict[str, dict[str, object]] = field(default_factory=dict)
     pending_item_audio_text_marks: dict[str, list[DuplexAssistantAudioTextMark]] = field(default_factory=dict)
     pending_item_input_commit_seqs: dict[str, int] = field(default_factory=dict)
     pending_truncations_ms: dict[str, int] = field(default_factory=dict)
+    hard_truncations_ms: dict[str, int] = field(default_factory=dict)
     last_assistant_full_message: dict[str, object] | None = None
     last_assistant_audio_text_marks: list[DuplexAssistantAudioTextMark] = field(default_factory=list)
     assistant_response_snapshots: dict[str, tuple[dict[str, object], tuple[DuplexAssistantAudioTextMark, ...], int]] = (
@@ -471,7 +473,8 @@ class DuplexSession:
 
     @property
     def history(self) -> tuple[dict[str, object], ...]:
-        return tuple(dict(message) for message in self._conversation.messages)
+        placeholders = {id(message) for message in self._conversation.history_item_placeholders.values()}
+        return tuple(dict(message) for message in self._conversation.messages if id(message) not in placeholders)
 
     @property
     def active_request_id(self) -> str | None:
@@ -521,6 +524,11 @@ class DuplexSession:
         return response_id in self._conversation.assistant_response_snapshots
 
     def playback_ack_is_too_late(self, response_id: str, item_id: str) -> bool:
+        # An earlier ACK reserved this item's exact history position.  Later
+        # ACKs update that slot in place, so they cannot append an old
+        # assistant turn after a newer user input.
+        if item_id in self._conversation.item_ids or item_id in self._conversation.history_item_placeholders:
+            return False
         input_commit_seq = self._conversation.pending_item_input_commit_seqs.get(item_id)
         if input_commit_seq is None:
             snapshot = self._conversation.assistant_response_snapshots.get(response_id)
@@ -530,6 +538,56 @@ class DuplexSession:
             input_commit_seq = self._response.active_response_input_commit_seq
         return input_commit_seq is not None and self.input_commit_seq > input_commit_seq
 
+    def reserve_history_item(self, item_id: str) -> None:
+        """Reserve the current history position for a response-owned item."""
+        if item_id in self._conversation.item_ids or item_id in self._conversation.history_item_placeholders:
+            return
+        placeholder: dict[str, object] = {}
+        self._conversation.history_item_placeholders[item_id] = placeholder
+        self._conversation.messages.append(placeholder)
+
+    def _store_history_item_message(
+        self,
+        item_id: str,
+        message: dict[str, object],
+    ) -> dict[str, object]:
+        """Materialize or update a response item without changing its order."""
+        existing = self._conversation.item_ids.get(item_id)
+        if existing is not None:
+            if existing is not message:
+                existing.clear()
+                existing.update(message)
+            return existing
+
+        placeholder = self._conversation.history_item_placeholders.pop(item_id, None)
+        if placeholder is not None:
+            for index, candidate in enumerate(self._conversation.messages):
+                if candidate is placeholder:
+                    self._conversation.messages[index] = message
+                    break
+            else:
+                self._conversation.messages.append(message)
+        elif not any(candidate is message for candidate in self._conversation.messages):
+            self._conversation.messages.append(message)
+        self._conversation.item_ids[item_id] = message
+        return message
+
+    def _discard_history_item_placeholder(self, item_id: str) -> bool:
+        placeholder = self._conversation.history_item_placeholders.pop(item_id, None)
+        if placeholder is None:
+            return False
+        self._conversation.messages = [
+            candidate for candidate in self._conversation.messages if candidate is not placeholder
+        ]
+        return True
+
+    def release_response_history_snapshot(self, response_id: str | None) -> None:
+        """Release a final response snapshot after its playback fully commits."""
+        if response_id is None or response_id == self.active_response_id:
+            return
+        self._conversation.assistant_response_snapshots.pop(response_id, None)
+        self._conversation.hard_truncations_ms.pop(f"item_{response_id}", None)
+
     @property
     def playback(self) -> DuplexPlaybackView:
         return self._playback.current.snapshot()
@@ -997,7 +1055,11 @@ class DuplexSession:
             committed_text = ""
         if commit_text and committed_text:
             message = {"role": "assistant", "content": committed_text}
-            self._conversation.messages.append(message)
+            item_id = f"item_{response_id}" if response_id is not None else None
+            if item_id is not None and item_id in self._conversation.history_item_placeholders:
+                self._store_history_item_message(item_id, message)
+            else:
+                self._conversation.messages.append(message)
         effective_playback_policy = playback_commit_policy or self.config.playback_commit_policy
         if (
             response_id is not None
@@ -1006,6 +1068,8 @@ class DuplexSession:
             and effective_playback_policy == DuplexPlaybackCommitPolicy.ACK_ONLY.value
         ):
             self.register_history_item(f"item_{response_id}", None)
+        elif response_id is not None and not assistant_text:
+            self._discard_history_item_placeholder(f"item_{response_id}")
         self._response.assistant_text_buffer.clear()
         if not preserve_request:
             self._response.active_request_id = None
@@ -1034,7 +1098,7 @@ class DuplexSession:
             self._conversation.pending_item_input_commit_seqs[item_id] = response_snapshot[2]
             if audio_text_marks:
                 self._conversation.pending_item_audio_text_marks[item_id] = list(copy.deepcopy(audio_text_marks))
-            pending_audio_ms = self._conversation.pending_truncations_ms.pop(item_id, None)
+            pending_audio_ms = self._conversation.pending_truncations_ms.get(item_id)
             if pending_audio_ms is not None:
                 self.truncate_history_item(
                     item_id,
@@ -1042,30 +1106,11 @@ class DuplexSession:
                     playback=self._playback_cursor_for_item_id(item_id),
                 )
             return
-        pending_audio_ms = self._conversation.pending_truncations_ms.pop(item_id, None)
-        if pending_audio_ms is not None:
-            self._truncate_message_to_audio_ms(
-                message,
-                audio_end_ms=pending_audio_ms,
-                marks=(
-                    response_audio_text_marks
-                    if response_audio_text_marks is not None
-                    else self._response.assistant_audio_text_marks or self._conversation.last_assistant_audio_text_marks
-                ),
-                playback=self._playback_cursor_for_item_id(item_id),
-            )
-            if self._message_text_len(message) <= 0:
-                try:
-                    self._conversation.messages.remove(message)
-                except ValueError:
-                    pass
-                return
-        self._conversation.item_ids[item_id] = message
+        message = self._store_history_item_message(item_id, message)
+        pending_audio_ms = self._conversation.pending_truncations_ms.get(item_id)
         self._conversation.pending_item_ids.pop(item_id, None)
         self._conversation.pending_item_audio_text_marks.pop(item_id, None)
         self._conversation.pending_item_input_commit_seqs.pop(item_id, None)
-        if response_id is not None:
-            self._conversation.assistant_response_snapshots.pop(response_id, None)
         if message.get("role") == "assistant":
             marks = (
                 response_audio_text_marks
@@ -1074,6 +1119,12 @@ class DuplexSession:
             )
             if marks:
                 self._conversation.item_audio_text_marks[item_id] = list(marks)
+        if pending_audio_ms is not None:
+            self.truncate_history_item(
+                item_id,
+                audio_end_ms=pending_audio_ms,
+                playback=self._playback_cursor_for_item_id(item_id),
+            )
 
     def delete_history_item(self, item_id: str) -> bool:
         response_id = item_id.removeprefix("item_") if item_id.startswith("item_") else None
@@ -1083,10 +1134,12 @@ class DuplexSession:
         self._conversation.pending_item_audio_text_marks.pop(item_id, None)
         self._conversation.pending_item_input_commit_seqs.pop(item_id, None)
         self._conversation.pending_truncations_ms.pop(item_id, None)
+        self._conversation.hard_truncations_ms.pop(item_id, None)
+        removed_placeholder = self._discard_history_item_placeholder(item_id)
         if response_id is not None:
             self._conversation.assistant_response_snapshots.pop(response_id, None)
         if message is None:
-            return pending is not None
+            return pending is not None or removed_placeholder
         try:
             self._conversation.messages.remove(message)
         except ValueError:
@@ -1099,8 +1152,43 @@ class DuplexSession:
         *,
         audio_end_ms: int,
         playback: DuplexPlaybackCursor | DuplexPlaybackView | None = None,
+        hard: bool = False,
     ) -> bool:
         playback = playback or self._playback_cursor_for_item_id(item_id)
+        audio_end_ms = max(0, int(audio_end_ms))
+        if hard:
+            previous_cap_ms = self._conversation.hard_truncations_ms.get(item_id)
+            self._conversation.hard_truncations_ms[item_id] = (
+                audio_end_ms if previous_cap_ms is None else min(previous_cap_ms, audio_end_ms)
+            )
+        hard_cap_ms = self._conversation.hard_truncations_ms.get(item_id)
+        if hard_cap_ms is not None:
+            audio_end_ms = min(audio_end_ms, hard_cap_ms)
+        response_id = item_id.removeprefix("item_") if item_id.startswith("item_") else None
+        response_snapshot = (
+            self._conversation.assistant_response_snapshots.get(response_id) if response_id is not None else None
+        )
+        if response_snapshot is not None:
+            full_message, full_marks, _ = response_snapshot
+            message = copy.deepcopy(full_message)
+            changed = self._truncate_message_to_audio_ms(
+                message,
+                audio_end_ms=audio_end_ms,
+                marks=list(full_marks),
+                playback=playback,
+            )
+            if not changed or self._message_text_len(message) <= 0:
+                self._conversation.pending_truncations_ms[item_id] = max(0, int(audio_end_ms))
+                return False
+            self._store_history_item_message(item_id, message)
+            if full_marks:
+                self._conversation.item_audio_text_marks[item_id] = list(copy.deepcopy(full_marks))
+            self._conversation.pending_item_ids.pop(item_id, None)
+            self._conversation.pending_item_audio_text_marks.pop(item_id, None)
+            self._conversation.pending_item_input_commit_seqs.pop(item_id, None)
+            self._conversation.pending_truncations_ms.pop(item_id, None)
+            return True
+
         message = self._conversation.item_ids.get(item_id)
         if message is None:
             pending = self._conversation.pending_item_ids.get(item_id)

vllm_omni/experimental/fullduplex/openai/serving.py

@@ -1709,6 +1709,10 @@ class OmniDuplexSessionHandler(
                     }
                 )
                 return
+            # Reserve the response's current history position before any later
+            # input commit can append a user item.  A 0 ms ACK is intentional:
+            # the response may be active but have no audio delta yet.
+            session.reserve_history_item(item_id)
         elif item_id is not None:
             await send_json(
                 {
@@ -1720,7 +1724,8 @@ class OmniDuplexSessionHandler(
                 }
             )
             return
-        if event.get("truncate") is True:
+        hard_truncate = event.get("truncate") is True
+        if hard_truncate:
             playback = session.acknowledge_playback(
                 int(played_ms),
                 committed_cursor,
@@ -1752,6 +1757,7 @@ class OmniDuplexSessionHandler(
                 item_id,
                 audio_end_ms=committed_cursor,
                 playback=playback,
+                hard=hard_truncate,
             )
         elif session.pending_history_item_ids:
             # A plain playback ack has no OpenAI item id. Commit the only
@@ -1764,6 +1770,7 @@ class OmniDuplexSessionHandler(
                     item_id,
                     audio_end_ms=committed_cursor,
                     playback=playback,
+                    hard=hard_truncate,
                 )
         elif session.active_response_id is not None:
             item_id = f"item_{session.active_response_id}"
@@ -1771,6 +1778,7 @@ class OmniDuplexSessionHandler(
                 item_id,
                 audio_end_ms=committed_cursor,
                 playback=playback,
+                hard=hard_truncate,
             )
         elif session.last_assistant_full_message is not None:
             if item_id is None and session.history_item_ids:
@@ -1786,6 +1794,7 @@ class OmniDuplexSessionHandler(
                     item_id,
                     audio_end_ms=committed_cursor,
                     playback=playback,
+                    hard=hard_truncate,
                 )
         await send_json(
             {
@@ -1802,6 +1811,7 @@ class OmniDuplexSessionHandler(
         )
         if committed_history and committed_cursor >= max(playback.sent_ms, playback.generated_ms):
             session.release_response_playback(response_id)
+            session.release_response_history_snapshot(response_id)
 
     async def _cancel_active_response(
         self,

vllm_omni/experimental/fullduplex/openai/session_runner.py

@@ -1261,6 +1261,7 @@ class DuplexSessionRunnerMixin:
                                 session.truncate_history_item(
                                     item_id,
                                     audio_end_ms=int(audio_end_ms) if isinstance(audio_end_ms, int | float) else 0,
+                                    hard=True,
                                 )
                                 if isinstance(item_id, str)
                                 else False