From 7ccbd6655c9aa71401faf581852835f44da4dfe6 Mon Sep 17 00:00:00 2001 From: Huzaifa Farooq Date: Mon, 14 Sep 2026 06:15:25 +0500 Subject: [PATCH 1/2] fix(middleware-node): accumulate SSE reads in tests instead of assuming one chunk streamableHttp.test.ts asserted on multiple SSE events after a single reader.read() call, assuming Node's fetch would coalesce them into one chunk. That assumption doesn't hold on newer Node versions, where closely-timed events can arrive as separate chunks across separate reads, failing the test before the later event is read. Adds a readUntilContains helper that accumulates decoded text across reads until the expected content appears (or times out), and uses it in the two tests that were asserting on multiple SSE events from a single read. --- .../node/test/streamableHttp.test.ts | 36 +++++++++++++++---- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 140717c6bb..209f06875e 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -34,6 +34,29 @@ async function getFreePort() { }); } +/** + * Reads chunks from an SSE stream reader until the accumulated decoded text contains every + * needle, or `timeoutMs` elapses. Node's fetch may deliver events emitted close together as + * separate chunks across separate `read()` calls (rather than coalesced into one), so tests + * asserting on multiple SSE events must accumulate across reads instead of reading once. + */ +async function readUntilContains(reader: ReadableStreamDefaultReader, needles: string[], timeoutMs = 2000): Promise { + const decoder = new TextDecoder(); + let text = ''; + const deadline = Date.now() + timeoutMs; + + while (!needles.every(needle => text.includes(needle))) { + if (Date.now() > deadline) { + throw new Error(`Timed out waiting for [${needles.join(', ')}] in SSE stream. Received so far:\n${text}`); + } + const { value, done } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + + return text; +} + /** * Test server configuration for NodeStreamableHTTPServerTransport tests */ @@ -727,9 +750,9 @@ describe('Zod v4', () => { const reader = response.body?.getReader(); - // The responses may come in any order or together in one chunk - const { value } = await reader!.read(); - const text = new TextDecoder().decode(value); + // The responses may come in any order, and Node's fetch may deliver them as + // separate chunks rather than coalesced into one, so accumulate until both arrive. + const text = await readUntilContains(reader!, ['"id":"req-1"', '"id":"req-2"']); // Check that both responses were sent on the same stream expect(text).toContain('"id":"req-1"'); @@ -1480,10 +1503,11 @@ describe('Zod v4', () => { // Send a server notification through the MCP server await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'First notification from MCP server' }); - // Read the notification from the SSE stream + // Read the notification from the SSE stream. Node's fetch may deliver the + // preceding connection event and this notification as separate chunks, so + // accumulate reads until the expected text shows up. const reader = sseResponse.body?.getReader(); - const { value } = await reader!.read(); - const text = new TextDecoder().decode(value); + const text = await readUntilContains(reader!, ['id: ', 'First notification from MCP server']); // Verify the notification was sent with an event ID expect(text).toContain('id: '); From 40d497b8c5e6987aae8881b3c6e0ce57ff8cdb95 Mon Sep 17 00:00:00 2001 From: Huzaifa Farooq Date: Mon, 14 Sep 2026 06:36:59 +0500 Subject: [PATCH 2/2] fix: race reader.read() against the timeout, not just check it beforehand readUntilContains only checked the deadline before calling reader.read(). If the expected event never arrives (e.g. a batch handler emits only one response), that read() call itself hangs indefinitely, so the helper's advertised timeoutMs never fires and the test instead waits for vitest's runner-level timeout. Race each read() against a per-iteration timer so a missing event fails at the intended deadline with the helper's diagnostic message. --- .../node/test/streamableHttp.test.ts | 20 +++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 209f06875e..da6fcdaeab 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -43,15 +43,23 @@ async function getFreePort() { async function readUntilContains(reader: ReadableStreamDefaultReader, needles: string[], timeoutMs = 2000): Promise { const decoder = new TextDecoder(); let text = ''; - const deadline = Date.now() + timeoutMs; while (!needles.every(needle => text.includes(needle))) { - if (Date.now() > deadline) { - throw new Error(`Timed out waiting for [${needles.join(', ')}] in SSE stream. Received so far:\n${text}`); + let timer: ReturnType; + const timeout = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`Timed out waiting for [${needles.join(', ')}] in SSE stream. Received so far:\n${text}`)), + timeoutMs + ); + }); + + try { + const { value, done } = await Promise.race([reader.read(), timeout]); + if (done) break; + text += decoder.decode(value, { stream: true }); + } finally { + clearTimeout(timer!); } - const { value, done } = await reader.read(); - if (done) break; - text += decoder.decode(value, { stream: true }); } return text;