From 262ff146935553d444702b583874da461b50c079 Mon Sep 17 00:00:00 2001 From: NATURE Date: Thu, 27 Aug 2026 19:51:47 +0800 Subject: [PATCH] [Bugfix][MiniCPM-o] Bound native auto-response continuation (#6630) Signed-off-by: natureofnature Co-authored-by: amy-why-3459 --- .../openai_api/test_duplex_handler.py | 102 +++++++++++++++++- .../fullduplex/openai/runtime_bridge.py | 85 +++++++++++++-- 2 files changed, 176 insertions(+), 11 deletions(-) diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index 6a2b9d224..7baced741 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -3148,7 +3148,7 @@ async def test_minicpmo_pre_response_continuation_drops_after_model_turn_ends(): @pytest.mark.asyncio -async def test_minicpmo_auto_response_continuation_has_no_semantic_unit_cap(): +async def test_minicpmo_auto_response_continuation_stops_at_large_safety_boundary(): request_id = "duplex-sid-long-response-e0-stage0" engine = FakeEngineClient() handler = OmniDuplexSessionHandler(chat_service=FakeChatService(engine)) @@ -3164,7 +3164,7 @@ async def test_minicpmo_auto_response_continuation_has_no_semantic_unit_cap(): _install_direct_silence_scheduler(handler, session) ws = TimedWebSocket() - for _ in range(9): + for _ in range(handler._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS): await handler._maybe_continue_native_response( ws.send_json, session=session, @@ -3172,9 +3172,103 @@ async def test_minicpmo_auto_response_continuation_has_no_semantic_unit_cap(): ) await asyncio.sleep(0.02) - assert len(engine.appended) == 9 + response_id = session.active_response_id + assert response_id is not None + assert len(engine.appended) == handler._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS assert all(payload["duplex_turn_id"] == 0 for _, _, payload, _ in engine.appended) - assert all("force_speak" not in payload for _, _, payload, _ in engine.appended) + assert all(not {"force_speak", "force_listen"} & payload.keys() for _, _, payload, _ in engine.appended[:-1]) + assert engine.appended[-1][2]["force_listen"] is True + assert handler._native_response_continuations_remaining(session, response_id) is False + + await handler._maybe_continue_native_response( + ws.send_json, + session=session, + expected_epoch=session.epoch, + ) + + assert len(engine.appended) == handler._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS + assert ws.sent_types()[-2:] == ["response.listen", "response.done"] + assert session.active_response_id is None + assert session.active_request_id == request_id + + +@pytest.mark.asyncio +async def test_minicpmo_auto_response_boundary_listen_closes_response(): + request_id = "duplex-sid-boundary-listen-e0-stage0" + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler(chat_service=FakeChatService(engine)) + session = DuplexSession( + session_id="sid-boundary-listen", + config=DuplexSessionConfig(extra_body={"auto_response": True}), + ) + session.capabilities = DuplexCapabilities.minicpmo45_native() + response_id = session.begin_response(turn_id=0) + session.bind_request(request_id) + native = handler._minicpmo_session_state(session) + native.continuation_owner_id = f"response:{response_id}" + native.continuation_units = handler._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS + _install_direct_silence_scheduler(handler, session) + ws = TimedWebSocket() + + close_reason, emitted = await handler._send_one_native_duplex_event( + ws.send_json, + { + "supported": True, + "stage_role": "llm", + "is_listen": True, + "model_listen": True, + "data_plane_request_id": request_id, + "end_of_turn": False, + "uses_model_runner_scheduler": True, + "runner_kv_backed": True, + "runtime_impl": "scheduler_data_plane", + }, + session=session, + expected_epoch=session.epoch, + ) + + assert close_reason is None + assert emitted is True + assert engine.appended == [] + assert ws.sent_types() == ["response.listen", "response.done"] + assert session.active_response_id is None + assert session.active_request_id == request_id + assert native.continuation_owner_id is None + + +@pytest.mark.asyncio +async def test_minicpmo_auto_response_boundary_does_not_close_new_epoch_response(): + request_id = "duplex-sid-boundary-race-e0-stage0" + handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient())) + session = DuplexSession( + session_id="sid-boundary-race", + config=DuplexSessionConfig(extra_body={"auto_response": True}), + ) + session.capabilities = DuplexCapabilities.minicpmo45_native() + old_response_id = session.begin_response(turn_id=0) + session.bind_request(request_id) + native = handler._minicpmo_session_state(session) + native.continuation_owner_id = f"response:{old_response_id}" + native.continuation_units = handler._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS + sent: list[dict[str, Any]] = [] + new_response_ids: list[str] = [] + + async def _barge_in_on_listen(payload: dict[str, Any]) -> None: + sent.append(payload) + if payload["type"] == "response.listen": + session.barge_in() + session.bind_request("duplex-sid-boundary-race-e1-stage0") + new_response_ids.append(session.begin_response(turn_id=1)) + + await handler._maybe_continue_native_response( + _barge_in_on_listen, + session=session, + expected_epoch=session.epoch, + ) + + assert [payload["type"] for payload in sent] == ["response.listen"] + assert len(new_response_ids) == 1 + assert session.active_response_id == new_response_ids[0] @pytest.mark.asyncio diff --git a/vllm_omni/experimental/fullduplex/openai/runtime_bridge.py b/vllm_omni/experimental/fullduplex/openai/runtime_bridge.py index 3f1d54b2a..39778a012 100644 --- a/vllm_omni/experimental/fullduplex/openai/runtime_bridge.py +++ b/vllm_omni/experimental/fullduplex/openai/runtime_bridge.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + from __future__ import annotations import asyncio @@ -321,6 +324,12 @@ class NativeRuntimeBridgeMixin: # Other native-duplex adapters may use a different frame-locked unit. _NATIVE_SILENCE_UNIT_PAYLOAD_AUDIO = base64.b64encode(bytes(16000 * 4)).decode("ascii") _NATIVE_RESPONSE_MAX_CONTINUATION_UNITS = 8 + # Each unit is one second of silence. Auto-response needs substantially + # more room than the eight decision beats above, but must still terminate + # well before the 300-second benchmark playback horizon. The final unit + # asks the model to LISTEN; a protocol fallback closes the response if that + # logit hint is ignored. + _NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS = 64 def _native_silence_unit_payload(self) -> dict[str, object]: samples = int(getattr(self._serving_runtime_adapter, "silence_continuation_samples", 16000)) @@ -337,12 +346,54 @@ class NativeRuntimeBridgeMixin: } def _native_response_continuations_remaining(self, session: DuplexSession, response_id: str) -> bool: - if self._session_auto_responds(session): - return True native = self._runtime_session_state(session) owner_id = f"response:{response_id}" count = native.continuation_units if native.continuation_owner_id == owner_id else 0 - return count < self._NATIVE_RESPONSE_MAX_CONTINUATION_UNITS + limit = ( + self._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS + if self._session_auto_responds(session) + else self._NATIVE_RESPONSE_MAX_CONTINUATION_UNITS + ) + return count < limit + + async def _finish_bounded_native_auto_response( + self, + send_json, + *, + session: DuplexSession, + expected_epoch: int | None, + ) -> None: + """Close an auto-response whose final silence unit did not end it.""" + native = self._runtime_session_state(session) + response_id = session.active_response_id + response_epoch = session.epoch + if expected_epoch is not None and response_epoch != expected_epoch: + return + native.clear_continuation() + await send_json( + { + "type": "response.listen", + "session_id": session.session_id, + "epoch": response_epoch, + "reason": "continuation_limit", + "model_listen": False, + } + ) + if response_id is None: + return + if session.epoch != response_epoch or session.active_response_id != response_id: + return + session.end_response(commit_text=False, preserve_request=True) + await send_json( + { + "type": "response.done", + "session_id": session.session_id, + "response_id": response_id, + "epoch": response_epoch, + "committed": False, + "playback": session.playback.as_dict(), + } + ) def _native_silence_continuation_is_stale( self, @@ -407,9 +458,22 @@ class NativeRuntimeBridgeMixin: owner_id = f"model-turn:{expected_model_turn_id}" payload_turn_id = expected_model_turn_id count = native.continuation_units if native.continuation_owner_id == owner_id else 0 - if not auto_response and count >= self._NATIVE_RESPONSE_MAX_CONTINUATION_UNITS: + continuation_limit = ( + self._NATIVE_AUTO_RESPONSE_MAX_CONTINUATION_UNITS + if auto_response + else self._NATIVE_RESPONSE_MAX_CONTINUATION_UNITS + ) + if count >= continuation_limit: + if auto_response: + await self._finish_bounded_native_auto_response( + send_json, + session=session, + expected_epoch=expected_epoch, + ) return payload = self._native_silence_unit_payload() + if auto_response and count + 1 == continuation_limit: + payload["force_listen"] = True payload["duplex_turn_id"] = payload_turn_id scheduler = native.silence_continuation_scheduler @@ -880,11 +944,17 @@ class NativeRuntimeBridgeMixin: and not session.active_response_accepts_model_turn(model_turn_id) ): return close_reason, emitted_response + auto_response = self._session_auto_responds(session) + active_response_id = session.active_response_id + auto_continuations_remaining = active_response_id is None or self._native_response_continuations_remaining( + session, active_response_id + ) non_terminal_auto_listen = ( - self._session_auto_responds(session) - and session.active_response_id is not None + auto_response + and active_response_id is not None and session.active_request_id is not None and native_result.get("end_of_turn") is not True + and auto_continuations_remaining ) if non_terminal_auto_listen: await self._maybe_continue_native_response( @@ -893,7 +963,6 @@ class NativeRuntimeBridgeMixin: expected_epoch=expected_epoch, ) return close_reason, emitted_response - auto_response = self._session_auto_responds(session) if not auto_response and data_plane_request_id == session.active_request_id: session.clear_request() model_listen = native_result.get("model_listen") @@ -932,6 +1001,8 @@ class NativeRuntimeBridgeMixin: expected_epoch=expected_epoch, ) return close_reason, emitted_response + if auto_response: + self._runtime_session_state(session).clear_continuation() session.end_response(commit_text=False, preserve_request=auto_response) await send_json( {