[Bugfix] Separate MiniCPM-o seeded instructions from explicit TTS

tests/e2e/online_serving/test_minicpmo_4_5_duplex.py

@@ -8,7 +8,10 @@
 import asyncio
 import base64
 import json
+import re
+import time
 from pathlib import Path
+from typing import TypedDict
 
 import pytest
 import websockets
@@ -30,7 +33,7 @@
     run_multi_session,
 )
 from tests.helpers.mark import hardware_test
-from vllm_omni.clients.duplex import build_realtime_url, metric_mean
+from vllm_omni.clients.duplex import DuplexClient, EventCollector, SessionConfig, build_realtime_url, metric_mean
 from vllm_omni.experimental.fullduplex.video_stacking import concat_frames_b64
 
 pytestmark = pytest.mark.omni
@@ -174,6 +177,13 @@ async def receive_outcome() -> dict[str, object]:
         return await asyncio.wait_for(receive_outcome(), timeout=timeout_s)
 
 
+class _SeededTextResult(TypedDict):
+    audio_bytes: int
+    transcript: str
+    output_text: str
+    event_types: list[str]
+
+
 async def _run_seeded_text_to_audio(
     *,
     url: str,
@@ -183,7 +193,7 @@ async def _run_seeded_text_to_audio(
     modalities: tuple[str, ...] = ("audio", "text"),
     silence_seconds: float = 12.0,
     timeout_s: float = 180.0,
-) -> dict[str, object]:
+) -> _SeededTextResult:
     """Speak a seeded text: the duplex route's text-to-speech shape.
 
     A model-native session takes its text once, in the session context
@@ -416,6 +426,68 @@ def test_duplex_seeded_text_to_text_needs_no_reference_voice(omni_server) -> Non
     assert "response.done" in result["event_types"], result["event_types"]
     produced_text = str(result["output_text"]) or str(result["transcript"])
     assert produced_text.strip(), f"text-only session produced nothing: {result['event_types']}"
+    assert re.search(r"\bparis\b", produced_text, re.IGNORECASE), (
+        f"text-only session did not answer the question: {produced_text!r}"
+    )
+
+
+@pytest.mark.advanced_model
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("with_reference", [False, True], ids=["no-reference", "with-reference"])
+def test_duplex_seeded_conditional_watch_remains_silent(omni_server, with_reference: bool) -> None:
+    async def observe() -> tuple[EventCollector, float, float]:
+        config = SessionConfig(
+            modalities=("text",),
+            temperature=0.0,
+            ref_audio=_ref_audio_data_url(str(resolve_ref_audio())) if with_reference else None,
+            extra_body={
+                "force_listen_count": 0,
+                "duplex_initial_user_text": (
+                    "Watch the video and speak only when a red ball appears. Otherwise remain silent."
+                ),
+            },
+        )
+        collector = EventCollector()
+        async with DuplexClient(
+            realtime_url(omni_server),
+            model=omni_server.model,
+            config=config,
+            reconnect=None,
+            heartbeat_interval_s=None,
+            handshake_timeout_s=60.0,
+        ) as client:
+            task = asyncio.create_task(collector.consume(client))
+            await asyncio.sleep(0)
+            started = time.monotonic()
+            try:
+                await client.stream_pcm(bytes(2 * 16_000 * 10), chunk_ms=200, realtime=True)
+                await asyncio.sleep(2.0)
+                observed_until = time.monotonic()
+            finally:
+                await client.close(timeout_s=20.0)
+                await asyncio.wait_for(task, timeout=5.0)
+        return collector, started, observed_until
+
+    collector, started, observed_until = asyncio.run(observe())
+    listen_times = [
+        received - started
+        for event, received in zip(collector.events, collector.event_received_at_s, strict=True)
+        if received <= observed_until
+        and event.get("type") == "response.listen"
+        and isinstance(response := event.get("response"), dict)
+        and isinstance(metadata := response.get("metadata"), dict)
+        and metadata.get("model_listen") is True
+    ]
+    assert not collector.errors(), collector.errors()
+    assert collector.count("session.closed") > 0
+    # A 1035 ms first unit plus steady 1000 ms units fits nine decisions, not ten.
+    assert len(listen_times) >= 9, listen_times
+    assert listen_times[-1] >= 5.0, listen_times
+    assert collector.count("response.speak") == 0
+    assert collector.count("response.output_audio.delta") == 0
+    text_types = {"response.output_text.delta", "response.text.delta", "response.output_audio_transcript.delta"}
+    assert not any(event.get("delta") for event in collector.events if event.get("type") in text_types)
 
 
 @pytest.mark.advanced_model

tests/engine/duplex/test_minicpmo_window_plugin.py

@@ -17,6 +17,50 @@ def _plugin():
     return module.MiniCPMO45DuplexPlugin(lambda *args: None)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("task_type", [None, "Base", "VoiceDesign"])
+@pytest.mark.parametrize("seed", [None, "What is the capital of France?"])
+async def test_initial_text_task_and_context_budget_are_prepared_together(monkeypatch, task_type, seed):
+    tokenizer = SimpleNamespace(
+        all_special_ids=[90],
+        unk_token_id=-1,
+        eos_token_id=99,
+        encode=lambda text, add_special_tokens=False: [ord(char) for char in text],
+        convert_tokens_to_ids=lambda token: {"<|listen|>": 91}.get(token, -1),
+    )
+    monkeypatch.setattr(module, "_load_tokenizer", lambda config: tokenizer)
+    config = DuplexSessionConfig(
+        modalities=("text",),
+        initial_user_text=seed,
+        extra_body={} if task_type is None else {"task_type": task_type},
+    )
+    runtime = await _plugin().prepare_runtime_config(config, model_config=None)
+    is_tts = seed is not None and task_type == "Base"
+    assert (runtime.get("initial_user_text_is_tts") is True) == is_tts
+    suffix = "".join(map(chr, runtime["duplex_window_suffix_token_ids"]))
+    assert suffix.endswith("\n<|im_start|>assistant\n<think>\n\n</think>\n\n<|tts_bos|>") == is_tts
+    assert runtime["duplex_first_append_context_tokens"] == runtime["duplex_window_prefix_tokens"] + len(suffix)
+
+
+def test_initial_text_task_selector_is_server_owned_and_cannot_change():
+    plugin = _plugin()
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError):
+        plugin.validate_client_extra_body({"initial_user_text_is_tts": True})
+    current = {"initial_user_text": "Read me.", "initial_user_text_is_tts": True}
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+        plugin.runtime_config_for_update(DuplexSessionConfig(extra_body={"task_type": "VoiceDesign"}), current)
+    assert exc.value.code == "initial_text_task_update_unsupported"
+    updated = plugin.runtime_config_for_update(DuplexSessionConfig(extra_body={"task_type": "Base"}), current)
+    assert updated["initial_user_text_is_tts"] is True
+    updated = plugin.runtime_config_for_update(DuplexSessionConfig(), current)
+    assert updated["initial_user_text_is_tts"] is True
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+        plugin.runtime_config_for_update(
+            DuplexSessionConfig(extra_body={"task_type": "Base"}), {"initial_user_text": "Read me."}
+        )
+    assert exc.value.code == "initial_text_task_update_unsupported"
+
+
 @pytest.mark.asyncio
 async def test_window_configuration_is_prepared_by_plugin(monkeypatch):
     tokenizer = SimpleNamespace(

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

@@ -9,14 +9,49 @@
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
 
-def test_seed_tts_initial_text_is_part_of_native_duplex_context():
+@pytest.mark.parametrize("has_ref_audio", [False, True])
+@pytest.mark.parametrize(
+    "initial_user_text",
+    [
+        "What is the capital of France?",
+        "Watch the video and only speak when a red ball appears.",
+        "The quick brown fox.",
+    ],
+)
+def test_initial_text_remains_a_completed_user_message_before_native_unit(has_ref_audio, initial_user_text):
     prefix, suffix = MiniCPMO45DuplexPolicy.session_context_texts(
         "Speak exactly.",
-        True,
-        "The quick brown fox.",
+        has_ref_audio,
+        initial_user_text,
+    )
+
+    assert prefix == "<|im_start|>system\nSpeak exactly." + ("\n<|audio_start|>" if has_ref_audio else "")
+    expected_suffix = "<|audio_end|>" if has_ref_audio else ""
+    expected_suffix += f"<|im_end|>\n<|im_start|>user\n{initial_user_text}<|im_end|>"
+    assert suffix == expected_suffix
+    assert (suffix + "<unit>").endswith(f"{initial_user_text}<|im_end|><unit>")
+    assert "<|im_start|>assistant" not in suffix
+
+
+@pytest.mark.parametrize("has_ref_audio", [False, True])
+@pytest.mark.parametrize("initial_user_text", [None, ""])
+@pytest.mark.parametrize("initial_user_text_is_tts", [False, True])
+def test_unseeded_native_duplex_context_is_unchanged(has_ref_audio, initial_user_text, initial_user_text_is_tts):
+    prefix, suffix = MiniCPMO45DuplexPolicy.session_context_texts(
+        None, has_ref_audio, initial_user_text, initial_user_text_is_tts=initial_user_text_is_tts
+    )
+
+    assert prefix == "<|im_start|>system\nStreaming Omni Conversation." + ("\n<|audio_start|>" if has_ref_audio else "")
+    assert suffix == ("<|audio_end|>" if has_ref_audio else "") + "<|im_end|>"
+
+
+@pytest.mark.parametrize("has_ref_audio", [False, True])
+def test_explicit_base_tts_opens_a_tts_response(has_ref_audio):
+    _, suffix = MiniCPMO45DuplexPolicy.session_context_texts(
+        "Speak exactly.", has_ref_audio, "The quick brown fox.", initial_user_text_is_tts=True
     )
 
-    assert prefix == "<|im_start|>system\nSpeak exactly.\n<|audio_start|>"
-    assert suffix == (
-        "<|audio_end|><|im_end|>\n<|im_start|>user\nThe quick brown fox.<|im_end|>\n<|im_start|>assistant\n"
+    assert suffix == ("<|audio_end|>" if has_ref_audio else "") + (
+        "<|im_end|>\n<|im_start|>user\nThe quick brown fox.<|im_end|>"
+        "\n<|im_start|>assistant\n<think>\n\n</think>\n\n<|tts_bos|>"
     )