[Bugfix] Fix video prewarm cache retention and cancel-restart delay

[psv666]

Purpose

Addresses D2 and D4 in #7223 (P0.5) for the streaming video session handler.

• D2 — evicted frames remain cached: a frame can leave the bounded frame buffer while its prewarm decode is still running. The completed task then re-inserts the PIL image into frame_pil_cache, where later buffer evictions no longer remove it. Check buffer membership after decoding before publishing either the PIL image or the failed-frame marker.
• D4 — cancel-then-restart adds a fixed 100 ms delay: after cancelling and awaiting the previous query task and awaiting abort(), the session still calls asyncio.sleep(0.1). Remove that extra wait while preserving the existing cancellation and task-exit ordering. AsyncOmni already awaits the Orchestrator abort acknowledgment; elapsed wall time is not an additional completion signal.

The prewarm cache/sentinel annotations are also made consistent across the base and Qwen handlers, existing test-output attributes are declared, and the engine-client/base64-string assumptions are explicit. These changes resolve the seven existing mypy errors in the touched code. SPDX headers follow the repository hook.

The D4 change removes the fixed delay. It does not claim to fix or disprove a deeper EngineCore scheduler race: Orchestrator acknowledgment is not a guarantee that all GPU execution has stopped. Existing abort-error handling is unchanged.

Test Plan

• D2 regression: delay frame A's real PIL decode until B is accepted. With max_frames=1, A's image must be collectible while the session is open; with max_frames=2, it stays cached until session teardown. The evicted case fails before the fix.
• D4 regression: exercise the actual session handler with immediate and delayed engine abort acknowledgments. Each case interrupts three successive queries, verifies the old generator closes and abort completes before the next generation, checks distinct request IDs and the final response, and records requested sleeps instead of using a machine-dependent wall-clock limit. Both cases fail before removing the sleep, recording [0.1, 0.1, 0.1].
• Focused regression: both regressions, frame acknowledgments/consumption, cancellation, and waiting for an in-flight query on video.done.
• Full relevant suite: all tests in the eight files listed below, excluding two model-weight-dependent request-ID hardware tests (test_diffusion_generate_request_id and test_omni_generate_request_id).
• Static checks: all applicable local pre-commit gates, including mypy for Python 3.10.

vLLM Version: 0.29.0
vLLM-Omni Commit: a44fb5fcb28e910e1d39a8a3dce2322733ec8c43 (merge base: bbac4df7d3392791ce64a574c4f78c18927e61fc)
Environment: Python 3.12.13, PyTorch 2.13.0+cu129, Transformers 5.14.1, pytest 9.1.1.

Prerequisites: the repository's matching vLLM and development environment. These checks use controlled engine outputs and require no model weights or deployed model server. On the shared CUDA host, pytest ran under gpu run --gpus 1 --nonblock --timeout 8m. No real-model scheduler stress test was performed. Logs retain the warning from stale installed Omni version metadata; imports resolve to the tested checkout.

Run from the repository root:

# D2 regression: the evicted case fails before the D2 fix.
.venv/bin/python -m pytest -q tests/entrypoints/openai_api/test_serving_video_stream.py \
  -k test_frame_prewarm_only_keeps_retained_images

# D4 regression: both cases fail with the fixed sleep still present.
.venv/bin/python -m pytest -q tests/entrypoints/openai_api/test_serving_video_stream.py \
  -k test_interrupted_queries_wait_for_abort_without_fixed_delay

# Combined focused regression.
.venv/bin/python -m pytest -q tests/entrypoints/openai_api/test_serving_video_stream.py \
  -k 'prewarm or frame_ack or frames_consumed or interrupted_queries or cancels_in_flight or waits_for_in_flight'

# Full relevant test suite.
.venv/bin/python -m pytest -q \
  tests/entrypoints/openai_api/test_serving_video_stream.py \
  tests/entrypoints/openai_api/test_video_stream_handler.py \
  tests/entrypoints/openai_api/test_video_stream_session.py \
  tests/entrypoints/openai_api/test_video_frame_filter.py \
  tests/entrypoints/openai_api/test_api_server_guards.py \
  tests/entrypoints/test_async_omni.py \
  tests/engine/test_async_omni_engine_abort_ack.py \
  tests/engine/test_orchestrator.py \
  -k 'not test_diffusion_generate_request_id and not test_omni_generate_request_id'

# CI-style L1 selection for the modified regression module.
.venv/bin/python -m pytest -q tests/entrypoints/openai_api/test_serving_video_stream.py \
  -m 'core_model and cpu' --run-level=core_model

# Static checks.
.venv/bin/python -m pre_commit run --files \
  vllm_omni/entrypoints/openai/video_stream_base.py \
  vllm_omni/entrypoints/openai/serving_video_stream.py \
  tests/entrypoints/openai_api/test_serving_video_stream.py

Test Result

Check	Result
D2 pre-fix regression	1 failed, 1 passed: evicted image remained alive
D4 pre-fix regression	2 failed, 21 deselected: each restart requested a 100 ms sleep
Combined focused regression	10 passed, 13 deselected
Full relevant suite	198 passed, 2 deselected
Pre-commit, including mypy	All applicable hooks passed

An additional controlled diagnostic using the real AsyncOmni.generate() and abort RPC transport passed 3/3 runs against the fixed source: withholding the acknowledgment for 150 ms prevented the next ADD; after acknowledgment, the previous frontend state was removed before the new ADD. The remote acknowledgment and model outputs were controlled test inputs, not a real-model scheduler run.

Author self-review: checked both prewarm publication paths, eviction/retention and session cleanup, consistent cache typing, cancellation/task-exit ordering, delayed acknowledgment, and repeated interrupts. No protocol event names or payload containers are changed.

Full captured logs follow; local checkout paths are replaced with <checkout>. The two red logs are from the respective pre-fix states. The focused/full/static logs are from the source committed as a44fb5fcb28e910e1d39a8a3dce2322733ec8c43.

D2 pre-fix regression (based on aff7d64)

Reserved 1 GPU(s): [7] for command execution (timeout: 0h 10m 0s)
Released automatically if no GPU usage is detected for 30m
Running 2 items in this shard
F.                                                                       [100%]
=================================== FAILURES ===================================
____________ test_frame_prewarm_only_keeps_retained_images[evicted] ____________

monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f876cc7d7c0>
max_frames = 1

    @pytest.mark.asyncio
    @pytest.mark.parametrize("max_frames", [1, 2], ids=["evicted", "retained"])
    async def test_frame_prewarm_only_keeps_retained_images(monkeypatch, max_frames):
        frame_a = _make_jpeg(255, 0, 0)
        frame_b = _make_jpeg(0, 255, 0)
        decode_started = asyncio.Event()
        release_decode = asyncio.Event()
        frame_b_accepted = asyncio.Event()
        decoded_images: dict[bytes, weakref.ReferenceType[Image.Image]] = {}
        prewarm_task: asyncio.Task | None = None
        original_to_thread = asyncio.to_thread
    
        async def controlled_to_thread(function, *args, **kwargs):
            nonlocal prewarm_task
            if function is video_stream_base._decode_frame_bytes:
                if args[0] == frame_a:
                    prewarm_task = asyncio.current_task()
                    decode_started.set()
                    await release_decode.wait()
                image = await original_to_thread(function, *args, **kwargs)
                decoded_images[args[0]] = weakref.ref(image)
                return image
            return await original_to_thread(function, *args, **kwargs)
    
        class AckWebSocket(TimedWebSocket):
            async def send_json(self, data):
                await super().send_json(data)
                if data.get("type") == "video.frame.ack" and data.get("frame_id") == "B":
                    frame_b_accepted.set()
    
        monkeypatch.setattr(video_stream_base.asyncio, "to_thread", controlled_to_thread)
        ws = AckWebSocket()
        handler = QwenOmniStreamingVideoHandler(chat_service=object(), idle_timeout=5.0)
        session_task = asyncio.create_task(handler.handle_session(ws))
        ws.put({"type": "session.config", "max_frames": max_frames, "enable_frame_filter": False})
        try:
            ws.put({"type": "video.frame", "frame_id": "A", "data": _b64(frame_a)})
            await asyncio.wait_for(decode_started.wait(), timeout=5.0)
            ws.put({"type": "video.frame", "frame_id": "B", "data": _b64(frame_b)})
            await asyncio.wait_for(frame_b_accepted.wait(), timeout=5.0)
            ack = next(message for message in ws.sent if message.get("frame_id") == "B")
            assert ack["accepted"] is True
            assert ack.get("dropped_frame_id") == ("A" if max_frames == 1 else None)
    
            # Finish A's real decode only after B has either evicted A or joined it.
            release_decode.set()
            assert prewarm_task is not None
            await asyncio.wait_for(asyncio.shield(prewarm_task), timeout=5.0)
            gc.collect()
            assert not session_task.done()
            # A finished task must not keep an evicted PIL image alive for the session.
>           assert (decoded_images[frame_a]() is not None) == (max_frames == 2)
E           AssertionError: assert (<PIL.Image.Image image mode=RGB size=64x64 at 0x7F876CAF7C80> is not None) == (1 == 2)
E            +  where <PIL.Image.Image image mode=RGB size=64x64 at 0x7F876CAF7C80> = <weakref at 0x7f876caf3ce0; to 'Image' at 0x7f876caf7c80>()

tests/entrypoints/openai_api/test_serving_video_stream.py:644: AssertionError
---------------------------- Captured stdout setup -----------------------------
INFO 09-10 15:04:22 [scheduler.py:277] Chunked prefill is enabled with max_num_batched_tokens=2048.
INFO 09-10 15:04:22 [kernel.py:369] Final IR op priority after setting platform defaults: IrOpPriorityConfig(rms_norm=['native'], fused_add_rms_norm=['native'])
--- Running test: test_frame_prewarm_only_keeps_retained_images[evicted]
------------------------------ Captured log setup ------------------------------
INFO     vllm.config.scheduler:scheduler.py:277 Chunked prefill is enabled with max_num_batched_tokens=2048.
INFO     vllm.config.kernel:kernel.py:369 Final IR op priority after setting platform defaults: IrOpPriorityConfig(rms_norm=['native'], fused_add_rms_norm=['native'])
=============================== warnings summary ===============================
.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: 14 warnings
  <checkout>/.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: DeprecationWarning: `torch.jit.script_method` is deprecated. Please switch to `torch.compile` or `torch.export`.
    warnings.warn(

vllm_omni/version.py:55
  <checkout>/vllm_omni/version.py:55: RuntimeWarning: vLLM and vLLM-Omni appear to have mismatched major/minor versions:
   --> vLLM-Omni version 0.28.0rc2.dev75+g58cb8de68
   --> vLLM version 0.29.0
  This will likely cause compatibility issues.
    warn_if_misaligned_vllm_version()

.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24
  <checkout>/.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24: DeprecationWarning: `vllm.entrypoints.openai.api_server` is deprecated and will likely beunsupported in a future version. Use the corresponding function from `vllm.entrypoints.launchers` instead.
    warnings.warn(

vllm_omni/entrypoints/duplex/audio.py:15
  <checkout>/vllm_omni/entrypoints/duplex/audio.py:15: DeprecationWarning: 'audioop' is deprecated and slated for removal in Python 3.13
    from audioop import alaw2lin, lin2alaw, lin2ulaw, ulaw2lin

vllm_omni/entrypoints/openai/protocol/audio.py:412
  <checkout>/vllm_omni/entrypoints/openai/protocol/audio.py:412: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.13/migration/
    class CreateAudio(BaseModel):

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
--- Running Summary
=========================== short test summary info ============================
FAILED tests/entrypoints/openai_api/test_serving_video_stream.py::test_frame_prewarm_only_keeps_retained_images[evicted]
1 failed, 1 passed, 19 deselected, 18 warnings in 11.09s

D4 pre-fix regression (D2 present, fixed sleep still present)

Reserved 1 GPU(s): [7] for command execution (timeout: 0h 3m 0s)
Released automatically if no GPU usage is detected for 30m
Running 2 items in this shard
FF                                                                       [100%]
=================================== FAILURES ===================================
__ test_interrupted_queries_wait_for_abort_without_fixed_delay[immediate-ack] __

monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f405c1381a0>
delay_abort = False

    @pytest.mark.asyncio
    @pytest.mark.parametrize("delay_abort", [False, True], ids=["immediate-ack", "delayed-ack"])
    async def test_interrupted_queries_wait_for_abort_without_fixed_delay(monkeypatch, delay_abort):
        started: asyncio.Queue[str] = asyncio.Queue()
        abort_started: asyncio.Queue[str] = asyncio.Queue()
        allow_abort = asyncio.Event()
        if not delay_abort:
            allow_abort.set()
        request_ids: list[str] = []
        closed: list[str] = []
        aborted: list[str] = []
        delays: list[float] = []
        original_sleep = asyncio.sleep
    
        async def record_sleep(delay, result=None):
            # Observe requested delays without a machine-dependent latency limit.
            delays.append(delay)
            return await original_sleep(0, result)
    
        monkeypatch.setattr(video_stream_base.asyncio, "sleep", record_sleep)
    
        class BlockingEngine:
            async def generate(self, *, request_id, **kwargs):
                if request_ids:
                    assert aborted[-1] == request_ids[-1]
                request_ids.append(request_id)
                started.put_nowait(request_id)
                try:
                    if len(request_ids) <= 3:
                        yield _text_result("partial")
                        await asyncio.Event().wait()
                    else:
                        yield _text_result("final")
                finally:
                    closed.append(request_id)
    
            async def abort(self, request_id):
                assert request_id in closed
                abort_started.put_nowait(request_id)
                await allow_abort.wait()
                aborted.append(request_id)
    
        class PreprocessedHandler(QwenOmniStreamingVideoHandler):
            async def _preprocess_to_engine_prompt(self, request):
                return {"prompt_token_ids": [1]}
    
        ws = TimedWebSocket()
        handler = PreprocessedHandler(chat_service=object(), engine_client=BlockingEngine(), idle_timeout=5.0)
        task = asyncio.create_task(handler.handle_session(ws))
        try:
            ws.put({"type": "session.config", "modalities": ["text"], "enable_frame_filter": False})
            ws.put({"type": "video.frame", "data": _b64(_make_jpeg())})
            ws.put({"type": "video.query", "text": "first"})
            previous_id = await asyncio.wait_for(started.get(), timeout=2.0)
    
            for _ in range(3):
                ws.put({"type": "video.query", "text": "interrupt"})
                assert await asyncio.wait_for(abort_started.get(), timeout=2.0) == previous_id
                if delay_abort:
                    assert previous_id not in aborted
                    assert started.empty()
                    allow_abort.set()
                previous_id = await asyncio.wait_for(started.get(), timeout=2.0)
                if delay_abort:
                    allow_abort.clear()
    
            ws.put({"type": "video.done"})
            await asyncio.wait_for(task, timeout=2.0)
    
            assert aborted == request_ids[:-1]
            assert len(set(request_ids)) == 4
            assert [msg["text"] for msg in ws.sent if msg["type"] == "response.text.done"] == ["final"]
            assert "error" not in ws.sent_types()
            assert "session.done" in ws.sent_types()
>           assert not [delay for delay in delays if delay > 0], "Restart must not add a timed grace period after abort"
E           AssertionError: Restart must not add a timed grace period after abort
E           assert not [0.1, 0.1, 0.1]

tests/entrypoints/openai_api/test_serving_video_stream.py:592: AssertionError
---------------------------- Captured stdout setup -----------------------------
INFO 09-10 16:22:02 [scheduler.py:277] Chunked prefill is enabled with max_num_batched_tokens=2048.
INFO 09-10 16:22:02 [kernel.py:369] Final IR op priority after setting platform defaults: IrOpPriorityConfig(rms_norm=['native'], fused_add_rms_norm=['native'])
--- Running test: test_interrupted_queries_wait_for_abort_without_fixed_delay[immediate-ack]
------------------------------ Captured log setup ------------------------------
INFO     vllm.config.scheduler:scheduler.py:277 Chunked prefill is enabled with max_num_batched_tokens=2048.
INFO     vllm.config.kernel:kernel.py:369 Final IR op priority after setting platform defaults: IrOpPriorityConfig(rms_norm=['native'], fused_add_rms_norm=['native'])
----------------------------- Captured stdout call -----------------------------
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-d2f1740b7793
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-52d94c1c216d
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-bf3208731542
INFO 09-10 16:22:02 [video_stream_base.py:826] [TIMING] mode=on total=0.00s first_text=0.00s first_audio=-1.00s audio_chunks=0
------------------------------ Captured log call -------------------------------
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-d2f1740b7793
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-52d94c1c216d
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-bf3208731542
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:826 [TIMING] mode=on total=0.00s first_text=0.00s first_audio=-1.00s audio_chunks=0
___ test_interrupted_queries_wait_for_abort_without_fixed_delay[delayed-ack] ___

monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f405c1af9e0>
delay_abort = True

    @pytest.mark.asyncio
    @pytest.mark.parametrize("delay_abort", [False, True], ids=["immediate-ack", "delayed-ack"])
    async def test_interrupted_queries_wait_for_abort_without_fixed_delay(monkeypatch, delay_abort):
        started: asyncio.Queue[str] = asyncio.Queue()
        abort_started: asyncio.Queue[str] = asyncio.Queue()
        allow_abort = asyncio.Event()
        if not delay_abort:
            allow_abort.set()
        request_ids: list[str] = []
        closed: list[str] = []
        aborted: list[str] = []
        delays: list[float] = []
        original_sleep = asyncio.sleep
    
        async def record_sleep(delay, result=None):
            # Observe requested delays without a machine-dependent latency limit.
            delays.append(delay)
            return await original_sleep(0, result)
    
        monkeypatch.setattr(video_stream_base.asyncio, "sleep", record_sleep)
    
        class BlockingEngine:
            async def generate(self, *, request_id, **kwargs):
                if request_ids:
                    assert aborted[-1] == request_ids[-1]
                request_ids.append(request_id)
                started.put_nowait(request_id)
                try:
                    if len(request_ids) <= 3:
                        yield _text_result("partial")
                        await asyncio.Event().wait()
                    else:
                        yield _text_result("final")
                finally:
                    closed.append(request_id)
    
            async def abort(self, request_id):
                assert request_id in closed
                abort_started.put_nowait(request_id)
                await allow_abort.wait()
                aborted.append(request_id)
    
        class PreprocessedHandler(QwenOmniStreamingVideoHandler):
            async def _preprocess_to_engine_prompt(self, request):
                return {"prompt_token_ids": [1]}
    
        ws = TimedWebSocket()
        handler = PreprocessedHandler(chat_service=object(), engine_client=BlockingEngine(), idle_timeout=5.0)
        task = asyncio.create_task(handler.handle_session(ws))
        try:
            ws.put({"type": "session.config", "modalities": ["text"], "enable_frame_filter": False})
            ws.put({"type": "video.frame", "data": _b64(_make_jpeg())})
            ws.put({"type": "video.query", "text": "first"})
            previous_id = await asyncio.wait_for(started.get(), timeout=2.0)
    
            for _ in range(3):
                ws.put({"type": "video.query", "text": "interrupt"})
                assert await asyncio.wait_for(abort_started.get(), timeout=2.0) == previous_id
                if delay_abort:
                    assert previous_id not in aborted
                    assert started.empty()
                    allow_abort.set()
                previous_id = await asyncio.wait_for(started.get(), timeout=2.0)
                if delay_abort:
                    allow_abort.clear()
    
            ws.put({"type": "video.done"})
            await asyncio.wait_for(task, timeout=2.0)
    
            assert aborted == request_ids[:-1]
            assert len(set(request_ids)) == 4
            assert [msg["text"] for msg in ws.sent if msg["type"] == "response.text.done"] == ["final"]
            assert "error" not in ws.sent_types()
            assert "session.done" in ws.sent_types()
>           assert not [delay for delay in delays if delay > 0], "Restart must not add a timed grace period after abort"
E           AssertionError: Restart must not add a timed grace period after abort
E           assert not [0.1, 0.1, 0.1]

tests/entrypoints/openai_api/test_serving_video_stream.py:592: AssertionError
---------------------------- Captured stdout setup -----------------------------
--- Running test: test_interrupted_queries_wait_for_abort_without_fixed_delay[delayed-ack]
----------------------------- Captured stdout call -----------------------------
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-d13898e800e1
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-83635c293678
INFO 09-10 16:22:02 [video_stream_base.py:291] Interrupt signaled for video-c09c920fe120
INFO 09-10 16:22:02 [video_stream_base.py:826] [TIMING] mode=on total=0.00s first_text=0.00s first_audio=-1.00s audio_chunks=0
------------------------------ Captured log call -------------------------------
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-d13898e800e1
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-83635c293678
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:291 Interrupt signaled for video-c09c920fe120
INFO     vllm_omni.entrypoints.openai.video_stream_base:video_stream_base.py:826 [TIMING] mode=on total=0.00s first_text=0.00s first_audio=-1.00s audio_chunks=0
=============================== warnings summary ===============================
.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: 14 warnings
  <checkout>/.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: DeprecationWarning: `torch.jit.script_method` is deprecated. Please switch to `torch.compile` or `torch.export`.
    warnings.warn(

vllm_omni/version.py:55
  <checkout>/vllm_omni/version.py:55: RuntimeWarning: vLLM and vLLM-Omni appear to have mismatched major/minor versions:
   --> vLLM-Omni version 0.28.0rc2.dev75+g58cb8de68
   --> vLLM version 0.29.0
  This will likely cause compatibility issues.
    warn_if_misaligned_vllm_version()

.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24
  <checkout>/.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24: DeprecationWarning: `vllm.entrypoints.openai.api_server` is deprecated and will likely beunsupported in a future version. Use the corresponding function from `vllm.entrypoints.launchers` instead.
    warnings.warn(

vllm_omni/entrypoints/duplex/audio.py:15
  <checkout>/vllm_omni/entrypoints/duplex/audio.py:15: DeprecationWarning: 'audioop' is deprecated and slated for removal in Python 3.13
    from audioop import alaw2lin, lin2alaw, lin2ulaw, ulaw2lin

vllm_omni/entrypoints/openai/protocol/audio.py:412
  <checkout>/vllm_omni/entrypoints/openai/protocol/audio.py:412: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.13/migration/
    class CreateAudio(BaseModel):

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
--- Running Summary
=========================== short test summary info ============================
FAILED tests/entrypoints/openai_api/test_serving_video_stream.py::test_interrupted_queries_wait_for_abort_without_fixed_delay[immediate-ack]
FAILED tests/entrypoints/openai_api/test_serving_video_stream.py::test_interrupted_queries_wait_for_abort_without_fixed_delay[delayed-ack]
2 failed, 21 deselected, 18 warnings in 5.96s

Combined focused regression

Running 10 items in this shard
..........                                                               [100%]
=============================== warnings summary ===============================
.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: 14 warnings
  <checkout>/.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: DeprecationWarning: `torch.jit.script_method` is deprecated. Please switch to `torch.compile` or `torch.export`.
    warnings.warn(

vllm_omni/version.py:55
  <checkout>/vllm_omni/version.py:55: RuntimeWarning: vLLM and vLLM-Omni appear to have mismatched major/minor versions:
   --> vLLM-Omni version 0.28.0rc2.dev75+g58cb8de68
   --> vLLM version 0.29.0
  This will likely cause compatibility issues.
    warn_if_misaligned_vllm_version()

.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24
  <checkout>/.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24: DeprecationWarning: `vllm.entrypoints.openai.api_server` is deprecated and will likely beunsupported in a future version. Use the corresponding function from `vllm.entrypoints.launchers` instead.
    warnings.warn(

vllm_omni/entrypoints/duplex/audio.py:15
  <checkout>/vllm_omni/entrypoints/duplex/audio.py:15: DeprecationWarning: 'audioop' is deprecated and slated for removal in Python 3.13
    from audioop import alaw2lin, lin2alaw, lin2ulaw, ulaw2lin

vllm_omni/entrypoints/openai/protocol/audio.py:412
  <checkout>/vllm_omni/entrypoints/openai/protocol/audio.py:412: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.13/migration/
    class CreateAudio(BaseModel):

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
--- Running Summary
10 passed, 13 deselected, 18 warnings in 7.58s

Full relevant suite

Running 198 items in this shard
........................................................................ [ 36%]
........................................................................ [ 72%]
......................................................                   [100%]
=============================== warnings summary ===============================
.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: 14 warnings
  <checkout>/.venv/lib/python3.12/site-packages/torch/jit/_script.py:365: DeprecationWarning: `torch.jit.script_method` is deprecated. Please switch to `torch.compile` or `torch.export`.
    warnings.warn(

vllm_omni/version.py:55
  <checkout>/vllm_omni/version.py:55: RuntimeWarning: vLLM and vLLM-Omni appear to have mismatched major/minor versions:
   --> vLLM-Omni version 0.28.0rc2.dev75+g58cb8de68
   --> vLLM version 0.29.0
  This will likely cause compatibility issues.
    warn_if_misaligned_vllm_version()

.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24
  <checkout>/.venv/lib/python3.12/site-packages/vllm/entrypoints/openai/api_server.py:24: DeprecationWarning: `vllm.entrypoints.openai.api_server` is deprecated and will likely beunsupported in a future version. Use the corresponding function from `vllm.entrypoints.launchers` instead.
    warnings.warn(

vllm_omni/entrypoints/duplex/audio.py:15
  <checkout>/vllm_omni/entrypoints/duplex/audio.py:15: DeprecationWarning: 'audioop' is deprecated and slated for removal in Python 3.13
    from audioop import alaw2lin, lin2alaw, lin2ulaw, ulaw2lin

vllm_omni/entrypoints/openai/protocol/audio.py:412
  <checkout>/vllm_omni/entrypoints/openai/protocol/audio.py:412: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.13/migration/
    class CreateAudio(BaseModel):

.venv/lib/python3.12/site-packages/fastapi/testclient.py:1
  <checkout>/.venv/lib/python3.12/site-packages/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
    from starlette.testclient import TestClient as TestClient  # noqa

tests/entrypoints/openai_api/test_video_stream_session.py: 29 warnings
tests/entrypoints/openai_api/test_video_frame_filter.py: 5 warnings
  <checkout>/tests/entrypoints/openai_api/conftest_video.py:28: DeprecationWarning: 'mode' parameter is deprecated and will be removed in Pillow 13 (2026-10-15)
    img = Image.fromarray(arr, "RGB")

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
--- Running Summary
198 passed, 2 deselected, 53 warnings in 14.67s

Local pre-commit, including mypy

check yaml................................................(no files to check)Skipped
debug statements (python).....................................................Passed
fix end of files..............................................................Passed
mixed line ending.............................................................Passed
trim trailing whitespace......................................................Passed
ruff check....................................................................Passed
ruff format...................................................................Passed
typos.........................................................................Passed
Lint GitHub Actions workflow files........................(no files to check)Skipped
markdownlint-cli2.........................................(no files to check)Skipped
Run mypy for Python 3.10......................................................Passed
Ensure test files have CI marks (no direct SKU)...............................Passed
Ratchet TTS per-model branches out of serving_speech.py...(no files to check)Skipped
- hook id: check-tts-adapter-migration
Lint shell scripts........................................(no files to check)Skipped
Check SPDX headers............................................................Passed
Check forbidden imports.......................................................Passed
Prevent new torch.cuda API calls..............................................Passed
Validate Buildkite Pipelines..............................(no files to check)Skipped
Suggestion....................................................................Passed
- hook id: suggestion
- duration: 0s

To bypass all the pre-commit hooks, add --no-verify to git commit. To skip a specific hook, prefix the commit command with SKIP=<hook-id>.

Additional controlled AsyncOmni / abort RPC diagnostic on fixed source

<checkout>/vllm_omni/version.py:55: RuntimeWarning: vLLM and vLLM-Omni appear to have mismatched major/minor versions:
 --> vLLM-Omni version 0.28.0rc2.dev75+g58cb8de68
 --> vLLM version 0.29.0
This will likely cause compatibility issues.
  warn_if_misaligned_vllm_version()
INFO 09-10 16:25:18 [patch.py:252] NVFP4 W4A4 weight_scale NaN-clamp: installed.
INFO 09-10 16:25:19 [patch.py:503] inductor factorable-divisibility patch: installed.
INFO 09-10 16:25:19 [patch.py:568] [cumem-cuda] CuMemAllocator._python_free_callback patched: asleep guard extended to all platforms.
INFO 09-10 16:25:25 [video_stream_base.py:291] Interrupt signaled for video-7d8e2d221006
PASS: ack held for 150ms; second ADD absent; first frontend state retained
INFO 09-10 16:25:25 [async_omni.py:665] [AsyncOmni] Request video-7d8e2d221006-a5284375 aborted.
INFO 09-10 16:25:25 [video_stream_base.py:825] [TIMING] mode=on total=0.00s first_text=-1.00s first_audio=-1.00s audio_chunks=0
PASS: first frontend state removed before second ADD; ack -> ADD = 1.256ms; skip_sleep=False
INFO 09-10 16:25:25 [video_stream_base.py:291] Interrupt signaled for video-9e0605e16196
PASS: ack held for 150ms; second ADD absent; first frontend state retained
INFO 09-10 16:25:25 [async_omni.py:665] [AsyncOmni] Request video-9e0605e16196-84f51939 aborted.
INFO 09-10 16:25:25 [video_stream_base.py:825] [TIMING] mode=on total=0.00s first_text=-1.00s first_audio=-1.00s audio_chunks=0
PASS: first frontend state removed before second ADD; ack -> ADD = 0.741ms; skip_sleep=False
INFO 09-10 16:25:25 [video_stream_base.py:291] Interrupt signaled for video-bc326f030b82
PASS: ack held for 150ms; second ADD absent; first frontend state retained
INFO 09-10 16:25:26 [async_omni.py:665] [AsyncOmni] Request video-bc326f030b82-8e4bebe3 aborted.
INFO 09-10 16:25:26 [video_stream_base.py:825] [TIMING] mode=on total=0.00s first_text=-1.00s first_audio=-1.00s audio_chunks=0
PASS: first frontend state removed before second ADD; ack -> ADD = 0.738ms; skip_sleep=False
Runs=3; delays_ms=[1.256, 0.741, 0.738]

[vllm-omni-review-bot]

This PR appears to belong to: docs/design/module/profiling.md, docs/design/module/entrypoints.md, docs/design/module/observability.md.

Module owners: @alex-jw-brooks @linyueqian @NickCao

Routing: @alex-jw-brooks via module of the changed files, semantic router, CODEOWNERS; @linyueqian via module of the changed files, semantic router, CODEOWNERS; @NickCao via module of the changed files, semantic router, CODEOWNERS

@psv666, please review your own changes and leave a short self-review comment describing what you checked. PRs without author self-review may not be assigned a reviewer.

Please take a look when you have a chance. If you would like an automated review, mention @vllm-omni-review-bot in a comment.

[vllm-omni-review-bot]

Omni ReviewBot triage note

Automated triage of commit ad2d01f161b9 produced:

• Priority: high. Prompt maintainer attention is suggested.

These are automated triage suggestions only — the final decision belongs to the maintainers.

[psv666]

@Gaohan123 @natureofnature PTAL

[natureofnature]

LGTM from static review of head a44fb5f. I found no blocking P1/P2 issues.

• The cache fix addresses the reported race. Checking frame-buffer membership after decoding prevents an evicted frame from repopulating the cache. There is no await between the membership check and publication. Both successful decoding and failure-marker publication are guarded.

• Removing the fixed 100 ms delay is reasonable. The handler still cancels and awaits the previous query task and awaits abort() before starting the next query. The actual AsyncOmni.generate() cancellation path also awaits Orchestrator abort acknowledgment. A fixed sleep provides no additional completion guarantee.

• The scope is appropriate. The additional typing changes clarify the cache/sentinel contract. The regression tests check actual image collectability, repeated interruptions, and delayed abort acknowledgment.

One non-blocking suggestion: add a regression where a frame is evicted while decoding is pending and decoding subsequently fails. This would exercise the new early-return branch and verify that it neither retains a _BAD_FRAME entry nor emits a stale decode-failure event.

[Gaohan123]

LGTM. Thanks