diff --git a/mobile/rpc-foundation/goldens/matrix-speech.dictation-chunk-speech.dictation.chunk-1.json b/mobile/rpc-foundation/goldens/matrix-speech.dictation-chunk-speech.dictation.chunk-1.json index 8ba233197bf..88ebf2c8bd7 100644 --- a/mobile/rpc-foundation/goldens/matrix-speech.dictation-chunk-speech.dictation.chunk-1.json +++ b/mobile/rpc-foundation/goldens/matrix-speech.dictation-chunk-speech.dictation.chunk-1.json @@ -4,7 +4,22 @@ "family": "speech.dictation-chunk", "namedDeltas": [], "values": { - "028afac3649e": { + "02cebe6f0062": { + "failures": [], + "pending": 0 + }, + "19c961e90970": { + "name": "dictation-failed", + "ordinal": 3, + "value": { + "id": "dictation-1" + } + }, + "2762d465637e": { + "failures": ["Unknown method"], + "pending": 0 + }, + "36a6b302a73d": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -17,7 +32,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -43,27 +59,7 @@ } } }, - "02cebe6f0062": { - "failures": [], - "pending": 0 - }, - "12521d219054": { - "name": "speech.dictation.chunk#1", - "ordinal": 2, - "json": "{\"id\":\"frame-1\",\"deviceToken\":\"recording-device\",\"method\":\"speech.dictation.chunk\",\"params\":{\"dictationId\":\"dictation-1\",\"audioBase64\":\"ACVKb5S53gM=\",\"sampleRate\":16000}}" - }, - "19c961e90970": { - "name": "dictation-failed", - "ordinal": 3, - "value": { - "id": "dictation-1" - } - }, - "2762d465637e": { - "failures": ["Unknown method"], - "pending": 0 - }, - "30ae620f58dc": { + "5133fafc8626": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -76,41 +72,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 - } - }, - { - "name": "options", - "value": { - "timeoutMs": 30000 - } - } - ], - "settlement": { - "status": "rejected", - "startedAt": 0, - "settledAt": 0, - "error": { - "category": "Error", - "message": "transport failure", - "isRpcDeliveryUnknown": true - } - } - }, - "62cfcef21f01": { - "name": "speech.dictation.chunk#1", - "ordinal": 1, - "args": [ - { - "name": "method", - "value": "speech.dictation.chunk" - }, - { - "name": "params", - "value": { - "audioBase64": "ACVKb5S53gM=", - "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -126,11 +89,14 @@ "settledAt": 0, "value": { "id": "frame-1", - "ok": true + "ok": true, + "result": { + "received": true + } } } }, - "66dece39b5ba": { + "65e5e348650c": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -143,7 +109,45 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 + } + }, + { + "name": "options", + "value": { + "timeoutMs": 30000 + } + } + ], + "settlement": { + "status": "fulfilled", + "startedAt": 0, + "settledAt": 0, + "value": { + "id": "frame-1", + "ok": true, + "result": { + "error": "refused" + } + } + } + }, + "66cc21847b0f": { + "name": "speech.dictation.chunk#1", + "ordinal": 1, + "args": [ + { + "name": "method", + "value": "speech.dictation.chunk" + }, + { + "name": "params", + "value": { + "audioBase64": "ACVKb5S53gM=", + "dictationId": "dictation-1", + "sampleRate": 16000, + "seq": 0 } }, { @@ -167,7 +171,7 @@ } } }, - "83a5bdf2ac1a": { + "67bd399925f2": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -180,7 +184,48 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 + } + }, + { + "name": "options", + "value": { + "timeoutMs": 30000 + } + } + ], + "settlement": { + "status": "rejected", + "startedAt": 0, + "settledAt": 0, + "error": { + "category": "Error", + "message": "transport failure", + "isRpcDeliveryUnknown": true + } + } + }, + "70f6f0244686": { + "name": "speech.dictation.chunk#1", + "ordinal": 2, + "json": "{\"id\":\"frame-1\",\"deviceToken\":\"recording-device\",\"method\":\"speech.dictation.chunk\",\"params\":{\"dictationId\":\"dictation-1\",\"audioBase64\":\"ACVKb5S53gM=\",\"sampleRate\":16000,\"seq\":0}}" + }, + "8c69a5dd9dd6": { + "name": "speech.dictation.chunk#1", + "ordinal": 1, + "args": [ + { + "name": "method", + "value": "speech.dictation.chunk" + }, + { + "name": "params", + "value": { + "audioBase64": "ACVKb5S53gM=", + "dictationId": "dictation-1", + "sampleRate": 16000, + "seq": 0 } }, { @@ -198,12 +243,12 @@ "id": "frame-1", "ok": true, "result": { - "error": "refused" + "$rpc": "null" } } } }, - "888a29b4fd97": { + "8da39029e37b": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -216,7 +261,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -232,15 +278,11 @@ "settledAt": 0, "value": { "id": "frame-1", - "ok": true, - "result": { - "error": "inner refused", - "ok": false - } + "ok": true } } }, - "99956eef8fff": { + "a280755140c5": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -253,43 +295,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 - } - }, - { - "name": "options", - "value": { - "timeoutMs": 30000 - } - } - ], - "settlement": { - "status": "fulfilled", - "startedAt": 0, - "settledAt": 0, - "value": { - "id": "frame-1", - "ok": true, - "result": { - "received": true - } - } - } - }, - "9d43743b1b10": { - "name": "speech.dictation.chunk#1", - "ordinal": 1, - "args": [ - { - "name": "method", - "value": "speech.dictation.chunk" - }, - { - "name": "params", - "value": { - "audioBase64": "ACVKb5S53gM=", - "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -314,24 +321,7 @@ "failures": ["outer refused"], "pending": 0 }, - "b5157118341f": { - "failures": [""], - "pending": 0 - }, - "bc459c132276": { - "status": "fulfilled", - "startedAt": 0, - "settledAt": 0, - "value": [ - { - "status": "fulfilled", - "value": { - "$rpc": "undefined" - } - } - ] - }, - "da1579f3254f": { + "b0ee39c0ea35": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -344,44 +334,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 - } - }, - { - "name": "options", - "value": { - "timeoutMs": 30000 - } - } - ], - "settlement": { - "status": "fulfilled", - "startedAt": 0, - "settledAt": 0, - "value": { - "error": { - "code": "method_not_found", - "message": "Unknown method" - }, - "id": "frame-1", - "ok": false - } - } - }, - "f6a8a6f550f5": { - "name": "speech.dictation.chunk#1", - "ordinal": 1, - "args": [ - { - "name": "method", - "value": "speech.dictation.chunk" - }, - { - "name": "params", - "value": { - "audioBase64": "ACVKb5S53gM=", - "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -405,7 +359,24 @@ } } }, - "f779babe2a18": { + "b5157118341f": { + "failures": [""], + "pending": 0 + }, + "bc459c132276": { + "status": "fulfilled", + "startedAt": 0, + "settledAt": 0, + "value": [ + { + "status": "fulfilled", + "value": { + "$rpc": "undefined" + } + } + ] + }, + "e37b9cab121b": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -418,7 +389,46 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 + } + }, + { + "name": "options", + "value": { + "timeoutMs": 30000 + } + } + ], + "settlement": { + "status": "fulfilled", + "startedAt": 0, + "settledAt": 0, + "value": { + "error": { + "code": "method_not_found", + "message": "Unknown method" + }, + "id": "frame-1", + "ok": false + } + } + }, + "f6826a08b96a": { + "name": "speech.dictation.chunk#1", + "ordinal": 1, + "args": [ + { + "name": "method", + "value": "speech.dictation.chunk" + }, + { + "name": "params", + "value": { + "audioBase64": "ACVKb5S53gM=", + "dictationId": "dictation-1", + "sampleRate": 16000, + "seq": 0 } }, { @@ -436,7 +446,8 @@ "id": "frame-1", "ok": true, "result": { - "$rpc": "null" + "error": "inner refused", + "ok": false } } } @@ -448,8 +459,8 @@ { "id": "speech-audio-chunk-acknowledged.normal:acknowledged", "observation": { - "sender": ["99956eef8fff"], - "payloads": ["12521d219054"], + "sender": ["5133fafc8626"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -460,8 +471,8 @@ { "id": "speech-audio-chunk-acknowledged.result-absent:acknowledged", "observation": { - "sender": ["62cfcef21f01"], - "payloads": ["12521d219054"], + "sender": ["8da39029e37b"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -472,8 +483,8 @@ { "id": "speech-audio-chunk-acknowledged.result-null:acknowledged", "observation": { - "sender": ["f779babe2a18"], - "payloads": ["12521d219054"], + "sender": ["8c69a5dd9dd6"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -484,8 +495,8 @@ { "id": "speech-audio-chunk-acknowledged.inner-ok-missing:acknowledged", "observation": { - "sender": ["83a5bdf2ac1a"], - "payloads": ["12521d219054"], + "sender": ["65e5e348650c"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -496,8 +507,8 @@ { "id": "speech-audio-chunk-acknowledged.inner-false-string-error:acknowledged", "observation": { - "sender": ["888a29b4fd97"], - "payloads": ["12521d219054"], + "sender": ["f6826a08b96a"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -508,8 +519,8 @@ { "id": "speech-audio-chunk-acknowledged.inner-false-object-error:acknowledged", "observation": { - "sender": ["028afac3649e"], - "payloads": ["12521d219054"], + "sender": ["36a6b302a73d"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -520,8 +531,8 @@ { "id": "speech-audio-chunk-acknowledged.outer-refused:acknowledged", "observation": { - "sender": ["66dece39b5ba"], - "payloads": ["12521d219054"], + "sender": ["66cc21847b0f"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -532,8 +543,8 @@ { "id": "speech-audio-chunk-acknowledged.outer-refused-no-message:acknowledged", "observation": { - "sender": ["f6a8a6f550f5"], - "payloads": ["12521d219054"], + "sender": ["b0ee39c0ea35"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -544,8 +555,8 @@ { "id": "speech-audio-chunk-acknowledged.method-not-found:acknowledged", "observation": { - "sender": ["da1579f3254f"], - "payloads": ["12521d219054"], + "sender": ["e37b9cab121b"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -556,8 +567,8 @@ { "id": "speech-audio-chunk-acknowledged.transport-rejection:acknowledged", "observation": { - "sender": ["30ae620f58dc"], - "payloads": ["12521d219054"], + "sender": ["67bd399925f2"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, @@ -568,8 +579,8 @@ { "id": "speech-audio-chunk-acknowledged.transport-rejection-no-message:acknowledged", "observation": { - "sender": ["9d43743b1b10"], - "payloads": ["12521d219054"], + "sender": ["a280755140c5"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, diff --git a/mobile/rpc-foundation/goldens/speech-audio-chunk-acknowledged.json b/mobile/rpc-foundation/goldens/speech-audio-chunk-acknowledged.json index 0f3839c8ed0..70bea28d688 100644 --- a/mobile/rpc-foundation/goldens/speech-audio-chunk-acknowledged.json +++ b/mobile/rpc-foundation/goldens/speech-audio-chunk-acknowledged.json @@ -8,12 +8,7 @@ "failures": [], "pending": 0 }, - "12521d219054": { - "name": "speech.dictation.chunk#1", - "ordinal": 2, - "json": "{\"id\":\"frame-1\",\"deviceToken\":\"recording-device\",\"method\":\"speech.dictation.chunk\",\"params\":{\"dictationId\":\"dictation-1\",\"audioBase64\":\"ACVKb5S53gM=\",\"sampleRate\":16000}}" - }, - "99956eef8fff": { + "5133fafc8626": { "name": "speech.dictation.chunk#1", "ordinal": 1, "args": [ @@ -26,7 +21,8 @@ "value": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 } }, { @@ -49,6 +45,11 @@ } } }, + "70f6f0244686": { + "name": "speech.dictation.chunk#1", + "ordinal": 2, + "json": "{\"id\":\"frame-1\",\"deviceToken\":\"recording-device\",\"method\":\"speech.dictation.chunk\",\"params\":{\"dictationId\":\"dictation-1\",\"audioBase64\":\"ACVKb5S53gM=\",\"sampleRate\":16000,\"seq\":0}}" + }, "bc459c132276": { "status": "fulfilled", "startedAt": 0, @@ -69,8 +70,8 @@ { "id": "acknowledged", "observation": { - "sender": ["99956eef8fff"], - "payloads": ["12521d219054"], + "sender": ["5133fafc8626"], + "payloads": ["70f6f0244686"], "settlements": { "chunk": "bc459c132276" }, diff --git a/mobile/rpc-foundation/pilot-scenarios.json b/mobile/rpc-foundation/pilot-scenarios.json index 2fff0ffed11..67d2919bf1e 100644 --- a/mobile/rpc-foundation/pilot-scenarios.json +++ b/mobile/rpc-foundation/pilot-scenarios.json @@ -14549,7 +14549,8 @@ "params": { "audioBase64": "ACVKb5S53gM=", "dictationId": "dictation-1", - "sampleRate": 16000 + "sampleRate": 16000, + "seq": 0 }, "reply": { "ok": true, diff --git a/mobile/src/hooks/mobile-dictation-audio-chunk.ts b/mobile/src/hooks/mobile-dictation-audio-chunk.ts index f0f29c0743b..41bfeb7b0f1 100644 --- a/mobile/src/hooks/mobile-dictation-audio-chunk.ts +++ b/mobile/src/hooks/mobile-dictation-audio-chunk.ts @@ -52,18 +52,26 @@ export function useMobileDictationTranscriptSoFar() { } // A request refused or still waiting while the connection moved never reached the host; send it again on the new one, within the host's grace. +// With resendWhile the host dedupes, so a request whose reply a drop lost goes again after each reconnect too, while it still holds. export async function sendAcrossConnectionMove( client: RpcClient, - send: () => Promise + send: () => Promise, + resendWhile?: () => boolean ): Promise { + const again = () => (resendWhile ? sendAcrossConnectionMove(client, send, resendWhile) : send()) try { return await send() } catch (err) { if (isLogicalClientCutoverError(err) && !isRpcDeliveryUnknown(err)) { - return send() + return again() } - if (isLogicalClientSuspendedError(err) && (await waitForReconnect(client))) { - return send() + // A timeout on a connection still up gets no second try there; a reply lost to a drop does. + const lostToDrop = isLogicalClientCutoverError(err) || client.getState() !== 'connected' + const resend = + isLogicalClientSuspendedError(err) || + (resendWhile !== undefined && isRpcDeliveryUnknown(err) && lostToDrop) + if (resend && (await waitForReconnect(client)) && (resendWhile?.() ?? true)) { + return again() } throw err } @@ -130,13 +138,18 @@ export function enqueueMobileDictationAudioChunk( const params = { dictationId, audioBase64: bytesToBase64(bytes), - sampleRate: MOBILE_DICTATION_PCM_SAMPLE_RATE + sampleRate: MOBILE_DICTATION_PCM_SAMPLE_RATE, + seq: queue.pendingAudioBudget.nextSeq() } - const sendChunk = sendAcrossConnectionMove(client, () => - // Wait for a reconnect only as long as the host keeps the session; past that, fail and keep the text so far. - dictationAudioChunkSend.request(client, params, { - timeoutMs: MOBILE_DICTATION_RECONNECT_GRACE_MS - }) + const sendChunk = sendAcrossConnectionMove( + client, + () => + // Wait for a reconnect only as long as the host keeps the session; past that, fail and keep the text so far. + dictationAudioChunkSend.request(client, params, { + timeoutMs: MOBILE_DICTATION_RECONNECT_GRACE_MS + }), + // A recording cancelled during the outage must not reach the host when it comes back. + queue.pendingAudioBudget.hostDedupesResends ? () => queue.isActive(dictationId) : undefined ) .then((reply) => { // Only newer hosts send the text, and only when it changed; any other result still acknowledges the chunk. @@ -147,7 +160,7 @@ export function enqueueMobileDictationAudioChunk( } }) .catch((err) => { - // The host keeps the session through a reconnect; only audio in flight at the drop is lost. + // The host keeps the session through a reconnect; a host that cannot dedupe loses only the audio in flight at the drop. if (!isRpcDeliveryUnknown(err)) { queue.failActiveDictation(dictationId, err) } diff --git a/mobile/src/hooks/mobile-dictation-desktop-start.ts b/mobile/src/hooks/mobile-dictation-desktop-start.ts index ea1f1511ab2..08f42de4c45 100644 --- a/mobile/src/hooks/mobile-dictation-desktop-start.ts +++ b/mobile/src/hooks/mobile-dictation-desktop-start.ts @@ -5,6 +5,7 @@ import { dictationSessionStart } from '../dictation/mobile-dictation-operations' import { MOBILE_DICTATION_RECONNECT_GRACE_MS } from '../../../src/shared/rpc-contract/speech-params' +import { rpcPayloadMember } from '../transport/rpc-reader-payload' import type { RpcClient } from '../transport/rpc-client' type StartMobileDictationDesktopSessionOptions = { @@ -16,7 +17,7 @@ type StartMobileDictationDesktopSessionOptions = { getActiveId: () => string | null clearActiveId: (dictationId: string) => void setIdle: () => void - commitRecordingStart: () => boolean + commitRecordingStart: (hostDedupesResends: boolean) => boolean rollbackRecordingStart: () => void } @@ -61,6 +62,7 @@ export async function startMobileDictationDesktopSession( options: StartMobileDictationDesktopSessionOptions ): Promise { const { client, dictationId } = options + let hostDedupesResends = false try { // The mic is already open, so a start stalled by a drop must give up rather than hold it. @@ -69,7 +71,9 @@ export async function startMobileDictationDesktopSession( { dictationId }, { timeoutMs: MOBILE_DICTATION_RECONNECT_GRACE_MS } ) - dictationSessionStart.interpret(reply) + const started = dictationSessionStart.interpret(reply) + hostDedupesResends = + started != null && rpcPayloadMember(started, 'dedupesResentChunks') === true } catch (err) { const wasCurrent = isCurrentStart(options) // Capture already holds the screen; roll both back only while this start is current. A stale start could otherwise stop its replacement’s microphone. Capture itself has no start identity. @@ -100,7 +104,7 @@ export async function startMobileDictationDesktopSession( try { // Commit before returning so queued cancellation cannot resurrect recording after cleanup. - if (!options.commitRecordingStart()) { + if (!options.commitRecordingStart(hostDedupesResends)) { throw new Error( translate( 'hooks.dictationDesktopStart.recordingFailed', diff --git a/mobile/src/hooks/mobile-dictation-pending-audio-budget.ts b/mobile/src/hooks/mobile-dictation-pending-audio-budget.ts index c90d0f313c5..507f47c1609 100644 --- a/mobile/src/hooks/mobile-dictation-pending-audio-budget.ts +++ b/mobile/src/hooks/mobile-dictation-pending-audio-budget.ts @@ -1,24 +1,18 @@ import { DICTATION_SAMPLE_RATE, - MOBILE_DICTATION_RECONNECT_GRACE_MS, - PCM_BYTES_PER_SAMPLE + MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES } from '../../../src/shared/rpc-contract/speech-params' export const MOBILE_DICTATION_PCM_SAMPLE_RATE = DICTATION_SAMPLE_RATE -// Hold the speech captured while a reconnect is pending, for as long as the host keeps the session. -const MAX_PENDING_AUDIO_SECONDS = MOBILE_DICTATION_RECONNECT_GRACE_MS / 1000 - -// Why: dictation chunk RPCs wait through reconnect before timing out, so -// cap retained raw microphone audio before it is expanded into base64 payloads. -const MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES = - MOBILE_DICTATION_PCM_SAMPLE_RATE * PCM_BYTES_PER_SAMPLE * MAX_PENDING_AUDIO_SECONDS - export const MOBILE_DICTATION_CONNECTION_SLOW_ERROR_MESSAGE = 'Connection is too slow for voice dictation. Try again when the connection improves.' export class MobileDictationPendingAudioBudget { private pendingBytes = 0 + private sentChunks = 0 + // Set at each start: a host that feeds each chunk seq once can take audio a drop left unanswered again. + hostDedupesResends = false constructor(private readonly maxPendingBytes = MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES) {} @@ -26,6 +20,10 @@ export class MobileDictationPendingAudioBudget { return this.pendingBytes } + nextSeq(): number { + return this.sentChunks++ + } + tryReserve(byteLength: number): boolean { const normalizedByteLength = normalizeByteLength(byteLength) if (this.pendingBytes + normalizedByteLength > this.maxPendingBytes) { @@ -39,8 +37,10 @@ export class MobileDictationPendingAudioBudget { this.pendingBytes = Math.max(0, this.pendingBytes - normalizeByteLength(byteLength)) } - reset(): void { + reset(hostDedupesResends = false): void { this.pendingBytes = 0 + this.sentChunks = 0 + this.hostDedupesResends = hostDedupesResends } } diff --git a/mobile/src/hooks/mobile-dictation-reconnect.test.tsx b/mobile/src/hooks/mobile-dictation-reconnect.test.tsx index ec1b94e82cc..234f22df15f 100644 --- a/mobile/src/hooks/mobile-dictation-reconnect.test.tsx +++ b/mobile/src/hooks/mobile-dictation-reconnect.test.tsx @@ -8,6 +8,7 @@ import { type SentRequest } from '../mobile-web-shell/bridge-host-test-fakes' import { markRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity' +import { rpcPayloadMember } from '../transport/rpc-reader-payload' import { LogicalClientCutoverError } from '../transport/stable-logical-rpc-client' import type { ConnectionState } from '../transport/types' import type { @@ -182,6 +183,71 @@ describe('dictation through a reconnect', () => { }) }) +describe('audio sent on a connection that died', () => { + async function dropMidChunk( + startResult: Record, + { cancelOffline = false } = {} + ) { + let state: ConnectionState = 'connected' + const rpc = createFakeRpcClient({ getState: () => state }) + const sent: SentRequest[] = [] + mount(rpc, [], []) + await act(async () => { + const started = dictation().start() + while (rpc.requests.length === 0) { + await Promise.resolve() + } + rpc.requests.splice(0)[0].resolve({ id: 'desktop', ok: true, result: startResult }) + await started + }) + const speak = () => { + for (const handler of seam.chunkHandlers) { + handler({ data: Uint8Array.from([1, 2]), droppedBytes: 0 }) + } + } + await act(async () => { + speak() + speak() + const [answered, lost] = rpc.requests.splice(0) + sent.push(answered, lost) + answered.resolve({ id: 'desktop', ok: true, result: {} }) + state = 'reconnecting' + lost.reject(markRpcDeliveryUnknown(new Error('Connection interrupted'))) + await Promise.resolve() + speak() + }) + if (cancelOffline) { + await act(async () => { + const cancelled = dictation().cancel() + await pump(rpc, sent, []) + await cancelled + }) + } + await act(async () => { + state = 'connected' + rpc.pushState('connected') + await pump(rpc, sent, []) + }) + return sent + .filter((request) => request.method === 'speech.dictation.chunk') + .map((request) => rpcPayloadMember(request.args[1], 'seq')) + } + + it('goes again to a host that feeds each chunk once', async () => { + await expect(dropMidChunk({ dedupesResentChunks: true })).resolves.toEqual([0, 1, 2, 1]) + }) + + it('is not resent once the recording was cancelled during the outage', async () => { + await expect( + dropMidChunk({ dedupesResentChunks: true }, { cancelOffline: true }) + ).resolves.toEqual([0, 1, 2]) + }) + + it('never goes twice to a host that would feed it twice', async () => { + await expect(dropMidChunk({})).resolves.toEqual([0, 1, 2]) + }) +}) + describe('the text so far', () => { it('shows what the host heard while recording and clears it as the final text lands', async () => { const rpc = createFakeRpcClient() diff --git a/mobile/src/hooks/use-mobile-dictation.ts b/mobile/src/hooks/use-mobile-dictation.ts index 19663bf7a3b..bab182b12bf 100644 --- a/mobile/src/hooks/use-mobile-dictation.ts +++ b/mobile/src/hooks/use-mobile-dictation.ts @@ -171,10 +171,10 @@ export function useMobileDictation(options: UseMobileDictationOptions): UseMobil } }, setIdle: () => applyStatus('idle'), - commitRecordingStart: () => { + commitRecordingStart: (hostDedupesResends) => { acceptingChunksRef.current = true pendingChunksRef.current.clear() - pendingAudioBudgetRef.current.reset() + pendingAudioBudgetRef.current.reset(hostDedupesResends) if (!capture.begin()) { return false } diff --git a/src/main/runtime/mobile-dictation-chunk-order.ts b/src/main/runtime/mobile-dictation-chunk-order.ts new file mode 100644 index 00000000000..cd4f5fbdf51 --- /dev/null +++ b/src/main/runtime/mobile-dictation-chunk-order.ts @@ -0,0 +1,51 @@ +import { MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES } from '../../shared/rpc-contract/speech-params' + +export type MobileDictationChunk = { audioBase64: string; sampleRate: number } + +// Past what the phone holds through a reconnect, a missing chunk is given up on. +const MAX_HELD_BASE64_LENGTH = Math.ceil(MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES / 3) * 4 +// The grace's worth of native microphone events, one chunk per ~32ms; tiny chunks must not pile up under the byte cap. +const MAX_HELD_CHUNKS = 1_000 + +// A chunk resent after a drop can land behind audio captured later, so release audio in seq order and each seq once. +export class MobileDictationChunkOrder { + private nextSeq = 0 + private heldLength = 0 + private readonly held = new Map() + + accept(seq: number, chunk: MobileDictationChunk): MobileDictationChunk[] { + if (seq < this.nextSeq || this.held.has(seq)) { + return [] + } + this.held.set(seq, chunk) + this.heldLength += chunk.audioBase64.length + const ready = this.releaseInOrder() + while (this.heldLength > MAX_HELD_BASE64_LENGTH || this.held.size > MAX_HELD_CHUNKS) { + this.nextSeq = Infinity + for (const held of this.held.keys()) { + this.nextSeq = Math.min(this.nextSeq, held) + } + ready.push(...this.releaseInOrder()) + } + return ready + } + + private releaseInOrder(): MobileDictationChunk[] { + const ready: MobileDictationChunk[] = [] + for (let next = this.held.get(this.nextSeq); next; next = this.held.get(this.nextSeq)) { + ready.push(next) + this.held.delete(this.nextSeq) + this.heldLength -= next.audioBase64.length + this.nextSeq += 1 + } + return ready + } + + // What is still held when the phone finishes, skipping chunks that never came. + drain(): MobileDictationChunk[] { + const rest = [...this.held].sort(([a], [b]) => a - b).map(([, chunk]) => chunk) + this.held.clear() + this.heldLength = 0 + return rest + } +} diff --git a/src/main/runtime/runtime-mobile-dictation-controller.test.ts b/src/main/runtime/runtime-mobile-dictation-controller.test.ts index 01b16db1e17..e6d66e96e06 100644 --- a/src/main/runtime/runtime-mobile-dictation-controller.test.ts +++ b/src/main/runtime/runtime-mobile-dictation-controller.test.ts @@ -18,18 +18,20 @@ const stopDictation = vi.fn(async (_owner: string, options?: { discard?: boolean await transcription } }) +const feedAudio = vi.fn() beforeEach(() => { vi.useFakeTimers() stopDictation.mockClear() transcription = Promise.resolve() + feedAudio.mockClear() const factories = { createModelManager: () => ({ getModelState: async () => ({ status: 'ready' }) }), createSttService: () => ({ startDictation: async (_modelId: string, next: Sink) => { sink = next }, - feedAudio: () => {}, + feedAudio, stopDictation }) } @@ -81,6 +83,51 @@ describe('mobile dictation across a phone reconnect', () => { expect(stopDictation).toHaveBeenCalledOnce() expect(() => controller.feed(chunk)).toThrow('dictation_stream_not_started') }) + + it('feeds resent audio once and in the order it was spoken', async () => { + const controller = await startPhoneDictation() + const spoken = (seq: number) => ({ + ...chunk, + seq, + audioBase64: Buffer.from(Int16Array.of(seq).buffer).toString('base64') + }) + + // 1 is resent after 2 overtook it; 0 and 1 arrive twice; 3 never comes. + for (const seq of [0, 2, 0, 1, 1, 4]) { + controller.feed(spoken(seq)) + } + await controller.finish({ dictationId: 'd-1', clientId: 'phone-a' }) + + const fed = feedAudio.mock.calls.map(([samples]) => Math.round(samples[0] * 32768)) + expect(fed).toEqual([0, 1, 2, 4]) + }) + + // 35s of speech, or more tiny chunks than 30s of native microphone events, is more than a reconnect holds back. + it.each([ + { held: 'long chunks', count: 7, audioBase64: Buffer.alloc(160_000).toString('base64') }, + { held: 'tiny chunks', count: 1001, audioBase64: 'AAA=' } + ])('stops waiting for a chunk that never comes behind $held', async ({ count, audioBase64 }) => { + const controller = await startPhoneDictation() + + for (let seq = 1; seq <= count; seq += 1) { + controller.feed({ ...chunk, audioBase64, seq }) + } + + expect(feedAudio).toHaveBeenCalledTimes(count) + }) + it('gives up on as many gaps as it takes to stay within what a reconnect holds back', async () => { + const controller = await startPhoneDictation() + const fiveSeconds = Buffer.alloc(160_000).toString('base64') + + // Every other chunk is missing: ten tiny ones, then ten of five seconds each. + for (let index = 0; index < 20; index += 1) { + const audioBase64 = index < 10 ? 'AAA=' : fiveSeconds + controller.feed({ ...chunk, audioBase64, seq: 2 * index + 1 }) + } + + // At most five 5s chunks stay held under the 30s cap. + expect(feedAudio).toHaveBeenCalledTimes(15) + }) }) describe('mobile dictation cancel', () => { @@ -92,6 +139,6 @@ describe('mobile dictation cancel', () => { await expect( controller.start({ dictationId: 'd-2', modelId: 'test-model', clientId: 'phone-a' }) - ).resolves.toEqual({ dictationId: 'd-2', modelId: 'test-model' }) + ).resolves.toEqual({ dictationId: 'd-2', modelId: 'test-model', dedupesResentChunks: true }) }) }) diff --git a/src/main/runtime/runtime-mobile-dictation-controller.ts b/src/main/runtime/runtime-mobile-dictation-controller.ts index 32f8e3ac369..4861628d56f 100644 --- a/src/main/runtime/runtime-mobile-dictation-controller.ts +++ b/src/main/runtime/runtime-mobile-dictation-controller.ts @@ -6,6 +6,10 @@ import { hasLocalSpeechRecognizer, LOCAL_SPEECH_UNAVAILABLE_ERROR } from '../speech/stt-worker-paths' +import { + MobileDictationChunkOrder, + type MobileDictationChunk +} from './mobile-dictation-chunk-order' import type { RuntimeStore } from './runtime-store-contract' type MobileDictationSession = { @@ -20,6 +24,7 @@ type MobileDictationSession = { textVersion: number reportedVersion: number reportedAt: number + chunkOrder: MobileDictationChunkOrder abandonTimer?: ReturnType } @@ -32,7 +37,7 @@ export class RuntimeMobileDictationController { dictationId: string modelId?: string clientId?: string - }): Promise<{ dictationId: string; modelId: string }> { + }): Promise<{ dictationId: string; modelId: string; dedupesResentChunks: true }> { const store = this.requireStore() const voice = store.getSettings().voice ?? getDefaultVoiceSettings() if (!voice.enabled) { @@ -72,7 +77,8 @@ export class RuntimeMobileDictationController { errors: [], textVersion: 0, reportedVersion: 0, - reportedAt: 0 + reportedAt: 0, + chunkOrder: new MobileDictationChunkOrder() } try { await getSpeechSttService(store).startDictation( @@ -97,13 +103,15 @@ export class RuntimeMobileDictationController { } throw error } - return { dictationId: params.dictationId, modelId } + // Tells the phone it may resend chunks a drop left unanswered. + return { dictationId: params.dictationId, modelId, dedupesResentChunks: true } } feed(params: { dictationId: string audioBase64: string sampleRate: number + seq?: number clientId?: string }): { dictationId: string; text?: string } { const session = this.requireOwnedSession(params) @@ -113,12 +121,12 @@ export class RuntimeMobileDictationController { if (session.errors.length > 0) { throw new Error(session.errors[0]) } - const pcm = Buffer.from(params.audioBase64, 'base64') - const samples = new Float32Array(Math.floor(pcm.length / 2)) - for (let i = 0; i < samples.length; i += 1) { - samples[i] = pcm.readInt16LE(i * 2) / 32768 + // Phones that send no seq never resend. + const ready = + params.seq === undefined ? [params] : session.chunkOrder.accept(params.seq, params) + for (const chunk of ready) { + this.feedChunk(session, chunk) } - getSpeechSttService(this.requireStore()).feedAudio(samples, params.sampleRate, session.owner) // The text so far lets the phone keep it if the connection never comes back; resent each second in case a reply was lost. const now = Date.now() if (session.reportedVersion === session.textVersion && now - session.reportedAt < 1000) { @@ -134,6 +142,9 @@ export class RuntimeMobileDictationController { clientId?: string }): Promise<{ dictationId: string; text: string }> { const session = this.requireOwnedSession(params) + for (const chunk of session.chunkOrder.drain()) { + this.feedChunk(session, chunk) + } await this.stopSession(session, { discard: false }) if (session.errors.length > 0) { throw new Error(session.errors[0]) @@ -152,6 +163,15 @@ export class RuntimeMobileDictationController { return { dictationId: params.dictationId } } + private feedChunk(session: MobileDictationSession, chunk: MobileDictationChunk): void { + const pcm = Buffer.from(chunk.audioBase64, 'base64') + const samples = new Float32Array(Math.floor(pcm.length / 2)) + for (let i = 0; i < samples.length; i += 1) { + samples[i] = pcm.readInt16LE(i * 2) / 32768 + } + getSpeechSttService(this.requireStore()).feedAudio(samples, chunk.sampleRate, session.owner) + } + private acceptEvent( dictationId: string, event: { type: string; text?: string; error?: string } diff --git a/src/shared/rpc-contract/speech-params.ts b/src/shared/rpc-contract/speech-params.ts index 043585c42f1..e73d653ed11 100644 --- a/src/shared/rpc-contract/speech-params.ts +++ b/src/shared/rpc-contract/speech-params.ts @@ -12,6 +12,10 @@ export const MAX_DICTATION_AUDIO_SECONDS = 5 // The host drops a phone dictation it hasn't heard from for this long, so the phone waits no longer for a reconnect. export const MOBILE_DICTATION_RECONNECT_GRACE_MS = 30_000 +// The speech a phone holds through a reconnect, and so what the host holds back behind a missing resend. +export const MOBILE_DICTATION_MAX_PENDING_AUDIO_BYTES = + (DICTATION_SAMPLE_RATE * PCM_BYTES_PER_SAMPLE * MOBILE_DICTATION_RECONNECT_GRACE_MS) / 1000 + export const MAX_DICTATION_AUDIO_CHUNK_BYTES = DICTATION_SAMPLE_RATE * PCM_BYTES_PER_SAMPLE * MAX_DICTATION_AUDIO_SECONDS @@ -37,7 +41,9 @@ export const DictationChunk = z.object({ ) // Buffer.from silently discards malformed base64; refuse corrupt PCM instead. .refine(isValidAudioBase64, 'Audio chunk must be base64'), - sampleRate: z.number().finite().positive() + sampleRate: z.number().finite().positive(), + // Counts up from 0 per dictation, so a chunk resent after a drop is fed once. + seq: z.number().int().nonnegative().optional() }) export const DictationHandle = z.object({