Fix dictation STT stream finalization

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);
@@ -410,4 +419,31 @@ 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");
+  });
 });

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

@@ -90,6 +90,7 @@ interface DictationStreamState {
   awaitingFinalCommit: boolean;
   finishRequested: boolean;
   finishSealed: boolean;
+  finalizing: boolean;
   finalSeq: number | null;
   finalTimeout: ReturnType<typeof setTimeout> | null;
 }
@@ -312,6 +313,7 @@ export class DictationStreamManager {
       awaitingFinalCommit: false,
       finishRequested: false,
       finishSealed: false,
+      finalizing: false,
       finalSeq: null,
       finalTimeout: null,
     });
@@ -714,6 +716,9 @@ export class DictationStreamManager {
     if (state.awaitingFinalCommit) {
       return;
     }
+    if (state.finalizing) {
+      return;
+    }
 
     const committedSet = new Set(state.committedSegmentIds);
     const orderedSegmentIds: string[] = [...state.committedSegmentIds];
@@ -724,6 +729,7 @@ export class DictationStreamManager {
     }
 
     if (orderedSegmentIds.length === 0) {
+      state.finalizing = true;
       void (async () => {
         const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
         this.emit({
@@ -763,6 +769,7 @@ export class DictationStreamManager {
       .join(" ")
       .trim();
 
+    state.finalizing = true;
     void (async () => {
       const debugRecordingPath = await this.maybePersistDictationStreamAudio(dictationId);
       this.emit({

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 new Promise((resolve) => setTimeout(resolve, 0));
+
+    session.commit();
+
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    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.test.ts

@@ -1,4 +1,5 @@
 import pino from "pino";
+import type { Readable } from "node:stream";
 import { afterEach, describe, expect, test, vi } from "vitest";
 
 const { openAiConstructorOptionsMock, transcriptionsCreateMock } = vi.hoisted(() => ({
@@ -21,6 +22,24 @@ vi.mock("openai", () => ({
 
 import { OpenAISTT } from "./stt.js";
 
+function createDeferred<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 };
+}
+
+async function drainStream(stream: NodeJS.ReadableStream): Promise<void> {
+  await new Promise<void>((resolve, reject) => {
+    stream.once("error", reject);
+    stream.once("end", resolve);
+    stream.resume();
+  });
+}
+
 describe("OpenAISTT", () => {
   afterEach(() => {
     openAiConstructorOptionsMock.mockReset();
@@ -43,11 +62,7 @@ describe("OpenAISTT", () => {
   test("passes transcription prompt to OpenAI REST STT", async () => {
     transcriptionsCreateMock.mockImplementation(
       async (request: { file: NodeJS.ReadableStream }) => {
-        await new Promise<void>((resolve, reject) => {
-          request.file.once("error", reject);
-          request.file.once("end", resolve);
-          request.file.resume();
-        });
+        await drainStream(request.file);
         return { text: "hello" };
       },
     );
@@ -87,4 +102,52 @@ describe("OpenAISTT", () => {
       }),
     );
   });
+
+  test("keeps audio appended while an OpenAI transcription is in flight", async () => {
+    const firstTranscription = createDeferred<{ text: string }>();
+    const transcripts: string[] = [];
+
+    transcriptionsCreateMock.mockImplementation(async (request: { file: Readable }) => {
+      await drainStream(request.file);
+      if (transcriptionsCreateMock.mock.calls.length === 1) {
+        await firstTranscription.promise;
+        return { text: "first" };
+      }
+      return { text: "second" };
+    });
+
+    const provider = new OpenAISTT({ apiKey: "sk-test" }, pino({ level: "silent" }));
+    const session = provider.createSession({
+      logger: pino({ level: "silent" }),
+      language: "en",
+    });
+
+    session.on("transcript", (event) => {
+      if (event.isFinal) {
+        transcripts.push(event.transcript);
+      }
+    });
+
+    await session.connect();
+    session.appendPcm16(Buffer.from([1, 0, 1, 0]));
+    session.commit();
+
+    await vi.waitFor(() => {
+      expect(transcriptionsCreateMock).toHaveBeenCalledTimes(1);
+    });
+
+    session.appendPcm16(Buffer.from([2, 0, 2, 0]));
+    firstTranscription.resolve({ text: "first" });
+
+    await vi.waitFor(() => {
+      expect(transcripts).toEqual(["first"]);
+    });
+
+    session.commit();
+
+    await vi.waitFor(() => {
+      expect(transcripts).toEqual(["first", "second"]);
+    });
+    expect(transcriptionsCreateMock).toHaveBeenCalledTimes(2);
+  });
 });

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);
           }
         })();
       },