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];