From 9c260b37743e86d69fabc82882121df28eef8234 Mon Sep 17 00:00:00 2001 From: Henry J <349043+henryj@users.noreply.github.com> Date: Tue, 23 Jun 2026 10:21:11 +1000 Subject: [PATCH] Qwen3-TTS full-payload: emit empty-finished payload on degenerate take instead of dropping (fixes Stage-1 300s hang) (#4465) Signed-off-by: jjw 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) --- .../qwen3_tts/test_qwen3_tts_code2wav.py | 16 ++++++++++ .../test_qwen3_tts_async_chunk.py | 29 +++++++++++++++++++ .../stage_input_processors/qwen3_tts.py | 17 +++++++++-- 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/tests/model_executor/models/qwen3_tts/test_qwen3_tts_code2wav.py b/tests/model_executor/models/qwen3_tts/test_qwen3_tts_code2wav.py index 6547fb29d..4c94eb520 100644 --- a/tests/model_executor/models/qwen3_tts/test_qwen3_tts_code2wav.py +++ b/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() diff --git a/tests/model_executor/stage_input_processors/test_qwen3_tts_async_chunk.py b/tests/model_executor/stage_input_processors/test_qwen3_tts_async_chunk.py index da25ee7bd..82f7da711 100644 --- a/tests/model_executor/stage_input_processors/test_qwen3_tts_async_chunk.py +++ b/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 diff --git a/vllm_omni/model_executor/stage_input_processors/qwen3_tts.py b/vllm_omni/model_executor/stage_input_processors/qwen3_tts.py index c67e4a1af..6cebc4c28 100644 --- a/vllm_omni/model_executor/stage_input_processors/qwen3_tts.py +++ b/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)