[Core][Model] Unified Full-Duplex MRv2 Support and MiniCPM-o 4.5 Adaptation

.buildkite/common/ci_source_file_dependencies.yml

@@ -303,7 +303,7 @@ source_file_dependencies:
     - tests/examples/test_minicpmo_realtime_web_static.py
     - tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
 
-  omni_minicpmo_4_5_duplex_function:
+  omni_minicpmo_4_5_duplex_function: &minicpmo_4_5_duplex_function
     - *minicpmo_4_5
     - *minicpmo_4_5_duplex
     - tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
@@ -328,6 +328,14 @@ source_file_dependencies:
     - tests/model_executor/models/minicpmo_4_5/duplex/
     - tests/protocol/
 
+  omni_minicpmo_4_5_duplex_mrv2_function:
+    - *minicpmo_4_5_duplex_function
+    - vllm_omni/worker_v2/
+    - vllm_omni/outputs/
+    - vllm_omni/deploy/minicpmo_4_5_duplex_mrv2.yaml
+    - tests/e2e/online_serving/test_minicpmo_4_5_duplex_mrv2.py
+    - tests/worker_v2/
+
   omni_minicpmo_4_5_duplex_perf:
     - *minicpmo_4_5
     - *minicpmo_4_5_duplex

.buildkite/cuda/test-ready.yml

@@ -207,6 +207,13 @@ steps:
             "
         mirror_hardwares: h100_1
 
+      - label: "Omni · MiniCPM-o 4.5 MRv2 Duplex Test"
+        source_file_dependencies: omni_minicpmo_4_5_duplex_mrv2_function
+        timeout_in_minutes: 40
+        commands:
+          - timeout 35m pytest -s -v tests/e2e/online_serving/test_minicpmo_4_5_duplex_mrv2.py -m 'advanced_model and cuda' --run-level advanced_model
+        mirror_hardwares: h100_1
+
       - label: "Omni · MiniCPM-o 4.5 Perf Test"
         timeout_in_minutes: 60
         source_file_dependencies: omni_minicpmo_4_5_perf

docs/design/minicpm_o45_mrv2_performance.md

@@ -17,8 +17,9 @@ ordinary TF32 for CFM DiT GEMMs within Code2Wav forward/capture only
 `input_precision="tf32"` on tiled attention). This is not compensated TF32x3.
 The previous process matmul policy is restored afterwards. cuDNN's TF32 policy
 is independent. HiFT stays IEEE FP32. TF32 changes rounding.
-Duplex keeps the mainline V1 session path. Turn results do not establish duplex
-performance or interruption correctness.
+Duplex defaults to the V1 session path; `minicpmo_4_5_duplex_mrv2.yaml` is the
+opt-in CUDA overlay that runs all three duplex stages on MRv2. Turn results do
+not establish duplex performance or interruption correctness.
 
 The shared Code2Wav backend defaults to the existing fused CFM body on CUDA
 when TF32 is allowed, subject to architecture and FP32 attention-cache
@@ -50,3 +51,15 @@ Disabling TF32 with `token2wav_allow_tf32: false` or
 `cfm_fused_body: true` overrides that choice and may use TF32 attention.
 Slot pooling and row-offset merging remain opt-in. Fusion changes floating-point
 rounding and does not promise identical waveforms.
+
+## Full-duplex MRv2
+
+`minicpmo_4_5_duplex_mrv2.yaml` opts all three CUDA stages into MRv2 and
+otherwise inherits the default duplex profile; non-CUDA platforms keep V1.
+Stage 0 keeps `async_chunk: false` because the Thinker has no async producer:
+as on V1, the orchestrator hands each finished segment to `llm2tts`.
+The Talker reuses the streaming prompt recipe (full attention extends its KV
+prefix until capacity; sliding recompute rebuilds the previous condition,
+confirmed codec ids and current condition) and keeps the 16-frame codec
+penalty history across conditions. A segment flushes only on an EOS sampled
+in the current output, never on `turn_end` alone.

tests/config/test_minicpmo_4_5_duplex_mrv2_deploy.py

@@ -0,0 +1,49 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Resolve the MiniCPM-o three-stage duplex MRv2 deployment contract."""
+
+import pytest
+
+from tests.helpers.stage_config import get_deploy_config_path
+from vllm_omni.config.stage_config import (
+    _apply_platform_overrides,
+    load_deploy_config,
+    merge_pipeline_deploy,
+)
+from vllm_omni.model_executor.models.minicpmo_4_5.pipeline import MINICPMO_4_5_PIPELINE
+from vllm_omni.platforms import current_omni_platform
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+_DEPLOY = "minicpmo_4_5_duplex_mrv2.yaml"
+
+
+def _resolve_cuda_stages(monkeypatch):
+    # Resolve CUDA overrides even when this test runs on a non-CUDA host.
+    monkeypatch.setattr(current_omni_platform, "device_name", "cuda")
+    config = _apply_platform_overrides(load_deploy_config(get_deploy_config_path(_DEPLOY)), platform="cuda")
+    return config, merge_pipeline_deploy(MINICPMO_4_5_PIPELINE, config)
+
+
+def test_duplex_mrv2_cuda_profile(monkeypatch) -> None:
+    config, stages = _resolve_cuda_stages(monkeypatch)
+    assert config.session_mode == "duplex"
+    # All three stages use MRv2; there is no hidden Talker fallback.
+    assert [s.yaml_engine_args["use_v2_model_runner"] for s in stages] == [True, True, True]
+    # Stage 0 duplex preprocessing stays on the synchronous path while the
+    # downstream stages keep streaming chunk transfer.
+    assert [s.yaml_engine_args["async_chunk"] for s in stages] == [False, True, True]
+    # Preserve asynchronous AR scheduling, including the Talker.
+    assert all(s.yaml_engine_args["async_scheduling"] for s in stages[:2])
+    assert stages[2].yaml_engine_args.get("async_scheduling") is not False
+    # Retain the base profile capacity and Talker KV budget.
+    assert [s.yaml_engine_args["max_num_seqs"] for s in stages] == [16, 16, 16]
+    assert stages[1].yaml_engine_args["kv_cache_memory_bytes"] == 4 * 1024**3
+
+
+@pytest.mark.parametrize("platform", ["npu", "xpu", "rocm", "musa"])
+def test_duplex_mrv2_profile_keeps_non_cuda_on_v1(monkeypatch, platform) -> None:
+    monkeypatch.setattr(current_omni_platform, "device_name", platform)
+    config = _apply_platform_overrides(load_deploy_config(get_deploy_config_path(_DEPLOY)), platform=platform)
+    stages = merge_pipeline_deploy(MINICPMO_4_5_PIPELINE, config)
+    assert [s.yaml_engine_args["use_v2_model_runner"] for s in stages] == [False, False, False]

tests/config/test_omni_config.py

@@ -2365,3 +2365,23 @@ def test_async_chunk_auto_disabled_without_processor():
     # since doing so will just raise a ValueError in validation.
     merge_pipeline_deploy(pipeline, deploy)
     assert not deploy.async_chunk
+
+
+def test_mrv2_duplex_session_validation():
+    from vllm_omni.config.stage_config import validate_native_mrv2_session
+
+    ps = StagePipelineConfig(
+        stage_id=1,
+        model_stage="tts",
+        input_sources=(0,),
+        supports_native_mrv2_data_plane=True,
+        supports_duplex_mrv2=True,
+    )
+    # Allowed when supports_duplex_mrv2 is True
+    deploy_duplex = DeployConfig(session_mode="duplex")
+    validate_native_mrv2_session(deploy_duplex, ps, "v2")
+
+    # Rejected when supports_duplex_mrv2 is False
+    ps_no_duplex = replace(ps, supports_duplex_mrv2=False)
+    with pytest.raises(ValueError, match="supports session_mode 'turn' only"):
+        validate_native_mrv2_session(deploy_duplex, ps_no_duplex, "v2")

tests/core/sched/test_omni_ar_scheduler_streaming.py

@@ -1587,3 +1587,52 @@ def fake_schedule(self, _throttle_prefills=False):
 
     assert observed_limits == [7 if native else 8]
     assert sched.max_num_active_reqs == 8
+
+
+@pytest.mark.parametrize(
+    ("attention_type", "prompt_len", "recompute"),
+    [("full_attention", 70, True), ("full_attention", 20, False), ("sliding_recompute", 20, True)],
+)
+def test_mrv2_talker_reuses_confirmed_prompt_window(attention_type, prompt_len, recompute) -> None:
+    sched = _make_scheduler(stage_id=1, session_mode="duplex")
+    sched._native_data_plane = True
+    sched.vllm_config.model_config.custom_process_next_stage_input_func = (
+        "vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.tts2code2wav_async_chunk"
+    )
+    sched.max_model_len = 100
+    sched.vllm_config.model_config.hf_config_name = "tts_config"
+    sched.vllm_config.model_config.max_model_len = 100
+    sched.vllm_config.model_config.hf_config = SimpleNamespace(
+        model_type="minicpmtts",
+        max_position_embeddings=100,
+        attention_type=attention_type,
+    )
+    sched.finish_requests = MagicMock()
+    session = _make_request()
+    session.external_req_id = "native-talker"
+    session.prompt_token_ids = [0] * prompt_len
+    session._all_token_ids.clear()
+    session._all_token_ids.extend(session.prompt_token_ids)
+    session.append_output_token_ids([7, 8, 9, 6561])
+    session.num_prompt_tokens = prompt_len
+    session.num_computed_tokens = prompt_len + 3  # EOS was sampled, not fed back into the model.
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    sched.num_waiting_for_streaming_input = 1
+    session.model_intermediate_buffer = {
+        "native_duplex": True,
+        "meta": {
+            "next_stage_prompt_len": 20,
+            "streaming_condition_seq": 0,
+        },
+    }
+    update = _make_talker_update(20, reserve=10, condition_seq=1)
+    sched._update_request_as_session(session, update)
+    assert session.num_computed_tokens == (0 if recompute else prompt_len + 3)
+    assert session.num_prompt_tokens == 43
+    assert update.model_intermediate_buffer["ids"]["streaming_prompt_previous_codes"] == [7, 8, 9]
+    assert update.model_intermediate_buffer["meta"]["streaming_prompt_recompute"] is recompute
+    sched.finish_requests.assert_not_called()
+    if recompute:
+        sched._free_request_blocks.assert_called_once_with(session)
+    else:
+        sched._free_request_blocks.assert_not_called()

tests/e2e/online_serving/test_minicpmo_4_5_duplex_mrv2.py

@@ -0,0 +1,121 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Real-weight MRv2 duplex concurrency guard, additional to the V1 CI jobs."""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
+    MODEL,
+    multi_session_args,
+    realtime_url,
+    resolve_ref_audio,
+    validated_input_wav,
+)
+from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import run_multi_session
+from tests.e2e.online_serving.test_minicpmo_4_5_duplex import _run_seeded_text_to_audio
+from tests.helpers.mark import hardware_test
+from tests.helpers.runtime import OmniServerParams
+from tests.helpers.stage_config import get_deploy_config_path
+
+pytestmark = [pytest.mark.omni, pytest.mark.advanced_model]
+
+_SERVER = OmniServerParams(
+    model=MODEL,
+    stage_config_path=get_deploy_config_path("minicpmo_4_5_duplex_mrv2.yaml"),
+    use_stage_cli=False,
+    server_args=["--trust-remote-code"],
+)
+
+
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", [pytest.param(_SERVER, id="mrv2-real-weights")], indirect=True)
+@pytest.mark.parametrize("sessions", [1, 2, 4])
+def test_mrv2_duplex_overlapping_turns(omni_server, tmp_path: Path, sessions: int):
+    args = multi_session_args(
+        omni_server=omni_server,
+        input_wav=validated_input_wav(),
+        ref_audio=resolve_ref_audio(),
+        output_dir=tmp_path / f"concurrency_{sessions}",
+        response_required=True,
+    )
+    args.sessions = sessions
+    args.turns = 2
+    args.turn_duration_ms = [args.first_turn_ms] * args.turns
+    args.synchronized_start = True
+    args.disconnect_session_index = None
+    args.takeover_session_index = None
+    result = asyncio.run(run_multi_session(args))
+    assert result["ok"] is True, result
+    assert result["session_count"] == sessions
+    assert result["identity_isolation_ok"] is True
+    assert not result["failures"]
+    for session in result["sessions"]:
+        assert session["audio_delta_count"] > 0, session
+        assert session["done_count"] == args.turns, session
+        assert session["error_count"] == 0, session
+
+
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", [pytest.param(_SERVER, id="mrv2-real-weights")], indirect=True)
+def test_mrv2_duplex_new_session_during_long_response(omni_server):
+    # Unequal prompts and a delayed second session prevent identical clients
+    # from advancing in lockstep and hiding shared condition-state bugs.
+    ref_audio = resolve_ref_audio()
+
+    async def speak(text, delay):
+        await asyncio.sleep(delay)
+        return await _run_seeded_text_to_audio(
+            url=realtime_url(omni_server),
+            model=omni_server.model,
+            ref_audio=ref_audio,
+            text=text,
+            silence_seconds=30.0,
+        )
+
+    async def overlap():
+        return await asyncio.gather(
+            speak("What is the capital of China? Answer in about 40 words.", 0),
+            speak("What is the capital of France? Answer in one short sentence.", 2),
+        )
+
+    first, second = asyncio.run(overlap())
+    for result in (first, second):
+        assert "response.done" in result["event_types"], result
+        assert result["audio_bytes"] > 0, result
+        assert str(result["transcript"]).strip(), result
+    assert first["audio_bytes"] > 96_000, first
+    assert str(first["transcript"]).strip() != str(second["transcript"]).strip()
+
+
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", [pytest.param(_SERVER, id="mrv2-real-weights")], indirect=True)
+def test_mrv2_duplex_server_answers_image_chat_from_the_image(omni_server, openai_client):
+    """The duplex Thinker also serves /v1/chat/completions: its image features must reach the prompt."""
+    import base64
+    import io
+
+    from PIL import Image
+
+    buffer = io.BytesIO()
+    Image.new("RGB", (224, 224), (255, 0, 0)).save(buffer, format="JPEG")
+    image_url = "data:image/jpeg;base64," + base64.b64encode(buffer.getvalue()).decode("ascii")
+    request_config = {
+        "model": omni_server.model,
+        "messages": [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "image_url", "image_url": {"url": image_url}},
+                    {"type": "text", "text": "What color is this image? Answer with one word."},
+                ],
+            }
+        ],
+        "stream": True,
+        "modalities": ["text"],
+        "key_words": {"text": ["red"]},
+        "extra_body": {"chat_template_kwargs": {"enable_thinking": False}},
+    }
+    openai_client.send_omni_request(request_config)

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

@@ -1479,6 +1479,72 @@ def _evict_window_units_for_reanchor(self, state, reanchor):
     assert torch.equal(k_pool, k_pool_after_first)
 
 
+@pytest.mark.parametrize("explicit_old_computed", [False, True])
+def test_mrv2_reanchor_uses_request_slot_without_persistent_input_batch(explicit_old_computed):
+    reanchor = {"delta": 16, "moved_from": 48, "sink_blocks": 2}
+    if explicit_old_computed:
+        reanchor["old_computed_tokens"] = 64
+    pool = torch.randn(8, 1, 16, 16)
+    original = pool.clone()
+    inv_freq = _get_inv_freq(head_dim=16)
+    block_ids = [0, 1, 2, 3]
+    info = {"duplex": {"stage0_reanchor": dict(reanchor)}}
+    runner = SimpleNamespace(
+        req_states=SimpleNamespace(req_id_to_index={"r": 2}, num_computed_tokens_np=np.array([0, 0, 48])),
+        model_state=SimpleNamespace(intermediate_buffer=SimpleNamespace(buffers=[{}, {}, info])),
+        block_tables=SimpleNamespace(
+            num_blocks=SimpleNamespace(np=np.array([[0, 0, 4]])),
+            block_tables=[SimpleNamespace(gpu=torch.tensor([[0] * 4, [0] * 4, block_ids]))],
+        ),
+        device=torch.device("cpu"),
+        cache_config=SimpleNamespace(block_size=16),
+        kv_caches=[pool],
+        _duplex_inv_freq=inv_freq,
+    )
+    scheduler = SimpleNamespace(num_scheduled_tokens={"r": 1}, scheduled_new_reqs=[])
+    MiniCPMO45DuplexWorkerHelper.maybe_apply_reanchor(runner, scheduler_output=scheduler)
+    # Only the retained tail in compacted block 2 is shifted; sink and
+    # unrelated pages stay untouched and scheduler counts are not decremented twice.
+    torch.testing.assert_close(pool[2], rotate_keys(original[2], 16, inv_freq), rtol=1e-5, atol=1e-6)
+    assert torch.equal(pool[:2], original[:2])
+    assert torch.equal(pool[3:], original[3:])
+    assert runner.req_states.num_computed_tokens_np.tolist() == [0, 0, 48]
+    assert "stage0_reanchor" not in info["duplex"]
+
+
+def test_mrv2_reanchor_reads_each_group_block_table_once(monkeypatch):
+    pools = [torch.randn(8, 1, 16, 16) for _ in range(3)]
+    originals = [pool.clone() for pool in pools]
+    inv_freq = _get_inv_freq(head_dim=16)
+    info = {"duplex": {"stage0_reanchor": {"delta": 16, "moved_from": 48, "sink_blocks": 2}}}
+    runner = SimpleNamespace(
+        req_states=SimpleNamespace(req_id_to_index={"r": 0}, num_computed_tokens_np=np.array([48])),
+        model_state=SimpleNamespace(intermediate_buffer=SimpleNamespace(buffers=[info])),
+        block_tables=SimpleNamespace(
+            num_blocks=SimpleNamespace(np=np.array([[4]])),
+            block_tables=[SimpleNamespace(gpu=torch.tensor([[0, 1, 2, 3]]))],
+        ),
+        device=torch.device("cpu"),
+        cache_config=SimpleNamespace(block_size=16),
+        kv_caches=pools,
+        _duplex_inv_freq=inv_freq,
+    )
+    resolve = MiniCPMO45DuplexWorkerHelper.resolve_group_block_ids
+    calls = []
+
+    def counted(runner_, req_id, req_idx, group_idx=0):
+        calls.append(group_idx)
+        return resolve(runner_, req_id, req_idx, group_idx=group_idx)
+
+    monkeypatch.setattr(MiniCPMO45DuplexWorkerHelper, "resolve_group_block_ids", staticmethod(counted))
+    MiniCPMO45DuplexWorkerHelper.maybe_apply_reanchor(
+        runner, scheduler_output=SimpleNamespace(num_scheduled_tokens={"r": 1}, scheduled_new_reqs=[])
+    )
+    assert calls == [0]
+    for pool, original in zip(pools, originals):
+        torch.testing.assert_close(pool[2], rotate_keys(original[2], 16, inv_freq), rtol=1e-5, atol=1e-6)
+
+
 def test_history_slicing_multi_row_embeddings_and_non_aligned_prefix():
     """Verify that history eviction properly handles:
     1. Non-aligned prefix: tokens in [prefix_tokens, sink_end) are kept in the sink!