From 42b6d99d60d57c2b423c00069fd816cf2479ede5 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 28 Aug 2026 15:29:42 +0200 Subject: [PATCH] Wait for in-flight dictation commits before finishing (#3968) * fix(dictation): preserve audio queued at stop Stopping capture disabled native recording before the final microphone buffer had crossed the bridge, so accepting immediately could omit the end of a dictation. Wait for the native producer to publish its final buffer before completing stop, and serialize stop with restart and destruction. * fix(voice): serialize capture lifecycle A capture start already awaiting permission could resume after teardown, and repeated destroy calls did not share completion. Order capture lifecycle operations behind one promise and build the native module declarations before clean-CI typechecking. * fix(dictation): drain queued web capture before finish * fix(dictation): await in-flight segment commits * Revert "fix(dictation): drain queued web capture before finish" This reverts commit 5168a6554d5cc793cf028c22979f6aae2781b571. * Revert "fix(voice): serialize capture lifecycle" This reverts commit 00fd3e3556a4c6f3bb96a51a850d4b5048d5fe56. * Revert "fix(dictation): preserve audio queued at stop" This reverts commit ec087dc9cdccc36e89b4f252398d5f9098ffdda1. --- .../dictation-stream-manager.test.ts | 79 ++++++++++++++++++- .../dictation/dictation-stream-manager.ts | 31 ++++++-- 2 files changed, 102 insertions(+), 8 deletions(-) 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..dde4fad734 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.test.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.test.ts @@ -298,6 +298,82 @@ describe("DictationStreamManager (provider-agnostic provider)", () => { } }); + it("waits for an in-flight auto-commit before finalizing", async () => { + const session = new FakeRealtimeSession(); + const emitted: Array<{ type: string; payload: unknown }> = []; + const manager = new DictationStreamManager({ + logger: pino({ level: "silent" }), + emit: (message) => emitted.push(message), + sessionId: "s1", + stt: new FakeSttProvider(session), + autoCommitSeconds: 1, + }); + + await manager.handleStart("d-delayed-auto-commit", "audio/pcm;rate=24000;bits=16"); + await manager.handleChunk({ + dictationId: "d-delayed-auto-commit", + seq: 0, + audioBase64: buildPcmBase64(2000, 24000), + format: "audio/pcm;rate=24000;bits=16", + }); + expect(session.commitCalls).toBe(1); + + await manager.handleFinish("d-delayed-auto-commit", 0); + await tick(); + + expect(emitted.find((message) => message.type === "dictation_stream_final")).toBeUndefined(); + expect(session.closed).toBe(false); + + session.emitCommitted("seg-tail"); + session.emitTranscript("seg-tail", "the final words", true); + await tick(); + + const final = emitted.find((message) => message.type === "dictation_stream_final"); + expect((final?.payload as { text?: string } | undefined)?.text).toBe("the final words"); + expect(session.closed).toBe(true); + }); + + it("commits tail audio appended while an auto-commit is in flight", async () => { + const session = new FakeRealtimeSession(); + const emitted: Array<{ type: string; payload: unknown }> = []; + const manager = new DictationStreamManager({ + logger: pino({ level: "silent" }), + emit: (message) => emitted.push(message), + sessionId: "s1", + stt: new FakeSttProvider(session), + autoCommitSeconds: 1, + }); + + await manager.handleStart("d-tail-during-commit", "audio/pcm;rate=24000;bits=16"); + await manager.handleChunk({ + dictationId: "d-tail-during-commit", + seq: 0, + audioBase64: buildPcmBase64(2000, 24000), + format: "audio/pcm;rate=24000;bits=16", + }); + await manager.handleChunk({ + dictationId: "d-tail-during-commit", + seq: 1, + audioBase64: buildPcmBase64(2000, 2400), + format: "audio/pcm;rate=24000;bits=16", + }); + + session.emitCommitted("seg-first"); + session.emitTranscript("seg-first", "the beginning", true); + await manager.handleFinish("d-tail-during-commit", 1); + + expect(session.commitCalls).toBe(2); + + session.emitCommitted("seg-tail"); + session.emitTranscript("seg-tail", "the final words", true); + await tick(); + + const final = emitted.find((message) => message.type === "dictation_stream_final"); + expect((final?.payload as { text?: string } | undefined)?.text).toBe( + "the beginning the final words", + ); + }); + it("adapts finish timeout based on pending committed segments", async () => { const session = new FakeRealtimeSession(); const emitted: Array<{ type: string; payload: unknown }> = []; @@ -374,13 +450,14 @@ describe("DictationStreamManager (provider-agnostic provider)", () => { sessionId: "s1", stt: new FakeSttProvider(session), finalTimeoutMs: 5000, + autoCommitSeconds: 1, }); await manager.handleStart("d-clear-tail", "audio/pcm;rate=24000;bits=16"); await manager.handleChunk({ dictationId: "d-clear-tail", seq: 0, - audioBase64: buildPcmBase64(2000, 2400), + audioBase64: buildPcmBase64(2000, 24000), format: "audio/pcm;rate=24000;bits=16", }); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.ts b/packages/server/src/server/dictation/dictation-stream-manager.ts index db3cab8aee..e62b975ef0 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.ts @@ -87,6 +87,7 @@ interface DictationStreamState { committedSegmentIds: string[]; transcriptsBySegmentId: Map; finalTranscriptSegmentIds: Set; + inFlightCommitCount: number; awaitingFinalCommit: boolean; finishRequested: boolean; finishSealed: boolean; @@ -193,9 +194,10 @@ export class DictationStreamManager { if (!state) { return; } + if (state.inFlightCommitCount > 0) { + state.inFlightCommitCount -= 1; + } state.committedSegmentIds.push(segmentId); - state.bytesSinceCommit = 0; - state.peakSinceCommit = 0; if (state.finishRequested && state.awaitingFinalCommit) { state.awaitingFinalCommit = false; @@ -235,6 +237,9 @@ export class DictationStreamManager { const message = err instanceof Error ? err.message : String(err); const state = this.streams.get(dictationId); if (state && state.finishRequested && isBufferTooSmallError(message)) { + if (state.inFlightCommitCount > 0) { + state.inFlightCommitCount -= 1; + } if (state.awaitingFinalCommit) { state.awaitingFinalCommit = false; } @@ -309,6 +314,7 @@ export class DictationStreamManager { committedSegmentIds: [], transcriptsBySegmentId: new Map(), finalTranscriptSegmentIds: new Set(), + inFlightCommitCount: 0, awaitingFinalCommit: false, finishRequested: false, finishSealed: false, @@ -583,9 +589,7 @@ export class DictationStreamManager { return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1; }, 0); const pendingSegments = - pendingCommittedSegments + - pendingUncommittedTranscriptSegments + - (state.awaitingFinalCommit ? 1 : 0); + pendingCommittedSegments + pendingUncommittedTranscriptSegments + state.inFlightCommitCount; const pendingAudioSeconds = Math.ceil(Math.max(0, state.bytesSinceCommit) / bytesPerSecond); const missingSeqCount = state.finalSeq === null ? 0 : Math.max(0, state.finalSeq - state.ackSeq); @@ -622,9 +626,19 @@ export class DictationStreamManager { return; } + this.requestDictationCommit(state); + } + + private requestDictationCommit(state: DictationStreamState): void { state.bytesSinceCommit = 0; state.peakSinceCommit = 0; - state.stt.commit(); + state.inFlightCommitCount += 1; + try { + state.stt.commit(); + } catch (error) { + state.inFlightCommitCount -= 1; + throw error; + } } private maybeSealDictationStreamFinish(dictationId: string): void { @@ -669,7 +683,7 @@ export class DictationStreamManager { } else { state.awaitingFinalCommit = true; try { - state.stt.commit(); + this.requestDictationCommit(state); } catch (error) { const message = error instanceof Error ? error.message : String(error); void this.failAndCleanupDictationStream(dictationId, message, true); @@ -714,6 +728,9 @@ export class DictationStreamManager { if (state.awaitingFinalCommit) { return; } + if (state.inFlightCommitCount > 0) { + return; + } const committedSet = new Set(state.committedSegmentIds); const orderedSegmentIds: string[] = [...state.committedSegmentIds];