[Bugfix] Add opt-in MiniCPM-o 4.5 Stage-0 sliding window (#7631)

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Signed-off-by: 0z5a <dezhen.lu@uni-tuebingen.de>
Co-authored-by: 0z5a <dezhen.lu@uni-tuebingen.de>

.buildkite/common/ci_source_file_dependencies.yml

@@ -255,6 +255,8 @@ source_file_dependencies:
     - *minicpmo_4_5_duplex
     - tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
     - tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py
+    - tests/e2e/online_serving/test_minicpmo_4_5_window.py
+    - tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
     - tests/e2e/online_serving/test_duplex_client_live.py
     - tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
     - tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py

.buildkite/cuda/test-merge.yml

@@ -130,7 +130,7 @@ steps:
         source_file_dependencies: omni_minicpmo_4_5_duplex_function
         timeout_in_minutes: 50
         commands:
-          - pytest -s -v tests/e2e/online_serving/test_minicpmo_4_5_duplex.py -m 'advanced_model and cuda' --run-level 'advanced_model'
+          - pytest -s -v tests/e2e/online_serving/test_minicpmo_4_5_duplex.py tests/e2e/online_serving/test_minicpmo_4_5_window.py -m 'advanced_model and cuda' --run-level 'advanced_model'
         mirror_hardwares: h100_1
 
       - label: "Diffusion · Qwen Image Test"

tests/core/sched/test_omni_ar_scheduler_streaming.py

@@ -69,6 +69,39 @@ def _make_update(prompt_token_ids: list[int] | None = None) -> StreamingUpdate:
     )
 
 
+def _make_minicpm_window_update(
+    *,
+    seq: int,
+    mode: str,
+    high: int = 15,
+    low: int = 10,
+    context_max_units: int = 1,
+) -> StreamingUpdate:
+    update = _make_update([0] * 8)
+    update.model_intermediate_buffer = {
+        "duplex": {
+            "data_plane": True,
+            "seq": seq,
+            "runtime_config": {
+                "duplex_scheduler_token_id": 0,
+                "duplex_first_append_context_tokens": 3,
+                "duplex_window_prefix_tokens": 2,
+                "duplex_window_suffix_token_ids": [3],
+                "duplex_window_previous_marker_token_ids": [70, 71],
+                "duplex_window_special_token_ids": [99],
+                "duplex_window_config": {
+                    "sliding_window_mode": mode,
+                    "basic_window_high_tokens": high,
+                    "basic_window_low_tokens": low,
+                    "context_previous_max_tokens": 4,
+                    "context_max_units": context_max_units,
+                },
+            },
+        }
+    }
+    return update
+
+
 def _make_talker_adapter(
     *,
     max_model_len: int = 100,
@@ -510,6 +543,225 @@ def test_stage0_streaming_update_keeps_all_computed_tokens_without_placeholder()
     assert sched._new_prompt_len_snapshot[session.request_id] == 2
 
 
+def test_stage0_basic_window_rebuilds_below_low_watermark() -> None:
+    sched = _make_scheduler(stage_id=0)
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids.clear()
+    session._all_token_ids.extend(session.prompt_token_ids)
+    session.num_prompt_tokens = 9
+    session.append_output_token_ids([40])
+    session.num_computed_tokens = 10
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+
+    sched._update_request_as_session(session, update)
+
+    assert session.prompt_token_ids == [0] * 9
+    assert session.num_computed_tokens == 0
+    assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
+    plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+    assert plan == {
+        "completed_token_ids": [40],
+        "replace": True,
+        "mode": "basic",
+        "drop_units": 1,
+        "dropped_tokens": 9,
+        "previous_token_ids": [],
+        "previous_marker_token_ids": [70, 71],
+        "replacement_prompt_len": 9,
+    }
+    sched._free_request_blocks.assert_called_once_with(session)
+
+
+def test_stage0_context_window_compacts_dropped_speech() -> None:
+    sched = _make_scheduler(stage_id=0)
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids.clear()
+    session._all_token_ids.extend(session.prompt_token_ids)
+    session.num_prompt_tokens = 9
+    session.append_output_token_ids([40])
+    session.num_computed_tokens = 10
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+
+    first = _make_minicpm_window_update(seq=2, mode="context")
+    sched._update_request_as_session(session, first)
+    assert session.num_prompt_tokens == 18
+    assert first.model_intermediate_buffer["duplex"]["stage0_window"] == {"completed_token_ids": [40]}
+
+    session.append_output_token_ids([50])
+    session.num_computed_tokens = 19
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    second = _make_minicpm_window_update(seq=3, mode="context")
+    sched._update_request_as_session(session, second)
+
+    assert session.prompt_token_ids == [0] * 21
+    plan = second.model_intermediate_buffer["duplex"]["stage0_window"]
+    assert plan["completed_token_ids"] == [50]
+    assert plan["drop_units"] == 1
+    assert plan["previous_token_ids"] == [40]
+    # The worker embeds this exact marker, not a re-tokenized one, so the
+    # previous region is len(marker) + len(previous).
+    assert plan["previous_marker_token_ids"] == [70, 71]
+    assert plan["replacement_prompt_len"] == 21
+    assert getattr(session, "_minicpmo45_window_previous_len") == 3
+
+
+@pytest.mark.parametrize("cleared_outputs", [False, True])
+@pytest.mark.parametrize("in_flight", [0, 1])
+def test_stage0_window_uses_confirmed_span_and_terminator(cleared_outputs, in_flight) -> None:
+    sched = _make_scheduler(stage_id=0)
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids[:] = session.prompt_token_ids
+    session.num_prompt_tokens = 9
+    session.append_output_token_ids([40, 99])
+    session.num_computed_tokens = 10 + in_flight
+    session.num_output_placeholders = in_flight
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    if cleared_outputs:
+        # A stop without a queued append clears this before the next update.
+        session._output_token_ids.clear()
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+
+    sched._update_request_as_session(session, update)
+
+    plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+    assert plan["completed_token_ids"] == [40]
+    assert plan["completed_terminator_token_id"] == 99
+    assert plan["replacement_prompt_len"] == 9
+    assert plan["dropped_tokens"] == 9
+
+
+def test_stage0_window_rebuild_that_overflows_max_model_len_finishes_the_session() -> None:
+    """A rebuilt window prompt is bounded by the client's window settings, not
+    by the model: a unit carrying camera frames is hundreds of tokens, and a
+    ``context`` window adds the ``previous`` region on top. The replacement
+    branch must therefore check ``replacement_prompt_len`` against
+    ``max_model_len - sample_room`` and finish the session with
+    ``context_length_exceeded``, exactly like a plain extension."""
+    sched = _make_live_session_scheduler(max_model_len=21)
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids[:] = [0] * 9
+    session.num_prompt_tokens = 9
+    session._output_token_ids[:] = [40]
+    session._all_token_ids.append(40)
+    session.num_computed_tokens = 10
+    session.num_output_placeholders = 0
+    _park_session(sched, session)
+    # The parked append state above is the frame before this update: one
+    # confirmed output token, fully computed.
+    session._output_token_ids[:] = [40]
+    session._all_token_ids[:] = [*session.prompt_token_ids, 40]
+    session.num_computed_tokens = 10
+    session.num_in_flight_tokens = 0
+    sched.num_sampled_tokens_per_step = 1
+    # The replacement plan rebuilds 21 tokens (retained context plus the
+    # append) and one step samples one more, so 21 leaves no room.
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+    update.prompt_token_ids = [0] * 20
+    original_prompt = list(session.prompt_token_ids)
+
+    sched._update_request_as_session(session, update)
+
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 21
+    assert session.prompt_token_ids == original_prompt
+    assert session.num_prompt_tokens == 9
+    assert session.status == RequestStatus.FINISHED_ERROR
+    assert session.request_id not in sched.requests
+    assert sched.finished_req_ids == {session.request_id}
+    sched._free_request_blocks.assert_called_once_with(session)
+    client_index, reason = sched._streaming_context_overflow[session.request_id]
+    assert client_index == session.client_index
+    assert reason.startswith("context_length_exceeded: ")
+    assert "21 tokens" in reason and "max_model_len 21" in reason
+
+    engine_core_outputs = _run_idle_step(sched)
+
+    (output,) = engine_core_outputs[session.client_index].outputs
+    assert output.request_id == session.request_id
+    assert output.finish_reason == FinishReason.ERROR
+    assert output.stop_reason == reason
+
+
+def test_stage0_window_rebuild_that_leaves_room_to_sample_replaces_the_prompt() -> None:
+    """One slot above the plan fits: the replacement applies through the normal
+    replacement path."""
+    sched = _make_scheduler(stage_id=0)
+    sched.max_model_len = 25
+    sched.num_sampled_tokens_per_step = 1
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids[:] = session.prompt_token_ids
+    session.num_prompt_tokens = 9
+    session.append_output_token_ids([40])
+    session.num_computed_tokens = 10
+    session.num_output_placeholders = 0
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+    update.prompt_token_ids = [0] * 20
+
+    sched._update_request_as_session(session, update)
+
+    assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 21
+    assert session.prompt_token_ids == [0] * 21
+    assert session.num_prompt_tokens == 21
+    assert session.num_computed_tokens == 0
+    assert session.status == RequestStatus.WAITING
+    sched._free_request_blocks.assert_called_once_with(session)
+    assert not getattr(sched, "_streaming_context_overflow", {})
+
+
+def test_stage0_window_open_start_zero_is_not_replaced_by_the_context_reserve() -> None:
+    """A recorded ``open_start`` of 0 is a legitimate empty context prefix. The
+    fallback must be an explicit ``is None`` check: treating 0 as missing
+    substitutes the context reserve, counts the suffix twice, and the worker's
+    rebuild-length check raises on the first replacement."""
+    session = SimpleNamespace(
+        num_prompt_tokens=9,
+        _all_token_ids=[0] * 9 + [40],
+        num_computed_tokens=10,
+        num_output_placeholders=0,
+        _minicpmo45_window_open_start=0,
+    )
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+    update.prompt_token_ids = [0] * 21
+
+    assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+        session, update, segment_output_ids=[40], completed_terminator=3
+    )
+
+    # open_start stays 0, so the reported unit spans the whole 12-token
+    # boundary instead of the 9 the context reserve would fold in.
+    plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+    assert plan["dropped_tokens"] == 12
+    assert plan["replacement_prompt_len"] == 19
+
+
+def test_stage0_window_open_start_falls_back_to_the_context_reserve_when_unset() -> None:
+    session = SimpleNamespace(
+        num_prompt_tokens=9,
+        _all_token_ids=[0] * 9 + [40],
+        num_computed_tokens=10,
+        num_output_placeholders=0,
+    )
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+    update.prompt_token_ids = [0] * 21
+
+    assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+        session, update, segment_output_ids=[40], completed_terminator=3
+    )
+
+    # fallback open_start is preserve_len 3 from duplex_first_append_context_tokens,
+    # so the unit is three tokens shorter.
+    plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+    assert plan["dropped_tokens"] == 9
+    assert plan["replacement_prompt_len"] == 22
+
+
 def test_explicit_streaming_payload_replaces_placeholder_prompt() -> None:
     sched = _make_scheduler(stage_id=1)
     sched.chunk_transfer_adapter = SimpleNamespace(

tests/e2e/online_serving/helpers/minicpmo_window_e2e.py

@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Drive complete native duplex turns while Stage-0 repeatedly rebuilds KV."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+from pathlib import Path
+
+from vllm_omni.clients.duplex import DuplexClient, EventCollector, read_pcm16_wav, wait_for_condition
+from vllm_omni.clients.minicpmo_4_5 import create_duplex_session_config
+
+
+async def run_window_turn(
+    *,
+    url: str,
+    model: str,
+    input_wav: Path,
+    mode: str,
+    ref_audio: Path | None = None,
+    repeats: int = 2,
+    video_frames: list[str] | None = None,
+    buffered_flush: bool = False,
+    timeout_s: float = 240.0,
+) -> dict[str, object]:
+    """Use aggressive limits to exercise multiple windows in one audio turn."""
+    pcm = read_pcm16_wav(input_wav) * repeats
+    collector = EventCollector()
+    reference = (
+        "data:audio/wav;base64," + base64.b64encode(ref_audio.read_bytes()).decode("ascii")
+        if ref_audio is not None
+        else None
+    )
+    config = create_duplex_session_config(
+        modalities=("text", "audio") if ref_audio is not None else ("text",),
+        ref_audio=reference,
+        temperature=0.0,
+        extra_body={
+            "sliding_window_mode": mode,
+            "context_max_units": 1,
+            "context_previous_max_tokens": 16,
+            "basic_window_high_tokens": 64,
+            "basic_window_low_tokens": 32,
+        },
+    )
+    client = DuplexClient(
+        url, model=model, config=config, reconnect=None, heartbeat_interval_s=None, handshake_timeout_s=timeout_s
+    )
+    async with client:
+        reader = asyncio.create_task(collector.consume(client))
+        collector.add({"type": "session.created", "session": client.session_info})
+        try:
+            frames_sent = await client.stream_pcm(
+                pcm,
+                chunk_ms=2500 if buffered_flush else 200,
+                realtime=not buffered_flush,
+                is_speech=True,
+                video_frames=video_frames,
+            )
+            await client.commit(final=True)
+            await wait_for_condition(
+                lambda: collector.count("input_audio_buffer.committed") > 0 or bool(collector.errors()),
+                timeout_s=timeout_s,
+                label="final input commit",
+            )
+            await wait_for_condition(
+                lambda: collector.count("response.done") > 0 or bool(collector.errors()),
+                timeout_s=timeout_s,
+                label="completed windowed response",
+            )
+            # Final execution follows the commit acknowledgement asynchronously.
+            await asyncio.sleep(3.0)
+            await client.close(timeout_s=timeout_s)
+            await asyncio.wait_for(reader, timeout=5.0)
+        finally:
+            if not reader.done():
+                reader.cancel()
+                await asyncio.gather(reader, return_exceptions=True)
+    return {
+        "created": collector.count("session.created"),
+        "committed": collector.count("input_audio_buffer.committed"),
+        "done": collector.count("response.done"),
+        "closed": collector.count("session.closed"),
+        "errors": collector.errors(),
+        "audio_bytes": len(collector.audio_bytes()),
+        "frames_sent": frames_sent,
+        "input_seconds": len(pcm) / 32000,
+    }

tests/e2e/online_serving/test_minicpmo_4_5_window.py

@@ -0,0 +1,88 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Real-model regression for the scheduler-owned Thinker sliding window."""
+
+import asyncio
+
+import pytest
+
+from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
+    SERVER_PARAMS,
+    duplex_camera_frames,
+    realtime_url,
+    resolve_ref_audio,
+    validated_input_wav,
+)
+from tests.e2e.online_serving.helpers.minicpmo_window_e2e import run_window_turn
+from tests.helpers.mark import hardware_test
+
+pytestmark = [pytest.mark.omni, pytest.mark.advanced_model]
+
+
+def _assert_complete(result, *, require_audio=False):
+    assert result["created"] == 1
+    assert result["committed"] > 0
+    assert result["done"] > 0
+    assert result["closed"] == 1
+    assert result["errors"] == []
+    if require_audio:
+        assert result["audio_bytes"] > 0
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode,reference", [("basic", False), ("context", False), ("context", True)])
+def test_window_rebuild_and_next_session(omni_server, mode, reference):
+    # Two independent sessions exercise admission after final-unit execution.
+    for _ in range(2):
+        result = asyncio.run(
+            run_window_turn(
+                url=realtime_url(omni_server),
+                model=omni_server.model,
+                input_wav=validated_input_wav(),
+                mode=mode,
+                ref_audio=resolve_ref_audio() if reference else None,
+            )
+        )
+        _assert_complete(result, require_audio=reference)
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode,camera", [("basic", False), ("context", False), ("context", True)])
+def test_window_continuous_input(omni_server, tmp_path, mode, camera):
+    # 24 repeats cover over two minutes and many canonical KV replacements.
+    frames = duplex_camera_frames(seconds=4, cache_dir=tmp_path / "camera") if camera else None
+    result = asyncio.run(
+        run_window_turn(
+            url=realtime_url(omni_server),
+            model=omni_server.model,
+            input_wav=validated_input_wav(),
+            mode=mode,
+            repeats=24,
+            ref_audio=resolve_ref_audio(),
+            video_frames=frames,
+        )
+    )
+    _assert_complete(result, require_audio=True)
+    assert isinstance(result["input_seconds"], (int, float)) and result["input_seconds"] > 120
+    if camera:
+        assert isinstance(result["frames_sent"], int) and result["frames_sent"] >= 120
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode", ["off", "basic", "context"])
+def test_window_buffered_flush(omni_server, mode):
+    # Burst appends cross processor chunk boundaries and leave a final tail.
+    result = asyncio.run(
+        run_window_turn(
+            url=realtime_url(omni_server),
+            model=omni_server.model,
+            input_wav=validated_input_wav(),
+            mode=mode,
+            ref_audio=resolve_ref_audio(),
+            buffered_flush=True,
+        )
+    )
+    _assert_complete(result, require_audio=True)

tests/engine/duplex/test_minicpmo_window_plugin.py

@@ -0,0 +1,117 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import base64
+from types import SimpleNamespace
+
+import pytest
+
+from vllm_omni.engine.duplex.config import DuplexSessionConfig
+from vllm_omni.engine.duplex.contracts import DuplexFence
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex import plugin as module
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _plugin():
+    return module.MiniCPMO45DuplexPlugin(lambda *args: None)
+
+
+@pytest.mark.asyncio
+async def test_window_configuration_is_prepared_by_plugin(monkeypatch):
+    tokenizer = SimpleNamespace(
+        all_special_ids=[90],
+        unk_token_id=-1,
+        eos_token_id=99,
+        encode=lambda text, add_special_tokens=False: [ord(char) for char in text],
+        convert_tokens_to_ids=lambda token: {"<|listen|>": 91}.get(token, -1),
+    )
+    monkeypatch.setattr(module, "_load_tokenizer", lambda config: tokenizer)
+    config = DuplexSessionConfig(
+        modalities=("text",),
+        extra_body={
+            "sliding_window_mode": "basic",
+            "basic_window_high_tokens": 120,
+            "basic_window_low_tokens": 80,
+        },
+    )
+    runtime = await _plugin().prepare_runtime_config(config, model_config=None)
+    assert runtime["duplex_window_config"]["sliding_window_mode"] == "basic"
+    assert "sliding_window_mode" not in config.extra_body
+    assert runtime["duplex_window_prefix_tokens"] > 0
+    assert runtime["duplex_window_suffix_token_ids"]
+    assert runtime["duplex_window_previous_marker_token_ids"]
+
+
+@pytest.mark.asyncio
+async def test_invalid_window_configuration_is_rejected():
+    config = DuplexSessionConfig(
+        modalities=("text",),
+        extra_body={
+            "basic_window_high_tokens": 80,
+            "basic_window_low_tokens": 80,
+        },
+    )
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+        await _plugin().prepare_runtime_config(config, model_config=None)
+    assert exc.value.code == "invalid_sliding_window_config"
+
+
+def test_window_update_cannot_change_created_configuration():
+    config = DuplexSessionConfig(extra_body={"sliding_window_mode": "basic"})
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+        _plugin().runtime_config_for_update(config, {"duplex_window_config": {"sliding_window_mode": "off"}})
+    assert exc.value.code == "sliding_window_update_unsupported"
+
+
+def test_window_internal_configuration_is_server_owned():
+    with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError):
+        _plugin().validate_client_extra_body({"duplex_window_config": {}})
+
+
+@pytest.mark.parametrize("final", [False, True])
+@pytest.mark.parametrize("seq,samples,expected", [(1, 32000, 16), (1, 48000, 28), (2, 16000, 13)])
+def test_first_and_final_append_reserve_exact_window_input(seq, samples, expected, final):
+    prompt = module.build_duplex_data_plane_prompt(
+        request_id="window-request",
+        fence=DuplexFence("sid", turn_id=1),
+        session_config={},
+        runtime_config={"duplex_first_append_context_tokens": 5},
+        seq=seq,
+        turn_seq=seq,
+        payload={
+            "audio": base64.b64encode(bytes(samples * 4)).decode(),
+            "format": "pcm_f32le",
+            "sample_rate_hz": 16000,
+        },
+        final=final,
+    )
+    assert len(prompt["prompt_token_ids"]) == expected
+
+
+@pytest.mark.parametrize("seq", [1, 2, 3])
+def test_final_exact_chunk_append_reserves_one_unit_not_two(seq):
+    """A final append whose audio is already an exact number of chunks must
+    reserve exactly what Stage0 feeds: one unit per chunk (``<unit>`` plus its
+    audio) and the closure pair for every unit after the first. Serving pads
+    the final residual itself, so the old extra 12-slot silent-unit
+    reservation only put 12 pad positions ahead of the audio in the KV."""
+
+    def _budget(*, seq, final):
+        prompt = module.build_duplex_data_plane_prompt(
+            request_id="final-request",
+            fence=DuplexFence("sid", turn_id=1),
+            session_config={},
+            runtime_config={"duplex_first_append_context_tokens": 0},
+            seq=seq,
+            turn_seq=seq,
+            payload={
+                "audio": base64.b64encode(bytes(16000 * 4)).decode(),
+                "format": "pcm_f32le",
+                "sample_rate_hz": 16000,
+            },
+            final=final,
+        )
+        return len(prompt["prompt_token_ids"])
+
+    assert _budget(seq=seq, final=True) == _budget(seq=seq, final=False)

tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py

@@ -142,6 +142,37 @@ def test_native_duplex_speak_segment_reaches_split_talker() -> None:
     assert info["duplex"]["turn_id"] == 7
 
 
+@pytest.mark.parametrize("folded_decisions", [0, 1, 2])
+@pytest.mark.parametrize("bos_in_prompt", [False, True])
+def test_native_duplex_tts_bos_aligns_after_window_rebuild(folded_decisions, bos_in_prompt) -> None:
+    prompt_ids = [101] * 10
+    output_ids = [9303] * folded_decisions + [9301, 21, 22, 9308]
+    if bos_in_prompt:
+        prompt_ids[-1] = 9301
+        output_ids = [21, 22, 9308]
+    rows = len(prompt_ids) + len(output_ids) - folded_decisions
+    latent = torch.arange(rows * 4, dtype=torch.float32).reshape(rows, 4)
+    source = _output(
+        prompt_ids=prompt_ids,
+        output_ids=output_ids,
+        latent=latent,
+        multimodal_output={
+            "duplex_prompt_token_ids": prompt_ids,
+            "meta": {
+                "tts_bos_token_id": 9301,
+                "tts_eos_token_id": 9302,
+                "listen_token_id": 9303,
+                "speak_token_id": 9304,
+                "chunk_eos_token_id": 9308,
+            },
+        },
+    )
+    converted = llm2tts([source], prompt=[{}], _streaming_context=SimpleNamespace(bridge_states={}))[0]
+    info = converted["model_intermediate_buffer"]
+    assert info["ids"]["tts"] == [21, 22]
+    torch.testing.assert_close(torch.as_tensor(info["hidden_states"]["tts"]), latent[-3:-1])
+
+
 def test_native_duplex_continuation_appends_only_new_talker_condition() -> None:
     prompt_ids = [101, 102]
     token_ids = {

tests/worker/test_native_duplex_hooks.py

@@ -848,6 +848,176 @@ def test_minicpmo_stage0_data_plane_prefill_matches_official_unit_format():
     assert result["prompt_suffix_len"] == 0
 
 
+@pytest.mark.parametrize("mode", [None, "off", "basic", "context"])
+def test_minicpmo_stage0_history_collection_starts_with_session_config(mode):
+    from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import _MiniCPMO45Stage0SessionState
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="history-lifecycle")
+    config = {} if mode is None else {"duplex_window_config": {"sliding_window_mode": mode}}
+    runtime._prepare_session_context(state, {}, runtime_config=config)
+    for seq in range(1, 101):
+        runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=seq)
+    enabled = mode in {"basic", "context"}
+    assert state.window_enabled is enabled
+    assert len(state.window_units) == (99 if enabled else 0)
+    assert (state.pending_window_unit is not None) is enabled
+    assert state.pending_window_generated_tokens == []
+
+
+@pytest.mark.parametrize("mode", ["basic", "context"])
+@pytest.mark.parametrize("buffered", [False, True])
+def test_minicpmo_stage0_multi_chunk_append_matches_scheduler_window(mode, buffered):
+    from vllm_omni.core.sched.omni_ar_scheduler import OmniARScheduler
+    from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import _MiniCPMO45Stage0SessionState
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="multi-chunk-window", window_enabled=True)
+    if buffered:
+        # Keep the tiny fake processor's initial chunk unpadded to model carry.
+        runtime._pad_first_audio_chunk_if_needed = lambda *args: None
+        assert runtime._stage_prefill_embeddings_only(state, np.zeros(2, dtype=np.float32), seq=0)["success"] is False
+    first = runtime._stage_prefill_embeddings_only(
+        state, np.zeros(6 if buffered else 8, dtype=np.float32), seq=1, final=buffered
+    )
+    assert first["input_token_ids"] == [1, 11, 2, 1, 11]
+    assert state.window_units == []
+    assert state.pending_window_unit.token_ids == first["input_token_ids"]
+    session = SimpleNamespace(num_prompt_tokens=5)
+    replaced = False
+    for seq in (2, 3):
+        info = {
+            "duplex": {
+                "data_plane": True,
+                "seq": seq,
+                "runtime_config": {
+                    "duplex_window_config": {
+                        "sliding_window_mode": mode,
+                        "basic_window_high_tokens": 1,
+                        "basic_window_low_tokens": 2,
+                        "context_max_units": 1,
+                        "context_previous_max_tokens": 16,
+                    },
+                    "duplex_window_previous_marker_token_ids": [201, 5],
+                },
+            }
+        }
+        update = SimpleNamespace(prompt_token_ids=[3, 2, 1, 11], model_intermediate_buffer=info)
+        planned = OmniARScheduler._prepare_minicpmo45_stage0_window(
+            session, update, segment_output_ids=[44], completed_terminator=3
+        )
+        result = runtime._stage_prefill_embeddings_only(
+            state, np.zeros(4, dtype=np.float32), seq=seq, stage0_window=info["duplex"]["stage0_window"]
+        )
+        assert result["stage0_window_replaced"] is planned
+        if planned:
+            assert result["num_input_tokens"] == len(update.prompt_token_ids)
+            replaced = True
+            break
+        session.num_prompt_tokens += 1 + len(update.prompt_token_ids)
+    assert replaced
+
+
+def test_minicpmo_stage0_basic_window_rebuilds_from_retained_units():
+    from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-basic-window", window_enabled=True)
+    first = runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+    assert first["input_token_ids"] == [1, 11]
+    state.pending_window_generated_tokens.append(5)
+    state.pending_terminator_token = 3
+
+    rebuilt = runtime._stage_prefill_embeddings_only(
+        state,
+        np.zeros(4, dtype=np.float32),
+        seq=2,
+        stage0_window={
+            "replace": True,
+            "mode": "basic",
+            "drop_units": 1,
+            "replacement_prompt_len": 2,
+        },
+    )
+
+    assert rebuilt["stage0_window_replaced"] is True
+    assert rebuilt["input_token_ids"] == [1, 11]
+    assert state.window_units == []
+
+
+def test_minicpmo_stage0_context_window_inserts_previous_before_suffix():
+    from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-context-window", window_enabled=True)
+    prefix = runtime._embed_token(200)
+    suffix = runtime._embed_token(202)
+    state.context_embeds = [prefix, suffix]
+    state.context_token_ids = [200, 202]
+    state.context_prefix_embeds = [prefix]
+    state.context_prefix_token_ids = [200]
+    state.context_suffix_embeds = [suffix]
+    state.context_suffix_token_ids = [202]
+    first = runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+    assert first["input_token_ids"] == [200, 202, 1, 11]
+    state.pending_window_generated_tokens.append(42)
+    state.pending_terminator_token = 3
+
+    rebuilt = runtime._stage_prefill_embeddings_only(
+        state,
+        np.zeros(4, dtype=np.float32),
+        seq=2,
+        stage0_window={
+            "replace": True,
+            "mode": "context",
+            "drop_units": 1,
+            "previous_token_ids": [42],
+            "previous_marker_token_ids": [201, 5],
+            "replacement_prompt_len": 7,
+        },
+    )
+
+    assert rebuilt["input_token_ids"] == [200, 201, 5, 42, 202, 1, 11]
+    assert rebuilt["num_input_tokens"] == 7
+
+
+@pytest.mark.parametrize("pending_terminator", [None, 3, 99])
+def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history(pending_terminator):
+    from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+        _MiniCPMO45Stage0SessionState,
+    )
+
+    runtime = _stage0_vision_runtime()
+    runtime._stage_audio_embeddings = lambda *args, **kwargs: torch.zeros((10, 2))
+    state = _MiniCPMO45Stage0SessionState(session_id="sid-accepted-window", window_enabled=True)
+    state.context_prefix_token_ids = [200] * 7
+    state.context_suffix_token_ids = [202]
+    state.context_token_ids = [*state.context_prefix_token_ids, 202]
+    state.context_prefix_embeds = [runtime._embed_token(token_id) for token_id in state.context_prefix_token_ids]
+    state.context_suffix_embeds = [runtime._embed_token(202)]
+    state.context_embeds = [*state.context_prefix_embeds, *state.context_suffix_embeds]
+    runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+    for seq in (2, 3):
+        state.pending_window_generated_tokens = [3, 40, 41]
+        state.pending_terminator_token = pending_terminator
+        plan = {"completed_token_ids": [], "completed_terminator_token_id": 3}
+        if seq == 3:
+            plan.update(replace=True, mode="context", drop_units=1, replacement_prompt_len=32)
+        result = runtime._stage_prefill_embeddings_only(
+            state, np.zeros(4, dtype=np.float32), seq=seq, stage0_window=plan
+        )
+
+    assert result["num_input_tokens"] == 32
+    assert [len(unit.token_ids) for unit in state.window_units] == [13]
+    assert len(state.pending_window_unit.token_ids) == 11
+    assert 40 not in result["input_token_ids"]
+    assert 41 not in result["input_token_ids"]
+
+
 def _stage0_vision_runtime():
     import torch
 

vllm_omni/core/sched/omni_ar_scheduler.py

@@ -848,6 +848,16 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         req_id = session.request_id
         self._new_prompt_len_snapshot[req_id] = len(update.prompt_token_ids)
         outstanding_async_tokens = getattr(session, "num_output_placeholders", 0)
+        # Use the same confirmed span that upstream preserves when extending
+        # a session. The segment output list is cleared on a resumable stop,
+        # and, for an already queued append, can still include its terminator.
+        confirmed_end = session.num_computed_tokens - outstanding_async_tokens
+        segment_output_ids = list(session._all_token_ids[session.num_prompt_tokens : confirmed_end])
+        completed_terminator = (
+            session._all_token_ids[confirmed_end]
+            if session.num_prompt_tokens <= confirmed_end < len(session._all_token_ids)
+            else None
+        )
         # Seed the stale share in SCHEDULED-token units (see the segment-stop
         # site in update_from_output): num_in_flight_tokens matches what each
         # pre-replacement frame will drain, so the counter reaches exactly
@@ -894,6 +904,23 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
                 if self.log_stats:
                     session.record_event(EngineCoreEventType.QUEUED)
                 return
+        if stage_id == 0 and self._prepare_minicpmo45_stage0_window(
+            session,
+            update,
+            segment_output_ids=segment_output_ids,
+            completed_terminator=completed_terminator,
+        ):
+            # The rebuilt prompt is bounded by the client's window settings,
+            # not by the model (a camera unit is hundreds of tokens), so it
+            # needs the same max_model_len check as a plain extension. The
+            # plan replaces the whole prompt, so the replacement length is the
+            # projection, not the session's current prompt plus an extension.
+            plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+            if self._streaming_update_overflows(session, update, projected_len=plan["replacement_prompt_len"]):
+                return
+            self._release_replaced_streaming_prompt_cache(session)
+            self._replace_streaming_session(session, update)
+            return
         streaming_prompt_payload = next(
             (
                 info
@@ -962,11 +989,178 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         if hasattr(update, "model_intermediate_buffer"):
             session.model_intermediate_buffer = update.model_intermediate_buffer
 
+    @staticmethod
+    def _prepare_minicpmo45_stage0_window(
+        session: Request,
+        update: StreamingUpdate,
+        *,
+        segment_output_ids: list[int],
+        completed_terminator: int | None = None,
+    ) -> bool:
+        """Plan an official-style MiniCPM Stage-0 window at unit boundaries.
+
+        vLLM owns a paged KV cache, so deleting a middle span and rotating the
+        retained K tensors in place is not a safe model hook.  Instead, record
+        completed unit lengths in the scheduler and request a full prompt
+        replacement when a watermark fires.  The worker rebuilds matching
+        embeddings from its unit history, which recomputes RoPE at the new
+        contiguous positions.
+        """
+        info = getattr(update, "model_intermediate_buffer", None)
+        if not isinstance(info, dict):
+            return False
+        duplex = info.get("duplex")
+        if not isinstance(duplex, dict) or duplex.get("data_plane") is not True:
+            return False
+        runtime_config = duplex.get("runtime_config")
+        runtime_config = runtime_config if isinstance(runtime_config, dict) else {}
+        window = runtime_config.get("duplex_window_config")
+        if not isinstance(window, dict):
+            return False
+        mode = window.get("sliding_window_mode", "off")
+        if mode == "off":
+            return False
+        if mode not in {"basic", "context"}:
+            return False
+
+        try:
+            seq = int(duplex.get("seq", 0) or 0)
+            preserve_len = int(runtime_config.get("duplex_first_append_context_tokens", 0) or 0)
+        except (TypeError, ValueError):
+            return False
+        if seq <= 1:
+            return False
+
+        # The sampled terminator is discarded by the normal session update and
+        # re-injected at the head of this append, followed by </unit>.
+        # generated_ids contains only the confirmed output span retained by
+        # the streaming update, excluding its final uncomputed sample.
+        generated_ids = segment_output_ids
+        # Sampler-side history can contain speculative or stale async output.
+        # The worker must rebuild from the same accepted ids used for lengths.
+        duplex["stage0_window"] = {"completed_token_ids": list(generated_ids)}
+        if completed_terminator is not None:
+            duplex["stage0_window"]["completed_terminator_token_id"] = int(completed_terminator)
+        base_len = int(getattr(session, "num_prompt_tokens", 0) or 0) + len(generated_ids)
+        boundary = base_len + 2
+        # A legitimate open_start of 0 (an empty context prefix) must not fall
+        # back to preserve_len, which would double-count the suffix below.
+        recorded_open_start = getattr(session, "_minicpmo45_window_open_start", None)
+        open_start = preserve_len if recorded_open_start is None else int(recorded_open_start)
+        unit_len = boundary - open_start
+        if unit_len <= 0:
+            return False
+
+        special_ids = {
+            int(token_id)
+            for token_id in runtime_config.get("duplex_window_special_token_ids", ())
+            if isinstance(token_id, int)
+        }
+        units = list(getattr(session, "_minicpmo45_window_units", ()))
+        units.append(
+            {
+                "length": unit_len,
+                "generated_token_ids": [int(token_id) for token_id in generated_ids if token_id not in special_ids],
+            }
+        )
+        projected_len = base_len + len(update.prompt_token_ids)
+        drop_count = 0
+        dropped_len = 0
+
+        if mode == "basic":
+            try:
+                high = int(window.get("basic_window_high_tokens", 8000))
+                low = int(window.get("basic_window_low_tokens", 6000))
+            except (TypeError, ValueError):
+                return False
+            if projected_len > high:
+                while units and projected_len - dropped_len > low:
+                    dropped_len += int(units[drop_count]["length"])
+                    drop_count += 1
+                    if drop_count >= len(units):
+                        break
+        else:
+            try:
+                max_units = int(window.get("context_max_units", 24))
+                previous_max = int(window.get("context_previous_max_tokens", 500))
+            except (TypeError, ValueError):
+                return False
+            drop_count = max(0, len(units) - max_units)
+            dropped_len = sum(int(unit["length"]) for unit in units[:drop_count])
+            previous = list(getattr(session, "_minicpmo45_window_previous_token_ids", ()))
+            for unit in units[:drop_count]:
+                previous.extend(unit["generated_token_ids"])
+            if len(previous) > previous_max:
+                previous = previous[-previous_max:]
+            marker = [
+                int(token_id)
+                for token_id in runtime_config.get("duplex_window_previous_marker_token_ids", ())
+                if isinstance(token_id, int)
+            ]
+            previous_with_marker = marker + previous if previous else []
+            old_previous_len = int(getattr(session, "_minicpmo45_window_previous_len", 0) or 0)
+            session._minicpmo45_window_previous_token_ids = previous
+            session._minicpmo45_window_previous_len = len(previous_with_marker)
+            projected_len += len(previous_with_marker) - old_previous_len
+
+        retained_units = units[drop_count:]
+        session._minicpmo45_window_units = retained_units
+        if drop_count == 0:
+            session._minicpmo45_window_open_start = boundary
+            return False
+
+        replacement_len = projected_len - dropped_len
+        if replacement_len <= 0:
+            return False
+        scheduler_token_id = runtime_config.get("duplex_scheduler_token_id", 0)
+        try:
+            scheduler_token_id = max(0, int(scheduler_token_id))
+        except (TypeError, ValueError):
+            scheduler_token_id = 0
+        update.prompt_token_ids = [scheduler_token_id] * replacement_len
+        meta = info.setdefault("meta", {})
+        if isinstance(meta, dict):
+            meta["replace_streaming_prompt"] = True
+        previous_ids = list(getattr(session, "_minicpmo45_window_previous_token_ids", ()))
+        # The worker embeds the marker ids the window plan carries: it must not
+        # re-tokenize the marker with its own tokenizer, which could differ
+        # from the ids this length is computed from.
+        marker_ids = [
+            int(token_id)
+            for token_id in runtime_config.get("duplex_window_previous_marker_token_ids", ())
+            if isinstance(token_id, int)
+        ]
+        duplex["stage0_window"] = {
+            "completed_token_ids": list(generated_ids),
+            "replace": True,
+            "mode": mode,
+            "drop_units": drop_count,
+            "dropped_tokens": dropped_len,
+            "previous_token_ids": previous_ids,
+            "previous_marker_token_ids": marker_ids,
+            "replacement_prompt_len": replacement_len,
+        }
+        if completed_terminator is not None:
+            duplex["stage0_window"]["completed_terminator_token_id"] = int(completed_terminator)
+        recorded_prefix_len = runtime_config.get("duplex_window_prefix_tokens")
+        prefix_len = preserve_len if recorded_prefix_len is None else int(recorded_prefix_len)
+        suffix_len = len(runtime_config.get("duplex_window_suffix_token_ids", ()) or ())
+        previous_len = int(getattr(session, "_minicpmo45_window_previous_len", 0) or 0)
+        new_preserve_len = prefix_len + previous_len + suffix_len if mode == "context" else preserve_len
+        session._minicpmo45_window_open_start = new_preserve_len + sum(int(unit["length"]) for unit in retained_units)
+        return True
+
     # Prefix of the stop_reason carried by the FinishReason.ERROR output, so the
     # serving side can map it to a stable error code.
     STREAMING_CONTEXT_OVERFLOW_STOP_REASON = "context_length_exceeded"
 
-    def _streaming_update_overflows(self, session: Request, update: StreamingUpdate) -> bool:
+    def _streaming_update_overflows(
+        self,
+        session: Request,
+        update: StreamingUpdate,
+        *,
+        projected_len: int | None = None,
+    ) -> bool:
         """Finish a streaming session whose next extension cannot fit the model.
 
         Upstream ``_update_request_as_session`` appends the update to the
@@ -991,6 +1185,10 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         finish to the next ``schedule()`` would leave the client waiting. The
         reason is emitted with the terminal output (see
         :meth:`_emit_streaming_context_overflow_outputs`).
+
+        ``projected_len`` overrides the extended-prompt projection for a caller
+        that replaces the prompt instead of growing it, such as the MiniCPM
+        Stage-0 window rebuild.
         """
         max_model_len = getattr(self, "max_model_len", None)
         if max_model_len is None:
@@ -1001,8 +1199,13 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         new_tokens = len(update.prompt_token_ids or ())
         # The extended prompt is the current prompt plus the computed output
         # tokens upstream keeps, then the update: num_computed_tokens covers
-        # both when the prompt was fully computed.
-        projected = max(int(session.num_prompt_tokens), int(session.num_computed_tokens)) + new_tokens
+        # both when the prompt was fully computed. A caller that already knows
+        # the post-update prompt length (a window rebuild) passes it instead.
+        projected = (
+            max(int(session.num_prompt_tokens), int(session.num_computed_tokens)) + new_tokens
+            if projected_len is None
+            else int(projected_len)
+        )
         # Room for the tokens one step samples on top of the prompt (1 without
         # speculative decoding). __new__-built test schedulers carry no
         # num_sampled_tokens_per_step.

vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py

@@ -45,7 +45,10 @@ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.data_plane import (
     MiniCPMO45DataPlaneContext,
     MiniCPMO45DataPlaneSession,
 )
-from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import (
+    MiniCPMO45DuplexPolicy,
+    MiniCPMO45DuplexWindowConfig,
+)
 from vllm_omni.model_executor.models.minicpmo_4_5.duplex.session import (
     MiniCPMO45ServingSessionState,
 )
@@ -85,6 +88,11 @@ PRIVATE_RUNTIME_CONFIG_KEYS = frozenset(
         "ref_audio_format",
         "ref_audio_sample_rate_hz",
         "initial_user_text",
+        "duplex_window_config",
+        "duplex_window_prefix_tokens",
+        "duplex_window_suffix_token_ids",
+        "duplex_window_previous_marker_token_ids",
+        "duplex_window_special_token_ids",
     }
 )
 
@@ -245,8 +253,8 @@ def build_duplex_data_plane_prompt(
             token_budget = context_reserve + first_units * 12 - 1 + vision_tokens
     if seq > 1 and duplex_payload_is_exact_chunks(payload):
         token_budget += 1
-    if final and duplex_payload_is_exact_chunks(payload):
-        token_budget += 12
+        # Serving already pads the final residual audio. Stage0 does not
+        # append another silent unit, so final must not reserve extra slots.
     extra_body = session_config.get("extra_body")
     raw_token_id = runtime_config.get("duplex_scheduler_token_id")
     try:
@@ -515,6 +523,16 @@ def _apply_first_append_context_tokens(
         return
     ref_tokens = MiniCPMO45DuplexPolicy.audio_token_count(ref_sample_count or 0)
     runtime_config["duplex_first_append_context_tokens"] = len(prefix_ids) + ref_tokens + len(suffix_ids)
+    runtime_config["duplex_window_prefix_tokens"] = len(prefix_ids) + ref_tokens
+    runtime_config["duplex_window_suffix_token_ids"] = [int(token_id) for token_id in suffix_ids]
+    marker_ids = tokenizer.encode("\n\nprevious: ", add_special_tokens=False)
+    runtime_config["duplex_window_previous_marker_token_ids"] = [int(token_id) for token_id in marker_ids]
+    runtime_config["duplex_window_special_token_ids"] = sorted(
+        MiniCPMO45DuplexPolicy.native_special_token_ids(
+            MiniCPMO45DuplexPolicy.token_ids_from_tokenizer(tokenizer),
+            tokenizer_special_ids=list(getattr(tokenizer, "all_special_ids", ()) or ()),
+        )
+    )
 
 
 def _model_vision_tile_pixels(model_config: ModelConfig | None) -> int | None:
@@ -722,6 +740,8 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
                 code="unsupported_ref_audio_path",
             )
         runtime_config: dict[str, object] = {"instructions": config.instructions}
+        window_config = self._pop_window_config(extra_body) or MiniCPMO45DuplexWindowConfig()
+        runtime_config["duplex_window_config"] = window_config.as_dict()
         # ``duplex_initial_user_text`` is the older extra_body spelling and
         # still works; the session field is the framework-level one.
         initial_user_text = extra_body.pop("duplex_initial_user_text", None)
@@ -788,6 +808,17 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
         current: Mapping[str, object],
     ) -> dict[str, object]:
         runtime_config = deepcopy(dict(current))
+        extra_body = dict(config.extra_body)
+        requested_window = self._pop_window_config(extra_body)
+        if requested_window is not None:
+            reject_changed_runtime_value(
+                requested_window.as_dict(),
+                runtime_config.get("duplex_window_config"),
+                message="sliding-window configuration cannot be changed after the session is created",
+                code="sliding_window_update_unsupported",
+                error_cls=MiniCPMO45ClientRuntimeConfigError,
+            )
+        config.extra_body = extra_body
         reject_changed_runtime_value(
             config.instructions,
             runtime_config.get("instructions"),
@@ -812,6 +843,23 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
         runtime_config["duplex_stage_sampling_params"] = stage_sampling
         return runtime_config
 
+    @staticmethod
+    def _pop_window_config(extra_body: dict[str, object]) -> MiniCPMO45DuplexWindowConfig | None:
+        names = (
+            "sliding_window_mode",
+            "basic_window_high_tokens",
+            "basic_window_low_tokens",
+            "context_previous_max_tokens",
+            "context_max_units",
+        )
+        provided = {name: extra_body.pop(name) for name in names if name in extra_body}
+        if not provided:
+            return None
+        try:
+            return MiniCPMO45DuplexWindowConfig.from_mapping(provided)
+        except ValueError as exc:
+            raise MiniCPMO45ClientRuntimeConfigError(str(exc), code="invalid_sliding_window_config") from exc
+
     def data_plane_context(
         self,
         *,

vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py

@@ -3,12 +3,63 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from transformers import PreTrainedTokenizerBase
 
 
+@dataclass(frozen=True, slots=True)
+class MiniCPMO45DuplexWindowConfig:
+    """Stage-0 window settings matching the released checkpoint defaults."""
+
+    sliding_window_mode: str = "off"
+    basic_window_high_tokens: int = 8000
+    basic_window_low_tokens: int = 6000
+    context_previous_max_tokens: int = 500
+    context_max_units: int = 24
+
+    @classmethod
+    def from_mapping(cls, value: object) -> MiniCPMO45DuplexWindowConfig:
+        source = value if isinstance(value, dict) else {}
+
+        def integer(name: str, default: int) -> int:
+            raw = source.get(name, default)
+            if isinstance(raw, bool):
+                raise ValueError(f"{name} must be an integer")
+            try:
+                parsed = int(raw)
+            except (TypeError, ValueError) as exc:
+                raise ValueError(f"{name} must be an integer") from exc
+            if parsed <= 0:
+                raise ValueError(f"{name} must be greater than zero")
+            return parsed
+
+        mode = source.get("sliding_window_mode", "off")
+        if mode not in {"off", "basic", "context"}:
+            raise ValueError("sliding_window_mode must be one of: off, basic, context")
+        config = cls(
+            sliding_window_mode=str(mode),
+            basic_window_high_tokens=integer("basic_window_high_tokens", 8000),
+            basic_window_low_tokens=integer("basic_window_low_tokens", 6000),
+            context_previous_max_tokens=integer("context_previous_max_tokens", 500),
+            context_max_units=integer("context_max_units", 24),
+        )
+        if config.basic_window_low_tokens >= config.basic_window_high_tokens:
+            raise ValueError("basic_window_low_tokens must be less than basic_window_high_tokens")
+        return config
+
+    def as_dict(self) -> dict[str, int | str]:
+        return {
+            "sliding_window_mode": self.sliding_window_mode,
+            "basic_window_high_tokens": self.basic_window_high_tokens,
+            "basic_window_low_tokens": self.basic_window_low_tokens,
+            "context_previous_max_tokens": self.context_previous_max_tokens,
+            "context_max_units": self.context_max_units,
+        }
+
+
 class MiniCPMO45DuplexPolicy:
     """MiniCPM-o 4.5 native duplex model policy.
 

vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py

@@ -27,6 +27,12 @@ _MINICPMO45_OPTIONAL_TOKEN_FIELDS = MiniCPMO45DuplexPolicy.OPTIONAL_TOKEN_FIELDS
 _MINICPMO45_PROCESSOR_LOAD_LOCK = Lock()
 
 
+@dataclass
+class _MiniCPMO45WindowUnit:
+    embeds: list[Any] = field(default_factory=list)
+    token_ids: list[int] = field(default_factory=list)
+
+
 @dataclass
 class _MiniCPMO45Stage0SessionState:
     session_id: str
@@ -36,6 +42,14 @@ class _MiniCPMO45Stage0SessionState:
     audio_chunk_idx: int = 0
     context_embeds: list[torch.Tensor] = field(default_factory=list)
     context_token_ids: list[int] = field(default_factory=list)
+    context_prefix_embeds: list[Any] = field(default_factory=list)
+    context_prefix_token_ids: list[int] = field(default_factory=list)
+    context_suffix_embeds: list[Any] = field(default_factory=list)
+    context_suffix_token_ids: list[int] = field(default_factory=list)
+    window_units: list[_MiniCPMO45WindowUnit] = field(default_factory=list)
+    window_enabled: bool = False
+    pending_window_unit: _MiniCPMO45WindowUnit | None = None
+    pending_window_generated_tokens: list[int] = field(default_factory=list)
     current_turn_ended: bool = True
     prepared_append_identity: tuple[int | None, int] | None = None
     prepared_inputs_embeds: torch.Tensor | None = None
@@ -142,6 +156,11 @@ class MiniCPMO45Stage0DuplexRuntime:
         *,
         runtime_config: dict[str, object] | None = None,
     ) -> None:
+        window_config = (runtime_config or {}).get("duplex_window_config")
+        state.window_enabled = isinstance(window_config, dict) and window_config.get("sliding_window_mode", "off") in {
+            "basic",
+            "context",
+        }
         if not self._stage_runtime_ready():
             return
         self._require_special_token_ids()
@@ -157,17 +176,26 @@ class MiniCPMO45Stage0DuplexRuntime:
             (runtime_config or {}).get("initial_user_text"),
         )
         for token_id in self._encode_text(prefix):
-            state.context_embeds.append(self._embed_token(token_id))
+            embed = self._embed_token(token_id)
+            state.context_embeds.append(embed)
             state.context_token_ids.append(token_id)
+            state.context_prefix_embeds.append(embed)
+            state.context_prefix_token_ids.append(token_id)
         if ref_audio is not None:
             ref_audio_embeds = self._stage_ref_audio_embeddings(ref_audio, state=state)
             if ref_audio_embeds is not None:
                 ref_audio_embeds = self._as_2d_tensor(ref_audio_embeds)
+                placeholder_ids = [self.unit_token_id] * int(ref_audio_embeds.shape[0])
                 state.context_embeds.append(ref_audio_embeds)
-                state.context_token_ids.extend([self.unit_token_id] * int(ref_audio_embeds.shape[0]))
+                state.context_token_ids.extend(placeholder_ids)
+                state.context_prefix_embeds.append(ref_audio_embeds)
+                state.context_prefix_token_ids.extend(placeholder_ids)
         for token_id in self._encode_text(suffix):
-            state.context_embeds.append(self._embed_token(token_id))
+            embed = self._embed_token(token_id)
+            state.context_embeds.append(embed)
             state.context_token_ids.append(token_id)
+            state.context_suffix_embeds.append(embed)
+            state.context_suffix_token_ids.append(token_id)
 
     def _stage_prefill_embeddings_only(
         self,
@@ -179,6 +207,7 @@ class MiniCPMO45Stage0DuplexRuntime:
         seq: int | None = None,
         is_speech: bool = False,
         final: bool = False,
+        stage0_window: dict[str, object] | None = None,
     ) -> dict[str, object]:
         """Build scheduler-owned Stage0 input embeddings for one audio append.
 
@@ -199,6 +228,16 @@ class MiniCPMO45Stage0DuplexRuntime:
             result["input_token_ids"] = list(state.prepared_input_token_ids)
             return result
         self._require_special_token_ids()
+        if isinstance(stage0_window, dict):
+            completed_ids = stage0_window.get("completed_token_ids")
+            if isinstance(completed_ids, list):
+                state.pending_window_generated_tokens = [int(token_id) for token_id in completed_ids]
+            completed_terminator = stage0_window.get("completed_terminator_token_id")
+            if isinstance(completed_terminator, int):
+                # Async sampling can overwrite the worker's pending token
+                # after the scheduler accepted a segment stop. Rebuild from
+                # the scheduler's canonical boundary, just like its content.
+                state.pending_terminator_token = completed_terminator
         if audio_waveform is None or len(audio_waveform) == 0:
             return self._stage_prefill_result(False, start_time, "empty audio")
         state.audio_buffer = np.concatenate([state.audio_buffer, np.asarray(audio_waveform, dtype=np.float32)])
@@ -279,12 +318,25 @@ class MiniCPMO45Stage0DuplexRuntime:
                 # the closure; the model's listen/speak policy depends on
                 # seeing its own past decisions in context.
                 pending_terminator = state.pending_terminator_token
+                closure_token_ids: list[int] = []
                 if pending_terminator is not None and units_built == 0:
                     state.pending_terminator_token = None
                     embed_parts.append(self._embed_token(pending_terminator))
                     token_ids.append(int(pending_terminator))
+                    closure_token_ids.append(int(pending_terminator))
                 embed_parts.append(self._embed_token(self.unit_end_token_id))
                 token_ids.append(self.unit_end_token_id)
+                closure_token_ids.append(self.unit_end_token_id)
+                if units_built == 0:
+                    self._finalize_window_unit(state, closure_token_ids)
+                elif state.pending_window_unit is not None:
+                    # Internal processor chunks belong to this same append.
+                    state.pending_window_unit.embeds.extend(
+                        self._embed_token(token_id) for token_id in closure_token_ids
+                    )
+                    state.pending_window_unit.token_ids.extend(closure_token_ids)
+            unit_embed_start = len(embed_parts)
+            unit_token_start = len(token_ids)
             embed_parts.append(self._embed_token(self.unit_token_id))
             token_ids.append(self.unit_token_id)
             if frame_blocks:
@@ -311,6 +363,11 @@ class MiniCPMO45Stage0DuplexRuntime:
             state.audio_buffer = state.audio_buffer[consumed_samples:]
             state.audio_chunk_idx += 1
             units_built += 1
+            if state.window_enabled:
+                if state.pending_window_unit is None:
+                    state.pending_window_unit = _MiniCPMO45WindowUnit()
+                state.pending_window_unit.embeds.extend(embed_parts[unit_embed_start:])
+                state.pending_window_unit.token_ids.extend(token_ids[unit_token_start:])
             chunk_size = self._streaming_chunk_size(processor)
         # 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
@@ -320,6 +377,9 @@ class MiniCPMO45Stage0DuplexRuntime:
 
         import torch
 
+        window_result = self._window_replacement_parts(state, stage0_window)
+        if window_result is not None:
+            embed_parts, token_ids = window_result
         inputs_embeds = torch.cat([self._as_2d_tensor(embed) for embed in embed_parts], dim=0)
         result = self._stage_prefill_result(True, start_time)
         result.update(
@@ -332,6 +392,7 @@ class MiniCPMO45Stage0DuplexRuntime:
                 "uses_model_runner_scheduler": True,
                 "runner_kv_backed": True,
                 "runtime_impl": "scheduler_data_plane",
+                "stage0_window_replaced": window_result is not None,
             }
         )
         if is_speech and (append_identity is None or state.pending_speech_append_identity != append_identity):
@@ -344,6 +405,85 @@ class MiniCPMO45Stage0DuplexRuntime:
             state.prepared_result = {k: v for k, v in result.items() if k not in {"inputs_embeds", "input_token_ids"}}
         return result
 
+    def _finalize_window_unit(
+        self,
+        state: _MiniCPMO45Stage0SessionState,
+        closure_token_ids: list[int],
+    ) -> None:
+        pending = state.pending_window_unit
+        if not state.window_enabled or pending is None:
+            state.pending_window_unit = None
+            state.pending_window_generated_tokens.clear()
+            return
+        generated = list(state.pending_window_generated_tokens)
+        token_ids = [*pending.token_ids, *generated, *closure_token_ids]
+        embeds = list(pending.embeds)
+        embeds.extend(self._embed_token(token_id) for token_id in generated)
+        embeds.extend(self._embed_token(token_id) for token_id in closure_token_ids)
+        state.window_units.append(_MiniCPMO45WindowUnit(embeds=embeds, token_ids=token_ids))
+        state.pending_window_unit = None
+        state.pending_window_generated_tokens.clear()
+
+    def _window_replacement_parts(
+        self,
+        state: _MiniCPMO45Stage0SessionState,
+        stage0_window: dict[str, Any] | None,
+    ) -> tuple[list[Any], list[int]] | None:
+        if not isinstance(stage0_window, dict) or stage0_window.get("replace") is not True:
+            return None
+        try:
+            drop_units = max(0, int(stage0_window.get("drop_units", 0)))
+        except (TypeError, ValueError):
+            return None
+        if drop_units > len(state.window_units):
+            raise RuntimeError(
+                "MiniCPM-o Stage-0 window history is shorter than the scheduler drop plan: "
+                f"drop={drop_units}, completed={len(state.window_units)}"
+            )
+        del state.window_units[:drop_units]
+
+        mode = stage0_window.get("mode")
+        embeds: list[Any] = []
+        token_ids: list[int] = []
+        previous_ids: list[int] = []
+        if mode == "context":
+            embeds.extend(state.context_prefix_embeds)
+            token_ids.extend(state.context_prefix_token_ids)
+            previous = stage0_window.get("previous_token_ids")
+            previous_ids = [int(token_id) for token_id in previous] if isinstance(previous, list) else []
+            if previous_ids:
+                # The plan carries the marker ids the scheduler sized the
+                # `previous` region from. Re-tokenizing here would be a second
+                # source of truth and could disagree with that length.
+                marker = stage0_window.get("previous_marker_token_ids")
+                marker_ids = [int(token_id) for token_id in marker] if isinstance(marker, list) else []
+                embeds.extend(self._embed_token(token_id) for token_id in [*marker_ids, *previous_ids])
+                token_ids.extend(marker_ids)
+                token_ids.extend(previous_ids)
+            embeds.extend(state.context_suffix_embeds)
+            token_ids.extend(state.context_suffix_token_ids)
+        else:
+            embeds.extend(state.context_embeds)
+            token_ids.extend(state.context_token_ids)
+        for unit in state.window_units:
+            embeds.extend(unit.embeds)
+            token_ids.extend(unit.token_ids)
+        if state.pending_window_unit is not None:
+            embeds.extend(state.pending_window_unit.embeds)
+            token_ids.extend(state.pending_window_unit.token_ids)
+        expected = stage0_window.get("replacement_prompt_len")
+        if isinstance(expected, int) and expected != len(token_ids):
+            unit_lengths = [len(unit.token_ids) for unit in state.window_units]
+            pending_len = len(state.pending_window_unit.token_ids) if state.pending_window_unit is not None else 0
+            raise RuntimeError(
+                "MiniCPM-o Stage-0 window rebuild length mismatch: "
+                f"worker={len(token_ids)}, scheduler={expected}, mode={mode}, "
+                f"prefix={len(state.context_prefix_token_ids)}, "
+                f"suffix={len(state.context_suffix_token_ids)}, previous={len(previous_ids)}, "
+                f"units={unit_lengths}, pending={pending_len}, drop={drop_units}"
+            )
+        return embeds, token_ids
+
     @staticmethod
     def _stage_prefill_result(success: bool, start_time: float, reason: str = "") -> dict[str, object]:
         return {

vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni.py

@@ -371,8 +371,18 @@ class MiniCPMO45OmniForConditionalGeneration(nn.Module, SupportsMultiModal, Supp
             seq=seq,
             is_speech=bool(payload.get("is_speech", False)),
             final=bool(duplex.get("final")),
+            stage0_window=(duplex.get("stage0_window") if isinstance(duplex.get("stage0_window"), dict) else None),
         )
         update_result = dict(result)
+        if result.get("stage0_window_replaced") is True:
+            window = duplex.get("stage0_window", {})
+            logger.info(
+                "MiniCPM-o Stage-0 window replaced: mode=%s drop_units=%s tokens=%s seq=%s",
+                window.get("mode"),
+                window.get("drop_units"),
+                result.get("num_input_tokens"),
+                seq,
+            )
         update_result.pop("inputs_embeds", None)
         if result.get("success") is not True:
             embeds = input_embeds if input_embeds is not None else self.get_input_embeddings(input_ids)

vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py

@@ -865,8 +865,25 @@ def llm2tts(
                     special_token_ids.get("chunk_eos_token_id"),
                     special_token_ids.get("chunk_tts_eos_token_id"),
                 }
-            tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
-            tts_hidden_slice = thinker_hidden_states[tts_bos_idx:end_idx].to(torch.float32).contiguous()
+            if is_native_duplex_handoff:
+                # Earlier unforwarded decisions may already be folded into
+                # the rebuilt prompt while still appearing in this delta.
+                # Align the explicit tts_bos path by the segment's end, just
+                # like the native speak/text paths below, rather than adding
+                # those decisions to the prompt a second time.
+                out_start = tts_bos_idx - prompt_token_ids_len
+                out_end = tts_eos_idx - prompt_token_ids_len if tts_eos_idx is not None else len(llm_output_ids)
+                hidden_base = int(thinker_hidden_states.shape[0]) - len(llm_output_ids)
+                if hidden_base >= 0 and out_end > out_start:
+                    tts_token_ids_slice = torch.tensor(llm_output_ids[out_start:out_end], dtype=torch.long)
+                    tts_hidden_slice = (
+                        thinker_hidden_states[hidden_base + out_start : hidden_base + out_end]
+                        .to(torch.float32)
+                        .contiguous()
+                    )
+            else:
+                tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
+                tts_hidden_slice = thinker_hidden_states[tts_bos_idx:end_idx].to(torch.float32).contiguous()
         elif is_native_duplex_handoff:
             # Official MiniCPM-o duplex does not prefill an assistant
             # <|tts_bos|> boundary before generation. A segment delta can