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 026103d1b..c41541cc8 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 @@ -16,6 +16,8 @@ from vllm_omni.model_executor.stage_input_processors.chunk_size_utils import ( ramp_cumulative, ) from vllm_omni.model_executor.stage_input_processors.qwen3_tts import ( + _NUM_QUANTIZERS_DEFAULT, + _filter_audio_codes_qwen3_tts, talker2code2wav_async_chunk, talker2code2wav_full_payload, talker2code2wav_token_only, @@ -534,17 +536,23 @@ def test_full_payload_omits_left_context_size_without_ref(): 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. +def test_full_payload_emits_placeholder_frame_on_degenerate_take(pooling_output): + """Regression for #4463 and #5471 (the producer half of #5196). - A degenerate talker take used to return ``None`` from - ``talker2code2wav_full_payload``. The connector treats ``None`` as "drop the + A degenerate talker take must not 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. + polls to ``connector_get_max_wait`` (~300s) and one stuck request stalls the + whole two-stage pipeline (#4463). It must not return an *empty* finished + payload either: zero codec frames produce a zero-token Stage-1 request, + which full-payload scheduling placeholder-schedules once and never + collects; the base-scheduler fallback then schedules it at a negative span, + which killed the stage EngineCore before #5269 and leaves the request + parked in ``running`` forever after it (#5196, #5471). Each degenerate case + (non-dict pooling_output, missing ``codes.audio``, all codec frames dropped + by the filter) must instead return a finished payload with at least one + frame that survives the codec validity filter, so the request runs the + normal one-shot path and finishes cleanly. """ request = SimpleNamespace(request_id="r", output_token_ids=[0, 1, 2]) @@ -552,7 +560,14 @@ def test_full_payload_emits_empty_finished_payload_on_degenerate_take(pooling_ou assert payload is not None assert payload["meta"]["finished"].item() is True - assert payload["codes"]["audio"].numel() == 0 + audio = payload["codes"]["audio"] + # Same wire format as the normal path: flat, codebook-major, one frame. + assert audio.ndim == 1 + assert audio.numel() == _NUM_QUANTIZERS_DEFAULT + # The placeholder must survive the same validity filter real takes go + # through; a frame the filter would drop re-creates the zero-token request. + frames = audio.reshape(-1, _NUM_QUANTIZERS_DEFAULT) + assert int(_filter_audio_codes_qwen3_tts(frames).shape[0]) >= 1 _RAMP = [1, 4, 8, 16, 25] 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 0d1b5ad74..9b5e5ea38 100644 --- a/vllm_omni/model_executor/stage_input_processors/qwen3_tts.py +++ b/vllm_omni/model_executor/stage_input_processors/qwen3_tts.py @@ -29,13 +29,29 @@ 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.""" +def _qwen3_tts_degenerate_finished_payload(): + """Single-placeholder-frame finished payload for a degenerate talker take. + + Returning ``None`` here makes the connector silently drop the request, and + Stage-1's wait gate then polls to ``connector_get_max_wait`` (#4463). + Returning an *empty* finished payload (zero codec frames) is no better on + full-payload deploys: it produces a zero-token Stage-1 request, which the + generation scheduler placeholder-schedules once and then never collects + (the ``required_tokens <= 0`` branch only finishes requests through the + async-chunk transfer adapter). The request is left parked in ``running`` + while the base-scheduler fallback schedules it at ``num_new_tokens = -1``, + which killed the whole stage EngineCore before #5269 and no-ops after it, + so the request never finishes either way (#5196, #5471). + + The placeholder is one all-ones frame, emitted flat (codebook-major, the + same wire format the normal path below produces). Its values are valid by + ``_filter_audio_codes_qwen3_tts`` (non-negative, not all-zero, below + ``_CODEBOOK_SIZE``), so the request runs the normal one-shot code2wav + path and finishes cleanly with a single frame (~80 ms at 12 Hz) of + placeholder audio. + """ return { - "codes": {"audio": torch.zeros(0, dtype=torch.long)}, + "codes": {"audio": torch.ones(_NUM_QUANTIZERS_DEFAULT, dtype=torch.long)}, "meta": {"finished": torch.tensor(True, dtype=torch.bool)}, } @@ -400,7 +416,7 @@ def talker2code2wav_full_payload( type(pooling_output).__name__, rid, ) - return _qwen3_tts_empty_finished_payload() + return _qwen3_tts_degenerate_finished_payload() # codes.audio — try flat dotted first (flatten_payload), then nested fallback. audio = pooling_output.get("codes.audio") @@ -415,7 +431,7 @@ def talker2code2wav_full_payload( list(pooling_output.keys()), rid, ) - return _qwen3_tts_empty_finished_payload() + return _qwen3_tts_degenerate_finished_payload() audio = audio.to(torch.long) audio = _filter_audio_codes_qwen3_tts(audio) if audio.numel() == 0: @@ -424,7 +440,7 @@ def talker2code2wav_full_payload( "filter (negative/all-zero/out-of-range rows dropped) for req=%s.", rid, ) - return _qwen3_tts_empty_finished_payload() + return _qwen3_tts_degenerate_finished_payload() output_token_ids = list(getattr(request, "output_token_ids", None) or []) seq_len = max(len(output_token_ids) - 1, 0)