[Rebase] Rebase to vllm 0.27.0

.buildkite/cuda/test-merge.yml

@@ -139,7 +139,12 @@ steps:
         timeout_in_minutes: 50
         commands:
           - pytest -s -v tests/e2e/offline_inference/test_minicpmo_4_5.py tests/e2e/online_serving/test_minicpmo_4_5.py -m 'advanced_model and cuda' --run-level 'advanced_model'
-        mirror_hardwares: h100_1
+        # h100_2, not h100_1: advanced_model asserts transcribe every audio
+        # response with Whisper, and tests/helpers/media.py only uses an
+        # accelerator for Whisper when a spare GPU exists. On a single card it
+        # grades on CPU: ~8-10 min per clip vs ~30 s, which alone overruns the
+        # 50-minute job timeout (build 2956).
+        mirror_hardwares: h100_2
 
       - label: "Omni · MiniCPM-o 4.5 Duplex Test"
         source_file_dependencies:

.buildkite/cuda/test-ready.yml

@@ -128,6 +128,21 @@ steps:
               image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
               always-pull: true
               propagate-environment: true
+              # vLLM 0.27 added a pre-flight capacity check in
+              # distributed/device_communicators/shm_broadcast.py. Without this
+              # the container gets Docker's 64 MiB default and every test in
+              # this step dies at engine start with
+              #   RuntimeError: Insufficient space in /dev/shm:
+              #                 240 MiB required, 64 MiB available
+              # (9 failed in 8.7s on build #2949). vLLM 0.26 had no such check
+              # and allocated regardless, so this step passed on main while
+              # being under-provisioned the whole time — 0.27 turned a silent
+              # risk into an explicit error.
+              #
+              # This is the only inline docker plugin block in .buildkite/cuda/;
+              # every other step inherits common/ci_mirror_hardwares.yml, which
+              # already sets 8gb. Matching that rather than inventing a value.
+              shm-size: "8gb"
               environment:
                 - "HF_HOME=/fsx/hf_cache"
               volumes:

docker/Dockerfile.ci

@@ -1,7 +1,24 @@
-# Target vllm version: v0.26.0 (releases/v0.26.0 branch).
+# Target vllm version: v0.27.0 (releases/v0.27.0 branch).
 # Keep VLLM_BASE_TAG aligned with the target vLLM release on every rebase.
+#
+# Release-boundary form: there is no precompiled-wheel pin here, because the
+# base image already ships exactly the target vLLM. Everything the runtime
+# needs — torch, flashinfer, nvJitLink, numpy — comes from that image as one
+# self-consistent set.
+#
+# This block reappears only when the rebase targets a vLLM commit that is NOT a
+# released tag. During the 0.27 rebase, targeting v0.27.0rc1 on top of the
+# v0.26.0 base image required reinstalling vLLM from wheels.vllm.ai and then
+# hand-repairing the fallout: pinning torch ahead of flashinfer (installing it
+# after bound flashinfer's TVM FFI runtime to the old torch and aborted engine
+# core init), registering libnvJitLink.so.13 with ldconfig, and re-pinning numpy
+# under numba's ceiling. All of it was compensation for a base image and a vLLM
+# that disagreed. At a real release boundary they agree, so none of it is
+# needed. If you pin an untagged commit again, restore the reinstall block —
+# agent/tasks/20_update_wheel_url.sh fails the run rather than letting the
+# Dockerfile silently ship the wrong vLLM.
 ARG VLLM_BASE_IMAGE=vllm/vllm-openai
-ARG VLLM_BASE_TAG=v0.26.0
+ARG VLLM_BASE_TAG=v0.27.0
 
 # We use registry-based layer caching for the dependencies,
 # where the tag is determined based on the hash of the contents

tests/core/sched/test_omni_ar_scheduler_logprobs.py

@@ -92,6 +92,9 @@ def __init__(self, request_id: str) -> None:
         self.num_computed_tokens = 0
         self.num_in_flight_tokens = 0
         self.num_output_placeholders = 0
+        # vLLM 0.27 (a0c092ee72): Request gained num_stale_output_tokens to
+        # track in-flight outputs discarded at preemption/streaming-stop.
+        self.num_stale_output_tokens = 0
         self.has_encoder_inputs = False
         self.pooling_params = None
         self.resumable = True

tests/core/sched/test_omni_ar_scheduler_stale_drain.py

@@ -0,0 +1,223 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+"""Investigation tests for the num_stale_output_tokens drain (59b9e719).
+
+The seed-TTS WER regression appeared with the vLLM 0.27 rebase (nightly
+2946 -> 2949: mean WER 0.2695 -> 0.3936, median 0.0 in both), whose prime
+suspect is the reimplemented async-token discard: vLLM 0.26 drained
+``async_tokens_to_discard`` one arriving frame at a time and suppressed
+only the token append; the 0.27 adaptation seeds
+``num_stale_output_tokens`` with the discarded *placeholder count* and
+drains it by each arriving frame's ``num_tokens_scheduled``, dropping the
+whole per-request update. Whenever seed and arrivals disagree, the counter
+outlives the old segment and the drain consumes *valid* frames of the new
+segment — a hole in the talker's codec stream (garbled audio, text/audio
+mismatch) — or underflows its own assert on the new segment's prefill
+frame (engine-core death mid-step).
+
+The fix seeds the counter from ``num_in_flight_tokens`` — the same
+scheduled-token units the drain subtracts — so every pre-discard frame
+drains exactly its own contribution and new-segment frames can never be
+swallowed. These tests pin that contract one frame at a time; the
+delivered / swallowed observable is whether ``_update_request_with_output``
+runs for the frame.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+# Imports must run in this order: vllm_omni applies patches to vllm.v1.request
+# before Request / StreamingUpdate are bound in this module.
+# isort: off
+import vllm_omni  # noqa: F401 - import for side effects (patch vLLM)
+from vllm.sampling_params import SamplingParams
+from vllm.v1.core.sched.output import SchedulerOutput
+from vllm.v1.outputs import ModelRunnerOutput
+from vllm.v1.request import Request, RequestStatus, StreamingUpdate
+from vllm_omni.core.sched.omni_ar_scheduler import OmniARScheduler
+
+# isort: on
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _make_session() -> Request:
+    session = Request(
+        request_id="req-stale-drain-test",
+        prompt_token_ids=[1, 2, 3],
+        sampling_params=SamplingParams(max_tokens=32),
+        pooling_params=None,
+        arrival_time=100.0,
+        block_hasher=None,
+    )
+    session.status = RequestStatus.RUNNING
+    if not hasattr(session, "num_stale_output_tokens"):
+        # vLLM 0.27 defines this Request field; 0.26 (local dev) does not.
+        session.num_stale_output_tokens = 0
+    if not hasattr(session, "num_in_flight_tokens"):
+        session.num_in_flight_tokens = 0
+    return session
+
+
+def _replace_streaming_session(session: Request) -> None:
+    """Run the real stage-0 streaming replacement (the discard/seed site)."""
+    sched = OmniARScheduler.__new__(OmniARScheduler)
+    sched._new_prompt_len_snapshot = {}
+    sched.vllm_config = SimpleNamespace(model_config=SimpleNamespace(stage_id=0))
+    sched.num_waiting_for_streaming_input = 0
+    sched.log_stats = False
+    sched.chunk_transfer_adapter = None
+    sched.skipped_waiting = set()
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    update = StreamingUpdate(
+        mm_features=None,
+        prompt_token_ids=[10, 20],
+        max_tokens=32,
+        arrival_time=200.0,
+        sampling_params=SamplingParams(max_tokens=16),
+    )
+    sched._update_request_as_session(session, update)
+    session.status = RequestStatus.RUNNING
+
+
+def _make_drain_sched(session: Request) -> MagicMock:
+    sched = MagicMock()
+    sched.requests = {session.request_id: session}
+    sched.perf_metrics = None
+    sched.structured_output_manager.should_advance.return_value = False
+    sched._update_request_with_output.return_value = ([42], False)
+    sched._process_kv_transfer_trigger.return_value = False
+    sched.chunk_transfer_adapter = MagicMock()
+    sched.running = [session]
+    sched.waiting_for_transfer_free = set()
+    sched.transfer_triggered_requests = set()
+    sched.active_kv_transfers = set()
+    sched.pending_stop_after_extraction = set()
+    sched.connector = None
+    sched.kv_cache_manager.take_events.return_value = None
+    sched.finished_req_ids_dict = {}
+    sched.make_stats.return_value = None
+    return sched
+
+
+def _run_step(sched: MagicMock, session: Request, *, num_scheduled: int, token: int) -> bool:
+    """Feed one model-runner frame through update_from_output.
+
+    Returns True when the frame's tokens were delivered (the append ran),
+    False when the frame was swallowed by the stale drain.
+    """
+    scheduler_output = MagicMock(spec=SchedulerOutput)
+    scheduler_output.num_scheduled_tokens = {session.request_id: num_scheduled}
+    scheduler_output.scheduled_spec_decode_tokens = {}
+    scheduler_output.num_invalid_spec_tokens = 0
+
+    model_runner_output = MagicMock(spec=ModelRunnerOutput)
+    model_runner_output.sampled_token_ids = [[token]]
+    model_runner_output.logprobs = None
+    model_runner_output.prompt_logprobs_dict = {}
+    model_runner_output.pooler_output = None
+    model_runner_output.num_nans_in_logits = None
+    model_runner_output.kv_connector_output = None
+    model_runner_output.cudagraph_stats = None
+    model_runner_output.req_id_to_index = {session.request_id: 0}
+    model_runner_output.routed_experts = None
+    model_runner_output.inter_stage_outputs = None
+
+    sched._update_request_with_output.reset_mock()
+    OmniARScheduler.update_from_output(sched, scheduler_output, model_runner_output)
+    return sched._update_request_with_output.called
+
+
+def test_exact_drain_delivers_new_segment_frame() -> None:
+    """One in-flight decode frame at replacement: its late output drains the
+    counter exactly and the new segment's first frame is delivered."""
+    session = _make_session()
+    session.append_output_token_ids([7, 8, 9])
+    session.num_computed_tokens = 6
+    session.num_output_placeholders = 1
+    session.num_in_flight_tokens = 1
+
+    _replace_streaming_session(session)
+    assert session.num_stale_output_tokens == 1
+
+    sched = _make_drain_sched(session)
+    assert _run_step(sched, session, num_scheduled=1, token=42) is False  # late frame dropped
+    assert session.num_stale_output_tokens == 0
+    assert _run_step(sched, session, num_scheduled=1, token=43) is True  # new segment survives
+
+
+def test_two_in_flight_frames_drain_without_swallowing() -> None:
+    """Two in-flight decode frames: both late frames drop, the new segment's
+    first valid frame is delivered. A placeholder-based seed swallowed it
+    whenever fewer frames arrived than placeholders counted (the driver of
+    the seed-TTS WER regression, nightly 2946 -> 2949: 0.2695 -> 0.3936)."""
+    session = _make_session()
+    session.append_output_token_ids([7, 8, 9])
+    session.num_computed_tokens = 7
+    session.num_output_placeholders = 2
+    session.num_in_flight_tokens = 2
+
+    _replace_streaming_session(session)
+    assert session.num_stale_output_tokens == 2
+
+    sched = _make_drain_sched(session)
+    assert _run_step(sched, session, num_scheduled=1, token=42) is False  # late frame 1
+    assert _run_step(sched, session, num_scheduled=1, token=44) is False  # late frame 2
+    assert session.num_stale_output_tokens == 0
+
+    delivered = _run_step(sched, session, num_scheduled=1, token=43)  # new segment, frame 1
+    assert delivered is True, (
+        "valid new-segment frame was swallowed by leftover stale count "
+        f"(num_stale_output_tokens left at {session.num_stale_output_tokens})"
+    )
+
+
+def test_spec_placeholder_asymmetry_does_not_swallow_valid_frame() -> None:
+    """Placeholder counts can exceed scheduled counts (e.g. a draft
+    invalidated before execution): 1 in-flight MTP frame scheduled 2 tokens
+    but contributed 3 placeholders. Seeding from num_in_flight_tokens keeps
+    seed == drain, so the new segment's valid frame is delivered."""
+    session = _make_session()
+    session.append_output_token_ids([7])
+    session.num_computed_tokens = 7
+    session.num_output_placeholders = 3  # 1 sampled + 2 spec placeholders
+    session.num_in_flight_tokens = 2  # the frame actually scheduled 1 + 1 tokens
+
+    _replace_streaming_session(session)
+    assert session.num_stale_output_tokens == 2
+
+    sched = _make_drain_sched(session)
+    assert _run_step(sched, session, num_scheduled=2, token=42) is False  # late MTP frame
+    assert session.num_stale_output_tokens == 0
+    delivered = _run_step(sched, session, num_scheduled=1, token=43)  # new segment, frame 1
+    assert delivered is True, (
+        "valid new-segment frame swallowed after spec placeholder asymmetry "
+        f"(num_stale_output_tokens left at {session.num_stale_output_tokens})"
+    )
+
+
+def test_in_flight_prefill_chunk_drains_exactly_without_underflow() -> None:
+    """An in-flight prefill chunk carries no placeholders but 5 scheduled
+    tokens. The placeholder-based seed missed it entirely (its late output
+    was appended to the rolled-back session) and any leftover counter
+    underflowed 'assert num_stale_output_tokens >= 0' on the chunk's arrival,
+    killing the engine-core step. Seeding from num_in_flight_tokens drops the
+    late chunk and drains to exactly zero."""
+    session = _make_session()
+    session.append_output_token_ids([7, 8, 9])
+    session.num_computed_tokens = 7
+    session.num_output_placeholders = 0  # prefill chunk: no placeholders
+    session.num_in_flight_tokens = 5  # the chunk's scheduled tokens
+
+    _replace_streaming_session(session)
+    assert session.num_stale_output_tokens == 5
+
+    sched = _make_drain_sched(session)
+    assert _run_step(sched, session, num_scheduled=5, token=42) is False  # late prefill chunk dropped
+    assert session.num_stale_output_tokens == 0
+    assert _run_step(sched, session, num_scheduled=1, token=43) is True  # new segment survives

tests/diffusion/distributed/test_expert_parallel_layout.py

@@ -64,6 +64,7 @@ def fake_init_vllm_model_parallel_group(
         local_rank,
         backend,
         group_name,
+        use_all2all=False,
     ):
         del backend
         group = _FakeGroup(
@@ -73,6 +74,7 @@ def fake_init_vllm_model_parallel_group(
             device_communicator=object(),
             reduce_scatter=lambda tensor, **kwargs: tensor,
         )
+        group.use_all2all = use_all2all
         created_groups.append(group)
         return group
 
@@ -121,6 +123,9 @@ def fake_init_vllm_model_parallel_group(
     assert parallel_state.vllm_parallel_state._PCP.device_communicator is not None
     assert parallel_state.vllm_parallel_state._DP.device_communicator is not None
     assert parallel_state.vllm_parallel_state._EP.device_communicator is not None
+    # vLLM 0.27's MoE oracle asserts the EP communicator has an all2all
+    # manager, which vLLM only builds for groups created with use_all2all.
+    assert parallel_state.vllm_parallel_state._EP.use_all2all is True
     assert hasattr(parallel_state.vllm_parallel_state._PCP, "reduce_scatter")
     assert hasattr(parallel_state.vllm_parallel_state._DP, "reduce_scatter")
     assert hasattr(parallel_state.vllm_parallel_state._EP, "reduce_scatter")
@@ -313,3 +318,35 @@ def fake_init_model_parallel_group(
     assert parallel_state.vllm_parallel_state._PCP is None
     assert parallel_state.vllm_parallel_state._EP is None
     assert parallel_state._EXPERT_PARALLEL_GROUP_RANKS is None
+
+
+@pytest.mark.cpu
+@pytest.mark.core_model
+def test_init_vllm_group_forwards_use_all2all_only_when_supported(monkeypatch):
+    """use_all2all reaches vLLM only on versions whose API accepts it.
+
+    vLLM 0.27 builds the EP all2all manager only for groups created with
+    use_all2all=True (its MoE oracle then asserts the manager exists —
+    build 2953, HunyuanImage3-DIT accuracy). Older vLLM has no such kwarg,
+    so forwarding it unconditionally would TypeError at init.
+    """
+    received = {}
+
+    def modern(group_ranks, local_rank, backend, group_name=None, use_device_communicator=True, use_all2all=False):
+        received["modern"] = use_all2all
+        return object()
+
+    monkeypatch.setattr(parallel_state.vllm_parallel_state, "init_model_parallel_group", modern)
+    parallel_state.init_vllm_model_parallel_group([[0]], 0, "gloo", "ep", use_all2all=True)
+    assert received["modern"] is True
+
+    parallel_state.init_vllm_model_parallel_group([[0]], 0, "gloo", "tp")
+    assert received["modern"] is False
+
+    def legacy(group_ranks, local_rank, backend, group_name=None, use_device_communicator=True):
+        received["legacy"] = True
+        return object()
+
+    monkeypatch.setattr(parallel_state.vllm_parallel_state, "init_model_parallel_group", legacy)
+    parallel_state.init_vllm_model_parallel_group([[0]], 0, "gloo", "ep", use_all2all=True)
+    assert received["legacy"] is True

tests/diffusion/layers/test_fused_moe.py

@@ -110,7 +110,7 @@ def test_new_returns_vllm_runner_and_registers_hooks(self, mocker, monkeypatch):
         monkeypatch.setitem(
             sys.modules,
             "vllm.model_executor.layers.fused_moe",
-            SimpleNamespace(FusedMoE=mock_vllm_fused_moe),
+            SimpleNamespace(FusedMoEFactory=mock_vllm_fused_moe),
         )
 
         result = fused_moe.FusedMoE(prefix="test", a=1, reduce_results=True)
@@ -139,7 +139,7 @@ def register_forward_pre_hook(self, hook, *, with_kwargs):
         monkeypatch.setitem(
             sys.modules,
             "vllm.model_executor.layers.fused_moe",
-            SimpleNamespace(FusedMoE=mocker.MagicMock(return_value=mock_runner)),
+            SimpleNamespace(FusedMoEFactory=mocker.MagicMock(return_value=mock_runner)),
         )
         mocker.patch.object(fused_moe, "current_omni_platform", mocker.MagicMock())
         mock_set_num_tokens = mocker.patch.object(fused_moe, "_set_forward_context_num_tokens")
@@ -169,7 +169,7 @@ def register_forward_pre_hook(self, hook, *, with_kwargs):
         monkeypatch.setitem(
             sys.modules,
             "vllm.model_executor.layers.fused_moe",
-            SimpleNamespace(FusedMoE=mocker.MagicMock(return_value=mock_runner)),
+            SimpleNamespace(FusedMoEFactory=mocker.MagicMock(return_value=mock_runner)),
         )
         mocker.patch.object(fused_moe, "current_omni_platform", mocker.MagicMock())
         mock_set_num_tokens = mocker.patch.object(fused_moe, "_set_forward_context_num_tokens")