VAPI-3991: implement the Bandwidth StartStream source and wire /bw/stream

AGENTS.md

@@ -74,7 +74,15 @@ 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 `<PUBLIC_BASE_URL>/bw/initiate`.
+at `<PUBLIC_BASE_URL>/bw/initiate`. If the customer app uses `<Connect><Stream>`,
+the host must also pass WebSocket upgrades through to the translator at
+`/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
@@ -112,8 +120,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. `<Parameter>`
+  - `Stream` — the translator rewrites the Twilio `<Stream url>` (the customer's
+    bot) to `wss://<PUBLIC_BASE_URL>/bw/stream?dest=<bot url>`, 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 `<dir>/streams/*.jsonl` (start, first 10 media, stop)
+    to refresh `test/fixtures/bandwidth/stream-frames.json`. `<Parameter>`
     children map to nested `<StreamParam/>` 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

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
@@ -198,7 +199,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)

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,16 @@ 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 <dir>/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;
+  /** 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;
 }
 
 export interface ServerDeps {
@@ -110,16 +124,135 @@ 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),
     // so they route to a dedicated egress endpoint rather than /bw/continue.
     if (kind === "recordingStatus") {
       return `${config.publicBaseUrl}/bw/recording-status?cb=${encodeURIComponent(absolute)}`;
     }
+    // A Twilio <Stream url> 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<TwilioStreamBridge>();
+
+  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<void> {
+    // 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)) {
+      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,
+            connectTimeoutMs: config.streamBotConnectTimeoutMs,
+          });
+          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 <StopStream wait="true"> 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);
+  });
+  // 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();
+  });
+
   function errorBxml(verbs: string[]): string {
     return bxmlDocument([
       {

src/server/capture.ts

@@ -1,7 +1,40 @@
-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<string, number>();
+
+/**
+ * Append one raw Bandwidth StartStream WebSocket frame to
+ * `<dir>/streams/<hash>.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);
+  } 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 });
+  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

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,6 +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: optionalMs("STREAM_PLAYOUT_LATENCY_PAD_MS"),
+    streamStartTimeoutMs: optionalMs("STREAM_START_TIMEOUT_MS"),
+    streamBotConnectTimeoutMs: optionalMs("STREAM_BOT_CONNECT_TIMEOUT_MS"),
   },
   {
     fetchImpl: fetch,

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)
@@ -168,6 +174,14 @@ export class TwilioStreamBridge {
       this.ws.close();
     });
 
+    // Bot hung up its side. On Twilio that ends <Connect> and TwiML resumes
+    // after it; here, closing the Bandwidth socket ends the StartStream so the
+    // translator's <StopStream wait="true"> 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) => {