[Core] Validate HSDP shard size before initialization

Signed-off-by: hsliuustc0106 <hsliuustc0106@users.noreply.github.com>
[Core] Identify HSDP replica ranks in worker titles
2026-08-29 06:00:36 +08:00 · 2026-08-28 18:27:56 +08:00 · 2026-08-28 18:19:10 +08:00
59 changed files with 413 additions and 2045 deletions
@@ -10,7 +10,7 @@ steps:
          DOCKER_BUILDKIT: "1"
          # Buildkite will automatically replace this with the actual commit hash
          VLLM_IMAGE_TAG: "${BUILDKITE_COMMIT}"
-          VLLM_VERSION: "v0.28.0"
+          VLLM_VERSION: "v0.27.0"
        priority: 100
        timeout_in_minutes: 60
        soft_fail: false
@@ -5,12 +5,7 @@ set -ex

 omni_source_dir=$(git rev-parse --show-toplevel)

-: "${VLLM_VERSION:?VLLM_VERSION must be set}"
-# Official upstream XPU image. It is published from vllm's own
-# docker/Dockerfile.xpu, which is exactly what build_vllm_base() clones and
-# builds below, so pulling it is equivalent to building the fallback base.
-upstream_base_image="vllm/vllm-openai-xpu:${VLLM_VERSION}"
-local_base_image="xpu/vllm-omni-ci-base:${VLLM_VERSION}"
+base_image_name="xpu/vllm-omni-ci-base:${VLLM_VERSION:?VLLM_VERSION must be set}"
 image_name="xpu/vllm-omni-ci:${BUILDKITE_COMMIT:?BUILDKITE_COMMIT must be set}"
 container_name="xpu_${BUILDKITE_COMMIT}_$(
    tr -dc A-Za-z0-9 </dev/urandom | head -c 10
@@ -54,36 +49,9 @@ build_vllm_base() (
        "${vllm_source_dir}"
 )

-# Resolve the vLLM base image, in priority order:
-#   1. the published upstream image for VLLM_VERSION, pulled fresh (so moving
-#      tags like nightly are refreshed rather than served stale from disk);
-#   2. a local copy of that same image, when the registry is unreachable;
-#   3. VLLM_BASE, if that image is already on disk;
-#   4. otherwise build VLLM_BASE from the matching vLLM source tag (~30min).
-# VLLM_BASE names the fallback base and defaults to ${local_base_image}; set it
-# to reuse or produce a base under a different tag.
-resolve_vllm_base() {
-    if docker pull "${upstream_base_image}"; then
-        base_image_name="${upstream_base_image}"
-        return
-    fi
-
-    # Registry unreachable, but a previous run may have left the image behind.
-    if [ -n "$(docker images -q "${upstream_base_image}")" ]; then
-        echo "WARNING: could not pull ${upstream_base_image}; using local copy" >&2
-        base_image_name="${upstream_base_image}"
-        return
-    fi
-
-    echo "WARNING: ${upstream_base_image} unavailable remotely and locally" >&2
-    base_image_name="${VLLM_BASE:-${local_base_image}}"
-    if [ -z "$(docker images -q "${base_image_name}")" ]; then
-        echo "WARNING: building ${base_image_name} from vLLM ${VLLM_VERSION} source" >&2
-        build_vllm_base
-    fi
-}
-
-resolve_vllm_base
+if [ -z "$(docker images -q "${base_image_name}")" ]; then
+    build_vllm_base
+fi

 # Try building the docker image
 docker_build "${image_name}" --build-arg "VLLM_BASE=${base_image_name}" --build-arg "VLLM_VERSION=${VLLM_VERSION}"
@@ -1,33 +1,140 @@
-# Target vLLM version: v0.28.0. Keep VLLM_VERSION aligned with the target vLLM
-# release on every rebase.
-#
-# This file layers vLLM-Omni on top of the published upstream XPU image, which
-# already ships a self-consistent XPU runtime (torch-xpu, oneCCL/UCX+NIXL, the
-# pinned UMD/compute-runtime stack, and vLLM itself) built from
-# https://github.com/vllm-project/vllm/blob/main/docker/Dockerfile.xpu. Docker
-# pulls that base automatically, so there is nothing to build here beyond the
-# Omni layers. Mirrors how docker/Dockerfile.ci consumes vllm/vllm-openai.
-#
-# If the tag does not exist yet (targeting an unreleased vLLM commit), build the
-# base from upstream's own Dockerfile and point VLLM_BASE at the result:
-#
-#   git clone --depth 1 --branch $VLLM_VERSION https://github.com/vllm-project/vllm /tmp/vllm
-#   docker build -t vllm-openai-xpu:local --target vllm-openai \
-#       -f /tmp/vllm/docker/Dockerfile.xpu /tmp/vllm
-#   docker build -f docker/Dockerfile.xpu --build-arg VLLM_BASE=vllm-openai-xpu:local .
-#
-# .buildkite/intel/scripts/run-xpu-test.sh automates that pull-then-build fallback.
-ARG VLLM_BASE_IMAGE=vllm/vllm-openai-xpu
-ARG VLLM_VERSION=v0.28.0
-ARG VLLM_BASE=${VLLM_BASE_IMAGE}:${VLLM_VERSION}
+# Argument to configure vllm base image if pre-built
+ARG VLLM_BASE=vllm-base
+
+FROM intel/deep-learning-essentials:2025.3.2-0-devel-ubuntu24.04 AS vllm-base
+
+WORKDIR /workspace/
+
+ARG PYTHON_VERSION=3.12
+ARG PIP_EXTRA_INDEX_URL="https://download.pytorch.org/whl/xpu"
+
+RUN wget -O- https://apt.repos.intel.com/intel-gpg-keys/GPG-PUB-KEY-INTEL-SW-PRODUCTS.PUB | gpg --dearmor | tee /usr/share/keyrings/oneapi-archive-keyring.gpg > /dev/null && \
+    echo "deb [signed-by=/usr/share/keyrings/oneapi-archive-keyring.gpg] https://apt.repos.intel.com/oneapi all main" | tee /etc/apt/sources.list.d/oneAPI.list
+
+RUN apt clean && apt-get update -y && \
+    apt-get install -y --no-install-recommends --fix-missing \
+    curl \
+    ffmpeg \
+    git \
+    libsndfile1 \
+    libsm6 \
+    libxext6 \
+    libgl1 \
+    lsb-release \
+    libaio-dev \
+    numactl \
+    wget \
+    vim \
+    python3.12 \
+    python3.12-dev \
+    python3-pip
+
+# Add oneAPI repo, pin oneAPI to 2025.3, then install pinned packages in one layer.
+RUN wget -O- https://apt.repos.intel.com/intel-gpg-keys/GPG-PUB-KEY-INTEL-SW-PRODUCTS.PUB | gpg --dearmor | tee /usr/share/keyrings/oneapi-archive-keyring.gpg > /dev/null && \
+    echo "deb [signed-by=/usr/share/keyrings/oneapi-archive-keyring.gpg] https://apt.repos.intel.com/oneapi all main" | tee /etc/apt/sources.list.d/oneAPI.list && \
+    printf '%s\n' \
+    'Package: intel-oneapi-* intel-deep-learning-essentials* intel-pti*' \
+    'Pin: version 2025.3*' \
+    'Pin-Priority: 1001' \
+    > /etc/apt/preferences.d/oneapi-2025.3.pref && \
+    apt-get update -y && \
+    apt-get install -y --no-install-recommends \
+    intel-oneapi-compiler-dpcpp-cpp-2025.3 \
+    intel-oneapi-mkl-devel-2025.3 \
+    intel-oneapi-dnnl-devel-2025.3 && \
+    rm -rf /var/lib/apt/lists/*
+
+# Install UMD with fixed version
+RUN mkdir neo && \
+    cd neo && \
+    wget https://github.com/intel/intel-graphics-compiler/releases/download/v2.24.8/intel-igc-core-2_2.24.8+20344_amd64.deb && \
+    wget https://github.com/intel/intel-graphics-compiler/releases/download/v2.24.8/intel-igc-opencl-2_2.24.8+20344_amd64.deb && \
+    wget https://github.com/intel/compute-runtime/releases/download/25.48.36300.8/intel-ocloc_25.48.36300.8-0_amd64.deb && \
+    wget https://github.com/intel/compute-runtime/releases/download/25.48.36300.8/intel-opencl-icd_25.48.36300.8-0_amd64.deb && \
+    wget https://github.com/intel/compute-runtime/releases/download/25.48.36300.8/libigdgmm12_22.8.2_amd64.deb && \
+    wget https://github.com/intel/compute-runtime/releases/download/25.48.36300.8/libze-intel-gpu1_25.48.36300.8-0_amd64.deb && \
+    wget https://github.com/oneapi-src/level-zero/releases/download/v1.26.0/level-zero_1.26.0+u24.04_amd64.deb && \
+    dpkg -i *.deb && \
+    cd .. && \
+    rm -rf neo
+
+ENV PATH="/root/.local/bin:$PATH"
+ENV VIRTUAL_ENV="/opt/venv"
+ENV UV_PYTHON_INSTALL_DIR=/opt/uv/python
+RUN curl -LsSf https://astral.sh/uv/install.sh | sh
+RUN uv venv --python ${PYTHON_VERSION} --seed ${VIRTUAL_ENV}
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+
+# This oneccl contains the BMG support which is not the case for default version of oneapi 2025.3.
+ARG ONECCL_INSTALLER="intel-oneccl-2021.15.9.14_offline.sh"
+RUN wget "https://github.com/uxlfoundation/oneCCL/releases/download/2021.15.9/${ONECCL_INSTALLER}" && \
+    bash "${ONECCL_INSTALLER}" -a --silent --eula accept && \
+    rm "${ONECCL_INSTALLER}" && \
+    echo "source /opt/intel/oneapi/setvars.sh --force" >> /root/.bashrc && \
+    echo "source /opt/intel/oneapi/ccl/2021.15/env/vars.sh --force" >> /root/.bashrc
+RUN rm -f /opt/intel/oneapi/ccl/latest && \
+    ln -s /opt/intel/oneapi/ccl/2021.15 /opt/intel/oneapi/ccl/latest
+
+SHELL ["bash", "-c"]
+CMD ["bash", "-c", "source /root/.bashrc && exec bash"]
+
+WORKDIR /workspace/
+
+ENV UV_HTTP_TIMEOUT=500
+
+# Configure package index for XPU
+ENV PIP_EXTRA_INDEX_URL=${PIP_EXTRA_INDEX_URL}
+ENV UV_EXTRA_INDEX_URL=${PIP_EXTRA_INDEX_URL}
+ENV UV_INDEX_STRATEGY="unsafe-best-match"
+ENV UV_LINK_MODE="copy"
+
+ARG VLLM_VERSION=v0.27.0
+RUN git clone -b ${VLLM_VERSION} https://github.com/vllm-project/vllm
+WORKDIR /workspace/vllm
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv pip install --upgrade pip && \
+    uv pip install -r requirements/xpu.txt && \
+    uv pip install setuptools_scm &&\
+    uv pip install grpcio-tools protobuf nanobind && \
+    source /opt/intel/oneapi/setvars.sh --force && \
+    source /opt/intel/oneapi/ccl/2021.15/env/vars.sh --force && \
+    export CMAKE_PREFIX_PATH="$(python3 -c 'import site; print(site.getsitepackages()[0])'):${CMAKE_PREFIX_PATH}"
+
+ENV LD_LIBRARY_PATH="/opt/intel/oneapi/ccl/2021.15/lib:$LD_LIBRARY_PATH:/usr/local/lib/"
+
+ENV VLLM_TARGET_DEVICE=xpu
+ENV VLLM_WORKER_MULTIPROC_METHOD=spawn
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv pip uninstall triton triton-xpu && \
+    uv pip install triton-xpu==3.7.2 && \
+    uv pip install --no-build-isolation --no-deps .
+
+CMD ["/bin/bash"]
+
+FROM vllm-base AS vllm-openai
+
+# install nixl from source code
+RUN uv pip install nixl
+
+# install additional dependencies for openai api server
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv pip install accelerate hf_transfer pytest pytest_asyncio lm_eval[api] modelscope
+
+# install development dependencies (for testing)
+RUN uv pip install -e tests/vllm_test_utils
+
+# ensure vllm is properly installed
+RUN python -c "import vllm, inspect; print(vllm.__file__)"
+RUN uv pip show vllm
+
+CMD ["/bin/bash"]
+
+ENTRYPOINT []

 FROM ${VLLM_BASE} AS vllm-omni

-# The base image ships ENTRYPOINT ["vllm", "serve"]; clear it so this stage stays
-# usable as a plain shell/test container regardless of which VLLM_BASE was used.
-ENTRYPOINT []
-CMD ["/bin/bash"]
-
 RUN apt-get update && \
    apt-get install -y espeak-ng ffmpeg git jq && \
    apt-get clean && \
@@ -23,19 +23,6 @@ vLLM-Omni currently recommends using the Docker image setup steps below.
 DOCKER_BUILDKIT=1 docker build -f docker/Dockerfile.xpu -t vllm-omni-xpu --shm-size=4g .
 ```

-This layers vLLM-Omni on top of the published `vllm/vllm-openai-xpu:<VLLM_VERSION>`
-base image, which Docker pulls automatically. To target a different vLLM release,
-pass `--build-arg VLLM_VERSION=<tag>`. If that tag has not been published yet, build
-the base from upstream's own Dockerfile first and point the build at it:
-
-```bash
-git clone --depth 1 --branch "$VLLM_VERSION" https://github.com/vllm-project/vllm /tmp/vllm
-DOCKER_BUILDKIT=1 docker build -t vllm-openai-xpu:local --target vllm-openai \
-  -f /tmp/vllm/docker/Dockerfile.xpu /tmp/vllm
-DOCKER_BUILDKIT=1 docker build -f docker/Dockerfile.xpu -t vllm-omni-xpu --shm-size=4g \
-  --build-arg VLLM_BASE=vllm-openai-xpu:local .
-```
-
 #### Launch the docker image

 ##### Launch with OpenAI API Server
@@ -26,15 +26,15 @@ th {
 | `InternVLAA1Pipeline` | InternVLA-A1 | `InternRobotics/InternVLA-A1-3B` | ✅︎ | ✅︎ | | | — |
 | `Gr00tN1d7Pipeline` | GR00T N1.7 | `nvidia/GR00T-N1.7-3B` | ✅︎ | | | | — |
 | `HunyuanImage3ForCausalMM` | HunyuanImage3.0 (DiT-only) | `tencent/HunyuanImage-3.0`, `tencent/HunyuanImage-3.0-Instruct` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
-| `QwenImagePipeline` | Qwen-Image | `Qwen/Qwen-Image` | ✅︎ | | | ✅︎ | [Published](https://recipes.vllm.ai/Qwen/Qwen-Image) |
-| `QwenImagePipeline` | Qwen-Image-2512 | `Qwen/Qwen-Image-2512` | ✅︎ | | | ✅︎ | [Repository](https://github.com/vllm-project/vllm-omni/blob/main/recipes/Qwen/Qwen-Image-2512.md) |
+| `QwenImagePipeline` | Qwen-Image | `Qwen/Qwen-Image` | ✅︎ | | | | [Published](https://recipes.vllm.ai/Qwen/Qwen-Image) |
+| `QwenImagePipeline` | Qwen-Image-2512 | `Qwen/Qwen-Image-2512` | ✅︎ | | | | [Repository](https://github.com/vllm-project/vllm-omni/blob/main/recipes/Qwen/Qwen-Image-2512.md) |
 | `QwenImageEditPipeline` | Qwen-Image-Edit | `Qwen/Qwen-Image-Edit` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `QwenImageEditPlusPipeline` | Qwen-Image-Edit-2509 | `Qwen/Qwen-Image-Edit-2509` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `QwenImageLayeredPipeline` | Qwen-Image-Layered | `Qwen/Qwen-Image-Layered` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `QwenImageEditPlusPipeline` | Qwen-Image-Edit-2511 | `Qwen/Qwen-Image-Edit-2511` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `GlmImagePipeline` | GLM-Image | `zai-org/GLM-Image` | ✅︎ | ✅︎ | | | — |
 | `ZImagePipeline` | Z-Image | `Tongyi-MAI/Z-Image-Turbo` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
-| `Krea2Pipeline` | Krea 2 (Raw + Turbo) | `krea/Krea-2-Raw`, `krea/Krea-2-Turbo` | ✅︎ | | | ✅︎ | — |
+| `Krea2Pipeline` | Krea 2 (Raw + Turbo) | `krea/Krea-2-Raw`, `krea/Krea-2-Turbo` | ✅︎ | | | | — |
 | `WanPipeline` | Wan2.1-T2V, Wan2.2-T2V, Wan2.2-TI2V | `Wan-AI/Wan2.1-T2V-1.3B-Diffusers`, `Wan-AI/Wan2.1-T2V-14B-Diffusers`, `Wan-AI/Wan2.2-T2V-A14B-Diffusers`, `Wan-AI/Wan2.2-TI2V-5B-Diffusers` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `WanImageToVideoPipeline` | Wan2.2-I2V | `Wan-AI/Wan2.2-I2V-A14B-Diffusers` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `LingBotWorldCausalDMDPipeline` | LingBot-World 2.0 v1 (experimental) | `robbyant/lingbot-world-v2-14b-causal-fast-diffusers` | Experimental | | | | — |
@@ -42,7 +42,7 @@ th {
 | `SanaWmPipeline` | SANA-WM | `BBBBruce/SANA-WM_bidirectional-stage1-diffusers` | ✅︎ | | | | [Repository](https://github.com/vllm-project/vllm-omni/blob/main/recipes/NVIDIA/SANA-WM.md) |
 | `Wan22S2VPipeline` | Wan2.2-S2V | `Wan-AI/Wan2.2-S2V-14B` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `Wan22VACEPipeline` | Wan2.1-VACE | `Wan-AI/Wan2.1-VACE-1.3B-diffusers`, `Wan-AI/Wan2.1-VACE-14B-diffusers` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
-| `Wan22VACEPipeline` | Wan2.2-VACE | `Pyros13/Wan2.2-VACE-Fun-A14B-Diffusers` | ✅︎ |   |   | ✅︎ | — |
+| `Wan22VACEPipeline` | Wan2.2-VACE | `Pyros13/Wan2.2-VACE-Fun-A14B-Diffusers` | ✅︎ |   |   |   | — |
 | `LTX2Pipeline` | LTX-2 / LTX-2.3 one-stage T2V and I2V | `Lightricks/LTX-2`, `diffusers/LTX-2.3-Diffusers` | ✅︎ | ✅︎ | | | — |
 | `LTX2TwoStagePipeline` | LTX-2 / LTX-2.3 ordinary two-stage T2V and I2V | `Lightricks/LTX-2`, `diffusers/LTX-2.3-Diffusers` + matching Lightricks LoRA and upsampler | ✅︎ | ✅︎ | | | — |
 | `LTX2DistilledOneStagePipeline` | LTX-2 / LTX-2.3 merged-distilled one-stage T2V and I2V | `rootonchair/LTX-2-19b-distilled`, `diffusers/LTX-2.3-Distilled-Diffusers` | ✅︎ | ✅︎ | | | — |
@@ -62,12 +62,12 @@ th {
 | `LongcatImagePipeline` | LongCat-Image | `meituan-longcat/LongCat-Image` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `LongCatImageEditPipeline` | LongCat-Image-Edit | `meituan-longcat/LongCat-Image-Edit` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `LongCatVideoAvatarPipeline` | LongCat-Video-Avatar-1.5 A2V/AI2V (native single-speaker and multi-speaker AI2V/AVC) | `meituan-longcat/LongCat-Video-Avatar-1.5` | ✅︎ | | | | [Repository](https://github.com/vllm-project/vllm-omni/blob/main/recipes/meituan-longcat/LongCat-Video-Avatar-1.5.md) |
-| `BooguImagePipeline` | Boogu-Image | `Boogu/Boogu-Image-0.1-Base`, `Boogu/Boogu-Image-0.1-Edit` | ✅︎ | | | ✅︎ | — |
+| `BooguImagePipeline` | Boogu-Image | `Boogu/Boogu-Image-0.1-Base`, `Boogu/Boogu-Image-0.1-Edit` | ✅︎ | | | | — |
 | `StableDiffusionXLPipeline` | Stable-Diffusion-XL | `stabilityai/stable-diffusion-xl-base-1.0` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `StableDiffusion3Pipeline` | Stable-Diffusion-3 | `stabilityai/stable-diffusion-3.5-medium` | ✅︎ | ✅︎ | | ✅︎ | — |
 | `CosyVoice3Model` | CosyVoice3 | `FunAudioLLM/Fun-CosyVoice3-0.5B-2512` | ✅︎ | ✅︎ | | ✅︎ | — |
-| `OmniVoiceModel` | OmniVoice | `k2-fsa/OmniVoice` | ✅︎ | | | ✅︎ | — |
-| `VoxCPM2TalkerForConditionalGeneration` | VoxCPM2 | `openbmb/VoxCPM2` | ✅︎ | | | ✅︎ | — |
+| `OmniVoiceModel` | OmniVoice | `k2-fsa/OmniVoice` | ✅︎ | | | | — |
+| `VoxCPM2TalkerForConditionalGeneration` | VoxCPM2 | `openbmb/VoxCPM2` | ✅︎ | | | | — |
 | `DotsTTSForConditionalGeneration` | dots.tts | `rednote-hilab/dots.tts-soar` | ✅︎ | | | | — |
 | `MammothModa2ForConditionalGeneration` | MammothModa2-Preview | `bytedance-research/MammothModa2-Preview` | ✅︎ | ✅︎ | | | — |
 | `MammothModa2ForConditionalGeneration` | MammothModa2-Dev (AR-only image understanding) | `bytedance-research/MammothModa2-Dev` | ✅︎ | | | | — |
@@ -78,10 +78,10 @@ th {
 | `OmniGen2Pipeline` | OmniGen2 | `OmniGen2/OmniGen2` | ✅︎ | ✅︎ | | ✅︎ | — |
 | `StableAudioPipeline` | Stable-Audio-Open | `stabilityai/stable-audio-open-1.0` | ✅︎ | ✅︎ | | ✅︎ | — |
 | `MiniMaxMusic3ForConditionalGeneration` | MiniMax Music 3 (text-to-music) | `MiniMaxAI/MiniMax-Music3` | ✅︎ | | | | — |
-| `Qwen3TTSForConditionalGeneration` | Qwen3-TTS-12Hz-1.7B-CustomVoice | `Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice` | ✅︎ | | | ✅︎ | [Published](https://recipes.vllm.ai/Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice) |
+| `Qwen3TTSForConditionalGeneration` | Qwen3-TTS-12Hz-1.7B-CustomVoice | `Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice` | ✅︎ | | | | [Published](https://recipes.vllm.ai/Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice) |
 | `Qwen3TTSForConditionalGeneration` | Qwen3-TTS-12Hz-1.7B-VoiceDesign | `Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
 | `Qwen3TTSForConditionalGeneration` | Qwen3-TTS-12Hz-1.7B-Base | `Qwen/Qwen3-TTS-12Hz-1.7B-Base` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
-| `MingTTSForConditionalGeneration` | Ming-omni-tts dense 0.5B | `inclusionAI/Ming-omni-tts-0.5B` | ✅︎ | | | ✅︎ | — |
+| `MingTTSForConditionalGeneration` | Ming-omni-tts dense 0.5B | `inclusionAI/Ming-omni-tts-0.5B` | ✅︎ | | | | — |
 | `GLMTTSForConditionalGeneration` | GLM-TTS | `zai-org/GLM-TTS` | ✅︎ | | | | — |
 | `MossTTSNanoForCausalLM` | MOSS-TTS-Nano | `OpenMOSS-Team/MOSS-TTS-Nano` | ✅︎ | | | | — |
 | `MossTTSDelayModel` | MOSS-TTS, MOSS-TTSD, MOSS-SoundEffect, MOSS-VoiceGenerator | `OpenMOSS-Team/MOSS-VoiceGenerator` | ✅︎ | | | | — |
@@ -101,10 +101,10 @@ th {
 | `HunyuanVideo15ImageToVideoPipeline` | HunyuanVideo-1.5-I2V | `hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-480p_i2v`, `hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-720p_i2v` | ✅︎ | ✅︎ | | | — |
 | `VoxtralTTSForConditionalGeneration` | Voxtral TTS | `mistralai/Voxtral-4B-TTS-2603` | ✅︎ | ✅︎ | | | — |
 | `CovoAudioForConditionalGeneration` | Covo-Audio-Chat | `tencent/Covo-Audio-Chat` | ✅︎ | | | | — |
-|`DyninOmniForConditionalGeneration` | Dynin-Omni | `snu-aidas/Dynin-Omni` | ✅︎ | | | ✅︎ | — |
+|`DyninOmniForConditionalGeneration` | Dynin-Omni | `snu-aidas/Dynin-Omni` | ✅︎ | | | | — |
 | `MiniCPMO45OmniForConditionalGeneration` | MiniCPM-o 4.5 | `openbmb/MiniCPM-o-4_5` | ✅︎ | | ✅︎ | | — |
 | `ErnieImagePipeline` | ERNIE-Image | `baidu/ERNIE-Image`, `baidu/ERNIE-Image-Turbo` | ✅︎ | ✅︎ | ✅︎ | ✅︎ | — |
-| `GepardTalkerForConditionalGeneration` | Gepard-1.0 | `nineninesix/gepard-1.0` | ✅︎ | | | ✅︎ | — |
+| `GepardTalkerForConditionalGeneration` | Gepard-1.0 | `nineninesix/gepard-1.0` | ✅︎ | | | | — |
 |`HiDreamImagePipeline` | HiDream-I1-Full | `HiDream-ai/HiDream-I1-Full` | ✅︎ | ✅︎ | | | — |
 | `HiDreamO1ImagePipeline` | HiDream-O1-Image | `HiDream-ai/HiDream-O1-Image` | ✅︎ | | | | — |

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 """
 Image-to-Video generation example using Wan2.2 I2V/TI2V models, LTX2/LTX-2.3,
@@ -111,13 +111,6 @@ def build_image_to_video_prompt(
    return result


-def _validate_video_output_type(output_type: str) -> None:
-    if output_type not in {"image", "video"}:
-        raise ValueError(
-            f"Unexpected output type '{output_type}', expected 'video' or legacy 'image' for video generation."
-        )
-
-
 def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
@@ -672,7 +665,10 @@ def main():
        frames = frames[0] if frames else None

    if isinstance(frames, OmniRequestOutput):
-        _validate_video_output_type(frames.final_output_type)
+        if frames.final_output_type != "image":
+            raise ValueError(
+                f"Unexpected output type '{frames.final_output_type}', expected 'image' for video generation."
+            )
        if frames.multimodal_output and "audio" in frames.multimodal_output:
            audio = frames.multimodal_output["audio"]
            audio_sample_rate = frames.multimodal_output.get("audio_sample_rate", audio_sample_rate)
@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 import argparse
 import json
@@ -174,13 +174,6 @@ def build_text_to_video_prompt(prompt: str, negative_prompt: str | None) -> dict
    return result


-def _validate_video_output_type(output_type: str) -> None:
-    if output_type not in {"image", "video"}:
-        raise ValueError(
-            f"Unexpected output type '{output_type}', expected 'video' or legacy 'image' for video generation."
-        )
-
-
 def _normalize_float_tensor(tensor: torch.Tensor, source_range: str) -> torch.Tensor:
    """Normalize decoded frames according to the pipeline's output contract."""
    if not tensor.is_floating_point():
@@ -627,7 +620,10 @@ def main():
        frames = frames[0] if frames else None

    if isinstance(frames, OmniRequestOutput):
-        _validate_video_output_type(frames.final_output_type)
+        if frames.final_output_type != "image":
+            raise ValueError(
+                f"Unexpected output type '{frames.final_output_type}', expected 'image' for video generation."
+            )
        if frames.multimodal_output and "audio" in frames.multimodal_output:
            audio = frames.multimodal_output["audio"]
            audio_sample_rate = frames.multimodal_output.get("audio_sample_rate", audio_sample_rate)
@@ -18,11 +18,11 @@ examples in this repository.

 Qwen3-TTS supports three task types, each backed by a dedicated model checkpoint:

-| Task Type     | Model                                    | Description                                                   |
-| ------------- | ---------------------------------------- | ------------------------------------------------------------- |
-| `CustomVoice` | `Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice`   | Predefined speaker voices with optional style/emotion control |
-| `VoiceDesign` | `Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign`   | Generate speech from a natural language voice description     |
-| `Base`        | `Qwen/Qwen3-TTS-12Hz-1.7B-Base`          | Voice cloning from reference audio + transcript               |
+| Task Type     | Model                                    | Description                                                  |
+| ------------- | ---------------------------------------- | ------------------------------------------------------------ |
+| `CustomVoice` | `Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice`  | Predefined speaker voices with optional style/emotion control |
+| `VoiceDesign` | `Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign`  | Generate speech from a natural language voice description     |
+| `Base`        | `Qwen/Qwen3-TTS-12Hz-1.7B-Base`         | Voice cloning from reference audio + transcript              |

 Smaller 0.6B variants are also available for `CustomVoice` and `Base`.

@@ -166,15 +166,6 @@ python examples/offline_inference/text_to_speech/qwen3_tts/end2end.py --query-ty
 - Key flags: `--omni` is required. `--deploy-config` points to the bundled two-stage pipeline config.
 - Async chunking: Enabled by default in `qwen3_tts.yaml` for streaming-friendly first-audio latency. Raw audio streaming requires `stream=true`, `stream_format="audio"`, and `response_format="pcm"`.
 - Task/model matching: Each task type requires its matching model checkpoint. Using a CustomVoice model for a Base (voice clone) request will fail.
- Base codec termination: Base requests without an explicit `max_new_tokens`
-  use a text-scaled safety ceiling (at least 192 codec frames and no more than
-  the configured model limit). If the Talker reaches that ceiling without
-  codec EOS, non-streaming serving discards the incomplete audio and retries
-  once with a fresh seed; an explicit `seed` or `max_new_tokens` disables the
-  retry. SSE and WebSocket clients receive structured errors and must discard
-  previously emitted audio when the error contains `"action":"discard"`;
-  raw PCM streams terminate with a connection error after any partial bytes
-  already sent.
 - Known limitations: The server serves one model variant at a time. To switch task types (e.g., CustomVoice to Base), restart the server with the corresponding model.

 ## Hardware Support
@@ -46,6 +46,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              3646.797
+            ],
            "mean_audio_rtf": [
              0.1098
            ]
@@ -79,6 +82,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              4059.0878
+            ],
            "mean_audio_rtf": [
              0.1335
            ]
@@ -114,6 +120,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              5639.5769
+            ],
            "mean_audio_rtf": [
              0.1798
            ]
@@ -45,6 +45,13 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              15793.1209,
+              30658.6632,
+              37463.4943,
+              47990.5839,
+              64980.2199
+            ],
            "mean_audio_rtf": [
              0.169,
              0.2041,
@@ -80,6 +87,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              4556.4347
+            ],
            "mean_audio_rtf": [
              0.138
            ]
@@ -113,6 +123,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              4768.0972
+            ],
            "mean_audio_rtf": [
              0.1609
            ]
@@ -148,6 +161,9 @@
        "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration",
        "baseline": {
          "H100": {
+            "mean_e2el_ms": [
+              6395.3517
+            ],
            "mean_audio_rtf": [
              0.2048
            ]
@@ -11,16 +11,13 @@ to the always-kept prefix segment before generating the block mask.
 """

 import dataclasses
-import sys
-import types

 import pytest
 import torch

-from vllm_omni.diffusion.attention.backends.abstract import AttentionMetadata, VideoTokenLayout, VideoTokenSpan
+from vllm_omni.diffusion.attention.backends.abstract import AttentionMetadata, VideoTokenLayout
 from vllm_omni.diffusion.attention.backends.rainfusion_attn import (
    _BLOCK_SIZE,
-    RainFusionAttentionBackend,
    RainFusionAttentionImpl,
    RainFusionPlan,
 )
@@ -100,62 +97,6 @@ def test_video_segment_must_be_the_tail_of_packed_document_zero():
    assert make_impl()._resolve_plan(metadata) is None


-def test_ref2va_multi_video_spans_resolve():
-    layout = VideoTokenLayout(
-        used_len=12000,
-        video_spans=(
-            VideoTokenSpan(start=128, latent_grid=(4, 16, 64), role="reference"),
-            VideoTokenSpan(start=5000, latent_grid=(4, 16, 64), role="target"),
-        ),
-    )
-    plan = make_impl()._resolve_plan(AttentionMetadata(extra={"max_seqlen_q": 12000}, video_layout=layout))
-
-    assert plan is not None
-    assert plan.used_len == 12000
-    assert plan.video_spans == [
-        {"start": 128, "latent_shape": [4, 16, 64]},
-        {"start": 5000, "latent_shape": [4, 16, 64]},
-    ]
-
-
-def test_invalid_ref2va_video_spans_fall_back_to_dense():
-    layout = VideoTokenLayout(
-        used_len=12000,
-        video_spans=(
-            VideoTokenSpan(start=128, latent_grid=(4, 16, 64), role="reference"),
-            VideoTokenSpan(start=4000, latent_grid=(4, 16, 64), role="target"),
-        ),
-    )
-    assert make_impl()._resolve_plan(AttentionMetadata(extra={"max_seqlen_q": 12000}, video_layout=layout)) is None
-
-
-def test_validate_available_rejects_legacy_mindiesd(monkeypatch):
-    mindiesd = types.ModuleType("mindiesd")
-
-    def sparse_attention(query, key, value, **kwargs):
-        return query
-
-    mindiesd.sparse_attention = sparse_attention
-    monkeypatch.setitem(sys.modules, "mindiesd", mindiesd)
-    monkeypatch.setattr("importlib.util.find_spec", lambda _: object())
-
-    with pytest.raises(ValueError, match="video_spans"):
-        RainFusionAttentionBackend.validate_available()
-
-
-def test_validate_available_accepts_new_mindiesd(monkeypatch):
-    mindiesd = types.ModuleType("mindiesd")
-
-    def sparse_attention(query, key, value, *, video_spans=None, **kwargs):
-        return query
-
-    mindiesd.sparse_attention = sparse_attention
-    monkeypatch.setitem(sys.modules, "mindiesd", mindiesd)
-    monkeypatch.setattr("importlib.util.find_spec", lambda _: object())
-
-    RainFusionAttentionBackend.validate_available()
-
-
@pytest.mark.parametrize("grid", [(4, 24, 40), (1, 24, 40)])
 def test_short_video_stays_dense(grid):
    assert make_impl()._resolve_plan(make_metadata(grid)) is None
@@ -1,62 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
-from collections.abc import Iterator
-
-import pytest
-import torch
-import torch.nn as nn
-
-from vllm_omni.diffusion.models.minimax_h3.encoder import MiniMaxH3Qwen3VLEncoder
-
-pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.diffusion]
-
-
-@pytest.fixture(autouse=True)
-def restore_cudnn_sdp_state() -> Iterator[None]:
-    original = torch.backends.cuda.cudnn_sdp_enabled()
-    try:
-        yield
-    finally:
-        torch.backends.cuda.enable_cudnn_sdp(original)
-
-
-def _encoder_without_weights() -> MiniMaxH3Qwen3VLEncoder:
-    encoder = MiniMaxH3Qwen3VLEncoder("unused", device=torch.device("cpu"), load_model=False)
-    encoder.text_model = nn.Linear(1, 1)
-    return encoder
-
-
-@pytest.mark.parametrize("initial_state", [False, True])
-def test_encode_ids_restores_cudnn_sdp_after_success(monkeypatch: pytest.MonkeyPatch, initial_state: bool) -> None:
-    encoder = _encoder_without_weights()
-    expected = torch.tensor([1.0])
-
-    def encode(*args, **kwargs):
-        assert torch.backends.cuda.cudnn_sdp_enabled()
-        return expected
-
-    monkeypatch.setattr(encoder, "_encode", encode)
-    torch.backends.cuda.enable_cudnn_sdp(initial_state)
-
-    actual = encoder.encode_ids(torch.tensor([1]))
-
-    torch.testing.assert_close(actual, expected)
-    assert torch.backends.cuda.cudnn_sdp_enabled() is initial_state
-
-
-@pytest.mark.parametrize("initial_state", [False, True])
-def test_encode_ids_restores_cudnn_sdp_after_failure(monkeypatch: pytest.MonkeyPatch, initial_state: bool) -> None:
-    encoder = _encoder_without_weights()
-
-    def fail(*args, **kwargs):
-        assert torch.backends.cuda.cudnn_sdp_enabled()
-        raise RuntimeError("encode failed")
-
-    monkeypatch.setattr(encoder, "_encode", fail)
-    torch.backends.cuda.enable_cudnn_sdp(initial_state)
-
-    with pytest.raises(RuntimeError, match="encode failed"):
-        encoder.encode_ids(torch.tensor([1]))
-
-    assert torch.backends.cuda.cudnn_sdp_enabled() is initial_state
@@ -74,8 +74,6 @@ def test_t2va_and_fl2va_packing_keep_update_rows_separate():
    assert fl2va["update_mask"].sum().item() == 12
    assert (~fl2va["update_mask"]).sum().item() == 6
    assert t2va["audio_pos"].numel() == fl2va["audio_pos"].numel() == 6
-    assert t2va["video_spans"] == ({"start": 10, "latent_grid": (2, 2, 3), "role": "target"},)
-    assert fl2va["video_spans"] == ({"start": 16, "latent_grid": (2, 2, 3), "role": "target"},)


 def test_ref2va_packing_tracks_video_and_audio_update_masks():
@@ -103,33 +101,6 @@ def test_ref2va_packing_tracks_video_and_audio_update_masks():
    assert packed["cu_seqlens"].tolist() == [0, 30, 64]


-def test_ref2va_packing_publishes_physical_video_spans_only():
-    from vllm_omni.diffusion.models.minimax_h3.packed_sequence import (
-        minimax_h3_packed_sequence_ref2va_blocks,
-    )
-
-    packed = minimax_h3_packed_sequence_ref2va_blocks(
-        text_len=4,
-        latent_t=2,
-        latent_h=4,
-        latent_w=6,
-        audio_t=3,
-        ref_blocks=[
-            {"kind": "image", "latent_h": 4, "latent_w": 4},
-            {"kind": "video_audio", "ref_audio_t": 2, "latent_t": 2, "latent_h": 4, "latent_w": 6},
-            {"kind": "audio", "ref_audio_t": 1},
-            {"kind": "video", "ref_audio_t": 0, "latent_t": 1, "latent_h": 4, "latent_w": 4},
-        ],
-    )
-
-    # Image and all audio rows intentionally stay out of the sparse spans.
-    assert packed["video_spans"] == (
-        {"start": 12, "latent_grid": (2, 2, 3), "role": "reference"},
-        {"start": 26, "latent_grid": (1, 2, 2), "role": "reference"},
-        {"start": 36, "latent_grid": (2, 2, 3), "role": "target"},
-    )
-
-
 def test_condition_noise_is_seeded_and_keeps_clean_anchor_at_timestep_one():
    from vllm_omni.diffusion.models.minimax_h3.condition_noise import (
        minimax_h3_audio_cond_noise_aug_rows,
@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 from pathlib import Path

@@ -187,8 +187,6 @@ def test_minimax_h3_quantization_quality(config: QualityTestConfig):
        "model": model_ref,
        "enforce_eager": True,
        "tensor_parallel_size": 2,
-        # The fused BF16 baseline only fits on H100-80GB with encoder TP.
-        "text_encoder_tp_size": 2,
        "vae_use_tiling": True,
    }

@@ -81,7 +81,7 @@ class QualityTestConfig:
    num_frames: int = 5  # only for t2v
    seed: int = 42
    gpu: str = "H100"  # minimum GPU requirement
-    negative_prompt: str | None = ""
+    negative_prompt: str = ""
    guidance_scale: float | None = None
    sigmas: list[float] | None = None
    enable_cpu_offload: bool = False
@@ -212,14 +212,24 @@ QUALITY_CONFIGS = [
        quantization="fp8",
        task="t2v",
        prompt="A serene lakeside sunrise with mist over the water",
-        max_lpips=0.20,
-        height=384,
-        width=512,
-        num_frames=73,
-        num_inference_steps=10,
-        # Do not override the recipe's negative conditioning, guidance, or
-        # scheduler trajectory: this gate follows the supported LTX default.
-        negative_prompt=None,
+        max_lpips=0.10,
+        height=256,
+        width=256,
+        num_frames=25,
+        num_inference_steps=8,
+        # Preserve the final scheduler trajectory used when this FP8 quality
+        # gate was established. Explicit LTX sigmas bypass dynamic shifting.
+        sigmas=[
+            1.0,
+            0.92185378074646,
+            0.8327768445014954,
+            0.7303033471107483,
+            0.6111654043197632,
+            0.4709382653236389,
+            0.30347853899002075,
+            0.10000002384185791,
+            0.0,
+        ],
    ),
 ]

@@ -286,11 +296,8 @@ def _generate_image(omni, config: QualityTestConfig):
    ).manual_seed(config.seed)
    torch.accelerator.reset_peak_memory_stats()

-    prompt = {"prompt": config.prompt}
-    if config.negative_prompt is not None:
-        prompt["negative_prompt"] = config.negative_prompt
    outputs = omni.generate(
-        prompt,
+        {"prompt": config.prompt, "negative_prompt": config.negative_prompt},
        OmniDiffusionSamplingParams(
            height=config.height,
            width=config.width,
@@ -321,11 +328,8 @@ def _generate_video(omni, config: QualityTestConfig):
    ).manual_seed(config.seed)
    torch.accelerator.reset_peak_memory_stats()

-    prompt = {"prompt": config.prompt}
-    if config.negative_prompt is not None:
-        prompt["negative_prompt"] = config.negative_prompt
    outputs = omni.generate(
-        prompt,
+        {"prompt": config.prompt, "negative_prompt": config.negative_prompt},
        OmniDiffusionSamplingParams(
            height=config.height,
            width=config.width,
@@ -461,7 +465,6 @@ def test_benchmark_generate_image_unwraps_nested_omni_request_output(monkeypatch


 def test_generate_video_forwards_sigmas(monkeypatch):
-    from vllm_omni.outputs import OmniRequestOutput
    from vllm_omni.platforms import current_omni_platform

    monkeypatch.setattr(current_omni_platform, "device_type", "cpu", raising=False)
@@ -472,12 +475,7 @@ def test_generate_video_forwards_sigmas(monkeypatch):
    class DummyOmni:
        def generate(self, _prompt, sampling):
            captured.sampling = sampling
-            return [
-                OmniRequestOutput.from_diffusion(
-                    request_id="req",
-                    images=[np.zeros((1, 2, 2, 3), dtype=np.float32)],
-                )
-            ]
+            return [SimpleNamespace(images=[np.zeros((1, 2, 2, 3), dtype=np.float32)])]

    config = QualityTestConfig(
        id="ltx-sigmas",
@@ -493,41 +491,6 @@ def test_generate_video_forwards_sigmas(monkeypatch):
    assert captured.sampling.sigmas == [1.0, 0.5]


-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_ltx_quality_gate_uses_official_eager_defaults(monkeypatch):
-    from vllm_omni.outputs import OmniRequestOutput
-    from vllm_omni.platforms import current_omni_platform
-
-    monkeypatch.setattr(current_omni_platform, "device_type", "cpu", raising=False)
-    monkeypatch.setattr(torch.accelerator, "reset_peak_memory_stats", lambda: None, raising=False)
-    monkeypatch.setattr(torch.accelerator, "max_memory_allocated", lambda: 0, raising=False)
-    captured = SimpleNamespace(prompt=None, sampling=None)
-
-    class DummyOmni:
-        def generate(self, prompt, sampling):
-            captured.prompt = prompt
-            captured.sampling = sampling
-            return [
-                OmniRequestOutput.from_diffusion(
-                    request_id="req",
-                    images=[np.zeros((1, 2, 2, 3), dtype=np.float32)],
-                )
-            ]
-
-    config = next(config for config in QUALITY_CONFIGS if config.id == "fp8_ltx2")
-    _generate_video(DummyOmni(), config)
-
-    assert (config.width, config.height, config.num_frames, config.num_inference_steps) == (512, 384, 73, 10)
-    assert config.max_lpips == 0.20
-    assert config.sigmas is None
-    assert config.guidance_scale is None
-    assert config.negative_prompt is None
-    assert "negative_prompt" not in captured.prompt
-    assert captured.sampling.sigmas is None
-    assert captured.sampling.guidance_scale is None
-
-
 _marks = hardware_marks(res={"cuda": "H100"})
 _OUTPUT_DIR = Path(os.environ["VLLM_OMNI_QUALITY_OUTPUT_DIR"]) if "VLLM_OMNI_QUALITY_OUTPUT_DIR" in os.environ else None

@@ -538,6 +501,8 @@ def _quality_param(c: QualityTestConfig):
        marks.append(
            pytest.mark.skip(reason="Qwen-Image FP8 quality gate temporarily disabled (see CI / issue tracker).")
        )
+    if c.id == "fp8_ltx2":
+        marks.append(pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/6245"))
    return pytest.param(c, id=c.id, marks=marks)


@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 """Unit tests for acknowledged abort on AsyncOmniEngine."""

 from __future__ import annotations
@@ -22,7 +19,6 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]

 def _make_engine(request_q: queue.Queue, rpc_q: queue.Queue) -> AsyncOmniEngine:
    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
    engine.request_queue = SimpleNamespace(sync_q=request_q)
    engine.rpc_output_queue = SimpleNamespace(sync_q=rpc_q)
    engine._correlated_rpc_client = CorrelatedRpcClient(request_q, rpc_q)
@@ -158,110 +154,3 @@ def test_abort_async_preserves_request_queue_backpressure(mocker: MockerFixture)
        asyncio.run(_run())
    finally:
        engine._correlated_rpc_client.close()
-
-
-def test_abort_async_is_noop_after_shutdown_starts():
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = True
-    engine.request_queue = None
-    engine._correlated_rpc_client = None
-
-    asyncio.run(engine.abort_async(["req-late"]))
-
-
-def test_abort_async_is_noop_for_empty_request_ids():
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = None
-    engine._correlated_rpc_client = None
-
-    asyncio.run(engine.abort_async([]))
-
-
-def test_abort_async_tolerates_request_queue_close_during_shutdown(
-    mocker: MockerFixture,
-):
-    class SyntheticSyncQueueShutDownError(Exception):
-        pass
-
-    rpc_q: queue.Queue = queue.Queue()
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-
-    class ClosingRequestQueue:
-        def put(self, item) -> None:
-            del item
-            engine._shutdown_called = True
-            raise SyntheticSyncQueueShutDownError
-
-    request_q = ClosingRequestQueue()
-    engine.request_queue = SimpleNamespace(sync_q=request_q)
-    engine._correlated_rpc_client = CorrelatedRpcClient(request_q, rpc_q)
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        SyntheticSyncQueueShutDownError,
-    )
-
-    try:
-        asyncio.run(engine.abort_async(["req-submit-race"], timeout=1))
-    finally:
-        engine._correlated_rpc_client.close()
-
-
-def test_abort_async_surfaces_request_queue_close_while_engine_is_live(
-    mocker: MockerFixture,
-):
-    class SyntheticSyncQueueShutDownError(Exception):
-        pass
-
-    class ClosedRequestQueue:
-        def put(self, item) -> None:
-            del item
-            raise SyntheticSyncQueueShutDownError
-
-    request_q = ClosedRequestQueue()
-    rpc_q: queue.Queue = queue.Queue()
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = SimpleNamespace(sync_q=request_q)
-    engine._correlated_rpc_client = CorrelatedRpcClient(request_q, rpc_q)
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        SyntheticSyncQueueShutDownError,
-    )
-
-    try:
-        with pytest.raises(SyntheticSyncQueueShutDownError):
-            asyncio.run(engine.abort_async(["req-live-submit"], timeout=1))
-    finally:
-        engine._correlated_rpc_client.close()
-
-
-def test_abort_async_tolerates_rpc_router_close_during_shutdown():
-    request_q: queue.Queue = queue.Queue()
-    rpc_q: queue.Queue = queue.Queue()
-    engine = _make_engine(request_q, rpc_q)
-
-    async def _run() -> None:
-        task = asyncio.create_task(engine.abort_async(["req-ack-race"], timeout=1))
-        await asyncio.to_thread(request_q.get, True, 1)
-        engine._shutdown_called = True
-        engine._correlated_rpc_client.close()
-        await task
-
-    asyncio.run(_run())
-
-
-def test_abort_async_surfaces_rpc_router_close_while_engine_is_live():
-    request_q: queue.Queue = queue.Queue()
-    rpc_q: queue.Queue = queue.Queue()
-    engine = _make_engine(request_q, rpc_q)
-
-    async def _run() -> None:
-        task = asyncio.create_task(engine.abort_async(["req-live"], timeout=1))
-        await asyncio.to_thread(request_q.get, True, 1)
-        engine._correlated_rpc_client.close()
-        with pytest.raises(RuntimeError, match="RPC result router closed"):
-            await task
-
-    asyncio.run(_run())
@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 import pytest
 import torch
 from pytest_mock import MockerFixture
@@ -17,104 +14,6 @@ from vllm_omni.model_executor.stage_input_processors.bagel import ExpandedPrompt
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]


-class _SyntheticSyncQueueShutDownError(Exception):
-    pass
-
-
-def test_abort_is_noop_for_empty_request_ids(mocker: MockerFixture):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = mocker.Mock()
-
-    engine.abort([])
-
-    engine.request_queue.sync_q.put.assert_not_called()
-
-
-def test_abort_is_noop_after_shutdown_starts(mocker: MockerFixture):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = True
-    engine.request_queue = mocker.Mock()
-
-    engine.abort(["req-1"])
-
-    engine.request_queue.sync_q.put.assert_not_called()
-
-
-def test_abort_tolerates_queue_close_race_during_shutdown(mocker: MockerFixture):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = mocker.Mock()
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        _SyntheticSyncQueueShutDownError,
-    )
-
-    def close_queue_during_abort(*args, **kwargs):
-        del args, kwargs
-        engine._shutdown_called = True
-        raise _SyntheticSyncQueueShutDownError
-
-    engine.request_queue.sync_q.put.side_effect = close_queue_during_abort
-
-    engine.abort(["req-1"])
-
-
-def test_abort_tolerates_legacy_janus_close_error_during_shutdown(mocker: MockerFixture):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = mocker.Mock()
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        None,
-    )
-
-    def close_queue_during_abort(*args, **kwargs):
-        del args, kwargs
-        engine._shutdown_called = True
-        raise RuntimeError("Operation on the closed queue is forbidden")
-
-    engine.request_queue.sync_q.put.side_effect = close_queue_during_abort
-
-    engine.abort(["req-1"])
-
-
-def test_abort_surfaces_unrelated_legacy_runtime_error_during_shutdown(
-    mocker: MockerFixture,
-):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = mocker.Mock()
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        None,
-    )
-
-    def fail_during_abort(*args, **kwargs):
-        del args, kwargs
-        engine._shutdown_called = True
-        raise RuntimeError("unrelated queue failure")
-
-    engine.request_queue.sync_q.put.side_effect = fail_during_abort
-
-    with pytest.raises(RuntimeError, match="unrelated queue failure"):
-        engine.abort(["req-1"])
-
-
-def test_abort_surfaces_unexpected_closed_queue(mocker: MockerFixture):
-    engine = object.__new__(AsyncOmniEngine)
-    engine._shutdown_called = False
-    engine.request_queue = mocker.Mock()
-    mocker.patch(
-        "vllm_omni.engine.async_engine_utils._JANUS_SYNC_QUEUE_SHUTDOWN",
-        _SyntheticSyncQueueShutDownError,
-    )
-    engine.request_queue.sync_q.put.side_effect = _SyntheticSyncQueueShutDownError
-
-    with pytest.raises(_SyntheticSyncQueueShutDownError):
-        engine.abort(["req-1"])
-
-
 def _make_engine_core_request(request_id: str = "req-1") -> EngineCoreRequest:
    return EngineCoreRequest(
        request_id=request_id,
@@ -2216,12 +2216,7 @@ def test_stage_pool_metrics_use_resumable_segment_token_count() -> None:
    )
    output = SimpleNamespace(
        request_id="req-stream",
-        outputs=[
-            SimpleNamespace(
-                cumulative_token_ids=list(range(11)),
-                finish_reason=FinishReason.LENGTH,
-            )
-        ],
+        outputs=[SimpleNamespace(cumulative_token_ids=list(range(11)))],
    )

    metrics = pool.build_stage_metrics(
@@ -2233,7 +2228,6 @@ def test_stage_pool_metrics_use_resumable_segment_token_count() -> None:

    assert metrics.num_tokens_out == 3
    assert metrics.output_unit_count == 3
-    assert metrics.finish_reason == "length"


 def test_image_ttfo_preserves_request_time_and_tracks_stage_time() -> None:
@@ -398,12 +398,6 @@ async def test_api_server_assembly_replaces_upstream_routes_and_mounts_omni_rout
    async def fake_omni_init_app_state(engine_client, state, args):
        state.engine_client = engine_client
        state.initialized_by_omni = True
-        state.openai_serving_video = SimpleNamespace(
-            shutdown=lambda: captured.__setitem__("video_shutdown", True),
-        )
-        state.openai_serving_speech = SimpleNamespace(
-            shutdown=lambda: captured.__setitem__("speech_shutdown", True),
-        )

    async def fake_storage_start():
        captured["storage_started"] = True
@@ -439,8 +433,6 @@ async def test_api_server_assembly_replaces_upstream_routes_and_mounts_omni_rout
    assert captured["supported_tasks"] == ("generate",)
    assert captured["storage_started"] is True
    assert captured["restrictions"] == {}
-    assert captured["video_shutdown"] is True
-    assert captured["speech_shutdown"] is True
    assert sock.closed is True
    assert served_app.state.initialized_by_omni is True

@@ -747,11 +739,7 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
    )

    state = State()
-    await api_server.omni_init_app_state(
-        engine,
-        state,
-        _minimal_args(),
-    )
+    await api_server.omni_init_app_state(engine, state, _minimal_args())

    _assert_app_state_snapshot(
        state,
@@ -762,55 +750,6 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
    assert state.diffusion_engine is engine


-@pytest.mark.asyncio
-async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) -> None:
-    stage = SimpleNamespace(engine_args={})
-    engine = _FakeEngineClient(stage_configs=[stage])
-    speech_kwargs = {}
-
-    def _for_diffusion_factory(label: str):
-        @classmethod
-        def _factory(cls, *args, **kwargs):
-            return _marker(label)
-
-        return _factory
-
-    @classmethod
-    def _speech_factory(cls, *args, **kwargs):
-        speech_kwargs.update(kwargs)
-        return _marker("speech")
-
-    monkeypatch.setattr(api_server, "get_stage_type", lambda _cfg: "diffusion")
-    monkeypatch.setattr(api_server.OmniOpenAIServingChat, "for_diffusion", _for_diffusion_factory("chat"))
-    monkeypatch.setattr(api_server.OmniOpenAIServingChatBatch, "for_diffusion", _for_diffusion_factory("chat_batch"))
-    monkeypatch.setattr(
-        api_server.OmniOpenAIServingAudioGenerate,
-        "for_diffusion",
-        _for_diffusion_factory("audio_generate"),
-    )
-    monkeypatch.setattr(api_server.OmniOpenAIServingVideo, "for_diffusion", _for_diffusion_factory("video"))
-    monkeypatch.setattr(api_server.OmniStreamingVideoOutputHandler, "__init__", lambda self, *a, **k: None)
-    monkeypatch.setattr(api_server.OmniOpenAIServingSpeech, "for_diffusion", _speech_factory)
-    monkeypatch.setattr(
-        api_server.ServingRealtimeRobotOpenPI,
-        "create_policy_server",
-        classmethod(lambda cls, *a, **k: _marker("openpi")),
-    )
-
-    state = State()
-    await api_server.omni_init_app_state(
-        engine,
-        state,
-        _minimal_args(
-            allowed_local_media_path="/allowed/media",
-            allowed_media_domains=["media.example.com"],
-        ),
-    )
-
-    assert speech_kwargs["allowed_local_media_path"] == "/allowed/media"
-    assert speech_kwargs["allowed_media_domains"] == ["media.example.com"]
-
-
@pytest.mark.asyncio
 async def test_multistage_app_state_key_snapshot(monkeypatch) -> None:
    """Lock multi-stage ``app.state`` keys after init, including live vs None.
@@ -45,15 +45,9 @@ from vllm_omni.entrypoints.openai.serving_speech import (
    OmniOpenAIServingSpeech,
    _create_wav_header,
 )
-from vllm_omni.entrypoints.openai.tts_adapters.base import (
-    DEFAULT_TTS_LANGUAGES,
-    PreparedRequest,
-    SpeechServingContext,
-    TTSGenerationError,
-)
+from vllm_omni.entrypoints.openai.tts_adapters.base import DEFAULT_TTS_LANGUAGES, PreparedRequest, SpeechServingContext
 from vllm_omni.entrypoints.openai.tts_adapters.capabilities import load_supported_speakers
 from vllm_omni.entrypoints.openai.tts_adapters.ming_tts import MingTTSAdapter
-from vllm_omni.entrypoints.openai.tts_adapters.qwen3_tts import Qwen3TTSCodecLimitError
 from vllm_omni.entrypoints.openai.tts_adapters.voxtral import VoxtralTTSAdapter
 from vllm_omni.model_executor.models.fish_speech.prompt_utils import (
    FISH_TEXT_ONLY_SYSTEM_PROMPT,
@@ -1251,46 +1245,6 @@ class TestTTSMethods:
            cache_key
        ) == speech_server._make_ref_audio_artifact_cache_key(np.asarray(first[0], dtype=np.float32), 24000)

-    @pytest.mark.asyncio
-    async def test_diffusion_ref_audio_uses_media_access_config(
-        self,
-        mocker: MockerFixture,
-        monkeypatch,
-        tmp_path,
-    ):
-        monkeypatch.setenv("SPEAKER_SAMPLES_DIR", str(tmp_path / "speakers"))
-        connector = mocker.MagicMock()
-        connector.fetch_audio_async = mocker.AsyncMock(return_value=(np.zeros(24000, dtype=np.float32), 24000))
-        connector_cls = mocker.patch.object(
-            serving_speech_module,
-            "MediaConnector",
-            return_value=connector,
-        )
-        media_dir = tmp_path / "media"
-        media_dir.mkdir()
-        ref_audio = media_dir / "reference.wav"
-        ref_audio.write_bytes(b"v1")
-
-        server = OmniOpenAIServingSpeech.for_diffusion(
-            diffusion_engine=mocker.MagicMock(),
-            model_name="test-model",
-            allowed_local_media_path=str(media_dir),
-            allowed_media_domains=["media.example.com"],
-        )
-        try:
-            first = await server._resolve_ref_audio(ref_audio.as_uri())
-            ref_audio.write_bytes(b"v2-with-different-size")
-            second = await server._resolve_ref_audio(ref_audio.as_uri())
-        finally:
-            server.shutdown()
-
-        connector_cls.assert_called_once_with(
-            allowed_local_media_path=str(media_dir),
-            allowed_media_domains=["media.example.com"],
-        )
-        assert first[2] != second[2]
-        assert connector.fetch_audio_async.await_count == 2
-
    # ── ref-audio cache key tests (static helper) ──

    def test_local_file_cache_key_invalidation(self):
@@ -2934,40 +2888,6 @@ class TestStreamingResponse:
        assert "text/event-stream" not in response.headers["content-type"]
        assert len(response.content) > 0

-    def test_raw_stream_passes_tts_params_to_common_guard(self, streaming_app, monkeypatch):
-        """The raw route must not bypass the shared Qwen3-TTS EOS guard."""
-        speech_server = streaming_app.state.speech_server
-        finalized_tts_params = {"_qwen3_tts_effective_max_tokens": [192]}
-        captured: dict = {}
-
-        async def prepare(_request, request_id=None):
-            return request_id, object(), finalized_tts_params
-
-        async def generate_chunks(
-            _generator,
-            _request_id,
-            _response_format="pcm",
-            raw_request=None,
-            request_start_s=None,
-            include_sample_rate=False,
-            usage_acc=None,
-            tts_params=None,
-            collect=None,
-        ):
-            captured["tts_params"] = tts_params
-            yield b"\x00\x00"
-
-        monkeypatch.setattr(speech_server, "_prepare_speech_generation", prepare)
-        monkeypatch.setattr(speech_server, "_generate_audio_chunks", generate_chunks)
-
-        response = TestClient(streaming_app).post(
-            "/v1/audio/speech",
-            json={"input": "Hello", "stream_format": "audio", "response_format": "pcm"},
-        )
-
-        assert response.status_code == 200
-        assert captured["tts_params"] is finalized_tts_params
-
    def test_sse_streaming(self, streaming_app):
        """stream_format=sse without stream=True returns audio deltas as SSE."""
        client = TestClient(streaming_app)
@@ -4888,57 +4808,6 @@ class TestTTSAsyncOffloading:
        assert ("artifact-a", False) not in qwen3_tts_server._ref_audio_model_artifact_ready
        assert "artifact-b" in {entry[3] for entry in qwen3_tts_server._ref_audio_resolve_cache.values()}

-    @pytest.mark.asyncio
-    async def test_qwen3_tts_nonstream_retries_codec_limit_once(self, qwen3_tts_server, mocker: MockerFixture) -> None:
-        qwen3_tts_server._check_model = mocker.AsyncMock(return_value=None)
-        qwen3_tts_server._generate_audio_bytes = mocker.AsyncMock(
-            side_effect=[
-                Qwen3TTSCodecLimitError("codec limit"),
-                (b"RIFF" + b"\x00" * 32, "audio/wav"),
-            ]
-        )
-        request = OpenAICreateSpeechRequest(input="Hello", task_type="Base")
-
-        response = await qwen3_tts_server.create_speech(request)
-
-        assert response.status_code == 200
-        assert qwen3_tts_server._generate_audio_bytes.await_count == 2
-        first_call, retry_call = qwen3_tts_server._generate_audio_bytes.await_args_list
-        assert first_call.args[0] is request
-        assert first_call.kwargs["request_id"] != retry_call.kwargs["request_id"]
-        assert retry_call.args[0].seed is not None
-        assert request.seed is None
-
-    @pytest.mark.asyncio
-    async def test_nonstream_does_not_retry_nonretryable_generation_error(
-        self, qwen3_tts_server, mocker: MockerFixture
-    ) -> None:
-        qwen3_tts_server._check_model = mocker.AsyncMock(return_value=None)
-        qwen3_tts_server._generate_audio_bytes = mocker.AsyncMock(side_effect=TTSGenerationError("invalid generation"))
-
-        response = await qwen3_tts_server.create_speech(OpenAICreateSpeechRequest(input="Hello", task_type="Base"))
-
-        assert response.status_code == 500
-        qwen3_tts_server._generate_audio_bytes.assert_awaited_once()
-
-    @pytest.mark.asyncio
-    @pytest.mark.parametrize("request_kwargs", [{"seed": 100423}, {"max_new_tokens": 192}])
-    async def test_qwen3_tts_nonstream_does_not_retry_explicit_sampling_controls(
-        self,
-        qwen3_tts_server,
-        mocker: MockerFixture,
-        request_kwargs: dict[str, int],
-    ) -> None:
-        qwen3_tts_server._check_model = mocker.AsyncMock(return_value=None)
-        qwen3_tts_server._generate_audio_bytes = mocker.AsyncMock(side_effect=Qwen3TTSCodecLimitError("codec limit"))
-
-        response = await qwen3_tts_server.create_speech(
-            OpenAICreateSpeechRequest(input="Hello", task_type="Base", **request_kwargs)
-        )
-
-        assert response.status_code == 500
-        qwen3_tts_server._generate_audio_bytes.assert_awaited_once()
-
    @pytest.mark.asyncio
    async def test_generate_audio_chunks_discards_ref_audio_artifact_warmup_on_error(self, qwen3_tts_server):
        async def failing_generator():
@@ -4953,80 +4822,6 @@ class TestTTSAsyncOffloading:
        assert "req-fail" not in qwen3_tts_server._request_ref_audio_artifact_keys
        assert ("artifact-fail", False) not in qwen3_tts_server._ref_audio_model_artifact_ready

-    @pytest.mark.asyncio
-    async def test_generate_audio_chunks_rejects_qwen3_codec_limit_before_success(self, qwen3_tts_server):
-        async def length_limited_generator():
-            yield SimpleNamespace(
-                multimodal_output={"audio": torch.zeros(16, dtype=torch.float32), "sr": 24000},
-                metrics={
-                    "stage_metrics": {
-                        "0": {
-                            "num_tokens_out": 191,
-                            "finish_reason": "length",
-                        }
-                    }
-                },
-            )
-
-        chunks = qwen3_tts_server._generate_audio_chunks(
-            length_limited_generator(),
-            "req-codec-limit",
-            tts_params={"task_type": ["Base"], "_qwen3_tts_effective_max_tokens": [192]},
-        )
-        with pytest.raises(Qwen3TTSCodecLimitError, match="191/192"):
-            async for _ in chunks:
-                pass
-
-    @pytest.mark.asyncio
-    async def test_sse_codec_limit_marks_partial_audio_for_discard(self, qwen3_tts_server):
-        async def length_limited_generator():
-            yield SimpleNamespace(
-                multimodal_output={"audio": torch.zeros(16, dtype=torch.float32), "sr": 24000},
-                metrics={
-                    "stage_metrics": {
-                        "0": {
-                            "num_tokens_out": 191,
-                            "finish_reason": "length",
-                        }
-                    }
-                },
-            )
-
-        events = [
-            event
-            async for event in qwen3_tts_server._generate_audio_sse_events(
-                length_limited_generator(),
-                "req-sse-codec-limit",
-                tts_params={"task_type": ["Base"], "_qwen3_tts_effective_max_tokens": [192]},
-            )
-        ]
-
-        assert any("event: speech.audio.delta" in event for event in events)
-        assert not any("event: speech.audio.done" in event for event in events)
-        error_event = next(event for event in events if "event: speech.audio.error" in event)
-        payload = json.loads(next(line for line in error_event.splitlines() if line.startswith("data: "))[6:])
-        assert payload["error"]["partial_audio"] is True
-        assert payload["error"]["action"] == "discard"
-
-    @pytest.mark.asyncio
-    async def test_sse_error_before_audio_is_not_marked_partial(self, qwen3_tts_server):
-        async def failing_generator():
-            raise RuntimeError("boom")
-            yield  # pragma: no cover
-
-        events = [
-            event
-            async for event in qwen3_tts_server._generate_audio_sse_events(
-                failing_generator(),
-                "req-sse-no-audio",
-            )
-        ]
-
-        error_event = next(event for event in events if "event: speech.audio.error" in event)
-        payload = json.loads(next(line for line in error_event.splitlines() if line.startswith("data: "))[6:])
-        assert "partial_audio" not in payload["error"]
-        assert "action" not in payload["error"]
-
    @pytest.mark.asyncio
    async def test_generate_audio_chunks_discards_ref_audio_artifact_warmup_on_close(self, qwen3_tts_server):
        async def pcm_generator():
@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 import asyncio
 import base64
 from types import SimpleNamespace
@@ -47,9 +44,7 @@ def _build_test_app(
        speech_service._prepare_speech_generation = mocker.AsyncMock(return_value=("req-1", object(), {}))
        speech_service.forced_aligner_enabled = False

-        async def mock_generate_pcm_chunks(
-            _generator, _request_id, *, include_sample_rate=False, tts_params=None, collect=None
-        ):
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, collect=None):
            for chunk in (b"\x01\x02", b"\x03\x04\x05"):
                yield (chunk, 24000) if include_sample_rate else chunk

@@ -249,7 +244,6 @@ class TestStreamingSpeechWebSocket:

    def test_streaming_multiple_binary_frames(self, mocker: MockerFixture):
        captured_requests = []
-        captured_tts_params = []

        speech_service = mocker.MagicMock(spec=OmniOpenAIServingSpeech)
        speech_service._generate_audio_bytes = mocker.AsyncMock(return_value=(b"", "audio/wav"))
@@ -259,12 +253,11 @@ class TestStreamingSpeechWebSocket:

        async def mock_prepare_speech_generation(request):
            captured_requests.append(request)
-            return "req-stream", object(), {"_qwen3_tts_effective_max_tokens": [192]}
+            return "req-stream", object(), {}

        speech_service._prepare_speech_generation = mock_prepare_speech_generation

-        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, tts_params=None):
-            captured_tts_params.append(tts_params)
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False):
            for chunk in (b"\x01\x02", b"\x03\x04\x05", b"\x06"):
                yield (chunk, 24000) if include_sample_rate else chunk

@@ -309,7 +302,6 @@ class TestStreamingSpeechWebSocket:
        assert captured_requests[0].stream is True
        assert captured_requests[0].response_format == "pcm"
        assert captured_requests[0].initial_codec_chunk_frames == 12
-        assert captured_tts_params == [{"_qwen3_tts_effective_max_tokens": [192]}]
        assert speech_service._generate_audio_bytes.await_count == 0

    def test_word_timestamps_requires_configured_aligner(self, mocker: MockerFixture):
@@ -352,9 +344,7 @@ class TestStreamingSpeechWebSocket:

        # The forced-aligner stage rides the same generator: its pooling output
        # is surfaced via the ``collect`` channel once the audio has streamed.
-        async def mock_generate_pcm_chunks(
-            _generator, _request_id, *, include_sample_rate=False, tts_params=None, collect=None
-        ):
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, collect=None):
            for chunk in (first_chunk, second_chunk):
                yield (chunk, 1000) if include_sample_rate else chunk
            if collect is not None:
@@ -437,9 +427,7 @@ class TestStreamingSpeechWebSocket:
        speech_service.forced_aligner_enabled = True
        speech_service._prepare_speech_generation = mocker.AsyncMock(return_value=("req", object(), {}))

-        async def mock_generate_pcm_chunks(
-            _generator, _request_id, *, include_sample_rate=False, tts_params=None, collect=None
-        ):
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, collect=None):
            chunk = b"\x01" * 1000
            yield (chunk, 1000) if include_sample_rate else chunk
            if collect is not None:
@@ -620,7 +608,7 @@ class TestStreamingSpeechWebSocket:
        speech_service.engine_client.abort = mocker.AsyncMock()
        speech_service.forced_aligner_enabled = False

-        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, tts_params=None):
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False):
            yield b"\x01\x02"
            raise RuntimeError("stream boom")

@@ -645,8 +633,6 @@ class TestStreamingSpeechWebSocket:
                assert ws.receive_json() == {
                    "type": "error",
                    "message": "Generation failed for utterance 0, sentence 0: stream boom",
-                    "partial_audio": True,
-                    "action": "discard",
                }
                assert ws.receive_json() == {
                    "type": "audio.done",
@@ -716,7 +702,7 @@ class TestStreamingSpeechWebSocket:
        speech_service.engine_client.abort = mocker.AsyncMock()
        speech_service.forced_aligner_enabled = False

-        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False, tts_params=None):
+        async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_rate=False):
            yield b"\x01\x02"

        speech_service._generate_pcm_chunks = mock_generate_pcm_chunks
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Unit tests for speech token-usage accounting (issue #4646).

 These exercise the pure usage logic without a model/server: input-token
@@ -206,14 +205,6 @@ def test_output_tokens_from_stage_metrics():
    assert acc.total() == 77


-def test_output_tokens_preserves_stage0_finish_reason():
-    acc = SpeechOutputTokenCounter()
-    res = _final_res(stage0_out=128)
-    res.metrics["stage_metrics"]["0"]["finish_reason"] = "length"
-    acc.observe(res)
-    assert acc.stage0_finish_reason == "length"
-
-
 def test_output_tokens_takes_max_across_stages():
    acc = SpeechOutputTokenCounter()
    acc.observe(_final_res(stage0_out=42, stage1_out=0))
@@ -10,14 +10,11 @@ from types import SimpleNamespace
 from typing import Any

 import pytest
-from vllm.sampling_params import SamplingParams

-from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
 from vllm_omni.entrypoints.openai.tts_adapters import (
    TTS_ADAPTER_REGISTRY,
    ARTTSAdapter,
    DiffusionTTSAdapter,
-    SpeechServingContext,
    all_tts_model_types,
    detect_tts_model_type,
    resolve_adapter,
@@ -32,11 +29,7 @@ from vllm_omni.entrypoints.openai.tts_adapters.moss_tts import (
    MossTTSAdapter,
    MossTTSNanoAdapter,
 )
-from vllm_omni.entrypoints.openai.tts_adapters.qwen3_tts import (
-    QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY,
-    Qwen3TTSAdapter,
-    Qwen3TTSCodecLimitError,
-)
+from vllm_omni.entrypoints.openai.tts_adapters.qwen3_tts import Qwen3TTSAdapter
 from vllm_omni.model_executor.models.indextts2 import prompt_utils
 from vllm_omni.model_executor.models.indextts2.tokenizer_v2_5 import (
    INDEXTTS25_TOKENIZER_FILE,
@@ -122,100 +115,11 @@ def test_moss_tts_applies_request_max_new_tokens(adapter_cls):
    assert stage_defaults[0].max_tokens == 4096


-def _build_moss_tts_request(adapter_cls, mocker, *, request_seed):
-    server = mocker.Mock()
-    server._build_moss_tts_params = mocker.AsyncMock(return_value={})
-    adapter = adapter_cls(SpeechServingContext(server=server))
-    request = OpenAICreateSpeechRequest(input="hello", seed=request_seed)
-
-    return asyncio.run(
-        adapter.build(
-            request,
-            [SamplingParams(seed=42)],
-            has_inline_ref_audio=False,
-        )
-    )
-
-
-# Full-family coverage pins the adapter contract; only Nano consumes this seed end to end today.
-@pytest.mark.parametrize("adapter_cls", [MossTTSAdapter, MossTTSNanoAdapter])
-@pytest.mark.parametrize("request_seed", [0, 1234])
-def test_moss_tts_request_seed_overrides_stage_default(adapter_cls, request_seed, mocker):
-    prepared = _build_moss_tts_request(
-        adapter_cls,
-        mocker,
-        request_seed=request_seed,
-    )
-
-    assert prepared.tts_params["seed"] == [request_seed]
-
-
-@pytest.mark.parametrize("adapter_cls", [MossTTSAdapter, MossTTSNanoAdapter])
-def test_moss_tts_seed_falls_back_to_stage_default(adapter_cls, mocker):
-    prepared = _build_moss_tts_request(
-        adapter_cls,
-        mocker,
-        request_seed=None,
-    )
-
-    assert prepared.tts_params["seed"] == [42]
-
-
 def test_qwen3_tts_metadata():
    assert Qwen3TTSAdapter.backend == "ar"
    assert issubclass(Qwen3TTSAdapter, ARTTSAdapter)


-@pytest.mark.parametrize(
-    ("task_type", "text_tokens", "request_cap", "expected_cap"),
-    [
-        ("Base", 0, None, 4096),
-        ("Base", 10, None, 192),
-        ("Base", 23, None, 276),
-        ("Base", 23, 128, 128),
-        ("Base", 23, 512, 512),
-        ("Base", 400, None, 4096),
-        ("CustomVoice", 10, None, 4096),
-        ("CustomVoice", 10, 256, 256),
-    ],
-)
-def test_qwen3_tts_applies_text_scaled_codec_safety_limit(task_type, text_tokens, request_cap, expected_cap, mocker):
-    server = mocker.Mock()
-    server._count_usage_text_tokens.return_value = text_tokens
-    adapter = Qwen3TTSAdapter(SpeechServingContext(server=server))
-    stage_defaults = [SamplingParams(max_tokens=4096, min_tokens=2)]
-    request = OpenAICreateSpeechRequest(
-        input="test text",
-        task_type=task_type,
-        max_new_tokens=request_cap,
-    )
-    prompt: dict[str, Any] = {"additional_information": {}}
-
-    overridden = adapter.apply_sampling_overrides(stage_defaults, request, prompt)
-
-    assert overridden[0].max_tokens == expected_cap
-    assert prompt["additional_information"][QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY] == [expected_cap]
-    assert stage_defaults[0].max_tokens == 4096
-
-
-def test_qwen3_tts_rejects_only_length_finished_base_audio(mocker):
-    adapter = Qwen3TTSAdapter(SpeechServingContext(server=mocker.Mock()))
-    params = {
-        "task_type": ["Base"],
-        QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY: [192],
-    }
-
-    # EOS at the exact budget is valid; the token count alone is not a
-    # sufficient failure signal.
-    adapter.validate_generation(params, stage0_finish_reason="stop", output_tokens=192)
-    adapter.validate_generation(params, stage0_finish_reason=None, output_tokens=192)
-
-    # Some frontends expose 191 decoded frames for max_new_tokens=192. The
-    # engine terminal reason still makes this an unambiguous limit failure.
-    with pytest.raises(Qwen3TTSCodecLimitError, match="191/192"):
-        adapter.validate_generation(params, stage0_finish_reason="length", output_tokens=191)
-
-
 def test_indextts_adapters_are_versioned():
    assert resolve_adapter("indextts2") is IndexTTS2Adapter
    assert resolve_adapter("indextts2_5") is IndexTTS25Adapter
@@ -3,7 +3,6 @@
 """Unit tests for OpenAI-compatible video API encoding helpers."""

 import base64
-import threading
 from io import BytesIO

 import av
@@ -250,95 +249,6 @@ def test_channel_first_video_tensor_uses_direct_planar_mux(monkeypatch):
    assert options["fps"] == 12.0


-def test_planar_converter_serial_and_parallel_outputs_match():
-    frames = [np.full((2, 3, 3), fill_value=value / 10, dtype=np.float32) for value in (1, 4, 7)]
-
-    serial_converter = video_api_utils._PlanarFrameConverter(max_workers=1)
-    parallel_converter = video_api_utils._PlanarFrameConverter(max_workers=2)
-    try:
-        serial = [
-            frame.to_ndarray(format="rgb24") for frame in serial_converter.iter_frames(frames, np.dtype(np.float32))
-        ]
-        parallel = [
-            frame.to_ndarray(format="rgb24") for frame in parallel_converter.iter_frames(frames, np.dtype(np.float32))
-        ]
-    finally:
-        serial_converter.shutdown()
-        parallel_converter.shutdown()
-
-    np.testing.assert_array_equal(np.stack(serial), np.stack(parallel))
-
-
-def test_planar_converter_preserves_fifo_order_and_bounds_pending_futures(monkeypatch):
-    converter = video_api_utils._PlanarFrameConverter(max_workers=2)
-    started = threading.Event()
-    release = threading.Event()
-    four_submitted = threading.Event()
-    outstanding = 0
-    peak_outstanding = 0
-    lock = threading.Lock()
-    real_submit = converter._executor.submit
-
-    def fake_build(frame, _common_dtype):
-        started.set()
-        release.wait(timeout=2)
-        return int(frame[0, 0, 0])
-
-    def submit(*args, **kwargs):
-        nonlocal outstanding, peak_outstanding
-        with lock:
-            outstanding += 1
-            peak_outstanding = max(peak_outstanding, outstanding)
-            if peak_outstanding >= 4:
-                four_submitted.set()
-        future = real_submit(*args, **kwargs)
-
-        def done(_future):
-            nonlocal outstanding
-            with lock:
-                outstanding -= 1
-
-        future.add_done_callback(done)
-        return future
-
-    monkeypatch.setattr(converter, "_build_frame", fake_build)
-    monkeypatch.setattr(converter._executor, "submit", submit)
-    frames = [np.full((1, 1, 3), value, dtype=np.uint8) for value in range(8)]
-    result: list[int] = []
-    iterator = converter.iter_frames(frames, np.dtype(np.uint8))
-    consumer = threading.Thread(target=lambda: result.extend(iterator))
-    consumer.start()
-    try:
-        assert started.wait(timeout=2)
-        assert four_submitted.wait(timeout=2)
-        release.set()
-        consumer.join(timeout=2)
-        assert not consumer.is_alive()
-    finally:
-        release.set()
-        iterator.close()
-        converter.shutdown()
-
-    assert result == list(range(8))
-    assert peak_outstanding <= 4
-
-
-def test_planar_converter_reuses_executor_and_shutdowns_after_cancel():
-    converter = video_api_utils._PlanarFrameConverter(max_workers=2)
-    executor = converter._executor
-    frames = [np.zeros((2, 2, 3), dtype=np.uint8) for _ in range(4)]
-    try:
-        list(converter.iter_frames(frames, np.dtype(np.uint8)))
-        iterator = converter.iter_frames(frames, np.dtype(np.uint8))
-        next(iterator)
-        iterator.close()
-        assert converter._executor is executor
-    finally:
-        converter.shutdown()
-
-    assert executor._shutdown
-
-
@pytest.mark.parametrize(
    ("video", "expected_path", "expected_reason", "expected_mux"),
    [
@@ -437,35 +347,6 @@ def test_video_only_does_not_log_default_audio_sample_rate(monkeypatch):
    assert log_messages == []


-def test_planar_mux_accepts_none_rate_and_defaults_audio_to_44100():
-    def make_frame():
-        return next(
-            video_api_utils._iter_planar_video_frames(
-                [np.zeros((2, 2, 3), dtype=np.uint8)],
-                np.dtype(np.uint8),
-            )
-        )
-
-    video_only = media_utils.mux_av_video_audio_bytes(
-        [make_frame()],
-        width=2,
-        height=2,
-        audio_sample_rate=None,
-    )
-    with av.open(BytesIO(video_only), mode="r", format="mp4") as container:
-        assert not container.streams.audio
-
-    with_audio = media_utils.mux_av_video_audio_bytes(
-        [make_frame()],
-        width=2,
-        height=2,
-        audio_waveform=np.zeros((1, 1024), dtype=np.float32),
-        audio_sample_rate=None,
-    )
-    with av.open(BytesIO(with_audio), mode="r", format="mp4") as container:
-        assert container.streams.audio[0].rate == 44100
-
-
 def test_interleaved_video_uses_legacy_fallback_automatically(monkeypatch):
    calls = []
    path_logs = []
@@ -502,7 +383,6 @@ def test_interleaved_video_uses_legacy_fallback_automatically(monkeypatch):
    assert "reason=non_contiguous_rgb_planes" in path_logs[0]
    assert "audio_present=False" in path_logs[0]
    assert "effective_audio_sample_rate=None" in path_logs[0]
-    assert "effective_frame_conversion_workers=0" in path_logs[0]
    assert len(coerce_calls) == 1


@@ -95,13 +95,12 @@ class FakeAsyncOmni:
        yield MockVideoResult(videos)


-def test_raw_and_base64_encoders_receive_persistent_converter(mocker: MockerFixture):
+def test_raw_and_base64_encoders_receive_no_policy_config(mocker: MockerFixture):
    engine = FakeAsyncOmni()
    handler = OmniOpenAIServingVideo.for_diffusion(
        engine,
        model_name="test-model",
    )
-    assert handler._video_frame_converter.max_workers == 8
    raw_encoder = mocker.patch(
        "vllm_omni.entrypoints.openai.serving_video._encode_video_bytes",
        return_value=b"encoded-video",
@@ -118,9 +117,8 @@ def test_raw_and_base64_encoders_receive_persistent_converter(mocker: MockerFixt

    asyncio.run(_generate_both_response_types())

-    assert raw_encoder.call_args.kwargs["frame_converter"] is handler._video_frame_converter
-    assert base64_encoder.call_args.kwargs["frame_converter"] is handler._video_frame_converter
-    handler.shutdown()
+    assert "encoding_config" not in raw_encoder.call_args.kwargs
+    assert "encoding_config" not in base64_encoder.call_args.kwargs


 def test_resolve_diffusion_od_config_falls_back_to_attribute():
@@ -1233,9 +1231,8 @@ def test_audio_sample_rate_comes_from_model_config(test_client, mocker: MockerFi
        audio=None,
        audio_sample_rate=None,
        video_codec_options=None,
-        frame_converter=None,
    ):
-        del video, fps, audio, video_codec_options, frame_converter
+        del video, fps, audio, video_codec_options
        audio_sample_rates.append(audio_sample_rate)
        return b"fake-video"

@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 import asyncio
 import re
 from types import SimpleNamespace
@@ -11,7 +8,6 @@ from vllm.sampling_params import RequestOutputKind, SamplingParams

 from tests.helpers.mark import hardware_test
 from tests.helpers.stage_config import get_deploy_config_path
-from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
 from vllm_omni.entrypoints.async_omni import AsyncOmni
 from vllm_omni.outputs import OmniRequestOutput

@@ -81,7 +77,7 @@ def test_generate_submits_randomized_id_to_engine():
    """Ensure the engine receives a UUID-suffixed ID, not the raw request ID"""

    async def run():
-        submitted_ids: list[str] = []
+        submitted_ids = []
        omni = get_async_omni_instance(fake_add_request=get_fake_add_request(submitted_ids))

        req_id = "my-req-1"
@@ -110,8 +106,8 @@ def test_generate_forwards_lora_request_to_engine():
    """

    async def run():
-        submitted_ids: list[str] = []
-        submitted_loras: list[LoRARequest | None] = []
+        submitted_ids = []
+        submitted_loras = []
        omni = get_async_omni_instance(fake_add_request=get_fake_add_request(submitted_ids, submitted_loras))

        lora = LoRARequest(lora_name="test", lora_int_id=1, lora_path="/tmp/fake")
@@ -147,7 +143,7 @@ def test_abort_handles_internal_request_mapping(req_ids: list[str], cancel_prefi
    user provided request ID will be aborted."""

    async def run():
-        aborted_batches: list[list[str]] = []
+        aborted_batches = []
        omni = get_async_omni_instance(
            fake_abort_request=get_fake_abort(aborted_batches),
        )
@@ -244,8 +240,8 @@ def test_abort_propagates_engine_errors_without_popping_state():
@pytest.mark.cpu
 def test_generate_accepts_request_after_repeated_cancellations():
    async def run_test():
-        submitted_request_ids: list[str] = []
-        aborted_request_batches: list[list[str]] = []
+        submitted_request_ids = []
+        aborted_request_batches = []

        async def collect_outputs(request_id):
            outputs = []
@@ -290,43 +286,6 @@ def test_generate_accepts_request_after_repeated_cancellations():
    asyncio.run(run_test())


-@pytest.mark.cpu
-def test_generate_cancellation_converges_after_engine_shutdown_starts(mocker):
-    async def run_test():
-        submitted_request_ids: list[str] = []
-        engine = object.__new__(AsyncOmniEngine)
-        engine._shutdown_called = True
-        engine.request_queue = mocker.Mock()
-        engine.num_stages = 1
-        engine.add_request_async = get_fake_add_request(submitted_request_ids)
-
-        omni = get_async_omni_instance()
-        omni.engine = engine
-
-        async def collect_outputs():
-            async for _ in omni.generate(
-                prompt={"prompt": "prompt"},
-                request_id="cancel-shutdown",
-                sampling_params_list=[SimpleNamespace()],
-                output_modalities=["image"],
-            ):
-                pass
-
-        task = asyncio.create_task(collect_outputs())
-        await asyncio.sleep(0)
-        task.cancel()
-
-        with pytest.raises(asyncio.CancelledError):
-            await task
-
-        assert len(submitted_request_ids) == 1
-        assert submitted_request_ids[0].startswith("cancel-shutdown-")
-        assert omni.request_states == {}
-        engine.request_queue.sync_q.put.assert_not_called()
-
-    asyncio.run(run_test())
-
-
@pytest.mark.cpu
 def test_generate_yields_streaming_diffusion_chunks_before_final():
    """AsyncOmni.generate yields every intermediate diffusion chunk before the final one."""
@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 from __future__ import annotations

@@ -65,38 +65,3 @@ def test_image_to_video_builds_canonical_prompt() -> None:
        "multi_modal_data": {"image": image},
        "negative_prompt": "flicker",
    }
-
-
-@pytest.mark.parametrize(
-    ("relative_path", "module_name"),
-    [
-        ("offline_inference/text_to_video/text_to_video.py", "text_to_video_output_example"),
-        ("offline_inference/image_to_video/image_to_video.py", "image_to_video_output_example"),
-    ],
-)
-@pytest.mark.parametrize("output_type", ["video", "image"])
-def test_video_examples_accept_semantic_and_legacy_output_types(
-    relative_path: str,
-    module_name: str,
-    output_type: str,
-) -> None:
-    mod = _load_example_module(relative_path, module_name)
-
-    mod._validate_video_output_type(output_type)
-
-
-@pytest.mark.parametrize(
-    ("relative_path", "module_name"),
-    [
-        ("offline_inference/text_to_video/text_to_video.py", "text_to_video_invalid_output_example"),
-        ("offline_inference/image_to_video/image_to_video.py", "image_to_video_invalid_output_example"),
-    ],
-)
-def test_video_examples_reject_unrelated_output_types(
-    relative_path: str,
-    module_name: str,
-) -> None:
-    mod = _load_example_module(relative_path, module_name)
-
-    with pytest.raises(ValueError, match="Unexpected output type 'audio'"):
-        mod._validate_video_output_type("audio")
@@ -198,32 +198,3 @@ class TestRequestLifecycleGauges:
        out = generate_latest(registry).decode()
        assert _sample_value(out, 'vllm_omni:num_requests_running{model_name="lifecycle-test-2"}') == 1.0
        assert _sample_value(out, 'vllm_omni:num_requests_waiting{model_name="lifecycle-test-2"}') == 0.0
-
-    def test_engine_queued_requests_count_as_waiting(self, registry: CollectorRegistry) -> None:
-        from types import SimpleNamespace
-
-        from vllm_omni.entrypoints.omni_base import OmniBase
-        from vllm_omni.metrics.prometheus import OmniPrometheusMetrics, OmniRequestCounter
-
-        obj = object.__new__(OmniBase)
-        obj.engine = SimpleNamespace(
-            _running_counter=OmniRequestCounter(),
-            _engines_waiting_counter=OmniRequestCounter(),
-        )
-        obj.prom_metrics = OmniPrometheusMetrics(model_name="lifecycle-test-3")
-        obj.request_states = {}
-        obj.log_stats = False
-
-        # Three requests dispatched to a stage engine whose scheduler admits
-        # one and queues two (e.g. a diffusion engine with
-        # max_num_running_reqs=1). The orchestrator reports the queued count
-        # via _engines_waiting_counter; they must show as waiting, not running.
-        for _ in range(3):
-            obj.engine._running_counter.increment()
-        obj.engine._engines_waiting_counter.value = 2
-
-        obj._publish_request_gauges(3)
-
-        out = generate_latest(registry).decode()
-        assert _sample_value(out, 'vllm_omni:num_requests_running{model_name="lifecycle-test-3"}') == 1.0
-        assert _sample_value(out, 'vllm_omni:num_requests_waiting{model_name="lifecycle-test-3"}') == 2.0
@@ -229,14 +229,6 @@ class TestDeployTopology:
            "vllm_omni.model_executor.stage_input_processors.minicpmo_4_5_omni.tts2code2wav_token_only"
        )

-    def test_no_async_chunk_selects_full_payload_processor(self) -> None:
-        deploy = load_deploy_config(_DEPLOY_DIR / "minicpmo_4_5.yaml")
-        deploy.async_chunk = False
-        stages = merge_pipeline_deploy(OMNI_PIPELINES[_PIPELINE_KEY], deploy)
-
-        assert stages[1].yaml_engine_args["async_chunk"] is False
-        assert stages[1].yaml_engine_args["custom_process_next_stage_input_func"].endswith(".tts2code2wav_full_payload")
-

 def test_code2wav_model_is_lazily_registered() -> None:
    assert _OMNI_MODELS["MiniCPMO45Code2Wav"] == (
@@ -2,7 +2,6 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 from types import SimpleNamespace
-from unittest.mock import MagicMock

 import pytest
 import torch
@@ -77,8 +76,8 @@ def test_step_audio2_token2wav_async_chunk_batch_guard():
        model._forward_async_chunk(
            input_ids=torch.tensor([1, 2, 3]),
            runtime_additional_information=[
-                {"meta": {"left_context_size": 0}},
-                {"meta": {"left_context_size": 1}},
+                {"left_context_size": 0},
+                {"left_context_size": 1},
            ],
        )

@@ -117,7 +116,7 @@ def test_step_audio2_token2wav_async_chunk_last_chunk_resets_state(monkeypatch):
    out = model.forward(
        input_ids=torch.tensor([10, 11, 12]),
        positions=torch.tensor([0, 1, 2]),
-        runtime_additional_information=[{"meta": {"left_context_size": 1}}],
+        runtime_additional_information=[{"left_context_size": 1}],
    )

    audio_list = out.multimodal_outputs["model_outputs"]
@@ -127,24 +126,12 @@ def test_step_audio2_token2wav_async_chunk_last_chunk_resets_state(monkeypatch):
    assert calls == {"setup": 1, "stream": 1, "reset": 1}


-def test_step_audio2_token2wav_routes_non_final_chunk_to_async_forward():
-    model = _build_model()
-    model._forward_async_chunk = MagicMock()
-
-    model.forward(
-        input_ids=torch.tensor([10, 11, 12]),
-        positions=torch.tensor([0, 1, 2]),
-        runtime_additional_information=[{"meta": {"left_context_size": 0}}],
-    )
-    model._forward_async_chunk.assert_called_once()
-
-
 def test_step_audio2_token2wav_async_chunk_empty_eof_returns_zero_chunk():
    model = _build_model()

    out = model._forward_async_chunk(
        input_ids=torch.tensor([], dtype=torch.int64),
-        runtime_additional_information=[{"meta": {"left_context_size": 1}}],
+        runtime_additional_information=[{"left_context_size": 1}],
    )

    audio_list = out.multimodal_outputs["model_outputs"]
@@ -72,16 +72,6 @@ def _codes(payload) -> list[int]:
    return payload.codes.audio.tolist()


-def test_empty_full_payload_releases_consumer_wait_gate() -> None:
-    payload = tts2code2wav_full_payload(_manager(), None, _request("req"))
-
-    assert _codes(payload) == []
-    assert payload.meta.code_flat_numel == 0
-    assert payload.meta.left_context_size == 0
-    assert payload.meta.last_chunk is True
-    assert payload.meta.finished.item() is True
-
-
@pytest.mark.parametrize(("count", "emitted"), [(24, False), (25, True), (26, True)])
 def test_first_chunk_threshold_is_25_generated_codes(count: int, emitted: bool) -> None:
    manager = _manager()
@@ -242,18 +242,19 @@ def tiny_flux_kontext_builder() -> str:

 def tiny_flux2_builder() -> str:
    def shrink_text_encoder(config: dict) -> dict:
+        # The real checkpoint ships language_model.lm_head.weight as its own
+        # tensor even though tie_word_embeddings defaults to True; keep it
+        # untied here too so transformers doesn't drop it on save.
        config["tie_word_embeddings"] = False
-        config["text_config"]["num_hidden_layers"] = 3
-        config["text_config"]["hidden_size"] = 32
+        config["text_config"]["num_hidden_layers"] = 31
        config["text_config"]["intermediate_size"] = 64
-        config["text_config"]["num_attention_heads"] = 2
-        config["text_config"]["head_dim"] = 16
+        config["text_config"]["num_attention_heads"] = 4
+        config["text_config"]["head_dim"] = 32
        config["text_config"]["num_key_value_heads"] = 2
-        config["text_config"]["text_encoder_out_layers"] = [0, 1, 2]
        config["vision_config"]["num_hidden_layers"] = 2
        config["vision_config"]["intermediate_size"] = 64
-        config["vision_config"]["num_attention_heads"] = 2
-        config["vision_config"]["head_dim"] = 16
+        config["vision_config"]["num_attention_heads"] = 4
+        config["vision_config"]["head_dim"] = 32
        return config

    return build_tiny_from_configs(
@@ -261,6 +262,6 @@ def tiny_flux2_builder() -> str:
        "black-forest-labs/FLUX.2-dev",
        transform={
            "text_encoder": shrink_text_encoder,
-            "transformer": partial(_shrink_dit_rope_config, num_single_layers=2, joint_attention_dim=96),
+            "transformer": partial(_shrink_dit_rope_config, num_single_layers=2),
        },
    )
@@ -3,7 +3,7 @@

 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Any, Generic, Literal, TypeVar
+from typing import Any, Generic, TypeVar

 import torch

@@ -125,22 +125,9 @@ class QueryRange:
    global_start: int


-@dataclass(frozen=True, slots=True)
-class VideoTokenSpan:
-    """One physical video-grid slice in packed document 0."""
-
-    start: int
-    latent_grid: tuple[int, int, int]
-    role: Literal["reference", "target"]
-
-    @property
-    def length(self) -> int:
-        return self.latent_grid[0] * self.latent_grid[1] * self.latent_grid[2]
-
-
@dataclass(frozen=True, slots=True)
 class VideoTokenLayout:
-    """Video-grid slices in a packed multimodal sequence.
+    """Where the video segment sits inside a packed multimodal sequence.

    A model that packs its sequence as ``[prefix | t*h*w video rows | padding]``
    publishes this so backends can recover spatiotemporal locality; the prefix
@@ -148,17 +135,11 @@ class VideoTokenLayout:
    Publishing it also asserts that any ``attn_mask`` masks only the trailing
    padding, so ``prefix_len + t*h*w`` is the used length of the sequence.

-    ``prefix_len``/``latent_grid`` retain the original one-tail contract. A
-    Ref2VA layout instead publishes ``used_len`` and every physical video
-    span, so audio and image rows between videos remain dense.
-
    Plain ints, so reading it never forces a device-to-host sync.
    """

-    prefix_len: int | None = None
-    latent_grid: tuple[int, int, int] | None = None
-    used_len: int | None = None
-    video_spans: tuple[VideoTokenSpan, ...] = ()
+    prefix_len: int
+    latent_grid: tuple[int, int, int]


@dataclass(frozen=True, slots=True)
@@ -3,7 +3,6 @@

 from __future__ import annotations

-import inspect
 import math
 from dataclasses import dataclass
 from typing import Any
@@ -51,11 +50,6 @@ _MISSING_MINDIESD = (
    "Otherwise, use FlashAttention by setting DIFFUSION_ATTENTION_BACKEND=FLASH_ATTN"
 )

-_INCOMPATIBLE_MINDIESD = (
-    "RAINFUSION_ATTN requires a MindIE-SD build whose sparse_attention supports the video_spans "
-    "argument. Please upgrade MindIE-SD or select FLASH_ATTN."
-)
-

 def _try_extract_layer_index(prefix: str) -> int | None:
    if not prefix:
@@ -66,13 +60,6 @@ def _try_extract_layer_index(prefix: str) -> int | None:
        return None


-def _supports_video_spans(sparse_attention: Any) -> bool:
-    try:
-        return "video_spans" in inspect.signature(sparse_attention).parameters
-    except (TypeError, ValueError):
-        return False
-
-
@dataclass(frozen=True)
 class RainFusionConfig:
    """Resolved RainFusion controls for one attention layer.
@@ -105,10 +92,9 @@ class RainFusionConfig:
 class RainFusionPlan:
    """Per-forward geometry handed to the rf_v2 kernel."""

+    prefix_len: int
    used_len: int
-    prefix_len: int | None = None
-    latent_shape: list[int] | None = None
-    video_spans: list[dict[str, object]] | None = None
+    latent_shape: list[int]


 class RainFusionAttentionBackend(AttentionBackend):
@@ -121,12 +107,6 @@ class RainFusionAttentionBackend(AttentionBackend):

        if find_spec("mindiesd") is None:
            raise ValueError(_MISSING_MINDIESD)
-        try:
-            from mindiesd import sparse_attention
-        except ImportError as exc:
-            raise ValueError(_MISSING_MINDIESD) from exc
-        if not _supports_video_spans(sparse_attention):
-            raise ValueError(_INCOMPATIBLE_MINDIESD)

    @staticmethod
    def get_supported_head_sizes() -> list[int]:
@@ -286,14 +266,6 @@ class RainFusionAttentionImpl(AttentionImpl):
            )
            return None

-        if layout.video_spans:
-            return self._resolve_multi_span_plan(layout, max_seqlen_q)
-
-        if layout.prefix_len is None or layout.latent_grid is None:
-            logger.warning_once(
-                "RAINFUSION_ATTN staying dense: video layout has neither a legacy video tail nor multi-video spans."
-            )
-            return None
        prefix_len = int(layout.prefix_len)
        latent_shape = [int(dim) for dim in layout.latent_grid]
        # rf_v2 splits the sequence as [prefix | t*h*w video rows]. Document 0 of
@@ -332,97 +304,11 @@ class RainFusionAttentionImpl(AttentionImpl):
            video_len,
        )
        return RainFusionPlan(
-            used_len=used_len,
            prefix_len=prefix_len,
+            used_len=used_len,
            latent_shape=latent_shape,
        )

-    def _resolve_multi_span_plan(self, layout, max_seqlen_q: int) -> RainFusionPlan | None:
-        """Validate Ref2VA's non-contiguous video grids before sparse dispatch."""
-        used_len = layout.used_len
-        if used_len is None or int(used_len) != int(max_seqlen_q):
-            logger.warning_once(
-                "RAINFUSION_ATTN staying dense: multi-video layout used_len=%r does not match packed document 0 (%d).",
-                used_len,
-                int(max_seqlen_q),
-            )
-            return None
-
-        spans: list[dict[str, object]] = []
-        span_summaries: list[str] = []
-        previous_end = 0
-        target_count = 0
-        video_seqlen = 0
-        boundary_dense_seqlen = 0
-        previous_video_length: int | None = None
-        for span in sorted(layout.video_spans, key=lambda item: item.start):
-            grid = tuple(int(dim) for dim in span.latent_grid)
-            length = math.prod(grid)
-            start = int(span.start)
-            if (
-                len(grid) != 3
-                or any(dim <= 0 for dim in grid)
-                or start < previous_end
-                or start + length > int(used_len)
-            ):
-                logger.warning_once(
-                    "RAINFUSION_ATTN staying dense: invalid multi-video span start=%d grid=%s used_len=%d.",
-                    start,
-                    grid,
-                    int(used_len),
-                )
-                return None
-            role = span.role
-            if role not in ("reference", "target"):
-                logger.warning_once("RAINFUSION_ATTN staying dense: unsupported multi-video span role %r.", role)
-                return None
-            if role == "target":
-                target_count += 1
-            if previous_video_length is not None:
-                # rf_v2 works on fixed 128-token blocks. The preceding clip
-                # needs these dense rows to complete its tail block before
-                # this clip begins, otherwise one sparse block would cross a
-                # clip boundary.
-                boundary_dense_seqlen += (-previous_video_length) % _BLOCK_SIZE
-            spans.append({"start": start, "latent_shape": list(grid)})
-            span_summaries.append(f"role={role}, start={start}, seqlen={length}, latent_shape={grid}")
-            previous_end = start + length
-            video_seqlen += length
-            previous_video_length = length
-
-        if target_count != 1:
-            logger.warning_once(
-                "RAINFUSION_ATTN staying dense: Ref2VA layout must contain exactly one target video span, got %d.",
-                target_count,
-            )
-            return None
-        if video_seqlen < _MIN_VIDEO_BLOCKS * _BLOCK_SIZE:
-            logger.warning_once(
-                "RAINFUSION_ATTN staying dense: multi-video seqlen=%d is under the sparse threshold seqlen=%d.",
-                video_seqlen,
-                _MIN_VIDEO_BLOCKS * _BLOCK_SIZE,
-            )
-            return None
-        dense_context_seqlen = int(used_len) - video_seqlen
-        if boundary_dense_seqlen > dense_context_seqlen:
-            logger.warning_once(
-                "RAINFUSION_ATTN staying dense: multi-video spans need dense_context_seqlen=%d "
-                "to isolate clip block boundaries, but this layout has only %d.",
-                boundary_dense_seqlen,
-                dense_context_seqlen,
-            )
-            return None
-        logger.info_once(
-            "RAINFUSION_ATTN multi-video active: sparsity=%.2f, spans=[%s], "
-            "video_seqlen=%d, dense_context_seqlen=%d, valid_packed_seqlen=%d.",
-            self.rainfusion.sparsity,
-            "; ".join(span_summaries),
-            video_seqlen,
-            dense_context_seqlen,
-            int(used_len),
-        )
-        return RainFusionPlan(used_len=int(used_len), video_spans=spans)
-
    def _forward_sparse_npu(
        self,
        query: torch.Tensor,
@@ -439,32 +325,21 @@ class RainFusionAttentionImpl(AttentionImpl):
        q, k, v = (tensor[:, :used] for tensor in (query, key, value))
        # Ulysses has already gathered the full sequence onto this rank and split
        # the heads, so read the head count off the tensor rather than num_heads.
-        common_kwargs: dict[str, object] = {
-            "scale": self.softmax_scale,
-            "head_num": query.shape[-2],
-            "input_layout": _INPUT_LAYOUT,
-            "inner_precise": _INNER_PRECISE,
-            "block_size": _BLOCK_SIZE,
-            "sparsity": self.rainfusion.sparsity,
-        }
-        if plan.video_spans is not None:
-            out = sparse_attention(
-                q,
-                k,
-                v,
-                sparse_type="rf_v2",
-                video_spans=plan.video_spans,
-                **common_kwargs,
-            )
-        else:
-            assert plan.prefix_len is not None and plan.latent_shape is not None
-            common_kwargs.update(
-                sparse_type="rf_v2",
-                txt_len=plan.prefix_len,
-                latent_shape_q=plan.latent_shape,
-                latent_shape_k=plan.latent_shape,
-            )
-            out = sparse_attention(q, k, v, **common_kwargs)
+        out = sparse_attention(
+            q,
+            k,
+            v,
+            scale=self.softmax_scale,
+            head_num=query.shape[-2],
+            input_layout=_INPUT_LAYOUT,
+            inner_precise=_INNER_PRECISE,
+            sparse_type="rf_v2",
+            txt_len=plan.prefix_len,
+            block_size=_BLOCK_SIZE,
+            latent_shape_q=plan.latent_shape,
+            latent_shape_k=plan.latent_shape,
+            sparsity=self.rainfusion.sparsity,
+        )
        if used == query.shape[1]:
            return out
        padded = torch.zeros_like(query)
@@ -438,12 +438,6 @@ class Flux2Pipeline(
        self.setup_diffusion_pipeline_profiler(
            enable_diffusion_pipeline_profiler=self.od_config.enable_diffusion_pipeline_profiler
        )
-
-        if hasattr(self.text_encoder.config, "text_encoder_out_layers"):
-            self.text_encoder_out_layers = tuple(self.text_encoder.config.text_encoder_out_layers)
-        else:
-            self.text_encoder_out_layers = (10, 20, 30)
-
        self._current_timestep = None
        self._interrupt = False

@@ -683,6 +677,7 @@ class Flux2Pipeline(
        num_images_per_prompt: int = 1,
        prompt_embeds: torch.Tensor | None = None,
        max_sequence_length: int = 512,
+        text_encoder_out_layers: tuple[int, ...] = (10, 20, 30),
    ):
        device = device or self._execution_device

@@ -699,7 +694,7 @@ class Flux2Pipeline(
                device=device,
                max_sequence_length=max_sequence_length,
                system_message=self.system_message,
-                hidden_states_layers=self.text_encoder_out_layers,
+                hidden_states_layers=text_encoder_out_layers,
            )

        batch_size, seq_len, _ = prompt_embeds.shape
@@ -891,6 +886,7 @@ class Flux2Pipeline(
        callback_on_step_end: Callable[[int, int, dict], None] | None = None,
        callback_on_step_end_tensor_inputs: list[str] = ["latents"],
        max_sequence_length: int = 512,
+        text_encoder_out_layers: tuple[int, ...] = (10, 20, 30),
        caption_upsample_temperature: float = None,
    ) -> DiffusionOutput:
        if len(req.prompts) > 1:
@@ -926,6 +922,7 @@ class Flux2Pipeline(
            else num_images_per_prompt
        )
        max_sequence_length = req.sampling_params.max_sequence_length or max_sequence_length
+        text_encoder_out_layers = req.sampling_params.extra_args.get("text_encoder_out_layers", text_encoder_out_layers)
        caption_upsample_temperature = req.sampling_params.extra_args.get(
            "caption_upsample_temperature", caption_upsample_temperature
        )
@@ -980,6 +977,7 @@ class Flux2Pipeline(
            device=device,
            num_images_per_prompt=num_images_per_prompt,
            max_sequence_length=max_sequence_length,
+            text_encoder_out_layers=text_encoder_out_layers,
        )

        has_neg_prompt = negative_prompt_embeds is not None or any(req_negative_prompt)
@@ -995,6 +993,7 @@ class Flux2Pipeline(
                device=device,
                num_images_per_prompt=num_images_per_prompt,
                max_sequence_length=max_sequence_length,
+                text_encoder_out_layers=text_encoder_out_layers,
            )

        # 4. process images
@@ -14,7 +14,7 @@ from typing import Any

 import torch

-from vllm_omni.diffusion.attention.backends.abstract import VideoTokenLayout, VideoTokenSpan
+from vllm_omni.diffusion.attention.backends.abstract import VideoTokenLayout
 from vllm_omni.diffusion.forward_context import (
    set_forward_context_denoise_step_idx,
    set_forward_context_denoise_timestep,
@@ -119,23 +119,11 @@ class MiniMaxH3DenoiseBranch:
        }
        # Where the video segment sits in the packed sequence. Resolved to plain
        # ints here so the attention layers never sync on it per step.
-        raw_spans = packed.get("video_spans")
-        if raw_spans:
-            spans = tuple(
-                VideoTokenSpan(
-                    start=int(span["start"]),
-                    latent_grid=tuple(int(dim) for dim in span["latent_grid"]),
-                    role=str(span["role"]),
-                )
-                for span in raw_spans
-            )
-            self.static_kwargs["video_token_layout"] = VideoTokenLayout(used_len=int(cu[1]), video_spans=spans)
-        else:
-            grid = packed["latent_grid"].tolist()
-            self.static_kwargs["video_token_layout"] = VideoTokenLayout(
-                prefix_len=int(packed["video_row_start"]),
-                latent_grid=(int(grid[0]), int(grid[1]), int(grid[2])),
-            )
+        grid = packed["latent_grid"].tolist()
+        self.static_kwargs["video_token_layout"] = VideoTokenLayout(
+            prefix_len=int(packed["video_row_start"]),
+            latent_grid=(int(grid[0]), int(grid[1]), int(grid[2])),
+        )

    def forward_kwargs(
        self,
@@ -1464,19 +1464,15 @@ class MiniMaxH3Qwen3VLEncoder(nn.Module):
        if next(self.parameters()).device.type != self.device_target.type:
            raise RuntimeError("call load_to_device() before encode_ids()")

-        cudnn_sdp_enabled = torch.backends.cuda.cudnn_sdp_enabled()
        torch.backends.cuda.enable_cudnn_sdp(True)
-        try:
-            hidden = self._encode(
-                input_ids,
-                pixel_values=pixel_values,
-                image_grid_thw=image_grid_thw,
-                pixel_values_videos=pixel_values_videos,
-                video_grid_thw=video_grid_thw,
-            )
-            return hidden.cpu()
-        finally:
-            torch.backends.cuda.enable_cudnn_sdp(cudnn_sdp_enabled)
+        hidden = self._encode(
+            input_ids,
+            pixel_values=pixel_values,
+            image_grid_thw=image_grid_thw,
+            pixel_values_videos=pixel_values_videos,
+            video_grid_thw=video_grid_thw,
+        )
+        return hidden.cpu()

    def forward(
        self,
@@ -124,7 +124,7 @@ def minimax_h3_packed_sequence(
    include_keyframe_cond: bool,
    keyframe_frame_indices: list[int] | tuple[int, ...] | None = None,
    frame_count: int | None = None,
-) -> dict[str, object]:
+) -> dict[str, torch.Tensor]:
    """Build the packed-sequence structural fields for one CFG branch.

    The used length is padded up to a multiple of 64.
@@ -239,14 +239,6 @@ def minimax_h3_packed_sequence(
        # Extent of the video segment within the packed sequence.
        "latent_grid": torch.tensor([latent_t, ph, pw], dtype=torch.int64),
        "video_row_start": torch.tensor(video_sl.start, dtype=torch.int64),
-        # The single target video uses the same span contract as Ref2VA.
-        "video_spans": (
-            {
-                "start": video_sl.start,
-                "latent_grid": (latent_t, ph, pw),
-                "role": "target",
-            },
-        ),
    }


@@ -308,7 +300,7 @@ def minimax_h3_packed_sequence_ref2va_blocks(
    ref_blocks: Sequence[Mapping[str, object]],
    audio_channel: int = 2,
    seq_len: int | None = None,
-) -> dict[str, object]:
+) -> dict[str, torch.Tensor]:
    """General ref2va-family packed layout.

    ``ref_blocks`` are consumed in request/plan order:
@@ -408,33 +400,6 @@ def minimax_h3_packed_sequence_ref2va_blocks(
    video_sl = slice(audio_sl.stop, audio_sl.stop + video_rows)
    pad_sl = slice(video_sl.stop, seq_len)

-    # This is deliberately tensor-free metadata. It is constructed once by
-    # the denoise branch and consumed by every attention layer without a
-    # device-to-host synchronization.
-    video_spans: list[dict[str, object]] = []
-    for item in block_slices:
-        if item["kind"] in ("video", "video_audio"):
-            visual_sl = item["visual_sl"]
-            assert isinstance(visual_sl, slice)
-            video_spans.append(
-                {
-                    "start": visual_sl.start,
-                    "latent_grid": (
-                        int(item["latent_t"]),
-                        int(item["latent_h"]) // _PATCH_H,
-                        int(item["latent_w"]) // _PATCH_W,
-                    ),
-                    "role": "reference",
-                }
-            )
-    video_spans.append(
-        {
-            "start": video_sl.start,
-            "latent_grid": (latent_t, ph, pw),
-            "role": "target",
-        }
-    )
-
    input_ids = torch.full((seq_len,), MINIMAX_H3_PAD_ID, dtype=torch.int64)
    input_ids[text_sl] = MINIMAX_H3_TEXT_ID
    image_mask = torch.zeros(seq_len, dtype=torch.bool)
@@ -595,7 +560,6 @@ def minimax_h3_packed_sequence_ref2va_blocks(
        # Extent of the video segment within the packed sequence.
        "latent_grid": torch.tensor([latent_t, ph, pw], dtype=torch.int64),
        "video_row_start": torch.tensor(video_sl.start, dtype=torch.int64),
-        "video_spans": tuple(video_spans),
    }


@@ -210,7 +210,7 @@ def mux_av_video_audio_bytes(
    audio_waveform: np.ndarray | None = None,
    *,
    fps: float = 25.0,
-    audio_sample_rate: int | None = None,
+    audio_sample_rate: int = 44100,
    video_codec: str = "h264",
    audio_codec: str = "aac",
    crf: str = "18",
@@ -236,7 +236,6 @@ def mux_av_video_audio_bytes(
        samples: np.ndarray | None = None
        layout: str | None = None
        if audio_waveform is not None:
-            effective_audio_sample_rate = 44100 if audio_sample_rate is None else audio_sample_rate
            samples = audio_waveform.astype(np.float32)
            if samples.ndim == 1:
                samples = samples.reshape(1, -1)
@@ -244,7 +243,7 @@ def mux_av_video_audio_bytes(
                samples = np.ascontiguousarray(samples.T)
            num_channels = samples.shape[0]
            layout = "stereo" if num_channels >= 2 else "mono"
-            a_stream = cast(av.AudioStream, container.add_stream(audio_codec, rate=effective_audio_sample_rate))
+            a_stream = cast(av.AudioStream, container.add_stream(audio_codec, rate=audio_sample_rate))
            a_stream.layout = layout

        for frame in video_frames:
@@ -257,12 +256,12 @@ def mux_av_video_audio_bytes(
            if samples is None or layout is None:
                raise ValueError("Audio samples were not prepared for muxing.")
            audio_frame = av.AudioFrame.from_ndarray(samples, format="fltp", layout=layout)
-            audio_frame.sample_rate = effective_audio_sample_rate
+            audio_frame.sample_rate = audio_sample_rate
            # AAC has a one-frame encoder delay. Mark the input waveform as
            # starting at t=0 so the MP4 muxer writes the corresponding negative
            # priming timestamp instead of exposing the delay as leading silence.
            audio_frame.pts = 0
-            audio_frame.time_base = Fraction(1, effective_audio_sample_rate)
+            audio_frame.time_base = Fraction(1, audio_sample_rate)
            for packet in a_stream.encode(audio_frame):
                container.mux(packet)
            for packet in a_stream.encode():
@@ -24,24 +24,6 @@ logger = init_logger(__name__)
 SHUTDOWN_ENQUEUE_TIMEOUT_S = 1.0
 SHUTDOWN_JOIN_TIMEOUT_S = 30.0
 _WEAK_SHUTDOWN_JOIN_TIMEOUT_S = 1.0
-_JANUS_SYNC_QUEUE_SHUTDOWN = getattr(janus, "SyncQueueShutDown", None)
-_LEGACY_JANUS_QUEUE_CLOSED_MESSAGE = "Operation on the closed queue is forbidden"
-_RPC_RESULT_ROUTER_CLOSED_MESSAGES = {
-    "RPC result router closed",
-    "RPC result router is closed",
-}
-
-
-def is_janus_sync_queue_shutdown(exc: Exception) -> bool:
-    if _JANUS_SYNC_QUEUE_SHUTDOWN is not None:
-        return isinstance(exc, _JANUS_SYNC_QUEUE_SHUTDOWN)
-    return isinstance(exc, RuntimeError) and str(exc) == _LEGACY_JANUS_QUEUE_CLOSED_MESSAGE
-
-
-def is_abort_transport_shutdown(exc: Exception) -> bool:
-    return is_janus_sync_queue_shutdown(exc) or (
-        isinstance(exc, RuntimeError) and str(exc) in _RPC_RESULT_ROUTER_CLOSED_MESSAGES
-    )


 def inject_global_id(target: Any, request_id: str) -> None:
@@ -52,8 +52,6 @@ from vllm_omni.engine.async_engine_utils import (
    apply_omni_final_stage_metadata,
    enqueue_orchestrator_shutdown,
    inject_global_id,
-    is_abort_transport_shutdown,
-    is_janus_sync_queue_shutdown,
    shutdown_runtime_after_orchestrator,
    upgrade_to_omni_request,
    weak_shutdown_async_omni_engine,
@@ -253,7 +251,6 @@ class AsyncOmniEngine:
        self._correlated_rpc_client: CorrelatedRpcClient | None = None
        self._duplex_control_client: DuplexControlClient | None = None
        self._running_counter = OmniRequestCounter()
-        self._engines_waiting_counter = OmniRequestCounter()

        logger.info(f"[AsyncOmniEngine] Launching Orchestrator thread with {self.num_stages} stages")

@@ -418,7 +415,6 @@ class AsyncOmniEngine:
                pd_config=pd_config,
                membership_controller=membership_controller,
                running_counter=self._running_counter,
-                engines_waiting_counter=self._engines_waiting_counter,
                transfer_emitter=self._transfer_emitter,
                prom_metrics=self._prom_metrics,
                log_stats=self._log_stats,
@@ -1905,16 +1901,9 @@ class AsyncOmniEngine:
        Prefer :meth:`abort_async` when the caller needs acknowledgment that
        stage aborts, binding release, and orchestrator request cleanup finished.
        """
-        if not request_ids or getattr(self, "_shutdown_called", False):
-            return
        if self.request_queue is None:
            raise RuntimeError("request_queue is not initialized")
-        try:
-            self.request_queue.sync_q.put(AbortRequestMessage(request_ids=request_ids))
-        except Exception as exc:
-            if getattr(self, "_shutdown_called", False) and is_janus_sync_queue_shutdown(exc):
-                return
-            raise
+        self.request_queue.sync_q.put(AbortRequestMessage(request_ids=request_ids))

    async def abort_async(
        self,
@@ -1927,8 +1916,6 @@ class AsyncOmniEngine:
        :class:`AbortResultMessage` via :class:`CorrelatedRpcClient`, and
        raises if the orchestrator reports failure or times out.
        """
-        if not request_ids or getattr(self, "_shutdown_called", False):
-            return
        if self.request_queue is None:
            raise RuntimeError("request_queue is not initialized")
        transport = self._correlated_rpc_client
@@ -1951,12 +1938,7 @@ class AsyncOmniEngine:
            return result_msg

        loop = asyncio.get_running_loop()
-        try:
-            result_msg = await loop.run_in_executor(None, _wait)
-        except Exception as exc:
-            if getattr(self, "_shutdown_called", False) and is_abort_transport_shutdown(exc):
-                return
-            raise
+        result_msg = await loop.run_in_executor(None, _wait)
        if not result_msg.success:
            raise RuntimeError(result_msg.error or "abort failed")

@@ -387,7 +387,6 @@ class Orchestrator:
    # Class-level defaults so tests that bypass __init__ via object.__new__
    # don't AttributeError when transfer / counter emit paths access them.
    _running_counter: OmniRequestCounter | None = None
-    _engines_waiting_counter: OmniRequestCounter | None = None
    _transfer_emitter: Any = None
    _prom_metrics: Any = None
    _stat_logger: OmniPrometheusStatLogger | None = None
@@ -404,7 +403,6 @@ class Orchestrator:
        pd_config: dict[str, Any] | None = None,
        membership_controller: MembershipController | None = None,
        running_counter: OmniRequestCounter | None = None,
-        engines_waiting_counter: OmniRequestCounter | None = None,
        transfer_emitter: Any = None,
        prom_metrics: Any = None,
        log_stats: bool = False,
@@ -441,13 +439,7 @@ class Orchestrator:
            self._pd_bootstrap_addr = pd_config.get("bootstrap_addr")
            self._pd_prefill_engine_id = pd_config.get("prefill_engine_id")
        self.request_states: dict[str, OrchestratorRequestState] = {}
-        self._init_metrics_state(
-            stage_pools,
-            running_counter,
-            transfer_emitter,
-            engines_waiting_counter=engines_waiting_counter,
-            log_stats=log_stats,
-        )
+        self._init_metrics_state(stage_pools, running_counter, transfer_emitter, log_stats=log_stats)

        self._cfg_tracker = CfgCompanionTracker()
        self._stage_input_processors: dict[int, Any] = {}
@@ -493,7 +485,6 @@ class Orchestrator:
        stage_pools: list[StagePool],
        running_counter: OmniRequestCounter | None,
        transfer_emitter: Any,
-        engines_waiting_counter: OmniRequestCounter | None = None,
        log_stats: bool = False,
    ) -> None:
        """Wire up all metric-related orchestrator state.
@@ -519,7 +510,6 @@ class Orchestrator:
        from iteration_stats independently of scheduler gauges.
        """
        self._running_counter = running_counter
-        self._engines_waiting_counter = engines_waiting_counter
        self._transfer_emitter = transfer_emitter

        # Flat engine_idx ↔ (stage, replica) maps. The reverse map is
@@ -1468,20 +1458,11 @@ class Orchestrator:
        """Update one replica snapshot and expose the stage-wide total."""
        self._stage_replica_waiting[(stage_id, replica_id)] = max(int(n_waiting), 0)
        self._set_stage_waiting_total(stage_id)
-        self._sync_engines_waiting_counter()

    def _remove_stage_replica_waiting(self, stage_id: int, replica_id: int) -> None:
        """Drop a dead replica's snapshot and refresh the stage total."""
        self._stage_replica_waiting.pop((stage_id, replica_id), None)
        self._set_stage_waiting_total(stage_id)
-        self._sync_engines_waiting_counter()
-
-    def _sync_engines_waiting_counter(self) -> None:
-        """Aggregate per-replica waiting into the counter shared with the
-        frontend so engine-queued requests show as waiting, not running."""
-        counter = self._engines_waiting_counter
-        if counter is not None:
-            counter.value = sum(self._stage_replica_waiting.values())

    def _set_stage_waiting_total(self, stage_id: int) -> None:
        if self._prom_metrics is None:
@@ -633,15 +633,6 @@ class StagePool:
            if callable(pop_native_text_metrics):
                native_text_metrics = pop_native_text_metrics(request_id)
        native_generation_tokens = native_text_metrics.get("num_generation_tokens")
-        finish_reason = next(
-            (
-                str(reason)
-                for request_output in reversed(request_outputs)
-                for completion in reversed(getattr(request_output, "outputs", []) or [])
-                if (reason := getattr(completion, "finish_reason", None)) is not None
-            ),
-            None,
-        )
        num_tokens_out = (
            max(int(native_generation_tokens), 0)
            if isinstance(native_generation_tokens, int) and not isinstance(native_generation_tokens, bool)
@@ -710,7 +701,6 @@ class StagePool:
            # happens inside the model runner and is not observable here.
            batch_size=1,
            replica_id=replica_id,
-            finish_reason=finish_reason,
            rx_decode_time_ms=0.0,
            rx_transfer_bytes=0,
            rx_in_flight_time_ms=0.0,
@@ -622,8 +622,6 @@ class AsyncOmni(EngineClient, OmniBase):
            submit_ts = time.time()
            req_state.metrics.stage_first_ts[0] = submit_ts
            req_start_ts[request_id] = submit_ts
-            # Refresh gauges on arrival.
-            self._publish_request_gauges(len(self.request_states))

            # Process results based on mode
            # Both sequential and async_chunk modes read the same message stream
@@ -406,26 +406,6 @@ class OmniBase(PDDisaggregationMixin):
        prom.request_failed()
        prom.inc_requests_failed(normalize_failure_reason(reason))

-    def _publish_request_gauges(self, total: int) -> None:
-        """Publish num_requests_running / num_requests_waiting for ``total`` in-flight requests.
-
-        ``_running_counter`` counts requests from the moment the orchestrator
-        dispatches them to a stage engine, which includes requests still queued
-        inside a stage scheduler. The orchestrator reports those via
-        ``_engines_waiting_counter``.
-        """
-        prom = getattr(self, "prom_metrics", None)
-        if prom is None:
-            return
-        engine = getattr(self, "engine", None)
-        counter = getattr(engine, "_running_counter", None)
-        waiting_counter = getattr(engine, "_engines_waiting_counter", None)
-        dispatched = counter.value if counter is not None else total
-        in_engine_waiting = waiting_counter.value if waiting_counter is not None else 0
-        in_engine_waiting = min(max(0, in_engine_waiting), dispatched)
-        prom.set_running(max(0, dispatched - in_engine_waiting))
-        prom.set_waiting(max(0, total - dispatched) + in_engine_waiting)
-
    def _log_summary_and_cleanup(self, request_id: str, reason: str = "stage_error") -> None:
        req_state = self.request_states.get(request_id)
        try:
@@ -451,7 +431,13 @@ class OmniBase(PDDisaggregationMixin):
            # Republish gauges so any stale value left by the per-stage
            # publish in _process_single_result (which runs while the request
            # is still in self.request_states) is corrected after the pop.
-            self._publish_request_gauges(len(self.request_states))
+            prom = getattr(self, "prom_metrics", None)
+            counter = getattr(getattr(self, "engine", None), "_running_counter", None)
+            if prom is not None:
+                total = len(self.request_states)
+                running = counter.value if counter is not None else total
+                prom.set_running(running)
+                prom.set_waiting(max(0, total - running))

    def _compute_final_stage_id(self, output_modalities: list[str] | None) -> int:
        return get_final_stage_id_for_e2e(
@@ -722,8 +708,12 @@ class OmniBase(PDDisaggregationMixin):
        # hasn't popped self.request_states yet — exclude the finalizing
        # request from `total` so waiting doesn't read 1 and stay stuck
        # there until the next request arrives.
+        counter = getattr(self.engine, "_running_counter", None)
        is_finalizing = finished and stage_id == final_stage_id_for_e2e
-        self._publish_request_gauges(max(0, len(self.request_states) - (1 if is_finalizing else 0)))
+        total = max(0, len(self.request_states) - (1 if is_finalizing else 0))
+        running = counter.value if counter is not None else total
+        self.prom_metrics.set_running(running)
+        self.prom_metrics.set_waiting(max(0, total - running))

        response_metrics: dict[str, Any] = {}
        stage_metrics: dict[str, dict[str, Any]] = {}
@@ -615,9 +615,6 @@ async def omni_run_server_worker(listen_address, sock, args, client_config=None,
            await shutdown_task
        finally:
            state = getattr(app, "state", None)
-            serving_video = getattr(state, "openai_serving_video", None) if state is not None else None
-            if serving_video is not None:
-                serving_video.shutdown()
            serving_speech = getattr(state, "openai_serving_speech", None) if state is not None else None
            if serving_speech is not None:
                serving_speech.shutdown()
@@ -828,8 +825,6 @@ async def omni_init_app_state(
            diffusion_engine=engine_client,
            model_name=model_name,
            stage_configs=diffusion_stage_configs,
-            allowed_local_media_path=getattr(args, "allowed_local_media_path", ""),
-            allowed_media_domains=getattr(args, "allowed_media_domains", None),
        )
        state.openai_serving_duplex = None
        state.openai_streaming_speech = None
@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 import asyncio
 import base64
 import hashlib
@@ -56,7 +53,6 @@ from vllm_omni.entrypoints.openai.speech_usage import (
 )
 from vllm_omni.entrypoints.openai.tts_adapters import (
    SpeechServingContext,
-    TTSGenerationError,
    all_tts_stage_keys,
    detect_tts_model_type,
    resolve_adapter,
@@ -79,7 +75,6 @@ from vllm_omni.utils.speaker_cache import get_speaker_cache

 logger = init_logger(__name__)

-
 _SPEECH_USAGE_INPUT_TOKENS_HEADER = "X-VLLM-OMNI-INPUT-TOKENS"
 _SPEECH_USAGE_OUTPUT_TOKENS_HEADER = "X-VLLM-OMNI-OUTPUT-TOKENS"
 _SPEECH_USAGE_TOTAL_TOKENS_HEADER = "X-VLLM-OMNI-TOTAL-TOKENS"
@@ -280,8 +275,6 @@ def _conditioning_cache_salt(request, tts_params: dict | None = None) -> str:

 class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
    _diffusion_mode: bool = False
-    _media_connector: MediaConnector | None = None
-    _allowed_local_media_path: str = ""
    _tts_executor: ThreadPoolExecutor | None = None

    def _init_speaker_storage(self) -> None:
@@ -398,8 +391,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        diffusion_engine: "Any",
        model_name: str,
        stage_configs: "list[Any] | None" = None,
-        allowed_local_media_path: str = "",
-        allowed_media_domains: list[str] | None = None,
    ) -> "OmniOpenAIServingSpeech":
        """Create a speech serving instance for pure diffusion TTS models.

@@ -411,11 +402,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        instance._diffusion_engine = diffusion_engine
        instance._diffusion_model_name = model_name
        instance._diffusion_stage_configs = stage_configs
-        instance._allowed_local_media_path = allowed_local_media_path
-        instance._media_connector = MediaConnector(
-            allowed_local_media_path=allowed_local_media_path,
-            allowed_media_domains=allowed_media_domains,
-        )
        instance._tts_model_type = "omnivoice"
        instance._is_tts = False
        # Diffusion-only instances don't have a TTS stage; set None so any
@@ -426,8 +412,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        return instance

    def __init__(self, *args, **kwargs):
-        self._media_connector = None
-        self._allowed_local_media_path = ""
        self.model_name = kwargs.pop("model_name", None)
        # True when the server was launched with --forced-aligner (a pooling
        # aligner stage is appended to the pipeline). Gates word_timestamps.
@@ -752,20 +736,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        details = self._compute_speech_input_details(request, tts_params)
        return build_speech_usage(details, output_tokens)

-    def _validate_tts_generation(
-        self,
-        tts_params: dict[str, Any],
-        usage_acc: SpeechOutputTokenCounter,
-    ) -> None:
-        adapter = self._get_tts_adapter()
-        if adapter is None:
-            return
-        adapter.validate_generation(
-            tts_params,
-            stage0_finish_reason=usage_acc.stage0_finish_reason,
-            output_tokens=usage_acc.total(),
-        )
-
    @staticmethod
    def _build_speech_usage_headers(usage: SpeechTokenUsage | None) -> dict[str, str]:
        """Map speech usage into non-streaming response headers.
@@ -1890,16 +1860,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        """
        # Pass the allowed-local-media-path so the stat is restricted to
        # paths the server operator has explicitly permitted.
-        allowed_path: str | None
-        if self._diffusion_mode:
-            allowed_path = self._allowed_local_media_path
-        else:
+        allowed_path = None
+        if not self._diffusion_mode:
            allowed_path = getattr(self.model_config, "allowed_local_media_path", None)

        wav_list: list[float] | None = None
        sr = 0
        artifact_key = ""
        post_key = ""
+        connector: MediaConnector | None = None
        for attempt in range(_REF_AUDIO_METADATA_FETCH_ATTEMPTS):
            cache_key = await self._ref_audio_cache_key(ref_audio_str, allowed_path)
            cached = self._ref_audio_resolve_cache.get(cache_key)
@@ -1914,15 +1883,19 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                )
                return wav_list, sr, cache_key

-            if self._media_connector is None:
-                model_config = self.model_config
-                self._media_connector = MediaConnector(
-                    allowed_local_media_path=model_config.allowed_local_media_path,
-                    allowed_media_domains=model_config.allowed_media_domains,
-                )
+            if connector is None:
+                # In diffusion mode, model_config may not be available
+                if self._diffusion_mode:
+                    connector = MediaConnector()
+                else:
+                    model_config = self.model_config
+                    connector = MediaConnector(
+                        allowed_local_media_path=model_config.allowed_local_media_path,
+                        allowed_media_domains=model_config.allowed_media_domains,
+                    )

            fetch_start_s = time.perf_counter()
-            wav_np, fetched_sr = await self._media_connector.fetch_audio_async(ref_audio_str)
+            wav_np, fetched_sr = await connector.fetch_audio_async(ref_audio_str)
            fetch_decode_ms = (time.perf_counter() - fetch_start_s) * 1000.0
            tolist_start_s = time.perf_counter()
            wav_list, sr, artifact_key, duration = self._finalize_fetched_ref_audio(wav_np, fetched_sr)
@@ -2189,7 +2162,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        request_start_s: float | None = None,
        include_sample_rate: bool = False,
        usage_acc: SpeechOutputTokenCounter | None = None,
-        tts_params: dict[str, Any] | None = None,
        collect: dict | None = None,
    ):
        """Generate audio chunks for streaming response.
@@ -2215,13 +2187,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        stream_start_s = request_start_s if request_start_s is not None else time.perf_counter()
        artifact_ready = False

-        # SSE supplies an accumulator for usage output. Raw-audio and WebSocket
-        # streams retain terminal metrics only when their model adapter needs
-        # generation validation.
-        adapter = self._get_tts_adapter()
-        if tts_params is not None and usage_acc is None and adapter is not None and adapter.validates_generation:
-            usage_acc = SpeechOutputTokenCounter()
-
        try:
            async for res in generator:
                # Tally generated codec tokens for usage (reads per-stage metrics
@@ -2299,11 +2264,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                # Audex contract: zero codec tokens must abort the stream, not
                # complete it cleanly with zero audio bytes.
                raise ValueError("Audex produced no audio (the thinker emitted zero or invalid codec tokens)")
-            # Check before committing the reference-audio artifact or logging
-            # success. Streaming protocols may already have emitted partial
-            # bytes, but they must terminate as an error rather than cleanly.
-            if tts_params is not None and usage_acc is not None:
-                self._validate_tts_generation(tts_params, usage_acc)
            self._mark_ref_audio_artifact_ready_for_request(request_id)
            artifact_ready = True
            total_ms = (time.perf_counter() - stream_start_s) * 1000.0
@@ -2388,7 +2348,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        ``input_tokens`` is computed from the request text + reference audio.
        """
        usage_acc = SpeechOutputTokenCounter()
-        emitted_audio = False
        try:
            async for chunk in self._generate_audio_chunks(
                generator,
@@ -2397,7 +2356,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                raw_request=raw_request,
                request_start_s=request_start_s,
                usage_acc=usage_acc,
-                tts_params=tts_params,
            ):
                payload = {
                    "type": "speech.audio.delta",
@@ -2405,7 +2363,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                    "response_format": response_format,
                }
                data = json.dumps(payload, separators=(",", ":"))
-                emitted_audio = True
                yield f"event: speech.audio.delta\ndata: {data}\n\n"
            done_payload: dict[str, Any] = {"type": "speech.audio.done"}
            if request is not None:
@@ -2417,19 +2374,16 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        except asyncio.CancelledError:
            raise
        except Exception as e:
-            error: dict[str, Any] = {
-                "message": str(e),
-                "type": "server_error",
-                "param": None,
-                "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
-            }
-            if emitted_audio:
-                error.update(partial_audio=True, action="discard")
-            error_payload: dict[str, Any] = {
+            payload = {
                "type": "speech.audio.error",
-                "error": error,
+                "error": {
+                    "message": str(e),
+                    "type": "server_error",
+                    "param": None,
+                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
+                },
            }
-            data = json.dumps(error_payload, separators=(",", ":"))
+            data = json.dumps(payload, separators=(",", ":"))
            yield f"event: speech.audio.error\ndata: {data}\n\n"

    @staticmethod
@@ -3144,13 +3098,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        return request_id, generator, tts_params

    async def _generate_pcm_chunks(
-        self,
-        generator,
-        request_id: str,
-        *,
-        include_sample_rate: bool = False,
-        tts_params: dict[str, Any] | None = None,
-        collect: dict | None = None,
+        self, generator, request_id: str, *, include_sample_rate: bool = False, collect: dict | None = None
    ):
        """Yield raw PCM byte chunks from the engine generator.

@@ -3164,16 +3112,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
            request_id,
            response_format="pcm",
            include_sample_rate=include_sample_rate,
-            tts_params=tts_params,
            collect=collect,
        ):
            yield chunk

    async def _iter_pcm_audio_bytes(self, request: OpenAICreateSpeechRequest):
        """Yield raw PCM bytes for a speech request as soon as chunks are decoded."""
-        request_id, generator, tts_params = await self._prepare_speech_generation(request)
+        request_id, generator, _ = await self._prepare_speech_generation(request)
        try:
-            async for chunk in self._generate_pcm_chunks(generator, request_id, tts_params=tts_params):
+            async for chunk in self._generate_pcm_chunks(generator, request_id):
                yield chunk
        finally:
            self._discard_ref_audio_artifact_warmup(request_id)
@@ -3245,8 +3192,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
            if final_output is None:
                raise ValueError("No output generated from the model.")

-            self._validate_tts_generation(bytes_tts_params or {}, usage_acc)
-
            # Extract audio from the audio-bearing res (not necessarily the last
            # yielded one, which may be the aligner's timestamps output).
            audio_source = audio_res if audio_res is not None else final_output
@@ -3571,7 +3516,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                    return error

                media_type = "audio/wav" if response_format == "wav" else "audio/pcm"
-                _, generator, raw_tts_params = await self._prepare_speech_generation(request, request_id=request_id)
+                _, generator, _ = await self._prepare_speech_generation(request, request_id=request_id)
                return StreamingResponse(
                    self._generate_audio_chunks(
                        generator,
@@ -3579,7 +3524,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                        response_format,
                        raw_request=raw_request,
                        request_start_s=request_start_s,
-                        tts_params=raw_tts_params,
                    ),
                    media_type=media_type,
                )
@@ -3608,35 +3552,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):

            collect: dict = {}
            usage_box: list[SpeechTokenUsage] = []
-            try:
-                audio_bytes, media_type = await self._generate_audio_bytes(
-                    request, request_id=request_id, usage_out=usage_box, collect=collect
-                )
-            except TTSGenerationError as error:
-                # An adapter can reject otherwise completed audio. Retry only
-                # retryable, stochastic, server-budgeted non-streaming
-                # requests: changing an explicit seed would violate
-                # reproducibility, and changing a caller-provided token limit
-                # would violate the requested budget.
-                if not error.retryable or request.seed is not None or request.max_new_tokens is not None:
-                    raise
-                retry_seed = int(random_uuid()[:8], 16)
-                retry_request = request.model_copy(update={"seed": retry_seed})
-                retry_request_id = f"speech-{random_uuid()}"
-                collect.clear()
-                usage_box.clear()
-                logger.warning(
-                    "TTS request %s failed generation validation; retrying once as %s with seed=%d",
-                    request_id,
-                    retry_request_id,
-                    retry_seed,
-                )
-                audio_bytes, media_type = await self._generate_audio_bytes(
-                    retry_request,
-                    request_id=retry_request_id,
-                    usage_out=usage_box,
-                    collect=collect,
-                )
+            audio_bytes, media_type = await self._generate_audio_bytes(
+                request, request_id=request_id, usage_out=usage_box, collect=collect
+            )
            total_ms = (time.perf_counter() - request_start_s) * 1000.0
            logger.info(
                "[SpeechE2E] request_id=%s stream=false status=ok total_ms=%.2f response_bytes=%d",
@@ -3689,16 +3607,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                e,
            )
            return self.create_error_response(e)
-        except TTSGenerationError as e:
-            total_ms = (time.perf_counter() - request_start_s) * 1000.0
-            logger.error(
-                "[SpeechE2E] request_id=%s stream=%s status=invalid_generation total_ms=%.2f error=%s",
-                request_id,
-                bool(request.stream),
-                total_ms,
-                e,
-            )
-            return self._diffusion_error_response(str(e), status_code=500)
        except Exception as e:
            total_ms = (time.perf_counter() - request_start_s) * 1000.0
            logger.exception(
@@ -1,6 +1,3 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-
 """WebSocket handler for streaming text input TTS.

 Accepts text incrementally via WebSocket, buffers it until input.done, and
@@ -332,7 +329,7 @@ class OmniStreamingSpeechHandler:
        request_id = None
        try:
            if config.stream_audio:
-                request_id, generator, tts_params = await self._speech_service._prepare_speech_generation(request)
+                request_id, generator, _ = await self._speech_service._prepare_speech_generation(request)
                if config.word_timestamps:
                    total_bytes = await self._stream_audio_with_alignments(
                        websocket=websocket,
@@ -342,16 +339,9 @@ class OmniStreamingSpeechHandler:
                        utterance_index=utterance_index,
                        sentence_index=sentence_index,
                        language=config.language,
-                        tts_params=tts_params,
                    )
                else:
-                    async with aclosing(
-                        self._speech_service._generate_pcm_chunks(
-                            generator,
-                            request_id,
-                            tts_params=tts_params,
-                        )
-                    ) as stream:
+                    async with aclosing(self._speech_service._generate_pcm_chunks(generator, request_id)) as stream:
                        async for chunk in stream:
                            total_bytes += len(chunk)
                            await websocket.send_bytes(chunk)
@@ -377,7 +367,6 @@ class OmniStreamingSpeechHandler:
            await self._send_error(
                websocket,
                f"Generation failed for utterance {utterance_index}, sentence {sentence_index}: {e}",
-                partial_audio=total_bytes > 0,
            )
        finally:
            try:
@@ -403,7 +392,6 @@ class OmniStreamingSpeechHandler:
        utterance_index: int,
        sentence_index: int,
        language: str | None = None,
-        tts_params: dict | None = None,
    ) -> int:
        """Stream PCM as JSON ``audio.chunk`` frames, aligned per sentence.

@@ -450,7 +438,6 @@ class OmniStreamingSpeechHandler:
                generator,
                request_id,
                include_sample_rate=True,
-                tts_params=tts_params,
                collect=collect,
            )
        ) as stream:
@@ -476,15 +463,14 @@ class OmniStreamingSpeechHandler:
        return total_bytes

    @staticmethod
-    async def _send_error(websocket: WebSocket, message: str, *, partial_audio: bool = False) -> None:
+    async def _send_error(websocket: WebSocket, message: str) -> None:
        """Send an error message to the client."""
        try:
-            payload: dict[str, object] = {
-                "type": "error",
-                "message": message,
-            }
-            if partial_audio:
-                payload.update(partial_audio=True, action="discard")
-            await websocket.send_json(payload)
+            await websocket.send_json(
+                {
+                    "type": "error",
+                    "message": message,
+                }
+            )
        except Exception:
            pass  # Connection may already be closed; safe to ignore
@@ -32,7 +32,6 @@ from vllm_omni.entrypoints.openai.stage_params import (
 from vllm_omni.entrypoints.openai.utils import is_video_generation_pipeline, parse_lora_request
 from vllm_omni.entrypoints.openai.video_api_utils import (
    _encode_video_bytes,
-    _PlanarFrameConverter,
    encode_video_base64,
 )
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams, OmniTextPrompt
@@ -45,8 +44,6 @@ from vllm_omni.outputs.output_metadata import (

 logger = init_logger(__name__)

-_VIDEO_RESPONSE_FRAME_CONVERSION_WORKERS = 8
-
 if TYPE_CHECKING:
    from vllm_omni.diffusion.data import OmniDiffusionConfig

@@ -99,11 +96,6 @@ class OmniOpenAIServingVideo:
        self._engine_client = engine_client
        self._model_name = model_name
        self._stage_configs = stage_configs
-        self._video_frame_converter = _PlanarFrameConverter(max_workers=_VIDEO_RESPONSE_FRAME_CONVERSION_WORKERS)
-        logger.info(
-            "Video response frame conversion pool configured: workers=%d",
-            self._video_frame_converter.max_workers,
-        )

    def _resolve_diffusion_od_config(self) -> OmniDiffusionConfig | SimpleNamespace | None:
        get_od_config = getattr(self._engine_client, "get_diffusion_od_config", None)
@@ -178,9 +170,6 @@ class OmniOpenAIServingVideo:
            stage_configs=stage_configs,
        )

-    def shutdown(self) -> None:
-        self._video_frame_converter.shutdown()
-
    async def _run_and_extract(
        self,
        request: VideoGenerationRequest,
@@ -372,7 +361,6 @@ class OmniOpenAIServingVideo:
                        video,
                        fps=artifacts.output_fps,
                        video_codec_options=video_codec_options,
-                        frame_converter=self._video_frame_converter,
                    )
                    if artifacts.audios[idx] is None
                    else encode_video_base64(
@@ -381,7 +369,6 @@ class OmniOpenAIServingVideo:
                        audio=artifacts.audios[idx],
                        audio_sample_rate=artifacts.audio_sample_rate,
                        video_codec_options=video_codec_options,
-                        frame_converter=self._video_frame_converter,
                    )
                ),
                action=artifacts.actions[idx],
@@ -438,7 +425,6 @@ class OmniOpenAIServingVideo:
            fps=artifacts.output_fps,
            **({"audio": audio, "audio_sample_rate": artifacts.audio_sample_rate} if audio is not None else {}),
            video_codec_options=video_codec_options,
-            frame_converter=self._video_frame_converter,
        )
        _t_encode_ms = (time.perf_counter() - _t_encode_start) * 1000
        logger.info("Video response encoding (MP4 bytes): %.2f ms", _t_encode_ms)
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Token-usage accounting for the Speech (``/v1/audio/speech``) API.

 Why this module exists (issue #4646)
@@ -161,7 +160,6 @@ class SpeechOutputTokenCounter:
    """

    output_tokens: int = 0
-    stage0_finish_reason: str | None = None

    def observe(self, res: Any) -> None:
        metrics = getattr(res, "metrics", None)
@@ -170,11 +168,6 @@ class SpeechOutputTokenCounter:
        stage_metrics = metrics.get("stage_metrics")
        if not isinstance(stage_metrics, dict):
            return
-        stage0 = stage_metrics.get("0", stage_metrics.get(0))
-        if isinstance(stage0, dict):
-            reason = stage0.get("finish_reason")
-            if reason is not None:
-                self.stage0_finish_reason = str(reason)
        for stage in stage_metrics.values():
            if not isinstance(stage, dict):
                continue
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Registry of TTS serving adapters, and the model detection built on it.

 Adapters register themselves by their ``name`` (the model-type discriminator)
@@ -20,7 +19,6 @@ from vllm_omni.entrypoints.openai.tts_adapters.base import (
    OutputPolicy,
    PreparedRequest,
    SpeechServingContext,
-    TTSGenerationError,
    TTSModelAdapter,
 )

@@ -128,7 +126,6 @@ __all__ = [
    "OutputPolicy",
    "PreparedRequest",
    "SpeechServingContext",
-    "TTSGenerationError",
    "TTSModelAdapter",
    "TTS_ADAPTER_REGISTRY",
    "all_tts_model_types",
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Base contract for per-model TTS serving adapters.

 This package factors the per-model ``if self._tts_model_type == ...`` dispatch
@@ -12,7 +11,7 @@ See the RFC for the full design (issue #4327).
 """

 from abc import ABC, abstractmethod
-from collections.abc import Callable, Mapping
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, ClassVar

@@ -70,14 +69,6 @@ def apply_max_new_tokens(
    return sampling_params_list


-class TTSGenerationError(RuntimeError):
-    """A completed TTS generation that must not be returned as valid audio."""
-
-    def __init__(self, message: str, *, retryable: bool = False) -> None:
-        super().__init__(message)
-        self.retryable = retryable
-
-
@dataclass
 class OutputPolicy:
    """How the orchestrator aggregates engine output for a model.
@@ -162,8 +153,6 @@ class TTSModelAdapter(ABC):
    detect_priority: ClassVar[int] = 100
    #: Serving backend: ``"ar"`` (engine_client) or ``"diffusion"``.
    backend: ClassVar[str] = "ar"
-    #: Whether streaming entrypoints must retain terminal metrics for validation.
-    validates_generation: ClassVar[bool] = False
    #: Whether the model consumes ``request.speed`` in its native parameters.
    native_speed_control: ClassVar[bool] = False

@@ -244,21 +233,6 @@ class TTSModelAdapter(ABC):
        """
        return sampling_params_list

-    def validate_generation(
-        self,
-        tts_params: Mapping[str, object],
-        *,
-        stage0_finish_reason: str | None,
-        output_tokens: int,
-    ) -> None:
-        """Reject a completed generation that is invalid for this model.
-
-        Adapters that override this hook must also set
-        :attr:`validates_generation` so streaming entrypoints retain the
-        terminal metrics needed by the validation without charging that cost
-        to unrelated TTS models.
-        """
-
    async def warmup(self) -> None:
        return

@@ -330,6 +304,5 @@ __all__ = [
    "OutputPolicy",
    "PreparedRequest",
    "SpeechServingContext",
-    "TTSGenerationError",
    "TTSModelAdapter",
 ]
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """MOSS-TTS serving adapters (Nano + full family).

 Both variants share the same build/validate flow (``_build_moss_tts_params``
@@ -98,13 +97,9 @@ class _MossTTSAdapterBase(ARTTSAdapter):
            if voice_lower in server.uploaded_speakers and not has_inline_ref_audio:
                tts_params["voice_name"] = [voice_lower]
                tts_params["voice_created_at"] = [server._voice_created_at(voice_lower)]
-        # MOSS samples internally from additional_information. build() runs
-        # before the shared path applies request.seed to SamplingParams.
-        seed = request.seed
-        if seed is None and sampling_params_list:
-            seed = getattr(sampling_params_list[0], "seed", None)
-        if seed is not None:
-            tts_params["seed"] = [int(seed)]
+        # MOSS reads the resolved seed at build time (it samples internally).
+        if sampling_params_list and getattr(sampling_params_list[0], "seed", None) is not None:
+            tts_params["seed"] = [sampling_params_list[0].seed]
        if isinstance(tts_params.get("prompt_token_ids"), list):
            prompt_token_ids = tts_params.pop("prompt_token_ids")
            prompt = tokens_input(prompt_token_ids=prompt_token_ids)
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Qwen3-TTS serving adapter."""

 from collections.abc import Mapping
@@ -13,7 +12,7 @@ from vllm_omni.entrypoints.openai.tts_adapters.base import (
    DEFAULT_TTS_LANGUAGES,
    ARTTSAdapter,
    PreparedRequest,
-    TTSGenerationError,
+    apply_max_new_tokens,
 )
 from vllm_omni.entrypoints.openai.tts_adapters.capabilities import load_precomputed_speakers
 from vllm_omni.utils.speaker_cache import validate_qwen3_tts_profile
@@ -23,23 +22,11 @@ if TYPE_CHECKING:

 logger = init_logger(__name__)

-QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY = "_qwen3_tts_effective_max_tokens"
-_MIN_CODEC_FRAMES = 192
-_MAX_CODEC_FRAMES_PER_TEXT_TOKEN = 12
-
-
-class Qwen3TTSCodecLimitError(TTSGenerationError):
-    """Qwen3-TTS Base exhausted its codec budget without emitting EOS."""
-
-    def __init__(self, message: str) -> None:
-        super().__init__(message, retryable=True)
-

@register_tts_adapter
 class Qwen3TTSAdapter(ARTTSAdapter):
    """Adapter for Qwen3-TTS (AR ``engine_client`` backend)."""

-    validates_generation = True
    stage_keys = frozenset({"qwen3_tts"})
    name = "qwen3_tts"

@@ -239,95 +226,4 @@ class Qwen3TTSAdapter(ARTTSAdapter):
        prompt: dict[str, Any] | None = None,
        request_id: str | None = None,
    ) -> list:
-        """Apply a text-scaled safety ceiling to Base codec generation.
-
-        Qwen3-TTS can rarely enter a repetitive state in which codec EOS is no
-        longer reachable through top-k sampling. A fixed 4096-frame ceiling
-        turns that into several minutes of unusable audio. Bound the default
-        Base-task budget by text length, while preserving an explicit caller
-        ``max_new_tokens`` override and the configured budget for other tasks.
-        """
-        import copy
-
-        del request_id
-        server = self.ctx.server
-        # Only scalar fields on stage 0 are changed below. Shallow-copy each
-        # stage so the shared defaults stay immutable without deep-copying the
-        # complete sampling configuration on every request.
-        sampling_params_list = [copy.copy(params) for params in sampling_params_list]
-        configured_cap = getattr(sampling_params_list[0], "max_tokens", None)
-        task_type = request.task_type or "CustomVoice"
-        text_tokens = None
-        dynamic_cap = None
-        effective_cap: int | None
-
-        if request.max_new_tokens is not None:
-            # An explicit request budget is an opt-out from the automatic
-            # ceiling. It remains an upper bound, and a length finish is still
-            # surfaced as an incomplete generation rather than valid audio.
-            effective_cap = int(request.max_new_tokens)
-        elif task_type == "Base":
-            counted_text_tokens = server._count_usage_text_tokens(request.input)
-            if counted_text_tokens > 0:
-                text_tokens = counted_text_tokens
-                dynamic_cap = max(_MIN_CODEC_FRAMES, text_tokens * _MAX_CODEC_FRAMES_PER_TEXT_TOKEN)
-                effective_cap = min(dynamic_cap, int(configured_cap)) if configured_cap is not None else dynamic_cap
-            else:
-                # Token counting is best-effort. If the tokenizer is missing or
-                # rejects the input, preserve the configured budget instead of
-                # truncating an otherwise valid request at the minimum ceiling.
-                effective_cap = int(configured_cap) if configured_cap is not None else None
-        else:
-            effective_cap = int(configured_cap) if configured_cap is not None else None
-
-        if effective_cap is not None:
-            effective_cap = max(1, effective_cap)
-            sampling_params_list[0].max_tokens = effective_cap
-            sampling_params_list[0].min_tokens = min(
-                int(getattr(sampling_params_list[0], "min_tokens", 0) or 0),
-                effective_cap,
-            )
-
-        if isinstance(prompt, dict):
-            additional_information = prompt.get("additional_information")
-            if isinstance(additional_information, dict) and effective_cap is not None:
-                additional_information[QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY] = [effective_cap]
-
-        logger.debug(
-            "Qwen3-TTS codec budget: task_type=%s text_tokens=%s dynamic_cap=%s "
-            "configured_cap=%s request_cap=%s effective_cap=%s",
-            task_type,
-            text_tokens,
-            dynamic_cap,
-            configured_cap,
-            request.max_new_tokens,
-            effective_cap,
-        )
-        return sampling_params_list
-
-    def validate_generation(
-        self,
-        tts_params: Mapping[str, object],
-        *,
-        stage0_finish_reason: str | None,
-        output_tokens: int,
-    ) -> None:
-        if QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY not in tts_params:
-            return
-        task_type = tts_params.get("task_type")
-        if isinstance(task_type, (list, tuple)):
-            task_type = task_type[0] if task_type else None
-        if task_type != "Base" or stage0_finish_reason != "length":
-            return
-
-        raw_limit = tts_params.get(QWEN3_TTS_EFFECTIVE_MAX_TOKENS_KEY)
-        if isinstance(raw_limit, (list, tuple)):
-            raw_limit = raw_limit[0] if raw_limit else None
-        try:
-            limit = int(raw_limit) if isinstance(raw_limit, (str, bytes, bytearray, int, float)) else 0
-        except (TypeError, ValueError):
-            limit = 0
-        raise Qwen3TTSCodecLimitError(
-            "Qwen3-TTS Base did not emit codec EOS before its token budget "
-            f"({output_tokens}/{limit} codec tokens); the generated audio is incomplete."
-        )
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -10,12 +10,9 @@ import base64
 import binascii
 import os
 import tempfile
-import threading
-from collections import deque
-from collections.abc import Generator
-from concurrent.futures import Future, ThreadPoolExecutor
+from collections.abc import Iterator
 from io import BytesIO
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast

 import httpx
 import numpy as np
@@ -579,112 +576,6 @@ def _direct_planar_fallback_reason(
    return None


-class _PlanarFrameConverter:
-    """Convert response frames with one reusable, bounded executor."""
-
-    def __init__(self, max_workers: int) -> None:
-        self._max_workers = max_workers
-        if self._max_workers < 1:
-            raise ValueError("max_workers must be positive")
-        self._executor = (
-            ThreadPoolExecutor(max_workers=self._max_workers, thread_name_prefix="video-planar")
-            if self._max_workers > 1
-            else None
-        )
-        self._scratch_local = threading.local()
-        self._shutdown = False
-
-    @property
-    def max_workers(self) -> int:
-        return self._max_workers
-
-    def _channel_scratch(self, height: int, width: int, common_dtype: np.dtype) -> np.ndarray:
-        scratch_dtype = np.dtype(common_dtype)
-        if not np.issubdtype(scratch_dtype, np.floating):
-            scratch_dtype = np.dtype(np.float64)
-        scratch = getattr(self._scratch_local, "channel", None)
-        if scratch is None or scratch.shape != (height, width) or scratch.dtype != scratch_dtype:
-            scratch = np.empty((height, width), dtype=scratch_dtype)
-            self._scratch_local.channel = scratch
-        return scratch
-
-    def _build_frame(self, frame: np.ndarray, common_dtype: np.dtype) -> av.VideoFrame:
-        """Build one quantized GBR PyAV frame."""
-        import av
-
-        height, width = frame.shape[:2]
-        scratch = None if frame.dtype == np.uint8 else self._channel_scratch(height, width, common_dtype)
-        av_frame = av.VideoFrame(width, height, format="gbrp")
-        for plane, channel in zip(av_frame.planes, (1, 2, 0)):
-            if plane.height < height or plane.line_size < width:
-                raise ValueError("PyAV video plane is smaller than the requested frame dimensions.")
-            plane_view = np.frombuffer(
-                memoryview(plane),
-                dtype=np.uint8,
-                count=plane.height * plane.line_size,
-            ).reshape(plane.height, plane.line_size)
-            plane_view.fill(0)
-            if frame.dtype == np.uint8:
-                plane_view[:height, :width] = frame[..., channel]
-            else:
-                np.copyto(scratch, frame[..., channel], casting="unsafe")
-                np.clip(scratch, 0.0, 1.0, out=scratch)
-                scratch *= 255.0
-                np.rint(scratch, out=scratch)
-                plane_view[:height, :width] = scratch
-        return av_frame
-
-    def iter_frames(
-        self,
-        frames: list[np.ndarray],
-        common_dtype: np.dtype,
-    ) -> Generator[av.VideoFrame, None, None]:
-        """Yield converted frames in order with at most two batches in flight."""
-        effective_workers = min(len(frames), self._max_workers)
-        if effective_workers <= 1:
-            for frame in frames:
-                yield self._build_frame(frame, common_dtype)
-            return
-
-        executor = self._executor
-        assert executor is not None
-        pending: deque[Future[av.VideoFrame]] = deque()
-        next_index = 0
-        max_pending = 2 * effective_workers
-        try:
-            while next_index < len(frames) and len(pending) < max_pending:
-                pending.append(executor.submit(self._build_frame, frames[next_index], common_dtype))
-                next_index += 1
-
-            while pending:
-                converted_frame = pending.popleft().result()
-                if next_index < len(frames):
-                    pending.append(executor.submit(self._build_frame, frames[next_index], common_dtype))
-                    next_index += 1
-                yield converted_frame
-        finally:
-            for future in pending:
-                future.cancel()
-
-    def shutdown(self) -> None:
-        if not self._shutdown:
-            self._shutdown = True
-            if self._executor is not None:
-                self._executor.shutdown(wait=True, cancel_futures=True)
-
-
-def _iter_planar_video_frames(
-    frames: list[np.ndarray],
-    common_dtype: np.dtype,
-) -> Generator[av.VideoFrame, None, None]:
-    """Yield serial planar frames for direct utility callers."""
-    converter = _PlanarFrameConverter(max_workers=1)
-    try:
-        yield from converter.iter_frames(frames, common_dtype)
-    finally:
-        converter.shutdown()
-
-
 def _log_video_encoding_path(
    *,
    selected_path: str,
@@ -694,13 +585,12 @@ def _log_video_encoding_path(
    fps: int,
    audio: AudioInput | None,
    audio_sample_rate: int | None,
-    effective_frame_conversion_workers: int,
    reason: str | None = None,
 ) -> None:
    reason_field = "" if reason is None else f" reason={reason}"
    logger.info(
        "Video response encoding route selected: selected_path=%s%s frames=%d frame_shape=%s dtype=%s fps=%s "
-        "audio_present=%s effective_audio_sample_rate=%s effective_frame_conversion_workers=%s",
+        "audio_present=%s effective_audio_sample_rate=%s",
        selected_path,
        reason_field,
        len(frames),
@@ -709,7 +599,6 @@ def _log_video_encoding_path(
        fps,
        audio is not None,
        audio_sample_rate,
-        effective_frame_conversion_workers,
    )


@@ -722,6 +611,40 @@ def _resolve_audio_sample_rate(audio: AudioInput | None, audio_sample_rate: int
    return audio_sample_rate or DEFAULT_AUDIO_SAMPLE_RATE


+def _iter_planar_video_frames(
+    frames: list[np.ndarray],
+    common_dtype: np.dtype,
+) -> Iterator[av.VideoFrame]:
+    """Yield planar PyAV frames while retaining only one channel scratch buffer."""
+    import av
+
+    height, width = frames[0].shape[:2]
+    scratch_dtype = np.float64 if np.issubdtype(common_dtype, np.bool_) else common_dtype
+    scratch = None if common_dtype == np.uint8 else np.empty((height, width), dtype=scratch_dtype)
+
+    for frame in frames:
+        av_frame = av.VideoFrame(width, height, format="gbrp")
+        for plane, channel in zip(av_frame.planes, (1, 2, 0)):
+            if plane.height < height or plane.line_size < width:
+                raise ValueError("PyAV video plane is smaller than the requested frame dimensions.")
+            plane_view = np.frombuffer(
+                plane,
+                dtype=np.uint8,
+                count=plane.height * plane.line_size,
+            ).reshape(plane.height, plane.line_size)
+            plane_view.fill(0)
+            if frame.dtype == np.uint8:
+                plane_view[:height, :width] = frame[..., channel]
+            else:
+                scratch_buffer = cast(np.ndarray, scratch)
+                np.copyto(scratch_buffer, frame[..., channel], casting="unsafe")
+                np.clip(scratch_buffer, 0.0, 1.0, out=scratch_buffer)
+                scratch_buffer *= 255.0
+                np.rint(scratch_buffer, out=scratch_buffer)
+                plane_view[:height, :width] = scratch_buffer
+        yield av_frame
+
+
 def _encode_prepared_video_bytes_legacy(
    frames: list[np.ndarray],
    frame_shape: tuple[int, ...],
@@ -770,9 +693,8 @@ def _encode_video_bytes(
    audio: Any | None = None,
    audio_sample_rate: int | None = None,
    video_codec_options: dict[str, str] | None = None,
-    frame_converter: _PlanarFrameConverter | None = None,
 ) -> bytes:
-    """Encode a video payload through the direct planar or legacy path."""
+    """Encode a video payload through the automatic capability dispatcher."""
    from vllm_omni.diffusion.utils.media_utils import mux_av_video_audio_bytes

    # Prepare once so validation is shared by both paths and malformed common
@@ -790,7 +712,6 @@ def _encode_video_bytes(
            fps=fps,
            audio=audio,
            audio_sample_rate=effective_audio_sample_rate,
-            effective_frame_conversion_workers=0,
        )
        return _encode_prepared_video_bytes_legacy(
            frames,
@@ -802,8 +723,6 @@ def _encode_video_bytes(
            video_codec_options=video_codec_options,
        )

-    owns_frame_converter = frame_converter is None
-    converter = frame_converter or _PlanarFrameConverter(max_workers=1)
    _log_video_encoding_path(
        selected_path="direct_planar",
        frames=frames,
@@ -812,24 +731,17 @@ def _encode_video_bytes(
        fps=fps,
        audio=audio,
        audio_sample_rate=effective_audio_sample_rate,
-        effective_frame_conversion_workers=min(len(frames), converter.max_workers),
    )
    audio_np = _coerce_audio_to_numpy(audio) if audio is not None else None
-    video_frames = converter.iter_frames(frames, common_dtype)
-    try:
-        return mux_av_video_audio_bytes(
-            video_frames,
-            width=frame_shape[1],
-            height=frame_shape[0],
-            audio_waveform=audio_np,
-            fps=float(fps),
-            audio_sample_rate=effective_audio_sample_rate,
-            video_codec_options=video_codec_options,
-        )
-    finally:
-        video_frames.close()
-        if owns_frame_converter:
-            converter.shutdown()
+    return mux_av_video_audio_bytes(
+        _iter_planar_video_frames(frames, common_dtype),
+        width=frame_shape[1],
+        height=frame_shape[0],
+        audio_waveform=audio_np,
+        fps=float(fps),
+        audio_sample_rate=effective_audio_sample_rate,
+        video_codec_options=video_codec_options,
+    )


 class FragmentedMP4VideoEncoder:
@@ -887,7 +799,6 @@ def encode_video_base64(
    audio: Any | None = None,
    audio_sample_rate: int | None = None,
    video_codec_options: dict[str, str] | None = None,
-    frame_converter: _PlanarFrameConverter | None = None,
 ) -> str:
    """Encode a video (frames/array/tensor) to base64 MP4."""
    video_bytes = _encode_video_bytes(
@@ -896,6 +807,5 @@ def encode_video_base64(
        audio=audio,
        audio_sample_rate=audio_sample_rate,
        video_codec_options=video_codec_options,
-        frame_converter=frame_converter,
    )
    return base64.b64encode(video_bytes).decode("utf-8")
@@ -45,7 +45,6 @@ class StageRequestStats:
    stage_id: int | None = None
    replica_id: int | None = None
    final_output_type: str | None = None
-    finish_reason: str | None = None
    request_id: str | None = None
    postprocess_time_ms: float = 0.0
    diffusion_metrics: dict[str, float] = None
@@ -127,7 +126,6 @@ STAGE_EXCLUDE = {
    "rx_decode_time_ms",
    "rx_in_flight_time_ms",
    "final_output_type",
-    "finish_reason",
    "pipeline_timings",
 }
 TRANSFER_EXCLUDE = {"from_stage", "to_stage", "request_id", "used_shm"}
@@ -475,7 +473,6 @@ class OrchestratorAggregator:
            current = {
                "stage_id": sid,
                "final_output_type": evt.final_output_type,
-                "finish_reason": evt.finish_reason,
                defs.NUM_TOKENS_IN: int(evt.num_tokens_in),
                defs.NUM_TOKENS_OUT: int(evt.num_tokens_out),
                defs.STAGE_GEN_TIME_MS: float(evt.stage_gen_time_ms),
@@ -501,8 +498,6 @@ class OrchestratorAggregator:

        current[defs.NUM_TOKENS_IN] = int(current.get(defs.NUM_TOKENS_IN, 0)) + int(evt.num_tokens_in)
        current[defs.NUM_TOKENS_OUT] = int(current.get(defs.NUM_TOKENS_OUT, 0)) + int(evt.num_tokens_out)
-        if evt.finish_reason is not None:
-            current["finish_reason"] = evt.finish_reason
        current[defs.STAGE_GEN_TIME_MS] = float(current.get(defs.STAGE_GEN_TIME_MS, 0.0)) + float(evt.stage_gen_time_ms)
        current[defs.POSTPROCESS_TIME_MS] = float(current.get(defs.POSTPROCESS_TIME_MS, 0.0)) + float(
            evt.postprocess_time_ms
@@ -2000,12 +2000,8 @@ class MossAudioTokenizerModel(MossAudioTokenizerPreTrainedModel):
        if codes_lengths is None:
            codes_lengths = torch.full((B,), T, device=device, dtype=torch.long)

-        # Keep eager execution and CUDA/NPU Graph capture on the same BF16 path.
-        # Autocast is required on NPU because #5235's decode-LUT uses .float()
-        # for precision, producing float32 embeddings.  Without autocast, these
-        # float32 tensors flow into bf16 LayerNorm weights and NPU's aclnnLayerNorm
-        # rejects the mixed dtype (CUDA auto-casts; NPU does not).
-        with torch.autocast(device_type=device.type, dtype=torch.bfloat16, enabled=device.type != "cpu"):
+        # Keep eager execution and CUDA Graph capture on the same BF16 path.
+        with torch.autocast(device_type=device.type, dtype=torch.bfloat16, enabled=device.type == "cuda"):
            quantizer = cast(MossAudioTokenizerResidualVQ | MossAudioTokenizerResidualLFQ, self.quantizer)
            zq = quantizer.decode_codes(codes)

@@ -44,14 +44,6 @@ from vllm_omni.model_executor.models.step_audio2.step_audio2_constants import (
 logger = init_logger(__name__)


-def _get_left_context_size(info: dict[str, Any]) -> Any | None:
-    """Return the async-chunk marker from a nested metadata payload."""
-    meta = info.get("meta")
-    if not isinstance(meta, dict):
-        return None
-    return meta.get("left_context_size")
-
-
 def fade_in_out(
    fade_in_mel: torch.Tensor,
    fade_out_mel: torch.Tensor,
@@ -548,8 +540,7 @@ class StepAudio2Token2WavForConditionalGeneration(nn.Module, SupportsPP):
            additional: Additional information dict (deprecated)
            runtime_additional_information: Per-request dicts from the
                async_chunk processor.  Each dict contains ``audio_tokens``
-                (list[int]) and ``meta.left_context_size`` as the
-                last-chunk marker (0 = non-final, 1 = final).
+                (list[int]) and ``last_chunk`` (bool).
            **kwargs: Other kwargs

        Returns:
@@ -562,7 +553,7 @@ class StepAudio2Token2WavForConditionalGeneration(nn.Module, SupportsPP):
        # [{}]), so a bare truthiness check would incorrectly enter this
        # branch in synchronous mode.
        if runtime_additional_information and any(
-            _get_left_context_size(info) is not None for info in runtime_additional_information
+            "left_context_size" in info for info in runtime_additional_information
        ):
            return self._forward_async_chunk(input_ids, runtime_additional_information, **kwargs)

@@ -631,15 +622,15 @@ class StepAudio2Token2WavForConditionalGeneration(nn.Module, SupportsPP):
        async_chunk payload to ``request.additional_information``).
        """
        # --- batch=1 guard ---
-        batch_size = sum(1 for info in runtime_additional_information if _get_left_context_size(info) is not None)
+        batch_size = sum(1 for info in runtime_additional_information if "left_context_size" in info)
        if batch_size != 1:
            raise RuntimeError(
                f"Token2Wav async_chunk only supports batch=1, got {batch_size}. "
                "Batch>1 requires framework support for batch_req_ids."
            )

-        info = next(info for info in runtime_additional_information if _get_left_context_size(info) is not None)
-        last_chunk = _get_left_context_size(info) == 1
+        info = next(info for info in runtime_additional_information if "left_context_size" in info)
+        last_chunk = info.get("left_context_size", 0) == 1

        # --- Manage single stream state ---
        # If the previous request was preempted (setup_done but not finished),
@@ -53,7 +53,7 @@ class NPUTorchProfilerWrapper(OmniTorchProfilerWrapper):
            export_type=torch_npu.profiler.ExportType.Text,
            profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
            msprof_tx=False,
-            aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
+            aic_metrics=torch_npu.profiler.AiCMetrics.AiCoreNone,
            l2_cache=False,
            op_attr=False,
            data_simplification=True,
Author	SHA1	Message	Date
hsliuustc0106	b0942b9c23	[Core] Validate HSDP shard size before initialization Signed-off-by: hsliuustc0106 <hsliuustc0106@users.noreply.github.com>	2026-08-29 06:00:36 +08:00
hsliuustc0106	5c8e0e85d8	[Core] Identify HSDP replica ranks in worker titles Signed-off-by: hsliuustc0106 <hsliuustc0106@users.noreply.github.com>	2026-08-28 18:27:56 +08:00
hsliuustc0106	65ac6794f9	[Core] Restore topology-aware diffusion worker titles Signed-off-by: hsliuustc0106 <hsliuustc0106@users.noreply.github.com>	2026-08-28 18:19:10 +08:00