diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index c20748d3b..ee9a054ee 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -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(); diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index c843b0616..39355f94c 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -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 { + async startDictationStream(dictationId: string, format: string): Promise { 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( + (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 | 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 { 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 f4085d0aa..e8ace8fab 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.test.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.test.ts @@ -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); + }); }); diff --git a/packages/server/src/server/dictation/dictation-stream-manager.ts b/packages/server/src/server/dictation/dictation-stream-manager.ts index d32993c42..40667903a 100644 --- a/packages/server/src/server/dictation/dictation-stream-manager.ts +++ b/packages/server/src/server/dictation/dictation-stream-manager.ts @@ -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; diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index 2041a9122..748beaac3 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -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,