[Bugfix] Preserve all video-stream audio deltas (#7532)

Signed-off-by: Deep Shah <deep@socratic.co>

docs/serving/video_stream_api.md

@@ -101,7 +101,7 @@ The server accepts these legacy field names and rewrites them before validation.
 | Variable | Values | Default | Description |
 | -------- | ------ | ------- | ----------- |
 | `VLLM_VIDEO_ASYNC_CHUNK` | `on`, `off` | `on` | Wire-level streaming switch. `off` buffers server-side deltas and emits coalesced outputs at the end of a query. |
-| `VLLM_VIDEO_AUDIO_DELTA_MODE` | `fast`, `slow` | `fast` | Audio delta extraction strategy. `fast` emits only newly produced chunks; `slow` recomputes from accumulated audio and exists for A/B verification. |
+| `VLLM_VIDEO_AUDIO_DELTA_MODE` | `fast`, `slow` | `fast` | Both settings forward every fresh engine audio delta. `slow` is retained as a compatibility alias for `fast`. |
 
 ## EVS Semantics
 

tests/entrypoints/openai_api/test_video_audio_delta.py

@@ -0,0 +1,242 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Preserve every sample from the engine's fresh DELTA audio payloads."""
+
+import asyncio
+import base64
+import io
+import wave
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import torch
+from vllm.sampling_params import RequestOutputKind, SamplingParams
+from vllm.v1.engine import FinishReason
+
+from tests.helpers.serving_chat import build_serving_chat
+from vllm_omni.entrypoints.omni_base import OmniBase
+from vllm_omni.entrypoints.openai.serving_video_stream import QwenOmniStreamingVideoHandler
+from vllm_omni.entrypoints.openai.video_stream_base import (
+    _CODEC_FRAME_SAMPLES,
+    OmniStreamingVideoHandler,
+    StreamingVideoSessionConfig,
+)
+from vllm_omni.outputs import OmniRequestOutput
+from vllm_omni.outputs.mm_outputs import MultimodalCompletionOutput, MultimodalPayload
+from vllm_omni.outputs.output_processor import OmniRequestState
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _audio_result(payload):
+    completion = MultimodalCompletionOutput(
+        index=0,
+        text="",
+        token_ids=[],
+        cumulative_logprob=None,
+        logprobs=None,
+        multimodal_output=MultimodalPayload.from_dict({"audio": payload}),
+    )
+    return OmniRequestOutput(
+        outputs=[completion],
+        final_output_type="audio",
+    )
+
+
+def _pcm(encoded):
+    assert encoded is not None
+    with wave.open(io.BytesIO(base64.b64decode(encoded)), "rb") as wav:
+        assert (wav.getframerate(), wav.getnchannels(), wav.getsampwidth()) == (24000, 1, 2)
+        return np.frombuffer(wav.readframes(wav.getnframes()), dtype="<i2")
+
+
+def _chunks():
+    # Exactly representable PCM16 values; distinct ordered samples catch loss,
+    # repetition, reordering, and trimming more than once.
+    return [torch.arange(start, start + 5000, dtype=torch.float32) / 32768 for start in (0, 5000, 10000)]
+
+
+@pytest.mark.parametrize("mode", ["fast", "slow"])
+@pytest.mark.parametrize("shape", ["tensor", "singleton_list", "batched_list"])
+def test_fresh_delta_samples_are_forwarded_once(monkeypatch, mode, shape):
+    monkeypatch.setenv("VLLM_VIDEO_AUDIO_DELTA_MODE", mode)
+    chunks = _chunks()
+    payloads = chunks if shape == "tensor" else [[chunk] for chunk in chunks]
+    if shape == "batched_list":
+        payloads = [[chunks[0]], chunks[1:]]
+    seen = 0
+    pcm = []
+    for payload in payloads:
+        encoded, seen = OmniStreamingVideoHandler._extract_audio_delta_b64(_audio_result(payload), seen)
+        pcm.append(_pcm(encoded))
+    np.testing.assert_array_equal(np.concatenate(pcm), np.arange(_CODEC_FRAME_SAMPLES, 15000))
+    assert seen == 3
+
+
+@pytest.mark.parametrize("mode", ["fast", "slow"])
+@pytest.mark.parametrize("empty", [None, [], torch.empty(0), [torch.empty(0)]])
+def test_empty_delta_does_not_consume_first_emission(monkeypatch, mode, empty):
+    monkeypatch.setenv("VLLM_VIDEO_AUDIO_DELTA_MODE", mode)
+    encoded, seen = OmniStreamingVideoHandler._extract_audio_delta_b64(_audio_result(empty), 0)
+    assert encoded is None
+    assert seen == 0
+    encoded, seen = OmniStreamingVideoHandler._extract_audio_delta_b64(_audio_result(_chunks()[0]), seen)
+    np.testing.assert_array_equal(_pcm(encoded), np.arange(_CODEC_FRAME_SAMPLES, 5000))
+    assert seen == 1
+    encoded, after = OmniStreamingVideoHandler._extract_audio_delta_b64(_audio_result(empty), seen)
+    assert encoded is None
+    assert after == seen
+
+
+@pytest.mark.parametrize("first_length", [1, 3840, 3841])
+def test_first_chunk_trim_threshold_is_preserved(first_length):
+    first = torch.arange(first_length, dtype=torch.float32) / 32768
+    later = torch.arange(5000, 10000, dtype=torch.float32).reshape(1, -1) / 32768
+    encoded, seen = OmniStreamingVideoHandler._delta_fast(first, 0)
+    start = _CODEC_FRAME_SAMPLES if first_length > 2 * _CODEC_FRAME_SAMPLES else 0
+    np.testing.assert_array_equal(_pcm(encoded), np.arange(start, first_length))
+    encoded, seen = OmniStreamingVideoHandler._delta_fast(later, seen)
+    np.testing.assert_array_equal(_pcm(encoded), np.arange(5000, 10000))
+    assert seen == 2
+
+
+@pytest.mark.parametrize("explicit", [False, True])
+def test_only_omitted_sampling_parameters_are_coerced(explicit):
+    base = object.__new__(OmniBase)
+    base.engine = SimpleNamespace(num_stages=1)
+    params = SamplingParams(output_kind=RequestOutputKind.CUMULATIVE)
+    base.default_sampling_params_list = [params]
+    base.sampling_constraints_list = [{}]
+    resolved = base.resolve_sampling_params_list([params] if explicit else None, allow_delta_coercion=True)
+    expected = RequestOutputKind.CUMULATIVE if explicit else RequestOutputKind.DELTA
+    assert resolved[0].output_kind == expected
+    assert params.output_kind == RequestOutputKind.CUMULATIVE
+
+
+def _state(output_kind):
+    return OmniRequestState(
+        request_id="audio-delta",
+        external_req_id="audio-delta",
+        parent_req=None,
+        request_index=0,
+        lora_request=None,
+        prompt=None,
+        prompt_token_ids=[0],
+        prompt_embeds=None,
+        logprobs_processor=None,
+        detokenizer=None,
+        max_tokens_param=None,
+        arrival_time=0.0,
+        queue=None,
+        log_stats=False,
+        stream_interval=1,
+        output_kind=output_kind,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("wire_mode", ["on", "off"])
+@pytest.mark.parametrize("delta_mode", ["fast", "slow"])
+@pytest.mark.parametrize("batch_later_chunks", [False, True])
+@pytest.mark.parametrize("finish_with_payload", [False, True])
+@pytest.mark.parametrize("explicit_sampling", [False, True])
+@pytest.mark.parametrize("lengths", [(5000, 5000, 5000), (0, 5000, 0, 2000, 5000)])
+async def test_real_producer_deltas_reach_video_client(
+    monkeypatch, wire_mode, delta_mode, batch_later_chunks, finish_with_payload, explicit_sampling, lengths
+):
+    monkeypatch.setenv("VLLM_VIDEO_ASYNC_CHUNK", wire_mode)
+    monkeypatch.setenv("VLLM_VIDEO_AUDIO_DELTA_MODE", delta_mode)
+    chunks = []
+    offset = 0
+    for length in lengths:
+        # Qwen's generation runner hands off [1, samples] CPU tensors.
+        chunks.append(torch.arange(offset, offset + length, dtype=torch.float32).reshape(1, -1) / 32768)
+        offset += length
+    groups = [[chunks[0]], chunks[1:]] if batch_later_chunks else [[chunk] for chunk in chunks]
+    state = _state(RequestOutputKind.DELTA)
+    snapshots = []
+    produced = []
+    # Complete production before consumption: retained DELTA snapshots must
+    # survive subsequent accumulation and draining under consumer backpressure.
+    for index, group in enumerate(groups):
+        for chunk in group:
+            state.add_multimodal_tensor(chunk, mm_type="audio")
+        finish = FinishReason.STOP if finish_with_payload and index == len(groups) - 1 else None
+        result = state.make_request_output([], None, finish, None)
+        assert result is not None
+        assert "audio" not in state.mm_accumulated
+        snapshots.append(result.outputs[0].multimodal_output["audio"])
+        produced.append(OmniRequestOutput.from_stage_output(result, final_output_type="audio"))
+    if not finish_with_payload:
+        result = state.make_request_output([], None, FinishReason.STOP, None)
+        assert result is not None
+        produced.append(OmniRequestOutput.from_stage_output(result, final_output_type="audio"))
+
+    class Engine:
+        async def generate(self, *, prompt, request_id, output_modalities, sampling_params_list=None):
+            if explicit_sampling:
+                assert sampling_params_list is not None
+                assert sampling_params_list[0].output_kind == RequestOutputKind.DELTA
+            else:
+                assert sampling_params_list is None
+            # AsyncOmni coerces omitted parameters to DELTA as well.
+            for result in produced:
+                yield result
+
+    class Handler(QwenOmniStreamingVideoHandler):
+        async def _preprocess_to_engine_prompt(self, request):
+            return {"prompt": "describe"}
+
+    class WebSocket:
+        def __init__(self):
+            self.sent = []
+
+        async def send_json(self, message):
+            self.sent.append(message)
+
+    ws = WebSocket()
+    handler = Handler(chat_service=build_serving_chat(), engine_client=Engine())
+    config = StreamingVideoSessionConfig(
+        model="test",
+        modalities=["text", "audio"],
+        sampling_params_list=[{"temperature": 0, "output_kind": 0}] if explicit_sampling else None,
+    )
+    await handler._process_query_engine(
+        ws,
+        config,
+        [],
+        bytearray(),
+        [],
+        "describe",
+        "audio-delta",
+        asyncio.Event(),
+        {},
+    )
+    assert not any(message["type"] == "error" for message in ws.sent)
+    audio = [message for message in ws.sent if message["type"] == "response.output_audio.delta"]
+    nonempty_groups = sum(any(chunk.numel() for chunk in group) for group in groups)
+    assert len(audio) == (nonempty_groups if wire_mode == "on" else 1)
+    np.testing.assert_array_equal(
+        np.concatenate([_pcm(message["data"]) for message in audio]),
+        np.arange(_CODEC_FRAME_SAMPLES, sum(lengths)),
+    )
+    assert sum(message["type"] == "response.output_audio.done" for message in ws.sent) == 1
+    assert torch.equal(snapshots[0], chunks[0])
+    last = torch.cat(snapshots[-1], dim=-1) if isinstance(snapshots[-1], list) else snapshots[-1]
+    assert torch.equal(last, torch.cat(groups[-1], dim=-1))
+
+
+def test_cumulative_producer_is_a_different_contract():
+    # The fix must not change the producer globally: explicit CUMULATIVE users
+    # still receive accumulated history, not fresh deltas.
+    state = _state(RequestOutputKind.CUMULATIVE)
+    chunks = _chunks()
+    for chunk in chunks:
+        state.add_multimodal_tensor(chunk, mm_type="audio")
+        result = state.make_request_output([], None, None, None)
+        assert result is not None
+        assert "audio" in state.mm_accumulated
+    payload = result.outputs[0].multimodal_output["audio"]
+    assert isinstance(payload, torch.Tensor)
+    assert torch.equal(payload, torch.cat(chunks))

tests/entrypoints/test_async_omni.py

@@ -14,6 +14,8 @@ from tests.helpers.stage_config import get_deploy_config_path
 from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
 from vllm_omni.entrypoints import async_omni as async_omni_mod
 from vllm_omni.entrypoints.async_omni import AsyncOmni
+from vllm_omni.entrypoints.utils import coerce_param_message_types
+from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE
 from vllm_omni.outputs import OmniRequestOutput
 
 pytestmark = [pytest.mark.core_model]
@@ -613,6 +615,46 @@ def test_output_kind_is_preserved_with_explicit_sampling_params(output_kind):
     assert captured_params[0].output_kind == output_kind
 
 
+@pytest.mark.cpu
+@pytest.mark.parametrize(
+    "input_kind", [None, RequestOutputKind.CUMULATIVE, RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY]
+)
+def test_qwen_video_sampling_keeps_all_three_stages_delta(input_kind):
+    """Real generate/resolution preserves the video handler's DELTA contract."""
+    captured = []
+
+    async def submit(*, sampling_params_list, **kwargs):
+        captured.extend(sampling_params_list)
+
+    async def run():
+        omni = get_async_omni_instance(fake_add_request=submit)
+        # The common fixture bypasses resolution; this test must execute it.
+        del omni.resolve_sampling_params_list
+        omni.engine.num_stages = 3
+        omni.default_sampling_params_list = [SamplingParams(output_kind=RequestOutputKind.CUMULATIVE) for _ in range(3)]
+        omni.sampling_constraints_list = omni._get_sampling_constraints_list(QWEN3_OMNI_PIPELINE.stages)
+        params = None
+        if input_kind is not None:
+            # The video handler applies this coercion to explicit overrides.
+            params = coerce_param_message_types(
+                [SamplingParams(output_kind=input_kind) for _ in range(3)], is_streaming=True
+            )
+        async for _ in omni.generate(
+            prompt={"prompt": "video"},
+            request_id="video-sampling",
+            sampling_params_list=params,
+            output_modalities=["text", "audio"],
+        ):
+            pass
+        assert len(captured) == 3
+        assert all(param.output_kind == RequestOutputKind.DELTA for param in captured)
+        assert captured[1].stop_token_ids == [2150]
+        assert [param.detokenize for param in captured] == [True, False, True]
+        assert all(param.output_kind == RequestOutputKind.CUMULATIVE for param in omni.default_sampling_params_list)
+
+    asyncio.run(run())
+
+
 # End to end tests for ensuring internal manipulation of request ID
 # in diffusion / Omni models don't leak back to the user.
 #

vllm_omni/entrypoints/openai/video_stream_base.py

@@ -693,8 +693,8 @@ class OmniStreamingVideoHandler:
         text_parts: list[str] = []
         text_done_sent = False
         audio_chunk_count = 0
-        # Number of per-step tensors in OmniRequestOutput.audio_data already
-        # drained. Used by the fast path to skip already-emitted history.
+        # Count emitted tensors only to identify the first audio emission.
+        # Default and explicit sampling parameters use DELTA audio outputs.
         audio_chunks_drained = 0
         previous_text = ""
         interrupted = False
@@ -784,9 +784,9 @@ class OmniStreamingVideoHandler:
                         audio_data = self._get_audio_data(output)
                         if audio_data is not None:
                             if isinstance(audio_data, list):
-                                audio_tail_tensors = list(audio_data)
+                                audio_tail_tensors.extend(audio_data)
                             else:
-                                audio_tail_tensors = [audio_data]
+                                audio_tail_tensors.append(audio_data)
                 else:
                     delta_text, previous_text = self._extract_text_delta(
                         output,
@@ -874,16 +874,12 @@ class OmniStreamingVideoHandler:
     ) -> tuple[str | None, int]:
         """Return (base64 WAV of new samples, updated chunks_drained).
 
-        `chunks_drained` is the number of per-step tensors in
-        ``audio_data`` that have already been emitted. Each engine step appends
-        one tensor, so new samples are ``audio_data[chunks_drained:]`` — no
-        matter how many steps accumulated between reads (handles backpressure
-        cleanly, unlike a simple ``audio_data[-1]``).
-
-        Two paths, selected at runtime by ``VLLM_VIDEO_AUDIO_DELTA_MODE``:
-          * fast — only D2H the new tail. Per-call cost ∝ new chunks.
-          * slow — full cat + D2H each call. Per-call cost ∝ total history.
-                   Retained for A/B; remove once downstream callers confirm.
+        Explicit sampling parameters are coerced to DELTA above, and AsyncOmni
+        applies the same coercion to defaults. The output processor drains audio
+        after each result, so every tensor in the current payload is new.
+        `chunks_drained` only identifies the first emission for the existing
+        leading-artifact trim.
+        The legacy slow setting remains accepted as an equivalent alias.
         """
         audio_data = cls._get_audio_data(result)
         if audio_data is None:
@@ -913,23 +909,20 @@ class OmniStreamingVideoHandler:
         audio_data,
         chunks_drained: int,
     ) -> tuple[str | None, int]:
-        """Emit only tensors appended since the last call."""
-        # Single tensor: output_processor hands us one tensor before it becomes a
-        # list (see output_processor.py:89). Treat it as chunk #0.
-        if not isinstance(audio_data, list):
-            if chunks_drained >= 1:
-                return None, chunks_drained
-            tail_np = cls._tensor_to_1d_np(audio_data)
-            return cls._encode_tail(tail_np, chunks_drained, new_drained=1, is_first=True)
-
-        n = len(audio_data)
-        if n <= chunks_drained:
+        """Encode every fresh tensor in this DELTA payload."""
+        new_chunks = audio_data if isinstance(audio_data, list) else [audio_data]
+        if not new_chunks:
             return None, chunks_drained
-
-        new_chunks = audio_data[chunks_drained:]
         tail = new_chunks[0] if len(new_chunks) == 1 else torch.cat(new_chunks, dim=-1)
         tail_np = cls._tensor_to_1d_np(tail)
-        return cls._encode_tail(tail_np, chunks_drained, new_drained=n, is_first=(chunks_drained == 0))
+        if tail_np is None or len(tail_np) == 0:
+            return None, chunks_drained
+        return cls._encode_tail(
+            tail_np,
+            chunks_drained,
+            new_drained=chunks_drained + len(new_chunks),
+            is_first=(chunks_drained == 0),
+        )
 
     @classmethod
     def _delta_slow(
@@ -937,36 +930,8 @@ class OmniStreamingVideoHandler:
         audio_data,
         chunks_drained: int,
     ) -> tuple[str | None, int]:
-        """Pre-fix behaviour: concat everything each call and slice on CPU."""
-        if isinstance(audio_data, list):
-            if not audio_data:
-                return None, chunks_drained
-            audio_tensor = torch.cat(audio_data, dim=-1)
-            new_drained = len(audio_data)
-        else:
-            audio_tensor = audio_data
-            new_drained = 1
-
-        full_np = cls._tensor_to_1d_np(audio_tensor)
-        if full_np is None:
-            return None, chunks_drained
-        # chunks_drained doesn't map directly to sample offset without tracking
-        # per-chunk lengths, so we re-derive: replay the tail that corresponds
-        # to chunks appended since last call by slicing off the part produced
-        # by the already-drained prefix. For slow path this is intentionally
-        # wasteful — the point is to reproduce the pre-fix hot loop.
-        if chunks_drained == 0:
-            tail_np = full_np
-        else:
-            # Recover prefix length by re-concatenating the already-drained
-            # prefix tensors (cost intentionally identical to the baseline
-            # implementation this was lifted from).
-            if isinstance(audio_data, list) and chunks_drained < len(audio_data):
-                prefix_len = sum(int(t.shape[-1]) for t in audio_data[:chunks_drained])
-                tail_np = full_np[prefix_len:]
-            else:
-                tail_np = full_np[0:0]
-        return cls._encode_tail(tail_np, chunks_drained, new_drained=new_drained, is_first=(chunks_drained == 0))
+        """Compatibility alias for the DELTA payload contract."""
+        return cls._delta_fast(audio_data, chunks_drained)
 
     @classmethod
     def _encode_tail(