[Bugfix][CI/Build] Add MiniCPM-o Stage-2 timeout diagnostics

.buildkite/amd/test-amd-nightly.yml

@@ -577,6 +577,7 @@ steps:
         env:
           SEED_TTS_WER_EVAL: "1"
           SEED_TTS_SIM_EVAL: "1"
+          VLLM_OMNI_CODE2WAV_WATCHDOG_TIMEOUT_S: "60"
         artifact_paths:
           - artifacts/rocm-minicpmo-accuracy/**/*
         commands:

docs/configuration/environment_variables.md

@@ -94,6 +94,7 @@ environment variables.
 | `VLLM_OMNI_ABORT_TIMEOUT` | Float seconds; default `2` | Engine abort wait for Videos DELETE and `generate()` cancel/error cleanup; read when `async_omni` imports | Environment-only setting. A non-float raises `ValueError` during import. | Experimental |
 | `VLLM_OMNI_API_SERVER_TORCH_THREADS` | Non-negative integer; default `4`; `0` disables the cap | Process-wide Torch intra-op pool; read at API worker startup, including inline diffusion stages | Caps the existing count without increasing a lower count (e.g. from `OMP_NUM_THREADS`). Set a higher cap for CPU-heavy preprocessing or postprocessing; `0` preserves the existing count. A non-integer or negative value raises `ValueError` at startup. Separate stage processes are unaffected. | Stable operational control |
 | `VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT` | Float seconds; default `600` | Diffusion engine async-output wait in `step_streaming`; resolved per call on the request path, not at import | Environment-only setting. A non-float or `<=0` value warns once and uses the default. | Experimental |
+| `VLLM_OMNI_CODE2WAV_WATCHDOG_TIMEOUT_S` | Positive finite seconds; unset or `<=0` disables | MiniCPM-o 4.5 Code2Wav stage; read when each Code2Wav model instance initializes | Reports the current phase, bounded request details, and the blocked owner-thread Python stack once per operation without interrupting it. Invalid or non-finite values raise `ValueError` at model startup. | Diagnostic |
 | `VLLM_OMNI_EVENT_DRIVEN_ORCH` | Defaults to on; `0`, `false`, `no` or `off` selects the legacy poll loop | Orchestrator stage-output loop; resolved at `Orchestrator` construction | Unset selects event-driven orchestration for every pipeline. Values are stripped and case-normalized; `1`, `true`, `yes` or `on` keep it enabled and any other value selects the legacy poll loop. The serving-side final-output loop always wakes on orchestrator output, independent of this setting. Set before server startup. | Experimental |
 | `VLLM_OMNI_INPUT_WAIT_TIMEOUT_S` | Float seconds; default `600`; `<=0` disables | Full-payload input coordinator, not async-chunk transfer; read when the scheduler module imports in each worker | Environment-only setting. A non-float logs a warning and uses `600`. | Stable operational control |
 | `VLLM_OMNI_ORCH_MONITOR_PATH` | Filesystem path; default `<current-working-directory>/vllm_omni_orch_monitor_<timestamp>.json` | Orchestrator monitor enabled by `--enable-orch-monitor`; read when the monitor is created | Environment-only path override. Parent directories are created; write errors are logged. | Diagnostic |

tests/benchmarks/patch/test_patch.py

@@ -32,6 +32,7 @@
     _attach_seed_tts_to_request_func_input,
     _build_benchmark_session,
     _extract_stage_durations_from_payload,
+    _gather_request_outputs,
     _iter_image_reference_inputs,
     _iter_video_reference_inputs,
     _omni_request_timeout_s,
@@ -1511,13 +1512,90 @@ async def handler(reader, writer):
             request_input.api_url = f"http://127.0.0.1:{port}/v1/chat/completions"
             output = await async_request_openai_chat_omni_completions(request_input, session)
             assert output.success is False
+            assert output.timed_out is True
             assert output.error
         finally:
             await session.close()
             server.close()
             await server.wait_closed()
 
 
+@pytest.mark.asyncio
+async def test_chat_timeout_is_structurally_marked(mocker: MockerFixture) -> None:
+    session = mocker.Mock()
+    session.post.side_effect = TimeoutError("request deadline expired")
+
+    output = await async_request_openai_chat_omni_completions(_seed_tts_request_func_input(), session)
+
+    assert output.success is False
+    assert output.timed_out is True
+    assert output.cancelled_after_timeout is False
+    assert "TimeoutError" in output.error
+
+
+@pytest.mark.asyncio
+async def test_chat_aiohttp_server_timeout_is_structurally_marked(mocker: MockerFixture) -> None:
+    session = mocker.Mock()
+    session.post.side_effect = patch.aiohttp.ServerTimeoutError("request deadline expired")
+
+    output = await async_request_openai_chat_omni_completions(_seed_tts_request_func_input(), session)
+
+    assert output.success is False
+    assert output.timed_out is True
+    assert output.cancelled_after_timeout is False
+    assert "ServerTimeoutError" in output.error
+
+
+@pytest.mark.asyncio
+async def test_timeout_gather_cancels_pending_tasks_and_retains_order(mocker: MockerFixture) -> None:
+    pending_started = [asyncio.Event(), asyncio.Event()]
+    cancellation_finished = [asyncio.Event(), asyncio.Event()]
+    completed_output = MixRequestFuncOutput(success=True, prompt_len=22)
+    timeout_output = MixRequestFuncOutput(success=False, prompt_len=33, timed_out=True, error="timed out")
+
+    async def pending(index: int) -> MixRequestFuncOutput:
+        pending_started[index].set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            await asyncio.sleep(0)
+            cancellation_finished[index].set()
+
+    async def completed() -> MixRequestFuncOutput:
+        return completed_output
+
+    async def timed_out() -> MixRequestFuncOutput:
+        await pending_started[0].wait()
+        await pending_started[1].wait()
+        await asyncio.sleep(0)
+        return timeout_output
+
+    request_inputs = []
+    for prompt_len in (11, 22, 33, 44):
+        request = _seed_tts_request_func_input()
+        request.prompt_len = prompt_len
+        request_inputs.append(request)
+    tasks = [
+        asyncio.create_task(pending(0)),
+        asyncio.create_task(completed()),
+        asyncio.create_task(timed_out()),
+        asyncio.create_task(pending(1)),
+    ]
+    pbar = mocker.Mock()
+
+    outputs = await _gather_request_outputs(tasks, request_inputs, pbar=pbar)
+
+    assert outputs[0].cancelled_after_timeout is True
+    assert outputs[0].prompt_len == 11
+    assert outputs[1] is completed_output
+    assert outputs[2] is timeout_output
+    assert outputs[3].cancelled_after_timeout is True
+    assert outputs[3].prompt_len == 44
+    assert all(event.is_set() for event in cancellation_finished)
+    assert all(task.done() for task in tasks)
+    assert pbar.update.call_count == 2
+
+
 def test_video_rtf_prefers_generation_time_over_poll_latency():
     """RTF must not grow with client poll_interval overshoot in output.latency."""
     payload = {

tests/benchmarks/test_accuracy_bench_utils.py

@@ -253,6 +253,39 @@ def test_seed_tts_floor_rejects_incomplete_sim_and_utmos_coverage():
     assert not any("seed_tts_utmos_mean=3.350000" in err for err in errs)
 
 
+@pytest.mark.parametrize(
+    ("timed_out_requests", "cancelled_after_timeout_requests"),
+    [(1, 0), (0, 3)],
+)
+def test_validate_seed_tts_rejects_high_scoring_partial_timeout_result(
+    timed_out_requests: int,
+    cancelled_after_timeout_requests: int,
+) -> None:
+    partial = {
+        "seed_tts_content_evaluated": 1,
+        "seed_tts_content_error_mean": 0.01,
+        "seed_tts_turn_count": 1,
+        "seed_tts_request_failed": 0,
+        "seed_tts_no_pcm": 0,
+        "seed_tts_sim_mean": 0.95,
+        "seed_tts_sim_evaluated": 1,
+        "seed_tts_sim_failed": 0,
+        "seed_tts_utmos_mean": 4.5,
+        "seed_tts_utmos_evaluated": 1,
+        "seed_tts_utmos_failed": 0,
+        "timed_out_requests": timed_out_requests,
+        "cancelled_after_timeout_requests": cancelled_after_timeout_requests,
+    }
+
+    errs = _validate_seed_tts(partial, max_mean_wer=0.05, min_mean_sim=0.80, min_mean_utmos=3.10)
+
+    assert any(
+        f"timed_out_requests={timed_out_requests}" in err
+        and f"cancelled_after_timeout_requests={cancelled_after_timeout_requests}" in err
+        for err in errs
+    )
+
+
 def test_seed_tts_floor_coverage_excludes_rows_the_scorer_never_sees():
     """ASR failures skip SIM, empty audio skips UTMOS; neither is a coverage gap."""
     result = {

tests/benchmarks/test_serve_cli.py

@@ -129,6 +129,7 @@ def test_extend_omni_choices_updates_tracking_parser_shadow(dataset_name: str) -
     ("argv", "dest", "expected"),
     [
         (["--print-stage"], "print_stage", True),
+        (["--omni-fail-fast-on-timeout"], "omni_fail_fast_on_timeout", True),
         (["--daily-omni-input-mode", "audio"], "daily_omni_input_mode", "audio"),
         (["--omniinteract-subsets", "1qna"], "omniinteract_subsets", ["1qna"]),
         (["--seed-tts-locale", "zh"], "seed_tts_locale", "zh"),
@@ -156,6 +157,7 @@ def test_add_omni_args_preserves_implicit_defaults() -> None:
 
     args = parser.parse_args([])
     assert args.print_stage is False
+    assert args.omni_fail_fast_on_timeout is False
     assert args.daily_omni_input_mode == "all"
     assert args.omniinteract_subsets == ["1q1a", "1q1a_math", "1qna"]
     assert args.seed_tts_locale == "en"
@@ -534,6 +536,68 @@ def test_omni_request_timeout_s_flag_defaults_and_parses() -> None:
     args = parser.parse_args(["--omni-request-timeout-s", "3600"])
     assert args.omni_request_timeout_s == 3600.0
 
+    args = parser.parse_args(["--omni-fail-fast-on-timeout"])
+    assert args.omni_fail_fast_on_timeout is True
+
+
+@pytest.mark.parametrize(
+    ("backend", "request_rate", "self_timed"),
+    [
+        ("openai-image-edits-omni", float("inf"), False),
+        ("openai-chat-omni", 1.0, False),
+        ("openai-chat-omni", float("inf"), True),
+    ],
+)
+def test_fail_fast_on_timeout_rejects_unsupported_dispatch_modes(
+    backend: str,
+    request_rate: float,
+    self_timed: bool,
+) -> None:
+    from vllm_omni.benchmarks.serve import _validate_fail_fast_on_timeout
+
+    args = argparse.Namespace(
+        omni_fail_fast_on_timeout=True,
+        backend=backend,
+        request_rate=request_rate,
+        self_timed=self_timed,
+    )
+
+    with pytest.raises(ValueError, match="requires --backend openai-chat-omni"):
+        _validate_fail_fast_on_timeout(args)
+
+
+def test_fail_fast_on_timeout_accepts_immediate_chat_omni_dispatch() -> None:
+    from vllm_omni.benchmarks.serve import _validate_fail_fast_on_timeout
+
+    args = argparse.Namespace(
+        omni_fail_fast_on_timeout=True,
+        backend="openai-chat-omni",
+        request_rate=float("inf"),
+        self_timed=False,
+    )
+
+    _validate_fail_fast_on_timeout(args)
+
+
+def test_fail_fast_on_timeout_rejects_finite_rate_before_dispatch(mocker) -> None:
+    from vllm_omni.benchmarks import serve
+
+    run_benchmark = mocker.patch.object(serve, "main_async")
+    enable_fail_fast = mocker.patch.object(serve, "set_fail_fast_on_timeout")
+    args = argparse.Namespace(
+        omni_fail_fast_on_timeout=True,
+        backend="openai-chat-omni",
+        endpoint="/v1/chat/completions",
+        request_rate=1.0,
+        self_timed=False,
+    )
+
+    with pytest.raises(ValueError, match="requires --backend openai-chat-omni"):
+        serve.main(args)
+
+    run_benchmark.assert_not_called()
+    enable_fail_fast.assert_not_called()
+
 
 def test_realtime_chat_choice_is_available_without_importing_runtime() -> None:
     parser = TrackingArgumentParser()

tests/e2e/accuracy/helpers.py

@@ -238,6 +238,14 @@ def _validate_seed_tts(
     if setup:
         errs.append(f"Seed-TTS eval setup failed: {setup}")
         return errs
+    timed_out = int(result.get("timed_out_requests", 0) or 0)
+    cancelled = int(result.get("cancelled_after_timeout_requests", 0) or 0)
+    if timed_out or cancelled:
+        errs.append(
+            "Seed-TTS benchmark coverage is incomplete: "
+            f"timed_out_requests={timed_out}, cancelled_after_timeout_requests={cancelled}; "
+            "accuracy metrics exclude unsuccessful outputs."
+        )
     n = int(result.get("seed_tts_content_evaluated", 0) or 0)
     if n <= 0:
         errs.append("seed_tts_content_evaluated is 0 (enable --seed-tts-wer-eval and check PCM capture).")
@@ -538,6 +546,8 @@ def run_seed_tts(ns: argparse.Namespace, vllm: str) -> Path:
         argv.append("--seed-tts-wer-save-items")
     if ns.seed_tts_file_ref_audio:
         argv.append("--seed-tts-file-ref-audio")
+    if getattr(ns, "omni_fail_fast_on_timeout", False):
+        argv.append("--omni-fail-fast-on-timeout")
     extra_env = seed_tts_eval_env(ns)
     print("\n$", vllm, *argv, "\n", flush=True)
     _omni_acc_bench_core().run_vllm_bench_subprocess(vllm, argv, extra_env=extra_env)
@@ -573,6 +583,11 @@ def build_arg_parser() -> argparse.ArgumentParser:
         help="If set, forwarded to ``vllm bench serve`` (probe first request until success). "
         "Omit to use upstream default (typically skip).",
     )
+    p.add_argument(
+        "--omni-fail-fast-on-timeout",
+        action="store_true",
+        help="Forward the benchmark timeout circuit breaker to Seed-TTS runs.",
+    )
     p.add_argument(
         "--trust-remote-code",
         action="store_true",

tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py

@@ -206,6 +206,7 @@ def _run_seed_tts_wer_bench(omni_server) -> None:
             str(_MIN_SEED_TTS_MEAN_SIM),
             "--seed-extra-body-json",
             json.dumps(_SEED_EXTRA_BODY, separators=(",", ":")),
+            "--omni-fail-fast-on-timeout",
             "--trust-remote-code",
         ]
     )