Skip to content

test(node): read SSE streams until expected events arrive instead of assuming one fetch chunk #2662

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
claude wants to merge 3 commits into
base: main
Choose a base branch
from
+48 −51
Open

test(node): read SSE streams until expected events arrive instead of assuming one fetch chunk #2662

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
29d9e2c
test(node): read SSE streams until expected events arrive instead of …
claude Aug 14, 2026
30a4cc6
test(node): correct comment — the standalone GET stream writes no pri…
claude Aug 14, 2026
3988a19
test(node): fold remaining single-read and inline read-loop sites int…
claude Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 48 additions & 51 deletions packages/middleware/node/test/streamableHttp.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,15 +95,48 @@ const TEST_MESSAGES = {

/**
* Helper to extract text from SSE response
* Note: Can only be called once per response stream. For multiple reads,
* get the reader manually and read multiple times.
* Note: Can only be called once per response stream. For multiple reads or
* when events may arrive across chunks, use readSSEUntil below.
*/
async function readSSEEvent(response: Response): Promise<string> {
const reader = response.body?.getReader();
const { value } = await reader!.read();
return new TextDecoder().decode(value);
}

/**
* Helper to read an SSE response stream until the accumulated text satisfies
* a predicate, the stream ends, or a timeout elapses. Returns whatever text
* was accumulated.
*
* fetch makes no guarantee that SSE events written separately by the server
* arrive in a single chunk — Node 26.7 delivers them in separate reads — so
* a test expecting more than one event must never assert on the result of a
* single read(). See #2661.
*
* The reader's lock is released on return, so the caller can still cancel
* the stream via `response.body.cancel()`.
*/
Comment thread
claude[bot] marked this conversation as resolved.
async function readSSEUntil(response: Response, predicate: (text: string) => boolean, timeoutMs = 2000): Promise<string> {
const reader = response.body!.getReader();
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 });
}
} finally {
clearTimeout(timeout);
reader.releaseLock();
}
return text;
}
Comment thread
claude[bot] marked this conversation as resolved.

/**
* Helper to send JSON-RPC request
*/
Expand Down Expand Up @@ -725,11 +758,9 @@ 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 each may arrive in its
// own chunk — read until both are in.
const text = await readSSEUntil(response, t => t.includes('"id":"req-1"') && t.includes('"id":"req-2"'));

// Check that both responses were sent on the same stream
expect(text).toContain('"id":"req-1"');
Comment thread
claude[bot] marked this conversation as resolved.
Expand Down Expand Up @@ -1314,9 +1345,7 @@ describe('Zod v4', () => {

expect(response.status).toBe(200);

const reader = response.body?.getReader();
const { value } = await reader!.read();
const text = new TextDecoder().decode(value);
const text = await readSSEUntil(response, t => t.includes('"id":"batch-1"'));

expect(text).toContain('"id":"batch-1"');
expect(text).toContain('"tools"');
Expand Down Expand Up @@ -1480,10 +1509,9 @@ 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
const reader = sseResponse.body?.getReader();
const { value } = await reader!.read();
const text = new TextDecoder().decode(value);
// Read the notification from the SSE stream (it may arrive split
// across multiple chunks)
const text = await readSSEUntil(sseResponse, t => t.includes('First notification from MCP server'));

// Verify the notification was sent with an event ID
expect(text).toContain('id: ');
Expand All @@ -1498,7 +1526,7 @@ describe('Zod v4', () => {
await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Second notification from MCP server' });

// Close the first SSE stream to simulate a disconnect
await reader!.cancel();
await sseResponse.body!.cancel();

// Reconnect with the Last-Event-ID to get missed messages
const reconnectResponse = await fetch(baseUrl, {
Expand All @@ -1513,10 +1541,9 @@ 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 (replayed events may span
// multiple chunks)
const reconnectText = await readSSEUntil(reconnectResponse, t => t.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');
Expand Down Expand Up @@ -1570,24 +1597,8 @@ describe('Zod v4', () => {

expect(reconnectResponse.status).toBe(200);

// Read replayed notifications with a timeout
const reconnectReader = reconnectResponse.body?.getReader();
let allText = '';

// Read chunks until we have all 3 notifications or timeout
const readWithTimeout = async () => {
const timeout = setTimeout(() => reconnectReader!.cancel(), 2000);
try {
while (!allText.includes('Missed notification 3')) {
const { value, done } = await reconnectReader!.read();
if (done) break;
allText += new TextDecoder().decode(value);
}
} finally {
clearTimeout(timeout);
}
};
await readWithTimeout();
const allText = await readSSEUntil(reconnectResponse, t => t.includes('Missed notification 3'));

// Verify we received ALL notifications that were sent while disconnected
expect(allText).toContain('Missed notification 1');
Expand Down Expand Up @@ -2369,21 +2380,7 @@ describe('Zod v4', () => {
expect(reconnectResponse.status).toBe(200);

// Read the replayed notification
const reconnectReader = reconnectResponse.body?.getReader();
let allText = '';
const readWithTimeout = async () => {
const timeout = setTimeout(() => reconnectReader!.cancel(), 5000);
try {
while (!allText.includes('Missed while disconnected')) {
const { value, done } = await reconnectReader!.read();
if (done) break;
allText += new TextDecoder().decode(value);
}
} finally {
clearTimeout(timeout);
}
};
await readWithTimeout();
const allText = await readSSEUntil(reconnectResponse, t => t.includes('Missed while disconnected'), 5000);

// Verify we received the notification that was sent while disconnected
expect(allText).toContain('Missed while disconnected');
Expand Down
Loading