fix(recorder): stream SSE responses instead of buffering (v1.14.5)

CHANGELOG.md

@@ -1,5 +1,11 @@
 # @copilotkit/aimock
 
+## 1.14.5
+
+### Fixed
+
+- Recorder no longer buffers SSE (`text/event-stream`) upstream responses before relaying to the client. `proxyAndRecord` accumulated all upstream chunks and replayed them via a single `res.end()`, collapsing multi-frame streams into one client-visible write and breaking progressive rendering for downstream consumers (notably showcase `--proxy-only` deployments). SSE responses now stream chunk-by-chunk to the client while still being tee'd into the recording buffer; non-SSE behavior is unchanged.
+
 ## 1.14.3
 
 ### Added

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@copilotkit/aimock",
-  "version": "1.14.3",
+  "version": "1.14.5",
   "description": "Mock infrastructure for AI application testing — LLM APIs, image generation, text-to-speech, transcription, video generation, MCP tools, A2A agents, AG-UI event streams, vector databases, search, rerank, and moderation. One package, one port, zero dependencies.",
   "license": "MIT",
   "keywords": [

src/__tests__/recorder.test.ts

@@ -3259,3 +3259,98 @@ describe("recorder binary EventStream relay integrity", () => {
     expect(fixtureContent.fixtures[0].response.content).toBe("Binary integrity test");
   });
 });
+
+// ---------------------------------------------------------------------------
+// SSE progressive streaming — recorder must tee upstream chunks to the
+// client as they arrive, not buffer and replay in a single write.
+// ---------------------------------------------------------------------------
+
+describe("recorder SSE progressive streaming", () => {
+  let rawServer: http.Server | undefined;
+
+  afterEach(async () => {
+    if (rawServer) {
+      await new Promise<void>((resolve) => rawServer!.close(() => resolve()));
+      rawServer = undefined;
+    }
+  });
+
+  it("streams SSE frames progressively to the client (not buffered)", async () => {
+    // Raw upstream that emits 5 SSE `data:` frames spaced by 50ms each.
+    // The recorder must relay each chunk as it arrives; if it buffers
+    // and replays via a single res.end(), the client observes all frames
+    // within microseconds of each other.
+    const FRAME_DELAY_MS = 50;
+    const NUM_FRAMES = 5;
+    rawServer = http.createServer((_req, res) => {
+      res.writeHead(200, {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+      });
+      let i = 0;
+      const emit = () => {
+        if (i < NUM_FRAMES) {
+          res.write(`data: {"chunk":${i}}\n\n`);
+          i++;
+          setTimeout(emit, FRAME_DELAY_MS);
+        } else {
+          res.write("data: [DONE]\n\n");
+          res.end();
+        }
+      };
+      setTimeout(emit, FRAME_DELAY_MS);
+    });
+    await new Promise<void>((resolve) => rawServer!.listen(0, "127.0.0.1", resolve));
+    const rawAddr = rawServer!.address() as { port: number };
+    const rawUrl = `http://127.0.0.1:${rawAddr.port}`;
+
+    tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "aimock-record-sse-"));
+    recorder = await createServer([], {
+      port: 0,
+      record: { providers: { openai: rawUrl }, fixturePath: tmpDir, proxyOnly: true },
+    });
+
+    // Issue request and capture the wall-clock arrival time of each
+    // client-visible data event.
+    const arrivalTimes: number[] = [];
+    const body = JSON.stringify({
+      model: "gpt-4",
+      messages: [{ role: "user", content: "stream me" }],
+      stream: true,
+    });
+    const parsedUrl = new URL(recorder.url);
+    await new Promise<void>((resolve, reject) => {
+      const clientReq = http.request(
+        {
+          hostname: parsedUrl.hostname,
+          port: parsedUrl.port,
+          path: "/v1/chat/completions",
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+            "Content-Length": Buffer.byteLength(body),
+          },
+        },
+        (res) => {
+          res.on("data", () => {
+            arrivalTimes.push(Date.now());
+          });
+          res.on("end", () => resolve());
+          res.on("error", reject);
+        },
+      );
+      clientReq.on("error", reject);
+      clientReq.write(body);
+      clientReq.end();
+    });
+
+    // We should observe multiple client-visible data events, and the span
+    // between first and last should reflect the upstream frame spacing.
+    // With buffer-and-replay, everything arrives in one write within ~0ms.
+    expect(arrivalTimes.length).toBeGreaterThanOrEqual(2);
+    const span = arrivalTimes[arrivalTimes.length - 1] - arrivalTimes[0];
+    // NUM_FRAMES frames spaced by 50ms = >=200ms expected span; allow
+    // slack for scheduler jitter but require clearly more than "all at once".
+    expect(span).toBeGreaterThanOrEqual(100);
+  });
+});

src/recorder.ts

@@ -103,12 +103,16 @@ export async function proxyAndRecord(
   let upstreamBody: string;
   let rawBuffer: Buffer;
 
+  // Track whether we streamed SSE progressively to the client; if so,
+  // skip the final res.writeHead/res.end relay at the bottom of this fn.
+  let streamedToClient = false;
   try {
-    const result = await makeUpstreamRequest(target, forwardHeaders, requestBody);
+    const result = await makeUpstreamRequest(target, forwardHeaders, requestBody, res);
     upstreamStatus = result.status;
     upstreamHeaders = result.headers;
     upstreamBody = result.body;
     rawBuffer = result.rawBuffer;
+    streamedToClient = result.streamedToClient;
   } catch (err) {
     const msg = err instanceof Error ? err.message : "Unknown proxy error";
     defaults.logger.error(`Proxy request failed: ${msg}`);
@@ -250,13 +254,17 @@ export async function proxyAndRecord(
     defaults.logger.info(`Proxied ${providerKey} request (proxy-only mode)`);
   }
 
-  // Relay upstream response to client
-  const relayHeaders: Record<string, string> = {};
-  if (ctString) {
-    relayHeaders["Content-Type"] = ctString;
+  // Relay upstream response to client (skip when SSE was already streamed
+  // progressively by makeUpstreamRequest — headers and body are already on
+  // the wire).
+  if (!streamedToClient) {
+    const relayHeaders: Record<string, string> = {};
+    if (ctString) {
+      relayHeaders["Content-Type"] = ctString;
+    }
+    res.writeHead(upstreamStatus, relayHeaders);
+    res.end(isBinaryStream ? rawBuffer : upstreamBody);
   }
-  res.writeHead(upstreamStatus, relayHeaders);
-  res.end(isBinaryStream ? rawBuffer : upstreamBody);
 
   return true;
 }
@@ -269,7 +277,14 @@ function makeUpstreamRequest(
   target: URL,
   headers: Record<string, string>,
   body: string,
-): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: string; rawBuffer: Buffer }> {
+  clientRes?: http.ServerResponse,
+): Promise<{
+  status: number;
+  headers: http.IncomingHttpHeaders;
+  body: string;
+  rawBuffer: Buffer;
+  streamedToClient: boolean;
+}> {
   return new Promise((resolve, reject) => {
     const transport = target.protocol === "https:" ? https : http;
     const UPSTREAM_TIMEOUT_MS = 30_000;
@@ -288,16 +303,39 @@ function makeUpstreamRequest(
         res.setTimeout(BODY_TIMEOUT_MS, () => {
           req.destroy(new Error(`Upstream response timed out after ${BODY_TIMEOUT_MS / 1000}s`));
         });
+        // Detect Server-Sent Events so we can tee upstream chunks to the
+        // client as they arrive rather than buffering the entire stream and
+        // replaying it in a single res.end() at the bottom of proxyAndRecord.
+        // Buffering collapses every SSE frame into one client-visible write,
+        // which defeats progressive rendering in downstream consumers.
+        const ct = res.headers["content-type"];
+        const ctStr = Array.isArray(ct) ? ct.join(", ") : (ct ?? "");
+        const isSSE = ctStr.toLowerCase().includes("text/event-stream");
+        let streamedToClient = false;
+        if (isSSE && clientRes && !clientRes.headersSent) {
+          const relayHeaders: Record<string, string> = {};
+          if (ctStr) relayHeaders["Content-Type"] = ctStr;
+          clientRes.writeHead(res.statusCode ?? 200, relayHeaders);
+          // Flush headers immediately so the client starts parsing frames
+          // before the first data chunk arrives.
+          if (typeof clientRes.flushHeaders === "function") clientRes.flushHeaders();
+          streamedToClient = true;
+        }
         const chunks: Buffer[] = [];
-        res.on("data", (chunk: Buffer) => chunks.push(chunk));
+        res.on("data", (chunk: Buffer) => {
+          chunks.push(chunk);
+          if (streamedToClient) clientRes!.write(chunk);
+        });
         res.on("error", reject);
         res.on("end", () => {
           const rawBuffer = Buffer.concat(chunks);
+          if (streamedToClient) clientRes!.end();
           resolve({
             status: res.statusCode ?? 500,
             headers: res.headers,
             body: rawBuffer.toString(),
             rawBuffer,
+            streamedToClient,
           });
         });
       },