```

feat: add audio buffering and interruption handling for voice input

Implement processing phase tracking (idle/transcribing/llm) and audio segment buffering to handle speech interruptions gracefully. Adds abort_request message type and buffer timeout mechanism to process accumulated audio segments together.
```

packages/voice-assistant/src/server/index.ts

@@ -265,7 +265,8 @@ async function main() {
         const transcriptText = result.text.trim();
         if (!transcriptText) {
           console.log("[STT] Transcription is empty or silence, skipping LLM processing");
-          // Return empty string without broadcasting or processing
+          // Reset to idle since we're not processing
+          wsServer.setPhaseForConversation(conversationId, 'idle');
           return "";
         }
 
@@ -281,6 +282,9 @@ async function main() {
           },
         });
 
+        // Set phase to LLM before processing
+        wsServer.setPhaseForConversation(conversationId, 'llm');
+
         // Process the transcribed text through the orchestrator WITH TTS enabled
         // Since this came from voice input, respond with voice output
         await processUserMessage({
@@ -291,8 +295,14 @@ async function main() {
           abortSignal,
         });
 
+        // Reset to idle after LLM processing completes
+        wsServer.setPhaseForConversation(conversationId, 'idle');
+
         return result.text;
       } catch (error: any) {
+        // Reset to idle on error
+        wsServer.setPhaseForConversation(conversationId, 'idle');
+
         console.error("[STT] Error transcribing audio:", error);
         wsServer.broadcastActivityLog({
           id: uuidv4(),

packages/voice-assistant/src/server/types.ts

@@ -16,7 +16,7 @@ export interface ActivityLogEntry {
 export interface WebSocketMessage {
   type: 'activity_log' | 'status' | 'ping' | 'pong' | 'user_message' | 'assistant_chunk'
        | 'audio_chunk' | 'audio_output' | 'recording_state' | 'transcription_result' | 'audio_played'
-       | 'artifact';
+       | 'artifact' | 'abort_request';
   payload: unknown;
 }
 

packages/voice-assistant/src/server/websocket-server.ts

@@ -9,10 +9,13 @@ import type {
 import { confirmAudioPlayed } from "./agent/tts-manager.js";
 import { createConversation, deleteConversation } from "./agent/orchestrator.js";
 
+type ProcessingPhase = 'idle' | 'transcribing' | 'llm';
+
 export class VoiceAssistantWebSocketServer {
   private wss: WebSocketServer;
   private clients: Map<WebSocket, string> = new Map(); // Map ws to client ID
   private conversationIds: Map<WebSocket, string> = new Map(); // Map ws to conversation ID
+  private conversationIdToWs: Map<string, WebSocket> = new Map(); // Reverse map: conversation ID to ws
   private abortControllers: Map<WebSocket, AbortController> = new Map(); // Map ws to AbortController
   private messageHandler?: (conversationId: string, message: string, abortSignal: AbortSignal) => Promise<void>;
   private audioHandler?: (conversationId: string, audio: Buffer, format: string, abortSignal: AbortSignal) => Promise<string>;
@@ -20,6 +23,11 @@ export class VoiceAssistantWebSocketServer {
     new Map();
   private clientIdCounter: number = 0;
 
+  // Audio buffering for interruption handling
+  private processingPhases: Map<WebSocket, ProcessingPhase> = new Map();
+  private pendingAudioSegments: Map<WebSocket, Array<{audio: Buffer, format: string}>> = new Map();
+  private bufferTimeouts: Map<WebSocket, NodeJS.Timeout> = new Map();
+
   constructor(server: HTTPServer) {
     this.wss = new WebSocketServer({ server, path: "/ws" });
 
@@ -38,11 +46,16 @@ export class VoiceAssistantWebSocketServer {
     // Create new conversation for this client
     const conversationId = createConversation();
     this.conversationIds.set(ws, conversationId);
+    this.conversationIdToWs.set(conversationId, ws); // Add reverse mapping
 
     // Create AbortController for this client
     const abortController = new AbortController();
     this.abortControllers.set(ws, abortController);
 
+    // Initialize processing state
+    this.processingPhases.set(ws, 'idle');
+    this.pendingAudioSegments.set(ws, []);
+
     console.log(
       `[WS] Client connected: ${clientId} with conversation ${conversationId} (total: ${this.clients.size})`
     );
@@ -72,6 +85,17 @@ export class VoiceAssistantWebSocketServer {
         console.log(`[WS] Aborted operations for ${clientId}`);
       }
 
+      // Clear buffer timeout
+      const timeout = this.bufferTimeouts.get(ws);
+      if (timeout) {
+        clearTimeout(timeout);
+        this.bufferTimeouts.delete(ws);
+      }
+
+      // Clean up state tracking
+      this.processingPhases.delete(ws);
+      this.pendingAudioSegments.delete(ws);
+
       if (clientId) {
         this.clients.delete(ws);
         console.log(
@@ -82,6 +106,7 @@ export class VoiceAssistantWebSocketServer {
       if (conversationId) {
         deleteConversation(conversationId);
         this.conversationIds.delete(ws);
+        this.conversationIdToWs.delete(conversationId); // Clean up reverse mapping
         console.log(`[WS] Conversation ${conversationId} deleted`);
       }
     });
@@ -98,6 +123,17 @@ export class VoiceAssistantWebSocketServer {
         this.abortControllers.delete(ws);
       }
 
+      // Clear buffer timeout
+      const timeout = this.bufferTimeouts.get(ws);
+      if (timeout) {
+        clearTimeout(timeout);
+        this.bufferTimeouts.delete(ws);
+      }
+
+      // Clean up state tracking
+      this.processingPhases.delete(ws);
+      this.pendingAudioSegments.delete(ws);
+
       if (clientId) {
         this.clients.delete(ws);
       }
@@ -105,6 +141,7 @@ export class VoiceAssistantWebSocketServer {
       if (conversationId) {
         deleteConversation(conversationId);
         this.conversationIds.delete(ws);
+        this.conversationIdToWs.delete(conversationId); // Clean up reverse mapping
       }
     });
   }
@@ -160,6 +197,11 @@ export class VoiceAssistantWebSocketServer {
           confirmAudioPlayed(audioPlayedPayload.id);
           break;
 
+        case "abort_request":
+          // Handle abort request from client (e.g., when VAD detects new speech)
+          await this.handleAbortRequest(ws);
+          break;
+
         default:
           console.warn(`[WS] Unknown message type: ${message.type}`);
       }
@@ -168,6 +210,40 @@ export class VoiceAssistantWebSocketServer {
     }
   }
 
+  private async handleAbortRequest(ws: WebSocket): Promise<void> {
+    const phase = this.processingPhases.get(ws);
+    const abortController = this.abortControllers.get(ws);
+
+    console.log(`[WS] Abort request received, current phase: ${phase}`);
+
+    if (phase === 'llm') {
+      // Already in LLM phase - abort immediately
+      if (abortController) {
+        abortController.abort();
+        console.log(`[WS] Aborted LLM processing`);
+      }
+
+      // Reset phase to idle
+      this.processingPhases.set(ws, 'idle');
+
+      // Clear any pending segments from previous interruptions
+      this.pendingAudioSegments.set(ws, []);
+
+      // Clear any pending timeout
+      const timeout = this.bufferTimeouts.get(ws);
+      if (timeout) {
+        clearTimeout(timeout);
+        this.bufferTimeouts.delete(ws);
+      }
+    } else if (phase === 'transcribing') {
+      // Still in STT phase - we'll buffer the next audio
+      // Don't abort yet, just set a flag by keeping the current abort controller
+      console.log(`[WS] Will buffer next audio segment (currently transcribing)`);
+      // Phase stays as 'transcribing', handleAudioChunk will handle buffering
+    }
+    // If idle, nothing to do
+  }
+
   private sendToClient(ws: WebSocket, message: WebSocketMessage): void {
     if (ws.readyState === 1) {
       // WebSocket.OPEN = 1
@@ -212,6 +288,16 @@ export class VoiceAssistantWebSocketServer {
     this.audioHandler = handler;
   }
 
+  public setPhaseForConversation(conversationId: string, phase: ProcessingPhase): void {
+    const ws = this.conversationIdToWs.get(conversationId);
+    if (ws) {
+      this.processingPhases.set(ws, phase);
+      console.log(`[WS] Phase set to '${phase}' for conversation ${conversationId}`);
+    } else {
+      console.warn(`[WS] Cannot set phase for unknown conversation ${conversationId}`);
+    }
+  }
+
   private async handleAudioChunk(
     ws: WebSocket,
     payload: AudioChunkPayload
@@ -248,49 +334,59 @@ export class VoiceAssistantWebSocketServer {
           `[WS] Buffered audio chunk (${audioBuffer.length} bytes, total chunks: ${buffer.chunks.length})`
         );
       } else {
-        // Last chunk - process complete audio
+        // Last chunk - complete audio segment received
         const buffer = this.audioBuffers.get(clientId);
         const allChunks = buffer
           ? [...buffer.chunks, audioBuffer]
           : [audioBuffer];
         const format = buffer?.format || payload.format;
 
-        // Concatenate all chunks
-        const completeAudio = Buffer.concat(allChunks);
+        // Concatenate all chunks for this segment
+        const currentSegmentAudio = Buffer.concat(allChunks);
         console.log(
-          `[WS] Complete audio received (${completeAudio.length} bytes, ${allChunks.length} chunks)`
+          `[WS] Complete audio segment received (${currentSegmentAudio.length} bytes, ${allChunks.length} chunks)`
         );
 
-        // Clear buffer
+        // Clear chunk buffer
         this.audioBuffers.delete(clientId);
 
-        // Abort any ongoing request
-        const oldAbortController = this.abortControllers.get(ws);
-        if (oldAbortController) {
-          oldAbortController.abort();
-        }
+        // Get current phase and pending segments
+        const currentPhase = this.processingPhases.get(ws) || 'idle';
+        const pendingSegments = this.pendingAudioSegments.get(ws) || [];
 
-        // Create new abort controller for this request
-        const newAbortController = new AbortController();
-        this.abortControllers.set(ws, newAbortController);
+        // Decision: buffer or process?
+        const shouldBuffer = currentPhase === 'transcribing' && pendingSegments.length === 0;
 
-        if (this.audioHandler) {
-          this.broadcastActivityLog({
-            id: Date.now().toString(),
-            timestamp: new Date(),
-            type: "system",
-            content: "Transcribing audio...",
-          });
+        if (shouldBuffer) {
+          // Currently transcribing first segment - buffer this one
+          console.log(`[WS] Buffering audio segment (phase: ${currentPhase})`);
+          pendingSegments.push({ audio: currentSegmentAudio, format });
+          this.pendingAudioSegments.set(ws, pendingSegments);
 
-          const transcript = await this.audioHandler(conversationId, completeAudio, format, newAbortController.signal);
+          // Set timeout to process buffer if no more audio arrives
+          this.setBufferTimeout(ws, conversationId);
+        } else if (pendingSegments.length > 0) {
+          // We have buffered segments - add this one and process all together
+          pendingSegments.push({ audio: currentSegmentAudio, format });
+          console.log(`[WS] Processing ${pendingSegments.length} buffered audio segments together`);
 
-          // Send transcription result back to client
-          this.sendToClient(ws, {
-            type: "transcription_result",
-            payload: { text: transcript },
-          });
+          // Clear pending segments and timeout
+          this.pendingAudioSegments.set(ws, []);
+          const timeout = this.bufferTimeouts.get(ws);
+          if (timeout) {
+            clearTimeout(timeout);
+            this.bufferTimeouts.delete(ws);
+          }
+
+          // Concatenate all segments
+          const allSegmentAudios = pendingSegments.map(s => s.audio);
+          const combinedAudio = Buffer.concat(allSegmentAudios);
+
+          // Process combined audio
+          await this.processAudio(ws, conversationId, combinedAudio, format);
         } else {
-          console.warn("[WS] No audio handler registered");
+          // Normal flow - no buffering needed
+          await this.processAudio(ws, conversationId, currentSegmentAudio, format);
         }
       }
     } catch (error: any) {
@@ -304,6 +400,85 @@ export class VoiceAssistantWebSocketServer {
     }
   }
 
+  private async processAudio(
+    ws: WebSocket,
+    conversationId: string,
+    audio: Buffer,
+    format: string
+  ): Promise<void> {
+    // Abort any ongoing request
+    const oldAbortController = this.abortControllers.get(ws);
+    if (oldAbortController) {
+      oldAbortController.abort();
+    }
+
+    // Create new abort controller for this request
+    const newAbortController = new AbortController();
+    this.abortControllers.set(ws, newAbortController);
+
+    // Set phase to transcribing
+    this.processingPhases.set(ws, 'transcribing');
+
+    if (this.audioHandler) {
+      this.broadcastActivityLog({
+        id: Date.now().toString(),
+        timestamp: new Date(),
+        type: "system",
+        content: "Transcribing audio...",
+      });
+
+      try {
+        const transcript = await this.audioHandler(conversationId, audio, format, newAbortController.signal);
+
+        // Send transcription result back to client
+        this.sendToClient(ws, {
+          type: "transcription_result",
+          payload: { text: transcript },
+        });
+
+        // Phase management is now handled by the audioHandler in index.ts
+        // It will set phase to 'llm' before LLM processing and 'idle' after completion
+      } catch (error: any) {
+        // If error occurs, reset to idle as safety net
+        this.processingPhases.set(ws, 'idle');
+        throw error;
+      }
+    } else {
+      console.warn("[WS] No audio handler registered");
+      this.processingPhases.set(ws, 'idle');
+    }
+  }
+
+  private setBufferTimeout(ws: WebSocket, conversationId: string): void {
+    // Clear any existing timeout
+    const existingTimeout = this.bufferTimeouts.get(ws);
+    if (existingTimeout) {
+      clearTimeout(existingTimeout);
+    }
+
+    // Set new timeout (10 seconds)
+    const timeout = setTimeout(async () => {
+      console.log(`[WS] Buffer timeout reached, processing pending segments`);
+
+      const pendingSegments = this.pendingAudioSegments.get(ws) || [];
+      if (pendingSegments.length > 0) {
+        // Concatenate all pending segments
+        const allSegmentAudios = pendingSegments.map(s => s.audio);
+        const combinedAudio = Buffer.concat(allSegmentAudios);
+        const format = pendingSegments[0].format;
+
+        // Clear pending segments
+        this.pendingAudioSegments.set(ws, []);
+        this.bufferTimeouts.delete(ws);
+
+        // Process combined audio
+        await this.processAudio(ws, conversationId, combinedAudio, format);
+      }
+    }, 10000); // 10 second timeout
+
+    this.bufferTimeouts.set(ws, timeout);
+  }
+
   public close(): void {
     this.clients.forEach((_clientId, ws) => {
       ws.close();

packages/voice-assistant/src/ui/App.tsx

@@ -435,6 +435,12 @@ function App() {
               audioPlayerRef.current.stop();
               setIsPlayingAudio(false);
               setCurrentAssistantMessage("");
+
+              // Send abort request to server immediately
+              ws.send({
+                type: "abort_request",
+                payload: {},
+              });
             },
             onSpeechEnd: async (audioData: Float32Array) => {
               console.log("[App] Speech ended, processing...");