Address video-audio example review feedback

Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Harden the shared video-audio example
2026-08-17 16:45:55 +08:00 · 2026-08-17 13:48:27 +08:00 · 2026-08-16 21:14:26 -07:00 · 2026-08-17 11:21:28 +08:00 · 2026-08-17 10:51:17 +08:00
76 changed files with 2002 additions and 905 deletions
@@ -159,19 +159,7 @@ steps:
          - export VLLM_ALLOW_LONG_MAX_MODEL_LEN="1"
          - export VLLM_USE_DEEP_GEMM="0"
          - export VLLM_MOE_USE_DEEP_GEMM="0"
-          - pytest -s -v tests/e2e/ -m "full_model and L4 and tts" --run-level "full_model" --ignore=tests/e2e/accuracy --ignore=tests/e2e/online_serving/test_voxcpm2_tts_expansion.py
-        mirror_hardwares: l4_1
-
-      - label: ":full_moon: TTS · Accuracy Test · VoxCPM2"
-        timeout_in_minutes: 30
-        commands:
-          - |
-            timeout 30m bash -c "
-              export VLLM_LOGGING_LEVEL=DEBUG
-              export VLLM_WORKER_MULTIPROC_METHOD=spawn
-              export VLLM_ALLOW_LONG_MAX_MODEL_LEN=1
-              pytest -s -v tests/e2e/online_serving/test_voxcpm2_tts_expansion.py -m 'full_model and cuda and tts' --run-level 'full_model'
-            "
+          - pytest -s -v tests/e2e/ -m "full_model and L4 and tts" --run-level "full_model" --ignore=tests/e2e/accuracy
        mirror_hardwares: l4_1

      - label: ":full_moon: TTS · Perf Test"
@@ -188,35 +176,6 @@ steps:
            exit $$EXIT
        mirror_hardwares: h100_2

-      - label: ":full_moon: TTS · VoxCPM2 · Perf Test"
-        key: nightly-tts-performance-voxcpm2
-        timeout_in_minutes: 180
-        commands:
-          - export BENCHMARK_DIR=tests/dfx/perf/results
-          - export VLLM_ALLOW_LONG_MAX_MODEL_LEN="1"
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_benchmark.py --test-config-file tests/dfx/perf/tests/test_voxcpm2.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/*.json"
-            exit $$EXIT
-        mirror_hardwares: h100_1
-
-      - label: ":full_moon: TTS · Higgs-Audio-V3 · Perf Test"
-        key: nightly-tts-performance-higgs-audio-v3
-        timeout_in_minutes: 180
-        commands:
-          - export BENCHMARK_DIR=tests/dfx/perf/results
-          - export VLLM_USE_DEEP_GEMM="0"
-          - export VLLM_MOE_USE_DEEP_GEMM="0"
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_benchmark.py --test-config-file tests/dfx/perf/tests/test_higgs_audio_v3.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/*.json"
-            exit $$EXIT
-        mirror_hardwares: h100_1
-
  - group: ":card_index_dividers: Tiny Model Tests [Multi-GPU]"
    key: nightly-tiny-model-test-group
    depends_on: upload-nightly-pipeline
@@ -246,8 +205,7 @@ steps:
      build.pull_request.labels includes "nightly-test" ||
      build.pull_request.labels includes "diffusion-x2iat-test"
    steps:
-      # Use explicit file shards so single-card cases do not reserve 2+ GPUs,
-      # and Bagel feature coverage is not duplicated via a broad tests/e2e sweep.
+      # Use explicit file shards so single-card cases do not reserve 2+ GPUs.

      - label: ":full_moon: Diffusion X2I(&A&T) · Function Test with H100 · Single-GPU"
        timeout_in_minutes: 120
@@ -255,15 +213,6 @@ steps:
          - >-
            pytest -sv
            tests/e2e/online_serving/test_qwen_image_expansion.py
-            tests/e2e/online_serving/test_qwen_image_edit_expansion.py
-            tests/e2e/online_serving/test_qwen_image_layered_expansion.py
-            tests/e2e/online_serving/test_boogu_image.py
-            tests/e2e/online_serving/test_boogu_image_edit.py
-            tests/e2e/online_serving/test_boogu_image_expansion.py
-            tests/e2e/online_serving/test_boogu_image_edit_expansion.py
-            tests/e2e/online_serving/test_bagel_expansion.py
-            tests/e2e/online_serving/test_krea2_expansion.py
-            tests/e2e/offline_inference/test_krea2_expansion.py
            -m "full_model and diffusion and H100 and not distributed_cuda"
            --run-level "full_model"
        mirror_hardwares: h100_1
@@ -274,47 +223,14 @@ steps:
          - >-
            pytest -sv
            tests/e2e/online_serving/test_qwen_image_expansion.py
-            tests/e2e/online_serving/test_qwen_image_edit_expansion.py
-            tests/e2e/offline_inference/test_krea2_expansion.py
            -m "full_model and diffusion and H100 and distributed_cuda"
            --run-level "full_model"
        mirror_hardwares: h100_2

-      - label: ":full_moon: Diffusion X2I(&A&T) · Function Test with H100 · Multi-GPU Layered/LongCat/FLUX"
-        timeout_in_minutes: 90
-        commands:
-          - >-
-            pytest -sv
-            tests/e2e/online_serving/test_qwen_image_layered_expansion.py
-            -m "full_model and diffusion and H100 and distributed_cuda"
-            --run-level "full_model"
-        mirror_hardwares: h100_2
-
-      - label: ":full_moon: Diffusion X2I(&A&T) · Function Test with H100 · Multi-GPU Bagel"
-        timeout_in_minutes: 120
-        commands:
-          - >-
-            pytest -sv
-            tests/e2e/online_serving/test_bagel_expansion.py
-            -m "full_model and diffusion and H100 and distributed_cuda"
-            --run-level "full_model"
-        mirror_hardwares: h100_2
-
-      - label: ":full_moon: Diffusion X2I(&A&T) · GLM-Image Function Test with H100"
-        timeout_in_minutes: 120
-        commands:
-          - >-
-            pytest -sv
-            tests/e2e/online_serving/test_glm_image_expansion.py
-            tests/e2e/offline_inference/test_glm_image_autoround_w4a16_expansion.py
-            -m "full_model and diffusion and H100"
-            --run-level "full_model"
-        mirror_hardwares: h100_2
-
      - label: ":full_moon: Diffusion X2I(&A&T) · Function Test with L4"
        timeout_in_minutes: 120
        commands:
-          - pytest -sv tests/e2e/ -k "not test_wan and not test_bagel_expansion and not hunyuan" -m "full_model and diffusion and L4" --run-level "full_model" --ignore=tests/e2e/accuracy
+          - pytest -sv tests/e2e/ -k "not test_wan and not hunyuan" -m "full_model and diffusion and L4" --run-level "full_model" --ignore=tests/e2e/accuracy
        mirror_hardwares: l4_4

      - label: ":full_moon: Diffusion X2I(&A&T) · Doc Test"
@@ -327,7 +243,7 @@ steps:
        timeout_in_minutes: 180
        commands:
          - export VLLM_HTTP_TIMEOUT_KEEP_ALIVE=120
-          - pytest -s -v tests/e2e/accuracy/test_qwen_image*.py --run-level full_model
+          - pytest -s -v tests/e2e/accuracy/test_qwen_image.py --run-level full_model
          - pytest -s -v tests/e2e/accuracy/test_diffusers_backend_similarity.py -k '2i_matches_diffusers' -m full_model --run-level full_model
        mirror_hardwares: h100_1

@@ -355,70 +271,6 @@ steps:
            exit $$EXIT
        mirror_hardwares: h100_4

-      - label: ":full_moon: Diffusion X2I(&A&T) · Perf Test · Qwen-Image-Edit"
-        key: nightly-diffusion-x2iat-performance-qwen-image-edit
-        timeout_in_minutes: 180
-        commands:
-          - export DIFFUSION_BENCHMARK_DIR=tests/dfx/perf/results
-          - export DIFFUSION_ATTENTION_BACKEND=FLASH_ATTN
-          - export CACHE_DIT_VERSION=1.5.0
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_qwen_image_edit_vllm_omni.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/diffusion_result_*.json"
-            buildkite-agent artifact upload "tests/dfx/perf/results/logs/*.log"
-            exit $$EXIT
-        mirror_hardwares: h100_4
-
-      - label: ":full_moon: Diffusion X2I(&A&T) · Perf Test · Qwen-Image-Edit-2511"
-        key: nightly-diffusion-x2iat-performance-qwen-image-edit-2511
-        timeout_in_minutes: 180
-        commands:
-          - export DIFFUSION_BENCHMARK_DIR=tests/dfx/perf/results
-          - export DIFFUSION_ATTENTION_BACKEND=FLASH_ATTN
-          - export CACHE_DIT_VERSION=1.5.0
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_qwen_image_edit_2511_vllm_omni.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/diffusion_result_*.json"
-            buildkite-agent artifact upload "tests/dfx/perf/results/logs/*.log"
-            exit $$EXIT
-        mirror_hardwares: h100_4
-
-      - label: ":full_moon: Diffusion X2I(&A&T) · Perf Test · Qwen-Image-Layered"
-        key: nightly-diffusion-x2iat-performance-qwen-image-layered
-        timeout_in_minutes: 120
-        commands:
-          - export DIFFUSION_BENCHMARK_DIR=tests/dfx/perf/results
-          - export DIFFUSION_ATTENTION_BACKEND=FLASH_ATTN
-          - export CACHE_DIT_VERSION=1.5.0
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_qwen_image_layered_vllm_omni.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/diffusion_result_*.json"
-            buildkite-agent artifact upload "tests/dfx/perf/results/logs/*.log"
-            exit $$EXIT
-        mirror_hardwares: h100_1
-
-      - label: ":full_moon: Diffusion X2I(&A&T) · Perf Test · BAGEL"
-        key: nightly-diffusion-x2iat-performance-bagel
-        timeout_in_minutes: 180
-        commands:
-          - export DIFFUSION_BENCHMARK_DIR=tests/dfx/perf/results
-          - export DIFFUSION_ATTENTION_BACKEND=FLASH_ATTN
-          - export CACHE_DIT_VERSION=1.5.0
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_bagel_vllm_omni.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/diffusion_result_*.json"
-            buildkite-agent artifact upload "tests/dfx/perf/results/logs/*.log"
-            exit $$EXIT
-        mirror_hardwares: h100_1
-
  - group: ":card_index_dividers: Diffusion X2V Model Test"
    key: nightly-diffusion-x2v-group
    depends_on: upload-nightly-pipeline
@@ -440,12 +292,6 @@ steps:
          - pytest -s -v tests/e2e/online_serving/test_wan22_expansion.py -k "i2v" -m "full_model and cuda" --run-level "full_model"
        mirror_hardwares: h100_2

-      - label: ":full_moon: Diffusion X2V · LingBot Video MoE Function Test"
-        timeout_in_minutes: 90
-        commands:
-          - pytest -s -v tests/e2e/online_serving/test_lingbot_video_moe.py -k "video_generation" -m "full_model and cuda" --run-level "full_model"
-        mirror_hardwares: h100_1
-
      - label: ":full_moon: Diffusion X2V · Doc Test"
        timeout_in_minutes: 60
        commands:
@@ -456,15 +302,8 @@ steps:
        timeout_in_minutes: 90
        commands:
          - pytest -s -v tests/e2e/offline_inference/test_wan22_autoround_w4a16_expansion.py -m "full_model and cuda" --run-level "full_model"
-          - pytest -s -v tests/e2e/online_serving/test_wan_2_1_vace_expansion.py tests/e2e/online_serving/test_ltx.py -m "full_model and cuda" --run-level "full_model"
        mirror_hardwares: h100_2

-      - label: ":full_moon: Diffusion X2V · LTX-2.5 Online Test"
-        timeout_in_minutes: 60
-        commands:
-          - pytest -s -v tests/e2e/online_serving/test_ltx25.py -m "full_model and cuda" --run-level "full_model"
-        mirror_hardwares: h100_1
-
      - label: ":full_moon: Diffusion X2V · Wan2.2 Accuracy Test"
        timeout_in_minutes: 180
        commands:
@@ -483,23 +322,8 @@ steps:
        commands:
          - pytest -s -v tests/e2e/accuracy/hunyuanvideo15_t2v/test_hunyuanvideo15_t2v_video_similarity.py -m full_model --run-level full_model
          - pytest -s -v tests/e2e/accuracy/hunyuanvideo15_i2v/test_hunyuanvideo15_i2v_video_similarity.py -m full_model --run-level full_model
-          - pytest -s -v tests/e2e/accuracy/test_video_streaming_output_similarity.py -m full_model --run-level full_model
        mirror_hardwares: h100_2

-      - label: ":full_moon: Diffusion X2V · LTX Accuracy Test"
-        timeout_in_minutes: 180
-        commands:
-          - pytest -s -v 'tests/e2e/accuracy/ltx/test_ltx_official_similarity.py::test_ltx_matches_official[ltx2_3_i2v]' -m full_model --run-level full_model
-          - pytest -s -v 'tests/e2e/accuracy/ltx/test_ltx_official_similarity.py::test_ltx_matches_official[ltx23_two_stage_layer_fused_i2v]' -m full_model --run-level full_model
-        mirror_hardwares: h100_1
-
-      - label: ":full_moon: Diffusion X2V · LTX-2.5 Official Parity Test"
-        timeout_in_minutes: 180
-        commands:
-          - pytest -s -v tests/e2e/accuracy/ltx/test_ltx25_official_similarity.py::test_ltx25_distilled_two_stage_matches_official -m full_model --run-level full_model
-          - pytest -s -v tests/e2e/accuracy/ltx/test_ltx25_official_similarity.py::test_ltx25_full_matches_official -m full_model --run-level full_model
-        mirror_hardwares: h100_1
-
      - label: ":full_moon: Diffusion X2V · Perf Test"
        key: nightly-diffusion-x2v-performance
        timeout_in_minutes: 180
@@ -575,16 +399,6 @@ steps:
          - pytest -sv tests/diffusion/quantization -m 'full_model and cuda and L4' --run-level "full_model"
        mirror_hardwares: l4_1

-  - label: ":bar_chart: Testcase Statistics"
-    key: nightly-testcase-statistics
-    timeout_in_minutes: 120
-    depends_on: upload-nightly-pipeline
-    if: build.env("NIGHTLY") == "1"
-    commands:
-      - python tools/nightly/buildkite_testcase_statistics.py -o tests/dfx/perf/results/buildkite_testcase_statistics.html
-      - buildkite-agent artifact upload "tests/dfx/perf/results/*.html"
-    mirror_hardwares: h100_1
-
  # NOTE: ":email: Nightly Collection & Email" is deprecated and will be removed soon.
  # Do not add new depends_on entries or extend this step.
  - label: ":email: Nightly Collection & Email"
@@ -595,32 +409,20 @@ steps:
      - nightly-omni-performance-vllm-text
      - nightly-omni-performance-multi-replicas
      - nightly-tts-performance
-      - nightly-tts-performance-higgs-audio-v3
      - nightly-diffusion-x2iat-performance-qwen-image
-      - nightly-diffusion-x2iat-performance-qwen-image-edit
-      - nightly-diffusion-x2iat-performance-qwen-image-edit-2511
-      - nightly-diffusion-x2iat-performance-qwen-image-layered
-      - nightly-diffusion-x2iat-performance-bagel
      - nightly-diffusion-x2v-performance
-      - nightly-testcase-statistics
    if: build.env("NIGHTLY") == "1" && build.env("EMAIL_DISTRIBUTION") == "1"
    commands:
      - pip install openpyxl
      - export DEFAULT_INPUT_DIR=tests/dfx/perf/results
      - export DEFAULT_OUTPUT_DIR=tests/dfx/perf/results
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-tts-performance
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-tts-performance-higgs-audio-v3
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-omni-performance-no-async-chunk
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-omni-performance-async-chunk
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-omni-performance-vllm-text
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-omni-performance-multi-replicas
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2iat-performance-qwen-image
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2iat-performance-qwen-image-edit
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2iat-performance-qwen-image-edit-2511
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2iat-performance-qwen-image-layered
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2iat-performance-bagel
      - buildkite-agent artifact download "tests/dfx/perf/results/*.json" . --step nightly-diffusion-x2v-performance
-      - buildkite-agent artifact download "tests/dfx/perf/results/*.html" . --step nightly-testcase-statistics
      - python tools/nightly/generate_nightly_perf_excel.py
      - python tools/nightly/generate_nightly_perf_html.py
      - python tools/nightly/send_nightly_email.py --report-file "tests/dfx/perf/results/*.xlsx, tests/dfx/perf/results/*.html"
@@ -149,4 +149,4 @@ steps:
            pytest -sv
            tests/e2e/ -m "slow and diffusion and H100 and distributed_cuda"
            --run-level "full_model"
-        mirror_hardwares: h100_2
+        mirror_hardwares: h100_4
@@ -96,33 +96,6 @@ steps:
            EXIT=$$?
            buildkite-agent artifact upload "tests/dfx/perf/results/*.json"
            exit $$EXIT
-  - group: ":card_index_dividers: Diffusion X2I(&A&T) Model Test"
-    key: nightly-diffusion-x2iat-group
-    depends_on: upload-nightly-pipeline
-    if: >-
-      build.env("NIGHTLY") == "1" ||
-      build.pull_request.labels includes "nightly-test" ||
-      build.pull_request.labels includes "diffusion-x2iat-test"
-    steps:
-      - label: ":full_moon: Diffusion X2I(&A&T) · Perf Test · Qwen-Image-Edit-2511"
-        key: nightly-diffusion-x2iat-performance-qwen-image-edit-2511
-        timeout_in_minutes: 180
-        mirror_hardwares: a3_npu_8
-        env:
-          VLLM_WORKER_MULTIPROC_METHOD: spawn
-          VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1"
-          HF_TOKEN: "${HF_TOKEN}"
-          DIFFUSION_ATTENTION_BACKEND: TORCH_SDPA
-        commands:
-          - export DIFFUSION_BENCHMARK_DIR=tests/dfx/perf/results
-          - export CACHE_DIT_VERSION=1.5.0
-          - |
-            set +e
-            pytest -s -v tests/dfx/perf/scripts/run_diffusion_benchmark.py -m npu --test-config-file tests/dfx/perf/tests/test_qwen_image_edit_2511_vllm_omni.json
-            EXIT=$$?
-            buildkite-agent artifact upload "tests/dfx/perf/results/diffusion_result_*.json"
-            buildkite-agent artifact upload "tests/dfx/perf/results/logs/*.log"
-            exit $$EXIT
  - group: ":card_index_dividers: Diffusion I2I Model Test"
    key: nightly-HunyuanImage3-x2i-test-group
    depends_on: upload-nightly-pipeline
@@ -66,7 +66,7 @@ repos:
      entry: python tools/pre_commit/check_tts_adapter.py
      language: python
      pass_filenames: false
-      files: ^(vllm_omni/entrypoints/openai/(serving_speech\.py|tts_adapters/__init__\.py)|tools/pre_commit/check_tts_adapter\.py)$
+      files: ^(vllm_omni/entrypoints/openai/serving_speech\.py|tools/pre_commit/check_tts_adapter\.py)$
      # The budget-is-now-slack reminder is printed on a PASSING run, and
      # pre-commit hides output from passing hooks unless the hook is verbose.
      verbose: true
@@ -3,7 +3,26 @@
 Source <https://github.com/vllm-project/vllm-omni/tree/main/examples/offline_inference/x_to_video_audio>.


-The `DreamID-Omni` pipeline generates short videos from text, image and video.
+The shared runner generates video with optional audio from text, image, and audio
+inputs. It can be extended through `vllm_omni.model_extras` without adding
+another model-specific runner.
+
+## Shared request and output contract
+
+The example builds one canonical request envelope: `prompt` is text,
+`modalities` is `["video"]`, and optional `multi_modal_data` contains RGB PIL
+images plus audio `(waveform, sample_rate)` tuples. Audio files are loaded in
+full by design; the runner does not apply a time window. Prepare a trimmed input
+file explicitly when only part of a recording should condition the model.
+DreamID-Omni's audio conditioning sequence grows with the complete input
+duration, so long recordings increase denoising memory and step time.
+
+Model-specific request translation (for example, DreamID-Omni's two negative
+prompts and its 16 kHz input requirement) is declared in `model_extras`. Model
+outputs must provide `fps` and, when audio is present, `audio_sample_rate`.
+`--fps` and `--audio-sample-rate` are available as explicit metadata overrides.
+The shared encoder normalizes supported video tensor layouts, so this runner
+does not branch on a model name or assume a model-specific frame layout.

 ## Local CLI Usage
 ### Download the Model locally
@@ -91,7 +110,8 @@ python x_to_video_audio.py \
 Key arguments:
 - `--prompt`: text description (string).
 - `--model`: path to the model local directory.
- `--height/--width`: output resolution (defaults 704 * 1024).
+- `--model-class-name`: optional diffusion pipeline class override.
+- `--height/--width`: output resolution (defaults 704 * 1280).
 - `--image-path`: path to the input image list.
 - `--audio-path`: path to the input audio list, indicate the timbre of the output video.
 - `--cfg-parallel-size`: number of parallel cfg parallel (defaults 1).
@@ -101,6 +121,9 @@ Key arguments:
 - `--num-inference-steps`: number of denoising steps (defaults 45).
 - `--video-negative-prompt`: negative prompt for video generation.
 - `--audio-negative-prompt`: negative prompt for audio generation.
+- `--extra-body`: JSON object containing parameters declared by the selected model.
+- `--fps`: explicit output FPS override when a model does not return that metadata.
+- `--audio-sample-rate`: explicit output audio sample-rate override.
 - `--enable-cpu-offload`: enable CPU offload (defaults False).

 ## Example materials
@@ -1,6 +1,25 @@
 # X-To-Video-Audio

-The `DreamID-Omni` pipeline generates short videos from text, image and video.
+The shared runner generates video with optional audio from text, image, and audio
+inputs. It currently documents DreamID-Omni and MagiHuman, and can be extended
+through `vllm_omni.model_extras` without adding another model-specific runner.
+
+## Shared request and output contract
+
+The example builds one canonical request envelope: `prompt` is text,
+`modalities` is `["video"]`, and optional `multi_modal_data` contains RGB PIL
+images plus audio `(waveform, sample_rate)` tuples. Audio files are loaded in
+full by design; the runner does not apply a time window. Prepare a trimmed input
+file explicitly when only part of a recording should condition the model.
+DreamID-Omni's audio conditioning sequence grows with the complete input
+duration, so long recordings increase denoising memory and step time.
+
+Model-specific request translation (for example, DreamID-Omni's two negative
+prompts and its 16 kHz input requirement) is declared in `model_extras`. Model
+outputs must provide `fps` and, when audio is present, `audio_sample_rate`.
+`--fps` and `--audio-sample-rate` are available as explicit metadata overrides.
+The shared encoder normalizes supported video tensor layouts, so this runner
+does not branch on a model name or assume a model-specific frame layout.

 ## Local CLI Usage
 ### Download the Model locally
@@ -147,7 +166,8 @@ in `vllm_omni/model_extras/magi_human.py`, routed via `extra_args`):
 Key arguments:
 - `--prompt`: text description (string).
 - `--model`: path to the model local directory.
- `--height/--width`: output resolution (defaults 704 * 1024).
+- `--model-class-name`: optional diffusion pipeline class override.
+- `--height/--width`: output resolution (defaults 704 * 1280).
 - `--image-path`: path to the input image list.
 - `--audio-path`: path to the input audio list, indicate the timbre of the output video.
 - `--cfg-parallel-size`: number of parallel cfg parallel (defaults 1).
@@ -157,6 +177,9 @@ Key arguments:
 - `--num-inference-steps`: number of denoising steps (defaults 45).
 - `--video-negative-prompt`: negative prompt for video generation.
 - `--audio-negative-prompt`: negative prompt for audio generation.
+- `--extra-body`: JSON object containing parameters declared by the selected model.
+- `--fps`: explicit output FPS override when a model does not return that metadata.
+- `--audio-sample-rate`: explicit output audio sample-rate override.
 - `--enable-cpu-offload`: enable CPU offload (defaults False).
 - `--cache-backend`: enable `cache_dit` for acceleration.
 - `--quantization`: online (dynamic) quantization method — `fp8` or `int8`.
@@ -3,32 +3,97 @@

 import argparse
 import json
+import math
 import re
 import time
+from collections.abc import Mapping
+from dataclasses import dataclass
+from pathlib import Path
 from typing import Any

 import numpy as np
+import torch
 from PIL import Image
 from vllm.multimodal.media.audio import load_audio

 from vllm_omni.diffusion.data import DiffusionParallelConfig
-from vllm_omni.diffusion.utils.media_utils import mux_video_audio_bytes
+from vllm_omni.diffusion.utils.param_utils import apply_declared_extra_args
 from vllm_omni.entrypoints.omni import Omni
+from vllm_omni.entrypoints.openai.video_api_utils import encode_video_bytes
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
+from vllm_omni.model_extras import (
+    build_x_to_video_audio_prompt as build_model_x_to_video_audio_prompt,
+)
+from vllm_omni.model_extras import (
+    get_extra_body_params,
+    get_input_audio_sample_rate,
+    get_model_class_name,
+    get_output_tensor_range,
+)

 IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".bmp", ".webp", ".gif"}
 AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".m4a", ".aac", ".ogg"}


+@dataclass(frozen=True)
+class XToVideoAudioOutput:
+    """Canonical generated video+audio payload and its required metadata."""
+
+    video: Any
+    audio: Any | None
+    fps: float
+    audio_sample_rate: int | None
+    output_tensor_range: str = "negative_one_to_one"
+
+
+def parse_json_object(value: str) -> dict[str, Any]:
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise argparse.ArgumentTypeError(f"--extra-body must be valid JSON: {exc}") from exc
+    if not isinstance(parsed, dict):
+        raise argparse.ArgumentTypeError("--extra-body must be a JSON object")
+    return parsed
+
+
+def build_x_to_video_audio_prompt(
+    prompt: str,
+    media_inputs: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build the canonical request envelope for image/audio-to-video models."""
+    result: dict[str, Any] = {"prompt": prompt, "modalities": ["video"]}
+    if media_inputs:
+        result["multi_modal_data"] = media_inputs
+    return result
+
+
+def _clean_official_prompt(prompt: str) -> str:
+    """Remove metadata tags used by the official DreamID prompt fixtures."""
+    prompt = re.sub(
+        r"\[SPEAKER_TIMESTAMPS_START\].*?\[SPEAKER_TIMESTAMPS_END\]",
+        "",
+        prompt,
+        flags=re.DOTALL,
+    ).strip()
+    prompt = re.sub(
+        r"\[AUDIO_DESCRIPTION_START].*?\[AUDIO_DESCRIPTION_END]",
+        "",
+        prompt,
+        flags=re.DOTALL,
+    ).strip()
+    prompt = re.sub(r"\[[A-Z_]+\]", "", prompt)
+    return re.sub(r"\n\s*\n", "\n", prompt).strip()
+
+
 def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Offline inference for X -> video+audio models.")
    parser.add_argument("--model", required=True, help="Model ckpt root directory.")
    parser.add_argument(
        "--model-type",
-        default="dreamid-omni",
-        choices=["dreamid-omni", "magi-human"],
-        help="Model type.",
+        default=None,
+        help="Optional legacy model type override (for example, magi-human).",
    )
+    parser.add_argument("--model-class-name", default=None, help="Optional diffusion pipeline class override.")
    parser.add_argument("--prompt", default=None, help="Text prompt.")

    parser.add_argument("--image-path", type=str, nargs="+", help="list of image-path")
@@ -36,9 +101,9 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument("--prompt-file", type=str, default=None, help="Text prompt in json format.")
    parser.add_argument(
        "--extra-body",
-        type=str,
+        type=parse_json_object,
        default=None,
-        help="[magi-human] JSON dict of model-specific extra params (declared in vllm_omni/model_extras/), "
+        help="JSON dict of model-specific extra params (declared in vllm_omni/model_extras/), "
        'merged into sampling extra_args. Example: \'{"image_path": "/path/to/img.jpg", "seconds": 5}\'.',
    )

@@ -71,13 +136,20 @@ def parse_args() -> argparse.Namespace:
        default="robotic, muffled, echo, distorted",
        help="Negative prompt for audio.",
    )
+    parser.add_argument("--fps", type=float, default=None, help="Override output FPS metadata.")
+    parser.add_argument(
+        "--audio-sample-rate",
+        type=int,
+        default=None,
+        help="Override output audio sample-rate metadata.",
+    )
    parser.add_argument("--output", default="dreamid_output.mp4", help="Output video path.")
    parser.add_argument(
        "--quantization",
        type=str,
        default=None,
        choices=["fp8", "int8"],
-        help=("Online (dynamic) quantization method for the DreamID-Omni transformer. "),
+        help="Online (dynamic) quantization method for the model transformer.",
    )
    parser.add_argument(
        "--enable-cpu-offload",
@@ -94,7 +166,7 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument(
        "--use-hsdp",
        action="store_true",
-        help="Enable HSDP for DreamID-Omni fused transformer blocks.",
+        help="Enable HSDP for supported transformer blocks.",
    )
    parser.add_argument(
        "--hsdp-shard-size",
@@ -111,20 +183,105 @@ def parse_args() -> argparse.Namespace:
    return parser.parse_args()


-def load_image_and_audio(image_paths, audio_paths):
-    image = []
-    audio = []
+def load_image_and_audio(
+    image_paths: list[str] | None,
+    audio_paths: list[str] | None,
+    *,
+    audio_sample_rate: int | None = None,
+) -> tuple[list[Image.Image], list[tuple[Any, int]]]:
+    """Load complete media inputs into the canonical multimodal representation."""
+    images = []
+    audios = []

-    for path in image_paths:
+    for path in image_paths or []:
        with Image.open(path) as img:
-            img = img.convert("RGB")
-            image.append(img)
+            images.append(img.convert("RGB"))

-    for path in audio_paths:
-        audio_array, sr = load_audio(path, sr=16000)
-        audio_array = audio_array[int(sr * 1) : int(sr * 3)]
-        audio.append(audio_array)
-    return image, audio
+    for path in audio_paths or []:
+        waveform, sample_rate = load_audio(path, sr=audio_sample_rate)
+        audios.append((waveform, sample_rate))
+    return images, audios
+
+
+def extract_x_to_video_audio_output(
+    outputs: Any,
+    *,
+    fps: float | None = None,
+    audio_sample_rate: int | None = None,
+    output_tensor_range: str = "negative_one_to_one",
+) -> XToVideoAudioOutput:
+    """Normalize engine output without depending on a model's tensor layout."""
+    if isinstance(outputs, list):
+        if not outputs:
+            raise RuntimeError("No output returned from the model.")
+        result = outputs[0]
+    else:
+        result = outputs
+    if result is None:
+        raise RuntimeError("No output returned from the model.")
+
+    multimodal_output = getattr(result, "multimodal_output", None) or {}
+    if not isinstance(multimodal_output, Mapping):
+        raise RuntimeError("Model multimodal output must be a mapping.")
+    images = getattr(result, "images", None)
+    video = images[0] if images else multimodal_output.get("video")
+    if video is None:
+        raise RuntimeError("No video payload found in model output.")
+
+    resolved_fps = fps if fps is not None else multimodal_output.get("fps")
+    try:
+        resolved_fps = float(resolved_fps)
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError("Model output must declare a valid 'fps', or pass --fps.") from exc
+    if not math.isfinite(resolved_fps) or resolved_fps <= 0:
+        raise RuntimeError("Output FPS must be finite and positive.")
+
+    audio = multimodal_output.get("audio")
+    resolved_sample_rate = audio_sample_rate
+    if resolved_sample_rate is None:
+        resolved_sample_rate = multimodal_output.get("audio_sample_rate")
+    if audio is not None:
+        try:
+            resolved_sample_rate = int(resolved_sample_rate)
+        except (TypeError, ValueError) as exc:
+            raise RuntimeError(
+                "Model output with audio must declare a valid 'audio_sample_rate', or pass --audio-sample-rate."
+            ) from exc
+        if resolved_sample_rate <= 0:
+            raise RuntimeError("Output audio sample rate must be positive.")
+    elif resolved_sample_rate is not None:
+        resolved_sample_rate = int(resolved_sample_rate)
+
+    return XToVideoAudioOutput(video, audio, resolved_fps, resolved_sample_rate, output_tensor_range)
+
+
+def _normalize_output_tensor_range(video: Any, source_range: str) -> Any:
+    """Normalize floating-point output according to the pipeline contract."""
+    if isinstance(video, torch.Tensor):
+        if not video.is_floating_point():
+            return video
+        video = video.detach().cpu().float().numpy()
+    if isinstance(video, np.ndarray):
+        if not np.issubdtype(video.dtype, np.floating):
+            return video
+        if source_range == "negative_one_to_one":
+            return np.clip(video, -1.0, 1.0) * 0.5 + 0.5
+        if source_range == "zero_to_one":
+            return np.clip(video, 0.0, 1.0)
+        raise ValueError(f"Unsupported floating-point tensor range: {source_range!r}")
+    if isinstance(video, list):
+        return [_normalize_output_tensor_range(frame, source_range) for frame in video]
+    return video
+
+
+def encode_x_to_video_audio_output(output: XToVideoAudioOutput) -> bytes:
+    """Encode any video layout supported by the shared video API utility."""
+    return encode_video_bytes(
+        _normalize_output_tensor_range(output.video, output.output_tensor_range),
+        output.fps,
+        output.audio,
+        output.audio_sample_rate,
+    )


 def _extract_peak_memory_mb(result: Any) -> float:
@@ -157,45 +314,10 @@ def main() -> None:
    if args.prompt_file:
        with open(args.prompt_file) as f:
            text_prompt = json.load(f)
-            text_prompt = re.sub(
-                r"\[SPEAKER_TIMESTAMPS_START\].*?\[SPEAKER_TIMESTAMPS_END\]", "", text_prompt, flags=re.DOTALL
-            ).strip()
-            text_prompt = re.sub(
-                r"\[AUDIO_DESCRIPTION_START].*?\[AUDIO_DESCRIPTION_END]", "", text_prompt, flags=re.DOTALL
-            ).strip()
-            text_prompt = re.sub(r"\[[A-Z_]+\]", "", text_prompt)
-            text_prompt = re.sub(r"\n\s*\n", "\n", text_prompt).strip()
-
-    if args.model_type == "magi-human":
-        prompt = text_prompt
-        extra_args = json.loads(args.extra_body) if args.extra_body else {}
-        sampling_params = OmniDiffusionSamplingParams(
-            height=args.height,
-            width=args.width,
-            num_inference_steps=args.num_inference_steps,
-            seed=args.seed,
-            extra_args=extra_args,
-        )
-    else:
-        image, audio = load_image_and_audio(args.image_path, args.audio_path)
-
-        prompt = {
-            "prompt": text_prompt,
-            "video_negative_prompt": args.video_negative_prompt,
-            "audio_negative_prompt": args.audio_negative_prompt,
-            "multi_modal_data": {"image": image, "audio": audio},
-        }
-
-        sampling_params = OmniDiffusionSamplingParams(
-            height=args.height,
-            width=args.width,
-            num_inference_steps=args.num_inference_steps,
-            seed=args.seed,
-            extra_args={
-                "solver_name": args.solver_name,
-                "shift": args.shift,
-            },
-        )
+            if isinstance(text_prompt, str):
+                text_prompt = _clean_official_prompt(text_prompt)
+    if not isinstance(text_prompt, str):
+        raise ValueError("Prompt content must be a JSON string or --prompt text.")

    parallel_config = DiffusionParallelConfig(
        cfg_parallel_size=args.cfg_parallel_size,
@@ -220,64 +342,83 @@ def main() -> None:
            "scm_steps_policy": "dynamic",
        }

-    omni_kwargs = dict(
+    omni_kwargs: dict[str, Any] = dict(
        model=args.model,
        parallel_config=parallel_config,
-        model_type=args.model_type,
        enable_cpu_offload=args.enable_cpu_offload,
        enable_layerwise_offload=args.enable_layerwise_offload,
        cache_backend=args.cache_backend,
        cache_config=cache_config,
    )
+    if args.model_type is not None:
+        omni_kwargs["model_type"] = args.model_type
+    if args.model_class_name is not None:
+        omni_kwargs["model_class_name"] = args.model_class_name
    if args.quantization is not None:
        omni_kwargs["quantization"] = args.quantization
    omni = Omni(**omni_kwargs)
-    start = time.perf_counter()
-    outputs = omni.generate(prompt, sampling_params)
-    elapsed = time.perf_counter() - start
+    try:
+        model_class_name = args.model_class_name or get_model_class_name(omni)
+        input_sample_rate = get_input_audio_sample_rate(model_class_name)
+        images, audios = load_image_and_audio(
+            args.image_path,
+            args.audio_path,
+            audio_sample_rate=input_sample_rate,
+        )
+        media_inputs: dict[str, Any] = {}
+        if images:
+            media_inputs["image"] = images
+        if audios:
+            media_inputs["audio"] = audios

-    if not outputs:
-        raise RuntimeError("No output returned from the model.")
-    result = outputs[0]
-    if not result.images:
-        raise RuntimeError("No video frames found in model output.")
-    generated_video = result.images[0]
-    mm = result.multimodal_output or {}
-    generated_audio = mm.get("audio")
+        canonical_prompt = build_x_to_video_audio_prompt(text_prompt, media_inputs)
+        prompt = build_model_x_to_video_audio_prompt(
+            model_class_name,
+            canonical_prompt,
+            {
+                "video_negative_prompt": args.video_negative_prompt,
+                "audio_negative_prompt": args.audio_negative_prompt,
+            },
+        )

-    if args.model_type == "magi-human":
-        # MagiHuman returns frames already as (F, H, W, C) uint8, ready to mux.
-        fps = float(mm.get("fps", 25))
-        sample_rate = int(mm.get("audio_sample_rate", 24000))
-        frames = generated_video
-    else:
-        fps = float(mm.get("fps", 24))
-        sample_rate = int(mm.get("audio_sample_rate", 16000))
-        # DreamID-Omni returns video as (C, F, H, W) float32 in [-1, 1].
-        # mux_video_audio_bytes expects (F, H, W, C) uint8.
-        if not isinstance(generated_video, np.ndarray) or generated_video.ndim != 4:
-            raise RuntimeError(f"Unexpected video shape: {getattr(generated_video, 'shape', None)}")
-        frames = generated_video.transpose(1, 2, 3, 0)
-        frames = (np.clip((frames + 1.0) / 2.0, 0.0, 1.0) * 255.0).round().astype(np.uint8)
+        sampling_params = OmniDiffusionSamplingParams(
+            height=args.height,
+            width=args.width,
+            num_inference_steps=args.num_inference_steps,
+            seed=args.seed,
+            extra_args={},
+        )
+        declared_extra_body_params = get_extra_body_params(model_class_name)
+        extra_body = dict(args.extra_body or {})
+        legacy_extra_body = {"solver_name": args.solver_name, "shift": args.shift}
+        for key, value in legacy_extra_body.items():
+            if key in declared_extra_body_params:
+                extra_body.setdefault(key, value)
+        if declared_extra_body_params:
+            apply_declared_extra_args(sampling_params, declared_extra_body_params, extra_body)
+        elif extra_body:
+            sampling_params.extra_args.update({key: value for key, value in extra_body.items() if value is not None})

-    audio_np = None
-    if generated_audio is not None:
-        audio_np = np.squeeze(np.asarray(generated_audio)).astype(np.float32)
+        start = time.perf_counter()
+        outputs = omni.generate(prompt, sampling_params)
+        elapsed = time.perf_counter() - start
+        output = extract_x_to_video_audio_output(
+            outputs,
+            fps=args.fps,
+            audio_sample_rate=args.audio_sample_rate,
+            output_tensor_range=get_output_tensor_range(model_class_name),
+        )

-    output_path = args.output
-    video_bytes = mux_video_audio_bytes(
-        frames,
-        audio_np,
-        fps=float(fps),
-        audio_sample_rate=sample_rate,
-    )
-    with open(output_path, "wb") as f:
-        f.write(video_bytes)
-    print(f"Saved generated video to {output_path}")
-    print(f"Total time: {elapsed:.2f}s")
-    peak_mb = _extract_peak_memory_mb(outputs)
-    if peak_mb:
-        print(f"Worker peak GPU memory (reserved): {peak_mb:.2f} MiB ({peak_mb / 1024:.2f} GiB)")
+        output_path = Path(args.output)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_bytes(encode_x_to_video_audio_output(output))
+        print(f"Saved generated video to {output_path}")
+        print(f"Total time: {elapsed:.2f}s")
+        peak_mb = _extract_peak_memory_mb(outputs)
+        if peak_mb:
+            print(f"Worker peak GPU memory (reserved): {peak_mb:.2f} MiB ({peak_mb / 1024:.2f} GiB)")
+    finally:
+        omni.close()


 if __name__ == "__main__":
@@ -4,7 +4,10 @@ from types import SimpleNamespace
 import pytest
 from vllm.v1.engine import FinishReason

-from vllm_omni.core.sched.omni_scheduler_mixin import OmniSchedulerMixin
+from vllm_omni.core.sched.omni_scheduler_mixin import (
+    DEFAULT_INPUT_WAIT_TIMEOUT_S,
+    OmniSchedulerMixin,
+)
 from vllm_omni.core.sched.output import OmniChunkRecvHandle

 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
@@ -22,13 +25,25 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths():
    scheduler.requests = {"request": object()}
    scheduler._consume_pending_connector_output = lambda mode: calls.append(("consume", mode))
    scheduler._process_pending_input_timeouts = lambda: calls.append(("timeouts",))
+
+    def _collect_timed_out(timeout_s):
+        calls.append(("chunk-timeouts", timeout_s))
+        return set()
+
+    def _collect_failed_sends():
+        calls.append(("failed-sends",))
+        return {}
+
    scheduler.chunk_transfer_adapter = SimpleNamespace(
+        receives_chunks=True,
        process_pending_chunks=lambda waiting, running, scheduler_requests: calls.append(
            ("process", waiting, running, scheduler_requests)
        ),
        restore_queues=lambda waiting, running, scheduler_requests: calls.append(
            ("restore-chunks", waiting, running, scheduler_requests)
        ),
+        collect_timed_out_request_ids=_collect_timed_out,
+        collect_failed_send_request_ids=_collect_failed_sends,
    )
    scheduler.input_coordinator = SimpleNamespace(
        restore_queues=lambda waiting: calls.append(("restore-full", waiting))
@@ -41,6 +56,10 @@ def test_schedule_lifecycle_helpers_process_and_restore_both_input_paths():
        ("consume", "ar"),
        ("timeouts",),
        ("process", scheduler.waiting, scheduler.running, scheduler.requests),
+        # The chunk deadline runs after chunks are applied, so a chunk that
+        # arrived this cycle resets the clock before it is measured (R1.1).
+        ("chunk-timeouts", DEFAULT_INPUT_WAIT_TIMEOUT_S),
+        ("failed-sends",),
        ("restore-chunks", scheduler.waiting, scheduler.running, scheduler.requests),
        ("restore-full", scheduler.waiting),
    ]
@@ -1,6 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Unit coverage for _process_pending_input_timeouts.
+"""Unit coverage for the two stage-input timeout hooks.
+
+``_process_pending_input_timeouts`` covers the full-payload transport;
+``_process_pending_chunk_timeouts`` covers async-chunk. Both share
+``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``.

 Verifies that the mixin correctly *delegates* timed-out requests to the
 base scheduler's ``finish_requests`` API with ``RequestStatus.FINISHED_ERROR``.
@@ -12,6 +16,10 @@ this file only asserts the wiring from the mixin to that API.

 from __future__ import annotations

+import importlib
+import logging
+import math
+import os
 from types import SimpleNamespace

 import pytest
@@ -102,5 +110,248 @@ def test_process_pending_input_timeouts_disabled_when_timeout_zero(monkeypatch):
    assert scheduler.finish_calls == []


+class _FakeChunkAdapter:
+    """Stands in for OmniChunkTransferAdapter's timeout-collection surface."""
+
+    def __init__(self, timed_out_ids, *, receives_chunks=True):
+        self._timed_out_ids = set(timed_out_ids)
+        self.receives_chunks = receives_chunks
+        self.calls = []
+
+    def collect_timed_out_request_ids(self, timeout_s):
+        self.calls.append(timeout_s)
+        return set(self._timed_out_ids)
+
+
+class _FakeChunkScheduler(OmniSchedulerMixin):
+    def __init__(self, requests, adapter):
+        self.requests = requests
+        self.chunk_transfer_adapter = adapter
+        self.finish_calls = []
+
+    def finish_requests(self, req_ids, status):
+        self.finish_calls.append((set(req_ids), status))
+
+
+def test_process_pending_chunk_timeouts_delegates_to_finish_requests():
+    """A request stalled in WAITING_FOR_CHUNK is failed, not left parked.
+
+    This is the async-chunk half of the safety net (RFC #4855 R1.1); before it
+    existed, the chunk path had no deadline at all and issue #3833 could park a
+    request forever.
+    """
+    req_id = "stalled-stream"
+    adapter = _FakeChunkAdapter(timed_out_ids={req_id})
+    scheduler = _FakeChunkScheduler({req_id: SimpleNamespace(request_id=req_id)}, adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert len(adapter.calls) == 1
+    assert adapter.calls[0] > 0
+    assert len(scheduler.finish_calls) == 1
+    finished_ids, status = scheduler.finish_calls[0]
+    assert finished_ids == {req_id}
+    assert getattr(status, "name", str(status)).endswith("FINISHED_ERROR")
+
+
+def test_process_pending_chunk_timeouts_skips_already_freed_request():
+    adapter = _FakeChunkAdapter(timed_out_ids={"already-freed"})
+    scheduler = _FakeChunkScheduler(requests={}, adapter=adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_noop_on_a_producer_stage():
+    """Stage 0 does not receive chunks, so it has nothing to expire."""
+    adapter = _FakeChunkAdapter(timed_out_ids={"r1"}, receives_chunks=False)
+    scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter)
+
+    scheduler._process_pending_chunk_timeouts()
+
+    assert adapter.calls == []
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_noop_without_adapter():
+    class _NoAdapter(OmniSchedulerMixin):
+        def __init__(self):
+            self.requests = {}
+            self.chunk_transfer_adapter = None
+            self.finish_calls = []
+
+        def finish_requests(self, req_ids, status):
+            self.finish_calls.append((set(req_ids), status))
+
+    scheduler = _NoAdapter()
+    scheduler._process_pending_chunk_timeouts()
+    assert scheduler.finish_calls == []
+
+
+def test_process_pending_chunk_timeouts_disabled_when_timeout_zero(monkeypatch):
+    """The shared knob disables both transports' nets together."""
+    from vllm_omni.core.sched import omni_scheduler_mixin
+
+    monkeypatch.setattr(omni_scheduler_mixin, "DEFAULT_INPUT_WAIT_TIMEOUT_S", 0.0)
+
+    adapter = _FakeChunkAdapter(timed_out_ids={"r1"})
+    scheduler = _FakeChunkScheduler({"r1": SimpleNamespace(request_id="r1")}, adapter)
+    scheduler._process_pending_chunk_timeouts()
+
+    assert adapter.calls == []
+    assert scheduler.finish_calls == []
+
+
+class _FakeSendFailureAdapter:
+    """Stands in for the adapter's drained send-failure map (R1.2)."""
+
+    def __init__(self, failures):
+        self._failures = dict(failures)
+        self.collect_calls = 0
+
+    def collect_failed_send_request_ids(self):
+        self.collect_calls += 1
+        failures, self._failures = self._failures, {}
+        return failures
+
+
+class _FakeSendFailureScheduler(OmniSchedulerMixin):
+    def __init__(self, requests, adapter):
+        self.requests = requests
+        self.chunk_transfer_adapter = adapter
+
+
+def test_log_failed_chunk_sends_reports_a_live_request(caplog):
+    adapter = _FakeSendFailureAdapter({"req-live": "OSError: No space left on device"})
+    scheduler = _FakeSendFailureScheduler({"req-live": SimpleNamespace(request_id="req-live")}, adapter)
+
+    with caplog.at_level(logging.ERROR):
+        scheduler._log_failed_chunk_sends()
+
+    assert "req-live" in caplog.text
+    assert "No space left on device" in caplog.text
+
+
+def test_log_failed_chunk_sends_still_reports_a_departed_request(caplog):
+    """``collect_*`` drains the map, so a failure for a request that already left
+    this scheduler is the one place a drop could still go unrecorded."""
+    adapter = _FakeSendFailureAdapter({"req-gone": "OSError: No space left on device"})
+    scheduler = _FakeSendFailureScheduler({}, adapter)
+
+    with caplog.at_level(logging.WARNING):
+        scheduler._log_failed_chunk_sends()
+
+    assert adapter.collect_calls == 1
+    assert "req-gone" in caplog.text
+    assert "No space left on device" in caplog.text
+    assert "finished or aborted" in caplog.text
+
+
+def test_log_failed_chunk_sends_noop_without_adapter():
+    scheduler = _FakeSendFailureScheduler({}, None)
+    scheduler._log_failed_chunk_sends()
+
+
+# --- R1.4: VLLM_OMNI_INPUT_WAIT_TIMEOUT_S is parsed at import time, so each of
+# these reloads the module under a different environment. One knob now gates the
+# deadline on both transports, which is why a silent 0 matters more than it did
+# when only the full-payload path used it.
+
+
+@pytest.fixture
+def reload_with(monkeypatch):
+    """Reload the mixin module under a given env value, then put it back.
+
+    ``importlib.reload`` re-executes into the live module dict, so whatever the
+    last test parsed is what every later test in the session sees -- ``-k
+    zero_warns`` alone would leave ``DEFAULT_INPUT_WAIT_TIMEOUT_S = 0.0`` and
+    silently disable both deadlines. The rejection cases are worse: the
+    ValueError fires *after* the rebind, so ``nan`` lingers in the module.
+    ``monkeypatch`` restores the environment but not the module, so reload once
+    more on the way out.
+    """
+    import vllm_omni.core.sched.omni_scheduler_mixin as mod
+
+    def _reload(raw):
+        if raw is None:
+            monkeypatch.delenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raising=False)
+        else:
+            monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", raw)
+        return importlib.reload(mod)
+
+    yield _reload
+
+    monkeypatch.undo()
+    importlib.reload(mod)
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        (None, 600.0),
+        ("30", 30.0),
+        ("0.5", 0.5),
+        ("0", 0.0),  # explicit opt-out is still honoured
+    ],
+)
+def test_timeout_accepts_valid_values(reload_with, raw, expected):
+    mod = reload_with(raw)
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == expected
+
+
+@pytest.mark.parametrize("raw", ["-1", "-600", "-0.5"])
+def test_negative_timeout_is_rejected(reload_with, raw):
+    """`-1` is a common "no limit" idiom; substituting 600 would give an operator
+    who meant "never time out" mysterious failures ten minutes in. Fail loudly."""
+    with pytest.raises(ValueError, match="negative"):
+        reload_with(raw)
+
+
+@pytest.mark.parametrize("raw", ["nan", "inf", "-inf", "1e999"])
+def test_non_finite_timeout_is_rejected(reload_with, raw):
+    """nan compares false against every deadline check and inf is unreachable;
+    both disable the net while looking like a number."""
+    with pytest.raises(ValueError, match="not finite"):
+        reload_with(raw)
+
+
+def test_zero_warns_and_names_both_transports(reload_with, caplog):
+    with caplog.at_level(logging.WARNING):
+        mod = reload_with("0")
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 0.0
+    text = caplog.text
+    assert "DISABLED" in text
+    assert "WAITING_FOR_INPUT" in text and "WAITING_FOR_CHUNK" in text
+
+
+def test_unparsable_timeout_falls_back(reload_with):
+    mod = reload_with("not-a-number")
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == 600.0
+
+
+def test_a_rejected_value_does_not_outlive_its_test(monkeypatch):
+    """The reason ``reload_with`` reloads on the way out, done here by hand.
+
+    ``nan`` is bound before the ValueError fires, so the reload that rejects it
+    still leaves it in the module -- and nan compares false against every
+    deadline check, which is exactly the silent-disable this item exists to
+    stop. Restoring the environment is not enough; the module has to be reloaded
+    too.
+    """
+    import vllm_omni.core.sched.omni_scheduler_mixin as mod
+
+    ambient = float(os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600"))
+
+    monkeypatch.setenv("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "nan")
+    with pytest.raises(ValueError, match="not finite"):
+        importlib.reload(mod)
+    assert math.isnan(mod.DEFAULT_INPUT_WAIT_TIMEOUT_S)
+
+    monkeypatch.undo()
+    importlib.reload(mod)
+    assert mod.DEFAULT_INPUT_WAIT_TIMEOUT_S == ambient
+
+
 if __name__ == "__main__":
    pytest.main([__file__, "-v"])
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-stage BAGEL (TP=1, CFG=1), t2i Task",
@@ -61,7 +61,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-stage BAGEL (TP=1, CFG=1), i2i Task",
@@ -113,7 +113,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Two-stage BAGEL (bagel.yaml), TP=1, CFG=1, t2i Task",
@@ -164,7 +164,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Two-stage BAGEL (bagel.yaml), TP=1, CFG=1, i2i Task",
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "tts"
    ],
    "server_params": {
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-device baseline with enforce-eager (no torch.compile)",
@@ -64,7 +64,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-device with torch.compile (default, no enforce-eager)",
@@ -117,7 +117,7 @@
          "num_cards": 2
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "CFG-parallel=2 with enforce-eager",
@@ -172,7 +172,7 @@
          "num_cards": 2
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "CFG-parallel=2 with torch.compile",
@@ -226,7 +226,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "CacheDiT with enforce-eager",
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-device baseline (two input images)",
@@ -76,7 +76,7 @@
          "num_cards": 4
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Ulysses SP=2 + CFG=2 + VAE patch parallel=4",
@@ -125,7 +125,7 @@
          "num_cards": 4
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Ulysses SP=2 + CFG=2 + CacheDiT",
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-device baseline",
@@ -73,7 +73,7 @@
          "num_cards": 4
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Ulysses SP=2 + CFG=2 + VAE patch parallel=4",
@@ -121,7 +121,7 @@
          "num_cards": 4
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Ulysses SP=2 + CFG=2 + CacheDiT",
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "diffusion"
    ],
    "description": "Single-device baseline",
@@ -10,7 +10,7 @@
          "num_cards": 1
        }
      },
-      "full_model",
+      "local_model",
      "tts"
    ],
    "server_params": {
@@ -1728,3 +1728,112 @@ def test_purge_is_noop_on_empty_deques(build_adapter):
    adapter.restore_queues(waiting_queue, running_queue, scheduler_requests={})
    assert running_queue == []
    assert waiting_queue == []
+
+
+# --------------------------------------------------------------------------- #
+#  Chunk-wait deadline (RFC #4855 R1.1, issue #3833)
+#
+#  Before this, the async-chunk path had no deadline of any kind, so a dropped
+#  terminal chunk or an upstream stage that died mid-stream parked the request
+#  in WAITING_FOR_CHUNK forever. None of these scenarios were covered.
+# --------------------------------------------------------------------------- #
+
+
+def _park_in_chunk_wait(adapter, request, *, waiting=True):
+    """Drive a request through one process_pending_chunks round into the wait."""
+    queue = DummyWaitingQueue([request]) if waiting else [request]
+    if waiting:
+        adapter.process_pending_chunks(queue, [])
+    else:
+        adapter.process_pending_chunks(DummyWaitingQueue(), queue)
+    return queue
+
+
+def test_chunk_wait_clock_starts_when_a_request_parks(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+
+    _park_in_chunk_wait(adapter, request)
+
+    assert request.status == RequestStatus.WAITING_FOR_CHUNK
+    assert "r1" in adapter._waiting_since
+
+
+def test_chunk_wait_does_not_expire_before_the_deadline(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+    assert "r1" in adapter._waiting_since
+
+
+def test_dropped_terminal_chunk_expires_the_request(build_adapter):
+    """The #3833 scenario: upstream stops sending and never marks the stream done."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    adapter._waiting_since["r1"] -= 601.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"r1"}
+    # Cleared, so a second sweep does not re-report the same request.
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+
+
+def test_arriving_chunk_resets_the_clock(build_adapter):
+    """A slow but healthy stream must never expire: the deadline measures stall
+    time between chunks, not the lifetime of the stream."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+    queue = _park_in_chunk_wait(adapter, request)
+    adapter.restore_queues(queue, [], scheduler_requests={"r1": request})
+
+    # Age the wait to just under the deadline, then deliver a chunk.
+    adapter._waiting_since["r1"] -= 599.0
+    adapter._finished_load_reqs.add("r1")
+    adapter.process_pending_chunks(queue, [])
+
+    assert "r1" not in adapter._waiting_since
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == set()
+
+
+def test_a_disabled_deadline_never_expires(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+    adapter._waiting_since["r1"] -= 10_000.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=0.0) == set()
+    assert adapter.collect_timed_out_request_ids(timeout_s=-1.0) == set()
+
+
+def test_finished_request_leaves_no_stale_timestamp(build_adapter):
+    """Otherwise a completed request would be reported as timed out later."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    request = _req("r1", RequestStatus.WAITING)
+    _park_in_chunk_wait(adapter, request)
+
+    adapter.finish_requests(["r1"], RequestStatus.FINISHED_STOPPED, {"r1": request})
+
+    assert "r1" not in adapter._waiting_since
+    assert adapter.collect_timed_out_request_ids(timeout_s=0.001) == set()
+
+
+def test_aborted_request_leaves_no_stale_timestamp(build_adapter):
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar")
+    _park_in_chunk_wait(adapter, _req("r1", RequestStatus.WAITING))
+
+    adapter.cleanup_receiver("r1")
+
+    assert "r1" not in adapter._waiting_since
+
+
+def test_expiry_is_per_request(build_adapter):
+    """One stalled stream must not take down its healthy neighbours."""
+    adapter, _ = build_adapter(stage_id=1, model_mode="ar", max_num_seqs=4)
+    stalled = _req("stalled", RequestStatus.WAITING)
+    healthy = _req("healthy", RequestStatus.WAITING)
+    adapter.process_pending_chunks(DummyWaitingQueue([stalled, healthy]), [])
+
+    adapter._waiting_since["stalled"] -= 601.0
+
+    assert adapter.collect_timed_out_request_ids(timeout_s=600.0) == {"stalled"}
+    assert "healthy" in adapter._waiting_since
@@ -463,7 +463,7 @@ def _audio_metrics(reference: np.ndarray, prediction: np.ndarray) -> dict[str, f
    }


-@pytest.mark.full_model
+@pytest.mark.slow
@pytest.mark.benchmark
@pytest.mark.diffusion
@hardware_test(res={"cuda": "H100"}, num_cards=1)
@@ -591,7 +591,7 @@ def test_ltx25_distilled_two_stage_matches_official(accuracy_artifact_root: Path
    assert audio_metrics["relative_l2"] <= LTX25_AUDIO_RELATIVE_L2_THRESHOLD


-@pytest.mark.full_model
+@pytest.mark.slow
@pytest.mark.benchmark
@pytest.mark.diffusion
@hardware_test(res={"cuda": "H100"}, num_cards=1)
@@ -544,7 +544,7 @@ def _audio_metrics(reference: np.ndarray, prediction: np.ndarray) -> dict[str, f


@pytest.mark.parametrize("case", CASES, ids=lambda case: case.name)
-@pytest.mark.full_model
+@pytest.mark.slow
@pytest.mark.benchmark
@pytest.mark.diffusion
@hardware_test(res={"cuda": "H100"}, num_cards=1)
@@ -17,7 +17,7 @@ from tests.helpers.mark import hardware_test
 from tests.helpers.media import get_asset_path
 from tests.helpers.runtime import OmniServer

-pytestmark = [pytest.mark.full_model, pytest.mark.diffusion]
+pytestmark = [pytest.mark.diffusion]


 SINGLE_MODEL = "Qwen/Qwen-Image-Edit"
@@ -253,6 +253,7 @@ def _diffusers_output_single_image(accuracy_artifact_root: Path, qwen_bear_image
    )


+@pytest.mark.slow
@pytest.mark.benchmark
@hardware_test(res={"cuda": "H100"}, num_cards=1)
 def test_qwen_image_edit_single_matches_diffusers(
@@ -278,6 +279,7 @@ def test_qwen_image_edit_single_matches_diffusers(
    )


+@pytest.mark.slow
@pytest.mark.benchmark
@hardware_test(res={"cuda": "H100"}, num_cards=1)
 def test_qwen_image_edit_2511_matches_diffusers_pixelwise(accuracy_artifact_root: Path) -> None:
@@ -17,7 +17,7 @@ from tests.helpers.env import run_post_test_cleanup, run_pre_test_cleanup
 from tests.helpers.mark import hardware_test
 from tests.helpers.runtime import OmniServer

-pytestmark = [pytest.mark.full_model, pytest.mark.diffusion]
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]


 MODEL_ID = "Qwen/Qwen-Image-Layered"
@@ -11,7 +11,7 @@ from tests.e2e.accuracy.helpers import assert_video_similarity_metrics, probe_vi
 from tests.helpers.mark import hardware_marks
 from tests.helpers.runtime import DiffusionResponse, OmniServer, OpenAIClientHandler

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "BestWishYsh/Helios-Distilled"
 PROMPT = "A serene lakeside sunrise with mist over the water."
@@ -138,7 +138,7 @@ def _make_file_lora_request(adapter_dir: Path) -> LoRARequest:


 pytestmark = [
-    pytest.mark.full_model,
+    pytest.mark.slow,
    pytest.mark.diffusion,
    pytest.mark.parametrize("omni_runner", [_OMNI_RUNNER_PARAM], indirect=True),
 ]
@@ -109,7 +109,7 @@ def _diffusion_sampling_params(


 pytestmark = [
-    pytest.mark.full_model,
+    pytest.mark.slow,
    pytest.mark.diffusion,
 ]

@@ -96,7 +96,7 @@ def _get_test_config() -> str:

 pytestmark = [
    pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/4700"),
-    pytest.mark.full_model,
+    pytest.mark.slow,
    pytest.mark.tts,
    pytest.mark.parametrize(
        "omni_runner",
@@ -271,7 +271,6 @@ def _collect_audio(omni_runner: OmniRunner, request: dict) -> tuple[torch.Tensor
 # ---------------------------------------------------------------------------


-@pytest.mark.advanced_model
@hardware_test(res={"cuda": "L4"}, num_cards=1)
 def test_moss_tts_realtime_english(omni_runner: OmniRunner, ref_audio_path: str) -> None:
    """MossTTSRealtime: English voice_clone produces non-empty 24 kHz audio."""
@@ -283,7 +282,6 @@ def test_moss_tts_realtime_english(omni_runner: OmniRunner, ref_audio_path: str)
    assert not torch.all(audio == 0), "Audio is silence"


-@pytest.mark.advanced_model
@hardware_test(res={"cuda": "L4"}, num_cards=1)
 def test_moss_tts_realtime_chinese(omni_runner: OmniRunner, ref_audio_path: str) -> None:
    """MossTTSRealtime: Chinese input produces non-empty audio."""
@@ -22,7 +22,7 @@ from tests.helpers.mark import hardware_marks
 from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler, dummy_messages_from_mix_data
 from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 PROMPT = "A futuristic city skyline at twilight, cyberpunk style, ultra-detailed, high resolution."
 NEGATIVE_PROMPT = "low quality, blurry, distorted, deformed, watermark"
@@ -1,6 +1,8 @@
-"""Nightly-only e2e coverage for BAGEL diffusion multi-replica serving.
+"""Weekly e2e coverage for BAGEL diffusion multi-replica serving.

 This test needs 4 H100 GPUs; keep it out of generic test_bagel_* jobs.
+It is covered by the weekly Diffusion · H100 · Multi-GPU lane
+(``.buildkite/cuda/test-weekly.yml``, ``mirror_hardwares: h100_4``).
 """

 import os
@@ -17,7 +19,7 @@ MODEL = "ByteDance-Seed/BAGEL-7B-MoT"
 MULTI_REPLICA_DEPLOY = get_deploy_config_path("ci/bagel_multi_replicas_4gpu.yaml")
 ROUTE_STRESS_REQUESTS = 6

-pytestmark = [pytest.mark.full_model, pytest.mark.diffusion]
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]

 test_params = [
    OmniServerParams(
@@ -16,7 +16,7 @@ parity vs. the Diffusers baseline is covered separately (see the support plan, s
 From ``tests/``::

    pytest -s -v e2e/online_serving/test_boogu_image.py \
-        -m "full_model and diffusion and H100" --run-level=full_model
+        -m "slow and diffusion and H100"
 """

 import os
@@ -28,7 +28,7 @@ from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHand

 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "Boogu/Boogu-Image-0.1-Base"
 T2I_PROMPT = "A mountain lake at sunset, photorealistic, cinematic lighting"
@@ -19,7 +19,7 @@ reference image is supported.
 From ``tests/``::

    pytest -s -v e2e/online_serving/test_boogu_image_edit.py \
-        -m "full_model and diffusion and H100" --run-level=full_model
+        -m "slow and diffusion and H100"
 """

 import base64
@@ -35,7 +35,7 @@ from tests.helpers.runtime import (
    dummy_messages_from_mix_data,
 )

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 EDIT_PROMPT = "Change the style to a colored pencil drawing."
 SINGLE_CARD_FEATURE_MARKS = hardware_marks(res={"cuda": "H100"})
@@ -25,7 +25,7 @@ model and are intentionally omitted.
 From ``tests/``::

    pytest -s -v e2e/online_serving/test_boogu_image_edit_expansion.py \
-        -m "full_model and diffusion and H100" --run-level=full_model
+        -m "slow and diffusion and H100"
 """

 import os
@@ -38,7 +38,7 @@ from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHand

 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "Boogu/Boogu-Image-0.1-Edit"
 EDIT_PROMPT = "Change the style to a colored pencil drawing."
@@ -30,7 +30,7 @@ omitted (add rows here when those features land).
 From ``tests/``::

    pytest -s -v e2e/online_serving/test_boogu_image_expansion.py \
-        -m "full_model and diffusion and H100" --run-level=full_model
+        -m "slow and diffusion and H100"
 """

 import os
@@ -42,7 +42,7 @@ from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHand

 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "Boogu/Boogu-Image-0.1-Base"
 T2I_PROMPT = "A mountain lake at sunset, photorealistic, cinematic lighting"
@@ -30,7 +30,7 @@ from tests.helpers.runtime import (
 )
 from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = os.environ.get("GLM_IMAGE_MODEL_PATH", "zai-org/GLM-Image")

@@ -63,7 +63,7 @@ def test_text_to_image_moe(omni_server: OmniServer, openai_client: OpenAIClientH
    )


-@pytest.mark.full_model
+@pytest.mark.slow
@pytest.mark.diffusion
@pytest.mark.parametrize("omni_server", _get_server_cases(MODEL), indirect=True)
 def test_video_generation_modes_moe(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
@@ -10,7 +10,7 @@ from tests.helpers.mark import hardware_marks
 from tests.helpers.media import generate_synthetic_image
 from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 DISTILLED_MODEL = os.getenv("VLLM_TEST_LTX23_DISTILLED_MODEL", "diffusers/LTX-2.3-Distilled-Diffusers")
 BASE_MODEL = os.getenv("VLLM_TEST_LTX23_MODEL", "diffusers/LTX-2.3-Diffusers")
@@ -20,7 +20,7 @@ MODEL = os.getenv("VLLM_TEST_LTX25_MODEL", DEFAULT_MODEL)
 MODEL_REVISION = os.getenv("VLLM_TEST_LTX25_MODEL_REVISION", DEFAULT_REVISION if MODEL == DEFAULT_MODEL else "")
 PROMPT = "A red fox walks through a snowy forest while the camera remains fixed."

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]
 SINGLE_CARD_MARKS = hardware_marks(res={"cuda": "H100"})


@@ -20,7 +20,7 @@ from tests.helpers.stage_config import get_deploy_config_path

 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

-pytestmark = [pytest.mark.omni, pytest.mark.full_model]
+pytestmark = [pytest.mark.omni, pytest.mark.slow]

 _SKIP_NEED_4_H100_NOT_CI = pytest.mark.skip(
    reason="Requires 4x H100 GPUs; skipped in CI for now.",
@@ -2,20 +2,19 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 """
-Online serving tests for the Qwen-Image-Edit family (image-to-image via chat completions).
+Online serving tests for ``Qwen/Qwen-Image-Edit-2511`` (image-to-image via chat completions).

- ``test_single_image_to_image_001``: ``Qwen/Qwen-Image-Edit`` — one reference image, fixed 512×512.
- ``test_multi_images_to_image_001``: ``Qwen/Qwen-Image-Edit-2511`` — two reference images, fixed 512×512.
- ``test_different_sizes_001``: ``Qwen/Qwen-Image-Edit-2511`` only, ``slow`` — mixed input
-  resolutions; ``extra_body`` uses per-output ``width``/``height`` lists; the test client sends one
-  scalar-size request per list index in parallel and merges images (see ``OpenAIClientHandler.send_diffusion_request``).
+- ``test_single_image_to_image_001``: one reference image, fixed 512×512.
+- ``test_multi_images_to_image_001``: two reference images, fixed 512×512.
+- ``test_different_sizes_001``: ``slow`` — mixed input resolutions; ``extra_body`` uses
+  per-output ``width``/``height`` lists; the test client sends one scalar-size request
+  per list index in parallel and merges images (see ``OpenAIClientHandler.send_diffusion_request``).

-``_get_diffusion_feature_cases`` registers a single ``default`` ``OmniServerParams`` row per model.
+``_get_diffusion_feature_cases`` registers a single ``default`` ``OmniServerParams`` row.

 From ``tests/``::

-    pytest -s -v e2e/online_serving/test_qwen_image_edit.py -m "core_model and diffusion" --run-level=core_model
-    pytest -s -v e2e/online_serving/test_qwen_image_edit.py -m "advanced_model and diffusion" --run-level=advanced_model
+    pytest -s -v e2e/online_serving/test_qwen_image_edit.py -m "slow and diffusion" --run-level=full_model
 """

 import pytest
@@ -29,6 +28,7 @@ from tests.helpers.runtime import (
    dummy_messages_from_mix_data,
 )

+MODEL = "Qwen/Qwen-Image-Edit-2511"
 EDIT_PROMPT = "Transform this modern, geometrist image into a Vincent van Gogh style impressionist painting."
 MULTI_EDIT_PROMPT = (
    "Transform the first image into a Dadaism collage art. "
@@ -39,7 +39,7 @@ NEGATIVE_PROMPT = "blurry, low quality, modern, geometrist"
 SINGLE_CARD_FEATURE_MARKS = hardware_marks(res={"cuda": "H100"})


-def _get_diffusion_feature_cases(model: str):
+def _get_diffusion_feature_cases(model: str = MODEL):
    """Return one ``default`` ``OmniServerParams`` row for ``model`` (no extra ``server_args``)."""
    return [
        pytest.param(
@@ -56,11 +56,11 @@ def _get_diffusion_feature_cases(model: str):
@pytest.mark.diffusion
@pytest.mark.parametrize(
    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit"),
+    _get_diffusion_feature_cases(),
    indirect=True,
 )
 def test_single_image_to_image_001(omni_server: OmniServer, openai_client: OpenAIClientHandler):
-    """Single-reference edit smoke for ``Qwen/Qwen-Image-Edit`` (CFG when negative prompt + ``true_cfg_scale`` > 1)."""
+    """Single-reference edit smoke for ``Qwen/Qwen-Image-Edit-2511`` (CFG when negative prompt + ``true_cfg_scale`` > 1)."""
    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(512, 512)['base64']}"

    messages = dummy_messages_from_mix_data(image_data_url=image_data_url, content_text=EDIT_PROMPT)
@@ -86,7 +86,7 @@ def test_single_image_to_image_001(omni_server: OmniServer, openai_client: OpenA
@pytest.mark.diffusion
@pytest.mark.parametrize(
    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit-2511"),
+    _get_diffusion_feature_cases(),
    indirect=True,
 )
 def test_multi_images_to_image_001(omni_server: OmniServer, openai_client: OpenAIClientHandler):
@@ -117,7 +117,7 @@ def test_multi_images_to_image_001(omni_server: OmniServer, openai_client: OpenA
@pytest.mark.diffusion
@pytest.mark.parametrize(
    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit-2511"),
+    _get_diffusion_feature_cases(),
    indirect=True,
 )
 def test_different_sizes_001(omni_server: OmniServer, openai_client: OpenAIClientHandler):
@@ -1,8 +1,6 @@
 """
 Comprehensive tests of diffusion features that are available in online serving mode
-and are supported by the following models:
- Qwen-Image-Edit: single image input
- Qwen-Image-Edit-2511: single image input and two image inputs
+and are supported by Qwen-Image-Edit-2511 (single image input and two image inputs).
 """

 import pytest
@@ -11,10 +9,10 @@ from tests.helpers.mark import hardware_marks
 from tests.helpers.media import generate_synthetic_image
 from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler, dummy_messages_from_mix_data

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

-EDIT_PROMPT = "Transform this modern, geometrist image into a Vincent van Gogh style impressionist painting."
-SINGLE_EDIT_PROMPT_2511 = "Restyle this image into a Vincent van Gogh style impressionist painting."
+MODEL = "Qwen/Qwen-Image-Edit-2511"
+SINGLE_EDIT_PROMPT = "Restyle this image into a Vincent van Gogh style impressionist painting."
 MULTI_EDIT_PROMPT = (
    "Transform the first image into a Dadaism collage art. "
    "Transform the second image into a Vincent van Gogh style painting. "
@@ -25,9 +23,7 @@ SINGLE_CARD_FEATURE_MARKS = hardware_marks(res={"cuda": "H100"})
 PARALLEL_FEATURE_MARKS = hardware_marks(res={"cuda": "H100"}, num_cards=2)


-# This test file targets two models, so I write a helper function.
-# If a similar test only involves one model, one can just define a global list variable.
-def _get_diffusion_feature_cases(model: str):
+def _get_diffusion_feature_cases(model: str = MODEL):
    return [
        pytest.param(
            OmniServerParams(
@@ -123,35 +119,7 @@ def _get_diffusion_feature_cases(model: str):

@pytest.mark.parametrize(
    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit"),
-    indirect=True,
-)
-def test_qwen_image_edit(omni_server: OmniServer, openai_client: OpenAIClientHandler):
-    """Test all diffusion features with Qwen-Image-Edit in regular end-user scenarios."""
-    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(512, 512)['base64']}"
-
-    messages = dummy_messages_from_mix_data(image_data_url=image_data_url, content_text=EDIT_PROMPT)
-
-    # CFG parallel is only activated when a negative prompt and true_cfg_scale > 1.0 are both present
-    request_config = {
-        "model": omni_server.model,
-        "messages": messages,
-        "extra_body": {
-            "height": 512,
-            "width": 512,
-            "num_inference_steps": 2,
-            "negative_prompt": NEGATIVE_PROMPT,
-            "true_cfg_scale": 4.0,
-            "seed": 42,
-        },
-    }
-
-    openai_client.send_diffusion_request(request_config)
-
-
-@pytest.mark.parametrize(
-    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit-2511"),
+    _get_diffusion_feature_cases(),
    indirect=True,
 )
 def test_qwen_image_edit_2511_single_image(omni_server: OmniServer, openai_client: OpenAIClientHandler):
@@ -166,7 +134,7 @@ def test_qwen_image_edit_2511_single_image(omni_server: OmniServer, openai_clien
    """
    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(512, 512)['base64']}"

-    messages = dummy_messages_from_mix_data(image_data_url=image_data_url, content_text=SINGLE_EDIT_PROMPT_2511)
+    messages = dummy_messages_from_mix_data(image_data_url=image_data_url, content_text=SINGLE_EDIT_PROMPT)

    request_config = {
        "model": omni_server.model,
@@ -186,7 +154,7 @@ def test_qwen_image_edit_2511_single_image(omni_server: OmniServer, openai_clien

@pytest.mark.parametrize(
    "omni_server",
-    _get_diffusion_feature_cases("Qwen/Qwen-Image-Edit-2511"),
+    _get_diffusion_feature_cases(),
    indirect=True,
 )
 def test_qwen_image_edit_2511_two_images(omni_server: OmniServer, openai_client: OpenAIClientHandler):
@@ -18,7 +18,7 @@ from tests.helpers.mark import hardware_marks
 from tests.helpers.media import decode_b64_image, generate_synthetic_image
 from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler, dummy_messages_from_mix_data

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "Qwen/Qwen-Image-Layered"
 EDIT_PROMPT = "Decompose this image into layers."
@@ -16,7 +16,7 @@ from tests.helpers.media import load_test_audio_data_url
 from tests.helpers.runtime import OmniServerParams
 from tests.helpers.stage_config import get_deploy_config_path

-pytestmark = [pytest.mark.full_model, pytest.mark.tts]
+pytestmark = [pytest.mark.slow, pytest.mark.tts]

 os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

@@ -26,7 +26,7 @@ import pytest
 from tests.helpers.mark import hardware_marks
 from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler

-pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
+pytestmark = [pytest.mark.diffusion, pytest.mark.slow]

 MODEL = "Wan-AI/Wan2.1-VACE-1.3B-diffusers"
 PROMPT = "A cat walking slowly across a sunlit garden path"
@@ -411,6 +411,51 @@ async def test_async_chunk_prewarm_failure_reports_failing_downstream_stage(orch
            orchestrator_fixture.thread.join(timeout=5)


+@pytest.mark.asyncio
+async def test_async_chunk_prewarm_without_prompt_token_ids_fails_only_that_request(
+    orchestrator_factory,
+) -> None:
+    """R1.3 of #4855: an async-chunk request whose stage-0 prompt carries no
+    token ids cannot prewarm anything, so it must fail with a client error
+    instead of stalling.
+
+    Non-fatal is the point: ``fatal=True`` reads as "the engine died" and the
+    entrypoints act on it (``OmniLLM.generate`` raises out of its batch loop,
+    the serving layer calls ``terminate_if_errored``). One client's embeds-only
+    prompt must not reach other requests.
+    """
+    stage0 = FakeStageClient(stage_type="llm", final_output=False)
+    stage1 = FakeStageClient(stage_type="llm", final_output=True)
+    orchestrator_fixture = orchestrator_factory([stage0, stage1], async_chunk=True)
+
+    try:
+        await _enqueue_add_request(
+            orchestrator_fixture,
+            request_id="req-embeds",
+            # No prompt_token_ids: what an embeds-only prompt looks like here.
+            prompt=SimpleNamespace(request_id="req-embeds"),
+            original_prompt={"prompt": "embeds only"},
+            sampling_params_list=[_sampling_params(), _sampling_params()],
+            final_stage_id=1,
+        )
+
+        error_msg = await _wait_for_error_message(orchestrator_fixture, request_id="req-embeds")
+        assert error_msg.fatal is False
+        assert error_msg.status_code == 400
+        assert error_msg.error_type == "BadRequestError"
+        assert error_msg.stage_id == 0
+        assert "prompt_token_ids" in error_msg.error
+
+        # Nothing was prewarmed downstream, and the request is gone.
+        assert stage1.add_request_calls == []
+        await _wait_for(lambda: "req-embeds" not in orchestrator_fixture.orchestrator.request_states)
+        assert orchestrator_fixture.thread.is_alive()
+    finally:
+        if orchestrator_fixture.thread.is_alive():
+            orchestrator_fixture.request_sync_q.put_nowait(ShutdownRequestMessage())
+            orchestrator_fixture.thread.join(timeout=5)
+
+
 # ───────── Direct unit tests for the fault-isolation helpers ─────────


@@ -256,12 +256,13 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None:
        duplex_identity=SimpleNamespace(),
    )

-    await orchestrator._prewarm_async_chunk_stages(
+    prewarmed = await orchestrator._prewarm_async_chunk_stages(
        "req-prewarm",
        SimpleNamespace(prompt_token_ids=[1, 2], resumable=True),
        req_state,
    )

+    assert prewarmed is True
    assert stage1.submitted == []
    assert len(stage2.submitted) == 1
    assert 1 not in req_state.stage_submit_ts
@@ -277,6 +278,7 @@ async def test_async_prewarm_skips_outgoing_only_stage() -> None:
@pytest.mark.asyncio
 async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None:
    port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission()
+    prewarm.return_value = True

    result = await port.submit(submission)

@@ -285,6 +287,30 @@ async def test_duplex_prewarm_runs_after_first_stage0_submission() -> None:
    prewarm.assert_awaited_once_with("req-duplex", ANY, request_states["req-duplex"])


+@pytest.mark.asyncio
+async def test_duplex_submit_bails_out_when_prewarm_failed_the_request() -> None:
+    """A failed prewarm has already aborted the request and popped its state.
+
+    Returning a success result here would hand the control plane a replica for a
+    request that no longer exists, and the trailing bookkeeping would re-register
+    a running counter the cleanup just released -- a leak that never decrements.
+    """
+    port, stage_pools, request_states, prewarm, submission = _duplex_stage_port_submission()
+    counter = MagicMock()
+    port._running_counter = counter
+    prewarm.return_value = False
+    request_state = request_states["req-duplex"]
+
+    with pytest.raises(RuntimeError, match="prewarm failed"):
+        await port.submit(submission)
+
+    prewarm.assert_awaited_once()
+    assert request_state.duplex_stage_fences == {}
+    assert request_state.stage_submit_ts == {}
+    assert request_state.running_counter_registered is False
+    counter.increment.assert_not_called()
+
+
@pytest.mark.asyncio
 async def test_async_route_forwards_to_outgoing_only_stage() -> None:
    orchestrator = object.__new__(Orchestrator)
@@ -26,8 +26,13 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]


 def _serving(model_type: str = "audex") -> OmniOpenAIServingSpeech:
+    from vllm_omni.entrypoints.openai.tts_adapters import resolve_adapter
+    from vllm_omni.entrypoints.openai.tts_adapters.base import SpeechServingContext
+
    serving = OmniOpenAIServingSpeech.__new__(OmniOpenAIServingSpeech)
    serving._tts_model_type = model_type
+    adapter_cls = resolve_adapter(model_type)
+    serving._adapter = adapter_cls(SpeechServingContext(server=serving)) if adapter_cls is not None else None
    serving._mark_ref_audio_artifact_ready_for_request = lambda request_id: None
    serving._discard_ref_audio_artifact_warmup = lambda request_id: None
    return serving
@@ -154,7 +159,7 @@ class TestAudexCfgValidation:
        assert _adapter().validate(_speech_request(cfg_scale=None)) is None
        serving = TestAudexCfgInjection()._serving_with_fake_tokenizer()
        params = SimpleNamespace(extra_args={"cfg_scale": None})
-        out = serving._inject_audex_cfg_pair_args("req-1", {"prompt": "p"}, params)
+        out = serving._adapter._inject_audex_cfg_pair_args("req-1", {"prompt": "p"}, params)
        assert out.extra_args == {}

    @pytest.mark.parametrize("key", ["cfg_role", "cfg_pair_id", "cfg_null_prompt"])
@@ -177,21 +182,21 @@ class TestAudexCfgInjection:
    def test_no_extra_args_is_untouched(self):
        serving = self._serving_with_fake_tokenizer()
        params = SimpleNamespace(extra_args=None)
-        out = serving._inject_audex_cfg_pair_args("req-1", {"prompt": self._cond_prompt()}, params)
+        out = serving._adapter._inject_audex_cfg_pair_args("req-1", {"prompt": self._cond_prompt()}, params)
        assert out.extra_args is None
        assert params.extra_args is None

    def test_scale_one_leaves_params_byte_identical(self):
        serving = self._serving_with_fake_tokenizer()
        params = SimpleNamespace(extra_args={"cfg_scale": 1.0})
-        out = serving._inject_audex_cfg_pair_args("req-1", {"prompt": self._cond_prompt()}, params)
+        out = serving._adapter._inject_audex_cfg_pair_args("req-1", {"prompt": self._cond_prompt()}, params)
        assert out.extra_args == {}

    def test_guided_request_gains_pair_contract(self):
        serving = self._serving_with_fake_tokenizer()
        params = SimpleNamespace(extra_args={"cfg_scale": 1.5})
        cond = self._cond_prompt("Some transcription text here.")
-        out = serving._inject_audex_cfg_pair_args("req-42", {"prompt": cond}, params)
+        out = serving._adapter._inject_audex_cfg_pair_args("req-42", {"prompt": cond}, params)

        extra = out.extra_args
        assert extra["cfg_scale"] == 1.5
@@ -207,14 +212,14 @@ class TestAudexCfgInjection:
        serving = self._serving_with_fake_tokenizer()
        params = SimpleNamespace(extra_args={"cfg_scale": 1.5})
        with pytest.raises(ValueError, match="adapter-built text prompt"):
-            serving._inject_audex_cfg_pair_args("req-1", {"prompt_token_ids": [1]}, params)
+            serving._adapter._inject_audex_cfg_pair_args("req-1", {"prompt_token_ids": [1]}, params)

    def test_injection_never_mutates_input_params(self):
        """The input may be the engine's SHARED default params (review P2):
        pair state must land only on the returned clone."""
        serving = self._serving_with_fake_tokenizer()
        shared = SimpleNamespace(extra_args={"cfg_scale": 1.5}, temperature=0.1)
-        out = serving._inject_audex_cfg_pair_args("req-7", {"prompt": self._cond_prompt()}, shared)
+        out = serving._adapter._inject_audex_cfg_pair_args("req-7", {"prompt": self._cond_prompt()}, shared)
        assert out is not shared
        assert shared.extra_args == {"cfg_scale": 1.5}
        assert shared.temperature == 0.1
@@ -252,11 +257,11 @@ def test_guided_injection_sets_guided_temperature():
    serving = TestAudexCfgInjection()._serving_with_fake_tokenizer()
    cond = TestAudexCfgInjection()._cond_prompt()
    params = SimpleNamespace(extra_args={"cfg_scale": 1.5}, temperature=0.1)
-    out = serving._inject_audex_cfg_pair_args("req-1", {"prompt": cond}, params)
+    out = serving._adapter._inject_audex_cfg_pair_args("req-1", {"prompt": cond}, params)
    assert out.temperature == 0.05

    unguided = SimpleNamespace(extra_args={"cfg_scale": 1.0}, temperature=0.1)
-    out = serving._inject_audex_cfg_pair_args("req-2", {"prompt": cond}, unguided)
+    out = serving._adapter._inject_audex_cfg_pair_args("req-2", {"prompt": cond}, unguided)
    assert out.temperature == 0.1


@@ -340,7 +345,7 @@ class TestAudexTTAInjection:
    def test_default_injection_applies_rvq_and_cfg3(self):
        serving = self._serving()
        params = SimpleNamespace(extra_args=None)
-        out = serving._inject_audex_tta_args("req-9", self._prompt(), params)
+        out = serving._adapter._inject_audex_tta_args("req-9", self._prompt(), params)

        extra = out.extra_args
        rvq = extra["tta_rvq"]
@@ -355,15 +360,15 @@ class TestAudexTTAInjection:
    def test_explicit_scale_one_disables_cfg_but_keeps_rvq(self):
        serving = self._serving()
        params = SimpleNamespace(extra_args={"cfg_scale": 1.0})
-        out = serving._inject_audex_tta_args("req-1", self._prompt(), params)
+        out = serving._adapter._inject_audex_tta_args("req-1", self._prompt(), params)
        assert "tta_rvq" in out.extra_args
        assert "cfg_scale" not in out.extra_args
        assert "cfg_role" not in out.extra_args

    def test_rvq_contract_cached_per_process(self):
        serving = self._serving()
-        out_a = serving._inject_audex_tta_args("a", self._prompt(), SimpleNamespace(extra_args=None))
-        out_b = serving._inject_audex_tta_args("b", self._prompt(), SimpleNamespace(extra_args=None))
+        out_a = serving._adapter._inject_audex_tta_args("a", self._prompt(), SimpleNamespace(extra_args=None))
+        out_b = serving._adapter._inject_audex_tta_args("b", self._prompt(), SimpleNamespace(extra_args=None))
        assert out_a.extra_args["tta_rvq"] is out_b.extra_args["tta_rvq"]

    def test_injection_never_mutates_input_params(self):
@@ -372,8 +377,8 @@ class TestAudexTTAInjection:
        per-request pair state must land only on the returned clone."""
        serving = self._serving()
        shared = SimpleNamespace(extra_args=None)
-        out_1 = serving._inject_audex_tta_args("req-1", self._prompt(), shared)
-        out_2 = serving._inject_audex_tta_args("req-2", self._prompt(), shared)
+        out_1 = serving._adapter._inject_audex_tta_args("req-1", self._prompt(), shared)
+        out_2 = serving._adapter._inject_audex_tta_args("req-2", self._prompt(), shared)
        assert shared.extra_args is None
        assert out_1 is not shared and out_2 is not shared
        assert out_1.extra_args["cfg_pair_id"] == "req-1"
@@ -52,6 +52,7 @@ from vllm_omni.model_executor.models.ming_tts.constants import (
    KEY_SPEAKER_WAVEFORM,
    KEY_SPEAKER_WAVEFORM_LENGTHS,
    SPEAKER_EMBEDDING_DIM,
+    TEXT_EOS_TOKEN_ID,
 )
 from vllm_omni.outputs import OmniRequestOutput

@@ -3618,7 +3619,21 @@ class TestCosyVoice3Serving:
        assert error is not None
        assert "max_new_tokens" in error

-    def test_prepare_speech_generation_cosyvoice3(self, cosyvoice3_server, mocker: MockerFixture):
+    @pytest.mark.parametrize(
+        ("max_new_tokens", "expected_min_tokens", "expected_max_tokens"),
+        [
+            (None, 10, 2000),
+            (5, 5, 5),
+        ],
+    )
+    def test_prepare_speech_generation_cosyvoice3(
+        self,
+        cosyvoice3_server,
+        mocker: MockerFixture,
+        max_new_tokens: int | None,
+        expected_min_tokens: int,
+        expected_max_tokens: int,
+    ):
        cosyvoice3_server._build_cosyvoice3_prompt = mocker.AsyncMock(
            return_value={
                "prompt": "Hello",
@@ -3626,15 +3641,23 @@ class TestCosyVoice3Serving:
                "mm_processor_kwargs": {"prompt_text": "ref text", "sample_rate": 24000},
            }
        )
-        cosyvoice3_server._apply_cosyvoice3_dynamic_tokens = mocker.MagicMock(side_effect=lambda spl, req: spl)
+        cosyvoice3_server.model_config.hf_config = SimpleNamespace(
+            min_token_text_ratio=1, max_token_text_ratio=200, allowed_special=True
+        )
+        cosyvoice3_server._cosyvoice3_tokenizer = mocker.MagicMock()
+        mocker.patch("vllm_omni.model_executor.models.cosyvoice3.utils.extract_text_token", return_value=[None, 10])

        request = OpenAICreateSpeechRequest(
            input="Hello",
            ref_audio="data:audio/wav;base64,abc",
            ref_text="Reference text",
+            max_new_tokens=max_new_tokens,
        )
        request_id, generator, tts_params = asyncio.run(cosyvoice3_server._prepare_speech_generation(request))
+        sampling_params = cosyvoice3_server.engine_client.generate.call_args.kwargs["sampling_params_list"][0]

+        assert sampling_params.min_tokens == expected_min_tokens
+        assert sampling_params.max_tokens == expected_max_tokens
        assert request_id.startswith("speech-")
        assert generator == "generator"
        assert tts_params == {}
@@ -3718,6 +3741,90 @@ class TestGLMTTSServing:
        assert text_token_len == 3
        load_tokenizer.assert_called_once()

+    def test_prepare_speech_generation_glm_tts(self, glm_tts_server, mocker: MockerFixture):
+        glm_tts_server._build_glm_tts_prompt = mocker.AsyncMock(
+            return_value={
+                "prompt": "Hello",
+                "multi_modal_data": {"audio": (np.zeros(24000), 24000)},
+                "mm_processor_kwargs": {"prompt_text": "ref text", "sample_rate": 24000},
+            }
+        )
+        request = OpenAICreateSpeechRequest(
+            input="Hello",
+            ref_audio="data:audio/wav;base64,abc",
+            ref_text="Reference text",
+        )
+        glm_tts_server.model_config.hf_config = SimpleNamespace(min_token_text_ratio=1, max_token_text_ratio=200)
+        glm_tts_server._estimate_glm_tts_text_token_len = mocker.MagicMock(return_value=10)
+        request_id, generator, tts_params = asyncio.run(glm_tts_server._prepare_speech_generation(request))
+        sampling_params = glm_tts_server.engine_client.generate.call_args.kwargs["sampling_params_list"][0]
+
+        assert sampling_params.min_tokens == 10
+        assert sampling_params.max_tokens == 2000
+        assert request_id.startswith("speech-")
+        assert generator == "generator"
+        assert tts_params == {}
+        glm_tts_server._build_glm_tts_prompt.assert_awaited_once()
+
+
+@pytest.fixture
+def ming_tts_server(mocker: MockerFixture):
+    mocker.patch.object(OmniOpenAIServingSpeech, "_load_supported_speakers", return_value=set())
+    mocker.patch.object(OmniOpenAIServingSpeech, "_load_codec_frame_rate", return_value=None)
+
+    mock_engine_client = mocker.MagicMock()
+    mock_engine_client.errored = False
+    mock_engine_client.model_config = mocker.MagicMock(
+        model="inclusionAI/Ming-omni-tts-0.5B",
+        hf_config=SimpleNamespace(min_token_text_ratio=2, max_token_text_ratio=20),
+    )
+    mock_engine_client.default_sampling_params_list = [
+        SimpleNamespace(max_tokens=2048, min_tokens=None, extra_args=None)
+    ]
+    mock_engine_client.tts_batch_max_items = 32
+    mock_engine_client.generate = mocker.MagicMock(return_value="generator")
+    mock_engine_client.stage_configs = [
+        SimpleNamespace(
+            engine_args=SimpleNamespace(
+                model_stage="llm", model_arch="MingTTSForConditionalGeneration", worker_type="ar"
+            ),
+            tts_args={},
+        )
+    ]
+
+    mock_models = mocker.MagicMock()
+    mock_models.is_base_model.return_value = True
+    return OmniOpenAIServingSpeech(
+        engine_client=mock_engine_client,
+        models=mock_models,
+        request_logger=mocker.MagicMock(),
+    )
+
+
+class TestMingTTSServing:
+    def test_prepare_speech_generation_ming(self, ming_tts_server, mocker: MockerFixture):
+        ming_tts_server._adapter.build = mocker.AsyncMock(
+            return_value=PreparedRequest(
+                prompt={"prompt": "hello"},
+                tts_params={},
+                model_type="ming_tts",
+            )
+        )
+
+        request = OpenAICreateSpeechRequest(
+            input="Hello",
+            ref_audio="data:audio/wav;base64,abc",
+            ref_text="Reference text",
+            max_new_tokens=7,
+        )
+        asyncio.run(ming_tts_server._prepare_speech_generation(request))
+
+        assert ming_tts_server._tts_model_type == "ming_tts"
+        ming_tts_server._adapter.build.assert_awaited_once()
+        sampling_params = ming_tts_server.engine_client.generate.call_args.kwargs["sampling_params_list"][0]
+        assert sampling_params.stop_token_ids == [TEXT_EOS_TOKEN_ID]
+        assert sampling_params.max_tokens == 8
+

@pytest.fixture
 def ming_flash_omni_tts_server(mocker: MockerFixture):
@@ -3901,6 +4008,9 @@ class TestTTSAsyncOffloading:
                    model_type=adapter_model_type,
                )

+            def apply_sampling_overrides(self, sampling_params_list, request, prompt, request_id=None):
+                return sampling_params_list
+
        voxtral_server._tts_model_type = legacy_tts_model_type
        mocker.patch.object(voxtral_server, "_get_tts_adapter", return_value=FakeAdapter())
        log_info = mocker.patch("vllm_omni.entrypoints.openai.serving_speech.logger.info")
@@ -14,7 +14,6 @@ import itertools
 import pytest

 from vllm_omni.entrypoints.openai.tts_adapters import (
-    LEGACY_TTS_DETECTORS,
    TTS_ADAPTER_REGISTRY,
    all_tts_stage_keys,
    detect_tts_model_type,
@@ -282,27 +281,17 @@ def test_entry_stage_archs_is_ming_only():
    assert tts_entry_stage_archs() == frozenset({"MingTTSForConditionalGeneration"})


-def test_every_detected_type_is_adapter_backed_or_declared_legacy():
-    """Detection may only name a model that has an adapter or an explicit
-    :data:`LEGACY_TTS_DETECTORS` entry — never an undeclared string."""
-    legacy_names = {d.name for d in LEGACY_TTS_DETECTORS}
+def test_every_detected_type_is_adapter_backed():
+    """Detection may only name a model that has a registered adapter."""
    for stage, arch in itertools.product(_STAGES, _ARCHS):  # full cross product on purpose
        detected = detect_tts_model_type(stage, arch)
        if detected is None:
            continue
-        assert detected in TTS_ADAPTER_REGISTRY or detected in legacy_names, (
+        assert detected in TTS_ADAPTER_REGISTRY, (
            f"detection produced undeclared model type {detected!r} for stage={stage!r} arch={arch!r}"
        )


-def test_legacy_detectors_have_no_adapter():
-    """A legacy entry that gained an adapter must be deleted from the list."""
-    for detector in LEGACY_TTS_DETECTORS:
-        assert resolve_adapter(detector.name) is None, (
-            f"{detector.name!r} now has an adapter; remove it from LEGACY_TTS_DETECTORS"
-        )
-
-
 def test_no_ambiguous_detectors_at_equal_priority():
    """Two detectors that can match the same deployment must be ordered.

@@ -0,0 +1,144 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from __future__ import annotations
+
+import io
+from pathlib import Path
+from types import SimpleNamespace
+
+import av
+import numpy as np
+import pytest
+from PIL import Image
+
+from examples.offline_inference.x_to_video_audio import x_to_video_audio
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_builds_canonical_image_audio_to_video_prompt() -> None:
+    image = Image.new("RGB", (16, 8), "red")
+    audio = (np.arange(8, dtype=np.float32), 24000)
+
+    prompt = x_to_video_audio.build_x_to_video_audio_prompt(
+        "A person speaks to the camera.",
+        {"image": [image], "audio": [audio]},
+    )
+
+    assert prompt == {
+        "prompt": "A person speaks to the camera.",
+        "modalities": ["video"],
+        "multi_modal_data": {"image": [image], "audio": [audio]},
+    }
+
+
+def test_canonical_prompt_omits_empty_media() -> None:
+    assert x_to_video_audio.build_x_to_video_audio_prompt("A landscape.") == {
+        "prompt": "A landscape.",
+        "modalities": ["video"],
+    }
+
+
+def test_official_prompt_cleanup_stays_in_prompt_file_handling() -> None:
+    prompt = (
+        "[SPEAKER_TIMESTAMPS_START]metadata[SPEAKER_TIMESTAMPS_END]\n\n"
+        "A person waves.\n[AUDIO_DESCRIPTION_START]noise[AUDIO_DESCRIPTION_END]"
+    )
+
+    assert x_to_video_audio._clean_official_prompt(prompt) == "A person waves."
+
+
+def test_media_loader_preserves_complete_audio(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    image_path = tmp_path / "reference.png"
+    Image.new("RGBA", (4, 3), "blue").save(image_path)
+    complete_waveform = np.arange(80000, dtype=np.float32)
+    requested_rates: list[int | None] = []
+
+    def fake_load_audio(path: str, *, sr: int | None) -> tuple[np.ndarray, int]:
+        assert path == "reference.wav"
+        requested_rates.append(sr)
+        return complete_waveform, 16000
+
+    monkeypatch.setattr(x_to_video_audio, "load_audio", fake_load_audio)
+
+    images, audios = x_to_video_audio.load_image_and_audio(
+        [str(image_path)],
+        ["reference.wav"],
+        audio_sample_rate=16000,
+    )
+
+    assert images[0].mode == "RGB"
+    assert requested_rates == [16000]
+    assert audios[0][0] is complete_waveform
+    assert audios[0][0].shape == (80000,)
+    assert audios[0][1] == 16000
+
+
+@pytest.mark.parametrize(
+    "video",
+    [
+        np.zeros((3, 2, 4, 4), dtype=np.float32),
+        np.zeros((2, 4, 4, 3), dtype=np.uint8),
+    ],
+    ids=["channel_first_float", "frame_first_uint8"],
+)
+def test_output_contract_keeps_model_video_layout_opaque(video: np.ndarray) -> None:
+    audio = np.zeros(16, dtype=np.float32)
+    result = SimpleNamespace(
+        images=[video],
+        multimodal_output={"audio": audio, "fps": 25, "audio_sample_rate": 44100},
+    )
+
+    output = x_to_video_audio.extract_x_to_video_audio_output([result])
+
+    assert output.video is video
+    assert output.audio is audio
+    assert output.fps == 25.0
+    assert output.audio_sample_rate == 44100
+
+
+def test_output_contract_accepts_video_from_multimodal_output() -> None:
+    video = object()
+    result = SimpleNamespace(images=[], multimodal_output={"video": video, "fps": 12})
+
+    output = x_to_video_audio.extract_x_to_video_audio_output(result)
+
+    assert output == x_to_video_audio.XToVideoAudioOutput(
+        video=video,
+        audio=None,
+        fps=12.0,
+        audio_sample_rate=None,
+    )
+
+
+def test_output_contract_requires_model_neutral_metadata() -> None:
+    result = SimpleNamespace(images=[object()], multimodal_output={"audio": np.zeros(4)})
+
+    with pytest.raises(RuntimeError, match="must declare a valid 'fps'"):
+        x_to_video_audio.extract_x_to_video_audio_output(result)
+
+    with pytest.raises(RuntimeError, match="must declare a valid 'audio_sample_rate'"):
+        x_to_video_audio.extract_x_to_video_audio_output(result, fps=24)
+
+
+def test_output_encoder_normalizes_declared_range_and_channel_first_layout() -> None:
+    video = np.zeros((3, 2, 4, 6), dtype=np.float32)
+    output = x_to_video_audio.XToVideoAudioOutput(
+        video=video,
+        audio=None,
+        fps=8,
+        audio_sample_rate=None,
+        output_tensor_range="negative_one_to_one",
+    )
+
+    payload = x_to_video_audio.encode_x_to_video_audio_output(output)
+
+    with av.open(io.BytesIO(payload)) as container:
+        decoded = [frame.to_ndarray(format="rgb24") for frame in container.decode(video=0)]
+    assert len(decoded) == 2
+    assert decoded[0].shape == (4, 6, 3)
+    np.testing.assert_allclose(decoded, 128, atol=2)
@@ -15,8 +15,10 @@ from vllm_omni.model_extras import (
    build_image_to_video_prompt,
    build_text_to_image_prompt,
    build_x_to_text_prompt,
+    build_x_to_video_audio_prompt,
    get_extra_body_params,
    get_extra_output_params,
+    get_input_audio_sample_rate,
    get_output_tensor_range,
    get_x_to_text_model_family,
    should_init_extra_args_for_non_diffusion_stages,
@@ -141,6 +143,50 @@ def test_lingbot_extra_registry_declares_request_params() -> None:
    assert should_init_extra_args_for_non_diffusion_stages("LingBotVideoPipeline") is False


+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_dreamid_x_to_video_audio_adapter() -> None:
+    image = Image.new("RGB", (16, 16))
+    waveform = object()
+    canonical_prompt = {
+        "prompt": "A person waves.",
+        "modalities": ["video"],
+        "multi_modal_data": {"image": [image], "audio": [(waveform, 16000)]},
+    }
+
+    result = build_x_to_video_audio_prompt(
+        "DreamIDOmniPipeline",
+        canonical_prompt,
+        {
+            "video_negative_prompt": "blur",
+            "audio_negative_prompt": "echo",
+        },
+    )
+
+    assert result == {
+        "prompt": "A person waves.",
+        "modalities": ["video"],
+        "video_negative_prompt": "blur",
+        "audio_negative_prompt": "echo",
+        "multi_modal_data": {"image": [image], "audio": [waveform]},
+    }
+    assert get_extra_body_params("DreamIDOmniPipeline") == frozenset({"solver_name", "shift"})
+    assert get_input_audio_sample_rate("DreamIDOmniPipeline") == 16000
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_generic_x_to_video_audio_adapter_is_model_neutral() -> None:
+    canonical_prompt = {
+        "prompt": "A person waves.",
+        "modalities": ["video"],
+        "multi_modal_data": {"audio": [(object(), 48000)]},
+    }
+
+    assert build_x_to_video_audio_prompt("FuturePipeline", canonical_prompt, {}) is canonical_prompt
+    assert get_input_audio_sample_rate("FuturePipeline") is None
+
+
@pytest.mark.core_model
@pytest.mark.cpu
@pytest.mark.parametrize("pipeline_name", ["Cosmos3OmniDiffusersPipeline", "Cosmos3OmniPipeline"])
@@ -17,9 +17,7 @@ import pytest
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tools" / "pre_commit"))

 from check_tts_adapter import (  # noqa: E402
-    MAX_LEGACY_DETECTORS,
    MAX_MODEL_TYPE_BRANCHES,
-    _legacy_detectors,
    _model_type_branches,
    main,
 )
@@ -60,12 +58,6 @@ def test_dict_dispatch_is_a_known_gap():
    assert _count('TABLE = {"foo": 1}\nvalue = TABLE[self._tts_model_type]') == 0


-def test_legacy_detector_counting():
-    src = 'X = (LegacyDetector(name="a", stage_keys=frozenset({"s"})), LegacyDetector(name="b"))'
-    found = _legacy_detectors(ast.parse(src))
-    assert [name for _, name in found] == ["a", "b"]
-
-
 def test_gate_passes_on_current_tree():
    """The recorded budgets must match reality, or the ratchet is decorative."""
    assert main([]) == 0
@@ -74,6 +66,4 @@ def test_gate_passes_on_current_tree():
 def test_budgets_are_the_real_counts():
    root = Path(__file__).resolve().parents[2]
    serving = root / "vllm_omni" / "entrypoints" / "openai" / "serving_speech.py"
-    adapters = root / "vllm_omni" / "entrypoints" / "openai" / "tts_adapters" / "__init__.py"
    assert len(_model_type_branches(ast.parse(serving.read_text()))) == MAX_MODEL_TYPE_BRANCHES
-    assert len(_legacy_detectors(ast.parse(adapters.read_text()))) == MAX_LEGACY_DETECTORS
@@ -2,6 +2,7 @@ import asyncio
 import copy
 from collections.abc import Sequence
 from types import SimpleNamespace
+from typing import Any

 import numpy as np
 import pytest
@@ -200,6 +201,15 @@ def test_speech_extra_params_reach_model_sampler_as_sampling_metadata(monkeypatc
        async def build(self, request, sampling_params_list, has_inline_ref_audio):
            return PreparedRequest(prompt={"prompt": request.input}, tts_params={}, model_type="higgs_audio_v3")

+        def apply_sampling_overrides(
+            self,
+            sampling_params_list: list,
+            request: "OpenAICreateSpeechRequest",
+            prompt: dict[str, Any] | None = None,
+            request_id: str | None = None,
+        ) -> list:
+            return sampling_params_list
+
    engine = SimpleNamespace(
        errored=False,
        default_sampling_params_list=[SamplingParams(temperature=1.0, top_p=0.95, top_k=50)],
@@ -1,8 +1,7 @@
 #!/usr/bin/env python3
 """
-Parse pytest commands from Buildkite test-ready.yml, test-merge.yml, test-nightly.yml,
-and test-weekly.yml; collect test cases (including parametrized) via pytest --collect-only -q
-and produce an HTML report.
+Parse pytest commands from Buildkite .buildkite/cuda/test-{ready,merge,nightly,weekly}.yml;
+collect test cases (including parametrized) via pytest --collect-only -q and produce an HTML report.

 Leaf steps may live under ``group:`` blocks (nested ``steps``); those are flattened with the
 group title carried into the report.
@@ -30,7 +29,7 @@ import yaml

 # Repo root (parent of the directory containing this script)
 REPO_ROOT = Path(__file__).resolve().parent.parent.parent
-BUILDKITE_DIR = REPO_ROOT / ".buildkite"
+BUILDKITE_DIR = REPO_ROOT / ".buildkite" / "cuda"
 PIPELINE_FILES = ["test-ready.yml", "test-merge.yml", "test-nightly.yml", "test-weekly.yml"]


@@ -256,13 +255,34 @@ def _resolve_pytest_target(target: str, repo_root: Path, extra: list[str], raw_l
        rel_dir = Path(target.rstrip("/"))
        dir_path = (repo_root / rel_dir).resolve()
        if dir_path.exists() and dir_path.is_dir():
-            return [str(dir_path)], target.replace("\\", "/").rstrip("/"), 120
+            return [str(dir_path)], target.replace("\\", "/").rstrip("/"), 300

    if not extra:
        raise RuntimeError(f"Failed to parse pytest sidecar args from line: {raw_line!r}")
    if not (repo_root / "tests").exists():
        raise FileNotFoundError("tests/ directory not found under repo root")
-    return ["tests/"], "tests/", 120
+    # Broad ``tests/`` collection loads many conftests; allow more time on busy CI agents.
+    return ["tests/"], "tests/", 300
+
+
+def _format_pytest_collect_failure(
+    path_args: list[str],
+    extra: list[str],
+    result: subprocess.CompletedProcess[str],
+) -> str:
+    """pytest often puts collection errors on stdout; include both streams + exit code."""
+    stdout = (result.stdout or "").strip()
+    stderr = (result.stderr or "").strip()
+    parts = [
+        f"pytest --collect-only failed for {path_args} with args {extra} (exit={result.returncode})",
+    ]
+    if stderr:
+        parts.append(f"stderr:\n{stderr}")
+    if stdout:
+        parts.append(f"stdout:\n{stdout}")
+    if not stderr and not stdout:
+        parts.append("(no stdout/stderr captured)")
+    return "\n".join(parts)


 def collect_test_names(target: str, repo_root: Path, raw_line: str) -> list[str]:
@@ -273,6 +293,8 @@ def collect_test_names(target: str, repo_root: Path, raw_line: str) -> list[str]
    flags ``--ignore``, ``--test-config-file``, ``-m``, ``-k``, ``--run-level`` are taken from the
    full ``raw_line`` and appended to the same invocation so collection matches CI (perf scripts
    parametrize from ``--test-config-file`` at import time).
+
+    Pytest exit code 5 (no tests collected) is treated as an empty result, not a hard failure.
    """
    extra = _pytest_collect_sidecar_args(raw_line)
    path_args, _fallback, timeout_quiet = _resolve_pytest_target(target, repo_root, extra, raw_line)
@@ -286,8 +308,15 @@ def collect_test_names(target: str, repo_root: Path, raw_line: str) -> list[str]
        text=True,
        timeout=timeout_quiet,
    )
+    # 5 == no tests collected under the given -m/-k filters.
+    if result.returncode == 5:
+        print(
+            f"pytest --collect-only: no tests matched for {path_args} with args {extra}",
+            file=sys.stderr,
+        )
+        return []
    if result.returncode != 0:
-        raise RuntimeError(f"pytest --collect-only failed for {path_args} with args {extra}: {result.stderr.strip()}")
+        raise RuntimeError(_format_pytest_collect_failure(path_args, extra, result))
    print(f"pytest --collect-only success for {path_args} with args {extra}: {result.stdout.strip()}")
    return _parse_collect_only_stdout(result.stdout or "", stderr=result.stderr or "")

@@ -962,7 +991,7 @@ def main() -> None:
        "--buildkite-dir",
        type=Path,
        default=BUILDKITE_DIR,
-        help="Path to .buildkite directory",
+        help="Path to Buildkite CUDA pipeline directory (default: .buildkite/cuda)",
    )
    default_out = REPO_ROOT / "buildkite_testcase_statistics.html"
    parser.add_argument(
@@ -1023,7 +1052,12 @@ def main() -> None:
            names = []
            skipped_ids: set[str] = set()
            for target, raw in targets:
-                collected = collect_test_names(target, REPO_ROOT, raw)
+                try:
+                    collected = collect_test_names(target, REPO_ROOT, raw)
+                except (RuntimeError, FileNotFoundError, subprocess.TimeoutExpired) as exc:
+                    # Keep generating the report for other steps; surface the root cause.
+                    print(f"WARNING: collect failed for {module_key} target={target!r}: {exc}", file=sys.stderr)
+                    collected = []
                skipped_ids.update(get_skip_status(target, raw, REPO_ROOT))
                names.extend(collected)
            count = len(names) - len(skipped_ids)
@@ -6,10 +6,8 @@ Per-model behaviour is being moved out of ``serving_speech.py`` and into
 nothing stopped a new model from adding one more ``self._tts_model_type == ...``
 branch faster than the migration removed them.

-This gate makes the two debt counters monotonically non-increasing:
-
-1. ``self._tts_model_type`` comparisons in ``serving_speech.py``
-2. entries in ``tts_adapters.LEGACY_TTS_DETECTORS`` (models with no adapter)
+This gate makes the number of ``self._tts_model_type`` comparisons in
+``serving_speech.py`` monotonically non-increasing.

 Lowering a budget after removing branches is expected and encouraged — the check
 insists on it, so the recorded numbers cannot silently drift upward. Raising one
@@ -28,11 +26,9 @@ from pathlib import Path

 REPO_ROOT = Path(__file__).resolve().parents[2]
 SERVING_SPEECH = Path("vllm_omni/entrypoints/openai/serving_speech.py")
-ADAPTERS_INIT = Path("vllm_omni/entrypoints/openai/tts_adapters/__init__.py")

 # Budgets. These may only go DOWN. See the module docstring.
-MAX_MODEL_TYPE_BRANCHES = 27
-MAX_LEGACY_DETECTORS = 0
+MAX_MODEL_TYPE_BRANCHES = 20

 _GUIDANCE = """
 Move the per-model behaviour into its adapter under
@@ -109,20 +105,6 @@ def _model_type_branches(tree: ast.AST) -> list[tuple[int, str]]:
    return found


-def _legacy_detectors(tree: ast.AST) -> list[tuple[int, str]]:
-    """Every ``LegacyDetector(...)`` construction."""
-    found: list[tuple[int, str]] = []
-    for node in ast.walk(tree):
-        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "LegacyDetector":
-            name = "?"
-            for kw in node.keywords:
-                if kw.arg == "name" and isinstance(kw.value, ast.Constant):
-                    name = str(kw.value.value)
-            found.append((node.lineno, name))
-    found.sort()
-    return found
-
-
 def _check(
    label: str,
    path: Path,
@@ -151,21 +133,18 @@ def _check(

 def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
-    # pre-commit passes the staged filenames; we always audit the same two files,
+    # pre-commit passes the staged filenames; we always audit the same file,
    # so they are accepted and ignored.
    parser.add_argument("filenames", nargs="*")
    parser.parse_args(argv)

    serving_path = REPO_ROOT / SERVING_SPEECH
-    adapters_path = REPO_ROOT / ADAPTERS_INIT
-    for path in (serving_path, adapters_path):
-        if not path.is_file():
-            print(f"check_tts_adapter: expected file not found: {path}", file=sys.stderr)
-            return 1
+    if not serving_path.is_file():
+        print(f"check_tts_adapter: expected file not found: {serving_path}", file=sys.stderr)
+        return 1

    notices: list[str] = []
    branches = _model_type_branches(ast.parse(serving_path.read_text()))
-    detectors = _legacy_detectors(ast.parse(adapters_path.read_text()))

    errors = _check(
        "`_tts_model_type` comparisons",
@@ -176,15 +155,6 @@ def main(argv: list[str] | None = None) -> int:
        "MAX_MODEL_TYPE_BRANCHES",
        notices,
    )
-    errors += _check(
-        "unmigrated model types in LEGACY_TTS_DETECTORS",
-        ADAPTERS_INIT,
-        len(detectors),
-        MAX_LEGACY_DETECTORS,
-        [f"line {lineno}: {name}" for lineno, name in detectors],
-        "MAX_LEGACY_DETECTORS",
-        notices,
-    )

    for notice in notices:
        print(f"check_tts_adapter: {notice}", file=sys.stderr)
@@ -1,5 +1,6 @@
 from __future__ import annotations

+import math
 import os
 import time
 from collections.abc import Iterable
@@ -42,25 +43,56 @@ logger = init_logger(__name__)

 _STATS_INTERVAL_S = 1.0

-# Upper bound on how long a request may sit in full-payload-input wait
-# (the state ``OmniSchedulingCoordinator`` records via ``_waiting_since``)
-# before the scheduler force-fails it.  Defends against stuck consumer-side
-# requests when the producer drops a full-payload, send fails, or recv
-# never arrives.  Override per-deployment via
-# VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set <=0 to disable the safety net.
+# Upper bound on how long a request may wait for stage input before the
+# scheduler force-fails it.  Defends against stuck consumer-side requests when
+# the producer drops a payload, send fails, or recv never arrives.  Override
+# per-deployment via VLLM_OMNI_INPUT_WAIT_TIMEOUT_S; set exactly 0 to disable
+# the safety net. Negative and non-finite values are rejected at startup.
 #
-# Scope: this constant only covers the full-payload coordinator path
-# (``input_coordinator``).  The async-chunk path uses
-# ``chunk_transfer_adapter`` and is not affected by this constant.
+# Scope: both transports.  The full-payload path measures time parked in
+# WAITING_FOR_INPUT (``OmniSchedulingCoordinator._waiting_since``); the
+# async-chunk path measures time stalled in WAITING_FOR_CHUNK
+# (``OmniChunkTransferAdapter._waiting_since``, reset on each chunk arrival).
+# One knob, because the question it answers -- "how long may a request wait for
+# stage input?" -- does not depend on which transport carries it.
+_INPUT_WAIT_TIMEOUT_DEFAULT = 600.0
 _INPUT_WAIT_TIMEOUT_RAW = os.environ.get("VLLM_OMNI_INPUT_WAIT_TIMEOUT_S", "600")
 try:
    DEFAULT_INPUT_WAIT_TIMEOUT_S: float = float(_INPUT_WAIT_TIMEOUT_RAW)
 except ValueError:
    logger.warning(
-        "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to 600 seconds.",
+        "Invalid VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=%r; falling back to %s seconds.",
        _INPUT_WAIT_TIMEOUT_RAW,
+        _INPUT_WAIT_TIMEOUT_DEFAULT,
+    )
+    DEFAULT_INPUT_WAIT_TIMEOUT_S = _INPUT_WAIT_TIMEOUT_DEFAULT
+
+if not math.isfinite(DEFAULT_INPUT_WAIT_TIMEOUT_S):
+    # nan compares false against everything, so every deadline check silently
+    # passes; inf makes the deadline unreachable. Both disable the net while
+    # looking like a number, which is the failure this item exists to stop.
+    raise ValueError(
+        f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is not finite. "
+        "Use a positive number of seconds, or exactly 0 to disable the deadline."
+    )
+if DEFAULT_INPUT_WAIT_TIMEOUT_S < 0:
+    # Fail at startup rather than substituting a default. `-1` is a common idiom
+    # for "no limit", so silently turning it into 600 would give an operator who
+    # meant "never time out" mysterious failures ten minutes in.
+    raise ValueError(
+        f"VLLM_OMNI_INPUT_WAIT_TIMEOUT_S={_INPUT_WAIT_TIMEOUT_RAW!r} is negative. "
+        "Use a positive number of seconds, or exactly 0 to disable the deadline."
+    )
+elif DEFAULT_INPUT_WAIT_TIMEOUT_S == 0:
+    # Zero stays a supported opt-out, but it is no longer silent: it disables
+    # the deadline on BOTH transports, so a request whose stage input never
+    # arrives waits forever. Say which nets just went away.
+    logger.warning(
+        "VLLM_OMNI_INPUT_WAIT_TIMEOUT_S=0: stage-input deadlines are DISABLED for "
+        "both the full-payload path (WAITING_FOR_INPUT) and the async-chunk path "
+        "(WAITING_FOR_CHUNK). A request whose input never arrives will wait "
+        "indefinitely instead of failing."
    )
-    DEFAULT_INPUT_WAIT_TIMEOUT_S = 600.0


 class OmniSchedulerMixin:
@@ -184,6 +216,8 @@ class OmniSchedulerMixin:
                self.running,
                scheduler_requests=self.requests,
            )
+            self._process_pending_chunk_timeouts()
+            self._log_failed_chunk_sends()

    def _restore_omni_wait_queues(self) -> None:
        """Restore requests temporarily parked by Omni input gates."""
@@ -211,9 +245,9 @@ class OmniSchedulerMixin:
        Disabled when ``DEFAULT_INPUT_WAIT_TIMEOUT_S`` is <= 0.

        Scope: only covers ``input_coordinator`` (full-payload path).
-        Async-chunk requests park in ``chunk_transfer_adapter`` instead
-        and are not handled here -- if a similar safety net is needed
-        for the chunk path, it belongs in the chunk adapter.
+        Async-chunk requests park in ``chunk_transfer_adapter`` instead and
+        are handled by ``_process_pending_chunk_timeouts`` below, which shares
+        this timeout.
        """
        if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0:
            return
@@ -234,6 +268,84 @@ class OmniSchedulerMixin:
        )
        self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR)

+    def _log_failed_chunk_sends(self) -> None:
+        """Surface chunks the sender gave up on (R1.2 of #4855).
+
+        ``save_loop`` pops a task with ``popleft()`` and never re-queues it, and
+        ``connector.put`` can report failure without raising, so a dropped chunk
+        is a give-up rather than a retry. Until this change the only trace was a
+        warning that printed ``None`` for the request id, because the task dict
+        has no ``request_id`` key.
+
+        This only reports. Failing the request from here does not work: the
+        producer is not the ``final_output`` stage, so an abort synthesized by
+        ``finish_requests`` is not turned into a client-visible termination by
+        ``Orchestrator._route_output`` (see the ``final_output`` guards there).
+        Making a producer-side drop client-visible needs an orchestrator-side
+        path like ``_fail_request_dead_stage``; the consumer's R1.1 deadline is
+        what actually ends the request today.
+        """
+        adapter = getattr(self, "chunk_transfer_adapter", None)
+        if adapter is None:
+            return
+        failures = adapter.collect_failed_send_request_ids()
+        for request_id, reason in failures.items():
+            if request_id in self.requests:
+                logger.error(
+                    "[OmniScheduler] req=%s: chunk send gave up (%s); the consumer will "
+                    "fail on the stage-input deadline",
+                    request_id,
+                    reason,
+                )
+            else:
+                # Drained but no longer scheduled here: the request finished or
+                # was aborted between the failed send and this sweep. Still say
+                # so -- collect_* empties the map, so staying quiet would make
+                # this the one drop path in a change whose point is visibility.
+                logger.warning(
+                    "[OmniScheduler] req=%s: chunk send gave up (%s) after the request "
+                    "left this scheduler (finished or aborted)",
+                    request_id,
+                    reason,
+                )
+
+    def _process_pending_chunk_timeouts(self) -> None:
+        """Force-fail requests stalled in ``WAITING_FOR_CHUNK`` too long.
+
+        The async-chunk counterpart of ``_process_pending_input_timeouts``,
+        called right after ``chunk_transfer_adapter.process_pending_chunks``.
+        Until now the chunk path had no safety net at all: the scheduler-side
+        comment on the full-payload net said a similar guard "belongs in the
+        chunk adapter", while the worker mixin assumed the full-payload net
+        already covered it.  Only the first was true, so a dropped terminal
+        chunk or a producer that exhausted its send retries parked the request
+        forever (vllm-project/vllm-omni#3833).  Since ``async_chunk: true`` is
+        the default for the flagship TTS pipelines, that was the default
+        streaming path.
+
+        Shares ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S`` with the full-payload net:
+        one knob for "how long may a request wait for stage input", regardless
+        of which transport carries it.  Disabled when the value is <= 0.
+        """
+        if DEFAULT_INPUT_WAIT_TIMEOUT_S <= 0:
+            return
+        adapter = getattr(self, "chunk_transfer_adapter", None)
+        if adapter is None or not getattr(adapter, "receives_chunks", False):
+            return
+        timed_out_ids = adapter.collect_timed_out_request_ids(timeout_s=DEFAULT_INPUT_WAIT_TIMEOUT_S)
+        if not timed_out_ids:
+            return
+        present_ids = {req_id for req_id in timed_out_ids if req_id in self.requests}
+        if not present_ids:
+            return
+        logger.warning(
+            "Marking %d request(s) as FINISHED_ERROR after stalling > %.0fs waiting for a chunk: %s",
+            len(present_ids),
+            DEFAULT_INPUT_WAIT_TIMEOUT_S,
+            sorted(present_ids),
+        )
+        self.finish_requests(present_ids, RequestStatus.FINISHED_ERROR)
+
    def _capture_omni_connector_output(self, model_runner_output: Any) -> None:
        """Stash the model runner's omni_connector_output for next schedule().

@@ -35,6 +35,8 @@ class OmniTransferAdapterBase:
        self.stop_event = threading.Event()
        self._recv_cond = threading.Condition()
        self._save_cond = threading.Condition()
+        self._send_failures: dict[str, str] = {}
+        self._send_failure_lock = threading.Lock()

        self.recv_thread = threading.Thread(target=self.recv_loop, daemon=True)
        self.recv_thread.start()
@@ -82,6 +84,24 @@ class OmniTransferAdapterBase:
                elif not any_success and not self.stop_event.is_set():
                    self._recv_cond.wait(timeout=0.001)

+    def record_send_failure(self, request_id: str | None, reason: str) -> None:
+        """Note that a chunk for *request_id* will never be delivered.
+
+        Called from the sender thread; drained by the scheduler thread, hence
+        the lock. ``_waiting_since`` needs none because only the scheduler
+        touches it.
+        """
+        if not request_id:
+            return
+        with self._send_failure_lock:
+            self._send_failures.setdefault(request_id, reason)
+
+    def collect_failed_send_request_ids(self) -> dict[str, str]:
+        """Drain and return the requests whose send gave up."""
+        with self._send_failure_lock:
+            failures, self._send_failures = self._send_failures, {}
+        return failures
+
    def save_loop(self):
        """Loop to send outgoing data."""
        while not self.stop_event.is_set():
@@ -90,7 +110,18 @@ class OmniTransferAdapterBase:
                try:
                    self._send_single_request(task)
                except Exception as e:
-                    logger.warning(f"Error saving data for {task.get('request_id')}: {e}")
+                    # R1.2 of #4855. The task was popleft()-ed and is not re-queued,
+                    # so this is a give-up, not a retry: the consumer will wait for a
+                    # chunk that is never coming. Record it so the scheduler can fail
+                    # the request now instead of leaving it to the wait deadline.
+                    # The task dict carries the Request object, not a bare id --
+                    # `task.get("request_id")` is always None (it was that way in the
+                    # original warning too, which logged "None" for every failure).
+                    # Use the scheduler-side id: that is what `self.requests` is keyed
+                    # by, and what `finish_requests` expects.
+                    failed = getattr(task.get("request"), "request_id", None)
+                    logger.error("Send gave up for %s: %s", failed, e)
+                    self.record_send_failure(failed, f"{type(e).__name__}: {e}")

            with self._save_cond:
                if not self._pending_save_reqs and not self.stop_event.is_set():
@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project

 import importlib
+import time
 from collections import defaultdict, deque
 from collections.abc import Callable, Mapping
 from typing import Any
@@ -94,6 +95,12 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
        self._held_non_active: deque[Any] = deque()
        self.requests_num_chunks_sent: dict[str, int] = defaultdict(int)
        self._pending_streaming_prefills: dict[str, dict] = {}
+        # Monotonic timestamp of when each request last began waiting for a
+        # chunk, refreshed every time one arrives.  Read by
+        # collect_timed_out_request_ids() so a stream that stops advancing
+        # becomes a client-visible error instead of parking forever.  Mirrors
+        # OmniSchedulingCoordinator._waiting_since on the full-payload path.
+        self._waiting_since: dict[str, float] = {}

    @staticmethod
    def _is_truthy_scalar(value: Any) -> bool:
@@ -383,6 +390,24 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
            # has been sent successfully. This avoids cleanup->save races.
            if is_payload_finished:
                self.cleanup(request.request_id, external_req_id)
+        else:
+            # R1.2 of #4855. connector.put returning False is a silent drop: no
+            # exception, no retry, and the caller's `if success:` block simply
+            # does not run. /dev/shm exhaustion in SharedMemoryConnector.put is
+            # one way to get here. Record it so the request fails now rather
+            # than parking in WAITING_FOR_CHUNK until the deadline.
+            logger.error(
+                "Chunk send failed for %s (stage %s -> %s); giving up on this chunk",
+                external_req_id,
+                stage_id,
+                next_stage_id,
+            )
+            # Key on the scheduler-side id. `external_req_id` is the user-facing id
+            # (InputProcessor renames request_id to an internal UUID and keeps the
+            # original in external_req_id, see async_omni_engine.py), while
+            # `self.requests` -- and therefore `finish_requests` -- is keyed by the
+            # internal one.
+            self.record_send_failure(request.request_id, "connector.put reported failure")

        if is_segment_finished:
            self.code_prompt_token_ids.pop(external_req_id, None)
@@ -438,6 +463,7 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):

        self._cancelled_load_reqs.add(request_id)
        self._finished_load_reqs.discard(request_id)
+        self._waiting_since.pop(request_id, None)

    @staticmethod
    def _discard_from_chunk_deque(deque_list: deque[Any], request_id: str) -> None:
@@ -579,6 +605,42 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
        self._active_streams[request_id] = request
        return True

+    def collect_timed_out_request_ids(self, timeout_s: float) -> set[str]:
+        """Return IDs whose chunk wait has exceeded *timeout_s*.
+
+        The async-chunk path had no deadline of any kind: a request parks in
+        ``WAITING_FOR_CHUNK`` and the receiver re-queues it on every failed
+        poll (``transfer_adapter/base.py``) with no attempt counter, so a
+        producer that crashed, gave up after its send retries, or simply never
+        emitted a terminal chunk left the request waiting forever
+        (vllm-project/vllm-omni#3833).  The full-payload path has had a net
+        since ``OmniSchedulingCoordinator.collect_timed_out_request_ids``; this
+        is its async-chunk counterpart, and both are driven by the same
+        ``VLLM_OMNI_INPUT_WAIT_TIMEOUT_S``.
+
+        The clock measures *stall* time, not stream lifetime: it starts when a
+        request begins waiting for a chunk and resets each time one arrives, so
+        a long but healthy stream is never failed.
+
+        Clears ``_waiting_since`` for the expired IDs.  The caller marks them
+        ``FINISHED_ERROR`` via the scheduler's ``finish_requests``, which routes
+        back through this adapter's own ``finish_requests`` and releases the
+        rest of the per-request state.
+        """
+        if timeout_s <= 0 or not self._waiting_since:
+            return set()
+        now = time.monotonic()
+        timed_out_ids = {req_id for req_id, started in self._waiting_since.items() if now - started > timeout_s}
+        for req_id in timed_out_ids:
+            self._waiting_since.pop(req_id, None)
+            logger.warning(
+                "[Stage-%s] Request %s timed out waiting for a chunk (stalled > %.0fs)",
+                self.connector.stage_id,
+                req_id,
+                timeout_s,
+            )
+        return timed_out_ids
+
    @property
    def num_running_waiting_for_chunk(self) -> int:
        """Count running requests temporarily removed while awaiting a chunk."""
@@ -624,11 +686,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
                # Requests that waiting for chunk
                self.load_async(request)
                request.status = RequestStatus.WAITING_FOR_CHUNK
+                self._waiting_since.setdefault(request.request_id, time.monotonic())
            else:
                if request.request_id in finished_load_reqs:
                    request.status = target_status
                    finished_load_reqs.remove(request.request_id)
                    self.requests_with_ready_chunks.add(request.request_id)
+                    # A chunk landed: restart the clock for the next one, so the
+                    # deadline measures stall time rather than total stream time.
+                    self._waiting_since.pop(request.request_id, None)
                    continue
            queue.remove(request)
            self.requests_origin_status[request.request_id] = target_status
@@ -767,11 +833,15 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
                # Requests that waiting for chunk
                self.load_async(request)
                request.status = RequestStatus.WAITING_FOR_CHUNK
+                self._waiting_since.setdefault(request.request_id, time.monotonic())
            else:
                if request.request_id in finished_load_reqs:
                    request.status = target_status
                    finished_load_reqs.remove(request.request_id)
                    self.requests_with_ready_chunks.add(request.request_id)
+                    # A chunk landed: restart the clock for the next one, so the
+                    # deadline measures stall time rather than total stream time.
+                    self._waiting_since.pop(request.request_id, None)
                    continue
            queue.remove(request)
            self.requests_origin_status[request.request_id] = target_status
@@ -826,5 +896,6 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
            self.upstream_exhausted_requests.discard(req_id)
            self._finished_load_reqs.discard(req_id)
            self._cancelled_load_reqs.add(req_id)
+            self._waiting_since.pop(req_id, None)

        return []
@@ -15,6 +15,7 @@ import asyncio
 import time as _time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
+from http import HTTPStatus
 from typing import TYPE_CHECKING, Any

 import janus
@@ -51,6 +52,7 @@ from vllm_omni.engine.messages import (
 from vllm_omni.engine.orchestrator_monitor import create_orch_monitor, replica_key
 from vllm_omni.engine.serialization import serialize_additional_information
 from vllm_omni.engine.stage_pool import StagePool, StageUnavailableError
+from vllm_omni.errors import DEFAULT_CLIENT_ERROR_TYPE
 from vllm_omni.metrics.prometheus import OmniRequestCounter
 from vllm_omni.metrics.stat_logger import OmniPrometheusStatLogger
 from vllm_omni.outputs import OmniRequestOutput
@@ -221,7 +223,7 @@ class _OrchestratorDuplexStagePort:
        async_chunk: bool,
        prewarm_async_chunk_stages: Callable[
            [str, Any, OrchestratorRequestState],
-            Awaitable[None],
+            Awaitable[bool],
        ],
    ) -> None:
        self._stage_pools = stage_pools
@@ -307,11 +309,23 @@ class _OrchestratorDuplexStagePort:
        else:
            replica_id = await pool.submit_initial(context.request_id, request_state, request, prompt_text=None)
            if self._async_chunk and context.stage_id == 0:
-                await self._prewarm_async_chunk_stages(
+                prewarmed = await self._prewarm_async_chunk_stages(
                    context.request_id,
                    request,
                    request_state,
                )
+                if not prewarmed:
+                    # The prewarm already failed the request, aborted it, and
+                    # popped its state. Falling through would write fences and
+                    # a submit timestamp onto that orphaned object and
+                    # re-increment the running counter the cleanup just
+                    # released, then hand the control plane a success result
+                    # for a request that no longer exists. Raise instead:
+                    # ``DuplexControlPlane.handle_append`` turns this into an
+                    # error result for the append.
+                    raise RuntimeError(
+                        f"async-chunk prewarm failed for duplex request {context.request_id}; the request was aborted"
+                    )
        request_state.duplex_stage_fences[context.stage_id] = context.fence
        request_state.stage_submit_ts[context.stage_id] = _time.time()
        if not request_state.running_counter_registered and self._running_counter is not None:
@@ -2217,18 +2231,54 @@ class Orchestrator:
        request_id: str,
        stage0_request: Any,
        req_state: OrchestratorRequestState,
-    ) -> None:
-        """Pre-submit downstream stages for async-chunk mode."""
+    ) -> bool:
+        """Pre-submit downstream stages for async-chunk mode.
+
+        Returns False when the request was failed and cleaned up in here, so a
+        caller still holding ``req_state`` stops instead of recording state on
+        an object the cleanup already popped from ``request_states``.
+        """
        if req_state.final_stage_id <= 0:
-            return
+            return True

        prompt_token_ids = getattr(stage0_request, "prompt_token_ids", None)
        if prompt_token_ids is None:
-            logger.warning(
-                "[Orchestrator] async_chunk prewarm skipped for req=%s: stage0 prompt_token_ids missing",
+            # R1.3 of #4855. Skipping the prewarm leaves every downstream stage
+            # unsubmitted, so the request produces no output and no error -- it
+            # just stops. An embeds-only prompt in async-chunk mode is a config
+            # mistake, and the client should hear about it now rather than wait
+            # for the stage-input deadline to notice nothing ever arrived.
+            #
+            # Non-fatal, with a 4xx: `fatal=True` means "the engine is dead", and
+            # the entrypoints act on it -- `OmniBase._handle_output_message`
+            # raises `OmniEngineDeadError` out of `OmniLLM.generate`'s batch loop,
+            # failing every other request in that batch, and the serving layer
+            # calls `terminate_if_errored`. One client's bad prompt must not do
+            # either. This is the same shape `_handle_stage_error` uses for a
+            # request-scoped client error.
+            logger.error(
+                "[Orchestrator] req=%s: async_chunk prewarm needs stage0 prompt_token_ids "
+                "and none were provided; failing the request",
                request_id,
            )
-            return
+            await self.output_async_queue.put(
+                ErrorMessage(
+                    error=(
+                        "async_chunk requires prompt_token_ids on the stage-0 request; "
+                        "an embeds-only prompt cannot prewarm downstream stages"
+                    ),
+                    status_code=HTTPStatus.BAD_REQUEST.value,
+                    error_type=DEFAULT_CLIENT_ERROR_TYPE,
+                    request_id=request_id,
+                    stage_id=0,
+                )
+            )
+            await self._cleanup_request_ids(
+                [request_id, *self._cfg_tracker.cleanup_parent(request_id)],
+                abort=True,
+                close_duplex_sessions=True,
+            )
+            return False

        for next_stage_id in range(1, req_state.final_stage_id + 1):
            next_pool = self.stage_pools[next_stage_id]
@@ -2303,7 +2353,7 @@ class Orchestrator:
            # Attribute the failure to the stage that failed, not stage 0; the
            # request is already cleaned up, so stop prewarming.
            if not submitted:
-                return
+                return False

            # The guard swallows submit_initial's return value; the pool binding
            # it recorded carries the same replica.
@@ -2329,6 +2379,8 @@ class Orchestrator:
                tx_ms=_tx_ms,
            )

+        return True
+
    def _build_kv_sender_info(
        self,
        sender_stage_ids: list[int],
@@ -92,19 +92,7 @@ logger = init_logger(__name__)
 # reference (issue #4644). Matches the offline example/test and upstream demo.
 _COSYVOICE3_PROMPT_DELIMITER = "<|endofprompt|>"
 _COSYVOICE3_PROMPT_PREFIX = f"You are a helpful assistant.{_COSYVOICE3_PROMPT_DELIMITER}"
-_SAMPLING_MAX_TOKENS_TTS_MODEL_TYPES = {
-    "fish_tts",
-    "qwen3_tts",
-    "voxtral_tts",
-    "cosyvoice3",
-    "voxcpm2",
-    "higgs_audio_v2",
-    "higgs_audio_v3",
-    "indextts2",
-    "indextts2_5",
-    "audex",
-    "audex_tta",
-}
+
 # Audex contract: zero-codec / invalid generations arrive as empty terminal
 # payloads and must fail the request, never serialize as a successful empty
 # WAV. Covers both the TTS ("audex") and TTA ("audex_tta") pipelines.
@@ -444,7 +432,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
        # Cached per process: the CosyVoice3 Qwen tokenizer + resolved model
        # path used for dynamic-token sizing. Without this, every request
        # re-ran snapshot_download + reloaded the tokenizer (~100 ms on the
-        # TTFP critical path) in _apply_cosyvoice3_dynamic_tokens.
+        # TTFP critical path) in the CosyVoice3 adapter's sampling override.
        self._cosyvoice3_tokenizer = None

        # Determine TTS model type or None
@@ -502,9 +490,8 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):

        # Resolve the per-model serving adapter (RFC #4327), keyed on the
        # detected model-type. Every dedicated TTS model has an adapter; the
-        # adapter owns request validation + prompt/param building. Sampling
-        # overrides and the model-type label remain in the orchestrator tail
-        # (keyed on ``_tts_model_type``) during this incremental migration.
+        # adapter owns request validation, prompt/param building, and sampling
+        # overrides; the model-type label remains available for compatibility.
        self._adapter = None
        if self._tts_stage is not None:
            adapter_cls = resolve_adapter(self._tts_model_type)
@@ -2825,154 +2812,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
            self._audex_tokenizer = AutoTokenizer.from_pretrained(os.path.join(root, folder))
        return self._audex_tokenizer

-    def _inject_audex_tta_args(self, request_id: str, prompt: Any, stage0_params: Any) -> Any:
-        """Complete the Audex TTA contract on a CLONE of the stage-0 params.
-
-        Always attaches the RVQ phase-mask contract (required for token
-        validity) and the CFG pair contract at the official default scale
-        3.0 unless the caller passed ``cfg_scale`` (1.0 disables guidance).
-
-        Returns the cloned params; the input is never mutated. On the
-        no-extra_params online path the input is an element of the engine's
-        SHARED ``default_sampling_params_list`` — writing per-request pair
-        state (``cfg_pair_id``/``cfg_null_prompt``) into it would race
-        concurrent requests and leak stale CFG metadata into later ones.
-        """
-        import copy
-
-        from vllm_omni.model_executor.models.audex.prompt import build_tta_null_prompt
-        from vllm_omni.model_executor.models.audex.tta import build_tta_phase_token_ids
-
-        cond_prompt = prompt.get("prompt") if isinstance(prompt, dict) else prompt
-        if not isinstance(cond_prompt, str) or not cond_prompt:
-            raise ValueError("Audex TTA requires the adapter-built caption prompt")
-
-        stage0_params = copy.deepcopy(stage0_params)
-        tokenizer = self._get_audex_tokenizer()
-        if self._audex_tta_rvq is None:
-            phase_token_ids, start_tid, end_tid = build_tta_phase_token_ids(tokenizer)
-            self._audex_tta_rvq = {
-                "phase_token_ids": phase_token_ids,
-                "start_tid": start_tid,
-                "end_tid": end_tid,
-                # Official generation cap (decode truncates at 500 frames).
-                "codec_cap": 4000,
-                # The TTA prompt ends with <audiogen_start>.
-                "start_in_prompt": True,
-            }
-
-        extra_args = getattr(stage0_params, "extra_args", None)
-        if extra_args is None:
-            extra_args = {}
-            stage0_params.extra_args = extra_args
-        extra_args["tta_rvq"] = self._audex_tta_rvq
-
-        cfg_scale = extra_args.get("cfg_scale")
-        cfg_scale = 3.0 if cfg_scale is None else float(cfg_scale)
-        if cfg_scale <= 1.0:
-            extra_args.pop("cfg_scale", None)
-            return stage0_params
-        extra_args.update(
-            {
-                "cfg_scale": cfg_scale,
-                "cfg_role": "cond",
-                "cfg_pair_id": request_id,
-                "cfg_null_prompt": build_tta_null_prompt(cond_prompt, tokenizer),
-            }
-        )
-        return stage0_params
-
-    def _inject_audex_cfg_pair_args(self, request_id: str, prompt: Any, stage0_params: Any) -> Any:
-        """Complete the Audex CFG contract on a CLONE of the stage-0 params.
-
-        The adapter validated ``cfg_scale``; here (where the final request id
-        exists) the cond role, pair id, and the length-matched null prompt are
-        attached. ``cfg_scale`` absent/1.0 returns a clone value-identical to
-        the non-CFG flow (the key is dropped rather than forwarded).
-
-        Returns the cloned params; the input is never mutated (it may be an
-        element of the engine's shared ``default_sampling_params_list`` —
-        see ``_inject_audex_tta_args``).
-        """
-        import copy
-
-        stage0_params = copy.deepcopy(stage0_params)
-        extra_args = getattr(stage0_params, "extra_args", None)
-        if not extra_args or extra_args.get("cfg_scale") is None:
-            if extra_args:
-                extra_args.pop("cfg_scale", None)
-            return stage0_params
-        cfg_scale = float(extra_args["cfg_scale"])
-        if cfg_scale <= 1.0:
-            extra_args.pop("cfg_scale", None)
-            return stage0_params
-
-        from vllm_omni.model_executor.models.audex.prompt import build_null_prompt
-
-        cond_prompt = prompt.get("prompt") if isinstance(prompt, dict) else prompt
-        if not isinstance(cond_prompt, str) or not cond_prompt:
-            raise ValueError("Audex CFG requires the adapter-built text prompt")
-        null_prompt = build_null_prompt(cond_prompt, self._get_audex_tokenizer())
-        extra_args.update(
-            {
-                "cfg_scale": cfg_scale,
-                "cfg_role": "cond",
-                "cfg_pair_id": request_id,
-                "cfg_null_prompt": null_prompt,
-            }
-        )
-        # Guided decoding sharpens the distribution; the unguided default
-        # temperature (0.1) adds excess sampling noise under CFG. Measured
-        # on the en-24 gate corpus at cfg 1.5: temp 0.1 -> CER 7.31%,
-        # temp 0.05 -> CER 6.87% (unguided baseline 7.24%).
-        stage0_params.temperature = 0.05
-        return stage0_params
-
-    def _apply_cosyvoice3_dynamic_tokens(
-        self,
-        sampling_params_list: list,
-        request: OpenAICreateSpeechRequest,
-    ) -> list:
-        """Set min/max tokens from tokenized text length (ratios target tokens, not chars)."""
-        import copy
-
-        from vllm_omni.model_executor.models.cosyvoice3.tokenizer import get_qwen_tokenizer
-        from vllm_omni.model_executor.models.cosyvoice3.utils import extract_text_token
-
-        sampling_params_list = copy.deepcopy(sampling_params_list)
-        hf_cfg = self.model_config.hf_config
-        # Build the Qwen tokenizer once per process (resolving the model dir via
-        # snapshot_download at most once) and reuse it across requests.
-        tokenizer = self._cosyvoice3_tokenizer
-        if tokenizer is None:
-            model_path = self.engine_client.model_config.model
-            if not os.path.isdir(model_path):
-                from huggingface_hub import snapshot_download
-
-                model_path = snapshot_download(model_path)
-            tokenizer = get_qwen_tokenizer(
-                token_path=os.path.join(model_path, hf_cfg.qwen_pretrain_path),
-                skip_special_tokens=hf_cfg.skip_special_tokens,
-                version=hf_cfg.version,
-            )
-            self._cosyvoice3_tokenizer = tokenizer
-        _, text_token_len = extract_text_token(
-            request.input,
-            tokenizer,
-            hf_cfg.allowed_special,
-        )
-        min_ratio = getattr(hf_cfg, "min_token_text_ratio", 2)
-        max_ratio = getattr(hf_cfg, "max_token_text_ratio", 20)
-        sampling_params_list[0].min_tokens = max(1, int(text_token_len * min_ratio))
-        sampling_params_list[0].max_tokens = min(2048, int(text_token_len * max_ratio))
-        logger.info(
-            "CosyVoice3 dynamic tokens: text_tokens=%d, min_tokens=%d, max_tokens=%d",
-            text_token_len,
-            sampling_params_list[0].min_tokens,
-            sampling_params_list[0].max_tokens,
-        )
-        return sampling_params_list
-
    # ---- GLM-TTS helpers ----

    async def _build_glm_tts_prompt(
@@ -3172,9 +3011,8 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):

        # Build prompt + tts_params via the per-model adapter (RFC #4327). Every
        # dedicated TTS model resolves to an adapter that owns its validation,
-        # uploaded-speaker handling, and prompt/param building. Sampling
-        # overrides and the model-type label remain in the orchestrator tail
-        # below (keyed on ``_tts_model_type``) during this incremental migration.
+        # uploaded-speaker handling, prompt/param building, and sampling
+        # overrides. The model-type label remains available for compatibility.
        # Non-TTS deployments (no adapter) fall through to the rejection below.
        # Capture inline-ref-audio status BEFORE validate(): several adapters
        # apply uploaded speakers inside validate(), which sets request.ref_audio
@@ -3225,63 +3063,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
            model_type,
        )

-        # CosyVoice3: set dynamic min/max tokens based on text length.
-        # The official model requires min_token_text_ratio to prevent early
-        # EOS and max_token_text_ratio to cap generation length.
-        if self._tts_model_type == "cosyvoice3" and sampling_params_list:
-            sampling_params_list = self._apply_cosyvoice3_dynamic_tokens(sampling_params_list, request)
-
-        # GLM-TTS: set dynamic min/max tokens based on text length.
-        if self._tts_model_type == "glm_tts" and sampling_params_list:
-            import copy
-
-            sampling_params_list = copy.deepcopy(sampling_params_list)
-            glm_metadata = prompt.get("additional_information") if isinstance(prompt, dict) else None
-            text_len_value = None
-            if isinstance(glm_metadata, dict):
-                text_len_value = glm_metadata.get("glm_tts_text_token_len")
-                if isinstance(text_len_value, list) and text_len_value:
-                    text_len_value = text_len_value[0]
-            text_token_len = (
-                int(text_len_value)
-                if text_len_value is not None
-                else self._estimate_glm_tts_text_token_len(request.input)
-            )
-            hf_cfg = self.model_config.hf_config
-            min_ratio = getattr(hf_cfg, "min_token_text_ratio", 2)
-            max_ratio = getattr(hf_cfg, "max_token_text_ratio", 20)
-            stage_min_tokens = getattr(sampling_params_list[0], "min_tokens", None)
-            stage_max_tokens = getattr(sampling_params_list[0], "max_tokens", None)
-            cap_candidates = [int(cap) for cap in (stage_max_tokens, request.max_new_tokens) if cap is not None]
-            hard_cap = min(cap_candidates) if cap_candidates else None
-
-            min_tokens = max(1, int(text_token_len * min_ratio))
-            if stage_min_tokens is not None:
-                min_tokens = max(min_tokens, int(stage_min_tokens))
-            if hard_cap is not None:
-                min_tokens = min(min_tokens, hard_cap)
-
-            max_tokens = max(min_tokens, int(text_token_len * max_ratio))
-            if hard_cap is not None:
-                max_tokens = min(max_tokens, hard_cap)
-            sampling_params_list[0].min_tokens = min_tokens
-            sampling_params_list[0].max_tokens = max_tokens
-            seed = getattr(request, "seed", None)
-            if seed is not None:
-                sampling_params_list[0].seed = seed
-            logger.info(
-                "GLM-TTS dynamic tokens: text_tokens=%d, min_ratio=%s, max_ratio=%s, "
-                "stage_min=%s, stage_max=%s, request_max=%s, min_tokens=%d, max_tokens=%d",
-                text_token_len,
-                min_ratio,
-                max_ratio,
-                stage_min_tokens,
-                stage_max_tokens,
-                request.max_new_tokens,
-                min_tokens,
-                max_tokens,
-            )
-
        # Apply model-specific extra parameters
        if request.extra_params is not None and sampling_params_list:
            if not isinstance(request.extra_params, dict):
@@ -3300,52 +3081,10 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
            sampling_params_list[0].extra_args.update(request.extra_params)
            logger.info("Applied extra_params: %s", request.extra_params)

-        # Audex CFG: turn an adapter-validated cfg_scale into the engine-side
-        # pair contract. This must run here (not in the adapter) because the
-        # pair id is the final request_id, which the adapter never sees.
-        # The injectors return a CLONE: without extra_params above,
-        # sampling_params_list still holds the engine's shared default params,
-        # which must never carry per-request CFG state.
-        if self._tts_model_type == "audex" and sampling_params_list:
-            sampling_params_list[0] = self._inject_audex_cfg_pair_args(request_id, prompt, sampling_params_list[0])
-        elif self._tts_model_type == "audex_tta" and sampling_params_list:
-            sampling_params_list[0] = self._inject_audex_tta_args(request_id, prompt, sampling_params_list[0])
-
-        # Some TTS model defaults come from deploy YAML. Their AR
-        # generation length is controlled by SamplingParams.max_tokens, so only
-        # override it when the caller explicitly requests max_new_tokens.
-        if (
-            self._tts_model_type in _SAMPLING_MAX_TOKENS_TTS_MODEL_TYPES
-            and request.max_new_tokens is not None
-            and sampling_params_list
-        ):
-            import copy
-
-            sampling_params_list = copy.deepcopy(sampling_params_list)
-            sampling_params_list[0].max_tokens = request.max_new_tokens
-            if self._tts_model_type == "cosyvoice3":
-                sampling_params_list[0].min_tokens = min(
-                    getattr(sampling_params_list[0], "min_tokens", 0),
-                    request.max_new_tokens,
-                )
-        elif self._tts_model_type == "ming_tts" and sampling_params_list:
-            import copy
-
-            from vllm_omni.model_executor.models.ming_tts.config_ming_tts import (
-                MOE_TEXT_EOS_TOKEN_ID,
-                TEXT_EOS_TOKEN_ID,
-            )
-
-            hf_config = self.engine_client.model_config.hf_config
-            is_moe = getattr(hf_config, "model_type", "") == "bailingmm"
-            stop_token_id = MOE_TEXT_EOS_TOKEN_ID if is_moe else TEXT_EOS_TOKEN_ID
-
-            sampling_params_list = copy.deepcopy(sampling_params_list)
-            sampling_params_list[0].stop_token_ids = [int(stop_token_id)]
-            if request.max_new_tokens is not None:
-                # Ming emits TEXT_EOS after the latent decode budget is exhausted, so
-                # Stage-0 needs one extra token beyond ming_max_decode_steps.
-                sampling_params_list[0].max_tokens = int(request.max_new_tokens) + 1
+        # Apply adapter-owned sampling overrides, including request-level token
+        # limits and model-specific dynamic token or stop-token configuration.
+        if sampling_params_list and (adapter := self._get_tts_adapter()) is not None:
+            sampling_params_list = adapter.apply_sampling_overrides(sampling_params_list, request, prompt, request_id)

        if request.seed is not None and sampling_params_list:
            import copy
@@ -9,8 +9,6 @@ key alone cannot distinguish.
 That metadata is the single source of truth. :func:`detect_tts_model_type` and
 :func:`all_tts_stage_keys` derive the whole stage -> model-type mapping from it,
 so a new TTS model is one adapter file and needs no edit to ``serving_speech.py``.
-Models still awaiting an adapter are listed explicitly in
-:data:`LEGACY_TTS_DETECTORS`.
 """

 from vllm.logger import init_logger
@@ -18,7 +16,6 @@ from vllm.logger import init_logger
 from vllm_omni.entrypoints.openai.tts_adapters.base import (
    ARTTSAdapter,
    DiffusionTTSAdapter,
-    LegacyDetector,
    OutputPolicy,
    PreparedRequest,
    SpeechServingContext,
@@ -29,10 +26,6 @@ logger = init_logger(__name__)

 TTS_ADAPTER_REGISTRY: dict[str, type[TTSModelAdapter]] = {}

-#: Model types the serving layer detects that have no adapter yet. See
-#: :class:`LegacyDetector`. This list must only ever shrink.
-LEGACY_TTS_DETECTORS: tuple[LegacyDetector, ...] = ()
-

 def register_tts_adapter(cls: type[TTSModelAdapter]) -> type[TTSModelAdapter]:
    """Class decorator: index ``cls`` under its ``name`` (model-type)."""
@@ -57,16 +50,13 @@ def resolve_adapter(model_type: str | None) -> type[TTSModelAdapter] | None:
    return TTS_ADAPTER_REGISTRY.get(model_type)


-def iter_tts_detectors() -> list[type[TTSModelAdapter] | LegacyDetector]:
+def iter_tts_detectors() -> list[type[TTSModelAdapter]]:
    """Every detector, in resolution order.

    Sorted by ``detect_priority`` then ``name``, so the order is total and does
    not depend on adapter import order.
    """
-    detectors: list[type[TTSModelAdapter] | LegacyDetector] = [
-        *TTS_ADAPTER_REGISTRY.values(),
-        *LEGACY_TTS_DETECTORS,
-    ]
+    detectors = list(TTS_ADAPTER_REGISTRY.values())
    detectors.sort(key=lambda d: (d.detect_priority, d.name))
    return detectors

@@ -132,12 +122,10 @@ from vllm_omni.entrypoints.openai.tts_adapters import (  # noqa: E402,F401
 __all__ = [
    "ARTTSAdapter",
    "DiffusionTTSAdapter",
-    "LegacyDetector",
    "OutputPolicy",
    "PreparedRequest",
    "SpeechServingContext",
    "TTSModelAdapter",
-    "LEGACY_TTS_DETECTORS",
    "TTS_ADAPTER_REGISTRY",
    "all_tts_model_types",
    "all_tts_stage_keys",
@@ -1,10 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """Audex (Nemotron-Labs-Audex-2B) TTS serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens
 from vllm_omni.model_executor.models.audex.prompt import build_cond_prompt

 if TYPE_CHECKING:
@@ -17,8 +17,8 @@ if TYPE_CHECKING:
 AUDEX_CFG_SCALE_MIN = 1.0
 AUDEX_CFG_SCALE_MAX = 10.0

-# Internal CFG pair-plumbing keys; injected by the serving layer once the
-# request id exists, never accepted from callers.
+# Internal CFG pair-plumbing keys; injected by this adapter once it receives
+# the final request id, never accepted from callers.
 _AUDEX_INTERNAL_CFG_KEYS = ("cfg_role", "cfg_pair_id", "cfg_null_prompt")


@@ -74,3 +74,65 @@ class AudexAdapter(ARTTSAdapter):
    ) -> PreparedRequest:
        prompt = {"prompt": build_cond_prompt(request.input)}
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="audex")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        if request_id is None:
+            raise ValueError("Audex CFG requires a request id")
+        stage0_params = sampling_params_list[0]
+        sampling_params_list[0] = self._inject_audex_cfg_pair_args(
+            request_id=request_id, prompt=prompt, stage0_params=stage0_params
+        )
+        return apply_max_new_tokens(sampling_params_list, request)
+
+    def _inject_audex_cfg_pair_args(self, request_id: str, prompt: Any, stage0_params: Any) -> Any:
+        """Complete the Audex CFG contract on a CLONE of the stage-0 params.
+
+        The adapter validated ``cfg_scale``; here (where the final request id
+        exists) the cond role, pair id, and the length-matched null prompt are
+        attached. ``cfg_scale`` absent/1.0 returns a clone value-identical to
+        the non-CFG flow (the key is dropped rather than forwarded).
+
+        Returns the cloned params; the input is never mutated (it may be an
+        element of the engine's shared ``default_sampling_params_list`` —
+        see ``_inject_audex_tta_args``).
+        """
+        server = self.ctx.server
+        import copy
+
+        stage0_params = copy.deepcopy(stage0_params)
+        extra_args = getattr(stage0_params, "extra_args", None)
+        if not extra_args or extra_args.get("cfg_scale") is None:
+            if extra_args:
+                extra_args.pop("cfg_scale", None)
+            return stage0_params
+        cfg_scale = float(extra_args["cfg_scale"])
+        if cfg_scale <= 1.0:
+            extra_args.pop("cfg_scale", None)
+            return stage0_params
+
+        from vllm_omni.model_executor.models.audex.prompt import build_null_prompt
+
+        cond_prompt = prompt.get("prompt") if isinstance(prompt, dict) else prompt
+        if not isinstance(cond_prompt, str) or not cond_prompt:
+            raise ValueError("Audex CFG requires the adapter-built text prompt")
+        null_prompt = build_null_prompt(cond_prompt, server._get_audex_tokenizer())
+        extra_args.update(
+            {
+                "cfg_scale": cfg_scale,
+                "cfg_role": "cond",
+                "cfg_pair_id": request_id,
+                "cfg_null_prompt": null_prompt,
+            }
+        )
+        # Guided decoding sharpens the distribution; the unguided default
+        # temperature (0.1) adds excess sampling noise under CFG. Measured
+        # on the en-24 gate corpus at cfg 1.5: temp 0.1 -> CER 7.31%,
+        # temp 0.05 -> CER 6.87% (unguided baseline 7.24%).
+        stage0_params.temperature = 0.05
+        return stage0_params
@@ -2,16 +2,16 @@
 """Audex (Nemotron-Labs-Audex-2B) text-to-audio serving adapter.

 Caption → general audio through the ``audex_tta`` pipeline.
-The adapter builds the TTA prompt; the serving layer completes the request
-after the final request id exists: the RVQ phase-mask contract
+The adapter builds the TTA prompt and, after receiving the final request id,
+completes the sampling parameters with the RVQ phase-mask contract
 (``extra_args["tta_rvq"]``) and the CFG pair contract (official default
 scale 3.0 — guidance is effectively mandatory for TTA quality).
 """

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens
 from vllm_omni.model_executor.models.audex.prompt import build_tta_cond_prompt

 if TYPE_CHECKING:
@@ -21,8 +21,7 @@ AUDEX_TTA_DEFAULT_CFG_SCALE = 3.0
 AUDEX_TTA_CFG_SCALE_MIN = 1.0
 AUDEX_TTA_CFG_SCALE_MAX = 10.0

-# Internal plumbing keys; injected by the serving layer, never accepted
-# from callers.
+# Internal plumbing keys; injected by this adapter, never accepted from callers.
 _INTERNAL_KEYS = ("cfg_role", "cfg_pair_id", "cfg_null_prompt", "tta_rvq")


@@ -67,3 +66,76 @@ class AudexTTAAdapter(ARTTSAdapter):
    ) -> PreparedRequest:
        prompt = {"prompt": build_tta_cond_prompt(request.input)}
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="audex_tta")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        if request_id is None:
+            raise ValueError("Audex TTA requires a request id")
+        stage0_params = sampling_params_list[0]
+        sampling_params_list[0] = self._inject_audex_tta_args(
+            request_id=request_id, prompt=prompt, stage0_params=stage0_params
+        )
+        return apply_max_new_tokens(sampling_params_list, request)
+
+    def _inject_audex_tta_args(self, request_id: str, prompt: Any, stage0_params: Any) -> Any:
+        """Complete the Audex TTA contract on a CLONE of the stage-0 params.
+
+        Always attaches the RVQ phase-mask contract (required for token
+        validity) and the CFG pair contract at the official default scale
+        3.0 unless the caller passed ``cfg_scale`` (1.0 disables guidance).
+
+        Returns the cloned params; the input is never mutated. On the
+        no-extra_params online path the input is an element of the engine's
+        SHARED ``default_sampling_params_list`` — writing per-request pair
+        state (``cfg_pair_id``/``cfg_null_prompt``) into it would race
+        concurrent requests and leak stale CFG metadata into later ones.
+        """
+        server = self.ctx.server
+        import copy
+
+        from vllm_omni.model_executor.models.audex.prompt import build_tta_null_prompt
+        from vllm_omni.model_executor.models.audex.tta import build_tta_phase_token_ids
+
+        cond_prompt = prompt.get("prompt") if isinstance(prompt, dict) else prompt
+        if not isinstance(cond_prompt, str) or not cond_prompt:
+            raise ValueError("Audex TTA requires the adapter-built caption prompt")
+
+        stage0_params = copy.deepcopy(stage0_params)
+        tokenizer = server._get_audex_tokenizer()
+        if server._audex_tta_rvq is None:
+            phase_token_ids, start_tid, end_tid = build_tta_phase_token_ids(tokenizer)
+            server._audex_tta_rvq = {
+                "phase_token_ids": phase_token_ids,
+                "start_tid": start_tid,
+                "end_tid": end_tid,
+                # Official generation cap (decode truncates at 500 frames).
+                "codec_cap": 4000,
+                # The TTA prompt ends with <audiogen_start>.
+                "start_in_prompt": True,
+            }
+
+        extra_args = getattr(stage0_params, "extra_args", None)
+        if extra_args is None:
+            extra_args = {}
+            stage0_params.extra_args = extra_args
+        extra_args["tta_rvq"] = server._audex_tta_rvq
+
+        cfg_scale = extra_args.get("cfg_scale")
+        cfg_scale = 3.0 if cfg_scale is None else float(cfg_scale)
+        if cfg_scale <= 1.0:
+            extra_args.pop("cfg_scale", None)
+            return stage0_params
+        extra_args.update(
+            {
+                "cfg_scale": cfg_scale,
+                "cfg_role": "cond",
+                "cfg_pair_id": request_id,
+                "cfg_null_prompt": build_tta_null_prompt(cond_prompt, tokenizer),
+            }
+        )
+        return stage0_params
@@ -7,9 +7,7 @@ model's request normalization, validation, prompt/param building, sampling
 overrides, and output policy, so adding a model means writing one adapter file
 instead of editing the shared serving module in ~10 scattered places.

-See the RFC for the full design (issue #4327). This is the foundation landed in
-the first migration PR; Qwen3-TTS is the first model routed through it while the
-remaining models stay on the legacy path until individually migrated.
+See the RFC for the full design (issue #4327).
 """

 from abc import ABC, abstractmethod
@@ -39,6 +37,21 @@ def conditioning_cache_salt(request: "OpenAICreateSpeechRequest", tts_params: di
    return _conditioning_cache_salt_fn(request, tts_params)


+def apply_max_new_tokens(
+    sampling_params_list: list,
+    request: "OpenAICreateSpeechRequest",
+) -> list:
+    """Apply a request-level ``max_new_tokens`` limit."""
+    if request.max_new_tokens is None:
+        return sampling_params_list
+
+    import copy
+
+    sampling_params_list = copy.deepcopy(sampling_params_list)
+    sampling_params_list[0].max_tokens = request.max_new_tokens
+    return sampling_params_list
+
+
@dataclass
 class OutputPolicy:
    """How the orchestrator aggregates engine output for a model.
@@ -184,6 +197,8 @@ class TTSModelAdapter(ABC):
        self,
        sampling_params_list: list,
        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
    ) -> list:
        """Apply model-specific sampling mutations.

@@ -228,37 +243,10 @@ class DiffusionTTSAdapter(TTSModelAdapter):
        return frozenset(params) if params is not None else frozenset()


-@dataclass(frozen=True)
-class LegacyDetector:
-    """A model type the serving layer detects but has no adapter for yet.
-
-    Detection has to keep naming these models so ``serving_speech.py`` can route
-    them down its legacy path, but they own none of the adapter contract. Each
-    entry is a migration debt: delete it when the model gets an adapter. The
-    pre-commit gate (``tools/pre_commit/check_tts_adapter.py``) fails if the list
-    grows.
-
-    Exposes the subset of the :class:`TTSModelAdapter` class surface that
-    detection reads, so both kinds live in one sorted sequence.
-    """
-
-    name: str
-    stage_keys: frozenset[str] = frozenset()
-    model_archs: frozenset[str] = frozenset()
-    arch_identifies_entry_stage: bool = False
-    detect_priority: int = 100
-
-    def matches(self, model_stage: str | None, model_arch: str | None) -> bool:
-        if model_arch is not None and model_arch in self.model_archs:
-            return True
-        return model_stage is not None and model_stage in self.stage_keys
-
-
 # Re-exported here to avoid import cycles at call sites.
 __all__ = [
    "ARTTSAdapter",
    "DiffusionTTSAdapter",
-    "LegacyDetector",
    "OutputPolicy",
    "PreparedRequest",
    "SpeechServingContext",
@@ -1,7 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """CosyVoice3 TTS serving adapter."""

-from typing import TYPE_CHECKING
+import os
+from typing import TYPE_CHECKING, Any
+
+from vllm.logger import init_logger

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
 from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
@@ -9,6 +12,8 @@ from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, Prepare
 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest

+logger = init_logger(__name__)
+

@register_tts_adapter
 class CosyVoice3Adapter(ARTTSAdapter):
@@ -47,6 +52,63 @@ class CosyVoice3Adapter(ARTTSAdapter):
        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
    ) -> PreparedRequest:
        prompt = await self.ctx.server._build_cosyvoice3_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
-        # NOTE: CosyVoice3 dynamic-token sampling stays in the orchestrator tail
-        # (keyed on _tts_model_type) during this incremental migration.
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="cosyvoice3")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        """Set dynamic min/max tokens based on tokenized text length.
+
+        The official model uses ``min_token_text_ratio`` to prevent early EOS
+        and ``max_token_text_ratio`` to cap the generated token count.
+        """
+        import copy
+
+        from vllm_omni.model_executor.models.cosyvoice3.tokenizer import get_qwen_tokenizer
+        from vllm_omni.model_executor.models.cosyvoice3.utils import extract_text_token
+
+        server = self.ctx.server
+        sampling_params_list = copy.deepcopy(sampling_params_list)
+        hf_cfg = server.model_config.hf_config
+        # Build the Qwen tokenizer once per process (resolving the model dir via
+        # snapshot_download at most once) and reuse it across requests.
+        tokenizer = server._cosyvoice3_tokenizer
+        if tokenizer is None:
+            model_path = server.engine_client.model_config.model
+            if not os.path.isdir(model_path):
+                from huggingface_hub import snapshot_download
+
+                model_path = snapshot_download(model_path)
+            tokenizer = get_qwen_tokenizer(
+                token_path=os.path.join(model_path, hf_cfg.qwen_pretrain_path),
+                skip_special_tokens=hf_cfg.skip_special_tokens,
+                version=hf_cfg.version,
+            )
+            server._cosyvoice3_tokenizer = tokenizer
+        _, text_token_len = extract_text_token(
+            request.input,
+            tokenizer,
+            hf_cfg.allowed_special,
+        )
+        min_ratio = getattr(hf_cfg, "min_token_text_ratio", 2)
+        max_ratio = getattr(hf_cfg, "max_token_text_ratio", 20)
+        sampling_params_list[0].min_tokens = max(1, int(text_token_len * min_ratio))
+        sampling_params_list[0].max_tokens = min(2048, int(text_token_len * max_ratio))
+
+        if request.max_new_tokens is not None:
+            sampling_params_list[0].max_tokens = request.max_new_tokens
+            sampling_params_list[0].min_tokens = min(
+                getattr(sampling_params_list[0], "min_tokens", 0),
+                request.max_new_tokens,
+            )
+        logger.info(
+            "CosyVoice3 dynamic tokens: text_tokens=%d, min_tokens=%d, max_tokens=%d",
+            text_token_len,
+            sampling_params_list[0].min_tokens,
+            sampling_params_list[0].max_tokens,
+        )
+        return sampling_params_list
@@ -1,10 +1,15 @@
 # SPDX-License-Identifier: Apache-2.0
 """Fish Speech serving adapter (retires the legacy ``_is_fish_speech`` flag)."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, conditioning_cache_salt
+from vllm_omni.entrypoints.openai.tts_adapters.base import (
+    ARTTSAdapter,
+    PreparedRequest,
+    apply_max_new_tokens,
+    conditioning_cache_salt,
+)

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -53,3 +58,12 @@ class FishSpeechAdapter(ARTTSAdapter):
        tts_params = {}
        prompt["cache_salt"] = conditioning_cache_salt(request, tts_params)
        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="fish_speech")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -1,7 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
 """GLM-TTS serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+
+from vllm.logger import init_logger

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
 from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
@@ -9,6 +11,8 @@ from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, Prepare
 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest

+logger = init_logger(__name__)
+

@register_tts_adapter
 class GlmTTSAdapter(ARTTSAdapter):
@@ -43,6 +47,63 @@ class GlmTTSAdapter(ARTTSAdapter):
        self, request: "OpenAICreateSpeechRequest", sampling_params_list: list, has_inline_ref_audio: bool
    ) -> PreparedRequest:
        prompt = await self.ctx.server._build_glm_tts_prompt(request, has_inline_ref_audio=has_inline_ref_audio)
-        # GLM-TTS dynamic-token sampling stays in the orchestrator tail
-        # (keyed on _tts_model_type) during this incremental migration.
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="glm_tts")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        # GLM-TTS: set dynamic min/max tokens based on text length.
+        import copy
+
+        server = self.ctx.server
+        sampling_params_list = copy.deepcopy(sampling_params_list)
+        glm_metadata = prompt.get("additional_information") if isinstance(prompt, dict) else None
+        text_len_value = None
+        if isinstance(glm_metadata, dict):
+            text_len_value = glm_metadata.get("glm_tts_text_token_len")
+            if isinstance(text_len_value, list) and text_len_value:
+                text_len_value = text_len_value[0]
+        text_token_len = (
+            int(text_len_value)
+            if text_len_value is not None
+            else server._estimate_glm_tts_text_token_len(request.input)
+        )
+        hf_cfg = server.model_config.hf_config
+        min_ratio = getattr(hf_cfg, "min_token_text_ratio", 2)
+        max_ratio = getattr(hf_cfg, "max_token_text_ratio", 20)
+        stage_min_tokens = getattr(sampling_params_list[0], "min_tokens", None)
+        stage_max_tokens = getattr(sampling_params_list[0], "max_tokens", None)
+        cap_candidates = [int(cap) for cap in (stage_max_tokens, request.max_new_tokens) if cap is not None]
+        hard_cap = min(cap_candidates) if cap_candidates else None
+
+        min_tokens = max(1, int(text_token_len * min_ratio))
+        if stage_min_tokens is not None:
+            min_tokens = max(min_tokens, int(stage_min_tokens))
+        if hard_cap is not None:
+            min_tokens = min(min_tokens, hard_cap)
+
+        max_tokens = max(min_tokens, int(text_token_len * max_ratio))
+        if hard_cap is not None:
+            max_tokens = min(max_tokens, hard_cap)
+        sampling_params_list[0].min_tokens = min_tokens
+        sampling_params_list[0].max_tokens = max_tokens
+        seed = getattr(request, "seed", None)
+        if seed is not None:
+            sampling_params_list[0].seed = seed
+        logger.info(
+            "GLM-TTS dynamic tokens: text_tokens=%d, min_ratio=%s, max_ratio=%s, "
+            "stage_min=%s, stage_max=%s, request_max=%s, min_tokens=%d, max_tokens=%d",
+            text_token_len,
+            min_ratio,
+            max_ratio,
+            stage_min_tokens,
+            stage_max_tokens,
+            request.max_new_tokens,
+            min_tokens,
+            max_tokens,
+        )
+        return sampling_params_list
@@ -1,10 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """Higgs-Audio v2 serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -99,3 +99,12 @@ class HiggsAudioV2Adapter(ARTTSAdapter):
                additional["voice_name"] = voice_lower
                additional["voice_created_at"] = server._voice_created_at(voice_lower)
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="higgs_audio_v2")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -1,10 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """Higgs-Audio v3 serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -36,3 +36,12 @@ class HiggsAudioV3Adapter(ARTTSAdapter):
    ) -> PreparedRequest:
        prompt = await self.ctx.server._build_higgs_audio_v3_params(request)
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="higgs_audio_v3")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -14,7 +14,7 @@ from vllm.inputs import tokens_input
 from vllm.utils import random_uuid

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens
 from vllm_omni.model_executor.models.indextts2.configuration_indextts2 import (
    INDEXTTS25_MAX_DURATION_FACTOR,
    INDEXTTS25_MIN_DURATION_FACTOR,
@@ -241,6 +241,15 @@ class IndexTTS2Adapter(ARTTSAdapter):
        prompt["cache_salt"] = indextts2_conditioning_cache_salt(request, tts_params)
        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type=self.name)

+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: OpenAICreateSpeechRequest,
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
+
    async def _build_params(self, request: OpenAICreateSpeechRequest) -> dict[str, Any]:
        server = self.ctx.server
        params: dict[str, Any] = {"text": [request.input]}
@@ -1,7 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 """Ming-TTS (dense) serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm.logger import init_logger

@@ -73,9 +73,36 @@ class MingTTSAdapter(ARTTSAdapter):
            voice_created_at=uploaded_audio_created_at,
        )
        tts_params = prompt.get("additional_information", {})
-        # Ming stop-token / max_tokens sampling stays in the orchestrator tail.
        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="ming_tts")

+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        import copy
+
+        server = self.ctx.server
+
+        from vllm_omni.model_executor.models.ming_tts.config_ming_tts import (
+            MOE_TEXT_EOS_TOKEN_ID,
+            TEXT_EOS_TOKEN_ID,
+        )
+
+        hf_config = server.engine_client.model_config.hf_config
+        is_moe = getattr(hf_config, "model_type", "") == "bailingmm"
+        stop_token_id = MOE_TEXT_EOS_TOKEN_ID if is_moe else TEXT_EOS_TOKEN_ID
+
+        sampling_params_list = copy.deepcopy(sampling_params_list)
+        sampling_params_list[0].stop_token_ids = [int(stop_token_id)]
+        if request.max_new_tokens is not None:
+            # Ming emits TEXT_EOS after the latent decode budget is exhausted, so
+            # Stage-0 needs one extra token beyond ming_max_decode_steps.
+            sampling_params_list[0].max_tokens = int(request.max_new_tokens) + 1
+        return sampling_params_list
+
    def _validate_ming_tts_single_speaker_request(self, request: "OpenAICreateSpeechRequest") -> str | None:
        server = self.ctx.server
        if request.ref_audio is not None:
@@ -2,12 +2,12 @@
 """Qwen3-TTS serving adapter."""

 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm.logger import init_logger

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -163,3 +163,12 @@ class Qwen3TTSAdapter(ARTTSAdapter):
            model_type=tts_params.get("task_type", ["unknown"])[0],
            warmup_artifact_key=warmup_key,
        )
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -1,10 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """VoxCPM2 serving adapter (AR base-LM + diffusion side-computation)."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -70,3 +70,12 @@ class VoxCPM2Adapter(ARTTSAdapter):
                additional["voice_name"] = voice_lower
                additional["voice_created_at"] = server._voice_created_at(voice_lower)
        return PreparedRequest(prompt=prompt, tts_params=tts_params, model_type="voxcpm2")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -1,10 +1,10 @@
 # SPDX-License-Identifier: Apache-2.0
 """Voxtral TTS serving adapter."""

-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any

 from vllm_omni.entrypoints.openai.tts_adapters import register_tts_adapter
-from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest
+from vllm_omni.entrypoints.openai.tts_adapters.base import ARTTSAdapter, PreparedRequest, apply_max_new_tokens

 if TYPE_CHECKING:
    from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest
@@ -48,3 +48,12 @@ class VoxtralTTSAdapter(ARTTSAdapter):
    ) -> PreparedRequest:
        prompt = await self.ctx.server._build_voxtral_prompt_async(request)
        return PreparedRequest(prompt=prompt, tts_params={}, model_type="voxtral_tts")
+
+    def apply_sampling_overrides(
+        self,
+        sampling_params_list: list,
+        request: "OpenAICreateSpeechRequest",
+        prompt: dict[str, Any] | None = None,
+        request_id: str | None = None,
+    ) -> list:
+        return apply_max_new_tokens(sampling_params_list, request)
@@ -509,9 +509,9 @@ def _coerce_video_to_uint8_frames(video: Any) -> np.ndarray:
    return frames_u8


-def _encode_video_bytes(
+def encode_video_bytes(
    video: Any,
-    fps: int,
+    fps: int | float,
    audio: Any | None = None,
    audio_sample_rate: int | None = None,
    video_codec_options: dict[str, str] | None = None,
@@ -530,6 +530,10 @@ def _encode_video_bytes(
    )


+# Backward-compatible private alias for existing internal callers.
+_encode_video_bytes = encode_video_bytes
+
+
 class FragmentedMP4VideoEncoder:
    """Normalize video chunks and append them to one fragmented MP4 stream."""

@@ -6,8 +6,10 @@ from vllm_omni.model_extras.registry import (
    build_image_to_video_prompt,
    build_text_to_image_prompt,
    build_x_to_text_prompt,
+    build_x_to_video_audio_prompt,
    get_extra_body_params,
    get_extra_output_params,
+    get_input_audio_sample_rate,
    get_model_class_name,
    get_output_tensor_range,
    get_transformer_config_subfolder,
@@ -21,8 +23,10 @@ __all__ = [
    "build_image_to_video_prompt",
    "build_text_to_image_prompt",
    "build_x_to_text_prompt",
+    "build_x_to_video_audio_prompt",
    "get_extra_body_params",
    "get_extra_output_params",
+    "get_input_audio_sample_rate",
    "get_model_class_name",
    "get_output_tensor_range",
    "get_transformer_config_subfolder",
@@ -0,0 +1,44 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+DREAMID_OMNI_EXTRA_BODY_PARAMS = frozenset({"solver_name", "shift"})
+DREAMID_OMNI_INPUT_AUDIO_SAMPLE_RATE = 16000
+
+
+def build_x_to_video_audio_prompt(
+    prompt: dict[str, Any],
+    request_options: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Translate the canonical X-to-video+audio envelope for DreamID-Omni."""
+    result = dict(prompt)
+
+    for key in ("video_negative_prompt", "audio_negative_prompt"):
+        if request_options.get(key) is not None:
+            result[key] = request_options[key]
+
+    media_inputs = prompt.get("multi_modal_data")
+    if isinstance(media_inputs, Mapping) and "audio" in media_inputs:
+        adapted_media = dict(media_inputs)
+        audio_inputs = media_inputs["audio"]
+        if not isinstance(audio_inputs, list):
+            audio_inputs = [audio_inputs]
+        adapted_media["audio"] = [_extract_waveform(item) for item in audio_inputs]
+        result["multi_modal_data"] = adapted_media
+    return result
+
+
+def _extract_waveform(audio: Any) -> Any:
+    if not isinstance(audio, tuple) or len(audio) != 2:
+        raise TypeError("Canonical audio inputs must be (waveform, sample_rate) tuples.")
+    waveform, sample_rate = audio
+    if sample_rate != DREAMID_OMNI_INPUT_AUDIO_SAMPLE_RATE:
+        raise ValueError(
+            "DreamID-Omni audio inputs must be resampled to "
+            f"{DREAMID_OMNI_INPUT_AUDIO_SAMPLE_RATE} Hz; got {sample_rate}."
+        )
+    return waveform
@@ -28,6 +28,13 @@ from vllm_omni.model_extras.cosmos3 import (
    COSMOS3_EXTRA_BODY_PARAMS,
    COSMOS3_EXTRA_OUTPUT_PARAMS,
 )
+from vllm_omni.model_extras.dreamid_omni import (
+    DREAMID_OMNI_EXTRA_BODY_PARAMS,
+    DREAMID_OMNI_INPUT_AUDIO_SAMPLE_RATE,
+)
+from vllm_omni.model_extras.dreamid_omni import (
+    build_x_to_video_audio_prompt as build_dreamid_x_to_video_audio_prompt,
+)
 from vllm_omni.model_extras.helios import (
    HELIOS_EXTRA_BODY_PARAMS,
    HELIOS_EXTRA_OUTPUT_PARAMS,
@@ -98,6 +105,7 @@ ImageToVideoPromptBuilder = Callable[
    dict[str, Any],
 ]
 XToTextPromptBuilder = Callable[[str, str, bool], tuple[dict[str, Any], list[int] | None]]
+XToVideoAudioPromptBuilder = Callable[[dict[str, Any], Mapping[str, Any]], dict[str, Any]]
 OutputTensorRange = Literal["negative_one_to_one", "zero_to_one"]


@@ -203,6 +211,11 @@ _EXTRA_SPECS: dict[str, dict[str, Any]] = {
        "extra_body_params": COSMOS3_EXTRA_BODY_PARAMS,
        "extra_output_params": COSMOS3_EXTRA_OUTPUT_PARAMS,
    },
+    "DreamIDOmniPipeline": {
+        "extra_body_params": DREAMID_OMNI_EXTRA_BODY_PARAMS,
+        "input_audio_sample_rate": DREAMID_OMNI_INPUT_AUDIO_SAMPLE_RATE,
+        "x_to_video_audio_prompt_builder": build_dreamid_x_to_video_audio_prompt,
+    },
    "MagiHumanPipeline": {
        "extra_body_params": MAGI_HUMAN_EXTRA_BODY_PARAMS,
        "extra_output_params": MAGI_HUMAN_EXTRA_OUTPUT_PARAMS,
@@ -295,6 +308,12 @@ def get_extra_output_params(model_class_name: str | None) -> frozenset[str]:
    return spec.get("extra_output_params", frozenset()) if spec is not None else frozenset()


+def get_input_audio_sample_rate(model_class_name: str | None) -> int | None:
+    """Return a pipeline's required input rate, or preserve the source rate."""
+    spec = _get_spec(model_class_name)
+    return spec.get("input_audio_sample_rate") if spec is not None else None
+
+
 def get_output_tensor_range(model_class_name: str | None) -> OutputTensorRange:
    """Return the declared range for floating-point tensor outputs.

@@ -401,3 +420,14 @@ def build_image_to_video_prompt(
        width=width,
        num_frames=num_frames,
    )
+
+
+def build_x_to_video_audio_prompt(
+    model_class_name: str | None,
+    prompt: dict[str, Any],
+    request_options: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Apply a model adapter to an example-owned video+audio envelope."""
+    spec = _get_spec(model_class_name)
+    builder: XToVideoAudioPromptBuilder | None = spec.get("x_to_video_audio_prompt_builder") if spec else None
+    return builder(prompt, request_options) if builder is not None else prompt
@@ -406,11 +406,13 @@ class OmniConnectorModelRunnerMixin:
        # thread by one execute_model() cycle, especially when the request was
        # added after the current scheduler_output snapshot.
        #
-        # Orphaned pending recv entries (e.g. from upstream stage crash)
-        # are handled by OmniSchedulingCoordinator.collect_timed_out_request_ids()
-        # which detects wait-time violations.  The scheduler then removes the
-        # request from its queues, sets FINISHED_ERROR, and calls _free_request()
-        # which ultimately triggers cleanup_finished_request() here.
+        # Orphaned pending recv entries (e.g. from upstream stage crash) are
+        # handled by collect_timed_out_request_ids() -- on
+        # OmniSchedulingCoordinator for full-payload requests, and on
+        # OmniChunkTransferAdapter for async-chunk ones -- which detect
+        # wait-time violations.  The scheduler then removes the request from
+        # its queues, sets FINISHED_ERROR, and calls _free_request() which
+        # ultimately triggers cleanup_finished_request() here.
        for attr_name in (
            "_request_ids_mapping",
            "_get_req_chunk",
Author	SHA1	Message	Date
Hongsheng Liu	f1bc6c8569	Address video-audio example review feedback Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>	2026-08-17 16:45:55 +08:00
Hongsheng Liu	6bdbaeb038	Harden the shared video-audio example Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>	2026-08-17 13:48:27 +08:00
boatman	65b3d41b3a	[Migrate] Move sampling parameter overrides from legacy dispatches to TTS adapters (#5272 ) Signed-off-by: boatman <109857087+sphinxkkkbc@users.noreply.github.com>	2026-08-16 21:14:26 -07:00
wangyuandCursor	4616823125	[CI]demote slow nightly cases and trim Qwen-Image-Edit coverage (#5944 ) Signed-off-by: wangyu <410167048@qq.com> Signed-off-by: [Your Name] <your.email@example.com> Co-authored-by: Cursor <cursoragent@cursor.com>	2026-08-17 11:21:28 +08:00
Yueqian LinandHongsheng Liu	611c083587	[BugFix][TTS] R1.1-R1.4: turn silent async-chunk failures into visible ones (#6033 ) Signed-off-by: Yueqian Lin <linyueqian@outlook.com> Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>	2026-08-17 10:51:17 +08:00