[Bugfix][MiniCPM-o] Bound native auto-response continuation (#6630)

Signed-off-by: natureofnature <wzliu@connect.hku.hk>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

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

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(
                     {