From b9b5769f61a8af85105c91804ff5982050402f05 Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Wed, 16 Sep 2026 12:22:05 +0800
Subject: [PATCH 1/8] feat: add MiniCPM-o Stage-0 sliding window

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../sched/test_omni_ar_scheduler_streaming.py |  94 +++++++++++
 .../duplex/test_minicpmo_window_plugin.py     |  89 ++++++++++
 tests/worker/test_native_duplex_hooks.py      |  98 +++++++++++
 vllm_omni/core/sched/omni_ar_scheduler.py     | 155 ++++++++++++++++++
 .../models/minicpmo_4_5/duplex/plugin.py      |  54 +++++-
 .../models/minicpmo_4_5/duplex/policy.py      |  51 ++++++
 .../models/minicpmo_4_5/duplex/stage0.py      | 121 +++++++++++++-
 .../models/minicpmo_4_5/minicpmo_4_5_omni.py  |  10 ++
 8 files changed, 666 insertions(+), 6 deletions(-)
 create mode 100644 tests/engine/duplex/test_minicpmo_window_plugin.py

diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py
index bca7d095c19..0ef2825749f 100644
--- a/tests/core/sched/test_omni_ar_scheduler_streaming.py
+++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py
@@ -65,6 +65,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,
@@ -506,6 +539,67 @@ 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": [],
+        "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]
+    assert plan["replacement_prompt_len"] == 21
+    assert getattr(session, "_minicpmo45_window_previous_len") == 3
+
+
 def test_explicit_streaming_payload_replaces_placeholder_prompt() -> None:
     sched = _make_scheduler(stage_id=1)
     sched.chunk_transfer_adapter = SimpleNamespace(
diff --git a/tests/engine/duplex/test_minicpmo_window_plugin.py b/tests/engine/duplex/test_minicpmo_window_plugin.py
new file mode 100644
index 00000000000..9b3bae2589d
--- /dev/null
+++ b/tests/engine/duplex/test_minicpmo_window_plugin.py
@@ -0,0 +1,89 @@
+# 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
diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index 388e334ccac..2cb92e9a3bc 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -848,6 +848,104 @@ def get_audio_hidden_states(self, _data):
     assert result["prompt_suffix_len"] == 0
 
 
+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")
+    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")
+    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],
+            "replacement_prompt_len": 7,
+        },
+    )
+
+    assert rebuilt["input_token_ids"] == [200, 201, 5, 42, 202, 1, 11]
+    assert rebuilt["num_input_tokens"] == 7
+
+
+def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history():
+    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")
+    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 = 3
+        plan = {"completed_token_ids": []}
+        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
 
diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py
index 329e8ad8190..0ae13cd6ca8 100644
--- a/vllm_omni/core/sched/omni_ar_scheduler.py
+++ b/vllm_omni/core/sched/omni_ar_scheduler.py
@@ -777,6 +777,9 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
         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)
+        segment_output_ids = list(getattr(session, "_output_token_ids", ()))
+        if outstanding_async_tokens > 0:
+            segment_output_ids = segment_output_ids[:-outstanding_async_tokens]
         # 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
@@ -823,6 +826,14 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
                 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,
+        ):
+            self._release_replaced_streaming_prompt_cache(session)
+            self._replace_streaming_session(session, update)
+            return
         streaming_prompt_payload = next(
             (
                 info
@@ -888,6 +899,150 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
         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],
+    ) -> 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>.
+        # Stop tokens are consumed by vLLM's stop handling before a resumable
+        # segment is updated.  The remaining row-local output ids are the
+        # generated content that official MiniCPM records for this unit.
+        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)}
+        base_len = int(getattr(session, "num_prompt_tokens", 0) or 0) + len(generated_ids)
+        boundary = base_len + 2
+        open_start = int(getattr(session, "_minicpmo45_window_open_start", preserve_len) or preserve_len)
+        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", ()))
+        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,
+            "replacement_prompt_len": replacement_len,
+        }
+        prefix_len = int(runtime_config.get("duplex_window_prefix_tokens", preserve_len) or preserve_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
+
     def _free_request(
         self, request: Request, delay_free_blocks: bool = False
     ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py
index 6736404610d..3b5023e9f37 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/plugin.py
@@ -45,7 +45,10 @@
     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,
 )
@@ -75,6 +78,11 @@
         "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",
     }
 )
 
@@ -194,8 +202,8 @@ def build_duplex_data_plane_prompt(
             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):
-        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:
@@ -464,6 +472,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 _apply_default_scheduler_policy(
@@ -645,6 +663,8 @@ async def prepare_runtime_config(
                 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)
@@ -711,6 +731,17 @@ def runtime_config_for_update(
         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"),
@@ -735,6 +766,23 @@ def runtime_config_for_update(
         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,
         *,
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py
index 0332c81c0a1..7b0aded4dae 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/policy.py
+++ b/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.
 
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
index bc13b58f2ca..048a9ee4ef6 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
@@ -27,6 +27,12 @@
 _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,13 @@ 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)
+    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
@@ -157,17 +170,26 @@ def _prepare_session_context(
             (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 +201,7 @@ def _stage_prefill_embeddings_only(
         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 +222,10 @@ def _stage_prefill_embeddings_only(
             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]
         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 +306,18 @@ def _stage_prefill_embeddings_only(
                 # 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)
+                self._finalize_window_unit(state, 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 +344,10 @@ def _stage_prefill_embeddings_only(
             state.audio_buffer = state.audio_buffer[consumed_samples:]
             state.audio_chunk_idx += 1
             units_built += 1
+            state.pending_window_unit = _MiniCPMO45WindowUnit(
+                embeds=list(embed_parts[unit_embed_start:]),
+                token_ids=list(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 +357,9 @@ def _stage_prefill_embeddings_only(
 
         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 +372,7 @@ def _stage_prefill_embeddings_only(
                 "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 +385,80 @@ def _stage_prefill_embeddings_only(
             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 pending is 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:
+                marker_ids = self._encode_text("\n\nprevious: ")
+                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 {
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni.py b/vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni.py
index b1c8db89c6e..03024bbe586 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/minicpmo_4_5_omni.py
@@ -371,8 +371,18 @@ def preprocess(
             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)

From 6817745cef38c5b77187b552c7bf050fdadc02d3 Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Wed, 16 Sep 2026 20:46:13 +0800
Subject: [PATCH 2/8] fix: preserve confirmed MiniCPM window boundaries

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../sched/test_omni_ar_scheduler_streaming.py | 26 +++++++++++++++++++
 tests/worker/test_native_duplex_hooks.py      |  7 ++---
 vllm_omni/core/sched/omni_ar_scheduler.py     | 24 ++++++++++++-----
 .../models/minicpmo_4_5/duplex/stage0.py      |  6 +++++
 4 files changed, 54 insertions(+), 9 deletions(-)

diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py
index 0ef2825749f..29846e27e8f 100644
--- a/tests/core/sched/test_omni_ar_scheduler_streaming.py
+++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py
@@ -600,6 +600,32 @@ def test_stage0_context_window_compacts_dropped_speech() -> None:
     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_explicit_streaming_payload_replaces_placeholder_prompt() -> None:
     sched = _make_scheduler(stage_id=1)
     sched.chunk_transfer_adapter = SimpleNamespace(
diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index 2cb92e9a3bc..c34fe286ab4 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -914,7 +914,8 @@ def test_minicpmo_stage0_context_window_inserts_previous_before_suffix():
     assert rebuilt["num_input_tokens"] == 7
 
 
-def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history():
+@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,
     )
@@ -931,8 +932,8 @@ def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history()
     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 = 3
-        plan = {"completed_token_ids": []}
+        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(
diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py
index 0ae13cd6ca8..119b3782b0d 100644
--- a/vllm_omni/core/sched/omni_ar_scheduler.py
+++ b/vllm_omni/core/sched/omni_ar_scheduler.py
@@ -777,9 +777,16 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
         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)
-        segment_output_ids = list(getattr(session, "_output_token_ids", ()))
-        if outstanding_async_tokens > 0:
-            segment_output_ids = segment_output_ids[:-outstanding_async_tokens]
+        # 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
@@ -830,6 +837,7 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
             session,
             update,
             segment_output_ids=segment_output_ids,
+            completed_terminator=completed_terminator,
         ):
             self._release_replaced_streaming_prompt_cache(session)
             self._replace_streaming_session(session, update)
@@ -905,6 +913,7 @@ def _prepare_minicpmo45_stage0_window(
         update: StreamingUpdate,
         *,
         segment_output_ids: list[int],
+        completed_terminator: int | None = None,
     ) -> bool:
         """Plan an official-style MiniCPM Stage-0 window at unit boundaries.
 
@@ -942,13 +951,14 @@ def _prepare_minicpmo45_stage0_window(
 
         # The sampled terminator is discarded by the normal session update and
         # re-injected at the head of this append, followed by </unit>.
-        # Stop tokens are consumed by vLLM's stop handling before a resumable
-        # segment is updated.  The remaining row-local output ids are the
-        # generated content that official MiniCPM records for this 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
         open_start = int(getattr(session, "_minicpmo45_window_open_start", preserve_len) or preserve_len)
@@ -1036,6 +1046,8 @@ def _prepare_minicpmo45_stage0_window(
             "previous_token_ids": previous_ids,
             "replacement_prompt_len": replacement_len,
         }
+        if completed_terminator is not None:
+            duplex["stage0_window"]["completed_terminator_token_id"] = int(completed_terminator)
         prefix_len = int(runtime_config.get("duplex_window_prefix_tokens", preserve_len) or preserve_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)
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
index 048a9ee4ef6..f27e8f437b0 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
@@ -226,6 +226,12 @@ def _stage_prefill_embeddings_only(
             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)])

From f8e646960dea18d1d83785a9c7acf06a7daff9c1 Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Wed, 16 Sep 2026 21:50:54 +0800
Subject: [PATCH 3/8] fix: align MiniCPM window TTS handoffs and cover live
 sessions

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../common/ci_source_file_dependencies.yml    |  2 +
 .buildkite/cuda/test-merge.yml                |  2 +-
 .../helpers/minicpmo_window_e2e.py            | 88 +++++++++++++++++++
 .../test_minicpmo_4_5_window.py               | 70 +++++++++++++++
 .../test_minicpmo_4_5_omni.py                 | 31 +++++++
 .../minicpmo_4_5_omni.py                      | 21 ++++-
 6 files changed, 211 insertions(+), 3 deletions(-)
 create mode 100644 tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
 create mode 100644 tests/e2e/online_serving/test_minicpmo_4_5_window.py

diff --git a/.buildkite/common/ci_source_file_dependencies.yml b/.buildkite/common/ci_source_file_dependencies.yml
index e29001d9a46..d6781af9d01 100644
--- a/.buildkite/common/ci_source_file_dependencies.yml
+++ b/.buildkite/common/ci_source_file_dependencies.yml
@@ -243,6 +243,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
diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml
index 87d4c536231..d4866dc4a7e 100644
--- a/.buildkite/cuda/test-merge.yml
+++ b/.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"
diff --git a/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py b/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
new file mode 100644
index 00000000000..4025eb6c33b
--- /dev/null
+++ b/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
@@ -0,0 +1,88 @@
+# 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,
+    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=200,
+                realtime=True,
+                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,
+    }
diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_window.py b/tests/e2e/online_serving/test_minicpmo_4_5_window.py
new file mode 100644
index 00000000000..b947a58f97e
--- /dev/null
+++ b/tests/e2e/online_serving/test_minicpmo_4_5_window.py
@@ -0,0 +1,70 @@
+# 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 result["input_seconds"] > 120
+    if camera:
+        assert result["frames_sent"] >= 120
diff --git a/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py b/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py
index 27037f02957..c5d4d0cde0b 100644
--- a/tests/model_executor/stage_input_processors/test_minicpmo_4_5_omni.py
+++ b/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 = {
diff --git a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py
index 1c374be8552..5af96011e68 100644
--- a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py
+++ b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py
@@ -862,8 +862,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

From f918052dbcdc9c4e9af5e961bfde8154f83c2cae Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Thu, 17 Sep 2026 03:33:24 +0000
Subject: [PATCH 4/8] Fix MiniCPM Stage-0 window history lifecycle and append
 boundaries

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../helpers/minicpmo_window_e2e.py            |  5 +-
 .../test_minicpmo_4_5_window.py               | 22 +++++-
 tests/worker/test_native_duplex_hooks.py      | 76 ++++++++++++++++++-
 .../models/minicpmo_4_5/duplex/stage0.py      | 27 +++++--
 4 files changed, 117 insertions(+), 13 deletions(-)

diff --git a/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py b/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
index 4025eb6c33b..553f0799219 100644
--- a/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
+++ b/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
@@ -21,6 +21,7 @@ async def run_window_turn(
     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."""
@@ -52,8 +53,8 @@ async def run_window_turn(
         try:
             frames_sent = await client.stream_pcm(
                 pcm,
-                chunk_ms=200,
-                realtime=True,
+                chunk_ms=2500 if buffered_flush else 200,
+                realtime=not buffered_flush,
                 is_speech=True,
                 video_frames=video_frames,
             )
diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_window.py b/tests/e2e/online_serving/test_minicpmo_4_5_window.py
index b947a58f97e..300205eb043 100644
--- a/tests/e2e/online_serving/test_minicpmo_4_5_window.py
+++ b/tests/e2e/online_serving/test_minicpmo_4_5_window.py
@@ -65,6 +65,24 @@ def test_window_continuous_input(omni_server, tmp_path, mode, camera):
         )
     )
     _assert_complete(result, require_audio=True)
-    assert result["input_seconds"] > 120
+    assert isinstance(result["input_seconds"], (int, float)) and result["input_seconds"] > 120
     if camera:
-        assert result["frames_sent"] >= 120
+        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)
diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index c34fe286ab4..14656a1a4fb 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -848,13 +848,83 @@ def get_audio_hidden_states(self, _data):
     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")
+    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)
@@ -883,7 +953,7 @@ def test_minicpmo_stage0_context_window_inserts_previous_before_suffix():
     )
 
     runtime = _stage0_vision_runtime()
-    state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-context-window")
+    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]
@@ -922,7 +992,7 @@ def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history(p
 
     runtime = _stage0_vision_runtime()
     runtime._stage_audio_embeddings = lambda *args, **kwargs: torch.zeros((10, 2))
-    state = _MiniCPMO45Stage0SessionState(session_id="sid-accepted-window")
+    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]
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
index f27e8f437b0..101a005a0de 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
@@ -47,6 +47,7 @@ class _MiniCPMO45Stage0SessionState:
     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
@@ -155,6 +156,11 @@ def _prepare_session_context(
         *,
         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()
@@ -321,7 +327,14 @@ def _stage_prefill_embeddings_only(
                 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)
-                self._finalize_window_unit(state, closure_token_ids)
+                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))
@@ -350,10 +363,11 @@ def _stage_prefill_embeddings_only(
             state.audio_buffer = state.audio_buffer[consumed_samples:]
             state.audio_chunk_idx += 1
             units_built += 1
-            state.pending_window_unit = _MiniCPMO45WindowUnit(
-                embeds=list(embed_parts[unit_embed_start:]),
-                token_ids=list(token_ids[unit_token_start:]),
-            )
+            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
@@ -397,7 +411,8 @@ def _finalize_window_unit(
         closure_token_ids: list[int],
     ) -> None:
         pending = state.pending_window_unit
-        if pending is None:
+        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)

From 4dd174f2dfacf372e8e1855ec4da7b1fe9e4351d Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Thu, 17 Sep 2026 17:06:10 +0800
Subject: [PATCH 5/8] [Bugfix] Check a rebuilt MiniCPM window prompt against
 max_model_len

The Stage-0 window replacement branch returned before
``_streaming_update_overflows``, so the rebuilt prompt was bounded only by
the client's window settings: in ``context`` mode it is prefix + marker +
``previous`` + up to ``context_max_units`` retained units (a camera unit is
hundreds of tokens) + the new append, and in ``basic`` mode it is
``basic_window_low_tokens`` + the append, with no upper bound beyond
``low < high``. A prompt over ``max_model_len - sample_room`` reached the
worker instead of finishing the session with ``context_length_exceeded``.

The plan already carries ``replacement_prompt_len``, so the guard now takes
that as an explicit projection: a replacement is not the session's current
prompt plus an extension. Two scheduler regressions cover the overflow and
the one-slot-larger replacement that still applies.

Also stop treating a legitimate zero ``_minicpmo45_window_open_start`` (an
empty context prefix) as missing: ``or preserve_len`` substituted the whole
first-append context length, which double-counts the suffix and makes the
worker's rebuild-length check raise on the first replacement. Same for the
recorded ``duplex_window_prefix_tokens``.

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../sched/test_omni_ar_scheduler_streaming.py | 109 ++++++++++++++++++
 vllm_omni/core/sched/omni_ar_scheduler.py     |  47 +++++++-
 2 files changed, 151 insertions(+), 5 deletions(-)

diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py
index a827d003c3d..383e8c05664 100644
--- a/tests/core/sched/test_omni_ar_scheduler_streaming.py
+++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py
@@ -567,6 +567,7 @@ def test_stage0_basic_window_rebuilds_below_low_watermark() -> None:
         "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)
@@ -599,6 +600,9 @@ def test_stage0_context_window_compacts_dropped_speech() -> None:
     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
 
@@ -629,6 +633,111 @@ def test_stage0_window_uses_confirmed_span_and_terminator(cleared_outputs, in_fl
     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. When that exceeds
+    ``max_model_len`` the replacement branch must finish the session with the
+    same explicit overflow error as a plain extension, not hand an
+    over-long prompt to the worker."""
+    sched = _make_live_session_scheduler(max_model_len=25)
+    session = _make_request()
+    session.prompt_token_ids = [0] * 9
+    session._all_token_ids[:] = [0] * 9
+    session._output_token_ids[:] = [40]
+    session._all_token_ids.append(40)
+    session.num_prompt_tokens = 9
+    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
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+    original_prompt = list(session.prompt_token_ids)
+
+    sched._update_request_as_session(session, update)
+
+    # The plan's own length (9 retained context + the 8-slot append + the 2
+    # closure tokens + the append's 6-token window = 25) is the projection,
+    # not the session prompt plus an extension.
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 25
+    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 "25 tokens" in reason and "max_model_len 25" 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 = 26
+    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")
+
+    sched._update_request_as_session(session, update)
+
+    assert session.prompt_token_ids == [0] * 25
+    assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 25
+    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, _minicpmo45_window_open_start=0)
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+
+    assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+        session, update, segment_output_ids=[40], completed_terminator=3
+    )
+
+    assert session._minicpmo45_window_units == [{"length": 10, "generated_token_ids": [40]}]
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["dropped_tokens"] == 10
+
+
+def test_stage0_window_open_start_falls_back_to_the_context_reserve_when_unset() -> None:
+    session = SimpleNamespace(num_prompt_tokens=9)
+    update = _make_minicpm_window_update(seq=2, mode="basic")
+
+    assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+        session, update, segment_output_ids=[40], completed_terminator=3
+    )
+
+    # preserve_len 3 from duplex_first_append_context_tokens.
+    assert session._minicpmo45_window_units == [{"length": 8, "generated_token_ids": [40]}]
+    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["dropped_tokens"] == 8
+
+
 def test_explicit_streaming_payload_replaces_placeholder_prompt() -> None:
     sched = _make_scheduler(stage_id=1)
     sched.chunk_transfer_adapter = SimpleNamespace(
diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py
index ebafdb78acc..685b1d67ca8 100644
--- a/vllm_omni/core/sched/omni_ar_scheduler.py
+++ b/vllm_omni/core/sched/omni_ar_scheduler.py
@@ -845,6 +845,14 @@ def _update_request_as_session(self, session: Request, update: StreamingUpdate)
             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
@@ -969,7 +977,10 @@ def _prepare_minicpmo45_stage0_window(
             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
-        open_start = int(getattr(session, "_minicpmo45_window_open_start", preserve_len) or preserve_len)
+        # 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
@@ -1045,6 +1056,14 @@ def _prepare_minicpmo45_stage0_window(
         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,
@@ -1052,21 +1071,30 @@ def _prepare_minicpmo45_stage0_window(
             "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)
-        prefix_len = int(runtime_config.get("duplex_window_prefix_tokens", preserve_len) or preserve_len)
+        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
@@ -1091,6 +1119,10 @@ def _streaming_update_overflows(self, session: Request, update: StreamingUpdate)
         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:
@@ -1101,8 +1133,13 @@ def _streaming_update_overflows(self, session: Request, update: StreamingUpdate)
         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.

From 6d393371f8ad98cdf63b41e70f181ba84fa66729 Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Thu, 17 Sep 2026 17:06:13 +0800
Subject: [PATCH 6/8] [Refactor] Carry the MiniCPM window previous-marker ids
 in the plan

The worker re-tokenized "\n\nprevious: " with its own tokenizer while the
scheduler sized the same block from
``duplex_window_previous_marker_token_ids``, which the plugin produced with
the runtime tokenizer. Any divergence only surfaced as the rebuild-length
RuntimeError. The window plan now carries the marker ids the scheduler
counted, so both sides use one source of truth; ``_encode_text`` stays for
the session context prefix and suffix.

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 tests/worker/test_native_duplex_hooks.py                    | 1 +
 .../model_executor/models/minicpmo_4_5/duplex/stage0.py     | 6 +++++-
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index 14656a1a4fb..117137a8b5e 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -976,6 +976,7 @@ def test_minicpmo_stage0_context_window_inserts_previous_before_suffix():
             "mode": "context",
             "drop_units": 1,
             "previous_token_ids": [42],
+            "previous_marker_token_ids": [201, 5],
             "replacement_prompt_len": 7,
         },
     )
diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
index 101a005a0de..cbc4d0b351c 100644
--- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
+++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/stage0.py
@@ -452,7 +452,11 @@ def _window_replacement_parts(
             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:
-                marker_ids = self._encode_text("\n\nprevious: ")
+                # 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)

From 6b384947f38fb24706896910caa653ca33e1c89f Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Thu, 17 Sep 2026 17:06:15 +0800
Subject: [PATCH 7/8] [Test] Pin the final MiniCPM append reservation to one
 unit

The unconditional extra 12 slots for a ``final`` exact-chunk append were
removed because Stage0 never fed the silent second unit. Serving pads the
final residual audio itself, so the reservation must equal one unit per
chunk (``<unit>`` + its audio) plus the closure pair for every unit after
the first. Add the off-mode regression that asserts a final append reserves
exactly what a non-final append of the same audio does.

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../duplex/test_minicpmo_window_plugin.py     | 28 +++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/tests/engine/duplex/test_minicpmo_window_plugin.py b/tests/engine/duplex/test_minicpmo_window_plugin.py
index 9b3bae2589d..d2afce19932 100644
--- a/tests/engine/duplex/test_minicpmo_window_plugin.py
+++ b/tests/engine/duplex/test_minicpmo_window_plugin.py
@@ -87,3 +87,31 @@ def test_first_and_final_append_reserve_exact_window_input(seq, samples, expecte
         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)

From 06d71be737aca9ff1a54d9ad7572c3300cfccf58 Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Thu, 17 Sep 2026 18:13:39 +0800
Subject: [PATCH 8/8] [Test] Cover the rebuilt MiniCPM window against the model
 length

The overflow regression needs a plan whose replacement length is at the
sample-room boundary, so pin num_sampled_tokens_per_step and the append
length instead of relying on the window defaults. The zero-open_start
regressions now drive the planner with the confirmed output span it reads,
and assert the dropped span and projection that the two fallbacks produce.

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 .../sched/test_omni_ar_scheduler_streaming.py | 65 +++++++++++++------
 1 file changed, 44 insertions(+), 21 deletions(-)

diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py
index 383e8c05664..781c2ee57ff 100644
--- a/tests/core/sched/test_omni_ar_scheduler_streaming.py
+++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py
@@ -636,17 +636,17 @@ def test_stage0_window_uses_confirmed_span_and_terminator(cleared_outputs, in_fl
 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. When that exceeds
-    ``max_model_len`` the replacement branch must finish the session with the
-    same explicit overflow error as a plain extension, not hand an
-    over-long prompt to the worker."""
-    sched = _make_live_session_scheduler(max_model_len=25)
+    ``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_prompt_tokens = 9
     session.num_computed_tokens = 10
     session.num_output_placeholders = 0
     _park_session(sched, session)
@@ -656,15 +656,16 @@ def test_stage0_window_rebuild_that_overflows_max_model_len_finishes_the_session
     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)
 
-    # The plan's own length (9 retained context + the 8-slot append + the 2
-    # closure tokens + the append's 6-token window = 25) is the projection,
-    # not the session prompt plus an extension.
-    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 25
+    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
@@ -674,7 +675,7 @@ def test_stage0_window_rebuild_that_overflows_max_model_len_finishes_the_session
     client_index, reason = sched._streaming_context_overflow[session.request_id]
     assert client_index == session.client_index
     assert reason.startswith("context_length_exceeded: ")
-    assert "25 tokens" in reason and "max_model_len 25" in reason
+    assert "21 tokens" in reason and "max_model_len 21" in reason
 
     engine_core_outputs = _run_idle_step(sched)
 
@@ -688,7 +689,8 @@ def test_stage0_window_rebuild_that_leaves_room_to_sample_replaces_the_prompt()
     """One slot above the plan fits: the replacement applies through the normal
     replacement path."""
     sched = _make_scheduler(stage_id=0)
-    sched.max_model_len = 26
+    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
@@ -698,12 +700,15 @@ def test_stage0_window_rebuild_that_leaves_room_to_sample_replaces_the_prompt()
     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 session.prompt_token_ids == [0] * 25
     assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
-    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 25
+    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", {})
@@ -714,28 +719,46 @@ def test_stage0_window_open_start_zero_is_not_replaced_by_the_context_reserve()
     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, _minicpmo45_window_open_start=0)
+    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
     )
 
-    assert session._minicpmo45_window_units == [{"length": 10, "generated_token_ids": [40]}]
-    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["dropped_tokens"] == 10
+    # 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)
+    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
     )
 
-    # preserve_len 3 from duplex_first_append_context_tokens.
-    assert session._minicpmo45_window_units == [{"length": 8, "generated_token_ids": [40]}]
-    assert update.model_intermediate_buffer["duplex"]["stage0_window"]["dropped_tokens"] == 8
+    # 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: