fix(dictation): retain audio recorded during transcription

Detach committed audio before asynchronous transcription so later chunks retain their own buffer and segment identity. Carry the contributor fix onto current main, retain its existing finalization protection, and verify uploaded PCM through the real SDK HTTP interface.

Co-authored-by: Zhou Yurui <github@yuruiz.com>

Co-authored-by: Joey Hotz <joeyhotz1@gmail.com>

packages/server/src/server/dictation/dictation-stream-manager.test.ts

@@ -63,6 +63,15 @@ class FakeSttProvider implements SpeechToTextProvider {
   }
 }
 
+class SyncFinalSession extends FakeRealtimeSession {
+  override commit(): void {
+    this.commitCalls += 1;
+    const segmentId = `seg-${this.commitCalls}`;
+    this.emitCommitted(segmentId);
+    this.emitTranscript(segmentId, `transcript ${this.commitCalls}`, true);
+  }
+}
+
 const buildPcmBase64 = (sampleValue: number, sampleCount: number): string => {
   const samples = new Int16Array(sampleCount);
   samples.fill(sampleValue);
@@ -533,6 +542,33 @@ describe("DictationStreamManager (provider-agnostic provider)", () => {
       vi.useRealTimers();
     }
   });
+
+  it("emits one final transcript when synchronous STT events re-enter finalization", async () => {
+    const session = new SyncFinalSession();
+    const emitted: Array<{ type: string; payload: unknown }> = [];
+    const manager = new DictationStreamManager({
+      logger: pino({ level: "silent" }),
+      emit: (msg) => emitted.push(msg),
+      sessionId: "s1",
+      stt: new FakeSttProvider(session),
+      autoCommitSeconds: 0,
+    });
+
+    await manager.handleStart("d-sync-final", "audio/pcm;rate=24000;bits=16");
+    await manager.handleChunk({
+      dictationId: "d-sync-final",
+      seq: 0,
+      audioBase64: buildPcmBase64(2000, 2400),
+      format: "audio/pcm;rate=24000;bits=16",
+    });
+
+    await manager.handleFinish("d-sync-final", 0);
+    await tick();
+
+    const finals = emitted.filter((msg) => msg.type === "dictation_stream_final");
+    expect(finals).toHaveLength(1);
+    expect((finals[0]?.payload as { text?: string } | undefined)?.text).toBe("transcript 1");
+  });
 });
 
 it("cancellation during STT bootstrap closes the producer and never acknowledges a late connection", async () => {

packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it } from "vitest";
+
+import { SherpaParakeetRealtimeTranscriptionSession } from "./sherpa-parakeet-realtime-session.js";
+import type { SherpaOfflineRecognizerEngine } from "./sherpa-offline-recognizer.js";
+
+function pcm16(value: number): Buffer {
+  const samples = new Int16Array([value]);
+  return Buffer.from(samples.buffer);
+}
+
+function createEngine(): SherpaOfflineRecognizerEngine {
+  return {
+    sampleRate: 16000,
+    createStream() {
+      return {
+        samples: new Float32Array(0),
+        acceptWaveform(input: { samples: Float32Array }) {
+          this.samples = input.samples;
+        },
+        free() {
+          // no-op
+        },
+      };
+    },
+    acceptWaveform(stream: { samples: Float32Array }, _sampleRate: number, samples: Float32Array) {
+      stream.samples = samples;
+    },
+    recognizer: {
+      decode() {
+        // no-op
+      },
+      getResult(stream: { samples: Float32Array }) {
+        const firstSample = Math.abs(stream.samples[0] ?? 0);
+        return { text: firstSample > 0.4 ? "second" : "first" };
+      },
+    },
+  } as unknown as SherpaOfflineRecognizerEngine;
+}
+
+describe("SherpaParakeetRealtimeTranscriptionSession", () => {
+  it("keeps audio appended after commit for the next segment", async () => {
+    const session = new SherpaParakeetRealtimeTranscriptionSession({
+      engine: createEngine(),
+      minDecodeIntervalMs: Number.MAX_SAFE_INTEGER,
+    });
+    const transcripts: string[] = [];
+
+    session.on("transcript", (event) => {
+      if (event.isFinal) {
+        transcripts.push(event.transcript);
+      }
+    });
+
+    await session.connect();
+    session.appendPcm16(pcm16(0));
+    session.commit();
+    session.appendPcm16(pcm16(24000));
+
+    await expect.poll(() => transcripts.length).toBe(1);
+
+    session.commit();
+
+    await expect.poll(() => transcripts.length).toBe(2);
+
+    expect(transcripts).toEqual(["first", "second"]);
+  });
+});

packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.ts

@@ -58,20 +58,19 @@ export class SherpaParakeetRealtimeTranscriptionSession
       return;
     }
 
+    const segmentId = this.currentSegmentId;
+    const previousSegmentId = this.previousSegmentId;
+    const committedPcm16 = this.pcm16;
+    this.previousSegmentId = segmentId;
+    this.currentSegmentId = uuidv4();
+    this.lastPartialText = "";
+    this.pcm16 = Buffer.alloc(0);
+    this.emit("committed", { segmentId, previousSegmentId });
+
     void (async () => {
       try {
-        await this.maybeDecode(true);
-        const finalText = this.lastPartialText;
-        const segmentId = this.currentSegmentId!;
-        const previousSegmentId = this.previousSegmentId;
-
-        this.emit("committed", { segmentId, previousSegmentId });
+        const finalText = await this.decodePcm16(committedPcm16);
         this.emit("transcript", { segmentId, transcript: finalText, isFinal: true });
-
-        this.previousSegmentId = segmentId;
-        this.currentSegmentId = uuidv4();
-        this.lastPartialText = "";
-        this.pcm16 = Buffer.alloc(0);
       } catch (err) {
         this.emit("error", err instanceof Error ? err : new Error(String(err)));
       }
@@ -110,12 +109,16 @@ export class SherpaParakeetRealtimeTranscriptionSession
 
     this.decoding = true;
     try {
-      const text = await this.decodeNow();
+      const segmentId = this.currentSegmentId;
+      const text = await this.decodePcm16(this.pcm16);
       this.lastDecodeAt = Date.now();
+      if (!this.connected || this.currentSegmentId !== segmentId) {
+        return;
+      }
       if (text !== this.lastPartialText) {
         this.lastPartialText = text;
         this.emit("transcript", {
-          segmentId: this.currentSegmentId,
+          segmentId,
           transcript: text,
           isFinal: false,
         });
@@ -129,12 +132,12 @@ export class SherpaParakeetRealtimeTranscriptionSession
     }
   }
 
-  private async decodeNow(): Promise<string> {
-    if (this.pcm16.length === 0) {
+  private async decodePcm16(pcm16: Buffer): Promise<string> {
+    if (pcm16.length === 0) {
       return "";
     }
 
-    const peak = pcm16lePeakAbs(this.pcm16);
+    const peak = pcm16lePeakAbs(pcm16);
     const peakFloat = peak / 32768.0;
     const targetPeak = 0.6;
     const maxGain = 50;
@@ -143,7 +146,7 @@ export class SherpaParakeetRealtimeTranscriptionSession
 
     const stream = this.engine.createStream();
     try {
-      const floatSamples = pcm16leToFloat32(this.pcm16, gain);
+      const floatSamples = pcm16leToFloat32(pcm16, gain);
       this.engine.acceptWaveform(stream, this.engine.sampleRate, floatSamples);
       this.engine.recognizer.decode(stream);
       const result = this.engine.recognizer.getResult(stream);

packages/server/src/server/speech/providers/openai/stt-segment-retention.test.ts

@@ -0,0 +1,66 @@
+import { createServer, type ServerResponse } from "node:http";
+import { once } from "node:events";
+import { expect, test } from "vitest";
+import pino from "pino";
+import { OpenAISTT } from "./stt.js";
+
+test.each(["before", "after"])(
+  "retains later PCM when stop is %s the first response",
+  async (stop) => {
+    const uploads: Buffer[] = [];
+    const responses: ServerResponse[] = [];
+    // A transcription service fake on the SDK's HTTP interface. The SDK, WAV encoding,
+    // temporary files, session buffer, and event handling all run unchanged.
+    const server = createServer(async (request, response) => {
+      const chunks: Buffer[] = [];
+      for await (const chunk of request) chunks.push(Buffer.from(chunk));
+      const body = Buffer.concat(chunks);
+      const wav = body.subarray(body.indexOf(Buffer.from("RIFF")));
+      uploads.push(wav.subarray(44, 44 + wav.readUInt32LE(40)));
+      responses.push(response);
+    });
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Missing server address");
+    const logger = pino({ level: "silent" });
+    const session = new OpenAISTT(
+      { apiKey: "test", baseUrl: `http://127.0.0.1:${address.port}/v1` },
+      logger,
+    ).createSession({ logger });
+    const transcripts: string[] = [];
+    session.on("transcript", (event) => {
+      if (event.isFinal) transcripts.push(event.transcript);
+    });
+    const finish = (index: number, text: string) => {
+      responses[index]!.writeHead(200, { "Content-Type": "application/json" });
+      responses[index]!.end(JSON.stringify({ text }));
+    };
+    try {
+      await session.connect();
+      session.appendPcm16(Buffer.from([1, 0, 2, 0]));
+      session.commit();
+      await expect.poll(() => uploads.length).toBe(1);
+      session.appendPcm16(Buffer.from([3, 0, 4, 0]));
+      if (stop === "after") {
+        finish(0, "first");
+        await expect.poll(() => transcripts.length).toBe(1);
+      }
+      session.commit();
+      await expect.poll(() => uploads.length, { timeout: 1000 }).toBe(2);
+      expect(uploads.map((pcm) => Array.from(pcm))).toEqual([
+        [1, 0, 2, 0],
+        [3, 0, 4, 0],
+      ]);
+      if (stop === "before") finish(0, "first");
+      finish(1, "second");
+      await expect.poll(() => transcripts.length).toBe(2);
+    } finally {
+      session.close();
+      for (const response of responses)
+        if (!response.writableEnded) response.end(JSON.stringify({ text: "" }));
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  },
+);

packages/server/src/server/speech/providers/openai/stt.ts

@@ -124,11 +124,15 @@ export class OpenAISTT implements SpeechToTextProvider {
 
         const committedId = segmentId;
         const prev = previousSegmentId;
+        const committedPcm16 = pcm16;
+        previousSegmentId = committedId;
+        segmentId = v4();
+        pcm16 = Buffer.alloc(0);
         emitter.emit("committed", { segmentId: committedId, previousSegmentId: prev });
 
         void (async () => {
           try {
-            if (pcm16.length === 0) {
+            if (committedPcm16.length === 0) {
               emitter.emit("transcript", {
                 segmentId: committedId,
                 transcript: "",
@@ -139,7 +143,7 @@ export class OpenAISTT implements SpeechToTextProvider {
               return;
             }
 
-            const wav = convertPCMToWavBuffer(pcm16);
+            const wav = convertPCMToWavBuffer(committedPcm16);
             const result = await transcribeAudio(
               wav,
               "audio/wav",
@@ -159,10 +163,6 @@ export class OpenAISTT implements SpeechToTextProvider {
             });
           } catch (err) {
             emitter.emit("error", err);
-          } finally {
-            previousSegmentId = committedId;
-            segmentId = v4();
-            pcm16 = Buffer.alloc(0);
           }
         })();
       },