[Bugfix][MiniCPM-o] Prevent native duplex Stage-1 prompt overflow

Signed-off-by: Sy03 <1370724210@qq.com>

.buildkite/cuda/test-merge.yml

@@ -166,6 +166,7 @@ steps:
           - tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
           - tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
           - tests/e2e/online_serving/run_minicpmo_realtime_duplex_multi_session.py
+          - tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py
           - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py
           - vllm_omni/deploy/minicpmo_4_5.yaml
           - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py

.buildkite/cuda/test-ready.yml

@@ -229,6 +229,7 @@ steps:
           - tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
           - tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
           - tests/e2e/online_serving/run_minicpmo_realtime_duplex_multi_session.py
+          - tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py
           - tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py
           - vllm_omni/deploy/minicpmo_4_5.yaml
           - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py

examples/online_serving/minicpmo/realtime_duplex_demo.py

@@ -388,7 +388,7 @@ def parse_args() -> argparse.Namespace:
         "--temperature",
         type=float,
         default=None,
-        help="Stage0 sampling temperature; response-required soft-interrupt uses 0.0.",
+        help="Stage0 sampling temperature; omitted to preserve the model default.",
     )
     parser.add_argument("--no-realtime-pacing", action="store_true")
     parser.add_argument("--require-audio", action="store_true")

pyproject.toml

@@ -74,6 +74,7 @@ dev = [
     "kernels==0.14.1", # for accuracy tests comparison with diffusers output (0.15.0 incompatible with diffusers 0.38.0)
     "s3tokenizer==0.3.0", # for tts test
     "step-audio2==1.0.0", # Step-Audio2 / MiniCPM-o 4.5 Token2Wav (flashcosyvoice / hyperpyyaml)
+    "silero-vad==6.1", # Realtime server-VAD duplex E2E
 ]
 
 demo = [
@@ -83,6 +84,10 @@ demo = [
     "cn2an>=0.5.22",  # Chinese number normalization for TTS (e.g. "1次" -> "一次")
 ]
 
+server-vad = [
+    "silero-vad==6.1",
+]
+
 forced-aligner = [
     "qwen-asr>=0.0.6",  # faithful Japanese/Korean word segmentation for the TTS forced aligner
 ]

tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py

@@ -0,0 +1,115 @@
+"""Tests for ``OmniARScheduler._drop_aborted_queued_requests``.
+
+``schedule()`` sweeps ``FINISHED_ABORTED`` requests out of the queues before
+handing control to upstream, because upstream ``Scheduler.schedule()`` raises
+``RuntimeError: Invalid request status: FINISHED_ABORTED`` for any request it
+admits in a finished state -- which kills the stage's engine core, not just
+the request.
+
+Regression for the sweep missing ``skipped_waiting``, the third queue
+upstream admits from. An aborted duplex session parked there was re-selected
+by ``_select_waiting_queue_for_scheduling`` on a later tick and crashed the
+stage.
+"""
+
+from __future__ import annotations
+
+import pytest
+from vllm.v1.core.sched.request_queue import SchedulingPolicy, create_request_queue
+from vllm.v1.request import RequestStatus
+
+from vllm_omni.core.sched.omni_ar_scheduler import OmniARScheduler
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+_POLICIES = [
+    pytest.param(SchedulingPolicy.FCFS, id="fcfs"),
+    pytest.param(SchedulingPolicy.PRIORITY, id="priority"),
+]
+
+
+class _StubRequest:
+    """Minimal ``Request`` stub with the surface the sweep exercises.
+
+    Carries the ``(priority, arrival_time)`` ordering ``PriorityRequestQueue``
+    relies on so the same stub works under both scheduling policies.
+    """
+
+    def __init__(self, request_id: str, status: RequestStatus, arrival_time: float = 0.0) -> None:
+        self.request_id = request_id
+        self.status = status
+        self.priority = 0
+        self.arrival_time = arrival_time
+
+    def __lt__(self, other: _StubRequest) -> bool:
+        return (self.priority, self.arrival_time) < (other.priority, other.arrival_time)
+
+
+def _make_scheduler(policy: SchedulingPolicy, *, waiting=(), skipped_waiting=(), running=()):
+    """Bare scheduler with just the surface the sweep reads."""
+    scheduler = OmniARScheduler.__new__(OmniARScheduler)
+    scheduler.waiting = create_request_queue(policy)
+    for req in waiting:
+        scheduler.waiting.add_request(req)
+    scheduler.skipped_waiting = create_request_queue(policy)
+    for req in skipped_waiting:
+        scheduler.skipped_waiting.add_request(req)
+    scheduler.running = list(running)
+    return scheduler
+
+
+@pytest.mark.parametrize("policy", _POLICIES)
+def test_sweep_drops_aborted_request_from_skipped_waiting(policy: SchedulingPolicy) -> None:
+    """The sweep must cover ``skipped_waiting``.
+
+    Upstream ``schedule()`` admits from ``skipped_waiting`` as readily as from
+    ``waiting``, so an aborted request left there is re-selected and raises.
+    """
+    aborted = _StubRequest("req-aborted", RequestStatus.FINISHED_ABORTED)
+    scheduler = _make_scheduler(policy, skipped_waiting=[aborted])
+
+    scheduler._drop_aborted_queued_requests()
+
+    assert list(scheduler.skipped_waiting) == []
+
+
+@pytest.mark.parametrize("policy", _POLICIES)
+def test_sweep_drops_aborted_requests_from_waiting_and_running(policy: SchedulingPolicy) -> None:
+    """``waiting`` and ``running`` stay covered alongside the new queue."""
+    aborted_waiting = _StubRequest("req-waiting", RequestStatus.FINISHED_ABORTED)
+    aborted_running = _StubRequest("req-running", RequestStatus.FINISHED_ABORTED)
+    scheduler = _make_scheduler(
+        policy,
+        waiting=[aborted_waiting],
+        running=[aborted_running],
+    )
+
+    scheduler._drop_aborted_queued_requests()
+
+    assert list(scheduler.waiting) == []
+    assert scheduler.running == []
+
+
+@pytest.mark.parametrize("policy", _POLICIES)
+def test_sweep_leaves_live_requests_intact(policy: SchedulingPolicy) -> None:
+    """The sweep must not over-sweep.
+
+    A request parked in ``skipped_waiting`` on a blocked waiting status is one
+    upstream still intends to schedule; only the aborted entry may go.
+    """
+    blocked = _StubRequest("req-blocked", RequestStatus.WAITING_FOR_STREAMING_REQ, arrival_time=0.0)
+    aborted = _StubRequest("req-aborted", RequestStatus.FINISHED_ABORTED, arrival_time=1.0)
+    live_waiting = _StubRequest("req-live-waiting", RequestStatus.WAITING)
+    live_running = _StubRequest("req-live-running", RequestStatus.RUNNING)
+    scheduler = _make_scheduler(
+        policy,
+        waiting=[live_waiting],
+        skipped_waiting=[blocked, aborted],
+        running=[live_running],
+    )
+
+    scheduler._drop_aborted_queued_requests()
+
+    assert list(scheduler.skipped_waiting) == [blocked]
+    assert list(scheduler.waiting) == [live_waiting]
+    assert scheduler.running == [live_running]

tests/core/sched/test_omni_ar_scheduler_streaming.py

@@ -28,7 +28,9 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 def _make_scheduler(*, stage_id: int = 0) -> OmniARScheduler:
     sched = OmniARScheduler.__new__(OmniARScheduler)
     sched._new_prompt_len_snapshot = {}
-    sched.vllm_config = SimpleNamespace(model_config=SimpleNamespace(stage_id=stage_id))
+    sched.vllm_config = SimpleNamespace(
+        model_config=SimpleNamespace(stage_id=stage_id, max_model_len=None),
+    )
     sched.num_waiting_for_streaming_input = 0
     sched.log_stats = False
     sched.chunk_transfer_adapter = None
@@ -210,6 +212,81 @@ def test_running_decode_step_without_inter_stage_payload_does_not_raise() -> Non
     sched.chunk_transfer_adapter.save_async.assert_not_called()
 
 
+def test_stale_async_frame_is_dropped_before_output_processing() -> None:
+    session = _make_request()
+    session.status = RequestStatus.RUNNING
+    session.num_in_flight_tokens = 2
+    session.num_computed_tokens = session.num_tokens + 1
+    session.num_output_placeholders = 1
+    session.async_tokens_to_discard = 1
+    session.sampling_params = SimpleNamespace(num_logprobs=1)
+    num_computed_tokens = session.num_computed_tokens
+    num_output_placeholders = session.num_output_placeholders
+
+    sched = MagicMock()
+    sched.requests = {session.request_id: session}
+    sched.perf_metrics = None
+    sched.structured_output_manager.should_advance.return_value = False
+
+    def discard_stale_output(request: Request, token_ids: list[int]) -> tuple[list[int], bool]:
+        request.async_tokens_to_discard = 0
+        return token_ids, False
+
+    sched._update_request_with_output.side_effect = discard_stale_output
+    sched._process_kv_transfer_trigger.return_value = False
+    sched.chunk_transfer_adapter = MagicMock()
+    sched.running = [session]
+    sched.waiting_for_transfer_free = set()
+    sched.transfer_triggered_requests = set()
+    sched.active_kv_transfers = set()
+    sched.pending_stop_after_extraction = set()
+    sched.connector = None
+    sched.kv_cache_manager.take_events.return_value = None
+    sched.finished_req_ids_dict = {}
+    sched.make_stats.return_value = None
+
+    scheduler_output = MagicMock(spec=SchedulerOutput)
+    scheduler_output.num_scheduled_tokens = {session.request_id: 1}
+    scheduler_output.scheduled_spec_decode_tokens = {}
+    scheduler_output.num_invalid_spec_tokens = 0
+
+    model_runner_output = MagicMock(spec=ModelRunnerOutput)
+    model_runner_output.sampled_token_ids = [[42]]
+    model_runner_output.logprobs = None
+    model_runner_output.prompt_logprobs_dict = {}
+    model_runner_output.pooler_output = None
+    model_runner_output.multimodal_outputs = None
+    model_runner_output.inter_stage_outputs = [{"hidden": object()}]
+    model_runner_output.num_nans_in_logits = None
+    model_runner_output.kv_connector_output = None
+    model_runner_output.cudagraph_stats = None
+    model_runner_output.req_id_to_index = {session.request_id: 0}
+    model_runner_output.routed_experts = None
+
+    OmniARScheduler.update_from_output(sched, scheduler_output, model_runner_output)
+
+    assert session.async_tokens_to_discard == 0
+    assert session.status == RequestStatus.RUNNING
+    assert session.num_computed_tokens == num_computed_tokens
+    assert session.num_output_placeholders == num_output_placeholders
+    sched.chunk_transfer_adapter.save_async.assert_not_called()
+
+    session.sampling_params = SamplingParams(max_tokens=8)
+    next_payload = {"hidden": object()}
+    sched._update_request_with_output.side_effect = None
+    sched._update_request_with_output.return_value = ([43], False)
+    model_runner_output.sampled_token_ids = [[43]]
+    model_runner_output.inter_stage_outputs = [next_payload]
+
+    OmniARScheduler.update_from_output(sched, scheduler_output, model_runner_output)
+
+    sched.chunk_transfer_adapter.save_async.assert_called_once_with(
+        next_payload,
+        session,
+        False,
+    )
+
+
 def test_stage0_streaming_update_discards_outstanding_async_placeholder_token() -> None:
     sched = _make_scheduler(stage_id=0)
     session = _make_request()
@@ -354,3 +431,34 @@ def test_ready_async_chunk_prompt_replacement_releases_stale_kv_once() -> None:
     assert sched.chunk_transfer_adapter.replaced_streaming_prompt_ids == set()
     assert sched.chunk_transfer_adapter.requests_with_ready_chunks == {session.request_id}
     assert sched.chunk_transfer_adapter.requests_num_chunks_sent == {}
+
+
+def test_native_duplex_prompt_reset_prevents_stage1_context_overflow() -> None:
+    sched = _make_scheduler(stage_id=1)
+    sched.vllm_config.model_config.max_model_len = 4096
+    sched.chunk_transfer_adapter = SimpleNamespace(
+        receives_chunks=False,
+        segment_finished_requests=set(),
+    )
+    session = _make_request()
+    session.status = RequestStatus.WAITING_FOR_STREAMING_REQ
+    session.resumable = True
+    session.prompt_token_ids = [0] * 3
+    session._all_token_ids.clear()
+    session._all_token_ids.extend([0] * 4095)
+    session._output_token_ids.extend([1] * 4080)
+    session.num_prompt_tokens = 3
+    session.num_computed_tokens = 4095
+
+    update = _make_update([0] * 16)
+    update.model_intermediate_buffer = {"native_duplex": True}
+
+    sched._update_request_as_session(session, update)
+
+    assert session.prompt_token_ids == [0] * 16
+    assert list(session._all_token_ids) == [0] * 16
+    assert session._output_token_ids == []
+    assert session.num_prompt_tokens == 16
+    assert session.num_computed_tokens == 0
+    assert session.model_intermediate_buffer == update.model_intermediate_buffer
+    assert session.status == RequestStatus.WAITING

tests/core/sched/test_omni_ar_scheduler_unified_decode_deferral.py

@@ -76,6 +76,9 @@ def _make_scheduler(
         else:
             hf_config.voxcpm2_runtime_config = runtime_config
     sched.vllm_config = SimpleNamespace(model_config=model_config)
+    # Upstream ``Scheduler.__init__`` always creates ``skipped_waiting``; the
+    # abort sweep at the top of ``schedule()`` reads it.
+    sched.skipped_waiting = _MockQueue()
     return sched
 
 

tests/core/sched/test_omni_scheduler_streaming_input_counter.py

@@ -0,0 +1,254 @@
+"""Tests for the ``num_waiting_for_streaming_input`` resync placed in
+``OmniARScheduler.finish_requests`` / ``OmniGenerationScheduler.finish_requests``
+and in ``OmniARScheduler.schedule``.
+
+Upstream keeps that counter incrementally (``+1`` when a resumable request
+parks in a waiting queue as ``WAITING_FOR_STREAMING_REQ``, ``-1`` when the
+next update arrives or the request is finished) and
+``Scheduler.get_num_unfinished_requests`` *subtracts* it from the waiting
+queues. Omni rewrites ``request.status`` outside both hooks -- the
+chunk-transfer adapter's park/restore and
+``_realign_request_status_to_queues`` -- so a counted request can leave
+``WAITING_FOR_STREAMING_REQ`` without the matching decrement and inflate the
+counter for the lifetime of the process.
+
+An inflated counter does not degrade throughput, it hangs the stage:
+``EngineCore.has_work()`` reads false while a live request sits in
+``waiting``, so the engine blocks in ``input_queue.get()`` and never calls
+``schedule()`` again. Regression for the Qwen3-Omni duplex symptom where the
+server answered the first conversation after a boot and was silent from the
+second on while still reporting healthy.
+"""
+
+from __future__ import annotations
+
+import pytest
+from vllm.v1.request import RequestStatus
+
+import vllm_omni.core.sched.omni_ar_scheduler as ar_sched_mod
+import vllm_omni.core.sched.omni_generation_scheduler as gen_sched_mod
+from vllm_omni.core.sched.omni_ar_scheduler import OmniARScheduler
+from vllm_omni.core.sched.omni_generation_scheduler import OmniGenerationScheduler
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _StubRequest:
+    def __init__(self, request_id: str, status: RequestStatus) -> None:
+        self.request_id = request_id
+        self.status = status
+
+    def is_finished(self) -> bool:
+        return RequestStatus.is_finished(self.status)
+
+
+_SCHEDULER_PARAMS = [
+    pytest.param(OmniARScheduler, ar_sched_mod, id="ar"),
+    pytest.param(OmniGenerationScheduler, gen_sched_mod, id="generation"),
+]
+
+
+def _make_scheduler(scheduler_cls, *, requests, running, waiting, counter, skipped=None):
+    scheduler = scheduler_cls.__new__(scheduler_cls)
+    scheduler.chunk_transfer_adapter = None
+    scheduler.input_coordinator = None
+    scheduler.requests = requests
+    scheduler.running = running
+    scheduler.waiting = waiting
+    scheduler.skipped_waiting = skipped if skipped is not None else []
+    scheduler.num_waiting_for_streaming_input = counter
+    return scheduler
+
+
+def _upstream_num_unfinished(scheduler) -> int:
+    """Mirror of ``Scheduler.get_num_unfinished_requests`` arithmetic."""
+    num_waiting = (
+        len(scheduler.waiting) + len(scheduler.skipped_waiting) - scheduler.num_waiting_for_streaming_input
+    )
+    return num_waiting + len(scheduler.running)
+
+
+@pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS)
+def test_finish_requests_clears_counter_left_by_a_stomped_status(
+    monkeypatch: pytest.MonkeyPatch,
+    scheduler_cls,
+    scheduler_mod,
+) -> None:
+    """The abort case that wedged the talker.
+
+    The request was counted while parked for streaming input, then had its
+    status rewritten to ``RUNNING`` by omni's queue surgery, so upstream's
+    ``finish_requests`` took the running branch and never decremented. With
+    no parked requests left, the counter must land at 0.
+    """
+    aborted = _StubRequest("req-aborted", RequestStatus.RUNNING)
+    scheduler = _make_scheduler(
+        scheduler_cls,
+        requests={},  # upstream already dropped it
+        running=[],
+        waiting=[],
+        counter=1,
+    )
+
+    monkeypatch.setattr(
+        scheduler_mod.VLLMScheduler,
+        "finish_requests",
+        lambda self, request_ids, finished_status: [aborted],
+    )
+
+    scheduler_cls.finish_requests(scheduler, [aborted.request_id], RequestStatus.FINISHED_ABORTED)
+
+    assert scheduler.num_waiting_for_streaming_input == 0
+
+
+@pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS)
+def test_next_request_is_visible_as_work_after_the_abort(
+    monkeypatch: pytest.MonkeyPatch,
+    scheduler_cls,
+    scheduler_mod,
+) -> None:
+    """The consequence, stated as the engine sees it.
+
+    A fresh request queued behind the leak must read as unfinished work.
+    Without the resync this is ``(1 + 0 - 1) + 0 == 0`` and the engine parks
+    in ``input_queue.get()`` with a live request in ``waiting``.
+    """
+    aborted = _StubRequest("req-aborted", RequestStatus.RUNNING)
+    incoming = _StubRequest("req-next-session", RequestStatus.WAITING)
+    scheduler = _make_scheduler(
+        scheduler_cls,
+        requests={incoming.request_id: incoming},
+        running=[],
+        waiting=[incoming],
+        counter=1,
+    )
+
+    monkeypatch.setattr(
+        scheduler_mod.VLLMScheduler,
+        "finish_requests",
+        lambda self, request_ids, finished_status: [aborted],
+    )
+
+    scheduler_cls.finish_requests(scheduler, [aborted.request_id], RequestStatus.FINISHED_ABORTED)
+
+    assert _upstream_num_unfinished(scheduler) == 1
+
+
+@pytest.mark.parametrize(("scheduler_cls", "scheduler_mod"), _SCHEDULER_PARAMS)
+def test_finish_requests_keeps_counting_a_genuinely_parked_request(
+    monkeypatch: pytest.MonkeyPatch,
+    scheduler_cls,
+    scheduler_mod,
+) -> None:
+    """The resync must not over-correct.
+
+    A request still parked as ``WAITING_FOR_STREAMING_REQ`` in ``waiting`` is
+    exactly what the counter is for -- it must stay counted, so the engine
+    does not spin on a request that has no input to process.
+    """
+    parked = _StubRequest("req-parked", RequestStatus.WAITING_FOR_STREAMING_REQ)
+    leaving = _StubRequest("req-leaving", RequestStatus.FINISHED_STOPPED)
+    scheduler = _make_scheduler(
+        scheduler_cls,
+        requests={parked.request_id: parked},
+        running=[],
+        waiting=[parked],
+        counter=1,
+    )
+
+    monkeypatch.setattr(
+        scheduler_mod.VLLMScheduler,
+        "finish_requests",
+        lambda self, request_ids, finished_status: [leaving],
+    )
+
+    scheduler_cls.finish_requests(scheduler, [leaving.request_id], RequestStatus.FINISHED_STOPPED)
+
+    assert scheduler.num_waiting_for_streaming_input == 1
+    assert _upstream_num_unfinished(scheduler) == 0
+
+
+def test_resync_counts_parked_requests_in_skipped_waiting() -> None:
+    """``get_num_unfinished_requests`` sums ``waiting`` and ``skipped_waiting``,
+    so the resync has to look at both or it under-counts and reintroduces the
+    spin the counter exists to prevent."""
+    parked = _StubRequest("req-skipped", RequestStatus.WAITING_FOR_STREAMING_REQ)
+    scheduler = _make_scheduler(
+        OmniARScheduler,
+        requests={parked.request_id: parked},
+        running=[],
+        waiting=[],
+        counter=0,
+        skipped=[parked],
+    )
+
+    scheduler._resync_streaming_input_counter()
+
+    assert scheduler.num_waiting_for_streaming_input == 1
+
+
+def test_resync_is_a_no_op_when_upstream_accounting_is_correct() -> None:
+    """In the healthy case the derived value equals upstream's incremental one,
+    so the helper must leave it alone rather than fight it."""
+    parked = _StubRequest("req-parked", RequestStatus.WAITING_FOR_STREAMING_REQ)
+    running = _StubRequest("req-running", RequestStatus.RUNNING)
+    scheduler = _make_scheduler(
+        OmniARScheduler,
+        requests={r.request_id: r for r in (parked, running)},
+        running=[running],
+        waiting=[parked],
+        counter=1,
+    )
+
+    scheduler._resync_streaming_input_counter()
+
+    assert scheduler.num_waiting_for_streaming_input == 1
+
+
+def test_resync_tolerates_a_scheduler_without_the_counter() -> None:
+    """Guard for vLLM versions that predate ``num_waiting_for_streaming_input``:
+    the helper must no-op rather than invent the attribute."""
+    scheduler = OmniARScheduler.__new__(OmniARScheduler)
+    scheduler.waiting = []
+    scheduler.skipped_waiting = []
+
+    scheduler._resync_streaming_input_counter()
+
+    assert not hasattr(scheduler, "num_waiting_for_streaming_input")
+
+
+def test_ar_schedule_resyncs_before_delegating_upstream(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Pins placement in ``schedule``: the chunk-transfer adapter rewrites
+    status during ``process_pending_chunks``, so the resync has to run after
+    that and before upstream schedules, while ``self.waiting`` is still the
+    real queue."""
+    parked = _StubRequest("req-resumed", RequestStatus.RUNNING)
+    scheduler = OmniARScheduler.__new__(OmniARScheduler)
+    scheduler.chunk_transfer_adapter = None
+    scheduler.input_coordinator = None
+    scheduler.requests = {parked.request_id: parked}
+    scheduler.running = [parked]
+    scheduler.waiting = []
+    scheduler.skipped_waiting = []
+    scheduler.num_waiting_for_streaming_input = 1  # leaked by the status rewrite
+    scheduler.policy = "fcfs"
+
+    seen: dict[str, int] = {}
+
+    monkeypatch.setattr(ar_sched_mod.OmniSchedulerMixin, "_consume_pending_connector_output", lambda self, model_mode: None)
+    monkeypatch.setattr(ar_sched_mod.OmniSchedulerMixin, "_process_pending_input_timeouts", lambda self: None)
+    monkeypatch.setattr(OmniARScheduler, "_should_defer_waiting_admission", lambda self: False)
+
+    def fake_schedule(self, throttle_prefills=False):
+        seen["counter"] = self.num_waiting_for_streaming_input
+        raise _StopScheduleError
+
+    class _StopScheduleError(Exception):
+        pass
+
+    monkeypatch.setattr(ar_sched_mod.VLLMScheduler, "schedule", fake_schedule)
+
+    with pytest.raises(_StopScheduleError):
+        OmniARScheduler.schedule(scheduler)
+
+    assert seen["counter"] == 0

tests/distributed/omni_connectors/test_chunk_transfer_adapter.py

@@ -546,7 +546,7 @@ def test_personaplex_sender_cleanup_drops_delayed_frame_state(build_adapter):
     assert second.codes is None
 
 
-def test_save_async_skips_stale_resumable_chunk_until_dedup_is_reset(build_adapter):
+def test_save_async_skips_stale_resumable_chunk_within_segment(build_adapter):
     adapter, _ = build_adapter(stage_id=1)
     request = _req("req-stream", RequestStatus.WAITING, external_req_id="ext-stream")
     request.resumable = True
@@ -565,6 +565,32 @@ def test_save_async_skips_stale_resumable_chunk_until_dedup_is_reset(build_adapt
     assert adapter.requests_num_chunks_sent["ext-stream"] == 0
 
 
+def test_save_async_drops_late_previous_segment_after_boundary_reset(build_adapter):
+    adapter, _ = build_adapter(stage_id=1)
+    request = _req("req-stream", RequestStatus.WAITING, external_req_id="ext-stream")
+    request.resumable = True
+    request._omni_segment_generation = 0
+    adapter.requests_num_chunks_sent["ext-stream"] = 111
+
+    adapter.save_async(multimodal_output=None, request=request, is_segment_finished=True)
+
+    late = _req("req-stream-late", RequestStatus.WAITING, external_req_id="ext-stream")
+    late.resumable = True
+    late.num_computed_tokens = 111
+    late._omni_segment_generation = 0
+    adapter.save_async(multimodal_output=None, request=late, is_segment_finished=False)
+
+    next_segment = _req("req-stream-next", RequestStatus.WAITING, external_req_id="ext-stream")
+    next_segment.resumable = True
+    next_segment.num_computed_tokens = 3
+    next_segment._omni_segment_generation = 1
+    adapter.save_async(multimodal_output=None, request=next_segment, is_segment_finished=False)
+
+    assert len(adapter._pending_save_reqs) == 2
+    assert adapter.requests_num_chunks_sent["ext-stream"] == 3
+    assert adapter._pending_save_reqs[-1]["request"] is next_segment
+
+
 def test_send_single_request_cleans_up_after_finished_payload(build_adapter, monkeypatch):
     adapter, _ = build_adapter(stage_id=1)
     request = _req("req-finished", RequestStatus.FINISHED_STOPPED, external_req_id="ext-finished")

tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py

@@ -0,0 +1,107 @@
+"""MiniCPM-o server-VAD hard-interruption acceptance driver."""
+
+from __future__ import annotations
+
+import base64
+from pathlib import Path
+
+from vllm_omni.experimental.fullduplex.client import (
+    PCM16_BYTES_PER_SAMPLE,
+    PCM16_SAMPLE_RATE,
+    RealtimeDuplexClient,
+    RealtimeEventCollector,
+    build_realtime_url,
+    read_pcm16_wav,
+    wait_for,
+)
+
+
+def _events(events: list[dict[str, object]], response_id: str, kind: str) -> list[dict[str, object]]:
+    return [
+        event
+        for event in events
+        if event.get("type") == kind and RealtimeEventCollector.response_id(event) == response_id
+    ]
+
+
+async def run_server_vad_interrupt(args) -> dict[str, object]:
+    initial = read_pcm16_wav(Path(args.input_wav))
+    interrupt = read_pcm16_wav(Path(args.interrupt_wav))
+    interrupt = interrupt[: PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * 4700 // 1000]
+    session_id = f"server-vad-hard-interrupt-{id(args)}"
+    client = RealtimeDuplexClient(build_realtime_url(args.url, args.model, autostart=False, session_id=session_id))
+    try:
+        await client.__aenter__()
+        await client.configure(
+            args.model,
+            ref_audio="data:audio/wav;base64," + base64.b64encode(Path(args.ref_audio).read_bytes()).decode(),
+            session_id=session_id,
+            turn_detection={
+                "type": "server_vad",
+                "interrupt_response": True,
+            },
+            timeout_s=args.timeout_s,
+        )
+        await client.stream_pcm16(initial + bytes(16_000 * 2 * 1600 // 1000), chunk_ms=200, realtime=True)
+        await client.commit()
+        await wait_for(
+            lambda: any(e.get("type") == "response.created" for e in client.events.events),
+            timeout_s=args.timeout_s,
+            label="response.created",
+        )
+        target = next(e for e in client.events.events if e.get("type") == "response.created")
+        target_id = RealtimeEventCollector.response_id(target)
+        assert target_id
+        await wait_for(
+            lambda: _events(client.events.events, target_id, "response.audio.delta"),
+            timeout_s=args.timeout_s,
+            label="response.audio.delta",
+        )
+        cursor = len(client.events.events)
+        await client.stream_pcm16(interrupt + bytes(16_000 * 2 * 800 // 1000), chunk_ms=args.chunk_ms, realtime=True)
+        await wait_for(
+            lambda: _events(client.events.events, target_id, "response.done"),
+            timeout_s=args.timeout_s,
+            label="cancelled response.done",
+        )
+        await client.commit()
+        await wait_for(
+            lambda: any(
+                e.get("type") == "response.done" and RealtimeEventCollector.response_id(e) != target_id
+                for e in client.events.events[cursor:]
+            ),
+            timeout_s=args.timeout_s,
+            label="follow-up response.done",
+        )
+        events = client.events.events
+    finally:
+        try:
+            await client.close_session(timeout_s=args.timeout_s)
+        finally:
+            await client.__aexit__(None, None, None)
+
+    done = _events(events, target_id, "response.done")
+    done_index = events.index(done[0]) if len(done) == 1 else len(events)
+    stale = (
+        any(
+            e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) == target_id
+            for e in events[done_index + 1 :]
+        )
+        if len(done) == 1
+        else False
+    )
+    followup_audio = any(
+        e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) != target_id
+        for e in events[cursor:]
+    )
+    ok = (
+        len(done) == 1
+        and isinstance(done[0].get("response"), dict)
+        and done[0]["response"].get("status") == "cancelled"
+        and done[0]["response"].get("status_details", {}).get("reason") == "turn_detected"
+        and not stale
+        and any(e.get("type") == "input_audio_buffer.committed" for e in events[cursor:])
+        and followup_audio
+        and not any(e.get("type") == "error" for e in events)
+    )
+    return {"ok": ok}

tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py

@@ -21,6 +21,7 @@ from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import (
 from tests.e2e.online_serving.run_minicpmo_realtime_duplex_multi_session import (
     run_lifecycle_probes,
 )
+from tests.e2e.online_serving.run_minicpmo_realtime_duplex_server_vad import run_server_vad_interrupt
 from tests.e2e.online_serving.run_minicpmo_realtime_duplex_soft_interrupt import (
     run_soft_interrupt,
 )
@@ -88,3 +89,22 @@ def test_duplex_soft_interrupt(omni_server, model_prefix: str, tmp_path: Path) -
     assert result["response_lifecycle_ok"] is True
     assert result["response_audio_contract_ok"] is True
     assert result["followup_response_transcript_ok"] is True
+
+
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server", SERVER_PARAMS, indirect=True)
+def test_duplex_server_vad_hard_interrupt(omni_server, model_prefix: str) -> None:
+    result = asyncio.run(
+        run_server_vad_interrupt(
+            SimpleNamespace(
+                url=realtime_url(omni_server),
+                model=omni_server.model,
+                input_wav=str(validated_input_wav()),
+                interrupt_wav=str(validated_soft_interrupt_wav()),
+                ref_audio=str(resolve_ref_audio(model_prefix)),
+                chunk_ms=200,
+                timeout_s=180.0,
+            )
+        )
+    )
+    assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2)

tests/engine/test_orchestrator.py

@@ -30,6 +30,7 @@ from vllm_omni.engine.messages import (
 from vllm_omni.engine.orchestrator import (
     Orchestrator,
     OrchestratorRequestState,
+    StreamingSegmentState,
     _build_terminal_empty_output,
 )
 from vllm_omni.engine.stage_pool import StagePool
@@ -1951,7 +1952,7 @@ async def test_resumable_segment_boundary_builds_stage_metrics() -> None:
         final_stage_id=0,
     )
     req_state.streaming.enabled = True
-    req_state.streaming.segment_finished = True
+    req_state.streaming.segments[0] = StreamingSegmentState(finished=True)
     req_state.stage_submit_ts[0] = time.time()
     orchestrator.request_states = {"req-stream": req_state}
     orchestrator.stage_pools = [pool]

tests/engine/test_orchestrator_error_handling.py

@@ -880,6 +880,39 @@ async def test_diffusion_client_error_output_propagates_status_code(orchestrator
         orchestrator_fixture.thread.join(timeout=5)
 
 
+@pytest.mark.asyncio
+async def test_duplex_input_processor_failure_is_request_scoped(orchestrator_factory, monkeypatch) -> None:
+    class FailingStage(FakeStageClient):
+        def process_engine_inputs(self, source_outputs, prompt=None, streaming_context=None):
+            del source_outputs, prompt, streaming_context
+            raise ValueError("No latent or hidden_states found in thinker output")
+
+    fixture = orchestrator_factory(
+        [FakeStageClient(stage_type="llm"), FailingStage(stage_type="llm", final_output=True)]
+    )
+    state = OrchestratorRequestState(
+        request_id="bad",
+        prompt=SimpleNamespace(request_id="bad", prompt_token_ids=[1]),
+        sampling_params_list=[_sampling_params(), _sampling_params()],
+        final_stage_id=1,
+        duplex_identity=SimpleNamespace(),
+    )
+
+    async def no_cleanup(*_args: object, **_kwargs: object) -> None:
+        return None
+
+    monkeypatch.setattr(fixture.orchestrator, "_cleanup_request_ids", no_cleanup)
+    try:
+        await fixture.orchestrator._forward_to_next_stage_unguarded(
+            "bad", 0, _build_request_output("raw", finished=True), state
+        )
+        error = await _wait_for_error_message(fixture, request_id="bad")
+        assert error.fatal is False
+    finally:
+        fixture.request_sync_q.put_nowait(ShutdownRequestMessage())
+        fixture.thread.join(timeout=5)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("status_code", "error_type"),

tests/engine/test_orchestrator_segment_stage_keying.py

@@ -0,0 +1,130 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Streaming segment-boundary state must be keyed by stage.
+
+``Orchestrator._orchestration_loop`` polls every stage into the same
+``OrchestratorRequestState``. While the segment-boundary fields lived in a single
+flat slot per request, whichever stage polled most recently overwrote the others,
+and the per-stage consumers then read back a different stage's boundary.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+import pytest
+from vllm.sampling_params import SamplingParams
+
+from vllm_omni.engine.orchestrator import (
+    Orchestrator,
+    OrchestratorRequestState,
+    StreamingSegmentState,
+)
+from vllm_omni.experimental.fullduplex.engine.contracts import DuplexRequestIdentity
+from vllm_omni.experimental.fullduplex.engine.messages import DuplexFence
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+THINKER_STAGE = 0
+TALKER_STAGE = 1
+
+
+@dataclass
+class _StageMetrics:
+    pipeline_timings: dict[str, float] = field(default_factory=dict)
+
+
+@dataclass
+class _StageOutput:
+    """Stand-in for a processed stage output, with the fields the router reads."""
+
+    request_id: str
+    finished: bool = False
+    error: object | None = None
+
+
+class _RecordingPool:
+    """Stage pool that records the requests it was asked to build metrics for."""
+
+    def __init__(self) -> None:
+        self.built_for: list[str] = []
+
+    def build_stage_metrics(self, outputs: list[_StageOutput], **_kwargs: object) -> _StageMetrics:
+        self.built_for.extend(output.request_id for output in outputs)
+        return _StageMetrics()
+
+
+def _duplex_req_state(request_id: str = "req-duplex") -> OrchestratorRequestState:
+    req_state = OrchestratorRequestState(
+        request_id=request_id,
+        sampling_params_list=[SamplingParams(max_tokens=1), SamplingParams(max_tokens=1)],
+        final_stage_id=TALKER_STAGE,
+    )
+    req_state.streaming.enabled = True
+    req_state.duplex_identity = DuplexRequestIdentity(
+        session_id="session-1",
+        fence=DuplexFence(session_id="session-1"),
+    )
+    return req_state
+
+
+def test_duplex_output_context_reads_only_its_own_stage_segment() -> None:
+    req_state = _duplex_req_state()
+    req_state.streaming.segments[THINKER_STAGE] = StreamingSegmentState(
+        finished=True,
+        token_ids=[11, 22],
+        output_metadata={"stage": "thinker"},
+    )
+    # The talker polls after the thinker and is still mid-segment.
+    req_state.streaming.segments[TALKER_STAGE] = StreamingSegmentState(finished=False)
+
+    thinker = Orchestrator._duplex_output_context(req_state, stage_id=THINKER_STAGE)
+    talker = Orchestrator._duplex_output_context(req_state, stage_id=TALKER_STAGE)
+
+    assert thinker is not None
+    assert talker is not None
+
+    # These are the assertions that fail on the flat field: the talker is written
+    # second, so the thinker loses its boundary to whichever stage polled last.
+    assert thinker.segment_finished is True
+    assert thinker.segment_token_ids == (11, 22)
+    assert thinker.segment_output_metadata == {"stage": "thinker"}
+
+    assert talker.segment_finished is False
+    assert talker.segment_token_ids == ()
+    assert talker.segment_output_metadata == {}
+
+
+@pytest.mark.asyncio
+async def test_segment_boundary_on_one_stage_does_not_build_metrics_on_another() -> None:
+    req_state = _duplex_req_state()
+    req_state.streaming.segments[THINKER_STAGE] = StreamingSegmentState(finished=True)
+
+    talker_pool = _RecordingPool()
+    orchestrator = object.__new__(Orchestrator)
+    orchestrator.request_states = {req_state.request_id: req_state}
+    orchestrator.stage_pools = [_RecordingPool(), talker_pool]
+
+    routed: list[_StageMetrics | None] = []
+
+    async def record_route(
+        _stage: int,
+        _replica: int,
+        _output: _StageOutput,
+        _state: OrchestratorRequestState,
+        metrics: _StageMetrics | None,
+    ) -> None:
+        routed.append(metrics)
+
+    orchestrator._route_output = record_route
+
+    # The talker's own output is neither finished nor at a segment boundary, so
+    # it must not be treated as one just because the thinker reached one.
+    await orchestrator._handle_processed_outputs(
+        TALKER_STAGE,
+        0,
+        [_StageOutput(request_id=req_state.request_id)],
+    )
+
+    assert talker_pool.built_for == []
+    assert routed == [None]

tests/engine/test_orchestrator_stage_input_bridge.py

@@ -17,6 +17,7 @@ from vllm.sampling_params import SamplingParams
 from vllm_omni.engine.orchestrator import (
     Orchestrator,
     OrchestratorRequestState,
+    StreamingSegmentState,
     _OrchestratorDuplexStagePort,
 )
 from vllm_omni.engine.stage_pool import StagePool
@@ -358,7 +359,7 @@ async def test_streaming_segment_does_not_complete_final_output_stage() -> None:
         final_output_stage_ids={0},
     )
     req_state.streaming.enabled = True
-    req_state.streaming.segment_finished = True
+    req_state.streaming.segments[0] = StreamingSegmentState(finished=True)
     output = SimpleNamespace(
         request_id=req_state.request_id,
         finished=True,

tests/entrypoints/openai/test_duplex_protocol.py

@@ -452,7 +452,7 @@ def test_minicpmo_native_capabilities_separate_model_state_from_core_kv_lease():
     assert caps["supports_core_resumable_request"] is True
     assert caps["supports_stage_connector_handoff"] is True
     assert caps["supports_audio_truncate"] is True
-    assert caps["supports_barge_in"] is False
+    assert caps["supports_barge_in"] is True
     assert caps["target_barge_in_latency_ms"] is None
     assert caps["supports_multi_session"] is True
     assert caps["supports_multi_session_same_replica"] is True

tests/entrypoints/openai_api/test_duplex_handler.py

@@ -652,7 +652,7 @@ async def test_native_realtime_protocol_preserves_input_turn_policy_hints():
 
 
 @pytest.mark.asyncio
-async def test_native_realtime_protocol_rejects_unimplemented_server_vad():
+async def test_native_realtime_protocol_rejects_server_vad_without_interrupt():
     ws = TimedWebSocket()
     protocol = NativeRealtimeSessionProtocol(ws)  # type: ignore[arg-type]
     protocol.bind_sender(ws.send_json)
@@ -675,7 +675,10 @@ async def test_native_realtime_protocol_rejects_unimplemented_server_vad():
     error = ws.sent[-1]
     assert error["type"] == "error"
     assert error["error"]["code"] == "unsupported_turn_detection"
-    assert "server_vad" in error["error"]["message"]
+    assert "interrupt_response=false" in error["error"]["message"]
+    assert "greater than 0.15" in str(
+        protocol._validate_realtime_turn_detection({"turn_detection": {"type": "server_vad", "threshold": 0.15}})
+    )
 
 
 @pytest.mark.asyncio
@@ -699,7 +702,7 @@ async def test_native_realtime_protocol_accepts_disabled_turn_detection():
 
 
 @pytest.mark.asyncio
-async def test_native_realtime_protocol_rejects_nested_vad_even_when_top_level_is_disabled():
+async def test_native_realtime_protocol_prefers_nested_vad_over_top_level():
     ws = TimedWebSocket()
     protocol = NativeRealtimeSessionProtocol(ws)  # type: ignore[arg-type]
     protocol.bind_sender(ws.send_json)
@@ -719,8 +722,18 @@ async def test_native_realtime_protocol_rejects_nested_vad_even_when_top_level_i
         }
     )
 
-    assert translated is None
-    assert ws.sent[-1]["error"]["code"] == "unsupported_turn_detection"
+    assert translated["session"]["extra_body"]["realtime_session_payload"]["overlap_policy"] == "barge_in_on_speech"
+
+
+@pytest.mark.asyncio
+async def test_realtime_vad_candidate_waits_for_session_ack():
+    protocol = NativeRealtimeSessionProtocol(TimedWebSocket())  # type: ignore[arg-type]
+    event = {"type": "session.update", "model": "test-model", "turn_detection": {"type": "server_vad"}}
+
+    for outbound, enabled in (({"type": "error"}, False), ({"type": "session.updated", "session": {}}, True)):
+        await protocol._to_duplex_event(event)
+        protocol.encode_outbound_event(outbound)
+        assert (protocol._server_vad is not None) is enabled
 
 
 def test_native_duplex_handler_has_no_fixed_session_admission_cap():
@@ -5674,7 +5687,7 @@ async def test_minicpmo_native_duplex_rejects_ref_audio_path():
     assert engine.opened == []
 
 
-def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen():
+def test_minicpmo_native_duplex_explicit_barge_in_request_interrupts():
     engine = FakeEngineClient()
     handler = OmniDuplexSessionHandler(
         chat_service=FakeChatService(engine),
@@ -5701,12 +5714,10 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen(
     )
 
     assert decision == {
-        "action": "listen",
-        "reason": "barge_in_unsupported",
+        "action": "barge_in",
+        "reason": "client_force_barge_in",
         "duration_ms": 1000,
-        "overlap_speech_ms": 1000,
         "buffer_audio": True,
-        "defer_runtime_append": True,
     }
 
 
@@ -5718,7 +5729,7 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen(
         {"type": "turn.signal", "event": "barge_in"},
     ],
 )
-async def test_minicpmo_native_duplex_rejects_unadvertised_barge_in_control(
+async def test_minicpmo_native_duplex_accepts_explicit_barge_in_control(
     barge_in_event: dict[str, object],
 ):
     engine = FakeEngineClient()
@@ -5734,9 +5745,8 @@ async def test_minicpmo_native_duplex_rejects_unadvertised_barge_in_control(
 
     await handler.handle_session(ws)
 
-    error = next(message for message in ws.sent if message.get("code") == "barge_in_unsupported")
-    assert error["session_id"] == "sid-native-barge-control-disabled"
-    assert ("sid-native-barge-control-disabled", "barge_in") not in engine.signals
+    assert not any(message.get("code") == "barge_in_unsupported" for message in ws.sent)
+    assert ("sid-native-barge-control-disabled", "barge_in") in engine.signals
 
 
 @pytest.mark.asyncio

vllm_omni/core/sched/omni_ar_scheduler.py

@@ -218,16 +218,21 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         if stop_after_transfer and req_id in self.requests_needing_kv_transfer:
             self.pending_stop_after_extraction.add(req_id)
 
+    def _drop_aborted_queued_requests(self) -> None:
+        for queue in (self.waiting, self.skipped_waiting):
+            aborted = [req for req in queue if req.status == RequestStatus.FINISHED_ABORTED]
+            if aborted:
+                queue.remove_requests(aborted)
+        self.running[:] = [req for req in self.running if req.status != RequestStatus.FINISHED_ABORTED]
+
     def schedule(self, throttle_prefills: bool = False) -> SchedulerOutput:
         # Remove FINISHED_ABORTED requests before the upstream scheduler sees
         # them. Upstream vllm raises RuntimeError on this status; omni allows
         # async abort (e.g. client disconnect during TTS streaming) to leave
         # requests in the waiting/running queues temporarily.
-        for queue in (self.waiting, self.running):
-            for req in list(queue):
-                if getattr(req, "status", None) == RequestStatus.FINISHED_ABORTED:
-                    queue.remove(req)
+        self._drop_aborted_queued_requests()
         self._process_pending_omni_inputs(model_mode="ar")
+        self._resync_streaming_input_counter()
 
         original_waiting = None
         if self._should_defer_waiting_admission():
@@ -354,6 +359,11 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
 
             req_index = model_runner_output.req_id_to_index[req_id]
             generated_token_ids = sampled_token_ids[req_index] if sampled_token_ids else []
+
+            if generated_token_ids and getattr(request, "async_tokens_to_discard", 0) > 0:
+                self._update_request_with_output(request, generated_token_ids)
+                continue
+
             status_before_stop = request.status
             new_logprobs = None
             logprob_validation_failed = False
@@ -651,6 +661,7 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
         if self.chunk_transfer_adapter and self.chunk_transfer_adapter.receives_chunks:
             self.chunk_transfer_adapter.requests_num_chunks_sent.pop(session.external_req_id, None)
             if stage_id != 0:
+                session._omni_segment_generation = int(getattr(session, "_omni_segment_generation", 0) or 0) + 1
                 # Downstream async-chunk stages receive real payloads from the
                 # connector. This update only resumes polling for the next segment.
                 self.chunk_transfer_adapter.segment_finished_requests.discard(session.request_id)
@@ -682,6 +693,30 @@ class OmniARScheduler(OmniSchedulerMixin, VLLMScheduler):
             self._release_replaced_streaming_prompt_cache(session)
             self._replace_streaming_session(session, update)
             return
+        native_duplex_update = any(
+            isinstance(info, dict) and info.get("native_duplex") is True for info in update_infos
+        )
+        if native_duplex_update:
+            max_model_len = int(getattr(self.vllm_config.model_config, "max_model_len", 0) or 0)
+            current_prompt_len = len(getattr(session, "prompt_token_ids", None) or ())
+            computed_tokens = int(getattr(session, "num_computed_tokens", 0) or 0)
+            kept_output_len = max(0, computed_tokens - current_prompt_len)
+            next_prompt_len = len(update.prompt_token_ids or ())
+            candidate_len = current_prompt_len + kept_output_len + next_prompt_len
+            # Keep one context position available for the next sampled codec token.
+            if max_model_len > 0 and candidate_len >= max_model_len:
+                logger.info(
+                    "Resetting native duplex Stage-1 prompt before context overflow: "
+                    "req=%s current=%s kept_output=%s next=%s max_model_len=%s",
+                    session.request_id,
+                    current_prompt_len,
+                    kept_output_len,
+                    next_prompt_len,
+                    max_model_len,
+                )
+                self._replace_streaming_session(session, update)
+                return
+        session._omni_segment_generation = int(getattr(session, "_omni_segment_generation", 0) or 0) + 1
         super()._update_request_as_session(session, update)
         if hasattr(update, "model_intermediate_buffer"):
             session.model_intermediate_buffer = update.model_intermediate_buffer

vllm_omni/core/sched/omni_scheduler_mixin.py

@@ -146,6 +146,7 @@ class OmniSchedulerMixin:
 
     def _replace_streaming_session(self, session: Request, update: StreamingUpdate) -> None:
         """Replace a downstream stage's placeholder with its next payload."""
+        session._omni_segment_generation = int(getattr(session, "_omni_segment_generation", 0) or 0) + 1
         adapter = getattr(self, "chunk_transfer_adapter", None)
         if adapter is not None:
             adapter.segment_finished_requests.discard(session.request_id)
@@ -645,6 +646,7 @@ class OmniSchedulerMixin:
         )
         finished = super().finish_requests(request_ids, finished_status)
         self._purge_finished_from_running(target_request_ids)
+        self._resync_streaming_input_counter()
 
         for request in finished:
             self._free_input_coordinator_request(request.request_id)
@@ -810,3 +812,18 @@ class OmniSchedulerMixin:
             return resumable_segment_stop and req.request_id not in target_request_ids
 
         self.running[:] = [req for req in self.running if keep_running(req)]
+
+    def _resync_streaming_input_counter(self) -> None:
+        counter = getattr(self, "num_waiting_for_streaming_input", None)
+        if counter is None:
+            return
+        parked = sum(
+            1
+            for queue in (getattr(self, "waiting", None), getattr(self, "skipped_waiting", None))
+            if queue
+            for request in queue
+            if request.status == RequestStatus.WAITING_FOR_STREAMING_REQ
+        )
+        if parked != counter:
+            logger.debug("[Omni] resynced streaming-input counter %s -> %s", counter, parked)
+            self.num_waiting_for_streaming_input = parked

vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py

@@ -95,6 +95,9 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
         #   been read").
         self._held_non_active: deque[Any] = deque()
         self.requests_num_chunks_sent: dict[str, int] = defaultdict(int)
+        # Boundary tasks advance this before the background sender runs, so a
+        # late old-segment frame cannot restore the previous watermark.
+        self._segment_generation: dict[str, int] = defaultdict(int)
         self._pending_streaming_prefills: dict[str, dict] = {}
         # Monotonic timestamp of when each request last began waiting for a
         # chunk, refreshed every time one arrives.  Read by
@@ -188,11 +191,31 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
             is_segment_finished: whether the segment of request is finished
         """
         is_finished = request.is_finished() and not request.resumable
+        if not hasattr(self, "_segment_generation"):
+            self._segment_generation = defaultdict(int)
+        external_req_id = request.external_req_id
+        raw_generation = getattr(request, "_omni_segment_generation", 0)
+        try:
+            generation = int(raw_generation)
+        except (TypeError, ValueError):
+            generation = 0
+        expected_generation = self._segment_generation.get(external_req_id, generation)
+        if generation < expected_generation:
+            logger.warning(
+                "Skip late save_async for request %s, segment_generation=%s, expected=%s",
+                external_req_id,
+                generation,
+                expected_generation,
+            )
+            return
+        self._segment_generation[external_req_id] = generation
 
         confirmed_num_computed_tokens = self._confirmed_num_computed_tokens(request)
 
         # If the request is preempted, skip the already saved chunks.
-        if confirmed_num_computed_tokens < self.requests_num_chunks_sent.get(request.external_req_id, 0):
+        if not is_segment_finished and confirmed_num_computed_tokens < self.requests_num_chunks_sent.get(
+            request.external_req_id, 0
+        ):
             logger.warning(
                 f"Enqueue save_async for request {request.external_req_id}, "
                 f"request.num_computed_tokens={request.num_computed_tokens}, "
@@ -201,13 +224,18 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
             )
             return
 
-        self.requests_num_chunks_sent[request.external_req_id] = confirmed_num_computed_tokens
         task = {
             "multimodal_output": multimodal_output,
             "request": request,
             "is_finished": is_finished,
             "is_segment_finished": is_segment_finished,
+            "segment_generation": generation,
         }
+        if is_segment_finished:
+            self._segment_generation[external_req_id] = generation + 1
+            self.requests_num_chunks_sent.pop(external_req_id, None)
+        else:
+            self.requests_num_chunks_sent[external_req_id] = confirmed_num_computed_tokens
         self._pending_save_reqs.append(task)
         with self._save_cond:
             self._save_cond.notify()
@@ -425,7 +453,6 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
 
         if is_segment_finished:
             self.code_prompt_token_ids.pop(external_req_id, None)
-            self.requests_num_chunks_sent.pop(external_req_id, None)
             self.ramp_chunk_count.pop(external_req_id, None)
             cached_ic = getattr(self, "_cached_ic", None)
             if cached_ic is not None:
@@ -501,6 +528,7 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase):
         self.request_payload.pop(external_req_id, None)
         self.code_prompt_token_ids.pop(external_req_id, None)
         self.requests_num_chunks_sent.pop(external_req_id, None)
+        self._segment_generation.pop(external_req_id, None)
         self.ramp_chunk_count.pop(external_req_id, None)
         self._pending_streaming_prefills.pop(external_req_id, None)
 

vllm_omni/engine/orchestrator.py

@@ -199,17 +199,37 @@ class OrchestratorRequestState:
 
 
 @dataclass
-class StreamingInputState:
-    # Flag of streaming input request
-    enabled: bool = False
+class StreamingSegmentState:
+    """Streaming segment-boundary state for one stage of a request.
+
+    Every stage of a multi-stage pipeline reaches its own segment boundaries
+    independently, so this is tracked per stage rather than per request.
+    """
+
     # Flag of segment of streaming input finished
-    segment_finished: bool = False
+    finished: bool = False
     # Tokens from the current raw segment boundary. The vLLM output processor
     # does not guarantee that EngineCoreOutput.new_token_ids survives on the
     # processed RequestOutput used by the routing layer.
-    segment_token_ids: list[int] = field(default_factory=list)
-    segment_output_metadata: dict[str, Any] = field(default_factory=dict)
-    # Streaming update prompt length
+    token_ids: list[int] = field(default_factory=list)
+    output_metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class StreamingInputState:
+    # Flag of streaming input request
+    enabled: bool = False
+    # Keyed by stage. ``_orchestration_loop`` records a boundary per stage poll
+    # into this one ``req_state``, while every consumer reads it for a specific
+    # ``stage_id``, so a flat slot is only correct as long as each read happens in
+    # the same iteration as its own stage's write. It is not for a stage that
+    # never records one (the ``poll_diffusion_output`` branch reaches
+    # ``_handle_processed_outputs`` without the raw-output loop) or for any
+    # consumer reading outside the writing iteration.
+    segments: dict[int, StreamingSegmentState] = field(default_factory=dict)
+    # Streaming update prompt length. NOT stage-keyed: its only consumer reads it
+    # through the streaming context on the stage-input-bridge path, not by stage
+    # id. Do not assume the rest of this struct is stage-safe.
     new_prompt_len_snapshot: int | None = None
     # Model/bridge-specific runtime states (e.g., thinker->talker)
     bridge_states: dict[str, Any] = field(default_factory=dict)
@@ -217,6 +237,14 @@ class StreamingInputState:
     # while the downstream input processor consumes upstream token output.
     source_token_decoder: Callable[..., str] | None = None
 
+    def segment(self, stage_id: int | None) -> StreamingSegmentState:
+        """Return ``stage_id``'s segment state, or a fresh empty one if unreported.
+
+        Fresh rather than shared: ``output_metadata`` is handed out by reference.
+        """
+        segment = self.segments.get(stage_id)
+        return segment if segment is not None else StreamingSegmentState()
+
 
 class _OrchestratorDuplexStagePort:
     """Adapts generic stage pools to the model-neutral duplex control plane."""
@@ -966,17 +994,18 @@ class Orchestrator:
                                 req_state = self.request_states.get(getattr(eco, "request_id", None))
                                 if req_state is None or not req_state.streaming.enabled:
                                     continue
-                                req_state.streaming.segment_finished = bool(getattr(eco, "is_segment_finished", False))
-                                req_state.streaming.segment_token_ids = (
-                                    self._coerce_int_list(getattr(eco, "new_token_ids", None))
-                                    if req_state.streaming.segment_finished
-                                    else []
-                                )
+                                segment_finished = bool(getattr(eco, "is_segment_finished", False))
                                 raw_mm = self._completion_multimodal_output(eco, None)
-                                req_state.streaming.segment_output_metadata = (
-                                    dict(raw_mm)
-                                    if req_state.streaming.segment_finished and isinstance(raw_mm, dict)
-                                    else {}
+                                req_state.streaming.segments[stage_id] = StreamingSegmentState(
+                                    finished=segment_finished,
+                                    token_ids=(
+                                        self._coerce_int_list(getattr(eco, "new_token_ids", None))
+                                        if segment_finished
+                                        else []
+                                    ),
+                                    output_metadata=(
+                                        dict(raw_mm) if segment_finished and isinstance(raw_mm, dict) else {}
+                                    ),
                                 )
                                 req_state.streaming.new_prompt_len_snapshot = getattr(
                                     eco,
@@ -1044,7 +1073,7 @@ class Orchestrator:
                 continue
 
             stage_metrics = None
-            segment_finished = req_state.streaming.enabled and req_state.streaming.segment_finished
+            segment_finished = req_state.streaming.enabled and req_state.streaming.segment(stage_id).finished
             if output.finished or segment_finished:
                 stage_metrics = pool.build_stage_metrics(
                     [output],
@@ -1378,6 +1407,7 @@ class Orchestrator:
         req_id = output.request_id
         finished = output.finished
         submit_ts = req_state.stage_submit_ts.get(stage_id)
+        segment_finished = req_state.streaming.enabled and req_state.streaming.segment(stage_id).finished
         # CFG companion: stash output so parent can bundle [parent, *companions]
         # into source_outputs for the bridge (e.g. thinker2imagegen).
         if finished and self._cfg_tracker.is_companion(req_id):
@@ -1387,11 +1417,7 @@ class Orchestrator:
             return
 
         request_finished = False
-        if (
-            finished
-            and self.stage_pools[stage_id].final_output
-            and not (req_state.streaming.enabled and req_state.streaming.segment_finished)
-        ):
+        if finished and self.stage_pools[stage_id].final_output and not segment_finished:
             req_state.finished_final_output_stage_ids.add(stage_id)
             final_output_stage_ids = req_state.final_output_stage_ids or {req_state.final_stage_id}
             request_finished = final_output_stage_ids.issubset(req_state.finished_final_output_stage_ids)
@@ -1403,7 +1429,7 @@ class Orchestrator:
         # filter out again (the official implementation returns exactly one
         # result per audio chunk).
         is_duplex_stage0_segment = (
-            stage_id == 0 and self._is_duplex_session_request(req_state) and req_state.streaming.segment_finished
+            stage_id == 0 and self._is_duplex_session_request(req_state) and req_state.streaming.segment(0).finished
         )
         if self.stage_pools[stage_id].final_output and not is_duplex_stage0_segment:
             await self.output_async_queue.put(
@@ -1415,7 +1441,10 @@ class Orchestrator:
                     metrics=stage_metrics,
                     finished=(
                         request_finished
-                        or (self._is_duplex_session_request(req_state) and req_state.streaming.segment_finished)
+                        or (
+                            self._is_duplex_session_request(req_state)
+                            and req_state.streaming.segment(stage_id).finished
+                        )
                     ),
                     stage_submit_ts=submit_ts,
                 )
@@ -1450,7 +1479,7 @@ class Orchestrator:
             return
 
         if (
-            (finished or (req_state.streaming.enabled and req_state.streaming.segment_finished))
+            (finished or segment_finished)
             and stage_id < req_state.final_stage_id
             and (not self.async_chunk or not self._stage_receives_async_chunks(stage_id + 1))
             and (not self._next_stage_already_submitted(stage_id, req_state) or req_state.streaming.enabled)
@@ -1589,7 +1618,9 @@ class Orchestrator:
     def _duplex_output_context(
         req_state: OrchestratorRequestState,
         *,
-        stage_id: int | None = None,
+        # Required: segment state is stage-keyed, so a caller that omitted this
+        # would silently read "no boundary" rather than the intended stage's.
+        stage_id: int | None,
     ) -> DuplexOutputContext | None:
         identity = req_state.duplex_identity
         if identity is None:
@@ -1600,15 +1631,16 @@ class Orchestrator:
         )
 
         fence = req_state.duplex_stage_fences.get(stage_id, identity.fence) if stage_id is not None else identity.fence
+        segment = req_state.streaming.segment(stage_id)
         return DuplexOutputContext(
             identity=DuplexRequestIdentity(
                 session_id=identity.session_id,
                 fence=fence,
             ),
             final_stage_id=req_state.final_stage_id,
-            segment_finished=req_state.streaming.enabled and req_state.streaming.segment_finished,
-            segment_token_ids=tuple(req_state.streaming.segment_token_ids),
-            segment_output_metadata=req_state.streaming.segment_output_metadata,
+            segment_finished=req_state.streaming.enabled and segment.finished,
+            segment_token_ids=tuple(segment.token_ids),
+            segment_output_metadata=segment.output_metadata,
         )
 
     @staticmethod
@@ -2144,13 +2176,32 @@ class Orchestrator:
                 req_state.prompt,
                 streaming_context=req_state.streaming,
             )
-        except Exception:
+        except Exception as exc:
             logger.exception(
                 "[Orchestrator] req=%s process_engine_inputs FAILED for stage-%s",
                 req_id,
                 next_logical,
             )
-            raise
+            if not self._is_duplex_session_request(req_state):
+                # Preserve the established failure semantics for ordinary
+                # pipelines.  This request-scoped recovery exists specifically
+                # for resumable full-duplex handoffs, where one malformed unit
+                # must not terminate the shared session orchestrator.
+                raise
+            await self.output_async_queue.put(
+                ErrorMessage(
+                    request_id=req_id,
+                    stage_id=next_logical,
+                    error=(f"Stage-{next_logical} input processor failed: {type(exc).__name__}: {exc}"),
+                    error_type=type(exc).__name__,
+                )
+            )
+            await self._cleanup_request_ids(
+                [req_id, *self._cfg_tracker.cleanup_parent(req_id)],
+                abort=True,
+                close_duplex_sessions=True,
+            )
+            return
         finally:
             req_state.streaming.source_token_decoder = previous_decoder
 

vllm_omni/experimental/fullduplex/DESIGN.md

@@ -58,13 +58,13 @@ The checkpoint keeps these verified contracts:
 - scheduler data-plane append over a resumable request;
 - stale epoch/turn/response fencing;
 - `/v1/duplex` and OpenAI Realtime projection;
+- optional OpenAI `server_vad` turn detection backed by per-session Silero VAD;
 - existing JoyVL behavior.
 
 The checkpoint does not claim:
 
 - scheduler-native KV append;
-- deterministic VAD-triggered interruption (the browser intentionally does not
-  run VAD; MiniCPM owns listen/speak decisions at model-unit boundaries);
+- an acoustic-onset-to-cancel latency target across arbitrary client chunk sizes;
 - production multi-session admission, fairness, capacity, or failure recovery;
 - bounded long-session KV;
 - video input or audio/video synchronization.
@@ -640,9 +640,8 @@ multi-turn runtime evidence.
 
 During auto-response overlap, `preserve_realtime_input` distinguishes "do not
 append this silent chunk to the native buffer" from "clear the open Realtime
-item". Silent overlap no longer discards earlier user PCM. This is an input
-ownership correction. It does not add a VAD policy: overlap is admitted to
-Stage0 and the model decides whether to listen or speak.
+item". Silent overlap no longer discards earlier user PCM. A separate explicit
+`server_vad` policy is documented in the Realtime contract above.
 
 The first chunk of one overlapping input item also reserves its target model
 turn. A later Realtime commit uses that reserved identity even if response EOS
@@ -1024,6 +1023,6 @@ Passing this checkpoint supports the statement:
 > Single-session, model-owned MiniCPM-o 4.5 native duplex is reviewable on the
 > validated H20 configuration.
 
-It does not support claims for deterministic VAD-triggered interruption,
-multi-session production concurrency, bounded long-session KV,
-scheduler-native append, or video input.
+It does not support claims for an acoustic-onset barge-in latency target across
+arbitrary client chunk sizes, multi-session production concurrency, bounded
+long-session KV, scheduler-native append, or video input.

vllm_omni/experimental/fullduplex/client.py

@@ -417,6 +417,7 @@ class RealtimeDuplexClient:
         auto_response: bool = True,
         temperature: float | None = None,
         extra_body: dict[str, object] | None = None,
+        turn_detection: dict[str, object] | None = None,
         session_id: str | None = None,
         timeout_s: float = 20.0,
     ) -> None:
@@ -436,8 +437,12 @@ class RealtimeDuplexClient:
             "modalities": ["audio", "text"],
             "input_audio_format": "pcm16",
             "output_audio_format": output_audio_format,
-            "turn_detection": None,
-            "overlap_policy": "listen_only",
+            "turn_detection": dict(turn_detection) if turn_detection is not None else None,
+            "overlap_policy": (
+                "barge_in_on_speech"
+                if turn_detection is not None and turn_detection.get("interrupt_response", True) is True
+                else "listen_only"
+            ),
             "playback_commit_policy": "ack_only",
             "extra_body": session_extra_body,
         }

vllm_omni/experimental/fullduplex/openai/protocol.py

@@ -9,7 +9,6 @@ from uuid import uuid4
 
 
 class DuplexOverlapPolicy(str, Enum):
-    AUTO = "auto"
     LISTEN_ONLY = "listen_only"
     BARGE_IN_ON_SPEECH = "barge_in_on_speech"
 
@@ -101,7 +100,7 @@ class DuplexCapabilities:
         supports_multi_session = max_sessions > 1
         return cls(
             supports_model_native_turn_policy=True,
-            supports_barge_in=False,
+            supports_barge_in=True,
             supports_input_append=True,
             supports_replace_latest_chunk=False,
             supports_reencode_context=False,
@@ -129,6 +128,9 @@ class DuplexCapabilities:
             signal_sources=["model_native", "client_event", "server_policy"],
             stage_handoff_transport="scheduler_data_plane",
             chunk_period_ms=1000,
+            # Barge-in is now supported, but the wire chunk duration is
+            # client-controlled and no hardware E2E latency has been recorded.
+            # Do not advertise an invented target until that measurement exists.
             target_barge_in_latency_ms=None,
         )
 

vllm_omni/experimental/fullduplex/openai/realtime_input.py

@@ -15,6 +15,13 @@ from vllm_omni.experimental.fullduplex.openai.realtime_state import (
     REALTIME_INPUT_AUDIO_FORMATS,
     REALTIME_OUTPUT_AUDIO_FORMATS,
 )
+from vllm_omni.experimental.fullduplex.openai.vad import (
+    SILERO_VAD_MIN_THRESHOLD,
+    ServerVADUnavailableError,
+    SileroStreamingVAD,
+    SileroVADConfig,
+    StreamingVADResult,
+)
 
 
 class RealtimeInputTranslator:
@@ -89,6 +96,7 @@ class RealtimeInputTranslator:
                     )
                 )
                 return None
+            self._prepare_realtime_turn_detection_update(session_payload)
             self._apply_realtime_session_defaults(session_payload)
             session_payload.update(self._realtime_overlap_fields(session_payload))
             if not self._opened:
@@ -275,16 +283,62 @@ class RealtimeInputTranslator:
                     )
                 )
                 return None
-            looks_like_speech = self._input_looks_like_speech(event, audio=audio, fmt=fmt)
+            vad_result: StreamingVADResult | None = None
+            if isinstance(self._server_vad, SileroStreamingVAD):
+                try:
+                    vad_result = self._server_vad.process_base64(
+                        audio,
+                        fmt=fmt,
+                        sample_rate_hz=sample_rate_hz,
+                    )
+                except ServerVADUnavailableError as exc:
+                    await self._send_realtime_payload(
+                        self._realtime_error_payload(
+                            "server_vad_unavailable",
+                            str(exc),
+                            event_id=event.get("event_id"),
+                            param="turn_detection",
+                        )
+                    )
+                    return None
+                except ValueError as exc:
+                    await self._send_realtime_payload(
+                        self._realtime_error_payload(
+                            "bad_audio",
+                            str(exc),
+                            event_id=event.get("event_id"),
+                            param="audio",
+                        )
+                    )
+                    return None
+            looks_like_speech = (
+                vad_result.is_speech
+                if vad_result is not None
+                else self._input_looks_like_speech(event, audio=audio, fmt=fmt)
+            )
             self._input_audio_buffer_has_audio = self._input_audio_buffer_has_audio or (
                 looks_like_speech and isinstance(audio, str) and bool(audio)
             )
             self._input_audio_buffer_had_non_speech = self._input_audio_buffer_had_non_speech or (
                 not looks_like_speech and isinstance(audio, str) and bool(audio)
             )
-            if looks_like_speech:
+            if vad_result is not None and vad_result.speech_started:
+                started_event = dict(event)
+                if vad_result.speech_start_ms is not None:
+                    started_event["audio_start_ms"] = vad_result.speech_start_ms
+                await self._emit_input_speech_started(started_event)
+            elif vad_result is None and looks_like_speech:
                 await self._emit_input_speech_started(event)
+            if looks_like_speech:
                 self._remember_input_transcript_hint(event)
+            if vad_result is not None and vad_result.speech_stopped:
+                stopped_event = dict(event)
+                if vad_result.speech_end_ms is not None:
+                    stopped_event["audio_end_ms"] = vad_result.speech_end_ms
+                await self._emit_input_speech_stopped(
+                    stopped_event,
+                    item_id=self._active_input_item_id or f"item_{uuid4().hex}",
+                )
             payload = {
                 "type": "input_audio_buffer.append",
                 "audio": audio,
@@ -306,14 +360,20 @@ class RealtimeInputTranslator:
                     return None
                 payload["video_frames"] = [frame for frame in video_frames if isinstance(frame, str) and frame]
             self._copy_realtime_input_hints(event, payload)
-            if not looks_like_speech:
-                payload["is_speech"] = False
+            payload["is_speech"] = looks_like_speech
+            if vad_result is not None:
+                payload["vad"] = vad_result.as_hint()
+                if self._server_vad_interrupts_response() and (vad_result.speech_active or vad_result.speech_started):
+                    # Hold the MiniCPM stream in listen mode for the remainder
+                    # of this utterance after the one-shot onset cancellation.
+                    payload["force_listen"] = True
             return payload
         if event_type == "input_audio_buffer.commit":
             if not self._input_audio_buffer_has_audio:
                 if self._input_audio_buffer_had_non_speech:
                     self._input_audio_buffer_had_non_speech = False
                     self._active_input_item_id = None
+                    self._reset_server_vad()
                     return {
                         "type": "input_audio_buffer.commit",
                         "final": event.get("final", True),
@@ -335,6 +395,7 @@ class RealtimeInputTranslator:
             self._active_input_item_id = None
             self._input_audio_buffer_has_audio = False
             self._input_audio_buffer_had_non_speech = False
+            self._reset_server_vad()
             payload = {
                 "type": "input_audio_buffer.commit",
                 "final": event.get("final", True),
@@ -350,6 +411,7 @@ class RealtimeInputTranslator:
             self._input_audio_buffer_has_audio = False
             self._input_audio_buffer_had_non_speech = False
             self._input_audio_buffer_transcript_parts.clear()
+            self._reset_server_vad()
             return {"type": "input_audio_buffer.clear", "reason": event_type}
         if event_type == "output_audio_buffer.clear":
             payload = {"type": "output_audio_buffer.clear", "reason": event_type}
@@ -991,8 +1053,8 @@ class RealtimeInputTranslator:
                 return "video_frames entries must be JPEG or PNG images"
         return None
 
-    @staticmethod
-    def _validate_realtime_turn_detection(session_payload: dict[str, object]) -> str | None:
+    @classmethod
+    def _validate_realtime_turn_detection(cls, session_payload: dict[str, object]) -> str | None:
         configured_values: list[tuple[str, object]] = []
         if "turn_detection" in session_payload:
             configured_values.append(("turn_detection", session_payload["turn_detection"]))
@@ -1004,13 +1066,152 @@ class RealtimeInputTranslator:
         for field_path, turn_detection in configured_values:
             if turn_detection is None:
                 continue
-            turn_detection_type = turn_detection.get("type") if isinstance(turn_detection, dict) else turn_detection
-            return (
-                f"{field_path}={turn_detection_type!r} is not implemented by the duplex Realtime adapter; "
-                "set turn_detection to null and commit input explicitly, or use the model-owned duplex policy"
-            )
+            if not isinstance(turn_detection, dict):
+                return f"{field_path} must be null or an object with type='server_vad'"
+            turn_detection_type = turn_detection.get("type")
+            if turn_detection_type != "server_vad":
+                return (
+                    f"{field_path}={turn_detection_type!r} is unsupported; "
+                    "use null for model-owned listen/speak or type='server_vad' for hard barge-in"
+                )
+            threshold = turn_detection.get("threshold", 0.5)
+            if (
+                not isinstance(threshold, int | float)
+                or isinstance(threshold, bool)
+                or not np.isfinite(float(threshold))
+                or not SILERO_VAD_MIN_THRESHOLD < float(threshold) <= 1.0
+            ):
+                return f"{field_path}.threshold must be greater than {SILERO_VAD_MIN_THRESHOLD} and at most 1"
+            for name in (
+                "prefix_padding_ms",
+                "silence_duration_ms",
+                "min_speech_duration_ms",
+            ):
+                value = turn_detection.get(name)
+                if value is not None and (
+                    not isinstance(value, int | float)
+                    or isinstance(value, bool)
+                    or not np.isfinite(float(value))
+                    or float(value) < 0
+                ):
+                    return f"{field_path}.{name} must be a non-negative number"
+            interrupt_response = turn_detection.get("interrupt_response", True)
+            if not isinstance(interrupt_response, bool):
+                return f"{field_path}.interrupt_response must be a boolean"
+            if interrupt_response is not True:
+                return (
+                    f"{field_path}.interrupt_response=false is unsupported; "
+                    "use turn_detection=null for model-owned listen/speak"
+                )
+
+        present, turn_detection = cls._configured_realtime_turn_detection(session_payload)
+        if not present:
+            if session_payload.get("overlap_policy") == "barge_in_on_speech":
+                return (
+                    "overlap_policy='barge_in_on_speech' requires "
+                    "turn_detection.type='server_vad' on the Realtime endpoint"
+                )
+            return None
+        desired_policy = (
+            "barge_in_on_speech"
+            if isinstance(turn_detection, dict) and turn_detection.get("interrupt_response", True) is True
+            else "listen_only"
+        )
+        overlap_policy = session_payload.get("overlap_policy")
+        if isinstance(overlap_policy, str) and overlap_policy != desired_policy:
+            return f"overlap_policy={overlap_policy!r} conflicts with turn_detection; expected {desired_policy!r}"
         return None
 
+    @staticmethod
+    def _configured_realtime_turn_detection(
+        session_payload: dict[str, object],
+    ) -> tuple[bool, dict[str, object] | None]:
+        present = "turn_detection" in session_payload
+        selected = session_payload.get("turn_detection")
+        audio_config = session_payload.get("audio")
+        if isinstance(audio_config, dict):
+            audio_input = audio_config.get("input")
+            if isinstance(audio_input, dict) and "turn_detection" in audio_input:
+                present = True
+                selected = audio_input.get("turn_detection")
+        return present, dict(selected) if isinstance(selected, dict) else None
+
+    def _prepare_realtime_turn_detection_update(self, session_payload: dict[str, object]) -> None:
+        """Stage a VAD replacement until the Session/runtime update commits."""
+        present, turn_detection = self._configured_realtime_turn_detection(session_payload)
+        if not present:
+            return
+        if self._turn_detection_update_pending:
+            raise RuntimeError("A Realtime turn-detection update is already pending")
+
+        pending_turn_detection: dict[str, object] | None = None
+        pending_server_vad: SileroStreamingVAD | None = None
+        if turn_detection is None:
+            session_payload["overlap_policy"] = "listen_only"
+        else:
+            normalized = dict(turn_detection)
+            normalized.setdefault("type", "server_vad")
+            normalized.setdefault("interrupt_response", True)
+            normalized.setdefault("threshold", 0.5)
+            normalized.setdefault("prefix_padding_ms", 300)
+            normalized.setdefault("silence_duration_ms", 500)
+            normalized.setdefault("min_speech_duration_ms", 96)
+            pending_turn_detection = normalized
+            pending_server_vad = SileroStreamingVAD(SileroVADConfig.from_turn_detection(normalized))
+            session_payload["turn_detection"] = dict(normalized)
+            session_payload["overlap_policy"] = "barge_in_on_speech"
+
+        self._pending_turn_detection = pending_turn_detection
+        self._pending_server_vad = pending_server_vad
+        self._turn_detection_update_pending = True
+        self._turn_detection_update_resolved = asyncio.Event()
+
+    async def wait_for_realtime_turn_detection_update(self) -> None:
+        """Preserve wire order while the runner accepts or rejects a candidate."""
+        resolved = self._turn_detection_update_resolved
+        if self._turn_detection_update_pending and resolved is not None:
+            await resolved.wait()
+
+    def commit_realtime_turn_detection_update(self) -> None:
+        """Publish the staged VAD only after the Session/runtime transaction ACKs."""
+        if not self._turn_detection_update_pending:
+            return
+        previous_server_vad = self._server_vad
+        self._turn_detection = (
+            dict(self._pending_turn_detection) if isinstance(self._pending_turn_detection, dict) else None
+        )
+        self._server_vad = self._pending_server_vad
+        self._clear_pending_realtime_turn_detection_update()
+        reset = getattr(previous_server_vad, "reset", None)
+        if callable(reset):
+            reset()
+
+    def reject_realtime_turn_detection_update(self) -> None:
+        """Discard a staged VAD after any rejected Session/runtime update."""
+        if not self._turn_detection_update_pending:
+            return
+        reset = getattr(self._pending_server_vad, "reset", None)
+        if callable(reset):
+            reset()
+        self._clear_pending_realtime_turn_detection_update()
+
+    def _clear_pending_realtime_turn_detection_update(self) -> None:
+        resolved = self._turn_detection_update_resolved
+        self._turn_detection_update_pending = False
+        self._pending_turn_detection = None
+        self._pending_server_vad = None
+        self._turn_detection_update_resolved = None
+        if resolved is not None:
+            resolved.set()
+
+    def _server_vad_interrupts_response(self) -> bool:
+        return isinstance(self._turn_detection, dict) and self._turn_detection.get("interrupt_response", True) is True
+
+    def _reset_server_vad(self) -> None:
+        reset = getattr(self._server_vad, "reset", None)
+        if callable(reset):
+            reset()
+
     @staticmethod
     def _input_audio_transcription_config(session_payload: dict[str, object]) -> dict[str, object] | None:
         transcription = session_payload.get("input_audio_transcription")

vllm_omni/experimental/fullduplex/openai/realtime_output.py

@@ -262,15 +262,10 @@ class RealtimeOutputProjector:
                         content_index=0,
                         audio_end_ms=committed_audio_ms,
                     )
-                payloads.append(
-                    {
-                        "type": "conversation.item.truncated",
-                        "item_id": item_id,
-                        "content_index": 0,
-                        "audio_end_ms": committed_audio_ms,
-                        "event": event,
-                    }
-                )
+                # Keep server-side conversation history aligned with the
+                # acknowledged playback cursor. ``conversation.item.truncated``
+                # is reserved for an explicit client item-truncate request;
+                # cancellation itself terminates through ``response.done``.
             payloads.extend(self._realtime_audio_done_events(event, response_id))
             payloads.extend(
                 self._realtime_response_terminal_events(

vllm_omni/experimental/fullduplex/openai/realtime_session.py

@@ -58,6 +58,7 @@ class NativeRealtimeSessionProtocol(
         while True:
             if not self._pending_outbound.empty():
                 return json.dumps(await self._pending_outbound.get())
+            await self.wait_for_realtime_turn_detection_update()
             raw = await websocket.receive_text()
             try:
                 event = json.loads(raw)
@@ -97,11 +98,22 @@ class NativeRealtimeSessionProtocol(
             return json.dumps(translated)
 
     def encode_outbound_event(self, data: dict[str, Any]) -> list[dict[str, object]]:
+        self._resolve_realtime_turn_detection_update(data)
         payloads = self._from_duplex_event(data)
         for payload in payloads:
             self._attach_event_id(payload)
         return payloads
 
+    def _resolve_realtime_turn_detection_update(self, data: dict[str, Any]) -> None:
+        """Publish or discard a staged VAD with the Realtime session ACK."""
+        if not self._turn_detection_update_pending:
+            return
+        event_type = data.get("type")
+        if event_type in {"session.created", "session.updated"}:
+            self.commit_realtime_turn_detection_update()
+        elif event_type == "error":
+            self.reject_realtime_turn_detection_update()
+
     @staticmethod
     def _attach_event_id(payload: dict[str, object]) -> None:
         payload.setdefault("event_id", f"event_{uuid4().hex}")

vllm_omni/experimental/fullduplex/openai/realtime_state.py

@@ -57,6 +57,7 @@ REALTIME_ERROR_TYPES_BY_CODE = {
     "native_text_append_unsupported": "invalid_request_error",
     "runtime_native_stage_role_required": "server_error",
     "runtime_native_runner_kv_required": "server_error",
+    "server_vad_unavailable": "server_error",
 }
 
 
@@ -105,6 +106,12 @@ class RealtimeSessionState:
     _input_sample_rate_hz: int = 16000
     _output_audio_format: str = "pcm16"
     _overlap_silence_rms: float = 0.003
+    _turn_detection: dict[str, object] | None = None
+    _server_vad: object | None = None
+    _turn_detection_update_pending: bool = False
+    _pending_turn_detection: dict[str, object] | None = None
+    _pending_server_vad: object | None = None
+    _turn_detection_update_resolved: asyncio.Event | None = None
     _send_realtime_json: Any = None
     _initial_session_update: bool = False
     _input_speech_started: bool = False
@@ -180,6 +187,12 @@ class RealtimeStateOwner:
     _input_sample_rate_hz: int = _RealtimeStateField()
     _output_audio_format: str = _RealtimeStateField()
     _overlap_silence_rms: float = _RealtimeStateField()
+    _turn_detection: dict[str, object] | None = _RealtimeStateField()
+    _server_vad: object | None = _RealtimeStateField()
+    _turn_detection_update_pending: bool = _RealtimeStateField()
+    _pending_turn_detection: dict[str, object] | None = _RealtimeStateField()
+    _pending_server_vad: object | None = _RealtimeStateField()
+    _turn_detection_update_resolved: asyncio.Event | None = _RealtimeStateField()
     _send_realtime_json: Any = _RealtimeStateField()
     _initial_session_update: bool = _RealtimeStateField()
     _input_speech_started: bool = _RealtimeStateField()

vllm_omni/experimental/fullduplex/openai/serving.py

@@ -494,6 +494,26 @@ class OmniDuplexSessionHandler(
             # playback only controls history ACKs, not model admission.
             if is_speech:
                 session.accumulate_overlap_speech(duration_ms)
+            vad_speech_started = self._vad_speech_started(event, payload)
+            if (
+                self._uses_native_input_append(session)
+                and session.capabilities.supports_barge_in
+                and is_speech
+                and session.config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value
+                and vad_speech_started is not False
+            ):
+                # The configured server overlap detector classified this chunk
+                # as speech while the assistant is still generating/playing.
+                # Fence the active response so the same interrupting chunk is
+                # retained as the start of the next turn.
+                return {
+                    "action": "barge_in",
+                    "reason": ("server_vad_speech_started" if vad_speech_started is True else "barge_in_on_speech"),
+                    "cancel_reason": "turn_detected" if vad_speech_started is True else "barge_in",
+                    "duration_ms": duration_ms,
+                    "overlap_speech_ms": session.overlap_speech_ms,
+                    "buffer_audio": True,
+                }
             return {
                 "action": "listen",
                 "reason": "auto_response_continuous",
@@ -501,7 +521,7 @@ class OmniDuplexSessionHandler(
                 "overlap_speech_ms": session.overlap_speech_ms,
                 "buffer_audio": True,
                 "defer_runtime_append": False,
-                "force_listen": False,
+                "force_listen": event.get("force_listen") is True or payload.get("force_listen") is True,
                 "preserve_realtime_input": True,
             }
         if bool(event.get("force_listen", False)):
@@ -553,9 +573,22 @@ class OmniDuplexSessionHandler(
 
         session.accumulate_overlap_speech(duration_ms)
         if policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value:
+            vad_speech_started = self._vad_speech_started(event, payload)
+            if vad_speech_started is False:
+                return {
+                    "action": "listen",
+                    "reason": "server_vad_utterance_active",
+                    "duration_ms": duration_ms,
+                    "overlap_speech_ms": session.overlap_speech_ms,
+                    "buffer_audio": True,
+                    "defer_runtime_append": False,
+                    "force_listen": True,
+                    "preserve_realtime_input": True,
+                }
             return {
                 "action": "barge_in",
-                "reason": "policy_barge_in_on_speech",
+                "reason": ("server_vad_speech_started" if vad_speech_started is True else "policy_barge_in_on_speech"),
+                "cancel_reason": "turn_detected" if vad_speech_started is True else "barge_in",
                 "duration_ms": duration_ms,
                 "overlap_speech_ms": session.overlap_speech_ms,
                 "buffer_audio": True,
@@ -599,6 +632,18 @@ class OmniDuplexSessionHandler(
             "defer_runtime_append": True,
         }
 
+    @staticmethod
+    def _vad_speech_started(
+        event: Mapping[str, object],
+        payload: Mapping[str, object],
+    ) -> bool | None:
+        """Return an onset edge, or None for legacy non-server-VAD input."""
+        for source in (event, payload):
+            vad = source.get("vad")
+            if isinstance(vad, Mapping) and isinstance(vad.get("speech_started"), bool):
+                return bool(vad["speech_started"])
+        return None
+
     @staticmethod
     def _event_requests_barge_in(event: Mapping[str, object]) -> bool:
         if event.get("force_barge_in") is True:

vllm_omni/experimental/fullduplex/openai/session_runner.py

@@ -18,6 +18,7 @@ from vllm_omni.experimental.fullduplex.openai.commit_policy import (
     decide_commit_action,
 )
 from vllm_omni.experimental.fullduplex.openai.protocol import (
+    DuplexOverlapPolicy,
     DuplexPlaybackCommitPolicy,
     DuplexSession,
     DuplexSessionState,
@@ -875,7 +876,11 @@ class DuplexSessionRunnerMixin:
 
                 if event_type in {"input.cancel", "response.cancel", "barge_in", "output_audio_buffer.clear"}:
                     cancel_reason = (
-                        "output_audio_buffer_clear" if event_type == "output_audio_buffer.clear" else "barge_in"
+                        "output_audio_buffer_clear"
+                        if event_type == "output_audio_buffer.clear"
+                        else "client_cancelled"
+                        if event_type == "response.cancel"
+                        else "barge_in"
                     )
                     cancelled_fence = DuplexFence(
                         session.session_id,
@@ -1249,7 +1254,16 @@ class DuplexSessionRunnerMixin:
                     buffer_overlap_audio = True
                     if self._uses_native_input_append(session):
                         mark_pending_silence_superseded()
-                        overlap_active = not self._session_auto_responds(session) and native_response_in_progress()
+                        overlap_active = native_response_in_progress() and (
+                            not self._session_auto_responds(session)
+                            or (
+                                session.capabilities.supports_barge_in
+                                and (
+                                    session.config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value
+                                    or self._event_requests_barge_in(event)
+                                )
+                            )
+                        )
                         if overlap_active:
                             decision = self._overlap_decision(session, event, payload)
                             await self._emit_overlap_decision(emit_event, session, decision)
@@ -1290,11 +1304,12 @@ class DuplexSessionRunnerMixin:
                                 native.speech_since_commit = False
                                 await actor.cancel_append_tasks()
                                 had_native_stream = native.data_plane_task is not None
+                                cancel_reason = str(decision.get("cancel_reason") or "barge_in")
                                 cancelled = await self._cancel_active_response(
                                     session,
                                     actor.active_response_task,
                                     emit_event,
-                                    reason="barge_in",
+                                    reason=cancel_reason,
                                 )
                                 had_native_stream = (
                                     await self._cancel_native_data_plane_stream(session) or had_native_stream
@@ -1310,7 +1325,7 @@ class DuplexSessionRunnerMixin:
                                             "type": "audio.cancelled",
                                             "session_id": session.session_id,
                                             "response_id": old_response_id,
-                                            "reason": "barge_in",
+                                            "reason": cancel_reason,
                                             "cancelled_epoch": old_epoch,
                                             "epoch": new_epoch,
                                             "committed_ms": committed_ms,
@@ -1330,7 +1345,7 @@ class DuplexSessionRunnerMixin:
                                             "type": "audio.cancelled",
                                             "session_id": session.session_id,
                                             "response_id": session.last_response_id,
-                                            "reason": "barge_in",
+                                            "reason": cancel_reason,
                                             "cancelled_epoch": old_epoch,
                                             "epoch": new_epoch,
                                             "committed_ms": committed_ms,

vllm_omni/experimental/fullduplex/openai/vad.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import binascii
+from collections.abc import Callable
+from dataclasses import dataclass
+
+import numpy as np
+import pybase64 as base64
+
+SILERO_VAD_MIN_THRESHOLD = 0.15
+
+
+class ServerVADUnavailableError(RuntimeError):
+    """Raised when the optional server VAD runtime is not installed."""
+
+
+@dataclass(frozen=True, slots=True)
+class SileroVADConfig:
+    threshold: float = 0.5
+    prefix_padding_ms: int = 300
+    silence_duration_ms: int = 500
+    min_speech_duration_ms: int = 96
+
+    @classmethod
+    def from_turn_detection(cls, turn_detection: dict[str, object]) -> SileroVADConfig:
+        defaults = cls()
+
+        def number(name: str, default: int | float) -> int | float:
+            value = turn_detection.get(name, default)
+            return value if isinstance(value, int | float) and not isinstance(value, bool) else default
+
+        return cls(
+            threshold=float(number("threshold", defaults.threshold)),
+            prefix_padding_ms=max(0, int(number("prefix_padding_ms", defaults.prefix_padding_ms))),
+            silence_duration_ms=max(0, int(number("silence_duration_ms", defaults.silence_duration_ms))),
+            min_speech_duration_ms=max(
+                32,
+                int(number("min_speech_duration_ms", defaults.min_speech_duration_ms)),
+            ),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class StreamingVADResult:
+    is_speech: bool
+    speech_active: bool
+    speech_started: bool = False
+    speech_stopped: bool = False
+    speech_probability: float = 0.0
+    speech_start_ms: int | None = None
+    speech_end_ms: int | None = None
+
+    def as_hint(self) -> dict[str, object]:
+        hint: dict[str, object] = {
+            "backend": "silero",
+            "is_speech": self.is_speech,
+            "speech_active": self.speech_active,
+            "speech_started": self.speech_started,
+            "speech_stopped": self.speech_stopped,
+            "speech_probability": self.speech_probability,
+        }
+        if self.speech_start_ms is not None:
+            hint["speech_start_ms"] = self.speech_start_ms
+        if self.speech_end_ms is not None:
+            hint["speech_end_ms"] = self.speech_end_ms
+        return hint
+
+
+class SileroStreamingVAD:
+    """Per-session streaming Silero VAD with onset debounce and utterance latch.
+
+    Silero's model is stateful, so an instance must never be shared across
+    Realtime sessions. Audio is normalized to 16 kHz and scored in the model's
+    512-sample (32 ms) windows. Loading is lazy so deployments which use the
+    default model-owned mode do not need the optional ``silero-vad`` package.
+    """
+
+    _SAMPLE_RATE_HZ = 16_000
+    _WINDOW_SAMPLES = 512
+    _NEGATIVE_THRESHOLD_GAP = SILERO_VAD_MIN_THRESHOLD
+
+    def __init__(
+        self,
+        config: SileroVADConfig,
+        *,
+        frame_scorer: Callable[[np.ndarray], float] | None = None,
+    ) -> None:
+        self.config = config
+        self._frame_scorer = frame_scorer
+        self._model: object | None = None
+        self._pending = np.empty(0, dtype=np.float32)
+        self._speech_active = False
+        self._candidate_samples = 0
+        self._candidate_start_sample: int | None = None
+        self._silence_samples = 0
+        self._processed_samples = 0
+
+    def reset(self) -> None:
+        self._pending = np.empty(0, dtype=np.float32)
+        self._speech_active = False
+        self._candidate_samples = 0
+        self._candidate_start_sample = None
+        self._silence_samples = 0
+        self._processed_samples = 0
+        reset_states = getattr(self._model, "reset_states", None)
+        if callable(reset_states):
+            reset_states()
+
+    def process_base64(
+        self,
+        audio: object,
+        *,
+        fmt: object,
+        sample_rate_hz: object,
+    ) -> StreamingVADResult:
+        if fmt != "pcm_f32le" or not isinstance(audio, str):
+            raise ValueError("Silero server VAD requires decoded pcm_f32le audio")
+        try:
+            raw = base64.b64decode(audio, validate=True)
+        except (binascii.Error, ValueError) as exc:
+            raise ValueError("Silero server VAD received invalid base64 audio") from exc
+        if len(raw) % 4:
+            raise ValueError("Silero server VAD received an incomplete pcm_f32le frame")
+        samples = np.frombuffer(raw, dtype="<f4").astype(np.float32, copy=False)
+        rate = int(sample_rate_hz) if isinstance(sample_rate_hz, int | float) else self._SAMPLE_RATE_HZ
+        if rate <= 0:
+            raise ValueError("Silero server VAD requires a positive sample rate")
+        if rate != self._SAMPLE_RATE_HZ and samples.size > 1:
+            target_size = max(1, int(round(samples.size * self._SAMPLE_RATE_HZ / rate)))
+            source_x = np.linspace(0.0, 1.0, num=samples.size, endpoint=True)
+            target_x = np.linspace(0.0, 1.0, num=target_size, endpoint=True)
+            samples = np.interp(target_x, source_x, samples).astype(np.float32)
+        return self.process(samples)
+
+    def process(self, samples: np.ndarray) -> StreamingVADResult:
+        samples = np.asarray(samples, dtype=np.float32).reshape(-1)
+        if samples.size:
+            self._pending = np.concatenate((self._pending, samples))
+
+        started = False
+        stopped = False
+        start_ms: int | None = None
+        end_ms: int | None = None
+        max_probability = 0.0
+        contained_speech = self._speech_active
+        min_speech_samples = int(self.config.min_speech_duration_ms * self._SAMPLE_RATE_HZ / 1000)
+        min_silence_samples = int(self.config.silence_duration_ms * self._SAMPLE_RATE_HZ / 1000)
+        negative_threshold = max(0.0, self.config.threshold - self._NEGATIVE_THRESHOLD_GAP)
+
+        while self._pending.size >= self._WINDOW_SAMPLES:
+            frame = self._pending[: self._WINDOW_SAMPLES]
+            self._pending = self._pending[self._WINDOW_SAMPLES :]
+            frame_start = self._processed_samples
+            self._processed_samples += self._WINDOW_SAMPLES
+            probability = min(1.0, max(0.0, float(self._score_frame(frame))))
+            max_probability = max(max_probability, probability)
+
+            if self._speech_active:
+                contained_speech = True
+                if probability < negative_threshold:
+                    self._silence_samples += self._WINDOW_SAMPLES
+                    if self._silence_samples >= min_silence_samples:
+                        speech_end_sample = self._processed_samples - self._silence_samples
+                        self._speech_active = False
+                        self._silence_samples = 0
+                        stopped = True
+                        end_ms = max(0, int(round(speech_end_sample * 1000 / self._SAMPLE_RATE_HZ)))
+                else:
+                    self._silence_samples = 0
+                continue
+
+            if probability >= self.config.threshold:
+                if self._candidate_samples == 0:
+                    self._candidate_start_sample = frame_start
+                self._candidate_samples += self._WINDOW_SAMPLES
+                if self._candidate_samples >= min_speech_samples:
+                    candidate_start = self._candidate_start_sample or 0
+                    prefix_samples = int(self.config.prefix_padding_ms * self._SAMPLE_RATE_HZ / 1000)
+                    self._speech_active = True
+                    self._candidate_samples = 0
+                    self._candidate_start_sample = None
+                    self._silence_samples = 0
+                    contained_speech = True
+                    started = True
+                    start_ms = max(
+                        0,
+                        int(round((candidate_start - prefix_samples) * 1000 / self._SAMPLE_RATE_HZ)),
+                    )
+            else:
+                self._candidate_samples = 0
+                self._candidate_start_sample = None
+
+        return StreamingVADResult(
+            is_speech=contained_speech,
+            speech_active=self._speech_active,
+            speech_started=started,
+            speech_stopped=stopped,
+            speech_probability=max_probability,
+            speech_start_ms=start_ms,
+            speech_end_ms=end_ms,
+        )
+
+    def _score_frame(self, frame: np.ndarray) -> float:
+        if self._frame_scorer is not None:
+            return float(self._frame_scorer(frame))
+        if self._model is None:
+            try:
+                import torch
+                from silero_vad import load_silero_vad
+            except ImportError as exc:
+                raise ServerVADUnavailableError(
+                    "server_vad requires the optional 'silero-vad' package; install vllm-omni[server-vad]"
+                ) from exc
+            self._model = load_silero_vad(onnx=False)
+            self._frame_scorer = lambda value: float(
+                self._model(torch.from_numpy(np.ascontiguousarray(value)), self._SAMPLE_RATE_HZ).item()
+            )
+        return float(self._frame_scorer(frame))

vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py

@@ -550,6 +550,35 @@ def _native_tts_boundary_token_ids(special_token_ids):
     }
 
 
+def _native_duplex_output_is_control_only(
+    output_ids: Sequence[int],
+    special_token_ids: Mapping[str, int],
+) -> bool:
+    """Whether a native output is a confirmed no-speech handoff.
+
+    Empty and control/terminal-only deltas intentionally have no Talker
+    conditioning. Any ordinary text or explicit speech boundary is not a
+    confirmed no-op and must retain the missing-hidden-state error path.
+    """
+    if not output_ids:
+        return True
+    control_ids = {
+        token_id
+        for token_id in (
+            special_token_ids.get("tts_eos_token_id"),
+            special_token_ids.get("tts_pad_token_id"),
+            special_token_ids.get("listen_token_id"),
+            special_token_ids.get("chunk_eos_token_id"),
+            special_token_ids.get("chunk_tts_eos_token_id"),
+            special_token_ids.get("turn_eos_token_id"),
+            special_token_ids.get("unit_token_id"),
+            special_token_ids.get("unit_end_token_id"),
+        )
+        if token_id is not None
+    }
+    return bool(control_ids) and all(token_id in control_ids for token_id in output_ids)
+
+
 def _decode_native_duplex_token_ids(
     token_ids: list[int],
     streaming_context,
@@ -757,10 +786,23 @@ def llm2tts(
             )
         prompt_token_ids_len = len(prompt_token_ids)
 
+        is_native_duplex_handoff = _has_native_duplex_prompt_metadata(mm_output)
+
         latent = mm_output.get("latent", None)
         if latent is None:
             latent = output.hidden_states if hasattr(output, "hidden_states") else None
             if latent is None:
+                if is_native_duplex_handoff and _native_duplex_output_is_control_only(
+                    llm_output_ids, special_token_ids
+                ):
+                    # Listen/control/terminal outputs have no Talker payload by
+                    # design. They are a successful no-op handoff, not a stage
+                    # failure, so do not let one such unit stop the orchestrator.
+                    logger.debug(
+                        "Skipping no-speech MiniCPM-o native duplex handoff for request_id=%s",
+                        llm_output.request_id,
+                    )
+                    continue
                 raise ValueError("No latent or hidden_states found in thinker output")
 
         thinker_hidden_states = latent.detach()
@@ -771,7 +813,6 @@ def llm2tts(
         full_token_ids = prompt_token_ids + llm_output_ids
 
         tts_bos_id = special_token_ids.get("tts_bos_token_id")
-        is_native_duplex_handoff = _has_native_duplex_prompt_metadata(mm_output)
         if is_native_duplex_handoff:
             _require_native_tts_boundary_metadata(special_token_ids)
         tts_end_ids = _native_tts_boundary_token_ids(special_token_ids)