[Bugfix] Refresh duplex client defaults after session updates (#7784)

Signed-off-by: wuli666 <421774554@qq.com>

tests/clients/test_duplex_client.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 import asyncio
 import base64
 import json
+import struct
 import wave
 from collections.abc import Callable
 from typing import Any
@@ -33,6 +34,7 @@ from vllm_omni.clients.duplex import (
     ReconnectPolicy,
     SessionConfig,
     SessionResumed,
+    SessionUpdated,
     build_realtime_url,
     chunk_period_ms,
     duplex_unit_boundary_ms,
@@ -797,6 +799,52 @@ async def test_inline_client_submits_typed_commands_in_order():
         assert ack.played_ms == 120 and ack.response_id == "resp-1"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("accepted", [True, False])
+@pytest.mark.parametrize(
+    "patch",
+    [
+        {"input_audio_format": "pcm_f32le", "sample_rate_hz": 24_000},
+        {"audio": {"input": {"format": "pcm_f32le", "sample_rate_hz": 24_000}}},
+    ],
+)
+async def test_inline_input_defaults_change_only_after_session_updated(accepted, patch):
+    client, omni = make_inline_client()
+    patch = {**patch, "overlap_silence_rms": 0.1}
+    async with client:
+        handle = omni.handles[0]
+        previous_session = dict(client.session_info)
+        await client.send({"type": "session.update", "session": patch})
+        # Sending a patch is not an acknowledgement of its acceptance.
+        assert client.session_info == previous_session
+        assert client._input_defaults().input_sample_rate_hz == 16_000
+        updated_session = {**previous_session, **patch}
+        event: dict[str, object] = (
+            {"type": "session.updated", "session": updated_session}
+            if accepted
+            else {"type": "error", "error": {"code": "invalid_session_update", "message": "rejected"}}
+        )
+        handle.feed(event)
+        received = await client.wait_for("session.updated" if accepted else "error", timeout_s=2.0)
+        assert isinstance(received, SessionUpdated if accepted else ErrorEvent)
+
+        pcm = struct.pack("<f", 0.05) * 16
+        await client.send({"type": "input_audio_buffer.append", "audio": _b64(pcm)})
+        command = handle.commands[-1]
+        assert isinstance(command, duplex_commands.AppendAudio)
+        assert command.format == "pcm_f32le"
+        if accepted:
+            assert client.session_info == updated_session
+            assert command.audio == pcm
+            assert command.sample_rate_hz == 24_000
+            assert command.is_speech is False
+        else:
+            assert client.session_info == previous_session
+            assert len(command.audio) == 2 * len(pcm)  # Still decoded as pcm16.
+            assert command.sample_rate_hz == 16_000
+            assert command.is_speech is True
+
+
 @pytest.mark.asyncio
 async def test_inline_session_close_maps_to_handle_close():
     client, omni = make_inline_client()

vllm_omni/clients/duplex.py

@@ -72,6 +72,7 @@ __all__ = [
     "ResponseHandle",
     "SessionConfig",
     "SessionCreated",
+    "SessionUpdated",
     "SpeakDecision",
     "WebSocketTransport",
     "TextDelta",
@@ -376,6 +377,15 @@ class SessionResumed(SessionCreated):
     pass
 
 
+class SessionUpdated(DuplexEvent):
+    """An acknowledged replacement of the public session configuration."""
+
+    @property
+    def session(self) -> dict[str, object]:
+        value = self.raw.get("session")
+        return value if isinstance(value, dict) else {}
+
+
 class SessionClosed(DuplexEvent):
     pass
 
@@ -456,6 +466,7 @@ class ErrorEvent(DuplexEvent):
 _EVENT_TYPES: dict[str, type[DuplexEvent]] = {
     "session.created": SessionCreated,
     "session.resumed": SessionResumed,
+    "session.updated": SessionUpdated,
     "session.closed": SessionClosed,
     "session.expired": SessionExpired,
     "response.created": ResponseCreated,
@@ -1011,6 +1022,9 @@ class DuplexClientBase(ABC):
             self._adopt_session(event)
         elif isinstance(event, SessionCreated):
             self._adopt_session(event)
+        elif isinstance(event, SessionUpdated):
+            if event.session:
+                self.session_info = event.session
         elif isinstance(event, ResponseCreated):
             response_id = event.response_id
             if response_id and response_id not in self._responses: