diff --git a/.buildkite/common/ci_source_file_dependencies.yml b/.buildkite/common/ci_source_file_dependencies.yml
index 99bbffa75..870221074 100644
--- a/.buildkite/common/ci_source_file_dependencies.yml
+++ b/.buildkite/common/ci_source_file_dependencies.yml
@@ -255,6 +255,8 @@ source_file_dependencies:
- *minicpmo_4_5_duplex
- tests/e2e/online_serving/test_minicpmo_4_5_duplex.py
- tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py
+ - tests/e2e/online_serving/test_minicpmo_4_5_window.py
+ - tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
- tests/e2e/online_serving/test_duplex_client_live.py
- tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
- tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml
index 86e7a7a35..e76487cce 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/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py
index 83c7038f2..ab373f5de 100644
--- a/tests/core/sched/test_omni_ar_scheduler_streaming.py
+++ b/tests/core/sched/test_omni_ar_scheduler_streaming.py
@@ -69,6 +69,39 @@ def _make_update(prompt_token_ids: list[int] | None = None) -> StreamingUpdate:
)
+def _make_minicpm_window_update(
+ *,
+ seq: int,
+ mode: str,
+ high: int = 15,
+ low: int = 10,
+ context_max_units: int = 1,
+) -> StreamingUpdate:
+ update = _make_update([0] * 8)
+ update.model_intermediate_buffer = {
+ "duplex": {
+ "data_plane": True,
+ "seq": seq,
+ "runtime_config": {
+ "duplex_scheduler_token_id": 0,
+ "duplex_first_append_context_tokens": 3,
+ "duplex_window_prefix_tokens": 2,
+ "duplex_window_suffix_token_ids": [3],
+ "duplex_window_previous_marker_token_ids": [70, 71],
+ "duplex_window_special_token_ids": [99],
+ "duplex_window_config": {
+ "sliding_window_mode": mode,
+ "basic_window_high_tokens": high,
+ "basic_window_low_tokens": low,
+ "context_previous_max_tokens": 4,
+ "context_max_units": context_max_units,
+ },
+ },
+ }
+ }
+ return update
+
+
def _make_talker_adapter(
*,
max_model_len: int = 100,
@@ -510,6 +543,225 @@ def test_stage0_streaming_update_keeps_all_computed_tokens_without_placeholder()
assert sched._new_prompt_len_snapshot[session.request_id] == 2
+def test_stage0_basic_window_rebuilds_below_low_watermark() -> None:
+ sched = _make_scheduler(stage_id=0)
+ session = _make_request()
+ session.prompt_token_ids = [0] * 9
+ session._all_token_ids.clear()
+ session._all_token_ids.extend(session.prompt_token_ids)
+ session.num_prompt_tokens = 9
+ session.append_output_token_ids([40])
+ session.num_computed_tokens = 10
+ session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+
+ sched._update_request_as_session(session, update)
+
+ assert session.prompt_token_ids == [0] * 9
+ assert session.num_computed_tokens == 0
+ assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
+ plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+ assert plan == {
+ "completed_token_ids": [40],
+ "replace": True,
+ "mode": "basic",
+ "drop_units": 1,
+ "dropped_tokens": 9,
+ "previous_token_ids": [],
+ "previous_marker_token_ids": [70, 71],
+ "replacement_prompt_len": 9,
+ }
+ sched._free_request_blocks.assert_called_once_with(session)
+
+
+def test_stage0_context_window_compacts_dropped_speech() -> None:
+ sched = _make_scheduler(stage_id=0)
+ session = _make_request()
+ session.prompt_token_ids = [0] * 9
+ session._all_token_ids.clear()
+ session._all_token_ids.extend(session.prompt_token_ids)
+ session.num_prompt_tokens = 9
+ session.append_output_token_ids([40])
+ session.num_computed_tokens = 10
+ session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+
+ first = _make_minicpm_window_update(seq=2, mode="context")
+ sched._update_request_as_session(session, first)
+ assert session.num_prompt_tokens == 18
+ assert first.model_intermediate_buffer["duplex"]["stage0_window"] == {"completed_token_ids": [40]}
+
+ session.append_output_token_ids([50])
+ session.num_computed_tokens = 19
+ session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+ second = _make_minicpm_window_update(seq=3, mode="context")
+ sched._update_request_as_session(session, second)
+
+ assert session.prompt_token_ids == [0] * 21
+ plan = second.model_intermediate_buffer["duplex"]["stage0_window"]
+ assert plan["completed_token_ids"] == [50]
+ assert plan["drop_units"] == 1
+ assert plan["previous_token_ids"] == [40]
+ # The worker embeds this exact marker, not a re-tokenized one, so the
+ # previous region is len(marker) + len(previous).
+ assert plan["previous_marker_token_ids"] == [70, 71]
+ assert plan["replacement_prompt_len"] == 21
+ assert getattr(session, "_minicpmo45_window_previous_len") == 3
+
+
+@pytest.mark.parametrize("cleared_outputs", [False, True])
+@pytest.mark.parametrize("in_flight", [0, 1])
+def test_stage0_window_uses_confirmed_span_and_terminator(cleared_outputs, in_flight) -> None:
+ sched = _make_scheduler(stage_id=0)
+ session = _make_request()
+ session.prompt_token_ids = [0] * 9
+ session._all_token_ids[:] = session.prompt_token_ids
+ session.num_prompt_tokens = 9
+ session.append_output_token_ids([40, 99])
+ session.num_computed_tokens = 10 + in_flight
+ session.num_output_placeholders = in_flight
+ session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+ if cleared_outputs:
+ # A stop without a queued append clears this before the next update.
+ session._output_token_ids.clear()
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+
+ sched._update_request_as_session(session, update)
+
+ plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+ assert plan["completed_token_ids"] == [40]
+ assert plan["completed_terminator_token_id"] == 99
+ assert plan["replacement_prompt_len"] == 9
+ assert plan["dropped_tokens"] == 9
+
+
+def test_stage0_window_rebuild_that_overflows_max_model_len_finishes_the_session() -> None:
+ """A rebuilt window prompt is bounded by the client's window settings, not
+ by the model: a unit carrying camera frames is hundreds of tokens, and a
+ ``context`` window adds the ``previous`` region on top. The replacement
+ branch must therefore check ``replacement_prompt_len`` against
+ ``max_model_len - sample_room`` and finish the session with
+ ``context_length_exceeded``, exactly like a plain extension."""
+ sched = _make_live_session_scheduler(max_model_len=21)
+ session = _make_request()
+ session.prompt_token_ids = [0] * 9
+ session._all_token_ids[:] = [0] * 9
+ session.num_prompt_tokens = 9
+ session._output_token_ids[:] = [40]
+ session._all_token_ids.append(40)
+ session.num_computed_tokens = 10
+ session.num_output_placeholders = 0
+ _park_session(sched, session)
+ # The parked append state above is the frame before this update: one
+ # confirmed output token, fully computed.
+ session._output_token_ids[:] = [40]
+ session._all_token_ids[:] = [*session.prompt_token_ids, 40]
+ session.num_computed_tokens = 10
+ session.num_in_flight_tokens = 0
+ sched.num_sampled_tokens_per_step = 1
+ # The replacement plan rebuilds 21 tokens (retained context plus the
+ # append) and one step samples one more, so 21 leaves no room.
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+ update.prompt_token_ids = [0] * 20
+ original_prompt = list(session.prompt_token_ids)
+
+ sched._update_request_as_session(session, update)
+
+ assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 21
+ assert session.prompt_token_ids == original_prompt
+ assert session.num_prompt_tokens == 9
+ assert session.status == RequestStatus.FINISHED_ERROR
+ assert session.request_id not in sched.requests
+ assert sched.finished_req_ids == {session.request_id}
+ sched._free_request_blocks.assert_called_once_with(session)
+ client_index, reason = sched._streaming_context_overflow[session.request_id]
+ assert client_index == session.client_index
+ assert reason.startswith("context_length_exceeded: ")
+ assert "21 tokens" in reason and "max_model_len 21" in reason
+
+ engine_core_outputs = _run_idle_step(sched)
+
+ (output,) = engine_core_outputs[session.client_index].outputs
+ assert output.request_id == session.request_id
+ assert output.finish_reason == FinishReason.ERROR
+ assert output.stop_reason == reason
+
+
+def test_stage0_window_rebuild_that_leaves_room_to_sample_replaces_the_prompt() -> None:
+ """One slot above the plan fits: the replacement applies through the normal
+ replacement path."""
+ sched = _make_scheduler(stage_id=0)
+ sched.max_model_len = 25
+ sched.num_sampled_tokens_per_step = 1
+ session = _make_request()
+ session.prompt_token_ids = [0] * 9
+ session._all_token_ids[:] = session.prompt_token_ids
+ session.num_prompt_tokens = 9
+ session.append_output_token_ids([40])
+ session.num_computed_tokens = 10
+ session.num_output_placeholders = 0
+ session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+ update.prompt_token_ids = [0] * 20
+
+ sched._update_request_as_session(session, update)
+
+ assert update.model_intermediate_buffer["meta"]["replace_streaming_prompt"] is True
+ assert update.model_intermediate_buffer["duplex"]["stage0_window"]["replacement_prompt_len"] == 21
+ assert session.prompt_token_ids == [0] * 21
+ assert session.num_prompt_tokens == 21
+ assert session.num_computed_tokens == 0
+ assert session.status == RequestStatus.WAITING
+ sched._free_request_blocks.assert_called_once_with(session)
+ assert not getattr(sched, "_streaming_context_overflow", {})
+
+
+def test_stage0_window_open_start_zero_is_not_replaced_by_the_context_reserve() -> None:
+ """A recorded ``open_start`` of 0 is a legitimate empty context prefix. The
+ fallback must be an explicit ``is None`` check: treating 0 as missing
+ substitutes the context reserve, counts the suffix twice, and the worker's
+ rebuild-length check raises on the first replacement."""
+ session = SimpleNamespace(
+ num_prompt_tokens=9,
+ _all_token_ids=[0] * 9 + [40],
+ num_computed_tokens=10,
+ num_output_placeholders=0,
+ _minicpmo45_window_open_start=0,
+ )
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+ update.prompt_token_ids = [0] * 21
+
+ assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+ session, update, segment_output_ids=[40], completed_terminator=3
+ )
+
+ # open_start stays 0, so the reported unit spans the whole 12-token
+ # boundary instead of the 9 the context reserve would fold in.
+ plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+ assert plan["dropped_tokens"] == 12
+ assert plan["replacement_prompt_len"] == 19
+
+
+def test_stage0_window_open_start_falls_back_to_the_context_reserve_when_unset() -> None:
+ session = SimpleNamespace(
+ num_prompt_tokens=9,
+ _all_token_ids=[0] * 9 + [40],
+ num_computed_tokens=10,
+ num_output_placeholders=0,
+ )
+ update = _make_minicpm_window_update(seq=2, mode="basic")
+ update.prompt_token_ids = [0] * 21
+
+ assert OmniARScheduler._prepare_minicpmo45_stage0_window(
+ session, update, segment_output_ids=[40], completed_terminator=3
+ )
+
+ # fallback open_start is preserve_len 3 from duplex_first_append_context_tokens,
+ # so the unit is three tokens shorter.
+ plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+ assert plan["dropped_tokens"] == 9
+ assert plan["replacement_prompt_len"] == 22
+
+
def test_explicit_streaming_payload_replaces_placeholder_prompt() -> None:
sched = _make_scheduler(stage_id=1)
sched.chunk_transfer_adapter = SimpleNamespace(
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 000000000..553f07992
--- /dev/null
+++ b/tests/e2e/online_serving/helpers/minicpmo_window_e2e.py
@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Drive complete native duplex turns while Stage-0 repeatedly rebuilds KV."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+from pathlib import Path
+
+from vllm_omni.clients.duplex import DuplexClient, EventCollector, read_pcm16_wav, wait_for_condition
+from vllm_omni.clients.minicpmo_4_5 import create_duplex_session_config
+
+
+async def run_window_turn(
+ *,
+ url: str,
+ model: str,
+ input_wav: Path,
+ mode: str,
+ ref_audio: Path | None = None,
+ repeats: int = 2,
+ video_frames: list[str] | None = None,
+ buffered_flush: bool = False,
+ timeout_s: float = 240.0,
+) -> dict[str, object]:
+ """Use aggressive limits to exercise multiple windows in one audio turn."""
+ pcm = read_pcm16_wav(input_wav) * repeats
+ collector = EventCollector()
+ reference = (
+ "data:audio/wav;base64," + base64.b64encode(ref_audio.read_bytes()).decode("ascii")
+ if ref_audio is not None
+ else None
+ )
+ config = create_duplex_session_config(
+ modalities=("text", "audio") if ref_audio is not None else ("text",),
+ ref_audio=reference,
+ temperature=0.0,
+ extra_body={
+ "sliding_window_mode": mode,
+ "context_max_units": 1,
+ "context_previous_max_tokens": 16,
+ "basic_window_high_tokens": 64,
+ "basic_window_low_tokens": 32,
+ },
+ )
+ client = DuplexClient(
+ url, model=model, config=config, reconnect=None, heartbeat_interval_s=None, handshake_timeout_s=timeout_s
+ )
+ async with client:
+ reader = asyncio.create_task(collector.consume(client))
+ collector.add({"type": "session.created", "session": client.session_info})
+ try:
+ frames_sent = await client.stream_pcm(
+ pcm,
+ chunk_ms=2500 if buffered_flush else 200,
+ realtime=not buffered_flush,
+ is_speech=True,
+ video_frames=video_frames,
+ )
+ await client.commit(final=True)
+ await wait_for_condition(
+ lambda: collector.count("input_audio_buffer.committed") > 0 or bool(collector.errors()),
+ timeout_s=timeout_s,
+ label="final input commit",
+ )
+ await wait_for_condition(
+ lambda: collector.count("response.done") > 0 or bool(collector.errors()),
+ timeout_s=timeout_s,
+ label="completed windowed response",
+ )
+ # Final execution follows the commit acknowledgement asynchronously.
+ await asyncio.sleep(3.0)
+ await client.close(timeout_s=timeout_s)
+ await asyncio.wait_for(reader, timeout=5.0)
+ finally:
+ if not reader.done():
+ reader.cancel()
+ await asyncio.gather(reader, return_exceptions=True)
+ return {
+ "created": collector.count("session.created"),
+ "committed": collector.count("input_audio_buffer.committed"),
+ "done": collector.count("response.done"),
+ "closed": collector.count("session.closed"),
+ "errors": collector.errors(),
+ "audio_bytes": len(collector.audio_bytes()),
+ "frames_sent": frames_sent,
+ "input_seconds": len(pcm) / 32000,
+ }
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 000000000..300205eb0
--- /dev/null
+++ b/tests/e2e/online_serving/test_minicpmo_4_5_window.py
@@ -0,0 +1,88 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Real-model regression for the scheduler-owned Thinker sliding window."""
+
+import asyncio
+
+import pytest
+
+from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
+ SERVER_PARAMS,
+ duplex_camera_frames,
+ realtime_url,
+ resolve_ref_audio,
+ validated_input_wav,
+)
+from tests.e2e.online_serving.helpers.minicpmo_window_e2e import run_window_turn
+from tests.helpers.mark import hardware_test
+
+pytestmark = [pytest.mark.omni, pytest.mark.advanced_model]
+
+
+def _assert_complete(result, *, require_audio=False):
+ assert result["created"] == 1
+ assert result["committed"] > 0
+ assert result["done"] > 0
+ assert result["closed"] == 1
+ assert result["errors"] == []
+ if require_audio:
+ assert result["audio_bytes"] > 0
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode,reference", [("basic", False), ("context", False), ("context", True)])
+def test_window_rebuild_and_next_session(omni_server, mode, reference):
+ # Two independent sessions exercise admission after final-unit execution.
+ for _ in range(2):
+ result = asyncio.run(
+ run_window_turn(
+ url=realtime_url(omni_server),
+ model=omni_server.model,
+ input_wav=validated_input_wav(),
+ mode=mode,
+ ref_audio=resolve_ref_audio() if reference else None,
+ )
+ )
+ _assert_complete(result, require_audio=reference)
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode,camera", [("basic", False), ("context", False), ("context", True)])
+def test_window_continuous_input(omni_server, tmp_path, mode, camera):
+ # 24 repeats cover over two minutes and many canonical KV replacements.
+ frames = duplex_camera_frames(seconds=4, cache_dir=tmp_path / "camera") if camera else None
+ result = asyncio.run(
+ run_window_turn(
+ url=realtime_url(omni_server),
+ model=omni_server.model,
+ input_wav=validated_input_wav(),
+ mode=mode,
+ repeats=24,
+ ref_audio=resolve_ref_audio(),
+ video_frames=frames,
+ )
+ )
+ _assert_complete(result, require_audio=True)
+ assert isinstance(result["input_seconds"], (int, float)) and result["input_seconds"] > 120
+ if camera:
+ assert isinstance(result["frames_sent"], int) and result["frames_sent"] >= 120
+
+
+@hardware_test(res={"cuda": "H100", "npu": "A3"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+@pytest.mark.parametrize("mode", ["off", "basic", "context"])
+def test_window_buffered_flush(omni_server, mode):
+ # Burst appends cross processor chunk boundaries and leave a final tail.
+ result = asyncio.run(
+ run_window_turn(
+ url=realtime_url(omni_server),
+ model=omni_server.model,
+ input_wav=validated_input_wav(),
+ mode=mode,
+ ref_audio=resolve_ref_audio(),
+ buffered_flush=True,
+ )
+ )
+ _assert_complete(result, require_audio=True)
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 000000000..d2afce199
--- /dev/null
+++ b/tests/engine/duplex/test_minicpmo_window_plugin.py
@@ -0,0 +1,117 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import base64
+from types import SimpleNamespace
+
+import pytest
+
+from vllm_omni.engine.duplex.config import DuplexSessionConfig
+from vllm_omni.engine.duplex.contracts import DuplexFence
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex import plugin as module
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def _plugin():
+ return module.MiniCPMO45DuplexPlugin(lambda *args: None)
+
+
+@pytest.mark.asyncio
+async def test_window_configuration_is_prepared_by_plugin(monkeypatch):
+ tokenizer = SimpleNamespace(
+ all_special_ids=[90],
+ unk_token_id=-1,
+ eos_token_id=99,
+ encode=lambda text, add_special_tokens=False: [ord(char) for char in text],
+ convert_tokens_to_ids=lambda token: {"<|listen|>": 91}.get(token, -1),
+ )
+ monkeypatch.setattr(module, "_load_tokenizer", lambda config: tokenizer)
+ config = DuplexSessionConfig(
+ modalities=("text",),
+ extra_body={
+ "sliding_window_mode": "basic",
+ "basic_window_high_tokens": 120,
+ "basic_window_low_tokens": 80,
+ },
+ )
+ runtime = await _plugin().prepare_runtime_config(config, model_config=None)
+ assert runtime["duplex_window_config"]["sliding_window_mode"] == "basic"
+ assert "sliding_window_mode" not in config.extra_body
+ assert runtime["duplex_window_prefix_tokens"] > 0
+ assert runtime["duplex_window_suffix_token_ids"]
+ assert runtime["duplex_window_previous_marker_token_ids"]
+
+
+@pytest.mark.asyncio
+async def test_invalid_window_configuration_is_rejected():
+ config = DuplexSessionConfig(
+ modalities=("text",),
+ extra_body={
+ "basic_window_high_tokens": 80,
+ "basic_window_low_tokens": 80,
+ },
+ )
+ with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+ await _plugin().prepare_runtime_config(config, model_config=None)
+ assert exc.value.code == "invalid_sliding_window_config"
+
+
+def test_window_update_cannot_change_created_configuration():
+ config = DuplexSessionConfig(extra_body={"sliding_window_mode": "basic"})
+ with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError) as exc:
+ _plugin().runtime_config_for_update(config, {"duplex_window_config": {"sliding_window_mode": "off"}})
+ assert exc.value.code == "sliding_window_update_unsupported"
+
+
+def test_window_internal_configuration_is_server_owned():
+ with pytest.raises(module.MiniCPMO45ClientRuntimeConfigError):
+ _plugin().validate_client_extra_body({"duplex_window_config": {}})
+
+
+@pytest.mark.parametrize("final", [False, True])
+@pytest.mark.parametrize("seq,samples,expected", [(1, 32000, 16), (1, 48000, 28), (2, 16000, 13)])
+def test_first_and_final_append_reserve_exact_window_input(seq, samples, expected, final):
+ prompt = module.build_duplex_data_plane_prompt(
+ request_id="window-request",
+ fence=DuplexFence("sid", turn_id=1),
+ session_config={},
+ runtime_config={"duplex_first_append_context_tokens": 5},
+ seq=seq,
+ turn_seq=seq,
+ payload={
+ "audio": base64.b64encode(bytes(samples * 4)).decode(),
+ "format": "pcm_f32le",
+ "sample_rate_hz": 16000,
+ },
+ final=final,
+ )
+ assert len(prompt["prompt_token_ids"]) == expected
+
+
+@pytest.mark.parametrize("seq", [1, 2, 3])
+def test_final_exact_chunk_append_reserves_one_unit_not_two(seq):
+ """A final append whose audio is already an exact number of chunks must
+ reserve exactly what Stage0 feeds: one unit per chunk (```` 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)
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 27037f029..c5d4d0cde 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/tests/worker/test_native_duplex_hooks.py b/tests/worker/test_native_duplex_hooks.py
index 1b1bf4b7e..d3da3bba0 100644
--- a/tests/worker/test_native_duplex_hooks.py
+++ b/tests/worker/test_native_duplex_hooks.py
@@ -848,6 +848,176 @@ def test_minicpmo_stage0_data_plane_prefill_matches_official_unit_format():
assert result["prompt_suffix_len"] == 0
+@pytest.mark.parametrize("mode", [None, "off", "basic", "context"])
+def test_minicpmo_stage0_history_collection_starts_with_session_config(mode):
+ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import _MiniCPMO45Stage0SessionState
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="history-lifecycle")
+ config = {} if mode is None else {"duplex_window_config": {"sliding_window_mode": mode}}
+ runtime._prepare_session_context(state, {}, runtime_config=config)
+ for seq in range(1, 101):
+ runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=seq)
+ enabled = mode in {"basic", "context"}
+ assert state.window_enabled is enabled
+ assert len(state.window_units) == (99 if enabled else 0)
+ assert (state.pending_window_unit is not None) is enabled
+ assert state.pending_window_generated_tokens == []
+
+
+@pytest.mark.parametrize("mode", ["basic", "context"])
+@pytest.mark.parametrize("buffered", [False, True])
+def test_minicpmo_stage0_multi_chunk_append_matches_scheduler_window(mode, buffered):
+ from vllm_omni.core.sched.omni_ar_scheduler import OmniARScheduler
+ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import _MiniCPMO45Stage0SessionState
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="multi-chunk-window", window_enabled=True)
+ if buffered:
+ # Keep the tiny fake processor's initial chunk unpadded to model carry.
+ runtime._pad_first_audio_chunk_if_needed = lambda *args: None
+ assert runtime._stage_prefill_embeddings_only(state, np.zeros(2, dtype=np.float32), seq=0)["success"] is False
+ first = runtime._stage_prefill_embeddings_only(
+ state, np.zeros(6 if buffered else 8, dtype=np.float32), seq=1, final=buffered
+ )
+ assert first["input_token_ids"] == [1, 11, 2, 1, 11]
+ assert state.window_units == []
+ assert state.pending_window_unit.token_ids == first["input_token_ids"]
+ session = SimpleNamespace(num_prompt_tokens=5)
+ replaced = False
+ for seq in (2, 3):
+ info = {
+ "duplex": {
+ "data_plane": True,
+ "seq": seq,
+ "runtime_config": {
+ "duplex_window_config": {
+ "sliding_window_mode": mode,
+ "basic_window_high_tokens": 1,
+ "basic_window_low_tokens": 2,
+ "context_max_units": 1,
+ "context_previous_max_tokens": 16,
+ },
+ "duplex_window_previous_marker_token_ids": [201, 5],
+ },
+ }
+ }
+ update = SimpleNamespace(prompt_token_ids=[3, 2, 1, 11], model_intermediate_buffer=info)
+ planned = OmniARScheduler._prepare_minicpmo45_stage0_window(
+ session, update, segment_output_ids=[44], completed_terminator=3
+ )
+ result = runtime._stage_prefill_embeddings_only(
+ state, np.zeros(4, dtype=np.float32), seq=seq, stage0_window=info["duplex"]["stage0_window"]
+ )
+ assert result["stage0_window_replaced"] is planned
+ if planned:
+ assert result["num_input_tokens"] == len(update.prompt_token_ids)
+ replaced = True
+ break
+ session.num_prompt_tokens += 1 + len(update.prompt_token_ids)
+ assert replaced
+
+
+def test_minicpmo_stage0_basic_window_rebuilds_from_retained_units():
+ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-basic-window", window_enabled=True)
+ first = runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+ assert first["input_token_ids"] == [1, 11]
+ state.pending_window_generated_tokens.append(5)
+ state.pending_terminator_token = 3
+
+ rebuilt = runtime._stage_prefill_embeddings_only(
+ state,
+ np.zeros(4, dtype=np.float32),
+ seq=2,
+ stage0_window={
+ "replace": True,
+ "mode": "basic",
+ "drop_units": 1,
+ "replacement_prompt_len": 2,
+ },
+ )
+
+ assert rebuilt["stage0_window_replaced"] is True
+ assert rebuilt["input_token_ids"] == [1, 11]
+ assert state.window_units == []
+
+
+def test_minicpmo_stage0_context_window_inserts_previous_before_suffix():
+ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-stage0-context-window", window_enabled=True)
+ prefix = runtime._embed_token(200)
+ suffix = runtime._embed_token(202)
+ state.context_embeds = [prefix, suffix]
+ state.context_token_ids = [200, 202]
+ state.context_prefix_embeds = [prefix]
+ state.context_prefix_token_ids = [200]
+ state.context_suffix_embeds = [suffix]
+ state.context_suffix_token_ids = [202]
+ first = runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+ assert first["input_token_ids"] == [200, 202, 1, 11]
+ state.pending_window_generated_tokens.append(42)
+ state.pending_terminator_token = 3
+
+ rebuilt = runtime._stage_prefill_embeddings_only(
+ state,
+ np.zeros(4, dtype=np.float32),
+ seq=2,
+ stage0_window={
+ "replace": True,
+ "mode": "context",
+ "drop_units": 1,
+ "previous_token_ids": [42],
+ "previous_marker_token_ids": [201, 5],
+ "replacement_prompt_len": 7,
+ },
+ )
+
+ assert rebuilt["input_token_ids"] == [200, 201, 5, 42, 202, 1, 11]
+ assert rebuilt["num_input_tokens"] == 7
+
+
+@pytest.mark.parametrize("pending_terminator", [None, 3, 99])
+def test_minicpmo_stage0_window_uses_accepted_output_not_async_sampler_history(pending_terminator):
+ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.stage0 import (
+ _MiniCPMO45Stage0SessionState,
+ )
+
+ runtime = _stage0_vision_runtime()
+ runtime._stage_audio_embeddings = lambda *args, **kwargs: torch.zeros((10, 2))
+ state = _MiniCPMO45Stage0SessionState(session_id="sid-accepted-window", window_enabled=True)
+ state.context_prefix_token_ids = [200] * 7
+ state.context_suffix_token_ids = [202]
+ state.context_token_ids = [*state.context_prefix_token_ids, 202]
+ state.context_prefix_embeds = [runtime._embed_token(token_id) for token_id in state.context_prefix_token_ids]
+ state.context_suffix_embeds = [runtime._embed_token(202)]
+ state.context_embeds = [*state.context_prefix_embeds, *state.context_suffix_embeds]
+ runtime._stage_prefill_embeddings_only(state, np.zeros(4, dtype=np.float32), seq=1)
+ for seq in (2, 3):
+ state.pending_window_generated_tokens = [3, 40, 41]
+ state.pending_terminator_token = pending_terminator
+ plan = {"completed_token_ids": [], "completed_terminator_token_id": 3}
+ if seq == 3:
+ plan.update(replace=True, mode="context", drop_units=1, replacement_prompt_len=32)
+ result = runtime._stage_prefill_embeddings_only(
+ state, np.zeros(4, dtype=np.float32), seq=seq, stage0_window=plan
+ )
+
+ assert result["num_input_tokens"] == 32
+ assert [len(unit.token_ids) for unit in state.window_units] == [13]
+ assert len(state.pending_window_unit.token_ids) == 11
+ assert 40 not in result["input_token_ids"]
+ assert 41 not in result["input_token_ids"]
+
+
def _stage0_vision_runtime():
import torch
diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py
index 4d55c55f8..f07aae021 100644
--- a/vllm_omni/core/sched/omni_ar_scheduler.py
+++ b/vllm_omni/core/sched/omni_ar_scheduler.py
@@ -848,6 +848,16 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
req_id = session.request_id
self._new_prompt_len_snapshot[req_id] = len(update.prompt_token_ids)
outstanding_async_tokens = getattr(session, "num_output_placeholders", 0)
+ # Use the same confirmed span that upstream preserves when extending
+ # a session. The segment output list is cleared on a resumable stop,
+ # and, for an already queued append, can still include its terminator.
+ confirmed_end = session.num_computed_tokens - outstanding_async_tokens
+ segment_output_ids = list(session._all_token_ids[session.num_prompt_tokens : confirmed_end])
+ completed_terminator = (
+ session._all_token_ids[confirmed_end]
+ if session.num_prompt_tokens <= confirmed_end < len(session._all_token_ids)
+ else None
+ )
# Seed the stale share in SCHEDULED-token units (see the segment-stop
# site in update_from_output): num_in_flight_tokens matches what each
# pre-replacement frame will drain, so the counter reaches exactly
@@ -894,6 +904,23 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
if self.log_stats:
session.record_event(EngineCoreEventType.QUEUED)
return
+ if stage_id == 0 and self._prepare_minicpmo45_stage0_window(
+ session,
+ update,
+ segment_output_ids=segment_output_ids,
+ completed_terminator=completed_terminator,
+ ):
+ # The rebuilt prompt is bounded by the client's window settings,
+ # not by the model (a camera unit is hundreds of tokens), so it
+ # needs the same max_model_len check as a plain extension. The
+ # plan replaces the whole prompt, so the replacement length is the
+ # projection, not the session's current prompt plus an extension.
+ plan = update.model_intermediate_buffer["duplex"]["stage0_window"]
+ if self._streaming_update_overflows(session, update, projected_len=plan["replacement_prompt_len"]):
+ return
+ self._release_replaced_streaming_prompt_cache(session)
+ self._replace_streaming_session(session, update)
+ return
streaming_prompt_payload = next(
(
info
@@ -962,11 +989,178 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
if hasattr(update, "model_intermediate_buffer"):
session.model_intermediate_buffer = update.model_intermediate_buffer
+ @staticmethod
+ def _prepare_minicpmo45_stage0_window(
+ session: Request,
+ update: StreamingUpdate,
+ *,
+ segment_output_ids: list[int],
+ completed_terminator: int | None = None,
+ ) -> bool:
+ """Plan an official-style MiniCPM Stage-0 window at unit boundaries.
+
+ vLLM owns a paged KV cache, so deleting a middle span and rotating the
+ retained K tensors in place is not a safe model hook. Instead, record
+ completed unit lengths in the scheduler and request a full prompt
+ replacement when a watermark fires. The worker rebuilds matching
+ embeddings from its unit history, which recomputes RoPE at the new
+ contiguous positions.
+ """
+ info = getattr(update, "model_intermediate_buffer", None)
+ if not isinstance(info, dict):
+ return False
+ duplex = info.get("duplex")
+ if not isinstance(duplex, dict) or duplex.get("data_plane") is not True:
+ return False
+ runtime_config = duplex.get("runtime_config")
+ runtime_config = runtime_config if isinstance(runtime_config, dict) else {}
+ window = runtime_config.get("duplex_window_config")
+ if not isinstance(window, dict):
+ return False
+ mode = window.get("sliding_window_mode", "off")
+ if mode == "off":
+ return False
+ if mode not in {"basic", "context"}:
+ return False
+
+ try:
+ seq = int(duplex.get("seq", 0) or 0)
+ preserve_len = int(runtime_config.get("duplex_first_append_context_tokens", 0) or 0)
+ except (TypeError, ValueError):
+ return False
+ if seq <= 1:
+ return False
+
+ # The sampled terminator is discarded by the normal session update and
+ # re-injected at the head of this append, followed by .
+ # generated_ids contains only the confirmed output span retained by
+ # the streaming update, excluding its final uncomputed sample.
+ generated_ids = segment_output_ids
+ # Sampler-side history can contain speculative or stale async output.
+ # The worker must rebuild from the same accepted ids used for lengths.
+ duplex["stage0_window"] = {"completed_token_ids": list(generated_ids)}
+ if completed_terminator is not None:
+ duplex["stage0_window"]["completed_terminator_token_id"] = int(completed_terminator)
+ base_len = int(getattr(session, "num_prompt_tokens", 0) or 0) + len(generated_ids)
+ boundary = base_len + 2
+ # A legitimate open_start of 0 (an empty context prefix) must not fall
+ # back to preserve_len, which would double-count the suffix below.
+ recorded_open_start = getattr(session, "_minicpmo45_window_open_start", None)
+ open_start = preserve_len if recorded_open_start is None else int(recorded_open_start)
+ unit_len = boundary - open_start
+ if unit_len <= 0:
+ return False
+
+ special_ids = {
+ int(token_id)
+ for token_id in runtime_config.get("duplex_window_special_token_ids", ())
+ if isinstance(token_id, int)
+ }
+ units = list(getattr(session, "_minicpmo45_window_units", ()))
+ units.append(
+ {
+ "length": unit_len,
+ "generated_token_ids": [int(token_id) for token_id in generated_ids if token_id not in special_ids],
+ }
+ )
+ projected_len = base_len + len(update.prompt_token_ids)
+ drop_count = 0
+ dropped_len = 0
+
+ if mode == "basic":
+ try:
+ high = int(window.get("basic_window_high_tokens", 8000))
+ low = int(window.get("basic_window_low_tokens", 6000))
+ except (TypeError, ValueError):
+ return False
+ if projected_len > high:
+ while units and projected_len - dropped_len > low:
+ dropped_len += int(units[drop_count]["length"])
+ drop_count += 1
+ if drop_count >= len(units):
+ break
+ else:
+ try:
+ max_units = int(window.get("context_max_units", 24))
+ previous_max = int(window.get("context_previous_max_tokens", 500))
+ except (TypeError, ValueError):
+ return False
+ drop_count = max(0, len(units) - max_units)
+ dropped_len = sum(int(unit["length"]) for unit in units[:drop_count])
+ previous = list(getattr(session, "_minicpmo45_window_previous_token_ids", ()))
+ for unit in units[:drop_count]:
+ previous.extend(unit["generated_token_ids"])
+ if len(previous) > previous_max:
+ previous = previous[-previous_max:]
+ marker = [
+ int(token_id)
+ for token_id in runtime_config.get("duplex_window_previous_marker_token_ids", ())
+ if isinstance(token_id, int)
+ ]
+ previous_with_marker = marker + previous if previous else []
+ old_previous_len = int(getattr(session, "_minicpmo45_window_previous_len", 0) or 0)
+ session._minicpmo45_window_previous_token_ids = previous
+ session._minicpmo45_window_previous_len = len(previous_with_marker)
+ projected_len += len(previous_with_marker) - old_previous_len
+
+ retained_units = units[drop_count:]
+ session._minicpmo45_window_units = retained_units
+ if drop_count == 0:
+ session._minicpmo45_window_open_start = boundary
+ return False
+
+ replacement_len = projected_len - dropped_len
+ if replacement_len <= 0:
+ return False
+ scheduler_token_id = runtime_config.get("duplex_scheduler_token_id", 0)
+ try:
+ scheduler_token_id = max(0, int(scheduler_token_id))
+ except (TypeError, ValueError):
+ scheduler_token_id = 0
+ update.prompt_token_ids = [scheduler_token_id] * replacement_len
+ meta = info.setdefault("meta", {})
+ if isinstance(meta, dict):
+ meta["replace_streaming_prompt"] = True
+ previous_ids = list(getattr(session, "_minicpmo45_window_previous_token_ids", ()))
+ # The worker embeds the marker ids the window plan carries: it must not
+ # re-tokenize the marker with its own tokenizer, which could differ
+ # from the ids this length is computed from.
+ marker_ids = [
+ int(token_id)
+ for token_id in runtime_config.get("duplex_window_previous_marker_token_ids", ())
+ if isinstance(token_id, int)
+ ]
+ duplex["stage0_window"] = {
+ "completed_token_ids": list(generated_ids),
+ "replace": True,
+ "mode": mode,
+ "drop_units": drop_count,
+ "dropped_tokens": dropped_len,
+ "previous_token_ids": previous_ids,
+ "previous_marker_token_ids": marker_ids,
+ "replacement_prompt_len": replacement_len,
+ }
+ if completed_terminator is not None:
+ duplex["stage0_window"]["completed_terminator_token_id"] = int(completed_terminator)
+ recorded_prefix_len = runtime_config.get("duplex_window_prefix_tokens")
+ prefix_len = preserve_len if recorded_prefix_len is None else int(recorded_prefix_len)
+ suffix_len = len(runtime_config.get("duplex_window_suffix_token_ids", ()) or ())
+ previous_len = int(getattr(session, "_minicpmo45_window_previous_len", 0) or 0)
+ new_preserve_len = prefix_len + previous_len + suffix_len if mode == "context" else preserve_len
+ session._minicpmo45_window_open_start = new_preserve_len + sum(int(unit["length"]) for unit in retained_units)
+ return True
+
# Prefix of the stop_reason carried by the FinishReason.ERROR output, so the
# serving side can map it to a stable error code.
STREAMING_CONTEXT_OVERFLOW_STOP_REASON = "context_length_exceeded"
- def _streaming_update_overflows(self, session: Request, update: StreamingUpdate) -> bool:
+ def _streaming_update_overflows(
+ self,
+ session: Request,
+ update: StreamingUpdate,
+ *,
+ projected_len: int | None = None,
+ ) -> bool:
"""Finish a streaming session whose next extension cannot fit the model.
Upstream ``_update_request_as_session`` appends the update to the
@@ -991,6 +1185,10 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
finish to the next ``schedule()`` would leave the client waiting. The
reason is emitted with the terminal output (see
:meth:`_emit_streaming_context_overflow_outputs`).
+
+ ``projected_len`` overrides the extended-prompt projection for a caller
+ that replaces the prompt instead of growing it, such as the MiniCPM
+ Stage-0 window rebuild.
"""
max_model_len = getattr(self, "max_model_len", None)
if max_model_len is None:
@@ -1001,8 +1199,13 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
new_tokens = len(update.prompt_token_ids or ())
# The extended prompt is the current prompt plus the computed output
# tokens upstream keeps, then the update: num_computed_tokens covers
- # both when the prompt was fully computed.
- projected = max(int(session.num_prompt_tokens), int(session.num_computed_tokens)) + new_tokens
+ # both when the prompt was fully computed. A caller that already knows
+ # the post-update prompt length (a window rebuild) passes it instead.
+ projected = (
+ max(int(session.num_prompt_tokens), int(session.num_computed_tokens)) + new_tokens
+ if projected_len is None
+ else int(projected_len)
+ )
# Room for the tokens one step samples on top of the prompt (1 without
# speculative decoding). __new__-built test schedulers carry no
# num_sampled_tokens_per_step.
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 e57377766..44ae9dd9a 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 @@ from vllm_omni.model_executor.models.minicpmo_4_5.duplex.data_plane import (
MiniCPMO45DataPlaneContext,
MiniCPMO45DataPlaneSession,
)
-from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import MiniCPMO45DuplexPolicy
+from vllm_omni.model_executor.models.minicpmo_4_5.duplex.policy import (
+ MiniCPMO45DuplexPolicy,
+ MiniCPMO45DuplexWindowConfig,
+)
from vllm_omni.model_executor.models.minicpmo_4_5.duplex.session import (
MiniCPMO45ServingSessionState,
)
@@ -85,6 +88,11 @@ PRIVATE_RUNTIME_CONFIG_KEYS = frozenset(
"ref_audio_format",
"ref_audio_sample_rate_hz",
"initial_user_text",
+ "duplex_window_config",
+ "duplex_window_prefix_tokens",
+ "duplex_window_suffix_token_ids",
+ "duplex_window_previous_marker_token_ids",
+ "duplex_window_special_token_ids",
}
)
@@ -245,8 +253,8 @@ def build_duplex_data_plane_prompt(
token_budget = context_reserve + first_units * 12 - 1 + vision_tokens
if seq > 1 and duplex_payload_is_exact_chunks(payload):
token_budget += 1
- if final and duplex_payload_is_exact_chunks(payload):
- token_budget += 12
+ # Serving already pads the final residual audio. Stage0 does not
+ # append another silent unit, so final must not reserve extra slots.
extra_body = session_config.get("extra_body")
raw_token_id = runtime_config.get("duplex_scheduler_token_id")
try:
@@ -515,6 +523,16 @@ def _apply_first_append_context_tokens(
return
ref_tokens = MiniCPMO45DuplexPolicy.audio_token_count(ref_sample_count or 0)
runtime_config["duplex_first_append_context_tokens"] = len(prefix_ids) + ref_tokens + len(suffix_ids)
+ runtime_config["duplex_window_prefix_tokens"] = len(prefix_ids) + ref_tokens
+ runtime_config["duplex_window_suffix_token_ids"] = [int(token_id) for token_id in suffix_ids]
+ marker_ids = tokenizer.encode("\n\nprevious: ", add_special_tokens=False)
+ runtime_config["duplex_window_previous_marker_token_ids"] = [int(token_id) for token_id in marker_ids]
+ runtime_config["duplex_window_special_token_ids"] = sorted(
+ MiniCPMO45DuplexPolicy.native_special_token_ids(
+ MiniCPMO45DuplexPolicy.token_ids_from_tokenizer(tokenizer),
+ tokenizer_special_ids=list(getattr(tokenizer, "all_special_ids", ()) or ()),
+ )
+ )
def _model_vision_tile_pixels(model_config: ModelConfig | None) -> int | None:
@@ -722,6 +740,8 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
code="unsupported_ref_audio_path",
)
runtime_config: dict[str, object] = {"instructions": config.instructions}
+ window_config = self._pop_window_config(extra_body) or MiniCPMO45DuplexWindowConfig()
+ runtime_config["duplex_window_config"] = window_config.as_dict()
# ``duplex_initial_user_text`` is the older extra_body spelling and
# still works; the session field is the framework-level one.
initial_user_text = extra_body.pop("duplex_initial_user_text", None)
@@ -788,6 +808,17 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
current: Mapping[str, object],
) -> dict[str, object]:
runtime_config = deepcopy(dict(current))
+ extra_body = dict(config.extra_body)
+ requested_window = self._pop_window_config(extra_body)
+ if requested_window is not None:
+ reject_changed_runtime_value(
+ requested_window.as_dict(),
+ runtime_config.get("duplex_window_config"),
+ message="sliding-window configuration cannot be changed after the session is created",
+ code="sliding_window_update_unsupported",
+ error_cls=MiniCPMO45ClientRuntimeConfigError,
+ )
+ config.extra_body = extra_body
reject_changed_runtime_value(
config.instructions,
runtime_config.get("instructions"),
@@ -812,6 +843,23 @@ class MiniCPMO45DuplexPlugin(DuplexModelPlugin):
runtime_config["duplex_stage_sampling_params"] = stage_sampling
return runtime_config
+ @staticmethod
+ def _pop_window_config(extra_body: dict[str, object]) -> MiniCPMO45DuplexWindowConfig | None:
+ names = (
+ "sliding_window_mode",
+ "basic_window_high_tokens",
+ "basic_window_low_tokens",
+ "context_previous_max_tokens",
+ "context_max_units",
+ )
+ provided = {name: extra_body.pop(name) for name in names if name in extra_body}
+ if not provided:
+ return None
+ try:
+ return MiniCPMO45DuplexWindowConfig.from_mapping(provided)
+ except ValueError as exc:
+ raise MiniCPMO45ClientRuntimeConfigError(str(exc), code="invalid_sliding_window_config") from exc
+
def data_plane_context(
self,
*,
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 0332c81c0..7b0aded4d 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 bc13b58f2..cbc4d0b35 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_OPTIONAL_TOKEN_FIELDS = MiniCPMO45DuplexPolicy.OPTIONAL_TOKEN_FIELDS
_MINICPMO45_PROCESSOR_LOAD_LOCK = Lock()
+@dataclass
+class _MiniCPMO45WindowUnit:
+ embeds: list[Any] = field(default_factory=list)
+ token_ids: list[int] = field(default_factory=list)
+
+
@dataclass
class _MiniCPMO45Stage0SessionState:
session_id: str
@@ -36,6 +42,14 @@ class _MiniCPMO45Stage0SessionState:
audio_chunk_idx: int = 0
context_embeds: list[torch.Tensor] = field(default_factory=list)
context_token_ids: list[int] = field(default_factory=list)
+ context_prefix_embeds: list[Any] = field(default_factory=list)
+ context_prefix_token_ids: list[int] = field(default_factory=list)
+ context_suffix_embeds: list[Any] = field(default_factory=list)
+ context_suffix_token_ids: list[int] = field(default_factory=list)
+ window_units: list[_MiniCPMO45WindowUnit] = field(default_factory=list)
+ window_enabled: bool = False
+ pending_window_unit: _MiniCPMO45WindowUnit | None = None
+ pending_window_generated_tokens: list[int] = field(default_factory=list)
current_turn_ended: bool = True
prepared_append_identity: tuple[int | None, int] | None = None
prepared_inputs_embeds: torch.Tensor | None = None
@@ -142,6 +156,11 @@ class MiniCPMO45Stage0DuplexRuntime:
*,
runtime_config: dict[str, object] | None = None,
) -> None:
+ window_config = (runtime_config or {}).get("duplex_window_config")
+ state.window_enabled = isinstance(window_config, dict) and window_config.get("sliding_window_mode", "off") in {
+ "basic",
+ "context",
+ }
if not self._stage_runtime_ready():
return
self._require_special_token_ids()
@@ -157,17 +176,26 @@ class MiniCPMO45Stage0DuplexRuntime:
(runtime_config or {}).get("initial_user_text"),
)
for token_id in self._encode_text(prefix):
- state.context_embeds.append(self._embed_token(token_id))
+ embed = self._embed_token(token_id)
+ state.context_embeds.append(embed)
state.context_token_ids.append(token_id)
+ state.context_prefix_embeds.append(embed)
+ state.context_prefix_token_ids.append(token_id)
if ref_audio is not None:
ref_audio_embeds = self._stage_ref_audio_embeddings(ref_audio, state=state)
if ref_audio_embeds is not None:
ref_audio_embeds = self._as_2d_tensor(ref_audio_embeds)
+ placeholder_ids = [self.unit_token_id] * int(ref_audio_embeds.shape[0])
state.context_embeds.append(ref_audio_embeds)
- state.context_token_ids.extend([self.unit_token_id] * int(ref_audio_embeds.shape[0]))
+ state.context_token_ids.extend(placeholder_ids)
+ state.context_prefix_embeds.append(ref_audio_embeds)
+ state.context_prefix_token_ids.extend(placeholder_ids)
for token_id in self._encode_text(suffix):
- state.context_embeds.append(self._embed_token(token_id))
+ embed = self._embed_token(token_id)
+ state.context_embeds.append(embed)
state.context_token_ids.append(token_id)
+ state.context_suffix_embeds.append(embed)
+ state.context_suffix_token_ids.append(token_id)
def _stage_prefill_embeddings_only(
self,
@@ -179,6 +207,7 @@ class MiniCPMO45Stage0DuplexRuntime:
seq: int | None = None,
is_speech: bool = False,
final: bool = False,
+ stage0_window: dict[str, object] | None = None,
) -> dict[str, object]:
"""Build scheduler-owned Stage0 input embeddings for one audio append.
@@ -199,6 +228,16 @@ class MiniCPMO45Stage0DuplexRuntime:
result["input_token_ids"] = list(state.prepared_input_token_ids)
return result
self._require_special_token_ids()
+ if isinstance(stage0_window, dict):
+ completed_ids = stage0_window.get("completed_token_ids")
+ if isinstance(completed_ids, list):
+ state.pending_window_generated_tokens = [int(token_id) for token_id in completed_ids]
+ completed_terminator = stage0_window.get("completed_terminator_token_id")
+ if isinstance(completed_terminator, int):
+ # Async sampling can overwrite the worker's pending token
+ # after the scheduler accepted a segment stop. Rebuild from
+ # the scheduler's canonical boundary, just like its content.
+ state.pending_terminator_token = completed_terminator
if audio_waveform is None or len(audio_waveform) == 0:
return self._stage_prefill_result(False, start_time, "empty audio")
state.audio_buffer = np.concatenate([state.audio_buffer, np.asarray(audio_waveform, dtype=np.float32)])
@@ -279,12 +318,25 @@ class MiniCPMO45Stage0DuplexRuntime:
# the closure; the model's listen/speak policy depends on
# seeing its own past decisions in context.
pending_terminator = state.pending_terminator_token
+ closure_token_ids: list[int] = []
if pending_terminator is not None and units_built == 0:
state.pending_terminator_token = None
embed_parts.append(self._embed_token(pending_terminator))
token_ids.append(int(pending_terminator))
+ closure_token_ids.append(int(pending_terminator))
embed_parts.append(self._embed_token(self.unit_end_token_id))
token_ids.append(self.unit_end_token_id)
+ closure_token_ids.append(self.unit_end_token_id)
+ if units_built == 0:
+ self._finalize_window_unit(state, closure_token_ids)
+ elif state.pending_window_unit is not None:
+ # Internal processor chunks belong to this same append.
+ state.pending_window_unit.embeds.extend(
+ self._embed_token(token_id) for token_id in closure_token_ids
+ )
+ state.pending_window_unit.token_ids.extend(closure_token_ids)
+ unit_embed_start = len(embed_parts)
+ unit_token_start = len(token_ids)
embed_parts.append(self._embed_token(self.unit_token_id))
token_ids.append(self.unit_token_id)
if frame_blocks:
@@ -311,6 +363,11 @@ class MiniCPMO45Stage0DuplexRuntime:
state.audio_buffer = state.audio_buffer[consumed_samples:]
state.audio_chunk_idx += 1
units_built += 1
+ if state.window_enabled:
+ if state.pending_window_unit is None:
+ state.pending_window_unit = _MiniCPMO45WindowUnit()
+ state.pending_window_unit.embeds.extend(embed_parts[unit_embed_start:])
+ state.pending_window_unit.token_ids.extend(token_ids[unit_token_start:])
chunk_size = self._streaming_chunk_size(processor)
# Match official streaming_prefill: per chunk feed ONLY +audio. The assistant
# turn is opened once at session init; re-emitting the turn-open prefix per chunk
@@ -320,6 +377,9 @@ class MiniCPMO45Stage0DuplexRuntime:
import torch
+ window_result = self._window_replacement_parts(state, stage0_window)
+ if window_result is not None:
+ embed_parts, token_ids = window_result
inputs_embeds = torch.cat([self._as_2d_tensor(embed) for embed in embed_parts], dim=0)
result = self._stage_prefill_result(True, start_time)
result.update(
@@ -332,6 +392,7 @@ class MiniCPMO45Stage0DuplexRuntime:
"uses_model_runner_scheduler": True,
"runner_kv_backed": True,
"runtime_impl": "scheduler_data_plane",
+ "stage0_window_replaced": window_result is not None,
}
)
if is_speech and (append_identity is None or state.pending_speech_append_identity != append_identity):
@@ -344,6 +405,85 @@ class MiniCPMO45Stage0DuplexRuntime:
state.prepared_result = {k: v for k, v in result.items() if k not in {"inputs_embeds", "input_token_ids"}}
return result
+ def _finalize_window_unit(
+ self,
+ state: _MiniCPMO45Stage0SessionState,
+ closure_token_ids: list[int],
+ ) -> None:
+ pending = state.pending_window_unit
+ if not state.window_enabled or pending is None:
+ state.pending_window_unit = None
+ state.pending_window_generated_tokens.clear()
+ return
+ generated = list(state.pending_window_generated_tokens)
+ token_ids = [*pending.token_ids, *generated, *closure_token_ids]
+ embeds = list(pending.embeds)
+ embeds.extend(self._embed_token(token_id) for token_id in generated)
+ embeds.extend(self._embed_token(token_id) for token_id in closure_token_ids)
+ state.window_units.append(_MiniCPMO45WindowUnit(embeds=embeds, token_ids=token_ids))
+ state.pending_window_unit = None
+ state.pending_window_generated_tokens.clear()
+
+ def _window_replacement_parts(
+ self,
+ state: _MiniCPMO45Stage0SessionState,
+ stage0_window: dict[str, Any] | None,
+ ) -> tuple[list[Any], list[int]] | None:
+ if not isinstance(stage0_window, dict) or stage0_window.get("replace") is not True:
+ return None
+ try:
+ drop_units = max(0, int(stage0_window.get("drop_units", 0)))
+ except (TypeError, ValueError):
+ return None
+ if drop_units > len(state.window_units):
+ raise RuntimeError(
+ "MiniCPM-o Stage-0 window history is shorter than the scheduler drop plan: "
+ f"drop={drop_units}, completed={len(state.window_units)}"
+ )
+ del state.window_units[:drop_units]
+
+ mode = stage0_window.get("mode")
+ embeds: list[Any] = []
+ token_ids: list[int] = []
+ previous_ids: list[int] = []
+ if mode == "context":
+ embeds.extend(state.context_prefix_embeds)
+ token_ids.extend(state.context_prefix_token_ids)
+ previous = stage0_window.get("previous_token_ids")
+ previous_ids = [int(token_id) for token_id in previous] if isinstance(previous, list) else []
+ if previous_ids:
+ # The plan carries the marker ids the scheduler sized the
+ # `previous` region from. Re-tokenizing here would be a second
+ # source of truth and could disagree with that length.
+ marker = stage0_window.get("previous_marker_token_ids")
+ marker_ids = [int(token_id) for token_id in marker] if isinstance(marker, list) else []
+ embeds.extend(self._embed_token(token_id) for token_id in [*marker_ids, *previous_ids])
+ token_ids.extend(marker_ids)
+ token_ids.extend(previous_ids)
+ embeds.extend(state.context_suffix_embeds)
+ token_ids.extend(state.context_suffix_token_ids)
+ else:
+ embeds.extend(state.context_embeds)
+ token_ids.extend(state.context_token_ids)
+ for unit in state.window_units:
+ embeds.extend(unit.embeds)
+ token_ids.extend(unit.token_ids)
+ if state.pending_window_unit is not None:
+ embeds.extend(state.pending_window_unit.embeds)
+ token_ids.extend(state.pending_window_unit.token_ids)
+ expected = stage0_window.get("replacement_prompt_len")
+ if isinstance(expected, int) and expected != len(token_ids):
+ unit_lengths = [len(unit.token_ids) for unit in state.window_units]
+ pending_len = len(state.pending_window_unit.token_ids) if state.pending_window_unit is not None else 0
+ raise RuntimeError(
+ "MiniCPM-o Stage-0 window rebuild length mismatch: "
+ f"worker={len(token_ids)}, scheduler={expected}, mode={mode}, "
+ f"prefix={len(state.context_prefix_token_ids)}, "
+ f"suffix={len(state.context_suffix_token_ids)}, previous={len(previous_ids)}, "
+ f"units={unit_lengths}, pending={pending_len}, drop={drop_units}"
+ )
+ return embeds, token_ids
+
@staticmethod
def _stage_prefill_result(success: bool, start_time: float, reason: str = "") -> dict[str, object]:
return {
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 b1c8db89c..03024bbe5 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 @@ class MiniCPMO45OmniForConditionalGeneration(nn.Module, SupportsMultiModal, Supp
seq=seq,
is_speech=bool(payload.get("is_speech", False)),
final=bool(duplex.get("final")),
+ stage0_window=(duplex.get("stage0_window") if isinstance(duplex.get("stage0_window"), dict) else None),
)
update_result = dict(result)
+ if result.get("stage0_window_replaced") is True:
+ window = duplex.get("stage0_window", {})
+ logger.info(
+ "MiniCPM-o Stage-0 window replaced: mode=%s drop_units=%s tokens=%s seq=%s",
+ window.get("mode"),
+ window.get("drop_units"),
+ result.get("num_input_tokens"),
+ seq,
+ )
update_result.pop("inputs_embeds", None)
if result.get("success") is not True:
embeds = input_embeds if input_embeds is not None else self.get_input_embeddings(input_ids)
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 4d818b79f..ad329c271 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
@@ -865,8 +865,25 @@ def llm2tts(
special_token_ids.get("chunk_eos_token_id"),
special_token_ids.get("chunk_tts_eos_token_id"),
}
- tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
- tts_hidden_slice = thinker_hidden_states[tts_bos_idx:end_idx].to(torch.float32).contiguous()
+ if is_native_duplex_handoff:
+ # Earlier unforwarded decisions may already be folded into
+ # the rebuilt prompt while still appearing in this delta.
+ # Align the explicit tts_bos path by the segment's end, just
+ # like the native speak/text paths below, rather than adding
+ # those decisions to the prompt a second time.
+ out_start = tts_bos_idx - prompt_token_ids_len
+ out_end = tts_eos_idx - prompt_token_ids_len if tts_eos_idx is not None else len(llm_output_ids)
+ hidden_base = int(thinker_hidden_states.shape[0]) - len(llm_output_ids)
+ if hidden_base >= 0 and out_end > out_start:
+ tts_token_ids_slice = torch.tensor(llm_output_ids[out_start:out_end], dtype=torch.long)
+ tts_hidden_slice = (
+ thinker_hidden_states[hidden_base + out_start : hidden_base + out_end]
+ .to(torch.float32)
+ .contiguous()
+ )
+ else:
+ tts_token_ids_slice = torch.tensor(full_token_ids[tts_bos_idx:end_idx], dtype=torch.long)
+ tts_hidden_slice = thinker_hidden_states[tts_bos_idx:end_idx].to(torch.float32).contiguous()
elif is_native_duplex_handoff:
# Official MiniCPM-o duplex does not prefill an assistant
# <|tts_bos|> boundary before generation. A segment delta can