Switch voice turn controller to streaming transcription

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
    */

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<void>;
+  stopVoiceTurnController(): Promise<void>;
+  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<unknown>;
   handleCheckoutMergeFromBaseRequest(params: unknown): Promise<unknown>;
   handleCheckoutCommitRequest(params: unknown): Promise<unknown>;
@@ -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<void> {}
+
+  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<void> {}
+
+  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<unknown>;
+        };
+      }
+    ).sttManager,
+    "transcribe",
+  );
+
+  return {
+    session,
+    internals,
+    messages,
+    detector,
+    sttSession,
+    sendAgentMessage,
+    transcribe,
+  };
+}
+
+async function settleVoiceSession(): Promise<void> {
+  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[] = [];
 

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<typeof setTimeout> | 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: {
+      stt,
-        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);
-        },
-      },
       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",
@@ -8021,7 +8013,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;
     }

packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.test.ts

@@ -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);
-  });
-});

packages/server/src/server/voice/fixed-duration-pcm-ring-buffer.ts

@@ -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;
-  }
-}

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<void> {
+    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<void> {
   await Promise.resolve();
   await Promise.resolve();
@@ -40,21 +100,34 @@ async function settleSerialQueue(): Promise<void> {
 
 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,
+    stt,
-    utteranceSink: {
-      submitUtterance,
-    },
     callbacks: {
       onSpeechStarted,
       onSpeechStopped,
+      onPartialTranscript,
+      onFinalTranscript,
       onError,
     },
   });
@@ -62,25 +135,17 @@ function createControllerHarness() {
   return {
     controller,
     detector,
+    sttSessions,
     onSpeechStarted,
     onSpeechStopped,
-    submitUtterance,
+    onPartialTranscript,
+    onFinalTranscript,
     onError,
   };
 }
 
-function createDeferredPromise<T>() {
-  let resolve!: (value: T | PromiseLike<T>) => void;
-  let reject!: (reason?: unknown) => void;
-  const promise = new Promise<T>((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.onFinalTranscript).not.toHaveBeenCalled();
-    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.onError).not.toHaveBeenCalled();
   });
 
-  it("finalizes one utterance when speech stops", async () => {
+  it("does not barge in on silence-only chunks", 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 () => {
     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({
+    harness.sttSessions[0]?.emitTranscript({
-      audioBase64: Buffer.from([5, 6, 7, 8]).toString("base64"),
+      segmentId: "segment-1",
-      format: "audio/pcm;rate=16000;bits=16",
+      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({
+    expect(harness.onPartialTranscript).toHaveBeenCalledTimes(1);
-      audioBase64: Buffer.from([9, 10, 11, 12]).toString("base64"),
+    expect(harness.onPartialTranscript).toHaveBeenCalledWith({
-      format: "audio/pcm;rate=16000;bits=16",
+      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({
+    for (const transcript of ["uh", "uh,", "Um", "uh um", "hmm"]) {
-      audioBase64: Buffer.from([13, 14, 15, 16]).toString("base64"),
+      harness.sttSessions[0]?.emitTranscript({
-      format: "audio/pcm;rate=16000;bits=16",
+        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.sttSessions[0]?.commitCount).toBe(1);
-    expect(harness.onSpeechStopped).toHaveBeenCalledTimes(2);
+  });
-    expect(harness.submitUtterance.mock.calls[0]?.[0]).toEqual(
+
+  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]),
+        segmentId: "segment-1",
-      }),
+        transcript: "hello there",
-    );
+        language: "en",
-    expect(harness.submitUtterance.mock.calls[1]?.[0]).toEqual(
+        avgLogprob: -0.2,
-      expect.objectContaining({
+        durationMs: expect.any(Number),
-        pcm16: Buffer.from([5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]),
       }),
     );
   });
 
-  it("continues detecting speech while a previous utterance submission is still pending", async () => {
+  it("assembles multiple committed final transcripts in commit order", async () => {
-    const deferred = createDeferredPromise<void>();
+    const harness = createControllerHarness();
-    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();
 
-    const controller = createVoiceTurnController({
+    await harness.controller.start();
-      logger: pino({ level: "silent" }),
+    harness.detector.emit("speech_started");
-      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 settleSerialQueue();
-    await controller.appendClientChunk({
+    harness.detector.emit("speech_stopped");
-      audioBase64: Buffer.from([5, 6, 7, 8]).toString("base64"),
-      format: "audio/pcm;rate=16000;bits=16",
-    });
-    detector.emit("speech_stopped");
     await settleSerialQueue();
 
-    await controller.appendClientChunk({
+    harness.sttSessions[0]?.emitCommitted({ segmentId: "segment-1", previousSegmentId: null });
-      audioBase64: Buffer.from([9, 10, 11, 12]).toString("base64"),
+    harness.sttSessions[0]?.emitCommitted({
-      format: "audio/pcm;rate=16000;bits=16",
+      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(harness.onFinalTranscript).toHaveBeenCalledTimes(1);
-    expect(onSpeechStopped).toHaveBeenCalledTimes(1);
+    expect(harness.onFinalTranscript).toHaveBeenCalledWith(
-    expect(submitUtterance).toHaveBeenCalledTimes(1);
+      expect.objectContaining({
-    expect(onError).not.toHaveBeenCalled();
+        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);
   });
 });

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 VOICE_FINAL_TRANSCRIPT_TIMEOUT_MS = 10_000;
-const PCM_BITS_PER_SAMPLE = 16;
+
-const DEFAULT_PREFIX_DURATION_MS = 1000;
+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<void>;
   onSpeechStopped(): Promise<void>;
+  onPartialTranscript(input: { segmentId: string; transcript: string }): Promise<void>;
+  onFinalTranscript(input: VoiceFinalTranscript): Promise<void>;
   onError(error: Error): void;
 }
 
-export interface DetectedVoiceUtterance {
-  pcm16: Buffer;
-  sampleRate: number;
-  format: string;
-  startedAt: number;
-  endedAt: number;
-}
-
-export interface VoiceUtteranceSink {
-  submitUtterance(utterance: DetectedVoiceUtterance): Promise<void>;
-}
-
 export interface VoiceTurnController {
   start(): Promise<void>;
   stop(): Promise<void>;
   appendClientChunk(input: { audioBase64: string; format: string }): Promise<void>;
 }
 
+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<string, string>;
+  finalTranscriptSegmentIds: Set<string>;
+  transcriptMetaBySegmentId: Map<string, TranscriptSegmentMeta>;
+  timeout: ReturnType<typeof setTimeout>;
+  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();
+  let activeTranscriptSegmentId: string | null = null;
-
+  let partialTranscriptFired = false;
-  function buildVoicePcmFormat(sampleRate: number): string {
+  let reconnectAttemptedForTurn = false;
-    return `audio/pcm;rate=${sampleRate};bits=16`;
+  const sealedTranscriptSegmentIds = new Set<string>();
-  }
+  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<void> {
+    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<void>): Promise<void> {
     queued = queued.then(task).catch((error) => {
       fail(error);
@@ -85,15 +344,57 @@ export function createVoiceTurnController(params: {
     return queued;
   }
 
-  function enqueueUtteranceSubmission(utterance: DetectedVoiceUtterance): void {
+  function updateDetectorResampler(parsedInputRate: number): void {
-    submissionQueue = submissionQueue
+    if (parsedInputRate === inputRate) {
-      .then(async () => {
+      return;
-        await params.utteranceSink.submitUtterance(utterance);
+    }
-        return;
+
-      })
+    inputRate = parsedInputRate;
-      .catch((error) => {
+    resampler =
-        fail(error);
+      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<void> {
@@ -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" };
-    state = { status: "listening", rollingPrefixBytes: prefixBuffer.byteLength };
+
+    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<void> {
+      sttSession = createSttSession();
+      await sttSession.connect();
       await detector.connect();
-      state = { status: "listening", rollingPrefixBytes: prefixBuffer.byteLength };
+      state = { status: "listening" };
     },
 
     async stop(): Promise<void> {
       await runSerial(async () => {
+        clearFinalizingTurnTimeout();
         detector.close();
-        prefixBuffer.clear();
+        sttSession?.close();
-        utteranceChunks = [];
         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) {
+        updateDetectorResampler(parsedInputRate);
-          inputRate = parsedInputRate;
+        const currentSttSession = sttSession;
-          resampler =
+        if (currentSttSession) {
-            inputRate === detector.requiredSampleRate
+          updateSttResampler(currentSttSession, parsedInputRate);
-              ? null
-              : new Pcm16MonoResampler({
-                  inputRate,
-                  outputRate: detector.requiredSampleRate,
-                });
         }
 
-        const normalized = resampler === null ? pcm16 : resampler.processChunk(pcm16);
+        const detectorPcm16 = pcmForDetector(pcm16);
-        if (normalized.length === 0) {
+        const sttPcm16 = currentSttSession
+          ? pcmForStt(currentSttSession, pcm16, detectorPcm16)
+          : null;
+        if (detectorPcm16.length === 0 && (!sttPcm16 || sttPcm16.length === 0)) {
           return;
         }
 
-        prefixBuffer.append(normalized);
+        if (detectorPcm16.length > 0) {
-
+          detector.appendPcm16(detectorPcm16);
-        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,
-          };
         }
 
-        detector.appendPcm16(normalized);
+        if (sttPcm16 && sttPcm16.length > 0) {
+          try {
+            currentSttSession?.appendPcm16(sttPcm16);
+          } catch (error) {
+            handleSttError(error);
+          }
+        }
       });
     },
   };