From a0d1f04454c2b16aba1ae72b75068f1307508651 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Wed, 22 Apr 2026 14:14:26 -0700 Subject: [PATCH 1/2] fix(recorder): stream SSE responses instead of buffering `proxyAndRecord` collected every upstream chunk into an array and replayed the entire body via a single `res.end()` at the bottom of the function. For `text/event-stream` responses this collapses all frames into one client-visible write, defeating progressive rendering in downstream consumers. When upstream declares `content-type: text/event-stream`, tee each incoming chunk to the client while still accumulating into the recording buffer. Non-SSE responses continue to use the existing buffered relay path unchanged. Latent since v1.6.0 (`900399b`); surfaced in production once showcase traffic began exercising `--proxy-only` more heavily under tightened v1.14.0+ fixture handling. --- src/__tests__/recorder.test.ts | 95 ++++++++++++++++++++++++++++++++++ src/recorder.ts | 56 ++++++++++++++++---- 2 files changed, 142 insertions(+), 9 deletions(-) diff --git a/src/__tests__/recorder.test.ts b/src/__tests__/recorder.test.ts index b28a9684..b591c7b1 100644 --- a/src/__tests__/recorder.test.ts +++ b/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((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((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((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); + }); +}); diff --git a/src/recorder.ts b/src/recorder.ts index 3b0c7430..e9d573b8 100644 --- a/src/recorder.ts +++ b/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 = {}; - 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 = {}; + 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, 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 = {}; + 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, }); }); }, From a6fcffd87ac912ad5b7cb1f92eee1fba89028d51 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Wed, 22 Apr 2026 14:14:30 -0700 Subject: [PATCH 2/2] chore: release v1.14.5 --- CHANGELOG.md | 6 ++++++ package.json | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b574ca53..e840545a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/package.json b/package.json index 47b0ef3f..161badfa 100644 --- a/package.json +++ b/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": [