From 5edee1393ad61fc0e04911c3d3fae64a1973e963 Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Mon, 24 Aug 2026 05:13:31 +0800 Subject: [PATCH] [Bugfix][MiniCPM-o] Prevent native duplex Stage-1 prompt overflow Signed-off-by: Sy03 <1370724210@qq.com> --- .buildkite/cuda/test-merge.yml | 1 + .buildkite/cuda/test-ready.yml | 1 + .../minicpmo/realtime_duplex_demo.py | 2 +- pyproject.toml | 5 + ...t_omni_ar_scheduler_aborted_queue_sweep.py | 115 ++++++++ .../sched/test_omni_ar_scheduler_streaming.py | 110 +++++++- ...ni_ar_scheduler_unified_decode_deferral.py | 3 + ..._omni_scheduler_streaming_input_counter.py | 254 ++++++++++++++++++ .../test_chunk_transfer_adapter.py | 28 +- ...run_minicpmo_realtime_duplex_server_vad.py | 107 ++++++++ .../test_minicpmo_4_5_duplex_expansion.py | 20 ++ tests/engine/test_orchestrator.py | 3 +- .../test_orchestrator_error_handling.py | 33 +++ .../test_orchestrator_segment_stage_keying.py | 130 +++++++++ .../test_orchestrator_stage_input_bridge.py | 3 +- .../openai/test_duplex_protocol.py | 2 +- .../openai_api/test_duplex_handler.py | 38 ++- vllm_omni/core/sched/omni_ar_scheduler.py | 43 ++- vllm_omni/core/sched/omni_scheduler_mixin.py | 17 ++ .../chunk_transfer_adapter.py | 34 ++- vllm_omni/engine/orchestrator.py | 115 +++++--- vllm_omni/experimental/fullduplex/DESIGN.md | 15 +- vllm_omni/experimental/fullduplex/client.py | 9 +- .../fullduplex/openai/protocol.py | 6 +- .../fullduplex/openai/realtime_input.py | 223 ++++++++++++++- .../fullduplex/openai/realtime_output.py | 13 +- .../fullduplex/openai/realtime_session.py | 12 + .../fullduplex/openai/realtime_state.py | 13 + .../experimental/fullduplex/openai/serving.py | 49 +++- .../fullduplex/openai/session_runner.py | 25 +- .../experimental/fullduplex/openai/vad.py | 218 +++++++++++++++ .../minicpmo_4_5_omni.py | 43 ++- 32 files changed, 1591 insertions(+), 99 deletions(-) create mode 100644 tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py create mode 100644 tests/core/sched/test_omni_scheduler_streaming_input_counter.py create mode 100644 tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py create mode 100644 tests/engine/test_orchestrator_segment_stage_keying.py create mode 100644 vllm_omni/experimental/fullduplex/openai/vad.py diff --git a/.buildkite/cuda/test-merge.yml b/.buildkite/cuda/test-merge.yml index 2d3be8bc4..51d752534 100644 --- a/.buildkite/cuda/test-merge.yml +++ b/.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 diff --git a/.buildkite/cuda/test-ready.yml b/.buildkite/cuda/test-ready.yml index ff55409b3..830e6aa63 100644 --- a/.buildkite/cuda/test-ready.yml +++ b/.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 diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py index b6c861747..bca251f4e 100644 --- a/examples/online_serving/minicpmo/realtime_duplex_demo.py +++ b/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") diff --git a/pyproject.toml b/pyproject.toml index 84a950719..a02825dce 100644 --- a/pyproject.toml +++ b/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 ] diff --git a/tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py b/tests/core/sched/test_omni_ar_scheduler_aborted_queue_sweep.py new file mode 100644 index 000000000..ee622a2f2 --- /dev/null +++ b/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] diff --git a/tests/core/sched/test_omni_ar_scheduler_streaming.py b/tests/core/sched/test_omni_ar_scheduler_streaming.py index 50fcbcf4e..d1930caf4 100644 --- a/tests/core/sched/test_omni_ar_scheduler_streaming.py +++ b/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 diff --git a/tests/core/sched/test_omni_ar_scheduler_unified_decode_deferral.py b/tests/core/sched/test_omni_ar_scheduler_unified_decode_deferral.py index 27087d253..0e0ed04ad 100644 --- a/tests/core/sched/test_omni_ar_scheduler_unified_decode_deferral.py +++ b/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 diff --git a/tests/core/sched/test_omni_scheduler_streaming_input_counter.py b/tests/core/sched/test_omni_scheduler_streaming_input_counter.py new file mode 100644 index 000000000..3a7a29139 --- /dev/null +++ b/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 diff --git a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py b/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py index d0fd00376..c72cc91d0 100644 --- a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py +++ b/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") diff --git a/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py b/tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py new file mode 100644 index 000000000..1bd3db703 --- /dev/null +++ b/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} diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py index 68204fe28..aa5320b8c 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py +++ b/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) diff --git a/tests/engine/test_orchestrator.py b/tests/engine/test_orchestrator.py index 3d63943a7..a1eab89bb 100644 --- a/tests/engine/test_orchestrator.py +++ b/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] diff --git a/tests/engine/test_orchestrator_error_handling.py b/tests/engine/test_orchestrator_error_handling.py index 4176764bc..0ec131e1a 100644 --- a/tests/engine/test_orchestrator_error_handling.py +++ b/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"), diff --git a/tests/engine/test_orchestrator_segment_stage_keying.py b/tests/engine/test_orchestrator_segment_stage_keying.py new file mode 100644 index 000000000..02bb9a338 --- /dev/null +++ b/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] diff --git a/tests/engine/test_orchestrator_stage_input_bridge.py b/tests/engine/test_orchestrator_stage_input_bridge.py index 25637913b..27a78ca7c 100644 --- a/tests/engine/test_orchestrator_stage_input_bridge.py +++ b/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, diff --git a/tests/entrypoints/openai/test_duplex_protocol.py b/tests/entrypoints/openai/test_duplex_protocol.py index 3c3666026..89ae03f6f 100644 --- a/tests/entrypoints/openai/test_duplex_protocol.py +++ b/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 diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index a37d9d2b1..cd7077c18 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/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 diff --git a/vllm_omni/core/sched/omni_ar_scheduler.py b/vllm_omni/core/sched/omni_ar_scheduler.py index f24edae38..53f559944 100644 --- a/vllm_omni/core/sched/omni_ar_scheduler.py +++ b/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 diff --git a/vllm_omni/core/sched/omni_scheduler_mixin.py b/vllm_omni/core/sched/omni_scheduler_mixin.py index 9920e3a27..38f8551fa 100644 --- a/vllm_omni/core/sched/omni_scheduler_mixin.py +++ b/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 diff --git a/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py b/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py index 9bcc74257..7c9267162 100644 --- a/vllm_omni/distributed/omni_connectors/transfer_adapter/chunk_transfer_adapter.py +++ b/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) diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index 0fb3fb14c..7829012c6 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/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 diff --git a/vllm_omni/experimental/fullduplex/DESIGN.md b/vllm_omni/experimental/fullduplex/DESIGN.md index 256214299..916a2a096 100644 --- a/vllm_omni/experimental/fullduplex/DESIGN.md +++ b/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. diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py index 97391a5c9..6c7e4787c 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/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, } diff --git a/vllm_omni/experimental/fullduplex/openai/protocol.py b/vllm_omni/experimental/fullduplex/openai/protocol.py index 0a87d55b4..9dfe83c9a 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/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, ) diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_input.py b/vllm_omni/experimental/fullduplex/openai/realtime_input.py index e3fe7f3ec..f07c3dd83 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_input.py +++ b/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") diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_output.py b/vllm_omni/experimental/fullduplex/openai/realtime_output.py index 57c07c852..317c5c253 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_output.py +++ b/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( diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_session.py b/vllm_omni/experimental/fullduplex/openai/realtime_session.py index 519c071cf..5b64b9fff 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_session.py +++ b/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}") diff --git a/vllm_omni/experimental/fullduplex/openai/realtime_state.py b/vllm_omni/experimental/fullduplex/openai/realtime_state.py index ac3b2a8de..880a67be6 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_state.py +++ b/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() diff --git a/vllm_omni/experimental/fullduplex/openai/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index 472df0256..63f486e08 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/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: diff --git a/vllm_omni/experimental/fullduplex/openai/session_runner.py b/vllm_omni/experimental/fullduplex/openai/session_runner.py index 4b5e4f2e6..8317a3b97 100644 --- a/vllm_omni/experimental/fullduplex/openai/session_runner.py +++ b/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, diff --git a/vllm_omni/experimental/fullduplex/openai/vad.py b/vllm_omni/experimental/fullduplex/openai/vad.py new file mode 100644 index 000000000..76b6ecef3 --- /dev/null +++ b/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=" 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)) diff --git a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py index 5bbbfbc4a..0ea269d9b 100644 --- a/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py +++ b/vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py @@ -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)