[Bugfix] Skip listen-terminated MiniCPM-o units in Talker handoff

tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py

@@ -341,6 +341,54 @@ def test_native_duplex_ledger_anchors_on_unit_after_reinjected_listens() -> None
     assert torch.equal(torch.tensor(info["hidden_states"]["tts"]), latent[9:11])
 
 
+@pytest.mark.parametrize("skipped_decision", [9304, 9301], ids=["speak", "tts-bos"])
+@pytest.mark.parametrize("current_decision", [9304, 9301], ids=["speak", "tts-bos"])
+def test_native_duplex_handoff_skips_prior_listen_terminated_units(
+    skipped_decision: int, current_decision: int
+) -> None:
+    # A unit may emit speak/turn_eos before choosing listen. The plugin sends
+    # that unit directly to the session, so it never advances the Talker cursor.
+    skipped_ids = [skipped_decision, 9310, 9303, 9303]
+    source, latent = _native_source(
+        prompt_ids=[101, 102],
+        output_ids=[*skipped_ids, current_decision, 21, 22, 9308],
+        row_ids=[101, 102, 9303, current_decision, 21, 22],
+    )
+
+    info = _native_handoff(source)
+
+    assert info["ids"]["tts"] == [21, 22]
+    torch.testing.assert_close(torch.tensor(info["hidden_states"]["tts"]), latent[4:6])
+    assert not info["meta"].get("turn_end", False)
+
+
+def test_native_duplex_listen_trimming_preserves_cumulative_handoff_cursor() -> None:
+    context = SimpleNamespace(bridge_states={"duplex": {"epoch": 3, "model_turn_id": 7}})
+    first_ids = [9304, 9310, 9303, 9304, 21, 22, 9308]
+    first, first_latent = _native_source(
+        prompt_ids=[101, 102],
+        output_ids=first_ids,
+        row_ids=[101, 102, 9304, 21, 22],
+    )
+    second, second_latent = _native_source(
+        prompt_ids=[101, 102],
+        # Another skipped listen must not reset the current turn's cursor.
+        output_ids=[*first_ids, 9303, 9304, 23, 24, 9308],
+        row_ids=[101, 102, 9304, 23, 24],
+    )
+
+    first_info = llm2tts([first], prompt=[{}], _streaming_context=context)[0]["model_intermediate_buffer"]
+    assert first_info["ids"]["tts"] == [21, 22]
+    torch.testing.assert_close(torch.tensor(first_info["hidden_states"]["tts"]), first_latent[3:5])
+    assert llm2tts([first], prompt=[{}], _streaming_context=context) == []
+
+    second_info = llm2tts([second], prompt=[{}], _streaming_context=context)[0]["model_intermediate_buffer"]
+    assert second_info["ids"]["tts"] == [23, 24]
+    torch.testing.assert_close(torch.tensor(second_info["hidden_states"]["tts"]), second_latent[3:5])
+    assert second_info["meta"]["turn_start"] is False
+    assert llm2tts([second], prompt=[{}], _streaming_context=context) == []
+
+
 @pytest.mark.parametrize("previous_terminator_forwarded", [False, True])
 @pytest.mark.parametrize("current_terminator_forwarded", [False, True])
 def test_native_duplex_ledger_prefers_latest_repeat_of_the_unit(

vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py

@@ -869,6 +869,15 @@ def llm2tts(
                 _streaming_context,
                 request_id=str(llm_output.request_id),
             )
+            # Listen-terminated units bypass the Talker, so they do not advance
+            # its handoff cursor. They can include speak/turn_eos (or tts_bos),
+            # not just listen tokens. Start at the current unit; earlier units
+            # may no longer have matching rows in the current ledger.
+            listen_id = special_token_ids.get("listen_token_id")
+            for idx in range(len(llm_output_ids) - 1, -1, -1):
+                if llm_output_ids[idx] == listen_id:
+                    llm_output_ids = llm_output_ids[idx + 1 :]
+                    break
         prompt_token_ids_len = len(prompt_token_ids)
 
         is_native_duplex_handoff = _has_native_duplex_prompt_metadata(mm_output)
@@ -932,27 +941,20 @@ def llm2tts(
                 tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
                 tts_hidden_slice = thinker_hidden_states[tts_bos_idx:end_idx].to(torch.float32).contiguous()
         else:
-            # Mirror the official streaming_generate unit loop. A segment delta
-            # can start with SEVERAL unit decisions (forced/model listens from
-            # chunks that produced no handoff accumulate ahead of the speak),
-            # so skip the leading listen run; the unit's own decisions start
-            # there. The first decision (<|speak|>, a <|tts_bos|> boundary, or
+            # Mirror the official streaming_generate unit loop. The first
+            # decision (<|speak|>, a <|tts_bos|> boundary, or
             # the first text token when no marker is emitted) is fed back but
             # is not part of total_hidden_in_unit; TTS conditions on the
             # tokens after it, up to the chunk terminator. <|turn_eos|> is not
             # a chunk terminator: it is fed like text, its hidden state is
             # handed to the Talker, and anything sampled after it is a stale
             # tail that must not enter TTS.
             out_ids = llm_output_ids
-            listen_id = special_token_ids.get("listen_token_id")
             turn_eos_id = special_token_ids.get("turn_eos_token_id")
-            unit_start = 0
-            while unit_start < len(out_ids) and out_ids[unit_start] == listen_id:
-                unit_start += 1
             if tts_bos_idx is not None:
-                out_start = max(unit_start, tts_bos_idx - prompt_token_ids_len)
-            elif unit_start < len(out_ids) and out_ids[unit_start] not in tts_end_ids:
-                out_start = unit_start + 1
+                out_start = max(0, tts_bos_idx - prompt_token_ids_len)
+            elif out_ids and out_ids[0] not in tts_end_ids:
+                out_start = 1
             else:
                 out_start = None
             if out_start is not None:
@@ -974,9 +976,9 @@ def llm2tts(
                     tts_hidden_slice = _native_duplex_forwarded_hidden_rows(
                         mm_output,
                         thinker_hidden_states,
-                        unit_ids=out_ids[unit_start:],
-                        slice_start=out_start - unit_start,
-                        slice_end=out_end - unit_start,
+                        unit_ids=out_ids,
+                        slice_start=out_start,
+                        slice_end=out_end,
                         request_id=str(llm_output.request_id),
                     )
         handoff_ids = _coerce_token_id_list(tts_token_ids_slice) if tts_token_ids_slice is not None else None