Merge branch 'v2/main' into v2/fix/2110-oauth-path-scoped-discovery

clients/web/src/test/core/mcp/remote/parseSSE.test.ts

@@ -0,0 +1,86 @@
+/**
+ * `parseSSE` cross-chunk frame loss.
+ *
+ * The reproduction needs no browser and no network: `parseSSE` takes a reader,
+ * so a test supplies the chunk boundary directly instead of hoping a real
+ * transport happens to produce one. That is what makes this defect testable at
+ * all — on the wire the split is a function of payload size, TCP segmentation
+ * and any proxy in between, none of which a test can steer.
+ *
+ * The bug: `currentEvent` / `currentData` are declared INSIDE the read loop, so
+ * a frame whose `data:` line arrives in one chunk and whose terminating blank
+ * line arrives in the next is discarded. `buffer` already carries a partial
+ * *line* across reads; nothing carries a partial *frame*.
+ */
+
+import { describe, expect, it } from "vitest";
+import { parseSSE } from "@inspector/core/mcp/remote/remoteClientTransport";
+
+/** A reader that yields exactly the chunks given, in order. */
+function readerFrom(chunks: string[]): ReadableStreamDefaultReader<Uint8Array> {
+  const encoder = new TextEncoder();
+  let i = 0;
+  return {
+    read: async () =>
+      i < chunks.length
+        ? { done: false as const, value: encoder.encode(chunks[i++]!) }
+        : { done: true as const, value: undefined },
+  } as ReadableStreamDefaultReader<Uint8Array>;
+}
+
+async function collect(chunks: string[]) {
+  const out: { event: string; data: string }[] = [];
+  for await (const frame of parseSSE(readerFrom(chunks))) out.push(frame);
+  return out;
+}
+
+const FRAME = 'event: message\ndata: {"id":1,"result":"payload"}\n\n';
+
+describe("parseSSE chunk boundaries", () => {
+  it("delivers a frame that arrives whole (the case that hides the bug)", async () => {
+    // Chromium and Firefox happen to produce this shape at our payload sizes,
+    // which is why the defect below has never been observed in practice.
+    expect(await collect([FRAME])).toEqual([
+      { event: "message", data: '{"id":1,"result":"payload"}' },
+    ]);
+  });
+
+  it("delivers a frame split between its data line and its terminator", async () => {
+    // THE BUG. Chunk 1 ends after the data line's newline; chunk 2 carries only
+    // the blank line that closes the frame. Nothing is malformed — this is one
+    // valid frame, delivered in two reads.
+    const split = ['event: message\ndata: {"id":1,"result":"payload"}\n', "\n"];
+    expect(split.join("")).toBe(FRAME); // same bytes as above, different reads
+    expect(await collect(split)).toEqual([
+      { event: "message", data: '{"id":1,"result":"payload"}' },
+    ]);
+  });
+
+  it("delivers a frame split mid-payload", async () => {
+    const split = ['event: message\ndata: {"id":1,"res', 'ult":"payload"}\n\n'];
+    expect(split.join("")).toBe(FRAME);
+    expect(await collect(split)).toEqual([
+      { event: "message", data: '{"id":1,"result":"payload"}' },
+    ]);
+  });
+
+  it("delivers every frame when a burst is split at each boundary", async () => {
+    // Bytes for three frames, re-cut so each split lands between a data line
+    // and its terminator — the shape a large response most plausibly produces.
+    const chunks: string[] = [];
+    for (const id of [1, 2, 3]) {
+      chunks.push(`event: message\ndata: {"id":${id}}\n`, "\n");
+    }
+    expect(await collect(chunks)).toEqual([
+      { event: "message", data: '{"id":1}' },
+      { event: "message", data: '{"id":2}' },
+      { event: "message", data: '{"id":3}' },
+    ]);
+  });
+
+  it("flushes a final frame when the stream ends without a blank line", async () => {
+    expect(await collect(['event: message\ndata: {"id":9}\n'])).toEqual([
+      { event: "message", data: '{"id":9}' },
+    ]);
+  });
+});

core/mcp/remote/remoteClientTransport.ts

@@ -165,14 +165,55 @@ function legacySessionId(json: RemoteConnectResponse): string | undefined {
 }
 
 /**
- * Parse SSE stream from a ReadableStream.
- * Yields { event, data } for each SSE message.
+ * Parse an SSE stream from a ReadableStream.
+ * Yields `{ event, data }` for each complete SSE message.
+ *
+ * ⚠️ **The in-progress event must live OUTSIDE the read loop.** A `read()`
+ * returns an arbitrary slice of bytes, not a whole frame, so a frame's `data:`
+ * line can arrive in one chunk and the blank line terminating it in the next.
+ * Declaring `currentEvent` / `currentData` inside the loop — which this did
+ * until #2134 — discards the accumulated payload at that boundary: the frame is
+ * never yielded, the JSON-RPC response it carried never settles, and the caller
+ * hangs forever with no error on any channel. `buffer` already carries a partial
+ * *line* across reads; these carry the partial *frame*.
+ *
+ * It survived because Chromium and Firefox happen to deliver whole frames at the
+ * payload sizes this transport sees. That is luck rather than a guarantee —
+ * chunk boundaries are a function of payload size, TCP segmentation and any
+ * intermediary — and the exposure grows with payload size, so a large
+ * `resources/read` is the most likely first victim.
+ *
+ * Exported for `clients/web/src/test/core/mcp/remote/parseSSE.test.ts`, which is
+ * the only way to reach the defect deterministically: driving a real transport
+ * cannot steer where the chunk boundary lands, but a test that supplies the
+ * reader can put it exactly on the terminator.
  */
-async function* parseSSE(
+export async function* parseSSE(
   reader: ReadableStreamDefaultReader<Uint8Array>,
 ): AsyncGenerator<{ event: string; data: string }> {
   const decoder = new TextDecoder();
   let buffer = "";
+  // Carried ACROSS reads, not per chunk — see the header.
+  let currentEvent = "message";
+  let currentData: string[] = [];
+
+  const consume = (line: string): { event: string; data: string } | null => {
+    if (line.startsWith("event:")) {
+      currentEvent = line.slice(6).trim();
+    } else if (line.startsWith("data:")) {
+      currentData.push(line.slice(5).trimStart());
+    } else if (line === "") {
+      const complete =
+        currentData.length > 0
+          ? { event: currentEvent, data: currentData.join("\n") }
+          : null;
+      currentEvent = "message";
+      currentData = [];
+      return complete;
+    }
+    // Anything else (a `:` keep-alive comment, `id:`, `retry:`) is ignored.
+    return null;
+  };
 
   while (true) {
     const { done, value } = await reader.read();
@@ -183,36 +224,21 @@ async function* parseSSE(
     /* v8 ignore next -- String.split always returns a non-empty array, so pop() is never undefined; the ?? "" fallback is unreachable. */
     buffer = lines.pop() ?? "";
 
-    let currentEvent = "message";
-    let currentData: string[] = [];
-
     for (const line of lines) {
-      if (line.startsWith("event:")) {
-        currentEvent = line.slice(6).trim();
-      } else if (line.startsWith("data:")) {
-        currentData.push(line.slice(5).trimStart());
-      } else if (line === "") {
-        if (currentData.length > 0) {
-          yield { event: currentEvent, data: currentData.join("\n") };
-        }
-        currentEvent = "message";
-        currentData = [];
-      }
+      const complete = consume(line);
+      if (complete) yield complete;
     }
   }
 
-  if (buffer.trim()) {
-    const lines = buffer.split("\n");
-    let currentEvent = "message";
-    const currentData: string[] = [];
-    for (const line of lines) {
-      if (line.startsWith("event:")) currentEvent = line.slice(6).trim();
-      else if (line.startsWith("data:"))
-        currentData.push(line.slice(5).trimStart());
-    }
-    if (currentData.length > 0) {
-      yield { event: currentEvent, data: currentData.join("\n") };
-    }
+  // Stream ended: flush the trailing partial line, then anything still held. A
+  // server that closes without a final blank line has still delivered a complete
+  // frame, and dropping it would lose the last message of every such stream.
+  if (buffer) {
+    const complete = consume(buffer);
+    if (complete) yield complete;
+  }
+  if (currentData.length > 0) {
+    yield { event: currentEvent, data: currentData.join("\n") };
   }
 }