From f4d7cef21a4d793a6a2042795231cc60ac1de73a Mon Sep 17 00:00:00 2001 From: Zhou Yurui Date: Sat, 8 Aug 2026 16:14:15 +0800 Subject: [PATCH] Fix dictation STT stream finalization --- .../dictation-stream-manager.test.ts | 36 +++++++++ .../dictation/dictation-stream-manager.ts | 7 ++ .../sherpa-parakeet-realtime-session.test.ts | 67 +++++++++++++++++ .../sherpa-parakeet-realtime-session.ts | 37 +++++----- .../speech/providers/openai/stt.test.ts | 73 +++++++++++++++++-- .../src/server/speech/providers/openai/stt.ts | 12 +-- 6 files changed, 204 insertions(+), 28 deletions(-) create mode 100644 packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.test.ts diff --git a/packages/server/src/server/dictation/dictation-stream-manager.test.ts b/packages/server/src/server/dictation/dictation-stream-manager.test.ts index bcba41b52a..f41ecc1ca7 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.test.ts +++ b/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"); + }); }); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.ts b/packages/server/src/server/dictation/dictation-stream-manager.ts index db3cab8aee..e1058fafb1 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.ts +++ b/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 | 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({ diff --git a/packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.test.ts b/packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.test.ts new file mode 100644 index 0000000000..801cdab8ee --- /dev/null +++ b/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"]); + }); +}); diff --git a/packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.ts b/packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.ts index a299b32beb..8c71643b5a 100644 --- a/packages/server/src/server/speech/providers/local/sherpa/sherpa-parakeet-realtime-session.ts +++ b/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 { - if (this.pcm16.length === 0) { + private async decodePcm16(pcm16: Buffer): Promise { + 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); diff --git a/packages/server/src/server/speech/providers/openai/stt.test.ts b/packages/server/src/server/speech/providers/openai/stt.test.ts index 72707be252..80ce982d2c 100644 --- a/packages/server/src/server/speech/providers/openai/stt.test.ts +++ b/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() { + 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 }; +} + +async function drainStream(stream: NodeJS.ReadableStream): Promise { + await new Promise((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((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); + }); }); diff --git a/packages/server/src/server/speech/providers/openai/stt.ts b/packages/server/src/server/speech/providers/openai/stt.ts index 6a143bac53..6d1ceebf82 100644 --- a/packages/server/src/server/speech/providers/openai/stt.ts +++ b/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); } })(); },