From da94c3ebe15c78f5c580936f6b18a6d97d03539b Mon Sep 17 00:00:00 2001
From: amy-why-3459 <wuhaiyan17@huawei.com>
Date: Fri, 28 Aug 2026 16:58:20 +0800
Subject: [PATCH] [Bugfix][MiniCPM-o] Bind omni-duplex video frames to the unit
 they close (#6404)

Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
---
 examples/online_serving/minicpmo/README.md    |  27 ++
 .../minicpmo/realtime_duplex_demo.py          | 313 +++++++++++++++++-
 .../benchmarks/test_daily_omni_pack_modes.py  |   9 +-
 tests/e2e/features/fullduplex/test_client.py  |  63 ++++
 .../helpers/minicpmo_4_5_duplex.py            |  15 +
 .../minicpmo_realtime_duplex_scenarios.py     | 179 ++++++++--
 .../test_minicpmo_4_5_duplex.py               |  47 ++-
 .../test_minicpmo_realtime_duplex_drivers.py  | 226 +++++++++++++
 ...st_minicpmo_realtime_duplex_simple_demo.py |  61 ++++
 tests/worker/test_native_duplex_hooks.py      | 160 ++++++++-
 .../data_modules/daily_omni_dataset.py        |  19 +-
 vllm_omni/experimental/fullduplex/client.py   |  88 ++++-
 .../fullduplex/minicpmo45/policy.py           |  14 +-
 .../fullduplex/minicpmo45/runtime.py          |  44 ++-
 .../fullduplex/minicpmo45/stage0.py           | 177 +++++++---
 .../fullduplex/openai/realtime_input.py       |  98 ++++--
 .../experimental/fullduplex/video_stacking.py | 134 ++++++++
 17 files changed, 1530 insertions(+), 144 deletions(-)
 create mode 100644 vllm_omni/experimental/fullduplex/video_stacking.py

diff --git a/examples/online_serving/minicpmo/README.md b/examples/online_serving/minicpmo/README.md
index 364c3a7a9..7cd60bda9 100644
--- a/examples/online_serving/minicpmo/README.md
+++ b/examples/online_serving/minicpmo/README.md
@@ -159,6 +159,33 @@ python examples/online_serving/minicpmo/realtime_duplex_demo.py \
     --output-dir /tmp/minicpmo_realtime_duplex_demo
 ```
 
+Video input uses the same session. PyAV demuxes JPEG frames at `--video-fps`
+(default 1.0) and vLLM `load_audio` extracts a 16 kHz mono WAV unless
+`--input-wav` overrides the soundtrack. Frames keep their capture size
+(`--frame-max-side 0`); the server `process_image` normalizes at 448.
+
+`--stack-frames N` raises the visual refresh rate the way official duplex does:
+each 1 s unit also samples the `N-1` sub-frames captured inside it and tiles
+them into a single composite image sent next to that unit's base frame. The
+audio cadence never changes — a unit is always one second — and the wire always
+carries 2 images per unit however large `N` is, because the sub-frames share one
+composite. Official uses 5 for high refresh rate mode:
+
+```bash
+python examples/online_serving/minicpmo/realtime_duplex_demo.py \
+    --url ws://localhost:8099/v1/realtime?duplex=1 \
+    --model openbmb/MiniCPM-o-4_5 \
+    --input-video /path/to/clip.mp4 \
+    --stack-frames 5 \
+    --ref-audio /path/to/MiniCPM-o-Demo/assets/ref_audio/ref_minicpm_signature.wav \
+    --output-dir /tmp/minicpmo_realtime_duplex_video_demo
+```
+
+Detail inside a composite is capped by `scale_resolution=448` at
+`max_slice_nums=1`: official suggests HD slicing (`max_slice_nums=[2, 1]`) for
+stacked frames, which the duplex adapter does not implement yet. Reading small
+text or digits out of a wide scene is limited by that, not by frame timing.
+
 ## Open the experimental browser client
 
 The browser UI serves the page and proxies the same-origin Realtime WebSocket to
diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py
index f726eec92..d1bca7aef 100644
--- a/examples/online_serving/minicpmo/realtime_duplex_demo.py
+++ b/examples/online_serving/minicpmo/realtime_duplex_demo.py
@@ -12,7 +12,9 @@ from __future__ import annotations
 import argparse
 import asyncio
 import base64
+import io
 import json
+import math
 import sys
 import time
 from pathlib import Path
@@ -22,8 +24,8 @@ if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
 from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
-    PCM16_BYTES_PER_SAMPLE,  # noqa: F401 - compatibility export
-    PCM16_SAMPLE_RATE,  # noqa: F401 - compatibility export
+    PCM16_BYTES_PER_SAMPLE,
+    PCM16_SAMPLE_RATE,
     RealtimeDuplexClient,
     RealtimeEventCollector,
     build_realtime_url,
@@ -38,6 +40,10 @@ from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
 from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
     reference_audio_data_url as _ref_audio_data_url,
 )
+from vllm_omni.experimental.fullduplex.video_stacking import (  # noqa: E402
+    concat_frames_b64,
+    unit_subframe_offsets,
+)
 
 
 class _StreamingOutputWriter:
@@ -159,14 +165,258 @@ def _event_count_after(
     return sum(event.get("type") == event_type for event in events[index + 1 :])
 
 
-async def run_demo(args: argparse.Namespace) -> dict[str, object]:
-    input_pcm16 = read_pcm16_wav(Path(args.input_wav))
-    if not input_pcm16:
-        raise ValueError("input WAV has no audio")
+def _probe_video_frames_and_fps(video_path: Path) -> tuple[int, float]:
+    """Frame count and average FPS via PyAV (same approach as Daily-Omni)."""
+    import av
 
+    with av.open(str(video_path)) as container:
+        stream = container.streams.video[0]
+        avg_fps = float(stream.average_rate) if stream.average_rate else 0.0
+        num_frames = stream.frames or 0
+        if num_frames <= 0:
+            duration: float | None = None
+            if stream.duration is not None and stream.time_base is not None:
+                duration = float(stream.duration * stream.time_base)
+            elif container.duration is not None:
+                duration = container.duration / av.time_base
+            if duration and avg_fps:
+                num_frames = int(math.floor(duration * avg_fps))
+        if num_frames <= 0:
+            stream.thread_type = "AUTO"
+            num_frames = sum(1 for _ in container.decode(stream))
+    if num_frames <= 0:
+        raise ValueError(f"No decodable video frames in {video_path}")
+    return num_frames, avg_fps or 1.0
+
+
+def _upright(image, rotation: int):
+    """Apply a container display-matrix rotation, in degrees, to ``image``.
+
+    PyAV hands back the stored raster and leaves the display matrix for the
+    caller, unlike the ffmpeg CLI which autorotates. Phone clips routinely
+    carry +/-90, so skipping this feeds the model a sideways frame: arrows and
+    digits stop being readable and the model mirrors the motion it reports.
+    """
+    from PIL import Image
+
+    quarter_turns = {
+        90: Image.Transpose.ROTATE_90,
+        180: Image.Transpose.ROTATE_180,
+        270: Image.Transpose.ROTATE_270,
+    }
+    turn = quarter_turns.get(int(rotation) % 360)
+    return image.transpose(turn) if turn is not None else image
+
+
+def _decode_video_frames_rgb(video_path: Path, frame_idx: list[int]) -> list:
+    """Decode presentation-order ``frame_idx`` into upright RGB ``PIL.Image``s."""
+    import av
+
+    wanted = sorted({int(i) for i in frame_idx})
+    if not wanted:
+        return []
+
+    decoded: dict[int, object] = {}
+    with av.open(str(video_path)) as container:
+        stream = container.streams.video[0]
+        stream.thread_type = "AUTO"
+        cursor = 0
+        for position, frame in enumerate(container.decode(stream)):
+            if wanted[cursor] > position:
+                continue
+            image = _upright(frame.to_image(), getattr(frame, "rotation", 0) or 0)
+            while cursor < len(wanted) and wanted[cursor] <= position:
+                decoded[wanted[cursor]] = image
+                cursor += 1
+            if cursor >= len(wanted):
+                break
+
+    if not decoded:
+        raise ValueError(f"Decoded no frames from {video_path} for indices {wanted[:8]}")
+    last_image = decoded[max(decoded)]
+    return [decoded.get(int(i), last_image) for i in frame_idx]
+
+
+def _resize_max_side(image, max_side: int):
+    width, height = image.size
+    longest = max(width, height)
+    if longest <= max_side or max_side <= 0:
+        return image
+    scale = max_side / float(longest)
+    new_size = (max(1, int(round(width * scale))), max(1, int(round(height * scale))))
+    return image.resize(new_size)
+
+
+def _jpeg_bytes(image, *, quality: int = 95) -> bytes:
+    buf = io.BytesIO()
+    image.convert("RGB").save(buf, format="JPEG", quality=quality)
+    return buf.getvalue()
+
+
+def _to_pcm16_bytes(audio_np) -> bytes:
+    import numpy as np
+
+    arr = np.asarray(audio_np).reshape(-1)
+    if arr.dtype != np.int16:
+        arr = (np.clip(arr, -1.0, 1.0) * 32767.0).astype(np.int16)
+    return arr.tobytes()
+
+
+def _pcm16_wav_duration_s(path: Path) -> float:
+    pcm16 = read_pcm16_wav(path)
+    return len(pcm16) / float(PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
+
+
+def _extract_video_frames(
+    video_path: Path,
+    work_dir: Path,
+    *,
+    fps: float,
+    max_side: int,
+    duration_s: float,
+    stack_frames: int = 1,
+) -> tuple[list[str], list[str | None]]:
+    """Demux JPEG frames only. Frame count follows ``duration_s``, not soundtrack.
+
+    Returns the base frames (one per model unit) and the parallel stacked track.
+    With ``stack_frames=N`` each unit also samples its ``N-1`` interior
+    sub-frames and tiles them into one composite, which is how official duplex
+    raises the visual refresh rate while keeping one second of audio per unit.
+    A unit whose interior falls past the end of the clip stacks nothing.
+    """
+    num_frames, avg_fps = _probe_video_frames_and_fps(video_path)
+    sample_fps = max(float(fps), 1e-6)
+    unit_s = 1.0 / sample_fps
+    timeline_s = max(float(duration_s), unit_s)
+    num_samples = max(1, int(math.ceil(timeline_s * sample_fps - 1e-9)))
+    offsets = unit_subframe_offsets(stack_frames, unit_s=unit_s)
+
+    def frame_at(seconds: float) -> int:
+        # Official get_video_frame_audio_segments uses truncating int(), not
+        # round(), so 29.97 fps lands on the same presentation index.
+        return min(int(seconds * avg_fps), num_frames - 1)
+
+    base_idx = [frame_at(unit * unit_s) for unit in range(num_samples)]
+    sub_times = [
+        [unit * unit_s + offset for offset in offsets if unit * unit_s + offset < timeline_s]
+        for unit in range(num_samples)
+    ]
+    sub_idx = [frame_at(seconds) for times in sub_times for seconds in times]
+    images = _decode_video_frames_rgb(video_path, base_idx + sub_idx)
+    sub_images = iter(images[len(base_idx) :])
+
+    frame_dir = work_dir / "frames"
+    frame_dir.mkdir(parents=True, exist_ok=True)
+    for stale in frame_dir.glob("*.jpg"):
+        stale.unlink()
+    frames_b64: list[str] = []
+    stacked_b64: list[str | None] = []
+    for index, (image, times) in enumerate(zip(images[: len(base_idx)], sub_times), start=1):
+        jpeg = _jpeg_bytes(_resize_max_side(image, max_side))
+        (frame_dir / f"frame_{index:05d}.jpg").write_bytes(jpeg)
+        frames_b64.append(base64.b64encode(jpeg).decode("ascii"))
+        interior = [_resize_max_side(next(sub_images), max_side) for _ in times]
+        if not interior:
+            stacked_b64.append(None)
+            continue
+        composite = concat_frames_b64(interior)
+        (frame_dir / f"stack_{index:05d}.jpg").write_bytes(base64.b64decode(composite))
+        stacked_b64.append(composite)
+    return frames_b64, stacked_b64
+
+
+def _extract_video_soundtrack(video_path: Path, work_dir: Path) -> Path:
+    """Write a 16 kHz mono WAV from the video soundtrack."""
+    try:
+        from vllm.multimodal.media.audio import load_audio
+    except ImportError as exc:  # pragma: no cover
+        raise SystemExit("--input-video needs vLLM audio helpers (load_audio)") from exc
+
+    audio_np, sr = load_audio(str(video_path), sr=PCM16_SAMPLE_RATE, mono=True)
+    work_dir.mkdir(parents=True, exist_ok=True)
+    audio_path = work_dir / "input_16k.wav"
+    write_pcm16_wav(audio_path, _to_pcm16_bytes(audio_np), sample_rate_hz=int(sr))
+    return audio_path
+
+
+def _extract_video_input(
+    video_path: Path,
+    work_dir: Path,
+    *,
+    fps: float,
+    max_side: int,
+    stack_frames: int,
+) -> tuple[Path, list[str], list[str | None]]:
+    """Demux a video into duplex inputs: 16 kHz mono WAV plus JPEG frames.
+
+    Use this only when the video soundtrack is the audio source. When the
+    caller already has ``--input-wav``, extract frames against that WAV
+    duration instead so a silent clip cannot fail ``load_audio``.
+    """
+    audio_path = _extract_video_soundtrack(video_path, work_dir)
+    frames_b64, stacked_b64 = _extract_video_frames(
+        video_path,
+        work_dir,
+        fps=fps,
+        max_side=max_side,
+        duration_s=_pcm16_wav_duration_s(audio_path),
+        stack_frames=stack_frames,
+    )
+    return audio_path, frames_b64, stacked_b64
+
+
+def _resolve_duplex_av_inputs(
+    *,
+    input_wav: str | None,
+    input_video: str | None,
+    work_dir: Path,
+    fps: float,
+    max_side: int,
+    stack_frames: int = 1,
+) -> tuple[str, list[str], list[str | None]]:
+    """Return ``(wav_path, frames_b64, stacked_b64)``. External WAV wins over video audio."""
+    if input_video:
+        video_path = Path(input_video)
+        if input_wav:
+            frames, stacked = _extract_video_frames(
+                video_path,
+                work_dir,
+                fps=fps,
+                max_side=max_side,
+                duration_s=_pcm16_wav_duration_s(Path(input_wav)),
+                stack_frames=stack_frames,
+            )
+            return input_wav, frames, stacked
+        extracted_wav, frames, stacked = _extract_video_input(
+            video_path,
+            work_dir,
+            fps=fps,
+            max_side=max_side,
+            stack_frames=stack_frames,
+        )
+        return str(extracted_wav), frames, stacked
+    if not input_wav:
+        raise SystemExit("provide --input-wav, --input-video, or both")
+    return input_wav, [], []
+
+
+async def run_demo(args: argparse.Namespace) -> dict[str, object]:
     output_dir = Path(args.output_dir)
     stream_writer = _StreamingOutputWriter(output_dir)
 
+    input_wav, video_frames, stacked_video_frames = _resolve_duplex_av_inputs(
+        input_wav=args.input_wav,
+        input_video=args.input_video,
+        work_dir=output_dir / "video_input",
+        fps=args.video_fps,
+        max_side=args.frame_max_side,
+        stack_frames=args.stack_frames,
+    )
+
+    input_pcm16 = read_pcm16_wav(Path(input_wav))
+    if not input_pcm16:
+        raise ValueError("input WAV has no audio")
+
     url = build_realtime_url(
         args.url,
         args.model,
@@ -184,10 +434,12 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
             timeout_s=args.timeout_s,
         )
         stream_event_cursor = len(client.events.events)
-        await client.stream_pcm16(
+        frames_sent = await client.stream_pcm16(
             input_pcm16,
             chunk_ms=args.chunk_ms,
             realtime=not args.no_realtime_pacing,
+            video_frames=video_frames,
+            stacked_video_frames=stacked_video_frames,
         )
         commit_event_cursor = len(client.events.events)
         stream_decision = _latest_model_decision(client.events.events, stream_event_cursor)
@@ -310,6 +562,10 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                     committed_index,
                 ),
             },
+            "input_video": args.input_video,
+            "video_frames_extracted": len(video_frames),
+            "video_frames_sent": frames_sent,
+            "video_stacked_frames_extracted": sum(frame is not None for frame in stacked_video_frames),
             "audio_bytes": len(audio),
             "audio_chunk_count": len(stream_writer.audio_chunk_paths),
             "audio_chunk_files": [str(path) for path in stream_writer.audio_chunk_paths],
@@ -356,7 +612,48 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--url", default="ws://localhost:8099/v1/realtime?duplex=1")
     parser.add_argument("--model", default="openbmb/MiniCPM-o-4_5")
     parser.add_argument("--session-id")
-    parser.add_argument("--input-wav", required=True)
+    parser.add_argument(
+        "--input-wav",
+        help="Mono 16 kHz PCM16 clip to stream. Optional when --input-video carries the audio track.",
+    )
+    parser.add_argument(
+        "--input-video",
+        help=(
+            "Video to stream as omni duplex input. PyAV demuxes frames and "
+            "vLLM load_audio extracts a 16 kHz mono WAV (unless --input-wav overrides the audio)."
+        ),
+    )
+    parser.add_argument(
+        "--video-fps",
+        type=float,
+        default=1.0,
+        help="Frame extraction rate. The wire contract carries ~1 frame per second of audio.",
+    )
+    parser.add_argument(
+        "--frame-max-side",
+        type=int,
+        default=0,
+        help=(
+            "Optional client-side downscale of the longest side, in pixels. "
+            "0 (default) keeps the capture resolution so the server "
+            "process_image can normalize at scale_resolution=448 from a "
+            "higher-quality source. The web demo never resizes on the client."
+        ),
+    )
+    parser.add_argument(
+        "--stack-frames",
+        type=int,
+        default=1,
+        help=(
+            "Frames sampled per 1 s audio unit. 1 (default) sends the unit's "
+            "frame alone. N > 1 also samples the N-1 sub-frames inside the unit "
+            "and tiles them into one composite sent next to the base frame, so "
+            "motion (arrows, doors, digits changing) is visible at N fps while "
+            "the audio stays 1 s per unit. Official uses 5 for high refresh "
+            "rate mode; the wire carries 2 images per unit at any N because the "
+            "sub-frames share one composite."
+        ),
+    )
     parser.add_argument(
         "--ref-audio",
         required=True,
diff --git a/tests/benchmarks/test_daily_omni_pack_modes.py b/tests/benchmarks/test_daily_omni_pack_modes.py
index 0fd231857..34d2a5637 100644
--- a/tests/benchmarks/test_daily_omni_pack_modes.py
+++ b/tests/benchmarks/test_daily_omni_pack_modes.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 """Tests for Daily-Omni ``--daily-omni-pack-mode`` packing recipes.
 
 The ``minicpm-interleave`` mode is an accuracy-critical protocol: MiniCPM-o only reaches
@@ -27,6 +30,8 @@ _MODULE_NAME = "vllm_omni.benchmarks.data_modules.daily_omni_dataset"
 
 if _MODULE_NAME not in sys.modules:
     _spec = importlib.util.spec_from_file_location(_MODULE_NAME, _MODULE_PATH)
+    if _spec is None or _spec.loader is None:
+        raise ImportError(f"Cannot load {_MODULE_NAME} from {_MODULE_PATH}")
     _mod = importlib.util.module_from_spec(_spec)
     sys.modules[_MODULE_NAME] = _mod
     _spec.loader.exec_module(_mod)
@@ -87,7 +92,9 @@ def _stub_extract(monkeypatch, dataset: DailyOmniDataset, counter: list[int]) ->
     def fake_extract(video_path, *, audio_path, include_audio, **_kw):
         counter.append(1)
         frames = [Image.new("RGB", (16, 16), color=(i, i, i)) for i in range(_NUM_FRAMES)]
-        segments = [np.zeros(1600, dtype=np.float32) for _ in range(_NUM_FRAMES)] if include_audio else []
+        segments: list[np.ndarray] = (
+            [np.zeros(1600, dtype=np.float32) for _ in range(_NUM_FRAMES)] if include_audio else []
+        )
         return frames, segments
 
     monkeypatch.setattr(dataset, "_extract_minicpm_frame_audio_segments", fake_extract)
diff --git a/tests/e2e/features/fullduplex/test_client.py b/tests/e2e/features/fullduplex/test_client.py
index 601bb1d1c..53965b89e 100644
--- a/tests/e2e/features/fullduplex/test_client.py
+++ b/tests/e2e/features/fullduplex/test_client.py
@@ -18,6 +18,10 @@ from vllm_omni.experimental.fullduplex.client import (
 from vllm_omni.experimental.fullduplex.minicpmo45.policy import (
     MiniCPMO45DuplexPolicy,
 )
+from vllm_omni.experimental.fullduplex.video_stacking import (
+    concat_frames,
+    unit_subframe_offsets,
+)
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -418,3 +422,62 @@ def test_realtime_client_pcm16_wav_round_trip(tmp_path):
         assert wav_file.getnchannels() == 1
         assert wav_file.getframerate() == 16_000
     assert read_pcm16_wav(path) == pcm16
+
+
+def _stack_test_image(color, size=(40, 30)):
+    from PIL import Image
+
+    return Image.new("RGB", size, color)
+
+
+def test_video_stacking_tiles_a_units_subframes_into_one_image():
+    frames = [_stack_test_image((255, 0, 0)), _stack_test_image((0, 255, 0))]
+
+    composite = concat_frames(frames)
+
+    # Landscape cells stack into a column because that canvas is squarer, and
+    # the interior seam carries the separator band that lets the model tell the
+    # sub-frames apart.
+    assert composite.size == (40, 30 * 2 + 6)
+    assert composite.getpixel((20, 30 + 3)) == (0, 0, 0)
+
+
+def test_video_stacking_picks_the_squarest_grid():
+    portrait = [_stack_test_image((255, 0, 0), size=(20, 60)) for _ in range(2)]
+    four = [_stack_test_image((0, 0, 255)) for _ in range(4)]
+
+    # A portrait clip tiles side by side, four frames always tile 2x2.
+    assert concat_frames(portrait).size == (20 * 2 + 6, 60)
+    assert concat_frames(four).size == (40 * 2 + 6, 30 * 2 + 6)
+
+
+def test_video_stacking_skips_the_subframe_that_duplicates_the_base_frame():
+    # Official samples stack_frames=5 as 0.2/0.4/0.6/0.8 s: offset 0 is the base
+    # frame, already sent as frame_list[0].
+    assert unit_subframe_offsets(5) == pytest.approx([0.2, 0.4, 0.6, 0.8])
+    assert unit_subframe_offsets(1) == []
+
+
+def test_realtime_client_sends_each_units_composite_beside_its_base_frame():
+    sent: list[dict] = []
+
+    class _Client(RealtimeDuplexClient):
+        def __init__(self):
+            pass
+
+        async def send(self, event):
+            sent.append(event)
+
+    asyncio.run(
+        _Client().stream_pcm16(
+            b"\x01\x00" * (16_000 * 3),
+            chunk_ms=200,
+            realtime=False,
+            video_frames=["f0", "f1"],
+            stacked_video_frames=["s0", None],
+        )
+    )
+
+    # A composite belongs to the unit it was captured in, so it rides the same
+    # append as that unit's base frame; a unit without one sends the base alone.
+    assert [event["video_frames"] for event in sent if "video_frames" in event] == [["f0", "s0"], ["f1"]]
diff --git a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
index b1575f14e..b201df660 100644
--- a/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
+++ b/tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py
@@ -91,6 +91,19 @@ def resolve_ref_audio() -> Path:
     return ref_audio
 
 
+def duplex_camera_frames(*, seconds: int, cache_dir: Path) -> list[str]:
+    """Base64 JPEG frames of a moving synthetic scene, one per second.
+
+    Mirrors what the realtime web client captures (1 fps, no client-side
+    resize: the server normalizes at ``scale_resolution=448``).
+    """
+    from tests.e2e.online_serving.helpers.minicpmo_realtime_duplex_scenarios import _video_frames_from_file
+    from tests.helpers.media import generate_synthetic_video
+
+    video = generate_synthetic_video(448, 448, 30 * seconds, cache_dir=cache_dir)
+    return _video_frames_from_file(Path(video["file_path"]))
+
+
 def realtime_url(omni_server) -> str:
     return f"ws://{omni_server.host}:{omni_server.port}/v1/realtime?duplex=1"
 
@@ -113,6 +126,8 @@ def demo_args(
         output_audio_format="pcm16",
         chunk_ms=200,
         realtime_input=True,
+        input_video=None,
+        stack_frames=1,
         first_turn_ms=1400,
         turn_duration_ms=[],
         first_turn_transcript="duplex CI speech",
diff --git a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
index 9ac816057..7776f2a0f 100644
--- a/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
+++ b/tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
@@ -37,8 +37,10 @@ import json
 import sys
 import time
 import wave
+from collections.abc import Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
+from types import SimpleNamespace
 
 _REPO_ROOT = Path(__file__).resolve().parents[4]
 if str(_REPO_ROOT) not in sys.path:
@@ -55,15 +57,100 @@ from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
     PCM16_SAMPLE_RATE,
     RealtimeEventCollector,
     build_realtime_url,
+    duplex_unit_boundary_ms,
     read_pcm16_wav,
     summarize_session_request_metrics,
 )
 from vllm_omni.experimental.fullduplex.client import (  # noqa: E402
     reference_audio_data_url as _ref_audio_data_url,
 )
+from vllm_omni.experimental.fullduplex.video_stacking import (  # noqa: E402
+    concat_frames_b64,
+    unit_subframe_offsets,
+)
 
 _url_with_model = build_realtime_url
 _read_wav_pcm16 = read_pcm16_wav
+DemoArgs = argparse.Namespace | SimpleNamespace
+
+
+def _video_frames_from_file(
+    path: Path,
+    *,
+    fps: float = 1.0,
+    jpeg_quality: int = 70,
+    stack_frames: int = 1,
+) -> tuple[list[str], list[str | None]]:
+    """Decode ``path`` into base64 JPEG frames at the omni duplex cadence.
+
+    One frame per second matches what the realtime web client captures, so the
+    stream stays representative of a live camera. ``stack_frames > 1`` also
+    samples each unit's interior sub-frames and tiles them into one composite,
+    the official way to raise visual refresh rate without changing the audio
+    cadence. Returns ``(base_frames, stacked_frames)`` with the tracks parallel.
+    """
+    import cv2
+
+    offsets = unit_subframe_offsets(stack_frames)
+    capture = cv2.VideoCapture(str(path))
+    if not capture.isOpened():
+        raise ValueError(f"cannot open video: {path}")
+
+    def encode(frame_bgr, index: int) -> str:
+        ok, jpeg = cv2.imencode(".jpg", frame_bgr, [int(cv2.IMWRITE_JPEG_QUALITY), jpeg_quality])
+        if not ok:
+            raise ValueError(f"cannot JPEG-encode frame {index} of {path}")
+        return base64.b64encode(jpeg.tobytes()).decode("ascii")
+
+    try:
+        source_fps = capture.get(cv2.CAP_PROP_FPS) or 0.0
+        stride = max(1, round(source_fps / fps)) if source_fps > 0 else 1
+        # Sub-frame k of a unit sits ``offset * stride`` frames past its base.
+        sub_strides = {max(1, round(offset * stride)): position for position, offset in enumerate(offsets)}
+        frames: list[str] = []
+        interiors: list[list[str]] = []
+        index = 0
+        while True:
+            ok, frame_bgr = capture.read()
+            if not ok:
+                break
+            position, within_unit = divmod(index, stride)
+            if within_unit == 0:
+                frames.append(encode(frame_bgr, index))
+                interiors.append([])
+            elif within_unit in sub_strides and position < len(interiors):
+                interiors[position].append(encode(frame_bgr, index))
+            index += 1
+    finally:
+        capture.release()
+    if not frames:
+        raise ValueError(f"video has no decodable frames: {path}")
+    stacked = [concat_frames_b64(interior) if interior else None for interior in interiors]
+    return frames, stacked
+
+
+def _resolve_video_frames(args: DemoArgs) -> tuple[list[str], list[str | None]]:
+    """Base64 JPEG frames for the duplex camera track, newest last.
+
+    Returns the base track plus the parallel stacked composites. Only a video
+    source has sub-frames to stack; an explicit frame list or a still image
+    stacks nothing.
+    """
+    frames = list(getattr(args, "video_frames_b64", None) or [])
+    if frames:
+        stacked = list(getattr(args, "video_stacked_frames_b64", None) or [])
+        stacked += [None] * (len(frames) - len(stacked))
+        return frames, stacked[: len(frames)]
+    input_video = getattr(args, "input_video", None)
+    if input_video:
+        return _video_frames_from_file(
+            Path(input_video).expanduser(),
+            stack_frames=max(1, int(getattr(args, "stack_frames", 1) or 1)),
+        )
+    frame_image = getattr(args, "frame_image", None)
+    if frame_image:
+        return [base64.b64encode(Path(frame_image).expanduser().read_bytes()).decode("ascii")], [None]
+    return [], []
 
 
 @dataclass
@@ -373,17 +460,20 @@ class DemoState:
         commit_indices = [
             index for index, event in enumerate(self.events) if event.get("type") == "input_audio_buffer.committed"
         ]
-        decision_indices = [
-            index
-            for index, event in enumerate(self.events)
-            if event.get("type") == "response.created"
-            or (
-                event.get("type") == "response.listen"
-                and isinstance(event.get("response"), dict)
-                and isinstance(event["response"].get("metadata"), dict)
-                and event["response"]["metadata"].get("model_listen") is True
-            )
-        ]
+        decision_indices: list[int] = []
+        for index, event in enumerate(self.events):
+            event_type = event.get("type")
+            if event_type == "response.created":
+                decision_indices.append(index)
+                continue
+            if event_type != "response.listen":
+                continue
+            response = event.get("response")
+            if not isinstance(response, dict):
+                continue
+            metadata = response.get("metadata")
+            if isinstance(metadata, dict) and metadata.get("model_listen") is True:
+                decision_indices.append(index)
         first_input_index = self.first_index("input_audio_buffer.speech_started")
         if first_input_index is None:
             first_input_index = commit_indices[0] if commit_indices else None
@@ -500,7 +590,7 @@ class DemoState:
         return stale
 
 
-def _session_update_event(args: argparse.Namespace) -> dict[str, object]:
+def _session_update_event(args: DemoArgs) -> dict[str, object]:
     session_payload: dict[str, object] = {
         "model": args.model,
         "modalities": ["audio", "text"],
@@ -836,7 +926,8 @@ async def _send_pcm16(
     hints: dict[str, object] | None = None,
     first_chunk_hints: dict[str, object] | None = None,
     on_model_unit_ready=None,
-    frame_b64: str | None = None,
+    frames_b64: Sequence[str] | None = None,
+    stacked_frames_b64: Sequence[str | None] | None = None,
 ) -> None:
     hints = hints or {}
     first_chunk_hints = first_chunk_hints or {}
@@ -844,6 +935,8 @@ async def _send_pcm16(
     model_unit_bytes = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE
     audio_ms = 0
     next_model_unit_bytes = model_unit_bytes
+    frames = list(frames_b64 or [])
+    stacked = list(stacked_frames_b64 or [])
     frames_sent = 0
     for offset in range(0, len(pcm16), chunk_bytes):
         chunk = pcm16[offset : offset + chunk_bytes]
@@ -852,9 +945,18 @@ async def _send_pcm16(
         chunk_hints = dict(hints)
         if offset == 0:
             chunk_hints.update(first_chunk_hints)
-        if frame_b64 is not None and audio_ms > frames_sent * 1000:
-            # Omni duplex cadence: one camera frame per 1 s of audio.
-            chunk_hints["video_frames"] = [frame_b64]
+        if frames and audio_ms >= duplex_unit_boundary_ms(frames_sent):
+            # Omni duplex cadence: one camera frame per model unit, riding the
+            # append that closes it, so the frame and the second of audio it was
+            # captured during enter the same unit. A video advances one frame per
+            # unit and holds its last frame when the audio outlives the clip; a
+            # still image is a one-element list and therefore repeats.
+            # A stacked composite of that unit's interior sub-frames rides the
+            # same append right after the base frame (wire maximum of 2).
+            index = min(frames_sent, len(frames) - 1)
+            composite = stacked[index] if index < len(stacked) else None
+            base = frames[index]
+            chunk_hints["video_frames"] = [base] if composite is None else [base, composite]
             frames_sent += 1
         await ws.send(
             json.dumps(
@@ -935,7 +1037,8 @@ async def _send_clean_turn(
     realtime_input: bool = False,
     model_policy_settle_s: float = 2.0,
     commit_input: bool = True,
-    frame_b64: str | None = None,
+    frames_b64: Sequence[str] | None = None,
+    stacked_frames_b64: Sequence[str | None] | None = None,
 ) -> tuple[str | None, str]:
     before_created = state.count("response.created")
     before_model_listen = state.model_listen_count
@@ -957,7 +1060,8 @@ async def _send_clean_turn(
         chunk_ms=chunk_ms,
         realtime_delay=realtime_input,
         hints={"transcript": transcript} if send_transcript_hint else {},
-        frame_b64=frame_b64,
+        frames_b64=frames_b64,
+        stacked_frames_b64=stacked_frames_b64,
     )
     if commit_input:
         state.input_commit_sent_at_s.append(time.monotonic())
@@ -1163,7 +1267,8 @@ async def _send_listen_only_overlap_pair(
     model_policy_settle_s: float,
     validation_mode: str,
     silence_ms: int,
-    frame_b64: str | None = None,
+    frames_b64: Sequence[str] | None = None,
+    stacked_frames_b64: Sequence[str | None] | None = None,
 ) -> tuple[list[str | None], list[str], bool]:
     if not realtime_input:
         raise ValueError("listen-only-overlap requires --realtime-input")
@@ -1194,7 +1299,8 @@ async def _send_listen_only_overlap_pair(
         realtime_delay=True,
         hints={"transcript": transcripts[0]} if send_transcript_hint else {},
         on_model_unit_ready=continue_until_first_speak,
-        frame_b64=frame_b64,
+        frames_b64=frames_b64,
+        stacked_frames_b64=stacked_frames_b64,
     )
     state.input_commit_sent_at_s.append(time.monotonic())
     await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True}))
@@ -1233,7 +1339,8 @@ async def _send_listen_only_overlap_pair(
         realtime_delay=True,
         hints={"transcript": transcripts[1]} if send_transcript_hint else {},
         on_model_unit_ready=record_model_unit_ready,
-        frame_b64=frame_b64,
+        frames_b64=frames_b64,
+        stacked_frames_b64=stacked_frames_b64,
     )
     state.input_commit_sent_at_s.append(time.monotonic())
     await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True}))
@@ -1297,10 +1404,8 @@ async def _send_listen_only_overlap_pair(
     return [first_response_id, second_response_id], ["speak", second_outcome], overlap_ok
 
 
-async def run_demo(args: argparse.Namespace) -> dict[str, object]:
-    demo_frame_b64: str | None = None
-    if getattr(args, "frame_image", None):
-        demo_frame_b64 = base64.b64encode(Path(args.frame_image).read_bytes()).decode("ascii")
+async def run_demo(args: DemoArgs) -> dict[str, object]:
+    demo_frames_b64, demo_stacked_b64 = _resolve_video_frames(args)
     turn_input_paths = _turn_input_paths(
         Path(args.input_wav),
         list(getattr(args, "turn_input_wav", []) or []),
@@ -1371,7 +1476,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                     model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
                     validation_mode=validation_mode,
                     silence_ms=max(0, args.silence_ms),
-                    frame_b64=demo_frame_b64,
+                    frames_b64=demo_frames_b64,
+                    stacked_frames_b64=demo_stacked_b64,
                 )
                 for turn_index in range(2, len(turn_specs)):
                     transcript, duration_ms = turn_specs[turn_index]
@@ -1389,7 +1495,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                         realtime_input=realtime_input,
                         model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
                         commit_input=not continuous_input,
-                        frame_b64=demo_frame_b64,
+                        frames_b64=demo_frames_b64,
+                        stacked_frames_b64=demo_stacked_b64,
                     )
                     turn_response_ids.append(response_id)
                     turn_outcomes.append(outcome)
@@ -1409,7 +1516,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
                         realtime_input=realtime_input,
                         model_policy_settle_s=max(0.0, args.model_policy_settle_ms / 1000),
                         commit_input=not continuous_input,
-                        frame_b64=demo_frame_b64,
+                        frames_b64=demo_frames_b64,
+                        stacked_frames_b64=demo_stacked_b64,
                     )
                     turn_response_ids.append(response_id)
                     turn_outcomes.append(outcome)
@@ -1584,6 +1692,8 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
         "continuous_input": continuous_input,
         "continuous_input_ok": continuous_input_ok,
         "input_chunk_ms": args.chunk_ms,
+        "video_frame_count": len(demo_frames_b64),
+        "video_stacked_frame_count": sum(frame is not None for frame in demo_stacked_b64),
         "model_policy_settle_ms": args.model_policy_settle_ms,
         "turn_outcomes": turn_outcomes,
         "turn_outcomes_ok": turn_outcomes_ok,
@@ -1641,6 +1751,19 @@ def parse_args() -> argparse.Namespace:
         default=None,
         help="Optional image file sent as an omni-duplex camera frame (one per 1 s of audio)",
     )
+    parser.add_argument(
+        "--input-video",
+        default=None,
+        help="Optional video file streamed as omni-duplex camera frames (decoded at 1 fps, "
+        "overrides --frame-image; the last frame is held if the audio outlives the clip)",
+    )
+    parser.add_argument(
+        "--stack-frames",
+        type=int,
+        default=1,
+        choices=[1, 2],
+        help="Frames per append: 2 also resends the previous frame for motion context.",
+    )
     parser.add_argument(
         "--turn-input-wav",
         action="append",
diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
index cff29e662..bbdf85c09 100644
--- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
+++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
@@ -16,6 +16,7 @@ import websockets
 from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
     SERVER_PARAMS,
     demo_args,
+    duplex_camera_frames,
     multi_session_args,
     realtime_url,
     resolve_ref_audio,
@@ -30,10 +31,16 @@ from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import
 )
 from tests.helpers.mark import hardware_test
 from vllm_omni.experimental.fullduplex.client import build_realtime_url
+from vllm_omni.experimental.fullduplex.video_stacking import concat_frames_b64
 
 pytestmark = pytest.mark.omni
 
 
+def _assert_positive_int(value: object) -> None:
+    assert isinstance(value, int)
+    assert value > 0
+
+
 def _assert_request_metrics(metrics: object, *, expected_count: int) -> None:
     assert isinstance(metrics, list)
     assert len(metrics) == expected_count
@@ -142,7 +149,45 @@ def test_duplex_single_session_response_required(omni_server, tmp_path: Path) ->
     args.turn_duration_ms = [args.first_turn_ms] * args.turns
     result = asyncio.run(run_demo(args))
     assert result["ok"] is True
-    assert result["audio_delta_count"] > 0
+    _assert_positive_int(result["audio_delta_count"])
+    assert result["done_count"] == 2
+    assert result["error_count"] == 0
+    assert result["all_audio_responses_have_transcript"] is True
+    assert result["transcript_delta_done_ok"] is True
+    _assert_request_metrics(result["request_metrics"], expected_count=2)
+    _assert_session_metrics(result["session_metrics"], expected_count=2)
+
+
+@pytest.mark.advanced_model
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+def test_duplex_single_session_video_input(omni_server, tmp_path: Path) -> None:
+    """Audio plus a 1 fps camera track, the omni-duplex video contract.
+
+    Frames ride the audio appends, so the response contract is unchanged: what
+    this covers is that interleaved vision input keeps the turn intact instead
+    of stalling or erroring out mid-segment.
+    """
+    args = demo_args(
+        omni_server=omni_server,
+        input_wav=validated_input_wav(),
+        ref_audio=resolve_ref_audio(),
+        output_dir=tmp_path / "video_input",
+    )
+    args.turns = 2
+    args.turn_duration_ms = [args.first_turn_ms] * args.turns
+    frames = duplex_camera_frames(seconds=4, cache_dir=tmp_path / "camera")
+    args.video_frames_b64 = frames
+    # Each unit also carries a composite of its interior sub-frames, so the
+    # append exercises the official two-image frame_list that carries motion.
+    args.video_stacked_frames_b64 = [concat_frames_b64([frames[index]] * 2) for index in range(len(frames))]
+
+    result = asyncio.run(run_demo(args))
+
+    assert result["ok"] is True
+    assert result["video_frame_count"] == 4
+    assert result["video_stacked_frame_count"] == 4
+    _assert_positive_int(result["audio_delta_count"])
     assert result["done_count"] == 2
     assert result["error_count"] == 0
     assert result["all_audio_responses_have_transcript"] is True
diff --git a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
index 5fd65bbcb..9452448c3 100644
--- a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
+++ b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
@@ -18,6 +18,7 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 DRIVER_DIR = Path(__file__).resolve().parent
 HELPERS_DIR = DRIVER_DIR / "helpers"
 DEMO_PATH = HELPERS_DIR / "minicpmo_realtime_duplex_scenarios.py"
+EXAMPLE_DEMO_PATH = DRIVER_DIR.parents[2] / "examples/online_serving/minicpmo/realtime_duplex_demo.py"
 MULTI_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_multi_session.py"
 PAIR_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_demo_pair.py"
 SOFT_INTERRUPT_DEMO_PATH = DRIVER_DIR / "run_minicpmo_realtime_duplex_soft_interrupt.py"
@@ -50,6 +51,18 @@ def _load_pair_demo_module():
     return module
 
 
+def _load_example_demo_module():
+    spec = importlib.util.spec_from_file_location(
+        "minicpmo_realtime_duplex_example_demo_test",
+        EXAMPLE_DEMO_PATH,
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
 def _load_soft_interrupt_demo_module():
     spec = importlib.util.spec_from_file_location(
         "minicpmo_realtime_duplex_soft_interrupt_test",
@@ -1813,6 +1826,219 @@ def test_realtime_duplex_demo_waits_at_each_model_unit_and_stops_after_speak():
     assert len(ws.messages) == 10
 
 
+def _sent_video_frames(messages: list[dict[str, object]]) -> list[list[str]]:
+    frames: list[list[str]] = []
+    for message in messages:
+        raw = message.get("video_frames")
+        if not isinstance(raw, list):
+            continue
+        frames.append([frame for frame in raw if isinstance(frame, str)])
+    return frames
+
+
+def _decode_jpeg_size(frame_b64: str) -> tuple[int, int]:
+    import io
+
+    from PIL import Image
+
+    return Image.open(io.BytesIO(base64.b64decode(frame_b64))).size
+
+
+class _RecordingWebSocket:
+    def __init__(self, demo):
+        self._demo = demo
+        self.messages: list[dict[str, object]] = []
+
+    async def send(self, payload):
+        self.messages.append(self._demo.json.loads(payload))
+
+
+def _send_seconds_of_audio(demo, *, seconds: int, frames: list[str], stacked: list[str | None] | None = None):
+    ws = _RecordingWebSocket(demo)
+    asyncio.run(
+        demo._send_pcm16(
+            ws,
+            b"\x01\x00" * (demo.PCM16_SAMPLE_RATE * seconds),
+            chunk_ms=200,
+            realtime_delay=False,
+            frames_b64=frames,
+            stacked_frames_b64=stacked,
+        )
+    )
+    return ws.messages
+
+
+def test_realtime_duplex_demo_streams_one_video_frame_per_model_unit():
+    demo = _load_demo_module()
+
+    messages = _send_seconds_of_audio(demo, seconds=3, frames=["f0", "f1", "f2", "f3"])
+
+    # Frame k rides the append that closes model unit k, so Stage0 can bind it
+    # to that unit's audio. 3 s of audio closes units at 1030 ms and 2030 ms and
+    # leaves a residual too short for a third, so only two frames go out.
+    assert _sent_video_frames(messages) == [["f0"], ["f1"]]
+    frame_indices = [index for index, message in enumerate(messages) if "video_frames" in message]
+    assert frame_indices == [5, 10]
+    # The carrying appends are exactly the ones that reach a unit boundary.
+    assert [messages[index]["audio_end_ms"] for index in frame_indices] == [1200, 2200]
+
+
+def test_realtime_duplex_demo_never_sends_a_frame_before_its_unit_can_close():
+    """Regression: frame 0 used to ride the very first 200 ms append.
+
+    Stage0 cannot close a unit until 1030 ms of audio has arrived, and it does
+    not carry frames across appends, so a frame sent that early was silently
+    dropped and every later frame ended up one unit ahead of its audio.
+    """
+    demo = _load_demo_module()
+
+    messages = _send_seconds_of_audio(demo, seconds=3, frames=["f0", "f1", "f2"])
+
+    carried = [message for message in messages if "video_frames" in message]
+    assert carried, "expected at least one frame to be sent"
+    for message in carried:
+        assert message["audio_end_ms"] >= demo.duplex_unit_boundary_ms(0)
+
+
+def test_realtime_duplex_demo_holds_the_last_video_frame_when_audio_outlives_the_clip():
+    demo = _load_demo_module()
+
+    # 4 s of audio closes three units (1030/2030/3030 ms); the two-frame clip
+    # holds its last frame for the third.
+    assert _sent_video_frames(_send_seconds_of_audio(demo, seconds=4, frames=["a", "b"])) == [["a"], ["b"], ["b"]]
+    # A still image is a one-element clip and therefore repeats every unit.
+    assert _sent_video_frames(_send_seconds_of_audio(demo, seconds=3, frames=["still"])) == [["still"], ["still"]]
+
+
+def test_realtime_duplex_demo_sends_each_units_stacked_composite_next_to_its_base_frame():
+    demo = _load_demo_module()
+
+    messages = _send_seconds_of_audio(
+        demo,
+        seconds=4,
+        frames=["f0", "f1", "f2"],
+        stacked=["s0", "s1", None],
+    )
+
+    # Official pairing is frame_list=[base, composite of that unit's interior],
+    # so the composite belongs to the same unit as the base beside it -- never
+    # to the previous one. A unit without interior sub-frames sends base alone.
+    assert _sent_video_frames(messages) == [["f0", "s0"], ["f1", "s1"], ["f2"]]
+
+
+def test_realtime_duplex_demo_decodes_a_video_file_into_one_frame_per_second(tmp_path):
+    pytest.importorskip("cv2")
+    pytest.importorskip("imageio")
+    from tests.helpers.media import generate_synthetic_video
+
+    demo = _load_demo_module()
+    # 30 fps synthetic clip: 90 frames is 3 s of camera input.
+    video = generate_synthetic_video(64, 64, 90, cache_dir=tmp_path)
+
+    frames, stacked = demo._resolve_video_frames(SimpleNamespace(input_video=video["file_path"]))
+
+    assert len(frames) == 3
+    assert all(base64.b64decode(frame, validate=True)[:3] == b"\xff\xd8\xff" for frame in frames)
+    assert len(set(frames)) == 3
+    # stack_frames defaults to 1, so there is nothing to stack.
+    assert stacked == [None, None, None]
+
+
+def test_realtime_duplex_demo_stacks_a_units_interior_subframes_into_one_composite(tmp_path):
+    pytest.importorskip("cv2")
+    pytest.importorskip("imageio")
+    from tests.helpers.media import generate_synthetic_video
+
+    demo = _load_demo_module()
+    video = generate_synthetic_video(64, 64, 90, cache_dir=tmp_path)
+
+    frames, stacked = demo._resolve_video_frames(SimpleNamespace(input_video=video["file_path"], stack_frames=5))
+
+    # stack_frames=5 raises the visual refresh rate to 5 fps, but the four
+    # interior sub-frames of each second collapse into a single composite, so
+    # the wire still carries 2 images per unit and the audio cadence is
+    # untouched.
+    assert len(frames) == 3
+    assert len(stacked) == 3
+    assert all(frame is not None for frame in stacked)
+    assert all(base64.b64decode(frame, validate=True)[:3] == b"\xff\xd8\xff" for frame in stacked)
+    composite = _decode_jpeg_size(stacked[0])
+    assert composite != _decode_jpeg_size(frames[0])
+
+
+def test_example_demo_uses_external_wav_duration_and_skips_video_soundtrack(tmp_path, monkeypatch):
+    demo = _load_example_demo_module()
+    wav = tmp_path / "external.wav"
+    demo.write_pcm16_wav(wav, b"\x00\x00" * (demo.PCM16_SAMPLE_RATE * 2), sample_rate_hz=demo.PCM16_SAMPLE_RATE)
+    soundtrack_calls: list[object] = []
+
+    def fake_soundtrack(*args, **kwargs):
+        soundtrack_calls.append(1)
+        raise AssertionError("video soundtrack must not be loaded when --input-wav is set")
+
+    def fake_frames(video_path, work_dir, *, fps, max_side, duration_s, stack_frames):
+        assert video_path == Path("/tmp/silent.mp4")
+        assert fps == 1.0
+        assert duration_s == pytest.approx(2.0)
+        return ["f0", "f1"], [None, None]
+
+    monkeypatch.setattr(demo, "_extract_video_soundtrack", fake_soundtrack)
+    monkeypatch.setattr(demo, "_extract_video_frames", fake_frames)
+
+    wav_out, frames, stacked = demo._resolve_duplex_av_inputs(
+        input_wav=str(wav),
+        input_video="/tmp/silent.mp4",
+        work_dir=tmp_path / "video_input",
+        fps=1.0,
+        max_side=0,
+    )
+
+    assert wav_out == str(wav)
+    assert frames == ["f0", "f1"]
+    assert stacked == [None, None]
+    assert soundtrack_calls == []
+
+
+def test_example_demo_extracts_video_soundtrack_only_when_wav_is_absent(tmp_path, monkeypatch):
+    demo = _load_example_demo_module()
+    extracted = tmp_path / "from_video.wav"
+    demo.write_pcm16_wav(extracted, b"\x00\x00" * demo.PCM16_SAMPLE_RATE, sample_rate_hz=demo.PCM16_SAMPLE_RATE)
+
+    def fake_extract(video_path, work_dir, *, fps, max_side, stack_frames):
+        assert video_path == Path("/tmp/clip.mp4")
+        return extracted, ["v0"], ["s0"]
+
+    monkeypatch.setattr(demo, "_extract_video_input", fake_extract)
+
+    wav_out, frames, stacked = demo._resolve_duplex_av_inputs(
+        input_wav=None,
+        input_video="/tmp/clip.mp4",
+        work_dir=tmp_path / "video_input",
+        fps=1.0,
+        max_side=0,
+    )
+
+    assert wav_out == str(extracted)
+    assert frames == ["v0"]
+    assert stacked == ["s0"]
+
+
+def test_realtime_duplex_demo_prefers_a_video_over_a_still_frame_image(tmp_path):
+    demo = _load_demo_module()
+    still = tmp_path / "frame.jpg"
+    still.write_bytes(b"\xff\xd8\xff-still")
+
+    # A still has no interior sub-frames, so it never stacks.
+    assert demo._resolve_video_frames(SimpleNamespace(frame_image=str(still))) == (
+        [base64.b64encode(still.read_bytes()).decode("ascii")],
+        [None],
+    )
+    assert demo._resolve_video_frames(
+        SimpleNamespace(frame_image=str(still), video_frames_b64=["preset"]),
+    ) == (["preset"], [None])
+    assert demo._resolve_video_frames(SimpleNamespace()) == ([], [])
+
+
 def test_realtime_duplex_demo_listen_only_overlap_accepts_silence_unit_before_first_done(monkeypatch):
     demo = _load_demo_module()
     state = demo.DemoState()
diff --git a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py b/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
index 93fc4db38..ca87dce03 100644
--- a/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
+++ b/tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 import base64
 import importlib.util
 import sys
@@ -169,6 +172,64 @@ def test_open_streaming_response_requires_post_commit_drain():
     )
 
 
+def _asymmetric_image():
+    from PIL import Image
+
+    image = Image.new("RGB", (4, 2), (0, 0, 0))
+    image.putpixel((0, 0), (255, 0, 0))
+    return image
+
+
+@pytest.mark.parametrize(
+    "rotation,expected_corner",
+    [
+        (0, (0, 0)),
+        (90, (0, 3)),
+        (180, (3, 1)),
+        (-90, (1, 0)),
+        (270, (1, 0)),
+        (-270, (0, 3)),
+    ],
+)
+def test_upright_applies_the_display_matrix_quarter_turn(rotation, expected_corner):
+    demo = _load_demo_module()
+
+    rotated = demo._upright(_asymmetric_image(), rotation)
+
+    assert rotated.getpixel(expected_corner) == (255, 0, 0)
+
+
+def test_decoded_frames_apply_the_container_display_matrix(monkeypatch, tmp_path):
+    """A -90 phone clip must reach the model upright, not on its side."""
+    demo = _load_demo_module()
+    source = _asymmetric_image()
+
+    class _Frame:
+        rotation = -90
+
+        def to_image(self):
+            return source
+
+    class _Container:
+        streams = type("_Streams", (), {"video": [type("_Stream", (), {})()]})()
+
+        def decode(self, stream):
+            yield _Frame()
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+    monkeypatch.setitem(sys.modules, "av", type("_Av", (), {"open": staticmethod(lambda path: _Container())}))
+
+    (image,) = demo._decode_video_frames_rgb(tmp_path / "clip.mp4", [0])
+
+    assert image.size == (2, 4)
+    assert image.getpixel((1, 0)) == (255, 0, 0)
+
+
 def test_streaming_output_writer_persists_audio_deltas_as_they_arrive(tmp_path, capsys):
     demo = _load_demo_module()
     writer = demo._StreamingOutputWriter(tmp_path)
diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index ba50be72c..a30dcdb58 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 from __future__ import annotations
 
 import ast
@@ -849,6 +852,161 @@ def test_minicpmo_stage0_data_plane_prefill_matches_official_unit_format():
     assert result["prompt_suffix_len"] == 0
 
 
+def _stage0_vision_runtime():
+    import torch
+
+    from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+        MiniCPMO45Stage0DuplexRuntime,
+    )
+
+    class _StageModel(torch.nn.Module):
+        def __init__(self):
+            super().__init__()
+            self.embed = torch.nn.Embedding(256, 2)
+
+        def get_input_embeddings(self):
+            return self.embed
+
+        def get_audio_hidden_states(self, _data):
+            return [torch.tensor([[0.5, 0.5]], dtype=torch.float32)]
+
+    runtime = MiniCPMO45Stage0DuplexRuntime.__new__(MiniCPMO45Stage0DuplexRuntime)
+    runtime.stage_model = _StageModel()
+    runtime.thinker = runtime.stage_model
+    runtime.tokenizer = SimpleNamespace(
+        unk_token_id=0,
+        convert_tokens_to_ids=lambda token: {
+            "<unit>": 1,
+            "</unit>": 2,
+            "<|listen|>": 3,
+            "<|speak|>": 4,
+            "<|tts_bos|>": 5,
+            "<|tts_eos|>": 6,
+            "<|tts_pad|>": 7,
+            "<|chunk_eos|>": 8,
+            "<|chunk_tts_eos|>": 9,
+            "<|turn_eos|>": 10,
+            "<|audio|>": 11,
+            "<image>": 12,
+            "</image>": 13,
+            "<slice>": 14,
+            "</slice>": 15,
+        }.get(token, 0),
+        encode=lambda text, add_special_tokens=False: [201, 5],
+    )
+    runtime.processor = SimpleNamespace(get_streaming_chunk_size=lambda: 4)
+    runtime.device = "cpu"
+    runtime._init_token_ids()
+    # One 2-row block per frame keeps the expected token_ids readable; the real
+    # resampler emits VISION_EMBEDS_PER_FRAME rows per slice.
+    runtime._stage_vision_embeddings = lambda frames: [
+        [torch.full((2, 2), float(index), dtype=torch.float32)] for index in range(len(frames))
+    ]
+    return runtime
+
+
+def test_minicpmo_stage0_puts_every_frame_of_an_append_in_one_unit():
+    """Official streaming_prefill feeds the whole frame_list into one unit.
+
+    Stacked pairs used to be spread one frame per unit, so a two-frame append
+    that closed a single unit left a frame over, failed the prefill, and still
+    advanced the session state.
+    """
+    from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-stacked-frames")
+
+    result = runtime._stage_prefill_embeddings_only(
+        state,
+        np.zeros(4, dtype=np.float32),
+        video_frames=["frame-prev", "frame-cur"],
+        seq=1,
+    )
+
+    assert result["success"] is True
+    # <unit>, then each frame as <image> + embeds + </image>, then the audio.
+    assert result["input_token_ids"] == [1, 12, 0, 0, 13, 12, 0, 0, 13, 11]
+    assert state.audio_chunk_idx == 1
+
+
+def test_minicpmo_stage0_wraps_hd_slices_after_the_source_image():
+    """Official stacked pair is max_slice_nums=[2, 1]: source + slices, then audio."""
+    import torch
+
+    from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-hd-slices")
+    runtime._stage_vision_embeddings = lambda frames: [
+        [torch.ones((2, 2)), torch.full((2, 2), 2.0), torch.full((2, 2), 3.0)],
+        [torch.full((2, 2), 4.0)],
+    ]
+
+    result = runtime._stage_prefill_embeddings_only(
+        state,
+        np.zeros(4, dtype=np.float32),
+        video_frames=["base", "stack"],
+        seq=1,
+    )
+
+    assert result["success"] is True
+    # <unit> <image> src </image> <slice> p0 </slice> <slice> p1 </slice>
+    # <image> stack </image> audio
+    assert result["input_token_ids"] == [
+        1,
+        12,
+        0,
+        0,
+        13,
+        14,
+        0,
+        0,
+        15,
+        14,
+        0,
+        0,
+        15,
+        12,
+        0,
+        0,
+        13,
+        11,
+    ]
+
+
+def test_minicpmo_stage0_reports_frames_that_close_no_unit_instead_of_dropping_them():
+    """A frame only binds to the unit its append closes.
+
+    Frames are not carried across appends, so one arriving early used to vanish
+    under a generic "audio not enough" buffering result.
+    """
+    from vllm_omni.experimental.fullduplex.minicpmo45.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-early-frame")
+
+    assert runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)["success"] is True
+
+    early = runtime._stage_prefill_embeddings_only(
+        state,
+        np.zeros(2, dtype=np.float32),
+        video_frames=["frame-cur"],
+        seq=2,
+    )
+
+    assert early["success"] is False
+    assert "closes no model unit" in early["reason"]
+    # The half-filled append must not advance the unit counter.
+    assert state.audio_chunk_idx == 1
+
+
 def test_minicpmo_stage0_speech_append_sets_pending_context_once():
     import torch
 
@@ -2514,7 +2672,7 @@ def _ar_runner_classdefs():
 
 
 def _method_calls(method):
-    calls = {}
+    calls: dict[str, int] = {}
     for node in ast.walk(method):
         if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
             calls.setdefault(node.func.attr, node.lineno)
diff --git a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py b/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
index 287a9cbca..abb02b1d5 100644
--- a/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
+++ b/vllm_omni/benchmarks/data_modules/daily_omni_dataset.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 """Daily-Omni Dataset loader for benchmark.
 
 Daily-Omni is an audio-visual reasoning benchmark with 684 videos
@@ -627,10 +630,13 @@ class DailyOmniDataset(BenchmarkDataset):
 
     def _load_from_local_json(self) -> None:
         """Load QA data from local JSON file."""
-        if not self.qa_json_path.exists():
-            raise FileNotFoundError(f"QA JSON file not found: {self.qa_json_path}")
+        qa_json_path = self.qa_json_path
+        if qa_json_path is None:
+            raise RuntimeError("qa_json_path is required for local JSON loading")
+        if not qa_json_path.exists():
+            raise FileNotFoundError(f"QA JSON file not found: {qa_json_path}")
 
-        with open(self.qa_json_path, encoding="utf-8") as f:
+        with open(qa_json_path, encoding="utf-8") as f:
             data = json.load(f)
 
         # Support both list format and dict with "train"/"test" splits
@@ -682,8 +688,11 @@ class DailyOmniDataset(BenchmarkDataset):
         one ``sample()`` / request-building implementation for both.
         """
         snapshot_error: Exception | None = None
+        dataset_path = self.dataset_path
+        if dataset_path is None:
+            raise ValueError("dataset_path is required for HuggingFace loading")
         try:
-            root = ensure_daily_omni_hub_root(self.dataset_path)
+            root = ensure_daily_omni_hub_root(dataset_path)
         except Exception as e:  # noqa: BLE001 - any failure just falls through to `datasets`
             snapshot_error = e
         else:
@@ -1256,7 +1265,7 @@ class DailyOmniDataset(BenchmarkDataset):
         num_video_frames, avg_fps = _probe_video_frames_and_fps(video_path)
         duration = num_video_frames / avg_fps
         num_seconds = max(1, int(math.ceil(duration)))
-        second_timestamps = list(range(num_seconds))
+        second_timestamps = [float(i) for i in range(num_seconds)]
 
         if duration > max_num_frames:
             timestamps = [round(i * 0.1, 1) for i in range(int(duration / 0.1))]
diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py
index 69ff91f99..630d7aef3 100644
--- a/vllm_omni/experimental/fullduplex/client.py
+++ b/vllm_omni/experimental/fullduplex/client.py
@@ -11,7 +11,7 @@ import json
 import math
 import time
 import wave
-from collections.abc import Callable
+from collections.abc import Callable, Iterator, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -28,6 +28,20 @@ except ImportError as exc:  # pragma: no cover - example dependency
 PCM16_SAMPLE_RATE = 16_000
 PCM16_BYTES_PER_SAMPLE = 2
 
+# Server-side model unit boundaries, in cumulative appended audio.
+# Stage0 configures the streaming mel processor with first_chunk_ms=1035 and
+# chunk_ms=1000; the processor aligns the first chunk down to a hop_length (160
+# samples) multiple, so unit 0 closes at 16480 samples and every later unit
+# closes 16000 samples after it. Camera frames must ride the append that closes
+# a unit, otherwise Stage0 cannot bind them to that unit's audio.
+DUPLEX_FIRST_UNIT_MS = 1030
+DUPLEX_UNIT_MS = 1000
+
+
+def duplex_unit_boundary_ms(unit_index: int) -> int:
+    """Cumulative appended audio, in ms, that closes model unit ``unit_index``."""
+    return DUPLEX_FIRST_UNIT_MS + max(0, int(unit_index)) * DUPLEX_UNIT_MS
+
 
 def _rounded_ms(value: float) -> float:
     return round(float(value), 3)
@@ -561,24 +575,71 @@ class RealtimeDuplexClient:
         *,
         chunk_ms: int = 200,
         realtime: bool = True,
-    ) -> None:
+        video_frames: Sequence[str] | None = None,
+        stacked_video_frames: Sequence[str | None] | None = None,
+    ) -> int:
+        """Append PCM16 audio, optionally interleaving omni camera frames.
+
+        ``video_frames`` holds base64 JPEG/PNG frames in capture order, one per
+        second of the clip. Frame ``k`` rides the append that closes model unit
+        ``k`` (see ``duplex_unit_boundary_ms``), which reproduces the official
+        ``streaming_prefill(audio_waveform=<1 s>, frame_list=[frame])`` pairing:
+        a second of audio and the picture captured during it enter the same
+        unit. Sending on whole-second boundaries instead would strand frame 0 on
+        an append that cannot close a unit yet, and shift every later frame one
+        unit ahead of its audio.
+
+        ``stacked_video_frames`` is the optional parallel track of composites
+        built by ``video_stacking.concat_frames``: entry ``k`` tiles the
+        sub-frames captured *inside* unit ``k``, and rides the same append right
+        after the base frame, giving the official ``frame_list=[base,
+        composite]``. The audio is untouched — a unit stays one second however
+        many sub-frames the composite carries. ``None`` entries send the base
+        frame alone.
+
+        Returns the number of base frames actually sent (a clip shorter than the
+        frame list leaves the tail unsent).
+        """
         chunk_bytes = max(
             PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * chunk_ms // 1000,
             PCM16_BYTES_PER_SAMPLE,
         )
-        await self.stream_av_units(
-            ((pcm16[offset : offset + chunk_bytes], None) for offset in range(0, len(pcm16), chunk_bytes)),
-            realtime=realtime,
-        )
+        frames = list(video_frames or [])
+        stacked = list(stacked_video_frames or [])
+
+        def units() -> Iterator[tuple[bytes, list[str] | None]]:
+            audio_end_ms = 0
+            frames_sent = 0
+            for offset in range(0, len(pcm16), chunk_bytes):
+                chunk = pcm16[offset : offset + chunk_bytes]
+                audio_end_ms += len(chunk) * 1000 // (PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
+                if not frames or audio_end_ms < duplex_unit_boundary_ms(frames_sent):
+                    yield chunk, None
+                    continue
+                # A video advances one frame per unit and holds its last frame
+                # when the audio outlives the clip; a still image is a
+                # one-element list and therefore repeats.
+                index = min(frames_sent, len(frames) - 1)
+                composite = stacked[index] if index < len(stacked) else None
+                frames_sent += 1
+                yield chunk, [frames[index]] if composite is None else [frames[index], composite]
+
+        return await self.stream_av_units(units(), realtime=realtime)
 
     async def stream_av_units(
         self,
         units: Any,
         *,
         realtime: bool = True,
-    ) -> None:
-        """Stream PCM16 units, optionally attaching a JPEG to each unit."""
+    ) -> int:
+        """Stream PCM16 units, optionally attaching camera frames to each unit.
+
+        A unit's frame slot takes a single JPEG/PNG (raw bytes or base64) or a
+        sequence of them, which the Realtime wire caps at two per append.
+        Returns the number of appends that carried at least one frame.
+        """
         audio_end_ms = 0
+        frames_sent = 0
         for chunk, frame in units:
             if not chunk:
                 continue
@@ -593,11 +654,18 @@ class RealtimeDuplexClient:
                 "duration_ms": duration_ms,
                 "audio_end_ms": audio_end_ms,
             }
-            if frame is not None:
-                event["video_frames"] = [base64.b64encode(frame).decode("ascii") if isinstance(frame, bytes) else frame]
+            encoded_frames = [
+                base64.b64encode(item).decode("ascii") if isinstance(item, bytes | bytearray) else item
+                for item in (frame if isinstance(frame, list | tuple) else [frame])
+                if item is not None
+            ]
+            if encoded_frames:
+                event["video_frames"] = encoded_frames
+                frames_sent += 1
             await self.send(event)
             if realtime:
                 await asyncio.sleep(duration_ms / 1000)
+        return frames_sent
 
     async def commit(self) -> None:
         await self.send({"type": "input_audio_buffer.commit", "final": True})
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/policy.py b/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
index d869e54c9..7a1779617 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/policy.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 from __future__ import annotations
 
 from typing import Any
@@ -24,7 +27,9 @@ class MiniCPMO45DuplexPolicy:
     # Vision framing contract (omni duplex). Official streaming_prefill feeds
     # each frame as <image> + 64 resampler embeddings + </image> inside the
     # unit, ahead of the unit's audio embeddings (max_slice_nums=1 in
-    # streaming, so exactly one 64-token block per frame).
+    # streaming, so exactly one 64-token block per frame). A unit may carry
+    # its base frame plus an optional stacked composite of the sub-frames
+    # captured inside that unit (2 blocks).
     VISION_EMBEDS_PER_FRAME = 64
     VISION_TOKENS_PER_FRAME = VISION_EMBEDS_PER_FRAME + 2  # <image> + embeds + </image>
     DEFAULT_MAX_NEW_SPEAK_TOKENS_PER_CHUNK = 20
@@ -77,6 +82,8 @@ class MiniCPMO45DuplexPolicy:
         "audio_placeholder_token_id": "<|audio|>",
         "image_start_token_id": "<image>",
         "image_end_token_id": "</image>",
+        "slice_start_token_id": "<slice>",
+        "slice_end_token_id": "</slice>",
     }
 
     @classmethod
@@ -92,7 +99,7 @@ class MiniCPMO45DuplexPolicy:
                     value = value[0] if len(value) == 1 else None
             unk_token_id = getattr(tokenizer, "unk_token_id", None)
             try:
-                candidate = int(value)
+                candidate = int(value) if value is not None else -1
             except (TypeError, ValueError):
                 candidate = -1
             if candidate >= 0 and candidate != unk_token_id:
@@ -118,7 +125,8 @@ class MiniCPMO45DuplexPolicy:
         if resolved.get("listen_token_id", -1) < 0:
             eos_id = getattr(tokenizer, "eos_token_id", None)
             try:
-                resolved["listen_token_id"] = int(eos_id)
+                if eos_id is not None:
+                    resolved["listen_token_id"] = int(eos_id)
             except (TypeError, ValueError):
                 pass
         return resolved
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py b/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
index b5fbd37a5..ecd634f73 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/runtime.py
@@ -1,11 +1,11 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 from __future__ import annotations
 
 from base64 import b64decode
 from binascii import Error as BinasciiError
-from typing import Any
+from typing import Any, cast
 
 from vllm.sampling_params import SamplingParams
 
@@ -22,6 +22,9 @@ _DUPLEX_SAMPLES_PER_AUDIO_TOKEN = 1600
 # <image> + 64 resampler embeddings + </image> per frame (max_slice_nums=1),
 # matching MiniCPMO45DuplexPolicy.VISION_TOKENS_PER_FRAME.
 _DUPLEX_VISION_TOKENS_PER_FRAME = 66
+# Official stacked pair uses max_slice_nums=[2, 1]: the current frame is HD
+# sliced (1 source + 2 patches on 960x540) and the composite is not.
+_DUPLEX_HD_SLICES_PER_BASE_FRAME = 3
 
 
 def _duplex_frame_count(payload: object) -> int:
@@ -33,6 +36,22 @@ def _duplex_frame_count(payload: object) -> int:
     return sum(1 for frame in frames if isinstance(frame, str) and frame)
 
 
+def _duplex_vision_tokens(payload: object) -> int:
+    """Scheduler slots for this append's camera track.
+
+    Audio is never stacked: a unit still carries one second of soundtrack.
+    ``stack_frames`` only adds a second *image*. Official HD on that pair is
+    ``[2, 1]``, so the base frame reserves three 66-token blocks and every
+    extra frame reserves one.
+    """
+    count = _duplex_frame_count(payload)
+    if count <= 0:
+        return 0
+    if count >= 2:
+        return (_DUPLEX_HD_SLICES_PER_BASE_FRAME + (count - 1)) * _DUPLEX_VISION_TOKENS_PER_FRAME
+    return count * _DUPLEX_VISION_TOKENS_PER_FRAME
+
+
 def _duplex_pcm_sample_count(payload: object) -> int | None:
     if not isinstance(payload, dict):
         return None
@@ -48,7 +67,7 @@ def _duplex_pcm_sample_count(payload: object) -> int | None:
 
 def duplex_payload_is_exact_chunks(payload: object) -> bool:
     sample_count = _duplex_pcm_sample_count(payload)
-    return bool(sample_count) and sample_count % _DUPLEX_CHUNK_SAMPLES == 0
+    return sample_count is not None and sample_count != 0 and sample_count % _DUPLEX_CHUNK_SAMPLES == 0
 
 
 def duplex_first_append_unit_count(payload: object) -> int | None:
@@ -59,7 +78,7 @@ def duplex_first_append_unit_count(payload: object) -> int | None:
 
 
 def duplex_scheduler_token_budget(payload: object, *, default: int = 64) -> int:
-    vision_tokens = _duplex_frame_count(payload) * _DUPLEX_VISION_TOKENS_PER_FRAME
+    vision_tokens = _duplex_vision_tokens(payload)
     sample_count = _duplex_pcm_sample_count(payload)
     if sample_count is None:
         return max(1, int(default)) + vision_tokens
@@ -114,9 +133,7 @@ def build_duplex_data_plane_prompt(
         token_budget += context_reserve
         first_units = duplex_first_append_unit_count(payload)
         if first_units is not None:
-            token_budget = (
-                context_reserve + first_units * 12 - 1 + _duplex_frame_count(payload) * _DUPLEX_VISION_TOKENS_PER_FRAME
-            )
+            token_budget = context_reserve + first_units * 12 - 1 + _duplex_vision_tokens(payload)
     if seq > 1 and duplex_payload_is_exact_chunks(payload):
         token_budget += 1
     if final and duplex_payload_is_exact_chunks(payload):
@@ -124,7 +141,7 @@ def build_duplex_data_plane_prompt(
     extra_body = session_config.get("extra_body")
     raw_token_id = runtime_config.get("duplex_scheduler_token_id")
     try:
-        token_id = max(0, int(raw_token_id))
+        token_id = 0 if raw_token_id is None else max(0, int(raw_token_id))
     except (TypeError, ValueError):
         token_id = 0
     force_listen_count = _duplex_force_listen_count(extra_body)
@@ -163,16 +180,17 @@ def build_duplex_data_plane_prompt(
 
 
 def _coerce_int(value: object) -> int | None:
-    if hasattr(value, "detach"):
+    detach = getattr(value, "detach", None)
+    if callable(detach):
         try:
-            value = value.detach().cpu().reshape(-1)
-            if value.numel() == 0:
+            flat: Any = detach().cpu().reshape(-1)
+            if flat.numel() == 0:
                 return None
-            value = value[0].item()
+            value = flat[0].item()
         except Exception:
             return None
     try:
-        return int(value)
+        return int(cast(Any, value))
     except (TypeError, ValueError):
         return None
 
diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py b/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
index 392484c98..fbea82188 100644
--- a/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
+++ b/vllm_omni/experimental/fullduplex/minicpmo45/stage0.py
@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 from __future__ import annotations
 
 import base64
@@ -42,6 +45,20 @@ class _MiniCPMO45Stage0SessionState:
 class MiniCPMO45Stage0DuplexRuntime:
     """Build scheduler-owned MiniCPM-o 4.5 Stage0 duplex inputs."""
 
+    unit_token_id: int = -1
+    unit_end_token_id: int = -1
+    listen_token_id: int = -1
+    speak_token_id: int = -1
+    tts_bos_token_id: int = -1
+    tts_eos_token_id: int = -1
+    tts_pad_token_id: int = -1
+    chunk_eos_token_id: int = -1
+    chunk_tts_eos_token_id: int = -1
+    turn_eos_token_id: int = -1
+    audio_placeholder_token_id: int = -1
+    image_start_token_id: int = -1
+    image_end_token_id: int = -1
+
     def __init__(self, stage_model: Any, *, model_path: str | None = None, device: str = "cuda") -> None:
         self.stage_model = stage_model
         self.model_path = model_path
@@ -81,7 +98,7 @@ class MiniCPMO45Stage0DuplexRuntime:
                 processor._streaming_mel_processor = copy.deepcopy(shared_mel)
             state.streaming_processor = processor
         if processor is None:
-            return
+            return None
         set_streaming_mode = getattr(processor, "set_streaming_mode", None)
         if callable(set_streaming_mode):
             set_streaming_mode(
@@ -175,24 +192,40 @@ class MiniCPMO45Stage0DuplexRuntime:
         self._require_special_token_ids()
         if audio_waveform is None or len(audio_waveform) == 0:
             return self._stage_prefill_result(False, start_time, "empty audio")
-        # Omni duplex: encode this append's camera frames up front so the unit
-        # loop can interleave one <image> block per unit, mirroring official
-        # streaming_prefill (feed <unit>, then image embeds, then audio).
-        frame_blocks: list[Any] = []
-        if video_frames:
-            self._require_vision_token_ids()
-            frame_blocks = self._stage_vision_embeddings(video_frames)
-            if frame_blocks is None or len(frame_blocks) != len(video_frames):
-                return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
         state.audio_buffer = np.concatenate([state.audio_buffer, np.asarray(audio_waveform, dtype=np.float32)])
         chunk_size = self._streaming_chunk_size(processor)
         self._pad_first_audio_chunk_if_needed(state, processor)
         if len(state.audio_buffer) < chunk_size:
+            if video_frames:
+                # Frames belong to the unit this append closes, and the
+                # scheduler already reserved their vision tokens against this
+                # append. A frame arriving before its unit can close has
+                # nowhere to go, so report it instead of dropping it under a
+                # generic buffering result.
+                return self._stage_prefill_result(
+                    False,
+                    start_time,
+                    f"{len(video_frames)} video frame(s) on an append that closes no model unit "
+                    f"(need {chunk_size} samples, only {len(state.audio_buffer)}); "
+                    "send frames on unit boundaries",
+                )
             return self._stage_prefill_result(
                 False,
                 start_time,
                 f"audio not enough: need {chunk_size} samples, only {len(state.audio_buffer)}",
             )
+        # Omni duplex: encode this append's camera frames so the first unit can
+        # carry them, mirroring official streaming_prefill (feed <unit>, then
+        # every frame_list entry as its own <image> block, then the audio).
+        frame_blocks: list[Any] = []
+        if video_frames:
+            self._require_vision_token_ids()
+            encoded_frames = self._stage_vision_embeddings(video_frames)
+            if encoded_frames is None or len(encoded_frames) != len(video_frames):
+                return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
+            if any(not slices for slices in encoded_frames):
+                return self._stage_prefill_result(False, start_time, "streaming vision embedding failed")
+            frame_blocks = encoded_frames
 
         embed_parts: list[Any] = []
         token_ids: list[int] = []
@@ -246,16 +279,22 @@ class MiniCPMO45Stage0DuplexRuntime:
             embed_parts.append(self._embed_token(self.unit_token_id))
             token_ids.append(self.unit_token_id)
             if frame_blocks:
-                # Official order inside a unit: <image> + 64 resampler
-                # embeddings + </image> ahead of the unit's audio embeddings
-                # (max_slice_nums=1 in streaming, one frame per unit).
-                vision_block = self._as_2d_tensor(frame_blocks.pop(0))
-                embed_parts.append(self._embed_token(self.image_start_token_id))
-                token_ids.append(int(self.image_start_token_id))
-                embed_parts.append(vision_block)
-                token_ids.extend([self._vision_embedding_placeholder_token_id()] * int(vision_block.shape[0]))
-                embed_parts.append(self._embed_token(self.image_end_token_id))
-                token_ids.append(int(self.image_end_token_id))
+                # Official order inside a unit: every frame_list entry as
+                # <image> + 64 embeds + </image>, then any HD patches as
+                # <slice> + 64 embeds + </slice>, all ahead of this unit's
+                # *one* second of audio. stack_frames never stacks audio.
+                vision_placeholder = self._vision_embedding_placeholder_token_id()
+                for frame_slices in frame_blocks:
+                    for slice_index, frame_block in enumerate(frame_slices):
+                        vision_block = self._as_2d_tensor(frame_block)
+                        start_id, end_id = self._vision_slice_token_ids(is_source=slice_index == 0)
+                        embed_parts.append(self._embed_token(start_id))
+                        token_ids.append(int(start_id))
+                        embed_parts.append(vision_block)
+                        token_ids.extend([vision_placeholder] * int(vision_block.shape[0]))
+                        embed_parts.append(self._embed_token(end_id))
+                        token_ids.append(int(end_id))
+                frame_blocks = []
             embed_parts.append(audio_embeds)
             token_ids.extend(
                 [self._audio_embedding_placeholder_token_id()] * int(self._as_2d_tensor(audio_embeds).shape[0])
@@ -264,14 +303,6 @@ class MiniCPMO45Stage0DuplexRuntime:
             state.audio_chunk_idx += 1
             units_built += 1
             chunk_size = self._streaming_chunk_size(processor)
-        if frame_blocks:
-            # Each frame reserves one image block in the append plan. A frame
-            # without a matching audio unit would desynchronize that plan.
-            return self._stage_prefill_result(
-                False,
-                start_time,
-                f"{len(frame_blocks)} video frame(s) left without a matching audio unit",
-            )
         # Match official streaming_prefill: per chunk feed ONLY <unit>+audio. The assistant
         # turn is opened once at session init; re-emitting the turn-open prefix per chunk
         # re-opened the turn each chunk -> degenerate repetition. tts_bos/listen/turn_eos are
@@ -331,6 +362,20 @@ class MiniCPMO45Stage0DuplexRuntime:
         embeds = embedder(token)
         return self._as_2d_tensor(embeds)
 
+    def _token_embedding_dtype(self) -> Any:
+        """dtype of the decoder token embeddings, the unit's reference dtype.
+
+        Official ``get_vllm_embedding`` casts vision hidden states to the token
+        embedding dtype before they enter the sequence, so the vision tower's
+        own dtype must not leak into the concatenated unit.
+        """
+        embedder = self._token_embedder()
+        weight = getattr(embedder, "weight", None)
+        dtype = getattr(weight, "dtype", None)
+        if dtype is not None:
+            return dtype
+        return getattr(next(self.thinker.parameters()), "dtype", None)
+
     def _token_embedder(self) -> Any:
         nested_embed = getattr(getattr(getattr(self.thinker, "llm", None), "model", None), "embed_tokens", None)
         if callable(nested_embed):
@@ -392,7 +437,7 @@ class MiniCPMO45Stage0DuplexRuntime:
         )
         if len(state.audio_buffer) >= first_chunk_samples:
             return
-        padding = np.zeros(first_chunk_samples - len(state.audio_buffer), dtype=np.float32)
+        padding: np.ndarray = np.zeros(first_chunk_samples - len(state.audio_buffer), dtype=np.float32)
         state.audio_buffer = np.concatenate([padding, state.audio_buffer])
 
     def _stage_param(self, name: str, default: Any) -> Any:
@@ -454,7 +499,7 @@ class MiniCPMO45Stage0DuplexRuntime:
         previous_audio_past_key_values = (
             getattr(self.thinker, "audio_past_key_values", None) if has_audio_cache else None
         )
-        if has_audio_cache:
+        if has_audio_cache and state is not None:
             self.thinker.audio_past_key_values = state.audio_past_key_values
         try:
             for target in (self.stage_model, self.thinker):
@@ -469,18 +514,18 @@ class MiniCPMO45Stage0DuplexRuntime:
                                 suffix_extra_frames=2,
                             )
                         )
-                        if has_audio_cache:
+                        if has_audio_cache and state is not None:
                             state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
                         return result
                     except TypeError:
                         result = self._cat_nested_tensors(get_streaming(batch_feature))
-                        if has_audio_cache:
+                        if has_audio_cache and state is not None:
                             state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
                         return result
                 get_hidden = getattr(target, "get_audio_hidden_states", None)
                 if callable(get_hidden):
                     result = self._cat_nested_tensors(get_hidden(batch_feature))
-                    if has_audio_cache:
+                    if has_audio_cache and state is not None:
                         state.audio_past_key_values = getattr(self.thinker, "audio_past_key_values", None)
                     return result
             return None
@@ -627,14 +672,14 @@ class MiniCPMO45Stage0DuplexRuntime:
                 return original_register(config_class, *args, **kwargs)
 
             with _MINICPMO45_PROCESSOR_LOAD_LOCK:
-                AutoImageProcessor.register = staticmethod(register_image_processor)
+                setattr(AutoImageProcessor, "register", staticmethod(register_image_processor))
                 try:
                     processor = AutoProcessor.from_pretrained(
                         model_path,
                         trust_remote_code=True,
                     )
                 finally:
-                    AutoImageProcessor.register = staticmethod(original_register)
+                    setattr(AutoImageProcessor, "register", staticmethod(original_register))
         except Exception as exc:
             raise RuntimeError(f"Failed to load MiniCPM-o duplex processor from {model_path!r}") from exc
         if getattr(processor, "tokenizer", None) is None:
@@ -764,21 +809,24 @@ class MiniCPMO45Stage0DuplexRuntime:
                 f"missing or unknown: {', '.join(missing)}"
             )
 
-    def _stage_vision_embeddings(self, frames: list[Any]) -> list[Any] | None:
-        """Encode camera frames for omni duplex via the loaded vision tower.
+    def _vision_slice_token_ids(self, *, is_source: bool) -> tuple[int, int]:
+        if is_source:
+            return int(self.image_start_token_id), int(self.image_end_token_id)
+        start = getattr(self, "slice_start_token_id", None)
+        end = getattr(self, "slice_end_token_id", None)
+        if isinstance(start, int) and start >= 0 and isinstance(end, int) and end >= 0:
+            return start, end
+        return int(self.image_start_token_id), int(self.image_end_token_id)
 
-        Semantics mirror ``MiniCPMODuplex.streaming_prefill`` (one 64-embedding
-        block per frame at ``max_slice_nums=1``), executed through the vLLM
-        wrapper's ``get_vision_hidden_states`` (vpm + resampler) since the
-        stage model does not expose the remote-code ``get_vision_embedding``.
-        """
-        process_image = getattr(self.processor, "process_image", None)
-        if not callable(process_image):
-            return None
-        try:
-            processed = process_image(frames, max_slice_nums=1)
-        except Exception:  # noqa: BLE001 - prefill fails with a reason
-            return None
+    def _official_max_slice_nums(self, frame_count: int) -> list[int]:
+        """Official stacked pair uses HD on the current frame only: ``[2, 1]``."""
+        if frame_count <= 0:
+            return []
+        if frame_count == 1:
+            return [1]
+        return [2] + [1] * (frame_count - 1)
+
+    def _encode_processed_vision(self, processed: Any) -> list[Any] | None:
         targets = (self.stage_model, self.thinker, getattr(self.stage_model, "model", None))
         for target in targets:
             if target is None:
@@ -801,17 +849,48 @@ class MiniCPMO45Stage0DuplexRuntime:
                         flat_pixels.append(slice_pixels.to(device=device, dtype=dtype))
                     tgt_tensor = image_tgt if hasattr(image_tgt, "reshape") else torch.tensor(image_tgt)
                     flat_tgt.append(tgt_tensor.reshape(-1, 2))
+                if not flat_pixels:
+                    return None
                 tgt_sizes = torch.cat(flat_tgt, dim=0).to(device=device, dtype=torch.int64)
                 with torch.no_grad():
                     hidden = get_hidden({"pixel_values": flat_pixels, "tgt_sizes": tgt_sizes})
             except Exception:  # noqa: BLE001 - prefill fails with a reason
                 return None
             expected = MiniCPMO45DuplexPolicy.VISION_EMBEDS_PER_FRAME
+            embed_dtype = self._token_embedding_dtype()
             out: list[Any] = []
             for block in hidden:
                 block_2d = self._as_2d_tensor(block)
                 if int(block_2d.shape[0]) != expected:
                     return None
+                if embed_dtype is not None and block_2d.dtype != embed_dtype:
+                    block_2d = block_2d.to(dtype=embed_dtype)
                 out.append(block_2d)
             return out
         return None
+
+    def _stage_vision_embeddings(self, frames: list[Any]) -> list[list[Any]] | None:
+        """Encode camera frames for omni duplex via the loaded vision tower.
+
+        Official ``streaming_prefill`` encodes each ``frame_list`` entry as its
+        own image (source + optional HD slices) and never stacks audio: the
+        unit still carries one second of soundtrack. A stacked pair uses
+        ``max_slice_nums=[2, 1]`` so the current frame keeps the elevator
+        display readable and the composite stays a single 448-normalized tile.
+        Frames are processed one at a time because ``process_image([a, b])``
+        packs both PILs into one batch item.
+        """
+        process_image = getattr(self.processor, "process_image", None)
+        if not callable(process_image):
+            return None
+        out: list[list[Any]] = []
+        for frame, max_slices in zip(frames, self._official_max_slice_nums(len(frames))):
+            try:
+                processed = process_image([frame], max_slice_nums=max_slices)
+            except Exception:  # noqa: BLE001 - prefill fails with a reason
+                return None
+            encoded = self._encode_processed_vision(processed)
+            if encoded is None:
+                return None
+            out.append(encoded)
+        return out
diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_input.py b/vllm_omni/experimental/fullduplex/openai/realtime_input.py
index bd8e8270b..3d77a4172 100644
--- a/vllm_omni/experimental/fullduplex/openai/realtime_input.py
+++ b/vllm_omni/experimental/fullduplex/openai/realtime_input.py
@@ -6,6 +6,7 @@ from __future__ import annotations
 import asyncio
 import base64
 import binascii
+from collections.abc import Callable
 from typing import Any
 from uuid import uuid4
 
@@ -17,6 +18,7 @@ from vllm_omni.experimental.fullduplex.openai.audio import (
 from vllm_omni.experimental.fullduplex.openai.realtime_state import (
     REALTIME_INPUT_AUDIO_FORMATS,
     REALTIME_OUTPUT_AUDIO_FORMATS,
+    RealtimeStateOwner,
 )
 from vllm_omni.experimental.fullduplex.openai.vad import (
     SILERO_VAD_MIN_THRESHOLD,
@@ -27,9 +29,42 @@ from vllm_omni.experimental.fullduplex.openai.vad import (
 )
 
 
-class RealtimeInputTranslator:
+class RealtimeInputTranslator(RealtimeStateOwner):
     """Translate and validate client Realtime events for the duplex core."""
 
+    _opened: bool
+    _initial_session_update: bool
+    _input_speech_started: bool
+    _input_audio_buffer_has_audio: bool
+    _input_audio_buffer_had_non_speech: bool
+    _input_audio_buffer_transcript_parts: list[str]
+    _active_input_item_id: str | None
+    _last_conversation_item_id: str | None
+    _active_response_id: str | None
+    _last_response_id: str | None
+    _input_audio_format: str
+    _output_audio_format: str
+    _input_sample_rate_hz: int
+    _conversation_items: dict[str, dict[str, object]]
+    _item_truncation_cursors: dict[str, tuple[int, int]]
+    _pending_outbound: asyncio.Queue[dict[str, object]]
+    _pending_commit_item_ids: asyncio.Queue[str]
+
+    async def _send_realtime_payload(self, payload: dict[str, object]) -> None:
+        raise NotImplementedError
+
+    def _realtime_error_payload(
+        self,
+        code: str,
+        message: str,
+        *,
+        event_id: object | None = None,
+        param: object | None = None,
+    ) -> dict[str, object]:
+        raise NotImplementedError
+
+    _response_is_done: Callable[[object], bool]
+
     async def discard_pending_input_audio(
         self,
         *,
@@ -77,7 +112,8 @@ class RealtimeInputTranslator:
     async def _to_duplex_event(self, event: dict[str, object]) -> dict[str, object] | None:
         event_type = event.get("type")
         if event_type == "session.update":
-            session_payload = event.get("session") if isinstance(event.get("session"), dict) else event
+            session = event.get("session")
+            session_payload: dict[str, object] = session if isinstance(session, dict) else event
             format_error = self._validate_realtime_session_audio_formats(session_payload)
             if format_error is not None:
                 await self._send_realtime_payload(
@@ -201,7 +237,7 @@ class RealtimeInputTranslator:
                 )
                 return None
             item = self._conversation_items.get(item_id)
-            if item is None:
+            if not isinstance(item, dict):
                 await self._send_realtime_payload(
                     self._realtime_error_payload(
                         "item_not_found",
@@ -357,7 +393,8 @@ class RealtimeInputTranslator:
                         )
                     )
                     return None
-                payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame]
+                if isinstance(video_frames, list):
+                    payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame]
             self._copy_realtime_input_hints(event, payload)
             payload["is_speech"] = looks_like_speech
             if vad_result is not None:
@@ -627,7 +664,8 @@ class RealtimeInputTranslator:
     def _remove_conversation_item(self, item_id: str) -> bool:
         removed = self._conversation_items.pop(item_id, None) is not None
         if self._last_conversation_item_id == item_id:
-            self._last_conversation_item_id = next(reversed(self._conversation_items), None)
+            remaining_ids = list(self._conversation_items)
+            self._last_conversation_item_id = remaining_ids[-1] if remaining_ids else None
         self._item_truncation_cursors.pop(item_id, None)
         return removed
 
@@ -713,29 +751,34 @@ class RealtimeInputTranslator:
         audio_config = session_payload.get("audio")
         audio_input = audio_config.get("input") if isinstance(audio_config, dict) else None
         audio_output = audio_config.get("output") if isinstance(audio_config, dict) else None
-        extra_body = (
-            dict(session_payload.get("extra_body")) if isinstance(session_payload.get("extra_body"), dict) else {}
-        )
+        extra_body_payload = session_payload.get("extra_body")
+        extra_body = dict(extra_body_payload) if isinstance(extra_body_payload, dict) else {}
         extra_body["realtime_session_payload"] = self._json_safe_realtime_payload(session_payload)
         if isinstance(session_payload.get("tools"), list):
             extra_body["realtime_tools"] = session_payload["tools"]
         if isinstance(session_payload.get("tool_choice"), str | dict):
             extra_body["realtime_tool_choice"] = session_payload["tool_choice"]
-        if isinstance(session_payload.get("metadata"), dict):
-            extra_body["realtime_metadata"] = dict(session_payload["metadata"])
-        if isinstance(session_payload.get("include"), list):
-            extra_body["realtime_include"] = list(session_payload["include"])
-        if isinstance(session_payload.get("prompt"), dict):
-            extra_body["realtime_prompt"] = dict(session_payload["prompt"])
+        metadata = session_payload.get("metadata")
+        if isinstance(metadata, dict):
+            extra_body["realtime_metadata"] = dict(metadata)
+        include = session_payload.get("include")
+        if isinstance(include, list):
+            extra_body["realtime_include"] = list(include)
+        prompt = session_payload.get("prompt")
+        if isinstance(prompt, dict):
+            extra_body["realtime_prompt"] = dict(prompt)
         input_audio_transcription = self._input_audio_transcription_config(session_payload)
         if isinstance(input_audio_transcription, dict):
             extra_body["realtime_input_audio_transcription"] = dict(input_audio_transcription)
-        if isinstance(session_payload.get("input_audio_noise_reduction"), dict):
-            extra_body["realtime_input_audio_noise_reduction"] = dict(session_payload["input_audio_noise_reduction"])
-        if isinstance(audio_input, dict) and isinstance(audio_input.get("noise_reduction"), dict):
-            extra_body["realtime_input_audio_noise_reduction"] = dict(audio_input["noise_reduction"])
-        if isinstance(session_payload.get("audio"), dict):
-            extra_body["realtime_audio"] = dict(session_payload["audio"])
+        noise_reduction = session_payload.get("input_audio_noise_reduction")
+        if isinstance(noise_reduction, dict):
+            extra_body["realtime_input_audio_noise_reduction"] = dict(noise_reduction)
+        audio_noise_reduction = audio_input.get("noise_reduction") if isinstance(audio_input, dict) else None
+        if isinstance(audio_noise_reduction, dict):
+            extra_body["realtime_input_audio_noise_reduction"] = dict(audio_noise_reduction)
+        audio = session_payload.get("audio")
+        if isinstance(audio, dict):
+            extra_body["realtime_audio"] = dict(audio)
         if isinstance(session_payload.get("tracing"), str | dict):
             extra_body["realtime_tracing"] = session_payload["tracing"]
         response_format = self._duplex_response_format(self._output_audio_format)
@@ -1032,10 +1075,14 @@ class RealtimeInputTranslator:
     def _validate_realtime_video_frames(video_frames: object, max_slice_nums: object) -> str | None:
         """Validate omni-duplex camera frames on input_audio_buffer.append.
 
-        Wire contract matches the official MiniCPM-o-Demo omni client: one
-        base64 JPEG per ~1 s audio chunk. HD slicing (max_slice_nums > 1) is
-        not implemented by the duplex adapter yet and is rejected explicitly
-        rather than silently ignored.
+        Wire contract matches the official MiniCPM-o duplex loop: one base
+        base64 JPEG per ~1 s audio chunk, optionally followed by that unit's
+        stacked composite tiling the sub-frames captured inside it (at most 2
+        images either way). HD slicing (max_slice_nums > 1) is not implemented
+        by the duplex adapter yet and is rejected explicitly rather than
+        silently ignored; note official suggests it for composites
+        (``max_slice_nums=[2, 1]``), so stacked detail is capped at
+        scale_resolution until that lands.
         """
         if max_slice_nums not in (None, 1):
             return "max_slice_nums > 1 (HD slicing) is not implemented by the duplex Realtime adapter"
@@ -1416,10 +1463,11 @@ class RealtimeInputTranslator:
         self._input_speech_started = True
         if self._active_input_item_id is None:
             self._active_input_item_id = f"item_{uuid4().hex}"
+        audio_start_ms = event.get("audio_start_ms", 0)
         await self._send_realtime_payload(
             {
                 "type": "input_audio_buffer.speech_started",
-                "audio_start_ms": int(event.get("audio_start_ms", 0) or 0),
+                "audio_start_ms": int(audio_start_ms) if isinstance(audio_start_ms, int | float) else 0,
                 "item_id": self._active_input_item_id,
             }
         )
diff --git a/vllm_omni/experimental/fullduplex/video_stacking.py b/vllm_omni/experimental/fullduplex/video_stacking.py
new file mode 100644
index 000000000..2a7706759
--- /dev/null
+++ b/vllm_omni/experimental/fullduplex/video_stacking.py
@@ -0,0 +1,134 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Intra-unit frame stacking for the omni duplex video track.
+
+Official MiniCPM-o duplex raises the visual refresh rate without touching the
+audio cadence: ``get_video_frame_audio_segments(stack_frames=N)`` samples the
+``N-1`` sub-frames captured inside each second (skipping the one that would
+duplicate the second's base frame) and tiles them into a *single* composite
+image. The duplex loop then feeds ``frame_list=[base, composite]`` alongside
+that second's audio, so a unit still carries one second of audio and at most
+two images however large ``N`` is.
+
+Tiling therefore belongs to media preprocessing, not to the wire: a caller
+turns its sub-frames into one composite here and sends it as the second frame
+of the append that closes the unit.
+"""
+
+from __future__ import annotations
+
+import io
+from collections.abc import Sequence
+from typing import Any
+
+import pybase64 as base64
+
+_BG_COLOR = (255, 255, 255)
+_LINE_COLOR = (0, 0, 0)
+_LINE_WIDTH = 6
+
+
+def _as_image(frame: Any) -> Any:
+    """Coerce a PIL image, encoded bytes, or base64/data-URL string to PIL RGB."""
+    from PIL import Image
+
+    if isinstance(frame, Image.Image):
+        return frame
+    if isinstance(frame, (bytes, bytearray)):
+        return Image.open(io.BytesIO(bytes(frame))).convert("RGB")
+    if isinstance(frame, str):
+        payload = frame.split(",")[-1] if ";base64," in frame else frame
+        return Image.open(io.BytesIO(base64.b64decode(payload))).convert("RGB")
+    raise TypeError(f"unsupported frame type for stacking: {type(frame)}")
+
+
+def _grid_shape(images: Sequence[Any], *, cell_w: int, cell_h: int) -> tuple[int, int]:
+    """Rows and columns for ``images``, matching official ``concat_images``.
+
+    4 frames tile 2x2; 2 and 3 frames pick the row/column arrangement whose
+    canvas is closest to square, so a landscape clip stacks vertically and a
+    portrait one horizontally; anything else falls back to a single row.
+    """
+    count = len(images)
+    if count == 4:
+        return 2, 2
+    if count == 1:
+        return 1, 1
+    if count not in (2, 3):
+        return 1, count
+
+    candidates = [(1, count), (count, 1)]
+
+    def squareness(rows: int, cols: int) -> float:
+        width = cols * cell_w + (cols - 1) * _LINE_WIDTH
+        height = rows * cell_h + (rows - 1) * _LINE_WIDTH
+        return abs(width / max(1, height) - 1.0)
+
+    ratios = [squareness(rows, cols) for rows, cols in candidates]
+    if count == 2 and ratios[0] == ratios[1]:
+        average_aspect = sum(im.width / max(1, im.height) for im in images) / count
+        return (1, 2) if average_aspect >= 1.0 else (2, 1)
+    return candidates[min(range(len(candidates)), key=lambda index: ratios[index])]
+
+
+def _letterbox(image: Any, target_w: int, target_h: int) -> Any:
+    from PIL import Image
+
+    image = image.convert("RGB")
+    width, height = image.size
+    scale = min(target_w / width, target_h / height)
+    new_size = (max(1, round(width * scale)), max(1, round(height * scale)))
+    resized = image.resize(new_size, Image.Resampling.BICUBIC)
+    canvas = Image.new("RGB", (target_w, target_h), _BG_COLOR)
+    canvas.paste(resized, ((target_w - new_size[0]) // 2, (target_h - new_size[1]) // 2))
+    return canvas
+
+
+def concat_frames(frames: Sequence[Any]) -> Any:
+    """Tile a unit's sub-frames into one PIL image, as official stacking does.
+
+    Cells are the largest input width and height, each frame is letterboxed to
+    keep its aspect ratio, and a separator band is drawn only in the interior
+    seams so the model can tell the sub-frames apart.
+    """
+    from PIL import Image
+
+    images = [_as_image(frame) for frame in frames]
+    if not images:
+        raise ValueError("cannot stack an empty frame list")
+
+    cell_w = max(image.width for image in images)
+    cell_h = max(image.height for image in images)
+    rows, cols = _grid_shape(images, cell_w=cell_w, cell_h=cell_h)
+
+    canvas = Image.new(
+        "RGB",
+        (cols * cell_w + (cols - 1) * _LINE_WIDTH, rows * cell_h + (rows - 1) * _LINE_WIDTH),
+        _LINE_COLOR,
+    )
+    for index, image in enumerate(images[: rows * cols]):
+        row, col = divmod(index, cols)
+        canvas.paste(
+            _letterbox(image, cell_w, cell_h),
+            (col * (cell_w + _LINE_WIDTH), row * (cell_h + _LINE_WIDTH)),
+        )
+    return canvas
+
+
+def concat_frames_b64(frames: Sequence[Any], *, quality: int = 95) -> str:
+    """``concat_frames`` re-encoded as a base64 JPEG, ready for the wire."""
+    buf = io.BytesIO()
+    concat_frames(frames).save(buf, format="JPEG", quality=quality)
+    return base64.b64encode(buf.getvalue()).decode("ascii")
+
+
+def unit_subframe_offsets(stack_frames: int, *, unit_s: float = 1.0) -> list[float]:
+    """Sub-frame capture offsets inside a unit, in seconds from its start.
+
+    ``stack_frames=5`` yields 0.2/0.4/0.6/0.8 s for a 1 s unit. Offset 0 is
+    skipped because official skips it: it is the unit's base frame, already
+    sent as ``frame_list[0]``.
+    """
+    count = max(1, int(stack_frames))
+    return [index / count * unit_s for index in range(1, count)]