Fix long dictation retry backpressure

Treat Node IPC backpressure as a queue signal instead of a failed send, and pace buffered dictation replay so finish waits for the backlog to flush.

packages/app/src/dictation/dictation-stream-sender.test.ts

@@ -137,4 +137,30 @@ describe("DictationStreamSender", () => {
 
     expect(client.chunks.map((c) => c.seq)).toEqual([0, 1]);
   });
+
+  it("does not replay long buffered native dictation in one synchronous burst", async () => {
+    const client = new FakeDaemonClient();
+    client.isConnected = false;
+    const sender = new DictationStreamSender({
+      client: client as unknown as DaemonClient,
+      format: "audio/pcm;rate=16000;bits=16",
+      createDictationId: () => "d1",
+    });
+
+    for (let seq = 0; seq < 480; seq += 1) {
+      sender.enqueueSegment(`native-frame-${seq}`);
+    }
+
+    client.isConnected = true;
+    const finish = sender.finish(sender.getFinalSeq());
+
+    await tick();
+
+    expect(client.chunks.length).toBeLessThanOrEqual(128);
+    await expect(finish).resolves.toEqual({
+      dictationId: "d1",
+      text: "ok",
+    });
+    expect(client.finishes).toEqual([{ dictationId: "d1", finalSeq: 479 }]);
+  });
 });

packages/app/src/dictation/dictation-stream-sender.ts

@@ -1,6 +1,10 @@
 import { generateMessageId } from "@/types/stream";
 import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
 
+const MAX_CHUNKS_PER_FLUSH_TURN = 128;
+
+const waitForNextFlushTurn = (): Promise<void> => new Promise((resolve) => setTimeout(resolve, 0));
+
 export interface DictationStreamSenderParams {
   client: DaemonClient | null;
   format: string;
@@ -33,6 +37,8 @@ export class DictationStreamSender {
   private sendSeq = 0;
   private segments: string[] = [];
   private streamReady = false;
+  private flushTimer: ReturnType<typeof setTimeout> | null = null;
+  private drainWaiters: Array<() => void> = [];
 
   private startGeneration = 0;
   private startPromise: Promise<void> | null = null;
@@ -64,6 +70,7 @@ export class DictationStreamSender {
   }
 
   clearAll(): void {
+    this.clearScheduledFlush();
     this.dictationId = null;
     this.sendSeq = 0;
     this.segments = [];
@@ -73,6 +80,7 @@ export class DictationStreamSender {
   }
 
   resetStreamForReplay(): void {
+    this.clearScheduledFlush();
     this.dictationId = null;
     this.sendSeq = 0;
     this.streamReady = false;
@@ -108,13 +116,18 @@ export class DictationStreamSender {
     }
 
     let sent = 0;
-    while (this.sendSeq < this.segments.length) {
+    while (this.sendSeq < this.segments.length && sent < MAX_CHUNKS_PER_FLUSH_TURN) {
       const seq = this.sendSeq;
       const audio = this.segments[seq];
       client.sendDictationStreamChunk(dictationId, seq, audio, this.format);
       this.sendSeq = seq + 1;
       sent += 1;
     }
+    if (this.hasPendingSegments()) {
+      this.scheduleFlush();
+    } else {
+      this.resolveDrainWaiters();
+    }
     return sent;
   }
 
@@ -184,6 +197,7 @@ export class DictationStreamSender {
     }
 
     this.flush();
+    await this.waitForFlushDrain();
     return client.finishDictationStream(dictationId, finalSeq);
   }
 
@@ -195,4 +209,47 @@ export class DictationStreamSender {
     }
     this.resetStreamForReplay();
   }
+
+  private hasPendingSegments(): boolean {
+    return this.sendSeq < this.segments.length;
+  }
+
+  private scheduleFlush(): void {
+    if (this.flushTimer) {
+      return;
+    }
+    this.flushTimer = setTimeout(() => {
+      this.flushTimer = null;
+      this.flush();
+    }, 0);
+  }
+
+  private clearScheduledFlush(): void {
+    if (!this.flushTimer) {
+      return;
+    }
+    clearTimeout(this.flushTimer);
+    this.flushTimer = null;
+  }
+
+  private async waitForFlushDrain(): Promise<void> {
+    while (this.hasPendingSegments()) {
+      const client = this.client;
+      if (!client?.isConnected || !this.dictationId || !this.streamReady) {
+        throw new Error("Failed to flush dictation stream");
+      }
+      await new Promise<void>((resolve) => {
+        this.drainWaiters.push(resolve);
+      });
+      await waitForNextFlushTurn();
+    }
+  }
+
+  private resolveDrainWaiters(): void {
+    const waiters = this.drainWaiters;
+    this.drainWaiters = [];
+    for (const resolve of waiters) {
+      resolve();
+    }
+  }
 }

packages/server/src/server/speech/providers/local/test-fixtures/paused-ipc-worker.cjs

@@ -0,0 +1,23 @@
+process.on("message", (message) => {
+  if (!message || message.type !== "session.create") {
+    if (message && message.requestId && process.send) {
+      process.send({ type: "response", requestId: message.requestId, ok: true });
+    }
+    return;
+  }
+
+  process.send(
+    {
+      type: "response",
+      requestId: message.requestId,
+      ok: true,
+      result: { requiredSampleRate: 16000 },
+    },
+    () => {
+      const lock = new Int32Array(new SharedArrayBuffer(4));
+      Atomics.wait(lock, 0, 0, 500);
+    },
+  );
+});
+
+setInterval(() => undefined, 1000);

packages/server/src/server/speech/providers/local/worker-client.test.ts

@@ -1,5 +1,7 @@
 import { EventEmitter } from "node:events";
 import { once } from "node:events";
+import { fork, type ChildProcess } from "node:child_process";
+import { fileURLToPath } from "node:url";
 import pino from "pino";
 import { describe, expect, it } from "vitest";
 
@@ -54,6 +56,50 @@ class FakeLocalSpeechWorker extends EventEmitter {
   }
 }
 
+class PausedIpcWorker {
+  private readonly child: ChildProcess;
+
+  constructor() {
+    this.child = fork(
+      fileURLToPath(new URL("./test-fixtures/paused-ipc-worker.cjs", import.meta.url)),
+      [],
+      { serialization: "advanced", stdio: ["ignore", "ignore", "ignore", "ipc"] },
+    );
+  }
+
+  get connected(): boolean {
+    return this.child.connected;
+  }
+
+  get killed(): boolean {
+    return this.child.killed;
+  }
+
+  send(message: LocalSpeechWorkerRequest, callback: (error: Error | null) => void): boolean {
+    return this.child.send(message, (error) => callback(error ?? null));
+  }
+
+  disconnect(): void {
+    this.child.disconnect();
+  }
+
+  kill(): boolean {
+    return this.child.kill();
+  }
+
+  on(event: "message", listener: (message: LocalSpeechWorkerToParentMessage) => void): this;
+  on(event: "exit", listener: (code: number | null, signal: NodeJS.Signals | null) => void): this;
+  on(
+    event: "message" | "exit",
+    listener:
+      | ((message: LocalSpeechWorkerToParentMessage) => void)
+      | ((code: number | null, signal: NodeJS.Signals | null) => void),
+  ): this {
+    this.child.on(event, listener as (...args: unknown[]) => void);
+    return this;
+  }
+}
+
 function createClient(options?: { idleTtlMs?: number }) {
   const workers: FakeLocalSpeechWorker[] = [];
   const client = new LocalSpeechWorkerClient({
@@ -168,6 +214,49 @@ describe("LocalSpeechWorkerClient", () => {
     ]);
   });
 
+  it("does not surface real IPC backpressure when replaying native-sized dictation frames", async () => {
+    const workers: PausedIpcWorker[] = [];
+    const client = new LocalSpeechWorkerClient({
+      config: {
+        modelsDir: "/tmp/models",
+        voiceSttModel: "parakeet-tdt-0.6b-v2-int8",
+        dictationSttModel: "parakeet-tdt-0.6b-v2-int8",
+        voiceTtsModel: "kokoro-en-v0_19",
+      },
+      requestTimeoutMs: 30_000,
+      idleTtlMs: 30_000,
+      forkWorker: () => {
+        const worker = new PausedIpcWorker();
+        workers.push(worker);
+        return worker;
+      },
+    });
+    const provider = new WorkerBackedSpeechToTextProvider(client, "dictationStt");
+    const session = provider.createSession({ logger: pino({ level: "silent" }) });
+    let observedError: Error | null = null;
+    (session as EventEmitter).on("error", (error: Error) => {
+      observedError = error;
+    });
+
+    try {
+      await session.connect();
+      const nativeFrame = Buffer.alloc(1024, 1);
+
+      for (let seq = 0; seq < 480; seq += 1) {
+        session.appendPcm16(nativeFrame);
+      }
+      session.commit();
+      await waitForMicrotasks();
+
+      expect(observedError?.message).not.toBe("Local speech worker IPC channel is not writable");
+    } finally {
+      client.shutdown();
+      for (const worker of workers) {
+        worker.kill();
+      }
+    }
+  });
+
   it("forwards VAD session events through the shared worker", async () => {
     const { client, workers } = createClient();
     const provider = new WorkerBackedTurnDetectionProvider(client);

packages/server/src/server/speech/providers/local/worker-client.ts

@@ -248,7 +248,7 @@ export class LocalSpeechWorkerClient {
         timeout,
       });
 
-      const sent = worker.send(message, (error) => {
+      worker.send(message, (error) => {
         if (!error) {
           return;
         }
@@ -262,16 +262,6 @@ export class LocalSpeechWorkerClient {
         this.scheduleIdleShutdownIfReady();
         pending.reject(error);
       });
-      if (!sent) {
-        const pending = this.pendingRequests.get(requestId);
-        if (pending) {
-          clearTimeout(pending.timeout);
-          this.pendingRequests.delete(requestId);
-          this.inFlightRequests = Math.max(0, this.inFlightRequests - 1);
-          this.scheduleIdleShutdownIfReady();
-          pending.reject(new Error("Local speech worker IPC channel is not writable"));
-        }
-      }
     });
   }