diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 140717c6bb..da6fcdaeab 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -34,6 +34,37 @@ 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 = ''; + + while (!needles.every(needle => text.includes(needle))) { + 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!); + } + } + + return text; +} + /** * Test server configuration for NodeStreamableHTTPServerTransport tests */ @@ -727,9 +758,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 +1511,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: ');