diff --git a/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py b/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py index 47b0f32c15c..0987030992c 100644 --- a/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py +++ b/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( diff --git a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py index 5855cf07996..8b3a42622d5 100644 --- a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py +++ b/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,11 +941,8 @@ 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 @@ -944,15 +950,11 @@ def llm2tts( # 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