jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dictation): add adaptive finish timeout with server-provided budget

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-13 10:15:44 +07:00
parent bf432b3d3a
commit b64fa4ca93
5 changed files with 341 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.test.ts
+88
View File
@@ -442,6 +442,94 @@ describe("DaemonClient", () => {
});
});
test("uses server-provided dictation finish timeout budget", async () => {
vi.useFakeTimers();
const logger = createMockLogger();
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
const finishPromise = client.finishDictationStream("dict-1", 0);
const finishError = finishPromise.then(
() => null,
(error) => error
);
expect(mock.sent).toHaveLength(1);
mock.triggerMessage(
wrapSessionMessage({
type: "dictation_stream_finish_accepted",
payload: {
dictationId: "dict-1",
timeoutMs: 100,
},
})
);
await vi.advanceTimersByTimeAsync(5_101);
const error = await finishError;
expect(error).toBeInstanceOf(Error);
expect((error as Error).message).toContain(
"Timeout waiting for dictation finalization (5100ms)"
);
vi.useRealTimers();
});
test("resolves dictation finish when final arrives after finish accepted", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
const finishPromise = client.finishDictationStream("dict-2", 1);
expect(mock.sent).toHaveLength(1);
mock.triggerMessage(
wrapSessionMessage({
type: "dictation_stream_finish_accepted",
payload: {
dictationId: "dict-2",
timeoutMs: 1000,
},
})
);
mock.triggerMessage(
wrapSessionMessage({
type: "dictation_stream_final",
payload: {
dictationId: "dict-2",
text: "hello",
},
})
);
await expect(finishPromise).resolves.toEqual({
dictationId: "dict-2",
text: "hello",
});
});
test("cancels waiters when send fails (no leaked timeouts)", async () => {
vi.useFakeTimers();
const logger = createMockLogger();
packages/server/src/client/daemon-client.ts
+128 -16
View File
@@ -207,6 +207,10 @@ type SetVoiceModePayload = Extract<
SessionOutboundMessage,
{ type: "set_voice_mode_response" }
>["payload"];
type DictationFinishAcceptedPayload = Extract<
SessionOutboundMessage,
{ type: "dictation_stream_finish_accepted" }
>["payload"];
type AgentPermissionResolvedPayload = AgentPermissionResolvedMessage["payload"];
type ListTerminalsPayload = ListTerminalsResponse["payload"];
type CreateTerminalPayload = CreateTerminalResponse["payload"];
@@ -267,6 +271,15 @@ const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000;
/** Default timeout for waiting for connection before sending queued messages */
const DEFAULT_SEND_QUEUE_TIMEOUT_MS = 10000;
const DEFAULT_DICTATION_FINISH_ACCEPT_TIMEOUT_MS = 15000;
const DEFAULT_DICTATION_FINISH_FALLBACK_TIMEOUT_MS = 5 * 60 * 1000;
const DEFAULT_DICTATION_FINISH_TIMEOUT_GRACE_MS = 5000;
function isWaiterTimeoutError(error: unknown): boolean {
return (
error instanceof Error && error.message.startsWith("Timeout waiting for message")
);
}
interface PendingSend {
message: SessionInboundMessage;
@@ -1416,7 +1429,7 @@ export class DaemonClient {
this.sendSessionMessage({ type: "voice_audio_chunk", audio, format, isLast });
}
startDictationStream(dictationId: string, format: string): Promise<void> {
async startDictationStream(dictationId: string, format: string): Promise<void> {
const ack = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_ack") {
@@ -1452,24 +1465,26 @@ export class DaemonClient {
throw new Error(payload.error);
});
const cleanupError = new Error("Cancelled dictation start waiter");
try {
this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format });
} catch (error) {
const err = error instanceof Error ? error : new Error(String(error));
ack.cancel(err);
streamError.cancel(err);
await Promise.race([ackPromise, errorPromise]);
} finally {
ack.cancel(cleanupError);
streamError.cancel(cleanupError);
void ackPromise.catch(() => undefined);
void errorPromise.catch(() => undefined);
throw err;
}
return Promise.race([ackPromise, errorPromise]);
}
sendDictationStreamChunk(dictationId: string, seq: number, audio: string, format: string): void {
this.sendSessionMessageStrict({ type: "dictation_stream_chunk", dictationId, seq, audio, format });
}
finishDictationStream(dictationId: string, finalSeq: number): Promise<{ dictationId: string; text: string }> {
async finishDictationStream(
dictationId: string,
finalSeq: number
): Promise<{ dictationId: string; text: string }> {
const final = this.waitForWithCancel(
(msg) => {
if (msg.type !== "dictation_stream_final") {
@@ -1480,7 +1495,7 @@ export class DaemonClient {
}
return msg.payload;
},
30000,
0,
{ skipQueue: true }
);
@@ -1494,7 +1509,21 @@ export class DaemonClient {
}
return msg.payload;
},
30000,
0,
{ skipQueue: true }
);
const finishAccepted = this.waitForWithCancel<DictationFinishAcceptedPayload>(
(msg) => {
if (msg.type !== "dictation_stream_finish_accepted") {
return null;
}
if (msg.payload.dictationId !== dictationId) {
return null;
}
return msg.payload;
},
DEFAULT_DICTATION_FINISH_ACCEPT_TIMEOUT_MS,
{ skipQueue: true }
);
@@ -1502,18 +1531,101 @@ export class DaemonClient {
const errorPromise = streamError.promise.then((payload) => {
throw new Error(payload.error);
});
const finishAcceptedPromise = finishAccepted.promise;
const finalOutcomePromise = finalPromise.then((payload) => ({
kind: "final" as const,
payload,
}));
const errorOutcomePromise = errorPromise.then(
() => ({
kind: "error" as const,
error: new Error("Unexpected dictation stream error state"),
}),
(error) => ({
kind: "error" as const,
error: error instanceof Error ? error : new Error(String(error)),
})
);
const finishAcceptedOutcomePromise = finishAcceptedPromise.then(
(payload) => ({ kind: "accepted" as const, payload }),
(error) => {
if (isWaiterTimeoutError(error)) {
return { kind: "accepted_timeout" as const };
}
return {
kind: "accepted_error" as const,
error: error instanceof Error ? error : new Error(String(error)),
};
}
);
const waitForFinalResult = async (
timeoutMs: number
): Promise<{ dictationId: string; text: string }> => {
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
const outcome = await Promise.race([finalOutcomePromise, errorOutcomePromise]);
if (outcome.kind === "error") {
throw outcome.error;
}
return outcome.payload;
}
let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
const timeoutPromise = new Promise<{ kind: "timeout" }>((resolve) => {
timeoutHandle = setTimeout(() => resolve({ kind: "timeout" }), timeoutMs);
});
const outcome = await Promise.race([
finalOutcomePromise,
errorOutcomePromise,
timeoutPromise,
]);
if (timeoutHandle) {
clearTimeout(timeoutHandle);
}
if (outcome.kind === "timeout") {
throw new Error(`Timeout waiting for dictation finalization (${timeoutMs}ms)`);
}
if (outcome.kind === "error") {
throw outcome.error;
}
return outcome.payload;
};
const cleanupError = new Error("Cancelled dictation finish waiter");
try {
this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq });
} catch (error) {
const err = error instanceof Error ? error : new Error(String(error));
final.cancel(err);
streamError.cancel(err);
const firstOutcome = await Promise.race([
finalOutcomePromise,
errorOutcomePromise,
finishAcceptedOutcomePromise,
]);
if (firstOutcome.kind === "final") {
return firstOutcome.payload;
}
if (firstOutcome.kind === "error") {
throw firstOutcome.error;
}
if (firstOutcome.kind === "accepted") {
return await waitForFinalResult(
firstOutcome.payload.timeoutMs + DEFAULT_DICTATION_FINISH_TIMEOUT_GRACE_MS
);
}
return await waitForFinalResult(DEFAULT_DICTATION_FINISH_FALLBACK_TIMEOUT_MS);
} finally {
final.cancel(cleanupError);
streamError.cancel(cleanupError);
finishAccepted.cancel(cleanupError);
void finalPromise.catch(() => undefined);
void errorPromise.catch(() => undefined);
throw err;
void finishAcceptedPromise.catch(() => undefined);
}
return Promise.race([finalPromise, errorPromise]);
}
cancelDictationStream(dictationId: string): void {
packages/server/src/server/dictation/dictation-stream-manager.test.ts
+31
View File
@@ -202,4 +202,35 @@ describe("DictationStreamManager (provider-agnostic provider)", () => {
}
}
});
it("adapts finish timeout based on pending committed segments", async () => {
const session = new FakeRealtimeSession();
const emitted: Array<{ type: string; payload: any }> = [];
const manager = new DictationStreamManager({
logger: pino({ level: "silent" }),
emit: (msg) => emitted.push(msg),
sessionId: "s1",
stt: new FakeSttProvider(session),
finalTimeoutMs: 5000,
});
await manager.handleStart("d-timeout", "audio/pcm;rate=24000;bits=16");
await manager.handleChunk({
dictationId: "d-timeout",
seq: 0,
audioBase64: buildPcmBase64(2000, 2400),
format: "audio/pcm;rate=24000;bits=16",
});
// Simulate a committed segment whose final transcript is still pending.
session.emitCommitted("seg-pending");
await manager.handleFinish("d-timeout", 0);
const finishAccepted = emitted.find(
(msg) => msg.type === "dictation_stream_finish_accepted"
);
expect(finishAccepted).toBeDefined();
expect(finishAccepted?.payload.timeoutMs).toBeGreaterThan(5000);
});
});
packages/server/src/server/dictation/dictation-stream-manager.ts
+85 -6
View File
@@ -17,6 +17,10 @@ const PCM_CHANNELS = 1;
const PCM_BITS_PER_SAMPLE = 16;
const DEFAULT_DICTATION_FINAL_TIMEOUT_MS = 10000;
const DEFAULT_DICTATION_AUTO_COMMIT_SECONDS = 15;
const DICTATION_FINAL_TIMEOUT_MAX_MS = 5 * 60 * 1000;
const DICTATION_FINAL_TIMEOUT_PER_PENDING_SEGMENT_MS = 15 * 1000;
const DICTATION_FINAL_TIMEOUT_PER_PENDING_AUDIO_SECOND_MS = 1500;
const DICTATION_FINAL_TIMEOUT_PER_MISSING_SEQ_MS = 250;
const DICTATION_SILENCE_PEAK_THRESHOLD = Number.parseInt(
process.env.PASEO_DICTATION_SILENCE_PEAK_THRESHOLD ?? "300",
10
@@ -91,6 +95,10 @@ type DictationStreamState = {
export type DictationStreamOutboundMessage =
| { type: "dictation_stream_ack"; payload: { dictationId: string; ackSeq: number } }
| {
type: "dictation_stream_finish_accepted";
payload: { dictationId: string; timeoutMs: number };
}
| { type: "dictation_stream_partial"; payload: { dictationId: string; text: string } }
| { type: "dictation_stream_final"; payload: { dictationId: string; text: string; debugRecordingPath?: string } }
| { type: "dictation_stream_error"; payload: { dictationId: string; error: string; retryable: boolean; debugRecordingPath?: string } }
@@ -369,19 +377,46 @@ export class DictationStreamManager {
return;
}
if (state.finalTimeout) {
clearTimeout(state.finalTimeout);
this.maybeSealDictationStreamFinish(dictationId);
this.maybeFinalizeDictationStream(dictationId);
const updatedState = this.streams.get(dictationId);
if (!updatedState) {
return;
}
state.finalTimeout = setTimeout(() => {
const timeoutEstimate = this.estimateFinalizationTimeout(updatedState);
if (updatedState.finalTimeout) {
clearTimeout(updatedState.finalTimeout);
}
updatedState.finalTimeout = setTimeout(() => {
void this.failAndCleanupDictationStream(
dictationId,
"Timed out waiting for final transcription",
true
);
}, this.finalTimeoutMs);
}, timeoutEstimate.timeoutMs);
this.maybeSealDictationStreamFinish(dictationId);
this.maybeFinalizeDictationStream(dictationId);
this.emit({
type: "dictation_stream_finish_accepted",
payload: {
dictationId,
timeoutMs: timeoutEstimate.timeoutMs,
},
});
this.logger.debug(
{
dictationId,
finalSeq,
ackSeq: updatedState.ackSeq,
pendingSegments: timeoutEstimate.pendingSegments,
pendingAudioSeconds: timeoutEstimate.pendingAudioSeconds,
missingSeqCount: timeoutEstimate.missingSeqCount,
timeoutMs: timeoutEstimate.timeoutMs,
},
"Accepted dictation finish request with adaptive timeout budget"
);
}
public handleCancel(dictationId: string): void {
@@ -482,6 +517,50 @@ export class DictationStreamManager {
this.streams.delete(dictationId);
}
private estimateFinalizationTimeout(
state: DictationStreamState
): {
timeoutMs: number;
pendingSegments: number;
pendingAudioSeconds: number;
missingSeqCount: number;
} {
const bytesPerSecond = Math.max(
1,
state.outputRate * PCM_CHANNELS * (PCM_BITS_PER_SAMPLE / 8)
);
const pendingCommittedSegments = state.committedSegmentIds.reduce((count, segmentId) => {
return state.finalTranscriptSegmentIds.has(segmentId) ? count : count + 1;
}, 0);
const pendingSegments =
pendingCommittedSegments + (state.awaitingFinalCommit ? 1 : 0);
const pendingAudioSeconds = Math.ceil(
Math.max(0, state.bytesSinceCommit) / bytesPerSecond
);
const missingSeqCount =
state.finalSeq === null ? 0 : Math.max(0, state.finalSeq - state.ackSeq);
const extraMs =
pendingSegments * DICTATION_FINAL_TIMEOUT_PER_PENDING_SEGMENT_MS +
pendingAudioSeconds * DICTATION_FINAL_TIMEOUT_PER_PENDING_AUDIO_SECOND_MS +
missingSeqCount * DICTATION_FINAL_TIMEOUT_PER_MISSING_SEQ_MS;
const timeoutMs = Math.max(
this.finalTimeoutMs,
Math.min(
DICTATION_FINAL_TIMEOUT_MAX_MS,
this.finalTimeoutMs + extraMs
)
);
return {
timeoutMs,
pendingSegments,
pendingAudioSeconds,
missingSeqCount,
};
}
private maybeAutoCommitDictationSegment(state: DictationStreamState): void {
if (state.finishRequested) {
return;
packages/server/src/shared/messages.ts
+9
View File
@@ -1112,6 +1112,14 @@ export const DictationStreamAckMessageSchema = z.object({
}),
});
export const DictationStreamFinishAcceptedMessageSchema = z.object({
type: z.literal("dictation_stream_finish_accepted"),
payload: z.object({
dictationId: z.string(),
timeoutMs: z.number().int().positive(),
}),
});
export const DictationStreamPartialMessageSchema = z.object({
type: z.literal("dictation_stream_partial"),
payload: z.object({
@@ -1817,6 +1825,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AudioOutputMessageSchema,
TranscriptionResultMessageSchema,
DictationStreamAckMessageSchema,
DictationStreamFinishAcceptedMessageSchema,
DictationStreamPartialMessageSchema,
DictationStreamFinalMessageSchema,
DictationStreamErrorMessageSchema,