From f18c728965b535993a13e430f563f2137f902a0b Mon Sep 17 00:00:00 2001 From: Vishnu Jayavel Date: Thu, 16 Jul 2026 22:02:46 -0700 Subject: [PATCH] fix(client): seed the SSE resumption tracker from resumptionToken A resumed stream (Last-Event-ID: e1) that disconnects again before any new id-bearing event arrives (LB idle timeout, server restart) was reconnecting with no Last-Event-ID header at all, because _handleSseStream() tracked the latest event id in a local variable that always started undefined instead of being seeded from the resumptionToken the stream was (re)opened with. The server then treated the reconnect as a brand-new stream instead of a resumption. Seed the tracker from options.resumptionToken so a reconnect that saw no new events re-sends the same token. Fixes #2499 --- .../resumed-stream-lost-last-event-id.md | 5 ++ packages/client/src/client/streamableHttp.ts | 6 ++- .../client/test/client/streamableHttp.test.ts | 52 +++++++++++++++++++ 3 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 .changeset/resumed-stream-lost-last-event-id.md diff --git a/.changeset/resumed-stream-lost-last-event-id.md b/.changeset/resumed-stream-lost-last-event-id.md new file mode 100644 index 0000000000..da690d43d3 --- /dev/null +++ b/.changeset/resumed-stream-lost-last-event-id.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport` no longer drops the resumption token when a resumed SSE stream disconnects before delivering any id-bearing event. `_handleSseStream()` tracked the latest event id in a local variable that always started `undefined`, instead of being seeded from the `resumptionToken` the stream was (re)opened with. A stream resumed with `Last-Event-ID: e1` that disconnected again before any new event arrived (load-balancer idle timeout, server restart) would therefore schedule its reconnect GET with no `Last-Event-ID` header at all, so the server treated it as a brand-new stream instead of a resumption — missed events were never replayed. The tracker is now seeded from `options.resumptionToken`, so a reconnect that saw no new events re-sends the same token instead of silently losing it. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 9067fd1ec4..deac866e26 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -694,7 +694,11 @@ 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 (re)opened with, so a + // resumed stream that disconnects again before any id-bearing event + // arrives still reconnects with the same `Last-Event-ID` instead of + // silently dropping it and starting a fresh, non-resumed stream. + 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 a36bbc0ad3..2676730be9 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2462,6 +2462,58 @@ describe('StreamableHTTPClientTransport', () => { const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers; expect(secondCallHeaders?.get('last-event-id')).toBe('evt-1'); }); + + it('should preserve the resumption token when a resumed stream disconnects before any id-bearing event', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + + // Resumed GET stream that closes immediately without ever sending + // an id-bearing event (e.g. LB idle timeout, server restart). + const emptyStream = new ReadableStream({ + start(controller) { + controller.close(); + } + }); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: emptyStream + }); + + // Second request for reconnection + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream() + }); + + await transport.start(); + // Resume from a previously-seen event id, mirroring resumeStream()/reconnect. + await transport['_startOrAuthSse']({ resumptionToken: 'event-1' }); + + // The resumed GET itself carries the token. + const firstCallHeaders = fetchMock.mock.calls[0]![1]?.headers; + expect(firstCallHeaders?.get('last-event-id')).toBe('event-1'); + + // Wait for the stream to close and the reconnection to be scheduled. + await vi.advanceTimersByTimeAsync(50); + + expect(fetchMock).toHaveBeenCalledTimes(2); + // The reconnect must carry the SAME token instead of dropping it + // because no id-bearing event arrived on the resumed stream. + const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers; + expect(secondCallHeaders?.get('last-event-id')).toBe('event-1'); + }); }); describe('Reconnection Logic with maxRetries 0', () => {