From 713c448e75ce20aca4af47eb2b9ce4bee12aebd1 Mon Sep 17 00:00:00 2001 From: Loi Nguyen <1948922+lntutor@users.noreply.github.com> Date: Sun, 19 Jul 2026 14:53:35 +0700 Subject: [PATCH] fix(client): thread onresumptiontoken and onRequestStreamEnd through the send() resume path The resumptionToken branch of StreamableHTTPClientTransport._send() reopened the SSE stream with the right Last-Event-ID but passed neither onresumptiontoken nor onRequestStreamEnd to _startOrAuthSse(), while the original POST path and resumeStream() both thread them. A caller resuming a long-running request therefore never saw newer resumption tokens (its persisted token went stale, causing replayed or lost events on the next resume) and never learned when the stream ended non-resumably. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01N6RtoHuxrDqTUo9Mw9h4Cv --- .changeset/send-resume-threads-callbacks.md | 5 ++ packages/client/src/client/streamableHttp.ts | 7 ++- .../client/test/client/streamableHttp.test.ts | 48 +++++++++++++++++++ 3 files changed, 59 insertions(+), 1 deletion(-) create mode 100644 .changeset/send-resume-threads-callbacks.md diff --git a/.changeset/send-resume-threads-callbacks.md b/.changeset/send-resume-threads-callbacks.md new file mode 100644 index 0000000000..02df5e4150 --- /dev/null +++ b/.changeset/send-resume-threads-callbacks.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport.send()` no longer drops `onresumptiontoken` and `onRequestStreamEnd` when called with a `resumptionToken`. The resume branch opened the GET with the correct `Last-Event-ID` but passed neither callback to the reopened stream, so a caller resuming a long-running request stopped receiving newer resumption tokens — its persisted token went permanently stale, and a later resume replayed already-delivered events or fell outside the server event store's retention window — and was never notified when the stream ended non-resumably. Both callbacks are now threaded through, matching the original POST path and `resumeStream()`. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 9067fd1ec4..61cf502d60 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -927,10 +927,15 @@ export class StreamableHTTPClientTransport implements Transport { // same per-request abort as the original POST — modern-era // cancel-via-stream-close routes through `requestSignal`, and // without it a resumed long-running request would not cancel. + // `onresumptiontoken` and `onRequestStreamEnd` must survive the + // resume as well: the caller keeps persisting newer tokens and + // must still learn when the stream ends non-resumably. this._startOrAuthSse({ resumptionToken, + onresumptiontoken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, - requestSignal: options?.requestSignal + requestSignal: options?.requestSignal, + onRequestStreamEnd: options?.onRequestStreamEnd }).catch(error => this.onerror?.(error)); return; } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..54c7459e97 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1984,6 +1984,54 @@ describe('StreamableHTTPClientTransport', () => { // Resumption token callback may be invoked, but the primary assertion // here is that no JSON parse errors occurred for the priming event. }); + + it('should thread onresumptiontoken and onRequestStreamEnd through the send() resume path', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 0 + } + }); + + const resumptionTokenSpy = vi.fn(); + const streamEndSpy = vi.fn(); + + // The resumed stream delivers one id-bearing event, then ends + // (maxRetries 0 above: the end is non-resumable, no reconnect). + const resumedStream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('id: evt-2\ndata: {"jsonrpc":"2.0","result":{},"id":"req-1"}\n\n')); + controller.close(); + } + }); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: resumedStream + }); + + await transport.start(); + transport.send( + { jsonrpc: '2.0', method: 'tools/call', id: 'req-1', params: {} }, + { resumptionToken: 'evt-1', onresumptiontoken: resumptionTokenSpy, onRequestStreamEnd: streamEndSpy } + ); + + await vi.advanceTimersByTimeAsync(50); + + // Sanity: the resume send() opened a GET carrying Last-Event-ID + expect(fetchMock.mock.calls[0]![1]?.method).toBe('GET'); + expect((fetchMock.mock.calls[0]![1]?.headers as Headers).get('last-event-id')).toBe('evt-1'); + // The caller's callbacks must stay wired across the resume: + // new tokens are reported for persistence… + expect(resumptionTokenSpy).toHaveBeenCalledWith('evt-2'); + // …and the non-resumable end of the request stream is surfaced. + expect(streamEndSpy).toHaveBeenCalled(); + }); }); it('invalidates all credentials on OAuthErrorCode.InvalidClient during auth', async () => {