Qwen3-TTS full-payload: emit empty-finished payload on degenerate take instead of dropping (fixes Stage-1 300s hang) (#4465)

Signed-off-by: jjw <m@jjw.in>
Signed-off-by: henryj <349043+henryj@users.noreply.github.com>
Co-authored-by: Yueqian Lin <70319226+linyueqian@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/model_executor/models/qwen3_tts/test_qwen3_tts_code2wav.py

@@ -375,6 +375,22 @@ def test_malformed_codec_length_warning_is_rate_limited():
     assert out2.multimodal_outputs["model_outputs"][0].numel() == 0
 
 
+def test_forward_emits_zero_samples_for_empty_codec_payload():
+    """Consumer side of #4463.
+
+    The empty-but-finished payload that ``talker2code2wav_full_payload`` now
+    returns on a degenerate take carries a zero-length ``codes.audio``. Code2Wav
+    must turn that into a 0-sample output (finishing the request) so the Stage-1
+    wait gate releases, rather than stalling the pipeline to the connector
+    timeout.
+    """
+    model = _make_model()
+
+    out = model.forward(input_ids=torch.zeros(0, dtype=torch.long))
+
+    assert out.multimodal_outputs["model_outputs"][0].numel() == 0
+
+
 def test_forward_batches_equal_length_requests_in_one_decoder_call():
     model = _make_model()
 

tests/model_executor/stage_input_processors/test_qwen3_tts_async_chunk.py

@@ -486,3 +486,32 @@ def test_full_payload_omits_left_context_size_without_ref():
     assert payload["meta"]["finished"].item() is True
     assert "left_context_size" not in payload["meta"]
     assert len(payload["codes"]["audio"]) == _Q * 2
+
+
+@pytest.mark.parametrize(
+    "pooling_output",
+    [
+        pytest.param(SimpleNamespace(codes="not-a-dict"), id="non_dict_output"),
+        pytest.param({}, id="missing_codes_audio"),
+        pytest.param({"codes.audio": torch.zeros((3, _Q), dtype=torch.long)}, id="all_codes_filtered"),
+    ],
+)
+def test_full_payload_emits_empty_finished_payload_on_degenerate_take(pooling_output):
+    """Regression for #4463.
+
+    A degenerate talker take used to return ``None`` from
+    ``talker2code2wav_full_payload``. The connector treats ``None`` as "drop the
+    request", but Stage-1 was already scheduled to receive it, so its wait gate
+    polls to ``connector_get_max_wait`` (~300s) and the orchestrator aborts — one
+    stuck request stalls the whole two-stage pipeline. Each degenerate case
+    (non-dict pooling_output, missing ``codes.audio``, all codec frames dropped by
+    the filter) must instead return an empty-but-finished payload so the gate
+    releases and the request finishes immediately with zero-length audio.
+    """
+    request = SimpleNamespace(request_id="r", output_token_ids=[0, 1, 2])
+
+    payload = talker2code2wav_full_payload(transfer_manager=None, pooling_output=pooling_output, request=request)
+
+    assert payload is not None
+    assert payload["meta"]["finished"].item() is True
+    assert payload["codes"]["audio"].numel() == 0

vllm_omni/model_executor/stage_input_processors/qwen3_tts.py

@@ -27,6 +27,17 @@ from vllm_omni.model_executor.stage_input_processors.tts_utils import (
 logger = init_logger(__name__)
 
 
+def _qwen3_tts_empty_finished_payload():
+    """Empty-but-finished stage payload. Returned instead of None on a degenerate
+    talker take so the Stage-1 wait gate releases (code2wav handles empty codec
+    input -> 0-sample audio, request finished) rather than the connector silently
+    dropping the request and Stage-1 hanging to connector_get_max_wait."""
+    return {
+        "codes": {"audio": torch.zeros(0, dtype=torch.long)},
+        "meta": {"finished": torch.tensor(True, dtype=torch.bool)},
+    }
+
+
 def talker2code2wav(
     source_outputs: list[Any],
     prompt: Any = None,
@@ -474,7 +485,7 @@ def talker2code2wav_full_payload(
             type(pooling_output).__name__,
             rid,
         )
-        return None
+        return _qwen3_tts_empty_finished_payload()
 
     # codes.audio — try flat dotted first (flatten_payload), then nested fallback.
     audio = pooling_output.get("codes.audio")
@@ -489,7 +500,7 @@ def talker2code2wav_full_payload(
             list(pooling_output.keys()),
             rid,
         )
-        return None
+        return _qwen3_tts_empty_finished_payload()
     audio = audio.to(torch.long)
     audio = _filter_audio_codes_qwen3_tts(audio)
     if audio.numel() == 0:
@@ -498,7 +509,7 @@ def talker2code2wav_full_payload(
             "filter (negative/all-zero/out-of-range rows dropped) for req=%s.",
             rid,
         )
-        return None
+        return _qwen3_tts_empty_finished_payload()
 
     output_token_ids = list(getattr(request, "output_token_ids", None) or [])
     seq_len = max(len(output_token_ids) - 1, 0)