diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index e51d31c79..2196a9447 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -2913,6 +2913,16 @@ class ClaudeAgentSession implements AgentSession { return; } if (message.subtype === "task_notification") { + // TODO: subagent timelines are best-effort. Subagent task_notifications + // arrive without parent_tool_use_id but with tool_use_id pointing at the + // parent's Task call, so they slip past the sidechain router and pollute + // the parent timeline. Drop them here; eventually thread them into the + // parent Task tool call's sub_agent log instead. + const taskUseId = message.tool_use_id; + const cachedTool = taskUseId ? this.toolUseCache.get(taskUseId) : undefined; + if (cachedTool?.name === "Task") { + return; + } const taskNotificationItem = mapTaskNotificationSystemRecordToToolCall(message); if (taskNotificationItem) { events.push({ diff --git a/packages/server/src/server/agent/stt-manager.ts b/packages/server/src/server/agent/stt-manager.ts index 084c816dc..66a32babd 100644 --- a/packages/server/src/server/agent/stt-manager.ts +++ b/packages/server/src/server/agent/stt-manager.ts @@ -130,6 +130,10 @@ export class STTManager { this.resolveStt = toResolver(stt); } + public getProvider(): SpeechToTextProvider | null { + return this.resolveStt(); + } + /** * Transcribe audio buffer to text */ diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index bab990bf3..4bdd6ef30 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -1,4 +1,5 @@ import { execSync } from "child_process"; +import { EventEmitter } from "events"; import { mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs"; import { homedir, tmpdir } from "os"; import { join } from "path"; @@ -21,8 +22,29 @@ import type { ManagedAgent } from "./agent/agent-manager.js"; import type { ProviderDefinition } from "./agent/provider-registry.js"; import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { SessionOptions } from "./session.js"; +import type { + SpeechToTextProvider, + StreamingTranscriptionCommittedEvent, + StreamingTranscriptionEvent, + StreamingTranscriptionSession, +} from "./speech/speech-provider.js"; +import type { + TurnDetectionProvider, + TurnDetectionSession, +} from "./speech/turn-detection-provider.js"; interface SessionHandlerInternals { + startVoiceTurnController(): Promise; + stopVoiceTurnController(): Promise; + handleSendAgentMessage( + agentId: string, + text: string, + messageId?: string, + images?: Array<{ data: string; mimeType: string }>, + attachments?: unknown[], + runOptions?: unknown, + options?: { spokenInput?: boolean }, + ): Promise<{ ok: true } | { ok: false; error: string }>; handleCheckoutMergeRequest(params: unknown): Promise; handleCheckoutMergeFromBaseRequest(params: unknown): Promise; handleCheckoutCommitRequest(params: unknown): Promise; @@ -234,6 +256,8 @@ interface SessionForTestOptions { getDaemonTcpPort?: () => number | null; getDaemonTcpHost?: () => string | null; providerSnapshotManager?: ProviderSnapshotManager; + stt?: SessionOptions["stt"]; + voice?: SessionOptions["voice"]; messages?: unknown[]; binaryMessages?: Uint8Array[]; } @@ -302,7 +326,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { })), onChange: vi.fn(() => () => {}), } as unknown as SessionOptions["daemonConfigStore"], - stt: null, + stt: options.stt ?? null, tts: null, terminalManager: (options.terminalManager ?? null) as SessionOptions["terminalManager"], providerSnapshotManager: options.providerSnapshotManager, @@ -310,9 +334,211 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { scriptRuntimeStore: options.scriptRuntimeStore as SessionOptions["scriptRuntimeStore"], getDaemonTcpPort: options.getDaemonTcpPort, getDaemonTcpHost: options.getDaemonTcpHost, + voice: options.voice, }); } +class FakeVoiceTurnDetectionSession extends EventEmitter implements TurnDetectionSession { + public readonly requiredSampleRate = 16000; + + async connect(): Promise {} + + appendPcm16(_chunk: Buffer): void {} + + flush(): void {} + reset(): void {} + close(): void {} +} + +class FakeVoiceSttSession extends EventEmitter implements StreamingTranscriptionSession { + public readonly requiredSampleRate = 16000; + public commitCount = 0; + + async connect(): Promise {} + + appendPcm16(_pcm16le: Buffer): void {} + + commit(): void { + this.commitCount += 1; + } + + clear(): void {} + close(): void {} + + emitCommitted(event: StreamingTranscriptionCommittedEvent): void { + this.emit("committed", event); + } + + emitTranscript(event: StreamingTranscriptionEvent): void { + this.emit("transcript", event); + } +} + +function createVoiceSessionHarness() { + const messages: unknown[] = []; + const detector = new FakeVoiceTurnDetectionSession(); + const sttSession = new FakeVoiceSttSession(); + const sttProvider: SpeechToTextProvider = { + id: "local", + createSession: vi.fn(() => sttSession), + }; + const turnDetection: TurnDetectionProvider = { + id: "local", + createSession: vi.fn(() => detector), + }; + const session = createSessionForTest({ + messages, + stt: sttProvider, + voice: { turnDetection }, + }); + Object.assign(session, { + isVoiceMode: true, + voiceModeAgentId: "11111111-1111-4111-8111-111111111111", + }); + const internals = asSessionInternals(session); + const sendAgentMessage = vi + .spyOn(internals, "handleSendAgentMessage") + .mockResolvedValue({ ok: true }); + const transcribe = vi.spyOn( + ( + session as unknown as { + sttManager: { + transcribe(audio: Buffer, format: string): Promise; + }; + } + ).sttManager, + "transcribe", + ); + + return { + session, + internals, + messages, + detector, + sttSession, + sendAgentMessage, + transcribe, + }; +} + +async function settleVoiceSession(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +describe("session voice mode streaming transcription", () => { + test("submits the streaming final transcript to the agent without batch transcribe", async () => { + const harness = createVoiceSessionHarness(); + + await harness.internals.startVoiceTurnController(); + harness.detector.emit("speech_started"); + await settleVoiceSession(); + harness.detector.emit("speech_stopped"); + await settleVoiceSession(); + harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSession.emitTranscript({ + segmentId: "segment-1", + transcript: "ship the streaming final", + isFinal: true, + language: "en", + avgLogprob: -0.1, + isLowConfidence: false, + }); + await settleVoiceSession(); + + expect(harness.sttSession.commitCount).toBe(1); + expect(harness.transcribe).not.toHaveBeenCalled(); + expect(harness.sendAgentMessage).toHaveBeenCalledWith( + "11111111-1111-4111-8111-111111111111", + "ship the streaming final", + undefined, + undefined, + undefined, + undefined, + { spokenInput: true }, + ); + expect(harness.messages).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ + text: "ship the streaming final", + language: "en", + avgLogprob: -0.1, + }), + }), + ); + + await harness.internals.stopVoiceTurnController(); + }); + + test("uses the finalization timeout empty transcript path without agent submission", async () => { + vi.useFakeTimers(); + try { + const harness = createVoiceSessionHarness(); + + await harness.internals.startVoiceTurnController(); + harness.detector.emit("speech_started"); + await settleVoiceSession(); + harness.detector.emit("speech_stopped"); + await settleVoiceSession(); + harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + + await vi.advanceTimersByTimeAsync(10_000); + await settleVoiceSession(); + + expect(harness.transcribe).not.toHaveBeenCalled(); + expect(harness.sendAgentMessage).not.toHaveBeenCalled(); + expect(harness.messages).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ + text: "", + }), + }), + ); + + await harness.internals.stopVoiceTurnController(); + } finally { + vi.useRealTimers(); + } + }); + + test("filters low-confidence streaming finals without agent submission", async () => { + const harness = createVoiceSessionHarness(); + + await harness.internals.startVoiceTurnController(); + harness.detector.emit("speech_started"); + await settleVoiceSession(); + harness.detector.emit("speech_stopped"); + await settleVoiceSession(); + harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSession.emitTranscript({ + segmentId: "segment-1", + transcript: "background noise", + isFinal: true, + avgLogprob: -2.5, + isLowConfidence: true, + }); + await settleVoiceSession(); + + expect(harness.transcribe).not.toHaveBeenCalled(); + expect(harness.sendAgentMessage).not.toHaveBeenCalled(); + expect(harness.messages).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ + text: "", + avgLogprob: -2.5, + isLowConfidence: true, + }), + }), + ); + + await harness.internals.stopVoiceTurnController(); + }); +}); + describe("file explorer binary responses", () => { const tempDirs: string[] = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 96f54bf80..af6a4b018 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -520,7 +520,6 @@ const MIN_STREAMING_SEGMENT_BYTES = Math.round( PCM_BYTES_PER_MS * MIN_STREAMING_SEGMENT_DURATION_MS, ); const AgentIdSchema = z.string().uuid(); -const VOICE_INTERRUPT_CONFIRMATION_MS = 500; const AVAILABLE_EDITOR_TARGETS_CACHE_TTL_MS = 60_000; const AVAILABLE_EDITOR_TARGETS_CACHE_KEY = "available"; @@ -764,8 +763,6 @@ export class Session { // Voice mode state private isVoiceMode = false; private speechInProgress = false; - private pendingVoiceSpeechStartAt: number | null = null; - private pendingVoiceSpeechTimer: ReturnType | null = null; private dictationStreamManager!: DictationStreamManager; private resolveVoiceTurnDetection!: () => TurnDetectionProvider | null; @@ -2783,6 +2780,10 @@ export class Session { if (!turnDetection) { throw new Error("Voice turn detection is not configured"); } + const stt = this.sttManager.getProvider(); + if (!stt) { + throw new Error("Voice speech-to-text is not configured"); + } this.sessionLogger.info( { providerId: turnDetection.id }, @@ -2792,27 +2793,69 @@ export class Session { const controller = createVoiceTurnController({ logger: this.sessionLogger.child({ component: "voice-turn-controller" }), turnDetection, - utteranceSink: { - submitUtterance: async ({ pcm16, format, sampleRate, startedAt, endedAt }) => { - this.sessionLogger.debug( - { - audioBytes: pcm16.length, - sampleRate, - startedAt, - endedAt, - durationMs: Math.max(0, endedAt - startedAt), - }, - "Submitting detected voice utterance", - ); - await this.processCompletedAudio(pcm16, format); - }, - }, + stt, callbacks: { onSpeechStarted: async () => { - this.handleProvisionalVoiceSpeechStarted(); + this.sessionLogger.debug("Voice VAD speech_started"); + }, + onPartialTranscript: async ({ segmentId, transcript }) => { + this.sessionLogger.info( + { segmentId, transcriptLength: transcript.trim().length }, + "voice_input_state emitting isSpeaking=true", + ); + this.emit({ + type: "voice_input_state", + payload: { + isSpeaking: true, + }, + }); + await this.handleVoiceSpeechStart(); }, onSpeechStopped: async () => { this.handleVoiceSpeechStopped(); + this.setPhase("transcribing"); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "system", + content: "Transcribing audio...", + }, + }); + }, + onFinalTranscript: async ({ + transcript, + language, + durationMs, + avgLogprob, + isLowConfidence, + }) => { + const requestId = uuidv4(); + const transcriptText = isLowConfidence ? "" : transcript.trim(); + if (isLowConfidence) { + this.sessionLogger.debug( + { text: transcript, avgLogprob }, + "Filtered low-confidence transcription (likely non-speech)", + ); + } + this.sessionLogger.info( + { + requestId, + isVoiceMode: this.isVoiceMode, + transcriptLength: transcriptText.length, + transcript: transcriptText, + }, + "Transcription result", + ); + await this.handleTranscriptionResultPayload({ + text: transcriptText, + requestId, + ...(language ? { language } : {}), + duration: durationMs, + ...(avgLogprob !== undefined ? { avgLogprob } : {}), + ...(isLowConfidence !== undefined ? { isLowConfidence } : {}), + }); }, onError: (error) => { this.sessionLogger.error({ err: error }, "Voice turn controller failed"); @@ -2831,63 +2874,12 @@ export class Session { return; } - this.clearPendingVoiceSpeechStart("turn-controller-stop"); const controller = this.voiceTurnController; this.voiceTurnController = null; await controller.stop(); } - private clearPendingVoiceSpeechStart(reason: string): void { - if (this.pendingVoiceSpeechTimer) { - clearTimeout(this.pendingVoiceSpeechTimer); - this.pendingVoiceSpeechTimer = null; - } - if (this.pendingVoiceSpeechStartAt !== null) { - this.sessionLogger.debug({ reason }, "Clearing provisional voice speech start"); - this.pendingVoiceSpeechStartAt = null; - } - } - - private handleProvisionalVoiceSpeechStarted(): void { - if (this.speechInProgress || this.pendingVoiceSpeechTimer) { - return; - } - - const startedAt = Date.now(); - this.pendingVoiceSpeechStartAt = startedAt; - this.sessionLogger.info( - { confirmationMs: VOICE_INTERRUPT_CONFIRMATION_MS }, - "Silero VAD provisional speech_started", - ); - this.pendingVoiceSpeechTimer = setTimeout(() => { - this.pendingVoiceSpeechTimer = null; - if (this.pendingVoiceSpeechStartAt !== startedAt || this.speechInProgress) { - return; - } - - this.pendingVoiceSpeechStartAt = null; - this.sessionLogger.info("voice_input_state emitting isSpeaking=true"); - this.emit({ - type: "voice_input_state", - payload: { - isSpeaking: true, - }, - }); - void this.handleVoiceSpeechStart(); - }, VOICE_INTERRUPT_CONFIRMATION_MS); - } - private handleVoiceSpeechStopped(): void { - if (this.pendingVoiceSpeechStartAt !== null) { - const durationMs = Date.now() - this.pendingVoiceSpeechStartAt; - this.clearPendingVoiceSpeechStart("speech-stopped-before-confirmation"); - this.sessionLogger.info( - { durationMs, confirmationMs: VOICE_INTERRUPT_CONFIRMATION_MS }, - "Ignoring provisional voice speech start that ended before confirmation", - ); - return; - } - this.sessionLogger.info("voice_input_state emitting isSpeaking=false"); this.emit({ type: "voice_input_state", @@ -8023,7 +8015,6 @@ export class Session { * Clear speech-in-progress flag once the user turn has completed */ private clearSpeechInProgress(reason: string): void { - this.clearPendingVoiceSpeechStart(`clear-speech-in-progress:${reason}`); if (!this.speechInProgress) { return; } diff --git a/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.test.ts b/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.test.ts deleted file mode 100644 index c56401762..000000000 --- a/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.test.ts +++ /dev/null @@ -1,37 +0,0 @@ -import { describe, expect, it } from "vitest"; - -import { FixedDurationPcmRingBuffer } from "./fixed-duration-pcm-ring-buffer.js"; - -describe("FixedDurationPcmRingBuffer", () => { - it("retains only the configured prefix window", () => { - const buffer = new FixedDurationPcmRingBuffer({ - sampleRate: 1000, - channels: 1, - bitsPerSample: 16, - durationMs: 100, - }); - - buffer.append(Buffer.alloc(120)); - buffer.append(Buffer.alloc(120, 1)); - - const drained = buffer.drain(); - expect(drained.length).toBe(120); - expect(Array.from(drained.slice(0, 4))).toEqual([1, 1, 1, 1]); - }); - - it("drains buffered chunks in arrival order", () => { - const buffer = new FixedDurationPcmRingBuffer({ - sampleRate: 1000, - channels: 1, - bitsPerSample: 16, - durationMs: 500, - }); - - buffer.append(Buffer.from([1, 2])); - buffer.append(Buffer.from([3, 4])); - buffer.append(Buffer.from([5, 6])); - - expect(Array.from(buffer.drain())).toEqual([1, 2, 3, 4, 5, 6]); - expect(buffer.drain().length).toBe(0); - }); -}); diff --git a/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.ts b/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.ts deleted file mode 100644 index 3f2ef75ec..000000000 --- a/packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.ts +++ /dev/null @@ -1,47 +0,0 @@ -export class FixedDurationPcmRingBuffer { - private readonly maxBytes: number; - private chunks: Buffer[] = []; - private totalBytes = 0; - - constructor(params: { - sampleRate: number; - channels: number; - bitsPerSample: number; - durationMs: number; - }) { - const bytesPerSecond = params.sampleRate * params.channels * (params.bitsPerSample / 8); - this.maxBytes = Math.max(1, Math.round((bytesPerSecond * params.durationMs) / 1000)); - } - - append(chunk: Buffer): void { - if (chunk.length === 0) { - return; - } - - this.chunks.push(chunk); - this.totalBytes += chunk.length; - - while (this.totalBytes > this.maxBytes && this.chunks.length > 0) { - const removed = this.chunks.shift(); - if (!removed) { - break; - } - this.totalBytes -= removed.length; - } - } - - drain(): Buffer { - const combined = Buffer.concat(this.chunks); - this.clear(); - return combined; - } - - get byteLength(): number { - return this.totalBytes; - } - - clear(): void { - this.chunks = []; - this.totalBytes = 0; - } -} diff --git a/packages/server/src/server/voice/voice-turn-controller.test.ts b/packages/server/src/server/voice/voice-turn-controller.test.ts index 92901539b..bc4ab0d3d 100644 --- a/packages/server/src/server/voice/voice-turn-controller.test.ts +++ b/packages/server/src/server/voice/voice-turn-controller.test.ts @@ -2,11 +2,17 @@ import { EventEmitter } from "node:events"; import { describe, expect, it, vi } from "vitest"; import pino from "pino"; +import type { + SpeechToTextProvider, + StreamingTranscriptionCommittedEvent, + StreamingTranscriptionEvent, + StreamingTranscriptionSession, +} from "../speech/speech-provider.js"; import type { TurnDetectionProvider, TurnDetectionSession, } from "../speech/turn-detection-provider.js"; -import { createVoiceTurnController, type DetectedVoiceUtterance } from "./voice-turn-controller.js"; +import { createVoiceTurnController } from "./voice-turn-controller.js"; class FakeTurnDetectionSession extends EventEmitter implements TurnDetectionSession { public readonly requiredSampleRate = 16000; @@ -23,6 +29,49 @@ class FakeTurnDetectionSession extends EventEmitter implements TurnDetectionSess close(): void {} } +class FakeSttSession extends EventEmitter implements StreamingTranscriptionSession { + public readonly requiredSampleRate: number; + public readonly appendedChunks: Buffer[] = []; + public connectCount = 0; + public closeCount = 0; + public commitCount = 0; + + constructor(requiredSampleRate = 16000) { + super(); + this.requiredSampleRate = requiredSampleRate; + } + + async connect(): Promise { + this.connectCount += 1; + } + + appendPcm16(chunk: Buffer): void { + this.appendedChunks.push(chunk); + } + + commit(): void { + this.commitCount += 1; + } + + clear(): void {} + + close(): void { + this.closeCount += 1; + } + + emitTranscript(event: StreamingTranscriptionEvent): void { + this.emit("transcript", event); + } + + emitCommitted(event: StreamingTranscriptionCommittedEvent): void { + this.emit("committed", event); + } + + emitError(error: unknown): void { + this.emit("error", error); + } +} + function createFakeTurnDetectionProvider(session: FakeTurnDetectionSession): TurnDetectionProvider { return { id: "local", @@ -32,6 +81,17 @@ function createFakeTurnDetectionProvider(session: FakeTurnDetectionSession): Tur }; } +function createFakeSttProvider(sessions: FakeSttSession[]): SpeechToTextProvider { + return { + id: "local", + createSession() { + const session = new FakeSttSession(); + sessions.push(session); + return session; + }, + }; +} + async function settleSerialQueue(): Promise { await Promise.resolve(); await Promise.resolve(); @@ -40,21 +100,34 @@ async function settleSerialQueue(): Promise { function createControllerHarness() { const detector = new FakeTurnDetectionSession(); + const sttSessions: FakeSttSession[] = []; + const stt = createFakeSttProvider(sttSessions); const onSpeechStarted = vi.fn(async () => {}); const onSpeechStopped = vi.fn(async () => {}); - const submitUtterance = vi.fn(async (_utterance: DetectedVoiceUtterance) => {}); + const onPartialTranscript = vi.fn( + async (_input: { segmentId: string; transcript: string }) => {}, + ); + const onFinalTranscript = vi.fn( + async (_input: { + segmentId: string; + transcript: string; + language?: string; + avgLogprob?: number; + isLowConfidence?: boolean; + durationMs: number; + }) => {}, + ); const onError = vi.fn(); const controller = createVoiceTurnController({ logger: pino({ level: "silent" }), turnDetection: createFakeTurnDetectionProvider(detector), - prefixDurationMs: 100, - utteranceSink: { - submitUtterance, - }, + stt, callbacks: { onSpeechStarted, onSpeechStopped, + onPartialTranscript, + onFinalTranscript, onError, }, }); @@ -62,25 +135,17 @@ function createControllerHarness() { return { controller, detector, + sttSessions, onSpeechStarted, onSpeechStopped, - submitUtterance, + onPartialTranscript, + onFinalTranscript, onError, }; } -function createDeferredPromise() { - let resolve!: (value: T | PromiseLike) => void; - let reject!: (reason?: unknown) => void; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - return { promise, resolve, reject }; -} - describe("voice turn controller", () => { - it("buffers audio before speech start and includes the prefix in the utterance", async () => { + it("forwards audio to the detector and streaming STT without submitting buffered utterances", async () => { const harness = createControllerHarness(); await harness.controller.start(); @@ -88,63 +153,30 @@ describe("voice turn controller", () => { audioBase64: Buffer.from([1, 2, 3, 4]).toString("base64"), format: "audio/pcm;rate=16000;bits=16", }); + harness.detector.emit("speech_started"); + await settleSerialQueue(); await harness.controller.appendClientChunk({ audioBase64: Buffer.from([5, 6, 7, 8]).toString("base64"), format: "audio/pcm;rate=16000;bits=16", }); - - harness.detector.emit("speech_started"); - await settleSerialQueue(); - - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([9, 10, 11, 12]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - harness.detector.emit("speech_stopped"); await settleSerialQueue(); + expect(harness.detector.appendedChunks).toEqual([ + Buffer.from([1, 2, 3, 4]), + Buffer.from([5, 6, 7, 8]), + ]); + expect(harness.sttSessions[0]?.appendedChunks).toEqual([ + Buffer.from([1, 2, 3, 4]), + Buffer.from([5, 6, 7, 8]), + ]); expect(harness.onSpeechStarted).toHaveBeenCalledTimes(1); expect(harness.onSpeechStopped).toHaveBeenCalledTimes(1); - expect(harness.submitUtterance).toHaveBeenCalledTimes(1); - expect(harness.submitUtterance).toHaveBeenCalledWith( - expect.objectContaining({ - pcm16: Buffer.from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]), - sampleRate: 16000, - format: "audio/pcm;rate=16000;bits=16", - }), - ); + expect(harness.onFinalTranscript).not.toHaveBeenCalled(); expect(harness.onError).not.toHaveBeenCalled(); }); - it("finalizes one utterance when speech stops", async () => { - const harness = createControllerHarness(); - - await harness.controller.start(); - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([1, 1, 1, 1]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - - harness.detector.emit("speech_started"); - await settleSerialQueue(); - - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([2, 2, 2, 2]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - - harness.detector.emit("speech_stopped"); - await settleSerialQueue(); - harness.detector.emit("speech_stopped"); - await settleSerialQueue(); - - expect(harness.submitUtterance).toHaveBeenCalledTimes(1); - expect(harness.onSpeechStopped).toHaveBeenCalledTimes(1); - expect(harness.onError).not.toHaveBeenCalled(); - }); - - it("does not barge in or emit an utterance on silence-only chunks", async () => { + it("does not barge in on silence-only chunks", async () => { const harness = createControllerHarness(); await harness.controller.start(); @@ -164,111 +196,320 @@ describe("voice turn controller", () => { ]); expect(harness.onSpeechStarted).not.toHaveBeenCalled(); expect(harness.onSpeechStopped).not.toHaveBeenCalled(); - expect(harness.submitUtterance).not.toHaveBeenCalled(); + expect(harness.onFinalTranscript).not.toHaveBeenCalled(); expect(harness.onError).not.toHaveBeenCalled(); }); - it("retains a rolling prefix for rapid follow-up utterances", async () => { + it("fires onPartialTranscript exactly once for the first non-empty partial in a turn", async () => { const harness = createControllerHarness(); await harness.controller.start(); - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([1, 2, 3, 4]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - harness.detector.emit("speech_started"); await settleSerialQueue(); - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([5, 6, 7, 8]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "hello", + isFinal: false, + }); + await settleSerialQueue(); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "hello again", + isFinal: false, }); - - harness.detector.emit("speech_stopped"); await settleSerialQueue(); - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([9, 10, 11, 12]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", + expect(harness.onPartialTranscript).toHaveBeenCalledTimes(1); + expect(harness.onPartialTranscript).toHaveBeenCalledWith({ + segmentId: "segment-1", + transcript: "hello", }); + }); + it("does not fire onPartialTranscript for filler-only partials, but fires once the partial grows past the filler", async () => { + const harness = createControllerHarness(); + + await harness.controller.start(); harness.detector.emit("speech_started"); await settleSerialQueue(); - await harness.controller.appendClientChunk({ - audioBase64: Buffer.from([13, 14, 15, 16]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); + for (const transcript of ["uh", "uh,", "Um", "uh um", "hmm"]) { + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript, + isFinal: false, + }); + await settleSerialQueue(); + } + expect(harness.onPartialTranscript).not.toHaveBeenCalled(); + + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "uh hello", + isFinal: false, + }); + await settleSerialQueue(); + + expect(harness.onPartialTranscript).toHaveBeenCalledTimes(1); + expect(harness.onPartialTranscript).toHaveBeenCalledWith({ + segmentId: "segment-1", + transcript: "uh hello", + }); + }); + + it("does not fire onPartialTranscript for empty or whitespace-only partials", async () => { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "", + isFinal: false, + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: " ", + isFinal: false, + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "ignored final", + isFinal: true, + }); + await settleSerialQueue(); + + expect(harness.onPartialTranscript).not.toHaveBeenCalled(); + }); + + it("does not fire onPartialTranscript from VAD speech_started alone", async () => { + vi.useFakeTimers(); + try { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + + await vi.advanceTimersByTimeAsync(5_000); + await settleSerialQueue(); + + expect(harness.onSpeechStarted).toHaveBeenCalledTimes(1); + expect(harness.onPartialTranscript).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("commits the streaming STT segment when speech stops", async () => { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); harness.detector.emit("speech_stopped"); await settleSerialQueue(); - expect(harness.submitUtterance).toHaveBeenCalledTimes(2); - expect(harness.onSpeechStopped).toHaveBeenCalledTimes(2); - expect(harness.submitUtterance.mock.calls[0]?.[0]).toEqual( + expect(harness.sttSessions[0]?.commitCount).toBe(1); + }); + + it("fires onFinalTranscript after speech stop, commit, and final transcript", async () => { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + harness.detector.emit("speech_stopped"); + await settleSerialQueue(); + + harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: " hello there ", + isFinal: true, + language: "en", + avgLogprob: -0.2, + isLowConfidence: false, + }); + await settleSerialQueue(); + + expect(harness.onFinalTranscript).toHaveBeenCalledTimes(1); + expect(harness.onFinalTranscript).toHaveBeenCalledWith( expect.objectContaining({ - pcm16: Buffer.from([1, 2, 3, 4, 5, 6, 7, 8]), - }), - ); - expect(harness.submitUtterance.mock.calls[1]?.[0]).toEqual( - expect.objectContaining({ - pcm16: Buffer.from([5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]), + segmentId: "segment-1", + transcript: "hello there", + language: "en", + avgLogprob: -0.2, + durationMs: expect.any(Number), }), ); }); - it("continues detecting speech while a previous utterance submission is still pending", async () => { - const deferred = createDeferredPromise(); - const detector = new FakeTurnDetectionSession(); - const onSpeechStarted = vi.fn(async () => {}); - const onSpeechStopped = vi.fn(async () => {}); - const submitUtterance = vi.fn(async () => { - await deferred.promise; - }); - const onError = vi.fn(); + it("assembles multiple committed final transcripts in commit order", async () => { + const harness = createControllerHarness(); - const controller = createVoiceTurnController({ - logger: pino({ level: "silent" }), - turnDetection: createFakeTurnDetectionProvider(detector), - prefixDurationMs: 100, - utteranceSink: { - submitUtterance, - }, - callbacks: { - onSpeechStarted, - onSpeechStopped, - onError, - }, - }); - - await controller.start(); - await controller.appendClientChunk({ - audioBase64: Buffer.from([1, 2, 3, 4]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - detector.emit("speech_started"); + await harness.controller.start(); + harness.detector.emit("speech_started"); await settleSerialQueue(); - await controller.appendClientChunk({ - audioBase64: Buffer.from([5, 6, 7, 8]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", - }); - detector.emit("speech_stopped"); + harness.detector.emit("speech_stopped"); await settleSerialQueue(); - await controller.appendClientChunk({ - audioBase64: Buffer.from([9, 10, 11, 12]).toString("base64"), - format: "audio/pcm;rate=16000;bits=16", + harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSessions[0]?.emitCommitted({ + segmentId: "segment-2", + previousSegmentId: "segment-1", + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-2", + transcript: " world ", + isFinal: true, + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: " hello ", + isFinal: true, }); - detector.emit("speech_started"); await settleSerialQueue(); - expect(onSpeechStarted).toHaveBeenCalledTimes(2); - expect(onSpeechStopped).toHaveBeenCalledTimes(1); - expect(submitUtterance).toHaveBeenCalledTimes(1); - expect(onError).not.toHaveBeenCalled(); + expect(harness.onFinalTranscript).toHaveBeenCalledTimes(1); + expect(harness.onFinalTranscript).toHaveBeenCalledWith( + expect.objectContaining({ + segmentId: "segment-1", + transcript: "hello world", + }), + ); + }); - deferred.resolve(); + it("fires the finalization timeout with whatever finals arrived", async () => { + vi.useFakeTimers(); + try { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + harness.detector.emit("speech_stopped"); + await settleSerialQueue(); + + harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSessions[0]?.emitCommitted({ + segmentId: "segment-2", + previousSegmentId: "segment-1", + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "hello", + isFinal: true, + }); + + await vi.advanceTimersByTimeAsync(10_000); + await settleSerialQueue(); + + expect(harness.onFinalTranscript).toHaveBeenCalledTimes(1); + expect(harness.onFinalTranscript).toHaveBeenCalledWith( + expect.objectContaining({ + segmentId: "segment-1", + transcript: "hello", + }), + ); + } finally { + vi.useRealTimers(); + } + }); + + it("does not include stale uncommitted finals once committed segments are known", async () => { + vi.useFakeTimers(); + try { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + harness.detector.emit("speech_stopped"); + await settleSerialQueue(); + + harness.sttSessions[0]?.emitTranscript({ + segmentId: "stale-segment", + transcript: "stale text", + isFinal: true, + }); + harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + harness.sttSessions[0]?.emitCommitted({ + segmentId: "segment-2", + previousSegmentId: "segment-1", + }); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "fresh text", + isFinal: true, + }); + + await vi.advanceTimersByTimeAsync(10_000); + await settleSerialQueue(); + + expect(harness.onFinalTranscript).toHaveBeenCalledTimes(1); + expect(harness.onFinalTranscript).toHaveBeenCalledWith( + expect.objectContaining({ + segmentId: "segment-1", + transcript: "fresh text", + }), + ); + } finally { + vi.useRealTimers(); + } + }); + + it("fires the finalization timeout with an empty transcript when a turn only has a partial", async () => { + vi.useFakeTimers(); + try { + const harness = createControllerHarness(); + + await harness.controller.start(); + harness.detector.emit("speech_started"); + await settleSerialQueue(); + harness.sttSessions[0]?.emitTranscript({ + segmentId: "segment-1", + transcript: "hello", + isFinal: false, + }); + await settleSerialQueue(); + harness.detector.emit("speech_stopped"); + await settleSerialQueue(); + harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + + await vi.advanceTimersByTimeAsync(10_000); + await settleSerialQueue(); + + expect(harness.onFinalTranscript).toHaveBeenCalledTimes(1); + expect(harness.onFinalTranscript).toHaveBeenCalledWith( + expect.objectContaining({ + segmentId: "segment-1", + transcript: "", + }), + ); + } finally { + vi.useRealTimers(); + } + }); + + it("reports STT errors and attempts one reconnect", async () => { + const harness = createControllerHarness(); + + await harness.controller.start(); + const firstSession = harness.sttSessions[0]; + firstSession?.emitError(new Error("stream failed")); await settleSerialQueue(); + + expect(harness.onError).toHaveBeenCalledTimes(1); + expect(harness.onError).toHaveBeenCalledWith( + expect.objectContaining({ message: "stream failed" }), + ); + expect(firstSession?.closeCount).toBe(1); + expect(harness.sttSessions).toHaveLength(2); + expect(harness.sttSessions[1]?.connectCount).toBe(1); }); }); diff --git a/packages/server/src/server/voice/voice-turn-controller.ts b/packages/server/src/server/voice/voice-turn-controller.ts index 5a38f03b9..2cab7f2ba 100644 --- a/packages/server/src/server/voice/voice-turn-controller.ts +++ b/packages/server/src/server/voice/voice-turn-controller.ts @@ -4,80 +4,339 @@ import { v4 as uuidv4 } from "uuid"; import { Pcm16MonoResampler } from "../agent/pcm16-resampler.js"; import { parsePcmRateFromFormat } from "../speech/audio.js"; +import type { + SpeechToTextProvider, + StreamingTranscriptionEvent, + StreamingTranscriptionSession, +} from "../speech/speech-provider.js"; import type { TurnDetectionProvider } from "../speech/turn-detection-provider.js"; -import { FixedDurationPcmRingBuffer } from "./fixed-duration-pcm-ring-buffer.js"; -const PCM_CHANNELS = 1; -const PCM_BITS_PER_SAMPLE = 16; -const DEFAULT_PREFIX_DURATION_MS = 1000; +const VOICE_FINAL_TRANSCRIPT_TIMEOUT_MS = 10_000; + +const FILLER_PARTIAL_WORDS = new Set([ + "uh", + "um", + "ah", + "eh", + "er", + "hmm", + "mm", + "mmm", + "mhm", + "huh", + "uhhuh", + "uh-huh", + "oh", +]); + +function isFillerOnlyPartial(transcript: string): boolean { + const tokens = transcript + .toLowerCase() + .replace(/[^\p{L}\p{N}\s'-]/gu, "") + .split(/\s+/) + .filter(Boolean); + if (tokens.length === 0) { + return true; + } + return tokens.every((token) => FILLER_PARTIAL_WORDS.has(token)); +} type VoiceInputState = | { status: "idle" } - | { status: "listening"; rollingPrefixBytes: number } + | { status: "listening" } | { status: "capturing"; utteranceId: string; startedAt: number; - rollingPrefixBytes: number; - utteranceBytes: number; }; export interface VoiceTurnControllerCallbacks { onSpeechStarted(): Promise; onSpeechStopped(): Promise; + onPartialTranscript(input: { segmentId: string; transcript: string }): Promise; + onFinalTranscript(input: VoiceFinalTranscript): Promise; onError(error: Error): void; } -export interface DetectedVoiceUtterance { - pcm16: Buffer; - sampleRate: number; - format: string; - startedAt: number; - endedAt: number; -} - -export interface VoiceUtteranceSink { - submitUtterance(utterance: DetectedVoiceUtterance): Promise; -} - export interface VoiceTurnController { start(): Promise; stop(): Promise; appendClientChunk(input: { audioBase64: string; format: string }): Promise; } +interface TranscriptSegmentMeta { + language?: string; + avgLogprob?: number; + isLowConfidence?: boolean; +} + +interface VoiceFinalTranscript { + segmentId: string; + transcript: string; + language?: string; + avgLogprob?: number; + isLowConfidence?: boolean; + durationMs: number; +} + +interface FinalizingVoiceTurn { + turnId: string; + startedAt: number; + committedSegmentIds: string[]; + transcriptsBySegmentId: Map; + finalTranscriptSegmentIds: Set; + transcriptMetaBySegmentId: Map; + timeout: ReturnType; + fired: boolean; +} + export function createVoiceTurnController(params: { logger: Logger; turnDetection: TurnDetectionProvider; - utteranceSink: VoiceUtteranceSink; + stt: SpeechToTextProvider; callbacks: VoiceTurnControllerCallbacks; - prefixDurationMs?: number; }): VoiceTurnController { const detector = params.turnDetection.createSession({ logger: params.logger.child({ component: "turn-detection" }), }); - const prefixBuffer = new FixedDurationPcmRingBuffer({ - sampleRate: detector.requiredSampleRate, - channels: PCM_CHANNELS, - bitsPerSample: PCM_BITS_PER_SAMPLE, - durationMs: params.prefixDurationMs ?? DEFAULT_PREFIX_DURATION_MS, - }); let state: VoiceInputState = { status: "idle" }; let resampler: Pcm16MonoResampler | null = null; + let sttSession: StreamingTranscriptionSession | null = null; + let sttResampler: Pcm16MonoResampler | null = null; let inputRate = detector.requiredSampleRate; - let utteranceChunks: Buffer[] = []; + let sttInputRate = 0; let queued = Promise.resolve(); - let submissionQueue = Promise.resolve(); - - function buildVoicePcmFormat(sampleRate: number): string { - return `audio/pcm;rate=${sampleRate};bits=16`; - } + let activeTranscriptSegmentId: string | null = null; + let partialTranscriptFired = false; + let reconnectAttemptedForTurn = false; + const sealedTranscriptSegmentIds = new Set(); + let currentFinalizingTurn: FinalizingVoiceTurn | null = null; function fail(error: unknown): void { params.callbacks.onError(error instanceof Error ? error : new Error(String(error))); } + function firePartialTranscript(segmentId: string, transcript: string): void { + if (partialTranscriptFired || state.status !== "capturing") { + return; + } + + partialTranscriptFired = true; + void runSerial(async () => { + await params.callbacks.onPartialTranscript({ segmentId, transcript }); + }); + } + + function clearFinalizingTurnTimeout(): void { + if (currentFinalizingTurn) { + clearTimeout(currentFinalizingTurn.timeout); + } + } + + function getFinalizingTurnForSegment(segmentId: string): FinalizingVoiceTurn | null { + if (!currentFinalizingTurn) { + return null; + } + + if (currentFinalizingTurn.committedSegmentIds.includes(segmentId)) { + return currentFinalizingTurn; + } + + if (currentFinalizingTurn.committedSegmentIds.length > 0) { + return null; + } + + if (activeTranscriptSegmentId && activeTranscriptSegmentId !== segmentId) { + return null; + } + + return currentFinalizingTurn; + } + + function getOrderedFinalSegmentIds(turn: FinalizingVoiceTurn): string[] { + if (turn.committedSegmentIds.length === 0) { + return [...turn.finalTranscriptSegmentIds]; + } + + return turn.committedSegmentIds.filter((segmentId) => + turn.finalTranscriptSegmentIds.has(segmentId), + ); + } + + function assembleFinalTranscript(turn: FinalizingVoiceTurn): VoiceFinalTranscript { + const orderedFinalSegmentIds = getOrderedFinalSegmentIds(turn); + const transcript = orderedFinalSegmentIds + .map((segmentId) => turn.transcriptsBySegmentId.get(segmentId)?.trim() ?? "") + .filter((segment) => segment.length > 0) + .join(" ") + .trim(); + const orderedFinalMeta = orderedFinalSegmentIds + .map((segmentId) => turn.transcriptMetaBySegmentId.get(segmentId)) + .filter((meta): meta is TranscriptSegmentMeta => Boolean(meta)); + const language = orderedFinalMeta.find((meta) => meta.language)?.language; + const singleSegmentMeta = orderedFinalMeta.length === 1 ? orderedFinalMeta[0] : null; + const allLowConfidence = + orderedFinalMeta.length > 0 && + orderedFinalMeta.every((meta) => meta.isLowConfidence === true); + + return { + segmentId: turn.committedSegmentIds[0] ?? orderedFinalSegmentIds[0] ?? turn.turnId, + transcript, + ...(language ? { language } : {}), + ...(singleSegmentMeta?.avgLogprob !== undefined + ? { avgLogprob: singleSegmentMeta.avgLogprob } + : {}), + ...(allLowConfidence ? { isLowConfidence: true } : {}), + durationMs: Math.max(0, Date.now() - turn.startedAt), + }; + } + + function fireFinalTranscript(turn: FinalizingVoiceTurn, reason: "complete" | "timeout"): void { + if (turn.fired || currentFinalizingTurn?.turnId !== turn.turnId) { + return; + } + + turn.fired = true; + clearTimeout(turn.timeout); + currentFinalizingTurn = null; + + const finalTranscript = assembleFinalTranscript(turn); + if (reason === "timeout") { + params.logger.warn( + { + turnId: turn.turnId, + committedSegments: turn.committedSegmentIds.length, + receivedFinals: turn.finalTranscriptSegmentIds.size, + timeoutMs: VOICE_FINAL_TRANSCRIPT_TIMEOUT_MS, + transcriptLength: finalTranscript.transcript.length, + }, + "voice_turn.final_transcript_timeout", + ); + } + + void runSerial(async () => { + await params.callbacks.onFinalTranscript(finalTranscript); + }); + } + + function maybeFireFinalTranscript(turn: FinalizingVoiceTurn): void { + if (turn.fired || turn.committedSegmentIds.length === 0) { + return; + } + + const allCommittedSegmentsFinal = turn.committedSegmentIds.every((segmentId) => + turn.finalTranscriptSegmentIds.has(segmentId), + ); + if (allCommittedSegmentsFinal) { + fireFinalTranscript(turn, "complete"); + } + } + + async function reconnectSttSession(): Promise { + const previousSession = sttSession; + sttSession = null; + sttResampler = null; + sttInputRate = 0; + previousSession?.close(); + + try { + const nextSession = createSttSession(); + await nextSession.connect(); + sttSession = nextSession; + params.logger.info("voice_turn.stt_reconnected"); + } catch (error) { + fail(error); + params.logger.warn({ err: error }, "voice_turn.stt_reconnect_failed"); + } + } + + function handleSttError(error: unknown): void { + fail(error); + params.logger.warn({ err: error }, "voice_turn.stt_error"); + if (reconnectAttemptedForTurn) { + sttSession?.close(); + sttSession = null; + return; + } + + reconnectAttemptedForTurn = true; + void runSerial(reconnectSttSession); + } + + function handleFinalSttTranscript(event: StreamingTranscriptionEvent): void { + const turn = getFinalizingTurnForSegment(event.segmentId); + if (!turn || turn.fired) { + return; + } + + turn.transcriptsBySegmentId.set(event.segmentId, event.transcript); + turn.finalTranscriptSegmentIds.add(event.segmentId); + turn.transcriptMetaBySegmentId.set(event.segmentId, { + ...(event.language ? { language: event.language } : {}), + ...(event.avgLogprob !== undefined ? { avgLogprob: event.avgLogprob } : {}), + ...(event.isLowConfidence !== undefined ? { isLowConfidence: event.isLowConfidence } : {}), + }); + maybeFireFinalTranscript(turn); + } + + function handlePartialSttTranscript(event: StreamingTranscriptionEvent): void { + if (state.status !== "capturing" || partialTranscriptFired) { + return; + } + + if (sealedTranscriptSegmentIds.has(event.segmentId)) { + return; + } + + if (activeTranscriptSegmentId && event.segmentId !== activeTranscriptSegmentId) { + return; + } + + const transcript = event.transcript.trim(); + if (!transcript) { + return; + } + + activeTranscriptSegmentId = event.segmentId; + + if (isFillerOnlyPartial(transcript)) { + return; + } + + firePartialTranscript(event.segmentId, transcript); + } + + function handleSttTranscript(event: StreamingTranscriptionEvent): void { + if (event.isFinal) { + handleFinalSttTranscript(event); + return; + } + + handlePartialSttTranscript(event); + } + + function createSttSession(): StreamingTranscriptionSession { + const session = params.stt.createSession({ + logger: params.logger.child({ component: "stt" }), + language: "en", + }); + session.on("transcript", handleSttTranscript); + session.on("committed", ({ segmentId }) => { + sealedTranscriptSegmentIds.add(segmentId); + if (state.status === "capturing" && !activeTranscriptSegmentId) { + activeTranscriptSegmentId = segmentId; + } + const turn = currentFinalizingTurn; + if (turn && !turn.committedSegmentIds.includes(segmentId)) { + turn.committedSegmentIds.push(segmentId); + maybeFireFinalTranscript(turn); + } + }); + session.on("error", handleSttError); + return session; + } + function runSerial(task: () => Promise): Promise { queued = queued.then(task).catch((error) => { fail(error); @@ -85,15 +344,57 @@ export function createVoiceTurnController(params: { return queued; } - function enqueueUtteranceSubmission(utterance: DetectedVoiceUtterance): void { - submissionQueue = submissionQueue - .then(async () => { - await params.utteranceSink.submitUtterance(utterance); - return; - }) - .catch((error) => { - fail(error); - }); + function updateDetectorResampler(parsedInputRate: number): void { + if (parsedInputRate === inputRate) { + return; + } + + inputRate = parsedInputRate; + resampler = + inputRate === detector.requiredSampleRate + ? null + : new Pcm16MonoResampler({ + inputRate, + outputRate: detector.requiredSampleRate, + }); + } + + function updateSttResampler( + session: StreamingTranscriptionSession, + parsedInputRate: number, + ): void { + if (parsedInputRate === sttInputRate) { + return; + } + + sttInputRate = parsedInputRate; + sttResampler = + sttInputRate === session.requiredSampleRate + ? null + : new Pcm16MonoResampler({ + inputRate: sttInputRate, + outputRate: session.requiredSampleRate, + }); + } + + function pcmForDetector(pcm16: Buffer): Buffer { + return resampler === null ? pcm16 : resampler.processChunk(pcm16); + } + + function pcmForStt( + session: StreamingTranscriptionSession, + pcm16: Buffer, + detectorPcm16: Buffer, + ): Buffer { + if (session.requiredSampleRate === detector.requiredSampleRate) { + return detectorPcm16; + } + + if (sttResampler === null) { + return pcm16; + } + + return sttResampler.processChunk(pcm16); } async function handleSpeechStarted(): Promise { @@ -103,21 +404,20 @@ export function createVoiceTurnController(params: { await params.callbacks.onSpeechStarted(); - const prefix = prefixBuffer.drain(); const startedAt = Date.now(); - utteranceChunks = prefix.length > 0 ? [prefix] : []; + activeTranscriptSegmentId = null; + partialTranscriptFired = false; + reconnectAttemptedForTurn = false; + clearFinalizingTurnTimeout(); + currentFinalizingTurn = null; state = { status: "capturing", utteranceId: uuidv4(), startedAt, - rollingPrefixBytes: prefixBuffer.byteLength, - utteranceBytes: prefix.length, }; params.logger.info( { utteranceId: state.utteranceId, - prefixBytes: prefix.length, - rollingPrefixBytes: prefixBuffer.byteLength, }, "voice_turn.speech_started", ); @@ -128,36 +428,41 @@ export function createVoiceTurnController(params: { return; } - const utterance = Buffer.concat(utteranceChunks); + const turnId = state.utteranceId; const startedAt = state.startedAt; const endedAt = Date.now(); - utteranceChunks = []; - state = { status: "listening", rollingPrefixBytes: prefixBuffer.byteLength }; + state = { status: "listening" }; + + const finalizingTurn: FinalizingVoiceTurn = { + turnId, + startedAt, + committedSegmentIds: [], + transcriptsBySegmentId: new Map(), + finalTranscriptSegmentIds: new Set(), + transcriptMetaBySegmentId: new Map(), + timeout: setTimeout(() => { + fireFinalTranscript(finalizingTurn, "timeout"); + }, VOICE_FINAL_TRANSCRIPT_TIMEOUT_MS), + fired: false, + }; + currentFinalizingTurn = finalizingTurn; detector.reset(); + try { + sttSession?.commit(); + } catch (error) { + handleSttError(error); + } await params.callbacks.onSpeechStopped(); params.logger.info( { - utteranceBytes: utterance.length, utteranceAgeMs: Math.max(0, endedAt - startedAt), }, "voice_turn.speech_stopped", ); - - if (utterance.length === 0) { - return; - } - - enqueueUtteranceSubmission({ - pcm16: utterance, - sampleRate: detector.requiredSampleRate, - format: buildVoicePcmFormat(detector.requiredSampleRate), - startedAt, - endedAt, - }); } detector.on("speech_started", () => { @@ -170,17 +475,23 @@ export function createVoiceTurnController(params: { return { async start(): Promise { + sttSession = createSttSession(); + await sttSession.connect(); await detector.connect(); - state = { status: "listening", rollingPrefixBytes: prefixBuffer.byteLength }; + state = { status: "listening" }; }, async stop(): Promise { await runSerial(async () => { + clearFinalizingTurnTimeout(); detector.close(); - prefixBuffer.clear(); - utteranceChunks = []; + sttSession?.close(); resampler?.reset(); + sttResampler?.reset(); resampler = null; + sttResampler = null; + sttSession = null; + currentFinalizingTurn = null; state = { status: "idle" }; }); }, @@ -200,39 +511,31 @@ export function createVoiceTurnController(params: { parsePcmRateFromFormat(input.format, detector.requiredSampleRate) ?? detector.requiredSampleRate; - if (parsedInputRate !== inputRate) { - inputRate = parsedInputRate; - resampler = - inputRate === detector.requiredSampleRate - ? null - : new Pcm16MonoResampler({ - inputRate, - outputRate: detector.requiredSampleRate, - }); + updateDetectorResampler(parsedInputRate); + const currentSttSession = sttSession; + if (currentSttSession) { + updateSttResampler(currentSttSession, parsedInputRate); } - const normalized = resampler === null ? pcm16 : resampler.processChunk(pcm16); - if (normalized.length === 0) { + const detectorPcm16 = pcmForDetector(pcm16); + const sttPcm16 = currentSttSession + ? pcmForStt(currentSttSession, pcm16, detectorPcm16) + : null; + if (detectorPcm16.length === 0 && (!sttPcm16 || sttPcm16.length === 0)) { return; } - prefixBuffer.append(normalized); - - if (state.status === "listening") { - state = { - status: "listening", - rollingPrefixBytes: prefixBuffer.byteLength, - }; - } else if (state.status === "capturing") { - utteranceChunks.push(normalized); - state = { - ...state, - rollingPrefixBytes: prefixBuffer.byteLength, - utteranceBytes: state.utteranceBytes + normalized.length, - }; + if (detectorPcm16.length > 0) { + detector.appendPcm16(detectorPcm16); } - detector.appendPcm16(normalized); + if (sttPcm16 && sttPcm16.length > 0) { + try { + currentSttSession?.appendPcm16(sttPcm16); + } catch (error) { + handleSttError(error); + } + } }); }, };