[Bugfix][MiniCPM-o] Isolate duplex handoff failures and barge-in (#6170)

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

.buildkite/cuda/test-merge.yml

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

.buildkite/cuda/test-ready.yml

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

examples/online_serving/minicpmo/realtime_duplex_demo.py

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

pyproject.toml

@@ -75,6 +75,7 @@ dev = [
     # Provides top-level cosyvoice2; do not co-install stepaudio2-minicpmo,
     # whose top-level stepaudio2 package conflicts with this distribution.
     "step-audio2==1.0.0", # Step-Audio2 / MiniCPM-o 4.5 Token2Wav (flashcosyvoice / hyperpyyaml)
+    "silero-vad==6.1", # Realtime server-VAD duplex E2E
 ]
 
 demo = [
@@ -84,6 +85,10 @@ demo = [
     "cn2an>=0.5.22",  # Chinese number normalization for TTS (e.g. "1次" -> "一次")
 ]
 
+server-vad = [
+    "silero-vad==6.1",
+]
+
 forced-aligner = [
     "qwen-asr>=0.0.6",  # faithful Japanese/Korean word segmentation for the TTS forced aligner
 ]

tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py

@@ -0,0 +1,85 @@
+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 [e for e in events if e.get("type") == kind and RealtimeEventCollector.response_id(e) == 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))[: PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * 47 // 10]
+    session_id = f"server-vad-hard-interrupt-{id(args)}"
+    client = RealtimeDuplexClient(build_realtime_url(args.url, args.model, autostart=False, session_id=session_id))
+
+    async def until(predicate, label: str) -> None:
+        await wait_for(predicate, timeout_s=args.timeout_s, label=label)
+
+    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 until(lambda: any(e.get("type") == "response.created" for e in client.events.events), "response.created")
+        target_id = RealtimeEventCollector.response_id(
+            next(e for e in client.events.events if e.get("type") == "response.created")
+        )
+        assert target_id
+        await until(lambda: _events(client.events.events, target_id, "response.audio.delta"), "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 until(lambda: _events(client.events.events, target_id, "response.done"), "cancelled response.done")
+        await client.commit()
+        await until(
+            lambda: any(
+                e.get("type") == "response.done" and RealtimeEventCollector.response_id(e) != target_id
+                for e in client.events.events[cursor:]
+            ),
+            "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")
+    terminal = done[0] if len(done) == 1 else None
+    trailing = events[events.index(terminal) + 1 :] if terminal is not None else events
+    stale = any(
+        e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) == target_id for e in trailing
+    )
+    followup_audio = any(
+        e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) != target_id
+        for e in events[cursor:]
+    )
+    ok = (
+        terminal is not None
+        and isinstance(terminal.get("response"), dict)
+        and terminal["response"].get("status") == "cancelled"
+        and terminal["response"].get("status_details", {}).get("reason") == "turn_detected"
+        and not stale
+        and sum(e.get("type") == "input_audio_buffer.speech_started" for e in events[cursor:]) == 1
+        and sum(e.get("type") == "input_audio_buffer.speech_stopped" for e in events[cursor:]) == 1
+        and any(e.get("type") == "input_audio_buffer.committed" for e in events[cursor:])
+        and followup_audio
+        and not any(e.get("type") == "error" for e in events)
+    )
+    return {"ok": ok}

tests/e2e/online_serving/test_minicpmo_4_5_duplex_expansion.py

@@ -25,6 +25,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,
 )
@@ -92,3 +93,22 @@ def test_duplex_soft_interrupt(omni_server, tmp_path: Path) -> None:
     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) -> 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()),
+                chunk_ms=200,
+                timeout_s=180.0,
+            )
+        )
+    )
+    assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2)

tests/engine/test_orchestrator_error_handling.py

@@ -935,6 +935,36 @@ 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, *_args, **_kwargs):
+            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, **_kwargs):
+        pass
+
+    monkeypatch.setattr(fixture.orchestrator, "_cleanup_request_ids", no_cleanup)
+    try:
+        await fixture.orchestrator._forward_to_next_stage_unguarded("bad", 0, _build_request_output("raw"), state)
+        error = await _wait_for_error_message(fixture, request_id="bad")
+        assert error.fatal is False
+    finally:
+        fixture.request_sync_q.put_nowait(ShutdownRequestMessage())
+        fixture.thread.join(timeout=5)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("status_code", "error_type"),

tests/entrypoints/openai/test_duplex_protocol.py

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

tests/entrypoints/openai_api/test_duplex_handler.py

@@ -33,6 +33,7 @@ from vllm_omni.experimental.fullduplex.minicpmo45.runtime import (
 from vllm_omni.experimental.fullduplex.minicpmo45.session import (
     MiniCPMO45ServingSessionState,
 )
+from vllm_omni.experimental.fullduplex.openai import vad as realtime_vad
 from vllm_omni.experimental.fullduplex.openai.protocol import (
     DuplexCapabilities,
     DuplexOverlapPolicy,
@@ -706,7 +707,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)
@@ -729,7 +730,11 @@ 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"]
+    threshold_error = protocol._validate_realtime_turn_detection(
+        {"turn_detection": {"type": "server_vad", "threshold": 0.15}}
+    )
+    assert threshold_error == "turn_detection.threshold must be greater than 0.15 and at most 1"
 
 
 @pytest.mark.asyncio
@@ -753,7 +758,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)
@@ -773,8 +778,55 @@ 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_edge_order_and_tiny_chunk_resampling(monkeypatch):
+    ws = TimedWebSocket()
+    protocol = NativeRealtimeSessionProtocol(ws)  # type: ignore[arg-type]
+    protocol.bind_sender(ws.send_json)
+    protocol._server_vad = realtime_vad.SileroStreamingVAD(realtime_vad.SileroVADConfig())
+    result = realtime_vad.StreamingVADResult(True, True, True, True)
+    monkeypatch.setattr(protocol._server_vad, "process_base64", lambda *args, **kwargs: result)
+    protocol._input_speech_started = True
+    protocol._hold_realtime_output_until_session_created = False
+    await protocol._to_duplex_event(
+        {"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(1), "format": "pcm_f32le"}
+    )
+    assert [event["type"] for event in ws.sent] == [
+        "input_audio_buffer.speech_stopped",
+        "input_audio_buffer.speech_started",
+    ]
+    ws.sent.clear()
+    protocol._input_speech_started = False
+    result = realtime_vad.StreamingVADResult(True, False, True, True)
+    await protocol._to_duplex_event({"type": "input_audio_buffer.append", "audio": "AAAA", "format": "pcm_f32le"})
+    assert [event["type"] for event in ws.sent] == [
+        "input_audio_buffer.speech_started",
+        "input_audio_buffer.speech_stopped",
+    ]
+    detector = realtime_vad.SileroStreamingVAD(realtime_vad.SileroVADConfig(), frame_scorer=lambda _: 0.0)
+    for _ in range(1024):
+        detector.process_base64(_pcm_f32_b64(1), fmt="pcm_f32le", sample_rate_hz=48_000)
+    assert detector._processed_samples + detector._pending.size == 341
+
+
+@pytest.mark.asyncio
+async def test_realtime_server_vad_runtime_failures_are_terminal(monkeypatch):
+    handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient()))
+    ws = TimedWebSocket()
+    ws.put(_server_vad_update("generic-vad", model="test-model", session_id="generic-vad"))
+    await handler.handle_session(ws, realtime_protocol=NativeRealtimeSessionProtocol({}))
+    assert ws.sent[-1]["error"]["code"] == "server_vad_requires_native_duplex"
+    monkeypatch.setattr(realtime_vad.SileroStreamingVAD, "process_base64", lambda *args, **kwargs: 1 / 0)
+    ws = TimedWebSocket()
+    create = _native_realtime_session_update("vad-inference-failure")
+    create["session"]["turn_detection"] = {"type": "server_vad"}
+    ws.put(create)
+    ws.put({"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(1), "format": "pcm_f32le"})
+    await asyncio.wait_for(handler.handle_session(ws, realtime_protocol=NativeRealtimeSessionProtocol({})), timeout=2)
+    assert any(e.get("error", {}).get("code") == "realtime_input_failed" for e in ws.sent)
 
 
 def test_native_duplex_handler_has_no_fixed_session_admission_cap():
@@ -1551,6 +1603,14 @@ def _native_realtime_session_update(
     }
 
 
+def _server_vad_update(event_id: str, **session: object) -> dict[str, object]:
+    return {
+        "type": "session.update",
+        "event_id": event_id,
+        "session": {"turn_detection": {"type": "server_vad"}, **session},
+    }
+
+
 def test_native_stage0_request_id_matches_engine_after_same_session_id_reopen():
     handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient()))
     first = handler._registry.create(session_id="sid-reused")
@@ -5311,7 +5371,7 @@ async def test_audio_clear_preserves_later_wire_order_append():
 
 
 @pytest.mark.asyncio
-async def test_native_append_is_cancelled_by_later_wire_order_input_cancel():
+async def test_native_append_is_cancelled_by_later_wire_order_input_cancel(mocker):
     class SlowAppendEngine(FakeEngineClient):
         def __init__(self) -> None:
             super().__init__()
@@ -5348,6 +5408,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel():
                 fence=fence,
             )
 
+    rollback = mocker.spy(MiniCPMO45PcmAppendBuffer, "_rollback_reservation")
     engine = SlowAppendEngine()
     handler = OmniDuplexSessionHandler(
         chat_service=FakeChatService(engine),
@@ -5378,6 +5439,7 @@ async def test_native_append_is_cancelled_by_later_wire_order_input_cancel():
     await asyncio.wait_for(handler_task, timeout=1)
 
     assert engine.append_cancelled.is_set()
+    rollback.assert_called_once()
     assert engine.signal_fences[-1] == DuplexFence("sid-cancel-slow-append")
     assert engine.signal_next_fences[-1] == DuplexFence("sid-cancel-slow-append", epoch=1)
 
@@ -5473,33 +5535,59 @@ async def test_native_session_update_commits_config_only_after_runtime_ack():
             return await super().signal_duplex_turn_async(session_id, **kwargs)
 
     engine = BlockingUpdateEngine()
-    handler = OmniDuplexSessionHandler(
-        chat_service=FakeChatService(engine),
-        config_timeout_s=0.1,
-        idle_timeout_s=1,
-    )
+    handler = OmniDuplexSessionHandler(chat_service=FakeChatService(engine))
     ws = TimedWebSocket()
-    ws.put(_session_create("sid-update-two-phase"))
-    ws.put(
-        {
-            "type": "turn.signal",
-            "event": "session.update",
-            "payload": {"instructions": "candidate instructions"},
-        }
-    )
+    protocol = NativeRealtimeSessionProtocol(ws)  # type: ignore[arg-type]
+    ws.put(_native_realtime_session_update("sid-update-two-phase"))
+    ws.put(_server_vad_update("update-two-phase"))
 
-    handler_task = asyncio.create_task(handler.handle_session(ws))
+    handler_task = asyncio.create_task(handler.handle_session(ws, realtime_protocol=protocol))
     await asyncio.wait_for(engine.update_started.wait(), timeout=1)
     live_session = handler._registry.get("sid-update-two-phase")
     assert live_session is not None
-    assert live_session.config.instructions != "candidate instructions"
+    assert live_session.config.overlap_policy == DuplexOverlapPolicy.LISTEN_ONLY.value
+    assert protocol._pending_turn_detection_update is not None
+    assert protocol._server_vad is None
+
+    protocol.encode_outbound_event({"type": "error", "code": "unrelated", "error": "unrelated"})
+    assert protocol._pending_turn_detection_update is not None
 
     engine.release_update.set()
     ws.put({"type": "session.close"})
     await asyncio.wait_for(handler_task, timeout=2)
 
-    updated = next(message for message in ws.sent if message.get("type") == "session.updated")
-    assert updated["session"]["instructions"] == "candidate instructions"
+    updated = next(message for message in reversed(ws.sent) if message.get("type") == "session.updated")
+    assert updated["session"]["overlap_policy"] == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value
+    assert isinstance(protocol._server_vad, realtime_vad.SileroStreamingVAD)
+
+
+@pytest.mark.parametrize(
+    ("append_fails", "expected_code"),
+    [(True, "session_update_aborted"), (False, "instructions_update_unsupported")],
+)
+@pytest.mark.asyncio
+async def test_realtime_vad_update_rejection_does_not_stall(append_fails, expected_code):
+    append_result = {"operation": "append", "ok": False, "error_count": 1} if append_fails else None
+    handler = OmniDuplexSessionHandler(chat_service=FakeChatService(FakeEngineClient(append_result=append_result)))
+    ws = TimedWebSocket()
+    protocol = NativeRealtimeSessionProtocol(ws)  # type: ignore[arg-type]
+    create = _native_realtime_session_update("sid-rejected-vad-update")
+    create["session"]["extra_body"]["auto_response"] = append_fails
+    ws.put(create)
+    if append_fails:
+        ws.put({"type": "input_audio_buffer.append", "audio": _pcm_f32_b64(16_000), "format": "pcm_f32le"})
+    update = _server_vad_update("rejected-vad-update")
+    if not append_fails:
+        update["session"]["instructions"] = "You are now a pirate."
+    ws.put(update)
+    ws.put({"type": "session.close"})
+    await asyncio.wait_for(handler.handle_session(ws, realtime_protocol=protocol), timeout=2)
+
+    errors = [message["error"] for message in ws.sent if isinstance(message.get("error"), dict)]
+    error = next(error for error in errors if error.get("code") == expected_code)
+    assert error["event_id"] == "rejected-vad-update"
+    assert protocol._pending_turn_detection_update is None
+    assert protocol._server_vad is None
 
 
 @pytest.mark.asyncio
@@ -5999,7 +6087,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),
@@ -6026,12 +6114,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,
     }
 
 
@@ -6043,7 +6129,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()
@@ -6059,9 +6145,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

tests/model_executor/stage_input_processors/test_nemotron_voicechat.py

@@ -414,8 +414,10 @@ def test_talker_drains_all_received_positions_per_wake() -> None:
     session = talker._sessions["req-0"]
     assert session["step"] == 1 and not session["sync_mode"]
 
-    # First decode wake: drains ALL prompt-region steps (t=1..P-1) at once.
-    _, embeds, update = talker.preprocess(ids, None, _omni_is_prefill=False, **info)
+    # First resumed wake: drains ALL prompt-region steps (t=1..P-1) at once.
+    # vLLM 0.28 reports each resumed one-token segment as prefill; the model
+    # session must still advance instead of restarting its EAR-TTS warmup.
+    _, embeds, update = talker.preprocess(ids, None, _omni_is_prefill=True, **info)
     assert session["step"] == prompt_len  # t advanced 1 -> P in one wake
     assert update["codes"]["audio"].shape == (prompt_len - 1, 31)
     assert float(embeds[0, 0]) == 0.0  # offline stream waits for upstream terminal

vllm_omni/engine/orchestrator.py

@@ -2638,13 +2638,28 @@ 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):
+                raise
+            await self.output_async_queue.put(
+                ErrorMessage(
+                    request_id=req_id,
+                    stage_id=next_logical,
+                    error=f"Stage-{next_logical} input processor failed: {type(exc).__name__}: {exc}",
+                    error_type=type(exc).__name__,
+                )
+            )
+            await self._cleanup_request_ids(
+                [req_id, *self._cfg_tracker.cleanup_parent(req_id)],
+                abort=True,
+                close_duplex_sessions=True,
+            )
+            return
         finally:
             req_state.streaming.source_token_decoder = previous_decoder
 

vllm_omni/experimental/fullduplex/DESIGN.md

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

vllm_omni/experimental/fullduplex/client.py

@@ -497,6 +497,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,
         idle_timeout_s: float | None = None,
         timeout_s: float = 20.0,
@@ -517,8 +518,12 @@ class RealtimeDuplexClient:
             "modalities": ["audio", "text"],
             "input_audio_format": "pcm16",
             "output_audio_format": output_audio_format,
-            "turn_detection": None,
-            "overlap_policy": "listen_only",
+            "turn_detection": dict(turn_detection) if turn_detection is not None else None,
+            "overlap_policy": (
+                "barge_in_on_speech"
+                if turn_detection is not None and turn_detection.get("interrupt_response", True) is True
+                else "listen_only"
+            ),
             "playback_commit_policy": "ack_only",
             "extra_body": session_extra_body,
         }

vllm_omni/experimental/fullduplex/openai/protocol.py

@@ -9,7 +9,6 @@ from uuid import uuid4
 
 
 class DuplexOverlapPolicy(str, Enum):
-    AUTO = "auto"
     LISTEN_ONLY = "listen_only"
     BARGE_IN_ON_SPEECH = "barge_in_on_speech"
 
@@ -101,7 +100,7 @@ class DuplexCapabilities:
         supports_multi_session = max_sessions > 1
         return cls(
             supports_model_native_turn_policy=True,
-            supports_barge_in=False,
+            supports_barge_in=True,
             supports_input_append=True,
             supports_replace_latest_chunk=False,
             supports_reencode_context=False,
@@ -129,6 +128,8 @@ class DuplexCapabilities:
             signal_sources=["model_native", "client_event", "server_policy"],
             stage_handoff_transport="scheduler_data_plane",
             chunk_period_ms=1000,
+            # Barge-in latency depends on client chunking and lacks hardware E2E
+            # measurement, so do not advertise an invented target.
             target_barge_in_latency_ms=None,
         )
 

vllm_omni/experimental/fullduplex/openai/realtime_input.py

@@ -18,6 +18,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:
@@ -92,6 +99,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:
@@ -102,6 +110,7 @@ class RealtimeInputTranslator:
                 "type": "turn.signal",
                 "event": "session.update",
                 "payload": session_payload,
+                "realtime_event_id": event.get("event_id"),
             }
         if event_type == "conversation.item.create":
             item = event.get("item")
@@ -278,15 +287,56 @@ class RealtimeInputTranslator:
                     )
                 )
                 return None
-            looks_like_speech = self._input_looks_like_speech(event, audio=audio, fmt=fmt)
+            detector = self._server_vad
+            vad_result: StreamingVADResult | None = None
+            if isinstance(detector, SileroStreamingVAD):
+                try:
+                    vad_result = detector.process_base64(audio, fmt=fmt, sample_rate_hz=sample_rate_hz)
+                except (ServerVADUnavailableError, ValueError) as exc:
+                    unavailable = isinstance(exc, ServerVADUnavailableError)
+                    await self._send_realtime_payload(
+                        self._realtime_error_payload(
+                            "server_vad_unavailable" if unavailable else "bad_audio",
+                            str(exc),
+                            event_id=event.get("event_id"),
+                            param="turn_detection" if unavailable else "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:
+            stopped_event: dict[str, object] | None = None
+            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
+            if stopped_event is not None and vad_result is not None and vad_result.speech_active:
+                await self._emit_input_speech_stopped(
+                    stopped_event,
+                    item_id=self._active_input_item_id or f"item_{uuid4().hex}",
+                )
+            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 stopped_event is not None and vad_result is not None and not vad_result.speech_active:
+                await self._emit_input_speech_stopped(
+                    stopped_event,
+                    item_id=self._active_input_item_id or f"item_{uuid4().hex}",
+                )
+            if looks_like_speech:
                 self._remember_input_transcript_hint(event)
             payload = {
                 "type": "input_audio_buffer.append",
@@ -309,14 +359,25 @@ 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"] = {
+                    "backend": "silero",
+                    "is_speech": vad_result.is_speech,
+                    "speech_active": vad_result.speech_active,
+                    "speech_started": vad_result.speech_started,
+                    "speech_stopped": vad_result.speech_stopped,
+                    "speech_probability": vad_result.speech_probability,
+                }
+                if vad_result.speech_active:
+                    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),
@@ -338,6 +399,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),
@@ -353,6 +415,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}
@@ -995,25 +1058,124 @@ class RealtimeInputTranslator:
         return None
 
     @staticmethod
-    def _validate_realtime_turn_detection(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"]))
+    def _configured_realtime_turn_detection(session_payload: dict[str, object]) -> tuple[str | None, object]:
+        field = "turn_detection" if "turn_detection" in session_payload else None
+        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:
-                configured_values.append(("audio.input.turn_detection", audio_input["turn_detection"]))
-        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"
-            )
+                field, selected = "audio.input.turn_detection", audio_input["turn_detection"]
+        return field, selected
+
+    @classmethod
+    def _validate_realtime_turn_detection(cls, session_payload: dict[str, object]) -> str | None:
+        field, turn_detection = cls._configured_realtime_turn_detection(session_payload)
+        if field is None:
+            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
+        if turn_detection is not None and not isinstance(turn_detection, dict):
+            return f"{field} must be null or an object with type='server_vad'"
+        if isinstance(turn_detection, dict):
+            if turn_detection.get("type") != "server_vad":
+                return f"{field}.type must be 'server_vad'"
+            threshold = turn_detection.get("threshold", 0.5)
+            if not cls._valid_vad_number(threshold, minimum=SILERO_VAD_MIN_THRESHOLD, strict=True, maximum=1):
+                return f"{field}.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 cls._valid_vad_number(value, minimum=0):
+                    return f"{field}.{name} must be a non-negative number"
+            if turn_detection.get("interrupt_response", True) is not True:
+                return (
+                    f"{field}.interrupt_response=false is unsupported; "
+                    "use turn_detection=null for model-owned listen/speak"
+                )
+        desired_policy = "barge_in_on_speech" if turn_detection is not None 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 _valid_vad_number(value: object, *, minimum: float, maximum: float | None = None, strict: bool = False) -> bool:
+        return (
+            isinstance(value, int | float)
+            and not isinstance(value, bool)
+            and np.isfinite(float(value))
+            and (float(value) > minimum if strict else float(value) >= minimum)
+            and (maximum is None or float(value) <= maximum)
+        )
+
+    def _prepare_realtime_turn_detection_update(self, session_payload: dict[str, object]) -> None:
+        field, configured = self._configured_realtime_turn_detection(session_payload)
+        if field is None:
+            return
+        if self._pending_turn_detection_update is not None:
+            raise RuntimeError("A Realtime turn-detection update is already pending")
+
+        turn_detection: dict[str, object] | None = None
+        detector: SileroStreamingVAD | None = None
+        if configured is None:
+            session_payload["overlap_policy"] = "listen_only"
+        else:
+            assert isinstance(configured, dict)
+            turn_detection = {
+                "type": "server_vad",
+                "interrupt_response": True,
+                "threshold": 0.5,
+                "prefix_padding_ms": 300,
+                "silence_duration_ms": 500,
+                "min_speech_duration_ms": 96,
+                **configured,
+            }
+            detector = SileroStreamingVAD(
+                SileroVADConfig(
+                    threshold=float(turn_detection["threshold"]),
+                    prefix_padding_ms=int(turn_detection["prefix_padding_ms"]),
+                    silence_duration_ms=int(turn_detection["silence_duration_ms"]),
+                    min_speech_duration_ms=max(32, int(turn_detection["min_speech_duration_ms"])),
+                )
+            )
+            session_payload["turn_detection"] = dict(turn_detection)
+            session_payload["overlap_policy"] = "barge_in_on_speech"
+
+        self._pending_turn_detection_update = (turn_detection, detector, asyncio.Event())
+
+    async def wait_for_realtime_turn_detection_update(self) -> None:
+        candidate = self._pending_turn_detection_update
+        if candidate is not None:
+            await candidate[2].wait()
+
+    def commit_realtime_turn_detection_update(self) -> None:
+        self._resolve_realtime_turn_detection_update(commit=True)
+
+    def reject_realtime_turn_detection_update(self) -> None:
+        self._resolve_realtime_turn_detection_update(commit=False)
+
+    def _resolve_realtime_turn_detection_update(self, *, commit: bool) -> None:
+        candidate = self._pending_turn_detection_update
+        if candidate is None:
+            return
+        turn_detection, detector, resolved = candidate
+        discarded = self._server_vad if commit else detector
+        if commit:
+            self._turn_detection, self._server_vad = turn_detection, detector
+        self._pending_turn_detection_update = None
+        resolved.set()
+        reset = getattr(discarded, "reset", None)
+        if callable(reset):
+            reset()
+
+    def _reset_server_vad(self) -> None:
+        reset = getattr(self._server_vad, "reset", None)
+        if callable(reset):
+            reset()
+
     @staticmethod
     def _input_audio_transcription_config(session_payload: dict[str, object]) -> dict[str, object] | None:
         transcription = session_payload.get("input_audio_transcription")

vllm_omni/experimental/fullduplex/openai/realtime_output.py

@@ -262,15 +262,8 @@ 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,
-                    }
-                )
+                # Align server history with acknowledged playback; only explicit
+                # client truncation emits ``conversation.item.truncated``.
             payloads.extend(self._realtime_audio_done_events(event, response_id))
             payloads.extend(
                 self._realtime_response_terminal_events(
@@ -413,7 +406,7 @@ class RealtimeOutputProjector:
                 return [event]
             message = str(raw_error or event.get("message") or "Duplex runtime error")
             code = str(event.get("code") or "duplex_error")
-            return [self._realtime_error_payload(code, message)]
+            return [self._realtime_error_payload(code, message, event_id=event.get("realtime_event_id"))]
         if event_type == "session.closed":
             return [{"type": "session.closed", "event": event}]
         return [{"type": f"duplex.{event_type}", "event": event}]

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)

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,9 @@ 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
+    _pending_turn_detection_update: tuple[dict[str, object] | None, object | None, asyncio.Event] | None = None
     _send_realtime_json: Any = None
     _initial_session_update: bool = False
     _input_speech_started: bool = False
@@ -180,6 +184,11 @@ 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()
+    _pending_turn_detection_update: tuple[dict[str, object] | None, object | None, asyncio.Event] | None = (
+        _RealtimeStateField()
+    )
     _send_realtime_json: Any = _RealtimeStateField()
     _initial_session_update: bool = _RealtimeStateField()
     _input_speech_started: bool = _RealtimeStateField()

vllm_omni/experimental/fullduplex/openai/serving.py

@@ -494,6 +494,22 @@ 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
+            ):
+                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 +517,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 +569,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 +628,17 @@ class OmniDuplexSessionHandler(
             "defer_runtime_append": True,
         }
 
+    @staticmethod
+    def _vad_speech_started(
+        event: Mapping[str, object],
+        payload: Mapping[str, object],
+    ) -> bool | None:
+        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:
@@ -1155,7 +1195,13 @@ class OmniDuplexSessionHandler(
         candidate_config: DuplexSessionConfig,
     ) -> dict[str, object] | None:
         if not self._uses_native_input_append(session):
-            return None
+            if candidate_config.overlap_policy != DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value:
+                return None
+            return {
+                "type": "error",
+                "code": "server_vad_requires_native_duplex",
+                "error": "Realtime server_vad requires a model-native duplex runtime",
+            }
         if (
             candidate_config.instructions != session.config.instructions
             and self._runtime_session_state(session).native_context_locked

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,
@@ -248,6 +249,9 @@ class DuplexSessionRunnerMixin:
                     await actor.enqueue_event(event)
             except WebSocketDisconnect:
                 await actor.enqueue_event({"type": "__disconnect__"})
+            except Exception as exc:
+                await emit_event({"type": "error", "error": str(exc), "code": "realtime_input_failed"})
+                await actor.enqueue_event({"type": "__disconnect__"})
 
         async def next_actor_event() -> dict[str, object]:
             nonlocal pending_turn_reservations, transport_detached
@@ -411,6 +415,8 @@ class DuplexSessionRunnerMixin:
                             )
                     return append_ok
                 except asyncio.CancelledError:
+                    if pcm_reservation is not None:
+                        pcm_reservation.rollback()
                     raise
                 except Exception as exc:
                     if pcm_reservation is not None:
@@ -670,6 +676,9 @@ class DuplexSessionRunnerMixin:
             if handshake is None:
                 return
             session = handshake.session
+            if initial_update_error := self._runtime_session_candidate_update_error(session, session.config):
+                await emit_event(initial_update_error)
+                return
             if handshake.resumed:
                 native = self._serving_runtime_adapter.session_states[session.session_id]
                 actor.tasks = self._session_tasks[session.session_id]
@@ -721,6 +730,8 @@ class DuplexSessionRunnerMixin:
                     )
                 if isinstance(open_result, dict):
                     created_payload["runtime_control"] = self._redact_runtime_control_result(open_result)
+                if realtime_protocol is not None:
+                    realtime_protocol.commit_realtime_turn_detection_update()
                 await emit_event(created_payload)
                 resume_credential_delivered = session.capabilities.supports_session_resume
                 reader_task = asyncio.create_task(read_event_loop(), name="duplex-session-reader")
@@ -903,7 +914,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,
@@ -1068,9 +1083,20 @@ class DuplexSessionRunnerMixin:
                             await emit_event(self._barge_in_unsupported_error(session))
                             continue
                         if turn_event == "session.update":
+                            realtime_event_id = event.get("realtime_event_id")
+
+                            async def emit_update_event(update_event: dict[str, object]) -> None:
+                                if isinstance(realtime_event_id, str) and realtime_event_id:
+                                    update_event = {**update_event, "realtime_event_id": realtime_event_id}
+                                await emit_event(update_event)
+
+                            def reject_update() -> None:
+                                if realtime_protocol is not None:
+                                    realtime_protocol.reject_realtime_turn_detection_update()
+
                             payload = event.get("payload")
                             if not isinstance(payload, dict):
-                                await emit_event(
+                                await emit_update_event(
                                     {
                                         "type": "error",
                                         "session_id": session.session_id,
@@ -1078,12 +1104,22 @@ class DuplexSessionRunnerMixin:
                                         "error": "session.update requires a session payload",
                                     }
                                 )
+                                reject_update()
                                 continue
                             if not await wait_for_native_append_tail():
+                                await emit_update_event(
+                                    {
+                                        "type": "error",
+                                        "code": "session_update_aborted",
+                                        "error": "session.update was not applied because the preceding append failed",
+                                    }
+                                )
+                                reject_update()
                                 continue
                             runtime_update_error = self._runtime_session_update_error(session, payload)
                             if runtime_update_error is not None:
-                                await emit_event(runtime_update_error)
+                                await emit_update_event(runtime_update_error)
+                                reject_update()
                                 continue
                             previous_config = session.config
                             candidate_config = deepcopy(previous_config)
@@ -1093,14 +1129,16 @@ class DuplexSessionRunnerMixin:
                             finally:
                                 session.replace_config(previous_config)
                             if update_error is not None:
-                                await emit_event(update_error)
+                                await emit_update_event(update_error)
+                                reject_update()
                                 continue
                             runtime_update_error = self._runtime_session_candidate_update_error(
                                 session,
                                 candidate_config,
                             )
                             if runtime_update_error is not None:
-                                await emit_event(runtime_update_error)
+                                await emit_update_event(runtime_update_error)
+                                reject_update()
                                 continue
                             try:
                                 candidate_runtime_config = self._runtime_config_for_session_update(
@@ -1108,7 +1146,7 @@ class DuplexSessionRunnerMixin:
                                     candidate_config,
                                 )
                             except ServingRuntimeConfigError as exc:
-                                await emit_event(
+                                await emit_update_event(
                                     {
                                         "type": "error",
                                         "session_id": session.session_id,
@@ -1116,18 +1154,22 @@ class DuplexSessionRunnerMixin:
                                         "error": str(exc),
                                     }
                                 )
+                                reject_update()
                                 continue
                             if not await self._signal_runtime_session(
                                 session,
                                 turn_event,
-                                emit_event,
+                                emit_update_event,
                                 session_config=candidate_config.as_dict(),
                                 runtime_config=candidate_runtime_config,
                             ):
+                                reject_update()
                                 continue
                             session.replace_config(candidate_config)
                             session.replace_runtime_config(candidate_runtime_config)
-                            await emit_event(
+                            if realtime_protocol is not None:
+                                realtime_protocol.commit_realtime_turn_detection_update()
+                            await emit_update_event(
                                 {
                                     "type": "session.updated",
                                     "session": session.as_public_dict(),
@@ -1337,7 +1379,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)
@@ -1378,11 +1429,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
@@ -1398,7 +1450,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,
@@ -1418,7 +1470,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,
@@ -1950,6 +2002,8 @@ class DuplexSessionRunnerMixin:
             with suppress(Exception):
                 await emit_event({"type": "error", "error": str(exc), "code": "internal_error"})
         finally:
+            if realtime_protocol is not None:
+                realtime_protocol.reject_realtime_turn_detection_update()
             if session is not None:
                 if reader_task is not None and not reader_task.done():
                     reader_task.cancel()

vllm_omni/experimental/fullduplex/openai/vad.py

@@ -0,0 +1,178 @@
+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):
+    pass
+
+
+@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
+
+
+@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
+
+
+class SileroStreamingVAD:
+    _SAMPLE_RATE_HZ = 16_000
+    _WINDOW_SAMPLES = 512
+
+    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._clear_stream_state()
+
+    def _clear_stream_state(self) -> 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
+        self._resample_rate_hz: int | None = None
+        self._resample_remainder = 0
+
+    def reset(self) -> None:
+        self._clear_stream_state()
+        reset_states = getattr(self._model, "reset_states", None)
+        if callable(reset_states):
+            reset_states()
+
+    def process_base64(
+        self,
+        audio: object,
+        *,
+        fmt: object,
+        sample_rate_hz: object,
+    ) -> StreamingVADResult:
+        if fmt != "pcm_f32le" or not isinstance(audio, str):
+            raise ValueError("Silero server VAD requires decoded pcm_f32le audio")
+        try:
+            raw = base64.b64decode(audio, validate=True)
+        except (binascii.Error, ValueError) as exc:
+            raise ValueError("Silero server VAD received invalid base64 audio") from exc
+        if len(raw) % 4:
+            raise ValueError("Silero server VAD received an incomplete pcm_f32le frame")
+        samples = np.frombuffer(raw, dtype="<f4").astype(np.float32, copy=False)
+        rate = int(sample_rate_hz) if isinstance(sample_rate_hz, int | float) else self._SAMPLE_RATE_HZ
+        if rate <= 0:
+            raise ValueError("Silero server VAD requires a positive sample rate")
+        if rate != self._SAMPLE_RATE_HZ:
+            if self._resample_rate_hz != rate:
+                self._resample_rate_hz = rate
+                self._resample_remainder = 0
+            numerator = samples.size * self._SAMPLE_RATE_HZ + self._resample_remainder
+            target_size, self._resample_remainder = divmod(numerator, rate)
+            if target_size == 0:
+                samples = np.empty(0, dtype=np.float32)
+            elif samples.size == 1:
+                samples = np.full(target_size, samples[0], dtype=np.float32)
+            else:
+                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)
+        else:
+            self._resample_rate_hz = None
+            self._resample_remainder = 0
+        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 = self.config.min_speech_duration_ms * 16
+        min_silence_samples = self.config.silence_duration_ms * 16
+        negative_threshold = max(0.0, self.config.threshold - SILERO_VAD_MIN_THRESHOLD)
+
+        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 / 16)))
+                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 = self.config.prefix_padding_ms * 16
+                    self._speech_active = True
+                    self._candidate_samples = self._silence_samples = 0
+                    self._candidate_start_sample = None
+                    contained_speech = True
+                    started = True
+                    start_ms = max(0, int(round((candidate_start - prefix_samples) / 16)))
+            else:
+                self._candidate_samples = 0
+                self._candidate_start_sample = None
+
+        return StreamingVADResult(
+            contained_speech, self._speech_active, started, stopped, max_probability, start_ms, end_ms
+        )
+
+    def _score_frame(self, frame: np.ndarray) -> float:
+        if self._frame_scorer is not None:
+            return float(self._frame_scorer(frame))
+        if self._model is None:
+            try:
+                import torch
+                from silero_vad import load_silero_vad
+            except ImportError as exc:
+                raise ServerVADUnavailableError(
+                    "server_vad requires the optional 'silero-vad' package; install vllm-omni[server-vad]"
+                ) from exc
+            self._model = load_silero_vad(onnx=False)
+            self._frame_scorer = lambda value: float(
+                self._model(torch.from_numpy(np.ascontiguousarray(value)), self._SAMPLE_RATE_HZ).item()
+            )
+        return float(self._frame_scorer(frame))

vllm_omni/model_executor/models/nemotron_voicechat/nemotron_voicechat_talker.py

@@ -392,14 +392,13 @@ class NemotronVoiceChatTalkerForConditionalGeneration(nn.Module):
         info = merge_runtime_info(info_dict)
         device = input_ids.device
         span = int(input_ids.shape[0])
-        is_prefill_raw = info.get("_omni_is_prefill")
-        is_prefill = bool(is_prefill_raw) if isinstance(is_prefill_raw, bool) else span > 1
         request_id = require_request_id(info, "talker")
 
         embeds = torch.zeros((span, self._hidden), device=device, dtype=self._dtype)
-        if is_prefill or request_id not in self._sessions:
-            # The 1-token vLLM prefill is the session boundary: run the vendored
-            # warmup here so the first decode step is a pure NeMo t=1 step.
+        if request_id not in self._sessions:
+            # The request id, not vLLM's per-segment prefill flag, is the
+            # session boundary. Resumable updates are prefill again in v0.28
+            # and must preserve the existing EAR-TTS cache and step.
             self._init_session(request_id, info, device)
             return input_ids, embeds, {}
 

vllm_omni/model_executor/stage_input_processors/minicpmo_4_5_omni.py

@@ -550,6 +550,29 @@ 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:
+    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 +780,16 @@ 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
+                ):
+                    continue
                 raise ValueError("No latent or hidden_states found in thinker output")
 
         thinker_hidden_states = latent.detach()
@@ -771,7 +800,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)