[Bugfix][MiniCPM-o] Align Daily-Omni offline loading and duplex soft-… (#6095)

Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/benchmarks/test_daily_omni_local_source.py

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Offline Daily-Omni sources: local mirrors must not fall back to the Hub id."""
+
+import json
+import os
+import tarfile
+from pathlib import Path
+
+import pytest
+
+from vllm_omni.benchmarks.data_modules.daily_omni_dataset import (
+    daily_omni_local_qa_json,
+    daily_omni_local_videos_dir,
+    resolve_daily_omni_local_root,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark]
+
+
+def _write_qa_json(root: Path) -> Path:
+    qa = root / "qa.json"
+    qa.write_text(json.dumps([{"Question": "q", "Answer": "A", "video_id": "vid0"}]), encoding="utf-8")
+    return qa
+
+
+def test_resolve_local_root_returns_none_for_hub_id() -> None:
+    assert resolve_daily_omni_local_root("liarliar/Daily-Omni") is None
+    assert resolve_daily_omni_local_root("") is None
+    assert resolve_daily_omni_local_root(None) is None
+
+
+def test_plain_local_mirror(tmp_path: Path) -> None:
+    _write_qa_json(tmp_path)
+    (tmp_path / "Videos" / "vid0").mkdir(parents=True)
+
+    root = resolve_daily_omni_local_root(str(tmp_path))
+    assert root == tmp_path.resolve()
+    assert daily_omni_local_qa_json(root) == (tmp_path / "qa.json").resolve()
+    assert daily_omni_local_videos_dir(root) == root / "Videos"
+
+
+def test_hf_cache_dir_resolves_to_snapshot(tmp_path: Path) -> None:
+    cache_dir = tmp_path / "datasets--liarliar--Daily-Omni"
+    snapshot = cache_dir / "snapshots" / "deadbeef"
+    snapshot.mkdir(parents=True)
+    _write_qa_json(snapshot)
+    (cache_dir / "refs").mkdir()
+    (cache_dir / "refs" / "main").write_text("deadbeef", encoding="utf-8")
+
+    root = resolve_daily_omni_local_root(str(cache_dir))
+    assert root == snapshot.resolve()
+    assert daily_omni_local_qa_json(root) == (snapshot / "qa.json").resolve()
+
+
+def test_hf_cache_dir_without_refs_picks_newest_snapshot(tmp_path: Path) -> None:
+    cache_dir = tmp_path / "datasets--liarliar--Daily-Omni"
+    older = cache_dir / "snapshots" / "aaa"
+    newer = cache_dir / "snapshots" / "bbb"
+    older.mkdir(parents=True)
+    newer.mkdir(parents=True)
+    os.utime(older, (1, 1))
+    os.utime(newer, (10, 10))
+
+    assert resolve_daily_omni_local_root(str(cache_dir)) == newer.resolve()
+
+
+def test_videos_tar_is_extracted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HF_HOME", str(tmp_path / "hf_home"))
+    root = tmp_path / "mirror"
+    (root / "Videos" / "vid0").mkdir(parents=True)
+    (root / "Videos" / "vid0" / "vid0_video.mp4").write_bytes(b"fake")
+    with tarfile.open(root / "Videos.tar", "w") as tf:
+        tf.add(root / "Videos", arcname="Videos")
+    # Only the tarball is left in the mirror.
+    for child in (root / "Videos").rglob("*"):
+        if child.is_file():
+            child.unlink()
+    (root / "Videos" / "vid0").rmdir()
+    (root / "Videos").rmdir()
+
+    videos_dir = daily_omni_local_videos_dir(root)
+    assert videos_dir is not None
+    assert (videos_dir / "vid0" / "vid0_video.mp4").is_file()
+
+    # Second call reuses the extraction instead of redoing it.
+    assert daily_omni_local_videos_dir(root) == videos_dir
+
+
+def test_missing_qa_json_reports_none(tmp_path: Path) -> None:
+    root = resolve_daily_omni_local_root(str(tmp_path))
+    assert daily_omni_local_qa_json(root) is None
+    assert daily_omni_local_videos_dir(root) is None

tests/benchmarks/test_daily_omni_pack_modes.py

@@ -2,9 +2,8 @@
 
 The ``minicpm-interleave`` mode is an accuracy-critical protocol: MiniCPM-o only reaches
 OpenBMB's reported Daily-Omni score when frames and 1s audio segments arrive strictly
-interleaved, with ``max_slice_nums=1`` / ``use_image_id=False`` and an empty MiniCPM
-system message (not the Qwen system prompt). These tests pin that contract so a refactor
-cannot silently regress the score.
+interleaved, with ``max_slice_nums=1`` / ``use_image_id=False`` and no Qwen system prompt.
+These tests pin that contract so a refactor cannot silently regress the score.
 
 vllm stubs are installed by tests/benchmarks/conftest.py before collection.
 """
@@ -99,22 +98,13 @@ def _stub_extract(monkeypatch, dataset: DailyOmniDataset, counter: list[int]) ->
 # ---------------------------------------------------------------------------
 
 
-@pytest.mark.parametrize(("n", "k"), [(10, 4), (100, 64), (1000, 64), (7, 7), (3, 5), (1, 1)])
-def test_uniform_sample_indices_matches_openbmb_midpoint(n: int, k: int) -> None:
-    """Pin OpenBMB midpoint-bin sampling (not np.linspace endpoints)."""
+@pytest.mark.parametrize(("n", "k"), [(10, 4), (100, 64), (7, 7), (3, 5), (1, 1)])
+def test_uniform_sample_indices_matches_numpy_linspace(n: int, k: int) -> None:
     got = _uniform_sample_indices(n, k)
     if n <= k:
         assert got == list(range(n))
     else:
-        gap = n / k
-        expected = [int(i * gap + gap / 2) for i in range(k)]
-        assert got == expected
-        # Endpoints differ from linspace for typical Daily-Omni downsample sizes.
-        linspace = [int(i) for i in np.linspace(0, n - 1, k, dtype=int)]
-        if n == 1000 and k == 64:
-            assert got[0] == 7 and got[-1] == 992
-            assert linspace[0] == 0 and linspace[-1] == 999
-            assert got != linspace
+        assert got == [int(i) for i in np.linspace(0, n - 1, k, dtype=int)]
 
 
 def test_uniform_sample_indices_non_positive_k() -> None:
@@ -223,16 +213,14 @@ def test_minicpm_interleave_missing_local_video_returns_none(qa_json, tmp_path)
 # ---------------------------------------------------------------------------
 
 
-def test_minicpm_interleave_keeps_empty_system_message(monkeypatch, qa_json, video_dir) -> None:
-    """Official MiniCPM omni mode still sends role=system with empty content."""
+def test_minicpm_interleave_omits_qwen_system_message(monkeypatch, qa_json, video_dir) -> None:
     ds = _make_dataset(qa_json, video_dir, pack_mode="minicpm-interleave", input_mode="all")
     _stub_extract(monkeypatch, ds, [])
     parts, _extra, _position = ds._compose_daily_omni_multimodal(_VIDEO_ID, None)
 
     messages = ds._build_daily_omni_openai_messages(parts, "q?", {"A": "a"})
 
-    assert [m["role"] for m in messages] == ["system", "user"]
-    assert messages[0]["content"][0]["text"] == ""
+    assert [m["role"] for m in messages] == ["user"]
 
 
 def test_qwen_pack_mode_keeps_system_message(qa_json, video_dir) -> None:

tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py

@@ -3,8 +3,8 @@
 
 Daily-Omni settings follow the MiniCPM interleaved AV recipe that reaches
 ~78% overall accuracy on Daily-Omni (``minicpm-interleave``, ``temperature=0``,
-text modalities with ``use_tts_template``, and server ``--interleave-mm-strings``
-+ 1fps / 128-frame media-io kwargs, also pinned in ``minicpmo_4_5.yaml``).
+text modalities, and server ``--interleave-mm-strings`` + 1fps / 128-frame
+media-io kwargs, also pinned in ``minicpmo_4_5.yaml``).
 """
 
 from __future__ import annotations
@@ -40,10 +40,7 @@ _MAX_SEED_TTS_MEAN_WER = 0.05
 # Match the validated Daily-Omni client body from daily_omni_bench.sh.
 _DAILY_EXTRA_BODY = {
     "modalities": ["text"],
-    "chat_template_kwargs": {
-        "enable_thinking": False,
-        "use_tts_template": True,
-    },
+    "chat_template_kwargs": {"enable_thinking": False},
 }
 _SEED_EXTRA_BODY = {
     "modalities": ["text", "audio"],

tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py

@@ -53,7 +53,11 @@ def test_duplex_admission_and_expiry_reaper(omni_server, model_prefix: str, tmp_
     assert result["admission"]["overflow_error_code"] == "resource_exhausted"
 
 
-@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+# CUDA-only for now: this contract asserts the model speaks while the user is
+# still talking, which only holds when the duplex pipeline sustains real-time
+# throughput. The current NPU stack runs several times slower than real time,
+# so it never reaches a mid-stream decision point.
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
 @pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
 def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) -> None:
     input_wav = validated_soft_interrupt_wav()
@@ -70,13 +74,8 @@ def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) -
                 timeout_s=180.0,
                 require_audio=True,
                 no_realtime_pacing=False,
-                # Single-GPU co-location cannot reliably finish the fixture's
-                # two-response soft-interrupt handoff before the WAV ends.
-                # model-policy + deterministic sampling checks streaming audio
-                # lifecycle instead of the diagnostic two-response contract.
-                validation_mode="model-policy",
-                temperature=0.0,
-                min_responses=1,
+                validation_mode="response-required",
+                min_responses=2,
                 min_audio_deltas_per_response=2,
                 input_sha256=SOFT_INTERRUPT_SHA256,
                 expect_followup_response_substring=None,
@@ -88,5 +87,4 @@ def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) -
     assert result["error_count"] == 0
     assert result["response_lifecycle_ok"] is True
     assert result["response_audio_contract_ok"] is True
-    assert result["response_before_final_commit"] is True
-    assert result["enough_responses"] is True
+    assert result["followup_response_transcript_ok"] is True