From 8c4be1bd74830d3a24539b0405f626296dc26456 Mon Sep 17 00:00:00 2001 From: allin2 <37659199+allin2@users.noreply.github.com> Date: Sun, 26 Jul 2026 00:19:51 +0800 Subject: [PATCH 1/6] fix(voice): propagate hard tail receive errors --- .../src/voice/__tests__/voice-spans.test.ts | 35 ++++++++++++------- javascript/src/voice/adapter.runtime.ts | 4 ++- javascript/src/voice/adapter.ts | 8 ++++- javascript/src/voice/adapters/composable.ts | 6 +++- javascript/src/voice/adapters/elevenlabs.ts | 9 +++-- javascript/src/voice/adapters/gemini-live.ts | 9 ++--- .../src/voice/adapters/openai-realtime.ts | 5 ++- javascript/src/voice/adapters/pipecat.ts | 7 +++- javascript/src/voice/adapters/twilio.ts | 7 +++- javascript/src/voice/receive-timeout-error.ts | 19 ++++++++++ 10 files changed, 85 insertions(+), 24 deletions(-) create mode 100644 javascript/src/voice/receive-timeout-error.ts diff --git a/javascript/src/voice/__tests__/voice-spans.test.ts b/javascript/src/voice/__tests__/voice-spans.test.ts index 6689d115c..758ca752e 100644 --- a/javascript/src/voice/__tests__/voice-spans.test.ts +++ b/javascript/src/voice/__tests__/voice-spans.test.ts @@ -6,10 +6,10 @@ * (`receiveAudio`/`sendAudio`) is faked. Mirrors the Python * `test_voice_spans.py` A1/A3/A4/A5/A6/A7. * - * H2 note: TS has no typed timeout, so a first-chunk timeout and a first-chunk - * transport error are indistinguishable at the drain — `voice.audio.receive` - * labels a first-chunk error `first_chunk_timeout` best-effort. The A4-negative - * (non-timeout first error NOT labelled) is Python-only. + * H2 note: the first-chunk path still labels every receive failure + * `first_chunk_timeout` best-effort; only the tail drain distinguishes + * `TimeoutError` from hard failures (#756). The A4-negative (non-timeout first + * error NOT labelled) remains Python-only. */ import { context, trace, SpanStatusCode } from "@opentelemetry/api"; import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; @@ -43,7 +43,7 @@ function tone(seconds: number, transcript = "agent"): AudioChunk { return new AudioChunk({ data, transcript }); } -type RecvAction = AudioChunk | "throw" | "empty"; +type RecvAction = AudioChunk | "throw" | "timeout" | "empty"; class ScriptedAdapter extends VoiceAgentAdapter { override role = AgentRole.AGENT; @@ -75,6 +75,11 @@ class ScriptedAdapter extends VoiceAgentAdapter { const a = this.actions.shift(); if (a === undefined || a === "empty") return silentChunk(0); if (a === "throw") throw new Error("recv failed"); + if (a === "timeout") { + const err = new Error("recv timed out"); + err.name = "TimeoutError"; + throw err; + } return a; } } @@ -106,7 +111,7 @@ describe("voice.* span instrumentation (base runtime)", () => { // A1 it("emits voice.-named spans from a real call()", async () => { - await new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput(tone(0.05))); + await new ScriptedAdapter([tone(0.1), "timeout"]).call(audioInput(tone(0.05))); const names = exporter.getFinishedSpans().map((s) => s.name); expect(names.some((n) => n.startsWith("voice."))).toBe(true); expect(names).toEqual( @@ -117,7 +122,7 @@ describe("voice.* span instrumentation (base runtime)", () => { // A3 it.each([ [["throw"] as RecvAction[], "first_chunk_timeout", true], - [[tone(0.1), "throw"] as RecvAction[], "tail_silence", false], + [[tone(0.1), "timeout"] as RecvAction[], "tail_silence", false], [[tone(0.1), "empty"] as RecvAction[], "terminal_chunk", false], ])("labels terminated_reason=%s#1", async (actions, reason, throws) => { const run = new ScriptedAdapter(actions).call(audioInput(tone(0.05))); @@ -127,6 +132,12 @@ describe("voice.* span instrumentation (base runtime)", () => { expect(recv.attributes["voice.audio.terminated_reason"]).toBe(reason); }); + it("propagates hard errors from the tail-silence receive", async () => { + await expect( + new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput(tone(0.05))), + ).rejects.toThrow("recv failed"); + }); + it("labels max/hard-ceiling as hard_ceiling in TS + sets first_chunk_latency_ms", async () => { const big = tone(20); // 20s each; two crosses the 30s*2 hard ceiling? use several await new ScriptedAdapter([big, big, big, big]).call(audioInput(tone(0.05))); @@ -152,14 +163,14 @@ describe("voice.* span instrumentation (base runtime)", () => { newMessages: [createAudioMessage(tone(0.05), "user")], scenarioState: { currentTurn: 3 }, } as unknown as AgentInput; - await new ScriptedAdapter([tone(0.1), "throw"]).call(input); + await new ScriptedAdapter([tone(0.1), "timeout"]).call(input); const turn = byName(exporter.getFinishedSpans())["voice.turn"]; expect(turn.attributes["voice.turn.index"]).toBe(3); }); // A5 it("nests send/receive under voice.turn and carries turn attrs", async () => { - await new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput(tone(0.05))); + await new ScriptedAdapter([tone(0.1), "timeout"]).call(audioInput(tone(0.05))); const spans = byName(exporter.getFinishedSpans()); const turn = spans["voice.turn"]; expect(turn.attributes["voice.adapter.class"]).toBe("ScriptedAdapter"); @@ -171,12 +182,12 @@ describe("voice.* span instrumentation (base runtime)", () => { // A6 it("emits voice.audio.send with bytes; a no-incoming turn emits none", async () => { const incoming = tone(0.05); - await new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput(incoming)); + await new ScriptedAdapter([tone(0.1), "timeout"]).call(audioInput(incoming)); const send = byName(exporter.getFinishedSpans())["voice.audio.send"]; expect(send.attributes["voice.audio.bytes"]).toBe(incoming.data.length); exporter.reset(); - await new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput()); // no incoming + await new ScriptedAdapter([tone(0.1), "timeout"]).call(audioInput()); // no incoming const names = exporter.getFinishedSpans().map((s) => s.name); expect(names).not.toContain("voice.audio.send"); expect(names).toContain("voice.turn"); @@ -200,7 +211,7 @@ describe("voice.* span instrumentation (base runtime)", () => { // safe via the SDK here; our voiceSpan guard is defense-in-depth and ALSO // guards the raw-throw path Python's use_span leaves open — proven in the // Python A7, where span.end() genuinely propagates and the guard WARNs.) - const result = await new ScriptedAdapter([tone(0.1), "throw"]).call( + const result = await new ScriptedAdapter([tone(0.1), "timeout"]).call( audioInput(tone(0.05)), ); expect(result).toBeTruthy(); diff --git a/javascript/src/voice/adapter.runtime.ts b/javascript/src/voice/adapter.runtime.ts index 1003b34a7..4c206039d 100644 --- a/javascript/src/voice/adapter.runtime.ts +++ b/javascript/src/voice/adapter.runtime.ts @@ -27,6 +27,7 @@ import type { VoiceAgentAdapter } from "./adapter"; import { PendingTransportError } from "./adapters/pending-transport-error"; import { AudioChunk, silentChunk } from "./audio-chunk"; import { createAudioMessage, extractAudio } from "./messages"; +import { isReceiveTimeoutError } from "./receive-timeout-error"; import { VoiceRecordingRuntime } from "./recording.runtime"; import type { AudioSegment, @@ -769,7 +770,8 @@ async function drainInner( let next: AudioChunk; try { next = await adapter.receiveAudio(tailSilence); - } catch { + } catch (err) { + if (!isReceiveTimeoutError(err)) throw err; terminatedReason = "tail_silence"; break; } diff --git a/javascript/src/voice/adapter.ts b/javascript/src/voice/adapter.ts index d63482f5f..6df416312 100644 --- a/javascript/src/voice/adapter.ts +++ b/javascript/src/voice/adapter.ts @@ -151,7 +151,13 @@ export abstract class VoiceAgentAdapter extends AgentAdapter { /** Transmit an {@link AudioChunk} to the agent under test. */ abstract sendAudio(chunk: AudioChunk): Promise; - /** Receive the next {@link AudioChunk} from the agent. */ + /** + * Receive the next {@link AudioChunk} from the agent. + * + * A deadline expiry must reject with an `Error` whose `name` is + * `"TimeoutError"`. The default response drain treats only that signal as + * expected tail silence and propagates every other receive failure. + */ abstract receiveAudio(timeout: number): Promise; /** diff --git a/javascript/src/voice/adapters/composable.ts b/javascript/src/voice/adapters/composable.ts index 5af9ba5f3..766d61692 100644 --- a/javascript/src/voice/adapters/composable.ts +++ b/javascript/src/voice/adapters/composable.ts @@ -28,6 +28,7 @@ import { AgentRole } from "../../domain/agents"; import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { ElevenLabsSTTProvider, type STTProvider, @@ -229,7 +230,10 @@ function withTimeout( message: string, ): Promise { return new Promise((resolve, reject) => { - const handle = setTimeout(() => reject(new Error(message)), timeoutSeconds * 1000); + const handle = setTimeout( + () => reject(new ReceiveTimeoutError(message)), + timeoutSeconds * 1000, + ); promise.then( (value) => { clearTimeout(handle); diff --git a/javascript/src/voice/adapters/elevenlabs.ts b/javascript/src/voice/adapters/elevenlabs.ts index 836169c47..7a34dd611 100644 --- a/javascript/src/voice/adapters/elevenlabs.ts +++ b/javascript/src/voice/adapters/elevenlabs.ts @@ -81,6 +81,7 @@ import { Logger } from "../../utils/logger"; import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes } from "../telemetry"; import { COMPOSABLE_VOICE_LLM_MODEL, @@ -872,7 +873,7 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { const onIdleTimeout = () => { cleanup(); - reject(new Error(idleTimeoutMessage(timeout))); + reject(new ReceiveTimeoutError(idleTimeoutMessage(timeout))); }; const onCeilingTimeout = () => { @@ -881,7 +882,11 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { return; } cleanup(); - reject(new Error(ceilingTimeoutMessage(timeout, ceilingS))); + reject( + new ReceiveTimeoutError( + ceilingTimeoutMessage(timeout, ceilingS), + ), + ); }; // Re-arm the IDLE deadline on every received message (pings included) so a diff --git a/javascript/src/voice/adapters/gemini-live.ts b/javascript/src/voice/adapters/gemini-live.ts index 08a8f9dbf..12551df76 100644 --- a/javascript/src/voice/adapters/gemini-live.ts +++ b/javascript/src/voice/adapters/gemini-live.ts @@ -35,6 +35,7 @@ import { Buffer } from "node:buffer"; import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes } from "../telemetry"; import { GEMINI_LIVE_MODEL } from "../voice-models"; @@ -613,11 +614,11 @@ export class GeminiLiveAgentAdapter extends VoiceAgentAdapter { const timer = setTimeout(() => { // Detach the pending resolver if we're the active waiter. if (this.resolveNext === wrapped) this.resolveNext = null; - const err = new Error( - `GeminiLiveAgentAdapter: no message within ${timeoutMs}ms`, + reject( + new ReceiveTimeoutError( + `GeminiLiveAgentAdapter: no message within ${timeoutMs}ms`, + ), ); - err.name = "TimeoutError"; - reject(err); }, timeoutMs); const wrapped = (item: QueueItem): void => { diff --git a/javascript/src/voice/adapters/openai-realtime.ts b/javascript/src/voice/adapters/openai-realtime.ts index b3a47eed2..798490f3f 100644 --- a/javascript/src/voice/adapters/openai-realtime.ts +++ b/javascript/src/voice/adapters/openai-realtime.ts @@ -34,6 +34,7 @@ import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; import { createAudioMessage, extractAudio } from "../messages"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes, voiceSpan } from "../telemetry"; import { OPENAI_REALTIME_MODEL, OPENAI_STT_MODEL } from "../voice-models"; @@ -1199,7 +1200,9 @@ export class OpenAIRealtimeAgentAdapter extends VoiceAgentAdapter { this._waitResolve = null; this._waitReject = null; reject( - new Error("OpenAIRealtimeAgentAdapter: receiveAudio timed out"), + new ReceiveTimeoutError( + "OpenAIRealtimeAgentAdapter: receiveAudio timed out", + ), ); }, Math.max(0, timeoutMs)); this._waitResolve = (evt) => { diff --git a/javascript/src/voice/adapters/pipecat.ts b/javascript/src/voice/adapters/pipecat.ts index f8d7756e6..52027c51e 100644 --- a/javascript/src/voice/adapters/pipecat.ts +++ b/javascript/src/voice/adapters/pipecat.ts @@ -29,6 +29,7 @@ import { Logger } from "../../utils/logger"; import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes, voiceReceiveSpanUnder } from "../telemetry"; import { sleep } from "../utils"; import { PendingTransportError } from "./pending-transport-error"; @@ -402,7 +403,11 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { return await new Promise((resolve, reject) => { const timer = setTimeout(() => { inbox.waiter = null; - reject(new Error(`PipecatAgentAdapter: receiveAudio timed out after ${timeout}s`)); + reject( + new ReceiveTimeoutError( + `PipecatAgentAdapter: receiveAudio timed out after ${timeout}s`, + ), + ); }, timeout * 1000); inbox.waiter = { resolve: (chunk) => { diff --git a/javascript/src/voice/adapters/twilio.ts b/javascript/src/voice/adapters/twilio.ts index d8d8eed52..af0ed087b 100644 --- a/javascript/src/voice/adapters/twilio.ts +++ b/javascript/src/voice/adapters/twilio.ts @@ -22,6 +22,7 @@ import { AgentRole } from "../../domain/agents"; import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes, voiceSpan } from "../telemetry"; import { sleep } from "../utils"; @@ -760,7 +761,11 @@ class InboundQueue { const timer = setTimeout(() => { const idx = this._waiters.findIndex((w) => w.timer === timer); if (idx >= 0) this._waiters.splice(idx, 1); - reject(new Error(`TwilioAgentAdapter: no audio received within ${timeoutMs}ms`)); + reject( + new ReceiveTimeoutError( + `TwilioAgentAdapter: no audio received within ${timeoutMs}ms`, + ), + ); }, timeoutMs); this._waiters.push({ resolve, reject, timer }); }); diff --git a/javascript/src/voice/receive-timeout-error.ts b/javascript/src/voice/receive-timeout-error.ts new file mode 100644 index 000000000..43ce947ef --- /dev/null +++ b/javascript/src/voice/receive-timeout-error.ts @@ -0,0 +1,19 @@ +/** + * Internal timeout signal shared by voice adapters and the response drain. + * + * `drainAgentResponse` treats only this error (or an adapter-defined error + * named `TimeoutError`) as expected tail silence. Every other receive failure + * must propagate so transport and adapter defects are not hidden as a normal + * end of turn. + */ +export class ReceiveTimeoutError extends Error { + constructor(message: string) { + super(message); + this.name = "TimeoutError"; + } +} + +export function isReceiveTimeoutError(error: unknown): boolean { + return error instanceof ReceiveTimeoutError || + (error instanceof Error && error.name === "TimeoutError"); +} From 8fb2f9bcb058414431ca915ccf1bdd234bd6386e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rog=C3=A9rio=20Chaves?= Date: Wed, 12 Aug 2026 13:57:30 +0200 Subject: [PATCH 2/6] test(voice): pin the receive-timeout contract and drop the first-chunk mislabel The tail-drain fix rested on one assertion that a hard error rejects with a given message. That leaves the two things worth proving untested: that the error reaches the caller intact rather than as a truncated turn, and that the built-in adapters still produce a rejection the drain reads as a deadline. The second gap is the dangerous one. An adapter that rejects its deadline with a plain Error is now read as a hard failure, so a normal end of turn aborts the run, and nothing in the suite notices because each adapter test asserts its own message rather than the class. - Add specs/voice-drain-error-propagation.feature. - Assert semantics on the tail split: the original error object reaches the caller with its class and fields, no agent messages are produced, the receive span is ERROR and not labelled tail_silence, a deadline still keeps the audio collected so far, and a deliberate agent hangup stays a clean end of turn. - Add an adapter contract test driving ElevenLabs, Pipecat and Gemini Live to their real deadlines, faked at the network-client boundary. - Label first_chunk_timeout only on an actual deadline, so a transport that dies before the first chunk is no longer traced as a quiet agent. Matches Python, and the comment above the catch already claimed this behaviour. - Classify by error name alone, so the DOMException from AbortSignal.timeout() qualifies on any runtime and a doubly loaded module cannot defeat instanceof. - Document both turn-ending signals on receiveAudio: a TimeoutError-named rejection, or an empty chunk for a terminal condition that is not a failure. --- .../drain-tail-error-propagation.test.ts | 257 ++++++++++++++++++ .../src/voice/__tests__/voice-spans.test.ts | 25 +- javascript/src/voice/adapter.runtime.ts | 17 +- javascript/src/voice/adapter.ts | 13 +- .../receive-timeout-contract.test.ts | 141 ++++++++++ javascript/src/voice/receive-timeout-error.ts | 32 ++- specs/voice-drain-error-propagation.feature | 82 ++++++ 7 files changed, 543 insertions(+), 24 deletions(-) create mode 100644 javascript/src/voice/__tests__/drain-tail-error-propagation.test.ts create mode 100644 javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts create mode 100644 specs/voice-drain-error-propagation.feature diff --git a/javascript/src/voice/__tests__/drain-tail-error-propagation.test.ts b/javascript/src/voice/__tests__/drain-tail-error-propagation.test.ts new file mode 100644 index 000000000..80b4a88a9 --- /dev/null +++ b/javascript/src/voice/__tests__/drain-tail-error-propagation.test.ts @@ -0,0 +1,257 @@ +/** + * Binds `specs/voice-drain-error-propagation.feature` (#756). + * + * The response drain ends a turn when the agent stops talking, and it learns + * that from a rejected `receiveAudio`. It used to end the turn on ANY rejection, + * so a dead transport, a misconfigured adapter or a tripped assertion was + * reported as a short but successful agent turn. That is the worst failure mode + * for a test framework: the scenario keeps running and asserts against a turn + * that never happened. It is what hid the #697 P0 through CI, five rounds of + * automated review and a human reproduction. + * + * These tests pin the split. A receive deadline still closes the turn cleanly + * and keeps the audio collected so far; anything else reaches the caller with + * its identity intact and produces NO agent messages. + * + * Mic-free and clock-free: the scripted adapter decides each `receiveAudio` + * outcome, so nothing here waits on a real deadline. + * + * Run with `pnpm test src/voice/__tests__/drain-tail-error-propagation.test.ts` + * from `javascript/`. + */ +import { context, trace, SpanStatusCode } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { + InMemorySpanExporter, + SimpleSpanProcessor, + type ReadableSpan, +} from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import { describe, it, expect, beforeEach, afterEach } from "vitest"; + +// Register a context manager ONCE so context.with propagates across awaits. +const _ctxManager = new AsyncLocalStorageContextManager(); +_ctxManager.enable(); +context.setGlobalContextManager(_ctxManager); + +import { AgentRole, type AgentInput } from "../../domain/agents"; +import { VoiceAgentAdapter } from "../adapter"; +import { AudioChunk } from "../audio-chunk"; +import { AdapterCapabilities } from "../capabilities"; +import { createAudioMessage } from "../messages"; +import { ReceiveTimeoutError } from "../receive-timeout-error"; + +const SR = 24000; // PCM16 mono 24kHz + +function tone(seconds: number, transcript = "agent"): AudioChunk { + const data = new Uint8Array(Math.round(seconds * SR) * 2); + for (let i = 0; i < data.length; i++) data[i] = (i % 250) + 1; + return new AudioChunk({ data, transcript }); +} + +/** The empty chunk every adapter uses to mean "end of stream, cleanly". */ +function terminal(): AudioChunk { + return new AudioChunk({ data: new Uint8Array(0) }); +} + +/** A transport failure of the kind the drain must never absorb. */ +class DeadTransportError extends Error { + readonly streamId = "stream-42"; + constructor() { + super("no live media stream"); + this.name = "DeadTransportError"; + } +} + +/** What a scripted `receiveAudio` call does: yield a chunk, or reject with this. */ +type RecvAction = AudioChunk | { rejectWith: unknown }; + +class ScriptedAdapter extends VoiceAgentAdapter { + override role = AgentRole.AGENT; + readonly capabilities = new AdapterCapabilities({ + streamingTranscripts: false, + nativeVad: true, + dtmf: false, + interruption: false, + inputFormats: ["pcm16/24000"], + outputFormats: ["pcm16/24000"], + }); + lastAgentTranscript: string | null = "agent"; + receiveCalls = 0; + private actions: RecvAction[]; + + constructor(actions: RecvAction[]) { + super(); + this.actions = [...actions]; + } + override isConnected(): boolean { + return true; + } + async connect(): Promise {} + async disconnect(): Promise {} + async sendAudio(_chunk: AudioChunk): Promise {} + async receiveAudio(_timeout: number): Promise { + this.receiveCalls += 1; + const action = this.actions.shift(); + // Running off the end means the drain asked for more than the script + // covers; a terminal chunk ends the turn rather than hanging the suite. + if (action === undefined) return terminal(); + if (action instanceof AudioChunk) return action; + throw action.rejectWith; + } +} + +function audioInput(incoming?: AudioChunk): AgentInput { + const newMessages = incoming ? [createAudioMessage(incoming, "user")] : []; + return { newMessages } as unknown as AgentInput; +} + +function byName(spans: ReadableSpan[]): Record { + return Object.fromEntries(spans.map((s) => [s.name, s])); +} + +describe("drain tail-receive error classification (#756)", () => { + let exporter: InMemorySpanExporter; + let provider: NodeTracerProvider; + + beforeEach(() => { + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + trace.setGlobalTracerProvider(provider); + }); + afterEach(async () => { + await provider.shutdown(); + trace.disable(); + }); + + describe("a hard error reaches the caller", () => { + it("rejects call() with the adapter's own error object, unwrapped", async () => { + const boom = new DeadTransportError(); + const adapter = new ScriptedAdapter([tone(0.1), { rejectWith: boom }]); + + // Identity, not just the message: the caller needs the original class, + // its custom fields and its stack to diagnose the transport. + const caught = await adapter.call(audioInput(tone(0.05))).then( + () => undefined, + (err: unknown) => err, + ); + expect(caught).toBe(boom); + expect(caught).toBeInstanceOf(DeadTransportError); + expect((caught as DeadTransportError).streamId).toBe("stream-42"); + expect((caught as Error).stack).toBeDefined(); + }); + + it("yields no agent turn, so the run cannot assert against audio the agent never sent", async () => { + const adapter = new ScriptedAdapter([ + tone(0.1), + { rejectWith: new DeadTransportError() }, + ]); + + // The pre-fix behaviour: call() RESOLVED here with the 0.1s of audio + // collected before the failure, and the scenario scored that truncated + // turn as a real one. + let resolvedWith: unknown = "did-not-resolve"; + await adapter.call(audioInput(tone(0.05))).then( + (value) => { + resolvedWith = value; + }, + () => {}, + ); + expect(resolvedWith).toBe("did-not-resolve"); + }); + + it("marks the receive span ERROR without claiming tail silence", async () => { + const adapter = new ScriptedAdapter([ + tone(0.1), + { rejectWith: new DeadTransportError() }, + ]); + await expect(adapter.call(audioInput(tone(0.05)))).rejects.toThrow( + "no live media stream", + ); + + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.status.code).toBe(SpanStatusCode.ERROR); + // The whole point: the trace must not read as a turn that ended normally. + expect(recv.attributes["voice.audio.terminated_reason"]).not.toBe("tail_silence"); + }); + + it("propagates a thrown non-Error value instead of reading it as a deadline", async () => { + // A `throw "string"` has no `name`, so the classifier must reject it. + const adapter = new ScriptedAdapter([tone(0.1), { rejectWith: "recv exploded" }]); + await expect(adapter.call(audioInput(tone(0.05)))).rejects.toBe("recv exploded"); + }); + + it("stops draining at the failure rather than swallowing and retrying", async () => { + const adapter = new ScriptedAdapter([ + tone(0.1), + { rejectWith: new DeadTransportError() }, + tone(0.1), + ]); + await expect(adapter.call(audioInput(tone(0.05)))).rejects.toThrow(); + expect(adapter.receiveCalls).toBe(2); + }); + }); + + describe("a receive deadline still ends the turn cleanly", () => { + it("closes the drain on the shared ReceiveTimeoutError and keeps the audio so far", async () => { + const adapter = new ScriptedAdapter([ + tone(0.1), + tone(0.1), + { rejectWith: new ReceiveTimeoutError("no audio received within 600ms") }, + ]); + + const messages = await adapter.call(audioInput(tone(0.05))); + + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.attributes["voice.audio.terminated_reason"]).toBe("tail_silence"); + expect(recv.status.code).not.toBe(SpanStatusCode.ERROR); + // Both chunks survive: a deadline ends the turn, it does not discard it. + expect(recv.attributes["voice.audio.chunk_count"]).toBe(2); + expect(messages).toBeTruthy(); + }); + + it("accepts a custom adapter's own error named TimeoutError", async () => { + // The documented contract for third-party adapters: no import from us. + const err = new Error("MyAdapter: receiveAudio timed out"); + err.name = "TimeoutError"; + const adapter = new ScriptedAdapter([tone(0.1), { rejectWith: err }]); + + await adapter.call(audioInput(tone(0.05))); + + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.attributes["voice.audio.terminated_reason"]).toBe("tail_silence"); + }); + + it("accepts the DOMException AbortSignal.timeout() rejects with", async () => { + // A custom adapter built on the platform primitive gets this for free, and + // a DOMException is not an Error subclass on every runtime. + const adapter = new ScriptedAdapter([ + tone(0.1), + { rejectWith: new DOMException("The operation timed out.", "TimeoutError") }, + ]); + + await adapter.call(audioInput(tone(0.05))); + + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.attributes["voice.audio.terminated_reason"]).toBe("tail_silence"); + }); + + it("keeps a deliberate agent hangup a clean end of turn, not a failure", async () => { + // #839/#849: an agent that hangs up wakes the parked receive with the + // empty end-of-stream chunk rather than throwing, so narrowing the catch + // must not turn correct agent behaviour into a failed run. + const adapter = new ScriptedAdapter([tone(0.1), terminal()]); + adapter.agentHungUp = true; + + const messages = await adapter.call(audioInput(tone(0.05))); + + expect(messages).toBeTruthy(); + expect(adapter.agentHungUp).toBe(true); + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.attributes["voice.audio.terminated_reason"]).toBe("terminal_chunk"); + expect(recv.status.code).not.toBe(SpanStatusCode.ERROR); + }); + }); +}); diff --git a/javascript/src/voice/__tests__/voice-spans.test.ts b/javascript/src/voice/__tests__/voice-spans.test.ts index 758ca752e..3367dd219 100644 --- a/javascript/src/voice/__tests__/voice-spans.test.ts +++ b/javascript/src/voice/__tests__/voice-spans.test.ts @@ -6,10 +6,10 @@ * (`receiveAudio`/`sendAudio`) is faked. Mirrors the Python * `test_voice_spans.py` A1/A3/A4/A5/A6/A7. * - * H2 note: the first-chunk path still labels every receive failure - * `first_chunk_timeout` best-effort; only the tail drain distinguishes - * `TimeoutError` from hard failures (#756). The A4-negative (non-timeout first - * error NOT labelled) remains Python-only. + * Both receive paths now read the typed timeout signal (#756), so the + * A4-negative — a non-timeout first-chunk error is NOT labelled + * `first_chunk_timeout` — holds in TypeScript too and is asserted below. + * `drain-tail-error-propagation.test.ts` covers the propagation itself. */ import { context, trace, SpanStatusCode } from "@opentelemetry/api"; import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; @@ -121,7 +121,7 @@ describe("voice.* span instrumentation (base runtime)", () => { // A3 it.each([ - [["throw"] as RecvAction[], "first_chunk_timeout", true], + [["timeout"] as RecvAction[], "first_chunk_timeout", true], [[tone(0.1), "timeout"] as RecvAction[], "tail_silence", false], [[tone(0.1), "empty"] as RecvAction[], "terminal_chunk", false], ])("labels terminated_reason=%s#1", async (actions, reason, throws) => { @@ -132,10 +132,15 @@ describe("voice.* span instrumentation (base runtime)", () => { expect(recv.attributes["voice.audio.terminated_reason"]).toBe(reason); }); - it("propagates hard errors from the tail-silence receive", async () => { + // A4-negative: a transport that dies before the first chunk is an ERROR span + // with no terminated_reason, never a turn the agent chose not to take. + it("does not label a hard first-chunk error as first_chunk_timeout", async () => { await expect( - new ScriptedAdapter([tone(0.1), "throw"]).call(audioInput(tone(0.05))), + new ScriptedAdapter(["throw"]).call(audioInput(tone(0.05))), ).rejects.toThrow("recv failed"); + const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; + expect(recv.attributes["voice.audio.terminated_reason"]).toBeUndefined(); + expect(recv.status.code).toBe(SpanStatusCode.ERROR); }); it("labels max/hard-ceiling as hard_ceiling in TS + sets first_chunk_latency_ms", async () => { @@ -147,10 +152,10 @@ describe("voice.* span instrumentation (base runtime)", () => { expect(recv.attributes["voice.audio.chunk_count"]).toBeGreaterThanOrEqual(1); }); - // A4 (positive; H2 note above — negative is Python-only) - it("marks voice.audio.receive ERROR + first_chunk_timeout on a first-chunk failure", async () => { + // A4 (positive; the negative is the test above) + it("marks voice.audio.receive ERROR + first_chunk_timeout on a first-chunk timeout", async () => { await expect( - new ScriptedAdapter(["throw"]).call(audioInput(tone(0.05))), + new ScriptedAdapter(["timeout"]).call(audioInput(tone(0.05))), ).rejects.toThrow(); const recv = byName(exporter.getFinishedSpans())["voice.audio.receive"]; expect(recv.status.code).toBe(SpanStatusCode.ERROR); diff --git a/javascript/src/voice/adapter.runtime.ts b/javascript/src/voice/adapter.runtime.ts index 4c206039d..5c891d864 100644 --- a/javascript/src/voice/adapter.runtime.ts +++ b/javascript/src/voice/adapter.runtime.ts @@ -722,14 +722,18 @@ async function drainInner( // The runaway backstop that replaces the old mid-utterance chop (#747). const hardCeiling = maxDuration * MAX_DURATION_CEILING_FACTOR; - // H2: TS has no typed timeout and this first receiveAudio was uncaught. Catch - // to attribute first_chunk_timeout (and let the guard set ERROR + re-throw); - // a NON-timeout first-chunk error is NOT labelled a timeout. + // Every first-chunk failure propagates — the catch only attributes it, and the + // span guard turns it into an ERROR span. A hard failure keeps + // `terminated_reason` unset rather than claiming a timeout, so a trace never + // reports a dead transport as an agent that stayed quiet. Mirrors Python's + // `_drain_agent_response`, which labels only `asyncio.TimeoutError` here. let first: AudioChunk; try { first = await adapter.receiveAudio(responseTimeout); } catch (err) { - span.setAttribute("voice.audio.terminated_reason", "first_chunk_timeout"); + if (isReceiveTimeoutError(err)) { + span.setAttribute("voice.audio.terminated_reason", "first_chunk_timeout"); + } throw err; } span.setAttribute( @@ -771,6 +775,11 @@ async function drainInner( try { next = await adapter.receiveAudio(tailSilence); } catch (err) { + // Only a receive deadline ends the turn. A dead transport, a + // misconfigured adapter or a tripped assertion propagates with its + // original message and stack, because a swallowed hard error downgrades a + // loud crash into a turn that silently reports audio the agent never sent + // (#756). if (!isReceiveTimeoutError(err)) throw err; terminatedReason = "tail_silence"; break; diff --git a/javascript/src/voice/adapter.ts b/javascript/src/voice/adapter.ts index 6df416312..88cc91471 100644 --- a/javascript/src/voice/adapter.ts +++ b/javascript/src/voice/adapter.ts @@ -154,9 +154,16 @@ export abstract class VoiceAgentAdapter extends AgentAdapter { /** * Receive the next {@link AudioChunk} from the agent. * - * A deadline expiry must reject with an `Error` whose `name` is - * `"TimeoutError"`. The default response drain treats only that signal as - * expected tail silence and propagates every other receive failure. + * Expiry of `timeout` MUST reject with an error whose `name` is + * `"TimeoutError"` — what `AbortSignal.timeout()` already produces, and what + * the built-in adapters get from `ReceiveTimeoutError`. That is the only + * rejection the default drain reads as the end of a turn; every other one + * propagates and fails the run, so a dead transport is never reported as an + * agent that simply stopped talking (#756). + * + * Returning an empty {@link AudioChunk} also ends the turn, cleanly. Use it + * for a terminal condition that is not a failure, such as the agent hanging + * up or the peer closing the stream after it finished speaking. */ abstract receiveAudio(timeout: number): Promise; diff --git a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts new file mode 100644 index 000000000..5131b84b1 --- /dev/null +++ b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts @@ -0,0 +1,141 @@ +/** + * Binds `specs/voice-drain-error-propagation.feature` (#756). + * + * The response drain ends a turn only on a receive deadline and propagates + * every other `receiveAudio` rejection. That split is worth nothing unless the + * adapters hold up their half: an adapter whose deadline rejects with a plain + * `Error` is read as a hard failure, so a perfectly normal end of turn crashes + * the run. + * + * Nothing about a plain `new Error(...)` at a `setTimeout` looks wrong in + * review, and no per-adapter test notices, because each one asserts its own + * message rather than the class. So these tests drive each adapter to its real + * deadline and assert only the drain's question: is this a receive timeout? + * + * **Adding a timeout site?** Reject with {@link ReceiveTimeoutError} and add the + * adapter here. Faked at the network-client boundary — a socket factory or the + * vendor SDK module — never by assigning adapter privates. + * + * Run with `pnpm test src/voice/adapters/__tests__/receive-timeout-contract.test.ts` + * from `javascript/`. + */ +import { EventEmitter } from "node:events"; + +import { describe, it, expect, vi } from "vitest"; + +import { isReceiveTimeoutError } from "../../receive-timeout-error"; +import { ElevenLabsAgentAdapter } from "../elevenlabs"; +import { GeminiLiveAgentAdapter } from "../gemini-live"; +import { PipecatAgentAdapter, type PipecatWebSocketLike } from "../pipecat"; +import { makeFakeConv } from "./fixtures/fake-elevenlabs-conversation"; + +// Mock the Gemini SDK so connect() never opens a real WebSocket. Scoped to this +// module, and the other adapters under test do not import it. +vi.mock("@google/genai", () => { + class FakeSession { + sendRealtimeInput = vi.fn(); + close = vi.fn(); + } + return { + Modality: { AUDIO: "AUDIO" }, + GoogleGenAI: class { + live = { connect: async () => new FakeSession() }; + constructor(_init: { apiKey?: string }) {} + }, + }; +}); + +/** + * Short enough to keep the suite fast, long enough that the adapter has really + * parked a waiter rather than failing some precondition on the way in. + */ +const DEADLINE_S = 0.05; + +/** Minimal Pipecat socket: opens, accepts frames, never delivers audio. */ +class SilentPipecatSocket extends EventEmitter implements PipecatWebSocketLike { + send(_data: string | Uint8Array): void {} + close(): void {} +} + +function silentPipecatSocket(): SilentPipecatSocket { + const socket = new SilentPipecatSocket(); + // The adapter registers `once('open')` during connect(); emit after that. + queueMicrotask(() => socket.emit("open")); + return socket; +} + +/** + * Each entry parks a `receiveAudio` against a transport that stays open and + * silent, so the adapter's own deadline is what ends the wait. + */ +const ADAPTERS: Array<{ + name: string; + parkReceive: () => Promise<{ receive: Promise; teardown: () => Promise }>; +}> = [ + { + name: "ElevenLabsAgentAdapter", + parkReceive: async () => { + const fake = makeFakeConv(); + const adapter = new ElevenLabsAgentAdapter({ + agentId: "agt-timeout-contract", + apiKey: "sk-timeout-contract", + webSocketFactory: fake.webSocketFactory, + conversationClient: fake.conversationClient, + }); + await adapter.connect(); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: () => adapter.disconnect(), + }; + }, + }, + { + name: "PipecatAgentAdapter", + parkReceive: async () => { + const adapter = new PipecatAgentAdapter({ + url: "ws://pipecat.test/ws", + streamSid: "MZtimeoutcontract", + realTimePacing: false, + webSocketFactory: () => silentPipecatSocket(), + }); + await adapter.connect(); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: () => adapter.disconnect(), + }; + }, + }, + { + name: "GeminiLiveAgentAdapter", + parkReceive: async () => { + const adapter = new GeminiLiveAgentAdapter({ apiKey: "test-key" }); + await adapter.connect(); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: () => adapter.disconnect(), + }; + }, + }, +]; + +describe("built-in adapters reject their receive deadline as a receive timeout (#756)", () => { + it.each(ADAPTERS)("$name", async ({ parkReceive }) => { + const { receive, teardown } = await parkReceive(); + try { + const error = await receive.then( + () => undefined, + (err: unknown) => err, + ); + + expect(error, "the deadline did not fire").toBeDefined(); + expect( + isReceiveTimeoutError(error), + `the drain reads this as a HARD failure and will abort the run: ${String(error)}`, + ).toBe(true); + // The diagnosis the user reads still has to survive the classification. + expect(String((error as Error).message)).not.toBe(""); + } finally { + await teardown(); + } + }); +}); diff --git a/javascript/src/voice/receive-timeout-error.ts b/javascript/src/voice/receive-timeout-error.ts index 43ce947ef..d3e7b7e3d 100644 --- a/javascript/src/voice/receive-timeout-error.ts +++ b/javascript/src/voice/receive-timeout-error.ts @@ -1,19 +1,37 @@ /** - * Internal timeout signal shared by voice adapters and the response drain. + * The receive-deadline signal shared by every built-in voice adapter and the + * response drain (#756). * - * `drainAgentResponse` treats only this error (or an adapter-defined error - * named `TimeoutError`) as expected tail silence. Every other receive failure - * must propagate so transport and adapter defects are not hidden as a normal - * end of turn. + * The drain has to tell "the agent stopped talking" from "the transport died", + * and both reach it as a rejected `receiveAudio`. Only a receive timeout ends a + * turn; anything else is a real failure and must propagate with its original + * message and stack, or a dead transport is reported as a short but successful + * agent turn. */ export class ReceiveTimeoutError extends Error { constructor(message: string) { super(message); + // `TimeoutError` rather than the class name: it is the web platform's own + // name for this condition, so custom adapters that already reject with + // `AbortSignal.timeout()` or a hand-rolled timeout satisfy the contract + // without importing anything from us. this.name = "TimeoutError"; } } +/** + * Whether `error` is the "no audio within the deadline" signal that ends a turn. + * + * Matched on `name`, not on the class: `AbortSignal.timeout()` rejects with a + * `DOMException`, which is only an `Error` subclass on some runtimes, and a + * bundle that loads this module twice would defeat `instanceof` on our own + * class. The name is the contract documented on + * {@link VoiceAgentAdapter.receiveAudio}, so read exactly that. + */ export function isReceiveTimeoutError(error: unknown): boolean { - return error instanceof ReceiveTimeoutError || - (error instanceof Error && error.name === "TimeoutError"); + return ( + typeof error === "object" && + error !== null && + (error as { name?: unknown }).name === "TimeoutError" + ); } diff --git a/specs/voice-drain-error-propagation.feature b/specs/voice-drain-error-propagation.feature new file mode 100644 index 000000000..3a1cd6852 --- /dev/null +++ b/specs/voice-drain-error-propagation.feature @@ -0,0 +1,82 @@ +Feature: The voice drain ends a turn on silence, not on failure + As a developer running a voice scenario + I want a dead transport or a broken adapter to fail my run loudly + So that a scenario never passes on a turn the agent never took + + # ROOT CAUSE (issue #756, verified on main): drainAgentResponse wrapped its + # tail-silence probe in a bare `catch { break }`. That cannot tell "no audio + # within responseTailSilence" — how every turn ends — from a hard error, so + # both closed the turn. In #697 the Twilio adapter threw "no live media + # stream" on the drain's follow-up call after the media-stream transport was + # torn down; Python surfaced it immediately (its drain catches only + # asyncio.TimeoutError) while TypeScript swallowed it and merely TRUNCATED + # the turn. The defect stayed invisible through CI, five rounds of automated + # review, and the original human reproduction. A swallowed hard error + # degrades a loud crash into silent data loss, which for a test framework is + # the worse failure: the scenario keeps running and asserts against a turn + # that never happened. + + Background: + Given a voice adapter whose receiveAudio the drain calls to collect a turn + + # AC1 — the fix: hard errors reach the caller. + @unit @ts-voice-drain + Scenario: A hard error from the tail-silence receive fails the turn + Given the agent has already sent one audio chunk + And the next receiveAudio rejects with a transport failure + When defaultVoiceCall drains the agent response + Then call() rejects with that same error object, unwrapped + And no agent messages are produced for the truncated turn + And the voice.audio.receive span is ERROR and is not labelled tail_silence + + # AC2 — no regression: silence is still how a turn ends. + @unit @ts-voice-drain + Scenario: A receive deadline closes the turn and keeps the audio collected + Given the agent has already sent two audio chunks + And the next receiveAudio rejects with a receive timeout + When defaultVoiceCall drains the agent response + Then the turn completes successfully with both chunks + And the voice.audio.receive span is labelled tail_silence + + # AC3 — the contract a custom adapter has to meet, with no import from us. + @unit @ts-voice-drain + Scenario Outline: Any error named TimeoutError is read as a receive deadline + Given the agent has already sent one audio chunk + And the next receiveAudio rejects with + When defaultVoiceCall drains the agent response + Then the turn completes successfully + And the voice.audio.receive span is labelled tail_silence + + Examples: + | rejection | + | the shared ReceiveTimeoutError | + | a custom Error whose name is TimeoutError | + | the DOMException AbortSignal.timeout() raises | + + # AC4 — every built-in adapter has to hold up its half of AC3, or a normal + # end of turn crashes the run. + @unit @ts-voice-drain + Scenario: A built-in adapter's own deadline is classified as a receive timeout + Given a built-in adapter connected to a transport that stays open and silent + When its receiveAudio deadline expires + Then the rejection is classified as a receive timeout + And the rejection still carries a diagnosis the developer can read + + # AC5 — the same distinction on the sibling call, matching Python's drain, + # which labels only asyncio.TimeoutError there. + @unit @ts-voice-drain + Scenario: A hard error before the first chunk is not attributed to a timeout + Given the first receiveAudio of the turn rejects with a transport failure + When defaultVoiceCall drains the agent response + Then the voice.audio.receive span is ERROR with no terminated_reason + And the span does not claim first_chunk_timeout + + # AC6 — #839/#849: an agent that hangs up wakes the parked receive with the + # empty end-of-stream chunk, so narrowing the catch must not punish it. + @unit @ts-voice-drain + Scenario: A deliberate agent hangup remains a clean end of turn + Given the agent has already sent one audio chunk + And the agent hung up, so the next receiveAudio returns an empty chunk + When defaultVoiceCall drains the agent response + Then the turn completes successfully with agentHungUp still set + And the voice.audio.receive span is labelled terminal_chunk From e41e4bd2219175be1ad8c25d04f803ce85754b25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rog=C3=A9rio=20Chaves?= Date: Wed, 12 Aug 2026 14:08:43 +0200 Subject: [PATCH 3/6] fix(voice): end a Pipecat turn when the bot closes the stream Narrowing the drain's catch turned an ordinary end of call into a failed run. Pipecat rejected every receive once its socket closed, and the broad catch used to absorb that. Now it propagates, so a bot that hangs up after speaking fails the scenario with "socket closed, no audio available". ElevenLabs (#648), OpenAI Realtime (#646) and Twilio (#695) all converged on the same answer: a stream that has ended yields the empty terminal chunk, and the shared drain exits cleanly. Pipecat was the last one still rejecting. - Resolve a parked receive with the terminal chunk on socket close, and keep returning it once the stream has ended. - Wake a parked receive on disconnect() the same way, matching #849. - Record a socket ERROR on the inbox so it keeps failing every later receive, carrying the underlying message and cause. A broken transport is not an end of turn however many times it is asked. - Cover all four paths, including a real call() whose stream ends mid-drain. --- .../receive-timeout-contract.test.ts | 93 +++++++++++++++++++ javascript/src/voice/adapters/pipecat.ts | 39 ++++++-- 2 files changed, 124 insertions(+), 8 deletions(-) diff --git a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts index 5131b84b1..e4f4fbbf0 100644 --- a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts +++ b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts @@ -19,10 +19,13 @@ * Run with `pnpm test src/voice/adapters/__tests__/receive-timeout-contract.test.ts` * from `javascript/`. */ +import { Buffer } from "node:buffer"; import { EventEmitter } from "node:events"; import { describe, it, expect, vi } from "vitest"; +import { AgentRole } from "../../../domain/agents"; +import { makeAgentInput } from "../../__tests__/helpers/drive-production"; import { isReceiveTimeoutError } from "../../receive-timeout-error"; import { ElevenLabsAgentAdapter } from "../elevenlabs"; import { GeminiLiveAgentAdapter } from "../gemini-live"; @@ -139,3 +142,93 @@ describe("built-in adapters reject their receive deadline as a receive timeout ( } }); }); + +/** + * The other half of the contract. A stream that ENDS is how a call finishes, so + * it has to reach the drain as the empty end-of-stream chunk. Rejecting instead + * used to be invisible, because the drain absorbed every rejection; now it + * aborts the run. Pipecat was the last adapter still rejecting — ElevenLabs + * (#648), OpenAI Realtime (#646) and Twilio (#695) already converged here. + */ +describe("a stream that ends is an end of turn, not a failure (#756)", () => { + async function connectedPipecat(): Promise<{ + adapter: PipecatAgentAdapter; + socket: SilentPipecatSocket; + }> { + let socket!: SilentPipecatSocket; + const adapter = new PipecatAgentAdapter({ + url: "ws://pipecat.test/ws", + streamSid: "MZstreamend", + realTimePacing: false, + webSocketFactory: () => { + socket = silentPipecatSocket(); + return socket; + }, + }); + await adapter.connect(); + return { adapter, socket }; + } + + it("hands a parked receive the terminal chunk when the bot closes the socket", async () => { + const { adapter, socket } = await connectedPipecat(); + const receive = adapter.receiveAudio(30); + await Promise.resolve(); + + socket.emit("close"); + + const chunk = await receive; + expect(chunk.data.length).toBe(0); + }); + + it("keeps returning the terminal chunk once the stream has ended", async () => { + // The drain's tail probe often lands AFTER the close rather than during it, + // which is the race that made this throw. + const { adapter, socket } = await connectedPipecat(); + socket.emit("close"); + + const chunk = await adapter.receiveAudio(30); + expect(chunk.data.length).toBe(0); + }); + + it("still fails every receive after a socket error", async () => { + const { adapter, socket } = await connectedPipecat(); + socket.emit("error", new Error("ECONNRESET")); + + // Now, and on the drain's next probe: a broken transport is not an end of + // turn however many times it is asked. + for (const attempt of ["first", "second"]) { + const error = await adapter.receiveAudio(30).then( + () => undefined, + (err: unknown) => err, + ); + expect(error, `${attempt} receive resolved after a socket error`).toBeInstanceOf(Error); + expect((error as Error).message).toContain("ECONNRESET"); + expect(isReceiveTimeoutError(error)).toBe(false); + } + }); + + it("completes a real call() whose stream ends mid-drain", async () => { + // The end-to-end shape of the regression: the bot speaks, then the call + // ends while the drain is between its first chunk and its tail probe. This + // drove the production wrapper straight to + // "socket closed, no audio available". + const { adapter, socket } = await connectedPipecat(); + adapter.role = AgentRole.AGENT; + + const payload = Buffer.alloc(160, 0xff).toString("base64"); + socket.emit( + "message", + Buffer.from( + JSON.stringify({ + event: "media", + streamSid: "MZstreamend", + media: { payload }, + }), + ), + ); + setTimeout(() => socket.emit("close"), 20); + + const messages = await adapter.call(makeAgentInput()); + expect(messages).toBeTruthy(); + }); +}); diff --git a/javascript/src/voice/adapters/pipecat.ts b/javascript/src/voice/adapters/pipecat.ts index 52027c51e..de0fb8834 100644 --- a/javascript/src/voice/adapters/pipecat.ts +++ b/javascript/src/voice/adapters/pipecat.ts @@ -108,6 +108,13 @@ interface AudioInbox { queue: AudioChunk[]; waiter: { resolve: (chunk: AudioChunk) => void; reject: (err: Error) => void } | null; closed: boolean; + /** + * Set when the socket ERRORED rather than ending. A stream that ends is a + * normal end of turn and yields the empty terminal chunk; a stream that broke + * has to keep failing every later receive, or the drain reads a transport + * fault as the agent finishing (#756). + */ + failure: Error | null; } /** @@ -251,7 +258,7 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { const factory = this.webSocketFactory ?? (await defaultWebSocketFactory()); const ws = factory(this.url); this.ws = ws; - this.inbox = { queue: [], waiter: null, closed: false }; + this.inbox = { queue: [], waiter: null, closed: false, failure: null }; this.mulawChunks = []; this.mulawChunksByteLength = 0; @@ -329,9 +336,10 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { if (this.inbox) { this.inbox.closed = true; - this.inbox.waiter?.reject( - new Error("PipecatAgentAdapter: disconnected while waiting for audio"), - ); + // Wake a receive already parked on this inbox before dropping it, with + // the empty end-of-stream chunk rather than a rejection: we are the ones + // tearing the call down, so the drain should finish, not fail (#849). + this.inbox.waiter?.resolve(new AudioChunk({ data: new Uint8Array(0) })); this.inbox = null; } @@ -397,8 +405,13 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { } const queued = inbox.queue.shift(); if (queued) return queued; + if (inbox.failure) throw inbox.failure; if (inbox.closed) { - throw new Error("PipecatAgentAdapter: socket closed, no audio available"); + // The bot closed the stream and everything it sent has been consumed: + // an end of turn, not a failure. The empty chunk is this codebase's + // end-of-stream signal, so the shared drain exits cleanly rather than + // waiting out the deadline (#648/#646/#695) or aborting the run (#756). + return new AudioChunk({ data: new Uint8Array(0) }); } return await new Promise((resolve, reject) => { const timer = setTimeout(() => { @@ -558,13 +571,19 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { } } - private onSocketError(_err: Error): void { + private onSocketError(err: Error): void { if (!this.inbox) return; this.inbox.closed = true; + // A broken transport must reach the caller, both now and on every later + // receive, so the drain never mistakes it for the agent finishing (#756). + this.inbox.failure = new Error( + `PipecatAgentAdapter: socket error: ${err.message}`, + { cause: err }, + ); const waiter = this.inbox.waiter; if (waiter) { this.inbox.waiter = null; - waiter.reject(new Error("PipecatAgentAdapter: socket error")); + waiter.reject(this.inbox.failure); } } @@ -575,7 +594,11 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { const waiter = this.inbox.waiter; if (waiter) { this.inbox.waiter = null; - waiter.reject(new Error("PipecatAgentAdapter: socket closed")); + // The bot closed the stream while the drain was parked. That is how a + // call ends, so hand over the empty end-of-stream chunk and let the turn + // finish; rejecting here would fail a run in which nothing went wrong. + // A socket that ERRORED took the branch above and still rejects. + waiter.resolve(new AudioChunk({ data: new Uint8Array(0) })); } } From 7af36dde1aaa38f71b7b7dc71ce71dcff7460f7d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rog=C3=A9rio=20Chaves?= Date: Wed, 12 Aug 2026 14:20:12 +0200 Subject: [PATCH 4/6] fix(voice): propagate hard errors from the realtime and ElevenLabs receive paths Three gaps found in review, all the same shape as #756. `_drainSpokenTurn` is a second drain loop, on the user-simulator side, and it absorbed every rejection from its own `receiveAudio`. A socket close or a server error handed back the audio collected so far as a complete spoken user line. It now breaks only on a receive deadline. Two span tests were ending their turn by pushing a synthetic server `error` event, which is exactly the rejection that must now propagate; they push an empty audio delta instead, which is the real end-of-stream signal and does not go green if the propagation breaks. The ElevenLabs session-error handler resolved parked receives with the empty terminal chunk, so a broken session read as a clean end of turn. It now rejects them, and records the failure so every later receive keeps failing. A clean session end still resolves with the terminal chunk, and keeps doing so once the stream has ended, which is the ordinary shape of an agent hangup (#839). The contract table covered three adapters. It now covers all six, so no timeout producer can drift back to a plain Error unnoticed. Each entry verified red against its own adapter. --- .../adapters/__tests__/elevenlabs.test.ts | 27 +++- .../__tests__/openai-realtime-spans.test.ts | 28 ++-- .../receive-timeout-contract.test.ts | 134 ++++++++++++++++++ javascript/src/voice/adapters/elevenlabs.ts | 74 ++++++++-- .../src/voice/adapters/openai-realtime.ts | 11 +- 5 files changed, 240 insertions(+), 34 deletions(-) diff --git a/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts b/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts index 4babb4e81..907a38d90 100644 --- a/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts +++ b/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts @@ -656,16 +656,21 @@ describe("ElevenLabsAgentAdapter wire-protocol (SDK-routed recv path)", () => { await adapter.disconnect(); }); - it("post-open socket error nulls the session and unblocks pending receivers", async () => { + it("post-open socket error nulls the session and fails pending receivers", async () => { const { adapter, socket } = await makeConnected(); const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); const recv = adapter.receiveAudio(2); socket.emit("error", new Error("connection lost")); - const chunk = await recv; - // Pending waiter resolves with an empty chunk so the executor unwinds - // rather than hanging. The session is cleared so subsequent sendAudio - // fails fast instead of writing to a dead socket. - expect(chunk.data.length).toBe(0); + + // A pending waiter is REJECTED, not resolved with an empty chunk. Either + // unwinds the executor, but the empty chunk reads as a clean end of turn, + // so the drain would report the audio collected so far as a complete turn + // and the broken session would never reach the run (#756). + await expect(recv).rejects.toThrow(/session error: connection lost/); + // And it keeps failing, rather than the next probe finding a quiet socket. + await expect(adapter.receiveAudio(2)).rejects.toThrow(/session error/); + // The session is cleared so subsequent sendAudio fails fast instead of + // writing to a dead socket. await expect(adapter.sendAudio(silentChunk(0.01))).rejects.toThrow(/not connected/); warn.mockRestore(); }); @@ -678,6 +683,16 @@ describe("ElevenLabsAgentAdapter wire-protocol (SDK-routed recv path)", () => { expect(chunk.data.length).toBe(0); }); + it("keeps returning the terminal chunk after a clean session end", async () => { + // The drain's tail probe often lands AFTER the close rather than during it. + // That is the ordinary shape of an agent hangup (#839), so it has to end the + // turn, not fail the run on "not connected". + const { adapter, socket } = await makeConnected(); + socket.emit("close", 1000, Buffer.from("closed")); + const chunk = await adapter.receiveAudio(2); + expect(chunk.data.length).toBe(0); + }); + it("receiveAudio rejects with timeout when no audio arrives in time", async () => { const { adapter } = await makeConnected(); await expect(adapter.receiveAudio(0.05)).rejects.toThrow(/timed out/); diff --git a/javascript/src/voice/adapters/__tests__/openai-realtime-spans.test.ts b/javascript/src/voice/adapters/__tests__/openai-realtime-spans.test.ts index cab6dbb98..fd82a0806 100644 --- a/javascript/src/voice/adapters/__tests__/openai-realtime-spans.test.ts +++ b/javascript/src/voice/adapters/__tests__/openai-realtime-spans.test.ts @@ -389,12 +389,14 @@ describe("OpenAIRealtimeAgentAdapter voice.realtime.* span instrumentation (#770 // _drainSpokenTurn's idle timeout is a hardcoded 15s (speakGeneratedUserTurn // is called with no override), not sourced from adapter.responseTailSilence — // so a real "just stop sending" tail-silence close would hang this test for - // 15 real seconds. Push a synthetic error event instead: the drain loop - // treats ANY receiveAudio rejection as "the model stopped talking" (see - // speakUserTurn's jsdoc), so this ends the ALREADY-drained turn (1 chunk) - // immediately. The R2 markers/attrs asserted below are already stamped - // (from the response.created/response.done events above) before this fires. - handle.push({ type: "error", error: { message: "test: end turn" } }); + // 15 real seconds. Push an EMPTY audio delta instead: a zero-length chunk is + // this codebase's end-of-stream signal, so it ends the ALREADY-drained turn + // (1 chunk) immediately. The R2 markers/attrs asserted below are already + // stamped (from the response.created/response.done events above) before this + // fires. Not an `error` event: a server error is a real failure that + // _drainSpokenTurn propagates, and a test that ends its turn with one would + // pass just as happily if the propagation broke (#756). + handle.push({ type: "response.output_audio.delta", delta: "" }); await callPromise; const spans = byName(exporter.getFinishedSpans()); @@ -472,13 +474,13 @@ describe("OpenAIRealtimeAgentAdapter voice.realtime.* span instrumentation (#770 handle.push({ type: "response.done" }); // _drainSpokenTurn's idle timeout is a hardcoded 15s default (speakUserTurn // is called with no override) — a real "just stop sending" tail-silence - // close would hang this test for 15 real seconds. Push a synthetic error - // event instead, exactly like the R4 test above: the drain loop treats ANY - // receiveAudio rejection as "the model stopped talking", so this ends the - // already-drained turn (1 chunk) immediately. No voice.audio.receive span - // is ever opened on this path (scope guard) — the assertions below are - // what that actually means: neither the receive nor the turn span exist. - handle.push({ type: "error", error: { message: "test: end turn" } }); + // close would hang this test for 15 real seconds. Push an EMPTY audio delta + // instead, exactly like the R4 test above: a zero-length chunk is the + // end-of-stream signal, so it ends the already-drained turn (1 chunk) + // immediately. No voice.audio.receive span is ever opened on this path + // (scope guard) — the assertions below are what that actually means: + // neither the receive nor the turn span exist. + handle.push({ type: "response.output_audio.delta", delta: "" }); await turnPromise; const spans = byName(exporter.getFinishedSpans()); diff --git a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts index e4f4fbbf0..2e7a8455f 100644 --- a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts +++ b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts @@ -22,15 +22,24 @@ import { Buffer } from "node:buffer"; import { EventEmitter } from "node:events"; +import type { LanguageModel } from "ai"; import { describe, it, expect, vi } from "vitest"; import { AgentRole } from "../../../domain/agents"; import { makeAgentInput } from "../../__tests__/helpers/drive-production"; import { isReceiveTimeoutError } from "../../receive-timeout-error"; +import type { STTProvider } from "../../stt"; +import { OPENAI_REALTIME_MODEL } from "../../voice-models"; +import { ComposableVoiceAgent } from "../composable"; import { ElevenLabsAgentAdapter } from "../elevenlabs"; import { GeminiLiveAgentAdapter } from "../gemini-live"; +import { OpenAIRealtimeAgentAdapter } from "../openai-realtime"; import { PipecatAgentAdapter, type PipecatWebSocketLike } from "../pipecat"; +import { TwilioAgentAdapter } from "../twilio"; +import type { MediaStreamWebSocket } from "../twilio-server"; +import { TwilioRESTHelper } from "../twilio-shared"; import { makeFakeConv } from "./fixtures/fake-elevenlabs-conversation"; +import { setupMockRealtimeServer } from "./fixtures/mock-realtime-server"; // Mock the Gemini SDK so connect() never opens a real WebSocket. Scoped to this // module, and the other adapters under test do not import it. @@ -67,6 +76,64 @@ function silentPipecatSocket(): SilentPipecatSocket { return socket; } +/** In-process stand-in for the OpenAI Realtime endpoint. Stays silent. */ +const realtimeServer = setupMockRealtimeServer(() => {}); + +/** Twilio REST with every network call stubbed, so connect() stays local. */ +function stubTwilioRest(): TwilioRESTHelper { + const stub = new TwilioRESTHelper("ACtest", "secret"); + stub.resolvePhoneNumberSid = async () => "PNtimeoutcontract"; + stub.readVoiceUrl = async () => null; + stub.writeVoiceUrl = async () => undefined; + stub.placeCall = async () => "CAtimeoutcontract"; + stub.sendDtmfOnCall = async () => undefined; + return stub; +} + +/** A media-stream socket that stays open and sends only what we hand it. */ +function twilioSocket(): MediaStreamWebSocket & { emit(text: string): void } { + const incoming: string[] = []; + let resolver: ((text: string | null) => void) | null = null; + return { + send() {}, + close() {}, + receiveText() { + const head = incoming.shift(); + if (head !== undefined) return Promise.resolve(head); + return new Promise((resolve) => { + resolver = resolve; + }); + }, + emit(text: string) { + if (resolver) { + const r = resolver; + resolver = null; + r(text); + } else { + incoming.push(text); + } + }, + } as MediaStreamWebSocket & { emit(text: string): void }; +} + +/** An ai-sdk model whose generation never settles, so the wrapper deadline wins. */ +function hangingLlm(): LanguageModel { + return { + specificationVersion: "v3" as const, + provider: "fake", + modelId: "hanging-model", + supportedUrls: {}, + doGenerate: () => new Promise(() => {}), + doStream: () => new Promise(() => {}), + } as unknown as LanguageModel; +} + +const stubStt: STTProvider = { + async transcribe(): Promise { + return "user said hello"; + }, +}; + /** * Each entry parks a `receiveAudio` against a transport that stays open and * silent, so the adapter's own deadline is what ends the wait. @@ -119,6 +186,73 @@ const ADAPTERS: Array<{ }; }, }, + { + name: "OpenAIRealtimeAgentAdapter", + parkReceive: async () => { + realtimeServer.arm(); + const adapter = new OpenAIRealtimeAgentAdapter({ + apiKey: "test-key", + url: `ws://127.0.0.1:${realtimeServer.port()}/realtime?model=${OPENAI_REALTIME_MODEL}`, + }); + await adapter.connect(); + await realtimeServer.socketReady(); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: () => adapter.disconnect(), + }; + }, + }, + { + name: "TwilioAgentAdapter", + parkReceive: async () => { + const adapter = new TwilioAgentAdapter({ + accountSid: "ACtest", + authToken: "secret", + phoneNumber: "+14155551234", + publicBaseUrl: "https://example.test", + validateSignature: false, + rest: stubTwilioRest(), + }); + await adapter.connect(); + const socket = twilioSocket(); + // Drive the media-stream loop and open a call, but never end it: the + // stream stays LIVE with an empty queue, which is the state whose only + // exit is the receive deadline. + const loop = adapter._driveMediaStream(socket); + socket.emit( + JSON.stringify({ + event: "start", + start: { streamSid: "MZtimeoutcontract", callSid: "CAtimeoutcontract" }, + }), + ); + await new Promise((resolve) => setTimeout(resolve, 10)); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: async () => { + socket.emit(JSON.stringify({ event: "stop", streamSid: "MZtimeoutcontract" })); + await loop; + await adapter.disconnect(); + }, + }; + }, + }, + { + name: "ComposableVoiceAgent", + parkReceive: async () => { + // No transport: the deadline wraps the STT/LLM/TTS work itself, and a + // generation that never settles is what a wedged provider looks like. + const adapter = new ComposableVoiceAgent({ + stt: stubStt, + llm: hangingLlm(), + tts: "openai/nova", + }); + await adapter.connect(); + return { + receive: adapter.receiveAudio(DEADLINE_S), + teardown: () => adapter.disconnect(), + }; + }, + }, ]; describe("built-in adapters reject their receive deadline as a receive timeout (#756)", () => { diff --git a/javascript/src/voice/adapters/elevenlabs.ts b/javascript/src/voice/adapters/elevenlabs.ts index 7a34dd611..6578c019a 100644 --- a/javascript/src/voice/adapters/elevenlabs.ts +++ b/javascript/src/voice/adapters/elevenlabs.ts @@ -382,10 +382,23 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { /** Queue of agent audio chunks the SDK pushed via `output()` ahead of a receiver. */ private readonly audioQueue: AudioChunk[] = []; - /** Resolvers waiting on the next agent audio chunk (FIFO). */ - private readonly waiters: Array<(chunk: AudioChunk) => void> = []; + /** + * Receivers waiting on the next agent audio chunk (FIFO). + * + * Both outcomes are needed: a session that ENDS resolves them with the empty + * terminal chunk, so the turn finishes; a session that ERRORS rejects them, + * so the drain fails the run instead of reporting a truncated turn (#756). + */ + private readonly waiters: Array<{ + resolve: (chunk: AudioChunk) => void; + reject: (err: Error) => void; + }> = []; /** Idle-timer-reset callbacks for active receiveAudio calls — called on every inbound frame. */ private readonly timerResetters: Array<() => void> = []; + /** Set once the session has ended cleanly, so later receives return the terminal chunk. */ + private streamEnded = false; + /** Set when the session ERRORED, so every later receive keeps failing with it. */ + private sessionFailure: Error | null = null; /** * Continuous mic pump outbound queue: 20 ms PCM frames enqueued by {@link @@ -477,6 +490,10 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { }); const client = new ElevenLabsClient({ apiKey: this.apiKey }); + // A previous session's terminal state must not decide this one's receives. + this.streamEnded = false; + this.sessionFailure = null; + // The adapter's NARROW prompt/first-message knobs build an `agent` override // that is always sent (an empty `agent` object is a no-op) so the handshake // shape is stable; it carries the prompt/first-message overrides when set. @@ -596,7 +613,7 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { if (pcm.length % 2 === 1) pcm = pcm.subarray(0, pcm.length - 1); const chunk = new AudioChunk({ data: new Uint8Array(pcm) }); const waiter = this.waiters.shift(); - if (waiter) waiter(chunk); + if (waiter) waiter.resolve(chunk); else this.audioQueue.push(chunk); } @@ -617,7 +634,14 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { // the executor unwinds rather than hanging, and null the session so subsequent // sendAudio/receiveAudio fail fast with a clear "not connected". this.stopPump(); - this.drainPendingWaiters(); + // Reject rather than resolve: the session broke, so a parked receive must + // surface that. Resolving with the terminal chunk would close the turn and + // report the partial audio as a complete one. + this.sessionFailure = new Error( + `ElevenLabsAgentAdapter: session error: ${err.message}`, + { cause: err }, + ); + this.failPendingWaiters(this.sessionFailure); this.inputCallback = null; this.conversation = null; } @@ -625,13 +649,22 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { /** Called on `session_ended` (clean endSession OR socket close). */ private onSessionEnded(): void { this.stopPump(); + this.streamEnded = true; this.drainPendingWaiters(); } + /** Clean end of stream: finish every parked receive with the terminal chunk. */ private drainPendingWaiters(): void { while (this.waiters.length > 0) { const waiter = this.waiters.shift(); - waiter?.(new AudioChunk({ data: new Uint8Array(0) })); + waiter?.resolve(new AudioChunk({ data: new Uint8Array(0) })); + } + } + + /** Broken session: fail every parked receive with the underlying error. */ + private failPendingWaiters(failure: Error): void { + while (this.waiters.length > 0) { + this.waiters.shift()?.reject(failure); } } @@ -832,13 +865,24 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { } async receiveAudio(timeout: number): Promise { + // Buffered audio first: it was produced while the session was live and is + // owed to the caller whether or not the session has since ended. + const queued = this.audioQueue.shift(); + if (queued) return queued; + + if (this.sessionFailure) throw this.sessionFailure; + if (this.streamEnded) { + // The session ended and everything it sent has been consumed. The drain + // reaches here when the close lands between its first chunk and its tail + // probe, which is the ordinary shape of an agent hangup (#839). Hand back + // the terminal chunk so the turn finishes rather than failing a run in + // which the agent behaved as designed. + return new AudioChunk({ data: new Uint8Array(0) }); + } if (!this.isConnected()) { throw new Error("ElevenLabsAgentAdapter: not connected"); } - const queued = this.audioQueue.shift(); - if (queued) return queued; - return await new Promise((resolve, reject) => { // Forward-declared so the timers, the resetter, and the waiter share them. let timer: ReturnType; @@ -899,9 +943,15 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { timer = setTimeout(onIdleTimeout, timeout * 1000); }; - const waiter = (chunk: AudioChunk) => { - cleanup(); - resolve(chunk); + const waiter = { + resolve: (chunk: AudioChunk) => { + cleanup(); + resolve(chunk); + }, + reject: (err: Error) => { + cleanup(); + reject(err); + }, }; timer = setTimeout(onIdleTimeout, timeout * 1000); @@ -971,7 +1021,7 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { // before the agent acts, so a mid-turn tool call finds one. Parity with the // close/error drain. const waiter = this.waiters.shift(); - if (waiter) waiter(new AudioChunk({ data: new Uint8Array(0) })); + if (waiter) waiter.resolve(new AudioChunk({ data: new Uint8Array(0) })); return; } diff --git a/javascript/src/voice/adapters/openai-realtime.ts b/javascript/src/voice/adapters/openai-realtime.ts index 798490f3f..cac3805cf 100644 --- a/javascript/src/voice/adapters/openai-realtime.ts +++ b/javascript/src/voice/adapters/openai-realtime.ts @@ -34,7 +34,7 @@ import { VoiceAgentAdapter } from "../adapter"; import { AudioChunk } from "../audio-chunk"; import { AdapterCapabilities } from "../capabilities"; import { createAudioMessage, extractAudio } from "../messages"; -import { ReceiveTimeoutError } from "../receive-timeout-error"; +import { isReceiveTimeoutError, ReceiveTimeoutError } from "../receive-timeout-error"; import { currentSpan, setSpanAttributes, voiceSpan } from "../telemetry"; import { OPENAI_REALTIME_MODEL, OPENAI_STT_MODEL } from "../voice-models"; @@ -900,8 +900,13 @@ export class OpenAIRealtimeAgentAdapter extends VoiceAgentAdapter { let chunk: AudioChunk; try { chunk = await this.receiveAudio(tailTimeoutS); - } catch { - break; // timeout / socket close = end of the model's spoken turn + } catch (err) { + // The idle timeout is the natural end of the model's spoken turn, and a + // closed stream arrives as the zero-length chunk below. Anything else is + // a real failure, and returning the audio collected so far would hand + // back a truncated user line as a complete one (#756). + if (!isReceiveTimeoutError(err)) throw err; + break; } if (chunk.data.length === 0) break; chunks.push(chunk.data); From 73df8107f35d9c0e00dadca3fe3221db7b090187 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rog=C3=A9rio=20Chaves?= Date: Wed, 12 Aug 2026 14:28:00 +0200 Subject: [PATCH 5/6] test(voice): assert the spoken user-turn drain fails on a server error The narrowed catch in `_drainSpokenTurn` had no direct coverage: the two existing tests end their turn on the idle deadline, which passes either way. This one pushes an error after the first audio delta and requires it to reach the caller, so the simulator can never hand back half a sentence as the user's real line. Records the rule as AC7 of the spec: it is per drain loop, not per adapter, and #623 adds more of them as agent-initiated turns reach the other adapters. --- .../openai-realtime-speak-user-turn.test.ts | 26 +++++++++++++++++++ specs/voice-drain-error-propagation.feature | 11 ++++++++ 2 files changed, 37 insertions(+) diff --git a/javascript/src/voice/adapters/__tests__/openai-realtime-speak-user-turn.test.ts b/javascript/src/voice/adapters/__tests__/openai-realtime-speak-user-turn.test.ts index 9842d00d1..841995111 100644 --- a/javascript/src/voice/adapters/__tests__/openai-realtime-speak-user-turn.test.ts +++ b/javascript/src/voice/adapters/__tests__/openai-realtime-speak-user-turn.test.ts @@ -131,4 +131,30 @@ describe("OpenAIRealtimeAgentAdapter.speakUserTurn (#705 bridge)", () => { await adapter.disconnect(); }); + + it("fails the turn on a server error instead of returning the partial line", async () => { + // #756 on the simulator side. The drain used to absorb every rejection, so + // a server error mid-utterance came back as a COMPLETE spoken turn made of + // whatever had arrived. The simulator then said half a sentence and the run + // scored it as the user's real line. + observed = []; + server.arm(); + const adapter = buildAdapter(server.port()); + await adapter.connect(); + await server.socketReady(); + await waitFor(() => observed.some((e) => e.type === "session.update")); + + const turnPromise = adapter.speakUserTurn("cancel my subscription", 1); + await waitFor(() => observed.some((e) => e.type === "response.create")); + + push({ + type: "response.output_audio.delta", + delta: Buffer.from(new Uint8Array([0x07, 0x00])).toString("base64"), + }); + push({ type: "error", error: { message: "upstream exploded" } }); + + await expect(turnPromise).rejects.toThrow(/upstream exploded/); + + await adapter.disconnect(); + }); }); diff --git a/specs/voice-drain-error-propagation.feature b/specs/voice-drain-error-propagation.feature index 3a1cd6852..3536cbf23 100644 --- a/specs/voice-drain-error-propagation.feature +++ b/specs/voice-drain-error-propagation.feature @@ -80,3 +80,14 @@ Feature: The voice drain ends a turn on silence, not on failure When defaultVoiceCall drains the agent response Then the turn completes successfully with agentHungUp still set And the voice.audio.receive span is labelled terminal_chunk + + # AC7 — the rule is per DRAIN LOOP, not per adapter. The user-simulator side + # has its own, and #623 will add more as agent-initiated turns spread to the + # other adapters. Each one has to make the same distinction. + @unit @ts-voice-drain + Scenario: The spoken user-turn drain fails on a server error + Given the simulator is speaking a scripted user line + And a server error arrives after the first audio delta + When the spoken turn drains + Then the error reaches the caller + And no partial spoken line is returned as the user's turn From 331e22f8c007abd469ecd751bbbe3a96a531a2c9 Mon Sep 17 00:00:00 2001 From: allin2 <37659199+allin2@users.noreply.github.com> Date: Sun, 16 Aug 2026 12:42:33 +0800 Subject: [PATCH 6/6] fix(voice): preserve transport error identity --- .../adapters/__tests__/elevenlabs.test.ts | 13 +++---- .../receive-timeout-contract.test.ts | 35 +++++++++++-------- javascript/src/voice/adapters/elevenlabs.ts | 18 +++++++--- javascript/src/voice/adapters/pipecat.ts | 16 ++++++--- 4 files changed, 53 insertions(+), 29 deletions(-) diff --git a/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts b/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts index 907a38d90..28bb643ad 100644 --- a/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts +++ b/javascript/src/voice/adapters/__tests__/elevenlabs.test.ts @@ -17,15 +17,15 @@ import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; -import type { LanguageModel } from "ai"; import { describeFeature, loadFeature } from "@amiceli/vitest-cucumber"; +import type { LanguageModel } from "ai"; import { afterEach, describe, it, expect, vi } from "vitest"; import { AgentRole } from "../../../domain/agents"; -import { AudioChunk, silentChunk } from "../../audio-chunk"; import { VoiceAgentAdapter } from "../../adapter"; -import { ELEVENLABS_DEFAULT_VOICE_ID } from "../../voice-models"; +import { AudioChunk, silentChunk } from "../../audio-chunk"; import { elevenLabsAgent } from "../../factories"; +import { ELEVENLABS_DEFAULT_VOICE_ID } from "../../voice-models"; import { ComposableVoiceAgent, ElevenLabsAgentAdapter, @@ -660,15 +660,16 @@ describe("ElevenLabsAgentAdapter wire-protocol (SDK-routed recv path)", () => { const { adapter, socket } = await makeConnected(); const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); const recv = adapter.receiveAudio(2); - socket.emit("error", new Error("connection lost")); + const transportError = new Error("connection lost"); + socket.emit("error", transportError); // A pending waiter is REJECTED, not resolved with an empty chunk. Either // unwinds the executor, but the empty chunk reads as a clean end of turn, // so the drain would report the audio collected so far as a complete turn // and the broken session would never reach the run (#756). - await expect(recv).rejects.toThrow(/session error: connection lost/); + await expect(recv).rejects.toBe(transportError); // And it keeps failing, rather than the next probe finding a quiet socket. - await expect(adapter.receiveAudio(2)).rejects.toThrow(/session error/); + await expect(adapter.receiveAudio(2)).rejects.toBe(transportError); // The session is cleared so subsequent sendAudio fails fast instead of // writing to a dead socket. await expect(adapter.sendAudio(silentChunk(0.01))).rejects.toThrow(/not connected/); diff --git a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts index 2e7a8455f..3dfed6c39 100644 --- a/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts +++ b/javascript/src/voice/adapters/__tests__/receive-timeout-contract.test.ts @@ -27,6 +27,7 @@ import { describe, it, expect, vi } from "vitest"; import { AgentRole } from "../../../domain/agents"; import { makeAgentInput } from "../../__tests__/helpers/drive-production"; +import { extractAudio } from "../../messages"; import { isReceiveTimeoutError } from "../../receive-timeout-error"; import type { STTProvider } from "../../stt"; import { OPENAI_REALTIME_MODEL } from "../../voice-models"; @@ -37,7 +38,7 @@ import { OpenAIRealtimeAgentAdapter } from "../openai-realtime"; import { PipecatAgentAdapter, type PipecatWebSocketLike } from "../pipecat"; import { TwilioAgentAdapter } from "../twilio"; import type { MediaStreamWebSocket } from "../twilio-server"; -import { TwilioRESTHelper } from "../twilio-shared"; +import { mulaw8kToPcm16At24k, TwilioRESTHelper } from "../twilio-shared"; import { makeFakeConv } from "./fixtures/fake-elevenlabs-conversation"; import { setupMockRealtimeServer } from "./fixtures/mock-realtime-server"; @@ -326,19 +327,19 @@ describe("a stream that ends is an end of turn, not a failure (#756)", () => { it("still fails every receive after a socket error", async () => { const { adapter, socket } = await connectedPipecat(); - socket.emit("error", new Error("ECONNRESET")); + const transportError = new Error("ECONNRESET"); + const inFlightReceive = adapter.receiveAudio(30); + await Promise.resolve(); - // Now, and on the drain's next probe: a broken transport is not an end of - // turn however many times it is asked. - for (const attempt of ["first", "second"]) { - const error = await adapter.receiveAudio(30).then( - () => undefined, - (err: unknown) => err, - ); - expect(error, `${attempt} receive resolved after a socket error`).toBeInstanceOf(Error); - expect((error as Error).message).toContain("ECONNRESET"); - expect(isReceiveTimeoutError(error)).toBe(false); - } + socket.emit("error", transportError); + + // A broken transport rejects the currently parked receive immediately, + // rather than letting its deadline relabel the failure as a timeout. + await expect(inFlightReceive).rejects.toBe(transportError); + expect(isReceiveTimeoutError(transportError)).toBe(false); + + // The drain's next probe must see the same transport failure too. + await expect(adapter.receiveAudio(30)).rejects.toBe(transportError); }); it("completes a real call() whose stream ends mid-drain", async () => { @@ -362,7 +363,11 @@ describe("a stream that ends is an end of turn, not a failure (#756)", () => { ); setTimeout(() => socket.emit("close"), 20); - const messages = await adapter.call(makeAgentInput()); - expect(messages).toBeTruthy(); + const message = await adapter.call(makeAgentInput()); + const audio = extractAudio(message); + expect(audio, "call() returned no assistant audio").not.toBeNull(); + expect(audio!.data.length).toBeGreaterThan(0); + const expectedPcm = mulaw8kToPcm16At24k(Buffer.from(payload, "base64")); + expect(Array.from(audio!.data)).toEqual(Array.from(expectedPcm)); }); }); diff --git a/javascript/src/voice/adapters/elevenlabs.ts b/javascript/src/voice/adapters/elevenlabs.ts index 6578c019a..86683840e 100644 --- a/javascript/src/voice/adapters/elevenlabs.ts +++ b/javascript/src/voice/adapters/elevenlabs.ts @@ -637,10 +637,10 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { // Reject rather than resolve: the session broke, so a parked receive must // surface that. Resolving with the terminal chunk would close the turn and // report the partial audio as a complete one. - this.sessionFailure = new Error( - `ElevenLabsAgentAdapter: session error: ${err.message}`, - { cause: err }, - ); + // Preserve the SDK error itself so callers retain its type, identity, + // stack, and any vendor-specific fields. The warning above supplies the + // adapter context without replacing the exception. + this.sessionFailure = err; this.failPendingWaiters(this.sessionFailure); this.inputCallback = null; this.conversation = null; @@ -864,6 +864,16 @@ export class ElevenLabsAgentAdapter extends VoiceAgentAdapter { // VAD measures to close the turn — until the agent responds and the pump pauses. } + /** + * Wait for the next PCM audio chunk from the hosted ElevenLabs session. + * + * @param timeout - Sliding idle deadline in seconds. An absolute ceiling of + * `max(timeout, 45s)` also bounds sessions that send keepalives but no audio. + * @returns A non-empty audio chunk, or a zero-length terminal chunk after a clean + * session end. + * @throws {ReceiveTimeoutError} When either receive deadline expires. + * @throws The original session or transport error when the session fails. + */ async receiveAudio(timeout: number): Promise { // Buffered audio first: it was produced while the session was live and is // owed to the caller whether or not the session has since ended. diff --git a/javascript/src/voice/adapters/pipecat.ts b/javascript/src/voice/adapters/pipecat.ts index de0fb8834..45e621a1d 100644 --- a/javascript/src/voice/adapters/pipecat.ts +++ b/javascript/src/voice/adapters/pipecat.ts @@ -389,6 +389,15 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { } } + /** + * Wait for the next decoded PCM audio chunk from the Pipecat stream. + * + * @param timeout - Maximum seconds to wait for audio before the receive deadline. + * @returns A non-empty audio chunk, or a zero-length terminal chunk after a clean + * stream close or interruption. + * @throws {ReceiveTimeoutError} When no audio arrives before `timeout`. + * @throws The original socket or transport error when the stream fails. + */ override async receiveAudio(timeout: number): Promise { this.assertConnected(); // Interrupt gate: `interrupt()` set this phase after clearing `inbox.queue`. @@ -576,10 +585,9 @@ export class PipecatAgentAdapter extends VoiceAgentAdapter { this.inbox.closed = true; // A broken transport must reach the caller, both now and on every later // receive, so the drain never mistakes it for the agent finishing (#756). - this.inbox.failure = new Error( - `PipecatAgentAdapter: socket error: ${err.message}`, - { cause: err }, - ); + // Preserve the transport error itself so callers keep its type, identity, + // stack, and any provider-specific fields. + this.inbox.failure = err; const waiter = this.inbox.waiter; if (waiter) { this.inbox.waiter = null;