[Rebase] Rebase to vllm releases/v0.22.0 (#3891)

Signed-off-by: tzhouam <tzhouam@connect.ust.hk>
Signed-off-by: ZhengWG <zwg0606@gmail.com>
Signed-off-by: david6666666 <530634352@qq.com>
Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Signed-off-by: nainiu258 <cperfect02@163.com>
Signed-off-by: nainiu258 <101917677+nainiu258@users.noreply.github.com>
Signed-off-by: JasonJ2021 <jasonj@zju.edu.cn>
Signed-off-by: Zhou Taichang <tzhouam@connect.ust.hk>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: skf <54565339+skf-1999@users.noreply.github.com>
Co-authored-by: Zheng Wengang <zwg0606@gmail.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Co-authored-by: WeiQing Chen <40507679+david6666666@users.noreply.github.com>
Co-authored-by: SYLAR <125541396+lishunyang12@users.noreply.github.com>
Co-authored-by: Yueqian Lin <linyueqian@outlook.com>
Co-authored-by: nainiu258 <101917677+nainiu258@users.noreply.github.com>
Co-authored-by: Jason <72191212+JasonJ2021@users.noreply.github.com>
Co-authored-by: Yueqian Lin <70319226+linyueqian@users.noreply.github.com>

.buildkite/test-merge.yml

@@ -107,7 +107,7 @@ steps:
     depends_on: upload-merge-pipeline
     steps:
       - label: "Entrypoint Test with H100"
-        timeout_in_minutes: 35
+        timeout_in_minutes: 60
         commands:
           - pytest -s -v tests/entrypoints/ -m "advanced_model and cuda and H100" --run-level "advanced_model"
         agents:

.buildkite/test-ready.yml

@@ -183,7 +183,7 @@ steps:
     depends_on: upload-ready-pipeline
     commands:
       - |
-        timeout 15m bash -c "
+        timeout 20m bash -c "
                   pytest -sv tests/distributed/ -m 'core_model and cuda' --run-level 'core_model'
         "
     agents:

benchmarks/tts/model_configs.yaml

@@ -14,6 +14,7 @@ models:
     supported_tasks: [default_voice, voice_design]
     backend: openai-audio-speech
     endpoint: /v1/audio/speech
+    trust_remote_code: true
     task_extra_body:
       default_voice:
         voice: Vivian
@@ -27,6 +28,7 @@ models:
     supported_tasks: [voice_clone]
     backend: openai-audio-speech
     endpoint: /v1/audio/speech
+    trust_remote_code: true
     task_extra_body:
       voice_clone:
         task_type: Base
@@ -36,6 +38,7 @@ models:
     trust_remote_code: true
     backend: openai-audio-speech
     endpoint: /v1/audio/speech
+    trust_remote_code: true
     task_extra_body:
       voice_clone: {}
       default_voice:

docker/Dockerfile.ci

@@ -1,5 +1,8 @@
+# Target vllm version: v0.21.0 (releases/v0.21.0 branch).
+# The pinned VLLM_PRECOMPILED_WHEEL_COMMIT and VLLM_BASE_TAG below must
+# be updated on every rebase to match the new upstream vllm commit/tag.
 ARG VLLM_BASE_IMAGE=vllm/vllm-openai
-ARG VLLM_BASE_TAG=v0.21.0
+ARG VLLM_BASE_TAG=v0.22.0
 FROM ${VLLM_BASE_IMAGE}:${VLLM_BASE_TAG}
 ARG APP_DIR=/workspace/vllm-omni
 WORKDIR ${APP_DIR}

docs/serving/speech_api.md

@@ -381,6 +381,44 @@ are cached in-process with a shared LRU so repeated requests with the same
 all TTS model types; deleting a voice invalidates every model-type slot at
 once.
 
+### Precomputed Custom Voices
+
+Qwen3-TTS Base and VoxCPM2 can load offline-precomputed voices at startup.
+Generate a directory containing `custom_voice_manifest.json` plus one
+`.safetensors` file per voice, then set the pipeline-wide deploy config field:
+
+```yaml
+custom_voice_dir: /path/to/custom_voices
+```
+
+Qwen3-TTS profiles are created with:
+
+```bash
+python examples/online_serving/text_to_speech/qwen3_tts/precompute_custom_voice.py \
+  --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \
+  --voice-name alice \
+  --ref-audio /path/to/reference.wav \
+  --ref-text "Original transcript of the reference audio" \
+  --mode icl \
+  --output-dir /path/to/custom_voices
+```
+
+VoxCPM2 profiles are created with:
+
+```bash
+python examples/online_serving/text_to_speech/voxcpm2/precompute_custom_voice.py \
+  --model openbmb/VoxCPM2 \
+  --voice-name alice \
+  --ref-audio /path/to/reference.wav \
+  --mode ref_continuation \
+  --prompt-text "Original transcript of the reference audio" \
+  --output-dir /path/to/custom_voices
+```
+
+Only profiles whose safetensors payload can be loaded and validated are exposed
+by `GET /v1/audio/voices`. Valid precomputed voices can be used in
+`POST /v1/audio/speech` by passing `voice="alice"` without `ref_audio`.
+
 **Configuration (environment variables):**
 
 | Variable | Default | Description |

docs/user_guide/examples/online_serving/text_to_speech.md

@@ -380,6 +380,31 @@ curl -X POST http://localhost:8091/v1/audio/voices \
 ```
 Uploaded voices are then usable as `voice="custom_voice_1"` on subsequent requests.
 
+### Precomputed custom voices
+For reused Base voice-cloning speakers, precompute the reference artifacts once and load them at server startup:
+```bash
+python qwen3_tts/precompute_custom_voice.py \
+    --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \
+    --voice-name alice \
+    --ref-audio /path/to/reference.wav \
+    --ref-text "Original transcript of the reference audio" \
+    --mode icl \
+    --output-dir /path/to/custom_voices
+```
+`--mode icl` stores both `speaker_embedding` and `ref_code`; `--mode xvec` stores only the speaker embedding. Add the output directory to a deploy config:
+```yaml
+custom_voice_dir: /path/to/custom_voices
+```
+Then start the server with that config and call the Speech API with only the voice name:
+```bash
+vllm serve Qwen/Qwen3-TTS-12Hz-1.7B-Base --omni --deploy-config /path/to/qwen3_tts_custom_voice.yaml
+
+curl -X POST http://localhost:8091/v1/audio/speech \
+    -H "Content-Type: application/json" \
+    -d '{"input":"Hello from a precomputed voice.","voice":"alice","task_type":"Base"}' \
+    --output alice.wav
+```
+
 ### Streaming PCM
 ```bash
 curl -X POST http://localhost:8091/v1/audio/speech \
@@ -419,7 +444,8 @@ python qwen3_tts/gradio_fastrtc_demo.py --api-base http://localhost:8000
 `qwen3_tts/batch_speech_client.py` issues many concurrent requests for throughput measurement.
 
 ### Notes
-- Base voice cloning has per-request reference-audio cost; the uniproc default keeps IPC overhead off the critical path. See the executor-backend section above for background.
+- Base voice cloning has uniproc-vs-mp tradeoffs depending on per-request reference audio cost; see the executor-backend section above.
+- With async chunking, Qwen3-TTS Base voice cloning sends the full reference context in the first Code2Wav packet, then caches that prefix on the Code2Wav stage for follow-up chunks in the same request.
 - `vllm_omni/deploy/qwen3_tts.yaml` is the default deploy config (loaded by HF `model_type`); per-stage runtime overrides are available via `--stage-N-<field> <value>`.
 
 ---
@@ -448,6 +474,23 @@ python voxcpm2/openai_speech_client.py \
 ```
 The `ref_audio` field accepts local file paths (auto-base64), HTTP URLs, or `data:audio/wav;base64,...` data URIs.
 
+### Precomputed custom voices
+For repeated VoxCPM2 speakers, precompute the prompt cache and load it through `custom_voice_dir`:
+```bash
+python voxcpm2/precompute_custom_voice.py \
+    --model openbmb/VoxCPM2 \
+    --voice-name alice \
+    --ref-audio /path/to/reference.wav \
+    --mode ref_continuation \
+    --prompt-text "Original transcript of the reference audio" \
+    --output-dir /path/to/custom_voices
+```
+Add the output directory to the deploy config:
+```yaml
+custom_voice_dir: /path/to/custom_voices
+```
+After startup, `/v1/audio/voices` lists `alice`, and `/v1/audio/speech` can use `voice="alice"` without sending `ref_audio`.
+
 ### Gradio demo (gapless streaming via AudioWorklet)
 ```bash
 python voxcpm2/gradio_demo.py

docs/user_guide/quantization/gguf.md

@@ -74,7 +74,7 @@ Online:
 vllm serve Qwen/Qwen-Image \
   --omni \
   --port 8000 \
-  --quantization-config '{"method":"gguf","gguf_model":"QuantStack/Qwen-Image-GGUF/Qwen_Image-Q4_K_M.gguf"}'
+  --diffusion-quantization-config '{"method":"gguf","gguf_model":"QuantStack/Qwen-Image-GGUF/Qwen_Image-Q4_K_M.gguf"}'
 ```
 
 ## Parameters

examples/online_serving/text_to_speech/README.md

@@ -412,6 +412,31 @@ curl -X POST http://localhost:8091/v1/audio/voices \
 ```
 Uploaded voices are then usable as `voice="custom_voice_1"` on subsequent requests.
 
+### Precomputed custom voices
+For reused Base voice-cloning speakers, precompute the reference artifacts once and load them at server startup:
+```bash
+python qwen3_tts/precompute_custom_voice.py \
+    --model Qwen/Qwen3-TTS-12Hz-1.7B-Base \
+    --voice-name alice \
+    --ref-audio /path/to/reference.wav \
+    --ref-text "Original transcript of the reference audio" \
+    --mode icl \
+    --output-dir /path/to/custom_voices
+```
+`--mode icl` stores both `speaker_embedding` and `ref_code`; `--mode xvec` stores only the speaker embedding. Add the output directory to a deploy config:
+```yaml
+custom_voice_dir: /path/to/custom_voices
+```
+Then start the server with that config and call the Speech API with only the voice name:
+```bash
+vllm serve Qwen/Qwen3-TTS-12Hz-1.7B-Base --omni --deploy-config /path/to/qwen3_tts_custom_voice.yaml
+
+curl -X POST http://localhost:8091/v1/audio/speech \
+    -H "Content-Type: application/json" \
+    -d '{"input":"Hello from a precomputed voice.","voice":"alice","task_type":"Base"}' \
+    --output alice.wav
+```
+
 ### Streaming PCM
 ```bash
 curl -X POST http://localhost:8091/v1/audio/speech \
@@ -447,7 +472,8 @@ python qwen3_tts/streaming_speech_client.py --text "..." --simulate-stt --stt-de
 `qwen3_tts/batch_speech_client.py` issues many concurrent requests for throughput measurement.
 
 ### Notes
-- Base voice cloning has per-request reference-audio cost; the uniproc default keeps IPC overhead off the critical path. See the executor-backend section above for background.
+- Base voice cloning has uniproc-vs-mp tradeoffs depending on per-request reference audio cost; see the executor-backend section above.
+- With async chunking, Qwen3-TTS Base voice cloning sends the full reference context in the first Code2Wav packet, then caches that prefix on the Code2Wav stage for follow-up chunks in the same request.
 - `vllm_omni/deploy/qwen3_tts.yaml` is the default deploy config (loaded by HF `model_type`); per-stage runtime overrides are available via `--stage-N-<field> <value>`.
 
 ---
@@ -474,6 +500,23 @@ python voxcpm2/openai_speech_client.py \
 ```
 The `ref_audio` field accepts local file paths (auto-base64), HTTP URLs, or `data:audio/wav;base64,...` data URIs.
 
+### Precomputed custom voices
+For repeated VoxCPM2 speakers, precompute the prompt cache and load it through `custom_voice_dir`:
+```bash
+python voxcpm2/precompute_custom_voice.py \
+    --model openbmb/VoxCPM2 \
+    --voice-name alice \
+    --ref-audio /path/to/reference.wav \
+    --mode ref_continuation \
+    --prompt-text "Original transcript of the reference audio" \
+    --output-dir /path/to/custom_voices
+```
+Add the output directory to the deploy config:
+```yaml
+custom_voice_dir: /path/to/custom_voices
+```
+After startup, `/v1/audio/voices` lists `alice`, and `/v1/audio/speech` can use `voice="alice"` without sending `ref_audio`.
+
 ### Gradio demo (gapless streaming via AudioWorklet)
 ```bash
 python voxcpm2/gradio_demo.py

examples/online_serving/text_to_speech/qwen3_tts/precompute_custom_voice.py

@@ -0,0 +1,243 @@
+"""Pre-compute Qwen3-TTS custom voice profiles.
+
+The generated directory can be passed to the server via
+``custom_voice_dir`` in ``vllm_omni/deploy/qwen3_tts.yaml``. Requests can then
+use ``/v1/audio/speech`` with ``voice="<name>"`` and no per-request ref_audio.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import torch
+from safetensors import safe_open
+from safetensors.torch import save_file
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from vllm_omni.utils.custom_voice_io import safe_voice_stem  # noqa: E402
+
+MANIFEST_NAME = "custom_voice_manifest.json"
+_SPEAKER_ENCODER_SAMPLE_RATE = 24000
+_SPEAKER_ENCODER_NUM_MELS = 128
+_SPEAKER_ENCODER_N_FFT = 1024
+_SPEAKER_ENCODER_HOP_SIZE = 256
+_SPEAKER_ENCODER_WIN_SIZE = 1024
+_SPEAKER_ENCODER_FMIN = 0
+_SPEAKER_ENCODER_FMAX = 12000
+
+
+def _resolve_model_dir(model: str) -> str:
+    if os.path.isdir(model):
+        return model
+    from transformers.utils.hub import cached_file
+
+    config_path = cached_file(model, "config.json")
+    if config_path is None:
+        raise FileNotFoundError(f"Could not resolve config.json for {model}")
+    return os.path.dirname(config_path)
+
+
+def _read_audio_mono(path: str) -> tuple[np.ndarray, int]:
+    import soundfile as sf
+
+    wav, sr = sf.read(path, dtype="float32")
+    if wav.ndim > 1:
+        wav = wav.mean(axis=-1)
+    return np.asarray(wav, dtype=np.float32), int(sr)
+
+
+def _resample(wav: np.ndarray, sr: int, target_sr: int) -> np.ndarray:
+    if sr == target_sr:
+        return wav.astype(np.float32)
+    from vllm.multimodal.audio import AudioResampler
+
+    return AudioResampler(target_sr=target_sr).resample(wav.astype(np.float32), orig_sr=int(sr))
+
+
+def _load_speaker_encoder(model_dir: str, device: torch.device):
+    from vllm_omni.model_executor.models.qwen3_tts.configuration_qwen3_tts import Qwen3TTSConfig
+    from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_talker import Qwen3TTSSpeakerEncoder
+
+    config = Qwen3TTSConfig.from_pretrained(model_dir)
+    encoder = Qwen3TTSSpeakerEncoder(config.speaker_encoder_config)
+    index_path = Path(model_dir) / "model.safetensors.index.json"
+    state: dict[str, torch.Tensor] = {}
+    if index_path.exists():
+        index = json.loads(index_path.read_text(encoding="utf-8"))
+        weight_map = index.get("weight_map", {})
+        shard_names = sorted({v for k, v in weight_map.items() if k.startswith("speaker_encoder.")})
+        shard_paths = [Path(model_dir) / name for name in shard_names]
+    else:
+        shard_paths = sorted(Path(model_dir).glob("model*.safetensors"))
+    for shard in shard_paths:
+        with safe_open(str(shard), framework="pt", device="cpu") as f:
+            for key in f.keys():
+                if key.startswith("speaker_encoder."):
+                    state[key.removeprefix("speaker_encoder.")] = f.get_tensor(key)
+    if not state:
+        raise RuntimeError(
+            f"No speaker_encoder.* tensors found under {model_dir}. "
+            "Use a Qwen3-TTS Base checkpoint for pre-computing custom voice profiles."
+        )
+    encoder.load_state_dict(state)
+    encoder.to(device=device, dtype=torch.bfloat16).eval()
+    return config, encoder
+
+
+def _speaker_embedding(
+    encoder: torch.nn.Module,
+    speaker_encoder_config: object,
+    wav: np.ndarray,
+    sr: int,
+    device: torch.device,
+) -> torch.Tensor:
+    from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import mel_spectrogram
+
+    target_sr = int(getattr(speaker_encoder_config, "sample_rate", _SPEAKER_ENCODER_SAMPLE_RATE))
+    mel_dim = int(getattr(speaker_encoder_config, "mel_dim", _SPEAKER_ENCODER_NUM_MELS))
+    if target_sr != _SPEAKER_ENCODER_SAMPLE_RATE or mel_dim != _SPEAKER_ENCODER_NUM_MELS:
+        raise ValueError(
+            "Unsupported Qwen3-TTS speaker encoder mel config for precompute: "
+            f"sample_rate={target_sr}, mel_dim={mel_dim}. "
+            f"Expected sample_rate={_SPEAKER_ENCODER_SAMPLE_RATE}, mel_dim={_SPEAKER_ENCODER_NUM_MELS} "
+            "to match the serving-time speaker embedding path."
+        )
+    wav = _resample(wav, sr, target_sr)
+    wav_t = torch.from_numpy(wav).to(device=device, dtype=torch.float32)
+    mels = mel_spectrogram(
+        wav_t.unsqueeze(0),
+        n_fft=_SPEAKER_ENCODER_N_FFT,
+        num_mels=_SPEAKER_ENCODER_NUM_MELS,
+        sampling_rate=_SPEAKER_ENCODER_SAMPLE_RATE,
+        hop_size=_SPEAKER_ENCODER_HOP_SIZE,
+        win_size=_SPEAKER_ENCODER_WIN_SIZE,
+        fmin=_SPEAKER_ENCODER_FMIN,
+        fmax=_SPEAKER_ENCODER_FMAX,
+    ).transpose(1, 2)
+    with torch.inference_mode():
+        return encoder(mels.to(device=device, dtype=torch.bfloat16))[0].float().cpu().contiguous()
+
+
+def _ref_code(model_dir: str, wav: np.ndarray, sr: int, device: torch.device) -> torch.Tensor:
+    from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_tokenizer import Qwen3TTSTokenizer
+
+    tok = Qwen3TTSTokenizer.from_pretrained(
+        str(Path(model_dir) / "speech_tokenizer"),
+        torch_dtype=torch.bfloat16,
+    )
+    try:
+        del tok.model.decoder
+        tok.model.decoder = None
+        tok.model.encoder.to(device)
+        tok.device = device
+    except Exception:
+        tok.device = device
+    with torch.inference_mode():
+        enc = tok.encode(wav, sr=int(sr), return_dict=True)
+    codes = getattr(enc, "audio_codes", None)
+    if isinstance(codes, list):
+        codes = codes[0] if codes else None
+    if not isinstance(codes, torch.Tensor):
+        raise RuntimeError("Speech tokenizer did not return audio_codes")
+    if codes.ndim == 3:
+        codes = codes[0]
+    return codes.to(dtype=torch.int32, device="cpu").contiguous()
+
+
+def _load_manifest(output_dir: Path, model: str, hidden_size: int) -> dict[str, Any]:
+    path = output_dir / MANIFEST_NAME
+    if path.exists():
+        return json.loads(path.read_text(encoding="utf-8"))
+    return {
+        "schema_version": 1,
+        "model_type": "qwen3_tts",
+        "model": model,
+        "hidden_size": hidden_size,
+        "voices": {},
+    }
+
+
+def _write_voice(
+    *,
+    model: str,
+    output_dir: Path,
+    voice_name: str,
+    ref_audio: str,
+    ref_text: str | None,
+    mode: str,
+    speaker_description: str | None,
+    device: torch.device,
+) -> None:
+    model_dir = _resolve_model_dir(model)
+    config, encoder = _load_speaker_encoder(model_dir, device)
+    wav, sr = _read_audio_mono(ref_audio)
+    if wav.size < 1024:
+        raise ValueError(f"Reference audio too short: {wav.size} samples")
+
+    tensors: dict[str, torch.Tensor] = {
+        "speaker_embedding": _speaker_embedding(encoder, config.speaker_encoder_config, wav, sr, device),
+    }
+    if mode == "icl":
+        if not ref_text or not ref_text.strip():
+            raise ValueError("--ref-text is required for --mode icl")
+        tensors["ref_code"] = _ref_code(model_dir, wav, sr, device)
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    filename = f"{safe_voice_stem(voice_name)}.safetensors"
+    save_file(tensors, str(output_dir / filename))
+
+    hidden_size = int(getattr(config.talker_config, "hidden_size", tensors["speaker_embedding"].numel()))
+    manifest = _load_manifest(output_dir, model, hidden_size)
+    entry: dict[str, Any] = {
+        "name": voice_name,
+        "file": filename,
+        "mode": mode,
+        "embedding_dim": int(tensors["speaker_embedding"].numel()),
+    }
+    if "ref_code" in tensors:
+        entry["ref_code_length"] = int(tensors["ref_code"].shape[0])
+    if ref_text:
+        entry["ref_text"] = ref_text
+    if speaker_description:
+        entry["speaker_description"] = speaker_description
+    manifest.setdefault("voices", {})[voice_name] = entry
+    (output_dir / MANIFEST_NAME).write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8")
+    print(f"Wrote {output_dir / filename}")
+    print(f"Updated {output_dir / MANIFEST_NAME}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Pre-compute Qwen3-TTS custom voice profile")
+    parser.add_argument("--model", required=True, help="Qwen3-TTS Base model path or Hugging Face ID")
+    parser.add_argument("--voice-name", required=True)
+    parser.add_argument("--ref-audio", required=True)
+    parser.add_argument("--ref-text", default=None)
+    parser.add_argument("--mode", choices=["xvec", "icl"], default="xvec")
+    parser.add_argument("--speaker-description", default=None)
+    parser.add_argument("--output-dir", required=True)
+    parser.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu")
+    args = parser.parse_args()
+
+    _write_voice(
+        model=args.model,
+        output_dir=Path(args.output_dir),
+        voice_name=args.voice_name,
+        ref_audio=args.ref_audio,
+        ref_text=args.ref_text,
+        mode=args.mode,
+        speaker_description=args.speaker_description,
+        device=torch.device(args.device),
+    )
+
+
+if __name__ == "__main__":
+    main()

examples/online_serving/text_to_speech/voxcpm2/precompute_custom_voice.py

@@ -0,0 +1,128 @@
+"""Pre-compute VoxCPM2 custom voice profiles.
+
+The generated directory can be passed to the server via
+``custom_voice_dir`` in ``vllm_omni/deploy/voxcpm2.yaml``. Requests can then
+use ``/v1/audio/speech`` with ``voice="<name>"`` and no per-request ref_audio.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+import torch
+from safetensors.torch import save_file
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from vllm_omni.utils.custom_voice_io import safe_voice_stem  # noqa: E402
+
+MANIFEST_NAME = "custom_voice_manifest.json"
+
+
+def _load_tts(model: str, device: torch.device):
+    from vllm_omni.model_executor.models.voxcpm2.voxcpm2_import_utils import import_voxcpm2_core
+
+    VoxCPM = import_voxcpm2_core()
+    native = VoxCPM.from_pretrained(model, load_denoiser=False, optimize=False)
+    return native.tts_model.to(device).eval()
+
+
+def _load_manifest(output_dir: Path, model: str) -> dict[str, Any]:
+    path = output_dir / MANIFEST_NAME
+    if path.exists():
+        return json.loads(path.read_text(encoding="utf-8"))
+    return {
+        "schema_version": 1,
+        "model_type": "voxcpm2",
+        "model": model,
+        "voices": {},
+    }
+
+
+def _write_voice(
+    *,
+    model: str,
+    output_dir: Path,
+    voice_name: str,
+    ref_audio: str,
+    prompt_text: str | None,
+    mode: str,
+    speaker_description: str | None,
+    device: torch.device,
+) -> None:
+    if mode in ("continuation", "ref_continuation") and not prompt_text:
+        raise ValueError("--prompt-text is required for continuation/ref_continuation modes")
+
+    tts = _load_tts(model, device)
+    tensors: dict[str, torch.Tensor] = {}
+    with torch.inference_mode():
+        if mode in ("reference", "ref_continuation"):
+            tensors["ref_audio_feat"] = tts._encode_wav(ref_audio, padding_mode="right").float().cpu().contiguous()
+        if mode in ("continuation", "ref_continuation"):
+            tensors["audio_feat"] = tts._encode_wav(ref_audio, padding_mode="left").float().cpu().contiguous()
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    filename = f"{safe_voice_stem(voice_name)}.safetensors"
+    save_file(tensors, str(output_dir / filename))
+
+    manifest = _load_manifest(output_dir, model)
+    entry: dict[str, Any] = {
+        "name": voice_name,
+        "file": filename,
+        "mode": mode,
+    }
+    if "ref_audio_feat" in tensors:
+        entry["ref_audio_feat_len"] = int(tensors["ref_audio_feat"].shape[0])
+    if "audio_feat" in tensors:
+        entry["audio_feat_len"] = int(tensors["audio_feat"].shape[0])
+    if prompt_text:
+        entry["prompt_text"] = prompt_text
+    if speaker_description:
+        entry["speaker_description"] = speaker_description
+
+    manifest.setdefault("voices", {})[voice_name] = entry
+    (output_dir / MANIFEST_NAME).write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8")
+    print(f"Wrote {output_dir / filename}")
+    print(f"Updated {output_dir / MANIFEST_NAME}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Pre-compute VoxCPM2 custom voice profile")
+    parser.add_argument("--model", default="openbmb/VoxCPM2", help="VoxCPM2 model path or Hugging Face ID")
+    parser.add_argument("--voice-name", required=True)
+    parser.add_argument("--ref-audio", required=True)
+    parser.add_argument(
+        "--prompt-text",
+        default=None,
+        help="Transcript of ref audio for continuation/ref_continuation modes",
+    )
+    parser.add_argument(
+        "--mode",
+        choices=["reference", "continuation", "ref_continuation"],
+        default="reference",
+    )
+    parser.add_argument("--speaker-description", default=None)
+    parser.add_argument("--output-dir", required=True)
+    parser.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu")
+    args = parser.parse_args()
+
+    _write_voice(
+        model=args.model,
+        output_dir=Path(args.output_dir),
+        voice_name=args.voice_name,
+        ref_audio=args.ref_audio,
+        prompt_text=args.prompt_text,
+        mode=args.mode,
+        speaker_description=args.speaker_description,
+        device=torch.device(args.device),
+    )
+
+
+if __name__ == "__main__":
+    main()

recipes/README.md

@@ -47,6 +47,7 @@ recipes/
 | [`Wan-AI/Wan2.2-S2V.md`](./Wan-AI/Wan2.2-S2V.md) | Speech-to-video serving (Wan2.2 14B) | 2x A100/H100 80GB |
 | [`zai-org/GLM-TTS.md`](./zai-org/GLM-TTS.md) | Online serving for Chinese/English zero-shot voice-cloned TTS | 1x A40 48GB |
 | [`GLM/GLM-Image.md`](./GLM/GLM-Image.md) | Online serving for image generation | 1x A800 80GB / 2x A800 80GB |
+|
 
 Within a single recipe file, include different hardware support sections such
 as `GPU`, `ROCm`, and `NPU`, and add concrete tested configurations like

requirements/common.txt

@@ -1,4 +1,10 @@
 # Common dependencies for all platforms
+# transformers: keep aligned with upstream vLLM constraint.
+# vLLM pins ">= 4.56.0, != 5.0.*, != 5.1.*, != 5.2.*, != 5.3.*, != 5.4.*, != 5.5.0".
+# 5.9.0 breaks models that define _keys_to_ignore_on_load_unexpected as a list
+# (e.g. OpenMOSS-Team/MOSS-TTS-Nano) because it changed
+# _adjust_missing_and_unexpected_keys from list+list to set|set.
+transformers >= 4.56.0, != 5.0.*, != 5.1.*, != 5.2.*, != 5.3.*, != 5.4.*, != 5.5.0, < 5.9.0
 av>=14.0.0
 omegaconf>=2.3.0
 diffusers==0.38.0

tests/dfx/perf/scripts/run_diffusion_benchmark.py

@@ -223,8 +223,12 @@ def _get_open_port() -> int:
         return s.getsockname()[1]
 
 
-def _wait_for_port(host: str, port: int, timeout: int = 1200) -> None:
-    """Block until the given host:port accepts connections or timeout expires."""
+def _wait_for_port(host: str, port: int, timeout: int = 1200, proc: subprocess.Popen | None = None) -> None:
+    """Block until the given host:port accepts connections or timeout expires.
+
+    If *proc* is provided, also monitors the process; raises RuntimeError
+    immediately if the server process exits before the port becomes available.
+    """
     start = time.time()
     while time.time() - start < timeout:
         try:
@@ -234,6 +238,10 @@ def _wait_for_port(host: str, port: int, timeout: int = 1200) -> None:
                     return
         except Exception:
             pass
+        if proc is not None:
+            ret = proc.poll()
+            if ret is not None:
+                raise RuntimeError(f"Server process exited with code {ret} before port {host}:{port} became ready")
         time.sleep(2)
     raise RuntimeError(f"Server did not start on {host}:{port} within {timeout}s")
 
@@ -331,7 +339,7 @@ class DiffusionServer:
             env=env,
             cwd=str(Path(__file__).parent.parent.parent.parent),
         )
-        _wait_for_port(self.host, self.port)
+        _wait_for_port(self.host, self.port, proc=self.proc)
         print(f"DiffusionServer ready on {self.host}:{self.port}")
 
     def __enter__(self):

tests/dfx/perf/tests/test_qwen3_omni_vllm_text.json

@@ -5,7 +5,8 @@
             "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
             "use_omni": false,
             "extra_cli_args": [
-                "--no-enable-prefix-caching"
+                "--no-enable-prefix-caching",
+                "--skip-mm-profiling"
             ]
         },
         "benchmark_params": [

tests/diffusion/offloader/test_diffusion_cpu_offload.py

@@ -17,7 +17,7 @@ AUDIO_MODEL = {
 }
 
 IMAGE_VIDEO_MODELS = {
-    "riverclouds/qwen_image_random": {"cuda": 2500, "rocm": 2100},
+    "riverclouds/qwen_image_random": {"cuda": 2200, "rocm": 2100},
 }
 
 MODELS = {**AUDIO_MODEL, **IMAGE_VIDEO_MODELS}

tests/diffusion/test_diffusion_model_runner.py

@@ -301,8 +301,4 @@ def test_vllm_set_forward_context_implementation(monkeypatch):
                 "vllm_config",
             ),
         ),
-        ("ir_op_priority", None),
-        ("enable_torch_wrap", vllm_config.compilation_config.ir_enable_torch_wrap),
-        ("enable_torch_wrap_exit", vllm_config.compilation_config.ir_enable_torch_wrap),
-        ("ir_op_priority_exit", None),
     ], ERROR_MESSAGE

tests/diffusion/test_stage_diffusion_proc.py

@@ -85,7 +85,7 @@ class MockOmniRequestOutput:
 BASE_HEIGHT = 512
 BASE_WIDTH = 512
 BASE_INFER_STEPS = 10
-DELAY_BASE = 2
+DELAY_BASE = 0.01
 
 
 class MockDiffusionEngine:
@@ -93,7 +93,7 @@ class MockDiffusionEngine:
         def simulate_step_delay(height, width, num_inference_steps) -> float:
             return (height / BASE_HEIGHT) * (width / BASE_WIDTH) * (num_inference_steps / BASE_INFER_STEPS)
 
-        DELAY_BASE = 2
+        DELAY_BASE = 0.01
         delay_scale = simulate_step_delay(
             request.sampling_params.height, request.sampling_params.width, request.sampling_params.num_inference_steps
         )
@@ -139,7 +139,7 @@ async def test_proc_process_request_with_batching_async_output():
     assert len(results) == len(test_requests)
     base_time = DELAY_BASE
     time_gap_std = DELAY_BASE * 2 * 2 * 1  # height/width/steps infer time scale
-    eps = 0.5
+    eps = 0.1
     for i, (res, elapsed_time) in enumerate(results):
         assert res.request_id == test_requests[i]["request_id"]
         assert isinstance(res, MockOmniRequestOutput)

tests/distributed/omni_connectors/test_chunk_transfer_adapter.py

@@ -703,6 +703,7 @@ def test_generation_scheduler_calls_cleanup_on_finished(monkeypatch, mocker: Moc
         kv_connector_output=None,
         cudagraph_stats=None,
         req_id_to_index={"req-s1": 0},
+        routed_experts=None,
         routed_experts_dict=None,
     )
 
@@ -789,6 +790,7 @@ def test_ar_scheduler_defers_cleanup_and_queues_save_on_finished(mocker: MockerF
         cudagraph_stats=None,
         req_id_to_index={"req-ar": 0},
         kv_extracted_req_ids=None,
+        routed_experts=None,
         routed_experts_dict=None,
     )
 

tests/distributed/omni_coordinator/test_omni_coordinator.py

@@ -68,13 +68,13 @@ def test_omni_coordinator_pub_coalescing_on_rapid_queue_updates():
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
-    sub.connect(pub_addr)
+    sub.connect(coordinator.pub_zmq_addr)
     sub.setsockopt(zmq.SUBSCRIBE, b"")
 
     time.sleep(0.3)  # PUB/SUB slow-joiner
 
     client = OmniCoordClientForStage(
-        router_addr,
+        coordinator.router_zmq_addr,
         "tcp://stage:coalesce",
         "tcp://stage:coalesce-out",
         0,
@@ -131,7 +131,7 @@ def test_omni_coordinator_registration_broadcast():
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
-    sub.connect(pub_addr)
+    sub.connect(coordinator.pub_zmq_addr)
     sub.setsockopt(zmq.SUBSCRIBE, b"")
 
     # ZMQ PUB/SUB slow-joiner: allow SUB to connect before clients register.
@@ -139,9 +139,9 @@ def test_omni_coordinator_registration_broadcast():
 
     # Create 3 stage clients; each auto-registers on init.
     clients = [
-        OmniCoordClientForStage(router_addr, "tcp://stage:10001", "tcp://stage:10001-out", 0),
-        OmniCoordClientForStage(router_addr, "tcp://stage:10002", "tcp://stage:10002-out", 0),
-        OmniCoordClientForStage(router_addr, "tcp://stage:10003", "tcp://stage:10003-out", 1),
+        OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:10001", "tcp://stage:10001-out", 0),
+        OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:10002", "tcp://stage:10002-out", 0),
+        OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:10003", "tcp://stage:10003-out", 1),
     ]
 
     msg = _wait_for_replica_list(sub, expected_count=3)
@@ -183,19 +183,19 @@ def test_omni_coordinator_heartbeat_timeout_handling():
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
-    sub.connect(pub_addr)
+    sub.connect(coordinator.pub_zmq_addr)
     sub.setsockopt(zmq.SUBSCRIBE, b"")
 
     time.sleep(0.3)
 
     # A and B: real clients that send heartbeats every 5s.
-    client_a = OmniCoordClientForStage(router_addr, "tcp://stage:a", "tcp://stage:a-out", 0)
-    client_b = OmniCoordClientForStage(router_addr, "tcp://stage:b", "tcp://stage:b-out", 0)
+    client_a = OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:a", "tcp://stage:a-out", 0)
+    client_b = OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:b", "tcp://stage:b-out", 0)
 
     # C: raw DEALER that sends only registration, no heartbeat.
     dealer_ctx = zmq.Context.instance()
     dealer_c = dealer_ctx.socket(zmq.DEALER)
-    dealer_c.connect(router_addr)
+    dealer_c.connect(coordinator.router_zmq_addr)
     reg_event = {
         "input_addr": "tcp://stage:c",
         "output_addr": "tcp://stage:c-out",
@@ -254,12 +254,12 @@ def test_omni_coordinator_replica_shutdown_handling():
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
-    sub.connect(pub_addr)
+    sub.connect(coordinator.pub_zmq_addr)
     sub.setsockopt(zmq.SUBSCRIBE, b"")
 
     time.sleep(0.3)  # PUB/SUB slow-joiner
 
-    client = OmniCoordClientForStage(router_addr, "tcp://stage:shutdown", "tcp://stage:shutdown-out", 0)
+    client = OmniCoordClientForStage(coordinator.router_zmq_addr, "tcp://stage:shutdown", "tcp://stage:shutdown-out", 0)
 
     msg = _wait_for_replica_list(sub, expected_count=1)
     assert msg is not None

tests/e2e/online_serving/test_qwen_image_expansion.py

@@ -97,7 +97,7 @@ def _get_diffusion_feature_cases(model: str):
                     "--vae-patch-parallel-size",
                     "2",
                     "--vae-use-tiling",
-                    "--quantization-config",
+                    "--diffusion-quantization-config",
                     '{"method":"fp8"}',
                 ],
             ),

tests/entrypoints/openai_api/test_serving_speech.py

@@ -1,9 +1,13 @@
 # tests/entrypoints/openai/test_serving_speech.py
 import asyncio
+import base64
+import hashlib
+import io
 import json
 import logging
 import os
 import struct
+import wave
 from inspect import Signature, signature
 from pathlib import Path
 from types import SimpleNamespace
@@ -170,6 +174,27 @@ def create_mock_audio_output_for_test(
     )
 
 
+def _write_custom_voice_manifest(root: Path, *, model_type: str, voices: dict) -> None:
+    payload = {
+        "schema_version": 1,
+        "model_type": model_type,
+        "voices": voices,
+    }
+    (root / "custom_voice_manifest.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def _wav_data_url(samples: np.ndarray, sample_rate: int) -> str:
+    pcm = (np.clip(samples, -1.0, 1.0) * 32767).astype("<i2")
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wav:
+        wav.setnchannels(1)
+        wav.setsampwidth(2)
+        wav.setframerate(sample_rate)
+        wav.writeframes(pcm.tobytes())
+    encoded = base64.b64encode(buf.getvalue()).decode("ascii")
+    return f"data:audio/wav;base64,{encoded}"
+
+
 @pytest.fixture
 def test_app(mocker: MockerFixture):
     # Mock the engine client
@@ -972,6 +997,233 @@ class TestTTSMethods:
         params = speech_server._build_tts_params(req)
         assert "voice_clone_prompt" not in params
 
+    @pytest.mark.asyncio
+    async def test_resolve_ref_audio_reuses_decoded_audio_for_same_source(self, speech_server):
+        wav = np.linspace(-0.5, 0.5, 48000, dtype=np.float32)
+        ref_audio = _wav_data_url(wav, 24000)
+        speech_server.model_config.allowed_local_media_path = ""
+        speech_server.model_config.allowed_media_domains = None
+
+        first = await speech_server._resolve_ref_audio(ref_audio)
+        second = await speech_server._resolve_ref_audio(ref_audio)
+
+        assert first[1] == 24000
+        assert second[1] == 24000
+        assert first[0] is second[0]
+        assert first[0][0] == pytest.approx(float(wav[0]), abs=1e-4)
+        assert speech_server._get_resolved_ref_audio_artifact_key(
+            ref_audio
+        ) == speech_server._make_ref_audio_artifact_cache_key(np.asarray(first[0], dtype=np.float32), 24000)
+
+    def test_precomputed_qwen3_voice_infers_base_without_ref_audio(self, speech_server):
+        """Precomputed Qwen3 voices are reusable by name without per-request ref_audio."""
+        speech_server._tts_model_type = "qwen3_tts"
+        speech_server.precomputed_speakers = {
+            "alice": {
+                "name": "Alice",
+                "model_type": "qwen3_tts",
+                "mode": "icl",
+                "ref_text": "reference transcript",
+                "ref_code_length": 3,
+            }
+        }
+        speech_server.supported_speakers = {"alice"}
+
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Alice")
+        assert speech_server._validate_tts_request(req) is None
+        assert req.task_type == "Base"
+
+        params = speech_server._build_tts_params(req)
+        assert params["task_type"] == ["Base"]
+        assert params["speaker"] == ["alice"]
+        assert params["x_vector_only_mode"] == [False]
+        assert params["ref_text"] == ["reference transcript"]
+        assert params["ref_code_length"] == [3]
+        assert "ref_audio" not in params
+
+    def test_uploaded_qwen3_voice_wins_over_same_named_precomputed_voice(self, speech_server, tmp_path):
+        """Qwen3 uploaded voices should take precedence over same-name precomputed voices."""
+        from safetensors.torch import save_file
+
+        uploaded_path = tmp_path / "alice.safetensors"
+        save_file({"speaker_embedding": torch.tensor([0.1] * 4)}, str(uploaded_path))
+        speech_server._tts_model_type = "qwen3_tts"
+        speech_server.uploaded_speakers_dir = tmp_path
+        speech_server.uploaded_speakers = {
+            "alice": {
+                "name": "alice",
+                "created_at": 123,
+                "file_path": str(uploaded_path),
+                "embedding_source": "direct",
+            }
+        }
+        speech_server.precomputed_speakers = {
+            "alice": {
+                "name": "Alice",
+                "model_type": "qwen3_tts",
+                "mode": "icl",
+                "ref_text": "precomputed transcript",
+                "ref_code_length": 3,
+            }
+        }
+
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Alice")
+        assert speech_server._validate_tts_request(req) is None
+
+        params = speech_server._build_tts_params(req)
+        assert params["speaker"] == ["alice"]
+        assert params["voice_created_at"] == [123]
+        assert params["task_type"] == ["Base"]
+        assert "voice_clone_prompt" in params
+        assert "ref_code_length" not in params
+        assert params.get("ref_text") != ["precomputed transcript"]
+
+    def test_precomputed_qwen3_missing_safetensors_is_not_registered(self, speech_server, tmp_path):
+        """Manifest entries are not supported speakers unless their safetensors load."""
+        _write_custom_voice_manifest(
+            tmp_path,
+            model_type="qwen3_tts",
+            voices={"Alice": {"file": "missing.safetensors", "mode": "xvec", "embedding_dim": 4}},
+        )
+        speech_server._tts_model_type = "qwen3_tts"
+        speech_server.engine_client.model_config = SimpleNamespace(
+            hf_config=SimpleNamespace(
+                custom_voice_dir=str(tmp_path),
+                talker_config=SimpleNamespace(hidden_size=4),
+            )
+        )
+
+        profiles = speech_server._load_precomputed_speakers()
+        assert profiles == {}
+
+        speech_server.precomputed_speakers = profiles
+        speech_server.supported_speakers = set(profiles)
+        assert "alice" not in speech_server.supported_speakers
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Alice")
+        assert speech_server._validate_tts_request(req) is not None
+
+    def test_precomputed_qwen3_icl_without_ref_code_is_not_registered(self, speech_server, tmp_path):
+        """Qwen3 ICL profiles without ref_code cannot be exposed by the API layer."""
+        from safetensors.torch import save_file
+
+        save_file({"speaker_embedding": torch.arange(4, dtype=torch.float32)}, str(tmp_path / "alice.safetensors"))
+        _write_custom_voice_manifest(
+            tmp_path,
+            model_type="qwen3_tts",
+            voices={
+                "Alice": {
+                    "file": "alice.safetensors",
+                    "mode": "icl",
+                    "ref_text": "reference transcript",
+                    "embedding_dim": 4,
+                }
+            },
+        )
+        speech_server._tts_model_type = "qwen3_tts"
+        speech_server.engine_client.model_config = SimpleNamespace(
+            hf_config=SimpleNamespace(
+                custom_voice_dir=str(tmp_path),
+                talker_config=SimpleNamespace(hidden_size=4),
+            )
+        )
+
+        profiles = speech_server._load_precomputed_speakers()
+        assert profiles == {}
+
+        speech_server.precomputed_speakers = profiles
+        speech_server.supported_speakers = set(profiles)
+        assert "alice" not in speech_server.supported_speakers
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Alice")
+        assert speech_server._validate_tts_request(req) is not None
+
+    def test_precomputed_voxcpm2_missing_safetensors_is_not_registered(self, speech_server, tmp_path):
+        """VoxCPM2 must not advertise a manifest-only voice that cannot hit prompt cache."""
+        _write_custom_voice_manifest(
+            tmp_path,
+            model_type="voxcpm2",
+            voices={"Bob": {"file": "missing.safetensors", "mode": "reference", "ref_audio_feat_len": 2}},
+        )
+        speech_server._tts_model_type = "voxcpm2"
+        speech_server.engine_client.model_config = SimpleNamespace(
+            hf_config=SimpleNamespace(custom_voice_dir=str(tmp_path))
+        )
+
+        profiles = speech_server._load_precomputed_speakers()
+        assert profiles == {}
+
+        speech_server.precomputed_speakers = profiles
+        speech_server.supported_speakers = set(profiles)
+        assert "bob" not in speech_server.supported_speakers
+        req = OpenAICreateSpeechRequest(input="Hello", voice="Bob")
+        assert speech_server._validate_tts_request(req) is not None
+        assert speech_server._validate_tts_request(OpenAICreateSpeechRequest(input="Hello")) is None
+
+    def test_prepare_voxcpm2_rejects_supported_speaker_without_voice_profile(self, speech_server, mocker):
+        """VoxCPM2 named voices must be uploaded or precomputed on the real request path."""
+        speech_server._tts_model_type = "voxcpm2"
+        speech_server.supported_speakers = {"bob", "default"}
+        speech_server.uploaded_speakers = {}
+        speech_server.precomputed_speakers = {}
+        speech_server.engine_client.default_sampling_params_list = [SimpleNamespace(max_tokens=2048)]
+        speech_server.engine_client.generate = mocker.MagicMock(return_value="generator")
+        speech_server._build_voxcpm2_prompt = mocker.AsyncMock(
+            return_value={"prompt_token_ids": [1], "additional_information": {}}
+        )
+
+        with pytest.raises(ValueError, match="Invalid voice 'bob'"):
+            asyncio.run(speech_server._prepare_speech_generation(OpenAICreateSpeechRequest(input="Hello", voice="Bob")))
+
+        speech_server._build_voxcpm2_prompt.assert_not_awaited()
+        speech_server.engine_client.generate.assert_not_called()
+
+    def test_prepare_voxcpm2_accepts_default_voice(self, speech_server, mocker):
+        """VoxCPM2 default voice preserves the built-in zero-shot request path."""
+        speech_server._tts_model_type = "voxcpm2"
+        speech_server.supported_speakers = {"default"}
+        speech_server.uploaded_speakers = {}
+        speech_server.precomputed_speakers = {}
+        speech_server.engine_client.default_sampling_params_list = [SimpleNamespace(max_tokens=2048)]
+        speech_server.engine_client.generate = mocker.MagicMock(return_value=iter(()))
+        speech_server._build_voxcpm2_prompt = mocker.AsyncMock(
+            return_value={"prompt_token_ids": [1], "additional_information": {}}
+        )
+
+        asyncio.run(speech_server._prepare_speech_generation(OpenAICreateSpeechRequest(input="Hello", voice="default")))
+
+        speech_server._build_voxcpm2_prompt.assert_awaited_once()
+        speech_server.engine_client.generate.assert_called_once()
+
+    def test_prepare_voxcpm2_precomputed_voice_sets_model_cache_key(self, speech_server, mocker):
+        """VoxCPM2 precomputed voices must carry voice metadata to the model cache lookup."""
+        speech_server._tts_model_type = "voxcpm2"
+        speech_server.supported_speakers = {"alice"}
+        speech_server.uploaded_speakers = {}
+        speech_server.precomputed_speakers = {
+            "alice": {
+                "name": "Alice",
+                "model_type": "voxcpm2",
+                "mode": "reference",
+                "ref_audio_feat_len": 2,
+            }
+        }
+        speech_server.engine_client.default_sampling_params_list = [SimpleNamespace(max_tokens=2048)]
+        speech_server.engine_client.generate = mocker.MagicMock(return_value=iter(()))
+        speech_server._build_voxcpm2_prompt = mocker.AsyncMock(
+            return_value={
+                "prompt_token_ids": [1],
+                "additional_information": {
+                    "voice_profile": speech_server.precomputed_speakers["alice"],
+                },
+            }
+        )
+
+        asyncio.run(speech_server._prepare_speech_generation(OpenAICreateSpeechRequest(input="Hello", voice="Alice")))
+
+        prompt = speech_server.engine_client.generate.call_args.kwargs["prompt"]
+        additional = prompt["additional_information"]
+        assert additional["voice_name"] == "alice"
+        assert additional["voice_created_at"] == 0
+
     def test_build_tts_params(self, speech_server):
         """Test TTS parameter building."""
         req = OpenAICreateSpeechRequest(input="Hello", voice="Ryan", language="English")
@@ -2497,6 +2749,95 @@ class TestTTSAsyncOffloading:
         qwen3_tts_server._build_tts_params.assert_called_once()
         qwen3_tts_server._estimate_prompt_len_async.assert_awaited_once()
 
+    def test_qwen3_repeated_ref_audio_hot_path_sends_cache_key_without_waveform(self, qwen3_tts_server):
+        """After a ref artifact is marked ready, repeated requests avoid ref_audio payload IPC."""
+        wav_list = [0.0] * 48000
+        artifact_key = "a" * 40
+        ref_audio = "data:audio/wav;base64,same"
+        qwen3_tts_server._put_resolved_ref_audio(
+            hashlib.sha1(ref_audio.encode("utf-8")).hexdigest(),
+            wav_list,
+            24000,
+            artifact_key,
+        )
+        qwen3_tts_server._ref_audio_model_artifact_ready.add(artifact_key)
+        qwen3_tts_server._codec_frame_rate = 25.0
+        qwen3_tts_server._tts_tokenizer = lambda _text, padding=False: {"input_ids": list(range(10))}
+        qwen3_tts_server.engine_client.model_config.hf_config.talker_config = SimpleNamespace(
+            codec_language_id={},
+            spk_is_dialect={},
+        )
+
+        request = OpenAICreateSpeechRequest(
+            input="hello",
+            task_type="Base",
+            ref_audio=ref_audio,
+            ref_text="reference",
+        )
+        request_id, _generator, tts_params = asyncio.run(
+            qwen3_tts_server._prepare_speech_generation(request, request_id="req-hot")
+        )
+
+        assert request_id == "req-hot"
+        assert "ref_audio" not in tts_params
+        assert tts_params["_qwen3_tts_ref_audio_cache_key"] == [artifact_key]
+        assert tts_params["ref_code_length"] == [50]
+        prompt = qwen3_tts_server.engine_client.generate.call_args.kwargs["prompt"]
+        assert prompt["additional_information"] is tts_params
+
+    def test_qwen3_ref_audio_artifact_ready_is_evicted_with_resolve_cache(self, qwen3_tts_server):
+        qwen3_tts_server._ref_audio_resolve_cache_max_entries = 1
+        qwen3_tts_server._ref_audio_resolve_cache_max_bytes = 1_000_000
+
+        qwen3_tts_server._put_resolved_ref_audio("ref-a", [0.0] * 8, 24000, "artifact-a")
+        qwen3_tts_server._ref_audio_model_artifact_ready.add("artifact-a")
+        qwen3_tts_server._put_resolved_ref_audio("ref-b", [0.0] * 8, 24000, "artifact-b")
+
+        assert "artifact-a" not in qwen3_tts_server._ref_audio_model_artifact_ready
+        assert "artifact-b" in {entry[3] for entry in qwen3_tts_server._ref_audio_resolve_cache.values()}
+
+    @pytest.mark.asyncio
+    async def test_generate_audio_chunks_discards_ref_audio_artifact_warmup_on_error(self, qwen3_tts_server):
+        async def failing_generator():
+            raise ValueError("boom")
+            yield  # pragma: no cover
+
+        qwen3_tts_server._request_ref_audio_artifact_keys["req-fail"] = "artifact-fail"
+
+        with pytest.raises(ValueError, match="boom"):
+            await anext(qwen3_tts_server._generate_audio_chunks(failing_generator(), "req-fail"))
+
+        assert "req-fail" not in qwen3_tts_server._request_ref_audio_artifact_keys
+        assert "artifact-fail" not in qwen3_tts_server._ref_audio_model_artifact_ready
+
+    @pytest.mark.asyncio
+    async def test_generate_audio_chunks_discards_ref_audio_artifact_warmup_on_close(self, qwen3_tts_server):
+        async def pcm_generator():
+            yield SimpleNamespace(
+                multimodal_output={
+                    "audio": torch.zeros(16, dtype=torch.float32),
+                    "sr": 24000,
+                }
+            )
+            await asyncio.sleep(0)
+
+        qwen3_tts_server._request_ref_audio_artifact_keys["req-close"] = "artifact-close"
+
+        stream = qwen3_tts_server._generate_audio_chunks(pcm_generator(), "req-close")
+        assert await anext(stream)
+        await stream.aclose()
+
+        assert "req-close" not in qwen3_tts_server._request_ref_audio_artifact_keys
+        assert "artifact-close" not in qwen3_tts_server._ref_audio_model_artifact_ready
+
+    def test_qwen3_ref_audio_artifact_ready_requires_live_resolve_cache_entry(self, qwen3_tts_server):
+        qwen3_tts_server._request_ref_audio_artifact_keys["req-evicted"] = "artifact-evicted"
+
+        qwen3_tts_server._mark_ref_audio_artifact_ready_for_request("req-evicted")
+
+        assert "req-evicted" not in qwen3_tts_server._request_ref_audio_artifact_keys
+        assert "artifact-evicted" not in qwen3_tts_server._ref_audio_model_artifact_ready
+
     def test_shutdown_is_idempotent(self, mocker: MockerFixture):
         """Calling shutdown() twice should not raise."""
         mocker.patch.object(OmniOpenAIServingSpeech, "_load_supported_speakers", return_value=set())

tests/entrypoints/test_stage_utils.py

@@ -4,7 +4,7 @@ import sys
 import pytest
 from pytest_mock import MockerFixture
 
-from vllm_omni.entrypoints.stage_utils import set_stage_devices
+from vllm_omni.entrypoints.stage_utils import _map_device_list, set_stage_devices
 
 
 def _make_dummy_torch(call_log):
@@ -130,3 +130,90 @@ def test_set_stage_devices_npu_platform(mocker: MockerFixture, monkeypatch: pyte
     set_stage_devices(stage_id=0, devices="0,1")
 
     assert os.environ["ASCEND_RT_VISIBLE_DEVICES"] == "4,5"
+
+
+# ---- _map_device_list unit tests ----
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_idempotency():
+    """Device IDs already in the visible set are returned as-is (idempotency)."""
+    result = _map_device_list(0, ["0", "1"], ["0", "1", "2", "3"])
+    assert result == ["0", "1"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_idempotency_single():
+    """Single device ID in visible set passes through."""
+    result = _map_device_list(1, ["1"], ["0", "1"])
+    assert result == ["1"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_index_mapping():
+    """Device IDs < num_visible not in set are treated as indices."""
+    result = _map_device_list(0, ["0", "1"], ["6", "7", "8", "9"])
+    assert result == ["6", "7"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_physical_fallback():
+    """Device IDs >= num_visible are treated as physical IDs and passed through."""
+    result = _map_device_list(1, ["2"], ["0", "1"])
+    assert result == ["2"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_physical_fallback_multi():
+    """Multiple device IDs >= num_visible all pass through as physical IDs."""
+    result = _map_device_list(1, ["2", "3"], ["0", "1"])
+    assert result == ["2", "3"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_physical_fallback_mixed():
+    """When some devices are >= num_visible but some < num_visible, index mapping applies for the subset."""
+    # "1" < num_visible(2) so index mapping applies: visible[1] = "5"
+    # "2" >= num_visible(2) so it's dropped via the partial mapping path
+    result = _map_device_list(1, ["1", "2"], ["0", "5"])
+    assert result == ["5"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_index_with_gaps():
+    """Index-based mapping works with non-contiguous visible device lists."""
+    result = _map_device_list(0, ["0", "1"], ["4", "5", "7"])
+    assert result == ["4", "5"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_index_mapping_no_idempotency():
+    """Device IDs < num_visible not in visible set map via index even when no ID matches literally."""
+    # "0","1" with visible ["2","3"]: idempotency check fails (none in set),
+    # but index mapping succeeds: visible[0]="2", visible[1]="3"
+    result = _map_device_list(0, ["0", "1"], ["2", "3"])
+    assert result == ["2", "3"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_partial_mapping():
+    """When only a subset can map, returns the mapped subset (no error)."""
+    result = _map_device_list(0, ["0", "1", "2"], ["5", "6"])
+    assert result == ["5", "6"]
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_map_device_list_raises_on_non_numeric():
+    """Non-numeric device IDs raise ValueError."""
+    with pytest.raises(ValueError, match="must be non-negative integers"):
+        _map_device_list(0, ["a"], ["0", "1"])

tests/helpers/env.py

@@ -160,9 +160,41 @@ def _print_gpu_processes() -> None:
     print("=" * 80)
 
 
+def _cleanup_stale_device_locks() -> None:
+    """Remove stale device-initialization lock files whose recorded PID is dead.
+
+    Lock files at ``/tmp/vllm_omni_device_*_init.lock`` may persist after a
+    crashed / killed test run and block subsequent orchestrator startups.
+    """
+    import glob as _glob
+
+    for lock_file in _glob.glob("/tmp/vllm_omni_device_*_init.lock"):
+        try:
+            with open(lock_file) as fh:
+                content = fh.read().strip()
+            if not content:
+                continue
+            pid = int(content)
+        except (OSError, ValueError):
+            continue
+
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            print(f"  Removing stale device lock {lock_file} (PID {pid} is dead)")
+            try:
+                os.unlink(lock_file)
+            except OSError:
+                pass
+        except PermissionError:
+            pass
+
+
 def run_pre_test_cleanup() -> None:
     print("Pre-test GPU status:")
 
+    _cleanup_stale_device_locks()
+
     num_gpus = current_omni_platform.device_count()
     if num_gpus > 0:
         try:

tests/helpers/mark.py

@@ -21,7 +21,7 @@ def cuda_marks(*, res: str, num_cards: int):
         return marks
     test_distributed = pytest.mark.distributed_cuda(num_cards=num_cards)
 
-    test_skipif = pytest.mark.skipif_cuda(
+    test_skipif = pytest.mark.skipif(
         not current_platform.is_cuda() or (current_platform.device_count() < num_cards),
         reason=f"Need at least {num_cards} CUDA GPUs to run the test.",
     )
@@ -52,7 +52,7 @@ def xpu_marks(*, res: str, num_cards: int):
         return marks
     test_distributed = pytest.mark.distributed_xpu(num_cards=num_cards)
 
-    test_skipif = pytest.mark.skipif_xpu(
+    test_skipif = pytest.mark.skipif(
         not current_platform.is_xpu() or (current_platform.device_count() < num_cards),
         reason=f"Need at least {num_cards} XPUs to run the test.",
     )

tests/helpers/runtime.py

@@ -305,40 +305,132 @@ class OmniServer:
             time.sleep(2)
         raise RuntimeError(f"Server failed to start within {max_wait} seconds")
 
+    @staticmethod
+    def _reap_zombie(proc: "psutil.Process") -> bool:
+        """Reap a zombie child process via ``os.waitpid``.
+
+        ``psutil.Process.wait()`` uses ``pidfd_open`` + ``poll()``, which
+        never fires for a zombie (the zombie is already dead and will not
+        change state).  Since the test process is the parent, we can reap
+        the zombie directly with ``os.waitpid(pid, os.WNOHANG)`` and
+        retrieve its exit code.
+
+        Returns True if the process was a zombie and was reaped.
+        """
+        if proc.status() != psutil.STATUS_ZOMBIE:
+            return False
+        try:
+            os.waitpid(proc.pid, os.WNOHANG)
+        except ChildProcessError:
+            pass
+        return True
+
+    @staticmethod
+    def _wait_or_reap(proc: "psutil.Process", timeout: int) -> None:
+        """Wait for *proc* to exit, handling zombie state transparently.
+
+        When the process is already a zombie (e.g. orchestrator thread
+        hung during shutdown and the main process exited), ``pidfd_open``
+        + ``poll()`` inside ``psutil`` will never see a state change.
+        Fall back to ``os.waitpid`` to reap the zombie.
+        """
+        if OmniServer._reap_zombie(proc):
+            return
+        try:
+            proc.wait(timeout=timeout)
+        except psutil.TimeoutExpired:
+            OmniServer._reap_zombie(proc)
+
     def _kill_process_tree(self, pid):
+        """Kill the process tree rooted at *pid*.
+
+        Terminate the parent **first** so the OmniServer can gracefully shut
+        down its stage-engine children through the orchestrator.  This avoids
+        the ``subprocess died unexpectedly`` ERROR that the APIServer monitor
+        thread logs when children are killed before the parent, which in turn
+        can cause CI watchdogs to false-trigger on the upstream ``Shutdown
+        initiated`` message.
+
+        When the parent does not exit within the grace period (e.g. CPU-
+        offloaded workers stuck in CUDA D-state), the method falls back to
+        killing children first so the parent can be reaped cleanly.
+        """
         try:
             parent = psutil.Process(pid)
             children = parent.children(recursive=True)
-            all_pids = [pid] + [child.pid for child in children]
-
-            for child in children:
-                try:
-                    child.terminate()
-                except psutil.NoSuchProcess:
-                    pass
-
-            _, still_alive = psutil.wait_procs(children, timeout=10)
-
-            for child in still_alive:
-                try:
-                    child.kill()
-                except psutil.NoSuchProcess:
-                    pass
 
+            # 1. Terminate the parent first — let it run its graceful
+            #    shutdown cascade (orchestrator → stage pools → engine cores).
             try:
                 parent.terminate()
-                parent.wait(timeout=10)
-            except (psutil.NoSuchProcess, psutil.TimeoutExpired):
+            except psutil.NoSuchProcess:
+                pass
+
+            # 2. Give the parent time to shut down its children cleanly.
+            parent_exited = False
+            try:
+                parent.wait(timeout=15)
+                parent_exited = True
+            except psutil.NoSuchProcess:
+                parent_exited = True
+            except psutil.TimeoutExpired:
+                parent_exited = OmniServer._reap_zombie(parent)
+
+            if not parent_exited:
+                # Parent is stuck — children (e.g. CPU-offloaded CFG workers)
+                # are likely in uninterruptible sleep.  Kill children first
+                # so the parent can be reaped without lingering as a zombie.
+                for child in children:
+                    try:
+                        child.kill()
+                    except psutil.NoSuchProcess:
+                        pass
+                psutil.wait_procs(children, timeout=5)
                 try:
                     parent.kill()
                 except psutil.NoSuchProcess:
                     pass
+                OmniServer._wait_or_reap(parent, timeout=5)
+            else:
+                # Parent exited cleanly — clean up any remaining children.
+                for child in children:
+                    try:
+                        if child.is_running():
+                            child.terminate()
+                    except psutil.NoSuchProcess:
+                        pass
 
+                gone, still_alive = psutil.wait_procs(children, timeout=10)
+
+                for child in still_alive:
+                    try:
+                        child.kill()
+                    except psutil.NoSuchProcess:
+                        pass
+
+                try:
+                    if parent.is_running() and not OmniServer._reap_zombie(parent):
+                        parent.kill()
+                        parent.wait(timeout=10)
+                except psutil.NoSuchProcess:
+                    pass
+
+            # 3. Final sweep — ``kill -9`` anything that escaped.
             time.sleep(1)
-            alive_processes = []
-            for check_pid in all_pids:
-                if psutil.pid_exists(check_pid):
-                    alive_processes.append(check_pid)
+            alive_processes: list[int] = []
+            for child in children:
+                try:
+                    if child.is_running():
+                        alive_processes.append(child.pid)
+                except psutil.NoSuchProcess:
+                    pass
+            # Only count the parent as alive if it is NOT a zombie
+            # (zombies are already dead — just waiting to be reaped).
+            try:
+                if parent.is_running() and parent.status() != psutil.STATUS_ZOMBIE:
+                    alive_processes.append(parent.pid)
+            except psutil.NoSuchProcess:
+                pass
 
             if alive_processes:
                 print(f"Warning: Processes still alive: {alive_processes}")

tests/model_executor/models/qwen3_tts/test_qwen3_tts_code2wav.py

@@ -25,6 +25,7 @@ class _FakeDecoder(nn.Module):
         self.total_upsample = total_upsample
         self.decode_calls: list[dict[str, int]] = []
         self.batched_decode_calls: list[dict[str, int]] = []
+        self.decode_codes: list[torch.Tensor] = []
         self.cudagraph_calls: list[dict[str, int | torch.device]] = []
 
     def to(self, *args, **kwargs):
@@ -37,6 +38,7 @@ class _FakeDecoder(nn.Module):
         chunk_size: int = 300,
         left_context_size: int = 25,
     ) -> torch.Tensor:
+        self.decode_codes.append(codes.detach().cpu().clone())
         self.decode_calls.append(
             {
                 "chunk_size": chunk_size,
@@ -181,6 +183,136 @@ def test_forward_trims_trailing_padding_without_context():
     torch.testing.assert_close(audio, expected)
 
 
+def test_forward_reuses_cached_ref_context_for_followup_chunk():
+    model = _make_model()
+
+    first_codes = torch.tensor(
+        [
+            9,
+            8,
+            1,
+            2,
+            9,
+            8,
+            3,
+            4,
+        ],
+        dtype=torch.long,
+    )
+    model.forward(
+        input_ids=first_codes,
+        runtime_additional_information=[
+            {
+                "meta": {
+                    "left_context_size": 2,
+                    "ref_context_size": 2,
+                    "ref_context_request_id": "rid",
+                    "ref_context_included": True,
+                }
+            }
+        ],
+    )
+
+    followup_codes = torch.tensor([5, 6, 7, 15, 16, 17], dtype=torch.long)
+    model.forward(
+        input_ids=followup_codes,
+        runtime_additional_information=[
+            {
+                "meta": {
+                    "left_context_size": 3,
+                    "ref_context_size": 2,
+                    "ref_context_request_id": "rid",
+                    "ref_context_included": False,
+                    "finished": torch.tensor(True),
+                }
+            }
+        ],
+    )
+
+    torch.testing.assert_close(
+        model.decoder.decode_codes[-1],
+        torch.tensor(
+            [
+                [
+                    [9, 8, 5, 6, 7],
+                    [9, 8, 15, 16, 17],
+                ]
+            ],
+            dtype=torch.long,
+        ),
+    )
+    assert "rid" not in model._ref_context_cache
+
+
+def test_forward_fails_fast_when_ref_context_cache_is_missing():
+    model = _make_model()
+
+    followup_codes = torch.tensor([5, 6, 7, 15, 16, 17], dtype=torch.long)
+    with pytest.raises(ValueError, match="Missing Qwen3-TTS ref context cache"):
+        model.forward(
+            input_ids=followup_codes,
+            runtime_additional_information=[
+                {
+                    "meta": {
+                        "left_context_size": 3,
+                        "ref_context_size": 2,
+                        "ref_context_request_id": "rid",
+                        "ref_context_included": False,
+                    }
+                }
+            ],
+        )
+
+
+def test_ref_context_cache_evicts_lru_entries_when_requests_abort():
+    model = _make_model()
+    model._ref_context_cache_max_entries = 2
+
+    first_codes = torch.tensor([9, 8, 1, 2, 9, 8, 3, 4], dtype=torch.long)
+    for rid in ("a", "b", "c"):
+        model.forward(
+            input_ids=first_codes,
+            runtime_additional_information=[
+                {
+                    "meta": {
+                        "left_context_size": 2,
+                        "ref_context_size": 2,
+                        "ref_context_request_id": rid,
+                        "ref_context_included": True,
+                    }
+                }
+            ],
+        )
+
+    assert list(model._ref_context_cache) == ["b", "c"]
+    assert model._ref_context_cache_bytes == sum(model._tensor_nbytes(t) for t in model._ref_context_cache.values())
+
+
+def test_ref_context_cache_evicts_to_byte_cap():
+    model = _make_model()
+    model._ref_context_cache_max_entries = 100
+    model._ref_context_cache_max_bytes = 33
+
+    first_codes = torch.tensor([9, 8, 1, 2, 9, 8, 3, 4], dtype=torch.long)
+    for rid in ("a", "b"):
+        model.forward(
+            input_ids=first_codes,
+            runtime_additional_information=[
+                {
+                    "meta": {
+                        "left_context_size": 2,
+                        "ref_context_size": 2,
+                        "ref_context_request_id": rid,
+                        "ref_context_included": True,
+                    }
+                }
+            ],
+        )
+
+    assert list(model._ref_context_cache) == ["b"]
+    assert model._ref_context_cache_bytes == 32
+
+
 def test_connector_codec_chunking_does_not_override_decode_chunking():
     model = _make_model(
         async_chunk=True,

tests/model_executor/models/qwen3_tts/test_qwen3_tts_talker_preprocess.py

@@ -2,6 +2,7 @@ from collections import OrderedDict
 from types import SimpleNamespace
 
 import numpy as np
+import pytest
 import torch
 
 from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import (
@@ -9,6 +10,7 @@ from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import (
     PRECOMPUTED_REF_CODE_KEY,
     PRECOMPUTED_REF_IDS_KEY,
     PRECOMPUTED_TEXT_IDS_KEY,
+    REF_AUDIO_CACHE_KEY,
     Qwen3TTSPromptEmbedsBuilder,
 )
 from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_talker import (
@@ -493,6 +495,41 @@ def test_base_voice_clone_batch_preprocess_skips_after_initial_prefill_state_exi
     assert NORMALIZED_REF_AUDIO_KEY not in buf["r1"]
 
 
+def test_base_voice_clone_batch_preprocess_uses_serving_artifact_cache_key_without_normalize():
+    builder = _make_minimal_builder()
+    ref_code = torch.full((2, 2), 5, dtype=torch.long)
+    builder.put_ref_audio_artifacts("same-ref", ref_code=ref_code)
+    builder.normalize_ref_audio = lambda _raw: (_ for _ in ()).throw(AssertionError("normalize not expected"))
+    builder._encode_ref_audio_batch_fn = lambda *a, **kw: (_ for _ in ()).throw(
+        AssertionError("speech tokenizer not expected")
+    )
+
+    class FakeTextTokenizer:
+        def __call__(self, texts, *, padding=False):
+            return {"input_ids": [[7, 8, 9] for _ in texts]}
+
+    builder._text_tokenizer = FakeTextTokenizer()
+    buf = {
+        "r1": {
+            "task_type": ["Base"],
+            "text": ["one"],
+            "ref_audio": ["a.wav"],
+            "ref_text": ["hello"],
+            "x_vector_only_mode": [False],
+            REF_AUDIO_CACHE_KEY: ["same-ref"],
+        }
+    }
+
+    builder.preprocess_batch(
+        req_ids=["r1"],
+        model_intermediate_buffer=buf,
+        device=torch.device("cpu"),
+    )
+
+    assert torch.equal(buf["r1"]["codes"][PRECOMPUTED_REF_CODE_KEY], ref_code)
+    assert NORMALIZED_REF_AUDIO_KEY not in buf["r1"]
+
+
 def test_base_voice_clone_uses_batched_ref_code_without_serial_encode():
     builder = _make_minimal_builder()
     device_param = torch.nn.Parameter(torch.empty(0))
@@ -536,3 +573,112 @@ def test_base_voice_clone_uses_batched_ref_code_without_serial_encode():
     assert speaker_wav_ids == [id(ref_audio)]
     assert ref_code_len == 2
     assert torch.equal(out_ref_code, ref_code)
+
+
+def test_voice_clone_prompt_artifacts_do_not_resolve_ref_audio_for_cache_fill():
+    builder = _make_minimal_builder()
+    device_param = torch.nn.Parameter(torch.empty(0))
+    builder._text_embedding = _stub_text_embedding(device_param)
+    builder._text_projection = lambda embeds: embeds
+    builder._codec_embed = lambda ids: torch.zeros((*ids.shape, 4), device=ids.device)
+
+    class FakeTokenizer:
+        def __call__(self, *_args, **_kwargs):
+            return {"input_ids": torch.arange(8, dtype=torch.long).reshape(1, -1)}
+
+    builder._text_tokenizer = FakeTokenizer()
+    builder._generate_icl_prompt = lambda **kwargs: (
+        torch.ones((1, 2, 4), device=kwargs["ref_code"].device),
+        torch.ones((1, 4), device=kwargs["ref_code"].device),
+    )
+    builder.normalize_ref_audio = lambda _raw: (_ for _ in ()).throw(AssertionError("normalize not expected"))
+    builder.encode_ref_audio_to_code = lambda _wav, _sr: (_ for _ in ()).throw(AssertionError("encode not expected"))
+    builder.extract_speaker_embedding = lambda _wav, _sr: (_ for _ in ()).throw(
+        AssertionError("speaker embedding not expected")
+    )
+    ref_code = torch.arange(4, dtype=torch.long).reshape(2, 2)
+
+    _prompt, _trailing, ref_code_len, out_ref_code = builder.build_prompt_embeds(
+        task_type="Base",
+        info_dict={
+            "text": ["hello"],
+            "ref_audio": ["ref.wav"],
+            "ref_ids": torch.arange(8, dtype=torch.long).reshape(1, -1),
+            "non_streaming_mode": [False],
+            "voice_clone_prompt": {
+                "ref_code": ref_code,
+                "ref_spk_embedding": torch.ones(4, dtype=torch.bfloat16),
+                "icl_mode": True,
+            },
+        },
+    )
+
+    assert ref_code_len == 2
+    assert torch.equal(out_ref_code, ref_code)
+
+
+def test_ref_audio_artifact_only_uses_cache_without_ref_audio_payload():
+    builder = _make_minimal_builder()
+    device_param = torch.nn.Parameter(torch.empty(0))
+    builder._text_embedding = _stub_text_embedding(device_param)
+    builder._text_projection = lambda embeds: embeds
+    builder._codec_embed = lambda ids: torch.zeros((*ids.shape, 4), device=ids.device)
+
+    class FakeTokenizer:
+        def __call__(self, *_args, **_kwargs):
+            return {"input_ids": torch.arange(8, dtype=torch.long).reshape(1, -1)}
+
+    builder._text_tokenizer = FakeTokenizer()
+    builder._generate_icl_prompt = lambda **kwargs: (
+        torch.ones((1, 2, 4), device=kwargs["ref_code"].device),
+        torch.ones((1, 4), device=kwargs["ref_code"].device),
+    )
+    builder.normalize_ref_audio = lambda _raw: (_ for _ in ()).throw(AssertionError("normalize not expected"))
+    builder.encode_ref_audio_to_code = lambda _wav, _sr: (_ for _ in ()).throw(AssertionError("encode not expected"))
+    builder.extract_speaker_embedding = lambda _wav, _sr: (_ for _ in ()).throw(
+        AssertionError("speaker embedding not expected")
+    )
+    ref_code = torch.arange(4, dtype=torch.long).reshape(2, 2)
+    builder.put_ref_audio_artifacts(
+        "same-ref",
+        ref_code=ref_code,
+        ref_spk_embedding=torch.ones(4, dtype=torch.bfloat16),
+    )
+
+    _prompt, _trailing, ref_code_len, out_ref_code = builder.build_prompt_embeds(
+        task_type="Base",
+        info_dict={
+            "text": ["hello"],
+            "ref_ids": torch.arange(8, dtype=torch.long).reshape(1, -1),
+            "non_streaming_mode": [False],
+            REF_AUDIO_CACHE_KEY: ["same-ref"],
+        },
+    )
+
+    assert ref_code_len == 2
+    assert torch.equal(out_ref_code, ref_code)
+
+
+def test_ref_audio_artifact_only_cache_miss_fails_fast():
+    builder = _make_minimal_builder()
+    device_param = torch.nn.Parameter(torch.empty(0))
+    builder._text_embedding = _stub_text_embedding(device_param)
+    builder._text_projection = lambda embeds: embeds
+    builder._codec_embed = lambda ids: torch.zeros((*ids.shape, 4), device=ids.device)
+
+    class FakeTokenizer:
+        def __call__(self, *_args, **_kwargs):
+            return {"input_ids": torch.arange(8, dtype=torch.long).reshape(1, -1)}
+
+    builder._text_tokenizer = FakeTokenizer()
+
+    with pytest.raises(RuntimeError, match="artifact cache miss"):
+        builder.build_prompt_embeds(
+            task_type="Base",
+            info_dict={
+                "text": ["hello"],
+                "ref_ids": torch.arange(8, dtype=torch.long).reshape(1, -1),
+                "non_streaming_mode": [False],
+                REF_AUDIO_CACHE_KEY: ["missing-ref"],
+            },
+        )

tests/model_executor/models/qwen3_tts/test_qwen3_tts_talker_prompt_len.py

@@ -79,6 +79,27 @@ def test_estimate_prompt_len_handles_1d_ref_code() -> None:
     assert est > 50, f"got {est}; 1D ref_code must contribute its own length"
 
 
+def test_estimate_prompt_len_uses_ref_code_length_without_ref_audio() -> None:
+    est = Qwen3TTSPromptEmbedsBuilder.estimate_prompt_len_from_additional_information(
+        additional_information={
+            "task_type": ["Base"],
+            "text": ["hello"],
+            "ref_text": ["reference transcript"],
+            "x_vector_only_mode": [False],
+            "ref_code_length": [6],
+            "_qwen3_tts_ref_audio_cache_key": ["same-ref"],
+            "language": ["English"],
+        },
+        task_type="Base",
+        tokenize_prompt=lambda _text: list(range(10)),
+        codec_language_id=None,
+        spk_is_dialect=None,
+        estimate_ref_code_len=lambda _ref_audio: None,
+    )
+
+    assert est == 15
+
+
 def test_estimate_prompt_len_uses_list_of_int_ref_ids_from_voice_clone_prompt() -> None:
     """`voice_clone_prompt.ref_ids` as a list-of-int (the natural pre-tokenized shape)
     must be read as a sequence, not unwrapped to its first element. Without the fix

tests/model_executor/stage_input_processors/test_qwen3_tts_async_chunk.py

@@ -269,11 +269,14 @@ def test_first_streaming_chunk_prepends_ref_code_context():
 
     assert payload is not None
     assert payload.meta.left_context_size == 2
+    assert payload.meta.ref_context_size == 2
+    assert payload.meta.ref_context_request_id == rid
+    assert payload.meta.ref_context_included is True
     assert len(payload.codes.audio) == _Q * 12
 
 
-def test_ref_code_context_applies_to_all_streaming_chunks():
-    """ref_code is prepended as decoder context on every chunk, not just the first."""
+def test_followup_ref_code_context_is_sent_as_metadata_handle():
+    """Follow-up chunks keep full ref context semantically without resending it."""
     tm = _tm()
     rid = "r-ref2"
     tm.code_prompt_token_ids[rid] = [_FRAME[:] for _ in range(35)]
@@ -289,9 +292,13 @@ def test_ref_code_context_applies_to_all_streaming_chunks():
     )
 
     assert payload is not None
-    # ref_code (2 frames) prepended as left context on second chunk too
+    # ref_code (2 frames) is represented in metadata so Code2Wav can restore it
+    # from its request-local cache. It must not be resent in codes.audio.
     assert payload.meta.left_context_size == 10 + 2
-    assert len(payload.codes.audio) == _Q * (35 + 2)
+    assert payload.meta.ref_context_size == 2
+    assert payload.meta.ref_context_request_id == rid
+    assert payload.meta.ref_context_included is False
+    assert len(payload.codes.audio) == _Q * 35
 
 
 def test_streaming_ref_code_context_is_bounded_for_batchable_shapes():

tests/test_arg_utils.py

@@ -164,7 +164,8 @@ def test_split_mixed_real_world():
     # orchestrator-only keys never reach engine
     assert "stage_init_timeout" not in engine
     assert "deploy_config" not in engine
-    assert "async_chunk" not in engine
+    # async_chunk is in SHARED_FIELDS so it flows to both
+    assert "async_chunk" in engine
 
 
 # ============================================================================
@@ -172,24 +173,47 @@ def test_split_mixed_real_world():
 # ============================================================================
 
 
-def test_user_typed_unclassified_warns(caplog):
+def test_user_typed_unclassified_warns(monkeypatch):
     """If the user types a flag we can't route, warn — don't silently drop."""
-    raw = {"bogus_flag": "value", "max_num_seqs": 64}
-    with caplog.at_level(logging.WARNING, logger="vllm_omni.engine.arg_utils"):
+    import io
+
+    buf = io.StringIO()
+    vllm_logger = logging.getLogger("vllm")
+    orig_handlers = list(vllm_logger.handlers)
+    vllm_logger.handlers.clear()
+    handler = logging.StreamHandler(buf)
+    handler.setLevel(logging.WARNING)
+    vllm_logger.addHandler(handler)
+    try:
+        raw = {"bogus_flag": "value", "max_num_seqs": 64}
         split_kwargs(raw, engine_cls=_FakeEngineArgs, user_typed={"bogus_flag"})
-    assert any("bogus_flag" in rec.message for rec in caplog.records), (
-        f"Expected warning mentioning 'bogus_flag', got: {[rec.message for rec in caplog.records]}"
-    )
+        assert "bogus_flag" in buf.getvalue(), f"Expected 'bogus_flag' in log output, got: {buf.getvalue()!r}"
+    finally:
+        vllm_logger.handlers.clear()
+        for h in orig_handlers:
+            vllm_logger.addHandler(h)
 
 
-def test_unclassified_without_user_typed_silent(caplog):
+def test_unclassified_without_user_typed_silent():
     """Without user_typed, unclassified keys drop silently (argparse defaults
     for server flags shouldn't spam logs on every launch)."""
-    raw = {"host": "0.0.0.0", "port": 8091, "max_num_seqs": 64}
-    with caplog.at_level(logging.WARNING, logger="vllm_omni.engine.arg_utils"):
+    import io
+
+    vllm_logger = logging.getLogger("vllm")
+    orig_handlers = list(vllm_logger.handlers)
+    vllm_logger.handlers.clear()
+    buf = io.StringIO()
+    handler = logging.StreamHandler(buf)
+    handler.setLevel(logging.WARNING)
+    vllm_logger.addHandler(handler)
+    try:
+        raw = {"host": "0.0.0.0", "port": 8091, "max_num_seqs": 64}
         split_kwargs(raw, engine_cls=_FakeEngineArgs, user_typed=None)
-    # No warnings because we don't know these were user-typed.
-    assert not any("host" in rec.message or "port" in rec.message for rec in caplog.records)
+        assert "host" not in buf.getvalue()
+    finally:
+        vllm_logger.handlers.clear()
+        for h in orig_handlers:
+            vllm_logger.addHandler(h)
 
 
 # ============================================================================
@@ -290,7 +314,7 @@ def test_internal_blacklist_keys_derived_from_orchestrator():
     # Spot-check expected entries
     assert "stage_init_timeout" in blacklist
     assert "deploy_config" in blacklist
-    assert "async_chunk" in blacklist
+    assert "async_chunk" not in blacklist  # now in SHARED_FIELDS
     # Shared fields must NOT appear — they flow to both orchestrator and engine
     assert "model" not in blacklist
     assert "log_stats" not in blacklist
@@ -339,18 +363,31 @@ def test_ambiguous_field_strict_raises():
         split_kwargs({"deploy_config": "x"}, engine_cls=_AmbiguousEngine, strict=True)
 
 
-def test_ambiguous_field_non_strict_routes_to_orchestrator(caplog):
+def test_ambiguous_field_non_strict_routes_to_orchestrator():
     """strict=False logs ERROR but routes the ambiguous field to orchestrator."""
 
     @dataclass
     class _AmbiguousEngine:
         deploy_config: str | None = None
 
-    with caplog.at_level(logging.ERROR, logger="vllm_omni.engine.arg_utils"):
+    import io
+
+    vllm_logger = logging.getLogger("vllm")
+    orig_handlers = list(vllm_logger.handlers)
+    vllm_logger.handlers.clear()
+    buf = io.StringIO()
+    handler = logging.StreamHandler(buf)
+    handler.setLevel(logging.ERROR)
+    vllm_logger.addHandler(handler)
+    try:
         orch, engine = split_kwargs({"deploy_config": "x"}, engine_cls=_AmbiguousEngine, strict=False)
-    assert orch.deploy_config == "x"
-    assert "deploy_config" not in engine
-    assert any("both OrchestratorArgs" in r.message for r in caplog.records)
+        assert orch.deploy_config == "x"
+        assert "deploy_config" not in engine
+        assert "both OrchestratorArgs" in buf.getvalue()
+    finally:
+        vllm_logger.handlers.clear()
+        for h in orig_handlers:
+            vllm_logger.addHandler(h)
 
 
 # Sentinel-default precedence invariants (#3035)
@@ -426,24 +463,20 @@ def test_help_text_preserves_default_after_nullify():
 
 _OMNIENGINEARGS_USER_INPUT_FIELDS = frozenset(
     {
-        "model_stage",
         "model_arch",
         "engine_output_type",
         "hf_config_name",
         "custom_process_next_stage_input_func",
         "subtalker_sampling_params",
-        "async_chunk",
         "omni_kv_config",
         "quantization_config",
         "worker_type",
         "task_type",
         "worker_cls",
-        "enable_sleep_mode",
         "omni_master_address",
         "omni_master_port",
         "stage_configs_path",
         "output_modalities",
-        "log_stats",
         "custom_pipeline_args",
     }
 )

tests/test_config_factory.py

@@ -1027,6 +1027,9 @@ class TestDeployConfigLoading:
             "async_scheduling",
             "compilation_config",
             "config_format",
+            "custom_voice_dir",
+            "data_parallel_size",
+            "devices",
             "disable_hybrid_kv_cache_manager",
             "enable_flashinfer_autotune",
             "enforce_eager",
@@ -1068,6 +1071,32 @@ class TestDeployConfigLoading:
             f"added={actual_fields - expected_fields}, removed={expected_fields - actual_fields}"
         )
 
+    def test_custom_voice_dir_is_pipeline_wide_engine_arg(self, tmp_path):
+        import vllm_omni.model_executor.models.qwen3_tts.pipeline  # noqa: F401
+        from vllm_omni.config.stage_config import load_deploy_config, merge_pipeline_deploy
+
+        deploy_path = tmp_path / "qwen3_tts_custom_voice.yaml"
+        custom_voice_dir = tmp_path / "voices"
+        deploy_path.write_text(
+            f"""
+async_chunk: true
+custom_voice_dir: {custom_voice_dir}
+stages:
+  - stage_id: 0
+    devices: "0"
+  - stage_id: 1
+    devices: "0"
+""",
+            encoding="utf-8",
+        )
+
+        deploy = load_deploy_config(deploy_path)
+        pipeline = _PIPELINE_REGISTRY["qwen3_tts"]
+        stages = merge_pipeline_deploy(pipeline, deploy)
+
+        assert deploy.custom_voice_dir == str(custom_voice_dir)
+        assert {s.yaml_engine_args.get("custom_voice_dir") for s in stages} == {str(custom_voice_dir)}
+
     def test_load_qwen3_omni_moe_deploy_config(self):
         deploy_path = Path(__file__).parent.parent / "vllm_omni" / "deploy" / "qwen3_omni_moe.yaml"
         deploy = load_deploy_config(deploy_path)

tests/test_tts_custom_voice_profiles.py

@@ -0,0 +1,283 @@
+import json
+from types import SimpleNamespace
+
+import pytest
+import torch
+from safetensors.torch import save_file
+
+from vllm_omni.utils.speaker_cache import SpeakerEmbeddingCache
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _write_manifest(root, *, model_type: str, voices: dict) -> None:
+    payload = {
+        "schema_version": 1,
+        "model_type": model_type,
+        "voices": voices,
+    }
+    (root / "custom_voice_manifest.json").write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_qwen3_custom_voice_profiles_warm_speaker_cache(tmp_path):
+    from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_talker import (
+        Qwen3TTSTalkerForConditionalGeneration,
+    )
+
+    save_file(
+        {
+            "speaker_embedding": torch.arange(4, dtype=torch.float32),
+            "ref_code": torch.arange(6, dtype=torch.int32).reshape(3, 2),
+        },
+        str(tmp_path / "alice.safetensors"),
+    )
+    _write_manifest(
+        tmp_path,
+        model_type="qwen3_tts",
+        voices={
+            "Alice": {
+                "file": "alice.safetensors",
+                "mode": "icl",
+                "ref_text": "reference transcript",
+                "embedding_dim": 4,
+            }
+        },
+    )
+
+    model = Qwen3TTSTalkerForConditionalGeneration.__new__(Qwen3TTSTalkerForConditionalGeneration)
+    model.config = SimpleNamespace(
+        custom_voice_dir=str(tmp_path),
+        speaker_encoder_config=SimpleNamespace(enc_dim=4),
+    )
+    model._speaker_cache = SpeakerEmbeddingCache()
+
+    model._load_custom_voice_profiles()
+
+    key = model._speaker_cache.make_cache_key("alice", model_type="qwen3_tts_icl", created_at=0)
+    cached = model._speaker_cache.get(key)
+    assert cached is not None
+    assert cached["icl_mode"] is True
+    assert cached["ref_text"] == "reference transcript"
+    torch.testing.assert_close(cached["ref_spk_embedding"], torch.arange(4, dtype=torch.float32))
+    torch.testing.assert_close(cached["ref_code"], torch.arange(6, dtype=torch.int32).reshape(3, 2))
+
+
+def test_qwen3_icl_profile_without_ref_code_is_not_downgraded(tmp_path):
+    from vllm_omni.model_executor.models.qwen3_tts.qwen3_tts_talker import (
+        Qwen3TTSTalkerForConditionalGeneration,
+    )
+
+    save_file({"speaker_embedding": torch.arange(4, dtype=torch.float32)}, str(tmp_path / "alice.safetensors"))
+    _write_manifest(
+        tmp_path,
+        model_type="qwen3_tts",
+        voices={
+            "Alice": {
+                "file": "alice.safetensors",
+                "mode": "icl",
+                "ref_text": "reference transcript",
+                "embedding_dim": 4,
+            }
+        },
+    )
+
+    model = Qwen3TTSTalkerForConditionalGeneration.__new__(Qwen3TTSTalkerForConditionalGeneration)
+    model.config = SimpleNamespace(
+        custom_voice_dir=str(tmp_path),
+        speaker_encoder_config=SimpleNamespace(enc_dim=4),
+    )
+    model._speaker_cache = SpeakerEmbeddingCache()
+
+    model._load_custom_voice_profiles()
+
+    icl_key = model._speaker_cache.make_cache_key("alice", model_type="qwen3_tts_icl", created_at=0)
+    xvec_key = model._speaker_cache.make_cache_key("alice", model_type="qwen3_tts_xvec", created_at=0)
+    assert model._speaker_cache.get(icl_key) is None
+    assert model._speaker_cache.get(xvec_key) is None
+
+
+def test_qwen3_prompt_len_uses_precomputed_ref_code_length():
+    from vllm_omni.model_executor.models.qwen3_tts.prompt_embeds_builder import (
+        Qwen3TTSPromptEmbedsBuilder,
+    )
+
+    prompt_len = Qwen3TTSPromptEmbedsBuilder.estimate_prompt_len_from_additional_information(
+        additional_information={
+            "text": ["hello"],
+            "task_type": ["Base"],
+            "speaker": ["alice"],
+            "x_vector_only_mode": [False],
+            "ref_code_length": [6],
+            "ref_text": ["reference transcript"],
+        },
+        task_type="Base",
+        tokenize_prompt=lambda _text: list(range(10)),
+        codec_language_id=None,
+        spk_is_dialect=None,
+        estimate_ref_code_len=lambda _ref_audio: None,
+    )
+
+    assert prompt_len == 15
+
+
+def test_custom_voice_profile_file_must_stay_under_manifest_dir(tmp_path):
+    from vllm_omni.utils.speaker_cache import iter_custom_voice_profiles
+
+    _write_manifest(
+        tmp_path,
+        model_type="qwen3_tts",
+        voices={"Alice": {"file": "../alice.safetensors", "mode": "xvec"}},
+    )
+
+    assert iter_custom_voice_profiles(tmp_path, expected_model_type="qwen3_tts") == []
+
+
+def test_custom_voice_profiles_drop_stale_provenance_metadata(tmp_path):
+    from vllm_omni.utils.speaker_cache import iter_custom_voice_profiles
+
+    save_file({"speaker_embedding": torch.arange(4, dtype=torch.float32)}, str(tmp_path / "alice.safetensors"))
+    _write_manifest(
+        tmp_path,
+        model_type="qwen3_tts",
+        voices={
+            "Alice": {
+                "file": "alice.safetensors",
+                "mode": "xvec",
+                "audio_codec_config_hash": "old-hash",
+                "audio_codec_version": "old-codec",
+                "model_revision": "old-revision",
+            }
+        },
+    )
+
+    profiles = iter_custom_voice_profiles(tmp_path, expected_model_type="qwen3_tts")
+
+    assert len(profiles) == 1
+    assert "audio_codec_config_hash" not in profiles[0]
+    assert "audio_codec_version" not in profiles[0]
+    assert "model_revision" not in profiles[0]
+
+
+def test_voxcpm2_custom_voice_profiles_warm_full_prompt_cache(tmp_path):
+    from vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker import (
+        VoxCPM2TalkerForConditionalGeneration,
+    )
+
+    save_file(
+        {
+            "ref_audio_feat": torch.ones(2, 3, 4, dtype=torch.float32),
+            "audio_feat": torch.full((5, 3, 4), 2.0, dtype=torch.float32),
+        },
+        str(tmp_path / "bob.safetensors"),
+    )
+    _write_manifest(
+        tmp_path,
+        model_type="voxcpm2",
+        voices={
+            "Bob": {
+                "file": "bob.safetensors",
+                "mode": "ref_continuation",
+                "prompt_text": "prompt transcript",
+                "ref_audio_feat_len": 2,
+                "audio_feat_len": 5,
+            }
+        },
+    )
+
+    model = VoxCPM2TalkerForConditionalGeneration.__new__(VoxCPM2TalkerForConditionalGeneration)
+    model.config = SimpleNamespace(custom_voice_dir=str(tmp_path))
+    model._speaker_cache = SpeakerEmbeddingCache()
+
+    model._load_custom_voice_profiles()
+
+    key = model._speaker_cache.make_cache_key("bob", model_type="voxcpm2", created_at=0)
+    cached = model._speaker_cache.get(key)
+    assert cached is not None
+    assert cached["mode"] == "ref_continuation"
+    assert cached["prompt_text"] == "prompt transcript"
+    torch.testing.assert_close(cached["ref_audio_feat"], torch.ones(2, 3, 4))
+    torch.testing.assert_close(cached["audio_feat"], torch.full((5, 3, 4), 2.0))
+
+
+def test_voxcpm2_custom_voice_profiles_skip_unloadable_manifest_entry(tmp_path):
+    from vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker import (
+        VoxCPM2TalkerForConditionalGeneration,
+    )
+
+    _write_manifest(
+        tmp_path,
+        model_type="voxcpm2",
+        voices={
+            "Bob": {
+                "file": "missing.safetensors",
+                "mode": "reference",
+                "ref_audio_feat_len": 2,
+            }
+        },
+    )
+
+    model = VoxCPM2TalkerForConditionalGeneration.__new__(VoxCPM2TalkerForConditionalGeneration)
+    model.config = SimpleNamespace(custom_voice_dir=str(tmp_path))
+    model._speaker_cache = SpeakerEmbeddingCache()
+
+    model._load_custom_voice_profiles()
+
+    key = model._speaker_cache.make_cache_key("bob", model_type="voxcpm2", created_at=0)
+    assert model._speaker_cache.get(key) is None
+
+
+def test_voxcpm2_precomputed_voice_cache_miss_raises_before_zero_shot():
+    from vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker import (
+        VoxCPM2TalkerForConditionalGeneration,
+    )
+
+    model = VoxCPM2TalkerForConditionalGeneration.__new__(VoxCPM2TalkerForConditionalGeneration)
+    model.config = SimpleNamespace(bos_token_id=1)
+    model._speaker_cache = SpeakerEmbeddingCache()
+    model._get_multichar_zh_split = lambda: {}
+    model._get_or_create_state = lambda _req_id: SimpleNamespace()
+
+    with pytest.raises(ValueError, match="not loaded in the model cache"):
+        model.preprocess(
+            torch.tensor([1, 2], dtype=torch.long),
+            None,
+            request_id="req-1",
+            text_token_ids=[[2]],
+            voice_name=["bob"],
+            voice_created_at=0,
+            voice_profile={"mode": "reference"},
+        )
+
+
+def test_voxcpm2_prompt_can_account_for_precomputed_profile_lengths():
+    from vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker import build_voxcpm2_prompt
+
+    class _Tokenizer:
+        bos_token_id = 1
+        unk_token_id = 0
+
+        def get_vocab(self):
+            return {}
+
+        def encode(self, text, add_special_tokens=True):
+            ids = [10 + i for i, _ in enumerate(text.split())]
+            return [self.bos_token_id, *ids] if add_special_tokens else ids
+
+    prompt = build_voxcpm2_prompt(
+        hf_config=SimpleNamespace(audio_vae_config={"sample_rate": 48000, "encoder_rates": [2]}, patch_size=3),
+        tokenizer=_Tokenizer(),
+        split_map={},
+        text="hello world",
+        voice_profile={
+            "mode": "ref_continuation",
+            "ref_audio_feat_len": 2,
+            "audio_feat_len": 5,
+            "prompt_text": "prompt words",
+        },
+    )
+
+    # text(2) + audio_start(1) + ref prefix(2 + ref_len) + prompt_text(2) + prompt_audio_len(5)
+    assert len(prompt["prompt_token_ids"]) == 14
+    additional = prompt["additional_information"]
+    assert additional["voice_profile"]["mode"] == "ref_continuation"
+    assert additional["prompt_text"] == ["prompt words"]

tools/nightly/generate_nightly_perf_html.py

@@ -250,7 +250,7 @@ def _collect_diffusion_records(input_dir: str) -> list[dict[str, Any]]:
 
 
 def _sort_omni_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    by_date_desc = sorted(records, key=lambda r: (r.get("date") or ""), reverse=True)
+    by_date_desc = sorted(records, key=lambda r: r.get("date") or "", reverse=True)
     return sorted(
         by_date_desc,
         key=lambda r: (
@@ -264,8 +264,8 @@ def _sort_omni_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
 
 
 def _sort_diffusion_records(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    by_date_desc = sorted(records, key=lambda r: (r.get("date") or ""), reverse=True)
-    return sorted(by_date_desc, key=lambda r: (r.get("test_name") or ""))
+    by_date_desc = sorted(records, key=lambda r: r.get("date") or "", reverse=True)
+    return sorted(by_date_desc, key=lambda r: r.get("test_name") or "")
 
 
 def _ensure_parent_dir(path: str) -> None:

vllm_omni/config/stage_config.py

@@ -516,6 +516,7 @@ class DeployConfig:
     enable_chunked_prefill: bool | None = None
     data_parallel_size: int | None = None
     pipeline_parallel_size: int | None = None
+    custom_voice_dir: str | None = None
 
 
 _STAGE_RESERVED_KEYS = frozenset(
@@ -693,6 +694,7 @@ def load_deploy_config(path: str | Path) -> DeployConfig:
         "enable_chunked_prefill",
         "data_parallel_size",
         "pipeline_parallel_size",
+        "custom_voice_dir",
     ):
         if name in raw_dict:
             kwargs[name] = raw_dict[name]
@@ -814,6 +816,7 @@ _PIPELINE_WIDE_ENGINE_FIELDS: tuple[str, ...] = (
     "enable_chunked_prefill",
     "data_parallel_size",
     "pipeline_parallel_size",
+    "custom_voice_dir",
 )
 
 

vllm_omni/core/sched/omni_ar_scheduler.py

@@ -24,9 +24,11 @@ from vllm_omni.core.sched.omni_scheduling_coordinator import (
     OmniSchedulingCoordinator,
     uses_full_payload_input_coordinator,
 )
+from vllm_omni.core.sched.utils import omni_routed_experts_for_request
 from vllm_omni.distributed.omni_connectors.transfer_adapter.chunk_transfer_adapter import (
     OmniChunkTransferAdapter,
 )
+from vllm_omni.engine import OmniEngineCoreOutput
 from vllm_omni.engine.serialization import deserialize_additional_information
 from vllm_omni.outputs import OmniConnectorOutput
 
@@ -321,7 +323,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
                     req = self.requests.get(req_id)
                     if req is not None and not req.is_finished():
                         outputs[req.client_index].append(
-                            EngineCoreOutput(
+                            OmniEngineCoreOutput(
                                 request_id=req_id,
                                 new_token_ids=[],
                                 kv_transfer_params={"kv_ready": True},
@@ -417,11 +419,8 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
                     stopped = True
 
             if stopped:
-                if (
-                    model_runner_output.routed_experts_dict is not None
-                    and req_id in model_runner_output.routed_experts_dict
-                ):
-                    routed_experts = model_runner_output.routed_experts_dict[req_id]
+                if model_runner_output.routed_experts is not None:
+                    routed_experts = omni_routed_experts_for_request(model_runner_output.routed_experts, request)
 
                 # Capture finish_reason BEFORE _handle_stopped_request, which may
                 # reset the status to WAITING for streaming requests that continue.
@@ -452,7 +451,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
                     stopped_preempted_reqs.add(request)
 
             # Extract sample logprobs if needed.
-            if request.sampling_params is not None and request.sampling_params.logprobs is not None and logprobs:
+            if request.sampling_params is not None and request.sampling_params.num_logprobs is not None and logprobs:
                 new_logprobs = logprobs.slice_request(req_index, len(new_token_ids))
 
             if num_nans_in_logits is not None and req_id in num_nans_in_logits:
@@ -463,7 +462,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
             if new_token_ids or pooler_output is not None or kv_transfer_params or stopped:
                 # Add EngineCoreOutput for this Request.
                 outputs[request.client_index].append(
-                    EngineCoreOutput(
+                    OmniEngineCoreOutput(
                         request_id=req_id,
                         new_token_ids=new_token_ids,
                         finish_reason=finish_reason,
@@ -501,7 +500,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
             self.finish_requests(failed_kv_load_req_ids, RequestStatus.FINISHED_ERROR)
             for request in requests:
                 outputs[request.client_index].append(
-                    EngineCoreOutput(
+                    OmniEngineCoreOutput(
                         request_id=request.request_id,
                         new_token_ids=[],
                         finish_reason=request.get_finished_reason(),
@@ -633,6 +632,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         """
         req_id = session.request_id
         self._new_prompt_len_snapshot[req_id] = len(update.prompt_token_ids)
+        session.discard_latest_async_tokens = False
         outstanding_async_tokens = getattr(session, "num_output_placeholders", 0)
         if outstanding_async_tokens > 0:
             # Async scheduling may already have sampled the previous

vllm_omni/core/sched/omni_generation_scheduler.py

@@ -29,9 +29,11 @@ from vllm_omni.core.sched.omni_scheduling_coordinator import (
     uses_full_payload_input_coordinator,
 )
 from vllm_omni.core.sched.output import OmniCachedRequestData, OmniNewRequestData
+from vllm_omni.core.sched.utils import omni_routed_experts_for_request
 from vllm_omni.distributed.omni_connectors.transfer_adapter.chunk_transfer_adapter import (
     OmniChunkTransferAdapter,
 )
+from vllm_omni.engine import OmniEngineCoreOutput
 from vllm_omni.outputs import OmniConnectorOutput, OmniModelRunnerOutput
 
 logger = init_logger(__name__)
@@ -499,11 +501,8 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler):
                 stopped = True
 
             if stopped:
-                if (
-                    model_runner_output.routed_experts_dict is not None
-                    and req_id in model_runner_output.routed_experts_dict
-                ):
-                    routed_experts = model_runner_output.routed_experts_dict[req_id]
+                if model_runner_output.routed_experts is not None:
+                    routed_experts = omni_routed_experts_for_request(model_runner_output.routed_experts, request)
                 finish_reason = request.get_finished_reason()
                 finished = self._handle_stopped_request(request)
                 if not finished:
@@ -524,7 +523,7 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler):
                     stopped_running_reqs.add(request)
 
             # Extract sample logprobs if needed.
-            if request.sampling_params is not None and request.sampling_params.logprobs is not None and logprobs:
+            if request.sampling_params is not None and request.sampling_params.num_logprobs is not None and logprobs:
                 new_logprobs = logprobs.slice_request(req_index, len(new_token_ids))
 
             if new_token_ids and self.structured_output_manager.should_advance(request):
@@ -543,7 +542,7 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler):
             prompt_logprobs_tensors = prompt_logprobs_dict.get(req_id)
             if new_token_ids or pooler_output is not None or kv_transfer_params or stopped:
                 outputs[request.client_index].append(
-                    EngineCoreOutput(
+                    OmniEngineCoreOutput(
                         request_id=req_id,
                         new_token_ids=new_token_ids,
                         finish_reason=finish_reason,
@@ -580,7 +579,7 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler):
                         getattr(request, "external_req_id", None),
                     )
             outputs[request.client_index].append(
-                EngineCoreOutput(
+                OmniEngineCoreOutput(
                     request_id=request.request_id,
                     new_token_ids=[],
                     finish_reason=finish_reason,
@@ -607,7 +606,7 @@ class OmniGenerationScheduler(OmniSchedulerMixin, VLLMScheduler):
             self.finish_requests(failed_kv_load_req_ids, RequestStatus.FINISHED_ERROR)
             for request in requests:
                 outputs[request.client_index].append(
-                    EngineCoreOutput(
+                    OmniEngineCoreOutput(
                         request_id=request.request_id,
                         new_token_ids=[],
                         finish_reason=request.get_finished_reason(),

vllm_omni/core/sched/utils.py

@@ -0,0 +1,23 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Shared utilities for omni schedulers."""
+
+import numpy as np
+from vllm.v1.outputs import RoutedExpertsLists
+
+
+def omni_routed_experts_for_request(routed_experts: RoutedExpertsLists, request) -> np.ndarray | None:
+    """Extract per-request routed experts from RoutedExpertsLists using slot_mapping.
+
+    Matches upstream RoutedExpertsManager.get() pattern — filters routing_data
+    rows whose slot_mapping entries belong to this request's block_table.
+    """
+    if routed_experts is None:
+        return None
+    slots = getattr(request, "block_table", None)
+    if slots is None:
+        return None
+    slot_set = set(slots)
+    mask = np.isin(routed_experts.slot_mapping, list(slot_set))
+    data = routed_experts.routing_data[mask]
+    return data if data.size > 0 else None

vllm_omni/data_entry_keys.py

@@ -77,6 +77,9 @@ class OmniPayloadMeta(TypedDict, total=False):
     decode_flag: bool
     codec_streaming: bool
     ref_code_len: int
+    ref_context_size: int
+    ref_context_request_id: str
+    ref_context_included: bool
     talker_prefill_offset: int
 
 
@@ -161,6 +164,9 @@ class MetaStruct(_StructBase):
     decode_flag: bool | None = None
     codec_streaming: bool | None = None
     ref_code_len: int | None = None
+    ref_context_size: int | None = None
+    ref_context_request_id: str | None = None
+    ref_context_included: bool | None = None
     talker_prefill_offset: int | None = None
     codec_chunk_frames: int | None = None
     codec_left_context_frames: int | None = None

vllm_omni/diffusion/diffusion_engine.py

@@ -867,7 +867,7 @@ class DiffusionEngine:
                 worker_thread.join(timeout=10)
             if worker_thread.is_alive():
                 logger.warning(
-                    "Worker thread did not terminate within 10s; scheduler and executor shutdown will be skipped."
+                    "Worker thread did not terminate within 10s; scheduler and executor shutdown will be deferred."
                 )
                 return
             else:

vllm_omni/diffusion/model_loader/hub_prefetch.py

@@ -64,6 +64,7 @@ import contextlib
 import errno
 import logging
 import os
+import time
 from collections.abc import Iterable, Iterator
 
 logger = logging.getLogger(__name__)
@@ -112,9 +113,43 @@ def _safe_repo_filename(model: str) -> str:
     return model.replace("/", "__").replace(os.sep, "__") + ".lock"
 
 
+def _dotfile_lock_acquire(lock_dir: str, model: str, timeout: float = 300.0, poll_interval: float = 0.5) -> bool:
+    """Acquire an exclusive lock via atomic directory creation.
+
+    ``os.makedirs(path, exist_ok=False)`` is atomic on POSIX filesystems
+    (including Lustre and NFS), making it a reliable fallback when
+    ``fcntl.flock`` is unsupported (``ENOLCK`` on FSx/Lustre mounts
+    without the ``flock`` mount option).
+
+    Returns ``True`` if the lock was acquired, ``False`` on timeout.
+    The caller must call ``os.rmdir(lock_path)`` to release.
+    """
+    lock_path = os.path.join(lock_dir, _safe_repo_filename(model) + ".dir")
+
+    deadline = time.monotonic() + timeout
+    while True:
+        try:
+            os.makedirs(lock_path, exist_ok=False)
+            logger.info("Acquired dotfile prefetch lock for %s at %s", model, lock_path)
+            return True
+        except FileExistsError:
+            if time.monotonic() >= deadline:
+                logger.warning(
+                    "Timed out waiting for dotfile prefetch lock %s after %.0fs; proceeding unlocked",
+                    lock_path,
+                    timeout,
+                )
+                return False
+            time.sleep(poll_interval)
+
+
 @contextlib.contextmanager
 def _repo_prefetch_lock(model: str) -> Iterator[None]:
-    """Hold an exclusive ``fcntl.flock`` keyed on the HF repo id.
+    """Hold an exclusive lock keyed on the HF repo id.
+
+    Tries ``fcntl.flock`` first; falls back to an atomic directory-creation
+    dotfile lock when flock is unsupported (e.g. FSx for Lustre without the
+    ``flock`` mount option, which returns ``ENOLCK``).
 
     Why we still need a node-wide lock on top of ``snapshot_download``'s
     own per-blob ``.lock`` files:
@@ -131,71 +166,93 @@ def _repo_prefetch_lock(model: str) -> Iterator[None]:
     * Multiple unrelated processes on the same node (concurrent CI jobs,
       multiple ``OmniServer`` instances, every ``DiffusionWorker``
       subprocess from the multiproc executor) can enter this code path
-      in parallel. ``flock`` on a per-repo file makes the whole
+      in parallel. An exclusive lock makes the whole
       ``snapshot_download(...)`` call a true node-wide critical
       section, so the *first* entrant fully populates the cache and
       every subsequent ``from_pretrained`` sees a warm, complete tree.
 
-    Best-effort: if we cannot create / acquire the lock (no ``fcntl``,
-    read-only FS, NFS without flock support) we log and run the download
-    anyway. Worst case behaviour reverts to "snapshot_download +
-    transformers v5 race" which is exactly the pre-fix state.
+    Best-effort: if we cannot acquire any lock (no ``fcntl``, read-only
+    FS, dotfile lock timeout) we log and run the download anyway. Worst
+    case behaviour reverts to "snapshot_download + transformers v5 race"
+    which is exactly the pre-fix state.
     """
+    lock_dir = None
+    lock_path = None
+    dotfile_held = None
+    fd = None
+    flock_held = False
+
+    # --- fcntl.flock path ---
     try:
         import fcntl  # type: ignore[import-not-found]
     except ImportError:  # pragma: no cover - non-POSIX (Windows)
-        logger.debug("fcntl unavailable on this platform; running prefetch unlocked")
-        yield
-        return
+        fcntl = None
 
-    try:
-        lock_dir = _node_lock_dir()
-    except OSError as exc:
-        logger.warning("Could not allocate lock dir for prefetch of %s (%s); skipping flock", model, exc)
-        yield
-        return
-
-    lock_path = os.path.join(lock_dir, _safe_repo_filename(model))
-
-    try:
-        fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o644)
-    except OSError as exc:
-        logger.warning("Could not open prefetch lock %s (%s); proceeding without flock", lock_path, exc)
-        yield
-        return
-
-    locked = False
-    try:
+    if fcntl is not None:
         try:
-            fcntl.flock(fd, fcntl.LOCK_EX)
-            locked = True
-            logger.debug("Acquired prefetch flock for %s at %s", model, lock_path)
+            lock_dir = _node_lock_dir()
         except OSError as exc:
-            # FS without flock support (ENOLCK / EOPNOTSUPP / EINVAL).
-            # EACCES on some Lustre configs without ``flock`` mount opt.
-            if exc.errno in (errno.ENOLCK, errno.EOPNOTSUPP, errno.EACCES, errno.EINVAL):
-                logger.warning(
-                    "fcntl.flock not supported on '%s' (%s: %s); running prefetch unlocked",
-                    lock_path,
-                    type(exc).__name__,
-                    exc,
-                )
-            else:
-                raise
+            logger.warning("Could not allocate lock dir for prefetch of %s (%s); skipping flock", model, exc)
+            fcntl = None  # force dotfile fallback
+
+    if fcntl is not None and lock_dir is not None:
+        lock_path = os.path.join(lock_dir, _safe_repo_filename(model))
+        try:
+            fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o644)
+        except OSError as exc:
+            logger.warning("Could not open prefetch lock %s (%s); falling back to dotfile lock", lock_path, exc)
+            fd = None
+
+        if fd is not None:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX)
+                flock_held = True
+                logger.info("Acquired flock prefetch lock for %s at %s", model, lock_path)
+            except OSError as exc:
+                if exc.errno in (errno.ENOLCK, errno.EOPNOTSUPP, errno.EACCES, errno.EINVAL):
+                    logger.warning(
+                        "fcntl.flock not supported on '%s' (%s: %s); falling back to dotfile lock",
+                        lock_path,
+                        type(exc).__name__,
+                        exc,
+                    )
+                    os.close(fd)
+                    fd = None
+                else:
+                    raise
+
+    # --- dotfile fallback ---
+    if not flock_held:
+        if lock_dir is None:
+            try:
+                lock_dir = _node_lock_dir()
+            except OSError as exc:
+                logger.warning("Could not allocate lock dir (%s); running prefetch unlocked", exc)
+                yield
+                return
+
+        if _dotfile_lock_acquire(lock_dir, model):
+            dotfile_held = os.path.join(lock_dir, _safe_repo_filename(model) + ".dir")
+
+    try:
         yield
     finally:
-        if locked:
+        if flock_held and fd is not None:
             with contextlib.suppress(OSError):
                 fcntl.flock(fd, fcntl.LOCK_UN)
-        with contextlib.suppress(OSError):
-            os.close(fd)
+        if fd is not None:
+            with contextlib.suppress(OSError):
+                os.close(fd)
+        if dotfile_held is not None:
+            with contextlib.suppress(OSError):
+                os.rmdir(dotfile_held)
 
 
 def prefetch_subfolders(
     model: str,
     subfolders: Iterable[str],
     *,
-    local_files_only: bool = False,
+    local_files_only: bool | None = None,
     include_root_metadata: bool = True,
 ) -> None:
     """Materialise ``model``'s ``subfolders`` in the HF cache before loading.
@@ -206,17 +263,21 @@ def prefetch_subfolders(
         subfolders: Iterable of subfolder names (e.g. ``["text_encoder",
             "vae"]``) whose contents need to be fully present before any
             worker calls ``from_pretrained(subfolder=...)``.
-        local_files_only: When True, skip the prefetch entirely. The caller
-            has explicitly promised the cache is already populated (as happens
-            for local model checkouts), so hitting the network would defeat
-            the intent and may fail in air-gapped environments.
+        local_files_only: When ``True``, skip the prefetch entirely.
+            When ``None`` (default), auto-detect: skip if *model* is a
+            local directory, run otherwise.
         include_root_metadata: When True, also pull ``*.json`` at the repo
             root so ``model_index.json`` / ``config.json`` resolution during
             ``from_pretrained`` also hits a warm cache.
     """
+    if local_files_only is None:
+        local_files_only = os.path.isdir(model)
+
     if local_files_only or not model or os.path.isdir(model):
         return
 
+    logger.info("Prefetching %s subfolders: %s", model, list(subfolders))
+
     try:
         from huggingface_hub import snapshot_download
     except ImportError:  # pragma: no cover - huggingface_hub is a hard dep
@@ -251,6 +312,7 @@ def prefetch_subfolders(
                 repo_id=model,
                 allow_patterns=allow_patterns,
             )
+        logger.info("Prefetch complete for %s", model)
     except Exception as exc:
         # Best-effort: propagate only via logging. The subsequent
         # ``from_pretrained`` call will raise a clearer, call-site-specific
@@ -312,3 +374,31 @@ def _looks_like_auth_error(exc: BaseException) -> bool:
     # only load-bearing signal.
     msg = str(exc).lower()
     return "401 client error" in msg or "403 client error" in msg or "gatedrepo" in msg
+
+
+def retry_on_missing_shard(load_fn, *, max_retries: int = 3, base_delay: float = 5.0):
+    """Call *load_fn* with retry on the transformers v5 shard-resolution race.
+
+    When the prefetch lock cannot be acquired (e.g. flock unsupported on
+    the filesystem and dotfile lock times out), ``from_pretrained`` may
+    still hit the ``cached_files`` race. This wrapper retries with
+    exponential backoff when the OSError message matches the specific
+    "does not appear to have a file named" pattern.
+    """
+    for attempt in range(max_retries):
+        try:
+            return load_fn()
+        except OSError as exc:
+            if "does not appear to have a file" not in str(exc):
+                raise
+            if attempt == max_retries - 1:
+                raise
+            delay = base_delay * (attempt + 1)
+            logger.warning(
+                "from_pretrained failed with shard-resolution race (%s); retrying in %.1fs (attempt %d/%d)",
+                exc,
+                delay,
+                attempt + 1,
+                max_retries,
+            )
+            time.sleep(delay)

vllm_omni/diffusion/models/qwen_image/pipeline_qwen_image.py

@@ -26,7 +26,7 @@ from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
 from vllm_omni.diffusion.distributed.autoencoders.autoencoder_kl_qwenimage import DistributedAutoencoderKLQwenImage
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
-from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders
+from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders, retry_on_missing_shard
 from vllm_omni.diffusion.models.dmd2 import DMD2PipelineMixin
 from vllm_omni.diffusion.models.qwen_image.cfg_parallel import (
     QwenImageCFGParallelMixin,
@@ -275,21 +275,22 @@ class QwenImagePipeline(nn.Module, QwenImageCFGParallelMixin, DiffusionPipelineP
         self.device = get_local_device()
         model = od_config.model
         # Check if model is a local path
-        local_files_only = os.path.exists(model)
+        local_files_only = os.path.isdir(model)
 
         # See pipeline_qwen_image_edit_plus: guard against transformers v5
         # multi-worker race on partial subfolder shard sets (Buildkite #1043).
         prefetch_subfolders(
             model,
             ["scheduler", "text_encoder", "vae", "tokenizer"],
-            local_files_only=local_files_only,
         )
 
         self.scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(
             model, subfolder="scheduler", local_files_only=local_files_only
         )
-        self.text_encoder = Qwen2_5_VLForConditionalGeneration.from_pretrained(
-            model, subfolder="text_encoder", local_files_only=local_files_only
+        self.text_encoder = retry_on_missing_shard(
+            lambda: Qwen2_5_VLForConditionalGeneration.from_pretrained(
+                model, subfolder="text_encoder", local_files_only=local_files_only
+            )
         )
         # Qwen2.5-VL ships a vision tower that text-to-image does not use.
         # Drop it while the model is still on CPU, before moving to GPU, so

vllm_omni/diffusion/models/qwen_image/pipeline_qwen_image_edit.py

@@ -27,7 +27,7 @@ from vllm.model_executor.models.utils import AutoWeightsLoader
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
-from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders
+from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders, retry_on_missing_shard
 from vllm_omni.diffusion.models.interface import SupportImageInput
 from vllm_omni.diffusion.models.qwen_image.cfg_parallel import (
     QwenImageCFGParallelMixin,
@@ -243,22 +243,23 @@ class QwenImageEditPipeline(nn.Module, SupportImageInput, QwenImageCFGParallelMi
         model = od_config.model
 
         # Check if model is a local path
-        local_files_only = os.path.exists(model)
+        local_files_only = os.path.isdir(model)
 
         # See pipeline_qwen_image_edit_plus: guard against transformers v5
         # multi-worker race on partial subfolder shard sets (Buildkite #1043).
         prefetch_subfolders(
             model,
             ["scheduler", "text_encoder", "vae", "tokenizer", "processor"],
-            local_files_only=local_files_only,
         )
 
         self.scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(
             model, subfolder="scheduler", local_files_only=local_files_only
         )
-        self.text_encoder = Qwen2_5_VLForConditionalGeneration.from_pretrained(
-            model, subfolder="text_encoder", local_files_only=local_files_only
-        ).to(self.device)
+        self.text_encoder = retry_on_missing_shard(
+            lambda: Qwen2_5_VLForConditionalGeneration.from_pretrained(
+                model, subfolder="text_encoder", local_files_only=local_files_only
+            ).to(self.device)
+        )
 
         self.vae = AutoencoderKLQwenImage.from_pretrained(model, subfolder="vae", local_files_only=local_files_only).to(
             self.device
@@ -266,8 +267,8 @@ class QwenImageEditPipeline(nn.Module, SupportImageInput, QwenImageCFGParallelMi
         transformer_kwargs = get_transformer_config_kwargs(od_config.tf_model_config, QwenImageTransformer2DModel)
         self.transformer = QwenImageTransformer2DModel(od_config=od_config, **transformer_kwargs)
         self.tokenizer = Qwen2Tokenizer.from_pretrained(model, subfolder="tokenizer", local_files_only=local_files_only)
-        self.processor = Qwen2VLProcessor.from_pretrained(
-            model, subfolder="processor", local_files_only=local_files_only
+        self.processor = retry_on_missing_shard(
+            lambda: Qwen2VLProcessor.from_pretrained(model, subfolder="processor", local_files_only=local_files_only)
         )
 
         self.stage = None

vllm_omni/diffusion/models/qwen_image/pipeline_qwen_image_edit_plus.py

@@ -25,7 +25,7 @@ from vllm.model_executor.models.utils import AutoWeightsLoader
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
-from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders
+from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders, retry_on_missing_shard
 from vllm_omni.diffusion.model_metadata import QWEN_IMAGE_EDIT_PLUS_MAX_INPUT_IMAGES
 from vllm_omni.diffusion.models.interface import SupportImageInput
 from vllm_omni.diffusion.models.qwen_image.cfg_parallel import (
@@ -208,7 +208,7 @@ class QwenImageEditPlusPipeline(
         model = od_config.model
 
         # Check if model is a local path
-        local_files_only = os.path.exists(model)
+        local_files_only = os.path.isdir(model)
 
         # Defend against a transformers v5 multi-worker race where a peer
         # worker's partially-written shard set makes our from_pretrained
@@ -219,15 +219,16 @@ class QwenImageEditPlusPipeline(
         prefetch_subfolders(
             model,
             ["scheduler", "text_encoder", "vae", "tokenizer", "processor"],
-            local_files_only=local_files_only,
         )
 
         self.scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(
             model, subfolder="scheduler", local_files_only=local_files_only
         )
-        self.text_encoder = Qwen2_5_VLForConditionalGeneration.from_pretrained(
-            model, subfolder="text_encoder", local_files_only=local_files_only
-        ).to(self.device)
+        self.text_encoder = retry_on_missing_shard(
+            lambda: Qwen2_5_VLForConditionalGeneration.from_pretrained(
+                model, subfolder="text_encoder", local_files_only=local_files_only
+            ).to(self.device)
+        )
 
         self.vae = AutoencoderKLQwenImage.from_pretrained(model, subfolder="vae", local_files_only=local_files_only).to(
             self.device
@@ -236,8 +237,8 @@ class QwenImageEditPlusPipeline(
         transformer_kwargs = get_transformer_config_kwargs(od_config.tf_model_config, QwenImageTransformer2DModel)
         self.transformer = QwenImageTransformer2DModel(od_config=od_config, **transformer_kwargs)
         self.tokenizer = Qwen2Tokenizer.from_pretrained(model, subfolder="tokenizer", local_files_only=local_files_only)
-        self.processor = Qwen2VLProcessor.from_pretrained(
-            model, subfolder="processor", local_files_only=local_files_only
+        self.processor = retry_on_missing_shard(
+            lambda: Qwen2VLProcessor.from_pretrained(model, subfolder="processor", local_files_only=local_files_only)
         )
 
         self.stage = None

vllm_omni/diffusion/models/qwen_image/pipeline_qwen_image_layered.py

@@ -24,7 +24,7 @@ from vllm.model_executor.models.utils import AutoWeightsLoader
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
-from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders
+from vllm_omni.diffusion.model_loader.hub_prefetch import prefetch_subfolders, retry_on_missing_shard
 from vllm_omni.diffusion.models.interface import SupportImageInput
 from vllm_omni.diffusion.models.qwen_image.autoencoder_kl_qwenimage import (
     AutoencoderKLQwenImage,
@@ -216,29 +216,30 @@ class QwenImageLayeredPipeline(nn.Module, SupportImageInput, QwenImageCFGParalle
         self.device = get_local_device()
         model = od_config.model
         # Check if model is a local path
-        local_files_only = os.path.exists(model)
+        local_files_only = os.path.isdir(model)
 
         # See pipeline_qwen_image_edit_plus: guard against transformers v5
         # multi-worker race on partial subfolder shard sets (Buildkite #1043).
         prefetch_subfolders(
             model,
             ["scheduler", "text_encoder", "vae", "tokenizer", "processor"],
-            local_files_only=local_files_only,
         )
 
         # modules keep same as transformers & diffusers
         self.scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(
             model, subfolder="scheduler", local_files_only=local_files_only
         )
-        self.text_encoder = Qwen2_5_VLForConditionalGeneration.from_pretrained(
-            model, subfolder="text_encoder", local_files_only=local_files_only
-        ).to(self.device)
+        self.text_encoder = retry_on_missing_shard(
+            lambda: Qwen2_5_VLForConditionalGeneration.from_pretrained(
+                model, subfolder="text_encoder", local_files_only=local_files_only
+            ).to(self.device)
+        )
         self.vae = AutoencoderKLQwenImage.from_pretrained(model, subfolder="vae", local_files_only=local_files_only).to(
             self.device
         )
         self.tokenizer = Qwen2Tokenizer.from_pretrained(model, subfolder="tokenizer", local_files_only=local_files_only)
-        self.processor = Qwen2VLProcessor.from_pretrained(
-            model, subfolder="processor", local_files_only=local_files_only
+        self.processor = retry_on_missing_shard(
+            lambda: Qwen2VLProcessor.from_pretrained(model, subfolder="processor", local_files_only=local_files_only)
         )
 
         # modules re-implemented for vLLM-Omni

vllm_omni/distributed/omni_coordinator/omni_coordinator.py

@@ -43,17 +43,20 @@ class OmniCoordinator:
             heartbeat_timeout: Seconds before a replica is considered
                 unhealthy if no heartbeat / update is received.
         """
-        self._router_zmq_addr = router_zmq_addr
-        self._pub_zmq_addr = pub_zmq_addr
         self._heartbeat_timeout = heartbeat_timeout
 
         # Dedicated ZMQ context for this coordinator instance.
         self._ctx = zmq.Context()
         self._router = self._ctx.socket(zmq.ROUTER)
-        self._router.bind(self._router_zmq_addr)
+        self._router.bind(router_zmq_addr)
+        # Recover the actual bound address (port=0 → real port).
+        self.router_zmq_addr = self._router.getsockopt_string(zmq.LAST_ENDPOINT)
+        self._pub_zmq_addr = pub_zmq_addr  # keep original for internal use
 
         self._pub = self._ctx.socket(zmq.PUB)
-        self._pub.bind(self._pub_zmq_addr)
+        self._pub.bind(pub_zmq_addr)
+        # Recover the actual bound address (port=0 → real port).
+        self.pub_zmq_addr = self._pub.getsockopt_string(zmq.LAST_ENDPOINT)
 
         self._replicas: dict[str, ReplicaInfo] = {}
         self._lock = threading.Lock()

vllm_omni/engine/arg_utils.py

@@ -144,7 +144,8 @@ class OmniEngineArgs(EngineArgs):
     force_cutlass_fp8: bool | None = None
     worker_type: str | None = None
     task_type: str | None = None
-    worker_cls: str = None
+    worker_cls: str = None  # type: ignore[assignment]  # Upstream default is "auto"; omni resolves
+    # in __post_init__ based on worker_type (ar/generation), so None is safe here.
     enable_sleep_mode: bool = False
     omni: bool = False
 
@@ -444,6 +445,48 @@ class OrchestratorArgs:
     # === Pre-built Objects ===
     parallel_config: Any = None
 
+    # === Diffusion model config ===
+    num_gpus: int | None = None
+    model_class_name: str | None = None
+    diffusion_load_format: str | None = None
+    diffusers_load_kwargs: str = "{}"
+    diffusers_call_kwargs: str = "{}"
+    ulysses_degree: int | None = None
+    ulysses_mode: str = "strict"
+    ring_degree: int | None = None
+    diffusion_quantization_config: str | None = None
+    use_hsdp: bool = False
+    hsdp_shard_size: int = -1
+    hsdp_replicate_size: int = 1
+    diffusion_attention_backend: str | None = None
+    diffusion_attention_config: str | None = None
+    cache_backend: str = "none"
+    cache_config: str | None = None
+    enable_cache_dit_summary: bool = False
+    step_execution: bool = False
+    vae_use_slicing: bool = False
+    vae_use_tiling: bool = False
+    enable_multithread_weight_load: bool = True
+    num_weight_load_threads: int = 4
+    enable_cpu_offload: bool = False
+    enable_layerwise_offload: bool = False
+    boundary_ratio: float | None = None
+    flow_shift: float | None = None
+    diffusion_kv_cache_dtype: str | None = None
+    diffusion_kv_cache_skip_steps: str | None = None
+    diffusion_kv_cache_skip_layers: str | None = None
+    cfg_parallel_size: int = 1
+    vae_patch_parallel_size: int = 1
+    default_sampling_params: str | None = None
+    max_generated_image_size: int | None = None
+    tts_max_instructions_length: int | None = None
+    enable_diffusion_pipeline_profiler: bool = False
+    enable_ar_profiler: bool = False
+    auxiliary_text_encoder: str | None = None
+    log_file: str | None = None
+    replica_id: int | None = None
+    omni_replica_address: str | None = None
+
     # === Multi-stage guards ===
     # --tokenizer is captured by the orchestrator and forwarded to stages
     # only when the stage does not define tokenizer/tokenizer_subdir itself.
@@ -459,6 +502,8 @@ SHARED_FIELDS: frozenset[str] = frozenset(
         "stage_id",  # orch: route (headless); engine: identity
         "log_stats",  # both want the flag
         "stage_configs_path",  # orch: load legacy YAML; engine: may reference for validation
+        "async_chunk",  # orch: read from CLI, redistribute; engine: per-stage flag
+        "tokenizer",  # orch: detect model type; engine: tokenization
     }
 )
 

vllm_omni/engine/async_omni_engine.py

@@ -2124,7 +2124,7 @@ class AsyncOmniEngine:
                             kwargs.get("diffusion_attention_config"),
                             attention_backend=kwargs.get("diffusion_attention_backend"),
                         )
-                quantization_config = kwargs.get("quantization_config")
+                quantization_config = kwargs.get("diffusion_quantization_config")
                 if quantization_config is not None:
                     if (
                         not hasattr(cfg.engine_args, "quantization_config")

vllm_omni/engine/output_processor.py

@@ -1,11 +1,7 @@
 from typing import Any
 
-import numpy as np
 import torch
 from vllm.logger import init_logger
-from vllm.model_executor.layers.fused_moe.routed_experts_capturer import (
-    split_routed_experts,
-)
 from vllm.outputs import PoolingRequestOutput
 from vllm.sampling_params import RequestOutputKind
 from vllm.tokenizers import TokenizerLike
@@ -172,7 +168,6 @@ class OmniRequestState(RequestState):
         finish_reason: FinishReason | None,
         stop_reason: int | str | None,
         kv_transfer_params: dict[str, Any] | None = None,
-        routed_experts: np.ndarray | None = None,
     ) -> OmniRequestOutput | PoolingRequestOutput | None:
         """Create a request output from generation results.
 
@@ -199,7 +194,6 @@ class OmniRequestState(RequestState):
                 finish_reason,
                 stop_reason,
                 kv_transfer_params,
-                routed_experts,
             )
 
         finished = finish_reason is not None
@@ -234,16 +228,7 @@ class OmniRequestState(RequestState):
 
         external_req_id = self.external_req_id
 
-        # Split routing data into prompt and generation portions, matching
-        # upstream make_request_output behaviour.
-        prompt_routed_experts = None
-        gen_routed_experts = None
-        if routed_experts is not None:
-            prompt_len = len(self.prompt_token_ids) if self.prompt_token_ids else 0
-            num_gen = self.detokenizer.num_output_tokens() if self.detokenizer is not None else None
-            prompt_routed_experts, gen_routed_experts = split_routed_experts(routed_experts, prompt_len, num_gen)
-
-        output = self._new_completion_output(new_token_ids, finish_reason, stop_reason, gen_routed_experts)
+        output = self._new_completion_output(new_token_ids, finish_reason, stop_reason)
 
         if self.parent_req is None:
             outputs = [output]
@@ -258,7 +243,6 @@ class OmniRequestState(RequestState):
             outputs,
             finished,
             kv_transfer_params,
-            prompt_routed_experts,
         )
 
     def _new_completion_output(
@@ -266,10 +250,11 @@ class OmniRequestState(RequestState):
         token_ids: list[int],
         finish_reason: FinishReason | None,
         stop_reason: int | str | None,
-        routed_experts: np.ndarray | None = None,
     ) -> Any:
         # Reuse base text/logprobs logic, then annotate with pooling_result.
-        base_output = super()._new_completion_output(token_ids, finish_reason, stop_reason, routed_experts)
+        # Note: upstream _new_completion_output no longer accepts routed_experts
+        # as a parameter; it reads from ``self.routed_experts_chunks`` internally.
+        base_output = super()._new_completion_output(token_ids, finish_reason, stop_reason)
 
         # Inter-stage processors need the full cumulative token sequence.
         # In DELTA mode, base_output.token_ids only has the latest step's

vllm_omni/engine/stage_engine_core_proc.py

@@ -101,6 +101,11 @@ class StageEngineCoreProc(EngineCoreProc):
             set_death_signal(signal.SIGTERM)
             set_process_title(f"StageEngineCoreProc_{stage_label}_replica{omni_replica_id}_DP{dp_rank}")
             decorate_logs()
+            # Workaround for flashinfer/jit-cache version mismatch in CI.
+            # The parent process handles this gracefully via ring_globals.py,
+            # but the subprocess hits an unprotected import in TopKTopPSampler.
+            # Setting this env var allows the same graceful fallback to work.
+            os.environ.setdefault("FLASHINFER_DISABLE_VERSION_CHECK", "1")
             os.environ["VLLM_OMNI_REPLICA_ID"] = str(max(int(omni_replica_id), 0))
 
             engine_core = StageEngineCoreProc(

vllm_omni/engine/stage_init_utils.py

@@ -824,6 +824,10 @@ def build_vllm_config(
     if upgraded is not vllm_config.quant_config:
         vllm_config = replace(vllm_config, quant_config=upgraded)
 
+    custom_voice_dir = engine_args_dict.get("custom_voice_dir")
+    if custom_voice_dir:
+        setattr(vllm_config.model_config.hf_config, "custom_voice_dir", custom_voice_dir)
+
     return vllm_config, executor_class
 
 
@@ -856,6 +860,45 @@ def build_stage0_input_processor(stage_vllm_config: Any) -> InputProcessor:
     return input_processor
 
 
+def _cleanup_stale_lock_if_dead(lock_file: str) -> bool:
+    """If *lock_file* exists and its recorded PID is dead, unlink the file.
+
+    Returns ``True`` if the stale lock was cleaned up (caller should retry),
+    ``False`` otherwise (lock holder appears alive, or file could not be read).
+    """
+    try:
+        with open(lock_file) as fh:
+            content = fh.read().strip()
+        if not content:
+            return False
+        pid = int(content)
+    except (OSError, ValueError):
+        return False
+
+    # Check whether the PID is still alive.
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        # PID does not exist — stale lock.
+        logger.info(
+            "Removing stale device lock %s (PID %s is dead)",
+            lock_file,
+            pid,
+        )
+        try:
+            os.unlink(lock_file)
+            return True
+        except OSError:
+            logger.debug("Failed to unlink stale lock %s", lock_file)
+            return False
+    except PermissionError:
+        # PID exists but we cannot signal it (different user) — treat as alive.
+        return False
+
+    # PID is alive — legitimate lock holder.
+    return False
+
+
 def acquire_device_locks(
     stage_id: int,
     engine_args_dict: dict[str, Any],
@@ -934,6 +977,7 @@ def acquire_device_locks(
         for device_id in devices_to_lock:
             lock_file = f"/tmp/vllm_omni_device_{device_id}_init.lock"
             lock_acquired = False
+            already_cleaned_stale = False  # only try stale cleanup once per device
 
             while not lock_acquired:
                 try:
@@ -948,6 +992,11 @@ def acquire_device_locks(
                         logger.debug("Acquired exclusive lock for device %s", device_id)
                     except BlockingIOError:
                         os.close(lock_fd)
+                        # Detect and clean stale locks from dead processes.
+                        if not already_cleaned_stale:
+                            already_cleaned_stale = True
+                            if _cleanup_stale_lock_if_dead(lock_file):
+                                continue  # retry flock immediately
                         if time.time() - wait_start > stage_init_timeout:
                             logger.warning(
                                 "Timeout waiting for device %s initialization lock, proceeding anyway",

vllm_omni/engine/stage_pool.py

@@ -22,6 +22,7 @@ from vllm_omni.engine.stage_client import (
     StagePoolDiffusionClient,
     StagePoolLLMClient,
 )
+from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 from vllm_omni.metrics.stats import StageRequestStats as StageRequestMetrics
 from vllm_omni.metrics.stats import StageStats
 from vllm_omni.metrics.utils import count_tokens_from_outputs
@@ -533,6 +534,10 @@ class StagePool:
     ) -> int:
         """Submit a stage-entry request into this pool."""
         params = params_override if params_override is not None else req_state.sampling_params_list[self.stage_id]
+        # Convert plain vllm SamplingParams for single-stage diffusion models
+        # that receive sampling params from the user/caller directly.
+        if self.stage_type == "diffusion" and not isinstance(params, OmniDiffusionSamplingParams):
+            params = OmniDiffusionSamplingParams()
         submit_kwargs = dict(submit_kwargs or {})
         if self.stage_type == "diffusion":
             replica_id = await self._pick_or_select(
@@ -593,6 +598,8 @@ class StagePool:
     ) -> int:
         """Submit a streaming update to an already admitted request."""
         params = req_state.sampling_params_list[self.stage_id]
+        if self.stage_type == "diffusion" and not isinstance(params, OmniDiffusionSamplingParams):
+            params = OmniDiffusionSamplingParams()
         replica_id = self.get_bound_replica_id(request_id)
         if replica_id is None or self.clients[replica_id] is None:
             replica_id = await self._pick_or_select(request_id)

vllm_omni/entrypoints/async_omni.py

@@ -295,6 +295,15 @@ class AsyncOmni(EngineClient, OmniBase):
             # Start final output dispatcher on the first call to generate()
             self._final_output_handler()
 
+            # Forward bare sampling_params (e.g. from /v1/completions) as the stage-0 entry.
+            if sampling_params_list is None and sampling_params is not None:
+                if self.num_stages == 1:
+                    sampling_params_list = [sampling_params]
+                else:
+                    default = list(self.default_sampling_params_list)
+                    default[0] = sampling_params
+                    sampling_params_list = default
+
             # Expand sampling params for PD disaggregation (user may provide N-1 params)
             if (
                 sampling_params_list is not None
@@ -919,8 +928,14 @@ class AsyncOmni(EngineClient, OmniBase):
         for result in results:
             if isinstance(result, dict) and result.get("todo"):
                 continue
-            if isinstance(result, list):
-                merged.update(result)
+            if isinstance(result, (list, set)):
+                for item in result:
+                    if isinstance(item, (list, set)):
+                        merged.update(item)
+                    elif isinstance(item, int):
+                        merged.add(item)
+            elif isinstance(result, int):
+                merged.add(result)
         return sorted(merged)
 
     async def pin_lora(self, adapter_id: int) -> bool:

vllm_omni/entrypoints/cli/serve.py

@@ -436,7 +436,7 @@ class OmniServeCommand(CLISubcommand):
             "Equivalent to setting DiffusionParallelConfig.ring_degree.",
         )
         omni_config_group.add_argument(
-            "--quantization-config",
+            "--diffusion-quantization-config",
             type=json.loads,
             default=None,
             help=(

vllm_omni/entrypoints/omni_base.py

@@ -463,7 +463,7 @@ class OmniBase(PDDisaggregationMixin):
             peak_memory_mb=peak_memory_mb,
         )
 
-    def shutdown(self) -> None:
+    def shutdown(self, timeout: float | None = None) -> None:
         logger.info("[%s] Shutting down", self.__class__.__name__)
         self._shutdown_base()
 

vllm_omni/entrypoints/openai/api_server.py

@@ -55,13 +55,8 @@ from vllm.entrypoints.openai.engine.protocol import (
 from vllm.entrypoints.openai.models.protocol import BaseModelPath
 from vllm.entrypoints.openai.models.serving import OpenAIServingModels
 from vllm.entrypoints.openai.orca_metrics import metrics_header
-from vllm.entrypoints.openai.realtime.serving import OpenAIServingRealtime
 from vllm.entrypoints.openai.responses.serving import OpenAIServingResponses
 from vllm.entrypoints.openai.server_utils import get_uvicorn_log_config
-from vllm.entrypoints.openai.speech_to_text.serving import (
-    OpenAIServingTranscription,
-    OpenAIServingTranslation,
-)
 from vllm.entrypoints.openai.utils import validate_json_request
 from vllm.entrypoints.pooling.classify.serving import ServingClassification
 from vllm.entrypoints.pooling.embed.serving import ServingEmbedding as OpenAIServingEmbedding
@@ -74,6 +69,13 @@ from vllm.entrypoints.serve.disagg.serving import ServingTokens
 from vllm.entrypoints.serve.instrumentator.basic import base
 from vllm.entrypoints.serve.render.serving import OpenAIServingRender
 from vllm.entrypoints.serve.tokenize.serving import OpenAIServingTokenization
+from vllm.entrypoints.speech_to_text.realtime.serving import OpenAIServingRealtime
+from vllm.entrypoints.speech_to_text.transcription.serving import (
+    OpenAIServingTranscription,
+)
+from vllm.entrypoints.speech_to_text.translation.serving import (
+    OpenAIServingTranslation,
+)
 from vllm.entrypoints.utils import (
     create_error_response,
     load_aware_call,
@@ -355,7 +357,7 @@ async def omni_run_server(args, **uvicorn_kwargs) -> None:
     warnings_module.filterwarnings("ignore", message=".*PydanticSerializationUnexpectedValue.*", category=UserWarning)
 
     # Add process-specific prefix to stdout and stderr.
-    decorate_logs("APIServer")
+    decorate_logs("APIServer", skip_if_decorated=True)
 
     listen_address, sock = setup_openai_server(args)
 
@@ -646,6 +648,19 @@ async def omni_init_app_state(
 
     state.vllm_config = vllm_config
 
+    # Propagate enable_in_reasoning to the API-server process. The engine core
+    # runs in a separate process, so the contextvar that backs
+    # `get_current_vllm_config_or_none()` is None on this stack. Tool parsers
+    # call `get_enable_structured_outputs_in_reasoning()` during request
+    # handling and need to see the real flag, otherwise they silently fall
+    # back to False and mismatch the engine-side bitmask gating.
+    if vllm_config is not None:
+        from vllm.tool_parsers.structural_tag_registry import (
+            set_enable_structured_outputs_in_reasoning,
+        )
+
+        set_enable_structured_outputs_in_reasoning(vllm_config.structured_outputs_config.enable_in_reasoning)
+
     # Get supported tasks
     supported_tasks: set[str] = {"generate"}
     if hasattr(engine_client, "get_supported_tasks"):
@@ -1531,6 +1546,7 @@ async def generate_images(request: ImageGenerationRequest, raw_request: Request)
                 prompt=request.prompt,
                 extra_body=extra_body,
                 request_id=f"img_gen-{random_uuid()}",
+                raw_request=raw_request,
             )
             if isinstance(generation_result, ErrorResponse):
                 return JSONResponse(
@@ -1937,6 +1953,7 @@ async def edit_images(
                 output_format=output_format,
                 output_compression=output_compression,
                 size=size_str,
+                raw_request=raw_request,
             )
             if stream and not isinstance(generation_result, ErrorResponse):
                 return StreamingResponse(

vllm_omni/entrypoints/openai/realtime_connection.py

@@ -9,8 +9,8 @@ from uuid import uuid4
 
 import numpy as np
 from vllm.entrypoints.openai.engine.protocol import UsageInfo
-from vllm.entrypoints.openai.realtime.connection import RealtimeConnection as VllmRealtimeConnection
-from vllm.entrypoints.openai.realtime.protocol import TranscriptionDelta, TranscriptionDone
+from vllm.entrypoints.speech_to_text.realtime.connection import RealtimeConnection as VllmRealtimeConnection
+from vllm.entrypoints.speech_to_text.realtime.protocol import TranscriptionDelta, TranscriptionDone
 from vllm.logger import init_logger
 
 from vllm_omni.entrypoints.async_omni import AsyncOmni

vllm_omni/entrypoints/openai/serving_chat.py

@@ -2574,6 +2574,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
         output_format: str = "png",
         output_compression: int = 100,
         size: str = "auto",
+        raw_request: Request | None = None,
     ) -> tuple[list[Image.Image], dict[str, Any], float, str | None] | ErrorResponse | AsyncIterator[str]:
         """Generate diffusion images and return raw images plus generation stats."""
         if request_id is None:
@@ -2635,6 +2636,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
                     output_format=output_format,
                     output_compression=output_compression,
                     size=size,
+                    raw_request=raw_request,
                 )
 
             result = None
@@ -2659,6 +2661,16 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
         peak_memory_mb = result.peak_memory_mb
         cot_output = None
 
+        req_out = getattr(result, "request_output", None)
+        if req_out:
+            prompt_obj = getattr(req_out, "prompt", None)
+            if isinstance(prompt_obj, dict):
+                extra = prompt_obj.get("extra", {})
+                if isinstance(extra, dict):
+                    ar_text = extra.get("ar_generated_text")
+                    if isinstance(ar_text, str) and ar_text.strip():
+                        cot_output = ar_text
+
         req_out = getattr(result, "request_output", None)
         if req_out:
             prompt_obj = getattr(req_out, "prompt", None)
@@ -2679,6 +2691,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
         output_format: str,
         output_compression: int,
         size: str,
+        raw_request: Request | None = None,
     ) -> AsyncIterator[str]:
         """Yield image edit SSE chunks from multi-stage diffusion outputs."""
         created = int(time.time())
@@ -2730,7 +2743,16 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
             logger.error("EngineDeadError during streaming image edit: %s", exc)
             data = self.create_streaming_error_response(exc)
             yield f"data: {data}\n\n"
-            terminate_if_errored(engine=self.engine_client)
+            if raw_request is not None:
+                terminate_if_errored(
+                    server=raw_request.app.state.server,
+                    engine=self.engine_client,
+                )
+            else:
+                logger.warning(
+                    "[OmniOpenAIServingChat] Engine dead during streaming image edit, "
+                    "but cannot terminate server (no raw_request context)."
+                )
         except Exception as exc:
             logger.exception("Streaming image edit failed: %s", exc)
             chunk = ImageEditStreamError(

vllm_omni/entrypoints/openai/serving_speech.py

@@ -8,6 +8,7 @@ import os
 import re
 import struct
 import time
+from collections import OrderedDict
 from concurrent.futures import ThreadPoolExecutor
 from http import HTTPStatus
 from pathlib import Path
@@ -55,7 +56,11 @@ from vllm_omni.model_executor.models.ming_flash_omni.prompt_utils import (
     create_instruction as ming_create_instruction,
 )
 from vllm_omni.outputs import OmniRequestOutput
-from vllm_omni.utils.speaker_cache import get_speaker_cache
+from vllm_omni.utils.speaker_cache import (
+    get_speaker_cache,
+    iter_custom_voice_profiles,
+    load_validated_profile_tensors,
+)
 
 logger = init_logger(__name__)
 
@@ -107,6 +112,9 @@ _TTS_LANGUAGES: set[str] = {
 }
 _REF_AUDIO_MIN_DURATION = 1.0  # seconds
 _REF_AUDIO_MAX_DURATION = 30.0  # seconds
+_REF_AUDIO_RESOLVE_CACHE_MAX_ENTRIES = 256
+_REF_AUDIO_RESOLVE_CACHE_MAX_BYTES = 256 * 1024 * 1024
+_QWEN3_TTS_REF_AUDIO_CACHE_KEY = "_qwen3_tts_ref_audio_cache_key"
 _TTS_MAX_INSTRUCTIONS_LENGTH = 500
 _TTS_MAX_NEW_TOKENS_MIN = 1
 _TTS_MAX_NEW_TOKENS_MAX = 4096
@@ -261,8 +269,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             logger.warning("Invalid SPEAKER_MAX_UPLOADED=%r; using default 1000", _raw_cap)
             self._max_uploaded_speakers = 1000
         self.uploaded_speakers: dict[str, dict] = {}
+        self.precomputed_speakers: dict[str, dict[str, Any]] = {}
         self.supported_speakers: set[str] = set()
         self._ref_audio_data_url_cache: dict[str, str] = {}
+        self._ref_audio_resolve_cache: OrderedDict[str, tuple[list[float], int, int, str]] = OrderedDict()
+        self._ref_audio_resolve_cache_bytes = 0
+        self._ref_audio_resolve_cache_max_entries = _REF_AUDIO_RESOLVE_CACHE_MAX_ENTRIES
+        self._ref_audio_resolve_cache_max_bytes = _REF_AUDIO_RESOLVE_CACHE_MAX_BYTES
+        self._ref_audio_model_artifact_ready: set[str] = set()
+        self._request_ref_audio_artifact_keys: dict[str, str] = {}
         self._speaker_cache = get_speaker_cache()
         self._last_upload_ts = 0
         self._upload_lock = asyncio.Lock()
@@ -395,6 +410,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         )
         # Determine TTS model type or None
         self._tts_model_type = self._detect_tts_model_type()
+        self.precomputed_speakers = self._load_precomputed_speakers()
 
         # GLM-TTS lazy-cached resources (populated on first GLM-TTS request)
         self._glm_tts_text_tokenizer: object | None = None
@@ -405,6 +421,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
 
         # Merge built-in speakers into the set initialized by _init_speaker_storage.
         self.supported_speakers |= self._load_supported_speakers()
+        self.supported_speakers |= set(self.precomputed_speakers)
         self._tts_tokenizer = None
         self._voxcpm2_tokenizer = None
         self._voxcpm2_split_map: dict[int, list[int]] = {}
@@ -570,6 +587,45 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             return "glm_tts"
         return None
 
+    def _get_custom_voice_dir(self) -> str | None:
+        try:
+            value = getattr(self.engine_client.model_config.hf_config, "custom_voice_dir", None)
+        except AttributeError:
+            return None
+        if isinstance(value, os.PathLike):
+            return os.fspath(value)
+        if isinstance(value, str) and value:
+            return value
+        return None
+
+    def _load_precomputed_speakers(self) -> dict[str, dict[str, Any]]:
+        """Load precomputed voice names from ``custom_voice_dir`` for API validation."""
+        if self._tts_model_type not in ("qwen3_tts", "voxcpm2"):
+            return {}
+        custom_voice_dir = self._get_custom_voice_dir()
+        if not custom_voice_dir:
+            return {}
+
+        profiles: dict[str, dict[str, Any]] = {}
+        qwen3_embedding_dim = self._get_qwen_tts_expected_speaker_embedding_dim()
+        for profile in iter_custom_voice_profiles(custom_voice_dir, expected_model_type=self._tts_model_type):
+            tensors = load_validated_profile_tensors(
+                profile,
+                expected_model_type=self._tts_model_type,
+                qwen3_embedding_dim=qwen3_embedding_dim,
+            )
+            if tensors is None:
+                continue
+            profiles[profile["voice_name_lower"]] = profile
+        if profiles:
+            logger.info(
+                "Loaded %d precomputed %s voice profile(s) from %s",
+                len(profiles),
+                self._tts_model_type,
+                custom_voice_dir,
+            )
+        return profiles
+
     def _compute_max_instructions_length(self) -> int:
         """Compute max instructions length with precedence: CLI > stage config > default.
 
@@ -748,11 +804,14 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         self._voxcpm2_encode("")  # lazy-init tokenizer + split_map
         ref_audio = None
         ref_sr = None
+        voice_profile = None
         if request.ref_audio is not None:
             ref_audio, ref_sr = await self._resolve_ref_audio(request.ref_audio)
         elif uploaded_ref is not None:
             wav_np, ref_sr = uploaded_ref
             ref_audio = wav_np.tolist()
+        elif request.voice is not None:
+            voice_profile = self.precomputed_speakers.get(request.voice.lower())
         return build_voxcpm2_prompt(
             hf_config=self.engine_client.model_config.hf_config,
             tokenizer=self._voxcpm2_tokenizer,
@@ -761,6 +820,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             ref_audio=ref_audio,
             ref_sr=ref_sr,
             ref_text=request.ref_text,
+            voice_profile=voice_profile,
         )
 
     def _load_uploaded_audio(self, voice_name: str) -> tuple[np.ndarray, int] | None:
@@ -1205,12 +1265,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         if self._tts_model_type == "cosyvoice3":
             return self._validate_cosyvoice3_request(request)
         if self._tts_model_type == "voxcpm2":
-            if request.max_new_tokens is not None:
-                if request.max_new_tokens < _TTS_MAX_NEW_TOKENS_MIN:
-                    return f"max_new_tokens must be at least {_TTS_MAX_NEW_TOKENS_MIN}"
-                if request.max_new_tokens > _TTS_MAX_NEW_TOKENS_MAX:
-                    return f"max_new_tokens cannot exceed {_TTS_MAX_NEW_TOKENS_MAX}"
-            return None  # VoxCPM2 accepts any text input
+            return self._validate_voxcpm2_request(request)
         if self._tts_model_type == "ming_flash_omni_tts":
             return self._validate_ming_tts_request(request)
         if self._tts_model_type == "moss_tts_nano":
@@ -1219,6 +1274,26 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             return self._validate_glm_tts_request(request)
         return self._validate_qwen_tts_request(request)
 
+    def _validate_voxcpm2_request(self, request: OpenAICreateSpeechRequest) -> str | None:
+        """Validate VoxCPM2 request parameters. Returns error message or None."""
+        if not request.input or not request.input.strip():
+            return "Input text cannot be empty"
+
+        if request.voice is not None:
+            request.voice = request.voice.lower()
+            available_voices = set(self.uploaded_speakers) | set(self.precomputed_speakers) | {"default"}
+            if request.voice not in available_voices:
+                supported = ", ".join(sorted(available_voices)) or "none"
+                return f"Invalid voice '{request.voice}'. Supported: {supported}"
+
+        if request.max_new_tokens is not None:
+            if request.max_new_tokens < _TTS_MAX_NEW_TOKENS_MIN:
+                return f"max_new_tokens must be at least {_TTS_MAX_NEW_TOKENS_MIN}"
+            if request.max_new_tokens > _TTS_MAX_NEW_TOKENS_MAX:
+                return f"max_new_tokens cannot exceed {_TTS_MAX_NEW_TOKENS_MAX}"
+
+        return None
+
     def _voxcpm2_encode(self, text: str) -> list[int]:
         """Tokenize text for VoxCPM2, splitting multichar Chinese tokens."""
         from vllm_omni.model_executor.models.voxcpm2.voxcpm2_talker import (
@@ -1315,11 +1390,13 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         # Infer Base task when ref_audio or ref_text is provided without explicit task_type.
         if request.task_type is None and (request.ref_audio is not None or request.ref_text is not None):
             request.task_type = "Base"
-        task_type = request.task_type or "CustomVoice"
 
         # Normalize voice to lowercase for case-insensitive matching
         if request.voice is not None:
             request.voice = request.voice.lower()
+            if request.task_type is None and request.voice in self.precomputed_speakers:
+                request.task_type = "Base"
+        task_type = request.task_type or "CustomVoice"
 
         # Validate input is not empty
         if not request.input or not request.input.strip():
@@ -1379,6 +1456,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                     file_path = Path(speaker_info["file_path"])
                     if not file_path.exists():
                         return f"Data file for uploaded speaker '{request.voice}' not found on disk"
+                elif voice_lower in self.precomputed_speakers:
+                    profile = self.precomputed_speakers[voice_lower]
+                    mode = str(profile.get("mode") or "xvec").lower()
+                    ref_text = request.ref_text or profile.get("ref_text")
+                    if mode == "icl" and (not isinstance(ref_text, str) or not ref_text.strip()):
+                        return (
+                            f"Precomputed voice '{request.voice}' uses ICL mode but has no ref_text in "
+                            "the request or custom voice manifest"
+                        )
                 else:
                     # need ref_audio for built-in speaker
                     if request.ref_audio is None:
@@ -1677,6 +1763,19 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         URLs, ``data:`` base64 URIs, and ``file:`` local paths (the latter
         gated by ``--allowed-local-media-path``).
         """
+        cache_key = hashlib.sha1(ref_audio_str.encode("utf-8")).hexdigest()
+        cached = self._ref_audio_resolve_cache.get(cache_key)
+        if cached is not None:
+            self._ref_audio_resolve_cache.move_to_end(cache_key)
+            wav_list, sr, _, _ = cached
+            logger.debug(
+                "Resolved ref_audio from cache: samples=%d sr=%d duration_s=%.3f",
+                len(wav_list),
+                sr,
+                len(wav_list) / sr if sr > 0 else 0.0,
+            )
+            return wav_list, sr
+
         # In diffusion mode, model_config may not be available
         if self._diffusion_mode:
             connector = MediaConnector()
@@ -1686,11 +1785,14 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                 allowed_local_media_path=model_config.allowed_local_media_path,
                 allowed_media_domains=model_config.allowed_media_domains,
             )
+        fetch_start_s = time.perf_counter()
         wav_np, sr = await connector.fetch_audio_async(ref_audio_str)
+        fetch_decode_ms = (time.perf_counter() - fetch_start_s) * 1000.0
         wav_np = np.asarray(wav_np, dtype=np.float32)
         if wav_np.ndim > 1:
             wav_np = np.mean(wav_np, axis=-1)
         sr = int(sr)
+        artifact_key = self._make_ref_audio_artifact_cache_key(wav_np, sr)
         duration = len(wav_np) / sr if sr > 0 else 0.0
         if duration < _REF_AUDIO_MIN_DURATION:
             raise ValueError(
@@ -1702,7 +1804,83 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                 f"Reference audio too long ({duration:.1f}s). "
                 f"Maximum {_REF_AUDIO_MAX_DURATION:.0f}s supported — use a shorter clip."
             )
-        return wav_np.tolist(), sr
+        tolist_start_s = time.perf_counter()
+        wav_list = wav_np.tolist()
+        tolist_ms = (time.perf_counter() - tolist_start_s) * 1000.0
+        logger.debug(
+            "Resolved ref_audio: fetch_decode_ms=%.3f tolist_ms=%.3f samples=%d sr=%d duration_s=%.3f",
+            fetch_decode_ms,
+            tolist_ms,
+            len(wav_np),
+            sr,
+            duration,
+        )
+        self._put_resolved_ref_audio(cache_key, wav_list, sr, artifact_key)
+        return wav_list, sr
+
+    @staticmethod
+    def _make_ref_audio_artifact_cache_key(wav: np.ndarray, sr: int) -> str:
+        wav_f32 = wav.astype(np.float32, copy=False).reshape(-1)
+        h = hashlib.sha1()
+        h.update(int(sr).to_bytes(4, byteorder="little", signed=False))
+        h.update(int(wav_f32.size).to_bytes(8, byteorder="little", signed=False))
+        h.update(wav_f32.tobytes(order="C"))
+        return h.hexdigest()
+
+    def _get_resolved_ref_audio_artifact_key(self, ref_audio_str: str) -> str | None:
+        source_key = hashlib.sha1(ref_audio_str.encode("utf-8")).hexdigest()
+        cached = self._ref_audio_resolve_cache.get(source_key)
+        if cached is None:
+            return None
+        self._ref_audio_resolve_cache.move_to_end(source_key)
+        return cached[3]
+
+    def _put_resolved_ref_audio(self, cache_key: str, wav_list: list[float], sr: int, artifact_key: str) -> None:
+        if self._ref_audio_resolve_cache_max_entries <= 0 or self._ref_audio_resolve_cache_max_bytes <= 0:
+            return
+        # Approximate list[float] storage. CPython float objects add per-element
+        # overhead, so max_entries remains the hard cache cap.
+        size = len(wav_list) * 40
+        if size > self._ref_audio_resolve_cache_max_bytes:
+            return
+        previous = self._ref_audio_resolve_cache.pop(cache_key, None)
+        if previous is not None:
+            self._ref_audio_resolve_cache_bytes -= previous[2]
+            if previous[3] != artifact_key:
+                self._discard_ref_audio_artifact_ready_if_unreferenced(previous[3])
+        self._ref_audio_resolve_cache[cache_key] = (wav_list, int(sr), size, artifact_key)
+        self._ref_audio_resolve_cache_bytes += size
+        while len(self._ref_audio_resolve_cache) > self._ref_audio_resolve_cache_max_entries:
+            _, (_, _, old_size, old_artifact_key) = self._ref_audio_resolve_cache.popitem(last=False)
+            self._ref_audio_resolve_cache_bytes -= old_size
+            self._discard_ref_audio_artifact_ready_if_unreferenced(old_artifact_key)
+        while self._ref_audio_resolve_cache_bytes > self._ref_audio_resolve_cache_max_bytes:
+            _, (_, _, old_size, old_artifact_key) = self._ref_audio_resolve_cache.popitem(last=False)
+            self._ref_audio_resolve_cache_bytes -= old_size
+            self._discard_ref_audio_artifact_ready_if_unreferenced(old_artifact_key)
+
+    def _discard_ref_audio_artifact_ready_if_unreferenced(self, artifact_key: str) -> None:
+        if artifact_key and all(entry[3] != artifact_key for entry in self._ref_audio_resolve_cache.values()):
+            self._ref_audio_model_artifact_ready.discard(artifact_key)
+
+    def _qwen3_tts_can_use_ref_audio_artifact_only(self, tts_params: dict[str, Any], artifact_key: str | None) -> bool:
+        if self._tts_model_type != "qwen3_tts":
+            return False
+        if not artifact_key or artifact_key not in self._ref_audio_model_artifact_ready:
+            return False
+        return (tts_params.get("task_type") or ["CustomVoice"])[0] == "Base"
+
+    def _track_ref_audio_artifact_warmup(self, request_id: str, artifact_key: str | None) -> None:
+        if artifact_key:
+            self._request_ref_audio_artifact_keys[request_id] = artifact_key
+
+    def _mark_ref_audio_artifact_ready_for_request(self, request_id: str) -> None:
+        artifact_key = self._request_ref_audio_artifact_keys.pop(request_id, None)
+        if artifact_key and any(entry[3] == artifact_key for entry in self._ref_audio_resolve_cache.values()):
+            self._ref_audio_model_artifact_ready.add(artifact_key)
+
+    def _discard_ref_audio_artifact_warmup(self, request_id: str) -> None:
+        self._request_ref_audio_artifact_keys.pop(request_id, None)
 
     async def _generate_audio_chunks(
         self,
@@ -1733,6 +1911,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         first_chunk = True
         first_audio_chunk_s: float | None = None
         stream_start_s = request_start_s if request_start_s is not None else time.perf_counter()
+        artifact_ready = False
 
         try:
             async for res in generator:
@@ -1787,6 +1966,8 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                     if first_audio_chunk_s is None:
                         first_audio_chunk_s = time.perf_counter()
                     yield self.create_audio(audio_obj).audio_data
+            self._mark_ref_audio_artifact_ready_for_request(request_id)
+            artifact_ready = True
             total_ms = (time.perf_counter() - stream_start_s) * 1000.0
             if first_audio_chunk_s is not None:
                 first_chunk_ms = (first_audio_chunk_s - stream_start_s) * 1000.0
@@ -1840,6 +2021,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             )
             logger.exception("Streaming speech generation failed for %s: %s", request_id, e)
             raise
+        finally:
+            if not artifact_ready:
+                self._discard_ref_audio_artifact_warmup(request_id)
 
     @staticmethod
     def _extract_audio_output(res) -> tuple[dict | None, str | None]:
@@ -1893,13 +2077,14 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
 
         # Speaker (voice)
         if request.voice is not None:
+            voice_lower = request.voice.lower()
             params["speaker"] = [request.voice]
-            params["voice_created_at"] = [self._voice_created_at(request.voice.lower())]
+            params["voice_created_at"] = [self._voice_created_at(voice_lower)]
 
             # Uploaded voices use task_type="Base" (CustomVoice requires built-in spk_id).
             # If ref_text was provided at upload time, use in-context cloning; otherwise x_vector only.
-            if request.voice.lower() in self.uploaded_speakers and request.ref_audio is None:
-                speaker_info = self.uploaded_speakers[request.voice.lower()]
+            if voice_lower in self.uploaded_speakers and request.ref_audio is None:
+                speaker_info = self.uploaded_speakers[voice_lower]
 
                 # Check if this voice was uploaded with a pre-computed embedding.
                 # Populate request.speaker_embedding so the existing code path
@@ -1907,6 +2092,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                 embedding = self._get_uploaded_speaker_embedding(request.voice)
                 if embedding is not None:
                     request.speaker_embedding = embedding
+                    params["speaker"] = [voice_lower]
                     params["task_type"] = ["Base"]
                     logger.info("Auto-set speaker_embedding for uploaded voice: %s", request.voice)
                 else:
@@ -1914,6 +2100,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                     if not audio_data:
                         raise ValueError(f"Audio file for uploaded voice '{request.voice}' is missing or corrupted")
                     stored_ref_text = speaker_info.get("ref_text")
+                    params["speaker"] = [voice_lower]
                     params["ref_audio"] = [audio_data]
                     params["task_type"] = ["Base"]
                     if stored_ref_text:
@@ -1924,6 +2111,19 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                     logger.info(
                         "Auto-set ref_audio for uploaded voice: %s (icl=%s)", request.voice, bool(stored_ref_text)
                     )
+            elif voice_lower in self.precomputed_speakers and request.ref_audio is None:
+                profile = self.precomputed_speakers[voice_lower]
+                mode = str(profile.get("mode") or "xvec").lower()
+                params["speaker"] = [voice_lower]
+                params["task_type"] = ["Base"]
+                params["x_vector_only_mode"] = [mode != "icl"]
+                ref_text = request.ref_text or profile.get("ref_text")
+                if isinstance(ref_text, str) and ref_text.strip():
+                    params["ref_text"] = [ref_text]
+                ref_code_length = profile.get("ref_code_length")
+                if mode == "icl" and ref_code_length:
+                    params["ref_code_length"] = [int(ref_code_length)]
+                logger.info("Using precomputed Qwen3-TTS custom voice profile: %s (mode=%s)", voice_lower, mode)
 
         elif params["task_type"][0] == "CustomVoice":
             params["speaker"] = ["Vivian"]  # Default for CustomVoice
@@ -2334,6 +2534,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         if self.engine_client.errored:
             raise self.engine_client.dead_error
 
+        request_id = request_id or f"speech-{random_uuid()}"
+        qwen3_ref_audio_warmup_artifact_key: str | None = None
+
         # If this is a streaming request, we need to coerce
         # cumulative outputs to delta outputs; this ensures
         # we don't emit redundant MM data & drain after emitting.
@@ -2412,12 +2615,13 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         elif self._tts_model_type == "voxcpm2":
             # voxcpm2 doesn't use `_apply_uploaded_speaker` because the prompt builder needs the
             # raw waveform tuple for prefill-length accounting, not a base64 data URL.
+            validation_error = self._validate_voxcpm2_request(request)
+            if validation_error:
+                raise ValueError(validation_error)
+
             uploaded_ref: tuple[np.ndarray, int] | None = None
             if request.voice:
                 voice_lower = request.voice.lower()
-                if voice_lower not in self.uploaded_speakers and voice_lower not in self.supported_speakers:
-                    all_voices = sorted(self.uploaded_speakers.keys() | self.supported_speakers)
-                    raise ValueError(f"Invalid voice '{request.voice}'. Supported: {', '.join(all_voices) or 'none'}")
                 if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
                     if self.uploaded_speakers[voice_lower].get("embedding_source") == "direct":
                         raise ValueError(
@@ -2430,7 +2634,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             tts_params = {}
             if request.voice:
                 voice_lower = request.voice.lower()
-                if voice_lower in self.uploaded_speakers and not has_inline_ref_audio:
+                if voice_lower in self.uploaded_speakers or voice_lower in self.precomputed_speakers:
                     additional = prompt.setdefault("additional_information", {})
                     additional["voice_name"] = voice_lower
                     additional["voice_created_at"] = self._voice_created_at(voice_lower)
@@ -2478,7 +2682,18 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                     ref_audio_source = tts_params["ref_audio"][0]
                 if ref_audio_source is not None and isinstance(ref_audio_source, str):
                     wav_list, sr = await self._resolve_ref_audio(ref_audio_source)
-                    tts_params["ref_audio"] = [[wav_list, sr]]
+                    artifact_key = self._get_resolved_ref_audio_artifact_key(ref_audio_source)
+                    if self._tts_model_type == "qwen3_tts" and artifact_key:
+                        tts_params[_QWEN3_TTS_REF_AUDIO_CACHE_KEY] = [artifact_key]
+                    ref_code_length = self._estimate_ref_code_len([wav_list, sr])
+                    if self._tts_model_type == "qwen3_tts" and ref_code_length is not None:
+                        tts_params["ref_code_length"] = [int(ref_code_length)]
+                    if self._qwen3_tts_can_use_ref_audio_artifact_only(tts_params, artifact_key):
+                        logger.debug("Using Qwen3-TTS ref_audio artifact-only path: %s", artifact_key)
+                    else:
+                        tts_params["ref_audio"] = [[wav_list, sr]]
+                        if self._tts_model_type == "qwen3_tts":
+                            qwen3_ref_audio_warmup_artifact_key = artifact_key
 
                 ph_len = await self._estimate_prompt_len_async(tts_params)
                 prompt = tokens_input(prompt_token_ids=[1] * ph_len)
@@ -2506,7 +2721,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             tts_params = {}
             prompt = {"prompt": request.input}
 
-        request_id = request_id or f"speech-{random_uuid()}"
         if self._is_fish_speech:
             model_type = "fish_speech"
         elif self._tts_model_type == "covo_audio":
@@ -2643,6 +2857,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
             sampling_params_list=sampling_params_list,
             output_modalities=["audio"],
         )
+        self._track_ref_audio_artifact_warmup(request_id, qwen3_ref_audio_warmup_artifact_key)
         return request_id, generator, tts_params
 
     async def _generate_pcm_chunks(self, generator, request_id: str):
@@ -2657,8 +2872,11 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
     async def _iter_pcm_audio_bytes(self, request: OpenAICreateSpeechRequest):
         """Yield raw PCM bytes for a speech request as soon as chunks are decoded."""
         request_id, generator, _ = await self._prepare_speech_generation(request)
-        async for chunk in self._generate_pcm_chunks(generator, request_id):
-            yield chunk
+        try:
+            async for chunk in self._generate_pcm_chunks(generator, request_id):
+                yield chunk
+        finally:
+            self._discard_ref_audio_artifact_warmup(request_id)
 
     async def _generate_audio_bytes(
         self,
@@ -2667,101 +2885,108 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         request_id: str | None = None,
     ) -> tuple[bytes | str, str]:
         request_id, generator, _ = await self._prepare_speech_generation(request, request_id=request_id)
+        artifact_ready = False
 
-        # MOSS-TTS-Nano emits delta chunks per yield (single-stage,
-        # async_chunk=false). The engine surfaces each yield as its own
-        # RequestOutput, so we need to accumulate across the async-for loop —
-        # final_output alone only carries the last (often empty) sentinel.
-        is_moss = self._tts_model_type == "moss_tts_nano"
-        moss_chunks: list[Any] = []
-        moss_sample_rate: int | None = None
+        try:
+            # MOSS-TTS-Nano emits delta chunks per yield (single-stage,
+            # async_chunk=false). The engine surfaces each yield as its own
+            # RequestOutput, so we need to accumulate across the async-for loop —
+            # final_output alone only carries the last (often empty) sentinel.
+            is_moss = self._tts_model_type == "moss_tts_nano"
+            moss_chunks: list[Any] = []
+            moss_sample_rate: int | None = None
 
-        final_output: OmniRequestOutput | None = None
-        async for res in generator:
-            final_output = res
-            if not is_moss:
-                continue
-            try:
-                step_audio, step_key = self._extract_audio_output(res)
-            except Exception:
-                continue
-            if step_key is None:
-                continue
-            chunk = step_audio[step_key]
-            candidates = chunk if isinstance(chunk, list) else [chunk]
-            for cand in candidates:
-                if hasattr(cand, "numel") and cand.numel() > 0:
-                    moss_chunks.append(cand)
-            sr_step = step_audio.get("sr")
-            if sr_step is not None:
-                sr_val_step = sr_step[-1] if isinstance(sr_step, list) and sr_step else sr_step
-                moss_sample_rate = int(sr_val_step.item()) if hasattr(sr_val_step, "item") else int(sr_val_step)
+            final_output: OmniRequestOutput | None = None
+            async for res in generator:
+                final_output = res
+                if not is_moss:
+                    continue
+                try:
+                    step_audio, step_key = self._extract_audio_output(res)
+                except Exception:
+                    continue
+                if step_key is None:
+                    continue
+                chunk = step_audio[step_key]
+                candidates = chunk if isinstance(chunk, list) else [chunk]
+                for cand in candidates:
+                    if hasattr(cand, "numel") and cand.numel() > 0:
+                        moss_chunks.append(cand)
+                sr_step = step_audio.get("sr")
+                if sr_step is not None:
+                    sr_val_step = sr_step[-1] if isinstance(sr_step, list) and sr_step else sr_step
+                    moss_sample_rate = int(sr_val_step.item()) if hasattr(sr_val_step, "item") else int(sr_val_step)
 
-        if final_output is None:
-            raise ValueError("No output generated from the model.")
+            if final_output is None:
+                raise ValueError("No output generated from the model.")
 
-        audio_output, audio_key = self._extract_audio_output(final_output)
-        if audio_key is None:
-            raise ValueError("TTS model did not produce audio output.")
+            audio_output, audio_key = self._extract_audio_output(final_output)
+            if audio_key is None:
+                raise ValueError("TTS model did not produce audio output.")
 
-        audio_tensor = audio_output[audio_key]
-        sr_raw = audio_output.get("sr", 24000)
-        sr_val = sr_raw[-1] if isinstance(sr_raw, list) and sr_raw else sr_raw
-        sample_rate = sr_val.item() if hasattr(sr_val, "item") else int(sr_val)
+            audio_tensor = audio_output[audio_key]
+            sr_raw = audio_output.get("sr", 24000)
+            sr_val = sr_raw[-1] if isinstance(sr_raw, list) and sr_raw else sr_raw
+            sample_rate = sr_val.item() if hasattr(sr_val, "item") else int(sr_val)
 
-        if is_moss:
-            # Prefer the engine's own consolidated audio when present. After the
-            # vllm 0.20 rebase non-stream requests resolve to FINAL_ONLY, so
-            # final_output already carries the full concatenated waveform; the
-            # delta-accumulator below is kept as a fallback for DELTA-style
-            # engines that surface chunks one yield at a time.
-            if isinstance(audio_tensor, list):
-                non_empty_final = [c for c in audio_tensor if hasattr(c, "numel") and c.numel() > 0]
-                final_audio = torch.cat(non_empty_final, dim=-1) if non_empty_final else None
-            elif hasattr(audio_tensor, "numel") and audio_tensor.numel() > 0:
-                final_audio = audio_tensor
-            else:
-                final_audio = None
+            if is_moss:
+                # Prefer the engine's own consolidated audio when present. After the
+                # vllm 0.20 rebase non-stream requests resolve to FINAL_ONLY, so
+                # final_output already carries the full concatenated waveform; the
+                # delta-accumulator below is kept as a fallback for DELTA-style
+                # engines that surface chunks one yield at a time.
+                if isinstance(audio_tensor, list):
+                    non_empty_final = [c for c in audio_tensor if hasattr(c, "numel") and c.numel() > 0]
+                    final_audio = torch.cat(non_empty_final, dim=-1) if non_empty_final else None
+                elif hasattr(audio_tensor, "numel") and audio_tensor.numel() > 0:
+                    final_audio = audio_tensor
+                else:
+                    final_audio = None
 
-            if final_audio is not None:
-                audio_tensor = final_audio
-            elif moss_chunks:
-                audio_tensor = torch.cat(moss_chunks, dim=-1)
-            else:
-                audio_tensor = np.zeros((0,), dtype=np.float32)
-            if moss_sample_rate is not None:
-                sample_rate = moss_sample_rate
-        elif isinstance(audio_tensor, list):
-            async_chunk = bool(getattr(self.engine_client.model_config, "async_chunk", False))
-            if async_chunk:
-                non_empty_chunks = [candidate for candidate in audio_tensor if candidate.numel() > 0]
-                audio_tensor = (
-                    torch.cat(non_empty_chunks, dim=-1) if non_empty_chunks else np.zeros((0,), dtype=np.float32)
-                )
-            else:
-                audio_history = audio_tensor
-                audio_tensor = np.zeros((0,), dtype=np.float32)
-                # Non-async Qwen3-TTS returns cumulative history snapshots, so keep the latest non-empty tensor.
-                for candidate in reversed(audio_history):
-                    if candidate.numel() > 0:
-                        audio_tensor = candidate
-                        break
-        if hasattr(audio_tensor, "float"):
-            audio_tensor = audio_tensor.float().detach().cpu().numpy()
+                if final_audio is not None:
+                    audio_tensor = final_audio
+                elif moss_chunks:
+                    audio_tensor = torch.cat(moss_chunks, dim=-1)
+                else:
+                    audio_tensor = np.zeros((0,), dtype=np.float32)
+                if moss_sample_rate is not None:
+                    sample_rate = moss_sample_rate
+            elif isinstance(audio_tensor, list):
+                async_chunk = bool(getattr(self.engine_client.model_config, "async_chunk", False))
+                if async_chunk:
+                    non_empty_chunks = [candidate for candidate in audio_tensor if candidate.numel() > 0]
+                    audio_tensor = (
+                        torch.cat(non_empty_chunks, dim=-1) if non_empty_chunks else np.zeros((0,), dtype=np.float32)
+                    )
+                else:
+                    audio_history = audio_tensor
+                    audio_tensor = np.zeros((0,), dtype=np.float32)
+                    # Non-async Qwen3-TTS returns cumulative history snapshots, so keep the latest non-empty tensor.
+                    for candidate in reversed(audio_history):
+                        if candidate.numel() > 0:
+                            audio_tensor = candidate
+                            break
+            if hasattr(audio_tensor, "float"):
+                audio_tensor = audio_tensor.float().detach().cpu().numpy()
 
-        if audio_tensor.ndim > 1:
-            audio_tensor = audio_tensor.squeeze()
+            if audio_tensor.ndim > 1:
+                audio_tensor = audio_tensor.squeeze()
 
-        audio_obj = CreateAudio(
-            audio_tensor=audio_tensor,
-            sample_rate=sample_rate,
-            response_format=request.response_format or "wav",
-            speed=request.speed or 1.0,
-            stream_format=request.stream_format,
-            base64_encode=base64_encode,
-        )
-        audio_response: AudioResponse = self.create_audio(audio_obj)
-        return audio_response.audio_data, audio_response.media_type
+            audio_obj = CreateAudio(
+                audio_tensor=audio_tensor,
+                sample_rate=sample_rate,
+                response_format=request.response_format or "wav",
+                speed=request.speed or 1.0,
+                stream_format=request.stream_format,
+                base64_encode=base64_encode,
+            )
+            audio_response: AudioResponse = self.create_audio(audio_obj)
+            self._mark_ref_audio_artifact_ready_for_request(request_id)
+            artifact_ready = True
+            return audio_response.audio_data, audio_response.media_type
+        finally:
+            if not artifact_ready:
+                self._discard_ref_audio_artifact_warmup(request_id)
 
     async def _create_diffusion_speech(
         self,

vllm_omni/entrypoints/stage_utils.py

@@ -114,6 +114,24 @@ def _map_device_list(stage_id: int, device_list: list[str], visible_device_list:
         return list(device_list)
 
     logical_ids = [int(device) for device in device_list]
+
+    # If ALL logical device IDs are >= num_visible, none can be 0-based indices
+    # into the visible list — they must be physical device IDs.  This happens
+    # during multi-replica stage initialization where split_devices_for_replicas
+    # assigns physical device IDs (e.g. "2") that are not part of the current
+    # process's CUDA_VISIBLE_DEVICES but are valid physical GPUs.  Pass them
+    # through directly so the replica subprocess inherits the correct device.
+    if logical_ids and all(d >= num_visible for d in logical_ids):
+        logger.warning(
+            "Stage %s has device IDs %s, none of which are < the visible device count %d "
+            "(visible=%s). Treating them as physical device IDs and passing through.",
+            stage_id,
+            device_list,
+            num_visible,
+            visible_device_list,
+        )
+        return list(device_list)
+
     mapped_devices = [visible_device_list[idx] for idx in logical_ids if idx < num_visible]
     mapping_pairs = [
         f"{logical_id}->{visible_device_list[logical_id]}" for logical_id in logical_ids if logical_id < num_visible

vllm_omni/model_executor/models/cosyvoice3/cosyvoice3.py

@@ -2,7 +2,6 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 import os
 from collections.abc import Iterable, Mapping, Sequence
-from dataclasses import replace
 from functools import partial
 from threading import Lock
 
@@ -365,6 +364,12 @@ class CosyVoice3Model(
         else:
             raise ValueError(f"Model stage not supported {self.model_stage}")
 
+    def get_language_model(self) -> "nn.Module":
+        """Return the language model for upstream MoE detection."""
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     def _create_llm_vllm_config(self, parent_config: VllmConfig) -> VllmConfig:
         """Create VllmConfig for the inner Qwen2 LLM.
 
@@ -542,17 +547,21 @@ class CosyVoice3Model(
         if self.model_stage != "cosyvoice3_talker":
             return None
 
-        sampler = getattr(self, "_talker_sampler", None)
-        if sampler is None:
-            sampler = Sampler()
-            self._talker_sampler = sampler
-
         if not self._cosyvoice3_ras_enabled(sampling_metadata):
+            sampler = getattr(self, "_talker_sampler", None)
+            if sampler is None:
+                sampler = Sampler()
+                self._talker_sampler = sampler
             return sampler(logits=logits, sampling_metadata=sampling_metadata)
 
         logits = logits.to(torch.float32)
-        sampling_for_processors = replace(sampling_metadata, no_penalties=True)
-        logits = sampler.apply_logits_processors(logits, sampling_for_processors, predict_bonus_token=False)
+        # Apply logits processors directly — RAS handles its own repetition
+        # logic.  We avoid instantiating Sampler() here because its import
+        # chain pulls in flashinfer / GPU deps that fail in CPU-only tests.
+        if sampling_metadata.allowed_token_ids_mask is not None:
+            logits.masked_fill_(sampling_metadata.allowed_token_ids_mask, float("-inf"))
+        for processor in sampling_metadata.logitsprocs.non_argmax_invariant:
+            logits = processor.apply(logits)
 
         sampling_cfg = dict(self.config.llm.get("sampling", {}))
         default_top_p = float(sampling_cfg.get("top_p", 0.8))

vllm_omni/model_executor/models/covo_audio/covo_audio.py

@@ -243,6 +243,12 @@ class CovoAudioForConditionalGeneration(
             else lambda *args, **kwargs: None
         )
 
+    def get_language_model(self) -> "nn.Module":
+        """Return the language model for upstream MoE detection."""
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     def fused_thinker_talker_preprocess(
         self,
         input_ids: torch.Tensor | None,

vllm_omni/model_executor/models/ming_flash_omni/ming_flash_omni.py

@@ -143,6 +143,12 @@ class MingFlashOmniForConditionalGeneration(
                 f"For image generation use stage_type: diffusion with model_arch: MingImagePipeline."
             )
 
+    def get_language_model(self) -> "nn.Module":
+        """Return the language model for upstream MoE detection."""
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     def forward(
         self,
         input_ids: torch.Tensor,

vllm_omni/model_executor/models/moss_tts_nano/modeling_moss_tts_nano.py

@@ -400,6 +400,13 @@ class MossTTSNanoForGeneration(nn.Module):
         runtime_additional_information: list[dict[str, Any]] | None = None,
         **kwargs: Any,
     ) -> OmniOutput:
+        if self._lm is None or self._audio_tokenizer is None:
+            # load_weights() loads the real LM from HuggingFace.  When
+            # load_format=dummy the vllm DummyModelLoader skips calling
+            # model.load_weights(), so we do it lazily on first forward.
+            # Must happen before _make_dummy_hidden so self._device is set.
+            self.load_weights([])
+
         sr = getattr(self.config, "audio_tokenizer_sample_rate", 48000)
         sr_tensor = torch.tensor(sr, dtype=torch.int32)
         empty = torch.zeros((0,), dtype=torch.float32)
@@ -418,9 +425,6 @@ class MossTTSNanoForGeneration(nn.Module):
                 },
             )
 
-        if self._lm is None or self._audio_tokenizer is None:
-            raise RuntimeError("MOSS-TTS-Nano model not loaded.  Was load_weights() called?")
-
         outputs: list[torch.Tensor] = []
         srs: list[torch.Tensor] = []
         last_chunk_flags: list[bool] = []

vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni.py

@@ -91,18 +91,17 @@ class Qwen2_5OmniForConditionalGeneration(
 
         elif self.model_stage == "talker":
             multimodal_config.skip_mm_profiling = True
-            # register the process function for the talker stage
             self.has_preprocess = True
             self.set_custom_preprocess(self.talker_preprocess)
             self.thinker = None
             # Initialize talker model wrapper (handles projection + LM)
-            self.talker = init_vllm_registered_model(
-                vllm_config=vllm_config,
-                prefix=maybe_prefix(prefix, "talker"),
-                hf_config=talker_config,
-                # Use registry architecture key
-                architectures=["Qwen2_5OmniTalkerModel"],
-            )
+            with self._mark_language_model(vllm_config=vllm_config):
+                self.talker = init_vllm_registered_model(
+                    vllm_config=vllm_config,
+                    prefix=maybe_prefix(prefix, "talker"),
+                    hf_config=talker_config,
+                    architectures=["Qwen2_5OmniTalkerModel"],
+                )
             self.model = self.talker
             self.token2wav = None
             # set suppress start id according to token2wav
@@ -130,12 +129,13 @@ class Qwen2_5OmniForConditionalGeneration(
             self.token2wav_config = getattr(config, "token2wav_config", None)
             self.token2wav = None
             if self.token2wav_config is not None:
-                self.token2wav = init_vllm_registered_model(
-                    vllm_config=vllm_config,
-                    prefix=maybe_prefix(prefix, "token2wav"),
-                    hf_config=self.token2wav_config,
-                    architectures=["Qwen2_5OmniToken2WavModel"],
-                )
+                with self._mark_language_model(vllm_config=vllm_config):
+                    self.token2wav = init_vllm_registered_model(
+                        vllm_config=vllm_config,
+                        prefix=maybe_prefix(prefix, "token2wav"),
+                        hf_config=self.token2wav_config,
+                        architectures=["Qwen2_5OmniToken2WavModel"],
+                    )
             # voice resources (loaded on demand)
             self._token2wav_conds: dict[str, torch.Tensor] = {}
             self._token2wav_ref_mels: dict[str, torch.Tensor] = {}
@@ -149,6 +149,18 @@ class Qwen2_5OmniForConditionalGeneration(
             (self.thinker.make_empty_intermediate_tensors) if self.model_stage == "thinker" else lambda: None
         )
 
+    def get_language_model(self) -> nn.Module:
+        """Return the language model for MoE detection in upstream load_model.
+
+        Upstream GPUModelRunner.load_model() calls get_language_model() to find
+        the MoE model inside VLM wrappers. For thinker (which IS MoE), delegate
+        to the child. For talker/code2wav (not MoE, not SupportsMultiModal),
+        return self.model directly to avoid NotImplementedError.
+        """
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     # -------------------- Device utilities --------------------
     @staticmethod
     def _module_device(module: nn.Module) -> torch.device:

vllm_omni/model_executor/models/qwen3_omni/qwen3_omni.py

@@ -278,6 +278,12 @@ class Qwen3OmniMoeForConditionalGeneration(
             return self.model.sampler
         return Sampler()
 
+    def get_language_model(self) -> torch.nn.Module:
+        """Delegate to the active stage's language model for upstream MoE resolution."""
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     def embed_input_ids(
         self,
         input_ids: torch.Tensor,

vllm_omni/model_executor/models/qwen3_omni/qwen3_omni_moe_thinker.py

@@ -64,6 +64,7 @@ from vllm.model_executor.layers.quantization import QuantizationConfig
 from vllm.model_executor.layers.vocab_parallel_embedding import ParallelLMHead
 from vllm.model_executor.models.interfaces import (
     MultiModalEmbeddings,
+    SupportsLoRA,
     SupportsMRoPE,
     SupportsMultiModal,
     SupportsPP,
@@ -1071,6 +1072,7 @@ class Qwen3OmniMoeThinkerForConditionalGeneration(
     nn.Module,
     SupportsMultiModal,
     SupportsPP,
+    SupportsLoRA,
     SupportsMRoPE,
     Qwen3OmniMoeConditionalGenerationMixin,
     SupportsTranscription,
@@ -1089,6 +1091,11 @@ class Qwen3OmniMoeThinkerForConditionalGeneration(
             "k_proj",
             "v_proj",
         ],
+        "attn.qkv": [
+            "attn.q",
+            "attn.k",
+            "attn.v",
+        ],
         "gate_up_proj": [
             "gate_proj",
             "up_proj",
@@ -1111,7 +1118,8 @@ class Qwen3OmniMoeThinkerForConditionalGeneration(
     def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
         super().__init__()
         self.vllm_config = vllm_config  # needed for torch compile forward context
-        thinker_config: Qwen3OmniMoeThinkerConfig = vllm_config.model_config.hf_config
+        hf_config = vllm_config.model_config.hf_config
+        thinker_config: Qwen3OmniMoeThinkerConfig = getattr(hf_config, "thinker_config", None) or hf_config
         quant_config = vllm_config.quant_config
         multimodal_config = vllm_config.model_config.multimodal_config
         self.config = thinker_config
@@ -1660,6 +1668,7 @@ class Qwen3OmniMoeThinkerForConditionalGeneration(
         self,
         input_tokens: list[int],
         mm_features: list[MultiModalFeatureSpec],
+        **kwargs,
     ) -> tuple[torch.Tensor, int]:
         """Compute M-RoPE input positions using mm_features directly."""
         seq_len = len(input_tokens)

vllm_omni/model_executor/models/qwen3_tts/prompt_embeds_builder.py

@@ -757,6 +757,14 @@ class Qwen3TTSPromptEmbedsBuilder:
             ref_audio_list = info_dict.get("ref_audio")
             if not isinstance(ref_audio_list, list) or not ref_audio_list:
                 continue
+            cache_key_from_serving = first_value(info_dict.get(REF_AUDIO_CACHE_KEY))
+            if isinstance(cache_key_from_serving, str) and cache_key_from_serving:
+                cached = self.get_ref_audio_artifacts(cache_key_from_serving)
+                if cached is not None:
+                    cached_ref_code = coerce_ref_code_tensor(cached.get("ref_code"), device=device)
+                    if isinstance(cached_ref_code, torch.Tensor):
+                        info_dict.setdefault("codes", {})[PRECOMPUTED_REF_CODE_KEY] = cached_ref_code
+                        continue
             try:
                 wav, sr = self.normalize_ref_audio(ref_audio_list[0])
             except Exception:
@@ -1029,6 +1037,19 @@ class Qwen3TTSPromptEmbedsBuilder:
             xvec_only = bool((info_dict.get("x_vector_only_mode") or [False])[0])
             in_context_mode = not xvec_only
             voice_clone_prompt = _normalize_voice_clone_prompt(info_dict.get("voice_clone_prompt"))
+            if voice_clone_prompt is not None and "icl_mode" in voice_clone_prompt:
+                icl_flag = _as_singleton(voice_clone_prompt.get("icl_mode"))
+                if isinstance(icl_flag, bool):
+                    in_context_mode = icl_flag
+                    xvec_only = not in_context_mode
+            prompt_has_ref_code = voice_clone_prompt is not None and self._has_ref_code_like(
+                voice_clone_prompt.get("ref_code")
+            )
+            prompt_spk = voice_clone_prompt.get("ref_spk_embedding") if voice_clone_prompt is not None else None
+            prompt_has_speaker = isinstance(prompt_spk, torch.Tensor) or (
+                isinstance(prompt_spk, (list, np.ndarray)) and len(prompt_spk) > 0
+            )
+            needs_ref_audio_cache_key = not (prompt_has_speaker and (not in_context_mode or prompt_has_ref_code))
             ref_audio_wav: np.ndarray | None = None
             ref_audio_sr: int | None = None
 
@@ -1056,7 +1077,8 @@ class Qwen3TTSPromptEmbedsBuilder:
             # from ref_audio: x_vector_only_mode, precomputed ref_code
             # with no ref_spk_embedding, voice_clone_prompt.icl_mode=False.
             ref_audio_cache_key = info_dict.pop(REF_AUDIO_CACHE_KEY, None)
-            if not (isinstance(ref_audio_cache_key, str) and ref_audio_cache_key):
+            ref_audio_cache_key = first_value(ref_audio_cache_key)
+            if needs_ref_audio_cache_key and not (isinstance(ref_audio_cache_key, str) and ref_audio_cache_key):
                 normalized = info_dict.get(NORMALIZED_REF_AUDIO_KEY)
                 wav_peek: np.ndarray | None = None
                 sr_peek: int | None = None
@@ -1075,13 +1097,25 @@ class Qwen3TTSPromptEmbedsBuilder:
             cached_artifacts = None
             if isinstance(ref_audio_cache_key, str) and ref_audio_cache_key:
                 cached_artifacts = self.get_ref_audio_artifacts(ref_audio_cache_key)
+            ref_audio_list = info_dict.get("ref_audio")
+            has_ref_audio_payload = isinstance(ref_audio_list, list) and bool(ref_audio_list)
+            artifact_only = (
+                isinstance(ref_audio_cache_key, str) and bool(ref_audio_cache_key) and not has_ref_audio_payload
+            )
+            if artifact_only and cached_artifacts is None:
+                raise RuntimeError(
+                    "Qwen3-TTS ref_audio artifact cache miss for artifact-only request; "
+                    "retry with ref_audio or precompute a custom voice profile."
+                )
 
             # Speaker cache: only for uploaded (named) speakers
             _speaker_cache_key = None
+            _cache_lookup_voice = None
             if voice_clone_prompt is None and self._speaker_cache is not None:
                 _speaker_list = info_dict.get("speaker")
                 if isinstance(_speaker_list, list) and _speaker_list:
                     _voice_name = str(_speaker_list[0]).lower()
+                    _cache_lookup_voice = _voice_name
                     # Per-mode namespace — xvec and icl produce different artifacts
                     # for the same voice, so they must not share a cache slot.
                     _mode = "xvec" if xvec_only else "icl"
@@ -1103,15 +1137,18 @@ class Qwen3TTSPromptEmbedsBuilder:
                             "ref_code": ref_code_cached,
                             "ref_spk_embedding": ref_spk_embed_cached,
                             "icl_mode": _cached.get("icl_mode"),
+                            "ref_text": _cached.get("ref_text"),
                         }
                         _speaker_cache_key = None  # hit → don't store again
 
-            # Official implementation may pass `voice_clone_prompt.icl_mode`.
-            if voice_clone_prompt is not None and "icl_mode" in voice_clone_prompt:
-                icl_flag = _as_singleton(voice_clone_prompt.get("icl_mode"))
-                if isinstance(icl_flag, bool):
-                    in_context_mode = icl_flag
-                    xvec_only = not in_context_mode
+            if voice_clone_prompt is None and _speaker_cache_key is not None:
+                ref_audio_list = info_dict.get("ref_audio")
+                if not isinstance(ref_audio_list, list) or not ref_audio_list:
+                    raise ValueError(
+                        f"Qwen3-TTS speaker '{_cache_lookup_voice}' was requested without ref_audio, "
+                        "but no precomputed cache entry was loaded"
+                    )
+
             ref_code = None
             if voice_clone_prompt is not None:
                 ref_code = _as_singleton(voice_clone_prompt.get("ref_code"))
@@ -1136,6 +1173,8 @@ class Qwen3TTSPromptEmbedsBuilder:
                 if isinstance(cached_ref_code, torch.Tensor):
                     ref_code_t = cached_ref_code
                     ref_code_len = int(ref_code_t.shape[0])
+            if ref_code_t is None and in_context_mode and artifact_only:
+                raise RuntimeError("Qwen3-TTS ref_audio artifact cache entry is missing ref_code.")
             if ref_code_t is None and in_context_mode:
                 # Compute ref_code from ref_audio if not provided.
                 wav_np, sr = _get_ref_audio()
@@ -1161,6 +1200,8 @@ class Qwen3TTSPromptEmbedsBuilder:
                     .to(device=input_ids.device, dtype=torch.bfloat16)
                     .view(1, 1, -1)
                 )
+            elif artifact_only:
+                raise RuntimeError("Qwen3-TTS ref_audio artifact cache entry is missing ref_spk_embedding.")
             else:
                 wav_np, sr = _get_ref_audio()
                 speaker_embed = self.extract_speaker_embedding(wav_np, sr).view(1, 1, -1)
@@ -1207,6 +1248,8 @@ class Qwen3TTSPromptEmbedsBuilder:
                     )
                 if ref_ids is None:
                     ref_text = _as_singleton(info_dict.get("ref_text"))
+                    if (not isinstance(ref_text, str) or not ref_text.strip()) and voice_clone_prompt is not None:
+                        ref_text = _as_singleton(voice_clone_prompt.get("ref_text"))
                     if isinstance(ref_text, str) and ref_text.strip():
                         ref_ids = tok(
                             build_ref_text(ref_text),
@@ -1489,6 +1532,13 @@ class Qwen3TTSPromptEmbedsBuilder:
                     except Exception:
                         ref_code_len = None
 
+                if ref_code_len is None:
+                    ref_code_length = _first(info.get("ref_code_length"), None)
+                    try:
+                        if ref_code_length is not None:
+                            ref_code_len = int(ref_code_length)  # type: ignore[arg-type]
+                    except (TypeError, ValueError):
+                        ref_code_len = None
                 if ref_code_len is None and estimate_ref_code_len is not None:
                     ref_code_len = estimate_ref_code_len(info.get("ref_audio"))
                 if ref_code_len is None:
@@ -1511,6 +1561,10 @@ class Qwen3TTSPromptEmbedsBuilder:
 
                     if ref_ids is None:
                         ref_text = _first(info.get("ref_text"), "")
+                        if (not isinstance(ref_text, str) or not ref_text.strip()) and isinstance(
+                            voice_clone_prompt, dict
+                        ):
+                            ref_text = _first(voice_clone_prompt.get("ref_text"), "")
                         if not isinstance(ref_text, str) or not ref_text.strip():
                             raise ValueError(
                                 "Base in-context non-streaming requires `ref_text` or tokenized `ref_ids`."

vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_code2wav.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from collections import Counter
+from collections import Counter, OrderedDict
 from collections.abc import Iterable
 from typing import Any
 
@@ -24,6 +24,9 @@ from .tokenizer_12hz.modeling_qwen3_tts_tokenizer_v2 import (
 
 logger = init_logger(__name__)
 
+_REF_CONTEXT_CACHE_MAX_ENTRIES = 4096
+_REF_CONTEXT_CACHE_MAX_BYTES = 64 * 1024 * 1024
+
 
 def _codec_ids_from_payload_or_input(
     input_ids: torch.Tensor,
@@ -100,6 +103,10 @@ class Qwen3TTSCode2Wav(nn.Module):
         self._output_sample_rate = int(tok_config.output_sample_rate)
         self._total_upsample = int(self.decoder.total_upsample)
         self._decoder_sliding_window = int(getattr(dec_config, "sliding_window", 0) or 0)
+        self._ref_context_cache: OrderedDict[str, torch.Tensor] = OrderedDict()
+        self._ref_context_cache_bytes = 0
+        self._ref_context_cache_max_entries = _REF_CONTEXT_CACHE_MAX_ENTRIES
+        self._ref_context_cache_max_bytes = _REF_CONTEXT_CACHE_MAX_BYTES
 
     def embed_input_ids(self, input_ids: torch.Tensor, **_: Any) -> torch.Tensor:
         # This stage ignores token embeddings. Keep a stable dummy embedding for vLLM runner.
@@ -205,6 +212,49 @@ class Qwen3TTSCode2Wav(nn.Module):
         self._batch_stats_actual_frames.update(actual_frames)
         self._batch_stats_bucket_groups[(group_size, bucket_frames)] += 1
 
+    @staticmethod
+    def _tensor_nbytes(tensor: torch.Tensor) -> int:
+        return int(tensor.numel() * tensor.element_size())
+
+    def _evict_ref_context_cache_if_needed(self) -> None:
+        evicted = 0
+        while len(self._ref_context_cache) > self._ref_context_cache_max_entries:
+            _, cached = self._ref_context_cache.popitem(last=False)
+            self._ref_context_cache_bytes -= self._tensor_nbytes(cached)
+            evicted += 1
+        while self._ref_context_cache_bytes > self._ref_context_cache_max_bytes and len(self._ref_context_cache) > 1:
+            _, cached = self._ref_context_cache.popitem(last=False)
+            self._ref_context_cache_bytes -= self._tensor_nbytes(cached)
+            evicted += 1
+        if evicted:
+            logger.debug(
+                "Evicted %d Qwen3-TTS ref context cache entries; entries=%d bytes=%d",
+                evicted,
+                len(self._ref_context_cache),
+                self._ref_context_cache_bytes,
+            )
+
+    def _cache_ref_context(self, request_id: str, tensor: torch.Tensor) -> None:
+        previous = self._ref_context_cache.pop(request_id, None)
+        if previous is not None:
+            self._ref_context_cache_bytes -= self._tensor_nbytes(previous)
+        cached = tensor.detach().contiguous()
+        self._ref_context_cache[request_id] = cached
+        self._ref_context_cache.move_to_end(request_id)
+        self._ref_context_cache_bytes += self._tensor_nbytes(cached)
+        self._evict_ref_context_cache_if_needed()
+
+    def _get_ref_context(self, request_id: str) -> torch.Tensor | None:
+        cached = self._ref_context_cache.get(request_id)
+        if cached is not None:
+            self._ref_context_cache.move_to_end(request_id)
+        return cached
+
+    def _pop_ref_context(self, request_id: str) -> None:
+        cached = self._ref_context_cache.pop(request_id, None)
+        if cached is not None:
+            self._ref_context_cache_bytes -= self._tensor_nbytes(cached)
+
     def log_decode_batch_stats(self) -> None:
         if not self._batch_stats_enabled or self._batch_stats_requests == 0:
             return
@@ -268,6 +318,32 @@ class Qwen3TTSCode2Wav(nn.Module):
         valid_codes_qf: list[torch.Tensor] = []
         valid_indices: list[int] = []
         left_context_size = [0] * len(request_ids_list)
+        ref_context_size = [0] * len(request_ids_list)
+        ref_context_request_ids: list[str | None] = [None] * len(request_ids_list)
+        ref_context_included = [False] * len(request_ids_list)
+        finished_flags = [False] * len(request_ids_list)
+
+        def _meta_int(value: Any) -> int:
+            if isinstance(value, list):
+                value = value[0] if value else 0
+            if isinstance(value, torch.Tensor):
+                value = value.reshape(-1)[0].item() if value.numel() > 0 else 0
+            return int(value or 0)
+
+        def _meta_str(value: Any) -> str | None:
+            if isinstance(value, list):
+                value = value[0] if value else None
+            if value is None:
+                return None
+            return str(value)
+
+        def _meta_bool(value: Any) -> bool:
+            if isinstance(value, list):
+                value = value[0] if value else False
+            if isinstance(value, torch.Tensor):
+                return bool(value.reshape(-1)[0].item()) if value.numel() > 0 else False
+            return bool(value)
+
         if runtime_infos:
             for i, info in enumerate(runtime_infos):
                 if i >= len(left_context_size):
@@ -276,7 +352,15 @@ class Qwen3TTSCode2Wav(nn.Module):
                     continue
                 meta = info.get("meta", {})
                 if "left_context_size" in meta:
-                    left_context_size[i] = meta["left_context_size"]
+                    left_context_size[i] = _meta_int(meta["left_context_size"])
+                if "ref_context_size" in meta:
+                    ref_context_size[i] = _meta_int(meta["ref_context_size"])
+                if "ref_context_request_id" in meta:
+                    ref_context_request_ids[i] = _meta_str(meta["ref_context_request_id"])
+                if "ref_context_included" in meta:
+                    ref_context_included[i] = _meta_bool(meta["ref_context_included"])
+                if "finished" in meta:
+                    finished_flags[i] = _meta_bool(meta["finished"])
         for i, req_ids in enumerate(request_ids_list):
             runtime_info = runtime_infos[i] if i < len(runtime_infos) else None
             req_ids = _codec_ids_from_payload_or_input(req_ids, runtime_info)
@@ -284,6 +368,7 @@ class Qwen3TTSCode2Wav(nn.Module):
                 parsed.append((0, 0))
                 continue
             ctx_frames = left_context_size[i]
+            ref_ctx_frames = ref_context_size[i]
             flat = req_ids
             n = flat.numel()
             if n == 0 or n % q != 0:
@@ -302,6 +387,25 @@ class Qwen3TTSCode2Wav(nn.Module):
             frames = n // q
             # [q*F] -> [Q, F] for direct decoder call (decoder expects [B, Q, F])
             codes_qf = flat.reshape(q, frames)
+            ref_req_id = ref_context_request_ids[i]
+            if ref_req_id is not None and ref_ctx_frames > 0:
+                if ref_context_included[i]:
+                    if frames < ref_ctx_frames:
+                        raise ValueError(
+                            "Qwen3-TTS ref context metadata says ref prefix is included, "
+                            f"but frames={frames} < ref_context_size={ref_ctx_frames}"
+                        )
+                    self._cache_ref_context(ref_req_id, codes_qf[:, :ref_ctx_frames])
+                else:
+                    cached_ref = self._get_ref_context(ref_req_id)
+                    if cached_ref is None:
+                        raise ValueError(
+                            "Missing Qwen3-TTS ref context cache for "
+                            f"request {ref_req_id!r}; first chunk must include ref_code"
+                        )
+                    cached_ref = cached_ref.to(device=codes_qf.device, dtype=codes_qf.dtype)
+                    codes_qf = torch.cat((cached_ref, codes_qf), dim=1)
+                    frames = int(codes_qf.shape[1])
             parsed.append((ctx_frames, frames))
             valid_codes_qf.append(codes_qf)
             valid_indices.append(i)
@@ -443,6 +547,10 @@ class Qwen3TTSCode2Wav(nn.Module):
             if wav.shape[0] > 0:
                 audios[idx] = wav.to(dtype=torch.float32).reshape(-1)
 
+        for req_id, finished in zip(ref_context_request_ids, finished_flags, strict=False):
+            if req_id is not None and finished:
+                self._pop_ref_context(req_id)
+
         return OmniOutput(
             text_hidden_states=None,
             multimodal_outputs={"model_outputs": audios, "sr": srs},

vllm_omni/model_executor/models/qwen3_tts/qwen3_tts_talker.py

@@ -23,7 +23,11 @@ from vllm.sequence import IntermediateTensors
 
 from vllm_omni.data_entry_keys import OmniPayload
 from vllm_omni.model_executor.models.output_templates import OmniOutput
-from vllm_omni.utils.speaker_cache import get_speaker_cache
+from vllm_omni.utils.speaker_cache import (
+    get_speaker_cache,
+    iter_custom_voice_profiles,
+    load_validated_profile_tensors,
+)
 
 from .configuration_qwen3_tts import Qwen3TTSConfig, Qwen3TTSSpeakerEncoderConfig, Qwen3TTSTalkerConfig
 from .prompt_embeds_builder import Qwen3TTSPromptEmbedsBuilder
@@ -441,6 +445,46 @@ class Qwen3TTSTalkerForConditionalGeneration(nn.Module):
             encode_ref_audio_batch=self._encode_ref_audio_batch,
             speaker_cache=self._speaker_cache,
         )
+        self._load_custom_voice_profiles()
+
+    # -------------------- custom voice profiles --------------------
+
+    def _load_custom_voice_profiles(self) -> None:
+        """Preload offline Qwen3-TTS custom voice profiles into speaker cache."""
+        custom_voice_dir = getattr(self.config, "custom_voice_dir", None)
+        if not custom_voice_dir:
+            return
+
+        expected_dim = int(getattr(self.config.speaker_encoder_config, "enc_dim", 0) or 0)
+        loaded = 0
+        for profile in iter_custom_voice_profiles(custom_voice_dir, expected_model_type="qwen3_tts"):
+            tensors = load_validated_profile_tensors(
+                profile,
+                expected_model_type="qwen3_tts",
+                qwen3_embedding_dim=expected_dim,
+            )
+            if tensors is None:
+                continue
+
+            speaker_embedding = tensors["speaker_embedding"].reshape(-1).contiguous().cpu()
+            mode = str(profile.get("mode") or "xvec").lower()
+            ref_code = tensors.get("ref_code")
+            artifacts: dict[str, Any] = {
+                "ref_spk_embedding": speaker_embedding,
+                "ref_code": ref_code.contiguous().cpu() if isinstance(ref_code, torch.Tensor) else None,
+                "icl_mode": mode == "icl",
+                "ref_text": profile.get("ref_text"),
+            }
+            key = self._speaker_cache.make_cache_key(
+                profile["voice_name_lower"],
+                model_type=f"qwen3_tts_{mode}",
+                created_at=0,
+            )
+            self._speaker_cache.put(key, artifacts)
+            loaded += 1
+
+        if loaded:
+            logger.info("Loaded %d precomputed Qwen3-TTS custom voice profile(s) from %s", loaded, custom_voice_dir)
 
     # -------------------- vLLM required hooks --------------------
 

vllm_omni/model_executor/models/registry.py

@@ -2,6 +2,7 @@ from vllm.model_executor.models.registry import (
     _VLLM_MODELS,
     _LazyRegisteredModel,
     _ModelRegistry,
+    _resolve_module_name,
 )
 
 _OMNI_MODELS = {
@@ -261,7 +262,7 @@ OmniModelRegistry = _ModelRegistry(
     {
         **{
             model_arch: _LazyRegisteredModel(
-                module_name=f"vllm.model_executor.models.{mod_relname}",
+                module_name=_resolve_module_name(mod_relname),
                 class_name=cls_name,
             )
             for model_arch, (mod_relname, cls_name) in _VLLM_MODELS.items()

vllm_omni/model_executor/models/voxcpm2/voxcpm2_talker.py

@@ -35,7 +35,11 @@ from vllm.platforms import current_platform
 from vllm.sequence import IntermediateTensors
 
 from vllm_omni.model_executor.models.output_templates import OmniOutput
-from vllm_omni.utils.speaker_cache import get_speaker_cache
+from vllm_omni.utils.speaker_cache import (
+    get_speaker_cache,
+    iter_custom_voice_profiles,
+    load_validated_profile_tensors,
+)
 
 from .minicpm4_paged import MiniCPM4PagedForVoxCPM2, MiniCPM4PagedResidualLM
 from .voxcpm2_import_utils import import_voxcpm2_core
@@ -97,6 +101,7 @@ def build_voxcpm2_prompt(
     ref_audio: Any | None = None,
     ref_sr: int | None = None,
     ref_text: str | None = None,
+    voice_profile: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build a VoxCPM2 prefill prompt whose ``prompt_token_ids`` length matches
     the talker-side prefill length.
@@ -110,7 +115,24 @@ def build_voxcpm2_prompt(
         ids = ids[1:]
     prefill_len = len(ids) + 1  # + audio_start
     additional: dict[str, Any] = {"text_token_ids": [ids]}
-    if ref_audio is not None:
+    if voice_profile is not None and ref_audio is None:
+        mode = str(voice_profile.get("mode") or "reference").lower()
+        ref_audio_feat_len = int(voice_profile.get("ref_audio_feat_len") or 0)
+        audio_feat_len = int(voice_profile.get("audio_feat_len") or 0)
+        prompt_text = voice_profile.get("prompt_text") or voice_profile.get("ref_text")
+        additional["voice_profile"] = dict(voice_profile)
+
+        if mode in ("reference", "ref_continuation"):
+            prefill_len += ref_audio_feat_len + 2  # ref_start / ref_end
+        if mode in ("continuation", "ref_continuation"):
+            if isinstance(prompt_text, str) and prompt_text:
+                additional["prompt_text"] = [prompt_text]
+                prompt_ids = split_multichar_chinese(tokenizer.encode(prompt_text, add_special_tokens=True), split_map)
+                if prompt_ids and prompt_ids[0] == bos:
+                    prompt_ids = prompt_ids[1:]
+                prefill_len += len(prompt_ids)
+            prefill_len += audio_feat_len
+    elif ref_audio is not None:
         vae = hf_config.audio_vae_config
         patch_samples = hf_config.patch_size * math.prod(vae["encoder_rates"])
         ref_len = math.ceil(math.ceil(len(ref_audio) * vae["sample_rate"] / ref_sr) / patch_samples)
@@ -450,6 +472,7 @@ class VoxCPM2TalkerForConditionalGeneration(nn.Module):
 
         # Speaker cache for ref_audio_feat across requests
         self._speaker_cache = get_speaker_cache()
+        self._load_custom_voice_profiles()
 
         self._perf = _PerfTimer(enabled=_ENABLE_PROFILING)
         self._cfm_buffers: _CFMBufferManager | None = None
@@ -472,6 +495,51 @@ class VoxCPM2TalkerForConditionalGeneration(nn.Module):
         # one-shot by design: fires at most once per process to avoid log spam.
         self._active_state_warned = False
 
+    # -------------------- custom voice profiles --------------------
+
+    @staticmethod
+    def _clone_prompt_cache(cache: dict[str, Any]) -> dict[str, Any]:
+        cloned: dict[str, Any] = {}
+        for key, value in cache.items():
+            cloned[key] = value.clone() if isinstance(value, torch.Tensor) else value
+        return cloned
+
+    def _load_custom_voice_profiles(self) -> None:
+        """Preload offline VoxCPM2 prompt caches into the shared speaker cache."""
+        custom_voice_dir = getattr(self.config, "custom_voice_dir", None)
+        if not custom_voice_dir:
+            return
+
+        loaded = 0
+        for profile in iter_custom_voice_profiles(custom_voice_dir, expected_model_type="voxcpm2"):
+            tensors = load_validated_profile_tensors(profile, expected_model_type="voxcpm2")
+            if tensors is None:
+                continue
+
+            ref_audio_feat = tensors.get("ref_audio_feat")
+            audio_feat = tensors.get("audio_feat")
+            mode = str(profile.get("mode") or "").lower()
+
+            prompt_cache: dict[str, Any] = {"mode": mode}
+            if ref_audio_feat is not None:
+                prompt_cache["ref_audio_feat"] = ref_audio_feat.contiguous().cpu()
+            if audio_feat is not None:
+                prompt_cache["audio_feat"] = audio_feat.contiguous().cpu()
+            prompt_text = profile.get("prompt_text") or profile.get("ref_text")
+            if isinstance(prompt_text, str) and prompt_text:
+                prompt_cache["prompt_text"] = prompt_text
+
+            key = self._speaker_cache.make_cache_key(
+                profile["voice_name_lower"],
+                model_type="voxcpm2",
+                created_at=0,
+            )
+            self._speaker_cache.put(key, prompt_cache)
+            loaded += 1
+
+        if loaded:
+            logger.info("Loaded %d precomputed VoxCPM2 custom voice profile(s) from %s", loaded, custom_voice_dir)
+
     @property
     def tts(self) -> nn.Module:
         return self._tts
@@ -1161,6 +1229,10 @@ class VoxCPM2TalkerForConditionalGeneration(nn.Module):
                 prompt_audio = prompt_audio[0] if prompt_audio else None
             if isinstance(prompt_text, list):
                 prompt_text = prompt_text[0] if prompt_text else None
+            voice_profile = info_dict.get("voice_profile")
+            if isinstance(voice_profile, list):
+                voice_profile = voice_profile[0] if voice_profile else None
+            requires_precomputed_cache = isinstance(voice_profile, dict)
 
             state.prompt_cache = None
             voice_name = info_dict.get("voice_name")
@@ -1168,20 +1240,29 @@ class VoxCPM2TalkerForConditionalGeneration(nn.Module):
                 voice_name = voice_name[0] if voice_name else None
             _created_at = int(info_dict.get("voice_created_at") or 0)
 
-            if ref_audio or (prompt_audio and prompt_text):
-                # Check speaker cache for reference-only mode
-                if voice_name and ref_audio and not prompt_audio:
-                    _cache_key = self._speaker_cache.make_cache_key(
-                        voice_name, model_type="voxcpm2", created_at=_created_at
-                    )
-                    cached = self._speaker_cache.get(_cache_key)
-                    if cached is not None:
+            if voice_name:
+                _cache_key = self._speaker_cache.make_cache_key(
+                    voice_name, model_type="voxcpm2", created_at=_created_at
+                )
+                cached = self._speaker_cache.get(_cache_key)
+                if cached is not None:
+                    if "mode" in cached:
+                        state.prompt_cache = self._clone_prompt_cache(cached)
+                    elif "ref_audio_feat" in cached:
                         state.prompt_cache = {
                             "mode": "reference",
                             "ref_audio_feat": cached["ref_audio_feat"].clone(),
                         }
-                        logger.debug("Speaker cache HIT for VoxCPM2 speaker '%s'", voice_name)
+                    logger.debug("Speaker cache HIT for VoxCPM2 speaker '%s'", voice_name)
 
+            if state.prompt_cache is None and requires_precomputed_cache:
+                requested_mode = voice_profile.get("mode") if isinstance(voice_profile, dict) else None
+                raise ValueError(
+                    f"Precomputed VoxCPM2 voice '{voice_name}' was accepted by serving but is not loaded "
+                    f"in the model cache (voice_created_at={_created_at}, mode={requested_mode})"
+                )
+
+            if state.prompt_cache is None and (ref_audio or (prompt_audio and prompt_text)):
                 if state.prompt_cache is None:
                     try:
                         state.prompt_cache = self._build_prompt_cache(

vllm_omni/model_executor/models/voxtral_tts/voxtral_tts.py

@@ -159,6 +159,13 @@ class VoxtralTTSForConditionalGeneration(
             raise ValueError("Invalid model stage")
 
     # -------------------- CUDA Graph for acoustic transformer --------------------
+
+    def get_language_model(self) -> "nn.Module":
+        """Return the language model for upstream MoE detection."""
+        if hasattr(self.model, "get_language_model"):
+            return self.model.get_language_model()
+        return self.model
+
     def _enable_acoustic_transformer_cudagraph(self):
         """Initialize and capture CUDA graphs for compute_mm_logits."""
         if self.model_stage != "audio_generation" or not hasattr(self, "_vllm_config"):

vllm_omni/model_executor/stage_input_processors/qwen3_tts.py

@@ -248,12 +248,29 @@ def talker2code2wav_async_chunk(
     left_context_size = max(0, end_index - context_length)
     window_frames = transfer_manager.code_prompt_token_ids[request_id][-end_index:]
 
-    # Prepend a bounded ref_code tail as decoder context for every chunk so the
-    # vocoder keeps voice-clone speaker identity without making Stage1 shapes
-    # depend on full reference-audio length. The decoder is causal with sliding
-    # attention, so frames older than this context window cannot affect the
-    # emitted chunk. Use `.get()` (not `.pop()`) to keep ref_code for later chunks.
+    # Prepend the bounded ref_code tail to the first emitted chunk so Code2Wav
+    # can cache the same reference context that mainline would otherwise send on
+    # every chunk. Follow-up chunks send only a metadata handle; Code2Wav restores
+    # this context locally before decode.
     ref_code = request_payload.get(request_id)
+    ref_context_size = 0
+    ref_context_request_id: str | None = None
+    ref_context_included = False
+    if isinstance(ref_code, torch.Tensor) and ref_code.numel() > 0:
+        if ref_code.ndim == 1:
+            num_quantizers = len(window_frames[0])
+            if ref_code.numel() % num_quantizers == 0:
+                ref_code = ref_code.reshape(-1, num_quantizers)
+            else:
+                logger.warning(
+                    "Ignoring malformed ref_code with %d elements not divisible by num_quantizers=%d",
+                    ref_code.numel(),
+                    num_quantizers,
+                )
+                ref_code = None
+        elif ref_code.ndim != 2:
+            logger.warning("Ignoring malformed ref_code shape %s", tuple(ref_code.shape))
+            ref_code = None
     if isinstance(ref_code, torch.Tensor) and ref_code.numel() > 0:
         ref_context = ref_code
         if ref_code_context_frames > 0 and int(ref_context.shape[0]) > ref_code_context_frames:
@@ -263,9 +280,15 @@ def talker2code2wav_async_chunk(
                 int(ref_context.shape[0]),
             )
             ref_context = ref_context[-ref_code_context_frames:]
-        ref_frames = ref_context.tolist()
-        window_frames = ref_frames + window_frames
-        left_context_size += len(ref_frames)
+        ref_context_size = int(ref_context.shape[0]) if ref_context.ndim > 1 else 0
+        if ref_context_size > 0:
+            ref_context_request_id = request_id
+            emitted_chunks = int(transfer_manager.put_req_chunk.get(request_id, 0))
+            if emitted_chunks <= 0:
+                ref_frames = ref_context.tolist()
+                window_frames = ref_frames + window_frames
+                ref_context_included = True
+            left_context_size += ref_context_size
 
     num_quantizers = len(window_frames[0])
     num_frames = len(window_frames)
@@ -274,12 +297,18 @@ def talker2code2wav_async_chunk(
         dtype=torch.long,
     )
 
+    meta = MetaStruct(
+        left_context_size=left_context_size,
+        finished=torch.tensor(finished, dtype=torch.bool),
+    )
+    if ref_context_size > 0 and ref_context_request_id is not None:
+        meta.ref_context_size = ref_context_size
+        meta.ref_context_request_id = ref_context_request_id
+        meta.ref_context_included = ref_context_included
+
     return OmniPayloadStruct(
         codes=CodesStruct(audio=code_predictor_codes),
-        meta=MetaStruct(
-            left_context_size=left_context_size,
-            finished=torch.tensor(finished, dtype=torch.bool),
-        ),
+        meta=meta,
         speaker=extract_speaker_from_request(request),
         language=extract_language_from_request(request),
     )

vllm_omni/patch.py

@@ -128,3 +128,25 @@ for module_name, module in list(sys.modules.items()):
         module.StreamingUpdate = OmniStreamingUpdate
     if hasattr(module, "EngineCoreRequest") and module.EngineCoreRequest == _OriginalEngineCoreRequest:
         module.EngineCoreRequest = OmniEngineCoreRequest
+
+
+# Patch: add qwen3_omni_moe to vllm's chat template fallback registry.
+# Qwen/Qwen3-Omni-30B-A3B-Instruct stores its chat_template in a standalone
+# chat_template.json (not in tokenizer_config.json).  transformers < 5.9.0
+# does not load this file, so the tokenizer has no chat_template attribute.
+# vllm's resolve_chat_template falls back to MODEL_TYPE_TO_CHAT_TEMPLATE
+# which has "qwen" but not "qwen3_omni_moe".  Register the same fallback.
+def _patch_chat_template_registry():
+    try:
+        from vllm.transformers_utils.chat_templates.registry import (
+            _MODEL_TYPE_TO_CHAT_TEMPLATE_FALLBACK,
+            _get_qwen_chat_template_fallback,
+        )
+
+        if "qwen3_omni_moe" not in _MODEL_TYPE_TO_CHAT_TEMPLATE_FALLBACK:
+            _MODEL_TYPE_TO_CHAT_TEMPLATE_FALLBACK["qwen3_omni_moe"] = _get_qwen_chat_template_fallback
+    except ImportError:
+        pass
+
+
+_patch_chat_template_registry()

vllm_omni/platforms/cuda/platform.py

@@ -229,4 +229,4 @@ class CudaOmniPlatform(OmniPlatform, CudaPlatformBase):
         if envs.VLLM_USE_OINK_OPS:
             rms_norm = ["oink"] + default
 
-        return IrOpPriorityConfig.with_default(default, rms_norm=rms_norm)
+        return IrOpPriorityConfig.with_default(default, rms_norm=rms_norm, fused_add_rms_norm=rms_norm)

vllm_omni/platforms/interface.py

@@ -11,6 +11,7 @@ from vllm.config import CUDAGraphMode, VllmConfig
 from vllm.forward_context import BatchDescriptor
 from vllm.logger import init_logger
 from vllm.platforms import Platform
+from vllm.platforms.interface import PlatformEnum
 
 logger = init_logger(__name__)
 
@@ -239,6 +240,7 @@ class OmniPlatform(Platform):
 
 class UnspecifiedOmniPlatform(OmniPlatform):
     _omni_enum = OmniPlatformEnum.UNSPECIFIED
+    _enum = PlatformEnum.UNSPECIFIED
     device_type = ""
 
     @classmethod

vllm_omni/platforms/npu/worker/npu_ar_model_runner.py

@@ -16,11 +16,6 @@ from vllm.distributed.kv_transfer import get_kv_transfer_group, has_kv_transfer_
 from vllm.distributed.parallel_state import get_pp_group, get_tp_group
 from vllm.forward_context import BatchDescriptor
 from vllm.logger import logger
-from vllm.model_executor.layers.fused_moe.routed_experts_capturer import (
-    extract_routed_experts_for_current_batch,
-    get_global_experts_capturer,
-    issue_routing_d2h_copy,
-)
 from vllm.sequence import IntermediateTensors
 from vllm.v1.core.sched.output import GrammarOutput, SchedulerOutput
 from vllm.v1.outputs import (
@@ -257,7 +252,7 @@ class NPUARModelRunner(OmniNPUModelRunner):
         intermediate_tensors: IntermediateTensors | None = None,
     ) -> OmniModelRunnerOutput | IntermediateTensors | None:
         if self.vllm_config.model_config.enable_return_routed_experts:
-            capturer = get_global_experts_capturer()
+            capturer = self.routed_experts_capturer
             if capturer is not None and hasattr(capturer, "finalize_pending_copy"):
                 capturer.finalize_pending_copy()
         if self.ascend_config.profiling_chunk_config.enabled:
@@ -728,12 +723,7 @@ class NPUARModelRunner(OmniNPUModelRunner):
             deferred_state_corrections_fn()
 
         if self.vllm_config.model_config.enable_return_routed_experts and hasattr(self, "_positions_cpu"):
-            issue_routing_d2h_copy(
-                input_batch_req_ids=self.input_batch.req_ids,
-                num_scheduled_tokens=scheduler_output.num_scheduled_tokens,
-                positions=self.positions,
-                positions_cpu=self._positions_cpu,
-            )
+            self._omni_routed_experts_d2h(scheduler_output)
 
         return None
 
@@ -911,17 +901,11 @@ class NPUARModelRunner(OmniNPUModelRunner):
             if self.speculative_config is not None:
                 self.finalize_kv_connector()
 
-        routed_experts_dict = None
+        routed_experts_lists = None
         if self.model_config.enable_return_routed_experts:
-            capturer = get_global_experts_capturer()
+            capturer = self.routed_experts_capturer
             if capturer is not None and hasattr(self.input_batch, "num_tokens_no_spec"):
-                routed_experts_dict = extract_routed_experts_for_current_batch(
-                    req_ids=req_ids_output_copy,
-                    requests=self.requests,
-                    req_id_to_index=self.input_batch.req_id_to_index,
-                    num_tokens_no_spec=self.input_batch.num_tokens_no_spec,
-                    max_model_len=self.max_model_len,
-                )
+                routed_experts_lists = self._omni_extract_routed_experts(scheduler_output)
 
         #  -------------------------------------- Omni-new -------------------------------------------------
         engine_output_type, downstream_req_ids = self._resolve_pooler_payload_req_ids(req_ids_output_copy)
@@ -1042,9 +1026,9 @@ class NPUARModelRunner(OmniNPUModelRunner):
             kv_connector_output=kv_connector_output,
             ec_connector_output=ec_connector_output if self.supports_mm_inputs else None,
             cudagraph_stats=cudagraph_stats,
-            routed_experts_dict=routed_experts_dict,
         )
         model_runner_output.kv_extracted_req_ids = kv_extracted_req_ids
+        model_runner_output.routed_experts = routed_experts_lists
         #  -------------------------------------- Omni-new -------------------------------------------------
 
         if self.ascend_config.profiling_chunk_config.enabled and hasattr(self, "_execution_start_time"):

vllm_omni/platforms/npu/worker/npu_generation_model_runner.py

@@ -15,11 +15,6 @@ from vllm.distributed.ec_transfer import get_ec_transfer, has_ec_transfer
 from vllm.distributed.kv_transfer import get_kv_transfer_group, has_kv_transfer_group
 from vllm.distributed.parallel_state import get_pp_group
 from vllm.logger import logger
-from vllm.model_executor.layers.fused_moe.routed_experts_capturer import (
-    extract_routed_experts_for_current_batch,
-    get_global_experts_capturer,
-    issue_routing_d2h_copy,
-)
 from vllm.sequence import IntermediateTensors
 from vllm.utils.math_utils import cdiv
 from vllm.v1.core.sched.output import GrammarOutput, SchedulerOutput
@@ -78,7 +73,7 @@ class NPUGenerationModelRunner(OmniNPUModelRunner):
         intermediate_tensors: IntermediateTensors | None = None,
     ) -> OmniModelRunnerOutput | IntermediateTensors | None:
         if self.vllm_config.model_config.enable_return_routed_experts:
-            capturer = get_global_experts_capturer()
+            capturer = self.routed_experts_capturer
             if capturer is not None and hasattr(capturer, "finalize_pending_copy"):
                 capturer.finalize_pending_copy()
         if self.ascend_config.profiling_chunk_config.enabled:
@@ -387,12 +382,7 @@ class NPUGenerationModelRunner(OmniNPUModelRunner):
             deferred_state_corrections_fn()
 
         if self.vllm_config.model_config.enable_return_routed_experts and hasattr(self, "_positions_cpu"):
-            issue_routing_d2h_copy(
-                input_batch_req_ids=self.input_batch.req_ids,
-                num_scheduled_tokens=scheduler_output.num_scheduled_tokens,
-                positions=self.positions,
-                positions_cpu=self._positions_cpu,
-            )
+            self._omni_routed_experts_d2h(scheduler_output)
 
         return None
 
@@ -479,17 +469,11 @@ class NPUGenerationModelRunner(OmniNPUModelRunner):
         # [Omni] Copy req_id mappings to avoid async scheduling mutation.
         req_ids_output_copy = self.input_batch.req_ids.copy()
         req_id_to_index_output_copy = self.input_batch.req_id_to_index.copy()
-        routed_experts_dict = None
+        routed_experts_lists = None
         if self.vllm_config.model_config.enable_return_routed_experts and hasattr(
             self.input_batch, "num_tokens_no_spec"
         ):
-            routed_experts_dict = extract_routed_experts_for_current_batch(
-                req_ids=req_ids_output_copy,
-                requests=self.requests,
-                req_id_to_index=self.input_batch.req_id_to_index,
-                num_tokens_no_spec=self.input_batch.num_tokens_no_spec,
-                max_model_len=self.max_model_len,
-            )
+            routed_experts_lists = self._omni_extract_routed_experts(scheduler_output)
         output = OmniModelRunnerOutput(
             req_ids=req_ids_output_copy,
             req_id_to_index=req_id_to_index_output_copy,
@@ -500,8 +484,8 @@ class NPUGenerationModelRunner(OmniNPUModelRunner):
             kv_connector_output=kv_connector_output,
             ec_connector_output=ec_connector_output if self.supports_mm_inputs else None,
             cudagraph_stats=cudagraph_stats,
-            routed_experts_dict=routed_experts_dict,
         )
+        output.routed_experts = routed_experts_lists
         #  -------------------------------------- Omni-new -------------------------------------------------
 
         if self.speculative_config is not None:

vllm_omni/platforms/rocm/platform.py

@@ -172,4 +172,4 @@ class RocmOmniPlatform(OmniPlatform, RocmPlatform):
         else:
             rms_norm = default
 
-        return IrOpPriorityConfig.with_default(default, rms_norm=rms_norm)
+        return IrOpPriorityConfig.with_default(default, rms_norm=rms_norm, fused_add_rms_norm=rms_norm)

vllm_omni/quantization/factory.py

@@ -10,19 +10,73 @@ different QuantizationConfig subclass in the OMNI context (e.g. GGUF).
 from __future__ import annotations
 
 import inspect
+import sys
 from collections.abc import Callable, Mapping
+from types import ModuleType
 from typing import Any
 
 from vllm.logger import init_logger
-from vllm.model_executor.layers.quantization import (
+
+
+# ---------------------------------------------------------------------------
+# Stub the ``humming`` package so that vLLM's lazy import inside
+# ``get_quantization_config()`` (which unconditionally does
+# ``from .humming import HummingConfig``) does not crash when the real
+# ``humming`` wheel is not installed.  Only populate the bare-minimum
+# names that ``humming.py`` accesses at module level.
+# ---------------------------------------------------------------------------
+def _register_humming_stubs() -> None:
+    """Register stub ``humming`` sub-modules so that the optional
+    humming quantization backend can be imported without the real wheel."""
+    if "humming" in sys.modules:
+        return  # already present (real or stub)
+
+    # --- sub-modules ---
+    submodules: dict[str, tuple[str, ...]] = {
+        "humming": (),
+        "humming.config": ("GemmType",),
+        "humming.dtypes": ("DataType",),
+        "humming.layer": ("HummingLayerMeta", "HummingMethod"),
+        "humming.schema": (
+            "BaseInputSchema",
+            "BaseWeightSchema",
+            "HummingInputSchema",
+            "HummingWeightSchema",
+        ),
+        "humming.utils": (),
+        "humming.utils.weight": ("quantize_weight",),
+    }
+
+    registry: dict[str, ModuleType] = {}
+    for name, attrs in submodules.items():
+        mod = ModuleType(name)
+        for attr in attrs:
+            setattr(mod, attr, type(attr, (), {}))
+        registry[name] = mod
+
+    # wire parent references
+    registry["humming"].config = registry["humming.config"]
+    registry["humming"].dtypes = registry["humming.dtypes"]
+    registry["humming"].layer = registry["humming.layer"]
+    registry["humming"].schema = registry["humming.schema"]
+    registry["humming"].utils = registry["humming.utils"]
+    registry["humming.utils"].weight = registry["humming.utils.weight"]
+
+    for name, mod in registry.items():
+        sys.modules[name] = mod
+
+
+_register_humming_stubs()
+
+from vllm.model_executor.layers.quantization import (  # noqa: E402
     QUANTIZATION_METHODS,
     get_quantization_config,
 )
-from vllm.model_executor.layers.quantization.base_config import (
+from vllm.model_executor.layers.quantization.base_config import (  # noqa: E402
     QuantizationConfig,
 )
 
-from .component_config import ComponentQuantizationConfig
+from .component_config import ComponentQuantizationConfig  # noqa: E402
 
 logger = init_logger(__name__)
 

vllm_omni/utils/custom_voice_io.py

@@ -0,0 +1,12 @@
+"""Shared helpers for custom voice profile tools."""
+
+from __future__ import annotations
+
+import re
+
+
+def safe_voice_stem(name: str) -> str:
+    stem = re.sub(r"[^a-zA-Z0-9_.-]+", "_", name.strip())
+    if not stem or stem in (".", ".."):
+        raise ValueError(f"Invalid voice name: {name!r}")
+    return stem[:200]

vllm_omni/utils/speaker_cache.py

@@ -6,8 +6,11 @@ has its own slot. Access via :func:`get_speaker_cache`.
 
 from __future__ import annotations
 
+import json
+import os
 import threading
 from collections import OrderedDict
+from pathlib import Path
 from typing import Any
 
 import torch
@@ -16,6 +19,8 @@ from vllm.logger import init_logger
 logger = init_logger(__name__)
 
 _MAX_BYTES = 512 * 1024**2  # 512 MiB
+_CUSTOM_VOICE_MANIFEST = "custom_voice_manifest.json"
+_CUSTOM_VOICE_SCHEMA_VERSION = 1
 
 _SINGLETON: SpeakerEmbeddingCache | None = None
 _SINGLETON_LOCK = threading.Lock()
@@ -31,6 +36,198 @@ def _estimate_tensor_bytes(obj: object) -> int:
     return 0
 
 
+def _as_voice_map(voices: object) -> dict[str, dict[str, Any]]:
+    if not isinstance(voices, dict):
+        return {}
+    result: dict[str, dict[str, Any]] = {}
+    for name, info in voices.items():
+        entry = dict(info) if isinstance(info, dict) else {}
+        entry.setdefault("name", str(name))
+        result[str(name)] = entry
+    return result
+
+
+def _load_custom_voice_manifest(custom_voice_dir: str | os.PathLike[str] | None) -> dict[str, Any] | None:
+    if not custom_voice_dir:
+        return None
+    manifest_path = Path(custom_voice_dir).expanduser() / _CUSTOM_VOICE_MANIFEST
+    if not manifest_path.exists():
+        return None
+    try:
+        return json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        logger.warning("Failed to load custom voice manifest %s: %s", manifest_path, exc)
+        return None
+
+
+def iter_custom_voice_profiles(
+    custom_voice_dir: str | os.PathLike[str] | None,
+    *,
+    expected_model_type: str | None = None,
+) -> list[dict[str, Any]]:
+    manifest = _load_custom_voice_manifest(custom_voice_dir)
+    if manifest is None:
+        return []
+
+    manifest_model_type = manifest.get("model_type")
+    if expected_model_type and manifest_model_type and manifest_model_type != expected_model_type:
+        logger.warning(
+            "Skipping custom voice manifest for model_type=%s while loading %s",
+            manifest_model_type,
+            expected_model_type,
+        )
+        return []
+
+    root = Path(custom_voice_dir).expanduser().resolve()  # type: ignore[arg-type]
+    profiles: list[dict[str, Any]] = []
+    for raw_name, raw_info in _as_voice_map(manifest.get("voices", {})).items():
+        name = str(raw_info.get("name") or raw_name).strip()
+        if not name:
+            continue
+        lower = name.lower()
+        info = dict(raw_info)
+        for stale_key in ("model_revision", "audio_codec_version", "audio_codec_config_hash"):
+            info.pop(stale_key, None)
+        filename = info.get("file") or info.get("filename") or f"{lower}.safetensors"
+        file_path = (root / str(filename)).resolve()
+        if not file_path.is_relative_to(root):
+            logger.warning("Skipping custom voice %s: file path escapes custom_voice_dir", name)
+            continue
+        info.update(
+            {
+                "name": name,
+                "voice_name_lower": lower,
+                "file": str(filename),
+                "file_path": str(file_path),
+                "custom_voice_dir": str(root),
+                "model_type": manifest_model_type or expected_model_type,
+                "schema_version": manifest.get("schema_version", _CUSTOM_VOICE_SCHEMA_VERSION),
+            }
+        )
+        profiles.append(info)
+    return profiles
+
+
+def _load_profile_tensors(profile: dict[str, Any]) -> dict[str, torch.Tensor] | None:
+    try:
+        from safetensors.torch import load_file
+    except ImportError:
+        logger.warning("safetensors is required to load custom voice profiles")
+        return None
+
+    file_path = profile.get("file_path")
+    if not file_path:
+        return None
+    try:
+        return load_file(str(file_path))
+    except Exception as exc:
+        logger.warning("Failed to load custom voice profile %s: %s", file_path, exc)
+        return None
+
+
+def _first_dim(tensor: torch.Tensor | None) -> int:
+    if tensor is None or tensor.ndim == 0:
+        return 0
+    return int(tensor.shape[0])
+
+
+def _validate_qwen3_tts_profile(
+    profile: dict[str, Any],
+    tensors: dict[str, torch.Tensor],
+    *,
+    expected_embedding_dim: int | None,
+) -> str | None:
+    # Successful validation also normalizes/augments `profile` for serving metadata.
+    speaker_embedding = tensors.get("speaker_embedding")
+    if not isinstance(speaker_embedding, torch.Tensor):
+        return "missing required tensor `speaker_embedding`"
+
+    embedding_dim = int(speaker_embedding.reshape(-1).numel())
+    if expected_embedding_dim and embedding_dim != expected_embedding_dim:
+        return f"speaker_embedding dim={embedding_dim}, expected={expected_embedding_dim}"
+
+    mode = str(profile.get("mode") or "xvec").lower()
+    if mode not in ("xvec", "icl"):
+        return f"invalid mode={mode!r}; expected 'xvec' or 'icl'"
+    profile["mode"] = mode
+    profile["embedding_dim"] = embedding_dim
+
+    if mode == "icl":
+        ref_code = tensors.get("ref_code")
+        if not isinstance(ref_code, torch.Tensor):
+            return "mode='icl' requires tensor `ref_code`"
+        ref_code_length = _first_dim(ref_code)
+        if ref_code_length <= 0:
+            return "mode='icl' requires non-empty tensor `ref_code`"
+        profile["ref_code_length"] = ref_code_length
+
+    return None
+
+
+def _validate_voxcpm2_profile(profile: dict[str, Any], tensors: dict[str, torch.Tensor]) -> str | None:
+    # Successful validation also normalizes/augments `profile` for serving metadata.
+    ref_audio_feat = tensors.get("ref_audio_feat")
+    audio_feat = tensors.get("audio_feat")
+    has_ref_audio_feat = isinstance(ref_audio_feat, torch.Tensor)
+    has_audio_feat = isinstance(audio_feat, torch.Tensor)
+    if not has_ref_audio_feat and not has_audio_feat:
+        return "missing required tensor `ref_audio_feat` or `audio_feat`"
+
+    mode = str(profile.get("mode") or "").lower()
+    if not mode:
+        if has_ref_audio_feat and has_audio_feat:
+            mode = "ref_continuation"
+        elif has_ref_audio_feat:
+            mode = "reference"
+        else:
+            mode = "continuation"
+    if mode not in ("reference", "continuation", "ref_continuation"):
+        return f"invalid mode={mode!r}; expected 'reference', 'continuation', or 'ref_continuation'"
+    if mode in ("reference", "ref_continuation") and not has_ref_audio_feat:
+        return f"mode={mode!r} requires tensor `ref_audio_feat`"
+    if mode in ("continuation", "ref_continuation") and not has_audio_feat:
+        return f"mode={mode!r} requires tensor `audio_feat`"
+
+    ref_audio_feat_len = _first_dim(ref_audio_feat)
+    audio_feat_len = _first_dim(audio_feat)
+    if mode in ("reference", "ref_continuation") and ref_audio_feat_len <= 0:
+        return f"mode={mode!r} requires non-empty tensor `ref_audio_feat`"
+    if mode in ("continuation", "ref_continuation") and audio_feat_len <= 0:
+        return f"mode={mode!r} requires non-empty tensor `audio_feat`"
+
+    profile["mode"] = mode
+    if has_ref_audio_feat:
+        profile["ref_audio_feat_len"] = ref_audio_feat_len
+    if has_audio_feat:
+        profile["audio_feat_len"] = audio_feat_len
+    return None
+
+
+def load_validated_profile_tensors(
+    profile: dict[str, Any],
+    *,
+    expected_model_type: str,
+    qwen3_embedding_dim: int | None = None,
+) -> dict[str, torch.Tensor] | None:
+    tensors = _load_profile_tensors(profile)
+    error = None
+    if tensors is None:
+        error = "safetensors file could not be loaded"
+    elif str(profile.get("model_type") or expected_model_type or "") != expected_model_type:
+        error = f"model_type={profile.get('model_type')!r}, expected={expected_model_type!r}"
+    elif expected_model_type == "qwen3_tts":
+        error = _validate_qwen3_tts_profile(profile, tensors, expected_embedding_dim=qwen3_embedding_dim)
+    elif expected_model_type == "voxcpm2":
+        error = _validate_voxcpm2_profile(profile, tensors)
+    else:
+        error = f"unsupported custom voice model_type={expected_model_type!r}"
+
+    if error is not None:
+        logger.warning("Skipping custom %s voice %s: %s", expected_model_type, profile.get("name"), error)
+        return None
+    return tensors
+
+
 class SpeakerEmbeddingCache:
     """Thread-safe in-memory LRU cache for speaker extraction artifacts."""
 

vllm_omni/worker/gpu_ar_model_runner.py

@@ -19,11 +19,6 @@ from vllm.distributed.kv_transfer import get_kv_transfer_group, has_kv_transfer_
 from vllm.distributed.parallel_state import get_pp_group, get_tp_group
 from vllm.forward_context import set_forward_context
 from vllm.logger import init_logger
-from vllm.model_executor.layers.fused_moe.routed_experts_capturer import (
-    extract_routed_experts_for_current_batch,
-    get_global_experts_capturer,
-    issue_routing_d2h_copy,
-)
 from vllm.v1.core.sched.output import GrammarOutput, SchedulerOutput
 from vllm.v1.outputs import AsyncModelRunnerOutput, make_empty_encoder_model_runner_output
 from vllm.v1.spec_decode.dflash import DFlashProposer
@@ -363,6 +358,9 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
         if self.execute_model_state is not None:
             raise RuntimeError("State error: sample_tokens() must be called after execute_model() returns None.")
 
+        if self.routed_experts_initialized:
+            self.routed_experts_capturer.clear_buffer()
+
         if not getattr(self, "_warmup_state_cleared", False):
             self._warmup_state_cleared = True
             if hasattr(self.model, "_clear_warmup_state"):
@@ -414,7 +412,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
             )
 
         if self.routed_experts_initialized:
-            capturer = get_global_experts_capturer()
+            capturer = self.routed_experts_capturer
             if capturer is not None and hasattr(capturer, "finalize_pending_copy"):
                 capturer.finalize_pending_copy()
 
@@ -679,7 +677,6 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
                 if not get_pp_group().is_last_rank:
                     # Return the intermediate tensors.
                     assert isinstance(hidden_states, IntermediateTensors)
-                    hidden_states.kv_connector_output = kv_connector_output
                     self.kv_connector_output = kv_connector_output
                     return hidden_states
 
@@ -692,7 +689,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
                         kv_connector_output,
                     )
 
-                sample_hidden_states = hidden_states[logits_indices]
+                sample_hidden_states = hidden_states[logits_indices.to(hidden_states.device)]
                 # Try with sampling_metadata first; fall back to without for models that don't support it
                 try:
                     logits = self.model.compute_logits(
@@ -704,7 +701,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
                 # Rare case.
                 assert not self.is_pooling_model
 
-                sample_hidden_states = hidden_states[logits_indices]
+                sample_hidden_states = hidden_states[logits_indices.to(hidden_states.device)]
                 if not get_pp_group().is_last_rank:
                     all_gather_tensors = {
                         "residual": not is_residual_scattered_for_sp(self.vllm_config, num_tokens_padded)
@@ -753,12 +750,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
             deferred_state_corrections_fn()
 
         if self.routed_experts_initialized and hasattr(self, "_positions_cpu"):
-            issue_routing_d2h_copy(
-                input_batch_req_ids=self.input_batch.req_ids,
-                num_scheduled_tokens=scheduler_output.num_scheduled_tokens,
-                positions=self.positions,
-                positions_cpu=self._positions_cpu,
-            )
+            self._omni_routed_experts_d2h(scheduler_output)
 
         return None
 
@@ -830,18 +822,14 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
         if self.execute_model_state is None:
             kv_connector_output = self.kv_connector_output
             self.kv_connector_output = None
-            # Nothing to do (PP non-final rank case), output isn't used.
-            if not kv_connector_output:
-                return None  # type: ignore[return-value]
-
+            # receive sampled token ids from the last PP rank.
+            if self.use_async_scheduling and not get_pp_group().is_last_rank:
+                self._pp_receive_prev_sampled_token_ids_to_input_batch()
             # In case of PP with kv transfer, we need to pass through the
             # kv_connector_output
-            if kv_connector_output.is_empty():
-                return self.attach_omni_connector_output(EMPTY_MODEL_RUNNER_OUTPUT)
-
-            output = copy(EMPTY_MODEL_RUNNER_OUTPUT)
-            output.kv_connector_output = kv_connector_output
-            return self.attach_omni_connector_output(output)
+            return self.attach_omni_connector_output(
+                OmniModelRunnerOutput.with_kv_conn_output_only(kv_connector_output)
+            )
 
         # Unpack ephemeral state.
         (
@@ -1108,15 +1096,9 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
                     self.accumulate_full_payload_output(rid, pooler_output[i], req_state)
 
         with record_function_or_nullcontext("gpu_model_runner: ModelRunnerOutput"):
-            routed_experts_dict = None
+            routed_experts_lists = None
             if self.routed_experts_initialized:
-                routed_experts_dict = extract_routed_experts_for_current_batch(
-                    req_ids=req_ids_output_copy,
-                    requests=self.requests,
-                    req_id_to_index=self.input_batch.req_id_to_index,
-                    num_tokens_no_spec=self.input_batch.num_tokens_no_spec,
-                    max_model_len=self.max_model_len,
-                )
+                routed_experts_lists = self._omni_extract_routed_experts(scheduler_output)
             output = OmniModelRunnerOutput(
                 req_ids=req_ids_output_copy,
                 req_id_to_index=req_id_to_index_output_copy,
@@ -1128,10 +1110,10 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin):
                 ec_connector_output=ec_connector_output if self.supports_mm_inputs else None,
                 num_nans_in_logits=num_nans_in_logits,
                 cudagraph_stats=cudagraph_stats,
-                routed_experts_dict=routed_experts_dict,
             )
             output.kv_extracted_req_ids = kv_extracted_req_ids
             output.omni_connector_output = self.get_omni_connector_output()
+            output.routed_experts = routed_experts_lists
 
         if not self.use_async_scheduling:
             return output

vllm_omni/worker/gpu_generation_model_runner.py

@@ -8,7 +8,6 @@ from __future__ import annotations
 
 import gc
 import logging
-from copy import copy
 from dataclasses import replace
 
 import numpy as np
@@ -18,11 +17,6 @@ from vllm.distributed.ec_transfer import get_ec_transfer, has_ec_transfer
 from vllm.distributed.kv_transfer import get_kv_transfer_group, has_kv_transfer_group
 from vllm.distributed.parallel_state import get_pp_group
 from vllm.forward_context import set_forward_context
-from vllm.model_executor.layers.fused_moe.routed_experts_capturer import (
-    extract_routed_experts_for_current_batch,
-    get_global_experts_capturer,
-    issue_routing_d2h_copy,
-)
 from vllm.utils.math_utils import cdiv
 from vllm.v1.core.sched.output import GrammarOutput, SchedulerOutput
 from vllm.v1.outputs import AsyncModelRunnerOutput, make_empty_encoder_model_runner_output
@@ -112,6 +106,9 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
         if self.execute_model_state is not None:
             raise RuntimeError("State error: sample_tokens() must be called after execute_model() returns None.")
 
+        if self.routed_experts_initialized:
+            self.routed_experts_capturer.clear_buffer()
+
         if hasattr(self, "_omni_connector"):
             for request in getattr(scheduler_output, "pending_input_registrations", []):
                 self.register_chunk_recv(request)
@@ -123,7 +120,7 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
                     self.flush_full_payload_outputs(flush_ids)
 
         if self.routed_experts_initialized:
-            capturer = get_global_experts_capturer()
+            capturer = self.routed_experts_capturer
             if capturer is not None and hasattr(capturer, "finalize_pending_copy"):
                 capturer.finalize_pending_copy()
 
@@ -369,12 +366,7 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
             deferred_state_corrections_fn()
 
         if self.routed_experts_initialized and hasattr(self, "_positions_cpu"):
-            issue_routing_d2h_copy(
-                input_batch_req_ids=self.input_batch.req_ids,
-                num_scheduled_tokens=scheduler_output.num_scheduled_tokens,
-                positions=self.positions,
-                positions_cpu=self._positions_cpu,
-            )
+            self._omni_routed_experts_d2h(scheduler_output)
 
         return None
 
@@ -390,18 +382,14 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
         self.kv_connector_output = None
 
         if self.execute_model_state is None:
-            # Nothing to do (PP non-final rank case), output isn't used.
-            if not kv_connector_output:
-                return None  # type: ignore[return-value]
-
+            # receive sampled token ids from the last PP rank.
+            if self.use_async_scheduling and not get_pp_group().is_last_rank:
+                self._pp_receive_prev_sampled_token_ids_to_input_batch()
             # In case of PP with kv transfer, we need to pass through the
             # kv_connector_output
-            if kv_connector_output.is_empty():
-                return self.attach_omni_connector_output(EMPTY_MODEL_RUNNER_OUTPUT)
-
-            output = copy(EMPTY_MODEL_RUNNER_OUTPUT)
-            output.kv_connector_output = kv_connector_output
-            return self.attach_omni_connector_output(output)
+            return self.attach_omni_connector_output(
+                OmniModelRunnerOutput.with_kv_conn_output_only(kv_connector_output)
+            )
 
         # Unpack ephemeral state.
         (
@@ -462,15 +450,9 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
         # [Omni] Copy req_id mappings to avoid async scheduling mutation.
         req_ids_output_copy = self.input_batch.req_ids.copy()
         req_id_to_index_output_copy = self.input_batch.req_id_to_index.copy()
-        routed_experts_dict = None
+        routed_experts_lists = None
         if self.routed_experts_initialized:
-            routed_experts_dict = extract_routed_experts_for_current_batch(
-                req_ids=req_ids_output_copy,
-                requests=self.requests,
-                req_id_to_index=self.input_batch.req_id_to_index,
-                num_tokens_no_spec=self.input_batch.num_tokens_no_spec,
-                max_model_len=self.max_model_len,
-            )
+            routed_experts_lists = self._omni_extract_routed_experts(scheduler_output)
         if self._should_accumulate_full_payload_output():
             for i, rid in enumerate(req_ids_output_copy):
                 req_state = self.requests.get(rid)
@@ -488,9 +470,9 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
             num_nans_in_logits={},
             cudagraph_stats=cudagraph_stats,
             ec_connector_output=ec_connector_output if self.supports_mm_inputs else None,
-            routed_experts_dict=routed_experts_dict,
         )
         output.omni_connector_output = self.get_omni_connector_output()
+        output.routed_experts = routed_experts_lists
 
         if not self.use_async_scheduling:
             return output
@@ -544,9 +526,9 @@ class GPUGenerationModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin
         )
 
     @torch.inference_mode()
-    def _dummy_sampler_run(self, hidden_states: torch.Tensor) -> None:
+    def _dummy_sampler_run(self, hidden_states: torch.Tensor) -> torch.Tensor:
         logger.warning("Dummy sampler run is not implemented for generation model")
-        return None
+        return torch.tensor([])
 
     @torch.inference_mode()
     def _dummy_run(

vllm_omni/worker/gpu_model_runner.py

@@ -36,6 +36,7 @@ from vllm_omni.platforms import current_omni_platform
 
 if TYPE_CHECKING:
     from vllm.v1.core.sched.output import SchedulerOutput
+    from vllm.v1.outputs import RoutedExpertsLists
 else:
     xgr = LazyLoader("xgr", globals(), "xgrammar")
     xgr_torch_compile = LazyLoader(
@@ -57,6 +58,52 @@ class OmniGPUModelRunner(GPUModelRunner):
         # when we initialize the metadata builders if enabled
         self.omni_prefix_cache = None
 
+    def _omni_routed_experts_d2h(self, scheduler_output) -> None:
+        """Issue routed-experts D2H copy matching upstream GPUModelRunner pattern.
+
+        Upstream does this inline in ``execute_model``:
+            buf = self.routed_experts_capturer.get_device_buffer()
+            total = scheduler_output.total_num_scheduled_tokens
+            self.routed_experts_cpu[:total].copy_(buf[:total], non_blocking=True)
+            self.routed_experts_slot_mapping_cpu[:total].copy_(
+                self.routed_experts_slot_mapping_device[:total], non_blocking=True)
+        """
+        if not self.routed_experts_initialized:
+            return
+        buf = self.routed_experts_capturer.get_device_buffer()
+        total = scheduler_output.total_num_scheduled_tokens
+        self.routed_experts_cpu[:total].copy_(buf[:total], non_blocking=True)
+        if hasattr(self, "routed_experts_slot_mapping_device"):
+            self.routed_experts_slot_mapping_cpu[:total].copy_(
+                self.routed_experts_slot_mapping_device[:total],
+                non_blocking=True,
+            )
+
+    def _omni_extract_routed_experts(self, scheduler_output) -> "RoutedExpertsLists | None":
+        """Extract routed experts matching upstream GPUModelRunner pattern.
+
+        Upstream (sync path, sample_tokens):
+            total = scheduler_output.total_num_scheduled_tokens
+            output.routed_experts = RoutedExpertsLists(
+                routing_data=self.routed_experts_cpu[:total].numpy(),
+                slot_mapping=self.routed_experts_slot_mapping_cpu[:total].numpy(),
+            )
+
+        Returns RoutedExpertsLists (batch-level, with slot_mapping) so that
+        downstream schedulers can use slot_mapping to map back to requests.
+        """
+        from vllm.v1.outputs import RoutedExpertsLists
+
+        if not self.routed_experts_initialized:
+            return None
+        total = scheduler_output.total_num_scheduled_tokens
+        if total <= 0:
+            return None
+        return RoutedExpertsLists(
+            routing_data=self.routed_experts_cpu[:total].numpy(),
+            slot_mapping=self.routed_experts_slot_mapping_cpu[:total].numpy(),
+        )
+
     def initialize_metadata_builders(self, kv_cache_config, kernel_block_sizes):
         """Override to fix scheduler_metadata buffer size for FA3 + CUDA graph.
 
@@ -747,6 +794,15 @@ class OmniGPUModelRunner(GPUModelRunner):
             raise ValueError(f"Invalid hidden states type: {type(hidden_states)}")
         return text_hidden_states, multimodal_outputs
 
+    def _dummy_sampler_run(self, hidden_states: torch.Tensor) -> torch.Tensor:
+        # Models loaded with load_format=dummy (e.g. MossTTSNano) may
+        # produce CPU hidden_states while the upstream sampler warmup
+        # expects GPU tensors.  Skip the warmup in that case — a
+        # meaningful warmup requires real weights anyway.
+        if hidden_states.device != self.device:
+            return torch.tensor([])
+        return super()._dummy_sampler_run(hidden_states=hidden_states)
+
     @torch.inference_mode()
     def _dummy_run(
         self,
@@ -1100,7 +1156,7 @@ class OmniGPUModelRunner(GPUModelRunner):
             self.eplb_step(is_dummy=True, is_profile=is_profile)
 
         logit_indices = np.cumsum(num_scheduled_tokens) - 1
-        logit_indices_device = torch.from_numpy(logit_indices).to(self.device, non_blocking=True)
+        logit_indices_device = torch.from_numpy(logit_indices).to(hidden_states.device, non_blocking=True)
         return hidden_states, hidden_states[logit_indices_device]
 
     # ------------------------------------------------------------------