diff --git a/.changeset/fix-sse-test-chunking.md b/.changeset/fix-sse-test-chunking.md new file mode 100644 index 0000000000..831c56c595 --- /dev/null +++ b/.changeset/fix-sse-test-chunking.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/node': patch +--- + +Test-only change: the node middleware SSE tests now accumulate a streamed response across multiple fetch chunks instead of assuming a complete event arrives in the first read, so chunked-delivery regressions fail the suite rather than passing intermittently. No runtime change. diff --git a/packages/middleware/node/test/streamableHttp.test.ts b/packages/middleware/node/test/streamableHttp.test.ts index 901b78446f..c1dbf75e51 100644 --- a/packages/middleware/node/test/streamableHttp.test.ts +++ b/packages/middleware/node/test/streamableHttp.test.ts @@ -104,6 +104,40 @@ async function readSSEEvent(response: Response): Promise { return new TextDecoder().decode(value); } +/** + * Read an SSE response until the expected messages arrive. + * + * Fetch is allowed to split a single logical SSE response across multiple + * chunks, so callers must not assume that the first read contains a complete + * event or the whole response. + */ +async function readSSEUntil(response: Response, predicate: (text: string) => boolean, timeoutMs = 2000): Promise { + const reader = response.body?.getReader(); + if (!reader) { + throw new Error('Expected an SSE response body'); + } + + const decoder = new TextDecoder(); + let text = ''; + const timeout = setTimeout(() => { + void reader.cancel(); + }, timeoutMs); + + try { + while (!predicate(text)) { + const { value, done } = await reader.read(); + if (done) { + break; + } + text += decoder.decode(value, { stream: true }); + } + return text + decoder.decode(); + } finally { + clearTimeout(timeout); + await reader.cancel().catch(() => {}); + } +} + /** * Helper to send JSON-RPC request */ @@ -736,11 +770,10 @@ describe('Zod v4', () => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); - 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 may be split across chunks. + const text = await readSSEUntil(response, accumulated => { + return accumulated.includes('"id":"req-1"') && accumulated.includes('"id":"req-2"'); + }); // Check that both responses were sent on the same stream expect(text).toContain('"id":"req-1"'); @@ -1524,10 +1557,11 @@ describe('Zod v4', () => { expect(reconnectResponse.status).toBe(200); - // Read the replayed notification - const reconnectReader = reconnectResponse.body?.getReader(); - const reconnectData = await reconnectReader!.read(); - const reconnectText = new TextDecoder().decode(reconnectData.value); + // Read the replayed notification, which may arrive after an earlier + // event in a separate fetch chunk. + const reconnectText = await readSSEUntil(reconnectResponse, accumulated => + accumulated.includes('Second notification from MCP server') + ); // Verify we received the second notification that was sent after our stored eventId expect(reconnectText).toContain('Second notification from MCP server');