From 2d0b03f717e96fc8030859dcadbf96e7f8ea8145 Mon Sep 17 00:00:00 2001 From: psv666 <150513104+psv666@users.noreply.github.com> Date: Sun, 20 Sep 2026 22:42:17 +0800 Subject: [PATCH] [Bugfix] Fix Qwen3-Omni realtime playback interruption and conversation history (#7791) Signed-off-by: psv666 <2693925048@qq.com> --- docs/serving/realtime_duplex_api.md | 33 +- .../realtime_web/app/static/app.js | 63 ++- .../app/static/profiles/qwen3_turn.js | 18 +- tests/engine/duplex/test_duplex_events.py | 69 ++- tests/engine/duplex/test_engine_session.py | 78 ++++ .../test_qwen3_omni_realtime_websocket.py | 410 +++++++++++++++++- tests/examples/test_realtime_web_profiles.cjs | 192 +++++++- .../models/qwen3_omni/test_duplex_plugin.py | 331 +++++++++++++- vllm_omni/engine/duplex/realtime_events.py | 16 +- .../engine/duplex/session/engine_session.py | 25 +- vllm_omni/engine/duplex/session/runner.py | 2 +- .../models/qwen3_omni/duplex/plugin.py | 118 +++-- .../models/qwen3_omni/duplex/session.py | 5 +- 13 files changed, 1292 insertions(+), 68 deletions(-) diff --git a/docs/serving/realtime_duplex_api.md b/docs/serving/realtime_duplex_api.md index 029f35f58..cd8832c82 100644 --- a/docs/serving/realtime_duplex_api.md +++ b/docs/serving/realtime_duplex_api.md @@ -8,12 +8,41 @@ history stays honest, and a dropped connection can resume the same session. This page covers how to run a duplex deployment, how to drive it from Python with `vllm_omni.clients.duplex.DuplexClient`, and the complete wire contract. -The endpoint is served for models that ship a duplex plugin — currently -MiniCPM-o 4.5 only — and just for deploy configurations that declare +The endpoint is served for models that ship a duplex plugin — including +MiniCPM-o 4.5 and Qwen3-Omni — and just for deploy configurations that declare `session_mode: duplex`. PersonaPlex and Nemotron VoiceChat are ported to the plugin contract in follow-up PRs and are not served over this endpoint yet. The runtime architecture is described in [Full-Duplex Runtime (MiniCPM-o 4.5)](../design/fullduplex.md). +## Qwen3-Omni conversation history + +Qwen3-Omni keeps incoming conversation items independently addressable by item +ID. Before applying the model's chat template, it merges consecutive user +items into one multimodal user message. An assistant message, including an +empty message reserved for an interrupted or unheard answer, separates turns. +The final user message places media before text instructions, following the +[official Qwen3-Omni demo's `format_history()`](https://github.com/QwenLM/Qwen3-Omni/blob/e4235853125589c789f06a2dd83e9f4126df5e9d/web_demo.py#L84-L179). + +A camera image or text item may arrive after an audio commit and before +`response.create`. Those items belong to the same pending user turn: the +committed audio remains associated with its history item even when that item +is no longer last. Prompt preparation snapshots the selected inputs; later +items do not modify an already prepared request. + +The duplex adapter retains its existing budgets: at most four audio payloads +(including the current input), an 8 MiB encoded-audio history budget (keeping +at least the newest payload), and at most eight prompt images. The recent +16-message window is rounded outwards to a complete turn. If any audio in an +older turn is no longer retained, the entire turn's model context, including +its images/text and assistant replies, is omitted. Images from that turn are +therefore **not** permanent session-wide model context. These limits differ +from the demo's one image-bearing turn and five audio-bearing turns; the +shared rule is grouping before pruning, rather than copying its UI budgets. + +Grouping and pruning affect only the model-input view. They do not merge or +delete the source conversation items exposed to clients, and do not change +playback acknowledgement or interruption handling. + ## Quick Start ### Start the Server diff --git a/examples/online_serving/realtime_web/app/static/app.js b/examples/online_serving/realtime_web/app/static/app.js index 4d9215ba0..1dd01d4b3 100644 --- a/examples/online_serving/realtime_web/app/static/app.js +++ b/examples/online_serving/realtime_web/app/static/app.js @@ -79,6 +79,13 @@ let echoTimer = null; let sessionGeneration = 0; let audioChain = Promise.resolve(); + // Retain interrupted IDs until startSession() resets the session. Queued + // events and pending audio decodes can outlive response.done, so removing + // IDs there could resume interrupted playback. This set grows with the + // interruptions in a call; sessionGeneration rejects old-session decodes + // after the set is cleared for a new session. + const interruptedResponses = new Set(); + const pendingEvents = new Set(); let assistantTextChannel = null; let connectionReady = false; let turnCounter = 0; @@ -389,8 +396,9 @@ currentResponseId = action.responseId || currentResponseId || `turn-${turnCounter}`; setModel('Speaking'); const generation = sessionGeneration; + const responseId = currentResponseId; const decoded = await decodeAudioDelta(action.event); - if (generation === sessionGeneration) feedPlayback(decoded, currentResponseId); + if (generation === sessionGeneration && !interruptedResponses.has(responseId)) feedPlayback(decoded, responseId); } function handleTranscriptEvent(action) { @@ -410,19 +418,32 @@ } else finishTranscript(action.role, action.text); } - async function handleEvent(event) { + async function handleEvent(event, action = profile.mapEvent(event)) { appendEventLog(event); - const action = profile.mapEvent(event); - const responseId = action.responseId; + const responseId = action.responseId || responseIdOf(event); + if (responseId && interruptedResponses.has(responseId) && action.kind !== 'interrupt') return; switch (action.kind) { case 'connected': setConnection('Connected', 'online'); if (!assistantActive) setModel(profile.waiting); break; - case 'interrupt': + case 'interrupt': { + if (responseId) interruptedResponses.add(responseId); + // A late cancellation belongs to its own response, not the newer one + // currently playing. Speech-start events interrupt the local playback. + if (responseId && responseId !== currentResponseId) break; + if (currentResponseId) interruptedResponses.add(currentResponseId); sessionGeneration += 1; - // A cancelled response never reaches 'done', so nothing else would - // disarm the turn watchdog it left armed. + audioChain = Promise.resolve(); + // Move queued events off the interrupted decode, preserving any newer + // response already received. Its chunks must not be lost with the old one. + const queued = [...pendingEvents]; + pendingEvents.clear(); + for (const pending of queued) enqueueEvent(pending.event, pending.action, pending.source); + clearTimeout(echoTimer); + echoTimer = null; + // Local playback completes immediately on interruption, including + // when the server already completed generation. clearTimeout(turnTimeout); if (playbackNode) playbackNode.port.postMessage({ type: 'clear' }); responseComplete = true; @@ -435,6 +456,7 @@ setPlayback('Idle'); setModel(profile.waiting); break; + } case 'listen': assistantActive = false; setModel(profile.waiting); @@ -538,6 +560,18 @@ await captureContext.resume(); } + function enqueueEvent(event, action, source) { + const pending = { event, action, source }; + pendingEvents.add(pending); + const generation = sessionGeneration; + audioChain = audioChain.then(() => { + if (!pendingEvents.delete(pending)) return; + if (socket === source || action.kind === 'closed') return handleEvent(event, action); + }).catch((error) => { + if (generation === sessionGeneration) return failSession(`Server event failed: ${error.message}`); + }); + } + function openSocket() { return new Promise((resolve, reject) => { const url = realtimeUrl(); @@ -591,11 +625,13 @@ runtimeDetail.textContent = `${captureRate} Hz capture / ${playbackRate} Hz playback`; resolve(); } - // Serialize decoding with terminal events: a drain must never overtake - // an asynchronously decoded WAV chunk. - audioChain = audioChain.then(() => { - if (socket === current || action.kind === 'closed') return handleEvent(event); - }).catch((error) => failSession(`Server event failed: ${error.message}`)); + // Stop the speaker immediately, even while an old WAV is decoding. + // Ordinary completion still waits for its preceding audio chunks. + if (action.kind === 'interrupt') { + void handleEvent(event, action); + return; + } + enqueueEvent(event, action, current); }; current.onerror = () => rejectOnce(`WebSocket connection failed. ${profile.connectionHint}`); current.onclose = (event) => { @@ -628,6 +664,8 @@ try { sessionGeneration += 1; audioChain = Promise.resolve(); + interruptedResponses.clear(); + pendingEvents.clear(); await openPlayback(); await openCapture(); await openSocket(); @@ -731,6 +769,7 @@ async function cleanupSession({ terminal = true } = {}) { sessionGeneration += 1; + pendingEvents.clear(); connectionReady = false; clearTimeout(echoTimer); clearTimeout(turnTimeout); diff --git a/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js b/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js index d6df32edd..5c4fa9d6f 100644 --- a/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js +++ b/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js @@ -3,6 +3,7 @@ const profiles = root.OmniRealtimeProfiles; profiles['qwen3-turn'] = (config) => { const vad = config.adapter === 'vad'; + let interruptOnSpeech = vad; let imageSeq = 0; let imageItems = []; // The engine refuses a ninth image (or 4 MiB of them) for the whole @@ -31,6 +32,7 @@ }, initialMessages(config, instructions) { imageItems = []; + interruptOnSpeech = vad; if (!vad) return [ { type: 'session.update', model: config.model }, { type: 'input_audio_buffer.commit', final: false }, @@ -70,6 +72,20 @@ ack: (responseId, playedMs) => vad ? { type: 'playback.ack', response_id: responseId, item_id: `item_${responseId}`, played_ms: playedMs, committed_ms: playedMs } : null, mapEvent(event) { + if (event.type === 'session.created' || event.type === 'session.updated') { + const input = event.session?.audio?.input; + const turnDetection = input && 'turn_detection' in input + ? input.turn_detection : event.session?.turn_detection; + if (turnDetection !== undefined) { + interruptOnSpeech = vad && turnDetection?.type === 'server_vad' + && turnDetection.interrupt_response !== false; + } + } + // Generation can finish long before the speaker drains its queue. In + // that case the server has no active response left to cancel for us. + if (interruptOnSpeech && event.type === 'input_audio_buffer.speech_started') { + return { kind: 'interrupt' }; + } if (event.type === 'playback.acknowledged') return { kind: 'ack', committedMs: (event.event || event).committed_ms || 0 }; if (event.type === 'response.output_text.delta' || event.type === 'transcription.delta') { // On the shipped STT route transcription.* is model-generated text, @@ -81,7 +97,7 @@ } if (vad && (event.type === 'output_audio_buffer.cleared' || (event.type === 'response.done' && event.response?.status === 'cancelled'))) { - return { kind: 'interrupt' }; + return { kind: 'interrupt', responseId: event.response_id || event.response?.id || null }; } if (event.type === 'input_audio_buffer.committed') return { kind: 'begin' }; if (event.type === 'input_audio_buffer.cleared') return { kind: 'backpressure', message: 'Input cleared. Model is answering; please wait.' }; diff --git a/tests/engine/duplex/test_duplex_events.py b/tests/engine/duplex/test_duplex_events.py index 11c822869..6330059a1 100644 --- a/tests/engine/duplex/test_duplex_events.py +++ b/tests/engine/duplex/test_duplex_events.py @@ -11,7 +11,7 @@ import inspect import pytest from vllm_omni.engine.duplex import events as events_module -from vllm_omni.engine.duplex.commands import TruncateItem +from vllm_omni.engine.duplex.commands import ClearOutputAudio, TruncateItem from vllm_omni.engine.duplex.events import ( REALTIME_ERROR_TYPES_BY_CODE, AudioDelta, @@ -41,6 +41,7 @@ from vllm_omni.engine.duplex.events import ( from vllm_omni.engine.duplex.realtime_events import ( RealtimeProjectionState, project_internal_event, + resolve_clear_output_audio, resolve_truncate_item, retrieve_item_events, ) @@ -376,6 +377,72 @@ def test_projection_of_output_audio_buffer_clear_emits_cleared_before_terminals( assert state.item_truncation_cursors["item_resp_1"] == (0, 250) +@pytest.mark.parametrize("reason", ["turn_detected", "barge_in", "output_audio_buffer_clear"]) +@pytest.mark.parametrize("committed_ms", [1000, 10000]) +def test_completed_response_cancellation_only_clears_pending_playback(reason, committed_ms): + state = RealtimeProjectionState(session_id="duplex-playback") + project_internal_event(state, {"type": "response.created", "response_id": "resp_1"}) + project_internal_event(state, {"type": "response.done", "response_id": "resp_1"}) + + cancelled = project_internal_event( + state, + { + "type": "audio.cancelled", + "response_id": "resp_1", + "reason": reason, + "committed_ms": committed_ms, + "playback": {"sent_ms": 10000, "committed_ms": committed_ms}, + }, + ) + + if reason == "output_audio_buffer_clear" or committed_ms < 10000: + assert _types(cancelled) == ["output_audio_buffer.cleared"] + assert cancelled[0].response_id == "resp_1" + else: + assert cancelled == [] + + +@pytest.mark.parametrize("explicit_response_id", [False, True]) +def test_clear_latest_completed_response_reaches_playback_owner(explicit_response_id): + state = RealtimeProjectionState(session_id="duplex-clear") + project_internal_event(state, {"type": "response.created", "response_id": "resp_1"}) + project_internal_event(state, {"type": "response.done", "response_id": "resp_1"}) + + control = resolve_clear_output_audio( + state, ClearOutputAudio(response_id="resp_1" if explicit_response_id else None, event_id="clear_1") + ) + + # Generation completion does not prove the client has drained its audio. + # A cleared event must follow the engine clearing its playback cursor. + assert control.events == [] + assert control.payloads == [ + { + "type": "output_audio_buffer.clear", + "reason": "output_audio_buffer.clear", + "response_id": "resp_1", + "realtime_event_id": "clear_1", + } + ] + + +@pytest.mark.parametrize("new_response_finished", [False, True]) +def test_late_clear_of_completed_response_does_not_touch_new_response(new_response_finished): + state = RealtimeProjectionState(session_id="duplex-clear") + project_internal_event(state, {"type": "response.created", "response_id": "resp_1"}) + project_internal_event(state, {"type": "response.done", "response_id": "resp_1"}) + project_internal_event(state, {"type": "response.created", "response_id": "resp_2"}) + if new_response_finished: + project_internal_event(state, {"type": "response.done", "response_id": "resp_2"}) + + control = resolve_clear_output_audio(state, ClearOutputAudio(response_id="resp_1")) + + assert control.payloads == [] + assert _types(control.events) == ["output_audio_buffer.cleared"] + assert control.events[0].response_id == "resp_1" + assert state.active_response_id == (None if new_response_finished else "resp_2") + assert state.last_response_id == "resp_2" + + def test_one_truncate_command_truncates_retrieved_transcript_once(): state = RealtimeProjectionState(session_id="duplex-truncate") state.conversation_items["item_resp_1"] = { diff --git a/tests/engine/duplex/test_engine_session.py b/tests/engine/duplex/test_engine_session.py index c90291a81..5ef4ab6b9 100644 --- a/tests/engine/duplex/test_engine_session.py +++ b/tests/engine/duplex/test_engine_session.py @@ -21,6 +21,7 @@ from vllm_omni.engine.duplex.config import ( from vllm_omni.engine.duplex.contracts import DuplexFence from vllm_omni.engine.duplex.events import TurnEvent from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession, DuplexFenceMismatchError +from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack from vllm_omni.model_executor.models.minicpmo_4_5.duplex.capabilities import ( minicpmo45_native_capabilities, ) @@ -241,6 +242,83 @@ def test_history_commit_with_ack_only_policy_defers_unacknowledged_text(): assert f"item_{response_id}" in session.pending_history_item_ids +@pytest.mark.parametrize("audio_complete", [False, True]) +def test_unaligned_response_keeps_empty_turn_before_later_user_input(audio_complete): + session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only")) + first = session.commit_audio_input(transcript="recite a poem") + response_id = session.begin_response() + session.reserve_history_item(f"item_{response_id}") + session.append_assistant_text("An answer the user has only partly heard") + session.mark_audio_sent( + duration_ms=10_000, + text_requires_complete_audio=True, + audio_complete=audio_complete, + ) + + assert session.end_response(commit_text=True) is None + session.barge_in() + session.clear_playback_cursor() + second = session.commit_audio_input(transcript="stop") + empty_answer = {"role": "assistant", "content": ""} + assert session.history == (first.message, empty_answer, second.message) + + ack = apply_playback_ack(session, {"response_id": response_id, "played_ms": 6_981}) + assert ack[0].to_realtime()["event"]["history_committed"] is False + assert session.history == (first.message, empty_answer, second.message) + + if audio_complete: + ack = apply_playback_ack(session, {"response_id": response_id, "played_ms": 10_000}) + assert ack[0].to_realtime()["event"]["history_committed"] is True + assert session.history == ( + first.message, + {"role": "assistant", "content": "An answer the user has only partly heard"}, + second.message, + ) + + +@pytest.mark.parametrize(("reserve_slot", "assistant_text"), [(True, ""), (False, "unplayed answer")]) +def test_unaligned_response_only_materializes_an_existing_answer_slot(reserve_slot, assistant_text): + session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only")) + first = session.commit_audio_input(transcript="first input") + response_id = session.begin_response() + if reserve_slot: + session.reserve_history_item(f"item_{response_id}") + session.append_assistant_text(assistant_text) + session.mark_audio_sent(duration_ms=10_000, text_requires_complete_audio=True, audio_complete=True) + + assert session.end_response(commit_text=reserve_slot) is None + assert session.history == (first.message,) + + +@pytest.mark.parametrize("operation", ["delete", "truncate"]) +def test_removing_one_empty_assistant_preserves_the_other_turn_boundary(operation): + session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only")) + user_inputs = [] + response_ids = [] + for transcript in ("first input", "second input"): + user_inputs.append(session.commit_audio_input(transcript=transcript).message) + response_id = session.begin_response() + response_ids.append(response_id) + session.reserve_history_item(f"item_{response_id}") + session.append_assistant_text("unplayed answer") + session.mark_audio_sent(duration_ms=10_000, text_requires_complete_audio=True, audio_complete=True) + session.end_response(commit_text=True) + + item_id = f"item_{response_ids[1]}" + if operation == "delete": + assert session.delete_history_item(item_id) is True + else: + # A full ACK releases the pending snapshot. Truncation then edits the + # stored message into another empty dict before removing that item. + ack = apply_playback_ack(session, {"response_id": response_ids[1], "played_ms": 10_000}) + assert ack[0].to_realtime()["event"]["history_committed"] is True + assert session.truncate_history_item(item_id, audio_end_ms=0, hard=True) is True + assert session.history == (user_inputs[0], {"role": "assistant", "content": ""}, user_inputs[1]) + ack = apply_playback_ack(session, {"response_id": response_ids[1], "played_ms": 10_000}) + assert ack[0].to_realtime()["error"]["code"] == "playback_item_not_found" + assert session.history == (user_inputs[0], {"role": "assistant", "content": ""}, user_inputs[1]) + + def test_truncate_history_item_uses_response_alignment_marks(): session = _session() response_id = session.begin_response() diff --git a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py index d0476b56c..7aaa975b5 100644 --- a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py +++ b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py @@ -18,20 +18,26 @@ import io import json import os import wave +from contextlib import asynccontextmanager, suppress +from pathlib import Path +from typing import Any, NamedTuple +import numpy as np import pytest import websockets +import yaml -from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import validated_input_wav +from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import validated_input_wav, validated_soft_interrupt_wav from tests.helpers.mark import hardware_test from tests.helpers.media import ( convert_audio_bytes_to_text, cosine_similarity_text, generate_synthetic_audio, ) -from tests.helpers.runtime import OmniServerParams +from tests.helpers.runtime import OmniServer, OmniServerParams, get_model_prefix from tests.helpers.stage_config import get_deploy_config_path, modify_stage_config from vllm_omni.config.stage_config import load_deploy_config +from vllm_omni.engine.duplex.contracts import duplex_resource_request_belongs_to_session from vllm_omni.engine.duplex.vad import ( SILERO_VAD_FILENAME, SILERO_VAD_REPO_ID, @@ -361,6 +367,15 @@ def _assert_realtime_accuracy( class TestQwen3OmniRealtimeWebSocket: + @pytest.fixture(scope="class") + def omni_server(self, request, run_level): + # Release the legacy turn deployment before the duplex class below + # starts another model server on the same pair of GPUs. + from tests.helpers.fixtures.runtime import omni_fixture_lock + from tests.helpers.runtime import iter_omni_server + + yield from iter_omni_server(request, run_level, omni_fixture_lock) + @pytest.mark.advanced_model @pytest.mark.omni @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) @@ -518,3 +533,394 @@ class TestQwen3OmniRealtimeWebSocket: _assert_realtime_smoke(result) _assert_realtime_accuracy(result) + + +# These regressions use the engine-owned Qwen duplex plugin. The older class +# above intentionally exercises the separate turn/STT deployment. +_DUPLEX_REPEATS = int(os.environ.get("VLLM_OMNI_TEST_QWEN_DUPLEX_REPEATS", "20")) +_DUPLEX_LONG_INSTRUCTIONS = "Answer the user's question in English using at least eight complete sentences." +_DUPLEX_SHORT_INSTRUCTIONS = "Answer the user's question in one short English sentence." + + +class _DuplexTestRuntime(NamedTuple): + server: OmniServer + processor: Any + artifacts: Path + + +@pytest.fixture(scope="class") +def qwen_duplex_server(tmp_path_factory, cached_silero_vad_artifact, run_level): + """Load real weights once for all interruption repetitions and history checks.""" + from transformers import AutoProcessor + + assert run_level in {"advanced_model", "full_model"}, "Duplex playback acceptance requires real weights" + artifacts = tmp_path_factory.mktemp("qwen_duplex_playback") + deploy = artifacts / "deploy.yaml" + # An absolute base keeps the production YAML's relative inheritance intact + # when the test's Silero override lives in a temporary directory. + deploy.write_text( + yaml.safe_dump( + { + "base_config": get_deploy_config_path("qwen3_omni_duplex.yaml"), + "duplex_session": {"server_vad_model_path": cached_silero_vad_artifact}, + } + ) + ) + assert load_deploy_config(deploy).session_mode == "duplex" + model = get_model_prefix() + MODEL + processor = AutoProcessor.from_pretrained(model, trust_remote_code=True) + with OmniServer( + model, + ["--deploy-config", str(deploy), "--init-timeout", "900", "--stage-init-timeout", "600"], + env_dict={"VLLM_OMNI_QWEN_VISUAL_DEBUG_DIR": str(artifacts)}, + ) as server: + yield _DuplexTestRuntime(server=server, processor=processor, artifacts=artifacts) + + +def _duplex_fixture_pcm(path: Path, *, duration_s: float | None = None) -> bytes: + """Replay checked-in speech at the web client's negotiated 24 kHz PCM16 rate.""" + pcm = np.frombuffer(_pcm16_mono_16k_from_wav_bytes(path.read_bytes()), dtype=" str | None: + return event.get("response_id") or event.get("response", {}).get("id") + + +def _duplex_response_events(events: list[dict], response_id: str, event_type: str) -> list[dict]: + return [event for event in events if event["type"] == event_type and _duplex_response_id(event) == response_id] + + +async def _duplex_receive_until(ws, events: list[dict], predicate, *, timeout_s: float = 180) -> None: + # One deadline covers the whole operation, including an endless stream that + # would otherwise reset a per-recv timeout forever. + async def receive() -> None: + while not predicate(): + raw = await ws.recv() + assert isinstance(raw, str), "Expected JSON Realtime events" + event = json.loads(raw) + events.append(event) + assert event.get("type") != "error", event + + await asyncio.wait_for(receive(), timeout=timeout_s) + + +async def _duplex_send_pcm(ws, pcm: bytes, *, paced: bool = False, trailing_silence: bool = True) -> None: + if trailing_silence: + pcm += bytes(24000 * 2) # One second exceeds the configured 500 ms endpoint. + chunk_bytes = 24000 * 2 // 5 + for offset in range(0, len(pcm), chunk_bytes): + await ws.send( + json.dumps( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(pcm[offset : offset + chunk_bytes]).decode(), + } + ) + ) + if paced: + await asyncio.sleep(0.2) + + +async def _duplex_add_image(ws) -> None: + from PIL import Image + + image = io.BytesIO() + Image.new("RGB", (448, 336), (32, 96, 160)).save(image, format="JPEG") + await ws.send( + json.dumps( + { + "type": "conversation.item.create", + "item": { + "id": "camera_1", + "type": "message", + "role": "user", + "content": [ + { + "type": "input_image", + "image_url": "data:image/jpeg;base64," + base64.b64encode(image.getvalue()).decode(), + } + ], + }, + } + ) + ) + + +@asynccontextmanager +async def _duplex_recorded_session(runtime, log_path: Path, *, instructions: str, vad: bool = True): + events: list[dict] = [] + server = runtime.server + async with websockets.connect( + f"ws://{server.host}:{server.port}/v1/realtime?duplex=1", max_size=64 * 1024 * 1024 + ) as ws: + failed = True + try: + await ws.send( + json.dumps( + { + "type": "session.update", + "session": { + "model": server.model, + "instructions": instructions, + "temperature": 0, + "max_output_tokens": 384 if vad else 64, + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": 24000}, + "turn_detection": { + "type": "server_vad", + "threshold": 0.5, + "prefix_padding_ms": 300, + "silence_duration_ms": 500, + "create_response": True, + "interrupt_response": True, + } + if vad + else None, + }, + "output": {"format": {"type": "audio/pcm", "rate": 24000}}, + }, + }, + } + ) + ) + await _duplex_receive_until(ws, events, lambda: any(e["type"] == "session.updated" for e in events)) + session = next(e["session"] for e in events if e["type"] == "session.updated") + if vad: + accepted = session["audio"]["input"]["turn_detection"] + assert accepted["interrupt_response"] is True + assert accepted["create_response"] is True + await _duplex_add_image(ws) + yield ws, events, session["id"] + failed = False + finally: + # Closing the transport alone leaves a resumable session occupying + # the deployment's single session slot. Always release the session. + try: + await ws.send(json.dumps({"type": "session.close"})) + await _duplex_receive_until( + ws, events, lambda: any(e["type"] == "session.closed" for e in events), timeout_s=15 + ) + except Exception: + if not failed: + raise + finally: + log_path.write_text(json.dumps(events, ensure_ascii=False, indent=2)) + + +def _duplex_assert_prompt( + runtime, + session_id: str, + *, + audio_count: int, + instructions: str, + request_count: int, + assistant_texts: list[str] | None = None, +): + captures = [] + for path in runtime.artifacts.glob("*/input.json"): + capture = json.loads(path.read_text()) + if duplex_resource_request_belongs_to_session(capture["request_id"], session_id): + captures.append((path.stat().st_mtime_ns, capture)) + assert len(captures) == request_count, captures + capture = max(captures, key=lambda item: item[0])[1] + assert capture["audio_count"] == audio_count + # The camera belongs to the first user turn and leaves with its audio. + image_retained = request_count <= 4 + assert len(capture["images"]) == int(image_retained) + messages: list[dict[str, Any]] = [{"role": "system", "content": instructions}] + if assistant_texts is None: + assistant_texts = [""] * (audio_count - 1) + assert len(assistant_texts) == audio_count - 1 + for index in range(audio_count): + if index: + messages.append({"role": "assistant", "content": assistant_texts[index - 1]}) + parts = [{"type": "image"}] if index == 0 and image_retained else [] + parts.append({"type": "audio"}) + messages.append({"role": "user", "content": parts}) + expected = runtime.processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) + assert capture["prompt"] == expected + # Use the checkpoint tokenizer too: a repr(messages) processor stub cannot + # establish that an empty assistant survives as a real turn separator. + tokenizer = runtime.processor.tokenizer + actual_ids = tokenizer.encode(capture["prompt"], add_special_tokens=False) + separator_ids = tokenizer.encode("<|im_start|>assistant\n<|im_end|>\n", add_special_tokens=False) + assert sum( + actual_ids[index : index + len(separator_ids)] == separator_ids for index in range(len(actual_ids)) + ) == assistant_texts.count("") + + +def _duplex_assert_completed_audio(events: list[dict], response_id: str) -> None: + terminal = _duplex_response_events(events, response_id, "response.done") + assert len(terminal) == 1, terminal + assert terminal[0]["response"]["status"] == "completed", terminal + audio = _duplex_response_events(events, response_id, "response.output_audio.delta") + assert sum(len(base64.b64decode(event["delta"])) for event in audio) > 4800 + assert any( + event.get("delta") + for event in events + if _duplex_response_id(event) == response_id + and event["type"] in {"response.output_text.delta", "response.output_audio_transcript.delta"} + ), "The follow-up must produce text as well as audio" + + +async def _run_duplex_playback_interruption(runtime, log_path: Path, *, generation_finished: bool) -> None: + async with _duplex_recorded_session(runtime, log_path, instructions=_DUPLEX_LONG_INSTRUCTIONS) as ( + ws, + events, + session_id, + ): + await _duplex_send_pcm(ws, _duplex_fixture_pcm(validated_input_wav())) + await _duplex_receive_until(ws, events, lambda: any(e["type"] == "response.created" for e in events)) + old_id = next(_duplex_response_id(e) for e in events if e["type"] == "response.created") + assert old_id + await _duplex_receive_until( + ws, + events, + lambda: ( + sum( + len(base64.b64decode(e["delta"])) + for e in _duplex_response_events(events, old_id, "response.output_audio.delta") + ) + > 4800 + ), + ) + if generation_finished: + await _duplex_receive_until(ws, events, lambda: _duplex_response_events(events, old_id, "response.done")) + await ws.send( + json.dumps({"type": "playback.ack", "response_id": old_id, "played_ms": 100, "committed_ms": 100}) + ) + await _duplex_receive_until(ws, events, lambda: any(e["type"] == "playback.acknowledged" for e in events)) + ack = next(e["event"] for e in events if e["type"] == "playback.acknowledged") + assert ack["item_id"] == f"item_{old_id}" + assert ack["committed_ms"] == 100 + assert ack["playback"]["sent_ms"] > 100 + cursor = len(events) + sender = asyncio.create_task( + _duplex_send_pcm(ws, _duplex_fixture_pcm(validated_soft_interrupt_wav(), duration_s=4.7), paced=True) + ) + try: + await _duplex_receive_until( + ws, events, lambda: any(e["type"] == "input_audio_buffer.speech_started" for e in events[cursor:]) + ) + speech_index = next( + i for i in range(cursor, len(events)) if events[i]["type"] == "input_audio_buffer.speech_started" + ) + if not generation_finished: + assert not _duplex_response_events(events[:speech_index], old_id, "response.done"), ( + "Generation completed before speech interruption; the in-progress scenario was not exercised" + ) + await _duplex_receive_until( + ws, + events, + lambda: any(e["type"] == "response.done" and _duplex_response_id(e) != old_id for e in events[cursor:]), + ) + await sender + finally: + sender.cancel() + with suppress(asyncio.CancelledError): + await sender + new_id = next( + _duplex_response_id(e) + for e in events[cursor:] + if e["type"] == "response.created" and _duplex_response_id(e) != old_id + ) + # Include everything observed through session.closed when checking duplicates. + terminals = _duplex_response_events(events, old_id, "response.done") + assert len(terminals) == 1, terminals + assert terminals[0]["response"]["status"] == ("completed" if generation_finished else "cancelled") + if generation_finished: + cleared = _duplex_response_events(events, old_id, "output_audio_buffer.cleared") + assert len(cleared) == 1, cleared + boundary = events.index(cleared[0]) + else: + boundary = events.index(terminals[0]) + assert not _duplex_response_events(events[boundary + 1 :], old_id, "response.output_audio.delta") + assert new_id and new_id != old_id + _duplex_assert_completed_audio(events, new_id) + _duplex_assert_prompt(runtime, session_id, audio_count=2, instructions=_DUPLEX_LONG_INSTRUCTIONS, request_count=2) + + +async def _run_duplex_history_pruning(runtime, log_path: Path) -> None: + async with _duplex_recorded_session(runtime, log_path, instructions=_DUPLEX_SHORT_INSTRUCTIONS, vad=False) as ( + ws, + events, + session_id, + ): + response_ids = [] + heard_texts: list[str] = [] + pcm = _duplex_fixture_pcm(validated_input_wav()) + for turn in range(6): + cursor = len(events) + await _duplex_send_pcm(ws, pcm, trailing_silence=False) + await ws.send(json.dumps({"type": "input_audio_buffer.commit"})) + await ws.send(json.dumps({"type": "response.create"})) + await _duplex_receive_until(ws, events, lambda: any(e["type"] == "response.done" for e in events[cursor:])) + response_id = next(_duplex_response_id(e) for e in events[cursor:] if e["type"] == "response.done") + assert response_id and response_id not in response_ids + response_ids.append(response_id) + _duplex_assert_completed_audio(events, response_id) + _duplex_assert_prompt( + runtime, + session_id, + audio_count=min(turn + 1, 4), + instructions=_DUPLEX_SHORT_INSTRUCTIONS, + request_count=turn + 1, + assistant_texts=heard_texts[-3:], + ) + # Complete playback so pruning is checked against real assistant + # answers paired with retained user audio. Repeated empty assistant + # turns can make Qwen emit EOS; interruption tests cover those slots. + done = _duplex_response_events(events, response_id, "response.done")[0]["response"] + played_ms = done["metadata"]["playback"]["sent_ms"] + heard_texts.append( + "".join( + e["delta"] + for e in _duplex_response_events(events, response_id, "response.output_audio_transcript.delta") + ).strip() + ) + assert heard_texts[-1] + cursor = len(events) + await ws.send( + json.dumps( + { + "type": "playback.ack", + "response_id": response_id, + "played_ms": played_ms, + "committed_ms": played_ms, + } + ) + ) + await _duplex_receive_until( + ws, events, lambda: any(e["type"] == "playback.acknowledged" for e in events[cursor:]) + ) + ack = next(e["event"] for e in events[cursor:] if e["type"] == "playback.acknowledged") + assert ack["item_id"] == f"item_{response_id}" + assert ack["committed_ms"] == ack["playback"]["sent_ms"] == played_ms + assert ack["history_committed"] is True + for response_id in response_ids: + _duplex_assert_completed_audio(events, response_id) + + +@pytest.mark.advanced_model +@pytest.mark.omni +@hardware_test(res={"cuda": "H100"}, num_cards=2) +class TestQwen3OmniDuplexPlayback: + @pytest.mark.parametrize("generation_finished", [False, True], ids=["generating", "queued-playback"]) + @pytest.mark.parametrize("repetition", range(_DUPLEX_REPEATS)) + def test_speech_interrupts_playback( + self, qwen_duplex_server, tmp_path: Path, generation_finished: bool, repetition: int + ): + """Reuse one live deployment; every repetition opens an independent call.""" + asyncio.run( + _run_duplex_playback_interruption( + qwen_duplex_server, tmp_path / f"events-{repetition}.json", generation_finished=generation_finished + ) + ) + + def test_mixed_image_audio_history_pruning(self, qwen_duplex_server, tmp_path: Path): + asyncio.run(_run_duplex_history_pruning(qwen_duplex_server, tmp_path / "events.json")) diff --git a/tests/examples/test_realtime_web_profiles.cjs b/tests/examples/test_realtime_web_profiles.cjs index 51d4b3972..1cf32e600 100644 --- a/tests/examples/test_realtime_web_profiles.cjs +++ b/tests/examples/test_realtime_web_profiles.cjs @@ -112,11 +112,28 @@ function shell(profileName, adapter = 'stt', options = {}) { constructor(options) { this.sampleRate = options.sampleRate; this.audioWorklet = { addModule: async () => {} }; this.destination = {}; } createMediaStreamSource() { return { connect() {} }; } createGain() { return { gain: {}, connect() {} }; } + decodeAudioData(bytes) { return options.decodeAudioData(bytes); } async resume() {} async close() {} } class AudioWorkletNode { - constructor(_context, name) { this.name = name; this.sent = []; nodes.push(this); this.port = { postMessage: (message) => this.sent.push(message) }; } + constructor(_context, name) { + this.name = name; this.sent = []; nodes.push(this); + this.port = { postMessage: (message) => { this.sent.push(message); this.player?.handleMessage(message); } }; + if (options.realPlayback && name === 'fullduplex-pcm-playback') { + let Playback; + const node = this; + const worker = vm.createContext({ + sampleRate: 24000, + AudioWorkletProcessor: class { + constructor() { this.port = { postMessage: message => queueMicrotask(() => node.port.onmessage?.({ data: message })) }; } + }, + registerProcessor: (_name, cls) => { Playback = cls; }, + }); + vm.runInContext(fs.readFileSync(path.join(root, 'playback_worklet.js'), 'utf8'), worker); + this.player = new Playback(); + } + } connect(target) { return target; } } const ctx = vm.createContext({ @@ -138,7 +155,7 @@ function shell(profileName, adapter = 'stt', options = {}) { // Expose closure controls only in the test VM; production has no test API. const source = fs.readFileSync(path.join(root, 'app.js'), 'utf8').replace(/\}\)\(\);\s*$/, ` globalThis.testUI = { startSession, stopSession, handleEvent, playbackDrained, - microphoneUploadEnabled, flushCapture, + microphoneUploadEnabled, flushCapture, awaitQueue: () => audioChain, capture() { pendingCapture.push(new Int16Array([100, 200])); }, state() { return { running, connectionReady, assistantActive }; } }; @@ -410,3 +427,174 @@ test('the camera retires its oldest image instead of exhausting the session budg assert.equal(plain(p.initialMessages(config, 'x')).length > 0, true); assert.equal(plain(p.imageMessages('JPEG')).length, 1); }); + +// Exercise the real WebSocket dispatch queue together with the actual playback +// processor: sending a clear message alone does not prove the speaker is silent. +function receive(app, event) { + app.sockets[0].onmessage({ data: JSON.stringify(event) }); +} + +function audioChunk(responseId, extra = {}) { + const pcm = new Int16Array(24000).fill(12000); + return { type: 'response.output_audio.delta', response_id: responseId, + delta: Buffer.from(pcm.buffer).toString('base64'), format: 'pcm16', sample_rate_hz: 24000, ...extra }; +} + +function renderPlayback(player) { + const output = new Float32Array(128); + player.process([], [[output]]); + return output; +} + +async function playingQwen(options = {}) { + const app = shell('qwen3-turn', 'vad', { realPlayback: true, ...options }); + await app.ui.startSession(); + receive(app, { type: 'response.created', response: { id: 'old' } }); + receive(app, audioChunk('old')); + await app.ui.awaitQueue(); + app.player = app.nodes.find(node => node.name === 'fullduplex-pcm-playback').player; + // Advance through the initial 400 ms playback buffer. + for (let index = 0; index < 75; index++) renderPlayback(app.player); + assert.ok(renderPlayback(app.player).some(sample => sample !== 0)); + return app; +} + +function assertSilent(player) { + assert.equal(player.bufferedFrames(), 0, 'interrupted audio must leave the playback queue'); + assert.ok(renderPlayback(player).every(sample => sample === 0), 'speaker must output silence'); +} + +for (const terminal of [false, true]) { + test(`Qwen speech interrupts actual playback after generation completed=${terminal}`, async () => { + const app = await playingQwen(); + if (terminal) { + receive(app, { type: 'response.output_audio.done', response_id: 'old' }); + receive(app, { type: 'response.done', response: { id: 'old', status: 'completed' } }); + await app.ui.awaitQueue(); + } + receive(app, { type: 'input_audio_buffer.speech_started', item_id: 'next', audio_start_ms: 1000 }); + await app.ui.awaitQueue(); + assertSilent(app.player); + assert.equal(app.ui.microphoneUploadEnabled(), true); + const ack = app.sockets[0].sent.find(event => event.type === 'playback.ack'); + assert.equal(ack.response_id, 'old'); + assert.equal(ack.played_ms, 5, 'ack only the 128 frames actually played, not all queued audio'); + await app.ui.stopSession({ terminal: false }); + }); +} + +test('Qwen cancellation preempts pending decode and does not delay the next response', async () => { + let releaseDecode; + const app = await playingQwen({ decodeAudioData: () => new Promise(resolve => { releaseDecode = resolve; }) }); + receive(app, audioChunk('old', { format: 'wav' })); + await flushTasks(); + assert.equal(typeof releaseDecode, 'function'); + receive(app, { type: 'response.done', response: { id: 'old', status: 'cancelled' } }); + await flushTasks(); + assertSilent(app.player); + receive(app, { type: 'response.created', response: { id: 'new' } }); + receive(app, audioChunk('new')); + await flushTasks(); + assert.equal(app.player.activeResponseId, 'new'); + assert.equal(app.player.bufferedFrames(), 24000); + releaseDecode({ sampleRate: 24000, getChannelData: () => new Float32Array(24000).fill(0.4) }); + await flushTasks(); + assert.equal(app.player.bufferedFrames(), 24000, 'old decoded audio must not enter the new response'); + await app.ui.stopSession({ terminal: false }); +}); + +test('Qwen interruption rejects queued and late audio for the cancelled response', async () => { + const app = await playingQwen(); + receive(app, audioChunk('old')); + receive(app, { type: 'response.done', response: { id: 'old', status: 'cancelled' } }); + receive(app, audioChunk('old')); + await app.ui.awaitQueue(); + assertSilent(app.player); + await app.ui.stopSession({ terminal: false }); +}); + +test('late cancellation of an old Qwen response does not clear a newer response', async () => { + const app = await playingQwen(); + receive(app, { type: 'response.done', response: { id: 'old', status: 'cancelled' } }); + await app.ui.awaitQueue(); + receive(app, { type: 'response.created', response: { id: 'new' } }); + receive(app, audioChunk('new')); + await app.ui.awaitQueue(); + receive(app, { type: 'output_audio_buffer.cleared', response_id: 'old' }); + receive(app, { type: 'response.done', response: { id: 'old', status: 'cancelled' } }); + await app.ui.awaitQueue(); + assert.equal(app.player.activeResponseId, 'new'); + assert.equal(app.player.bufferedFrames(), 24000); + assert.equal(app.ui.state().assistantActive, true); + await app.ui.stopSession({ terminal: false }); +}); + +test('Qwen speech interruption follows accepted turn detection; MiniCPM mapping stays unchanged', () => { + const qwen = profiles['qwen3-turn']({ adapter: 'vad' }); + const speech = { type: 'input_audio_buffer.speech_started', item_id: 'next' }; + assert.equal(qwen.mapEvent(speech).kind, 'interrupt'); + qwen.mapEvent({ type: 'session.updated', session: { audio: { input: { turn_detection: { + type: 'server_vad', interrupt_response: false, + } } } } }); + assert.equal(qwen.mapEvent(speech).kind, 'ignore'); + qwen.mapEvent({ type: 'session.updated', session: { audio: { input: { turn_detection: null } } } }); + assert.equal(qwen.mapEvent(speech).kind, 'ignore'); + qwen.mapEvent({ type: 'session.updated', session: { audio: { input: { turn_detection: { + type: 'server_vad', interrupt_response: true, + } } } } }); + assert.equal(qwen.mapEvent(speech).kind, 'interrupt'); + assert.equal(profiles['qwen3-turn']({ adapter: 'stt' }).mapEvent(speech).kind, 'ignore'); + assert.equal(profiles['minicpm-native']().mapEvent(speech).kind, 'ignore'); +}); + +test('late old-response controls cannot invalidate a newer pending decode', async () => { + let releaseDecode; + const app = await playingQwen({ decodeAudioData: () => new Promise(resolve => { releaseDecode = resolve; }) }); + receive(app, { type: 'input_audio_buffer.speech_started', item_id: 'next', audio_start_ms: 1000 }); + await app.ui.awaitQueue(); + receive(app, { type: 'response.created', response: { id: 'new' } }); + receive(app, audioChunk('new', { format: 'wav' })); + await flushTasks(); + assert.equal(typeof releaseDecode, 'function'); + receive(app, { type: 'output_audio_buffer.cleared', response_id: 'old' }); + receive(app, { type: 'response.done', response: { id: 'old', status: 'cancelled' } }); + releaseDecode({ sampleRate: 24000, getChannelData: () => new Float32Array(24000).fill(0.4) }); + await app.ui.awaitQueue(); + assert.equal(app.player.activeResponseId, 'new'); + assert.equal(app.player.bufferedFrames(), 24000); + assert.equal(app.ui.state().assistantActive, true); + await app.ui.stopSession({ terminal: false }); +}); + +test('normal completion still waits for decoding before draining playback', async () => { + let releaseDecode; + const app = await playingQwen({ decodeAudioData: () => new Promise(resolve => { releaseDecode = resolve; }) }); + receive(app, audioChunk('old', { format: 'wav' })); + await flushTasks(); + receive(app, { type: 'response.output_audio.done', response_id: 'old' }); + receive(app, { type: 'response.done', response: { id: 'old', status: 'completed' } }); + await flushTasks(); + assert.equal(app.player.drain, null); + releaseDecode({ sampleRate: 24000, getChannelData: () => new Float32Array(24000).fill(0.4) }); + await app.ui.awaitQueue(); + assert.equal(app.player.bufferedFrames(), 47872); + assert.equal(app.player.drain.responseId, 'old'); + await app.ui.stopSession({ terminal: false }); +}); + +test('interrupting an old decode preserves a newer response already waiting in the event queue', async () => { + let releaseDecode; + const app = await playingQwen({ decodeAudioData: () => new Promise(resolve => { releaseDecode = resolve; }) }); + receive(app, audioChunk('old', { format: 'wav' })); + await flushTasks(); + receive(app, { type: 'response.created', response: { id: 'new' } }); + receive(app, audioChunk('new')); + receive(app, { type: 'output_audio_buffer.cleared', response_id: 'old' }); + await flushTasks(); + assert.equal(app.player.activeResponseId, 'new'); + assert.equal(app.player.bufferedFrames(), 24000); + releaseDecode({ sampleRate: 24000, getChannelData: () => new Float32Array(24000).fill(0.4) }); + await flushTasks(); + assert.equal(app.player.bufferedFrames(), 24000); + await app.ui.stopSession({ terminal: false }); +}); diff --git a/tests/model_executor/models/qwen3_omni/test_duplex_plugin.py b/tests/model_executor/models/qwen3_omni/test_duplex_plugin.py index c57d5ff14..908ae6cf6 100644 --- a/tests/model_executor/models/qwen3_omni/test_duplex_plugin.py +++ b/tests/model_executor/models/qwen3_omni/test_duplex_plugin.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +import ast import asyncio import base64 from types import SimpleNamespace @@ -17,7 +18,14 @@ from tests.engine.duplex.test_session_runner import ( tts_output, ) from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig -from vllm_omni.engine.duplex.commands import AckPlayback, CancelResponse, Commit, UpdateSession +from vllm_omni.engine.duplex.commands import ( + AckPlayback, + CancelResponse, + ClearOutputAudio, + Commit, + CreateResponse, + UpdateSession, +) from vllm_omni.engine.duplex.config import DuplexSessionConfig from vllm_omni.engine.duplex.messages import OpenDuplexSessionMessage from vllm_omni.engine.duplex.plugin import DuplexRuntimeConfigError @@ -120,6 +128,72 @@ async def test_two_committed_turns_stream_text_and_audio_and_release_requests(): await h.manager.shutdown() +@pytest.mark.asyncio +@pytest.mark.parametrize("played_ms", [0, 500, 1000]) +async def test_clear_completed_qwen_playback_allows_next_manual_turn(played_ms): + h = await open_qwen() + try: + await h.run(UpdateSession(patch={"turn_detection": None})) + assert h.session.config.overlap_policy == "listen_only" + await h.run(append_audio()) + await h.run(Commit(final=True, create_response=False)) + await h.run(CreateResponse()) + old = h.port.submissions[-1].context + response_id = h.session.active_response_id + await h.deliver_and_settle( + tts_output(old.request_id, samples=0, text="first answer", finished=True), stage_id=0 + ) + await h.deliver_and_settle(tts_output(old.request_id, finished=True), stage_id=2) + if played_ms: + await h.run(AckPlayback(response_id=response_id, played_ms=played_ms, committed_ms=played_ms)) + assert h.session.active_response_id is None + assert h.session.playback.sent_ms == 1000 + + events = await h.run(ClearOutputAudio(response_id=response_id)) + cleared = [event for event in events if event.type == "output_audio_buffer.cleared"] + assert len(cleared) == 1 + assert cleared[0].response_id == response_id + assert h.session.playback.sent_ms == h.session.playback.committed_ms == 0 + # Repeated clear after the queue is empty still identifies the old turn. + repeated = await h.run(ClearOutputAudio(response_id=response_id)) + assert [event.response_id for event in repeated if event.type == "output_audio_buffer.cleared"] == [response_id] + + await h.run(append_audio(value=-0.25)) + await h.run(Commit(final=True, create_response=False)) + await h.run(CreateResponse()) + assert len(h.port.submissions) == 2, [event.to_realtime() for event in h.events] + latest = h.port.submissions[-1] + assert latest.context.request_id != old.request_id + assert latest.prompt["prompt"] == repr( + [ + {"role": "user", "content": [{"type": "audio"}]}, + {"role": "assistant", "content": "first answer" if played_ms == 1000 else ""}, + {"role": "user", "content": [{"type": "audio"}]}, + ] + ) + newest_audio, sample_rate = latest.prompt["multi_modal_data"]["audio"][-1] + assert sample_rate == 16000 + np.testing.assert_array_equal(newest_audio, np.full(16000, -0.25, dtype=np.float32)) + + new_response_id = h.session.active_response_id + new_epoch = h.session.epoch + await h.deliver_and_settle(tts_output(latest.context.request_id, finished=False), stage_id=2) + new_playback = h.session.playback.as_dict() + late_clear = await h.run(ClearOutputAudio(response_id=response_id)) + assert [event.response_id for event in late_clear if event.type == "output_audio_buffer.cleared"] == [ + response_id + ] + assert h.session.active_response_id == new_response_id + assert h.session.active_request_id == latest.context.request_id + assert h.session.epoch == new_epoch + assert h.session.playback.as_dict() == new_playback + terminals = [event for event in h.events if event.type == "response.done" and event.response_id == response_id] + assert len(terminals) == 1 and terminals[0].status == "completed" + assert not any(event.to_realtime().get("type") == "error" for event in h.events) + finally: + await h.manager.shutdown() + + @pytest.mark.asyncio async def test_cancel_drops_old_output_and_accepts_next_turn(): h = await open_qwen() @@ -157,6 +231,60 @@ async def test_vad_commits_and_speech_interrupts_active_qwen_response(): await h.manager.shutdown() +@pytest.mark.asyncio +@pytest.mark.parametrize("generation_finished", [False, True]) +async def test_vad_interrupts_queued_qwen_audio_and_keeps_the_new_utterance(generation_finished): + h = await open_qwen() + try: + h.runner.control._detector = _speech_then_stop() + await h.run(append_audio(value=0.125)) + await h.run(append_audio(is_speech=False, value=0.0)) + old = h.port.submissions[-1].context + response_id = h.session.active_response_id + await h.deliver_and_settle( + tts_output(old.request_id, samples=0, text="first answer", finished=True), stage_id=0 + ) + await h.deliver_and_settle(tts_output(old.request_id, samples=240000, finished=generation_finished), stage_id=2) + await h.run(AckPlayback(response_id=response_id, played_ms=1000, committed_ms=1000)) + assert h.session.playback.sent_ms == 10000 + assert h.session.playback.committed_ms == 1000 + if generation_finished: + assert h.session.active_response_id is None + assert h.session.active_request_id is None + + h.runner.control._detector = _speech_then_stop() + events = await h.run(append_audio(value=-0.25)) + if generation_finished: + cleared = [event for event in events if event.type == "output_audio_buffer.cleared"] + assert len(cleared) == 1, [event.to_realtime() for event in events] + assert cleared[0].response_id == response_id + else: + assert any(event.type == "response.done" and event.status == "cancelled" for event in events) + assert h.session.epoch > old.fence.epoch + # Generation already completed is still completed: stopping playback + # must not emit a second, cancelled terminal for the same response. + terminals = [event for event in h.events if event.type == "response.done" and event.response_id == response_id] + assert len(terminals) == 1 + assert terminals[0].status == ("completed" if generation_finished else "cancelled") + + await h.run(append_audio(is_speech=False, value=0.0)) + assert len(h.port.submissions) == 2 + latest = h.port.submissions[-1] + assert latest.context.request_id != old.request_id + assert latest.prompt["prompt"] == repr( + [ + {"role": "user", "content": [{"type": "audio"}]}, + {"role": "assistant", "content": ""}, + {"role": "user", "content": [{"type": "audio"}]}, + ] + ) + newest_audio, sample_rate = latest.prompt["multi_modal_data"]["audio"][-1] + assert sample_rate == 16000 + np.testing.assert_array_equal(newest_audio, np.full(16000, -0.25, dtype=np.float32)) + finally: + await h.manager.shutdown() + + def test_audio_projector_handles_delta_tensor_lists_and_sample_rate(): import torch @@ -180,6 +308,49 @@ def test_audio_projector_handles_delta_tensor_lists_and_sample_rate(): assert events[0]["end_of_turn"] is True +def test_format_history_merges_user_parts_but_preserves_empty_assistant_boundary(): + text = {"type": "text", "text": "first question"} + image = {"type": "image"} + audio = {"type": "audio"} + source = [ + {"role": "system", "content": "instructions"}, + {"role": "user", "content": [image]}, + {"role": "user", "content": "first question"}, + {"role": "user", "content": [audio]}, + {"role": "assistant", "content": ""}, + {"role": "user", "content": "next question"}, + {"role": "user", "content": [image, audio]}, + ] + before = repr(source) + assert Qwen3OmniDuplexPlugin.format_history(source) == [ + source[0], + {"role": "user", "content": [image, text, audio]}, + {"role": "assistant", "content": ""}, + {"role": "user", "content": [image, audio, {"type": "text", "text": "next question"}]}, + ] + assert repr(source) == before + + +def test_audio_byte_budget_evicts_whole_multimodal_turn(): + plugin = Qwen3OmniDuplexPlugin(lambda *args: None) + state = plugin.create_session_state() + first = {"role": "user", "content": [{"type": "audio_url"}]} + image = {"role": "user", "content": [{"type": "image_url"}]} + source = [first, image] + plugin.prepare_prompt_config({"conversation": source}, state=state, payload={"audio": "A" * (5 * 1024 * 1024)}) + source.extend( + [ + {"role": "assistant", "content": "old answer"}, + {"role": "user", "content": [{"type": "audio_url"}]}, + ] + ) + current = {"audio": "B" * (4 * 1024 * 1024)} + config = plugin.prepare_prompt_config({"conversation": source}, state=state, payload=current) + assert config["qwen_messages"] == [{"role": "user", "audio_payload": current}] + assert len(state.audio_history) == 1 + assert source[1] is image + + def test_audio_history_is_bounded_and_does_not_cross_sessions(): plugin = Qwen3OmniDuplexPlugin(lambda *args: None) state, other = plugin.create_session_state(), plugin.create_session_state() @@ -198,6 +369,27 @@ def test_audio_history_is_bounded_and_does_not_cross_sessions(): assert other.audio_history == [] +@pytest.mark.parametrize("assistant_text", ["", "previous answer"]) +def test_dropped_audio_removes_its_image_and_assistant(assistant_text): + plugin = Qwen3OmniDuplexPlugin(lambda *args: None) + state = plugin.create_session_state() + image = {"role": "user", "content": [{"type": "image_url"}]} + history = [image] + for i in range(6): + history.append({"role": "user", "content": [{"type": "audio_url"}]}) + config = plugin.prepare_prompt_config({"conversation": history}, state=state, payload={"audio": str(i)}) + history.append({"role": "assistant", "content": assistant_text}) + assert config["qwen_messages"] == [ + {"role": "user", "audio_payload": {"audio": "2"}}, + {"role": "assistant", "content": assistant_text}, + {"role": "user", "audio_payload": {"audio": "3"}}, + {"role": "assistant", "content": assistant_text}, + {"role": "user", "audio_payload": {"audio": "4"}}, + {"role": "assistant", "content": assistant_text}, + {"role": "user", "audio_payload": {"audio": "5"}}, + ] + + @pytest.mark.asyncio async def test_late_playback_ack_keeps_answer_before_new_user_input(): h = await open_qwen() @@ -208,10 +400,18 @@ async def test_late_playback_ack_keeps_answer_before_new_user_input(): response_id = h.session.active_response_id await h.deliver_and_settle(tts_output(request_id, samples=0, text="first answer", finished=True), stage_id=0) await h.deliver_and_settle(tts_output(request_id, finished=True), stage_id=2) - # Playback has not drained yet. A new utterance commits first. - assert [m["role"] for m in h.session.history] == ["user"] + # The unplayed answer reserves its turn without exposing unheard text. + assert [m["role"] for m in h.session.history] == ["user", "assistant"] + assert h.session.history[1]["content"] == "" await h.run(append_audio()) await h.run(Commit(final=True, create_response=True)) + assert h.port.submissions[-1].prompt["prompt"] == repr( + [ + {"role": "user", "content": [{"type": "audio"}]}, + {"role": "assistant", "content": ""}, + {"role": "user", "content": [{"type": "audio"}]}, + ] + ) current_response = h.session.active_response_id events = await h.run(AckPlayback(response_id=response_id, played_ms=1000)) assert not any(e.to_realtime().get("type") == "error" for e in events) @@ -286,6 +486,45 @@ def test_current_audio_is_not_replaced_by_history_ending_with_assistant(): assert config["qwen_messages"][-1]["audio_payload"] is current +@pytest.mark.parametrize("current_in_history", [False, True]) +def test_empty_assistant_keeps_turn_boundary_in_audio_prompt(current_in_history): + plugin = Qwen3OmniDuplexPlugin(lambda *args: None) + plugin.processor = SimpleNamespace(apply_chat_template=lambda messages, **kwargs: repr(messages)) + state = plugin.create_session_state() + first = {"audio": base64.b64encode(np.full(160, 0.1, dtype=" list[D if not isinstance(response_id, str) or not response_id: return events if response_is_done(state, response_id): + # Generation may finish long before the client drains its audio. + # A later barge-in still needs to stop that queued playback, but + # must not produce a second response.done for the completed turn. + playback = event.get("playback") + if ( + not events + and isinstance(playback, Mapping) + and _int_or(playback.get("sent_ms")) > _int_or(playback.get("committed_ms")) + ): + events.append(OutputAudioCleared(response_id=response_id)) return events committed_ms = event.get("committed_ms") if isinstance(committed_ms, int | float): @@ -1320,7 +1330,11 @@ def resolve_cancel_response(state: RealtimeProjectionState, command: CancelRespo def resolve_clear_output_audio(state: RealtimeProjectionState, command: ClearOutputAudio) -> ResolvedControl: payload: dict[str, object] = {"type": "output_audio_buffer.clear", "reason": "output_audio_buffer.clear"} response_id = command.response_id or state.active_response_id or state.last_response_id - if response_is_done(state, response_id): + # A completed response can still own queued playback. Clear the latest + # response in the engine, while keeping late clears away from newer turns. + if response_is_done(state, response_id) and ( + state.active_response_id is not None or response_id != state.last_response_id + ): return ResolvedControl(payloads=[], events=[OutputAudioCleared(response_id=_str_or_none(response_id))]) if isinstance(response_id, str) and response_id: payload["response_id"] = response_id diff --git a/vllm_omni/engine/duplex/session/engine_session.py b/vllm_omni/engine/duplex/session/engine_session.py index f64ad1f93..f181aaaea 100644 --- a/vllm_omni/engine/duplex/session/engine_session.py +++ b/vllm_omni/engine/duplex/session/engine_session.py @@ -1029,6 +1029,16 @@ class DuplexEngineSession: last_message, audio_text_marks, _ = response_snapshot self._conversation.pending_item_ids[item_id] = copy.deepcopy(last_message) self._conversation.pending_item_input_commit_seqs[item_id] = response_snapshot[2] + playback = self._playback_cursor_for_item_id(item_id) + if ( + item_id in self._conversation.history_item_placeholders + and playback is not None + and playback.text_requires_complete_audio + ): + # Keep the assistant turn between its user inputs even when + # unaligned audio cannot establish any heard text. A later + # full playback ACK fills this same slot from the snapshot. + self._store_history_item_message(item_id, _object_dict(role="assistant", content="")) if audio_text_marks: self._conversation.pending_item_audio_text_marks[item_id] = list(copy.deepcopy(audio_text_marks)) pending_audio_ms = self._conversation.pending_truncations_ms.get(item_id) @@ -1073,10 +1083,10 @@ class DuplexEngineSession: self._conversation.assistant_response_snapshots.pop(response_id, None) if message is None: return pending is not None or removed_placeholder - try: - self._conversation.messages.remove(message) - except ValueError: - pass + # Distinct response items can have identical (including empty) content. + self._conversation.messages = [ + candidate for candidate in self._conversation.messages if candidate is not message + ] return True def truncate_history_item( @@ -1167,10 +1177,9 @@ class DuplexEngineSession: if changed and self._message_text_len(message) <= 0: self._conversation.item_ids.pop(item_id, None) self._conversation.item_audio_text_marks.pop(item_id, None) - try: - self._conversation.messages.remove(message) - except ValueError: - pass + self._conversation.messages = [ + candidate for candidate in self._conversation.messages if candidate is not message + ] if item_id.startswith("item_"): self._conversation.assistant_response_snapshots.pop(item_id.removeprefix("item_"), None) elif changed and item_id.startswith("item_"): diff --git a/vllm_omni/engine/duplex/session/runner.py b/vllm_omni/engine/duplex/session/runner.py index 24e99d28a..1ca248fef 100644 --- a/vllm_omni/engine/duplex/session/runner.py +++ b/vllm_omni/engine/duplex/session/runner.py @@ -1313,7 +1313,7 @@ class DuplexSessionRunner: { "type": "audio.cancelled", "session_id": session.session_id, - "response_id": session.active_response_id, + "response_id": event.get("response_id") or session.active_response_id, "reason": cancel_reason, "cancelled_epoch": session.epoch, "epoch": session.epoch, diff --git a/vllm_omni/model_executor/models/qwen3_omni/duplex/plugin.py b/vllm_omni/model_executor/models/qwen3_omni/duplex/plugin.py index 90cb41f8c..89b2c6ffd 100644 --- a/vllm_omni/model_executor/models/qwen3_omni/duplex/plugin.py +++ b/vllm_omni/model_executor/models/qwen3_omni/duplex/plugin.py @@ -213,16 +213,27 @@ class Qwen3OmniDuplexPlugin(DuplexModelPlugin): # The history accessor copies message dicts but preserves content identity. # Keep payloads privately, never put PCM into public conversation events. has_audio = bool(payload.get("audio")) - last_content = history[-1].get("content") if history else None - current_is_in_history = ( - has_audio - and isinstance(last_content, list) - and any(part.get("type") == "audio_url" for part in last_content) - ) + # A committed audio item need not be last: camera/text items can arrive + # before response.create. Search only the unanswered user turn so a new + # payload can never overwrite an earlier, already answered utterance. + current_contents = [] + if has_audio: + for message in reversed(history): + if message.get("role") != "user": + break + content = message.get("content") + if isinstance(content, list) and any(part.get("type") == "audio_url" for part in content): + current_contents.append(content) + current_is_in_history = bool(current_contents) if current_is_in_history: - content = history[-1].get("content") - state.audio_history = [(key, value) for key, value in state.audio_history if key is not content] - state.audio_history.append((content, payload)) + # The runner concatenates multiple commits made before a response. + # Associate that single payload with all of its source audio items. + state.audio_history = [ + (keys, value) + for keys, value in state.audio_history + if not any(key is content for key in keys for content in current_contents) + ] + state.audio_history.append((tuple(current_contents), payload)) # Reserve one of the four audio slots for the actual submitted input, # including when deferred input has no trailing user history entry. history_limit = 4 if current_is_in_history or not has_audio else 3 @@ -230,33 +241,75 @@ class Qwen3OmniDuplexPlugin(DuplexModelPlugin): while len(retained) > 1 and sum(len(value.get("audio", "")) for _, value in retained) > 8 * 1024 * 1024: retained.pop(0) state.audio_history = retained + # Match the official demo's unit of history: consecutive user items + # followed by their assistant reply(s). Prune whole turns, before + # formatting, so a camera item cannot rescue an orphaned answer. + turns = [] + for index, message in enumerate(history): + role = message.get("role") + if not turns or role == "system" or (role == "user" and turns[-1][1][-1].get("role") != "user"): + turns.append((index, [])) + turns[-1][1].append(message) + messages = [] - - def has_image(message): - content = message.get("content") - return isinstance(content, list) and any(p.get("type") == "image_url" for p in content) - - selected = [m for m in history[:-16] if has_image(m)] + history[-16:] - for message in selected: - content = message.get("content") - audio = next((value for key, value in retained if key is content), None) - if audio is not None: - messages.append({"role": "user", "audio_payload": audio}) - elif isinstance(content, list) and any(part.get("type") == "image_url" for part in content): - messages.append(message) - elif isinstance(content, str) and content: - messages.append(message) - elif message.get("role") == "user": - # Do not keep an assistant answer after dropping its user audio. - messages = [m for m in messages if has_image(m)] - while messages and messages[0].get("role") == "assistant": - messages.pop(0) + for index, turn in turns: + # Keep the existing recent-message window, rounded outwards to a + # complete turn. Never retain an answer whose input was cut off. + if index + len(turn) <= len(history) - 16: + continue + if turn[0].get("role") == "assistant": + continue + resolved = [] + included_audio = set() + for message in turn: + content = message.get("content") + if message.get("role") == "user" and isinstance(content, list): + if any(part.get("type") == "audio_url" for part in content): + audio = next((value for keys, value in retained if any(key is content for key in keys)), None) + if audio is None: + break + if id(audio) not in included_audio: + resolved.append({"role": "user", "audio_payload": audio}) + included_audio.add(id(audio)) + # Preserve text/images in mixed source items as well. + other_parts = [part for part in content if part.get("type") != "audio_url"] + if other_parts: + resolved.append({"role": "user", "content": other_parts}) + continue + resolved.append(message) + else: + messages.extend(resolved) if has_audio and not current_is_in_history: # History is context, not a replacement for the current request. # Re-encoding only history here makes Qwen answer an old question. messages.append({"role": "user", "audio_payload": payload}) return {**config, "qwen_messages": messages} + @staticmethod + def format_history(messages): + """Merge user items as Qwen's official web_demo.format_history does. + + This is a model-input view, not a mutation of addressable session items. + Empty assistant messages remain boundaries for interrupted/unheard turns. + """ + formatted = [] + pending = [] + for message in messages: + if message.get("role") == "user": + content = message.get("content") + pending.extend([{"type": "text", "text": content}] if isinstance(content, str) else content or []) + continue + if pending: + formatted.append({"role": "user", "content": pending}) + pending = [] + formatted.append(message) + if pending: + # The official demo puts media before the final turn's instruction. + media = [part for part in pending if part.get("type") != "text"] + text = [part for part in pending if part.get("type") == "text"] + formatted.append({"role": "user", "content": media + text}) + return formatted + @staticmethod def _trim_prompt_images(messages, images): """Drop the oldest images, with their placeholders, down to the prompt limit. @@ -323,7 +376,7 @@ class Qwen3OmniDuplexPlugin(DuplexModelPlugin): raise DuplexRuntimeConfigError("Qwen generation requires committed audio or conversation history") audios = [] images = [] - for index, message in enumerate(history): + for message in history: if "audio_payload" not in message: content = message.get("content") if isinstance(content, list): @@ -343,10 +396,11 @@ class Qwen3OmniDuplexPlugin(DuplexModelPlugin): continue content, audio = self._audio_content(message["audio_payload"]) audios.append((audio, 16000)) - if runtime_config.get("initial_user_text") and index == len(history) - 1: - content.append({"type": "text", "text": runtime_config["initial_user_text"]}) messages.append({"role": "user", "content": content}) + if payload.get("audio") and runtime_config.get("initial_user_text"): + messages.append({"role": "user", "content": runtime_config["initial_user_text"]}) messages, images = self._trim_prompt_images(messages, images) + messages = self.format_history(messages) prompt = self.processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) mm = {"audio": audios} if audios else {} if images: diff --git a/vllm_omni/model_executor/models/qwen3_omni/duplex/session.py b/vllm_omni/model_executor/models/qwen3_omni/duplex/session.py index de19e404c..038474e9b 100644 --- a/vllm_omni/model_executor/models/qwen3_omni/duplex/session.py +++ b/vllm_omni/model_executor/models/qwen3_omni/duplex/session.py @@ -17,8 +17,9 @@ class QwenDuplexSessionState(DuplexModelSessionState): """Mutable model-owned state of one Qwen duplex session (owned by the session runner).""" audio_buffer: QwenPcmBuffer = field(default_factory=QwenPcmBuffer) - # Keep content objects alongside payloads so identity survives history copies. - audio_history: list[tuple[object, dict]] = field(default_factory=list) + # Each payload can combine several audio commits from one pending turn. + # Keep their content objects so associations survive shallow history copies. + audio_history: list[tuple[tuple[object, ...], dict]] = field(default_factory=list) input_since_commit: bool = False speech_since_commit: bool = False context_locked: bool = False