puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-29 04:32:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Wait for in-flight dictation commits before finishing (#3968)

Browse Source 
* 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 5168a6554d.

* Revert "fix(voice): serialize capture lifecycle"

This reverts commit 00fd3e3556.

* Revert "fix(dictation): preserve audio queued at stop"

This reverts commit ec087dc9cd.
This commit is contained in:
Mohamed Boudra
2026-08-28 15:29:42 +02:00
committed by GitHub
parent 23debf3af1
commit 42b6d99d60
2 changed files with 102 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/dictation/dictation-stream-manager.test.ts
+78 -1
View File
@@ -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",
});
packages/server/src/server/dictation/dictation-stream-manager.ts
+24 -7
View File
@@ -87,6 +87,7 @@ interface DictationStreamState {
committedSegmentIds: string[];
transcriptsBySegmentId: Map<string, string>;
finalTranscriptSegmentIds: Set<string>;
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];