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 975d6677b8..cabb29b414 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); @@ -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 () => { 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..1d1f1bc5b1 --- /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 expect.poll(() => transcripts.length).toBe(1); + + session.commit(); + + await expect.poll(() => transcripts.length).toBe(2); + + 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-segment-retention.test.ts b/packages/server/src/server/speech/providers/openai/stt-segment-retention.test.ts new file mode 100644 index 0000000000..e18b5cd84b --- /dev/null +++ b/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((resolve) => server.close(() => resolve())); + } + }, +); 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); } })(); },