[perf] Minicpm o4.5 optimization (#5228)

Signed-off-by: R2-Y <ruiruyang2@gmail.com>
Co-authored-by: Sy03 <1370724210@qq.com>

tests/model_executor/models/minicpmo_4_5/test_code2wav_batching.py

@@ -0,0 +1,431 @@
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+import torch.nn as nn
+
+from vllm_omni.model_executor.models.minicpmo_4_5.batched_token2wav import (
+    BatchedToken2Wav,
+)
+from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_code2wav import (
+    MiniCPMO45Code2Wav,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _FakeEncoder(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.calls: list[int] = []
+
+    def forward_chunk(self, xs, last_chunk=False, cnn_cache=None, att_cache=None):
+        batch, length, _ = xs.shape
+        self.calls.append(batch)
+        old_length = 0 if att_cache is None else att_cache.shape[3]
+        output = xs[:, : max(1, length - 1)]
+        cnn = xs[:, :1, :].transpose(1, 2).contiguous()
+        marker = xs[:, 0, 0].reshape(1, batch, 1, 1, 1)
+        att = marker.expand(1, batch, 1, old_length + output.shape[1], 1).clone()
+        return output, cnn, att
+
+
+class _FakeBlock:
+    def __init__(self):
+        conv1 = SimpleNamespace(causal_padding=(1, 0))
+        self.conv = SimpleNamespace(
+            in_channels=1,
+            out_channels=1,
+            block=[None, conv1],
+        )
+        self.attn = SimpleNamespace(num_heads=1, head_dim=1)
+
+
+class _FakeEstimator(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.blocks = [_FakeBlock()]
+        self.cfg_batches: list[int] = []
+        self.speaker_order: list[list[float]] = []
+
+    def t_embedder(self, time):
+        return time[:, None]
+
+    def blocks_forward_chunk(
+        self,
+        inputs,
+        time,
+        mask,
+        cnn_cache,
+        att_cache,
+        cnn_out,
+        att_out,
+    ):
+        del time, mask, cnn_cache, att_cache
+        self.cfg_batches.append(inputs.shape[0])
+        self.speaker_order.append(inputs[:, 2, 0].tolist())
+        marker = inputs[:, 1, 0]
+        cnn_out.copy_(marker.reshape(1, -1, 1, 1).expand_as(cnn_out))
+        att_out.copy_(marker.reshape(1, -1, 1, 1, 1).expand_as(att_out))
+        return inputs[:, 1:2]
+
+
+class _FakeDecoder(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.estimator = _FakeEstimator()
+        self.inference_cfg_rate = 0.7
+        self.register_buffer("rand_noise", torch.zeros(1, 1, 100), persistent=False)
+
+
+class _FakeFlow(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.encoder = _FakeEncoder()
+        self.encoder_proj = nn.Identity()
+        self.decoder = _FakeDecoder()
+        self.spk_embed_affine_layer = nn.Identity()
+
+    def input_embedding(self, tokens):
+        return tokens.to(torch.float32).unsqueeze(-1)
+
+
+class _FakeHiFT(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.calls: list[int] = []
+
+    def forward(self, mel, source):
+        del source
+        self.calls.append(mel.shape[0])
+        speech = mel[:, 0].repeat_interleave(3, dim=1)
+        generated_source = speech[:, None]
+        return speech, generated_source
+
+
+class _FakeToken2Wav:
+    def __init__(self):
+        self.flow = _FakeFlow()
+        self.hift = _FakeHiFT()
+        self.float16 = False
+        self.n_timesteps = 2
+        self.mel_cache_len = 1
+        self.source_cache_len = 2
+        self.speech_window = torch.hamming_window(4, periodic=False)
+        self.prompt_calls = 0
+
+    def _prepare_prompt(self, prompt_wav):
+        del prompt_wav
+        self.prompt_calls += 1
+        return (
+            torch.tensor([[5, 6]], dtype=torch.long),
+            torch.tensor([2], dtype=torch.int32),
+            torch.ones(1, 1),
+            torch.ones(1, 4, 1),
+            torch.tensor([4], dtype=torch.int32),
+        )
+
+    def stream(self, *args, **kwargs):
+        raise AssertionError("sequential stream fallback must never be called")
+
+    def __call__(self, *args, **kwargs):
+        raise AssertionError("sequential __call__ fallback must never be called")
+
+
+def _config(minimum: int = 1):
+    return SimpleNamespace(
+        model_config=SimpleNamespace(
+            model="/fake/model",
+            stage_connector_config={
+                "extra": {
+                    "code2wav_min_batch_size": minimum,
+                    "prompt_cache_id": "shared",
+                    "prompt_wav": "/fake/prompt.wav",
+                }
+            },
+        )
+    )
+
+
+def _model():
+    token2wav = _FakeToken2Wav()
+    backend = BatchedToken2Wav(token2wav)
+    model = MiniCPMO45Code2Wav(vllm_config=_config())
+    model.backend = backend
+    return model, token2wav
+
+
+def _info(
+    request_id: str,
+    chunk_seq: int,
+    codes: list[int],
+    *,
+    last_chunk: bool = False,
+    cache_epoch: int = 0,
+):
+    return {
+        "codes": {"audio": torch.tensor(codes, dtype=torch.long)},
+        "meta": {
+            "request_id": request_id,
+            "chunk_seq": chunk_seq,
+            "cache_epoch": cache_epoch,
+            "last_chunk": last_chunk,
+            "prompt_cache_id": "shared",
+        },
+    }
+
+
+def _forward(model, infos, placeholder_counts=None, request_ids=None):
+    placeholder_counts = placeholder_counts or [1] * len(infos)
+    input_ids = torch.zeros(sum(placeholder_counts), dtype=torch.long)
+    return model(
+        input_ids=input_ids,
+        seq_token_counts=placeholder_counts,
+        runtime_additional_information=infos,
+        request_ids=request_ids,
+    )
+
+
+def test_adapter_runs_true_batch_cfg_and_splits_request_caches():
+    token2wav = _FakeToken2Wav()
+    adapter = BatchedToken2Wav(token2wav)
+    prompt = adapter.prepare_prompt("shared", "/fake/prompt.wav")
+    states = adapter.setup_batch(prompt, 2)
+    audios, states = adapter.decode_batch(
+        torch.tensor([[10, 11], [20, 21]]),
+        prompt,
+        states,
+        last_chunk=False,
+    )
+
+    assert token2wav.prompt_calls == 1
+    assert token2wav.flow.encoder.calls == [2, 2]
+    assert token2wav.flow.decoder.estimator.cfg_batches == [4, 4, 4, 4]
+    assert all(order == [1.0, 1.0, 0.0, 0.0] for order in token2wav.flow.decoder.estimator.speaker_order)
+    assert token2wav.hift.calls == [2]
+    assert len(audios) == 2
+    cache0 = states[0].flow_cache["estimator_cnn_cache"]
+    cache1 = states[1].flow_cache["estimator_cnn_cache"]
+    assert cache0.data_ptr() != cache1.data_ptr()
+    assert cache0[0, 0, 0, 0, 0].item() == 10
+    assert cache1[0, 0, 0, 0, 0].item() == 20
+
+
+def test_estimator_cache_stack_split_round_trip_preserves_cfg_rows():
+    token2wav = _FakeToken2Wav()
+    adapter = BatchedToken2Wav(token2wav)
+    prompt = adapter.prepare_prompt("shared", "/fake/prompt.wav")
+    states = adapter.setup_batch(prompt, 2)
+    _, states = adapter.decode_batch(
+        torch.tensor([[10, 11], [20, 21]]),
+        prompt,
+        states,
+        last_chunk=False,
+    )
+
+    stacked = adapter._stack_flow_cache(states)
+    assert stacked["estimator_cnn_cache"].shape[2] == 4
+    assert stacked["estimator_att_cache"].shape[2] == 4
+    restored = adapter._split_flow_cache(stacked, 2)
+    for original, round_tripped in zip(states, restored, strict=True):
+        torch.testing.assert_close(
+            round_tripped["estimator_cnn_cache"],
+            original.flow_cache["estimator_cnn_cache"],
+        )
+        torch.testing.assert_close(
+            round_tripped["estimator_att_cache"],
+            original.flow_cache["estimator_att_cache"],
+        )
+
+
+def test_model_preserves_output_slots_and_prefers_runtime_codes():
+    model, token2wav = _model()
+    output = _forward(
+        model,
+        [_info("a", 0, [10, 11]), _info("b", 0, [20, 21])],
+        placeholder_counts=[3, 1],
+    )
+
+    audios = output.multimodal_outputs["model_outputs"]
+    assert len(audios) == 2
+    assert len(output.multimodal_outputs["sr"]) == 2
+    assert all(sr.item() == 24000 for sr in output.multimodal_outputs["sr"])
+    assert all(audio.dtype == torch.float32 for audio in audios)
+    # Fake CFM uses two Euler steps whose deltas sum to one. Its conditional
+    # row is mu and its unconditional row is zero, so CFG produces 1.7 * mu.
+    torch.testing.assert_close(audios[0][0], torch.tensor(1.7 * 10))
+    torch.testing.assert_close(audios[1][0], torch.tensor(1.7 * 20))
+    assert token2wav.flow.encoder.calls[-1] == 2
+
+
+def test_mixed_final_exact_buckets_keep_order_and_release_only_final_states():
+    model, _ = _model()
+    _forward(
+        model,
+        [_info(name, 0, [index + 1, index + 2]) for index, name in enumerate(("a", "b", "c", "d"))],
+    )
+    output = _forward(
+        model,
+        [
+            _info("a", 1, [11, 12]),
+            _info("c", 1, [31, 32, 33], last_chunk=True),
+            _info("b", 1, [21, 22]),
+            _info("d", 1, [41, 42, 43], last_chunk=True),
+        ],
+    )
+
+    audios = output.multimodal_outputs["model_outputs"]
+    window = torch.hamming_window(4, periodic=False)
+    overlap_scale = 1.7 * (window[0] + window[2])
+    expected = torch.tensor([1, 3, 2, 4], dtype=torch.float32) * overlap_scale
+    actual = torch.stack([audio[0] for audio in audios])
+    torch.testing.assert_close(actual, expected)
+    assert set(model._states) == {"a", "b"}
+
+
+def test_empty_final_sentinel_emits_empty_and_releases_state_without_compute():
+    model, token2wav = _model()
+    _forward(model, [_info("a", 0, [1, 2]), _info("b", 0, [3, 4])])
+    hift_calls = list(token2wav.hift.calls)
+    output = _forward(
+        model,
+        [
+            _info("a", 1, [], last_chunk=True),
+            _info("b", 1, [], last_chunk=True),
+        ],
+    )
+
+    assert [audio.numel() for audio in output.multimodal_outputs["model_outputs"]] == [0, 0]
+    assert model._states == {}
+    assert token2wav.hift.calls == hift_calls
+
+
+def test_empty_final_ignores_generation_scheduler_placeholder_token():
+    model, _ = _model()
+    _forward(model, [_info("a", 0, [1, 2]), _info("b", 0, [3, 4])])
+    infos = [_info("a", 1, [], last_chunk=True), _info("b", 1, [], last_chunk=True)]
+    for info in infos:
+        info.pop("codes")
+        info["meta"]["code_flat_numel"] = 0
+
+    output = _forward(model, infos, placeholder_counts=[1, 1])
+
+    assert [audio.numel() for audio in output.multimodal_outputs["model_outputs"]] == [0, 0]
+    assert model._states == {}
+
+
+@pytest.mark.parametrize(
+    ("info", "reason"),
+    [
+        (_info("a", 0, [1, 2], cache_epoch=-1), "negative_stream_position"),
+        (_info("a", 0, [1, 2]), "stale_or_reordered_chunk"),
+        (_info("a", 2, [1, 2]), "stale_or_reordered_chunk"),
+    ],
+)
+def test_stale_epoch_and_reordered_chunks_are_rejected(info, reason):
+    model, _ = _model()
+    _forward(model, [_info("a", 0, [1, 2]), _info("b", 0, [3, 4])])
+
+    with pytest.raises(RuntimeError, match=reason):
+        _forward(model, [info, _info("b", 1, [3, 4])])
+
+
+def test_singleton_and_mixed_shape_buckets_use_same_batched_backend_without_fallback():
+    model, token2wav = _model()
+    _forward(model, [_info("a", 0, [1, 2]), _info("b", 0, [3, 4])])
+    output = _forward(model, [_info("a", 1, [5, 6]), _info("b", 1, [7, 8, 9])])
+
+    assert len(output.multimodal_outputs["model_outputs"]) == 2
+    # Exact-shape buckets execute independently but both use the same vectorized
+    # adapter; there is no Token2wav.stream/__call__ fallback.
+    assert token2wav.hift.calls[-2:] == [1, 1]
+
+
+def test_backend_failure_does_not_commit_any_request_state(monkeypatch):
+    model, _ = _model()
+    _forward(
+        model,
+        [_info(name, 0, [index + 1, index + 2]) for index, name in enumerate(("a", "b", "c", "d"))],
+    )
+    before = dict(model._states)
+    original = model.backend.decode_batch
+    call_count = 0
+
+    def fail(*args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 2:
+            raise RuntimeError("injected failure")
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(model.backend, "decode_batch", fail)
+    with pytest.raises(RuntimeError, match="injected failure"):
+        _forward(
+            model,
+            [
+                _info("a", 1, [5, 6]),
+                _info("b", 1, [7, 8]),
+                _info("c", 1, [9, 10, 11]),
+                _info("d", 1, [12, 13, 14]),
+            ],
+        )
+    assert call_count == 2
+    assert model._states == before
+
+
+def test_cleanup_and_profile_output_are_aligned():
+    model, _ = _model()
+    _forward(model, [_info("a", 0, [1, 2]), _info("b", 0, [3, 4])])
+    model.on_requests_finished(["a"])
+    assert set(model._states) == {"b"}
+
+    profile = model(
+        input_ids=torch.zeros(5, dtype=torch.long),
+        seq_token_counts=[2, 3],
+    )
+    assert [audio.numel() for audio in profile.multimodal_outputs["model_outputs"]] == [0, 0]
+    assert set(model._states) == {"b"}
+
+
+def test_cleanup_uses_generation_runner_internal_request_ids():
+    model, _ = _model()
+    _forward(
+        model,
+        [_info("external-a", 0, [1, 2]), _info("external-b", 0, [3, 4])],
+        request_ids=["internal-a", "internal-b"],
+    )
+
+    model.on_requests_finished(["internal-a"])
+
+    assert set(model._states) == {"internal-b"}
+
+
+def test_reference_voice_and_duplex_metadata_follow_request_lifecycle():
+    model, _ = _model()
+    first = _info("voice-a", 0, [1, 2])
+    first["codes"]["ref"] = torch.linspace(-0.1, 0.1, 160)
+    first["meta"].update(
+        ref_audio_sr=16000,
+        native_duplex_segment_text="hello",
+        duplex_turn_id=7,
+        duplex_epoch=3,
+    )
+    first["meta"].pop("prompt_cache_id")
+
+    output = _forward(model, [first])
+    prompt_cache_id, prompt_wav = model._owned_prompt_wavs["voice-a"]
+    assert prompt_cache_id.startswith("voice-a:")
+    assert Path(prompt_wav).is_file()
+    assert bytes(output.multimodal_outputs["meta.llm_output_text_utf8"][0].tolist()).decode() == "hello"
+    assert output.multimodal_outputs["meta.duplex_turn_id"][0].item() == 7
+    assert output.multimodal_outputs["meta.duplex_epoch"][0].item() == 3
+
+    final = _info("voice-a", 1, [3, 4], last_chunk=True)
+    final["meta"].pop("prompt_cache_id")
+    output = _forward(model, [final])
+
+    assert output.multimodal_outputs["meta.tts_is_last_chunk"][0].item() is True
+    assert "voice-a" not in model._owned_prompt_wavs
+    assert not Path(prompt_wav).exists()
+    assert (prompt_cache_id, prompt_wav) not in model.backend._prompt_features

tests/model_executor/models/minicpmo_4_5/test_hf_model_path.py

@@ -1,74 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Regression test for HF model-ID resolution and path caching in MiniCPM-o 4.5 TTS."""
-
-from types import SimpleNamespace
-
-import pytest
-
-from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni_tts import (
-    MiniCPMO45OmniTTSForConditionalGeneration,
-)
-
-pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
-
-
-class _FakeMiniCPMTTS:
-    def __init__(self, config, audio_tokenizer):
-        self.config = config
-        self.audio_tokenizer = audio_tokenizer
-        self.emb_text = object()
-        self.projector_semantic = object()
-
-    def generate(self):
-        raise NotImplementedError
-
-
-def test_hf_model_id_path_caching(mocker, tmp_path):
-    """Verify the resolved HF path is cached and reused for default ref audio."""
-    resolved_path = str(tmp_path / "resolved_hf_model")
-    assets_dir = tmp_path / "resolved_hf_model" / "assets"
-    assets_dir.mkdir(parents=True, exist_ok=True)
-    ref_audio_file = assets_dir / "HT_ref_audio.wav"
-    ref_audio_file.write_bytes(b"dummy wav content")
-
-    mock_download = mocker.patch(
-        "vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni_tts.download_weights_from_hf_specific",
-        return_value=resolved_path,
-    )
-
-    mocker.patch(
-        "transformers.dynamic_module_utils.get_class_from_dynamic_module",
-        return_value=_FakeMiniCPMTTS,
-    )
-
-    # Build fake vllm_config with HF model ID
-    model_id = "openbmb/MiniCPM-o-4_5"
-    tts_cfg = SimpleNamespace(
-        audio_bos_token_id=151687,
-        text_eos_token_id=151692,
-        num_audio_tokens=6562,
-        hidden_size=768,
-        normalize_projected_hidden=True,
-        top_p=0.8,
-        top_k=100,
-        repetition_penalty=1.02,
-        attn_implementation="sdpa",
-    )
-    hf_config = SimpleNamespace(tts_config=tts_cfg)
-    vllm_config = SimpleNamespace(
-        model_config=SimpleNamespace(
-            model=model_id,
-            hf_config=hf_config,
-        )
-    )
-
-    tts_model = MiniCPMO45OmniTTSForConditionalGeneration(vllm_config=vllm_config)
-
-    tts_model._lazy_init_tts()
-    tts_model._lazy_init_tts()
-
-    mock_download.assert_called_once_with(model_id, None, ["*"])
-    assert tts_model._model_path == resolved_path
-    assert tts_model._assets_loaded is True
-    assert tts_model._resolve_prompt_wav_path(None, None) == (str(ref_audio_file), None)

tests/model_executor/models/minicpmo_4_5/test_llm2tts.py

@@ -10,7 +10,7 @@ Covers ``vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.llm2t
   - structured model_intermediate_buffer carries the thinker ids and text
   - scheduler prompt tokens follow the selected TTS region or output tokens
   - MiniCPM-o 4.5 TTS region detection on 151703 / 151704 tokens
-  - No TTS markers present -> no ``tts_token_ids`` / ``tts_hidden_states`` keys
+  - plain chat without TTS markers conditions on the generated assistant span
   - prompt arg is normalized to a list and ``multi_modal_data`` is gated by
     ``requires_multimodal_data``
 """
@@ -33,6 +33,7 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 _HIDDEN_DIM = 4
 
 
+@pytest.mark.skip(reason="Legacy fused Talker metadata is replaced by separate Talker and Code2Wav stage payloads.")
 def test_native_duplex_talker_emits_empty_text_metadata_to_clear_previous_segment() -> None:
     class _Talker:
         _ar_last_chunk_flags = [True]
@@ -91,13 +92,15 @@ def _make_thinker_output(
     latent: torch.Tensor | None = None,
     hidden_states: torch.Tensor | None = None,
     multimodal_output: dict[str, object] | None = None,
+    request_multimodal_output: dict[str, object] | None = None,
 ):
     """Construct a minimal mock of a thinker engine output entry.
 
     The real ``llm2tts`` only reads a tight slice of fields:
-      - top-level: ``request_id``, ``prompt_token_ids``, ``outputs[0]``
-      - per-output: ``multimodal_output`` (dict), ``hidden_states`` (opt),
-        ``text``, ``token_ids``
+      - top-level: ``request_id``, ``prompt_token_ids``, optional
+        ``multimodal_output``, and ``outputs[0]``
+      - per-output: fallback ``multimodal_output``, ``hidden_states`` (opt),
+        ``text``, and ``token_ids``
     """
     mm_output = dict(multimodal_output or {})
     if latent is not None:
@@ -109,11 +112,14 @@ def _make_thinker_output(
     )
     if hidden_states is not None:
         output.hidden_states = hidden_states
-    return SimpleNamespace(
+    request_output = SimpleNamespace(
         request_id=request_id,
         prompt_token_ids=prompt_token_ids,
         outputs=[output],
     )
+    if request_multimodal_output is not None:
+        request_output.multimodal_output = request_multimodal_output
+    return request_output
 
 
 class TestInputValidation:
@@ -131,6 +137,19 @@ class TestInputValidation:
 
 
 class TestBasicShape:
+    def test_request_level_multimodal_output_feeds_orchestrator_bridge(self) -> None:
+        hidden = torch.zeros((2, _HIDDEN_DIM))
+        thinker_output = _make_thinker_output(
+            prompt_token_ids=[10],
+            output_token_ids=[20],
+            request_multimodal_output={"latent": hidden},
+        )
+
+        out = llm2tts([thinker_output], prompt=None)
+
+        assert out[0]["prompt_token_ids"] == [0, 0, 0]
+        assert torch.equal(torch.tensor(out[0]["model_intermediate_buffer"]["hidden_states"]["tts"]), hidden[1:])
+
     def test_returns_one_entry_per_input(self) -> None:
         hidden = torch.zeros((3, _HIDDEN_DIM))
         out = llm2tts(
@@ -144,13 +163,13 @@ class TestBasicShape:
         )
         assert len(out) == 2
 
-    def test_talker_scheduler_prompt_uses_output_tokens(self) -> None:
+    def test_talker_scheduler_prompt_matches_condition_length(self) -> None:
         hidden = torch.zeros((2, _HIDDEN_DIM))
         out = llm2tts(
             [_make_thinker_output(prompt_token_ids=[10], output_token_ids=[20], hidden_states=hidden)],
             prompt=None,
         )
-        assert out[0]["prompt_token_ids"] == [20]
+        assert out[0]["prompt_token_ids"] == [0, 0, 0]
         assert "stream_output" not in out[0]["model_intermediate_buffer"]
 
     def test_model_intermediate_buffer_carries_thinker_outputs(self) -> None:
@@ -174,6 +193,24 @@ class TestBasicShape:
         assert buffer["ids"]["output"] == out_ids
         assert buffer["llm_output_text"] == ["hello"]
 
+    def test_strict_talker_receives_canonical_nested_handoff(self) -> None:
+        hidden = torch.arange(5 * _HIDDEN_DIM, dtype=torch.float32).reshape(5, _HIDDEN_DIM)
+        result = llm2tts(
+            [
+                _make_thinker_output(
+                    prompt_token_ids=[10],
+                    output_token_ids=[151703, 20, 21, 151704],
+                    hidden_states=hidden,
+                )
+            ],
+            prompt=None,
+        )[0]
+
+        buffer = result["model_intermediate_buffer"]
+        assert result["prompt_token_ids"] == [0, 0, 0, 0]
+        assert buffer["ids"]["tts"] == [20, 21]
+        assert torch.equal(torch.tensor(buffer["hidden_states"]["tts"]), hidden[2:4])
+
     def test_latent_in_multimodal_output_takes_precedence(self) -> None:
         # When both ``multimodal_output["latent"]`` and ``hidden_states`` are
         # present, the latent payload must win (this is the steady-state path
@@ -241,13 +278,10 @@ class TestTtsRegionDetection:
         assert buffer["ids"]["tts"] == [30, 31]
         assert torch.equal(torch.tensor(buffer["hidden_states"]["tts"]), hidden[3:5])
 
-    def test_no_tts_markers_omits_slice_keys(self) -> None:
-        # If neither marker pair is present, the bridge should NOT populate
-        # ``tts_token_ids`` / ``tts_hidden_states`` — the talker should fall
-        # through to the dummy path.
-        buffer, _ = self._run([10, 11], [20, 21, 22])
-        assert "tts" not in buffer["ids"]
-        assert "tts" not in buffer.get("hidden_states", {})
+    def test_plain_chat_without_tts_markers_uses_assistant_span(self) -> None:
+        buffer, hidden = self._run([10, 11], [20, 21, 22])
+        assert buffer["ids"]["tts"] == [20, 21, 22]
+        assert torch.equal(torch.tensor(buffer["hidden_states"]["tts"]), hidden[2:5])
 
 
 class TestPromptAndMultiModal:

tests/model_executor/models/minicpmo_4_5/test_pipeline.py

@@ -5,8 +5,8 @@
 Covers:
   - pipeline declared in the central registry
   - lazy loader returns the expected ``PipelineConfig``
-  - 2-stage topology (thinker LLM_AR + talker LLM_AR with audio output)
-  - stage 1 routes through ``llm2tts`` custom input processor
+  - split 3-stage topology with no fused compatibility registration
+  - stage 1 routes through ``llm2tts`` and the async Code2Wav producer
   - ``hf_architectures`` covers both the shared ``MiniCPMO`` alias and the
     explicit 4.5 arch
   - ``hf_config_predicate`` selects MiniCPM-o 4.5 only and rejects 2.6
@@ -15,6 +15,7 @@ Covers:
 
 from __future__ import annotations
 
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
@@ -23,20 +24,29 @@ from vllm_omni.config.pipeline_registry import OMNI_PIPELINES
 from vllm_omni.config.stage_config import (
     PipelineConfig,
     StageExecutionType,
+    load_deploy_config,
+    merge_pipeline_deploy,
 )
+from vllm_omni.model_executor.models.registry import _OMNI_MODELS
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
 
 _PIPELINE_KEY = "minicpmo_4_5"
+_DEPLOY_DIR = Path(__file__).resolve().parents[4] / "vllm_omni" / "deploy"
+_DEPLOY_LAYOUTS = {
+    "minicpmo_4_5.yaml": ["0", "0", "0"],
+    "minicpmo_4_5_batching.yaml": ["0", "1", "1"],
+    "minicpmo_4_5_2gpu.yaml": ["0", "1", "1"],
+    "minicpmo_4_5_3gpu.yaml": ["0", "1", "2"],
+    "minicpmo_4_5_8x4090.yaml": ["0,1,2,3", "4", "5"],
+}
 
 
 class TestRegistryDeclaration:
     def test_declared_in_omni_pipelines(self) -> None:
         assert _PIPELINE_KEY in OMNI_PIPELINES
-
-    def test_visible_in_central_registry(self) -> None:
-        assert _PIPELINE_KEY in OMNI_PIPELINES
+        assert "minicpmo_4_5_fused" not in OMNI_PIPELINES
 
     def test_lazy_load_returns_pipeline_config(self) -> None:
         pipeline = OMNI_PIPELINES[_PIPELINE_KEY]
@@ -64,9 +74,9 @@ class TestPipelineTopology:
     def pipeline(self) -> PipelineConfig:
         return OMNI_PIPELINES[_PIPELINE_KEY]
 
-    def test_two_stages(self, pipeline: PipelineConfig) -> None:
-        assert len(pipeline.stages) == 2
-        assert [s.stage_id for s in pipeline.stages] == [0, 1]
+    def test_three_stages(self, pipeline: PipelineConfig) -> None:
+        assert len(pipeline.stages) == 3
+        assert [s.stage_id for s in pipeline.stages] == [0, 1, 2]
 
     def test_topology_validates(self, pipeline: PipelineConfig) -> None:
         # ``validate`` returns a list of structural errors; empty == valid.
@@ -90,11 +100,15 @@ class TestPipelineTopology:
         assert talker.execution_type == StageExecutionType.LLM_AR
         # talker consumes thinker output
         assert talker.input_sources == (0,)
-        assert talker.final_output is True
-        assert talker.final_output_type == "audio"
-        assert talker.engine_output_type == "audio"
+        assert talker.final_output is False
+        assert talker.final_output_type is None
+        assert talker.engine_output_type == "latent"
         # scope KV cache / mrope sizing to talker sub-config
         assert talker.hf_config_name == "tts_config"
+        assert talker.custom_process_next_stage_input_func is None
+        assert talker.async_chunk_process_next_stage_input_func == (
+            "vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.tts2code2wav_async_chunk"
+        )
 
     def test_talker_routes_through_llm2tts(self, pipeline: PipelineConfig) -> None:
         talker = pipeline.get_stage(1)
@@ -106,6 +120,73 @@ class TestPipelineTopology:
             "vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.llm2tts"
         )
 
+    def test_code2wav_stage(self, pipeline: PipelineConfig) -> None:
+        code2wav = pipeline.get_stage(2)
+        assert code2wav is not None
+        assert code2wav.model_stage == "code2wav"
+        assert code2wav.execution_type == StageExecutionType.LLM_GENERATION
+        assert code2wav.input_sources == (1,)
+        assert code2wav.final_output is True
+        assert code2wav.final_output_type == "audio"
+        assert code2wav.engine_output_type == "audio"
+        assert code2wav.model_arch == "MiniCPMO45Code2Wav"
+        assert code2wav.sync_process_input_func is None
+
+
+class TestDeployTopology:
+    @pytest.mark.parametrize(("filename", "devices"), _DEPLOY_LAYOUTS.items())
+    def test_deploy_resolves_three_stage_pipeline(self, filename: str, devices: list[str]) -> None:
+        deploy = load_deploy_config(_DEPLOY_DIR / filename)
+        pipeline = OMNI_PIPELINES[deploy.pipeline]
+        stages = merge_pipeline_deploy(pipeline, deploy)
+
+        assert deploy.pipeline == _PIPELINE_KEY
+        assert deploy.async_chunk is True
+        assert [stage.stage_id for stage in stages] == [0, 1, 2]
+        assert [stage.yaml_runtime["devices"] for stage in stages] == devices
+        assert "additional_config" not in stages[1].yaml_engine_args
+        assert "skip_mm_profiling" not in stages[0].yaml_engine_args
+        assert stages[1].yaml_engine_args["skip_mm_profiling"] is True
+        assert stages[2].yaml_engine_args["skip_mm_profiling"] is True
+        assert stages[1].yaml_extras["output_connectors"]["to_stage_2"] == "connector_of_shared_memory"
+        assert stages[2].yaml_extras["input_connectors"]["from_stage_1"] == "connector_of_shared_memory"
+        connector = deploy.connectors["connector_of_shared_memory"]
+        assert connector["name"] == "SharedMemoryConnector"
+        assert connector["extra"]["codec_chunk_frames"] == 25
+        assert connector["extra"]["codec_left_context_frames"] == 3
+        assert connector["extra"]["connector_get_max_wait_first_chunk"] == 3000
+        assert connector["extra"]["connector_get_max_wait"] == 300
+        if filename == "minicpmo_4_5.yaml":
+            assert [stage.yaml_engine_args["max_num_seqs"] for stage in stages] == [4, 4, 4]
+            assert [stage.yaml_engine_args["gpu_memory_utilization"] for stage in stages] == [
+                0.65,
+                0.15,
+                0.15,
+            ]
+        elif filename in {"minicpmo_4_5_batching.yaml", "minicpmo_4_5_2gpu.yaml"}:
+            assert [stage.yaml_engine_args["gpu_memory_utilization"] for stage in stages] == [
+                0.9,
+                0.55,
+                0.35,
+            ]
+
+    def test_pipeline_exposes_no_full_payload_or_token_placeholder_hooks(self) -> None:
+        pipeline = OMNI_PIPELINES[_PIPELINE_KEY]
+        talker = pipeline.get_stage(1)
+        code2wav = pipeline.get_stage(2)
+        assert talker is not None
+        assert code2wav is not None
+        assert talker.custom_process_next_stage_input_func is None
+        assert code2wav.sync_process_input_func is None
+
+
+def test_code2wav_model_is_lazily_registered() -> None:
+    assert _OMNI_MODELS["MiniCPMO45Code2Wav"] == (
+        "minicpmo_4_5",
+        "minicpmo_4_5_code2wav",
+        "MiniCPMO45Code2Wav",
+    )
+
 
 class TestArchAliases:
     """``hf_architectures`` must cover both the shared and explicit names."""

tests/model_executor/models/minicpmo_4_5/test_talker_batching.py

@@ -0,0 +1,248 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Request-alignment tests for MiniCPM-o 4.5's native Talker."""
+
+from __future__ import annotations
+
+import pytest
+import torch
+import torch.nn as nn
+
+from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni import (
+    MiniCPMO45OmniForConditionalGeneration,
+)
+from vllm_omni.model_executor.models.minicpmo_4_5.minicpmo_4_5_omni_tts import (
+    MiniCPMO45OmniTTSForConditionalGeneration,
+    _max_audio_tokens,
+)
+from vllm_omni.utils.mm_outputs import to_payload_element
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _FakeNativeTalker(nn.Module):
+    has_preprocess = True
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.forward_kwargs = None
+
+    def forward(self, **kwargs):
+        self.forward_kwargs = kwargs
+        return torch.ones(2, 4)
+
+
+def test_wrapper_always_delegates_talker_to_native_ar_path() -> None:
+    model = MiniCPMO45OmniForConditionalGeneration.__new__(MiniCPMO45OmniForConditionalGeneration)
+    nn.Module.__init__(model)
+    model.model_stage = "tts"
+    model.talker = _FakeNativeTalker()
+
+    output = model(
+        input_ids=torch.tensor([1, 2]),
+        positions=torch.arange(2),
+        model_intermediate_buffer=[{"request_id": "req"}],
+    )
+
+    assert output.shape == (2, 4)
+    assert model.talker.forward_kwargs["model_intermediate_buffer"][0]["request_id"] == "req"
+
+
+def _make_talker() -> MiniCPMO45OmniTTSForConditionalGeneration:
+    talker = MiniCPMO45OmniTTSForConditionalGeneration.__new__(MiniCPMO45OmniTTSForConditionalGeneration)
+    nn.Module.__init__(talker)
+    talker._num_audio_tokens = 8
+    talker._batch_stop_logits = None
+    talker._request_generators = {}
+    talker._request_audio_states = {}
+    talker._deferred_cleanup_ids = set()
+    return talker
+
+
+def _routed(output, index: int):
+    return to_payload_element(
+        output.multimodal_outputs,
+        index,
+        index,
+        index + 1,
+        seq_len=2,
+        scheduled_seq_len=2,
+    )
+
+
+@pytest.mark.parametrize(
+    ("condition_tokens", "expected"),
+    [(3, 64), (100, 1000), (1000, 2048)],
+)
+def test_audio_token_limit_scales_with_condition_length(
+    condition_tokens: int,
+    expected: int,
+) -> None:
+    assert _max_audio_tokens(condition_tokens) == expected
+
+
+def test_talker_emits_request_aligned_codec_deltas_after_compaction(mocker) -> None:
+    talker = _make_talker()
+    seen: list[tuple[str, list[float], list[int]]] = []
+
+    def sample(hidden, history, request_id, step):
+        assert step == 0
+        seen.append((request_id, hidden.reshape(-1).tolist(), history.tolist()))
+        return torch.tensor(2 if request_id == "req-a" else 3)
+
+    mocker.patch.object(talker, "_sample_audio_code", side_effect=sample)
+    infos = [
+        {"request_id": "req-a", "audio_codes": {"accumulated": torch.tensor([1])}},
+        {"request_id": "req-b", "audio_codes": {"accumulated": torch.empty(0, dtype=torch.long)}},
+    ]
+
+    output = talker.make_omni_output(
+        torch.tensor([[1.0, 0.0], [2.0, 0.0], [3.0, 0.0]]),
+        model_intermediate_buffer=infos,
+        request_token_spans=[(0, 2), (2, 3)],
+    )
+
+    assert seen == [
+        ("req-a", [2.0, 0.0], [1]),
+        ("req-b", [3.0, 0.0], []),
+    ]
+    assert infos[0]["audio_codes"]["accumulated"].tolist() == [1, 2]
+    assert infos[1]["audio_codes"]["accumulated"].tolist() == [3]
+    assert set(output.multimodal_outputs) == {"codes", "meta"}
+    assert "model_outputs" not in output.multimodal_outputs
+    assert "sr" not in output.multimodal_outputs
+    assert _routed(output, 0)["codes"]["audio"].tolist() == [[2]]
+    assert _routed(output, 1)["codes"]["audio"].tolist() == [[3]]
+    assert _routed(output, 0)["meta"]["finished"].item() is False
+    assert talker.compute_logits(output.text_hidden_states).argmax(dim=-1).tolist() == [0, 0]
+
+
+def test_incomplete_prefill_emits_no_code_and_does_not_advance_state(mocker) -> None:
+    talker = _make_talker()
+    sample = mocker.patch.object(talker, "_sample_audio_code", return_value=torch.tensor(2))
+    infos = [
+        {
+            "request_id": "req-prefill",
+            "audio_state": {"step": 0},
+            "audio_codes": {"accumulated": torch.empty(0, dtype=torch.long)},
+        },
+        {
+            "request_id": "req-decode",
+            "audio_state": {"step": 4},
+            "audio_codes": {"accumulated": torch.tensor([1])},
+        },
+    ]
+
+    output = talker.make_omni_output(
+        torch.tensor([[1.0, 0.0], [2.0, 0.0], [3.0, 0.0]]),
+        model_intermediate_buffer=infos,
+        request_token_spans=[(0, 2), (2, 3)],
+        request_sample_eligible=[False, True],
+    )
+
+    sample.assert_called_once()
+    assert sample.call_args.args[2] == "req-decode"
+    assert infos[0]["audio_state"]["step"] == 0
+    assert infos[0]["audio_codes"]["accumulated"].numel() == 0
+    assert infos[1]["audio_state"]["step"] == 5
+    assert _routed(output, 0)["codes"]["audio"].shape == (0, 1)
+    assert _routed(output, 1)["codes"]["audio"].tolist() == [[2]]
+
+
+def test_eos_is_terminal_once_and_never_enters_codec_history(mocker) -> None:
+    talker = _make_talker()
+    sample = mocker.patch.object(talker, "_sample_audio_code", return_value=torch.tensor(7))
+    info = {
+        "request_id": "req-stop",
+        "audio_state": {"step": 3},
+        "audio_codes": {"accumulated": torch.tensor([4, 5])},
+    }
+
+    first = talker.make_omni_output(
+        torch.ones(1, 2),
+        model_intermediate_buffer=[info],
+        request_token_spans=[(0, 1)],
+    )
+    first_logits = talker.compute_logits(first.text_hidden_states)
+    second = talker.make_omni_output(
+        torch.ones(1, 2),
+        model_intermediate_buffer=[info],
+        request_token_spans=[(0, 1)],
+    )
+
+    sample.assert_called_once()
+    assert info["audio_codes"]["accumulated"].tolist() == [4, 5]
+    assert first.multimodal_outputs["codes"]["audio"][0].shape == (0, 1)
+    assert first.multimodal_outputs["meta"]["finished"][0].item() is True
+    assert second.multimodal_outputs["meta"]["finished"][0].item() is False
+    assert first_logits.argmax(dim=-1).tolist() == [1]
+    assert talker.compute_logits(second.text_hidden_states).argmax(dim=-1).tolist() == [1]
+
+
+def test_max_token_terminal_includes_only_new_codec_delta(mocker) -> None:
+    talker = _make_talker()
+    mocker.patch.object(talker, "_sample_audio_code", return_value=torch.tensor(3))
+    info = {
+        "request_id": "req-limit",
+        "audio_state": {"step": 1, "max_tokens": 2},
+        "audio_codes": {"accumulated": torch.tensor([4, 5])},
+    }
+
+    output = talker.make_omni_output(
+        torch.ones(1, 2),
+        model_intermediate_buffer=[info],
+        request_token_spans=[(0, 1)],
+    )
+
+    assert info["audio_codes"]["accumulated"].tolist() == [4, 5, 3]
+    assert output.multimodal_outputs["codes"]["audio"][0].tolist() == [[3]]
+    assert output.multimodal_outputs["meta"]["finished"][0].item() is True
+    assert talker.compute_logits(output.text_hidden_states).argmax(dim=-1).tolist() == [1]
+
+
+def test_request_local_state_survives_missing_runner_buffer_update(mocker) -> None:
+    talker = _make_talker()
+    mocker.patch.object(talker, "_sample_audio_code", return_value=torch.tensor(3))
+    first_info = {
+        "request_id": "req-local-state",
+        "audio_state": {"step": 1, "max_tokens": 3},
+        "audio_codes": {"accumulated": torch.tensor([4])},
+    }
+
+    talker.make_omni_output(
+        torch.ones(1, 2),
+        model_intermediate_buffer=[first_info],
+        request_token_spans=[(0, 1)],
+    )
+    second = talker.make_omni_output(
+        torch.ones(1, 2),
+        model_intermediate_buffer=[{"request_id": "req-local-state"}],
+        request_token_spans=[(0, 1)],
+    )
+
+    assert second.multimodal_outputs["meta"]["finished"][0].item() is True
+    assert talker._request_audio_states["req-local-state"]["step"] == 3
+
+
+def test_missing_conditioning_fails_clearly() -> None:
+    talker = _make_talker()
+
+    with pytest.raises(ValueError, match="tts_token_ids and tts_hidden_states"):
+        talker.preprocess(
+            torch.tensor([0]),
+            None,
+            _omni_is_prefill=True,
+            request_id="req-invalid",
+        )
+
+
+def test_request_cleanup_evicts_ar_rng_and_decode_state() -> None:
+    talker = _make_talker()
+    talker._request_generators["req-done"] = torch.Generator()
+    talker._request_audio_states["req-done"] = {"step": 1}
+
+    talker.on_requests_finished(["req-done"])
+    talker._flush_deferred_cleanup()
+
+    assert "req-done" not in talker._request_generators
+    assert "req-done" not in talker._request_audio_states