From 705e781b0efc52bea75930ecd584690eafcc204a Mon Sep 17 00:00:00 2001 From: Gao Han Date: Mon, 31 Aug 2026 23:18:29 +0800 Subject: [PATCH] [Bugfix][MiniCPM-o] Checkpoint playback before final input commit (#6821) Signed-off-by: Gao Han --- .../minicpmo/realtime_duplex_demo.py | 20 ++- .../openai_api/test_duplex_handler.py | 161 ++++++++++++++++++ ...st_minicpmo_realtime_duplex_simple_demo.py | 24 +++ vllm_omni/experimental/fullduplex/client.py | 7 +- .../fullduplex/openai/protocol.py | 138 ++++++++++++--- .../experimental/fullduplex/openai/serving.py | 12 +- .../fullduplex/openai/session_runner.py | 1 + 7 files changed, 334 insertions(+), 29 deletions(-) diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py index d1bca7aef..b12c20a56 100644 --- a/examples/online_serving/minicpmo/realtime_duplex_demo.py +++ b/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 diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index b42ace688..c75f1e599 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/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( diff --git a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py b/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py index ca87dce03..dfb7b32b1 100644 --- a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py +++ b/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 diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py index 630d7aef3..7c114f6c8 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/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) diff --git a/vllm_omni/experimental/fullduplex/openai/protocol.py b/vllm_omni/experimental/fullduplex/openai/protocol.py index 45e702d2b..8b3812bc3 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/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) diff --git a/vllm_omni/experimental/fullduplex/openai/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index ce73684c0..0b3bd246c 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/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, diff --git a/vllm_omni/experimental/fullduplex/openai/session_runner.py b/vllm_omni/experimental/fullduplex/openai/session_runner.py index b7694c6a0..e3a8a8cdf 100644 --- a/vllm_omni/experimental/fullduplex/openai/session_runner.py +++ b/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