Update PR #2742

Author: lforst

dev-packages/seinfeld/src/cassette.ts

@@ -46,7 +46,12 @@ export interface RecordedRequest {
 /** A single recorded response, as it appeared on the wire. */
 export interface RecordedResponse {
   /** Original wire chunk sizes and inter-chunk delays for interactive streams. */
-  streamChunks?: Array<{ size: number; delayMs: number }>;
+  streamChunks?: Array<{
+    size: number;
+    delayMs: number;
+    /** Complete Connect request frames received before this response chunk. */
+    requestFrames?: number;
+  }>;
   status: number;
   statusText?: string;
   headers: Record<string, string>;

dev-packages/seinfeld/src/format/v1.ts

@@ -34,6 +34,7 @@ const recordedResponseSchema = z.object({
       z.object({
         size: z.number().int().positive(),
         delayMs: z.number().nonnegative(),
+        requestFrames: z.number().int().nonnegative().optional(),
       }),
     )
     .optional(),

dev-packages/seinfeld/src/http.ts

@@ -107,7 +107,11 @@ export async function recordResponseDraft(
 /** Build a Fetch API `Response` from a recorded response body. */
 export async function buildResponse(
   recorded: RecordedResponse | RecordedResponseOrDraft,
-  ctx?: { store: CassetteStore; name: string },
+  ctx?: {
+    store: CassetteStore;
+    name: string;
+    waitForRequestFrames?: (count: number) => Promise<void>;
+  },
 ): Promise<Response> {
   // Expand \n-joined set-cookie back into multiple header entries.
   // Strip encoding headers: stored bytes are already decoded by the fetch
@@ -150,9 +154,9 @@ export async function buildResponse(
       : await decodeBody(recorded.body, ctx);
   if (bytes.length === 0) return new Response(null, init);
   if (recorded.streamChunks) {
-    // Interactive providers wait for client tool results between chunks.
-    // Preserve the recorded pacing instead of closing the response while
-    // the SDK still has pending writes to its request stream.
+    // Interactive providers wait for client messages between chunks. Frame
+    // counts preserve that ordering even if replay's client is slower than
+    // recording's client; recorded delays alone cannot enforce it.
     const chunks = recorded.streamChunks;
     let index = 0;
     let offset = 0;
@@ -162,7 +166,13 @@ export async function buildResponse(
           controller.close();
           return;
         }
-        const { size, delayMs } = chunks[index++]!;
+        const { size, delayMs, requestFrames } = chunks[index++]!;
+        if (requestFrames !== undefined) {
+          if (!ctx?.waitForRequestFrames) {
+            throw new Error("Connect replay requires request frame tracking");
+          }
+          await ctx.waitForRequestFrames(requestFrames);
+        }
         await new Promise((resolve) => setTimeout(resolve, delayMs));
         controller.enqueue(bytes.slice(offset, offset + size));
         offset += size;

dev-packages/seinfeld/src/internal/connect-request.ts

@@ -0,0 +1,67 @@
+import type { IncomingMessage } from "node:http";
+
+/** Count complete Connect envelopes without retaining their payloads. */
+export function trackConnectRequest(req: IncomingMessage) {
+  const header = Buffer.alloc(5);
+  let headerSize = 0;
+  let remaining = 0;
+  let frames = 0;
+  let closed = false;
+  let notify = () => {};
+
+  const onData = (data: Buffer) => {
+    let offset = 0;
+    while (offset < data.length) {
+      if (headerSize < header.length) {
+        const size = Math.min(header.length - headerSize, data.length - offset);
+        data.copy(header, headerSize, offset, offset + size);
+        headerSize += size;
+        offset += size;
+        if (headerSize < header.length) continue;
+        remaining = header.readUInt32BE(1);
+      }
+      const size = Math.min(remaining, data.length - offset);
+      remaining -= size;
+      offset += size;
+      if (remaining === 0) {
+        frames++;
+        headerSize = 0;
+      }
+    }
+    notify();
+  };
+  const onClose = () => {
+    closed = true;
+    notify();
+  };
+  req.on("data", onData);
+  req.on("end", onClose);
+  req.on("close", onClose);
+  req.on("error", onClose);
+
+  return {
+    get frames() {
+      return frames;
+    },
+    // Response stream pulls are sequential, so there is only one waiter.
+    async waitFor(count: number) {
+      while (frames < count) {
+        if (closed) {
+          throw new Error(
+            `Connect request ended after ${frames} frames; replay requires ${count}`,
+          );
+        }
+        await new Promise<void>((resolve) => {
+          notify = resolve;
+        });
+      }
+    },
+    dispose() {
+      req.off("data", onData);
+      req.off("end", onClose);
+      req.off("close", onClose);
+      req.off("error", onClose);
+      onClose();
+    },
+  };
+}

dev-packages/seinfeld/src/recorder.ts

@@ -17,7 +17,9 @@ import type {
   CassetteEntry,
   CassetteMode,
   RecordedRequest,
+  RecordedResponse,
 } from "./cassette";
+import { trackConnectRequest } from "./internal/connect-request";
 import { AggregateCassetteMissError, CassetteMissError } from "./errors";
 import { applyFilters } from "./normalizer";
 import type { FilterSpec } from "./normalizer";
@@ -46,7 +48,11 @@ export type RequestUrlMatcher = string | RegExp;
 export type IgnoredRequestMatcher = RequestUrlMatcher;
 export type StreamingRequestMatcher =
   | RequestUrlMatcher
-  | { url: RequestUrlMatcher; httpVersion: "2" };
+  | {
+      url: RequestUrlMatcher;
+      httpVersion?: "1.1" | "2";
+      requestFraming?: "connect";
+    };
 
 // ---- Internal draft types ------------------------------------------------
 
@@ -126,18 +132,32 @@ async function handleReplayWithoutRequestBody(
   req: IncomingMessage,
   res: ServerResponse,
   upstreamUrl: string,
+  requestFraming?: "connect",
 ): Promise<void> {
   // Some duplex APIs keep the request body open while waiting for response
   // messages. For opted-in URLs, replay can respond immediately because the
   // cassette match is configured to ignore the request body.
+  const tracker =
+    requestFraming === "connect" ? trackConnectRequest(req) : undefined;
+  const onClose = () => tracker?.dispose();
+  res.on("close", onClose);
   req.resume();
-  const response = await handleReplay(ctx, {
-    method: req.method ?? "GET",
-    url: upstreamUrl,
-    headers: headersToRecord(req.headers),
-    body: { kind: "empty" },
-  });
-  await writeFetchResponse(res, response);
+  try {
+    const response = await handleReplay(
+      ctx,
+      {
+        method: req.method ?? "GET",
+        url: upstreamUrl,
+        headers: headersToRecord(req.headers),
+        body: { kind: "empty" },
+      },
+      tracker?.waitFor,
+    );
+    await writeFetchResponse(res, response);
+  } finally {
+    res.off("close", onClose);
+    tracker?.dispose();
+  }
 }
 
 async function handleStreamingRecord(
@@ -146,10 +166,13 @@ async function handleStreamingRecord(
   res: ServerResponse,
   upstreamUrl: string,
   httpVersion: "1.1" | "2",
+  requestFraming?: "connect",
 ): Promise<void> {
   const upstream = new URL(upstreamUrl);
   const responseChunks: Buffer[] = [];
-  const streamChunks: Array<{ size: number; delayMs: number }> = [];
+  const streamChunks: NonNullable<RecordedResponse["streamChunks"]> = [];
+  const tracker =
+    requestFraming === "connect" ? trackConnectRequest(req) : undefined;
   let lastChunkAt = performance.now();
   let responseHeaders: IncomingHttpHeaders = {};
   let responseStatus = 502;
@@ -202,6 +225,7 @@ async function handleStreamingRecord(
           streamChunks.push({
             size: bytes.length,
             delayMs: Math.round(now - lastChunkAt),
+            ...(tracker ? { requestFrames: tracker.frames } : {}),
           });
           lastChunkAt = now;
           responseChunks.push(bytes);
@@ -239,6 +263,7 @@ async function handleStreamingRecord(
       upstreamUrl,
     });
   } finally {
+    tracker?.dispose();
     session?.destroy();
   }
 }
@@ -249,7 +274,7 @@ function pushStreamingEntry(
     requestHeaders: Record<string, number | string | string[] | undefined>;
     requestMethod: string;
     responseBody: Uint8Array;
-    streamChunks: Array<{ size: number; delayMs: number }>;
+    streamChunks: NonNullable<RecordedResponse["streamChunks"]>;
     responseHeaders: Record<string, number | string | string[] | undefined>;
     responseStatus: number;
     responseStatusText?: string;
@@ -284,6 +309,7 @@ function pushStreamingEntry(
 function handleReplay(
   ctx: CassetteContext,
   recorded: RecordedRequest,
+  waitForRequestFrames?: (count: number) => Promise<void>,
 ): Promise<Response> {
   const filtered = asNormalized(applyFilters(recorded, ctx.filters));
   const matchKey = filtered.matchKey;
@@ -310,7 +336,11 @@ function handleReplay(
     );
   }
 
-  return buildResponse(match.response, { store: ctx.store, name: ctx.name });
+  return buildResponse(match.response, {
+    store: ctx.store,
+    name: ctx.name,
+    waitForRequestFrames,
+  });
 }
 
 async function handleRecord(
@@ -575,8 +605,18 @@ export function createCassetteServer(
       ),
     );
     if (streamingRequest) {
+      const options =
+        typeof streamingRequest === "object" && "url" in streamingRequest
+          ? streamingRequest
+          : undefined;
       if (ctx.mode === "replay") {
-        await handleReplayWithoutRequestBody(ctx, req, res, upstreamUrl);
+        await handleReplayWithoutRequestBody(
+          ctx,
+          req,
+          res,
+          upstreamUrl,
+          options?.requestFraming,
+        );
         return;
       }
       if (ctx.mode === "record") {
@@ -585,10 +625,8 @@ export function createCassetteServer(
           req,
           res,
           upstreamUrl,
-          typeof streamingRequest === "object" &&
-            "httpVersion" in streamingRequest
-            ? streamingRequest.httpVersion
-            : "1.1",
+          options?.httpVersion ?? "1.1",
+          options?.requestFraming,
         );
         return;
       }