fix(voice): prevent choppy audio in realtime calls (#125620)

* fix(voice-call): pace realtime audio from stream clock

* fix(voice): preserve resampler state across audio chunks

* fix(voice-call): honor telephony TTS audio formats

* fix(voice-call): await acknowledged stream playback

* fix(voice): bound realtime input audio backlog

* chore(voice-call): shrink assertion baseline and localize tts format error

* test(voice-call): split playback-mark coverage into its own suite

* test(voice-call): drop helper left unused by suite split

* test(openai): scope queued audio copy assertion

* fix(openai): flush realtime resampler at response end

* fix(voice-call): reject containerized mulaw TTS

config/assertion-safety-baseline.txt

@@ -1389,7 +1389,7 @@ extensions/voice-call/src/media-stream.ts	1
 extensions/voice-call/src/providers/mock.ts	7
 extensions/voice-call/src/providers/shared/guarded-json-api.ts	2
 extensions/voice-call/src/providers/shared/response-body.ts	1
-extensions/voice-call/src/providers/twilio.ts	3
+extensions/voice-call/src/providers/twilio.ts	2
 extensions/voice-call/src/providers/twilio/api.ts	3
 extensions/voice-call/src/realtime-agent-context.ts	1
 extensions/voice-call/src/response-generator.ts	3

extensions/openai/realtime-audio-buffer-ownership.test.ts

@@ -117,7 +117,7 @@ describe("OpenAI realtime queued audio buffer ownership", () => {
         const copyBuffer = vi.spyOn(Buffer, "from");
         try {
           bridge.sendAudio(oversized);
-          expect(copyBuffer).not.toHaveBeenCalled();
+          expect(copyBuffer.mock.calls.some(([source]) => source === oversized)).toBe(false);
         } finally {
           copyBuffer.mockRestore();
         }

extensions/openai/realtime-quicksilver-bridge.test.ts

@@ -85,6 +85,7 @@ function createHarness(params?: {
   const onError = vi.fn();
   const onClose = vi.fn();
   const onEvent = vi.fn();
+  const logger = { warn: vi.fn() };
   const bridge = new OpenAIQuicksilverVoiceBridge({
     providerConfig: {},
     model: "gpt-live-1-codex",
@@ -104,10 +105,12 @@ function createHarness(params?: {
     onError,
     onClose,
     onEvent,
+    logger,
   });
   return {
     bridge,
     connections,
+    logger,
     onAudio,
     onClose,
     onError,
@@ -205,7 +208,11 @@ describe("OpenAIQuicksilverVoiceBridge", () => {
     expect(audioEvents).toHaveLength(2);
     expect(
       audioEvents.map((event) => Buffer.from(String(event.audio), "base64").byteLength),
-    ).toEqual([512 * 1024, 512 * 1024]);
+    ).toEqual([512 * 1024, Buffer.byteLength("overflow")]);
+    expect(harness.logger.warn).toHaveBeenCalledOnce();
+    expect(harness.logger.warn).toHaveBeenCalledWith(
+      "OpenAI GPT-Live input audio queue overflow; keeping newest audio",
+    );
     harness.bridge.close();
   });
 
@@ -447,13 +454,34 @@ describe("OpenAIQuicksilverVoiceBridge", () => {
 
     const inputEvent = sentEvents(harness.socket).at(-1);
     expect(inputEvent?.type).toBe("input_audio.append");
-    expect(Buffer.from(String(inputEvent?.audio), "base64")).toHaveLength(960);
+    expect(Buffer.from(String(inputEvent?.audio), "base64")).toHaveLength(870);
 
     harness.socket.serverEvent({
       type: "output_audio.delta",
       audio: Buffer.alloc(960).toString("base64"),
     });
-    expect(harness.onAudio).toHaveBeenCalledWith(Buffer.alloc(160, 0xff));
+    harness.socket.serverEvent({
+      type: "turn.done",
+      turn: { role: "assistant", transcript: "first response" },
+    });
+    harness.socket.serverEvent({
+      type: "output_audio.delta",
+      audio: Buffer.alloc(960).toString("base64"),
+    });
+    harness.socket.serverEvent({
+      type: "turn.done",
+      turn: { role: "assistant", transcript: "second response" },
+    });
+
+    expect(harness.onAudio.mock.calls.map(([audio]) => audio)).toEqual([
+      Buffer.alloc(155, 0xff),
+      Buffer.alloc(5, 0xff),
+      Buffer.alloc(155, 0xff),
+      Buffer.alloc(5, 0xff),
+    ]);
+    expect(harness.onAudio.mock.invocationCallOrder.at(-1)).toBeLessThan(
+      harness.onEvent.mock.invocationCallOrder.at(-1) ?? 0,
+    );
   });
 
   it("uses session context for forced consult results without a provider delegation", async () => {

extensions/openai/realtime-quicksilver-bridge.ts

@@ -1,17 +1,18 @@
 // GPT-Live backend bridge over the Frameless Bidi WebSocket protocol used by Codex realtime v3.
 import { randomUUID } from "node:crypto";
 import { canonicalizeBase64 } from "openclaw/plugin-sdk/media-runtime";
+import type { PluginLogger } from "openclaw/plugin-sdk/plugin-entry";
 import {
   captureWsEvent,
   createDebugProxyWebSocketAgent,
   resolveDebugProxySettings,
 } from "openclaw/plugin-sdk/proxy-capture";
 import {
-  convertPcmToMulaw8k,
+  createStreamingPcmResampler,
   mulawToPcm,
+  pcmToMulaw,
   REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
   RealtimeVoiceSessionLifecycle,
-  resamplePcm,
   type RealtimeVoiceBridge,
   type RealtimeVoiceBridgeCreateRequest,
   type RealtimeVoiceSessionConnection,
@@ -44,6 +45,7 @@ type OpenAIQuicksilverVoiceBridgeConfig = RealtimeVoiceBridgeCreateRequest & {
   model: string;
   voice?: string;
   resolveAuth: () => Promise<OpenAIQuicksilverAuth>;
+  logger?: Pick<PluginLogger, "warn">;
   webSocketFactory?: OpenAIQuicksilverSocketFactory;
 };
 
@@ -73,7 +75,15 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
   readonly handlesInputAudioBargeIn = false;
 
   private socket: OpenAIQuicksilverSocket | undefined;
-  private readonly lifecycle = new RealtimeVoiceSessionLifecycle("OpenAI");
+  private readonly lifecycle: RealtimeVoiceSessionLifecycle;
+  private inboundTelephonyResampler = createStreamingPcmResampler(
+    8_000,
+    OPENAI_QUICKSILVER_SAMPLE_RATE,
+  );
+  private outboundTelephonyResampler = createStreamingPcmResampler(
+    OPENAI_QUICKSILVER_SAMPLE_RATE,
+    8_000,
+  );
   private activeDelegations = new Set<string>();
   private readonly flowId = randomUUID();
   private readonly requestIds: OpenAIQuicksilverRequestIds = {
@@ -82,7 +92,15 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
     threadId: randomUUID(),
   };
 
-  constructor(private readonly config: OpenAIQuicksilverVoiceBridgeConfig) {}
+  constructor(private readonly config: OpenAIQuicksilverVoiceBridgeConfig) {
+    this.lifecycle = new RealtimeVoiceSessionLifecycle("OpenAI", {
+      pendingAudioOverflowPolicy: "drop-oldest",
+      onPendingAudioOverflow: () =>
+        (config.logger?.warn ?? console.warn)(
+          "OpenAI GPT-Live input audio queue overflow; keeping newest audio",
+        ),
+    });
+  }
 
   async connect(): Promise<void> {
     await this.lifecycle.connect((connection) => this.connectConnection(connection));
@@ -416,15 +434,31 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
         return;
       }
       const pcm = Buffer.from(canonical, "base64");
-      this.config.onAudio(
+      const output =
         this.config.audioFormat?.encoding === "g711_ulaw"
-          ? convertPcmToMulaw8k(pcm, OPENAI_QUICKSILVER_SAMPLE_RATE)
-          : pcm,
-      );
+          ? pcmToMulaw(this.outboundTelephonyResampler.process(pcm))
+          : pcm;
+      if (output.length > 0) {
+        this.config.onAudio(output);
+      }
       this.config.onEvent?.({ direction: "server", type: "output_audio.delta" });
       return;
     }
     if (event.kind === "transcript-delta" || event.kind === "transcript-done") {
+      if (
+        event.kind === "transcript-done" &&
+        event.role === "assistant" &&
+        this.config.audioFormat?.encoding === "g711_ulaw"
+      ) {
+        const tail = pcmToMulaw(this.outboundTelephonyResampler.flush());
+        this.outboundTelephonyResampler = createStreamingPcmResampler(
+          OPENAI_QUICKSILVER_SAMPLE_RATE,
+          8_000,
+        );
+        if (tail.length > 0) {
+          this.config.onAudio(tail);
+        }
+      }
       this.config.onTranscript?.(event.role, event.text, event.kind === "transcript-done");
       this.config.onEvent?.({
         direction: "server",
@@ -468,8 +502,11 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
   private sendAudioNow(audio: Buffer): void {
     const pcm =
       this.config.audioFormat?.encoding === "g711_ulaw"
-        ? resamplePcm(mulawToPcm(audio), 8_000, OPENAI_QUICKSILVER_SAMPLE_RATE)
+        ? this.inboundTelephonyResampler.process(mulawToPcm(audio))
         : audio;
+    if (pcm.length === 0) {
+      return;
+    }
     this.sendEvent({ type: "input_audio.append", audio: pcm.toString("base64") });
   }
 
@@ -528,6 +565,14 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge {
 
   private resetTerminalState(): void {
     this.activeDelegations.clear();
+    this.inboundTelephonyResampler = createStreamingPcmResampler(
+      8_000,
+      OPENAI_QUICKSILVER_SAMPLE_RATE,
+    );
+    this.outboundTelephonyResampler = createStreamingPcmResampler(
+      OPENAI_QUICKSILVER_SAMPLE_RATE,
+      8_000,
+    );
   }
 
   private closeSocket(reason: string, socket = this.socket): void {

extensions/openai/realtime-quicksilver-gateway-bridge.test.ts

@@ -249,7 +249,11 @@ describe("GPT-Live werift audio peer", () => {
 
       expect(decodeOrder).toEqual([20, "plc", 22, 23, 24, 25]);
       expect(decodePacketLoss).toHaveBeenCalledWith(960);
-      expect(Buffer.concat(onAudio.mock.calls.map(([audio]) => audio))).toHaveLength(6 * 480 * 2);
+      // The centered streaming filter retains seven 24 kHz samples of right-edge
+      // context until the next packet instead of fabricating a boundary per packet.
+      expect(Buffer.concat(onAudio.mock.calls.map(([audio]) => audio))).toHaveLength(
+        (6 * 480 - 7) * 2,
+      );
       expect(onError).not.toHaveBeenCalled();
     } finally {
       peer.close();

extensions/openai/realtime-quicksilver-peer.runtime.ts

@@ -1,7 +1,7 @@
 // Lazy GPT-Live media runtime: werift peer plus WASM Opus framing and PCM conversion.
 import { randomInt } from "node:crypto";
 import { toErrorObject } from "openclaw/plugin-sdk/error-runtime";
-import { resamplePcm } from "openclaw/plugin-sdk/realtime-voice";
+import { createStreamingPcmResampler, resamplePcm } from "openclaw/plugin-sdk/realtime-voice";
 import {
   OpenAIQuicksilverPendingAudio,
   OPENAI_QUICKSILVER_RELAY_FRAME_BYTES,
@@ -12,6 +12,9 @@ const RELAY_SAMPLE_RATE = 24_000;
 const QUICKSILVER_CHANNELS = 2;
 const OPUS_FRAME_SAMPLES = 960;
 const OPUS_FRAME_DURATION_MS = 20;
+// The centered 31-tap filter withholds 15 input samples. Prime the 2x path with
+// the matching 30-sample silence so every Opus tick still receives one full frame.
+const OUTBOUND_RESAMPLE_PREROLL_SAMPLES = 30;
 const INBOUND_REORDER_DEPTH = 4;
 // More than two seconds behind cannot be useful 20 ms reordering; fail instead of corrupting Opus state.
 const INBOUND_MAX_LATE_PACKETS = 100;
@@ -59,9 +62,11 @@ function pcmBufferToInt16(pcm: Buffer): Int16Array {
 }
 
 function convertRelayPcmToQuicksilverPcm(pcm24kMono: Buffer): Int16Array {
-  const mono48k = pcmBufferToInt16(
-    resamplePcm(pcm24kMono, RELAY_SAMPLE_RATE, QUICKSILVER_SAMPLE_RATE),
-  );
+  return duplicateMonoToStereo(resamplePcm(pcm24kMono, RELAY_SAMPLE_RATE, QUICKSILVER_SAMPLE_RATE));
+}
+
+function duplicateMonoToStereo(pcm48kMono: Buffer): Int16Array {
+  const mono48k = pcmBufferToInt16(pcm48kMono);
   const stereo48k = new Int16Array(mono48k.length * QUICKSILVER_CHANNELS);
   for (let index = 0; index < mono48k.length; index += 1) {
     const sample = mono48k[index] ?? 0;
@@ -167,6 +172,15 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
   private inboundRtpState: InboundRtpState = { pendingPackets: new Map() };
   private mediaTimer: ReturnType<typeof setInterval> | undefined;
   private pendingAudio = new OpenAIQuicksilverPendingAudio();
+  private pendingResampledAudio = Buffer.alloc(OUTBOUND_RESAMPLE_PREROLL_SAMPLES * 2);
+  private readonly inboundResampler = createStreamingPcmResampler(
+    QUICKSILVER_SAMPLE_RATE,
+    RELAY_SAMPLE_RATE,
+  );
+  private readonly outboundResampler = createStreamingPcmResampler(
+    RELAY_SAMPLE_RATE,
+    QUICKSILVER_SAMPLE_RATE,
+  );
   private sequenceNumber = randomInt(0x1_0000);
   private subscribedTracks = new Set<string>();
   private timestamp = randomInt(0x1_0000_0000);
@@ -248,6 +262,9 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
       this.mediaTimer = undefined;
     }
     this.pendingAudio.clear();
+    this.pendingResampledAudio = Buffer.alloc(0);
+    this.inboundResampler.flush();
+    this.outboundResampler.flush();
     this.resetInboundRtpState();
     this.state.encoder.free();
     this.state.decoder.free();
@@ -396,7 +413,14 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
   }
 
   private emitInboundPcm(decoded: Int16Array): void {
-    const relayPcm = convertQuicksilverPcmToRelayPcm(decoded);
+    const frameCount = Math.floor(decoded.length / QUICKSILVER_CHANNELS);
+    const mono48k = Buffer.alloc(frameCount * 2);
+    for (let frame = 0; frame < frameCount; frame += 1) {
+      const left = decoded[frame * 2] ?? 0;
+      const right = decoded[frame * 2 + 1] ?? 0;
+      mono48k.writeInt16LE(Math.round((left + right) / 2), frame * 2);
+    }
+    const relayPcm = this.inboundResampler.process(mono48k);
     if (relayPcm.length > 0) {
       this.state.callbacks.onAudio(relayPcm);
     }
@@ -418,7 +442,13 @@ export class OpenAIQuicksilverAudioPeer implements OpenAIQuicksilverAudioPeerCon
     }
     const frame = this.takeNextRelayFrame();
     try {
-      const opusPacket = this.state.encoder.encode(convertRelayPcmToQuicksilverPcm(frame), {
+      const resampled = this.outboundResampler.process(frame);
+      this.pendingResampledAudio = Buffer.concat([this.pendingResampledAudio, resampled]);
+      const monoFrameBytes = OPUS_FRAME_SAMPLES * 2;
+      const monoFrame = Buffer.alloc(monoFrameBytes);
+      this.pendingResampledAudio.copy(monoFrame, 0, 0, monoFrameBytes);
+      this.pendingResampledAudio = Buffer.from(this.pendingResampledAudio.subarray(monoFrameBytes));
+      const opusPacket = this.state.encoder.encode(duplicateMonoToStereo(monoFrame), {
         frameSize: OPUS_FRAME_SAMPLES,
       });
       const rtp = new this.state.werift.RtpPacket(

extensions/openai/realtime-voice-bridge-connection.test.ts

@@ -291,7 +291,70 @@ describe("OpenAI realtime voice bridge connection", () => {
     expect(audioEvents).toHaveLength(2);
     expect(
       audioEvents.map((event) => Buffer.from(String(event.audio), "base64").byteLength),
-    ).toEqual([512 * 1024, 512 * 1024]);
+    ).toEqual([512 * 1024, Buffer.byteLength("overflow")]);
+    bridge.close();
+  });
+
+  it("drops stalled input audio and rate-limits aggregate warnings", async () => {
+    vi.useFakeTimers();
+    try {
+      const logger = { debug: vi.fn(), warn: vi.fn() };
+      const provider = buildOpenAIRealtimeVoiceProvider({ logger });
+      const bridge = provider.createBridge({
+        providerConfig: { apiKey: "test-api-key-test" },
+        onAudio: vi.fn(),
+        onClearAudio: vi.fn(),
+      });
+      const { connecting, socket } = beginBridgeConnection(bridge);
+      openSocket(socket);
+      emitSessionUpdated(socket);
+      await connecting;
+      socket.bufferedAmount = 1024 * 1024 + 1;
+
+      bridge.sendAudio(Buffer.from("first"));
+      await vi.advanceTimersByTimeAsync(4_000);
+      bridge.sendAudio(Buffer.from("second"));
+      await vi.advanceTimersByTimeAsync(1_000);
+      bridge.sendAudio(Buffer.from("third"));
+
+      expect(
+        parseSent(socket).filter((event) => event.type === "input_audio_buffer.append"),
+      ).toHaveLength(0);
+      expect(logger.warn).toHaveBeenNthCalledWith(
+        1,
+        "OpenAI realtime input audio backpressure; droppedFrames=1",
+      );
+      expect(logger.warn).toHaveBeenNthCalledWith(
+        2,
+        "OpenAI realtime input audio backpressure; droppedFrames=2",
+      );
+      bridge.close();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("routes readiness-drained audio through websocket backpressure", async () => {
+    const logger = { debug: vi.fn(), warn: vi.fn() };
+    const provider = buildOpenAIRealtimeVoiceProvider({ logger });
+    const bridge = provider.createBridge({
+      providerConfig: { apiKey: "test-api-key-test" },
+      onAudio: vi.fn(),
+      onClearAudio: vi.fn(),
+    });
+    const { connecting, socket } = beginBridgeConnection(bridge);
+    openSocket(socket);
+    bridge.sendAudio(Buffer.from("queued-before-ready"));
+    socket.bufferedAmount = 1024 * 1024 + 1;
+    emitSessionUpdated(socket);
+    await connecting;
+
+    expect(
+      parseSent(socket).filter((event) => event.type === "input_audio_buffer.append"),
+    ).toHaveLength(0);
+    expect(logger.warn).toHaveBeenCalledWith(
+      "OpenAI realtime input audio backpressure; droppedFrames=1",
+    );
     bridge.close();
   });
 

extensions/openai/realtime-voice-bridge.ts

@@ -38,9 +38,13 @@ import {
   resolveOpenAIRealtimeEnvApiKey,
   resolveOpenAIRealtimeSecretInput,
   type OpenAIRealtimeUserMessageOptions,
+  type OpenAIRealtimeVoiceBridgeConfig,
   type RealtimeEvent,
 } from "./realtime-voice-session-policy.js";
 
+const OPENAI_REALTIME_MAX_BUFFERED_AUDIO_BYTES = 1024 * 1024;
+const OPENAI_REALTIME_AUDIO_DROP_WARN_INTERVAL_MS = 5_000;
+
 export class OpenAIRealtimeBridge extends OpenAIRealtimeEvents implements RealtimeVoiceBridge {
   private static readonly DEFAULT_MODEL = OPENAI_REALTIME_DEFAULT_MODEL;
 
@@ -52,7 +56,7 @@ export class OpenAIRealtimeBridge extends OpenAIRealtimeEvents implements Realti
 
   private ws: WebSocket | null = null;
 
-  private readonly lifecycle = new RealtimeVoiceSessionLifecycle("OpenAI");
+  private readonly lifecycle: RealtimeVoiceSessionLifecycle;
 
   private connectionUrl = "";
 
@@ -66,6 +70,19 @@ export class OpenAIRealtimeBridge extends OpenAIRealtimeEvents implements Realti
 
   private terminalError: Error | undefined;
 
+  private droppedInputAudioFrames = 0;
+
+  private lastInputAudioDropWarningAt = Number.NEGATIVE_INFINITY;
+
+  constructor(config: OpenAIRealtimeVoiceBridgeConfig) {
+    super(config);
+    this.lifecycle = new RealtimeVoiceSessionLifecycle("OpenAI", {
+      pendingAudioOverflowPolicy: "drop-oldest",
+      onPendingAudioOverflow: () =>
+        this.config.logger.warn("OpenAI realtime input audio queue overflow; keeping newest audio"),
+    });
+  }
+
   async connect(): Promise<void> {
     if (this.terminalError) {
       throw this.terminalError;
@@ -81,6 +98,18 @@ export class OpenAIRealtimeBridge extends OpenAIRealtimeEvents implements Realti
       this.lifecycle.enqueuePendingAudio(audio);
       return;
     }
+    if (this.ws.bufferedAmount > OPENAI_REALTIME_MAX_BUFFERED_AUDIO_BYTES) {
+      this.droppedInputAudioFrames += 1;
+      const now = Date.now();
+      if (now - this.lastInputAudioDropWarningAt >= OPENAI_REALTIME_AUDIO_DROP_WARN_INTERVAL_MS) {
+        this.config.logger.warn(
+          `OpenAI realtime input audio backpressure; droppedFrames=${this.droppedInputAudioFrames}`,
+        );
+        this.droppedInputAudioFrames = 0;
+        this.lastInputAudioDropWarningAt = now;
+      }
+      return;
+    }
     this.sendEvent({
       type: "input_audio_buffer.append",
       audio: audio.toString("base64"),

extensions/openai/realtime-voice-provider.ts

@@ -170,6 +170,7 @@ function buildOpenAIRealtimeBrowserSessionConfig(
 async function createOpenAIRealtimeBrowserSession(
   req: OpenAIInternalRealtimeBrowserSessionCreateRequest,
   quicksilverBroker: OpenAIQuicksilverBrowserSessionBroker | undefined,
+  logger: Pick<PluginLogger, "warn">,
 ): Promise<RealtimeVoiceBrowserSession> {
   const rawConfig = resolveOpenAIProviderConfigRecord(req.providerConfig);
   const config = normalizeProviderConfig(req.providerConfig);
@@ -242,6 +243,7 @@ async function createOpenAIRealtimeBrowserSession(
                 gatewayControl.onClose?.(reason);
                 onTerminal();
               },
+              logger,
             });
             gatewayControl.bindBridge(bridge);
             return bridge;
@@ -385,6 +387,7 @@ export function buildOpenAIRealtimeVoiceProvider(options?: {
           model,
           voice: config.voice,
           instructions: buildOpenAIQuicksilverInstructions(req.instructions),
+          logger: options?.logger ?? { warn: () => undefined },
           resolveAuth: async () => ({
             type: "api-key",
             token: (
@@ -413,12 +416,14 @@ export function buildOpenAIRealtimeVoiceProvider(options?: {
         azureEndpoint: config.azureEndpoint,
         azureDeployment: config.azureDeployment,
         azureApiVersion: config.azureApiVersion,
+        logger: options?.logger ?? { warn: () => undefined },
       });
     },
     createBrowserSession: (req) =>
       createOpenAIRealtimeBrowserSession(
         req as OpenAIInternalRealtimeBrowserSessionCreateRequest,
         options?.quicksilverBrowserSessionBroker,
+        options?.logger ?? { warn: () => undefined },
       ),
   };
   const internalApi: OpenAIInternalRealtimeVoiceProviderApi = {

extensions/openai/realtime-voice-session-policy.ts

@@ -82,6 +82,7 @@ export type OpenAIRealtimeVoiceBridgeConfig = RealtimeVoiceBridgeCreateRequest &
   azureEndpoint?: string;
   azureDeployment?: string;
   azureApiVersion?: string;
+  logger: Pick<import("openclaw/plugin-sdk/plugin-entry").PluginLogger, "warn">;
 };
 
 export const OPENAI_REALTIME_DEFAULT_MODEL = "gpt-realtime-2.1";

extensions/openai/realtime-voice-test-support.ts

@@ -18,6 +18,7 @@ export function createOpenAIRealtimeMockState() {
 
     readonly listeners = new Map<string, Listener[]>();
     readyState = 0;
+    bufferedAmount = 0;
     sent: string[] = [];
     closed = false;
     terminated = false;

extensions/voice-call/src/media-stream.playback-marks.test.ts

@@ -0,0 +1,179 @@
+// Voice Call tests cover Twilio playback-mark acknowledgement behavior.
+import type {
+  RealtimeTranscriptionProviderPlugin,
+  RealtimeTranscriptionSession,
+} from "openclaw/plugin-sdk/realtime-transcription";
+import type { TalkEvent } from "openclaw/plugin-sdk/realtime-voice";
+import { rawDataToString } from "openclaw/plugin-sdk/webhook-ingress";
+import { describe, expect, it, vi } from "vitest";
+import { WebSocket } from "ws";
+import { MediaStreamHandler } from "./media-stream.js";
+import {
+  connectWs,
+  startUpgradeWsServer,
+  waitForClose,
+  withTimeout,
+} from "./websocket-test-support.js";
+
+const createStubSession = (): RealtimeTranscriptionSession => ({
+  connect: async () => {},
+  sendAudio: () => {},
+  close: () => {},
+  isConnected: () => true,
+});
+
+const createStubSttProvider = (): RealtimeTranscriptionProviderPlugin =>
+  ({
+    createSession: () => createStubSession(),
+    id: "openai",
+    label: "OpenAI",
+    isConfigured: () => true,
+  }) as unknown as RealtimeTranscriptionProviderPlugin;
+
+const requireRecord = (value: unknown, label: string): Record<string, unknown> => {
+  if (value === null || typeof value !== "object" || Array.isArray(value)) {
+    throw new Error(`Expected ${label} to be a record`);
+  }
+  return value as Record<string, unknown>;
+};
+
+const nextWsMessage = (ws: WebSocket): Promise<Record<string, unknown>> =>
+  new Promise((resolve, reject) => {
+    ws.once("message", (data) => {
+      try {
+        resolve(requireRecord(JSON.parse(rawDataToString(data)), "WebSocket message"));
+      } catch (error) {
+        reject(error instanceof Error ? error : new Error(String(error)));
+      }
+    });
+  });
+
+const startWsServer = async (
+  handler: MediaStreamHandler,
+): Promise<{
+  url: string;
+  close: () => Promise<void>;
+}> =>
+  startUpgradeWsServer({
+    urlPath: "/voice/stream",
+    onUpgrade: (request, socket, head) => {
+      handler.handleUpgrade(request, socket, head);
+    },
+  });
+
+describe("MediaStreamHandler playback marks", () => {
+  it("completes queued playback only after Twilio echoes its mark", async () => {
+    const onConnect = vi.fn();
+    const handler = new MediaStreamHandler({
+      transcriptionProvider: createStubSttProvider(),
+      providerConfig: {},
+      shouldAcceptStream: () => true,
+      onConnect,
+    });
+    const server = await startWsServer(handler);
+    let ws: WebSocket | undefined;
+
+    try {
+      ws = await connectWs(server.url);
+      ws.send(
+        JSON.stringify({
+          event: "start",
+          streamSid: "MZ-mark",
+          start: { callSid: "CA-mark" },
+        }),
+      );
+      await vi.waitFor(() => expect(onConnect).toHaveBeenCalledOnce());
+
+      const outboundMark = nextWsMessage(ws);
+      let completed = false;
+      const playback = handler
+        .queueTts("MZ-mark", async (signal) => {
+          await handler.sendMarkAndWait("MZ-mark", "tts-complete", 100, signal);
+        })
+        .then(() => {
+          completed = true;
+        });
+      expect(await withTimeout(outboundMark)).toMatchObject({
+        event: "mark",
+        mark: { name: "tts-complete" },
+      });
+      await Promise.resolve();
+      expect(completed).toBe(false);
+
+      ws.send(
+        JSON.stringify({
+          event: "mark",
+          streamSid: "MZ-mark",
+          mark: { name: "tts-complete" },
+        }),
+      );
+      await withTimeout(playback);
+      expect(completed).toBe(true);
+
+      ws.close();
+      await waitForClose(ws);
+    } finally {
+      ws?.terminate();
+      await server.close();
+    }
+  });
+
+  it("ignores a playback mark echoed after clear", async () => {
+    const onConnect = vi.fn();
+    const talkEvents: TalkEvent[] = [];
+    const handler = new MediaStreamHandler({
+      transcriptionProvider: createStubSttProvider(),
+      providerConfig: {},
+      shouldAcceptStream: () => true,
+      onConnect,
+      onTalkEvent: (_callId, _streamSid, event) => talkEvents.push(event),
+    });
+    const server = await startWsServer(handler);
+    let ws: WebSocket | undefined;
+
+    try {
+      ws = await connectWs(server.url);
+      ws.send(
+        JSON.stringify({
+          event: "start",
+          streamSid: "MZ-clear-mark",
+          start: { callSid: "CA-clear-mark" },
+        }),
+      );
+      await vi.waitFor(() => expect(onConnect).toHaveBeenCalledOnce());
+
+      const outboundMark = nextWsMessage(ws);
+      const playback = handler.queueTts("MZ-clear-mark", async (signal) => {
+        await handler.sendMarkAndWait("MZ-clear-mark", "tts-cleared", 100, signal);
+      });
+      await withTimeout(outboundMark);
+      handler.clearTtsQueue("MZ-clear-mark", "barge-in");
+      await withTimeout(playback);
+      expect(talkEvents.some((event) => event.type === "output.audio.done")).toBe(false);
+
+      const state = handler as unknown as {
+        ignoredPlaybackMarks: Map<string, Set<string>>;
+      };
+      expect(state.ignoredPlaybackMarks.get("MZ-clear-mark")).toContain("tts-cleared");
+      ws.send(
+        JSON.stringify({
+          event: "mark",
+          streamSid: "MZ-clear-mark",
+          mark: { name: "tts-cleared" },
+        }),
+      );
+      await vi.waitFor(() => {
+        expect(state.ignoredPlaybackMarks.get("MZ-clear-mark")?.has("tts-cleared") ?? false).toBe(
+          false,
+        );
+      });
+      expect(talkEvents.some((event) => event.type === "output.audio.done")).toBe(false);
+
+      ws.close();
+      await waitForClose(ws);
+    } finally {
+      ws?.terminate();
+      await server.close();
+    }
+  });
+});

extensions/voice-call/src/media-stream.ts

@@ -83,6 +83,10 @@ type TtsQueueEntry = {
   reject: (error: unknown) => void;
 };
 
+type PendingPlaybackMark = {
+  settle: (error?: Error, ignoreLateAck?: boolean) => void;
+};
+
 type StreamSendResult = {
   sent: boolean;
   readyState?: number;
@@ -102,6 +106,8 @@ const DEFAULT_MAX_CONNECTIONS = 128;
 const MAX_INBOUND_MESSAGE_BYTES = 64 * 1024;
 const MAX_WS_BUFFERED_BYTES = 1024 * 1024;
 const MAX_PENDING_TTS_OPERATIONS_PER_STREAM = 8;
+const MAX_IGNORED_PLAYBACK_MARKS_PER_STREAM = 64;
+const PLAYBACK_MARK_TIMEOUT_GRACE_MS = 2_000;
 const CLOSE_REASON_LOG_MAX_CHARS = 120;
 
 function sanitizeLogText(value: string, maxChars: number): string {
@@ -158,6 +164,8 @@ export class MediaStreamHandler {
   private ttsPlaying = new Map<string, boolean>();
   /** Active TTS playback controllers per stream */
   private ttsActiveControllers = new Map<string, AbortController>();
+  private pendingPlaybackMarks = new Map<string, Map<string, PendingPlaybackMark>>();
+  private ignoredPlaybackMarks = new Map<string, Set<string>>();
 
   constructor(config: MediaStreamConfig) {
     this.config = config;
@@ -316,8 +324,13 @@ export class MediaStreamHandler {
             }
             break;
 
-          case "clear":
           case "mark":
+            if (session && message.mark?.name) {
+              this.acknowledgePlaybackMark(session.streamSid, message.mark.name);
+            }
+            break;
+
+          case "clear":
             break;
         }
       } catch (error) {
@@ -691,10 +704,74 @@ export class MediaStreamHandler {
     });
   }
 
+  /** Send a completion mark and wait until Twilio reports that buffered playback reached it. */
+  async sendMarkAndWait(
+    streamSid: string,
+    name: string,
+    audioDurationMs: number,
+    signal: AbortSignal,
+  ): Promise<void> {
+    signal.throwIfAborted();
+    const marks = this.getPendingPlaybackMarks(streamSid);
+    if (marks.has(name)) {
+      throw new Error(`Telephony playback mark is already pending: ${name}`);
+    }
+    this.ignoredPlaybackMarks.get(streamSid)?.delete(name);
+
+    let pending!: PendingPlaybackMark;
+    const acknowledgement = new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(
+        () => {
+          console.warn(`[MediaStream] Playback mark timed out; continuing stream=${streamSid}`);
+          pending.settle();
+        },
+        Math.max(1, audioDurationMs + PLAYBACK_MARK_TIMEOUT_GRACE_MS),
+      );
+      timeout.unref?.();
+      const onAbort = () => {
+        const reason =
+          signal.reason instanceof Error
+            ? signal.reason
+            : new Error("Telephony playback mark wait aborted");
+        pending.settle(reason, true);
+      };
+      pending = {
+        settle: (error, ignoreLateAck = false) => {
+          if (marks.get(name) !== pending) {
+            return;
+          }
+          clearTimeout(timeout);
+          signal.removeEventListener("abort", onAbort);
+          marks.delete(name);
+          if (marks.size === 0) {
+            this.pendingPlaybackMarks.delete(streamSid);
+          }
+          if (ignoreLateAck) {
+            this.ignorePlaybackMark(streamSid, name);
+          }
+          if (error) {
+            reject(error);
+          } else {
+            resolve();
+          }
+        },
+      };
+      marks.set(name, pending);
+      signal.addEventListener("abort", onAbort, { once: true });
+    });
+
+    const result = this.sendMark(streamSid, name);
+    if (!result.sent) {
+      pending.settle(new Error("Telephony stream playback failed: completion mark not delivered"));
+    }
+    return acknowledgement;
+  }
+
   /**
    * Clear audio buffer (interrupt playback).
    */
   clearAudio(streamSid: string): StreamSendResult {
+    this.invalidatePlaybackMarks(streamSid);
     return this.sendToStream(streamSid, { event: "clear", streamSid });
   }
 
@@ -762,6 +839,51 @@ export class MediaStreamHandler {
     return queue;
   }
 
+  private getPendingPlaybackMarks(streamSid: string): Map<string, PendingPlaybackMark> {
+    const existing = this.pendingPlaybackMarks.get(streamSid);
+    if (existing) {
+      return existing;
+    }
+    const marks = new Map<string, PendingPlaybackMark>();
+    this.pendingPlaybackMarks.set(streamSid, marks);
+    return marks;
+  }
+
+  private acknowledgePlaybackMark(streamSid: string, name: string): void {
+    const ignored = this.ignoredPlaybackMarks.get(streamSid);
+    if (ignored?.delete(name)) {
+      if (ignored.size === 0) {
+        this.ignoredPlaybackMarks.delete(streamSid);
+      }
+      return;
+    }
+    this.pendingPlaybackMarks.get(streamSid)?.get(name)?.settle();
+  }
+
+  private invalidatePlaybackMarks(streamSid: string): void {
+    const marks = this.pendingPlaybackMarks.get(streamSid);
+    if (!marks) {
+      return;
+    }
+    // Map iteration tolerates settle() deleting entries mid-walk.
+    for (const pending of marks.values()) {
+      pending.settle(new Error("Telephony playback cleared before completion"), true);
+    }
+  }
+
+  private ignorePlaybackMark(streamSid: string, name: string): void {
+    const ignored = this.ignoredPlaybackMarks.get(streamSid) ?? new Set<string>();
+    ignored.add(name);
+    while (ignored.size > MAX_IGNORED_PLAYBACK_MARKS_PER_STREAM) {
+      const oldest = ignored.values().next().value;
+      if (oldest === undefined) {
+        break;
+      }
+      ignored.delete(oldest);
+    }
+    this.ignoredPlaybackMarks.set(streamSid, ignored);
+  }
+
   /**
    * Process the TTS queue for a stream.
    * Uses iterative approach to avoid stack accumulation from recursion.
@@ -868,6 +990,8 @@ export class MediaStreamHandler {
     this.ttsActiveControllers.delete(streamSid);
     this.ttsPlaying.delete(streamSid);
     this.ttsQueues.delete(streamSid);
+    this.invalidatePlaybackMarks(streamSid);
+    this.ignoredPlaybackMarks.delete(streamSid);
   }
 
   private resolveQueuedTtsEntries(queue: TtsQueueEntry[]): void {

extensions/voice-call/src/providers/twilio.test.ts

@@ -745,7 +745,7 @@ describe("TwilioProvider", () => {
       provider.registerCallStream("CA-timeout", "MZ-timeout");
 
       const sendAudio = vi.fn();
-      const sendMark = vi.fn();
+      const sendMarkAndWait = vi.fn();
       const mediaStreamHandler = {
         queueTts: async (
           _streamSid: string,
@@ -754,7 +754,8 @@ describe("TwilioProvider", () => {
           await playFn(new AbortController().signal);
         },
         sendAudio,
-        sendMark,
+        sendMarkAndWait,
+        clearAudio: vi.fn(),
       };
 
       provider.setMediaStreamHandler(mediaStreamHandler as never);
@@ -773,20 +774,25 @@ describe("TwilioProvider", () => {
       await vi.advanceTimersByTimeAsync(5_100);
       await playExpectation;
       expect(sendAudio).toHaveBeenCalled();
-      expect(sendMark).not.toHaveBeenCalled();
+      expect(sendMarkAndWait).not.toHaveBeenCalled();
     } finally {
       vi.useRealTimers();
     }
   });
 
-  it("fails stream playback when all audio sends and completion mark are dropped", async () => {
+  it("stops and clears playback on the first failed audio chunk", async () => {
     vi.useFakeTimers();
     try {
       const provider = createProvider();
       provider.registerCallStream("CA-dropped", "MZ-dropped");
 
-      const sendAudio = vi.fn(() => ({ sent: false }));
-      const sendMark = vi.fn(() => ({ sent: false }));
+      const sendAudio = vi
+        .fn<() => { sent: boolean }>()
+        .mockReturnValueOnce({ sent: true })
+        .mockReturnValueOnce({ sent: true })
+        .mockReturnValue({ sent: false });
+      const sendMarkAndWait = vi.fn();
+      const clearAudio = vi.fn();
       const mediaStreamHandler = {
         queueTts: async (
           _streamSid: string,
@@ -795,13 +801,14 @@ describe("TwilioProvider", () => {
           await playFn(new AbortController().signal);
         },
         sendAudio,
-        sendMark,
+        sendMarkAndWait,
+        clearAudio,
       };
 
       provider.setMediaStreamHandler(mediaStreamHandler as never);
       provider.setTTSProvider({
         synthesisTimeoutMs: 5000,
-        synthesizeForTelephony: async () => Buffer.alloc(320),
+        synthesizeForTelephony: async () => Buffer.alloc(480),
       });
 
       const playback = provider.playTts({
@@ -812,8 +819,9 @@ describe("TwilioProvider", () => {
       const playExpectation = expect(playback).rejects.toThrow("Telephony stream playback failed");
       await vi.advanceTimersByTimeAsync(100);
       await playExpectation;
-      expect(sendAudio).toHaveBeenCalled();
-      expect(sendMark).toHaveBeenCalledTimes(1);
+      expect(sendAudio).toHaveBeenCalledTimes(3);
+      expect(clearAudio).toHaveBeenCalledWith("MZ-dropped");
+      expect(sendMarkAndWait).not.toHaveBeenCalled();
     } finally {
       vi.useRealTimers();
     }
@@ -824,7 +832,7 @@ describe("TwilioProvider", () => {
     provider.registerCallStream("CA-empty", "MZ-empty");
 
     const sendAudio = vi.fn();
-    const sendMark = vi.fn();
+    const sendMarkAndWait = vi.fn();
     const mediaStreamHandler = {
       queueTts: async (
         _streamSid: string,
@@ -833,7 +841,8 @@ describe("TwilioProvider", () => {
         await playFn(new AbortController().signal);
       },
       sendAudio,
-      sendMark,
+      sendMarkAndWait,
+      clearAudio: vi.fn(),
     };
 
     provider.setMediaStreamHandler(mediaStreamHandler as never);
@@ -850,7 +859,7 @@ describe("TwilioProvider", () => {
       }),
     ).rejects.toThrow("Telephony TTS produced no audio");
     expect(sendAudio).toHaveBeenCalled();
-    expect(sendMark).not.toHaveBeenCalled();
+    expect(sendMarkAndWait).not.toHaveBeenCalled();
   });
 
   it("exits chunk pacing early when the abort signal fires after the first chunk", async () => {
@@ -859,7 +868,7 @@ describe("TwilioProvider", () => {
       const provider = createProvider();
       provider.registerCallStream("CA-abort-chunk", "MZ-abort-chunk");
 
-      const sendMark = vi.fn(() => ({ sent: true }));
+      const sendMarkAndWait = vi.fn(async () => {});
       const controller = new AbortController();
       const sendAudio = vi.fn(() => {
         // The first send is the synthesis keepalive; the second is the first real audio chunk.
@@ -877,7 +886,8 @@ describe("TwilioProvider", () => {
           await playFn(controller.signal);
         },
         sendAudio,
-        sendMark,
+        sendMarkAndWait,
+        clearAudio: vi.fn(),
       };
 
       provider.setMediaStreamHandler(mediaStreamHandler as never);
@@ -899,10 +909,105 @@ describe("TwilioProvider", () => {
 
       expect(Date.now()).toBe(startedAt);
       expect(sendAudio).toHaveBeenCalledTimes(2);
-      expect(sendMark).not.toHaveBeenCalled();
+      expect(sendMarkAndWait).not.toHaveBeenCalled();
       expect(vi.getTimerCount()).toBe(0);
     } finally {
       vi.useRealTimers();
     }
   });
+
+  it("waits for the provider playback mark before completing stream playback", async () => {
+    vi.useFakeTimers();
+    try {
+      const provider = createProvider();
+      provider.registerCallStream("CA-mark", "MZ-mark");
+      let acknowledgeMark!: () => void;
+      const markAcknowledgement = new Promise<void>((resolve) => {
+        acknowledgeMark = resolve;
+      });
+      const sendMarkAndWait = vi.fn(async () => await markAcknowledgement);
+      provider.setMediaStreamHandler({
+        queueTts: async (_streamSid: string, playFn: (signal: AbortSignal) => Promise<void>) =>
+          await playFn(new AbortController().signal),
+        sendAudio: () => ({ sent: true }),
+        sendMarkAndWait,
+        clearAudio: vi.fn(),
+      } as never);
+      provider.setTTSProvider({
+        synthesisTimeoutMs: 5_000,
+        synthesizeForTelephony: async () => Buffer.alloc(320, 0x80),
+      });
+
+      let completed = false;
+      const playback = provider
+        .playTts({ callId: "call-mark", providerCallId: "CA-mark", text: "hello" })
+        .then(() => {
+          completed = true;
+        });
+      await vi.advanceTimersByTimeAsync(100);
+
+      expect(sendMarkAndWait).toHaveBeenCalledOnce();
+      expect(completed).toBe(false);
+      acknowledgeMark();
+      await playback;
+      expect(completed).toBe(true);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("releases serialized playback immediately when barge-in aborts synthesis", async () => {
+    const provider = createProvider();
+    provider.registerCallStream("CA-synth-abort", "MZ-synth-abort");
+    let activeController: AbortController | undefined;
+    let queueTail = Promise.resolve();
+    const mediaStreamHandler = {
+      queueTts: (_streamSid: string, playFn: (signal: AbortSignal) => Promise<void>) => {
+        const controller = new AbortController();
+        const operation = queueTail.then(async () => {
+          activeController = controller;
+          try {
+            await playFn(controller.signal);
+          } catch (error) {
+            if (!controller.signal.aborted) {
+              throw error;
+            }
+          } finally {
+            if (activeController === controller) {
+              activeController = undefined;
+            }
+          }
+        });
+        queueTail = operation.catch(() => {});
+        return operation;
+      },
+      clearTtsQueue: () => activeController?.abort(),
+      sendAudio: () => ({ sent: true }),
+      sendMarkAndWait: async () => {},
+      clearAudio: vi.fn(),
+    };
+    provider.setMediaStreamHandler(mediaStreamHandler as never);
+    const synthesizeForTelephony = vi
+      .fn<() => Promise<Buffer>>()
+      .mockImplementationOnce(async () => await new Promise<Buffer>(() => {}))
+      .mockResolvedValueOnce(Buffer.alloc(160, 0x80));
+    provider.setTTSProvider({ synthesisTimeoutMs: 30_000, synthesizeForTelephony });
+
+    const cancelled = provider.playTts({
+      callId: "call-synth-abort",
+      providerCallId: "CA-synth-abort",
+      text: "cancel me",
+    });
+    await vi.waitFor(() => expect(synthesizeForTelephony).toHaveBeenCalledTimes(1));
+    provider.clearTtsQueue("CA-synth-abort", "barge-in");
+    const next = provider.playTts({
+      callId: "call-synth-abort",
+      providerCallId: "CA-synth-abort",
+      text: "play next",
+    });
+
+    await expect(cancelled).resolves.toBeUndefined();
+    await expect(next).resolves.toBeUndefined();
+    expect(synthesizeForTelephony).toHaveBeenCalledTimes(2);
+  });
 });

extensions/voice-call/src/providers/twilio.ts

@@ -94,6 +94,7 @@ export class TwilioProvider implements VoiceCallProvider {
 
   /** Optional media stream handler for sending audio */
   private mediaStreamHandler: MediaStreamHandler | null = null;
+  private playbackMarkSequence = 0;
 
   /** Map of call SID to stream SID for media streams */
   private callStreamMap = new Map<string, string>();
@@ -705,14 +706,6 @@ export class TwilioProvider implements VoiceCallProvider {
       return normalizeSendResult(raw);
     };
 
-    const sendPlaybackMark = (name: string): StreamSendResult => {
-      const raw = (handler as { sendMark: (sid: string, markName: string) => unknown }).sendMark(
-        streamSid,
-        name,
-      );
-      return normalizeSendResult(raw);
-    };
-
     await handler.queueTts(streamSid, async (signal) => {
       const sendKeepAlive = () => {
         sendAudioChunk(SILENCE_CHUNK);
@@ -727,6 +720,7 @@ export class TwilioProvider implements VoiceCallProvider {
       // Generate audio with core TTS (returns mu-law at 8kHz)
       let muLawAudio: Buffer;
       let synthTimeout: ReturnType<typeof setTimeout> | null = null;
+      let removeAbortListener = () => {};
       const synthTimeoutMs = ttsProvider.synthesisTimeoutMs;
       try {
         const synthPromise = ttsProvider.synthesizeForTelephony(text);
@@ -735,12 +729,27 @@ export class TwilioProvider implements VoiceCallProvider {
             reject(new Error(`Telephony TTS synthesis timed out after ${synthTimeoutMs}ms`));
           }, synthTimeoutMs);
         });
-        muLawAudio = await Promise.race([synthPromise, timeoutPromise]);
+        const abortPromise = new Promise<never>((_, reject) => {
+          const onAbort = () => {
+            reject(
+              signal.reason instanceof Error
+                ? signal.reason
+                : new Error("Telephony TTS synthesis aborted"),
+            );
+          };
+          signal.addEventListener("abort", onAbort, { once: true });
+          removeAbortListener = () => signal.removeEventListener("abort", onAbort);
+          if (signal.aborted) {
+            onAbort();
+          }
+        });
+        muLawAudio = await Promise.race([synthPromise, timeoutPromise, abortPromise]);
       } finally {
         if (synthTimeout) {
           clearTimeout(synthTimeout);
         }
         clearInterval(keepAlive);
+        removeAbortListener();
       }
 
       if (muLawAudio.length === 0) {
@@ -756,9 +765,13 @@ export class TwilioProvider implements VoiceCallProvider {
         }
         chunkAttempts += 1;
         const chunkResult = sendAudioChunk(chunk);
-        if (chunkResult.sent) {
-          chunkDelivered += 1;
+        if (!chunkResult.sent) {
+          handler.clearAudio(streamSid);
+          throw new Error(
+            `Telephony stream playback failed: audio chunk ${chunkAttempts} not delivered`,
+          );
         }
+        chunkDelivered += 1;
 
         // Drift-corrected pacing: schedule against an absolute clock to avoid cumulative delay.
         const waitMs = nextChunkDueAt - Date.now();
@@ -778,22 +791,14 @@ export class TwilioProvider implements VoiceCallProvider {
         }
       }
 
-      let markSent = true;
-      if (!signal.aborted) {
-        // Send a mark to track when audio finishes
-        markSent = sendPlaybackMark(`tts-${Date.now()}`).sent;
+      if (signal.aborted) {
+        return;
       }
-
-      if (!signal.aborted && chunkAttempts > 0 && (chunkDelivered === 0 || !markSent)) {
-        const failures: string[] = [];
-        if (chunkDelivered === 0) {
-          failures.push("no audio chunks delivered");
-        }
-        if (!markSent) {
-          failures.push("completion mark not delivered");
-        }
-        throw new Error(`Telephony stream playback failed: ${failures.join("; ")}`);
+      if (chunkAttempts === 0 || chunkDelivered !== chunkAttempts) {
+        throw new Error("Telephony stream playback failed: incomplete audio delivery");
       }
+      const markName = `tts-${Date.now()}-${++this.playbackMarkSequence}`;
+      await handler.sendMarkAndWait(streamSid, markName, muLawAudio.length / 8, signal);
     });
   }
 

extensions/voice-call/src/telephony-tts.test.ts

@@ -36,6 +36,67 @@ function createRuntime(
 }
 
 describe("createTelephonyTtsProvider", () => {
+  it.each([
+    ["Azure", "raw-8khz-8bit-mono-mulaw"],
+    ["Gradium", "ulaw_8000"],
+  ])("passes through %s 8 kHz mu-law output", async (providerName, outputFormat) => {
+    const audioBuffer = Buffer.from([0x00, 0x7f, 0xff]);
+    const provider = await createTelephonyTtsProvider({
+      coreConfig: createCoreConfig(),
+      runtime: createRuntime(async () => ({
+        success: true,
+        audioBuffer,
+        outputFormat,
+        sampleRate: 8_000,
+        provider: providerName.toLowerCase(),
+      })),
+    });
+
+    await expect(provider.synthesizeForTelephony("hello")).resolves.toBe(audioBuffer);
+  });
+
+  it("converts provider PCM output to 8 kHz mu-law", async () => {
+    const provider = await createTelephonyTtsProvider({
+      coreConfig: createCoreConfig(),
+      runtime: createRuntime(async () => ({
+        success: true,
+        audioBuffer: Buffer.alloc(480 * 2),
+        outputFormat: "pcm",
+        sampleRate: 24_000,
+        provider: "openai",
+      })),
+    });
+
+    await expect(provider.synthesizeForTelephony("hello")).resolves.toEqual(
+      Buffer.alloc(160, 0xff),
+    );
+  });
+
+  it.each([
+    ["mp3", 24_000],
+    ["riff-8khz-8bit-mono-mulaw", 8_000],
+  ])(
+    "rejects unsupported %s container output with provider context",
+    async (outputFormat, sampleRate) => {
+      const provider = await createTelephonyTtsProvider({
+        coreConfig: createCoreConfig(),
+        runtime: createRuntime(async () => ({
+          success: true,
+          audioBuffer: Buffer.from("container"),
+          outputFormat,
+          sampleRate,
+          provider: "example-provider",
+        })),
+      });
+
+      const synthesis = provider.synthesizeForTelephony("hello");
+      await expect(synthesis).rejects.toMatchObject({
+        name: "UnsupportedTelephonyTtsOutputFormatError",
+        message: `Unsupported telephony TTS output format "${outputFormat}" from provider "example-provider"`,
+      });
+    },
+  );
+
   it("uses shared preparation for the surface override and request text", async () => {
     const effectiveConfig: OpenClawConfig = {
       tts: { provider: "openai", timeoutMs: 15_000 },

extensions/voice-call/src/telephony-tts.ts

@@ -27,6 +27,7 @@ export type TelephonyTtsRuntime = {
     audioBuffer?: Buffer;
     sampleRate?: number;
     provider?: string;
+    outputFormat?: string;
     fallbackFrom?: string;
     attemptedProviders?: string[];
     error?: string;
@@ -42,6 +43,45 @@ export type TelephonyTtsProvider = {
 /** Default timeout for one telephony synthesis request. */
 export const TELEPHONY_DEFAULT_TTS_TIMEOUT_MS = 8000;
 
+class UnsupportedTelephonyTtsOutputFormatError extends Error {
+  constructor(
+    readonly outputFormat: string,
+    readonly provider: string,
+  ) {
+    super(`Unsupported telephony TTS output format "${outputFormat}" from provider "${provider}"`);
+    this.name = "UnsupportedTelephonyTtsOutputFormatError";
+  }
+}
+
+function convertTelephonyTtsOutput(result: {
+  audioBuffer: Buffer;
+  outputFormat?: string;
+  provider?: string;
+  sampleRate: number;
+}): Buffer {
+  const format = result.outputFormat?.trim().toLowerCase();
+  // Bundled provider contracts: Azure/Gradium emit raw-8khz-8bit-mono-mulaw/ulaw_8000;
+  // ElevenLabs/OpenAI emit pcm_22050/pcm. An absent format is the shipped PCM default.
+  const isRawMulaw = format === "raw-8khz-8bit-mono-mulaw" || format === "ulaw_8000";
+  if (isRawMulaw && result.sampleRate === 8_000) {
+    return result.audioBuffer;
+  }
+  const isPcm =
+    !format ||
+    format === "pcm" ||
+    /^pcm[_-]\d+$/.test(format) ||
+    (format.includes("raw") &&
+      (format.includes("16bit") || format.includes("16-bit")) &&
+      format.includes("pcm"));
+  if (isPcm) {
+    return convertPcmToMulaw8k(result.audioBuffer, result.sampleRate);
+  }
+  throw new UnsupportedTelephonyTtsOutputFormatError(
+    result.outputFormat ?? "absent",
+    result.provider ?? "unknown",
+  );
+}
+
 /** Create a TTS provider that honors voice-call overrides and converts PCM to mulaw. */
 export async function createTelephonyTtsProvider(params: {
   coreConfig: OpenClawConfig;
@@ -98,7 +138,12 @@ export async function createTelephonyTtsProvider(params: {
         );
       }
 
-      return convertPcmToMulaw8k(result.audioBuffer, result.sampleRate);
+      return convertTelephonyTtsOutput({
+        audioBuffer: result.audioBuffer,
+        outputFormat: result.outputFormat,
+        provider: result.provider,
+        sampleRate: result.sampleRate,
+      });
     },
   };
 }

extensions/voice-call/src/webhook/realtime-audio-pacer.test.ts

@@ -46,8 +46,8 @@ describe("RealtimeAudioPacer", () => {
     vi.useRealTimers();
   });
 
-  it("paces realtime audio as 20ms telephony frames before marks (Twilio shape)", async () => {
-    vi.useFakeTimers();
+  it("primes an eight-frame lead and then advances without timer drift", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance", "Date"] });
     const sent: unknown[] = [];
     const pacer = new RealtimeAudioPacer({
       serializer: createTwilioSerializer("MZ-test"),
@@ -57,26 +57,70 @@ describe("RealtimeAudioPacer", () => {
       },
     });
 
-    pacer.sendAudio(Buffer.alloc(320, 0x7f));
-    pacer.sendMark("audio-1");
+    pacer.sendAudio(Buffer.alloc(50 * 160, 0x7f));
 
-    expect(sent).toHaveLength(1);
-    expect(
-      Buffer.from((sent[0] as { media: { payload: string } }).media.payload, "base64"),
-    ).toHaveLength(160);
+    expect(sent).toHaveLength(8);
 
-    await vi.advanceTimersByTimeAsync(20);
-    expect(sent).toHaveLength(2);
-    expect(
-      Buffer.from((sent[1] as { media: { payload: string } }).media.payload, "base64"),
-    ).toHaveLength(160);
+    await vi.advanceTimersByTimeAsync(100);
+    expect(sent).toHaveLength(13);
+  });
 
-    await vi.advanceTimersByTimeAsync(20);
-    expect(sent[2]).toEqual({
-      event: "mark",
-      streamSid: "MZ-test",
-      mark: { name: "audio-1" },
+  it("catches up all overdue frames when a pump runs late", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance", "Date"] });
+    let now = 0;
+    vi.spyOn(performance, "now").mockImplementation(() => now);
+    const sent: string[] = [];
+    const pacer = new RealtimeAudioPacer({
+      serializer: createCompactSerializer(),
+      send: (message) => {
+        sent.push(message);
+        return true;
+      },
     });
+
+    pacer.sendAudio(createSequencedAudio(50));
+    expect(sent).toHaveLength(8);
+
+    now = 100;
+    await vi.runOnlyPendingTimersAsync();
+    expect(sent).toHaveLength(13);
+  });
+
+  it("starts a fresh lead window after a genuine silence gap", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance", "Date"] });
+    const sent: string[] = [];
+    const pacer = new RealtimeAudioPacer({
+      serializer: createCompactSerializer(),
+      send: (message) => {
+        sent.push(message);
+        return true;
+      },
+    });
+
+    pacer.sendAudio(createSequencedAudio(5));
+    expect(sent).toHaveLength(5);
+    await vi.advanceTimersByTimeAsync(500);
+    pacer.sendAudio(createSequencedAudio(20));
+    expect(sent).toHaveLength(13);
+  });
+
+  it("preserves marks after the audio they follow", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance", "Date"] });
+    const sent: string[] = [];
+    const pacer = new RealtimeAudioPacer({
+      serializer: createCompactSerializer(),
+      send: (message) => {
+        sent.push(message);
+        return true;
+      },
+    });
+
+    pacer.sendAudio(createSequencedAudio(10));
+    pacer.sendMark("audio-1");
+    pacer.sendMark("audio-2");
+    await vi.advanceTimersByTimeAsync(100);
+
+    expect(sent.slice(-2)).toEqual(["mark:audio-1", "mark:audio-2"]);
   });
 
   it("clears queued audio immediately (Twilio shape)", async () => {
@@ -94,8 +138,27 @@ describe("RealtimeAudioPacer", () => {
     pacer.clearAudio();
     await vi.advanceTimersByTimeAsync(100);
 
-    expect(sent).toHaveLength(2);
-    expect(sent[1]).toEqual({ event: "clear", streamSid: "MZ-test" });
+    expect(sent).toHaveLength(4);
+    expect(sent[3]).toEqual({ event: "clear", streamSid: "MZ-test" });
+  });
+
+  it("closes without sending the remaining lead-window backlog", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "performance", "Date"] });
+    const sent: string[] = [];
+    const pacer = new RealtimeAudioPacer({
+      serializer: createCompactSerializer(),
+      send: (message) => {
+        sent.push(message);
+        return true;
+      },
+    });
+
+    pacer.sendAudio(createSequencedAudio(20));
+    pacer.close();
+    await vi.advanceTimersByTimeAsync(1_000);
+
+    expect(sent).toHaveLength(8);
+    expect(pacer.hasPendingAudio()).toBe(false);
   });
 
   it("stops instead of buffering unbounded realtime audio", async () => {
@@ -183,12 +246,12 @@ describe("RealtimeAudioPacer", () => {
     pacer.sendAudio(createSequencedAudio(frameCount));
     await vi.advanceTimersByTimeAsync(500 * 20);
 
-    expect(sent).toHaveLength(501);
-    expect(inspectQueue(pacer)).toEqual({ length: 399, head: 100 });
-    expect(pacer.clearAudio()).toBe(299 * 160);
+    expect(sent).toHaveLength(508);
+    expect(inspectQueue(pacer)).toEqual({ length: 399, head: 107 });
+    expect(pacer.clearAudio()).toBe(292 * 160);
     await vi.advanceTimersByTimeAsync(frameCount * 20);
 
-    expect(sent).toHaveLength(502);
+    expect(sent).toHaveLength(509);
     expect(sent.at(-1)).toBe("clear");
     expect(inspectQueue(pacer)).toEqual({ length: 0, head: 0 });
     expect(pacer.hasPendingAudio()).toBe(false);

extensions/voice-call/src/webhook/realtime-audio-pacer.ts

@@ -2,7 +2,9 @@
 
 const TELEPHONY_SAMPLE_RATE = 8_000;
 const TELEPHONY_CHUNK_BYTES = 160;
-const TELEPHONY_CHUNK_MS = 20;
+// The lead absorbs event-loop timer lateness and network jitter in the telephony edge buffer.
+// Barge-in clear flushes both queues, so this cushion does not add interruption latency.
+const LEAD_MS = 160;
 const DEFAULT_MAX_QUEUED_AUDIO_BYTES = TELEPHONY_SAMPLE_RATE * 120;
 const QUEUE_COMPACT_HEAD_THRESHOLD = 256;
 
@@ -35,6 +37,7 @@ export class RealtimeAudioPacer {
   private timer: ReturnType<typeof setTimeout> | null = null;
   private queuedAudioBytes = 0;
   private closed = false;
+  private streamClockMs: number | null = null;
 
   constructor(
     private readonly params: {
@@ -60,7 +63,7 @@ export class RealtimeAudioPacer {
       this.queue.push({
         type: "audio",
         chunk,
-        durationMs: Math.max(1, Math.round((chunk.length / TELEPHONY_SAMPLE_RATE) * 1000)),
+        durationMs: chunk.length / 8,
       });
       this.queuedAudioBytes += chunk.length;
     }
@@ -85,6 +88,7 @@ export class RealtimeAudioPacer {
     this.clearTimer();
     this.resetQueue();
     this.queuedAudioBytes = 0;
+    this.streamClockMs = null;
     this.params.send(this.params.serializer.clear());
     return clearedAudioBytes;
   }
@@ -100,6 +104,7 @@ export class RealtimeAudioPacer {
     this.clearTimer();
     this.resetQueue();
     this.queuedAudioBytes = 0;
+    this.streamClockMs = null;
   }
 
   /** Clear the scheduled pump timer. */
@@ -153,34 +158,45 @@ export class RealtimeAudioPacer {
     this.queueHead = 0;
   }
 
-  /** Send one queued item and schedule the next send based on audio duration. */
+  /** Fill the provider playout cushion, then wake at the next timeline boundary. */
   private pump(): void {
     this.timer = null;
     if (this.closed) {
       return;
     }
-    const item = this.takeNextItem();
-    if (!item) {
-      return;
+    const now = performance.now();
+    this.streamClockMs ??= now;
+
+    while (this.pendingQueueSize > 0 && this.streamClockMs < now + LEAD_MS) {
+      const item = this.takeNextItem();
+      if (!item) {
+        break;
+      }
+
+      const sent =
+        item.type === "audio"
+          ? this.sendAudioItem(item)
+          : this.params.send(this.params.serializer.mark(item.name));
+      if (!sent) {
+        this.resetQueue();
+        this.queuedAudioBytes = 0;
+        this.streamClockMs = null;
+        return;
+      }
     }
 
-    let delayMs = 0;
-    let sent;
-    if (item.type === "audio") {
-      this.queuedAudioBytes = Math.max(0, this.queuedAudioBytes - item.chunk.length);
-      sent = this.params.send(this.params.serializer.media(item.chunk.toString("base64")));
-      delayMs = item.durationMs || TELEPHONY_CHUNK_MS;
-    } else {
-      sent = this.params.send(this.params.serializer.mark(item.name));
-    }
-
-    if (!sent) {
-      this.resetQueue();
-      this.queuedAudioBytes = 0;
+    if (this.pendingQueueSize === 0) {
+      this.streamClockMs = null;
       return;
     }
-    if (this.pendingQueueSize > 0) {
-      this.timer = setTimeout(() => this.pump(), delayMs);
-    }
+    const delayMs = Math.max(1, this.streamClockMs - LEAD_MS - performance.now());
+    this.timer = setTimeout(() => this.pump(), delayMs);
+  }
+
+  private sendAudioItem(item: Extract<RealtimeAudioQueueItem, { type: "audio" }>): boolean {
+    this.queuedAudioBytes = Math.max(0, this.queuedAudioBytes - item.chunk.length);
+    const sent = this.params.send(this.params.serializer.media(item.chunk.toString("base64")));
+    this.streamClockMs = (this.streamClockMs ?? performance.now()) + item.durationMs;
+    return sent;
   }
 }

src/plugin-sdk/realtime-voice.test.ts

@@ -263,6 +263,27 @@ describe("RealtimeVoiceSessionLifecycle", () => {
     lifecycle.cancel();
     expect(lifecycle.drainPendingAudio()).toEqual([]);
   });
+
+  it("keeps the freshest queued speech and warns once per overflow episode", () => {
+    const onPendingAudioOverflow = vi.fn();
+    const lifecycle = new RealtimeVoiceSessionLifecycle("Test", {
+      pendingAudioOverflowPolicy: "drop-oldest",
+      onPendingAudioOverflow,
+    });
+
+    for (let index = 0; index < 322; index += 1) {
+      expect(lifecycle.enqueuePendingAudio(Buffer.from([index % 256]))).toBe(true);
+    }
+    expect(onPendingAudioOverflow).toHaveBeenCalledOnce();
+    expect(lifecycle.drainPendingAudio()).toEqual(
+      Array.from({ length: 320 }, (_, index) => Buffer.from([(index + 2) % 256])),
+    );
+
+    for (let index = 0; index < 321; index += 1) {
+      lifecycle.enqueuePendingAudio(Buffer.from([index % 256]));
+    }
+    expect(onPendingAudioOverflow).toHaveBeenCalledTimes(2);
+  });
 });
 
 describe("normalizeRealtimeVoiceResponseOutcome", () => {

src/plugin-sdk/realtime-voice.ts

@@ -221,6 +221,7 @@ export {
 } from "../talk/audio-energy.js";
 export {
   convertPcmToMulaw8k,
+  createStreamingPcmResampler,
   mulawToPcm,
   pcmToMulaw,
   resamplePcm,

src/talk/audio-codec.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it } from "vitest";
+import { createStreamingPcmResampler, resamplePcm } from "./audio-codec.js";
+
+function createSine(sampleRate: number, durationMs: number): Buffer {
+  const sampleCount = Math.floor((sampleRate * durationMs) / 1_000);
+  const pcm = Buffer.alloc(sampleCount * 2);
+  for (let index = 0; index < sampleCount; index += 1) {
+    pcm.writeInt16LE(
+      Math.round(Math.sin((2 * Math.PI * 440 * index) / sampleRate) * 20_000),
+      index * 2,
+    );
+  }
+  return pcm;
+}
+
+function processInFrames(input: Buffer, inputRate: number, outputRate: number) {
+  const resampler = createStreamingPcmResampler(inputRate, outputRate);
+  const frameBytes = (inputRate / 50) * 2;
+  const chunks: Buffer[] = [];
+  for (let offset = 0; offset < input.length; offset += frameBytes) {
+    chunks.push(resampler.process(input.subarray(offset, offset + frameBytes)));
+  }
+  chunks.push(resampler.flush());
+  return chunks;
+}
+
+describe("createStreamingPcmResampler", () => {
+  it("matches whole-buffer resampling across 20 ms chunk boundaries", () => {
+    const input = createSine(48_000, 1_000);
+    const streamed = Buffer.concat(processInFrames(input, 48_000, 24_000));
+    const oneShot = resamplePcm(input, 48_000, 24_000);
+
+    expect(Math.abs(streamed.length - oneShot.length)).toBeLessThanOrEqual(2);
+    for (let offset = 32; offset < Math.min(streamed.length, oneShot.length) - 32; offset += 2) {
+      expect(
+        Math.abs(streamed.readInt16LE(offset) - oneShot.readInt16LE(offset)),
+      ).toBeLessThanOrEqual(3);
+    }
+  });
+
+  it("keeps chunk-seam jumps within the continuous signal envelope", () => {
+    const input = createSine(8_000, 1_000);
+    const chunks = processInFrames(input, 8_000, 24_000);
+    const streamed = Buffer.concat(chunks);
+    const seamSamples = new Set<number>();
+    let samples = 0;
+    for (const chunk of chunks.slice(0, -1)) {
+      samples += chunk.length / 2;
+      seamSamples.add(samples);
+    }
+
+    let maxSeamJump = 0;
+    let maxInteriorJump = 0;
+    for (let sample = 1; sample < streamed.length / 2; sample += 1) {
+      const jump = Math.abs(
+        streamed.readInt16LE(sample * 2) - streamed.readInt16LE((sample - 1) * 2),
+      );
+      if (seamSamples.has(sample)) {
+        maxSeamJump = Math.max(maxSeamJump, jump);
+      } else {
+        maxInteriorJump = Math.max(maxInteriorJump, jump);
+      }
+    }
+
+    expect(maxSeamJump).toBeLessThanOrEqual(maxInteriorJump + 3);
+  });
+});

src/talk/audio-codec.ts

@@ -20,6 +20,14 @@ type ResampleKernel = {
   phaseCount: number;
 };
 
+type ResamplePlan = {
+  cutoffCyclesPerSample: number;
+  inputSampleRate: number;
+  kernel: ResampleKernel | undefined;
+  outputSampleRate: number;
+  ratio: number;
+};
+
 const HOST_IS_LITTLE_ENDIAN = new Uint16Array(new Uint8Array([1, 0]).buffer)[0] === 1;
 
 /** Clamp an intermediate sample to signed 16-bit PCM range. */
@@ -158,6 +166,70 @@ function sampleBandlimited(
   return weighted / weightSum;
 }
 
+function createResamplePlan(inputSampleRate: number, outputSampleRate: number): ResamplePlan {
+  const ratio = inputSampleRate / outputSampleRate;
+  const maxCutoff = 0.5;
+  const downsampleCutoff = ratio > 1 ? maxCutoff / ratio : maxCutoff;
+  const cutoffCyclesPerSample = Math.max(0.01, downsampleCutoff * RESAMPLE_CUTOFF_GUARD);
+  return {
+    cutoffCyclesPerSample,
+    inputSampleRate,
+    kernel: buildResampleKernel(inputSampleRate, outputSampleRate, cutoffCyclesPerSample),
+    outputSampleRate,
+    ratio,
+  };
+}
+
+function sampleResampledPcm(
+  input: Int16Array,
+  inputStartSample: number,
+  outputIndex: number,
+  plan: ResamplePlan,
+): number {
+  const sourcePosition = (outputIndex * plan.inputSampleRate) / plan.outputSampleRate;
+  return Math.round(
+    plan.kernel
+      ? sampleBandlimitedWithCoefficients(
+          input,
+          Math.floor(sourcePosition) - inputStartSample,
+          expectDefined(
+            plan.kernel.coefficients[
+              (outputIndex * plan.kernel.inputStep) % plan.kernel.phaseCount
+            ],
+            "coefficients entry at (output index * kernel input step) % kernel phase count",
+          ) ?? plan.kernel.coefficients[0],
+        )
+      : sampleBandlimited(
+          input,
+          outputIndex * plan.ratio - inputStartSample,
+          plan.cutoffCyclesPerSample,
+        ),
+  );
+}
+
+function renderResampledPcm(
+  input: Buffer,
+  inputStartSample: number,
+  firstOutputIndex: number,
+  outputSamples: number,
+  plan: ResamplePlan,
+): Buffer {
+  const output = Buffer.alloc(outputSamples * 2);
+  const inputView = readInt16Samples(input);
+  const outputView = canUseInt16View(output) ? int16View(output) : undefined;
+  for (let offset = 0; offset < outputSamples; offset += 1) {
+    const sample = clamp16(
+      sampleResampledPcm(inputView, inputStartSample, firstOutputIndex + offset, plan),
+    );
+    if (outputView) {
+      outputView[offset] = sample;
+    } else {
+      output.writeInt16LE(sample, offset * 2);
+    }
+  }
+  return output;
+}
+
 /** Resample little-endian signed 16-bit PCM to another integer sample rate. */
 export function resamplePcm(
   input: Buffer,
@@ -172,38 +244,93 @@ export function resamplePcm(
     return Buffer.alloc(0);
   }
 
-  const ratio = inputSampleRate / outputSampleRate;
-  const outputSamples = Math.floor(inputSamples / ratio);
-  const output = Buffer.alloc(outputSamples * 2);
-  const maxCutoff = 0.5;
-  const downsampleCutoff = ratio > 1 ? maxCutoff / ratio : maxCutoff;
-  const cutoffCyclesPerSample = Math.max(0.01, downsampleCutoff * RESAMPLE_CUTOFF_GUARD);
-  const kernel = buildResampleKernel(inputSampleRate, outputSampleRate, cutoffCyclesPerSample);
+  const plan = createResamplePlan(inputSampleRate, outputSampleRate);
+  const outputSamples = Math.floor(inputSamples / plan.ratio);
+  return renderResampledPcm(input, 0, 0, outputSamples, plan);
+}
 
-  const inputView = readInt16Samples(input);
-  const outputView = canUseInt16View(output) ? int16View(output) : undefined;
-
-  for (let i = 0; i < outputSamples; i += 1) {
-    const sample = Math.round(
-      kernel
-        ? sampleBandlimitedWithCoefficients(
-            inputView,
-            Math.floor((i * inputSampleRate) / outputSampleRate),
-            expectDefined(
-              kernel.coefficients[(i * kernel.inputStep) % kernel.phaseCount],
-              "coefficients entry at (i * kernel.input step) % kernel.phase count",
-            ) ?? kernel.coefficients[0],
-          )
-        : sampleBandlimited(inputView, i * ratio, cutoffCyclesPerSample),
-    );
-    if (outputView) {
-      outputView[i] = clamp16(sample);
-    } else {
-      output.writeInt16LE(clamp16(sample), i * 2);
-    }
+/** Create a chunk-safe PCM resampler that preserves filter and fractional phase state. */
+export function createStreamingPcmResampler(
+  inputSampleRate: number,
+  outputSampleRate: number,
+): {
+  process(chunk: Buffer): Buffer;
+  flush(): Buffer;
+} {
+  if (inputSampleRate === outputSampleRate) {
+    return {
+      process: (chunk) => Buffer.from(chunk),
+      flush: () => Buffer.alloc(0),
+    };
   }
 
-  return output;
+  const plan = createResamplePlan(inputSampleRate, outputSampleRate);
+  let bufferedInput = Buffer.alloc(0);
+  let inputStartSample = 0;
+  let totalInputSamples = 0;
+  let nextOutputIndex = 0;
+  let trailingByte = Buffer.alloc(0);
+  let flushed = false;
+
+  const renderAvailable = (includeRightEdge: boolean): Buffer => {
+    const targetOutputCount = Math.floor(totalInputSamples / plan.ratio);
+    let endOutputIndex = nextOutputIndex;
+    while (endOutputIndex < targetOutputCount) {
+      const center = Math.floor((endOutputIndex * plan.inputSampleRate) / plan.outputSampleRate);
+      if (!includeRightEdge && center + RESAMPLE_HALF_TAPS >= totalInputSamples) {
+        break;
+      }
+      endOutputIndex += 1;
+    }
+
+    const output = renderResampledPcm(
+      bufferedInput,
+      inputStartSample,
+      nextOutputIndex,
+      endOutputIndex - nextOutputIndex,
+      plan,
+    );
+    nextOutputIndex = endOutputIndex;
+
+    const nextCenter = Math.floor((nextOutputIndex * plan.inputSampleRate) / plan.outputSampleRate);
+    const retainFromSample = Math.max(0, nextCenter - RESAMPLE_HALF_TAPS);
+    const dropSamples = retainFromSample - inputStartSample;
+    if (dropSamples > 0) {
+      bufferedInput = Buffer.from(bufferedInput.subarray(dropSamples * 2));
+      inputStartSample = retainFromSample;
+    }
+    return output;
+  };
+
+  return {
+    process(chunk) {
+      if (flushed) {
+        throw new Error("Cannot process PCM after the streaming resampler was flushed");
+      }
+      const combined = trailingByte.length > 0 ? Buffer.concat([trailingByte, chunk]) : chunk;
+      const completeBytes = combined.length - (combined.length % 2);
+      trailingByte = Buffer.from(combined.subarray(completeBytes));
+      if (completeBytes > 0) {
+        const completePcm = combined.subarray(0, completeBytes);
+        bufferedInput =
+          bufferedInput.length > 0
+            ? Buffer.concat([bufferedInput, completePcm])
+            : Buffer.from(completePcm);
+        totalInputSamples += completeBytes / 2;
+      }
+      return renderAvailable(false);
+    },
+    flush() {
+      if (flushed) {
+        return Buffer.alloc(0);
+      }
+      flushed = true;
+      trailingByte = Buffer.alloc(0);
+      const output = renderAvailable(true);
+      bufferedInput = Buffer.alloc(0);
+      return output;
+    },
+  };
 }
 
 /** Resample little-endian signed 16-bit PCM to the telephony 8 kHz rate. */

src/talk/realtime-session-lifecycle.ts

@@ -12,6 +12,7 @@ export type RealtimeVoiceAudioQueue = {
 
 export function createRealtimeVoiceAudioQueue(
   overflowPolicy: RealtimeVoiceAudioOverflowPolicy,
+  onOverflow?: () => void,
 ): RealtimeVoiceAudioQueue {
   let chunks: Buffer[] = [];
   let bytes = 0;
@@ -37,6 +38,7 @@ export function createRealtimeVoiceAudioQueue(
     },
     enqueue: (audio) => {
       if (audio.byteLength > REALTIME_VOICE_MAX_PENDING_AUDIO_BYTES) {
+        onOverflow?.();
         return false;
       }
       if (
@@ -44,12 +46,14 @@ export function createRealtimeVoiceAudioQueue(
         (chunks.length >= REALTIME_VOICE_MAX_PENDING_AUDIO_CHUNKS ||
           bytes + audio.byteLength > REALTIME_VOICE_MAX_PENDING_AUDIO_BYTES)
       ) {
+        onOverflow?.();
         return false;
       }
       while (
         chunks.length >= REALTIME_VOICE_MAX_PENDING_AUDIO_CHUNKS ||
         bytes + audio.byteLength > REALTIME_VOICE_MAX_PENDING_AUDIO_BYTES
       ) {
+        onOverflow?.();
         const dropped = chunks.shift();
         if (!dropped) {
           return false;
@@ -110,9 +114,26 @@ type RealtimeVoiceConnectAttemptOptions = {
 export class RealtimeVoiceSessionLifecycle {
   private state: RealtimeVoiceIdleState | RealtimeVoiceConnectionState = { phase: "idle" };
   private connectPromise: Promise<void> | undefined;
-  private readonly pendingAudio = createRealtimeVoiceAudioQueue("reject-newest");
+  private readonly pendingAudio: RealtimeVoiceAudioQueue;
+  private pendingAudioOverflowReported = false;
 
-  constructor(private readonly label: string) {}
+  constructor(
+    private readonly label: string,
+    options: {
+      pendingAudioOverflowPolicy?: RealtimeVoiceAudioOverflowPolicy;
+      onPendingAudioOverflow?: () => void;
+    } = {},
+  ) {
+    this.pendingAudio = createRealtimeVoiceAudioQueue(
+      options.pendingAudioOverflowPolicy ?? "reject-newest",
+      () => {
+        if (!this.pendingAudioOverflowReported) {
+          this.pendingAudioOverflowReported = true;
+          options.onPendingAudioOverflow?.();
+        }
+      },
+    );
+  }
 
   connect(start: (connection: RealtimeVoiceSessionConnection) => Promise<void>): Promise<void> {
     if (this.isReady()) {
@@ -270,7 +291,7 @@ export class RealtimeVoiceSessionLifecycle {
       return false;
     }
     this.connectPromise = undefined;
-    this.pendingAudio.clear();
+    this.clearPendingAudio();
     if (!("controller" in state)) {
       this.state = { phase: "terminal", terminalOutcome: "completed" };
       return true;
@@ -286,7 +307,7 @@ export class RealtimeVoiceSessionLifecycle {
     if (!state || state.terminalOutcome) {
       return false;
     }
-    this.pendingAudio.clear();
+    this.clearPendingAudio();
     state.phase = "terminal";
     state.terminalOutcome = "error";
     state.controller.abort(new Error(`${this.label} realtime voice session failed`));
@@ -301,7 +322,7 @@ export class RealtimeVoiceSessionLifecycle {
     if (!state) {
       return undefined;
     }
-    this.pendingAudio.clear();
+    this.clearPendingAudio();
     if (!state.terminalOutcome) {
       state.phase = "terminal";
       state.terminalOutcome = outcome;
@@ -346,7 +367,14 @@ export class RealtimeVoiceSessionLifecycle {
   }
 
   drainPendingAudio(): Buffer[] {
-    return this.pendingAudio.drain();
+    const drained = this.pendingAudio.drain();
+    this.pendingAudioOverflowReported = false;
+    return drained;
+  }
+
+  private clearPendingAudio(): void {
+    this.pendingAudio.clear();
+    this.pendingAudioOverflowReported = false;
   }
 
   private createFreshConnection(): RealtimeVoiceSessionConnection {