diff --git a/packages/voice-assistant/src/server/index.ts b/packages/voice-assistant/src/server/index.ts index 08c90f282..6b4ac01b9 100644 --- a/packages/voice-assistant/src/server/index.ts +++ b/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(), diff --git a/packages/voice-assistant/src/server/types.ts b/packages/voice-assistant/src/server/types.ts index 2f5fb1170..2e07e2f2e 100644 --- a/packages/voice-assistant/src/server/types.ts +++ b/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; } diff --git a/packages/voice-assistant/src/server/websocket-server.ts b/packages/voice-assistant/src/server/websocket-server.ts index 418467b2f..5e25bd2ec 100644 --- a/packages/voice-assistant/src/server/websocket-server.ts +++ b/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 = new Map(); // Map ws to client ID private conversationIds: Map = new Map(); // Map ws to conversation ID + private conversationIdToWs: Map = new Map(); // Reverse map: conversation ID to ws private abortControllers: Map = new Map(); // Map ws to AbortController private messageHandler?: (conversationId: string, message: string, abortSignal: AbortSignal) => Promise; private audioHandler?: (conversationId: string, audio: Buffer, format: string, abortSignal: AbortSignal) => Promise; @@ -20,6 +23,11 @@ export class VoiceAssistantWebSocketServer { new Map(); private clientIdCounter: number = 0; + // Audio buffering for interruption handling + private processingPhases: Map = new Map(); + private pendingAudioSegments: Map> = new Map(); + private bufferTimeouts: Map = 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 { + 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 { + // 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(); diff --git a/packages/voice-assistant/src/ui/App.tsx b/packages/voice-assistant/src/ui/App.tsx index 33e9d9320..a6ec9ab3a 100644 --- a/packages/voice-assistant/src/ui/App.tsx +++ b/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...");