diff --git a/examples/online_serving/minicpmo/realtime_duplex_demo.py b/examples/online_serving/minicpmo/realtime_duplex_demo.py index bca251f4e..b6c861747 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; omitted to preserve the model default.", + help="Stage0 sampling temperature; response-required soft-interrupt uses 0.0.", ) parser.add_argument("--no-realtime-pacing", action="store_true") parser.add_argument("--require-audio", action="store_true") diff --git a/tests/engine/test_orchestrator_error_handling.py b/tests/engine/test_orchestrator_error_handling.py index 0ec131e1a..4176764bc 100644 --- a/tests/engine/test_orchestrator_error_handling.py +++ b/tests/engine/test_orchestrator_error_handling.py @@ -880,39 +880,6 @@ 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/entrypoints/openai/test_duplex_protocol.py b/tests/entrypoints/openai/test_duplex_protocol.py index 41be4a0b0..68f27c91c 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 True + assert caps["supports_barge_in"] is False 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 3f0d98b35..a37d9d2b1 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -5674,7 +5674,7 @@ async def test_minicpmo_native_duplex_rejects_ref_audio_path(): assert engine.opened == [] -def test_minicpmo_native_duplex_explicit_barge_in_request_interrupts(): +def test_minicpmo_native_duplex_explicit_barge_in_request_is_deferred_as_listen(): engine = FakeEngineClient() handler = OmniDuplexSessionHandler( chat_service=FakeChatService(engine), @@ -5701,10 +5701,12 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_interrupts(): ) assert decision == { - "action": "barge_in", - "reason": "client_force_barge_in", + "action": "listen", + "reason": "barge_in_unsupported", "duration_ms": 1000, + "overlap_speech_ms": 1000, "buffer_audio": True, + "defer_runtime_append": True, } @@ -5716,7 +5718,7 @@ def test_minicpmo_native_duplex_explicit_barge_in_request_interrupts(): {"type": "turn.signal", "event": "barge_in"}, ], ) -async def test_minicpmo_native_duplex_accepts_explicit_barge_in_control( +async def test_minicpmo_native_duplex_rejects_unadvertised_barge_in_control( barge_in_event: dict[str, object], ): engine = FakeEngineClient() @@ -5732,8 +5734,9 @@ async def test_minicpmo_native_duplex_accepts_explicit_barge_in_control( await handler.handle_session(ws) - 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 + 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 @pytest.mark.asyncio diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index 7829012c6..f254cf499 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -2176,32 +2176,13 @@ class Orchestrator: req_state.prompt, streaming_context=req_state.streaming, ) - except Exception as exc: + except Exception: logger.exception( "[Orchestrator] req=%s process_engine_inputs FAILED for stage-%s", req_id, next_logical, ) - 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 + raise 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 916a2a096..256214299 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; -- an acoustic-onset-to-cancel latency target across arbitrary client chunk sizes; +- deterministic VAD-triggered interruption (the browser intentionally does not + run VAD; MiniCPM owns listen/speak decisions at model-unit boundaries); - production multi-session admission, fairness, capacity, or failure recovery; - bounded long-session KV; - video input or audio/video synchronization. @@ -640,8 +640,9 @@ 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. A separate explicit -`server_vad` policy is documented in the Realtime contract above. +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. 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 @@ -1023,6 +1024,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 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. +It does not support claims for deterministic VAD-triggered interruption, +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 6c7e4787c..97391a5c9 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/vllm_omni/experimental/fullduplex/client.py @@ -417,7 +417,6 @@ 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: @@ -437,12 +436,8 @@ class RealtimeDuplexClient: "modalities": ["audio", "text"], "input_audio_format": "pcm16", "output_audio_format": output_audio_format, - "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" - ), + "turn_detection": None, + "overlap_policy": "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 40fdf880b..0a87d55b4 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/vllm_omni/experimental/fullduplex/openai/protocol.py @@ -101,7 +101,7 @@ class DuplexCapabilities: supports_multi_session = max_sessions > 1 return cls( supports_model_native_turn_policy=True, - supports_barge_in=True, + supports_barge_in=False, supports_input_append=True, supports_replace_latest_chunk=False, supports_reencode_context=False, @@ -129,9 +129,6 @@ 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_output.py b/vllm_omni/experimental/fullduplex/openai/realtime_output.py index 317c5c253..57c07c852 100644 --- a/vllm_omni/experimental/fullduplex/openai/realtime_output.py +++ b/vllm_omni/experimental/fullduplex/openai/realtime_output.py @@ -262,10 +262,15 @@ class RealtimeOutputProjector: content_index=0, audio_end_ms=committed_audio_ms, ) - # 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.append( + { + "type": "conversation.item.truncated", + "item_id": item_id, + "content_index": 0, + "audio_end_ms": committed_audio_ms, + "event": event, + } + ) 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/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index 15a804f86..472df0256 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/vllm_omni/experimental/fullduplex/openai/serving.py @@ -501,7 +501,7 @@ class OmniDuplexSessionHandler( "overlap_speech_ms": session.overlap_speech_ms, "buffer_audio": True, "defer_runtime_append": False, - "force_listen": event.get("force_listen") is True or payload.get("force_listen") is True, + "force_listen": False, "preserve_realtime_input": True, } if bool(event.get("force_listen", False)): diff --git a/vllm_omni/experimental/fullduplex/openai/session_runner.py b/vllm_omni/experimental/fullduplex/openai/session_runner.py index 8317a3b97..4b5e4f2e6 100644 --- a/vllm_omni/experimental/fullduplex/openai/session_runner.py +++ b/vllm_omni/experimental/fullduplex/openai/session_runner.py @@ -18,7 +18,6 @@ from vllm_omni.experimental.fullduplex.openai.commit_policy import ( decide_commit_action, ) from vllm_omni.experimental.fullduplex.openai.protocol import ( - DuplexOverlapPolicy, DuplexPlaybackCommitPolicy, DuplexSession, DuplexSessionState, @@ -876,11 +875,7 @@ 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 "client_cancelled" - if event_type == "response.cancel" - else "barge_in" + "output_audio_buffer_clear" if event_type == "output_audio_buffer.clear" else "barge_in" ) cancelled_fence = DuplexFence( session.session_id, @@ -1254,16 +1249,7 @@ class DuplexSessionRunnerMixin: buffer_overlap_audio = True if self._uses_native_input_append(session): mark_pending_silence_superseded() - 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) - ) - ) - ) + overlap_active = not self._session_auto_responds(session) and native_response_in_progress() if overlap_active: decision = self._overlap_decision(session, event, payload) await self._emit_overlap_decision(emit_event, session, decision) @@ -1304,12 +1290,11 @@ 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=cancel_reason, + reason="barge_in", ) had_native_stream = ( await self._cancel_native_data_plane_stream(session) or had_native_stream @@ -1325,7 +1310,7 @@ class DuplexSessionRunnerMixin: "type": "audio.cancelled", "session_id": session.session_id, "response_id": old_response_id, - "reason": cancel_reason, + "reason": "barge_in", "cancelled_epoch": old_epoch, "epoch": new_epoch, "committed_ms": committed_ms, @@ -1345,7 +1330,7 @@ class DuplexSessionRunnerMixin: "type": "audio.cancelled", "session_id": session.session_id, "response_id": session.last_response_id, - "reason": cancel_reason, + "reason": "barge_in", "cancelled_epoch": old_epoch, "epoch": new_epoch, "committed_ms": committed_ms, 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 0ea269d9b..5bbbfbc4a 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,35 +550,6 @@ 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, @@ -786,23 +757,10 @@ 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() @@ -813,6 +771,7 @@ 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)