fix(dictation): drain queued web capture before finish

.github/workflows/ci.yml

@@ -123,9 +123,6 @@ jobs:
       - name: Build server stack
         run: npm run build:server
 
-      - name: Build native app dependency
-        run: npm run build --workspace=@getpaseo/expo-two-way-audio
-
       - name: Typecheck all packages
         run: npm run typecheck
 

packages/app/src/hooks/use-dictation-audio-source.web.test.tsx

@@ -0,0 +1,171 @@
+// @vitest-environment jsdom
+
+import { Blob as NodeBlob } from "node:buffer";
+import { act, renderHook } from "@testing-library/react";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { useDictationAudioSource } from "./use-dictation-audio-source.web";
+
+function pcm16Wav(samples: Int16Array, sampleRate = 16000): Blob {
+  const bytes = new Uint8Array(samples.buffer, samples.byteOffset, samples.byteLength);
+  const wav = new ArrayBuffer(44 + bytes.byteLength);
+  const view = new DataView(wav);
+  const header = new TextEncoder();
+  new Uint8Array(wav, 0, 4).set(header.encode("RIFF"));
+  view.setUint32(4, 36 + bytes.byteLength, true);
+  new Uint8Array(wav, 8, 4).set(header.encode("WAVE"));
+  new Uint8Array(wav, 12, 4).set(header.encode("fmt "));
+  view.setUint32(16, 16, true);
+  view.setUint16(20, 1, true);
+  view.setUint16(22, 1, true);
+  view.setUint32(24, sampleRate, true);
+  view.setUint32(28, sampleRate * 2, true);
+  view.setUint16(32, 2, true);
+  view.setUint16(34, 16, true);
+  new Uint8Array(wav, 36, 4).set(header.encode("data"));
+  view.setUint32(40, bytes.byteLength, true);
+  new Uint8Array(wav, 44).set(bytes);
+  const blob = new Blob([wav], { type: "audio/wav" });
+  Object.defineProperty(blob, "arrayBuffer", {
+    configurable: true,
+    value: async () => wav,
+  });
+  return blob;
+}
+
+describe("web dictation audio source", () => {
+  const originalAudioContext = globalThis.AudioContext;
+  const originalBlob = globalThis.Blob;
+  const originalMediaRecorder = globalThis.MediaRecorder;
+  const originalMediaDevices = navigator.mediaDevices;
+
+  afterEach(() => {
+    Object.defineProperty(globalThis, "AudioContext", {
+      configurable: true,
+      value: originalAudioContext,
+    });
+    Object.defineProperty(globalThis, "Blob", {
+      configurable: true,
+      value: originalBlob,
+    });
+    Object.defineProperty(globalThis, "MediaRecorder", {
+      configurable: true,
+      value: originalMediaRecorder,
+    });
+    Object.defineProperty(navigator, "mediaDevices", {
+      configurable: true,
+      value: originalMediaDevices,
+    });
+  });
+
+  it("delivers PCM still queued behind the web audio callback when capture stops", async () => {
+    Object.defineProperty(globalThis, "Blob", {
+      configurable: true,
+      value: NodeBlob,
+    });
+    const completeRecording = new Int16Array(32000);
+    completeRecording.fill(1000);
+    const completeWav = pcm16Wav(completeRecording);
+    const track = { stop: vi.fn() };
+    const stream = { getTracks: () => [track] } as unknown as MediaStream;
+    const queuedSecond = new Float32Array(16000);
+    queuedSecond.fill(1000 / 32768);
+
+    let processor: ScriptProcessorNode | null = null;
+    const deliverQueuedSecond = () => {
+      processor?.onaudioprocess?.({
+        inputBuffer: { getChannelData: () => queuedSecond },
+      } as unknown as AudioProcessingEvent);
+    };
+    class FakeAudioContext {
+      sampleRate = 16000;
+      destination = {} as AudioDestinationNode;
+
+      createMediaStreamSource() {
+        return { connect: vi.fn(), disconnect: vi.fn() } as unknown as MediaStreamAudioSourceNode;
+      }
+
+      createScriptProcessor() {
+        processor = {
+          onaudioprocess: null,
+          connect: vi.fn(),
+          disconnect: vi.fn(),
+        } as unknown as ScriptProcessorNode;
+        return processor;
+      }
+
+      createGain() {
+        return {
+          gain: { value: 1 },
+          connect: vi.fn(),
+          disconnect: vi.fn(),
+        } as unknown as GainNode;
+      }
+
+      close() {
+        return Promise.resolve();
+      }
+    }
+
+    class FakeMediaRecorder extends EventTarget {
+      state: RecordingState = "inactive";
+      mimeType = "audio/wav";
+      ondataavailable: ((event: BlobEvent) => void) | null = null;
+
+      start() {
+        this.state = "recording";
+      }
+
+      stop() {
+        this.state = "inactive";
+        queueMicrotask(() => {
+          this.ondataavailable?.({ data: completeWav } as BlobEvent);
+          this.dispatchEvent(new Event("stop"));
+        });
+      }
+    }
+
+    Object.defineProperty(globalThis, "AudioContext", {
+      configurable: true,
+      value: FakeAudioContext,
+    });
+    Object.defineProperty(globalThis, "MediaRecorder", {
+      configurable: true,
+      value: FakeMediaRecorder,
+    });
+    Object.defineProperty(navigator, "mediaDevices", {
+      configurable: true,
+      value: { getUserMedia: vi.fn(async () => stream) },
+    });
+
+    const segments: string[] = [];
+    const { result } = renderHook(() =>
+      useDictationAudioSource({ onPcmSegment: (segment) => segments.push(segment) }),
+    );
+
+    await act(async () => result.current.start());
+    expect(processor).not.toBeNull();
+
+    const firstSecond = new Float32Array(16000);
+    firstSecond.fill(1000 / 32768);
+    act(() => {
+      processor?.onaudioprocess?.({
+        inputBuffer: { getChannelData: () => firstSecond },
+      } as unknown as AudioProcessingEvent);
+    });
+
+    let didStop = false;
+    const stopped = result.current.stop().then(() => {
+      didStop = true;
+      return undefined;
+    });
+    await new Promise<void>((resolve) => setTimeout(resolve, 0));
+    expect(didStop).toBe(false);
+
+    act(deliverQueuedSecond);
+    await act(async () => stopped);
+
+    const deliveredBytes = segments.reduce((total, segment) => total + atob(segment).length, 0);
+    expect(deliveredBytes).toBe(completeRecording.byteLength);
+  });
+});

packages/app/src/hooks/use-dictation-audio-source.web.ts

@@ -90,6 +90,16 @@ interface RecorderRefs {
   stoppedReject: ((error: unknown) => void) | null;
 }
 
+function emptyRecorderRefs(): RecorderRefs {
+  return {
+    recorder: null,
+    audioChunks: [],
+    stoppedPromise: null,
+    stoppedResolve: null,
+    stoppedReject: null,
+  };
+}
+
 function safeDisconnectNode(node: AudioNode | null): void {
   if (!node) return;
   try {
@@ -137,34 +147,33 @@ function stopMediaRecorderIfActive(recorder: MediaRecorder | null): void {
   }
 }
 
-async function finalizeRecorderStoppedPromise(input: {
+async function decodeStoppedRecorder(input: {
   stoppedPromise: Promise<Blob>;
   context: AudioContext | null;
   decodeAudioData: (context: AudioContext, arrayBuffer: ArrayBuffer) => Promise<AudioBuffer>;
-  emitPcmSegments: (pcm: Int16Array) => void;
   onError: ((err: Error) => void) | undefined;
-}): Promise<void> {
-  const { stoppedPromise, context, decodeAudioData, emitPcmSegments, onError } = input;
+}): Promise<Int16Array> {
+  const { stoppedPromise, context, decodeAudioData, onError } = input;
   try {
     const blob = await stoppedPromise;
     const arrayBuffer = await blob.arrayBuffer();
     if (arrayBuffer.byteLength === 0) {
-      return;
+      return new Int16Array(0);
     }
     const parsedWav = parsePcm16Wav(arrayBuffer);
     if (parsedWav) {
       const floatPcm = int16ToFloat32(parsedWav.samples);
-      emitPcmSegments(resampleToPcm16(floatPcm, parsedWav.sampleRate, 16000));
-      return;
+      return resampleToPcm16(floatPcm, parsedWav.sampleRate, 16000);
     }
     if (context) {
       const decoded = await decodeAudioData(context, arrayBuffer);
       const floatPcm = decoded.getChannelData(0);
-      emitPcmSegments(resampleToPcm16(floatPcm, decoded.sampleRate, 16000));
+      return resampleToPcm16(floatPcm, decoded.sampleRate, 16000);
     }
   } catch (err) {
     onError?.(err instanceof Error ? err : new Error(String(err)));
   }
+  return new Int16Array(0);
 }
 
 export function useDictationAudioSource(config: DictationAudioSourceConfig): DictationAudioSource {
@@ -185,8 +194,9 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
     processor: ScriptProcessorNode | null;
     gain: GainNode | null;
     pending: Int16Array;
+    emittedSamples: number;
+    pcmWaiters: Array<() => void>;
     started: boolean;
-    mode: "pcm" | "recorder";
     recorder: RecorderRefs;
   }>({
     stream: null,
@@ -195,15 +205,10 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
     processor: null,
     gain: null,
     pending: new Int16Array(0),
+    emittedSamples: 0,
+    pcmWaiters: [],
     started: false,
-    mode: "pcm",
-    recorder: {
-      recorder: null,
-      audioChunks: [],
-      stoppedPromise: null,
-      stoppedResolve: null,
-      stoppedReject: null,
-    },
+    recorder: emptyRecorderRefs(),
   });
 
   const decodeAudioData = useCallback(
@@ -233,6 +238,27 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
     }
   }, []);
 
+  const emitPcm = useCallback(
+    (pcm: Int16Array) => {
+      if (pcm.length === 0) return;
+      emitPcmSegments(pcm);
+      refs.current.emittedSamples += pcm.length;
+    },
+    [emitPcmSegments],
+  );
+
+  const resolvePcmWaiters = useCallback(() => {
+    const waiters = refs.current.pcmWaiters;
+    refs.current.pcmWaiters = [];
+    for (const resolve of waiters) resolve();
+  }, []);
+
+  const waitForPcmSamples = useCallback(async (targetSamples: number) => {
+    while (refs.current.emittedSamples + refs.current.pending.length < targetSamples) {
+      await new Promise<void>((resolve) => refs.current.pcmWaiters.push(resolve));
+    }
+  }, []);
+
   const start = useCallback(async () => {
     const missingNavigator =
       typeof navigator === "undefined" ||
@@ -280,6 +306,55 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
 
     const context = new AudioContextCtor();
 
+    let recorderRefs = emptyRecorderRefs();
+    const RecorderCtor =
+      typeof window !== "undefined"
+        ? (window as Window & { MediaRecorder?: typeof MediaRecorder }).MediaRecorder
+        : undefined;
+    if (RecorderCtor) {
+      try {
+        const preferredMimeType = "audio/webm;codecs=opus";
+        const options =
+          typeof RecorderCtor.isTypeSupported !== "function" ||
+          RecorderCtor.isTypeSupported(preferredMimeType)
+            ? { mimeType: preferredMimeType }
+            : undefined;
+        const recorder = new RecorderCtor(stream, options);
+        recorderRefs = {
+          recorder,
+          audioChunks: [],
+          stoppedPromise: null,
+          stoppedResolve: null,
+          stoppedReject: null,
+        };
+        recorder.ondataavailable = (event: BlobEvent) => {
+          const data: Blob | undefined = event?.data;
+          if (data) recorderRefs.audioChunks.push(data);
+        };
+        recorder.addEventListener("error", (event: Event) => {
+          recorderRefs.stoppedReject?.(event);
+        });
+        recorder.addEventListener("stop", () => {
+          try {
+            const blob =
+              recorderRefs.audioChunks.length > 0
+                ? new Blob(recorderRefs.audioChunks, { type: recorder.mimeType })
+                : new Blob([], { type: recorder.mimeType });
+            recorderRefs.stoppedResolve?.(blob);
+          } catch (err) {
+            recorderRefs.stoppedReject?.(err);
+          }
+        });
+        recorderRefs.stoppedPromise = new Promise<Blob>((resolve, reject) => {
+          recorderRefs.stoppedResolve = resolve;
+          recorderRefs.stoppedReject = reject;
+        });
+        recorder.start();
+      } catch {
+        recorderRefs = emptyRecorderRefs();
+      }
+    }
+
     try {
       const source = context.createMediaStreamSource(stream);
       const processor = context.createScriptProcessor(4096, 1, 1);
@@ -290,7 +365,6 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
       const chunkSamples = outputRate; // ~1s
 
       refs.current.started = true;
-      refs.current.mode = "pcm";
 
       processor.onaudioprocess = (event) => {
         if (!refs.current.started) {
@@ -313,8 +387,9 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
         while (refs.current.pending.length >= chunkSamples) {
           const chunk = refs.current.pending.slice(0, chunkSamples);
           refs.current.pending = refs.current.pending.slice(chunkSamples);
-          onPcmSegmentRef.current(int16ToBase64(chunk));
+          emitPcm(chunk);
         }
+        resolvePcmWaiters();
       };
 
       source.connect(processor);
@@ -329,7 +404,10 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
         processor,
         gain,
         pending: new Int16Array(0),
+        emittedSamples: 0,
+        pcmWaiters: [],
         started: true,
+        recorder: recorderRefs,
       };
       return;
     } catch {
@@ -337,58 +415,10 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
       // isn't available (e.g., Playwright tests with a stubbed getUserMedia).
     }
 
-    const RecorderCtor =
-      typeof window !== "undefined"
-        ? (window as Window & { MediaRecorder?: typeof MediaRecorder }).MediaRecorder
-        : undefined;
-    if (!RecorderCtor) {
+    if (!recorderRefs.recorder) {
       throw new Error("MediaRecorder unavailable");
     }
 
-    const recorder = new RecorderCtor(stream, {
-      mimeType: "audio/webm;codecs=opus",
-    } as MediaRecorderOptions);
-
-    const recorderRefs: RecorderRefs = {
-      recorder,
-      audioChunks: [],
-      stoppedPromise: null,
-      stoppedResolve: null,
-      stoppedReject: null,
-    };
-
-    recorder.ondataavailable = (event: BlobEvent) => {
-      const data: Blob | undefined = event?.data;
-      if (data) {
-        recorderRefs.audioChunks.push(data);
-      }
-    };
-    recorder.addEventListener("error", (event: Event) => {
-      recorderRefs.stoppedReject?.(event);
-    });
-    recorder.addEventListener("stop", () => {
-      try {
-        const blob =
-          recorderRefs.audioChunks.length > 0
-            ? new Blob(recorderRefs.audioChunks, { type: recorder.mimeType })
-            : new Blob([], { type: recorder.mimeType });
-        recorderRefs.stoppedResolve?.(blob);
-      } catch (err) {
-        recorderRefs.stoppedReject?.(err);
-      }
-    });
-
-    recorderRefs.stoppedPromise = new Promise<Blob>((resolve, reject) => {
-      recorderRefs.stoppedResolve = resolve;
-      recorderRefs.stoppedReject = reject;
-    });
-
-    try {
-      recorder.start();
-    } catch (err) {
-      throw err instanceof Error ? err : new Error(String(err));
-    }
-
     refs.current = {
       ...refs.current,
       stream,
@@ -397,33 +427,38 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
       processor: null,
       gain: null,
       pending: new Int16Array(0),
+      emittedSamples: 0,
+      pcmWaiters: [],
       started: true,
-      mode: "recorder",
       recorder: recorderRefs,
     };
-  }, []);
+  }, [emitPcm, resolvePcmWaiters]);
 
   const stop = useCallback(async () => {
-    refs.current.started = false;
     setVolume(0);
 
-    const { processor, source, gain, context, stream, pending, mode, recorder } = refs.current;
+    const { processor, source, gain, context, stream, recorder } = refs.current;
 
-    disconnectDictationAudioGraph({ processor, source, gain, stream });
+    stopMediaRecorderIfActive(recorder.recorder);
 
-    if (mode === "recorder") {
-      stopMediaRecorderIfActive(recorder.recorder);
-      if (recorder.stoppedPromise) {
-        await finalizeRecorderStoppedPromise({
-          stoppedPromise: recorder.stoppedPromise,
-          context,
-          decodeAudioData,
-          emitPcmSegments,
-          onError: onErrorRef.current,
-        });
-      }
+    let recordedPcm: Int16Array<ArrayBufferLike> = new Int16Array(0);
+    if (recorder.stoppedPromise) {
+      recordedPcm = await decodeStoppedRecorder({
+        stoppedPromise: recorder.stoppedPromise,
+        context,
+        decodeAudioData,
+        onError: onErrorRef.current,
+      });
     }
 
+    if (processor && recordedPcm.length > 0) {
+      await waitForPcmSamples(recordedPcm.length);
+    }
+
+    refs.current.started = false;
+    resolvePcmWaiters();
+    disconnectDictationAudioGraph({ processor, source, gain, stream });
+
     if (context) {
       try {
         await context.close();
@@ -432,8 +467,12 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
       }
     }
 
-    if (pending.length > 0) {
-      onPcmSegmentRef.current(int16ToBase64(pending));
+    if (processor) {
+      if (refs.current.pending.length > 0) {
+        emitPcm(refs.current.pending);
+      }
+    } else {
+      emitPcm(recordedPcm);
     }
 
     refs.current = {
@@ -443,17 +482,12 @@ export function useDictationAudioSource(config: DictationAudioSourceConfig): Dic
       processor: null,
       gain: null,
       pending: new Int16Array(0),
+      emittedSamples: 0,
+      pcmWaiters: [],
       started: false,
-      mode: "pcm",
-      recorder: {
-        recorder: null,
-        audioChunks: [],
-        stoppedPromise: null,
-        stoppedResolve: null,
-        stoppedReject: null,
-      },
+      recorder: emptyRecorderRefs(),
     };
-  }, [decodeAudioData, emitPcmSegments]);
+  }, [decodeAudioData, emitPcm, resolvePcmWaiters, waitForPcmSamples]);
 
   useEffect(() => {
     return () => {

packages/app/src/voice/audio-engine.native.test.ts

@@ -38,8 +38,6 @@ const nativeAudio = vi.hoisted(() => {
   };
 });
 
-vi.mock("@getpaseo/expo-two-way-audio", () => nativeAudio);
-
 import { createAudioEngine } from "./audio-engine.native";
 
 function emitMicrophoneData(data: Uint8Array): void {
@@ -59,10 +57,13 @@ describe("native audio capture", () => {
 
   it("drains microphone events queued while capture stops", async () => {
     const captured: Uint8Array[] = [];
-    const engine = createAudioEngine({
-      onCaptureData: (pcm) => captured.push(pcm),
-      onVolumeLevel: vi.fn(),
-    });
+    const engine = createAudioEngine(
+      {
+        onCaptureData: (pcm) => captured.push(pcm),
+        onVolumeLevel: vi.fn(),
+      },
+      { nativeModule: nativeAudio },
+    );
 
     await engine.startCapture();
     let didStop = false;
@@ -88,10 +89,13 @@ describe("native audio capture", () => {
   });
 
   it("waits for capture to stop before destroying the native engine", async () => {
-    const engine = createAudioEngine({
-      onCaptureData: vi.fn(),
-      onVolumeLevel: vi.fn(),
-    });
+    const engine = createAudioEngine(
+      {
+        onCaptureData: vi.fn(),
+        onVolumeLevel: vi.fn(),
+      },
+      { nativeModule: nativeAudio },
+    );
 
     await engine.startCapture();
     const destroyed = engine.destroy();
@@ -113,10 +117,13 @@ describe("native audio capture", () => {
           resolvePermission = resolve;
         }),
     );
-    const engine = createAudioEngine({
-      onCaptureData: vi.fn(),
-      onVolumeLevel: vi.fn(),
-    });
+    const engine = createAudioEngine(
+      {
+        onCaptureData: vi.fn(),
+        onVolumeLevel: vi.fn(),
+      },
+      { nativeModule: nativeAudio },
+    );
 
     const started = engine.startCapture();
     await Promise.resolve();
@@ -131,10 +138,13 @@ describe("native audio capture", () => {
   });
 
   it("makes concurrent destruction wait for the same capture drain", async () => {
-    const engine = createAudioEngine({
-      onCaptureData: vi.fn(),
-      onVolumeLevel: vi.fn(),
-    });
+    const engine = createAudioEngine(
+      {
+        onCaptureData: vi.fn(),
+        onVolumeLevel: vi.fn(),
+      },
+      { nativeModule: nativeAudio },
+    );
 
     await engine.startCapture();
     let firstFinished = false;
@@ -160,10 +170,13 @@ describe("native audio capture", () => {
   });
 
   it("does not restart capture when destruction overtakes a pending stop", async () => {
-    const engine = createAudioEngine({
-      onCaptureData: vi.fn(),
-      onVolumeLevel: vi.fn(),
-    });
+    const engine = createAudioEngine(
+      {
+        onCaptureData: vi.fn(),
+        onVolumeLevel: vi.fn(),
+      },
+      { nativeModule: nativeAudio },
+    );
 
     await engine.startCapture();
     const stopped = engine.stopCapture();

packages/app/src/voice/audio-engine.native.ts

@@ -1,4 +1,3 @@
-import * as native from "@getpaseo/expo-two-way-audio";
 import type {
   AudioEngine,
   AudioEngineCallbacks,
@@ -13,6 +12,24 @@ interface QueuedAudio {
 
 interface AudioEngineTraceOptions {
   traceLabel?: string;
+  nativeModule?: object;
+}
+
+interface NativeAudioModule {
+  addExpoTwoWayAudioEventListener(
+    eventName: string,
+    handler: (event: { data: never }) => void,
+  ): { remove(): void };
+  getMicrophonePermissionsAsync(): Promise<{ granted: boolean }>;
+  requestMicrophonePermissionsAsync(): Promise<{ granted: boolean }>;
+  initialize(): Promise<boolean>;
+  toggleRecording(active: boolean): boolean;
+  stopRecording(): Promise<void>;
+  releaseAudioSession(): void;
+  resumePlayback(): void;
+  playPCMData(data: Uint8Array): void;
+  tearDown(): void;
+  stopPlayback(): void;
 }
 
 function parsePcmSampleRate(mimeType: string): number | null {
@@ -71,6 +88,9 @@ export function createAudioEngine(
   callbacks: AudioEngineCallbacks,
   _options?: AudioEngineTraceOptions,
 ): AudioEngine {
+  const native = (_options?.nativeModule ??
+    require("@getpaseo/expo-two-way-audio")) as NativeAudioModule;
+
   const refs: {
     initialized: boolean;
     captureActive: boolean;