fix(serving): scope no-async-chunk FINAL_ONLY streaming to qwen3_tts

async_chunk defaults to False in OmniModelConfig, so coercing every
async_chunk=False deploy to FINAL_ONLY for streaming requests would
silently disable incremental delivery for non-Qwen3 models that stream
DELTA-style today. Gate the coercion on _tts_model_type == "qwen3_tts"
(the model this PR validated) and pin the non-qwen3 behavior with a test.

Signed-off-by: Yueqian Lin <linyueqian@outlook.com>

tests/entrypoints/openai_api/test_serving_speech.py

@@ -3414,6 +3414,27 @@ class TestTTSAsyncOffloading:
 
         mock_coerce.assert_called_once_with(qwen3_tts_server.engine_client.default_sampling_params_list, False)
 
+    def test_prepare_speech_generation_no_async_chunk_stream_keeps_delta_for_non_qwen3(
+        self, voxtral_server, mocker: MockerFixture
+    ):
+        """FINAL_ONLY streaming for async_chunk=False is scoped to qwen3_tts only."""
+        voxtral_server.engine_client.model_config.async_chunk = False
+        voxtral_server._build_voxtral_prompt_async = mocker.AsyncMock(
+            return_value={
+                "prompt_token_ids": [1, 2, 3],
+                "additional_information": {"voice": ["test"]},
+            }
+        )
+        mock_coerce = mocker.patch(
+            "vllm_omni.entrypoints.openai.serving_speech.coerce_param_message_types",
+            return_value=voxtral_server.engine_client.default_sampling_params_list,
+        )
+        request = OpenAICreateSpeechRequest(input="hello", voice="test", stream_format="audio", response_format="pcm")
+
+        asyncio.run(voxtral_server._prepare_speech_generation(request))
+
+        mock_coerce.assert_called_once_with(voxtral_server.engine_client.default_sampling_params_list, True)
+
     def test_prepare_speech_generation_qwen3_voicedesign_non_streaming_mode_false(
         self, qwen3_tts_server, mocker: MockerFixture
     ):

vllm_omni/entrypoints/openai/serving_speech.py

@@ -3497,14 +3497,16 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
 
         # If this is a streaming request with real async chunks, we need to
         # coerce cumulative outputs to delta outputs; this ensures we don't
-        # emit redundant MM data & drain after emitting. Full-payload
-        # (async_chunk=False) TTS has no incremental audio chunks, so keep
+        # emit redundant MM data & drain after emitting. Qwen3-TTS full-payload
+        # (async_chunk=False) has no incremental audio chunks, so keep
         # FINAL_ONLY semantics and let the streaming response send the final
-        # waveform once.
+        # waveform once. Scoped to qwen3_tts: other async_chunk=False models
+        # keep the DELTA coercion they stream with today.
         # list() makes a copy to avoid mutating the params.
         sampling_params_list = list(self.engine_client.default_sampling_params_list)
         async_chunk = getattr(self.engine_client.model_config, "async_chunk", True)
-        is_streaming_request = request.is_streaming() and bool(async_chunk)
+        qwen3_full_payload = self._tts_model_type == "qwen3_tts" and not bool(async_chunk)
+        is_streaming_request = request.is_streaming() and not qwen3_full_payload
         sampling_params_list = coerce_param_message_types(sampling_params_list, is_streaming_request)
 
         # Build prompt + tts_params via the per-model adapter (RFC #4327). Every