From 7284efd0fe7401c1ccf8b06dddf8f28861d37386 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 15 Nov 2025 19:06:53 +0100 Subject: [PATCH] Prototype streaming TTS chunks --- packages/app/src/contexts/session-context.tsx | 20 +- .../server/src/server/agent/tts-manager.ts | 183 +++++++++++++----- .../server/src/server/agent/tts-openai.ts | 37 +--- packages/server/src/server/messages.ts | 3 + plan.md | 7 +- 5 files changed, 163 insertions(+), 87 deletions(-) diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 9a2eda032..2871601d5 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -273,6 +273,7 @@ export function SessionProvider({ children, serverUrl }: SessionProviderProps) { const [pendingPermissions, setPendingPermissions] = useState>(new Map()); const [gitDiffs, setGitDiffs] = useState>(new Map()); const [fileExplorer, setFileExplorer] = useState>(new Map()); + const activeAudioGroupsRef = useRef>(new Set()); const updateExplorerState = useCallback( (agentId: string, updater: (state: AgentFileExplorerState) => AgentFileExplorerState) => { @@ -463,10 +464,14 @@ export function SessionProvider({ children, serverUrl }: SessionProviderProps) { const unsubAudioOutput = ws.on("audio_output", async (message) => { if (message.type !== "audio_output") return; const data = message.payload; + const playbackGroupId = data.groupId ?? data.id; + const isFinalChunk = data.isLastChunk ?? true; + activeAudioGroupsRef.current.add(playbackGroupId); + setIsPlayingAudio(true); + + let playbackFailed = false; try { - setIsPlayingAudio(true); - // Create blob-like object with correct mime type (React Native compatible) const mimeType = data.format === "mp3" ? "audio/mpeg" : `audio/${data.format}`; @@ -499,11 +504,9 @@ export function SessionProvider({ children, serverUrl }: SessionProviderProps) { }, }; ws.send(confirmMessage); - - setIsPlayingAudio(false); } catch (error: any) { + playbackFailed = true; console.error("[Session] Audio playback error:", error); - setIsPlayingAudio(false); // Still send confirmation even on error to prevent server from waiting const confirmMessage: WSInboundMessage = { @@ -514,6 +517,13 @@ export function SessionProvider({ children, serverUrl }: SessionProviderProps) { }, }; ws.send(confirmMessage); + } finally { + if (isFinalChunk || playbackFailed) { + activeAudioGroupsRef.current.delete(playbackGroupId); + } + if (activeAudioGroupsRef.current.size === 0) { + setIsPlayingAudio(false); + } } }); diff --git a/packages/server/src/server/agent/tts-manager.ts b/packages/server/src/server/agent/tts-manager.ts index 8c2bc2a93..8973bfa9c 100644 --- a/packages/server/src/server/agent/tts-manager.ts +++ b/packages/server/src/server/agent/tts-manager.ts @@ -5,6 +5,8 @@ import type { SessionOutboundMessage } from "../messages.js"; interface PendingPlayback { resolve: () => void; reject: (error: Error) => void; + pendingChunks: number; + streamEnded: boolean; } /** @@ -36,8 +38,8 @@ export class TTSManager { return; } - // Generate TTS audio - const { audio, format } = await synthesizeSpeech(text); + // Generate TTS audio stream + const { stream, format } = await synthesizeSpeech(text); if (abortSignal.aborted) { console.log( @@ -46,58 +48,129 @@ export class TTSManager { return; } - // Create unique ID for this audio segment const audioId = uuidv4(); + let playbackResolve!: () => void; + let playbackReject!: (error: Error) => void; - // Store abort handler reference outside Promise constructor - let onAbort: (() => void) | undefined; - - // Create promise that will be resolved when client confirms playback const playbackPromise = new Promise((resolve, reject) => { - // Store handlers (no timeout - will resolve when client confirms or connection closes) - this.pendingPlaybacks.set(audioId, { resolve, reject }); - - // Handle abort signal - onAbort = () => { - console.log( - `[TTS-Manager ${this.sessionId}] Aborted while waiting for playback` - ); - // Clean up pending playback - this.pendingPlaybacks.delete(audioId); - // Reject with abort error - resolve(); - }; - - // Listen for abort (once: true for auto-cleanup if abort fires) - abortSignal.addEventListener("abort", onAbort, { once: true }); + playbackResolve = resolve; + playbackReject = reject; }); - // Clean up abort listener when promise settles (in case abort never fires) - if (onAbort) { - playbackPromise.finally(() => { - abortSignal!.removeEventListener("abort", onAbort!); - }); + const pendingPlayback: PendingPlayback = { + resolve: playbackResolve, + reject: playbackReject, + pendingChunks: 0, + streamEnded: false, + }; + + this.pendingPlaybacks.set(audioId, pendingPlayback); + + let onAbort: (() => void) | undefined; + const destroyStream = () => { + if (typeof stream.destroy === "function" && !stream.destroyed) { + stream.destroy(); + } + }; + + onAbort = () => { + console.log( + `[TTS-Manager ${this.sessionId}] Aborted while waiting for playback` + ); + pendingPlayback.streamEnded = true; + pendingPlayback.pendingChunks = 0; + this.pendingPlaybacks.delete(audioId); + playbackResolve(); + destroyStream(); + }; + + abortSignal.addEventListener("abort", onAbort, { once: true }); + + try { + const iterator = stream[Symbol.asyncIterator](); + let chunkIndex = 0; + let current = await iterator.next(); + + if (!current.done) { + let next = await iterator.next(); + + while (true) { + if (abortSignal.aborted) { + console.log( + `[TTS-Manager ${this.sessionId}] Aborted during stream emission` + ); + break; + } + + const chunkBuffer = Buffer.isBuffer(current.value) + ? current.value + : Buffer.from(current.value); + + const chunkId = `${audioId}:${chunkIndex}`; + pendingPlayback.pendingChunks += 1; + + emitMessage({ + type: "audio_output", + payload: { + id: chunkId, + groupId: audioId, + chunkIndex, + isLastChunk: next.done, + audio: chunkBuffer.toString("base64"), + format, + isRealtimeMode, + }, + }); + + console.log( + `[TTS-Manager ${this.sessionId}] ${new Date().toISOString()} Sent audio chunk ${chunkId}${ + next.done ? " (last)" : "" + }` + ); + + chunkIndex += 1; + + if (next.done) { + break; + } + + current = next; + next = await iterator.next(); + } + } + + pendingPlayback.streamEnded = true; + + if (pendingPlayback.pendingChunks === 0) { + this.pendingPlaybacks.delete(audioId); + playbackResolve(); + } + + await playbackPromise; + } catch (error) { + if (abortSignal.aborted) { + console.log( + `[TTS-Manager ${this.sessionId}] Audio stream closed after abort` + ); + } else { + console.error( + `[TTS-Manager ${this.sessionId}] Error streaming audio`, + error + ); + this.pendingPlaybacks.delete(audioId); + pendingPlayback.reject(error as Error); + throw error; + } + } finally { + if (onAbort) { + abortSignal.removeEventListener("abort", onAbort); + } + destroyStream(); } - // Emit audio output message (include mode for drift protection) - emitMessage({ - type: "audio_output", - payload: { - id: audioId, - audio: audio.toString("base64"), - format, - isRealtimeMode, - }, - }); - - console.log( - `[TTS-Manager ${ - this.sessionId - }] ${new Date().toISOString()} Sent audio ${audioId}, waiting for playback...` - ); - - // Wait for playback confirmation - await playbackPromise; + if (abortSignal.aborted) { + return; + } console.log( `[TTS-Manager ${ @@ -110,19 +183,25 @@ export class TTSManager { * Called when client confirms audio playback completed * Resolves the corresponding promise */ - public confirmAudioPlayed(audioId: string): void { + public confirmAudioPlayed(chunkId: string): void { + const [audioId] = chunkId.includes(":") + ? chunkId.split(":") + : [chunkId]; const pending = this.pendingPlaybacks.get(audioId); if (!pending) { console.warn( - `[TTS-Manager ${this.sessionId}] Received confirmation for unknown audio ID: ${audioId}` + `[TTS-Manager ${this.sessionId}] Received confirmation for unknown audio ID: ${chunkId}` ); return; } - // Resolve promise and cleanup - pending.resolve(); - this.pendingPlaybacks.delete(audioId); + pending.pendingChunks = Math.max(0, pending.pendingChunks - 1); + + if (pending.pendingChunks === 0 && pending.streamEnded) { + pending.resolve(); + this.pendingPlaybacks.delete(audioId); + } } /** diff --git a/packages/server/src/server/agent/tts-openai.ts b/packages/server/src/server/agent/tts-openai.ts index 57408cdb0..d5d113ea2 100644 --- a/packages/server/src/server/agent/tts-openai.ts +++ b/packages/server/src/server/agent/tts-openai.ts @@ -8,8 +8,8 @@ export interface TTSConfig { responseFormat?: "mp3" | "opus" | "aac" | "flac" | "wav" | "pcm"; } -export interface SpeechResult { - audio: Buffer; +export interface SpeechStreamResult { + stream: Readable; format: string; } @@ -31,7 +31,9 @@ export function initializeTTS(ttsConfig: TTSConfig): void { ); } -export async function synthesizeSpeech(text: string): Promise { +export async function synthesizeSpeech( + text: string +): Promise { if (!openaiClient || !config) { throw new Error("TTS not initialized. Call initializeTTS() first."); } @@ -58,18 +60,13 @@ export async function synthesizeSpeech(text: string): Promise { response_format: config.responseFormat as "mp3" | "opus" | "aac" | "flac" | "wav" | "pcm", }); - // Response is a stream, convert to buffer - const audioBuffer = await streamToBuffer( - response.body as unknown as Readable - ); + const audioStream = response.body as unknown as Readable; const duration = Date.now() - startTime; - console.log( - `[TTS] Speech synthesis complete in ${duration}ms (${audioBuffer.length} bytes)` - ); + console.log(`[TTS] Speech synthesis stream ready in ${duration}ms`); return { - audio: audioBuffer, + stream: audioStream, format: config.responseFormat || "mp3", }; } catch (error: any) { @@ -78,24 +75,6 @@ export async function synthesizeSpeech(text: string): Promise { } } -async function streamToBuffer(stream: Readable): Promise { - const chunks: Buffer[] = []; - - return new Promise((resolve, reject) => { - stream.on("data", (chunk) => { - chunks.push(Buffer.from(chunk)); - }); - - stream.on("end", () => { - resolve(Buffer.concat(chunks)); - }); - - stream.on("error", (error) => { - reject(error); - }); - }); -} - export function isTTSInitialized(): boolean { return openaiClient !== null && config !== null; } diff --git a/packages/server/src/server/messages.ts b/packages/server/src/server/messages.ts index 02341169e..ee9f40850 100644 --- a/packages/server/src/server/messages.ts +++ b/packages/server/src/server/messages.ts @@ -472,6 +472,9 @@ export const AudioOutputMessageSchema = z.object({ format: z.string(), id: z.string(), isRealtimeMode: z.boolean(), // Mode when audio was generated (for drift protection) + groupId: z.string().optional(), // Logical utterance id + chunkIndex: z.number().int().nonnegative().optional(), + isLastChunk: z.boolean().optional(), }), }); diff --git a/plan.md b/plan.md index 03a706cd9..d8db5ef3b 100644 --- a/plan.md +++ b/plan.md @@ -66,5 +66,10 @@ - `useSpeechmaticsAudio` now streams ~1s PCM payloads throughout a speech turn (plus a final tail chunk) and `RealtimeContext` forwards them with `format: "audio/pcm;rate=16000;bits=16"` + accurate `isLast` so the server sees continuous frames instead of one big WAV. On the backend `Session.handleAudioChunk` tracks PCM bytes, converts buffered samples into WAV once the 1s threshold—or final chunk—arrives, and runs the prior buffering/timeout logic so STT fires repeatedly without waiting for speech end. `npm run typecheck --workspaces --if-present` still fails due to the existing cross-workspace `packages/app/src/types/stream.ts` errors (missing module resolution, tool snapshot typing) that occur on main as well. - [x] Investigate streaming STT/server-side VAD (Whisper or equivalent) and document requirements for production use. - Captured the current audio pipeline, design goals, candidate approaches (self-hosted Whisper vs managed APIs), and production requirements (transport, VAD, streaming STT engine, infra, observability, compliance) in `docs/streaming-stt-vad.md`, along with an implementation phase plan and open questions for the follow-up agent. -- [ ] Prototype streaming TTS output (chunked `audio_output` messages) and update the player to start playback as soon as chunks arrive. +- [x] Prototype streaming TTS output (chunked `audio_output` messages) and update the player to start playback as soon as chunks arrive. + - Split OpenAI TTS responses into chunked `audio_output` payloads via a streaming `TTSManager` pipeline (`packages/server/src/server/agent/tts-manager.ts`, `packages/server/src/server/agent/tts-openai.ts`) that tracks per-chunk acknowledgements, adds utterance IDs, and resolves once every chunk finishes. The app’s session context now keeps audio groups active until the last chunk (or an error) lands so playback begins as soon as each chunk arrives without waiting for the full clip (`packages/app/src/contexts/session-context.tsx`). `npm run typecheck --workspace=@voice-dev/server` and `npm run typecheck --workspace=@voice-dev/app` still fail in the pre-existing cross-workspace stream harness files (see earlier tasks); no new regressions were introduced. - [ ] Add telemetry for barge-in latency (speech start → playback stop, chunk arrival → LLM abort) to measure improvements per milestone. +- [ ] Codex tool calls aren’t visible in the live stream (they only show up after hydrating/refreshing); fix the reducer/rendering path so Codex tool events appear in real time the same way Claude’s do. +- [ ] Update the AgentInput “Cancel” control to use a square stop icon button (consistent with media players) so interrupting an agent is visually distinct from other actions. +- [ ] When the agent input has text and we’re not in the cancellable state, replace both the mic (Dictate) and Realtime buttons with a single Send button—only surface Dictate/Realtime when the input is empty or the agent is running. +- [ ] Remove the agent status pill next to the permission selector; instead show a “working” indicator inside the chat scroll view itself (three bouncing dots animation) whenever the agent is busy.