From b103e65efc1e126525989c7a6f3d5ca367cd3bd5 Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Tue, 25 Aug 2026 00:40:09 +0800 Subject: [PATCH] [Bugfix][Full Duplex] Preserve shared runtime compatibility Signed-off-by: Sy03 <1370724210@qq.com> --- .../test_chunk_transfer_adapter.py | 6 ++--- .../openai/test_duplex_protocol.py | 10 +++++++ .../openai_api/test_duplex_handler.py | 27 +++++++++++++++++++ .../chunk_transfer_adapter.py | 15 ----------- .../fullduplex/openai/protocol.py | 1 + .../experimental/fullduplex/openai/serving.py | 17 ++++++++++++ 6 files changed, 58 insertions(+), 18 deletions(-) diff --git a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py b/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py index 8fb6278c1..c3234b138 100644 --- a/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py +++ b/tests/distributed/omni_connectors/test_chunk_transfer_adapter.py @@ -591,7 +591,7 @@ def test_save_async_drops_late_previous_segment_after_boundary_reset(build_adapt assert adapter._pending_save_reqs[-1]["request"] is next_segment -def test_send_single_request_drops_queued_old_segment_after_boundary(build_adapter): +def test_send_single_request_preserves_queued_preboundary_chunk(build_adapter): adapter, connector = build_adapter(stage_id=1) adapter.custom_process_next_stage_input_func = lambda **kwargs: OmniPayloadStruct(meta=MetaStruct()) request = _req("req-queued-old", RequestStatus.WAITING, external_req_id="ext-queued-old") @@ -603,11 +603,11 @@ def test_send_single_request_drops_queued_old_segment_after_boundary(build_adapt old_task = adapter._pending_save_reqs.popleft() adapter._send_single_request(old_task) - connector.put.assert_not_called() + connector.put.assert_called_once() boundary_task = adapter._pending_save_reqs.popleft() adapter._send_single_request(boundary_task) - connector.put.assert_called_once() + assert connector.put.call_count == 2 def test_send_single_request_cleans_up_after_finished_payload(build_adapter, monkeypatch): diff --git a/tests/entrypoints/openai/test_duplex_protocol.py b/tests/entrypoints/openai/test_duplex_protocol.py index 89ae03f6f..41be4a0b0 100644 --- a/tests/entrypoints/openai/test_duplex_protocol.py +++ b/tests/entrypoints/openai/test_duplex_protocol.py @@ -462,6 +462,16 @@ def test_minicpmo_native_capabilities_separate_model_state_from_core_kv_lease(): assert caps["stage_handoff_transport"] == "scheduler_data_plane" +def test_duplex_overlap_policy_auto_remains_a_compatibility_value(): + from vllm_omni.experimental.fullduplex.openai.protocol import ( + DuplexOverlapPolicy, + DuplexSessionConfig, + ) + + assert DuplexOverlapPolicy.AUTO.value == "auto" + assert DuplexSessionConfig._normalize_overlap_policy("auto") == "auto" + + def test_minicpmo_native_capabilities_do_not_overclaim_single_session_deployment(): caps = DuplexCapabilities.minicpmo45_native(max_sessions=1).as_dict() diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index cd7077c18..070bcc1f9 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -5439,6 +5439,33 @@ async def test_minicpmo_native_duplex_open_unsupported_fails_session_create(): assert ws.sent[0]["runtime_control"]["unsupported_count"] == 1 +@pytest.mark.asyncio +async def test_native_runtime_rejects_server_vad_without_barge_in(monkeypatch: pytest.MonkeyPatch): + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler( + chat_service=FakeChatService(engine), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + capabilities = DuplexCapabilities.minicpmo45_native() + capabilities.supports_barge_in = False + monkeypatch.setattr( + handler._serving_runtime_adapter, + "capabilities", + lambda *, max_sessions: capabilities, + ) + event = _native_session_create("sid-native-no-barge-in") + event["session"]["overlap_policy"] = "barge_in_on_speech" + ws = TimedWebSocket() + ws.put(event) + + await handler.handle_session(ws) + + assert ws.sent_types() == ["error"] + assert ws.sent[0]["code"] == "unsupported_turn_detection" + assert engine.opened == [] + + @pytest.mark.asyncio async def test_minicpmo_native_duplex_rejects_engine_without_fence_contract(): class LegacyEngineClient(FakeEngineClient): 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 7668fa5a0..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 @@ -370,21 +370,6 @@ class OmniChunkTransferAdapter(OmniTransferAdapterBase): stage_id = self.connector.stage_id next_stage_id = stage_id + 1 external_req_id = request.external_req_id - task_generation = task.get("segment_generation") - expected_generation = self._segment_generation.get(external_req_id, task_generation) - if ( - isinstance(task_generation, int) - and isinstance(expected_generation, int) - and task_generation < expected_generation - and not is_segment_finished - ): - logger.warning( - "Skip queued late chunk for request %s, segment_generation=%s, expected=%s", - external_req_id, - task_generation, - expected_generation, - ) - return chunk_id = self.put_req_chunk[external_req_id] connector_put_key = f"{external_req_id}_{stage_id}_{chunk_id}" # Process payload in save_loop thread diff --git a/vllm_omni/experimental/fullduplex/openai/protocol.py b/vllm_omni/experimental/fullduplex/openai/protocol.py index 9dfe83c9a..40fdf880b 100644 --- a/vllm_omni/experimental/fullduplex/openai/protocol.py +++ b/vllm_omni/experimental/fullduplex/openai/protocol.py @@ -9,6 +9,7 @@ from uuid import uuid4 class DuplexOverlapPolicy(str, Enum): + AUTO = "auto" LISTEN_ONLY = "listen_only" BARGE_IN_ON_SPEECH = "barge_in_on_speech" diff --git a/vllm_omni/experimental/fullduplex/openai/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index 63f486e08..1cf7dedf3 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/vllm_omni/experimental/fullduplex/openai/serving.py @@ -961,6 +961,23 @@ class OmniDuplexSessionHandler( max_sessions=self._duplex_session_config.max_sessions, ) ) + if ( + config.overlap_policy == DuplexOverlapPolicy.BARGE_IN_ON_SPEECH.value + and not session.capabilities.supports_barge_in + ): + if realtime_protocol is not None: + realtime_protocol.reject_realtime_turn_detection_update() + await send_json( + { + "type": "error", + "error": ( + "server_vad with interrupt_response=true requires a runtime " + "that supports barge-in" + ), + "code": "unsupported_turn_detection", + } + ) + return None session.replace_runtime_config(runtime_config) return _DuplexSessionHandshake(session=session)