From e41ec50074cab1903ee56b4fbc43ed8066e0de8a Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Fri, 25 Sep 2026 11:52:10 -0400 Subject: [PATCH 1/4] VAPI-3991: implement the Bandwidth StartStream source and wire /bw/stream The Media Streams bridge spoke Twilio's protocol to the bot, but the Bandwidth side was only an interface: nothing in src/ implemented BwStreamSource and the server never wired a StartStream destination to the bridge. A translated Connect/Stream therefore opened a Bandwidth stream to a destination that could not relay anything. src/streams/bw-source.ts (new): BwWebSocketSource speaks Bandwidth's StartStream WebSocket protocol over the socket Bandwidth opens to us. - start: parses metadata (accountId, callId, streamId, streamName, tracks) and streamParams; waitForStart() with a timeout for the server. - media: relays the inbound (caller) track; outbound-track frames and frames without a string payload are ignored. Frames arriving before release() are buffered (bounded to 10 s) so caller audio spoken while the bot's socket is still opening is delivered in order, not dropped. - stop: emitted exactly once, on a stop event or on socket close. - sendMedia -> playAudio (audio/pcmu); flush -> clear; close -> ws.close. - onFrame hook for fixture capture. No dtmf: Bandwidth delivers digits via BXML Gather webhooks, not the media socket. src/server/app.ts: the translator now rewrites a Twilio to wss:///bw/stream?dest= (ws:// for http bases). The HTTP upgrade is taken off Fastify's underlying server and handed to ws. The upgrade must carry the same Basic-auth credentials as the /bw/* webhooks (401 otherwise), dest is required (400), must be ws/wss (400) and passes the egress guard (400), and any other path is 404. After Bandwidth's start event a TwilioStreamBridge is built with callSid from callId, accountSid from config, customParameters from streamParams, and the configured playout latency pad; caller audio is released once the bot's socket is open. No start within streamStartTimeoutMs (default 5 s), or an unreachable bot, closes the Bandwidth socket so the translator's returns and the call's BXML moves on. Bridges are closed on app close. src/streams/bridge.ts: when the bot closes its socket, close the Bandwidth source. On Twilio that ends and TwiML resumes; here it ends the StartStream so BXML resumes likewise. src/translator/translate.ts: stamp destinationUsername/destinationPassword on StartStream alongside the existing callback credentials. Bandwidth presents them as the Authorization header on the WebSocket upgrade. src/twilio/egress-guard.ts: assertPublicUrl accepts a schemes option so the stream route can require ws/wss while HTTP callers keep http/https. src/server/capture.ts: captureStreamFrame appends raw Bandwidth frames to /streams/.jsonl (start, first 10 media, stop) so the first live call yields checked-in fixtures. Same hashed-filename and private-mode guardrails as captureTwiml; the key is a per-connection UUID, never Bandwidth input. test/fixtures/bandwidth/stream-frames.json: start.metadata is the real event recorded on a live call 2026-09-17 (ids redacted); streamParams, media, and stop follow the documented shapes, with the media payload being 20 ms of synthetic PCMU silence. The _meta block says so and how to replace it from a capture. Tests: streams-bw-source.test.ts covers parsing, buffering and release, track filtering, the buffer cap, stop once, playAudio/clear, and onFrame. server-stream.test.ts drives a fake Bandwidth client through the listening server to a fake Twilio bot: both directions end to end, buffered audio order, bot hangup and unreachable bot ending the stream, start timeout, all four upgrade rejections, the URL rewrite with destination credentials, and frame capture. translate-callback-auth.test.ts covers the new StartStream stamping. Docs: AGENTS.md describes the route, auth, capture, and the tunnel requirement; README notes the source. The README capability rows stay at Unit until an end-to-end call passes. --- AGENTS.md | 19 +- README.md | 2 +- src/server/app.ts | 125 +++++++- src/server/capture.ts | 32 ++- src/server/index.ts | 3 + src/streams/bridge.ts | 8 + src/streams/bw-source.ts | 215 ++++++++++++++ src/translator/translate.ts | 9 +- src/twilio/egress-guard.ts | 12 +- test/fixtures/bandwidth/stream-frames.json | 64 +++++ test/server-stream.test.ts | 315 +++++++++++++++++++++ test/streams-bw-source.test.ts | 232 +++++++++++++++ test/translate-callback-auth.test.ts | 13 + 13 files changed, 1037 insertions(+), 12 deletions(-) create mode 100644 src/streams/bw-source.ts create mode 100644 test/fixtures/bandwidth/stream-frames.json create mode 100644 test/server-stream.test.ts create mode 100644 test/streams-bw-source.test.ts diff --git a/AGENTS.md b/AGENTS.md index ef95a94..3665667 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,7 +74,11 @@ Set these env vars (the server reads **these exact names** — note the checked- ### Phase 4 — Deploy 🧍 **Human required:** provide a public HTTPS host (or tunnel) for `PUBLIC_BASE_URL`. The BW Voice Application's callback (set in Phase 2) must point -at `/bw/initiate`. +at `/bw/initiate`. If the customer app uses ``, +the host must also pass WebSocket upgrades through to the translator at +`/bw/stream` (ngrok and cloudflared do by default). Optional: +`STREAM_PLAYOUT_LATENCY_PAD_MS` delays mark acknowledgements to the bot by that +many ms to absorb network and jitter-buffer latency; default 0. ### Phase 5 — Verify ```bash @@ -112,8 +116,17 @@ Translation is a fixed rulebook (`src/matrix/twilio-voice.json`), not a guess. after the stream ends, as on Twilio. A stream name is generated when the TwiML omits one. `ConversationRelay` and `VirtualAgent` are unsupported (separate IoV). - - `Stream` — Twilio's WS message schema is emulated by the translator's stream - bridge; live Bandwidth-side binding requires fixture capture. `` + - `Stream` — the translator rewrites the Twilio `` (the customer's + bot) to `wss:///bw/stream?dest=`, so Bandwidth's + StartStream WebSocket lands on the translator, which bridges it to the bot in + Twilio's Media Streams protocol (`src/streams/bw-source.ts` speaks Bandwidth's + side, `src/streams/bridge.ts` Twilio's). The upgrade requires the same + Basic-auth credentials as the `/bw/*` webhooks, stamped on `StartStream` as + `destinationUsername`/`destinationPassword`, and `dest` passes the egress + guard. Only the inbound (caller) track is relayed; Bandwidth delivers no DTMF + over the media socket. With `TRANSLATOR_CAPTURE_DIR` set, raw Bandwidth + frames are written to `/streams/*.jsonl` (start, first 10 media, stop) + to refresh `test/fixtures/bandwidth/stream-frames.json`. `` children map to nested `` elements in order (Bandwidth allows at most 12; extras are dropped with a warning). Bandwidth echoes them in its `start` event as `streamParams`, and the bridge forwards them to the bot as diff --git a/README.md b/README.md index abb131e..f65f0f2 100644 --- a/README.md +++ b/README.md @@ -198,7 +198,7 @@ All translation is driven by one declarative compatibility matrix (`src/matrix/t ``` src/translator TwiML → BXML translation src/twilio Twilio-shaped REST facade + signed webhooks (egress) -src/streams Media Streams bridge +src/streams Media Streams bridge (Twilio protocol to the bot) + Bandwidth StartStream source src/server the proxy (Fastify) wiring it together, readiness check (/readyz, npm run doctor) src/compatibility-check static migration-complexity report (npm run compatibility-check) src/bxml-generator batch BXML generation + coverage report (npm run bxml-generator; see AGENTS.md Phase 1) diff --git a/src/server/app.ts b/src/server/app.ts index 3974d95..92308c9 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -1,6 +1,12 @@ import Fastify, { type FastifyInstance, type FastifyReply, type FastifyRequest } from "fastify"; import formbody from "@fastify/formbody"; +import type { IncomingMessage } from "node:http"; +import type { Duplex } from "node:stream"; +import { randomUUID } from "node:crypto"; +import { WebSocketServer } from "ws"; import { translateTwiml, type UrlKind } from "../translator/translate.js"; +import { TwilioStreamBridge, customParametersFromBwStart } from "../streams/bridge.js"; +import { BwWebSocketSource } from "../streams/bw-source.js"; import { bxmlDocument } from "../xml/build-xml.js"; import { initiateParams, @@ -9,7 +15,7 @@ import { recordingStatusParams, postToCustomer, } from "../twilio/egress.js"; -import { EgressBlockedError } from "../twilio/egress-guard.js"; +import { EgressBlockedError, assertPublicUrl } from "../twilio/egress-guard.js"; import { toCallSid, toRecordingSid } from "../twilio/call-sid.js"; import { createdCallResource, bwStateToTwilioStatus, twilioErrors } from "../twilio/call-resource.js"; import { serverErrors } from "./errors.js"; @@ -23,7 +29,7 @@ import { CallStore, type CallRecord } from "./call-store.js"; import { isSafeBwId, type BwClient } from "../bw/client.js"; import { checkReadiness } from "./readiness.js"; import { safeEqual } from "./safe-equal.js"; -import { captureTwiml } from "./capture.js"; +import { captureTwiml, captureStreamFrame } from "./capture.js"; export interface ServerConfig { accountSid: string; @@ -37,8 +43,13 @@ export interface ServerConfig { /** Basic-auth credentials Bandwidth presents on inbound webhooks (must match the app's CallbackCreds). */ webhookUser: string; webhookPassword: string; - /** Opt-in dir to persist each customer TwiML response for the later BXML Generator. Undefined → no capture. */ + /** Opt-in dir to persist each customer TwiML response for the later BXML Generator, and + * raw Bandwidth StartStream frames under /streams/. Undefined → no capture. */ captureDir?: string; + /** How long a Bandwidth StartStream WebSocket may sit without a "start" event before it is closed. Default 5000. */ + streamStartTimeoutMs?: number; + /** Passed to TwilioStreamBridge.playoutLatencyPadMs for every stream. Default 0. */ + streamPlayoutLatencyPadMs?: number; } export interface ServerDeps { @@ -110,6 +121,14 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc "Basic " + Buffer.from(`${config.accountSid}:${config.authToken}`).toString("base64"); const authOk = (req: FastifyRequest) => safeEqual(req.headers.authorization ?? "", expectedAuth); + // The WebSocket form of PUBLIC_BASE_URL: Bandwidth's StartStream destination + // must be ws(s)://, so https → wss and http → ws (local dev). + const publicWsBase = (() => { + const u = new URL(config.publicBaseUrl); + u.protocol = u.protocol === "http:" ? "ws:" : "wss:"; + return u.toString().replace(/\/$/, ""); + })(); + const rewriter = (base: string) => (url: string, kind: UrlKind) => { const absolute = new URL(url, base).toString(); // Recording-available events are async, fire-and-forget (no BXML continuation), @@ -117,9 +136,109 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc if (kind === "recordingStatus") { return `${config.publicBaseUrl}/bw/recording-status?cb=${encodeURIComponent(absolute)}`; } + // A Twilio is the customer's bot. Bandwidth speaks its own + // StartStream protocol, so the stream must come to us first; /bw/stream + // bridges it to the bot in Twilio's Media Streams protocol. + if (kind === "stream") { + return `${publicWsBase}/bw/stream?dest=${encodeURIComponent(absolute)}`; + } return `${config.publicBaseUrl}/bw/continue?next=${encodeURIComponent(absolute)}`; }; + // ── Media Streams: Bandwidth → /bw/stream → TwilioStreamBridge → bot ─────── + // Fastify has no WebSocket routing of its own, so the HTTP upgrade is taken + // off the underlying server and handed to ws. The upgrade must carry the same + // Basic-auth credentials as the /bw/* webhooks (the translator stamps them on + // StartStream as destinationUsername/destinationPassword), and `dest` goes + // through the egress guard like any customer-supplied URL. Once Bandwidth's + // "start" event arrives, a bridge is built per stream; caller audio is held + // until the bot's socket is open so the first words are not lost. + const streamServer = new WebSocketServer({ noServer: true }); + const bridges = new Set(); + + const rejectUpgrade = (socket: Duplex, status: number, reason: string, extraHeaders = ""): void => { + if (socket.destroyed) return; + socket.write(`HTTP/1.1 ${status} ${reason}\r\n${extraHeaders}Connection: close\r\nContent-Length: 0\r\n\r\n`); + socket.destroy(); + }; + + async function handleStreamUpgrade(req: IncomingMessage, socket: Duplex, head: Buffer): Promise { + const url = new URL(req.url ?? "/", "http://placeholder.invalid"); + if (url.pathname !== "/bw/stream") return rejectUpgrade(socket, 404, "Not Found"); + if (!safeEqual(req.headers.authorization ?? "", expectedWebhookAuth)) { + return rejectUpgrade(socket, 401, "Unauthorized", "WWW-Authenticate: Basic\r\n"); + } + const dest = url.searchParams.get("dest"); + if (!dest) return rejectUpgrade(socket, 400, "Bad Request"); + try { + await assertPublicUrl(dest, { + allowPrivate: config.allowPrivateEgress, + allowHosts: config.egressAllowHosts, + schemes: ["ws:", "wss:"], + }); + } catch (err) { + app.log.error({ dest, err }, "stream dest blocked"); + return err instanceof EgressBlockedError + ? rejectUpgrade(socket, 400, "Bad Request") + : rejectUpgrade(socket, 500, "Internal Server Error"); + } + if (socket.destroyed) return; + + streamServer.handleUpgrade(req, socket, head, (ws) => { + // Per-connection key for capture; never derived from Bandwidth input. + const captureKey = randomUUID(); + const source = new BwWebSocketSource(ws, { + onFrame: config.captureDir + ? (raw) => { + try { + captureStreamFrame(config.captureDir!, captureKey, raw); + } catch (err) { + app.log.error({ err }, "stream capture failed"); + } + } + : undefined, + }); + let bridge: TwilioStreamBridge | undefined; + source + .waitForStart(config.streamStartTimeoutMs ?? 5000) + .then((start) => { + bridge = new TwilioStreamBridge({ + botUrl: dest, + callSid: toCallSid(start.callId), + accountSid: config.accountSid, + customParameters: customParametersFromBwStart(start.raw), + source, + playoutLatencyPadMs: config.streamPlayoutLatencyPadMs, + }); + const b = bridge; + bridges.add(b); + source.once("stop", () => bridges.delete(b)); + app.log.info({ callId: start.callId, streamId: start.streamId, dest }, "stream bridge started"); + return b.ready(); + }) + .then(() => source.release()) + .catch((err) => { + // No start, or the bot could not be reached: end the Bandwidth stream so + // the returns and the call's BXML moves on. + app.log.error({ dest, err }, "stream bridge setup failed"); + if (bridge) { + bridges.delete(bridge); + bridge.close(); + } + source.close(); + }); + }); + } + + app.server.on("upgrade", (req, socket, head) => { + void handleStreamUpgrade(req, socket, head); + }); + app.addHook("onClose", async () => { + for (const b of bridges) b.close(); + bridges.clear(); + streamServer.close(); + }); + function errorBxml(verbs: string[]): string { return bxmlDocument([ { diff --git a/src/server/capture.ts b/src/server/capture.ts index 01edfa5..c247bb4 100644 --- a/src/server/capture.ts +++ b/src/server/capture.ts @@ -1,7 +1,37 @@ -import { mkdirSync, writeFileSync, existsSync } from "node:fs"; +import { mkdirSync, writeFileSync, existsSync, appendFileSync } from "node:fs"; import { createHash } from "node:crypto"; import { join } from "node:path"; +/** Media frames kept per stream capture; start and stop are always written. */ +const MAX_CAPTURED_MEDIA_FRAMES = 10; +const capturedMedia = new Map(); + +/** + * Append one raw Bandwidth StartStream WebSocket frame to + * `/streams/.jsonl`, one JSON frame per line, so a live call yields + * checked-in fixtures for the bridge tests (see test/fixtures/bandwidth/). + * + * `streamKey` is a per-connection identifier chosen by the server (never + * Bandwidth-supplied), hashed like captureTwiml so nothing untrusted reaches the + * filesystem path. Only the first MAX_CAPTURED_MEDIA_FRAMES media frames are + * kept; a call is 50 frames a second and the shapes are identical. Same + * private-by-default modes and synchronous I/O as captureTwiml. + * + * Returns the path written, or undefined when the frame was skipped by the cap. + */ +export function captureStreamFrame(dir: string, streamKey: string, rawFrame: string): string | undefined { + const streamsDir = join(dir, "streams"); + const file = join(streamsDir, `${createHash("sha256").update(streamKey).digest("hex").slice(0, 16)}.jsonl`); + if (/"eventType"\s*:\s*"media"/.test(rawFrame)) { + const n = capturedMedia.get(file) ?? 0; + if (n >= MAX_CAPTURED_MEDIA_FRAMES) return undefined; + capturedMedia.set(file, n + 1); + } + mkdirSync(streamsDir, { recursive: true, mode: 0o700 }); + appendFileSync(file, rawFrame.replace(/\r?\n/g, " ") + "\n", { mode: 0o600 }); + return file; +} + /** * Persist a raw TwiML response the translator fetched from the customer app, so a * later `npm run bxml-generator` over the capture dir can turn the paths a test call diff --git a/src/server/index.ts b/src/server/index.ts index 8afdea7..ee5e1a8 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -20,6 +20,9 @@ const app = buildApp( allowPrivateEgress: process.env.EGRESS_ALLOW_PRIVATE === "1", egressAllowHosts: process.env.EGRESS_ALLOW_HOSTS?.split(",").map((s) => s.trim()).filter(Boolean), captureDir: process.env.TRANSLATOR_CAPTURE_DIR, + streamPlayoutLatencyPadMs: process.env.STREAM_PLAYOUT_LATENCY_PAD_MS + ? Number(process.env.STREAM_PLAYOUT_LATENCY_PAD_MS) + : undefined, }, { fetchImpl: fetch, diff --git a/src/streams/bridge.ts b/src/streams/bridge.ts index 2b48a82..b3911fe 100644 --- a/src/streams/bridge.ts +++ b/src/streams/bridge.ts @@ -168,6 +168,14 @@ export class TwilioStreamBridge { this.ws.close(); }); + // Bot hung up its side. On Twilio that ends and TwiML resumes + // after it; here, closing the Bandwidth socket ends the StartStream so the + // translator's returns and BXML resumes likewise. + this.ws.on("close", () => { + this.dropPendingMarks(); + this.opts.source.close(); + }); + // Bot → bridge inbound message handler. The bot only ever sends // media / mark / clear back to us (per Twilio's protocol). this.ws.on("message", (data) => { diff --git a/src/streams/bw-source.ts b/src/streams/bw-source.ts new file mode 100644 index 0000000..a051b09 --- /dev/null +++ b/src/streams/bw-source.ts @@ -0,0 +1,215 @@ +import { EventEmitter } from "node:events"; +import { WebSocket } from "ws"; +import type { BwStreamSource } from "./bridge.js"; + +/** + * The Bandwidth side of the Media Streams bridge: a BwStreamSource over the + * WebSocket Bandwidth opens to a (VAPI-3991). + * + * Protocol (StartStream docs, confirmed on real calls 2026-09-17), all JSON: + * + * Bandwidth → us + * { eventType: "start", metadata: { accountId, callId, streamId, streamName, + * tracks: [{ name, mediaFormat: { encoding: "PCMU", sampleRate: 8000 } }] }, + * streamParams?: { name: value } } + * { eventType: "media", track: "inbound" | "outbound", payload: , + * sequenceNumber: "1" } + * { eventType: "stop", metadata: { ...same shape as start } } + * + * us → Bandwidth (bidirectional streams only) + * { eventType: "playAudio", media: { contentType: "audio/pcmu", payload } } + * { eventType: "clear" } + * + * Bandwidth does not deliver DTMF over this socket (it arrives via BXML Gather + * webhooks), so this source never emits "dtmf". + * + * Only the "inbound" track (the caller) is relayed: the bridge presents a single + * inbound track to the bot, matching what the translator emits (tracks="inbound"). + * Outbound-track frames, if a BXML author asked for them, are ignored. + * + * Media that arrives before release() is buffered (bounded), so the server can + * wait for the bot's WebSocket to open before the first caller audio flows and + * nothing said in the first few hundred milliseconds is lost. + */ + +export interface BwStreamTrack { + name: string; + mediaFormat?: { encoding?: string; sampleRate?: number }; +} + +export interface BwStreamStart { + accountId: string; + callId: string; + streamId: string; + streamName: string; + tracks: BwStreamTrack[]; + /** values echoed by Bandwidth, as a flat map. */ + streamParams: Record; + /** The start event exactly as received (input to customParametersFromBwStart). */ + raw: unknown; +} + +export interface BwWebSocketSourceOpts { + /** Receives every inbound frame's raw text before parsing; used for fixture capture. */ + onFrame?: (raw: string) => void; +} + +/** Frames of caller audio held before release(): 500 × 20 ms = 10 s, oldest dropped first. */ +const MAX_BUFFERED_FRAMES = 500; + +function str(v: unknown): string { + return typeof v === "string" ? v : v === undefined || v === null ? "" : String(v); +} + +function parseStart(msg: Record): BwStreamStart { + const md = (msg.metadata ?? {}) as Record; + const tracks: BwStreamTrack[] = Array.isArray(md.tracks) + ? (md.tracks as unknown[]).flatMap((t) => { + if (!t || typeof t !== "object") return []; + const tr = t as Record; + const mf = (tr.mediaFormat ?? undefined) as Record | undefined; + return [ + { + name: str(tr.name), + ...(mf ? { mediaFormat: { encoding: str(mf.encoding), sampleRate: Number(mf.sampleRate) || undefined } } : {}), + }, + ]; + }) + : []; + const streamParams: Record = Object.create(null); + const sp = msg.streamParams; + if (sp && typeof sp === "object" && !Array.isArray(sp)) { + for (const [k, v] of Object.entries(sp as Record)) { + if (typeof v === "string") streamParams[k] = v; + else if (typeof v === "number" || typeof v === "boolean") streamParams[k] = String(v); + } + } + return { + accountId: str(md.accountId), + callId: str(md.callId), + streamId: str(md.streamId), + streamName: str(md.streamName), + tracks, + streamParams, + raw: msg, + }; +} + +export class BwWebSocketSource extends EventEmitter implements BwStreamSource { + /** Set once Bandwidth's start event has been parsed. */ + start: BwStreamStart | undefined; + private released = false; + private buffered: string[] = []; + private stopped = false; + + constructor( + private ws: WebSocket, + private opts: BwWebSocketSourceOpts = {}, + ) { + super(); + ws.on("message", (data) => this.onMessage(String(data))); + // "close" always follows "error" on ws; nothing to do here except keep an + // unhandled error from crashing the process. + ws.on("error", () => {}); + ws.on("close", () => this.emitStop()); + } + + /** + * Resolve with the parsed start event, or reject if the socket closes or + * `timeoutMs` passes first. Bandwidth sends start as its first frame, so a + * timeout means whatever connected is not a Bandwidth stream. + */ + waitForStart(timeoutMs: number): Promise { + if (this.start) return Promise.resolve(this.start); + if (this.stopped) return Promise.reject(new Error("stream closed before start")); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`no start event within ${timeoutMs} ms`)); + }, timeoutMs); + const onStart = (s: BwStreamStart) => { + cleanup(); + resolve(s); + }; + const onStop = () => { + cleanup(); + reject(new Error("stream closed before start")); + }; + const cleanup = () => { + clearTimeout(timer); + this.off("start", onStart); + this.off("stop", onStop); + }; + this.once("start", onStart); + this.once("stop", onStop); + }); + } + + /** Start emitting "media"; frames received so far are emitted first, in order. */ + release(): void { + if (this.released) return; + this.released = true; + const pending = this.buffered; + this.buffered = []; + for (const p of pending) this.emit("media", p); + } + + /** Frames of caller audio currently held back (0 once released). */ + bufferedFrames(): number { + return this.buffered.length; + } + + sendMedia(payloadB64: string): void { + this.sendJson({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: payloadB64 } }); + } + + flush(): void { + this.sendJson({ eventType: "clear" }); + } + + close(): void { + if (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING) this.ws.close(); + } + + private sendJson(obj: unknown): void { + if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(obj)); + } + + private onMessage(raw: string): void { + this.opts.onFrame?.(raw); + let msg: unknown; + try { + msg = JSON.parse(raw); + } catch { + return; // not one of ours; ignore rather than tear the stream down + } + if (!msg || typeof msg !== "object") return; + const m = msg as Record; + switch (m.eventType) { + case "start": + this.start = parseStart(m); + this.emit("start", this.start); + break; + case "media": { + if (typeof m.payload !== "string") return; + if (m.track !== undefined && m.track !== "inbound") return; + if (this.released) { + this.emit("media", m.payload); + } else { + if (this.buffered.length >= MAX_BUFFERED_FRAMES) this.buffered.shift(); + this.buffered.push(m.payload); + } + break; + } + case "stop": + this.emitStop(); + break; + } + } + + private emitStop(): void { + if (this.stopped) return; + this.stopped = true; + this.emit("stop"); + } +} diff --git a/src/translator/translate.ts b/src/translator/translate.ts index 5c10ec6..e342c28 100644 --- a/src/translator/translate.ts +++ b/src/translator/translate.ts @@ -178,13 +178,20 @@ const CALLBACK_URL_ATTRS = [ /** Walks the built element tree and stamps username/password onto any element * carrying a rewritten translator callback URL, so Bandwidth Basic-auths the - * continuation request instead of hitting it unauthenticated. */ + * continuation request instead of hitting it unauthenticated. A StartStream's + * destination is the translator's own /bw/stream WebSocket after rewriting, so + * it gets the same credentials as destinationUsername/destinationPassword, + * which Bandwidth presents as the Authorization header on the upgrade. */ function stampCallbackAuth(els: XmlEl[], auth: { username: string; password: string }): void { for (const el of els) { if (el.attrs && CALLBACK_URL_ATTRS.some((a) => el.attrs![a] !== undefined)) { el.attrs.username = auth.username; el.attrs.password = auth.password; } + if (el.name === "StartStream" && el.attrs?.destination !== undefined) { + el.attrs.destinationUsername = auth.username; + el.attrs.destinationPassword = auth.password; + } if (el.children) { const childEls = el.children.filter( (c): c is XmlEl => typeof c !== "string" && !("raw" in c), diff --git a/src/twilio/egress-guard.ts b/src/twilio/egress-guard.ts index e028723..3bf1cd6 100644 --- a/src/twilio/egress-guard.ts +++ b/src/twilio/egress-guard.ts @@ -40,7 +40,13 @@ export function isBlockedAddress(ip: string): boolean { export async function assertPublicUrl( rawUrl: string, - opts: { allowPrivate?: boolean; allowHosts?: string[]; lookup?: (host: string) => Promise } = {}, + opts: { + allowPrivate?: boolean; + allowHosts?: string[]; + lookup?: (host: string) => Promise; + /** Accepted URL schemes (with colon). Default http/https; the stream route passes ws/wss. */ + schemes?: string[]; + } = {}, ): Promise { let url: URL; try { @@ -48,8 +54,8 @@ export async function assertPublicUrl( } catch { throw new EgressBlockedError(`Malformed URL: ${rawUrl}`); } - if (url.protocol !== "http:" && url.protocol !== "https:") - throw new EgressBlockedError(`Disallowed scheme: ${url.protocol}`); + const schemes = opts.schemes ?? ["http:", "https:"]; + if (!schemes.includes(url.protocol)) throw new EgressBlockedError(`Disallowed scheme: ${url.protocol}`); // Opt-in host allowlist ("full remediation"): default-deny by host. Listed // hosts are explicitly trusted, so they bypass the range denylist below. diff --git a/test/fixtures/bandwidth/stream-frames.json b/test/fixtures/bandwidth/stream-frames.json new file mode 100644 index 0000000..0a5f611 --- /dev/null +++ b/test/fixtures/bandwidth/stream-frames.json @@ -0,0 +1,64 @@ +{ + "_meta": { + "captured": "2026-09-17", + "source": "start.metadata: live Bandwidth on a real call, recorded by the local repro app. Field names, nesting, and the PCMU/8000 track format are exactly as received.", + "note": "streamParams was absent on that call because the BXML carried no ; the value here follows the StartStream docs example. media and stop follow the docs' documented shapes; the media payload is 160 bytes (20 ms) of synthetic PCMU silence, not recorded audio. Replace this file with frames the translator captures to /streams/*.jsonl on the first live end-to-end call.", + "redacted": [ + "accountId", + "callId", + "streamId" + ] + }, + "start": { + "eventType": "start", + "metadata": { + "accountId": "99XXXXX", + "callId": "c-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamId": "s-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamName": "connect-stream-1", + "tracks": [ + { + "name": "inbound", + "mediaFormat": { + "encoding": "PCMU", + "sampleRate": 8000 + } + } + ] + }, + "streamParams": { + "callSid": "CA123", + "tenant": "acme" + } + }, + "media": { + "eventType": "media", + "track": "inbound", + "payload": "/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////w==", + "sequenceNumber": "1" + }, + "mediaOutbound": { + "eventType": "media", + "track": "outbound", + "payload": "/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////w==", + "sequenceNumber": "1" + }, + "stop": { + "eventType": "stop", + "metadata": { + "accountId": "99XXXXX", + "callId": "c-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamId": "s-xxxxxxxx-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "streamName": "connect-stream-1", + "tracks": [ + { + "name": "inbound", + "mediaFormat": { + "encoding": "PCMU", + "sampleRate": 8000 + } + } + ] + } + } +} diff --git a/test/server-stream.test.ts b/test/server-stream.test.ts new file mode 100644 index 0000000..6d009c0 --- /dev/null +++ b/test/server-stream.test.ts @@ -0,0 +1,315 @@ +/** + * /bw/stream: the translator's StartStream destination (VAPI-3991). + * + * Bandwidth (played by a ws client) connects to the running Fastify server with + * the fixture frames; a local WebSocketServer plays the customer's Twilio bot. + * Everything runs on 127.0.0.1 with allowPrivateEgress so ports are ephemeral. + */ +import { describe, it, expect, vi } from "vitest"; +import { WebSocketServer, WebSocket } from "ws"; +import type { AddressInfo } from "node:net"; +import { readFileSync, mkdtempSync, rmSync, readdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { buildApp } from "../src/server/app.js"; +import { toCallSid } from "../src/twilio/call-sid.js"; + +const frames = JSON.parse( + readFileSync(new URL("./fixtures/bandwidth/stream-frames.json", import.meta.url), "utf8"), +); +const webhookAuth = "Basic " + Buffer.from("u:p").toString("base64"); + +function config(extra: Record = {}) { + return { + accountSid: "AC123", + authToken: "tok", + publicBaseUrl: "https://translator.test", + voiceUrl: "https://customer.test/voice", + allowPrivateEgress: true, + webhookUser: "u", + webhookPassword: "p", + streamStartTimeoutMs: 300, + ...extra, + }; +} + +function makeApp(cfg = config(), twimlByUrl: Record = {}) { + const fetchImpl = vi.fn(async (url: any) => { + const twiml = twimlByUrl[String(url)]; + return twiml ? new Response(twiml, { status: 200 }) : new Response("not found", { status: 404 }); + }) as unknown as typeof fetch; + const bwClient = { + createCall: vi.fn(), modifyCall: vi.fn(), getCall: vi.fn(), listRecordings: vi.fn(), + getRecording: vi.fn(), getRecordingMedia: vi.fn(), updateRecording: vi.fn(), + }; + return buildApp(cfg, { fetchImpl, bwClient }); +} + +async function listen(app: ReturnType): Promise { + await app.listen({ port: 0, host: "127.0.0.1" }); + return (app.server.address() as AddressInfo).port; +} + +/** The customer's bot, speaking Twilio Media Streams. */ +async function fakeBot() { + const wss = new WebSocketServer({ port: 0, host: "127.0.0.1" }); + await new Promise((r) => wss.once("listening", r)); + const port = (wss.address() as AddressInfo).port; + const messages: any[] = []; + const socket = new Promise((res) => + wss.on("connection", (ws) => { + ws.on("message", (d) => messages.push(JSON.parse(String(d)))); + res(ws); + }), + ); + return { url: `ws://127.0.0.1:${port}/bot`, messages, socket, close: () => wss.close() }; +} + +/** Connect as Bandwidth would: the rewritten destination plus destination creds. */ +function bandwidthClient(port: number, dest: string, headers: Record = { authorization: webhookAuth }) { + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/stream?dest=${encodeURIComponent(dest)}`, { headers }); + const received: any[] = []; + ws.on("message", (d) => received.push(JSON.parse(String(d)))); + return { ws, received, send: (obj: unknown) => ws.send(JSON.stringify(obj)) }; +} + +/** 101 on a completed upgrade, otherwise the HTTP status the server rejected with. */ +function upgradeStatus(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + ws.once("open", () => resolve(101)); + ws.once("unexpected-response", (_req, res) => { + res.resume(); + resolve(res.statusCode ?? 0); + }); + ws.once("error", reject); + }); +} + +async function waitFor(check: () => boolean, ms = 1000): Promise { + const deadline = Date.now() + ms; + while (!check()) { + if (Date.now() > deadline) throw new Error("waitFor timeout"); + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("/bw/stream end to end", () => { + it("bridges a Bandwidth stream to the bot in Twilio's protocol, both directions, then stops", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + expect(await upgradeStatus(bw.ws)).toBe(101); + + // Bandwidth → bot: start becomes connected + start with our ids and customParameters. + bw.send(frames.start); + const botWs = await bot.socket; + await waitFor(() => bot.messages.length >= 2); + expect(bot.messages[0]).toMatchObject({ event: "connected", protocol: "Call" }); + expect(bot.messages[1].event).toBe("start"); + expect(bot.messages[1].start.callSid).toBe(toCallSid(frames.start.metadata.callId)); + expect(bot.messages[1].start.accountSid).toBe("AC123"); + expect(bot.messages[1].start.customParameters).toEqual({ callSid: "CA123", tenant: "acme" }); + expect(bot.messages[1].start.mediaFormat).toEqual({ encoding: "audio/x-mulaw", sampleRate: 8000, channels: 1 }); + + // Caller audio reaches the bot as a Twilio media frame. + bw.send(frames.media); + await waitFor(() => bot.messages.some((m) => m.event === "media")); + expect(bot.messages.find((m) => m.event === "media").media).toMatchObject({ + track: "inbound", + payload: frames.media.payload, + }); + + // Bot → Bandwidth: media becomes playAudio, clear becomes clear. + const streamSid = bot.messages[1].streamSid; + botWs.send(JSON.stringify({ event: "media", streamSid, media: { payload: "QUJD" } })); + botWs.send(JSON.stringify({ event: "clear", streamSid })); + await waitFor(() => bw.received.length >= 2); + expect(bw.received[0]).toEqual({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: "QUJD" } }); + expect(bw.received[1]).toEqual({ eventType: "clear" }); + + // Bandwidth ends the stream: the bot gets stop and its socket is closed. + const botClosed = new Promise((r) => botWs.once("close", r)); + bw.send(frames.stop); + await botClosed; + expect(bot.messages.at(-1)).toMatchObject({ event: "stop", stop: { accountSid: "AC123" } }); + + bw.ws.close(); + bot.close(); + await app.close(); + }); + + it("holds caller audio sent before the bot's socket is open and delivers it in order", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + + bw.send(frames.start); + for (const p of ["AAAA", "BBBB", "CCCC"]) bw.send({ ...frames.media, payload: p }); + + await waitFor(() => bot.messages.filter((m) => m.event === "media").length === 3); + expect(bot.messages.filter((m) => m.event === "media").map((m) => m.media.payload)).toEqual(["AAAA", "BBBB", "CCCC"]); + // And they arrive after connected/start, never before. + expect(bot.messages.findIndex((m) => m.event === "media")).toBeGreaterThan(1); + + bw.ws.close(); + bot.close(); + await app.close(); + }); + + it("ends the Bandwidth stream when the bot closes its socket", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + bw.send(frames.start); + const botWs = await bot.socket; + await waitFor(() => bot.messages.length >= 2); + + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + botWs.close(); + await bwClosed; + + bot.close(); + await app.close(); + }); + + it("ends the Bandwidth stream when the bot cannot be reached", async () => { + const app = makeApp(); + const port = await listen(app); + // Nothing listens on this port; the bridge's connect fails. + const bw = bandwidthClient(port, "ws://127.0.0.1:1/bot"); + await upgradeStatus(bw.ws); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + bw.send(frames.start); + await bwClosed; + await app.close(); + }); + + it("closes a socket that sends no start event within the timeout", async () => { + const app = makeApp(); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + const openedAt = Date.now(); + await new Promise((r) => bw.ws.once("close", r)); + expect(Date.now() - openedAt).toBeGreaterThanOrEqual(250); + expect(Date.now() - openedAt).toBeLessThan(2000); + bot.close(); + await app.close(); + }); +}); + +describe("/bw/stream upgrade gate", () => { + it("rejects an upgrade without the webhook credentials (401)", async () => { + const app = makeApp(); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot", {}); + expect(await upgradeStatus(bw.ws)).toBe(401); + const wrong = bandwidthClient(port, "ws://127.0.0.1:9/bot", { + authorization: "Basic " + Buffer.from("u:wrong").toString("base64"), + }); + expect(await upgradeStatus(wrong.ws)).toBe(401); + await app.close(); + }); + + it("rejects a missing dest (400)", async () => { + const app = makeApp(); + const port = await listen(app); + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/stream`, { headers: { authorization: webhookAuth } }); + expect(await upgradeStatus(ws)).toBe(400); + await app.close(); + }); + + it("rejects a dest that is not a WebSocket URL (400)", async () => { + const app = makeApp(); + const port = await listen(app); + const bw = bandwidthClient(port, "https://bot.example/audio"); + expect(await upgradeStatus(bw.ws)).toBe(400); + await app.close(); + }); + + it("rejects a private-network dest when private egress is not allowed (400)", async () => { + const app = makeApp(config({ allowPrivateEgress: false })); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot"); + expect(await upgradeStatus(bw.ws)).toBe(400); + await app.close(); + }); + + it("rejects an upgrade on any other path (404)", async () => { + const app = makeApp(); + const port = await listen(app); + const ws = new WebSocket(`ws://127.0.0.1:${port}/bw/other?dest=ws%3A%2F%2Fx`, { headers: { authorization: webhookAuth } }); + expect(await upgradeStatus(ws)).toBe(404); + await app.close(); + }); +}); + +describe("translator side of the stream route", () => { + it("rewrites a Connect/Stream destination to wss:///bw/stream?dest= and stamps destination creds", async () => { + const app = makeApp(config(), { + "https://customer.test/voice": ``, + }); + const res = await app.inject({ + method: "POST", + url: "/bw/initiate", + headers: { authorization: webhookAuth }, + payload: { eventType: "initiate", callId: "c-stream-1", from: "+1", to: "+2", direction: "inbound" }, + }); + expect(res.statusCode).toBe(200); + expect(res.body).toContain(`destination="wss://translator.test/bw/stream?dest=wss%3A%2F%2Fbot.example%2Faudio"`); + expect(res.body).toContain(`destinationUsername="u"`); + expect(res.body).toContain(`destinationPassword="p"`); + expect(res.body).toContain(``); + expect(res.body).toContain(``); + }); + + it("uses ws:// when PUBLIC_BASE_URL is plain http (local dev)", async () => { + const app = makeApp(config({ publicBaseUrl: "http://localhost:3000" }), { + "https://customer.test/voice": ``, + }); + const res = await app.inject({ + method: "POST", + url: "/bw/initiate", + headers: { authorization: webhookAuth }, + payload: { eventType: "initiate", callId: "c-stream-2", from: "+1", to: "+2", direction: "inbound" }, + }); + expect(res.body).toContain(`destination="ws://localhost:3000/bw/stream?dest=`); + }); +}); + +describe("stream frame capture", () => { + it("writes start, media, and stop frames to /streams/*.jsonl when captureDir is set", async () => { + const dir = mkdtempSync(join(tmpdir(), "stream-capture-")); + try { + const app = makeApp(config({ captureDir: dir })); + const port = await listen(app); + const bot = await fakeBot(); + const bw = bandwidthClient(port, bot.url); + await upgradeStatus(bw.ws); + bw.send(frames.start); + const botWs = await bot.socket; + bw.send(frames.media); + const botClosed = new Promise((r) => botWs.once("close", r)); + bw.send(frames.stop); + await botClosed; + bw.ws.close(); + bot.close(); + await app.close(); + + const files = readdirSync(join(dir, "streams")); + expect(files).toHaveLength(1); + expect(files[0]).toMatch(/^[0-9a-f]{16}\.jsonl$/); + const lines = readFileSync(join(dir, "streams", files[0]), "utf8").trim().split("\n").map((l) => JSON.parse(l)); + expect(lines.map((l) => l.eventType)).toEqual(["start", "media", "stop"]); + expect(lines[0].metadata.streamId).toBe(frames.start.metadata.streamId); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/streams-bw-source.test.ts b/test/streams-bw-source.test.ts new file mode 100644 index 0000000..f18106f --- /dev/null +++ b/test/streams-bw-source.test.ts @@ -0,0 +1,232 @@ +/** + * BwWebSocketSource: the Bandwidth side of the Media Streams bridge (VAPI-3991). + * + * Pattern: a local WebSocketServer stands in for the translator's /bw/stream + * endpoint and a ws client plays "Bandwidth", sending the frames in + * test/fixtures/bandwidth/stream-frames.json. + */ +import { describe, it, expect } from "vitest"; +import { WebSocketServer, WebSocket } from "ws"; +import { readFileSync } from "node:fs"; +import type { AddressInfo } from "node:net"; +import { BwWebSocketSource, type BwWebSocketSourceOpts } from "../src/streams/bw-source.js"; +import { customParametersFromBwStart } from "../src/streams/bridge.js"; + +const frames = JSON.parse( + readFileSync(new URL("./fixtures/bandwidth/stream-frames.json", import.meta.url), "utf8"), +); + +async function pair(opts?: BwWebSocketSourceOpts) { + const wss = new WebSocketServer({ port: 0, host: "127.0.0.1" }); + await new Promise((r) => wss.once("listening", r)); + const port = (wss.address() as AddressInfo).port; + const sourceP = new Promise((res) => + wss.once("connection", (ws) => res(new BwWebSocketSource(ws, opts))), + ); + const bandwidth = new WebSocket(`ws://127.0.0.1:${port}/bw/stream`); + const received: any[] = []; + bandwidth.on("message", (d) => received.push(JSON.parse(String(d)))); + await new Promise((r) => bandwidth.once("open", r)); + const source = await sourceP; + const send = (obj: unknown) => bandwidth.send(typeof obj === "string" ? obj : JSON.stringify(obj)); + return { + source, + bandwidth, + received, + send, + close: () => { + bandwidth.close(); + wss.close(); + }, + }; +} + +async function waitFor(check: () => boolean, ms = 500): Promise { + const deadline = Date.now() + ms; + while (!check()) { + if (Date.now() > deadline) throw new Error("waitFor timeout"); + await new Promise((r) => setTimeout(r, 10)); + } +} + +describe("BwWebSocketSource: start", () => { + it("parses the recorded start event: ids, tracks, and streamParams", async () => { + const t = await pair(); + t.send(frames.start); + const start = await t.source.waitForStart(1000); + t.close(); + + expect(start.accountId).toBe(frames.start.metadata.accountId); + expect(start.callId).toBe(frames.start.metadata.callId); + expect(start.streamId).toBe(frames.start.metadata.streamId); + expect(start.streamName).toBe("connect-stream-1"); + expect(start.tracks).toEqual([{ name: "inbound", mediaFormat: { encoding: "PCMU", sampleRate: 8000 } }]); + expect(start.streamParams).toEqual({ callSid: "CA123", tenant: "acme" }); + // The raw event feeds the bridge's customParameters mapper unchanged. + expect(customParametersFromBwStart(start.raw)).toEqual({ callSid: "CA123", tenant: "acme" }); + expect(t.source.start).toBe(start); + }); + + it("tolerates a start event with no metadata or streamParams", async () => { + const t = await pair(); + t.send({ eventType: "start" }); + const start = await t.source.waitForStart(1000); + t.close(); + expect(start).toMatchObject({ accountId: "", callId: "", streamId: "", streamName: "", tracks: [] }); + expect(start.streamParams).toEqual({}); + }); + + it("waitForStart resolves immediately if start already arrived", async () => { + const t = await pair(); + t.send(frames.start); + await waitFor(() => t.source.start !== undefined); + const start = await t.source.waitForStart(1); + t.close(); + expect(start.callId).toBe(frames.start.metadata.callId); + }); + + it("waitForStart rejects when no start arrives in time", async () => { + const t = await pair(); + await expect(t.source.waitForStart(50)).rejects.toThrow(/no start event within 50 ms/); + t.close(); + }); + + it("waitForStart rejects when Bandwidth closes first", async () => { + const t = await pair(); + const p = t.source.waitForStart(1000); + t.bandwidth.close(); + await expect(p).rejects.toThrow(/closed before start/); + t.close(); + }); +}); + +describe("BwWebSocketSource: media (Bandwidth → bot)", () => { + it("buffers inbound frames until release(), then emits them in order and streams the rest live", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + + t.send({ ...frames.media, payload: "AAAA", sequenceNumber: "1" }); + t.send({ ...frames.media, payload: "BBBB", sequenceNumber: "2" }); + t.send({ ...frames.media, payload: "CCCC", sequenceNumber: "3" }); + await waitFor(() => t.source.bufferedFrames() === 3); + expect(got).toEqual([]); + + t.source.release(); + expect(got).toEqual(["AAAA", "BBBB", "CCCC"]); + expect(t.source.bufferedFrames()).toBe(0); + + t.send({ ...frames.media, payload: "DDDD", sequenceNumber: "4" }); + await waitFor(() => got.length === 4); + t.close(); + expect(got[3]).toBe("DDDD"); + }); + + it("relays the recorded media frame's payload verbatim", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.send(frames.media); + await waitFor(() => got.length === 1); + t.close(); + expect(got[0]).toBe(frames.media.payload); + expect(Buffer.from(got[0], "base64")).toHaveLength(160); // 20 ms of PCMU + }); + + it("ignores outbound-track frames and frames without a string payload", async () => { + const t = await pair(); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.send(frames.mediaOutbound); + t.send({ eventType: "media", track: "inbound", payload: 123 }); + t.send({ eventType: "media", track: "inbound" }); + t.send({ ...frames.media, payload: "ZZZZ" }); + await waitFor(() => got.length === 1); + await new Promise((r) => setTimeout(r, 30)); + t.close(); + expect(got).toEqual(["ZZZZ"]); + }); + + it("drops the oldest buffered frames beyond the cap (10 s of audio)", async () => { + const t = await pair(); + for (let i = 0; i < 505; i++) t.send({ ...frames.media, payload: `f${i}`, sequenceNumber: String(i + 1) }); + await waitFor(() => t.source.bufferedFrames() === 500, 2000); + const got: string[] = []; + t.source.on("media", (p: string) => got.push(p)); + t.source.release(); + t.close(); + expect(got).toHaveLength(500); + expect(got[0]).toBe("f5"); + expect(got[499]).toBe("f504"); + }); + + it("ignores non-JSON frames and unknown event types", async () => { + const t = await pair(); + t.send("not json at all"); + t.send({ eventType: "somethingNew", data: 1 }); + t.send(frames.start); + const start = await t.source.waitForStart(1000); + t.close(); + expect(start.streamName).toBe("connect-stream-1"); + }); +}); + +describe("BwWebSocketSource: stop", () => { + it("emits stop exactly once for a stop event followed by the socket closing", async () => { + const t = await pair(); + let stops = 0; + t.source.on("stop", () => stops++); + t.send(frames.stop); + await waitFor(() => stops === 1); + t.bandwidth.close(); + await new Promise((r) => setTimeout(r, 50)); + t.close(); + expect(stops).toBe(1); + }); + + it("emits stop when the socket closes without a stop event", async () => { + const t = await pair(); + let stops = 0; + t.source.on("stop", () => stops++); + t.bandwidth.close(); + await waitFor(() => stops === 1); + t.close(); + expect(stops).toBe(1); + }); +}); + +describe("BwWebSocketSource: bot → Bandwidth", () => { + it("sendMedia becomes playAudio audio/pcmu and flush becomes clear", async () => { + const t = await pair(); + t.source.sendMedia("QUJD"); + t.source.flush(); + await waitFor(() => t.received.length >= 2); + t.close(); + expect(t.received[0]).toEqual({ eventType: "playAudio", media: { contentType: "audio/pcmu", payload: "QUJD" } }); + expect(t.received[1]).toEqual({ eventType: "clear" }); + }); + + it("close() closes Bandwidth's socket and is safe to call twice", async () => { + const t = await pair(); + const closed = new Promise((r) => t.bandwidth.once("close", r)); + t.source.close(); + t.source.close(); + await closed; + t.close(); + }); + + it("onFrame receives every raw frame before parsing", async () => { + const raws: string[] = []; + const t = await pair({ onFrame: (raw) => raws.push(raw) }); + t.send(frames.start); + t.send("garbage"); + t.send(frames.media); + await waitFor(() => raws.length === 3); + t.close(); + expect(raws[0]).toContain('"eventType":"start"'); + expect(raws[1]).toBe("garbage"); + expect(raws[2]).toContain('"eventType":"media"'); + }); +}); diff --git a/test/translate-callback-auth.test.ts b/test/translate-callback-auth.test.ts index b161f32..b76ff14 100644 --- a/test/translate-callback-auth.test.ts +++ b/test/translate-callback-auth.test.ts @@ -17,6 +17,19 @@ describe("callback auth stamping", () => { const { bxml } = translateTwiml(`https://c.test/next`, { rewriteUrl, callbackAuth: auth }); expect(bxml).toContain(`username="bw-user"`); }); + it("adds destinationUsername/destinationPassword to a StartStream (Bandwidth sends them on the WebSocket upgrade)", () => { + const { bxml } = translateTwiml( + ``, + { rewriteUrl: (u) => `wss://translator.test/bw/stream?dest=${encodeURIComponent(u)}`, callbackAuth: auth }, + ); + expect(bxml).toMatch(/]*destinationUsername="bw-user"[^>]*destinationPassword="bw-pass"/); + // The StopStream that follows has no URL and gets nothing. + expect(bxml).toMatch(//); + }); + it("omits destination credentials from StartStream when callbackAuth is not provided", () => { + const { bxml } = translateTwiml(``, { rewriteUrl }); + expect(bxml).not.toContain("destinationUsername="); + }); it("omits credentials when callbackAuth is not provided", () => { const { bxml } = translateTwiml(`https://c.test/next`, { rewriteUrl }); expect(bxml).not.toContain("username="); From 3f45e2fc9898966fe67f8c4810b68f283acfea76 Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Fri, 25 Sep 2026 11:52:10 -0400 Subject: [PATCH 2/4] VAPI-3991: harden the upgrade path, bound the bot handshake, tidy shutdown Review follow-ups: - Attach an error listener to the raw upgrade socket at the top of the handler. Node hands it over with none, ws only adds one inside handleUpgrade, and the egress check awaits a DNS lookup in between; a reset in that window was an uncaught exception that would kill the server. - Bound the bot's WebSocket handshake. TwilioStreamBridge takes connectTimeoutMs (default 10 s) and passes it to ws as handshakeTimeout, so a bot that accepts TCP but never answers the upgrade makes ready() reject and the server ends the Bandwidth stream instead of holding the caller in silence until the OS gave up. Wired as ServerConfig.streamBotConnectTimeoutMs / STREAM_BOT_CONNECT_TIMEOUT_MS; STREAM_START_TIMEOUT_MS added alongside for symmetry. - Move stream teardown from onClose to preClose. Fastify waits for HTTP connections to drain before onClose runs and an upgraded socket counts as one, so close() hung until Bandwidth hung up. preClose also terminates streamServer.clients (noServer mode does not on close()), covering a socket that upgraded but never sent start. The waitForStart timer is unref'd. - captureStreamFrame forgets a stream's media tally when its stop frame is captured, so the module map no longer grows per call. - State the credential-stamping invariant in a comment: callbackAuth only travels with the server's rewriter, which always points stream URLs at the translator. Tests: black-hole bot at both the bridge and server level, app.close() with a pre-start socket, the 10-frame capture cap (12 sent, 10 kept), and a loosened upper bound on the start-timeout wall-clock assertion. --- AGENTS.md | 10 ++++--- src/server/app.ts | 16 ++++++++++- src/server/capture.ts | 3 +++ src/server/index.ts | 14 +++++++--- src/streams/bridge.ts | 8 +++++- src/streams/bw-source.ts | 1 + src/translator/translate.ts | 4 +++ test/server-stream.test.ts | 53 +++++++++++++++++++++++++++++++++---- test/streams-wire.test.ts | 25 +++++++++++++++++ 9 files changed, 121 insertions(+), 13 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 3665667..0c73757 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -76,9 +76,13 @@ Set these env vars (the server reads **these exact names** — note the checked- `PUBLIC_BASE_URL`. The BW Voice Application's callback (set in Phase 2) must point at `/bw/initiate`. If the customer app uses ``, the host must also pass WebSocket upgrades through to the translator at -`/bw/stream` (ngrok and cloudflared do by default). Optional: -`STREAM_PLAYOUT_LATENCY_PAD_MS` delays mark acknowledgements to the bot by that -many ms to absorb network and jitter-buffer latency; default 0. +`/bw/stream` (ngrok and cloudflared do by default). Optional stream tuning, all +in ms: `STREAM_PLAYOUT_LATENCY_PAD_MS` delays mark acknowledgements to the bot +to absorb network and jitter-buffer latency (default 0); +`STREAM_START_TIMEOUT_MS` bounds the wait for Bandwidth's `start` event +(default 5000); `STREAM_BOT_CONNECT_TIMEOUT_MS` bounds the bot's WebSocket +handshake (default 10000). Either timeout ends the Bandwidth stream so the +call's BXML moves on. ### Phase 5 — Verify ```bash diff --git a/src/server/app.ts b/src/server/app.ts index 92308c9..fc36ec1 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -48,6 +48,9 @@ export interface ServerConfig { captureDir?: string; /** How long a Bandwidth StartStream WebSocket may sit without a "start" event before it is closed. Default 5000. */ streamStartTimeoutMs?: number; + /** How long the bridge waits for the bot's WebSocket handshake before giving up and ending the + * Bandwidth stream. Bounds a bot that accepts TCP but never completes the upgrade. Default 10000. */ + streamBotConnectTimeoutMs?: number; /** Passed to TwilioStreamBridge.playoutLatencyPadMs for every stream. Default 0. */ streamPlayoutLatencyPadMs?: number; } @@ -163,6 +166,10 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc }; async function handleStreamUpgrade(req: IncomingMessage, socket: Duplex, head: Buffer): Promise { + // Node hands over the raw socket with no error listener; ws attaches one only + // inside handleUpgrade, and the egress check below can await a DNS lookup for + // seconds. A reset in that window would otherwise be an uncaught exception. + socket.on("error", () => socket.destroy()); const url = new URL(req.url ?? "/", "http://placeholder.invalid"); if (url.pathname !== "/bw/stream") return rejectUpgrade(socket, 404, "Not Found"); if (!safeEqual(req.headers.authorization ?? "", expectedWebhookAuth)) { @@ -209,6 +216,7 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc customParameters: customParametersFromBwStart(start.raw), source, playoutLatencyPadMs: config.streamPlayoutLatencyPadMs, + connectTimeoutMs: config.streamBotConnectTimeoutMs, }); const b = bridge; bridges.add(b); @@ -233,9 +241,15 @@ export function buildApp(config: ServerConfig, deps: ServerDeps): FastifyInstanc app.server.on("upgrade", (req, socket, head) => { void handleStreamUpgrade(req, socket, head); }); - app.addHook("onClose", async () => { + // preClose, not onClose: Fastify waits for the HTTP server's connections to + // drain before onClose runs, and an upgraded socket counts as one of them, so + // it must be torn down first or close() would wait for Bandwidth to hang up. + app.addHook("preClose", async () => { for (const b of bridges) b.close(); bridges.clear(); + // In noServer mode close() does not end established clients, and a socket + // that upgraded but has not sent "start" yet has no bridge to close it. + for (const client of streamServer.clients) client.terminate(); streamServer.close(); }); diff --git a/src/server/capture.ts b/src/server/capture.ts index c247bb4..b868230 100644 --- a/src/server/capture.ts +++ b/src/server/capture.ts @@ -26,6 +26,9 @@ export function captureStreamFrame(dir: string, streamKey: string, rawFrame: str const n = capturedMedia.get(file) ?? 0; if (n >= MAX_CAPTURED_MEDIA_FRAMES) return undefined; capturedMedia.set(file, n + 1); + } else if (/"eventType"\s*:\s*"stop"/.test(rawFrame)) { + // The stream is over; forget its tally so the map does not grow per call. + capturedMedia.delete(file); } mkdirSync(streamsDir, { recursive: true, mode: 0o700 }); appendFileSync(file, rawFrame.replace(/\r?\n/g, " ") + "\n", { mode: 0o600 }); diff --git a/src/server/index.ts b/src/server/index.ts index ee5e1a8..2f7ea66 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -9,6 +9,14 @@ function env(name: string): string { const bwEnv = process.env.BW_ENVIRONMENT === "test" ? "test" : "prod"; +/** Optional millisecond setting; unset or non-numeric leaves the code default in place. */ +function optionalMs(name: string): number | undefined { + const v = process.env[name]; + if (!v) return undefined; + const n = Number(v); + return Number.isFinite(n) && n >= 0 ? n : undefined; +} + const app = buildApp( { accountSid: env("TRANSLATOR_ACCOUNT_SID"), @@ -20,9 +28,9 @@ const app = buildApp( allowPrivateEgress: process.env.EGRESS_ALLOW_PRIVATE === "1", egressAllowHosts: process.env.EGRESS_ALLOW_HOSTS?.split(",").map((s) => s.trim()).filter(Boolean), captureDir: process.env.TRANSLATOR_CAPTURE_DIR, - streamPlayoutLatencyPadMs: process.env.STREAM_PLAYOUT_LATENCY_PAD_MS - ? Number(process.env.STREAM_PLAYOUT_LATENCY_PAD_MS) - : undefined, + streamPlayoutLatencyPadMs: optionalMs("STREAM_PLAYOUT_LATENCY_PAD_MS"), + streamStartTimeoutMs: optionalMs("STREAM_START_TIMEOUT_MS"), + streamBotConnectTimeoutMs: optionalMs("STREAM_BOT_CONNECT_TIMEOUT_MS"), }, { fetchImpl: fetch, diff --git a/src/streams/bridge.ts b/src/streams/bridge.ts index b3911fe..f368ecd 100644 --- a/src/streams/bridge.ts +++ b/src/streams/bridge.ts @@ -50,8 +50,14 @@ export interface BridgeOpts { * return slightly early in the same direction as the original bug. Default 0 * until real measurements from VAPI-3991 give a value worth setting. */ playoutLatencyPadMs?: number; + /** Bot WebSocket handshake timeout in ms; ready() rejects when it elapses. Default 10000. + * Without it a bot that accepts TCP but never answers the upgrade holds the stream + * open until the OS gives up, which can be a minute or more of silence for the caller. */ + connectTimeoutMs?: number; } +const DEFAULT_CONNECT_TIMEOUT_MS = 10_000; + /** * Map Bandwidth's StartStream WebSocket "start" event to Twilio `customParameters`. * @@ -100,7 +106,7 @@ export class TwilioStreamBridge { constructor(private opts: BridgeOpts) { this.streamSid = "MZ" + randomBytes(16).toString("hex"); - this.ws = new WebSocket(opts.botUrl); + this.ws = new WebSocket(opts.botUrl, { handshakeTimeout: opts.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS }); this.readyPromise = new Promise((resolve, reject) => { this.ws.on("open", () => { // 1. connected — protocol handshake (no sequenceNumber per Twilio spec) diff --git a/src/streams/bw-source.ts b/src/streams/bw-source.ts index a051b09..0b95a6c 100644 --- a/src/streams/bw-source.ts +++ b/src/streams/bw-source.ts @@ -127,6 +127,7 @@ export class BwWebSocketSource extends EventEmitter implements BwStreamSource { cleanup(); reject(new Error(`no start event within ${timeoutMs} ms`)); }, timeoutMs); + timer.unref?.(); // never the only thing keeping the process (or a test) alive const onStart = (s: BwStreamStart) => { cleanup(); resolve(s); diff --git a/src/translator/translate.ts b/src/translator/translate.ts index e342c28..3ad9e3f 100644 --- a/src/translator/translate.ts +++ b/src/translator/translate.ts @@ -188,6 +188,10 @@ function stampCallbackAuth(els: XmlEl[], auth: { username: string; password: str el.attrs.username = auth.username; el.attrs.password = auth.password; } + // INVARIANT: callbackAuth is only passed together with the server's rewriter, + // which points every "stream" URL at the translator's own /bw/stream. These + // are the translator's webhook credentials; a rewriter that ever left the + // customer's bot URL in place would hand them to a third party. if (el.name === "StartStream" && el.attrs?.destination !== undefined) { el.attrs.destinationUsername = auth.username; el.attrs.destinationPassword = auth.password; diff --git a/test/server-stream.test.ts b/test/server-stream.test.ts index 6d009c0..34a4c72 100644 --- a/test/server-stream.test.ts +++ b/test/server-stream.test.ts @@ -7,7 +7,7 @@ */ import { describe, it, expect, vi } from "vitest"; import { WebSocketServer, WebSocket } from "ws"; -import type { AddressInfo } from "node:net"; +import { createServer, type AddressInfo } from "node:net"; import { readFileSync, mkdtempSync, rmSync, readdirSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -189,6 +189,29 @@ describe("/bw/stream end to end", () => { await app.close(); }); + it("ends the Bandwidth stream when the bot accepts TCP but never completes the WebSocket handshake", async () => { + // A black-hole bot: accepts the connection and says nothing. Without the + // handshake timeout the stream would hang until the OS gave up. + const blackHole = createServer((sock) => sock.on("error", () => {})); + await new Promise((r) => blackHole.listen(0, "127.0.0.1", r)); + const holePort = (blackHole.address() as AddressInfo).port; + + const app = makeApp(config({ streamBotConnectTimeoutMs: 300 })); + const port = await listen(app); + const bw = bandwidthClient(port, `ws://127.0.0.1:${holePort}/bot`); + await upgradeStatus(bw.ws); + const sentAt = Date.now(); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + bw.send(frames.start); + await bwClosed; + // Long enough to be the timeout, not an immediate refusal; generous upper bound for slow CI. + expect(Date.now() - sentAt).toBeGreaterThanOrEqual(250); + expect(Date.now() - sentAt).toBeLessThan(5000); + + blackHole.close(); + await app.close(); + }); + it("closes a socket that sends no start event within the timeout", async () => { const app = makeApp(); const port = await listen(app); @@ -198,10 +221,23 @@ describe("/bw/stream end to end", () => { const openedAt = Date.now(); await new Promise((r) => bw.ws.once("close", r)); expect(Date.now() - openedAt).toBeGreaterThanOrEqual(250); - expect(Date.now() - openedAt).toBeLessThan(2000); + // Upper bound is deliberately loose: it only guards against "never closes". + expect(Date.now() - openedAt).toBeLessThan(5000); bot.close(); await app.close(); }); + + it("app.close() terminates a stream that upgraded but has not sent start yet", async () => { + const app = makeApp(config({ streamStartTimeoutMs: 60_000 })); + const port = await listen(app); + const bw = bandwidthClient(port, "ws://127.0.0.1:9/bot"); + await upgradeStatus(bw.ws); + const bwClosed = new Promise((r) => bw.ws.once("close", r)); + const closingAt = Date.now(); + await app.close(); + await bwClosed; + expect(Date.now() - closingAt).toBeLessThan(2000); + }); }); describe("/bw/stream upgrade gate", () => { @@ -284,7 +320,7 @@ describe("translator side of the stream route", () => { }); describe("stream frame capture", () => { - it("writes start, media, and stop frames to /streams/*.jsonl when captureDir is set", async () => { + it("writes start, the first 10 media frames, and stop to /streams/*.jsonl", async () => { const dir = mkdtempSync(join(tmpdir(), "stream-capture-")); try { const app = makeApp(config({ captureDir: dir })); @@ -294,7 +330,9 @@ describe("stream frame capture", () => { await upgradeStatus(bw.ws); bw.send(frames.start); const botWs = await bot.socket; - bw.send(frames.media); + // 12 frames sent; the cap keeps 10 so a capture stays small. + for (let i = 1; i <= 12; i++) bw.send({ ...frames.media, sequenceNumber: String(i) }); + await waitFor(() => bot.messages.filter((m) => m.event === "media").length === 12); const botClosed = new Promise((r) => botWs.once("close", r)); bw.send(frames.stop); await botClosed; @@ -306,8 +344,13 @@ describe("stream frame capture", () => { expect(files).toHaveLength(1); expect(files[0]).toMatch(/^[0-9a-f]{16}\.jsonl$/); const lines = readFileSync(join(dir, "streams", files[0]), "utf8").trim().split("\n").map((l) => JSON.parse(l)); - expect(lines.map((l) => l.eventType)).toEqual(["start", "media", "stop"]); + expect(lines[0].eventType).toBe("start"); expect(lines[0].metadata.streamId).toBe(frames.start.metadata.streamId); + expect(lines.filter((l) => l.eventType === "media").map((l) => l.sequenceNumber)).toEqual( + ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"], + ); + expect(lines.at(-1).eventType).toBe("stop"); + expect(lines).toHaveLength(12); } finally { rmSync(dir, { recursive: true, force: true }); } diff --git a/test/streams-wire.test.ts b/test/streams-wire.test.ts index 4f0b821..0bdb4f5 100644 --- a/test/streams-wire.test.ts +++ b/test/streams-wire.test.ts @@ -192,6 +192,31 @@ describe("start message", () => { }); }); +// ─── bot connect timeout ──────────────────────────────────────────────────── + +describe("bot connect timeout", () => { + it("ready() rejects when the bot accepts TCP but never completes the handshake", async () => { + const { createServer } = await import("node:net"); + const blackHole = createServer((sock) => sock.on("error", () => {})); + await new Promise((r) => blackHole.listen(0, "127.0.0.1", r)); + const port = (blackHole.address() as import("node:net").AddressInfo).port; + + const source = new FakeBwSource(); + const bridge = new TwilioStreamBridge({ + botUrl: `ws://127.0.0.1:${port}`, + callSid: "CAhole", + accountSid: "AChole", + source, + connectTimeoutMs: 200, + }); + const startedAt = Date.now(); + await expect(bridge.ready()).rejects.toThrow(/handshake/i); + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(150); + bridge.close(); + blackHole.close(); + }); +}); + // ─── stop message ─────────────────────────────────────────────────────────── describe("stop message", () => { From 5400ad2da95e126dbf2844c69368a16c4ef0534a Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Fri, 25 Sep 2026 11:53:07 -0400 Subject: [PATCH 3/4] VAPI-3991: mark Connect/Stream and the bridge Live after an end-to-end call A real inbound call on 2026-09-25 ran through the translator: Bandwidth opened its StartStream WebSocket to /bw/stream, the bridge relayed it to a Twilio-protocol echo bot, and the caller heard the bot's tone and then their own voice echoed back. Over 17.6 s: 878 caller frames in, 853 bot frames out, StreamParam values delivered as customParameters, the bot's greeting mark returned after 999 ms (held for the 1 s of audio queued ahead of it), and stop propagated when the call ended. README: split the stream row so Connect -> Stream and the Media Streams bridge read Live; Start/Stop Stream forks and transcription stay Unit. Fixture: frames unchanged. The _meta note now records that the live frames matched these shapes field for field, and that the recorded frames are deliberately not checked in because they came from a personal line and this repository is public. --- README.md | 5 +++-- test/fixtures/bandwidth/stream-frames.json | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index f65f0f2..e4dfe02 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,8 @@ real calls. | Record (+ `recordingStatusCallback`) | ✅ | Live | | Dial → Number / SIP | ✅ | Live | | Dial → Conference (mute + events; no hold music / lifecycle) | ⚠️ | Live | -| Connect/Start/Stop Stream, Start/Stop Transcription | ✅ | Unit | +| Connect → Stream (bidirectional, via the Media Streams bridge) | ✅ | Live | +| Start/Stop Stream (fork), Start/Stop Transcription | ✅ | Unit | | Reject, Refer (SIP) | ⚠️ | Unit | | Queue, Client, Enqueue, Leave, Pay | ❌ | — (no BW primitive — fail loudly) | @@ -128,7 +129,7 @@ real calls. | List / fetch / download recordings; pause-resume | ✅ | Live/Unit | | Status callback (call completion) | ✅ | Unit | | Recording callback (`recordingStatusCallback`) | ✅ | Unit | -| Media Streams bridge (AI-voice path) | ✅ | Unit | +| Media Streams bridge (AI-voice path) | ✅ | Live | Number provisioning (search/order/activate) isn't part of this toolkit — it's handled by the [`band` CLI](#pairs-with-the-band-cli); see the Phase 2 runbook diff --git a/test/fixtures/bandwidth/stream-frames.json b/test/fixtures/bandwidth/stream-frames.json index 0a5f611..8c4f135 100644 --- a/test/fixtures/bandwidth/stream-frames.json +++ b/test/fixtures/bandwidth/stream-frames.json @@ -1,8 +1,8 @@ { "_meta": { "captured": "2026-09-17", - "source": "start.metadata: live Bandwidth on a real call, recorded by the local repro app. Field names, nesting, and the PCMU/8000 track format are exactly as received.", - "note": "streamParams was absent on that call because the BXML carried no ; the value here follows the StartStream docs example. media and stop follow the docs' documented shapes; the media payload is 160 bytes (20 ms) of synthetic PCMU silence, not recorded audio. Replace this file with frames the translator captures to /streams/*.jsonl on the first live end-to-end call.", + "source": "start.metadata: live Bandwidth on a real call, recorded by the local repro app. Field names, nesting, and the PCMU/8000 track format are exactly as received; ids redacted.", + "note": "streamParams follows the StartStream docs example; media and stop follow the docs' documented shapes; the media payload is 160 bytes (20 ms) of synthetic PCMU silence, not recorded audio. On 2026-09-25 a live inbound call ran end to end through /bw/stream with frame capture on: the real start, media, and stop frames matched these shapes field for field (Bandwidth orders eventType last, which the parser does not depend on), streamParams arrived as the flat map modelled here, and media frames were 160-byte inbound PCMU with sequenceNumber counting from \"1\". Those recorded frames are deliberately not checked in: they came from a personal line and this repo is public. To refresh from a shared test account, run a call with TRANSLATOR_CAPTURE_DIR set and copy /streams/*.jsonl.", "redacted": [ "accountId", "callId", From 962607acfe10b58cf24d7f1d0d248c078611b415 Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Fri, 25 Sep 2026 11:53:26 -0400 Subject: [PATCH 4/4] Use the docs' example account id in the bridge test fixture The Bandwidth start-event literal in streams-wire.test.ts carried a real account id from the live repro. Replace it with the StartStream docs' example value; the test only needs the shape. --- test/streams-wire.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/streams-wire.test.ts b/test/streams-wire.test.ts index 0bdb4f5..6b624f5 100644 --- a/test/streams-wire.test.ts +++ b/test/streams-wire.test.ts @@ -121,7 +121,7 @@ describe("start message", () => { // Shape per the StartStream docs' start-event example. const bwStart = { eventType: "start", - metadata: { accountId: "9900778", callId: "c-abc", to: "+15550001111", from: "+15550002222" }, + metadata: { accountId: "5555555", callId: "c-abc", to: "+15550001111", from: "+15550002222" }, streamParams: { callSid: "CA123", tenant: "acme" }, }; const port = nextPort();