[Model] Add JoyAI speech input to unified duplex sessions

docs/design/fullduplex.md

@@ -452,6 +452,28 @@ Frame-based plugins use `engine/duplex/intermediate.py::build_duplex_append_prom
 for the shared request identity, sequencing and config snapshots. Token budgets,
 PCM framing and model-specific worker fields remain in the plugins.
 
+Turn plugins may bind their stage identities in `configure_pipeline` and select
+an append's entry with `input_stage_id(payload)` (default 0). The request context
+and orchestrator retain that entry stage. Session prompt preparation receives
+the request's complete `stage_sampling_params`, so a pipeline with an ASR prefix
+budgets the JoyAI prompt against JoyAI's generation settings. Session model config
+comes from the first tokenizer-owning stage, matching frontend comprehension
+ownership. JoyAI's audio-free requests enter its `joyvl` stage directly; speech
+enters ASR and carries the original observation context through the handoff.
+
+While idle, JoyAI retains the latest camera frame and one input chunk of audio
+pre-roll. Once speech starts, it keeps the utterance until commit. An idle
+camera commit enters JoyAI directly; reserved-byte accounting includes the
+pre-roll until the commit is consumed or rolled back.
+
+Completed input transcriptions travel with stage outputs through the runner
+mailbox. After checking the request fence, the runner updates the exact committed
+audio history item and emits its transcription event. The append captures the
+original native audio part before yielding and retains that identity on its
+request resource; a reused public item ID does not confer transcript ownership.
+Cancellation, item deletion or replacement cannot apply a late transcript to a
+newer turn.
+
 See [supported models and deployments](../serving/full_duplex_api.md#enable-full-duplex)
 for the current plugin integrations and deployment configurations.
 

examples/online_serving/realtime_web/README.md

@@ -11,6 +11,8 @@ This implements [RFC #7222](https://github.com/vllm-project/vllm-omni/issues/722
 | `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 |
 | `joyai-turn` | Latest observation once per second while idle, or **Send observation**; waits for response completion and playback | Camera frames without audio input | Yes |
+| `joyai-turn --speech-input` | Hold to talk; release commits speech, then camera observations resume while idle | Latest frame with speech and camera-only observations | Yes |
+| `joyai-turn --speech-input --vad` | Server detects trailing silence; speech interrupts replies | Latest frame with each utterance; no automatic camera-only commits | Yes |
 
 Qwen3 VAD uses an engine-owned duplex plugin: microphone upload continues
 while replies stream, and speech can interrupt generation and playback. The
@@ -80,9 +82,24 @@ next turn starts after both `response.done` and local playback finish.
 
 Replies appear as text and play through the existing speaker component, which
 sends `playback.ack`. This profile uses camera input and text instructions;
-microphone input will follow the unified Session ASR integration. The browser
-only requests camera permission when **Camera** is enabled. Use localhost or
-HTTPS for browser camera access.
+the default does not request microphone access. The browser only requests camera
+permission when **Camera** is enabled. Use localhost or HTTPS for browser capture.
+
+For spoken instructions, start the `joyai_vl_interaction_asr_duplex.yaml`
+deployment and add `--speech-input` to the UI command. **Hold to talk** captures
+speech with the latest camera frame; releasing it commits only when that hold
+uploaded audio. Empty or muted holds do not submit a turn. Camera-only
+observations resume after the response and playback finish. Incoming speech may
+be buffered during playback; holding the button does not cancel the response.
+
+Add `--speech-input --vad` to commit speech automatically with shared server VAD
+(Silero must be available on the server). Speech interrupts an active reply,
+using the server's supported `interrupt_response: true` setting. Each utterance
+includes the latest camera frame. In this mode automatic camera-only observations
+are paused, so continuous silent watching uses the default camera mode or
+push-to-talk mode.
+The three-stage deployment rejects audio input; use the ASR deployment for both
+microphone modes.
 
 ## Qwen3: explicit-turn Realtime
 

examples/online_serving/realtime_web/app/static/app.js

@@ -54,7 +54,7 @@
     }
   }
   sendTurnButton.hidden = !profile.clientCommit;
-  if (profile.videoOnly) sendTurnButton.textContent = 'Send observation';
+  if (profile.cameraObservations) sendTurnButton.textContent = 'Send observation';
   pttButton.hidden = !profile.pushToTalk;
   promptPreset.replaceChildren();
   for (const name of [...Object.keys(PROMPT_PRESETS), 'custom']) {
@@ -89,6 +89,7 @@
   let visionTurnLocked = false;
   let suppressedPlaybackId = null;
   let pttHeld = false;
+  let pttUploadedInput = false;
   const cameraCanvas = document.createElement('canvas');
   let playbackRate = OUTPUT_RATE;
   let pendingCapture = [];
@@ -101,6 +102,9 @@
   let responseComplete = false;
   let playbackComplete = true;
   let turnSubmitted = false;
+  const pendingCommits = new Set();
+  let pendingCommitsOwnBusy = false;
+  let commitCounter = 0;
   let echoTimer = null;
   let sessionGeneration = 0;
   let audioChain = Promise.resolve();
@@ -124,6 +128,25 @@
     if (profile.halfDuplex) pendingCapture = [];
   }
 
+  function settleCommitWithoutResponse(eventId) {
+    if (!pendingCommits.delete(eventId) || pendingCommits.size !== 0) return;
+    // Release this group's idle-to-busy state only once every commit has
+    // settled without a response. Another pending commit may still succeed.
+    if (pendingCommitsOwnBusy && !currentResponseId) {
+      clearTimeout(turnTimeout);
+      turnSubmitted = false;
+      assistantActive = false;
+      responseComplete = false;
+      sendTurnButton.disabled = !profile.clientCommit;
+      if (!pttHeld) setModel(profile.waiting);
+    }
+    pendingCommitsOwnBusy = false;
+    if (responseComplete) {
+      clearTimeout(turnTimeout);
+      finishResponseIfReady();
+    }
+  }
+
   function armTurnTimeout() {
     clearTimeout(turnTimeout);
     turnTimeout = window.setTimeout(() => {
@@ -139,11 +162,11 @@
   }
 
   function finishResponseIfReady() {
-    if (!responseComplete || !playbackComplete || echoTimer !== null) return;
+    if (!responseComplete || !playbackComplete || pendingCommits.size || echoTimer !== null) return;
     const generation = sessionGeneration;
     echoTimer = window.setTimeout(async () => {
       echoTimer = null;
-      if (!running || generation !== sessionGeneration) return;
+      if (!running || generation !== sessionGeneration || pendingCommits.size) return;
       if (profile.reconnectEachTurn) {
         connectionReady = false;
         const previous = socket;
@@ -156,6 +179,8 @@
       assistantActive = false;
       responseComplete = false;
       turnSubmitted = false;
+      pendingCommits.clear();
+      pendingCommitsOwnBusy = false;
       currentResponseId = null;
       responseHasAudio = false;
       assistantTextChannel = null;
@@ -407,6 +432,7 @@
 
     if (profile.pushToTalk && !pttHeld) {
       pendingCapture = [];
+      if (profile.cameraObservations && cameraPendingFrame) submitObservation(cameraPendingFrame);
       return;
     }
 
@@ -424,12 +450,14 @@
     }
     pendingCapture = [];
     const pcm = resampleInt16(merged, captureRate, profile.inputSampleRate || INPUT_RATE);
+    if (!pcm.length) return;
     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));
+    if (profile.pushToTalk) pttUploadedInput = true;
   }
 
   function submitObservation(frame) {
@@ -440,9 +468,18 @@
   }
 
   function submitTurn() {
-    for (const event of profile.commitMessages()) socket.send(JSON.stringify(event));
+    for (const event of profile.commitMessages()) {
+      if (event.type === 'input_audio_buffer.commit') {
+        event.event_id = `web-commit-${++commitCounter}`;
+        pendingCommits.add(event.event_id);
+        if (!assistantActive && !turnSubmitted) pendingCommitsOwnBusy = true;
+      }
+      socket.send(JSON.stringify(event));
+    }
     turnSubmitted = true;
-    responseComplete = false;
+    // A new hold can be rejected while the previous reply is still playing.
+    // Its terminal state belongs to that reply until a new response begins.
+    if (!currentResponseId) responseComplete = false;
     markBusy();
     armTurnTimeout();
     sendTurnButton.disabled = true;
@@ -454,7 +491,8 @@
     if (!profile.pushToTalk || !running || pttHeld === held) return;
     if (held) {
       pttHeld = true;
-      suppressLocalPlayback();
+      pttUploadedInput = false;
+      if (!profile.cameraObservations) suppressLocalPlayback();
       pttButton.classList.toggle('is-active', true);
       pttButton.textContent = 'Release to send';
       setModel('Talking');
@@ -463,24 +501,28 @@
     }
     // 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));
+    const committed = pttUploadedInput && socket && socket.readyState === WebSocket.OPEN;
+    if (committed) {
+      if (profile.cameraObservations) submitTurn();
+      else 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) {
+    if (committed && 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');
+    appendLog(committed ? 'PTT up · commit' : 'PTT up · no input');
   }
 
   function beginAssistant(responseId) {
+    pendingCommits.clear();
+    pendingCommitsOwnBusy = false;
     currentResponseId = responseId || currentResponseId;
     responseHasAudio = false;
     assistantActive = true;
@@ -523,8 +565,10 @@
 
   function playbackDrained(message) {
     const responseId = message.responseId || currentResponseId;
-    if (currentResponseId && responseId !== currentResponseId) return;
     if (profile.playbackAck) sendPlaybackAck(responseId, Number(message.playedMs) || 0);
+    // Generation may advance before the previous response finishes playing.
+    // ACK that response's own cursor without completing the newer UI turn.
+    if (currentResponseId && responseId !== currentResponseId) return;
     setPlayback('Idle');
     playbackComplete = true;
     if (message.underrunMs > 0) appendLog(`playback underrun ${message.underrunMs} ms`);
@@ -656,7 +700,11 @@
         if (action.kind === 'text-final' && action.role === 'assistant') releaseVisionTurn(responseId);
         break;
       case 'done':
-        clearTimeout(turnTimeout);
+        if (!currentResponseId) {
+          pendingCommits.clear();
+          pendingCommitsOwnBusy = false;
+        }
+        if (!pendingCommits.size) clearTimeout(turnTimeout);
         responseComplete = true;
         finishTranscript('assistant');
         releaseVisionTurn(responseId);
@@ -676,8 +724,12 @@
         if (sessionCloseResolver) sessionCloseResolver();
         sessionCloseResolver = null;
         break;
+      case 'commit-settled':
+        settleCommitWithoutResponse(action.eventId);
+        break;
       case 'error':
         if (action.fatal) { await failSession(action.message); break; }
+        settleCommitWithoutResponse(event.error?.event_id);
         // A rejected frame or a refused update costs one operation, not the
         // call: report it and keep listening.
         appendLog(action.message, true);
@@ -968,11 +1020,14 @@
     responseComplete = false;
     playbackComplete = true;
     turnSubmitted = false;
+    pendingCommits.clear();
+    pendingCommitsOwnBusy = false;
     assistantTextChannel = null;
     assistantRowByResponse.clear();
     lastClosedAssistantText = '';
     sendTurnButton.disabled = true;
     pttHeld = false;
+    pttUploadedInput = false;
     pttButton.disabled = true;
     pttButton.classList.remove('is-active');
     pttButton.textContent = 'Hold to talk';
@@ -1044,7 +1099,7 @@
   });
   sendTurnButton.addEventListener('click', () => {
     if (!running || !connectionReady || !profile.clientCommit || turnSubmitted) return;
-    if (profile.videoOnly) {
+    if (profile.videoOnly || (profile.cameraObservations && !pttHeld)) {
       submitObservation(cameraPendingFrame || cameraLastFrame);
       return;
     }

examples/online_serving/realtime_web/app/static/playback_worklet.js

@@ -2,6 +2,7 @@ class FullDuplexPcmPlayback extends AudioWorkletProcessor {
   constructor() {
     super();
     this.queue = [];
+    this.pendingMessages = [];
     this.offset = 0;
     this.playedFrames = 0;
     this.underrunFrames = 0;
@@ -18,6 +19,13 @@ class FullDuplexPcmPlayback extends AudioWorkletProcessor {
   }
 
   handleMessage(message) {
+    // Keep each response's samples and drain marker together. The next
+    // response can finish generating before this one finishes playing.
+    if (message.type !== 'clear' && this.activeResponseId
+        && message.responseId && message.responseId !== this.activeResponseId) {
+      this.pendingMessages.push(message);
+      return;
+    }
     if (message.type === 'audio' && message.pcm) {
       const wasEmpty = this.queue.length === 0;
       if (!this.started && !this.activeResponseId) {
@@ -47,6 +55,7 @@ class FullDuplexPcmPlayback extends AudioWorkletProcessor {
         });
       }
       this.queue = [];
+      this.pendingMessages = [];
       this.offset = 0;
       this.playedFrames = 0;
       this.underrunFrames = 0;
@@ -92,6 +101,9 @@ class FullDuplexPcmPlayback extends AudioWorkletProcessor {
     this.bufferWaitFrames = this.initialBufferFrames;
     this.rebuffering = false;
     this.fadeInFrames = 0;
+    const pending = this.pendingMessages;
+    this.pendingMessages = [];
+    for (const message of pending) this.handleMessage(message);
   }
 
   reportUnderrun() {