diff --git a/tests/e2e/features/fullduplex/personaplex/test_unified_runtime.py b/tests/e2e/features/fullduplex/personaplex/test_unified_runtime.py index f4a3c994e..f2970a240 100644 --- a/tests/e2e/features/fullduplex/personaplex/test_unified_runtime.py +++ b/tests/e2e/features/fullduplex/personaplex/test_unified_runtime.py @@ -18,6 +18,10 @@ from vllm_omni.experimental.fullduplex.engine.duplex_runtime import ( DuplexInputMode, ) from vllm_omni.experimental.fullduplex.engine.messages import DuplexFence +from vllm_omni.experimental.fullduplex.openai.runtime_adapter import ( + ServingRuntimeConfigError, +) +from vllm_omni.experimental.fullduplex.personaplex.config import DEFAULT_PERSONA from vllm_omni.experimental.fullduplex.personaplex.data_plane import ( PersonaPlexDataPlaneContext, PersonaPlexDataPlaneSession, @@ -242,6 +246,53 @@ def test_personaplex_capabilities_are_honest() -> None: assert multi.supports_barge_in is False +def test_personaplex_runtime_config_update_rejects_changed_persona() -> None: + adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None) + short_persona = "You are a helpful assistant." + longer_persona = ( + "You are now a swashbuckling pirate captain who speaks entirely in " + "nautical slang, references buried treasure constantly, and never " + "breaks character no matter what the user asks." + ) + current = {"personaplex_persona": short_persona, "personaplex_voice_prompt": "NATF2.pt"} + + with pytest.raises(ServingRuntimeConfigError, match="persona"): + adapter.runtime_config_for_update( + SimpleNamespace(instructions=longer_persona, extra_body={}), + current, + ) + + unchanged = adapter.runtime_config_for_update( + SimpleNamespace(instructions=short_persona, extra_body={}), + current, + ) + assert unchanged["personaplex_persona"] == short_persona + + +def test_personaplex_runtime_config_update_rejects_changed_voice() -> None: + adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None) + current = {"personaplex_persona": DEFAULT_PERSONA, "personaplex_voice_prompt": "NATF2.pt"} + + with pytest.raises(ServingRuntimeConfigError, match="voice"): + adapter.runtime_config_for_update( + SimpleNamespace(instructions=None, voice="OtherVoice.pt", extra_body={}), + current, + ) + + unchanged = adapter.runtime_config_for_update( + SimpleNamespace(instructions=None, voice="NATF2.pt", extra_body={}), + current, + ) + assert unchanged["personaplex_voice_prompt"] == "NATF2.pt" + + +def test_personaplex_rejects_minicpmo_only_opt_out_flag() -> None: + adapter = PersonaPlexServingRuntimeAdapter(lambda *_: None) + + with pytest.raises(ServingRuntimeConfigError, match="minicpmo45_native_duplex"): + adapter.validate_client_extra_body({"minicpmo45_native_duplex": False}) + + def test_pcm_buffer_emits_one_80ms_frame_transactionally() -> None: buffer = PersonaPlexPcmAppendBuffer() reservation = buffer.prepare_append( diff --git a/tests/entrypoints/openai_api/test_duplex_handler.py b/tests/entrypoints/openai_api/test_duplex_handler.py index d45cac20c..617a8a101 100644 --- a/tests/entrypoints/openai_api/test_duplex_handler.py +++ b/tests/entrypoints/openai_api/test_duplex_handler.py @@ -49,6 +49,9 @@ from vllm_omni.experimental.fullduplex.openai.serving import ( ) from vllm_omni.experimental.fullduplex.openai.websocket import DuplexWebSocketActor from vllm_omni.experimental.fullduplex.output import attach_duplex_output_decision +from vllm_omni.experimental.fullduplex.personaplex.serving_adapter import ( + PersonaPlexServingRuntimeAdapter, +) from vllm_omni.outputs import OmniRequestOutput pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -991,6 +994,178 @@ async def test_minicpmo_native_session_update_requires_ref_audio_before_enabling assert ("sid-runtime-update-ref-required", "session.update") not in engine.signals +def test_personaplex_candidate_update_does_not_leak_minicpmo_ref_audio_check(): + handler = OmniDuplexSessionHandler( + chat_service=SimpleNamespace(engine_client=SimpleNamespace()), + serving_runtime_adapter=PersonaPlexServingRuntimeAdapter(lambda *_: None), + ) + session = DuplexSession( + session_id="sid-personaplex-voice-change", + config=DuplexSessionConfig(modalities=["audio"], voice="NATF2.pt"), + capabilities=PersonaPlexServingRuntimeAdapter.capabilities(max_sessions=1), + ) + candidate_config = DuplexSessionConfig(modalities=["audio"], voice="OtherVoice.pt") + + error = handler._runtime_session_candidate_update_error(session, candidate_config) + + assert error is None + + +@pytest.mark.asyncio +async def test_minicpmo_native_session_update_rejects_instructions_after_native_context_locked(): + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler( + chat_service=FakeChatService(engine), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + ws = TimedWebSocket() + ws.put(_native_session_create("sid-instructions-locked", modalities=["text"])) + ws.put( + { + "type": "input_audio_buffer.append", + "audio": _pcm_f32_b64(16000, value=0.05), + "format": "pcm_f32le", + "sample_rate_hz": 16000, + "duration_ms": 1000, + "is_speech": True, + } + ) + ws.put( + { + "type": "turn.signal", + "event": "session.update", + "payload": {"instructions": "You are now a pirate."}, + } + ) + ws.put({"type": "session.close"}) + + await handler.handle_session(ws) + + error = next(message for message in ws.sent if message.get("type") == "error") + assert error["code"] == "instructions_update_unsupported" + assert "session.updated" not in ws.sent_types() + assert ("sid-instructions-locked", "session.update") not in engine.signals + + +@pytest.mark.asyncio +async def test_minicpmo_native_session_update_allows_unchanged_instructions_after_lock(): + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler( + chat_service=FakeChatService(engine), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + ws = TimedWebSocket() + ws.put(_native_session_create("sid-instructions-unchanged", modalities=["text"])) + ws.put( + { + "type": "input_audio_buffer.append", + "audio": _pcm_f32_b64(16000, value=0.05), + "format": "pcm_f32le", + "sample_rate_hz": 16000, + "duration_ms": 1000, + "is_speech": True, + } + ) + ws.put( + { + "type": "turn.signal", + "event": "session.update", + "payload": {"instructions": "You are a concise assistant.", "temperature": 0.3}, + } + ) + ws.put({"type": "session.close"}) + + await handler.handle_session(ws) + + assert "session.updated" in ws.sent_types() + assert not any( + message.get("type") == "error" and message.get("code") == "instructions_update_unsupported" + for message in ws.sent + ) + + +@pytest.mark.asyncio +async def test_minicpmo_native_session_update_rejects_instructions_before_first_real_append(): + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler( + chat_service=FakeChatService(engine), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + ws = TimedWebSocket() + ws.put(_native_session_create("sid-buffered-before-lock", modalities=["text"])) + ws.put( + { + "type": "input_audio_buffer.append", + "audio": _pcm_f32_b64(3200, value=0.05), + "format": "pcm_f32le", + "sample_rate_hz": 16000, + "duration_ms": 200, + "is_speech": True, + } + ) + ws.put( + { + "type": "turn.signal", + "event": "session.update", + "payload": { + "instructions": ( + "You are now a swashbuckling pirate captain who speaks entirely in " + "nautical slang, references buried treasure constantly, and never " + "breaks character no matter what the user asks." + ) + }, + } + ) + ws.put({"type": "input_audio_buffer.commit", "final": True, "response_create": True}) + ws.put({"type": "session.close"}) + + await handler.handle_session(ws) + + error = next(message for message in ws.sent if message.get("type") == "error") + assert error["code"] == "instructions_update_unsupported" + assert "session.updated" not in ws.sent_types() + assert engine.appended, "the commit must still flush the buffered chunk despite the rejected update" + + +@pytest.mark.asyncio +async def test_minicpmo_native_session_update_rejects_native_duplex_mode_flip(): + engine = FakeEngineClient() + handler = OmniDuplexSessionHandler( + chat_service=FakeChatService(engine), + config_timeout_s=0.1, + idle_timeout_s=1, + ) + ws = TimedWebSocket() + ws.put(_native_session_create("sid-native-duplex-flag-flip", modalities=["text"])) + ws.put( + { + "type": "turn.signal", + "event": "session.update", + "payload": {"extra_body": {"minicpmo45_native_duplex": False}}, + } + ) + ws.put( + { + "type": "turn.signal", + "event": "session.update", + "payload": {"instructions": "You are now a pirate."}, + } + ) + ws.put({"type": "session.close"}) + + await handler.handle_session(ws) + + errors = [message for message in ws.sent if message.get("type") == "error"] + assert [error["code"] for error in errors] == [ + "native_duplex_mode_update_unsupported", + "instructions_update_unsupported", + ] + assert "session.updated" not in ws.sent_types() + + def test_native_realtime_protocol_audio_delta_preserves_sample_rate_hz(): ws = TimedWebSocket() protocol = NativeRealtimeSessionProtocol(ws) # type: ignore[arg-type] diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py b/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py index d3d81f31d..0afffa03c 100644 --- a/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py +++ b/vllm_omni/experimental/fullduplex/minicpmo45/adapter.py @@ -14,6 +14,7 @@ from vllm_omni.experimental.fullduplex.minicpmo45.policy import MiniCPMO45Duplex from vllm_omni.experimental.fullduplex.openai.protocol import DuplexSessionConfig from vllm_omni.experimental.fullduplex.openai.runtime_adapter import ( ServingRuntimeConfigError, + reject_changed_runtime_value, ) @@ -69,7 +70,13 @@ class MiniCPMO45NativeDuplexServingAdapter: current: object, ) -> dict[str, object]: runtime_config = deepcopy(dict(current)) if isinstance(current, dict) else {} - runtime_config["instructions"] = config.instructions + reject_changed_runtime_value( + config.instructions, + runtime_config.get("instructions"), + message="instructions cannot be changed after the session is created", + code="instructions_update_unsupported", + error_cls=MiniCPMO45ClientRuntimeConfigError, + ) stage_max_tokens = runtime_config.get("duplex_stage_max_tokens") stage_max_tokens = deepcopy(stage_max_tokens) if isinstance(stage_max_tokens, dict) else {} stage_max_tokens["0"] = ( diff --git a/vllm_omni/experimental/fullduplex/minicpmo45/session.py b/vllm_omni/experimental/fullduplex/minicpmo45/session.py index 763bcf8b3..1848ee81b 100644 --- a/vllm_omni/experimental/fullduplex/minicpmo45/session.py +++ b/vllm_omni/experimental/fullduplex/minicpmo45/session.py @@ -16,6 +16,7 @@ class MiniCPMO45ServingSessionState: audio_buffer: MiniCPMO45PcmAppendBuffer = field(default_factory=MiniCPMO45PcmAppendBuffer) input_since_commit: bool = False speech_since_commit: bool = False + native_context_locked: bool = False committed_audio_payload: dict[str, object] | None = None committed_audio_operation_id: str | None = None committed_audio_reserved_bytes: int = 0 diff --git a/vllm_omni/experimental/fullduplex/openai/runtime_adapter.py b/vllm_omni/experimental/fullduplex/openai/runtime_adapter.py index f88cffd3b..710b10022 100644 --- a/vllm_omni/experimental/fullduplex/openai/runtime_adapter.py +++ b/vllm_omni/experimental/fullduplex/openai/runtime_adapter.py @@ -14,6 +14,18 @@ class ServingRuntimeConfigError(ValueError): self.code = code +def reject_changed_runtime_value( + new_value: object, + current_value: object, + *, + message: str, + code: str, + error_cls: type[ServingRuntimeConfigError] = ServingRuntimeConfigError, +) -> None: + if new_value != current_value: + raise error_cls(message, code=code) + + class PcmAppendReservation(Protocol): operation_id: str payload: dict[str, object] | None @@ -64,6 +76,7 @@ class ServingRuntimeSessionState(Protocol): audio_buffer: PcmAppendBuffer input_since_commit: bool speech_since_commit: bool + native_context_locked: bool committed_audio_payload: dict[str, object] | None committed_audio_operation_id: str | None committed_audio_reserved_bytes: int diff --git a/vllm_omni/experimental/fullduplex/openai/serving.py b/vllm_omni/experimental/fullduplex/openai/serving.py index 472df0256..d7eb31214 100644 --- a/vllm_omni/experimental/fullduplex/openai/serving.py +++ b/vllm_omni/experimental/fullduplex/openai/serving.py @@ -1156,6 +1156,18 @@ class OmniDuplexSessionHandler( ) -> dict[str, object] | None: if not self._uses_native_input_append(session): return None + if ( + candidate_config.instructions != session.config.instructions + and self._runtime_session_state(session).native_context_locked + ): + return { + "type": "error", + "session_id": session.session_id, + "code": "instructions_update_unsupported", + "error": "session.update cannot change instructions after the native duplex context is initialized", + } + if session.config.extra_body.get("minicpmo45_native_duplex") is not True: + return None if not self._config_requests_audio_output(candidate_config): return None if "ref_audio_data" in session.runtime_config: @@ -1326,6 +1338,19 @@ class OmniDuplexSessionHandler( "code": "ref_audio_update_unsupported", "error": "session.update cannot change ref_audio after the native duplex runtime is open", } + extra_body_payload = payload.get("extra_body") + if isinstance(extra_body_payload, dict) and "minicpmo45_native_duplex" in extra_body_payload: + current_native_duplex = session.config.extra_body.get("minicpmo45_native_duplex") + if ( + current_native_duplex is not None + and extra_body_payload["minicpmo45_native_duplex"] != current_native_duplex + ): + return { + "type": "error", + "session_id": session.session_id, + "code": "native_duplex_mode_update_unsupported", + "error": "session.update cannot change minicpmo45_native_duplex after the session is created", + } if isinstance(payload.get("instructions"), str): session.config.instructions = str(payload["instructions"]) elif "instructions" in payload and payload.get("instructions") is None: diff --git a/vllm_omni/experimental/fullduplex/openai/session_runner.py b/vllm_omni/experimental/fullduplex/openai/session_runner.py index 4b5e4f2e6..d94dfb43c 100644 --- a/vllm_omni/experimental/fullduplex/openai/session_runner.py +++ b/vllm_omni/experimental/fullduplex/openai/session_runner.py @@ -28,6 +28,7 @@ from vllm_omni.experimental.fullduplex.openai.realtime_session import ( ) from vllm_omni.experimental.fullduplex.openai.runtime_adapter import ( PcmAppendReservation, + ServingRuntimeConfigError, ServingRuntimeSessionState, payload_turn_id, ) @@ -357,6 +358,7 @@ class DuplexSessionRunnerMixin: expected_epoch=append_epoch, ) if append_ok: + native.native_context_locked = True if pcm_reservation is not None: pcm_reservation.commit() session.release_input_bytes(pcm_reservation.byte_count) @@ -1072,10 +1074,21 @@ class DuplexSessionRunnerMixin: if runtime_update_error is not None: await emit_event(runtime_update_error) continue - candidate_runtime_config = self._runtime_config_for_session_update( - session, - candidate_config, - ) + try: + candidate_runtime_config = self._runtime_config_for_session_update( + session, + candidate_config, + ) + except ServingRuntimeConfigError as exc: + await emit_event( + { + "type": "error", + "session_id": session.session_id, + "code": exc.code, + "error": str(exc), + } + ) + continue if not await self._signal_runtime_session( session, turn_event, diff --git a/vllm_omni/experimental/fullduplex/personaplex/serving_adapter.py b/vllm_omni/experimental/fullduplex/personaplex/serving_adapter.py index 66244ca68..d4f920d44 100644 --- a/vllm_omni/experimental/fullduplex/personaplex/serving_adapter.py +++ b/vllm_omni/experimental/fullduplex/personaplex/serving_adapter.py @@ -15,6 +15,7 @@ from vllm_omni.experimental.fullduplex.openai.protocol import ( ) from vllm_omni.experimental.fullduplex.openai.runtime_adapter import ( ServingRuntimeConfigError, + reject_changed_runtime_value, ) from vllm_omni.experimental.fullduplex.personaplex.config import DEFAULT_PERSONA from vllm_omni.experimental.fullduplex.personaplex.data_plane import ( @@ -31,6 +32,7 @@ _PRIVATE_RUNTIME_CONFIG_KEYS = frozenset( { "personaplex_prefill_slots", "personaplex_model_path", + "minicpmo45_native_duplex", } ) @@ -40,6 +42,7 @@ class PersonaPlexServingSessionState: audio_buffer: PersonaPlexPcmAppendBuffer = field(default_factory=PersonaPlexPcmAppendBuffer) input_since_commit: bool = False speech_since_commit: bool = False + native_context_locked: bool = False committed_audio_payload: dict[str, object] | None = None committed_audio_operation_id: str | None = None committed_audio_reserved_bytes: int = 0 @@ -190,9 +193,23 @@ class PersonaPlexServingRuntimeAdapter: current: Mapping[str, object], ) -> dict[str, object]: cls.validate_client_extra_body(getattr(config, "extra_body", None)) + new_persona = str(getattr(config, "instructions", None) or DEFAULT_PERSONA) + reject_changed_runtime_value( + new_persona, + current.get("personaplex_persona"), + message="PersonaPlex persona (instructions) cannot be changed after the session is created", + code="persona_update_unsupported", + ) + new_voice = cls._voice_name(getattr(config, "voice", None)) + reject_changed_runtime_value( + new_voice, + current.get("personaplex_voice_prompt"), + message="PersonaPlex voice cannot be changed after the session is created", + code="voice_update_unsupported", + ) runtime_config = deepcopy(dict(current)) - runtime_config["personaplex_voice_prompt"] = cls._voice_name(getattr(config, "voice", None)) - runtime_config["personaplex_persona"] = str(getattr(config, "instructions", None) or DEFAULT_PERSONA) + runtime_config["personaplex_voice_prompt"] = new_voice + runtime_config["personaplex_persona"] = new_persona return runtime_config @staticmethod