diff --git a/tests/engine/duplex/test_deadline_pacing.py b/tests/engine/duplex/test_deadline_pacing.py new file mode 100644 index 000000000..2b1bf09d6 --- /dev/null +++ b/tests/engine/duplex/test_deadline_pacing.py @@ -0,0 +1,88 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Deterministic unit tests for deadline-aligned silence continuation. + +The engine-resident duplex scheduler in +``vllm_omni/engine/duplex/session/runner.py`` aligns each silence +continuation to ``submission_time_N + chunk_period`` and sleeps only the +remaining budget. These tests cover the pure deadline arithmetic and the +per-session reset semantics of the shared ``DuplexModelSessionState`` +contract. +""" + +from __future__ import annotations + +import pytest + +from vllm_omni.engine.duplex.session.runner import ( + compute_silence_continuation_deadline, +) +from vllm_omni.model_executor.models.minicpmo_4_5.duplex.session import ( + MiniCPMO45ServingSessionState, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +@pytest.mark.parametrize( + ("chunk_period_s", "now", "last_submit", "current_deadline", "delay_s", "next_silence_deadline"), + [ + # First continuation anchors to the last real submission: unit N was + # submitted at t=1.0, audio N produced at 1.4 -> 0.6 s of sleep left. + (1.0, 1.4, 1.0, None, 0.6, 3.0), + # Only the remaining budget is slept (audio produced at 1.7). + (1.0, 1.7, 1.0, None, 0.3, 3.0), + # Overdue deadline: no sleep; the chain still advances by the period. + (1.0, 2.5, 1.0, None, 0.0, 3.0), + # No submission yet (None): the first continuation anchors to now. + (1.0, 0.4, None, None, 1.0, 2.4), + # Deadlines advance from the current deadline, not from now, so pipeline + # processing time does not accumulate as timer drift. + (1.0, 4.2, 3.0, 5.0, 0.8, 6.0), + # Two consecutive units that overrun their deadline by 0.2 s keep the + # 1 s cadence: no sleep, and the chain advances one period each. + (1.0, 2.2, 1.0, None, 0.0, 3.0), + (1.0, 3.2, 2.0, 3.0, 0.0, 4.0), + # A short overshoot (within one period of the current deadline) still + # chases the stale deadline: immediate submit, chain advances. + (1.0, 2.5, 1.0, 2.0, 0.0, 3.0), + # A long stall (more than one period past the current deadline) submits + # one continuation immediately and restarts from that submission. + (1.0, 5.0, 1.0, 2.0, 0.0, 6.0), + # The same recovery applies to the first continuation after a stall. + (1.0, 5.0, 1.0, None, 0.0, 6.0), + # Zero chunk period (guarded upstream): total, no sleep. + (0.0, 10.0, None, None, 0.0, 10.0), + ], +) +def test_compute_silence_continuation_deadline( + chunk_period_s: float, + now: float, + last_submit: float | None, + current_deadline: float | None, + delay_s: float, + next_silence_deadline: float, +) -> None: + delay, next_dl = compute_silence_continuation_deadline( + chunk_period_s=chunk_period_s, + now=now, + last_submit=last_submit, + current_deadline=current_deadline, + ) + assert delay == pytest.approx(delay_s) + assert next_dl == pytest.approx(next_silence_deadline) + + +class TestSilenceDeadlineSessionState: + def test_clear_continuation_resets_the_deadline_chain(self) -> None: + # The native-duplex session state feeds the shared runner, so it must + # start with an unset chain and drop it at turn boundaries. + state = MiniCPMO45ServingSessionState() + assert state.last_native_submit_monotonic is None + assert state.silence_deadline_monotonic is None + state.last_native_submit_monotonic = 5.0 + state.silence_deadline_monotonic = 6.0 + state.clear_continuation() + assert state.last_native_submit_monotonic is None + assert state.silence_deadline_monotonic is None diff --git a/tests/engine/duplex/test_deadline_pacing_scheduler.py b/tests/engine/duplex/test_deadline_pacing_scheduler.py new file mode 100644 index 000000000..451af0bfd --- /dev/null +++ b/tests/engine/duplex/test_deadline_pacing_scheduler.py @@ -0,0 +1,370 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Focused scheduler tests for deadline-aligned silence continuation. + +These tests exercise the production ``_schedule_silence_continuation`` path +(and the append acceptance callback it installs) rather than copying the +scheduling arithmetic. The monotonic clock is a fake so the deadline math is +deterministic, and ``asyncio.sleep`` is simulated by advancing that clock. +Append completion is controlled by gating the stage port's ``submit``. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from typing import Any + +import pytest + +import vllm_omni.engine.duplex.session.model_channel as model_channel_module +import vllm_omni.engine.duplex.session.runner as runner_module +from tests.engine.duplex.test_session_runner import ( + Harness, + append_audio, + close_harness, + open_harness, + tts_output, +) +from vllm_omni.engine.duplex.contracts import DuplexStageSubmission + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +CHUNK_PERIOD_S = 1.0 + + +class FakeClock: + """A monotonic clock whose value the test controls directly.""" + + def __init__(self, start: float = 0.0) -> None: + self.value = start + + def __call__(self) -> float: + return self.value + + +async def _active_response_harness() -> Harness: + """Open a harness and drive a TTS segment so a response is active.""" + h = await open_harness() + await h.run(append_audio()) + request_id = h.stage0_request_id() + await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hello")) + assert h.session.active_response_id is not None + return h + + +def _continuation_kwargs(h: Harness) -> dict[str, object]: + """The scheduler arguments the model channel would pass for this session.""" + return { + "request_id": h.session.active_request_id, + "owner_id": f"response:{h.session.active_response_id}", + "response_id": h.session.active_response_id, + "response_owned": True, + "expected_epoch": h.session.epoch, + "expected_model_turn_id": h.session.turn_id, + } + + +def _install_fake_clock( + monkeypatch: pytest.MonkeyPatch, + *, + clock: FakeClock, + real_sleep: Callable[..., Any], +) -> list[float]: + """Point both modules at the fake clock and simulate ``asyncio.sleep``. + + ``real_sleep`` is the unpatched ``asyncio.sleep`` captured before patching; + the simulated sleep advances the fake clock by the requested delay and then + yields control so the append task can run. The returned ``stall`` list is + added once to the next *positive* sleep so a test can model a wake that + happens late (after the planned deadline); zero-duration scheduling yields + do not consume it. + """ + monkeypatch.setattr(runner_module.time, "monotonic", clock) + monkeypatch.setattr(model_channel_module.time, "monotonic", clock) + stall: list[float] = [0.0] + + async def fake_sleep(delay: float) -> None: + clock.value += delay + if delay > 0: + clock.value += stall[0] + stall[0] = 0.0 + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + return stall + + +@pytest.mark.asyncio +async def test_normal_cadence_keeps_deadlines_aligned(monkeypatch: pytest.MonkeyPatch) -> None: + """Processing time reduces the remaining sleep; consecutive deadlines align. + + Each accepted unit becomes the new anchor (its submission time), and the + following continuation advances from the stored deadline rather than + resetting to now. + """ + clock = FakeClock(start=100.0) + real_sleep = asyncio.sleep + _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + try: + # Real input accepted at t=100.0; the model processes for 0.4 s. + h.runner.model_state.last_native_submit_monotonic = 100.0 + clock.value = 100.4 + + # First continuation: due at 101.0, so only 0.6 s of sleep remains. + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + first_task = h.runner.tasks.append_tail + assert first_task is not None + assert await first_task + # The accepted unit re-anchored the chain to its own submission time + # (101.0) and stored the following deadline (102.0). + assert h.runner.model_state.last_native_submit_monotonic == pytest.approx(101.0) + assert h.runner.model_state.silence_deadline_monotonic == pytest.approx(102.0) + + # Model produces the continuation's audio; the next unit is planned + # from the stored deadline (no drift): the following deadline is 103.0. + clock.value = 101.3 + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + second_task = h.runner.tasks.append_tail + assert second_task is not None + assert await second_task + assert h.runner.model_state.silence_deadline_monotonic == pytest.approx(103.0) + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_in_flight_real_append_skips_outdated_silence(monkeypatch: pytest.MonkeyPatch) -> None: + """A real append accepted after the silence was planned re-anchors the chain. + + The planned silence queues behind the real append; once the real append + completes and re-anchors, ``before_append`` detects the changed anchor + (numeric timestamp) and the outdated silence is skipped (no submission, + no callback overwrite). + """ + clock = FakeClock(start=200.0) + real_sleep = asyncio.sleep + _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + state = h.runner.model_state + submissions_before = len(h.port.submissions) + + original_submit = h.port.submit + entered = asyncio.Event() + release = asyncio.Event() + + async def gated_submit(submission: DuplexStageSubmission) -> Any: + entered.set() + await release.wait() + return await original_submit(submission) + + monkeypatch.setattr(h.port, "submit", gated_submit) + try: + # Existing anchor at t=200.0. + state.last_native_submit_monotonic = 200.0 + state.silence_deadline_monotonic = None + + # The real append starts at t+0.8 and blocks inside the gate. + clock.value = 200.8 + real_append_task = await h.runner._start_append( + h.runner.model.silence_unit_payload(), + final=False, + silence_continuation=False, + ) + await entered.wait() + assert not real_append_task.done() + + # At t+1.0 the silence is planned using the existing anchor; its append + # task waits behind the real append. + clock.value = 201.0 + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + silence_task = h.runner.tasks.append_tail + assert silence_task is not None + assert not silence_task.done() + + # At t+1.1 the real append is released; both appends finish. + clock.value = 201.1 + release.set() + assert await real_append_task + assert await silence_task + + # Exactly one additional submission happened: the real append. The + # outdated silence was skipped, so its callback never overwrote the + # new anchor (the real append's submission time t+0.8), and the + # deadline the real append cleared stays cleared. + assert len(h.port.submissions) == submissions_before + 1 + assert state.last_native_submit_monotonic == pytest.approx(200.8) + assert state.silence_deadline_monotonic is None + + # A fresh continuation uses the new anchor. The clock is left at t+1.1: + # the scheduler sleeps the remaining 0.7 s (deadline t+1.8), which + # verifies re-anchoring controls the wait. Following deadline t+2.8. + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + fresh_task = h.runner.tasks.append_tail + assert fresh_task is not None + assert await fresh_task + assert state.last_native_submit_monotonic == pytest.approx(201.8) + assert state.silence_deadline_monotonic == pytest.approx(202.8) + finally: + release.set() + await close_harness(h) + + +@pytest.mark.asyncio +async def test_long_stall_saves_future_deadline_from_actual_submission( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A submission more than one period past the planned deadline restarts. + + The schedule plans the following deadline at ``planned + period``, but the + actual submission happens much later. The accepted continuation must save + ``submit_time + period`` so the next unit is not immediately due. + """ + clock = FakeClock(start=300.0) + real_sleep = asyncio.sleep + stall = _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + state = h.runner.model_state + try: + # Plan a continuation whose deadline is 1.0 s in the future. The wake + # is delayed 2 s past the planned submission (anchor + 3.0 during the + # scheduler's sleep), so the append accepts 2 s late. + anchor = clock.value + state.last_native_submit_monotonic = anchor + state.silence_deadline_monotonic = None + stall[0] = 2.0 + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + append_task = h.runner.tasks.append_tail + assert append_task is not None + assert await append_task + # anchor + 3.0 > anchor + 2.0 (planned deadline + period) -> save + # submit_time + period = anchor + 4.0. + assert state.last_native_submit_monotonic == pytest.approx(anchor + 3.0) + assert state.silence_deadline_monotonic == pytest.approx(anchor + 4.0) + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_small_wakeup_delay_preserves_planned_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ordinary wakeup jitter (within one period) keeps the planned cadence. + + A submission 0.2 s past the planned deadline must not reset the chain: the + stored deadline stays at ``planned + period`` so drift does not accumulate. + """ + clock = FakeClock(start=400.0) + real_sleep = asyncio.sleep + stall = _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + state = h.runner.model_state + try: + anchor = clock.value + state.last_native_submit_monotonic = anchor + state.silence_deadline_monotonic = None + stall[0] = 0.2 + + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + append_task = h.runner.tasks.append_tail + assert append_task is not None + assert await append_task + + # Submitted at anchor + period + 0.2 (within one period): the planned + # following deadline is kept. + assert state.last_native_submit_monotonic == pytest.approx(anchor + CHUNK_PERIOD_S + 0.2) + assert state.silence_deadline_monotonic == pytest.approx(anchor + 2 * CHUNK_PERIOD_S) + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_exact_one_period_late_preserves_planned_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Exactly one period late is not "more than one period late". + + The strict ``>`` comparison means a submission exactly at the planned + following deadline keeps the planned cadence (no extra catch-up unit). + """ + clock = FakeClock(start=500.0) + real_sleep = asyncio.sleep + stall = _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + state = h.runner.model_state + try: + anchor = clock.value + state.last_native_submit_monotonic = anchor + state.silence_deadline_monotonic = None + stall[0] = 1.0 + + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + append_task = h.runner.tasks.append_tail + assert append_task is not None + assert await append_task + + # Submitted exactly at the planned following deadline: kept. + assert state.last_native_submit_monotonic == pytest.approx(anchor + 2 * CHUNK_PERIOD_S) + assert state.silence_deadline_monotonic == pytest.approx(anchor + 2 * CHUNK_PERIOD_S) + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_failed_append_leaves_timing_unchanged(monkeypatch: pytest.MonkeyPatch) -> None: + """A failed append never advances the timing chain.""" + clock = FakeClock(start=600.0) + real_sleep = asyncio.sleep + _install_fake_clock(monkeypatch, clock=clock, real_sleep=real_sleep) + h = await _active_response_harness() + state = h.runner.model_state + try: + state.last_native_submit_monotonic = 600.0 + state.silence_deadline_monotonic = 601.0 + before_anchor = state.last_native_submit_monotonic + before_deadline = state.silence_deadline_monotonic + h.port.fail_submit = RuntimeError("boom") + + scheduled = await h.runner._schedule_silence_continuation( + h.runner.model.silence_unit_payload(), + **_continuation_kwargs(h), + ) + assert scheduled is True + append_task = h.runner.tasks.append_tail + assert append_task is not None + # The append fails: its task reports failure, and timing is untouched. + assert await append_task is False + assert state.last_native_submit_monotonic == before_anchor + assert state.silence_deadline_monotonic == before_deadline + finally: + await close_harness(h) diff --git a/tests/engine/duplex/test_session_manager.py b/tests/engine/duplex/test_session_manager.py index 0323f16d8..4f1962ee5 100644 --- a/tests/engine/duplex/test_session_manager.py +++ b/tests/engine/duplex/test_session_manager.py @@ -120,6 +120,8 @@ class FakeModelSessionState(DuplexModelSessionState): self.continuation_units = 0 self.pending_silence_task = None self.pending_silence_owner_id = None + self.last_native_submit_monotonic = None + self.silence_deadline_monotonic = None def retain_committed_audio(self, payload, *, operation_id, reserved_bytes=0) -> None: self.committed_audio_payload = payload @@ -136,6 +138,8 @@ class FakeModelSessionState(DuplexModelSessionState): def clear_continuation(self) -> None: self.continuation_owner_id = None self.continuation_units = 0 + self.last_native_submit_monotonic = None + self.silence_deadline_monotonic = None class FakeDataPlane(DuplexDataPlane): diff --git a/vllm_omni/engine/duplex/plugin.py b/vllm_omni/engine/duplex/plugin.py index 130d5204d..f9f9be283 100644 --- a/vllm_omni/engine/duplex/plugin.py +++ b/vllm_omni/engine/duplex/plugin.py @@ -123,6 +123,11 @@ class DuplexModelSessionState(ABC): continuation_units: int pending_silence_task: asyncio.Task[bool] | None pending_silence_owner_id: str | None + # Deadline-aligned silence continuation state: the monotonic submission + # time of the most recent native input unit and the next silence + # continuation deadline. A real (non-silence) input resets the chain. + last_native_submit_monotonic: float | None + silence_deadline_monotonic: float | None @abstractmethod def retain_committed_audio( diff --git a/vllm_omni/engine/duplex/session/append_task.py b/vllm_omni/engine/duplex/session/append_task.py index 724dc7f82..2bc060d63 100644 --- a/vllm_omni/engine/duplex/session/append_task.py +++ b/vllm_omni/engine/duplex/session/append_task.py @@ -53,8 +53,11 @@ class AppendAttempt: retained_committed_payload: dict[str, object] | None #: Response reserved before submission, to be failed if submission does not happen. precreated_response_id: str | None + #: Commits timing state once the runtime accepts the append (before any + #: returned output event can clear the continuation chain). + on_append_accepted: Callable[[float], None] | None = None #: Last chance to call the append off, checked once the predecessor is done. - before_append: Callable[[], bool] | None + before_append: Callable[[], bool] | None = None # ------------------------------------------------------------------ # # Compensation # @@ -140,6 +143,7 @@ class AppendAttempt: ), final=self.final, expected_epoch=self.epoch, + on_append_accepted=self.on_append_accepted, ) if append_ok: model_state.context_locked = True diff --git a/vllm_omni/engine/duplex/session/model_channel.py b/vllm_omni/engine/duplex/session/model_channel.py index 322db7ede..13f0f7781 100644 --- a/vllm_omni/engine/duplex/session/model_channel.py +++ b/vllm_omni/engine/duplex/session/model_channel.py @@ -22,6 +22,7 @@ tracked task, and aborting a stage request in the background. from __future__ import annotations import asyncio +import time import uuid from collections.abc import Awaitable, Callable, Mapping from dataclasses import replace @@ -146,12 +147,16 @@ class ModelChannel: operation_id: str | None = None, final: bool, expected_epoch: int | None = None, + on_append_accepted: Callable[[float], None] | None = None, ) -> tuple[bool, bool]: session = self._ctx.session if not session.capabilities.supports_input_append: return True, False if expected_epoch is not None and session.epoch != expected_epoch: return True, False + # Anchor the submission time before the RPC; the acceptance callback + # commits timing state only if the append actually submitted. + submit_time = time.monotonic() try: result = await self._append_via_data_plane( payload, @@ -169,6 +174,10 @@ class ModelChannel: return True, False if expected_epoch is not None and session.epoch != expected_epoch: return True, False + # Commit timing state before returned output events can clear the + # silence-continuation chain (e.g. a terminal turn-end). + if on_append_accepted is not None: + on_append_accepted(submit_time) request_id, _ = duplex_data_plane_request_info(result) if request_id is not None: self._ctx.plugin.data_plane.begin_request(request_id) diff --git a/vllm_omni/engine/duplex/session/runner.py b/vllm_omni/engine/duplex/session/runner.py index 0b8af49ab..ef464a5fa 100644 --- a/vllm_omni/engine/duplex/session/runner.py +++ b/vllm_omni/engine/duplex/session/runner.py @@ -21,6 +21,7 @@ orchestrator loop (the session is never touched from another thread): from __future__ import annotations import asyncio +import time import uuid from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field @@ -126,6 +127,38 @@ class _Internal: _CANCEL_EVENTS = frozenset({"input.cancel", "response.cancel", "barge_in", "output_audio_buffer.clear"}) +def compute_silence_continuation_deadline( + *, + chunk_period_s: float, + now: float, + last_submit: float | None, + current_deadline: float | None, +) -> tuple[float, float]: + """Return the silence continuation schedule ``(delay_s, next_silence_deadline)``. + + The first continuation anchors to the latest accepted append's submission + time (``last_submit + chunk_period_s``), falling back to ``now`` when no + submission exists. Later continuations advance from the current deadline + instead of from ``now``, so pipeline processing time does not accumulate + as timer drift. When the schedule is more than one period overdue it is + stale: one continuation submits immediately and the schedule restarts + from ``now`` (``next_silence_deadline = now + chunk_period_s``) instead of + firing a burst of catch-ups. + + ``delay_s`` is the wait before this continuation and + ``next_silence_deadline`` the deadline for the following one. + """ + if current_deadline is None: + base = last_submit if last_submit is not None else now + deadline = base + chunk_period_s + else: + deadline = current_deadline + if now - deadline > chunk_period_s: + deadline = now + delay_s = max(0.0, deadline - now) + return delay_s, deadline + chunk_period_s + + class DuplexSessionRunner: """Owns one ``DuplexEngineSession`` on the orchestrator loop (see module docstring).""" @@ -918,12 +951,22 @@ class DuplexSessionRunner: operation_id: str | None = None, retained_committed_payload: dict[str, object] | None = None, silence_continuation: bool = False, + on_append_accepted: Callable[[float], None] | None = None, before_append: Callable[[], bool] | None = None, ) -> asyncio.Task[bool]: session = self.session model_state = self.model_state if not silence_continuation: self._mark_pending_silence_superseded() + if on_append_accepted is None: + # A real (non-silence) input re-anchors the silence pacing + # chain: its submission becomes the anchor and the stored + # deadline is cleared until the next continuation sets one. + def _reanchor_chain(submit_time: float) -> None: + model_state.last_native_submit_monotonic = submit_time + model_state.silence_deadline_monotonic = None + + on_append_accepted = _reanchor_chain append_epoch = session.epoch append_turn_id = payload_turn_id(payload) if append_turn_id is None: @@ -949,6 +992,7 @@ class DuplexSessionRunner: operation_id=operation_id, retained_committed_payload=retained_committed_payload, precreated_response_id=session.active_response_id if precreate_response else None, + on_append_accepted=on_append_accepted, before_append=before_append, ) @@ -1032,29 +1076,60 @@ class DuplexSessionRunner: append_tail = self.tasks.append_tail if (append_tail is None or append_tail.done()) and self._real_input_waiting(): return False - continuation_delay_s = max(0.0, float(session.capabilities.chunk_period_ms or 1000) / 1000.0) - if continuation_delay_s > 0: - await asyncio.sleep(continuation_delay_s) - if ( - self.tasks.append_tail is not append_tail - or ((append_tail is None or append_tail.done()) and self._real_input_waiting()) - or self.model.silence_continuation_is_stale( + chunk_period_s = max(0.0, float(session.capabilities.chunk_period_ms or 1000) / 1000.0) + # Snapshot the anchor after any wait for pending silence. A real append + # accepted between the snapshot and this continuation's submission + # re-anchors the chain; _still_valid() then skips the stale unit. + anchor = model_state.last_native_submit_monotonic + # Align the next silence unit to submission_time_N + chunk_period and + # sleep only the remaining budget. The deadline is stored by the + # acceptance callback when the append actually submits, so skipped or + # stale continuations never advance the clock. + delay_s, next_silence_deadline = compute_silence_continuation_deadline( + chunk_period_s=chunk_period_s, + now=time.monotonic(), + last_submit=anchor, + current_deadline=model_state.silence_deadline_monotonic, + ) + if delay_s > 0: + await asyncio.sleep(delay_s) + if ( + self.tasks.append_tail is not append_tail + or ((append_tail is None or append_tail.done()) and self._real_input_waiting()) + or self.model.silence_continuation_is_stale( + request_id=request_id, + response_id=response_id, + response_owned=response_owned, + expected_epoch=expected_epoch, + expected_model_turn_id=expected_model_turn_id, + ) + ): + return False + + def _still_valid() -> bool: + return ( + # The anchor changed (a real append was accepted) after this + # continuation was planned; the unit is outdated. + model_state.last_native_submit_monotonic == anchor + and not self._real_input_waiting() + and not self.model.silence_continuation_is_stale( request_id=request_id, response_id=response_id, response_owned=response_owned, expected_epoch=expected_epoch, expected_model_turn_id=expected_model_turn_id, ) - ): - return False + ) - def _still_valid() -> bool: - return not self._real_input_waiting() and not self.model.silence_continuation_is_stale( - request_id=request_id, - response_id=response_id, - response_owned=response_owned, - expected_epoch=expected_epoch, - expected_model_turn_id=expected_model_turn_id, + def _on_append_accepted(submit_time: float) -> None: + # Commit timing state once the runtime accepts the append. If the + # submission is more than one chunk period past the planned + # deadline, the schedule is stale: restart from the actual + # submission. Small delays keep the planned cadence so ordinary + # jitter does not accumulate as drift. + model_state.last_native_submit_monotonic = submit_time + model_state.silence_deadline_monotonic = ( + submit_time + chunk_period_s if submit_time > next_silence_deadline else next_silence_deadline ) model_state.pending_silence_owner_id = owner_id @@ -1062,6 +1137,7 @@ class DuplexSessionRunner: dict(payload) if isinstance(payload, dict) else {}, final=False, silence_continuation=True, + on_append_accepted=_on_append_accepted, before_append=_still_valid, ) return task is not None diff --git a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py index 49d0caab6..fe53ec653 100644 --- a/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py +++ b/vllm_omni/model_executor/models/minicpmo_4_5/duplex/session.py @@ -29,6 +29,11 @@ class MiniCPMO45ServingSessionState(DuplexModelSessionState): continuation_units: int = 0 pending_silence_task: asyncio.Task[bool] | None = None pending_silence_owner_id: str | None = None + # Deadline-aligned silence continuation state: the monotonic submission + # time of the most recent native input unit and the next silence + # continuation deadline. A real (non-silence) input resets the chain. + last_native_submit_monotonic: float | None = None + silence_deadline_monotonic: float | None = None def retain_committed_audio( self, @@ -55,3 +60,5 @@ class MiniCPMO45ServingSessionState(DuplexModelSessionState): self.continuation_units = 0 self.pending_silence_task = None self.pending_silence_owner_id = None + self.last_native_submit_monotonic = None + self.silence_deadline_monotonic = None