From 62b15142bdab8c80eb83473f65e51774d1a2521b Mon Sep 17 00:00:00 2001 From: NumberWan Date: Tue, 22 Sep 2026 00:26:54 +0800 Subject: [PATCH] [Model] Integrate AURA into Full-Duplex Runtime (#7633) --- .../examples/online_serving/aura_omni.md | 67 +- examples/online_serving/aura_omni/README.md | 176 ++--- .../aura_omni/aura_omni_duplex_smoke.yaml | 119 +++ .../aura_omni/realtime_web/__init__.py | 4 + .../aura_omni/realtime_web/__main__.py | 7 + .../aura_omni/run_duplex_smoke_serve.sh | 29 + .../aura_omni/smoke_duplex_realtime_client.py | 153 ++++ .../online_serving/realtime_web/README.md | 38 +- .../realtime_web/app/index.html | 2 + .../realtime_web/app/static/app.js | 219 +++++- .../app/static/profiles/aura_ptt.js | 79 ++ .../realtime_web/app/static/styles.css | 12 + .../online_serving/realtime_web/server.py | 20 +- .../test_aura_omniinteract_clock.py | 157 ++++ tests/config/test_environment_variables.py | 4 +- .../duplex/test_aura_resumable_hooks.py | 725 ++++++++++++++++++ tests/engine/duplex/test_engine_session.py | 105 +++ tests/engine/duplex/test_input_transcribed.py | 49 ++ tests/engine/duplex/test_overlap_policy.py | 33 + tests/engine/duplex/test_session_manager.py | 43 +- tests/engine/duplex/test_session_runner.py | 259 ++++++- .../test_session_runner_ephemeral_bind.py | 28 + tests/engine/test_duplex_orchestrator.py | 94 ++- .../test_orchestrator_stage_input_bridge.py | 36 + .../entrypoints/duplex/test_startup_warmup.py | 56 ++ .../test_minicpmo_realtime_web_server.py | 13 +- .../test_minicpmo_realtime_web_static.py | 15 + tests/examples/test_realtime_web_profiles.cjs | 157 +++- .../aura_omni/duplex/test_aura_data_plane.py | 165 ++++ .../duplex/test_aura_data_plane_multimodal.py | 115 +++ .../duplex/test_aura_plugin_contract.py | 598 +++++++++++++++ .../aura_omni/duplex/test_sentence_decode.py | 62 ++ .../test_qwen3_tts_talker_preprocess.py | 32 + .../stage_input_processors/test_aura_omni.py | 113 ++- .../test_aura_omni_duplex_history.py | 273 +++++++ .../test_realtime_codec_single_source.py | 5 +- vllm_omni/benchmarks/omniinteract.py | 70 +- .../config/environment_variable_inventory.py | 2 + vllm_omni/config/stage_config.py | 10 +- vllm_omni/deploy/aura_omni.yaml | 17 +- vllm_omni/engine/duplex/commands.py | 7 +- vllm_omni/engine/duplex/config.py | 33 + vllm_omni/engine/duplex/contracts.py | 81 +- vllm_omni/engine/duplex/pcm_reservation.py | 53 ++ vllm_omni/engine/duplex/plugin.py | 112 +++ vllm_omni/engine/duplex/realtime_events.py | 41 +- .../engine/duplex/session/append_task.py | 13 +- vllm_omni/engine/duplex/session/context.py | 4 + .../engine/duplex/session/engine_session.py | 204 ++++- vllm_omni/engine/duplex/session/helpers.py | 28 +- vllm_omni/engine/duplex/session/manager.py | 17 +- .../engine/duplex/session/model_channel.py | 270 ++++++- .../engine/duplex/session/overlap_policy.py | 15 + vllm_omni/engine/duplex/session/runner.py | 357 ++++++--- vllm_omni/engine/duplex_orchestrator.py | 98 ++- vllm_omni/engine/orchestrator.py | 4 + vllm_omni/entrypoints/duplex/warmup.py | 202 ++++- vllm_omni/entrypoints/openai/api_server.py | 27 +- .../models/aura_omni/benchmarks/__init__.py | 4 + .../benchmarks/omniinteract_clock.py | 287 +++++++ .../models/aura_omni/duplex/__init__.py | 6 + .../models/aura_omni/duplex/capabilities.py | 48 ++ .../models/aura_omni/duplex/data_plane.py | 381 +++++++++ .../models/aura_omni/duplex/history.py | 128 ++++ .../models/aura_omni/duplex/input.py | 186 +++++ .../models/aura_omni/duplex/plugin.py | 520 +++++++++++++ .../models/aura_omni/duplex/sentence_tts.py | 189 +++++ .../models/aura_omni/duplex/session.py | 58 ++ .../models/aura_omni/pipeline.py | 1 + .../minicpmo_4_5/duplex/capabilities.py | 3 + .../models/minicpmo_4_5/duplex/input.py | 30 +- .../models/qwen3_tts/qwen3_tts_talker.py | 15 +- .../stage_input_processors/aura_omni.py | 515 ++++++++++++- vllm_omni/protocol/duplex/commands.py | 4 +- vllm_omni/protocol/duplex/errors.py | 1 + vllm_omni/protocol/realtime/audio_input.py | 46 +- vllm_omni/protocol/realtime/commands.py | 3 +- 77 files changed, 7695 insertions(+), 457 deletions(-) create mode 100644 examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml create mode 100644 examples/online_serving/aura_omni/realtime_web/__init__.py create mode 100644 examples/online_serving/aura_omni/realtime_web/__main__.py create mode 100755 examples/online_serving/aura_omni/run_duplex_smoke_serve.sh create mode 100755 examples/online_serving/aura_omni/smoke_duplex_realtime_client.py create mode 100644 examples/online_serving/realtime_web/app/static/profiles/aura_ptt.js create mode 100644 tests/benchmarks/test_aura_omniinteract_clock.py create mode 100644 tests/engine/duplex/test_aura_resumable_hooks.py create mode 100644 tests/engine/duplex/test_input_transcribed.py create mode 100644 tests/engine/duplex/test_session_runner_ephemeral_bind.py create mode 100644 tests/entrypoints/duplex/test_startup_warmup.py create mode 100644 tests/model_executor/models/aura_omni/duplex/test_aura_data_plane.py create mode 100644 tests/model_executor/models/aura_omni/duplex/test_aura_data_plane_multimodal.py create mode 100644 tests/model_executor/models/aura_omni/duplex/test_aura_plugin_contract.py create mode 100644 tests/model_executor/models/aura_omni/duplex/test_sentence_decode.py create mode 100644 tests/model_executor/stage_input_processors/test_aura_omni_duplex_history.py create mode 100644 vllm_omni/engine/duplex/pcm_reservation.py create mode 100644 vllm_omni/model_executor/models/aura_omni/benchmarks/__init__.py create mode 100644 vllm_omni/model_executor/models/aura_omni/benchmarks/omniinteract_clock.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/__init__.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/capabilities.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/data_plane.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/history.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/input.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/plugin.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/sentence_tts.py create mode 100644 vllm_omni/model_executor/models/aura_omni/duplex/session.py diff --git a/docs/user_guide/examples/online_serving/aura_omni.md b/docs/user_guide/examples/online_serving/aura_omni.md index 8b7f56ac5..2042d0d37 100644 --- a/docs/user_guide/examples/online_serving/aura_omni.md +++ b/docs/user_guide/examples/online_serving/aura_omni.md @@ -6,29 +6,29 @@ ASR -> AURA -> Qwen3-TTS Talker -> Code2Wav ``` -Qwen3-TTS remains two engine stages so the pipeline reuses the existing native -Talker and Code2Wav implementation. +**Supported online path:** Realtime duplex at `/v1/realtime?duplex=1`. + +The default deploy profile (`vllm_omni/deploy/aura_omni.yaml`) sets +`session_mode: duplex`. The `aura_omni` pipeline declares `duplex_plugin`, and +`DuplexOmniEngine` requires that mode (same pattern as MiniCPM-o duplex +deploys). Turn-based `chat/completions` is not a supported AURA online mode +against this profile. + +`async_chunk` defaults to `false` in that file (safer if a turn-based +`OmniOrchestrator` ever loads it). Duplex Realtime Stage2→3 still needs +async chunks — use the smoke deploy below, or set `async_chunk: true` +yourself. ```bash -vllm serve aurateam/AURA \ - --omni \ - --deploy-config vllm_omni/deploy/aura_omni.yaml \ - --served-model-name aurateam/AURA \ - --trust-remote-code +bash examples/online_serving/aura_omni/run_duplex_smoke_serve.sh +python examples/online_serving/aura_omni/smoke_duplex_realtime_client.py ``` -Configure local checkpoints by editing per-stage `model` values in -`vllm_omni/deploy/aura_omni.yaml`. The deploy file sets -`pipeline: aura_omni`, so the four-stage topology is used even if the -command-line `--model` points at one of the component checkpoints. +Configure local checkpoints by editing per-stage `model` values in the deploy +or smoke YAML. The file sets `pipeline: aura_omni`, so the four-stage topology +is used even if the command-line `--model` points at one component checkpoint. -Send requests with `"model": "aurateam/AURA"`. The ASR, AURA, and Qwen3-TTS -checkpoint paths are internal stage models from the deploy YAML, not the -OpenAI-facing served model name. - -The AURA stage can emit `<|silent|>`. Silent outputs are treated as a gate: -they produce no Qwen3-TTS Talker input, so no audio is synthesized for that -turn. +Silent Stage1 outputs (`<|silent|>` / id `151669`) skip TTS for that turn. ## GPU Utilization Recommendation @@ -40,26 +40,21 @@ VRAM each stage can reserve. Start with this split for a single GPU: - Stage 2 (Qwen3-TTS Talker): `0.20` - Stage 3 (Qwen3-TTS Code2Wav): `0.20` -## TTS Modes +## TTS modes (stage extras) -`aura_omni` can pass AURA text to Qwen3-TTS in two task modes: +When the duplex session supplies TTS extras, AURA text can feed Qwen3-TTS as: -- `Base`: voice clone from `tts_ref_audio` with ICL enabled in the AURA - pipeline. Provide both `tts_ref_audio` and `tts_ref_text`. Set - `tts_x_vector_only_mode=true` to disable ICL and use speaker embedding only. -- `CustomVoice`: predefined speaker mode. Use a Qwen3-TTS CustomVoice - checkpoint for stages 2 and 3 in `aura_omni.yaml`, then pass - `tts_task_type=CustomVoice` and `tts_speaker`. +- `Base`: voice clone from `tts_ref_audio` (optional x-vector-only mode) +- `CustomVoice`: predefined speaker (`tts_speaker`) with a CustomVoice checkpoint + on stages 2 and 3 -By default, AURA responses are passed to Qwen3-TTS as text. Set -`additional_information.tts_pass_token_ids=true` to pass AURA-generated -assistant token ids directly instead. Even when token passthrough is disabled, -the stage processor uses AURA token ids when available to estimate the Talker -prefill length, so scheduling does not rely on raw character length. +Optional `tts_pass_token_ids` passes AURA assistant token ids into Talker +instead of detokenized text. -The example client exposes this as: +## Old chat / Gradio / curl scripts (unsupported) -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --tts-pass-token-ids -``` +`examples/online_serving/aura_omni/` still contains chat-completions, curl, and +Gradio helpers from an older turn-based path. They do **not** work against the +shipped duplex profile. That path was an incomplete stand-in before streaming +I/O and session history were available; it is not a supported AURA mode. Use +the Realtime duplex smoke client for online verification. diff --git a/examples/online_serving/aura_omni/README.md b/examples/online_serving/aura_omni/README.md index 5f246c336..903850469 100644 --- a/examples/online_serving/aura_omni/README.md +++ b/examples/online_serving/aura_omni/README.md @@ -6,139 +6,89 @@ Qwen3-ASR -> AURA/Qwen3-VL -> Qwen3-TTS Talker -> Qwen3-TTS Code2Wav ``` -The pipeline has three semantic modules, but four engine stages because the -existing Qwen3-TTS implementation is natively split into Talker and Code2Wav. +**AURA's supported online path is Realtime duplex** (`/v1/realtime?duplex=1`). +The default deploy profile sets `session_mode: duplex` because the pipeline +declares `duplex_plugin` and `DuplexOmniEngine` requires that mode. -Start the server with the deploy profile: +This is AURA v1 / Qwen3-VL (silent / ChatML ids `151669` / `151645`), not +AURA v2 / Qwen3.5-VL. + +## Duplex Realtime (only supported online path) + +Stage2→3 codec handoff needs `async_chunk: true`. The default +`vllm_omni/deploy/aura_omni.yaml` keeps `async_chunk: false` as a safer +baseline (turn-based `OmniOrchestrator` never sees the duplex async-chunk +gate). For Realtime, use the smoke deploy (already `async_chunk: true`) or +set that flag yourself: ```bash +bash examples/online_serving/aura_omni/run_duplex_smoke_serve.sh +python examples/online_serving/aura_omni/smoke_duplex_realtime_client.py +``` + +Smoke deploy file: `examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml`. + +Equivalent serve with the default topology and async chunks enabled: + +```bash +# Copy or edit so async_chunk: true, then: vllm serve aurateam/AURA \ --omni \ --port 8091 \ - --deploy-config vllm_omni/deploy/aura_omni.yaml \ + --deploy-config examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml \ --served-model-name aurateam/AURA \ --trust-remote-code ``` -The deploy file sets per-stage model repos: +Connect clients to `/v1/realtime?duplex=1`. Silent Stage1 outputs gate TTS (no +audio for that turn). Overlapped input and vision-follow are AURA duplex +capabilities. + +### Browser UI + +AURA has no client VAD: control is **push-to-talk** on the shared shell from +[#7585](https://github.com/vllm-project/vllm-omni/pull/7585) +([`examples/online_serving/realtime_web/`](../realtime_web/README.md)), profile +`aura-ptt` (hold = `is_speech=true` PCM + sticky frames; release = `commit`; +silent+frames for vision-follow). Thin wrapper: + +```bash +python -m examples.online_serving.aura_omni.realtime_web \ + --ws-backend ws://127.0.0.1:8099 --model aurateam/AURA --port 7862 +``` + +Do **not** fork a separate `aura_omni/realtime_web` asset tree; MiniCPM / Qwen +profiles are unchanged and hide the PTT control. + +### Per-stage models + +Edit `model` on each stage in `vllm_omni/deploy/aura_omni.yaml` (or the smoke +YAML) for local checkpoints: - Stage 0 ASR: `Qwen/Qwen3-ASR-1.7B` - Stage 1 AURA: `aurateam/AURA` -- Stage 2/3 TTS: `Qwen/Qwen3-TTS-12Hz-1.7B-Base` +- Stage 2/3 TTS: `Qwen/Qwen3-TTS-12Hz-1.7B-Base` (or CustomVoice) -For local weights, edit the `model` value on each stage in -`vllm_omni/deploy/aura_omni.yaml`. The deploy profile includes -`pipeline: aura_omni`, so the server uses this four-stage topology even when -the command-line model path points at one component checkpoint. +### GPU utilization -Expected request shape: - -- Send microphone audio as the Stage 0 multimodal audio input. -- Include video frames in the original request `multi_modal_data`; the - `asr2aura` processor carries them forward to AURA. -- Optional `additional_information` keys: - - `aura_system_prompt` - - `tts_task_type` - - `tts_language` - - `tts_speaker` - - `tts_instruct` - - `tts_ref_audio` - - `tts_ref_text` - - `tts_x_vector_only_mode` - - `tts_pass_token_ids` - -If AURA emits `<|silent|>`, the `aura2tts` processor returns no TTS request, so -the TTS stages are skipped for that turn. - -## GPU Utilization Recommendation - -Tune `gpu_memory_utilization` per stage in `vllm_omni/deploy/aura_omni.yaml`. -Recommended baseline on one GPU for H200 +Tune `gpu_memory_utilization` per stage. Baseline on one large GPU: - Stage 0 (ASR): `0.10` -- Stage 1 (AURA): `0.4` -- Stage 2 (Qwen3-TTS Talker): `0.20` -- Stage 3 (Qwen3-TTS Code2Wav): `0.20` +- Stage 1 (AURA): `0.40` +- Stage 2 (Talker): `0.20` +- Stage 3 (Code2Wav): `0.20` -## Python Client +## Old chat / Gradio / curl scripts (unsupported) -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --host localhost \ - --port 8091 \ - --model aurateam/AURA \ - --modalities text,audio -``` +`openai_chat_completion_client.py`, `run_curl_multimodal_generation.sh`, and +`run_gradio_demo.sh` target the older turn-based Omni path. They do **not** +work against the shipped `session_mode: duplex` profile. -Use local media: - -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --audio-path /path/to/input.wav \ - --video-path /path/to/video.mp4 \ - --output-dir output_aura_omni_online -``` - -Base voice clone mode (default, recommended as x-vector while debugging ICL): - -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --tts-task-type Base \ - --tts-ref-audio vllm-omni/tests/assets/qwen3_tts/clone_2.wav \ - --tts-ref-text "Okay. Yeah. I resent you. I love you. I respect you. But you know what? You blew it! And thanks to you." -``` - -Enable AURA token-id passthrough explicitly: - -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --tts-pass-token-ids -``` - -CustomVoice mode requires stages 2 and 3 in `aura_omni.yaml` to point at a -Qwen3-TTS CustomVoice checkpoint: - -```bash -python examples/online_serving/aura_omni/openai_chat_completion_client.py \ - --tts-task-type CustomVoice \ - --tts-speaker Vivian -``` - -By default, AURA responses are passed to Qwen3-TTS as text. Set -`tts_pass_token_ids=true` to pass AURA-generated assistant token ids directly -to Qwen3-TTS instead. The processor still uses AURA token ids, when available, -to estimate the Talker prompt length in the default text path. - -## Curl - -```bash -cd examples/online_serving/aura_omni -bash run_curl_multimodal_generation.sh -``` - -Set `PORT`, `MODEL`, or `OUTPUT_DIR` to override defaults: - -```bash -PORT=8666 MODEL=aurateam/AURA bash run_curl_multimodal_generation.sh -TTS_PASS_TOKEN_IDS=true PORT=8666 MODEL=aurateam/AURA bash run_curl_multimodal_generation.sh -``` - -## Gradio - -Launch the server and Gradio UI together: - -```bash -cd examples/online_serving/aura_omni -bash run_gradio_demo.sh -``` - -If the server is already running: - -```bash -python examples/online_serving/aura_omni/gradio_demo.py \ - --model aurateam/AURA \ - --api-base http://localhost:8091/v1 -``` +That turn-based path was an incomplete June-era stand-in: vLLM-Omni did not +yet support streaming I/O for this stack, and adding session history then would +have required a much larger orchestrator change. It is not a supported AURA +mode. Do not run those scripts against the default deploy. Prefer the duplex +smoke client above. ## Offline diff --git a/examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml b/examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml new file mode 100644 index 000000000..2be7ea0e7 --- /dev/null +++ b/examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml @@ -0,0 +1,119 @@ +# AURA duplex Realtime smoke (local weights, session_mode: duplex). +# Stage1 = /workspace/models/AURA + AuraQwen3VLForConditionalGeneration (AURA v1 / Qwen3-VL). +pipeline: aura_omni +# Qwen3-TTS Stage2→3 codec handoff requires async_chunk=true (same as native / +# duplex explore). async_chunk=false reaches talker/code_predictor then never +# delivers audio to Stage3/client. +async_chunk: true +session_mode: duplex +duplex_session: + max_sessions: 1 + +connectors: + connector_of_shared_memory: + name: SharedMemoryConnector + extra: + codec_streaming: true + connector_get_sleep_s: 0.01 + connector_get_max_wait_first_chunk: 3000 + connector_get_max_wait: 300 + codec_chunk_frames: 25 + codec_left_context_frames: 72 + # Match native/duplex explore. initial=1 can stall Code2Wav on the first + # undersized chunk (observed hang after "Code2Wav codec: frames=1"). + initial_codec_chunk_frames: 8 + decode_cudagraph_capture_sizes: [25, 73, 97, 169, 325] + +stages: + - stage_id: 0 + max_num_seqs: 1 + gpu_memory_utilization: 0.08 + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: true + async_scheduling: true + max_model_len: 4096 + max_num_batched_tokens: 2048 + devices: "0" + model: /workspace/models/Qwen3-ASR-1.7B + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 256 + + - stage_id: 1 + max_num_seqs: 1 + gpu_memory_utilization: 0.55 + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: true + async_scheduling: true + max_model_len: 32768 + max_num_batched_tokens: 8192 + devices: "0" + model: /workspace/models/AURA + model_arch: AuraQwen3VLForConditionalGeneration + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 256 + repetition_penalty: 1.0 + stop_token_ids: [151669, 151645] + + - stage_id: 2 + # Allow prior TTS to drain while the next Talker starts (not assumed 1). + max_num_seqs: 2 + gpu_memory_utilization: 0.12 + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: false + async_scheduling: true + max_num_batched_tokens: 4096 + max_model_len: 4096 + devices: "0" + # Stage1→2 is orchestrator-fed via aura2tts (custom_process_input_func). + # A from_stage_1 SHM edge parks Talker in WAITING_FOR_CHUNK and clears + # additional_information.text → Missing Qwen3-TTS text conditioning. + # Only Stage2→3 codec handoff needs a connector. + output_connectors: + to_stage_3: connector_of_shared_memory + model: /workspace/models/Qwen3-TTS-12Hz-1.7B-CustomVoice + default_sampling_params: + temperature: 0.5 + top_k: 50 + # 12Hz × 240 ≈ 20s. Stay under the smoke 25s audio gate so a missed + # 2150 (English/quotes humming) cannot overlap the next vision-follow. + max_tokens: 240 + min_tokens: 8 + repetition_penalty: 1.05 + # Qwen3-TTS Talker codec EOS — without this Talker can run to max_tokens + # (~26s garbage audio while Stage1 text was correct). + stop_token_ids: [2150] + subtalker_sampling_params: + do_sample: true + temperature: 0.5 + top_k: 50 + top_p: 1.0 + + - stage_id: 3 + max_num_seqs: 2 + gpu_memory_utilization: 0.12 + enforce_eager: true + trust_remote_code: true + enable_prefix_caching: false + async_scheduling: true + max_num_batched_tokens: 16384 + max_model_len: 16384 + devices: "0" + input_connectors: + from_stage_2: connector_of_shared_memory + model: /workspace/models/Qwen3-TTS-12Hz-1.7B-CustomVoice + default_sampling_params: + temperature: 0.0 + top_p: 1.0 + top_k: -1 + max_tokens: 16384 + seed: 42 + repetition_penalty: 1.0 diff --git a/examples/online_serving/aura_omni/realtime_web/__init__.py b/examples/online_serving/aura_omni/realtime_web/__init__.py new file mode 100644 index 000000000..d305da15b --- /dev/null +++ b/examples/online_serving/aura_omni/realtime_web/__init__.py @@ -0,0 +1,4 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""AURA push-to-talk browser call UI.""" diff --git a/examples/online_serving/aura_omni/realtime_web/__main__.py b/examples/online_serving/aura_omni/realtime_web/__main__.py new file mode 100644 index 000000000..1f89c5d41 --- /dev/null +++ b/examples/online_serving/aura_omni/realtime_web/__main__.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from examples.online_serving.realtime_web.server import main + +if __name__ == "__main__": + main(default_profile="aura-ptt") diff --git a/examples/online_serving/aura_omni/run_duplex_smoke_serve.sh b/examples/online_serving/aura_omni/run_duplex_smoke_serve.sh new file mode 100755 index 000000000..ed8f3dc40 --- /dev/null +++ b/examples/online_serving/aura_omni/run_duplex_smoke_serve.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# Minimal AURA duplex Realtime server for smoke. +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "$REPO_ROOT" + +PORT="${PORT:-8099}" +MODEL="${MODEL:-/workspace/models/AURA}" +DEPLOY_CONFIG="${DEPLOY_CONFIG:-$REPO_ROOT/examples/online_serving/aura_omni/aura_omni_duplex_smoke.yaml}" + +# AURA v1 / Qwen3-VL ChatML ids (do not use AURA v2 248070/248046 here). +export VLLM_AURA_SILENT_TOKEN_ID="${VLLM_AURA_SILENT_TOKEN_ID:-151669}" +export VLLM_AURA_IM_END_TOKEN_ID="${VLLM_AURA_IM_END_TOKEN_ID:-151645}" +export VLLM_AURA_IM_START_TOKEN_ID="${VLLM_AURA_IM_START_TOKEN_ID:-151644}" +export VLLM_AURA_ASSISTANT_TOKEN_ID="${VLLM_AURA_ASSISTANT_TOKEN_ID:-77091}" +# Multi-stage colocated init can make earlier stages' Process.sentinel spuriously +# readable; prefer spawn for stage engine subprocesses (override if needed). +export VLLM_WORKER_MULTIPROC_METHOD="${VLLM_WORKER_MULTIPROC_METHOD:-spawn}" +# Do NOT set TORCHDYNAMO_DISABLE here: it breaks Stage0/1 AOT when those +# stages are not enforce_eager. Stage2/3 use enforce_eager; code_predictor +# honors that and skips its own torch.compile. + +exec "$REPO_ROOT/.venv/bin/vllm-omni" serve "$MODEL" \ + --omni \ + --deploy-config "$DEPLOY_CONFIG" \ + --trust-remote-code \ + --host 0.0.0.0 \ + --port "$PORT" diff --git a/examples/online_serving/aura_omni/smoke_duplex_realtime_client.py b/examples/online_serving/aura_omni/smoke_duplex_realtime_client.py new file mode 100755 index 000000000..431e13d90 --- /dev/null +++ b/examples/online_serving/aura_omni/smoke_duplex_realtime_client.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 +"""Minimal AURA duplex Realtime smoke via DuplexClient.""" + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import json +import sys +import time +from io import BytesIO +from pathlib import Path + +from PIL import Image + +REPO_ROOT = Path(__file__).resolve().parents[3] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from vllm_omni.clients.duplex import ( # noqa: E402 + DuplexClient, + EventCollector, + SessionConfig, + read_pcm16_wav, +) + +PROGRESS_TYPES = { + "response.created", + "response.done", + "response.audio.delta", + "response.output_audio.delta", + "response.audio_transcript.delta", + "response.output_text.delta", + "response.text.delta", + "response.listen", + "response.model_listen", + "error", +} + + +def _tiny_jpeg_b64() -> str: + img = Image.new("RGB", (64, 64), color=(200, 40, 40)) + buf = BytesIO() + img.save(buf, format="JPEG", quality=85) + return base64.b64encode(buf.getvalue()).decode("ascii") + + +async def run(url: str, model: str, wav: Path, timeout_s: float) -> int: + pcm16 = read_pcm16_wav(wav) + if not pcm16: + print("empty wav", file=sys.stderr) + return 2 + + frame = _tiny_jpeg_b64() + collector = EventCollector() + config = SessionConfig( + modalities=("text", "audio"), + instructions="You are AURA. Reply briefly in Chinese.", + extra_body={ + "aura_system_prompt": "You are AURA. Reply briefly in Chinese.", + "tts_task_type": "CustomVoice", + "tts_language": "Chinese", + "tts_speaker": "Vivian", + }, + ) + + started = time.time() + client = DuplexClient( + url, + model=model, + config=config, + reconnect=None, + heartbeat_interval_s=None, + handshake_timeout_s=min(60.0, timeout_s), + ) + + async with client: + consume_task = asyncio.create_task(collector.consume(client)) + collector.add({"type": "session.created", "session": getattr(client, "session_info", {})}) + + await client.append_audio(pcm16, is_speech=True, video_frames=[frame]) + await client.commit(create_response=True) + + progress = False + got_done = False + while time.time() - started < timeout_s: + types = [e.get("type") for e in collector.events] + if any(t in PROGRESS_TYPES for t in types): + progress = True + if "response.done" in types or "response.completed" in types: + got_done = True + progress = True + break + for e in collector.events: + meta = e.get("metadata") if isinstance(e.get("metadata"), dict) else {} + if meta.get("model_listen") or meta.get("listen_source") == "aura_silent": + progress = True + break + if collector.count("error") > 0 or collector.count("session.closed") > 0: + break + # Keep draining until response.done so TTS is not truncated. + await asyncio.sleep(0.2) + + try: + await client.close(timeout_s=30.0) + except Exception as exc: # noqa: BLE001 + print(f"close_warn={exc}", file=sys.stderr) + try: + await asyncio.wait_for(consume_task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError): + consume_task.cancel() + + event_types = [e.get("type") for e in collector.events] + errors = [json.dumps(e, ensure_ascii=False)[:500] for e in collector.errors()] + result = { + "ok": progress and (got_done or collector.count("response.done") > 0), + "got_response_progress": progress, + "got_response_done": got_done or collector.count("response.done") > 0, + "errors": errors, + "event_types": event_types, + "n_events": len(event_types), + "audio_bytes": len(collector.audio_bytes()), + "llm_text": collector.response_text(collector.response_ids[0]) if collector.response_ids else "", + "elapsed_s": round(time.time() - started, 2), + } + print(json.dumps(result, ensure_ascii=False, indent=2)) + log_path = Path("/tmp/aura_duplex_smoke/client_events.jsonl") + log_path.parent.mkdir(parents=True, exist_ok=True) + log_path.write_text( + "".join(json.dumps(e, ensure_ascii=False) + "\n" for e in collector.events), + encoding="utf-8", + ) + print(f"events_log={log_path}", file=sys.stderr) + return 0 if progress else 1 + + +def main() -> int: + p = argparse.ArgumentParser() + p.add_argument("--url", default="ws://127.0.0.1:8099/v1/realtime") + p.add_argument("--model", default="/workspace/models/AURA") + p.add_argument( + "--wav", + default=str(REPO_ROOT / "tests/assets/minicpmo_4_5/response_required_16k.wav"), + ) + p.add_argument("--timeout-s", type=float, default=300.0) + args = p.parse_args() + return asyncio.run(run(args.url, args.model, Path(args.wav), args.timeout_s)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/online_serving/realtime_web/README.md b/examples/online_serving/realtime_web/README.md index 854aa4718..7aba2ee10 100644 --- a/examples/online_serving/realtime_web/README.md +++ b/examples/online_serving/realtime_web/README.md @@ -9,6 +9,7 @@ This implements [RFC #7222](https://github.com/vllm-project/vllm-omni/issues/722 | `minicpm-native` | Model-controlled listen/speak, continuous audio input | Frames accompany audio | Yes | | `qwen3-turn --stt` (default) | User presses **Send turn** | No | No | | `qwen3-turn --vad` | Server detects trailing silence; speech interrupts replies | Sampled frames with each spoken turn | Yes | +| `aura-ptt` | Hold **Hold to talk**; release commits. Hold also stops local playback | Sticky frames with speech; vision-follow every 2 frames only while unlocked (locked from PTT release until text-final / listen) | Yes | Qwen3 VAD uses an engine-owned duplex plugin: microphone upload continues while replies stream, and speech can interrupt generation and playback. The @@ -29,6 +30,38 @@ The compatibility wrapper serves the shared assets with `minicpm-native`. The current native query, `extra_body`, reference voice, continuous microphone upload, camera frames and playback acknowledgements are retained. +## AURA: push-to-talk duplex + +AURA has no client VAD. Start the duplex serve, then the thin wrapper: + +```bash +bash examples/online_serving/aura_omni/run_duplex_smoke_serve.sh + +python -m examples.online_serving.aura_omni.realtime_web \ + --ws-backend ws://127.0.0.1:8099 --model aurateam/AURA --port 7862 +``` + +Or select the profile on the shared host: + +```bash +python -m examples.online_serving.realtime_web --profile aura-ptt \ + --ws-backend ws://127.0.0.1:8099 --model aurateam/AURA --port 7862 +``` + +Open the UI, start a session, optionally enable **Camera**, then **hold** +**Hold to talk** to stream `is_speech=true` PCM (with sticky camera frames). +**Release** to `input_audio_buffer.commit`. Holding the button stops playback +in the browser only; the in-flight response is not cancelled on the server. +With the camera on, frames are still captured at 2 fps. While the button is +up they are committed once per two frames (`is_speech=false` + silent PCM). +Both frames of that commit are sent and packed as one video (`<|video_pad|>`). +Turns still open from this client commit, not from a server-side auto trigger. +but only after the prior spoken turn unlocks (text-final or `response.listen`). +Release itself locks vision immediately so a follow-up frame cannot abort the +just-committed speech turn on Stage0/1 (`max_num_seqs=1`). Audio still playing +does not hold the next vision turn once text is done. +MiniCPM / Qwen profiles do not set `pushToTalk`, so the PTT control stays hidden. + ## Qwen3: explicit-turn STT adapter Start the backend, then the UI in a second terminal: @@ -49,6 +82,7 @@ accepts only model selection, so system-prompt controls are hidden in this mode. The adapter follows the shipped `qwen3_omni/openai_realtime_client.py` and `vllm_omni/entrypoints/openai/realtime_connection.py`: + 1. Explicit `duplex=0` selects the legacy STT handler on a turn deployment. It does not enable STT on a duplex deployment. 2. Send `{type: "session.update", model: ...}` and `commit(final=false)`. @@ -181,7 +215,7 @@ conversation API. This demo does not add WebRTC, semantic VAD, or tool calling. ## Shared host options -The shared entry point can select either profile explicitly: +The shared entry point can select a profile explicitly: ```bash python -m examples.online_serving.realtime_web --profile qwen3-turn \ @@ -193,7 +227,7 @@ python -m examples.online_serving.realtime_web --profile qwen3-turn \ - `--public-realtime-url`: optional browser-visible WebSocket URL; otherwise the static host proxies `/v1/realtime` on the same origin. - `--host`, `--port`: UI bind address and port (default port 7862). -- `--ref-audio`: required for MiniCPM; rejected for Qwen3. +- `--ref-audio`: required for MiniCPM; rejected for Qwen3 and AURA. Microphone and camera access require `localhost` or HTTPS. For a remote backend, use an SSH tunnel to the UI host or an HTTPS reverse proxy with WebSocket support. diff --git a/examples/online_serving/realtime_web/app/index.html b/examples/online_serving/realtime_web/app/index.html index a38a731e4..d4662d5a1 100644 --- a/examples/online_serving/realtime_web/app/index.html +++ b/examples/online_serving/realtime_web/app/index.html @@ -22,6 +22,7 @@
+ @@ -75,6 +76,7 @@ + diff --git a/examples/online_serving/realtime_web/app/static/app.js b/examples/online_serving/realtime_web/app/static/app.js index 1dd01d4b3..f57a3f064 100644 --- a/examples/online_serving/realtime_web/app/static/app.js +++ b/examples/online_serving/realtime_web/app/static/app.js @@ -5,6 +5,7 @@ const profile = window.OmniRealtimeProfiles[config.profile || 'minicpm-native'](config); const callButton = document.getElementById('callButton'); const sendTurnButton = document.getElementById('sendTurnButton'); + const pttButton = document.getElementById('pttButton'); const muteButton = document.getElementById('muteButton'); const cameraButton = document.getElementById('cameraButton'); const cameraPreview = document.getElementById('cameraPreview'); @@ -28,6 +29,7 @@ const ECHO_GUARD_MS = 300; const INITIAL_PLAYBACK_BUFFER_MS = 400; const SESSION_CLOSE_TIMEOUT_MS = 1000; + const SILENT_PCM_SAMPLES = 160; const PROMPT_PRESETS = profile.presets; document.title = profile.title; @@ -36,7 +38,21 @@ document.getElementById('profileDescription').textContent = profile.description; document.getElementById('policyLabel').textContent = profile.policy; cameraButton.hidden = !profile.camera; + if (profile.cameraPreviewLarge) { + cameraPreview.classList.add('camera-preview-large'); + // Inline, so a cached stylesheet cannot leave the 96×72 HTML default. + cameraPreview.style.width = '100%'; + cameraPreview.style.height = 'auto'; + cameraPreview.style.maxHeight = '480px'; + cameraPreview.style.flex = '0 0 100%'; + const cameraActions = cameraPreview.closest('.call-actions'); + if (cameraActions) { + cameraActions.style.flexWrap = 'wrap'; + cameraActions.style.gridColumn = '1 / -1'; + } + } sendTurnButton.hidden = !profile.clientCommit; + pttButton.hidden = !profile.pushToTalk; promptPreset.replaceChildren(); for (const name of [...Object.keys(PROMPT_PRESETS), 'custom']) { const option = document.createElement('option'); @@ -64,6 +80,12 @@ let cameraStream = null; let cameraTimer = null; let cameraPendingFrame = null; + let cameraLastFrame = null; + // AURA vision clock only. Other profiles leave this empty and never read it. + let visionFollowQueue = []; + let visionTurnLocked = false; + let suppressedPlaybackId = null; + let pttHeld = false; const cameraCanvas = document.createElement('canvas'); let playbackRate = OUTPUT_RATE; let pendingCapture = []; @@ -87,6 +109,8 @@ const interruptedResponses = new Set(); const pendingEvents = new Set(); let assistantTextChannel = null; + const assistantRowByResponse = new Map(); + let lastClosedAssistantText = ''; let connectionReady = false; let turnCounter = 0; let stopping = null; @@ -134,7 +158,7 @@ assistantTextChannel = null; if (profile.halfDuplex) pendingCapture = []; sendTurnButton.disabled = !profile.clientCommit; - setModel(profile.waiting); + if (!pttHeld) setModel(profile.waiting); }, ECHO_GUARD_MS); } @@ -240,6 +264,7 @@ current.value = finalText; current.text.textContent = finalText; } + if (role === 'assistant' && current.value) lastClosedAssistantText = current.value; current.row.classList.remove('turn-live'); if (role === 'user') liveUserTurn = null; else liveAssistantTurn = null; @@ -317,11 +342,67 @@ } function microphoneUploadEnabled() { - return running && connectionReady && !muted && (!profile.halfDuplex || !assistantActive); + if (!running || !connectionReady || muted) return false; + if (profile.pushToTalk) return pttHeld; + return !profile.halfDuplex || !assistantActive; + } + + function queueVisionFrame(frame) { + visionFollowQueue.push(frame); + if (visionFollowQueue.length > 2) visionFollowQueue.shift(); + } + + function commitVisionFollow() { + // Two frames still gate the turn (client commit). Both frames are one clip; + // the server packs them as a single <|video_pad|>, same as Native. + const frames = visionFollowQueue.slice(); + visionFollowQueue = []; + const silent = new Int16Array(SILENT_PCM_SAMPLES); + const event = profile.append(int16ToBase64(silent), frames, { isSpeech: false }); + socket.send(JSON.stringify(event)); + for (const message of profile.commitMessages()) socket.send(JSON.stringify(message)); + } + + function releaseVisionTurn(responseId) { + if (!profile.visionFollowWhileSpeaking || !visionTurnLocked) return; + if (responseId && currentResponseId && responseId !== currentResponseId) return; + visionTurnLocked = false; + } + + function suppressLocalPlayback() { + // Hold-to-talk cuts the speaker only. The in-flight response stays on the server. + if (currentResponseId) suppressedPlaybackId = currentResponseId; + if (playbackNode) playbackNode.port.postMessage({ type: 'clear' }); + playbackComplete = true; + responseHasAudio = false; + setPlayback('Idle'); + } + + function playbackSuppressed(responseId) { + return Boolean(suppressedPlaybackId) && responseId === suppressedPlaybackId; } function flushCapture() { - if (!socket || socket.readyState !== WebSocket.OPEN || pendingCapture.length === 0) return; + if (!socket || socket.readyState !== WebSocket.OPEN) return; + + // AURA: capture stays 2 fps, but a turn opens only every 2 frames, and + // not until the previous response's text is done (or it listened). + // Playback still running is not the lock. PTT speech is not batched here. + if (profile.pushToTalk && profile.visionFollowWhileSpeaking && !pttHeld) { + if (cameraPendingFrame) { + queueVisionFrame(cameraPendingFrame); + cameraPendingFrame = null; + } + if (!visionTurnLocked && visionFollowQueue.length >= 2) commitVisionFollow(); + return; + } + + if (profile.pushToTalk && !pttHeld) { + pendingCapture = []; + return; + } + + if (pendingCapture.length === 0) return; if (!microphoneUploadEnabled()) { pendingCapture = []; return; @@ -335,11 +416,44 @@ } pendingCapture = []; const pcm = resampleInt16(merged, captureRate, profile.inputSampleRate || INPUT_RATE); - const appendEvent = profile.append(int16ToBase64(pcm), cameraPendingFrame); + const frame = cameraPendingFrame || (profile.stickyCamera ? cameraLastFrame : null); cameraPendingFrame = null; + const appendEvent = profile.pushToTalk + ? profile.append(int16ToBase64(pcm), frame, { isSpeech: true }) + : profile.append(int16ToBase64(pcm), frame); socket.send(JSON.stringify(appendEvent)); } + function setPttHeld(held) { + if (!profile.pushToTalk || !running || pttHeld === held) return; + if (held) { + pttHeld = true; + suppressLocalPlayback(); + pttButton.classList.toggle('is-active', true); + pttButton.textContent = 'Release to send'; + setModel('Talking'); + appendLog('PTT down'); + return; + } + // Flush remaining speech while still held, then commit. + flushCapture(); + if (socket && socket.readyState === WebSocket.OPEN) { + for (const event of profile.commitMessages()) socket.send(JSON.stringify(event)); + } + // Lock vision before clearing pttHeld. Stage0/1 are max_num_seqs=1: a + // vision-follow commit in the gap before response.begin aborts the speech + // turn (server sees a new ephemeral and kills the prior ASR request). + if (profile.visionFollowWhileSpeaking) { + visionTurnLocked = true; + visionFollowQueue = []; + } + pttHeld = false; + pttButton.classList.toggle('is-active', false); + pttButton.textContent = 'Hold to talk'; + setModel(profile.waiting); + appendLog('PTT up · commit'); + } + function beginAssistant(responseId) { currentResponseId = responseId || currentResponseId; responseHasAudio = false; @@ -351,6 +465,7 @@ } function feedPlayback(decoded, responseId) { + if (playbackSuppressed(responseId || currentResponseId)) return; if (!decoded || !decoded.pcm || decoded.pcm.length === 0 || !playbackNode) return; const pcm = resampleInt16(decoded.pcm, decoded.sourceRate, playbackRate); responseHasAudio = true; @@ -392,13 +507,18 @@ } async function handleAudioEvent(action) { + const responseId = action.responseId || currentResponseId || `turn-${turnCounter}`; + if (playbackSuppressed(responseId)) return; markBusy(); - currentResponseId = action.responseId || currentResponseId || `turn-${turnCounter}`; + currentResponseId = responseId; setModel('Speaking'); const generation = sessionGeneration; - const responseId = currentResponseId; const decoded = await decodeAudioDelta(action.event); - if (generation === sessionGeneration && !interruptedResponses.has(responseId)) feedPlayback(decoded, responseId); + if ( + generation === sessionGeneration + && !interruptedResponses.has(responseId) + && !playbackSuppressed(responseId) + ) feedPlayback(decoded, responseId); } function handleTranscriptEvent(action) { @@ -408,12 +528,43 @@ if (profile.deduplicateTranscript && assistantTextChannel && action.channel !== assistantTextChannel) return; if (action.text) assistantTextChannel = action.channel || assistantTextChannel; } - if (action.kind === 'text') addTranscript(action.role, action.text); - else if (profile.deduplicateTranscript && action.role === 'assistant') { + const responseId = action.responseId || null; + const existing = responseId && action.role === 'assistant' ? assistantRowByResponse.get(responseId) : null; + if (existing) { + if (action.kind === 'text') { + if (!action.text) return; + existing.value += action.text; + } else if (action.text) { + existing.value = action.text; + } + existing.text.textContent = existing.value; + liveAssistantTurn = existing; + return; + } + if ( + action.role === 'assistant' + && !liveAssistantTurn + && action.text + && action.text === lastClosedAssistantText + ) { + // response.done already closed this sentence. A later full transcript, + // including one stamped with a newer vision-follow response id, must + // not open a second bubble with the same words. + return; + } + if (action.kind === 'text') { + addTranscript(action.role, action.text); + if (action.role === 'assistant' && responseId && liveAssistantTurn) { + assistantRowByResponse.set(responseId, liveAssistantTurn); + } + return; + } + if (profile.deduplicateTranscript && action.role === 'assistant') { if (action.text) { const turn = ensureTurn('assistant'); turn.value = action.text; turn.text.textContent = action.text; + if (responseId) assistantRowByResponse.set(responseId, turn); } } else finishTranscript(action.role, action.text); } @@ -459,12 +610,14 @@ } case 'listen': assistantActive = false; + releaseVisionTurn(responseId); setModel(profile.waiting); break; case 'begin': if (profile.halfDuplex) armTurnTimeout(); if (echoTimer !== null) { clearTimeout(echoTimer); echoTimer = null; } beginAssistant(responseId); + if (profile.visionFollowWhileSpeaking) visionTurnLocked = true; break; case 'audio': await handleAudioEvent(action); @@ -474,11 +627,13 @@ break; case 'text': case 'text-final': handleTranscriptEvent(action); + if (action.kind === 'text-final' && action.role === 'assistant') releaseVisionTurn(responseId); break; case 'done': clearTimeout(turnTimeout); responseComplete = true; finishTranscript('assistant'); + releaseVisionTurn(responseId); requestPlaybackDrain(responseId); finishResponseIfReady(); break; @@ -671,7 +826,11 @@ await openSocket(); running = true; muted = false; + pttHeld = false; assistantActive = false; + visionFollowQueue = []; + visionTurnLocked = false; + suppressedPlaybackId = null; sendTimer = window.setInterval(flushCapture, profile.sendIntervalMs || SEND_INTERVAL_MS); startClock(); callButton.textContent = 'End session'; @@ -679,6 +838,7 @@ muteButton.disabled = false; cameraButton.disabled = !profile.camera; sendTurnButton.disabled = !profile.clientCommit; + pttButton.disabled = !profile.pushToTalk; setConnection('Connected', 'online'); setModel(profile.waiting); appendLog('session started'); @@ -698,8 +858,9 @@ cameraPreview.srcObject = cameraStream; cameraPreview.style.display = ''; await cameraPreview.play().catch(() => {}); - // Official omni-duplex cadence: one JPEG (quality 0.7) per ~1 s chunk, - // no client-side resize (the server normalizes at scale_resolution=448). + // JPEG quality 0.7; no client-side resize (server scale_resolution=448). + // Interval is per-profile: AURA 2 fps, MiniCPM/Qwen stay at 1 s. + const cameraIntervalMs = profile.cameraIntervalMs || 1000; const captureCameraFrame = () => { if (!cameraStream || cameraPreview.videoWidth === 0) return; const scale = profile.cameraMaxDimension @@ -712,14 +873,17 @@ if (connectionReady && socket?.readyState === WebSocket.OPEN) { for (const event of profile.imageMessages(frame)) socket.send(JSON.stringify(event)); } - } else cameraPendingFrame = frame; + } else { + cameraPendingFrame = frame; + if (profile.stickyCamera) cameraLastFrame = frame; + } }; // Do not make the first spoken turn race a one-second timer. captureCameraFrame(); - cameraTimer = window.setInterval(captureCameraFrame, 1000); + cameraTimer = window.setInterval(captureCameraFrame, cameraIntervalMs); cameraButton.textContent = 'Camera off'; cameraButton.classList.add('is-active'); - appendLog('camera on (1 fps omni frames)'); + appendLog(`camera on (${(1000 / cameraIntervalMs).toFixed(0)} fps omni frames)`); } function stopCamera() { @@ -730,6 +894,7 @@ } cameraStream = null; cameraPendingFrame = null; + cameraLastFrame = null; cameraPreview.srcObject = null; cameraPreview.style.display = 'none'; cameraButton.textContent = 'Camera'; @@ -778,9 +943,18 @@ playbackComplete = true; turnSubmitted = false; assistantTextChannel = null; + assistantRowByResponse.clear(); + lastClosedAssistantText = ''; sendTurnButton.disabled = true; + pttHeld = false; + pttButton.disabled = true; + pttButton.classList.remove('is-active'); + pttButton.textContent = 'Hold to talk'; running = false; assistantActive = false; + visionFollowQueue = []; + visionTurnLocked = false; + suppressedPlaybackId = null; pendingCapture = []; if (sendTimer !== null) clearInterval(sendTimer); if (clockTimer !== null) clearInterval(clockTimer); @@ -854,6 +1028,23 @@ setModel('Thinking'); appendLog('turn submitted'); }); + const pttDown = (event) => { + event.preventDefault(); + setPttHeld(true); + }; + const pttUp = (event) => { + event.preventDefault(); + setPttHeld(false); + }; + pttButton.addEventListener('pointerdown', pttDown); + pttButton.addEventListener('pointerup', pttUp); + pttButton.addEventListener('pointercancel', pttUp); + pttButton.addEventListener('pointerleave', () => { + if (pttHeld) setPttHeld(false); + }); + window.addEventListener('pointerup', () => { + if (pttHeld) setPttHeld(false); + }); muteButton.addEventListener('click', toggleMute); clearLogButton.addEventListener('click', () => { eventLog.textContent = ''; diff --git a/examples/online_serving/realtime_web/app/static/profiles/aura_ptt.js b/examples/online_serving/realtime_web/app/static/profiles/aura_ptt.js new file mode 100644 index 000000000..242aa1b6a --- /dev/null +++ b/examples/online_serving/realtime_web/app/static/profiles/aura_ptt.js @@ -0,0 +1,79 @@ +(function (root) { + 'use strict'; + const profiles = root.OmniRealtimeProfiles; + profiles['aura-ptt'] = () => ({ + title: 'AURA Realtime', + eyebrow: 'Push-to-talk duplex', + policy: 'Hold to talk • release to commit', + description: + 'Hold to talk, release to send. Camera stays on whether you speak or not. ' + + 'Holding the button stops local playback.', + waiting: 'Ready', + camera: true, + // Display only. Shared CSS puts .camera-preview-large on its own row. + // Other profiles keep the 96×72 preview in index.html. + cameraPreviewLarge: true, + playbackAck: true, + clientCommit: false, + pushToTalk: true, + stickyCamera: true, + // OmniInteract / Native AURA clock: 2 fps. Other profiles omit this and stay at 1 s. + cameraIntervalMs: 500, + visionFollowWhileSpeaking: true, + deduplicateTranscript: true, + halfDuplex: false, + closeSession: true, + reconnectEachTurn: false, + readyEvent: 'session.updated', + instructions: true, + sendIntervalMs: 200, + presets: { + omni: 'You are receiving a live video stream where the final frame is the present moment. Respond only when a response is needed based on the user\'s message or the visual context. Otherwise, output `<|silent|>` to signify silence.', + }, + url(config, location) { + const url = profiles.url(config, location); + url.searchParams.set('duplex', '1'); + url.searchParams.set('model', config.model || 'aurateam/AURA'); + return url.toString(); + }, + initialMessages(config, instructions) { + const session = { + modalities: ['audio', 'text'], + extra_body: { auto_response: true }, + }; + if (instructions) session.instructions = instructions; + return [{ type: 'session.update', session }]; + }, + append(audio, frame, opts) { + const isSpeech = !opts || opts.isSpeech !== false; + const event = { + type: 'input_audio_buffer.append', + audio, + format: 'pcm16', + sample_rate_hz: 16000, + is_speech: isSpeech, + }; + const frames = Array.isArray(frame) ? frame.filter(Boolean) : (frame ? [frame] : []); + if (frames.length) event.video_frames = frames; + return event; + }, + commitMessages: () => [{ type: 'input_audio_buffer.commit', create_response: true }], + ack(responseId, playedMs) { + return { + type: 'playback.ack', + response_id: responseId, + item_id: `item_${responseId}`, + played_ms: playedMs, + committed_ms: playedMs, + }; + }, + mapEvent(event) { + if (event.type === 'response.listen') return { kind: 'listen' }; + if (event.type === 'playback.acknowledged') { + return { kind: 'ack', committedMs: (event.event || event).committed_ms || 0 }; + } + return profiles.event(event); + }, + connectionHint: 'Check the AURA duplex serve (run_duplex_smoke_serve.sh) and --ws-backend port.', + }); +})(globalThis); diff --git a/examples/online_serving/realtime_web/app/static/styles.css b/examples/online_serving/realtime_web/app/static/styles.css index bf25c3eff..53f98df7c 100644 --- a/examples/online_serving/realtime_web/app/static/styles.css +++ b/examples/online_serving/realtime_web/app/static/styles.css @@ -71,6 +71,18 @@ main { width: min(1040px, calc(100% - 40px)); margin: 28px auto 48px; } } .call-actions { display: flex; gap: 8px; } +/* AURA only: class is added solely by the aura-ptt profile. */ +.call-bar:has(.camera-preview-large) .call-actions { grid-column: 1 / -1; } +.call-actions:has(.camera-preview-large) { flex-wrap: wrap; align-items: center; } +.call-actions:has(.camera-preview-large) .camera-preview-large { + order: -1; + flex: 0 0 100%; + width: 100% !important; + height: auto !important; + max-height: 480px; + aspect-ratio: 4 / 3; +} +.call-actions:has(.camera-preview-large) button { flex: 0 0 auto; } button { border-radius: 5px; border: 1px solid transparent; cursor: pointer; } button:disabled { opacity: .46; cursor: default; } .primary { padding: 10px 15px; color: #ffffff; background: #16834b; font-weight: 700; } diff --git a/examples/online_serving/realtime_web/server.py b/examples/online_serving/realtime_web/server.py index 85e4b63d5..f6a10c4d3 100644 --- a/examples/online_serving/realtime_web/server.py +++ b/examples/online_serving/realtime_web/server.py @@ -76,15 +76,20 @@ def build_app( profile: str = "minicpm-native", adapter: str = "stt", ) -> FastAPI: - if profile not in {"minicpm-native", "qwen3-turn"}: + if profile not in {"minicpm-native", "qwen3-turn", "aura-ptt"}: raise ValueError(f"Unknown profile: {profile}") if adapter not in {"stt", "vad"}: raise ValueError(f"Unknown adapter: {adapter}") if profile == "qwen3-turn" and ref_audio: raise ValueError("Qwen3 profile does not accept --ref-audio") + if profile == "aura-ptt" and ref_audio: + raise ValueError("AURA profile does not accept --ref-audio") native = profile == "minicpm-native" - ws_backend = ws_backend or ("ws://127.0.0.1:8099" if native else "ws://127.0.0.1:8091") - model = model or ("openbmb/MiniCPM-o-4_5" if native else "Qwen/Qwen3-Omni-30B-A3B-Instruct") + aura = profile == "aura-ptt" + ws_backend = ws_backend or ("ws://127.0.0.1:8099" if native or aura else "ws://127.0.0.1:8091") + model = model or ( + "openbmb/MiniCPM-o-4_5" if native else "aurateam/AURA" if aura else "Qwen/Qwen3-Omni-30B-A3B-Instruct" + ) app = FastAPI(title="Omni Realtime Web UI") index_path = APP_DIR / "index.html" app_version_hash = hashlib.sha256() @@ -182,7 +187,7 @@ def main(default_profile: str = "minicpm-native") -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=7862) - parser.add_argument("--profile", choices=("minicpm-native", "qwen3-turn"), default=default_profile) + parser.add_argument("--profile", choices=("minicpm-native", "qwen3-turn", "aura-ptt"), default=default_profile) parser.add_argument("--ws-backend", "--backend", dest="ws_backend") mode = parser.add_mutually_exclusive_group() mode.add_argument("--stt", dest="adapter", action="store_const", const="stt") @@ -202,12 +207,15 @@ def main(default_profile: str = "minicpm-native") -> None: ) args = parser.parse_args() native = args.profile == "minicpm-native" + aura = args.profile == "aura-ptt" if native and not args.ref_audio: parser.error("--ref-audio is required for the minicpm-native profile") if not native and args.ref_audio: parser.error("--ref-audio is only supported by minicpm-native") - args.model = args.model or ("openbmb/MiniCPM-o-4_5" if native else "Qwen/Qwen3-Omni-30B-A3B-Instruct") - args.ws_backend = args.ws_backend or ("ws://127.0.0.1:8099" if native else "ws://127.0.0.1:8091") + args.model = args.model or ( + "openbmb/MiniCPM-o-4_5" if native else "aurateam/AURA" if aura else "Qwen/Qwen3-Omni-30B-A3B-Instruct" + ) + args.ws_backend = args.ws_backend or ("ws://127.0.0.1:8099" if native or aura else "ws://127.0.0.1:8091") logging.basicConfig(level=logging.INFO) uvicorn.run( diff --git a/tests/benchmarks/test_aura_omniinteract_clock.py b/tests/benchmarks/test_aura_omniinteract_clock.py new file mode 100644 index 000000000..5e815f5fe --- /dev/null +++ b/tests/benchmarks/test_aura_omniinteract_clock.py @@ -0,0 +1,157 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest + +from vllm_omni.benchmarks.omniinteract import _session_capabilities, _wants_annotation_video_clock +from vllm_omni.model_executor.models.aura_omni.benchmarks.omniinteract_clock import ( + stream_annotation_clock, + wants_annotation_video_clock, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +class _Events: + def __init__(self, events): + self.events = events + + +class _Client: + def __init__(self, capabilities): + self.events = _Events([{"type": "session.created", "session": {"capabilities": capabilities}}]) + self._client = None + + +def test_annotation_clock_follows_capabilities_not_model_name(): + aura = {"required_input_modalities": ["video"], "supports_turn_commit_only": True} + minicpm = {"required_input_modalities": ["audio"], "supports_turn_commit_only": False} + audio_commit = {"required_input_modalities": ["audio"], "supports_turn_commit_only": True} + assert wants_annotation_video_clock(aura) + assert not wants_annotation_video_clock(minicpm) + assert not wants_annotation_video_clock(audio_commit) + assert not wants_annotation_video_clock(None) + assert _wants_annotation_video_clock(_Client(minicpm)) is False + assert _wants_annotation_video_clock(_Client(aura)) is True + assert _session_capabilities(_Client(aura))["required_input_modalities"] == ["video"] + + +def test_annotation_clock_commits_only_the_due_wav(monkeypatch, tmp_path: Path): + video = tmp_path / "videos" / "0002.mp4" + video.parent.mkdir() + video.write_bytes(b"") + sent = [] + + class Client: + async def send(self, event): + sent.append(event) + + async def commit(self): + sent.append({"type": "commit"}) + + class Playback: + async def acknowledge(self, client): + del client + + monkeypatch.setattr( + "vllm_omni.model_executor.models.aura_omni.benchmarks.omniinteract_clock.load_annotation_utterances", + lambda path: [{"at_sec": 0.0, "pcm": b"\x01\x00" * 160, "sent": False, "qa_index": 0}], + ) + chunks, frames, _, _ = asyncio.run( + stream_annotation_clock( + Client(), + ["frame-a", "frame-b"], + Playback(), + video_path=video, + fps=2.0, + ) + ) + assert (chunks, frames) == (2, 2) + assert sent[0]["is_speech"] is True + assert sent[0]["video_frames"] == ["frame-a"] + assert sent[1] == {"type": "commit"} + # The trailing silent frame is one vision turn, committed after the clip. + assert sent[2]["is_speech"] is False + assert sent[2]["video_frames"] == ["frame-b"] + assert sent[3] == {"type": "commit"} + assert sent.count({"type": "commit"}) == 2 + + +def test_annotation_clock_commits_silent_frames_in_pairs(monkeypatch, tmp_path: Path): + video = tmp_path / "videos" / "0002.mp4" + video.parent.mkdir() + video.write_bytes(b"") + sent = [] + + class Client: + async def send(self, event): + sent.append(event) + + async def commit(self): + sent.append({"type": "commit"}) + + class Playback: + async def acknowledge(self, client): + del client + + monkeypatch.setattr( + "vllm_omni.model_executor.models.aura_omni.benchmarks.omniinteract_clock.load_annotation_utterances", + lambda path: [{"at_sec": 100.0, "pcm": b"\x01\x00" * 160, "sent": False, "qa_index": 0}], + ) + asyncio.run( + stream_annotation_clock( + Client(), + ["f1", "f2", "f3", "f4"], + Playback(), + video_path=video, + fps=2.0, + ) + ) + vision = [event for event in sent if event.get("is_speech") is False] + assert [event["video_frames"] for event in vision] == [["f1", "f2"], ["f3", "f4"]] + assert sent.count({"type": "commit"}) == 2 + + +def test_annotation_clock_holds_vision_while_a_response_is_open(monkeypatch, tmp_path: Path): + video = tmp_path / "videos" / "0002.mp4" + video.parent.mkdir() + video.write_bytes(b"") + sent = [] + + class Events: + def __init__(self): + self.events = [{"type": "response.created", "response": {"id": "resp-open"}}] + + class Client: + def __init__(self): + self.events = Events() + + async def send(self, event): + sent.append(event) + + async def commit(self): + sent.append({"type": "commit"}) + + class Playback: + async def acknowledge(self, client): + del client + + monkeypatch.setattr( + "vllm_omni.model_executor.models.aura_omni.benchmarks.omniinteract_clock.load_annotation_utterances", + lambda path: [{"at_sec": 100.0, "pcm": b"\x01\x00" * 160, "sent": False, "qa_index": 0}], + ) + asyncio.run( + stream_annotation_clock( + Client(), + ["f1", "f2", "f3", "f4"], + Playback(), + video_path=video, + fps=2.0, + ) + ) + assert sent == [] diff --git a/tests/config/test_environment_variables.py b/tests/config/test_environment_variables.py index e1561dab8..94f6c0647 100644 --- a/tests/config/test_environment_variables.py +++ b/tests/config/test_environment_variables.py @@ -167,7 +167,7 @@ def test_inventory_matches_reviewed_snapshot_counts(): EnvironmentVariableCategory.PUBLIC_OMNI: 29, EnvironmentVariableCategory.INHERITED_VLLM: 20, EnvironmentVariableCategory.PLATFORM_EXTERNAL: 27, - EnvironmentVariableCategory.MODEL_SPECIFIC: 57, + EnvironmentVariableCategory.MODEL_SPECIFIC: 59, EnvironmentVariableCategory.BENCHMARK_TRANSITIONAL: 21, EnvironmentVariableCategory.INTERNAL: 2, } @@ -181,7 +181,7 @@ def test_inventory_matches_reviewed_snapshot_counts(): ModelEnvironmentVariableDisposition.PROMOTE: 31, ModelEnvironmentVariableDisposition.REQUEST_SCOPE: 5, ModelEnvironmentVariableDisposition.EXTERNAL: 0, - ModelEnvironmentVariableDisposition.INTERNALIZE: 14, + ModelEnvironmentVariableDisposition.INTERNALIZE: 16, ModelEnvironmentVariableDisposition.DEPRECATE_REMOVE: 7, } diff --git a/tests/engine/duplex/test_aura_resumable_hooks.py b/tests/engine/duplex/test_aura_resumable_hooks.py new file mode 100644 index 000000000..e2ed43372 --- /dev/null +++ b/tests/engine/duplex/test_aura_resumable_hooks.py @@ -0,0 +1,725 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Public duplex hook: resumable defaults True; ephemeral ids when False.""" + +from __future__ import annotations + +import pytest +from vllm.sampling_params import SamplingParams + +from vllm_omni.engine.duplex.config import DuplexCapabilities +from vllm_omni.engine.duplex.contracts import ( + DuplexFence, + DuplexStageRequestContext, + DuplexStageSubmission, + duplex_ephemeral_stage_request_id, +) +from vllm_omni.engine.duplex.session.manager import DuplexSessionManager + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def test_submission_resumable_default_preserves_minicpm() -> None: + ctx = DuplexStageRequestContext( + request_id="r", + session_id="s", + fence=DuplexFence("s"), + stage_id=0, + final_stage_id=2, + config_generation=0, + sampling_params=(SamplingParams(max_tokens=1),), + ) + submission = DuplexStageSubmission( + context=ctx, + prompt={"prompt_token_ids": [1, 2]}, + already_submitted=False, + ) + assert submission.resumable is True + + +def test_stage_request_id_respects_resumable_flag() -> None: + fence = DuplexFence("sess", epoch=3, turn_id=9) + resumable_id = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=True) + ephemeral_id = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=False) + assert ephemeral_id == duplex_ephemeral_stage_request_id(fence, stage_id=0) + assert resumable_id != ephemeral_id + assert ephemeral_id.endswith("stage0-turn9") + from vllm_omni.engine.duplex.contracts import duplex_turn_id_from_request_id + + assert duplex_turn_id_from_request_id(ephemeral_id) == 9 + assert duplex_turn_id_from_request_id(resumable_id) is None + + +def test_capabilities_expose_concurrent_turn_requests_default_false() -> None: + caps = DuplexCapabilities() + assert caps.supports_concurrent_turn_requests is False + assert caps.allows_video_without_audio() is False + assert caps.required_input_modalities == frozenset({"audio"}) + assert caps.optional_input_modalities == frozenset({"video"}) + assert caps.supports_core_resumable_request is False + payload = caps.as_dict() + assert payload["supports_concurrent_turn_requests"] is False + assert payload["required_input_modalities"] == ["audio"] + assert payload["optional_input_modalities"] == ["video"] + + +def test_next_commit_allowed_soft_opens_on_r4_release() -> None: + from types import SimpleNamespace + + from vllm_omni.engine.duplex.session import helpers + + caps = DuplexCapabilities(supports_concurrent_turn_requests=True) + session = SimpleNamespace(active_response_id="resp", capabilities=caps) + tasks = SimpleNamespace( + active_response_task=None, + has_response_bound_append_tasks=lambda: False, + ) + # Monkeypatch assistant_playback_active via response_in_progress path: active_response_id set. + assert helpers.response_in_progress(session, tasks) is True + assert helpers.next_commit_allowed(session, tasks, concurrent_turn_requests_released=False) is False + assert helpers.next_commit_allowed(session, tasks, concurrent_turn_requests_released=True) is True + + caps_off = DuplexCapabilities(supports_concurrent_turn_requests=False) + session_off = SimpleNamespace(active_response_id="resp", capabilities=caps_off) + assert helpers.next_commit_allowed(session_off, tasks, concurrent_turn_requests_released=True) is False + + +def test_active_response_accepts_own_turn_when_concurrent_turn_requests() -> None: + """Approach A: each response owns its turn; draining TTS uses request→response map.""" + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + + session = DuplexEngineSession( + session_id="s", + config=DuplexSessionConfig(model="m", modalities=["text"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True), + ) + session.begin_response(turn_id=3) + assert session.active_response_accepts_model_turn(3) + assert not session.active_response_accepts_model_turn(4) + assert not session.active_response_accepts_model_turn(2) + session.bind_draining_request("req-old", "resp-old") + assert session.response_id_for_request("req-old") == "resp-old" + assert session.is_draining_request("req-old") + + # Ending the active response must not wipe older draining bindings. + session.bind_draining_request("req-newer-tts", session.active_response_id or "resp-active") + ended = session.active_response_id + session.end_response(commit_text=False) + assert session.is_draining_request("req-old") + assert session.response_id_for_request("req-old") == "resp-old" + assert not session.is_draining_request("req-newer-tts") + session.clear_draining_for_response("resp-old") + assert not session.is_draining_request("req-old") + del ended + + strict = DuplexEngineSession( + session_id="s2", + config=DuplexSessionConfig(model="m", modalities=["text"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=False), + ) + strict.begin_response(turn_id=3) + assert strict.active_response_accepts_model_turn(3) + assert not strict.active_response_accepts_model_turn(4) + + +def test_draining_request_exempt_from_completed_turn_filter() -> None: + """Draining ownership must win over the model_turn_id < turn_id late-audio guard.""" + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + + session = DuplexEngineSession( + session_id="s-drain", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + assert r1 is not None + session.bind_draining_request("req-r1-tts", r1) + # Overlapping R2 becomes active, then finishes — must not wipe R1 draining. + session.begin_response(turn_id=2) + session.end_response(commit_text=False) + session.turn_id = 2 + assert session.active_response_id is None + assert session.is_draining_request("req-r1-tts") + assert session.response_id_for_request("req-r1-tts") == r1 + draining_response_id = session.response_id_for_request("req-r1-tts") + model_turn_id = 1 + drop = ( + draining_response_id is None + and session.active_response_id is None + and model_turn_id is not None + and model_turn_id < session.turn_id + ) + assert drop is False + + +def test_ephemeral_turn_id_parser_accepts_stage_turn_ids() -> None: + from vllm_omni.engine.duplex.contracts import duplex_turn_id_from_request_id + + assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0-turn12") == 12 + assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage2-turn3") == 3 + assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0_t9") is None + assert duplex_turn_id_from_request_id("duplex-s.x.e.0.r.stage0") is None + + +def test_same_turn_request_ids_skip_other_turns_and_epochs() -> None: + from vllm_omni.engine.duplex.contracts import duplex_same_turn_request_ids + + stage3 = "duplex-s.x.e.0.r.stage3-turn4" + found = duplex_same_turn_request_ids( + stage3, + [ + "duplex-s.x.e.0.r.stage0-turn4", + "duplex-s.x.e.0.r.stage1-turn4", + "duplex-s.x.e.0.r.stage2-turn4", + stage3, + "duplex-s.x.e.0.r.stage2-turn5", + "duplex-s.x.e.1.r.stage0-turn4", + ], + ) + assert found == [ + "duplex-s.x.e.0.r.stage0-turn4", + "duplex-s.x.e.0.r.stage1-turn4", + "duplex-s.x.e.0.r.stage2-turn4", + ] + from vllm_omni.engine.duplex.contracts import duplex_session_id_from_request_id + + fence_session = "sess id" + import base64 + + encoded = base64.urlsafe_b64encode(fence_session.encode()).decode("ascii").rstrip("=") + assert duplex_session_id_from_request_id(f"duplex-s.{encoded}.e.0.r.stage1-turn2") == fence_session + assert duplex_session_id_from_request_id("not-a-request") is None + + +def test_stale_keys_skip_already_draining_request_ids() -> None: + """Rebinding must not move an older draining Stage2/3 id onto the newer response.""" + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + + session = DuplexEngineSession( + session_id="s-stale", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + assert r1 is not None + session.bind_draining_request("req-r1-talker", r1) + session.begin_response(turn_id=2) + r2 = session.active_response_id + assert r2 is not None and r2 != r1 + # Simulate the concurrent-turn rebind loop: already-draining ids keep R1. + for rid in ("req-r1-talker", "req-r2-talker"): + if session.is_draining_request(rid): + continue + session.bind_draining_request(rid, r2) + assert session.response_id_for_request("req-r1-talker") == r1 + assert session.response_id_for_request("req-r2-talker") == r2 + + +def test_end_response_clears_only_own_draining_entries() -> None: + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + + session = DuplexEngineSession( + session_id="s-end", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + session.bind_draining_request("req-r1", r1) + session.begin_response(turn_id=2) + r2 = session.active_response_id + session.bind_draining_request("req-r2", r2) + session.end_response(commit_text=False) + assert session.is_draining_request("req-r1") + assert session.response_id_for_request("req-r1") == r1 + assert not session.is_draining_request("req-r2") + + +def test_on_stage_failure_resolves_draining_response_before_active() -> None: + """Mirrors runner.on_stage_failure: draining request_id owns the failed done.""" + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + + session = DuplexEngineSession( + session_id="s-fail", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + assert r1 is not None + session.bind_draining_request("req-r1-talker", r1) + session.begin_response(turn_id=2) + r2 = session.active_response_id + assert r2 is not None and r2 != r1 + + request_id = "req-r1-talker" + draining_response_id = ( + session.response_id_for_request(request_id) if session.is_draining_request(request_id) else None + ) + response_id = draining_response_id or session.active_response_id + assert response_id == r1 + assert response_id != r2 + session.clear_draining_for_response(r1) + assert not session.is_draining_request(request_id) + assert session.active_response_id == r2 + + +def test_concurrent_turn_requests_released_resets_when_new_stage0_binds() -> None: + """After R4 release, a new ephemeral Stage0 bind must clear the gate flag.""" + from types import SimpleNamespace + + run = SimpleNamespace(concurrent_turn_requests_released=True) + # Simulate the concurrent-turn rebind arm in model_channel._append_via_data_plane. + concurrent_turn = True + if concurrent_turn: + run.concurrent_turn_requests_released = False + assert run.concurrent_turn_requests_released is False + + +def test_draining_stage_ids_come_from_the_plugin() -> None: + from types import SimpleNamespace + + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + channel = ModelChannel.__new__(ModelChannel) + channel._ctx = SimpleNamespace( + plugin=SimpleNamespace(draining_stage_ids=lambda *, stage_count: {stage_count - 1}), + stage_port=SimpleNamespace(stage_count=5), + ) + assert channel._draining_stage_ids() == frozenset({4}) + + channel._ctx.plugin = SimpleNamespace() + assert channel._draining_stage_ids() == frozenset() + + +def test_draining_empty_eos_completes_owning_response_before_shortcut() -> None: + """R2 finishing first must not swallow R1's empty Stage3 EOS.""" + import asyncio + from types import SimpleNamespace + + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + session = DuplexEngineSession( + session_id="s-empty-eos", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + assert r1 is not None + r1_request = "duplex-s.x.e.0.r.stage3-turn1" + still_running = "duplex-s.x.e.0.r.stage1-turn2" + session.bind_draining_request(r1_request, r1) + session.begin_response(turn_id=2) + session.end_response(commit_text=False) + session.turn_id = 2 + assert session.active_response_id is None + + class _Plane: + def is_terminal(self, request_id: str | None) -> bool: + del request_id + return False + + def close_stream(self, request_id: str) -> None: + del request_id + + def mark_terminal(self, request_id: str) -> None: + del request_id + + class _Port: + def __init__(self) -> None: + self.cleanups: list[list[str]] = [] + + async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None: + del abort + self.cleanups.append(list(request_ids)) + + class _Out: + def __init__(self) -> None: + self.events: list[dict[str, object]] = [] + + def auto_responds(self) -> bool: + return True + + def emit(self, payload: dict[str, object]) -> None: + self.events.append(payload) + + port = _Port() + out = _Out() + + async def _noop(*_args: object, **_kwargs: object) -> None: + return None + + async def _schedule(*_args: object, **_kwargs: object) -> bool: + return False + + channel = ModelChannel( + SimpleNamespace( + session=session, + plugin=SimpleNamespace(data_plane=_Plane()), + stage_port=port, + model_state=SimpleNamespace(clear_continuation=lambda: None), + services=SimpleNamespace(spawn=lambda *_a, **_k: None), + ), + out, + close_from_runtime=_noop, + schedule_silence_continuation=_schedule, + abort_request=_noop, + ) + asyncio.run( + channel._send_one_model_output_event( + { + "data_plane_request_id": r1_request, + "end_of_turn": True, + "model_turn_id": 1, + "text": "", + "stage_role": "tts", + } + ) + ) + done = [event for event in out.events if event.get("type") == "response.done"] + assert done and done[-1].get("response_id") == r1 + assert port.cleanups == [[r1_request]] + assert all(still_running not in ids for ids in port.cleanups) + assert not session.is_draining_request(r1_request) + + +def test_draining_completion_drops_finished_response_books_only() -> None: + """R1 drain finish drops R1 books and leaves the live R2 snapshot.""" + import asyncio + from types import SimpleNamespace + + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + session = DuplexEngineSession( + session_id="s-drain-books", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False), + ) + session.begin_response(turn_id=1) + r1 = session.active_response_id + assert r1 is not None + session.append_assistant_text("from-r1") + session.snapshot_active_response_for_drain() + r1_request = "duplex-s.x.e.0.r.stage3-turn1" + session.bind_draining_request(r1_request, r1) + session.begin_response(turn_id=2) + r2 = session.active_response_id + assert r2 is not None and r2 != r1 + session.append_assistant_text("from-r2") + session.snapshot_active_response_for_drain() + + class _Plane: + def is_terminal(self, request_id: str | None) -> bool: + del request_id + return False + + def close_stream(self, request_id: str) -> None: + del request_id + + def mark_terminal(self, request_id: str) -> None: + del request_id + + class _Port: + async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None: + del request_ids, abort + + class _Out: + def __init__(self) -> None: + self.events: list[dict[str, object]] = [] + + def auto_responds(self) -> bool: + return True + + def emit(self, payload: dict[str, object]) -> None: + self.events.append(payload) + + out = _Out() + + async def _noop(*_args: object, **_kwargs: object) -> None: + return None + + async def _schedule(*_args: object, **_kwargs: object) -> bool: + return False + + channel = ModelChannel( + SimpleNamespace( + session=session, + plugin=SimpleNamespace(data_plane=_Plane()), + stage_port=_Port(), + model_state=SimpleNamespace(clear_continuation=lambda: None), + services=SimpleNamespace(spawn=lambda *_a, **_k: None), + ), + out, + close_from_runtime=_noop, + schedule_silence_continuation=_schedule, + abort_request=_noop, + ) + asyncio.run( + channel._send_one_model_output_event( + { + "data_plane_request_id": r1_request, + "end_of_turn": True, + "model_turn_id": 1, + "text": "", + "stage_role": "tts", + } + ) + ) + assert r1 not in session._conversation.assistant_response_snapshots + assert r1 not in session._playback.by_response + assert f"item_{r1}" not in session._conversation.history_item_placeholders + assert r2 in session._conversation.assistant_response_snapshots + assert session.active_response_id == r2 + done = [event for event in out.events if event.get("type") == "response.done"] + assert done and done[-1].get("response_id") == r1 + + +def test_stale_continue_does_not_close_the_new_response() -> None: + import asyncio + from types import SimpleNamespace + + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + session = DuplexEngineSession( + session_id="s-stale-continue", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"], extra_body={"auto_response": True}), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False), + ) + session.begin_response(turn_id=2) + live = session.active_response_id + session.epoch = 1 + + class _Out: + def __init__(self) -> None: + self.events: list[dict[str, object]] = [] + + def auto_responds(self) -> bool: + return True + + def emit(self, payload: dict[str, object]) -> None: + self.events.append(payload) + + out = _Out() + cleared: list[bool] = [] + + async def _noop(*_args: object, **_kwargs: object) -> None: + return None + + async def _schedule(*_args: object, **_kwargs: object) -> bool: + return False + + channel = ModelChannel( + SimpleNamespace( + session=session, + plugin=SimpleNamespace(data_plane=SimpleNamespace()), + stage_port=SimpleNamespace(), + model_state=SimpleNamespace(clear_continuation=lambda: cleared.append(True)), + services=SimpleNamespace(spawn=lambda *_a, **_k: None), + run=SimpleNamespace(closing=False), + ), + out, + close_from_runtime=_noop, + schedule_silence_continuation=_schedule, + abort_request=_noop, + ) + asyncio.run(channel.maybe_continue_response(expected_epoch=0)) + assert session.active_response_id == live + assert out.events == [] + assert cleared == [] + + session.bind_draining_request("duplex-s.x.e.1.r.stage3-turn2", live or "") + asyncio.run(channel.maybe_continue_response(expected_epoch=1)) + assert session.active_response_id == live + assert out.events == [] + + session.clear_draining_requests() + asyncio.run(channel.maybe_continue_response(expected_epoch=1)) + assert session.active_response_id is None + assert [event.get("type") for event in out.events] == ["response.done"] + assert out.events[0].get("response_id") == live + + +def test_silent_listen_with_continuation_releases_ephemeral_request() -> None: + import asyncio + from collections.abc import Coroutine + from types import SimpleNamespace + from typing import Any + + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + session = DuplexEngineSession( + session_id="s-silent-continue", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False), + ) + session.begin_response(turn_id=1) + response_id = session.active_response_id + request_id = "duplex-s.x.e.0.r.stage1-turn1" + session.bind_request(request_id) + + class _Plane: + def __init__(self) -> None: + self.terminal: list[str] = [] + + def is_terminal(self, request_id: str | None) -> bool: + return request_id in self.terminal + + def mark_terminal(self, request_id: str) -> None: + self.terminal.append(request_id) + + class _Port: + def __init__(self) -> None: + self.cleanups: list[list[str]] = [] + + async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None: + del abort + self.cleanups.append(list(request_ids)) + + class _Out: + def __init__(self) -> None: + self.events: list[dict[str, object]] = [] + + def auto_responds(self) -> bool: + return False + + def emit(self, payload: dict[str, object]) -> None: + self.events.append(payload) + + port = _Port() + out = _Out() + spawned: list[Coroutine[Any, Any, None]] = [] + + async def _noop(*_args: object, **_kwargs: object) -> None: + return None + + async def _schedule(*_args: object, **_kwargs: object) -> bool: + return False + + channel = ModelChannel( + SimpleNamespace( + session=session, + plugin=SimpleNamespace(data_plane=_Plane()), + stage_port=port, + model_state=SimpleNamespace( + continuation_owner_id=None, continuation_units=0, clear_continuation=lambda: None + ), + services=SimpleNamespace(spawn=lambda coro, **_k: spawned.append(coro)), + run=SimpleNamespace(closing=False), + ), + out, + close_from_runtime=_noop, + schedule_silence_continuation=_schedule, + abort_request=_noop, + ) + asyncio.run( + channel._send_one_model_output_event( + { + "data_plane_request_id": request_id, + "is_listen": True, + "end_of_turn": True, + "model_turn_id": 1, + "reason": "model_listen", + } + ) + ) + assert port.cleanups == [[request_id]] + assert session.active_request_id is None + assert spawned, "silent listen with unused continuation must schedule maybe_continue" + asyncio.run(spawned[0]) + assert port.cleanups == [[request_id]] + assert session.active_response_id is None + assert [event.get("type") for event in out.events] == ["response.listen", "response.done"] + assert out.events[0].get("response_id") == response_id + assert out.events[1].get("response_id") == response_id + + +def test_silent_listen_aborts_data_plane_request_with_full_id() -> None: + """abort_data_plane_request must pass [request_id], not a bare str.""" + import asyncio + from types import SimpleNamespace + + from vllm_omni.engine.duplex.config import DuplexCapabilities, DuplexSessionConfig + from vllm_omni.engine.duplex.session.engine_session import DuplexEngineSession + from vllm_omni.engine.duplex.session.model_channel import ModelChannel + + session = DuplexEngineSession( + session_id="s-silent-abort", + config=DuplexSessionConfig(model="m", modalities=["text", "audio"]), + capabilities=DuplexCapabilities(supports_concurrent_turn_requests=True, supports_core_resumable_request=False), + ) + request_id = "duplex-s.x.e.0.r.stage0-turn7" + session.bind_request(request_id) + aborts: list[list[str]] = [] + + class _Plane: + def is_terminal(self, request_id: str | None) -> bool: + return False + + def mark_terminal(self, request_id: str) -> None: + del request_id + + class _Port: + async def cleanup(self, request_ids: list[str], *, abort: bool = False) -> None: + del request_ids, abort + + class _Out: + def auto_responds(self) -> bool: + return True + + def emit(self, payload: dict[str, object]) -> None: + del payload + + async def _noop(*_args: object, **_kwargs: object) -> None: + return None + + async def _abort(request_ids: list[str], *, notify: bool = False) -> None: + del notify + aborts.append(list(request_ids)) + + channel = ModelChannel( + SimpleNamespace( + session=session, + plugin=SimpleNamespace(data_plane=_Plane()), + stage_port=_Port(), + model_state=SimpleNamespace( + continuation_owner_id=None, continuation_units=0, clear_continuation=lambda: None + ), + services=SimpleNamespace(spawn=lambda coro, **_k: None), + run=SimpleNamespace(closing=False), + ), + _Out(), + close_from_runtime=_noop, + schedule_silence_continuation=_noop, + abort_request=_abort, + ) + asyncio.run( + channel._send_one_model_output_event( + { + "data_plane_request_id": request_id, + "is_listen": True, + "end_of_turn": True, + "model_turn_id": 1, + "reason": "model_listen", + "abort_data_plane_request": True, + } + ) + ) + assert aborts == [[request_id]] + assert all(len(item) > 1 for batch in aborts for item in batch), "must not split request id into chars" diff --git a/tests/engine/duplex/test_engine_session.py b/tests/engine/duplex/test_engine_session.py index 5ef4ab6b9..358ee2d9c 100644 --- a/tests/engine/duplex/test_engine_session.py +++ b/tests/engine/duplex/test_engine_session.py @@ -242,6 +242,19 @@ def test_history_commit_with_ack_only_policy_defers_unacknowledged_text(): assert f"item_{response_id}" in session.pending_history_item_ids +def test_history_commit_with_ack_only_policy_keeps_only_acknowledged_prefix(): + session = _session(config=DuplexSessionConfig(playback_commit_policy="ack_only")) + session.begin_response() + session.append_assistant_text("hello world") + session.mark_audio_sent(duration_ms=1_000, text_chars=6) + session.mark_audio_sent(duration_ms=2_000, text_chars=11) + session.acknowledge_playback(played_ms=1_000, committed_ms=1_000) + + committed = session.end_response(commit_text=True) + + assert committed == {"role": "assistant", "content": "hello"} + + @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")) @@ -469,6 +482,36 @@ def test_cancel_fence_releases_stage_requests_and_advances_identity(): session.cancel_fence(DuplexFence("sid-other", epoch=1, turn_id=0), DuplexFence("sid-cancel", epoch=2)) +def test_request_resource_keys_are_stage_id_and_request_id(): + session = _session("sid-keys") + session.bind_stage_request(0, "req-a", fence=session.fence) + session.bind_stage_request(1, "req-b", fence=session.fence) + session.bind_stage_request(2, "req-tts", fence=session.fence) + stale_keys = list(session.request_resources.keys()) + stale_ids = list(dict.fromkeys(rid for _, rid in stale_keys)) + assert stale_ids == ["req-a", "req-b", "req-tts"] + for sid, rid in stale_keys: + if sid < 2: + session.request_resources.pop((sid, rid), None) + assert session.resource_request_ids() == ["req-tts"] + with pytest.raises(TypeError, match="unhashable"): + dict.fromkeys(rid for _, rid in session.request_resources.items()) + + +def test_release_resources_for_request_ids_keeps_other_ids_on_the_same_fence(): + session = _session("sid-drain-release") + fence = session.fence + session.bind_stage_request(2, "drain", fence=fence) + session.bind_stage_request(3, "drain", fence=fence) + session.bind_stage_request(0, "live", fence=fence) + + released = session.release_resources_for_request_ids(["drain"]) + + assert released == ["drain"] + assert session.resource_request_ids() == ["live"] + assert session.release_resources_for_request_ids([]) == [] + + # ---- public view ---- @@ -688,3 +731,65 @@ def test_minicpmo_native_capabilities_do_not_overclaim_single_session_deployment assert caps["supports_multi_session"] is False assert caps["supports_multi_session_same_replica"] is False + + +def test_draining_response_stays_ack_admissible_after_next_begin_response() -> None: + from vllm_omni.engine.duplex.events import ErrorEvent + from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack + + session = _session() + first = session.begin_response(turn_id=1) + session.append_assistant_text("hello") + session.mark_audio_sent(400, text_chars=5) + session.snapshot_active_response_for_drain() + second = session.begin_response(turn_id=2) + assert second != first + assert session.playback.sent_ms == 0 + events = apply_playback_ack( + session, + { + "type": "playback.ack", + "response_id": first, + "item_id": f"item_{first}", + "played_ms": 100, + "committed_ms": 100, + }, + ) + assert not any(isinstance(event, ErrorEvent) and event.code == "playback_item_not_found" for event in events) + session.mark_audio_sent(900, text_chars=5, response_id=first) + assert session.playback_for_response(first).sent_ms == 900 + assert session.playback.sent_ms == 0 + + +def test_finished_drain_keeps_sent_audio_ackable() -> None: + from vllm_omni.engine.duplex.events import ErrorEvent + from vllm_omni.engine.duplex.session.playback_ledger import apply_playback_ack + + session = _session() + first = session.begin_response(turn_id=1) + session.append_assistant_text("hello") + session.mark_audio_sent(400, text_chars=5) + session.snapshot_active_response_for_drain() + second = session.begin_response(turn_id=2) + session.release_finished_drain_response(first) + assert first in session._conversation.assistant_response_snapshots + assert f"item_{first}" in session._conversation.history_item_placeholders + events = apply_playback_ack( + session, + { + "type": "playback.ack", + "response_id": first, + "item_id": f"item_{first}", + "played_ms": 400, + "committed_ms": 400, + }, + ) + assert not any(isinstance(event, ErrorEvent) for event in events) + assert session.active_response_id == second + + silent = session.begin_response(turn_id=3) + session.snapshot_active_response_for_drain() + session.begin_response(turn_id=4) + session.release_finished_drain_response(silent) + assert silent not in session._conversation.assistant_response_snapshots + assert f"item_{silent}" not in session._conversation.history_item_placeholders diff --git a/tests/engine/duplex/test_input_transcribed.py b/tests/engine/duplex/test_input_transcribed.py new file mode 100644 index 000000000..98914b899 --- /dev/null +++ b/tests/engine/duplex/test_input_transcribed.py @@ -0,0 +1,49 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import pytest + +from vllm_omni.engine.duplex.realtime_events import ( + RealtimeProjectionState, + project_internal_event, +) + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def test_input_transcribed_is_a_user_transcript_event() -> None: + events = project_internal_event( + RealtimeProjectionState(session_id="s"), + {"type": "input.transcribed", "transcript": " 你好 "}, + ) + assert len(events) == 1 + assert events[0].wire_type == "conversation.item.input_audio_transcription.completed" + assert events[0].transcript == "你好" + + +def test_input_transcribed_updates_the_committed_item() -> None: + state = RealtimeProjectionState(session_id="s") + project_internal_event( + state, + { + "type": "input.committed", + "realtime_item_id": "item_user_1", + "is_speech": True, + "message": {"role": "user"}, + }, + ) + events = project_internal_event( + state, + {"type": "input.transcribed", "transcript": "你好", "realtime_item_id": "item_user_1"}, + ) + assert events[0].item_id == "item_user_1" + item = state.conversation_items["item_user_1"] + assert item["content"][0]["transcript"] == "你好" + + +def test_blank_input_transcribed_emits_nothing() -> None: + events = project_internal_event( + RealtimeProjectionState(session_id="s"), + {"type": "input.transcribed", "transcript": " "}, + ) + assert events == [] diff --git a/tests/engine/duplex/test_overlap_policy.py b/tests/engine/duplex/test_overlap_policy.py index 096b7c7c9..61840e2bb 100644 --- a/tests/engine/duplex/test_overlap_policy.py +++ b/tests/engine/duplex/test_overlap_policy.py @@ -79,6 +79,39 @@ def test_an_explicit_client_overlap_action_requests_barge_in() -> None: assert decision["reason"] == "client_overlap_action" +def test_turn_mode_keeps_silent_vision_follow_during_overlap() -> None: + """Vision-follow is is_speech=False + a frame; must not drop while TTS plays.""" + session = _session() + session.capabilities = type(session.capabilities)( + required_input_modalities=frozenset({"video"}), + optional_input_modalities=frozenset({"audio"}), + ) + payload = {"format": "pcm16", "audio": _pcm16(0.0), "video_frames": ["frame"], "is_speech": False} + decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False) + assert decision["action"] == "listen" + assert decision["reason"] == "vision_follow" + assert decision["buffer_audio"] is True + assert decision["defer_runtime_append"] is False + assert decision["force_listen"] is False + + +def test_turn_mode_drops_silent_vision_without_capability() -> None: + session = _session() + assert session.capabilities.allows_video_without_audio() is False + payload = {"format": "pcm16", "audio": _pcm16(0.0), "video_frames": ["frame"], "is_speech": False} + decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False) + assert decision["action"] == "drop" + assert decision["reason"] == "silence_or_noise" + + +def test_turn_mode_still_drops_silent_overlap_without_vision() -> None: + session = _session() + payload = {"format": "pcm16", "audio": _pcm16(0.0), "is_speech": False} + decision = overlap_policy.decide(session, {"is_speech": False}, payload, auto_responds=False) + assert decision["action"] == "drop" + assert decision["reason"] == "silence_or_noise" + + def test_barge_in_is_deferred_when_the_session_cannot_support_it() -> None: session = _session() session.capabilities = type(session.capabilities)(supports_barge_in=False) diff --git a/tests/engine/duplex/test_session_manager.py b/tests/engine/duplex/test_session_manager.py index bccce18f1..cb00737d9 100644 --- a/tests/engine/duplex/test_session_manager.py +++ b/tests/engine/duplex/test_session_manager.py @@ -196,7 +196,11 @@ class FakePlugin(DuplexModelPlugin): def capabilities(self, *, max_sessions: int) -> DuplexCapabilities: del max_sessions - return DuplexCapabilities(supports_input_append=True, supports_core_resumable_request=True) + # Resident Stage0 ids (MiniCPM-shaped); AURA opts out via supports_core_resumable_request=False. + return DuplexCapabilities( + supports_input_append=True, + supports_core_resumable_request=True, + ) def validate_client_extra_body(self, extra_body: object) -> None: pass @@ -409,7 +413,8 @@ async def test_open_answers_with_capabilities_and_emits_session_created() -> Non assert result.session_id == "sid-open" assert result.lease_generation == 0 assert result.capabilities == DuplexCapabilities( - supports_input_append=True, supports_core_resumable_request=True + supports_input_append=True, + supports_core_resumable_request=True, ) assert result.public_session is not None assert result.public_session["id"] == "sid-open" @@ -1028,6 +1033,40 @@ async def test_append_bytes_are_reserved_at_admission_until_the_runner_dequeues( gate.set() +async def test_append_admission_counts_audio_and_video_frame_bytes() -> None: + """Manager reserves len(audio)+Σlen(frame); video bytes count toward the same limit.""" + async with Harness.create(max_sessions=1, max_pending_input_bytes_per_session=20) as harness: + await harness.open("sid-av") + harness.events() + session = harness.session("sid-av") + runner = harness.manager.runners["sid-av"] + gate = asyncio.Event() + runner._mailbox.put_nowait(_Internal("wait", {"gate": gate})) + runner._on_internal = lambda item: gate.wait() # type: ignore[method-assign] + await asyncio.sleep(0) + + # Default caps require audio; attach video to a non-empty audio unit. + frame_a = "aaaa" + frame_b = "bbbbbb" + audio = b"1234" + expected = len(audio) + len(frame_a) + len(frame_b) + harness.command( + "sid-av", + AppendAudio(audio=audio, video_frames=(frame_a, frame_b), event_id="evt-av"), + ) + assert session.pending_input_bytes == expected + # Second append that would exceed the limit is backpressured. + harness.command( + "sid-av", + AppendAudio(audio=b"x" * 10, video_frames=("yyyyyyyyyy",), event_id="evt-over"), + ) + errors = [event for event in harness.events("sid-av") if isinstance(event, ErrorEvent)] + assert [error.code for error in errors] == ["input_backpressure"] + assert errors[0].related_event_id == "evt-over" + assert session.pending_input_bytes == expected + gate.set() + + async def test_expired_session_retains_the_admission_slot_until_cleanup_succeeds() -> None: async with Harness.create(max_sessions=1, idle_ttl_s=1.0) as harness: await harness.open("sid-expired") diff --git a/tests/engine/duplex/test_session_runner.py b/tests/engine/duplex/test_session_runner.py index 51711f52a..f48154a48 100644 --- a/tests/engine/duplex/test_session_runner.py +++ b/tests/engine/duplex/test_session_runner.py @@ -180,7 +180,10 @@ class Harness: segment_token_ids=tuple(segment_token_ids), segment_output_metadata=dict(segment_output_metadata or {}), ) - return self.runner.on_stage_output(stage_id, output, metrics, request_id=output.request_id, context=context) + request_id = getattr(output, "request_id", None) + if not isinstance(request_id, str): + raise AssertionError("stage output is missing request_id") + return self.runner.on_stage_output(stage_id, output, metrics, request_id=request_id, context=context) async def deliver_and_settle(self, output: object, **kwargs: Any) -> list[DuplexEvent]: self.deliver(output, **kwargs) @@ -520,6 +523,28 @@ async def test_commit_without_audio_is_rejected() -> None: await close_harness(h) +@pytest.mark.asyncio +async def test_empty_silent_commit_acks_without_opening_response() -> None: + """#3: empty is_speech=False commit keeps the silent-ack pair, even auto-respond.""" + h = await open_harness(auto_response=True) + try: + # resolve_commit only admits empty silent commits after non-speech was seen. + projector = h.runner._require_projector() + projector.input_audio_buffer_had_non_speech = True + events = await h.run(commands.Commit()) + assert "error" not in types(events), types(events) + assert "input_audio_buffer.committed" in types(events), types(events) + assert "response.listen" in types(events), types(events) + listen = next(event for event in events if event.type == "response.listen") + assert listen.details["reason"] == "silence_or_noise" + committed = next(event for event in events if event.type == "input_audio_buffer.committed") + assert committed.details.get("empty") is True + assert committed.details.get("no_response") is True + assert h.port.submissions == [] + finally: + await close_harness(h) + + @pytest.mark.asyncio async def test_commit_projects_the_user_item_and_does_not_resubmit_consumed_audio() -> None: h = await open_harness() @@ -674,6 +699,45 @@ async def test_stale_epoch_output_is_dropped_after_barge_in() -> None: await close_harness(h) +async def test_barge_in_aborts_draining_tts_as_well_as_the_active_request() -> None: + h = await open_harness() + try: + await h.run(append_audio()) + request_id = h.stage0_request_id() + await h.deliver_and_settle(tts_output(request_id, samples=24000, text="he")) + h.session.bind_draining_request("duplex-drain-tts", "resp-old") + # Draining TTS sits on an older turn fence. cancel_fence only drops + # the fence being cancelled, so these bindings must be popped by id. + from vllm_omni.engine.duplex.contracts import DuplexFence + from vllm_omni.engine.duplex.session.engine_session import DuplexRequestResource + + older = DuplexFence(h.session.session_id, epoch=h.session.epoch, turn_id=h.session.turn_id + 5) + h.session.request_resources[(2, "duplex-drain-tts")] = DuplexRequestResource( + stage_id=2, request_id="duplex-drain-tts", fence=older, submitted=True + ) + h.session.request_resources[(3, "duplex-drain-tts")] = DuplexRequestResource( + stage_id=3, request_id="duplex-drain-tts", fence=older, submitted=True + ) + h.session.request_resources[(1, "still-live")] = DuplexRequestResource( + stage_id=1, request_id="still-live", fence=older, submitted=True + ) + events = await h.run(commands.BargeIn()) + assert h.port.aborts == [[request_id, "duplex-drain-tts"]] + assert not h.session.is_draining_request("duplex-drain-tts") + done_ids = [ + event.response["id"] + for event in events + if getattr(event, "wire_type", "") == "response.done" and isinstance(getattr(event, "response", None), dict) + ] + assert "resp-old" in done_ids + assert (2, "duplex-drain-tts") not in h.session.request_resources + assert (3, "duplex-drain-tts") not in h.session.request_resources + assert (0, request_id) not in h.session.request_resources + assert (1, "still-live") in h.session.request_resources + finally: + await close_harness(h) + + @pytest.mark.asyncio async def test_cancel_response_without_active_response_is_rejected() -> None: h = await open_harness() @@ -778,6 +842,46 @@ async def test_listen_decision_on_a_resumable_request_closes_the_bounded_respons await close_harness(h) +@pytest.mark.asyncio +async def test_direct_response_listen_still_emits_response_done_after_continuation_clear() -> None: + """AURA-like silent DIRECT_RESPONSE: after continuation budget is spent, emit response.done. + + Non-resumable turn-commit still closes the active response on a listen/direct + decision so the client is not left without a terminal event. + """ + from dataclasses import replace + + h = await open_harness() + try: + h.session.capabilities = replace(h.session.capabilities, supports_core_resumable_request=False) + await h.run(append_audio()) + assert h.port.submissions, "ephemeral Stage0 must submit" + request_id = h.port.submissions[-1].context.request_id + assert "-turn" in request_id + await h.deliver_and_settle(tts_output(request_id, samples=24000, text="ok")) + response_id = h.session.active_response_id + assert response_id is not None + model_state = h.runner.model_state + model_state.continuation_owner_id = f"response:{response_id}" + model_state.continuation_units = h.runner.model._AUTO_RESPONSE_MAX_CONTINUATION_UNITS + + listen = listen_output(request_id) + listen.finished = True + events = await h.deliver_and_settle( + listen, + stage_id=0, + segment_finished=True, + segment_token_ids=[11, 12, LISTEN_TOKEN_ID], + segment_output_metadata={"meta.listen_token_id": LISTEN_TOKEN_ID}, + ) + assert "response.listen" in types(events) + done = find(events, "response.done") + assert done.response_id == response_id + assert h.session.active_response_id is None + finally: + await close_harness(h) + + @pytest.mark.asyncio async def test_listen_decision_on_a_resumable_request_keeps_the_turn_going() -> None: """Same unfinished listen, budget left: it must schedule the next unit. @@ -860,6 +964,38 @@ async def test_stage_failure_fails_the_active_response() -> None: await close_harness(h) +@pytest.mark.asyncio +async def test_stage_failure_on_draining_request_fails_that_response_only() -> None: + """Overlapped: a draining older request failure must not wipe the active response.""" + from dataclasses import replace + + h = await open_harness() + try: + h.session.capabilities = replace(h.session.capabilities, supports_concurrent_turn_requests=True) + await h.run(append_audio()) + request_id = h.stage0_request_id() + await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hello")) + r1 = h.session.active_response_id + assert r1 is not None + draining_req = "req-r1-talker-drain" + h.session.bind_draining_request(draining_req, r1) + # Open a newer active response while R1 TTS is still draining. + r2 = h.session.begin_response(turn_id=(h.session.turn_id or 0) + 1) + assert h.session.active_response_id == r2 + assert r2 != r1 + + h.runner.on_stage_failure(2, RuntimeError("drain-boom"), request_id=draining_req) + events = await h.settle() + assert types(events)[0] == "error" + done = find(events, "response.done") + assert done.response_id == r1 + assert done.status == "failed" + assert h.session.active_response_id == r2 + assert not h.session.is_draining_request(draining_req) + finally: + await close_harness(h) + + # --------------------------------------------------------------------------- # # Turn mode (no auto response) # # --------------------------------------------------------------------------- # @@ -877,6 +1013,91 @@ async def test_turn_mode_skips_silent_chunks() -> None: await close_harness(h) +@pytest.mark.asyncio +async def test_turn_mode_keeps_silent_chunks_with_video_frames() -> None: + """Engine vision-follow buffers silent+frames; MiniCPM commit stays speech-gated.""" + from dataclasses import replace + + h = await open_harness(auto_response=False) + try: + h.session.capabilities = replace( + h.session.capabilities, + required_input_modalities=frozenset({"video"}), + optional_input_modalities=frozenset({"audio"}), + ) + # Minimal 1x1 JPEG (base64) — wire validation only checks non-empty str. + frame = ( + "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkS" + "Ew8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJ" + "CQwLDBgNDRgyIRwhMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIy" + "MjIyMjIyMjIyMjIyMjL/wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAA" + "AAAAAAAAAAj/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/8QAFQEBAQAAAAAAAAAAAAAA" + "AAAAAAD/xAAUEQEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAGfAP/EABQQ" + "AQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAQUCf//EABQRAQAAAAAAAAAAAAAAAAAA" + "AAD/2gAIAQMBAT8Bf//EABQRAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQIBAT8Bf//E" + "ABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEABj8Cf//EABQQAQAAAAAAAAAAAAAA" + "AAAAAAD/2gAIAQEAAT8hf//Z" + ) + cmd = commands.AppendAudio( + audio=pcm_f32(160, value=0.0), + format="pcm_f32le", + sample_rate_hz=16000, + is_speech=False, + video_frames=(frame,), + ) + events = await h.run(cmd) + assert "response.listen" not in types(events) + assert h.runner.model_state.audio_buffer.has_pending() + events = await h.run(commands.Commit(create_response=True)) + # MiniCPM prepare_commit is speech-gated (no frames-only Stage0). + assert "input_audio_buffer.committed" in types(events), types(events) + assert h.port.submissions == [] + finally: + await close_harness(h) + + +@pytest.mark.asyncio +async def test_turn_mode_keeps_silent_vision_while_response_in_progress() -> None: + """TTS still playing: vision-follow must buffer, not drop as silence_or_noise.""" + from dataclasses import replace + + h = await open_harness(auto_response=False) + try: + h.session.capabilities = replace( + h.session.capabilities, + required_input_modalities=frozenset({"video"}), + optional_input_modalities=frozenset({"audio"}), + ) + h.session._response.active_response_id = "resp-tts" + frame = ( + "/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkS" + "Ew8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJ" + "CQwLDBgNDRgyIRwhMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIy" + "MjIyMjIyMjIyMjIyMjL/wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAA" + "AAAAAAAAAAj/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/8QAFQEBAQAAAAAAAAAAAAAA" + "AAAAAAD/xAAUEQEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIQAxAAAAGfAP/EABQQ" + "AQAAAAAAAAAAAAAAAAAAAAD/2gAIAQMBAT8Bf//EABQRAQAAAAAAAAAAAAAAAAAA" + "AAD/2gAIAQIBAT8Bf//EABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEABj8Cf//E" + "ABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAT8hf//Z" + ) + events = await h.run( + commands.AppendAudio( + audio=pcm_f32(160, value=0.0), + format="pcm_f32le", + sample_rate_hz=16000, + is_speech=False, + video_frames=(frame,), + ) + ) + listen_reasons = [ + getattr(event, "details", {}) or {} for event in events if getattr(event, "type", None) == "response.listen" + ] + assert all(details.get("reason") != "silence_or_noise" for details in listen_reasons) + assert h.runner.model_state.audio_buffer.has_pending() + finally: + await close_harness(h) + + @pytest.mark.asyncio async def test_turn_mode_commit_with_response_create_starts_one_response() -> None: h = await open_harness(auto_response=False) @@ -1033,7 +1254,7 @@ async def test_conversation_items_can_be_injected_and_deleted() -> None: await close_harness(h) -def _stage_metrics_of(event: object) -> dict[str, dict[str, object]]: +def _stage_metrics_of(event: DuplexEvent) -> dict[str, dict[str, object]]: """Per-stage engine metrics as the client reads them off one wire event.""" payload = event.to_realtime() metadata = payload.get("metadata") @@ -1081,6 +1302,40 @@ async def test_stage0_metrics_reach_the_response_even_though_its_output_feeds_tt await close_harness(h) +@pytest.mark.asyncio +async def test_projected_stage1_metrics_reach_the_spoken_audio_event() -> None: + """Thinker text is projected, but its TTFT/TPOT still have to ride the audio event.""" + h = await open_harness(stage_count=3) + try: + plugin = h.runner.model._ctx.plugin + + def project_stage1(*, stage_id: int, output: object, context: object) -> bool: + del output, context + return stage_id == 1 + + plugin.project_intermediate_output = project_stage1 + await h.run(append_audio()) + request_id = h.stage0_request_id() + thinker = SimpleNamespace( + request_id=request_id, + finished=True, + outputs=[SimpleNamespace(text="hello", token_ids=[7, 8], multimodal_output={})], + multimodal_output={}, + ) + forwarded = h.deliver( + thinker, + stage_id=1, + segment_finished=False, + metrics=stage_stats(stage_id=1, request_id=request_id, num_tokens_out=2, itls_ms=[6.0]), + ) + assert forwarded is False + events = await h.deliver_and_settle(tts_output(request_id, samples=24000, text="hello"), stage_id=2) + stage_metrics = _stage_metrics_of(find(events, "response.output_audio.delta")) + assert stage_metrics["1"]["num_tokens_out"] == 2 + finally: + await close_harness(h) + + @pytest.mark.asyncio async def test_stage0_metrics_from_several_units_are_summed_into_one_response() -> None: """Two pass-through segments before the first audio are one response's tokens.""" diff --git a/tests/engine/duplex/test_session_runner_ephemeral_bind.py b/tests/engine/duplex/test_session_runner_ephemeral_bind.py new file mode 100644 index 000000000..cbde18ec3 --- /dev/null +++ b/tests/engine/duplex/test_session_runner_ephemeral_bind.py @@ -0,0 +1,28 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Ephemeral Stage0 bind uses r.stage0-turn{N}, not resumable r.stage0.""" + +from __future__ import annotations + +from vllm_omni.engine.duplex.contracts import ( + DuplexFence, + duplex_ephemeral_stage_request_id, + duplex_resource_request_id, +) +from vllm_omni.engine.duplex.session.manager import DuplexSessionManager + + +def test_stage0_request_id_ephemeral_when_not_resumable() -> None: + fence = DuplexFence("duplex-test", epoch=0, turn_id=3) + rid = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=False) + expected = duplex_ephemeral_stage_request_id(fence, stage_id=0) + assert rid == expected + assert rid.endswith(".r.stage0-turn3") + assert not rid.endswith(".r.stage0") + + +def test_stage0_request_id_resumable_keeps_stable_role() -> None: + fence = DuplexFence("duplex-test", epoch=1, turn_id=3) + rid = DuplexSessionManager.stage_request_id(fence, stage_id=0, resumable=True) + expected = duplex_resource_request_id(fence, "stage0") + assert rid == expected + assert rid.endswith(".r.stage0") diff --git a/tests/engine/test_duplex_orchestrator.py b/tests/engine/test_duplex_orchestrator.py index 703af6a35..1bffda919 100644 --- a/tests/engine/test_duplex_orchestrator.py +++ b/tests/engine/test_duplex_orchestrator.py @@ -55,6 +55,7 @@ def _build( stages: int = 1, running_counter: FakeRunningCounter | None = None, runtime_config: DuplexSessionRuntimeConfig | None = None, + plugin=None, ) -> tuple[DuplexOrchestrator, list[FakeStageClient], asyncio.Queue, asyncio.Queue]: clients = [FakeStageClient(stage_type="llm", final_output=index == stages - 1) for index in range(stages)] pools = _build_stage_pools( @@ -70,7 +71,7 @@ def _build( rpc_async_queue=rpc_q, stage_pools=pools, running_counter=running_counter, - plugin=MiniCPMO45DuplexPlugin(_encode_audio), + plugin=plugin or MiniCPMO45DuplexPlugin(_encode_audio), duplex_session_config=runtime_config or DuplexSessionRuntimeConfig(reaper_interval_s=0.01), model_config=None, ) @@ -197,7 +198,8 @@ async def test_open_preregisters_the_stage0_request_and_close_releases_it() -> N assert request_state.session_owned is True assert request_state.session_id == SESSION_ID assert request_state.fence == DuplexFence(SESSION_ID) - assert request_state.streaming.enabled is True + # Set on submit from submission.resumable, not again at preregister. + assert request_state.streaming.enabled is False # Preregistration reserves the id; nothing is running until an append submits. assert counter.value == 0 assert clients[0].add_request_calls == [] @@ -211,6 +213,18 @@ async def test_open_preregisters_the_stage0_request_and_close_releases_it() -> N assert orchestrator.session_manager.active_count() == 0 +@pytest.mark.asyncio +async def test_aura_ephemeral_preregister_disables_streaming() -> None: + from vllm_omni.model_executor.models.aura_omni.duplex.plugin import AuraDuplexPlugin + + orchestrator, _, rpc_q, _ = _build(plugin=AuraDuplexPlugin(_encode_audio)) + result = await _open(orchestrator, rpc_q) + assert result.ok is True + request_state = next(iter(orchestrator.request_states.values())) + assert request_state.streaming.enabled is False + await orchestrator.session_manager.shutdown() + + @pytest.mark.asyncio async def test_open_failure_rolls_back_session_and_reserved_request() -> None: orchestrator, _, rpc_q, _ = _build() @@ -275,6 +289,7 @@ async def test_append_submits_the_resumable_stage0_request_and_counts_it_running assert submitted.resumable is True assert submitted.sampling_params.max_tokens == 20 request_state = orchestrator.request_states[request_id] + assert request_state.streaming.enabled is True assert request_state.stage_fences[0] == DuplexFence(SESSION_ID) assert 0 in request_state.stage_submit_ts assert counter.value == 1 @@ -390,6 +405,31 @@ async def test_forward_failure_closes_the_owning_session() -> None: assert types[-1] == "session.expired" +@pytest.mark.asyncio +async def test_aura_forward_failure_keeps_the_session() -> None: + from vllm_omni.model_executor.models.aura_omni.duplex.plugin import AuraDuplexPlugin + + orchestrator, clients, rpc_q, output_q = _build(stages=2, plugin=AuraDuplexPlugin(_encode_audio)) + await _open(orchestrator, rpc_q) + request_id = next(iter(orchestrator.request_states)) + await _submit(orchestrator, _append_audio()) + request_state = orchestrator.request_states[request_id] + session = orchestrator.session_manager.get(SESSION_ID) + assert session is not None + + absorbed = await orchestrator._handle_forward_failure(request_id, 1, request_state, ValueError("stale talker")) + await _settle(orchestrator) + + assert absorbed is True + assert request_id not in orchestrator.request_states + assert SESSION_ID in orchestrator.session_manager.runners + assert session.state != DuplexSessionState.CLOSED + types = [message.event.type for message in [output_q.get_nowait() for _ in range(output_q.qsize())]] + assert "error" in types + assert "session.expired" not in types + await orchestrator.session_manager.shutdown() + + @pytest.mark.asyncio async def test_request_cleanup_failure_is_deferred_and_retried_by_the_reaper(monkeypatch) -> None: orchestrator, clients, rpc_q, _ = _build() @@ -560,3 +600,53 @@ async def test_turn_plugin_processes_multimodal_prompt_before_stage_submission(m assert not orchestrator.request_states[submitted.request_id].streaming.enabled finally: await _close(orchestrator, rpc_q) + + +@pytest.mark.asyncio +async def test_sentence_partial_does_not_legacy_forward_the_full_stage_output() -> None: + from vllm_omni.engine.duplex.plugin import PartialStageForward + + orchestrator, *_ = _build(stages=4) + forwarded: list[object] = [] + + async def record_forward(req_id, stage_id, output, req_state, **kwargs): + del req_id, stage_id, req_state, kwargs + forwarded.append(output) + + async def no_intercept(*args, **kwargs): + del args, kwargs + return False + + orchestrator._forward_to_next_stage = record_forward # type: ignore[method-assign] + orchestrator._intercept_stage_output = no_intercept # type: ignore[method-assign] + orchestrator.async_chunk = True + orchestrator._stage_receives_async_chunks = lambda stage_id: False # type: ignore[method-assign] + + full = SimpleNamespace(request_id="r", finished=True, text="FULL") + chunk = SimpleNamespace(request_id="r", finished=True, text="CHUNK") + req_state = DuplexOrchestratorRequestState( + request_id="r", + final_stage_id=3, + session_owned=True, + sampling_params_list=[SimpleNamespace() for _ in range(4)], + ) + orchestrator.plugin.plan_partial_stage_output = lambda *args, **kwargs: PartialStageForward( # type: ignore[method-assign] + output=chunk, is_final_update=True + ) + orchestrator.plugin.partial_stage_followup = lambda *args, **kwargs: None # type: ignore[method-assign] + + await orchestrator._route_output(1, 0, full, req_state, None) + + assert forwarded == [chunk] + assert req_state.skip_legacy_stage_forward is False + + forwarded.clear() + orchestrator.plugin.plan_partial_stage_output = lambda *args, **kwargs: None # type: ignore[method-assign] + await orchestrator._route_output(1, 0, full, req_state, None) + + assert forwarded == [full] + + +def test_default_plugin_declares_no_draining_stages() -> None: + plugin = MiniCPMO45DuplexPlugin(_encode_audio) + assert plugin.draining_stage_ids(stage_count=4) == frozenset() diff --git a/tests/engine/test_orchestrator_stage_input_bridge.py b/tests/engine/test_orchestrator_stage_input_bridge.py index 842824c2d..3c1dd6d2b 100644 --- a/tests/engine/test_orchestrator_stage_input_bridge.py +++ b/tests/engine/test_orchestrator_stage_input_bridge.py @@ -14,6 +14,7 @@ import pytest from vllm.outputs import CompletionOutput, RequestOutput from vllm.sampling_params import SamplingParams +from vllm_omni.engine.duplex_orchestrator import DuplexOrchestrator from vllm_omni.engine.orchestrator import ( Orchestrator, OrchestratorRequestState, @@ -249,6 +250,41 @@ async def test_async_prewarm_skips_outgoing_only_stage(payload_sender_info) -> N ) +@pytest.mark.asyncio +async def test_async_prewarm_skips_stage_with_custom_process_input_func() -> None: + """AURA Stage1 has asr2aura: must not be zero-prewarmed under async_chunk. + + The custom-process-input gate lives on DuplexOrchestrator (AURA path), not + the turn-based Orchestrator base. + """ + orchestrator = object.__new__(DuplexOrchestrator) + stage0 = FakePrewarmPool("sender") + stage1 = FakePrewarmPool("receiver") # would receive chunks if role alone decided + stage1.stage_client = SimpleNamespace(custom_process_input_func=lambda *a, **k: None) + stage2 = FakePrewarmPool("receiver") + orchestrator.stage_pools = [stage0, stage1, stage2] + orchestrator._emit_tx_edge = lambda **_kwargs: None + orchestrator._on_stage_submitted = MagicMock() + req_state = OrchestratorRequestState( + request_id="req-prewarm-custom", + prompt={"prompt_token_ids": [1, 2]}, + sampling_params_list=[SamplingParams(max_tokens=1) for _ in range(3)], + final_stage_id=2, + ) + + prewarmed = await orchestrator._prewarm_async_chunk_stages( + "req-prewarm-custom", + SimpleNamespace(prompt_token_ids=[1, 2], resumable=True), + req_state, + ) + + assert prewarmed is True + assert stage1.submitted == [] + assert len(stage2.submitted) == 1 + assert 1 not in req_state.stage_submit_ts + assert 2 in req_state.stage_submit_ts + + @pytest.mark.asyncio async def test_async_route_forwards_to_outgoing_only_stage() -> None: orchestrator = object.__new__(Orchestrator) diff --git a/tests/entrypoints/duplex/test_startup_warmup.py b/tests/entrypoints/duplex/test_startup_warmup.py new file mode 100644 index 000000000..0971e317d --- /dev/null +++ b/tests/entrypoints/duplex/test_startup_warmup.py @@ -0,0 +1,56 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import pybase64 as base64 + +from vllm_omni.engine.duplex.config import DuplexCapabilities +from vllm_omni.entrypoints.duplex.warmup import ( + _WARMUP_AUDIO_SECONDS, + _WARMUP_FRAME_HEIGHT, + _WARMUP_FRAME_WIDTH, + _WARMUP_SAMPLE_RATE_HZ, + _short_pcm_f32le_b64, + _warmup_jpeg_b64, + startup_warmup_kind, +) + + +class _Plugin: + def __init__(self, required: frozenset[str]) -> None: + self._required = required + + def capabilities(self, *, max_sessions: int) -> DuplexCapabilities: + del max_sessions + return DuplexCapabilities(required_input_modalities=self._required) + + +def test_video_required_model_warms_up_one_turn_by_default() -> None: + plugin = _Plugin(frozenset({"video"})) + assert startup_warmup_kind(plugin, 0) == "video_turn" + + +def test_negative_warmup_frames_disables_the_video_turn() -> None: + plugin = _Plugin(frozenset({"video"})) + assert startup_warmup_kind(plugin, -1) is None + + +def test_audio_primary_model_stays_opt_in() -> None: + plugin = _Plugin(frozenset({"audio"})) + assert startup_warmup_kind(plugin, 0) is None + assert startup_warmup_kind(plugin, 4) == "silent_frames" + + +def test_startup_turn_audio_is_short_and_not_silent() -> None: + raw = base64.b64decode(_short_pcm_f32le_b64()) + assert len(raw) == int(_WARMUP_SAMPLE_RATE_HZ * _WARMUP_AUDIO_SECONDS) * 4 + assert any(raw) + + +def test_startup_turn_frame_is_a_camera_sized_jpeg() -> None: + from io import BytesIO + + from PIL import Image + + image = Image.open(BytesIO(base64.b64decode(_warmup_jpeg_b64()))) + assert image.size == (_WARMUP_FRAME_WIDTH, _WARMUP_FRAME_HEIGHT) + assert image.format == "JPEG" diff --git a/tests/examples/test_minicpmo_realtime_web_server.py b/tests/examples/test_minicpmo_realtime_web_server.py index 44f04eb1b..0f7149a34 100644 --- a/tests/examples/test_minicpmo_realtime_web_server.py +++ b/tests/examples/test_minicpmo_realtime_web_server.py @@ -113,12 +113,23 @@ def test_qwen_config_does_not_inject_reference_voice(): assert '"profile": "qwen3-turn"' in page assert '"adapter": "vad"' in page assert '"refAudio": null' in page - for name in ("common", "minicpm_native", "qwen3_turn"): + for name in ("common", "minicpm_native", "qwen3_turn", "aura_ptt"): assert client.get(f"/static/profiles/{name}.js").status_code == 200 with pytest.raises(ValueError, match="ref-audio"): build_app(profile="qwen3-turn", ref_audio="unused.wav") +def test_aura_ptt_config_defaults_and_rejects_ref_audio(): + client = TestClient(build_app(profile="aura-ptt", model="aurateam/AURA")) + page = client.get("/").text + assert '"profile": "aura-ptt"' in page + assert '"refAudio": null' in page + assert "aura_ptt.js" in page + assert 'id="pttButton"' in page + with pytest.raises(ValueError, match="ref-audio"): + build_app(profile="aura-ptt", ref_audio="unused.wav") + + def test_config_cannot_close_its_script_element(): page = TestClient(build_app(model="")).get("/").text assert "