[Bugfix][MiniCPM-o] Forward a speech unit closed by LISTEN after turn_eos to the Talker (#8227)

Signed-off-by: chickeyton <ngton2014@gmail.com>
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tests/model_executor/models/minicpmo_4_5/duplex/test_turn_eos_listen_unit.py

@@ -0,0 +1,93 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""A unit that closes its speech with ``<|turn_eos|>`` and then LISTEN is speech, not a LISTEN decision.
+
+On silent input the policy forces the unit terminator after ``<|turn_eos|>`` to
+be LISTEN. Treating that unit as a model LISTEN keeps its final speech and
+``<|turn_eos|>`` from the Talker, so the response never ends.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.data_plane import MiniCPMO45DataPlaneSession
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.plugin import MiniCPMO45DuplexPlugin
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+CHUNK_EOS, CHUNK_TTS_EOS, LISTEN, TURN_EOS = 5, 6, 7, 9
+SPECIAL_TOKEN_IDS = {
+    "chunk_eos_token_id": CHUNK_EOS,
+    "chunk_tts_eos_token_id": CHUNK_TTS_EOS,
+    "listen_token_id": LISTEN,
+    "turn_eos_token_id": TURN_EOS,
+}
+TEXT = 100
+
+
+@pytest.mark.parametrize(
+    ("token_ids", "closes_speech"),
+    [
+        ([TEXT, TEXT, TURN_EOS, LISTEN], True),
+        ([TURN_EOS, LISTEN], True),
+        ([LISTEN, LISTEN, TEXT, TURN_EOS, LISTEN], True),
+        ([LISTEN], False),
+        ([TEXT, TEXT, LISTEN], False),
+        # A <|turn_eos|> from an earlier unit does not count.
+        ([TEXT, TURN_EOS, CHUNK_EOS, LISTEN], False),
+        ([TEXT, TURN_EOS, LISTEN, LISTEN], False),
+        ([TEXT, TURN_EOS, CHUNK_EOS], False),
+        ([], False),
+    ],
+)
+def test_speech_unit_closed_by_listen(token_ids: list[int], closes_speech: bool) -> None:
+    assert MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(token_ids, SPECIAL_TOKEN_IDS) is closes_speech
+
+
+def _stage0_output(token_ids: list[int], *, finished: bool = True) -> SimpleNamespace:
+    completion = SimpleNamespace(token_ids=token_ids, stop_reason=token_ids[-1], text="", multimodal_output={})
+    return SimpleNamespace(
+        request_id="duplex-s.c2Vzc2lvbg==.e.0.r.stage0",
+        outputs=[completion],
+        finished=finished,
+        multimodal_output={"special_token_ids": dict(SPECIAL_TOKEN_IDS)},
+    )
+
+
+def _decide(token_ids: list[int]):
+    return MiniCPMO45DuplexPlugin(lambda *args: None).decide_output(
+        stage_id=0,
+        final_stage_id=2,
+        segment_finished=True,
+        segment_token_ids=tuple(token_ids),
+        segment_output_metadata={},
+        output=_stage0_output(token_ids),
+    )
+
+
+def test_plugin_forwards_speech_closed_by_listen_to_the_talker() -> None:
+    assert _decide([TEXT, TEXT, TURN_EOS, LISTEN]) is None
+
+
+def test_plugin_keeps_a_plain_listen_unit_a_listen_decision() -> None:
+    decision = _decide([LISTEN])
+
+    assert decision is not None
+    assert decision.metadata["duplex_native_decision"] == "listen"
+
+
+def _projected_model_listens(token_ids: list[int]) -> list[dict[str, object]]:
+    session = MiniCPMO45DataPlaneSession(lambda *args, **kwargs: "")
+    return [result for result in session.project_output(_stage0_output(token_ids)) if result.get("model_listen")]
+
+
+def test_data_plane_does_not_project_speech_closed_by_listen_as_model_listen() -> None:
+    assert _projected_model_listens([TEXT, TEXT, TURN_EOS, LISTEN]) == []
+
+
+def test_data_plane_projects_a_plain_listen_unit_as_model_listen() -> None:
+    assert len(_projected_model_listens([LISTEN])) == 1

vllm_omni/model_executor/models/minicpmo_4_5/duplex/data_plane.py

@@ -13,6 +13,7 @@ from vllm_omni.engine.duplex.contracts import (
     duplex_resource_request_belongs_to_session,
 )
 from vllm_omni.engine.duplex.plugin import DuplexDataPlane, DuplexDataPlaneContext
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy
 from vllm_omni.outputs.duplex import get_duplex_output_decision
 
 logger = init_logger(__name__)
@@ -638,9 +639,13 @@ def _native_decision(
         return None
     if mm_output.get("duplex_native_decision") == "listen" or mm_output.get("model_listen") is True:
         return "listen"
-    listen_id = _special_token_ids(mm_output).get("listen_token_id")
+    special_token_ids = _special_token_ids(mm_output)
+    listen_id = special_token_ids.get("listen_token_id")
     if listen_id is None:
         return None
+    unit_ids = max(token_ids, _coerce_int_list(getattr(completion, "cumulative_token_ids", None)), key=len)
+    if MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(unit_ids, special_token_ids):
+        return None
     stop_reason = getattr(completion, "stop_reason", None) if completion is not None else None
     if coerce_int(stop_reason) == listen_id:
         return "listen"

vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py

@@ -639,6 +639,14 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
         token_ids = _completion_token_ids(completion) or list(segment_token_ids)
         if _coerce_int(stop_reason) != listen_id and (not token_ids or token_ids[-1] != listen_id):
             return None
+        unit_ids = max(
+            (token_ids, _coerce_int_list(getattr(completion, "cumulative_token_ids", None)), list(segment_token_ids)),
+            key=len,
+        )
+        if MiniCPMO45DuplexPolicy.speech_unit_closed_by_listen(unit_ids, special_token_ids):
+            # The unit's final speech and <|turn_eos|> must reach the Talker,
+            # or the response never ends.
+            return None
 
         metadata = dict(output_metadata)
         for key, value in special_token_ids.items():

vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py

@@ -7,6 +7,8 @@ from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+
     from transformers import PreTrainedTokenizerBase
 
 
@@ -93,6 +95,33 @@ class MiniCPMO45DuplexPolicy:
     DEFAULT_MIN_NEW_SPEAK_TOKENS_BEFORE_CHUNK_BOUNDARY = 8
     REPETITION_HISTORY_SIZE = 512
 
+    @staticmethod
+    def speech_unit_closed_by_listen(token_ids: Sequence[int], special_token_ids: Mapping[str, int]) -> bool:
+        """Whether a unit ending in LISTEN closes speech rather than deciding to listen.
+
+        ``<|turn_eos|>`` ends the turn but not the unit: the model keeps
+        decoding until a unit terminator, and on silent input the policy forces
+        that terminator to be LISTEN. Such a unit still carries the turn's final
+        speech and ``<|turn_eos|>`` for the Talker, so it is not a LISTEN
+        decision. Only the current unit counts: the scan stops at the previous
+        unit terminator.
+        """
+        listen_id = special_token_ids.get("listen_token_id")
+        turn_eos_id = special_token_ids.get("turn_eos_token_id")
+        if listen_id is None or turn_eos_id is None or not token_ids or token_ids[-1] != listen_id:
+            return False
+        unit_terminators = {
+            listen_id,
+            special_token_ids.get("chunk_eos_token_id"),
+            special_token_ids.get("chunk_tts_eos_token_id"),
+        }
+        for token_id in reversed(token_ids[:-1]):
+            if token_id == turn_eos_id:
+                return True
+            if token_id in unit_terminators:
+                return False
+        return False
+
     @classmethod
     def audio_token_count(cls, sample_count: int) -> int:
         """Audio embedding count for a clip of ``sample_count`` samples.