diff --git a/.changeset/sse-resume-token-seed.md b/.changeset/sse-resume-token-seed.md new file mode 100644 index 0000000000..d4ea72e0ca --- /dev/null +++ b/.changeset/sse-resume-token-seed.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport` now re-sends the `Last-Event-ID` header when a resumed SSE stream disconnects before any id-bearing event arrives (LB idle timeout, server restart). Previously the reconnect GET was sent without the header, so the server treated it as a brand-new stream and never replayed the missed events, hanging long-running requests. Fixes #2499 diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index c91c6db008..b0e4c060b0 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -739,7 +739,12 @@ export class StreamableHTTPClientTransport implements Transport { // caller just tore down. const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; - let lastEventId: string | undefined; + // Seed from the resumption token the stream was opened with: if the + // stream disconnects before any id-bearing event arrives, the reconnect + // must re-send that token (replay is idempotent). Without this, the + // reconnect GET goes out without Last-Event-ID and the server treats it + // as a brand-new stream, never replaying the missed events. + let lastEventId: string | undefined = options.resumptionToken; // Track whether we've received a priming event (event with ID) // Per spec, server SHOULD send a priming event with ID before closing let hasPrimingEvent = false; diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 17fd2df276..7d90e7f07d 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2625,6 +2625,52 @@ describe('StreamableHTTPClientTransport', () => { const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers; expect(secondCallHeaders?.get('last-event-id')).toBe('evt-1'); }); + + it('re-sends Last-Event-ID when a resumed stream closes before any id-bearing event', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + + // The resumed stream is accepted but closes immediately without any + // event carrying an id (LB idle timeout, server restart). + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.close(); + } + }) + }); + // The reconnecting stream stays open so no further reconnection is + // scheduled after the one under test. + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream() + }); + + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'evt-1' }); + + await vi.advanceTimersByTimeAsync(50); + await vi.advanceTimersByTimeAsync(150); + + // The reconnect GET must re-send the token the stream was opened + // with — replay is idempotent, and without it the server treats the + // reconnect as a brand-new stream and never replays missed events. + expect(fetchMock).toHaveBeenCalledTimes(2); + const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers; + expect(secondCallHeaders?.get('last-event-id')).toBe('evt-1'); + }); }); describe('Reconnection Logic with maxRetries 0', () => {