[Edge][WP5] Keep DELTA outputs for talker-only (codec-token) deployments

The speech path forced FINAL_ONLY for Qwen3-TTS deployments without async
chunks, so a talker-only server emitted one engine output per request and all
codec frames arrived at once (measured: 1 engine output per request, first
codes at output #1). A talker-only stage (final_output + latent) now keeps
DELTA outputs so codec frames stream per step.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tests/entrypoints/openai_api/test_codec_stream_protocol.py

@@ -181,3 +181,21 @@ def test_input_cancel_mid_stream_aborts_request(mocker):
                 assert payload["cancelled"] is True
         assert "codec.done" in types and types[-1] == "session.done"
     svc.engine_client.abort.assert_awaited()
+
+
+@pytest.mark.core_model
+@pytest.mark.cpu
+def test_talker_only_deployment_detection():
+    from types import SimpleNamespace
+
+    from vllm_omni.entrypoints.openai.serving_speech import OmniOpenAIServingSpeech
+
+    probe = OmniOpenAIServingSpeech.__new__(OmniOpenAIServingSpeech)
+    probe._tts_stage = SimpleNamespace(final_output=True, final_output_type="latent")
+    assert probe._is_talker_only_deployment() is True
+    probe._tts_stage = SimpleNamespace(final_output=False, final_output_type="latent")
+    assert probe._is_talker_only_deployment() is False
+    probe._tts_stage = SimpleNamespace(final_output=True, final_output_type="audio")
+    assert probe._is_talker_only_deployment() is False
+    probe._tts_stage = None
+    assert probe._is_talker_only_deployment() is False

vllm_omni/entrypoints/openai/serving_speech.py

@@ -586,6 +586,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
                 return stage
         return None
 
+    def _is_talker_only_deployment(self) -> bool:
+        """True when the TTS stage is the final stage and emits latents (codec tokens) per step."""
+        stage = self._tts_stage
+        if stage is None:
+            return False
+        return bool(getattr(stage, "final_output", False)) and (
+            str(getattr(stage, "final_output_type", "") or "").lower() == "latent"
+        )
+
     def _detect_tts_model_type(self) -> str | None:
         """Detect TTS model type from the resolved stage's deployment metadata.
 
@@ -3096,7 +3105,13 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
         # list() makes a copy to avoid mutating the params.
         sampling_params_list = list(self.engine_client.default_sampling_params_list)
         async_chunk = getattr(self.model_config, "async_chunk", True)
-        qwen3_full_payload = self._tts_model_type == "qwen3_tts" and not bool(async_chunk)
+        # A talker-only (codec-token) deployment has no code2wav stage and no
+        # async chunks either, but its per-step ``codes.audio`` payloads are the
+        # stream itself, so it must keep DELTA outputs (otherwise the engine
+        # emits one FINAL_ONLY output and every codec frame arrives at once).
+        qwen3_full_payload = (
+            self._tts_model_type == "qwen3_tts" and not bool(async_chunk) and not self._is_talker_only_deployment()
+        )
         is_streaming_request = request.is_streaming() and not qwen3_full_payload
         sampling_params_list = coerce_param_message_types(sampling_params_list, is_streaming_request)