From b1f829860132da30caa01baf62fe036908c3630b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 05:23:10 +0000 Subject: [PATCH 01/10] fix(client): persist SSE reconnection attempt count across idle-close cycles The standby GET/SSE reconnect path hardcoded attempt count 0, so a server that gracefully idle-closes the stream kept the client reconnecting forever at initialReconnectionDelay with maxRetries never tripping. Thread the attempt number through to the stream it produces and only reset it once the stream delivers a message. Fixes #2682 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- .changeset/bound-standby-sse-reconnects.md | 7 + packages/client/src/client/streamableHttp.ts | 56 ++++++-- .../client/test/client/streamableHttp.test.ts | 128 ++++++++++++++++++ 3 files changed, 181 insertions(+), 10 deletions(-) create mode 100644 .changeset/bound-standby-sse-reconnects.md diff --git a/.changeset/bound-standby-sse-reconnects.md b/.changeset/bound-standby-sse-reconnects.md new file mode 100644 index 0000000000..c3ca9a7868 --- /dev/null +++ b/.changeset/bound-standby-sse-reconnects.md @@ -0,0 +1,7 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Bound standby SSE reconnects when the server keeps idle-closing the stream. `StreamableHTTPClientTransport` reset its reconnection attempt count to `0` every time a stream ended, so a server that gracefully idle-closes the standby GET/SSE stream (spec-compliant behavior) kept the client reconnecting forever at `initialReconnectionDelay` — `maxRetries` never tripped, and every cycle re-ran the authenticated fetch path. + +The attempt count now persists across connect-then-close cycles that deliver no messages: after `maxRetries` consecutive fruitless reconnects the transport stops and surfaces `onerror` ("Maximum reconnection attempts exceeded"), exactly as it already did for reconnects that fail outright. A stream that delivers a message still resets the count, so healthy long-lived streams reconnect indefinitely as before. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..7d38e47571 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -116,7 +116,11 @@ export interface StreamableHTTPReconnectionOptions { reconnectionDelayGrowFactor: number; /** - * Maximum number of reconnection attempts before giving up. + * Maximum number of consecutive reconnection attempts before giving up. + * An attempt counts against this limit when it fails outright or when the + * reconnected stream ends again without having delivered a message (for + * example, a server that gracefully idle-closes its standby SSE stream); + * a stream that delivers a message resets the count. * Default is 2. */ maxRetries: number; @@ -517,7 +521,7 @@ export class StreamableHTTPClientTransport implements Transport { return typeof v === 'string' && isModernProtocolVersion(v); } - private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise { + private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0, reconnectAttempt = 0): Promise { const { resumptionToken, requestSignal } = options; // Same guard as `_handleSseStream`: a resurrected listen stream (the // POST-SSE → GET reconnect path threads `requestSignal` through @@ -575,7 +579,7 @@ export class StreamableHTTPClientTransport implements Transport { } await response.text?.().catch(() => {}); // Purposely _not_ awaited, so we don't call onerror twice - return this._startOrAuthSse(options, true, stepUpRetries); + return this._startOrAuthSse(options, true, stepUpRetries, reconnectAttempt); } await response.text?.().catch(() => {}); if (isAuthRetry) { @@ -600,7 +604,7 @@ export class StreamableHTTPClientTransport implements Transport { if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); } - return this._startOrAuthSse(options, isAuthRetry, stepUpRetries + 1); + return this._startOrAuthSse(options, isAuthRetry, stepUpRetries + 1, reconnectAttempt); } } @@ -627,7 +631,7 @@ export class StreamableHTTPClientTransport implements Transport { }); } - this._handleSseStream(response.body, options, true); + this._handleSseStream(response.body, options, true, reconnectAttempt); } catch (error) { if (!isIntentionalAbort()) { this.onerror?.(error as Error); @@ -684,7 +688,13 @@ export class StreamableHTTPClientTransport implements Transport { // (a listen subscription closed during the backoff delay): do not // resurrect a stream the caller already tore down. if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return; - this._startOrAuthSse(options).catch(error => { + // Thread the attempt number into the stream this attempt produces: + // if the server accepts the connection but the stream ends again + // without delivering a message, `_handleSseStream` continues the + // count instead of restarting it — otherwise a server that + // gracefully idle-closes the stream on every reconnect would keep + // the transport looping forever with `maxRetries` never tripping. + this._startOrAuthSse(options, false, 0, attemptCount + 1).catch(error => { if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return; this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`)); try { @@ -704,7 +714,19 @@ export class StreamableHTTPClientTransport implements Transport { } } - private _handleSseStream(stream: ReadableStream | null, options: StartSSEOptions, isReconnectable: boolean): void { + /** + * @param reconnectAttempt - Which reconnection attempt produced this + * stream (0 for an original stream). Carried into the next + * `_scheduleReconnection` call when the stream ends without having + * delivered a message, so consecutive fruitless reconnects are bounded by + * `maxRetries`; a stream that delivered a message resets the count. + */ + private _handleSseStream( + stream: ReadableStream | null, + options: StartSSEOptions, + isReconnectable: boolean, + reconnectAttempt = 0 + ): void { if (!stream) { // A null body on a per-request stream (or its GET resume) is the // same terminal non-resumable outcome as a 405 — fire the @@ -728,6 +750,12 @@ export class StreamableHTTPClientTransport implements Transport { // Track whether we've received a response - if so, no need to reconnect // Reconnection is for when server disconnects BEFORE sending response let receivedResponse = false; + // Track whether the stream delivered any message at all. A stream that + // did is genuinely working, so its next reconnection starts a fresh + // attempt count; a stream that ended without one continues the count, + // keeping repeated connect/idle-close cycles bounded by `maxRetries` + // (#2682). + let receivedMessage = false; const processStream = async () => { // this is the closest we can get to trying to catch network errors // if something happens reader will throw @@ -775,6 +803,7 @@ export class StreamableHTTPClientTransport implements Transport { message.id = replayMessageId; } } + receivedMessage = true; this.onmessage?.(message); } catch (error) { this.onerror?.(error as Error); @@ -789,6 +818,11 @@ export class StreamableHTTPClientTransport implements Transport { const canResume = isReconnectable || hasPrimingEvent; const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { + // A stream that delivered a message was genuinely working — + // restart the attempt count. A graceful close without one is + // retry-equivalent to a failed attempt: continue the count so + // a server that idle-closes every standby stream cannot keep + // the transport reconnecting forever (#2682). this._scheduleReconnection( { resumptionToken: lastEventId, @@ -797,7 +831,7 @@ export class StreamableHTTPClientTransport implements Transport { requestSignal, onRequestStreamEnd }, - 0 + receivedMessage ? 0 : reconnectAttempt ); } else if (!isIntentionalAbort()) { // The per-request stream ended without reconnecting (no @@ -820,7 +854,9 @@ export class StreamableHTTPClientTransport implements Transport { const canResume = isReconnectable || hasPrimingEvent; const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { - // Use the exponential backoff reconnection strategy + // Use the exponential backoff reconnection strategy. Same + // accounting as the graceful-close path: only a stream that + // delivered a message restarts the attempt count. try { this._scheduleReconnection( { @@ -830,7 +866,7 @@ export class StreamableHTTPClientTransport implements Transport { requestSignal, onRequestStreamEnd }, - 0 + receivedMessage ? 0 : reconnectAttempt ); } catch (error) { this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..4a6e7c8690 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2526,6 +2526,134 @@ describe('StreamableHTTPClientTransport', () => { }); }); + describe('Reconnection attempt accounting (#2682)', () => { + let transport: StreamableHTTPClientTransport; + + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + /** An SSE response whose stream delivers the given chunks and then closes gracefully. */ + const sseResponse = (chunks: string[]) => ({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)); + } + controller.close(); + } + }) + }); + + const notificationEvent = 'data: {"jsonrpc":"2.0","method":"notifications/message","params":{}}\n\n'; + + it('bounds repeated graceful idle-closes of the standby stream by maxRetries', async () => { + // Regression test for #2682: a server that gracefully idle-closes + // the standby GET/SSE stream on every (re)connect must not keep the + // transport reconnecting forever — the attempt count has to persist + // across successful-connect-then-idle-close cycles. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 2 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const fetchMock = globalThis.fetch as Mock; + // Every GET opens fine and closes gracefully without delivering anything. + fetchMock.mockImplementation(async () => sseResponse([])); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(500); + + // Original stream + exactly maxRetries reconnection attempts. + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ + message: 'Maximum reconnection attempts (2) exceeded.' + }) + ); + + // And it stays stopped — no further reconnects are scheduled. + await vi.advanceTimersByTimeAsync(5000); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it('does not count streams that delivered messages against maxRetries', async () => { + // A stream that delivers a message is genuinely working; when it + // later closes gracefully, reconnection accounting starts fresh. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const messageSpy = vi.fn(); + transport.onmessage = messageSpy; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementation(async () => sseResponse([notificationEvent])); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(200); + + // Far more streams than maxRetries + 1 — every one delivered a + // message, so the reconnect loop keeps going by design. + expect(fetchMock.mock.calls.length).toBeGreaterThan(5); + expect(messageSpy).toHaveBeenCalled(); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('resets the attempt count after a productive stream, then bounds fruitless reconnects again', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 2 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const fetchMock = globalThis.fetch as Mock; + // GET 1: closes empty. GET 2: delivers a message (resets the + // count). GETs 3+: close empty until the limit trips again. + const bodies: string[][] = [[], [notificationEvent]]; + fetchMock.mockImplementation(async () => sseResponse(bodies.shift() ?? [])); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(500); + + // GET 1 (empty, attempt 0 scheduled) → GET 2 (productive, count + // resets) → GETs 3-4 (empty, attempts 0-1 of the fresh count) → + // limit trips. Without the reset this would stop after 3 fetches. + expect(fetchMock).toHaveBeenCalledTimes(4); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ + message: 'Maximum reconnection attempts (2) exceeded.' + }) + ); + + await vi.advanceTimersByTimeAsync(5000); + expect(fetchMock).toHaveBeenCalledTimes(4); + }); + }); + describe('prevent infinite recursion when server returns 401 after successful auth', () => { it('should throw error when server returns 401 after successful auth', async () => { const message: JSONRPCMessage = { From 1d296bd70e26cc856d3358f8cf6610cb7c073828 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 05:36:41 +0000 Subject: [PATCH 02/10] review: count long-lived idle streams as progress, keep resumption token, dedupe scheduling - A stream that stays open at least maxReconnectionDelay resets the attempt count even without messages, so healthy-but-quiet sessions whose standby stream is periodically idle-closed (e.g. behind an ALB) keep reconnecting indefinitely; only rapid connect-then-close cycles are bounded by maxRetries. - Hoist a single scheduleNext() closure used by both the graceful-close and mid-stream-error branches (the duplicated block is how the original bug happened). - Fall back to the stream's own resumptionToken when it ended before any event arrived, so empty reconnects no longer drop Last-Event-ID. - Update the migration guide bullet that claimed exhaustion accounting was unchanged from v1. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- .changeset/bound-standby-sse-reconnects.md | 6 +- docs/migration/upgrade-to-v2.md | 17 ++-- packages/client/src/client/streamableHttp.ts | 82 +++++++++++-------- .../client/test/client/streamableHttp.test.ts | 75 +++++++++++++++++ 4 files changed, 140 insertions(+), 40 deletions(-) diff --git a/.changeset/bound-standby-sse-reconnects.md b/.changeset/bound-standby-sse-reconnects.md index c3ca9a7868..a4908f74f1 100644 --- a/.changeset/bound-standby-sse-reconnects.md +++ b/.changeset/bound-standby-sse-reconnects.md @@ -2,6 +2,8 @@ '@modelcontextprotocol/client': patch --- -Bound standby SSE reconnects when the server keeps idle-closing the stream. `StreamableHTTPClientTransport` reset its reconnection attempt count to `0` every time a stream ended, so a server that gracefully idle-closes the standby GET/SSE stream (spec-compliant behavior) kept the client reconnecting forever at `initialReconnectionDelay` — `maxRetries` never tripped, and every cycle re-ran the authenticated fetch path. +Bound standby SSE reconnects when the server rapidly idle-closes the stream. `StreamableHTTPClientTransport` reset its reconnection attempt count to `0` every time a stream ended, so a server that gracefully idle-closes the standby GET/SSE stream immediately after every reconnect (spec-compliant behavior) kept the client reconnecting forever at `initialReconnectionDelay` — `maxRetries` never tripped, and every cycle re-ran the authenticated fetch path. -The attempt count now persists across connect-then-close cycles that deliver no messages: after `maxRetries` consecutive fruitless reconnects the transport stops and surfaces `onerror` ("Maximum reconnection attempts exceeded"), exactly as it already did for reconnects that fail outright. A stream that delivers a message still resets the count, so healthy long-lived streams reconnect indefinitely as before. +The attempt count now persists across connect-then-close cycles that make no progress: after `maxRetries` consecutive fruitless reconnects the transport stops and surfaces `onerror` ("Maximum reconnection attempts exceeded"), exactly as it already did for reconnects that fail outright. A stream counts as having made progress — and resets the count — when it delivers a message or stays open for at least `maxReconnectionDelay`, so healthy sessions whose idle standby stream is periodically closed by the server or an intermediary keep reconnecting indefinitely as before. + +Also fixed in the same path: a reconnected stream that ended before any event arrived no longer drops the `Last-Event-ID` resumption token it was opened with — the next attempt resumes from the same token instead of silently starting a fresh stream. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 8f4cd5990d..d2783ec997 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1511,11 +1511,18 @@ rewrite required unless noted. no `notifications/cancelled` goes on the wire — the spec forbids cancelling `initialize`, and v1 sent one anyway. v1 tests asserting that notification need re-baselining. -- **Also unchanged: SSE reconnection exhaustion.** `StreamableHTTPClientTransport`'s - standalone GET-stream reconnection behavior and its exhaustion signal carry over from - v1: when retries run out, the transport emits `onerror` with a plain `Error` whose - message is `Maximum reconnection attempts (N) exceeded.` — there is no typed error - class for this condition, so monitors that match the message text keep working. +- **Changed: SSE reconnection exhaustion accounting.** The exhaustion _signal_ carries + over from v1: when retries run out, `StreamableHTTPClientTransport` emits `onerror` + with a plain `Error` whose message is `Maximum reconnection attempts (N) exceeded.` — + there is no typed error class for this condition, so monitors that match the message + text keep working. The _accounting_ changed: v1 reset the retry counter every time a + stream closed, so a server that gracefully idle-closed the standby GET stream + immediately after each reconnect kept the client reconnecting forever and + `maxRetries` never fired. v2 counts consecutive reconnects that make no progress — + no message delivered and the connection lasted less than `maxReconnectionDelay` — so + such loops now stop with the error above after `maxRetries` attempts. Streams that + deliver a message or stay open at least `maxReconnectionDelay` reset the counter and + reconnect indefinitely, exactly as in v1. - **Also unchanged: elicitation response validation.** `elicitInput`'s local validation of elicitation responses against `requestedSchema`, the resulting `-32602` error message wording (`Elicitation response content does not match requested schema: …`), diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 7d38e47571..7a97043d1d 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -118,9 +118,14 @@ export interface StreamableHTTPReconnectionOptions { /** * Maximum number of consecutive reconnection attempts before giving up. * An attempt counts against this limit when it fails outright or when the - * reconnected stream ends again without having delivered a message (for - * example, a server that gracefully idle-closes its standby SSE stream); - * a stream that delivers a message resets the count. + * reconnected stream ends again without making progress — no message + * delivered and the connection lasted less than + * {@linkcode maxReconnectionDelay} (for example, a server that gracefully + * idle-closes its standby SSE stream immediately after every reconnect). + * A stream that delivers a message or stays open at least + * {@linkcode maxReconnectionDelay} resets the count, so healthy sessions + * whose standby stream is periodically idle-closed keep reconnecting + * indefinitely. * Default is 2. */ maxRetries: number; @@ -717,9 +722,10 @@ export class StreamableHTTPClientTransport implements Transport { /** * @param reconnectAttempt - Which reconnection attempt produced this * stream (0 for an original stream). Carried into the next - * `_scheduleReconnection` call when the stream ends without having - * delivered a message, so consecutive fruitless reconnects are bounded by - * `maxRetries`; a stream that delivered a message resets the count. + * `_scheduleReconnection` call when the stream ends without having made + * progress (no message delivered, connection shorter than + * `maxReconnectionDelay`), so consecutive fruitless reconnects are bounded + * by `maxRetries`; a stream that made progress resets the count. */ private _handleSseStream( stream: ReadableStream | null, @@ -756,6 +762,40 @@ export class StreamableHTTPClientTransport implements Transport { // keeping repeated connect/idle-close cycles bounded by `maxRetries` // (#2682). let receivedMessage = false; + const streamOpenedAt = Date.now(); + + // Single scheduling site for both the graceful-close and the + // mid-stream-error paths below — the #2682 bug existed precisely + // because the two branches carried separate copies of this block. + const scheduleNext = (): void => { + // A stream counts as having made progress when it delivered a + // message, or when it stayed open for at least + // `maxReconnectionDelay` before ending — an idle standby stream + // that a server (or intermediary) periodically closes is healthy, + // and resetting the count for it can never produce a reconnect + // rate faster than the maximum backoff already permits. Without + // progress, a connect-then-close cycle is retry-equivalent to a + // failed attempt: the count continues so a server that idle-closes + // every standby stream right away cannot keep the transport + // looping forever (#2682). Priming events alone deliberately do + // not reset the count — a server can send one and still close + // immediately, which would re-arm exactly that loop. + const madeProgress = receivedMessage || Date.now() - streamOpenedAt >= this._reconnectionOptions.maxReconnectionDelay; + this._scheduleReconnection( + { + // A reconnected stream that ended before any event arrived + // must not drop the token the stream was opened with — + // fall back to it so the next attempt still resumes. + resumptionToken: lastEventId ?? options.resumptionToken, + onresumptiontoken, + replayMessageId, + requestSignal, + onRequestStreamEnd + }, + madeProgress ? 0 : reconnectAttempt + ); + }; + const processStream = async () => { // this is the closest we can get to trying to catch network errors // if something happens reader will throw @@ -818,21 +858,7 @@ export class StreamableHTTPClientTransport implements Transport { const canResume = isReconnectable || hasPrimingEvent; const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { - // A stream that delivered a message was genuinely working — - // restart the attempt count. A graceful close without one is - // retry-equivalent to a failed attempt: continue the count so - // a server that idle-closes every standby stream cannot keep - // the transport reconnecting forever (#2682). - this._scheduleReconnection( - { - resumptionToken: lastEventId, - onresumptiontoken, - replayMessageId, - requestSignal, - onRequestStreamEnd - }, - receivedMessage ? 0 : reconnectAttempt - ); + scheduleNext(); } else if (!isIntentionalAbort()) { // The per-request stream ended without reconnecting (no // priming event for a POST stream, or response already @@ -855,19 +881,9 @@ export class StreamableHTTPClientTransport implements Transport { const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { // Use the exponential backoff reconnection strategy. Same - // accounting as the graceful-close path: only a stream that - // delivered a message restarts the attempt count. + // accounting as the graceful-close path. try { - this._scheduleReconnection( - { - resumptionToken: lastEventId, - onresumptiontoken, - replayMessageId, - requestSignal, - onRequestStreamEnd - }, - receivedMessage ? 0 : reconnectAttempt - ); + scheduleNext(); } catch (error) { this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); onRequestStreamEnd?.(); diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 4a6e7c8690..5e9be76600 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2652,6 +2652,81 @@ describe('StreamableHTTPClientTransport', () => { await vi.advanceTimersByTimeAsync(5000); expect(fetchMock).toHaveBeenCalledTimes(4); }); + + it('treats a long-lived idle stream as progress, so periodic idle-closes reconnect indefinitely', async () => { + // A healthy-but-quiet session behind e.g. a load balancer with an + // idle timeout: the standby stream delivers nothing but stays open + // well past maxReconnectionDelay before each close. That must NOT + // count against maxRetries — the notification channel stays alive. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 2 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const controllers: ReadableStreamDefaultController[] = []; + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementation(async () => ({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controllers.push(controller); + } + }) + })); + + await transport.start(); + await transport['_startOrAuthSse']({}); + + // Five cycles: each stream lives longer than maxReconnectionDelay + // (fake timers also drive Date.now), then idle-closes empty. + for (let cycle = 0; cycle < 5; cycle++) { + await vi.advanceTimersByTimeAsync(1500); + controllers[cycle]!.close(); + await vi.advanceTimersByTimeAsync(50); + } + + // Well past 1 + maxRetries fetches, and no exhaustion error. + expect(fetchMock.mock.calls.length).toBeGreaterThan(3); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('carries the resumption token through reconnected streams that ended before any event', async () => { + // Stream 1 delivers a priming event (id only), so the first + // reconnect resumes from it. Stream 2 ends empty — the next + // attempt must still send the same Last-Event-ID instead of + // dropping it and starting a fresh stream. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 3 + } + }); + + const fetchMock = globalThis.fetch as Mock; + const bodies: string[][] = [['id: event-1\ndata: \n\n']]; + fetchMock.mockImplementation(async () => sseResponse(bodies.shift() ?? [])); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(100); + + expect(fetchMock.mock.calls.length).toBeGreaterThanOrEqual(3); + const secondCallHeaders = fetchMock.mock.calls[1]![1]?.headers as Headers; + const thirdCallHeaders = fetchMock.mock.calls[2]![1]?.headers as Headers; + expect(secondCallHeaders.get('last-event-id')).toBe('event-1'); + // Before the fix this was null: the empty second stream dropped the token. + expect(thirdCallHeaders.get('last-event-id')).toBe('event-1'); + }); }); describe('prevent infinite recursion when server returns 401 after successful auth', () => { From b3d8c77aa322dd028bbc6bf79acbfc302d6a07af Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 05:49:04 +0000 Subject: [PATCH 03/10] review: thread stream callbacks through resumed send(), guard graceful-close scheduling - The send()-resume branch now forwards onresumptiontoken and onRequestStreamEnd to the resumed GET, so new event IDs keep reaching the caller's persistence hook and the pending request settles when reconnection attempts are exhausted instead of hanging to timeout. - Guard the graceful-close scheduleNext() call the same way as the error branch: exhaustion now runs user callbacks from graceful close, and a throwing handler must not fall into the outer catch and double-fire or surface a misleading disconnect error. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 24 +++++- .../client/test/client/streamableHttp.test.ts | 85 +++++++++++++++++++ 2 files changed, 107 insertions(+), 2 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 7a97043d1d..cafcac95f2 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -858,7 +858,19 @@ export class StreamableHTTPClientTransport implements Transport { const canResume = isReconnectable || hasPrimingEvent; const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { - scheduleNext(); + // Same guard as the error path below. With the fruitless + // accounting, exhaustion — which synchronously invokes + // `onerror`/`onRequestStreamEnd` and any custom + // `ReconnectionScheduler` — is reachable from a graceful + // close, and a throwing user callback must not fall into + // the outer catch (which would surface a misleading + // disconnect error and schedule a second time). + try { + scheduleNext(); + } catch (error) { + this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); + onRequestStreamEnd?.(); + } } else if (!isIntentionalAbort()) { // The per-request stream ended without reconnecting (no // priming event for a POST stream, or response already @@ -1006,10 +1018,18 @@ 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. + // Thread the caller's stream callbacks through as well: + // `onresumptiontoken` so new event IDs on the resumed stream + // keep reaching the caller's persistence hook, and + // `onRequestStreamEnd` so the pending request settles instead + // of hanging when the resumed stream ends for good (e.g. + // reconnection attempts are exhausted). 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 5e9be76600..b04eb146e9 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2727,6 +2727,91 @@ describe('StreamableHTTPClientTransport', () => { // Before the fix this was null: the empty second stream dropped the token. expect(thirdCallHeaders.get('last-event-id')).toBe('event-1'); }); + + it('threads the stream callbacks through a resumed send(), settling the caller on exhaustion', async () => { + // A send() with a resumptionToken resumes via GET. New event IDs on + // the resumed stream must reach the caller's onresumptiontoken, and + // when reconnection attempts are exhausted the caller's + // onRequestStreamEnd must fire so the pending request settles + // instead of hanging until its timeout. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const tokenSpy = vi.fn(); + const streamEndSpy = vi.fn(); + + const fetchMock = globalThis.fetch as Mock; + // Resumed stream delivers a fresh priming event then idle-closes; + // every stream after that closes empty until exhaustion. + const bodies: string[][] = [['id: event-next\ndata: \n\n']]; + fetchMock.mockImplementation(async () => sseResponse(bodies.shift() ?? [])); + + const requestMessage: JSONRPCRequest = { + jsonrpc: '2.0', + method: 'long_running_tool', + id: 'request-1', + params: {} + }; + + await transport.start(); + await transport.send(requestMessage, { + resumptionToken: 'event-0', + onresumptiontoken: tokenSpy, + onRequestStreamEnd: streamEndSpy + }); + await vi.advanceTimersByTimeAsync(200); + + // The resume goes out as a GET with the caller's token. + expect(fetchMock.mock.calls[0]![1]?.method).toBe('GET'); + expect((fetchMock.mock.calls[0]![1]?.headers as Headers).get('last-event-id')).toBe('event-0'); + // The fresh priming event reached the caller's persistence hook. + expect(tokenSpy).toHaveBeenCalledWith('event-next'); + // Exhaustion settled the caller instead of leaving it hanging. + expect(streamEndSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ + message: 'Maximum reconnection attempts (1) exceeded.' + }) + ); + }); + + it('does not double-fire exhaustion callbacks when a user onerror handler throws on graceful close', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 0 // exhaustion trips on the first graceful close + } + }); + const errorSpy = vi.fn().mockImplementationOnce(() => { + throw new Error('user handler exploded'); + }); + transport.onerror = errorSpy; + const streamEndSpy = vi.fn(); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementation(async () => sseResponse([])); + + await transport.start(); + await transport['_startOrAuthSse']({ onRequestStreamEnd: streamEndSpy }); + await vi.advanceTimersByTimeAsync(100); + + // The throwing handler is contained by the graceful branch's guard: + // the caller still settles exactly once, no reconnect is scheduled, + // and no misleading 'SSE stream disconnected' error is emitted. + expect(streamEndSpy).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledTimes(1); + const messages = errorSpy.mock.calls.map(args => (args[0] as Error).message); + expect(messages.some(m => m.includes('SSE stream disconnected'))).toBe(false); + }); }); describe('prevent infinite recursion when server returns 401 after successful auth', () => { From f83507f51dbb1fa02798ccc27f677d8f50930651 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 06:04:44 +0000 Subject: [PATCH 04/10] review: settle a resumed send() when the initial resume GET fails outright A resume GET that rejects before any stream exists bypasses every downstream settlement path (405/null-body, exhaustion), leaving the pending request hanging until its timeout. Fire the caller's onRequestStreamEnd from the resume branch's catch for non-abort failures. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 14 ++++++- .../client/test/client/streamableHttp.test.ts | 39 +++++++++++++++++++ 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index cafcac95f2..44efe3e60b 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -1030,7 +1030,19 @@ export class StreamableHTTPClientTransport implements Transport { replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, requestSignal: options?.requestSignal, onRequestStreamEnd: options?.onRequestStreamEnd - }).catch(error => this.onerror?.(error)); + }).catch(error => { + this.onerror?.(error as Error); + // The resume GET failed outright (network rejection, HTTP + // error, auth failure) before any stream existed, so none + // of the downstream settlement paths (405/null-body, + // reconnection exhaustion) can run — the per-request + // stream is terminally gone and the caller must settle. + // Never on an intentional abort, matching the + // `onRequestStreamEnd` contract. + if (this._abortController?.signal.aborted !== true && options?.requestSignal?.aborted !== true) { + options?.onRequestStreamEnd?.(); + } + }); return; } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index b04eb146e9..1981acd2a4 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2812,6 +2812,45 @@ describe('StreamableHTTPClientTransport', () => { const messages = errorSpy.mock.calls.map(args => (args[0] as Error).message); expect(messages.some(m => m.includes('SSE stream disconnected'))).toBe(false); }); + + it('settles a resumed send() when the initial resume GET fails outright', async () => { + // If the resume GET itself rejects (network failure, session + // expired, auth failure) no stream ever exists, so none of the + // downstream settlement paths can run — the caller's + // onRequestStreamEnd must still fire or the pending request hangs + // until its timeout. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const streamEndSpy = vi.fn(); + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockRejectedValue(new Error('connection refused')); + + const requestMessage: JSONRPCRequest = { + jsonrpc: '2.0', + method: 'long_running_tool', + id: 'request-1', + params: {} + }; + + await transport.start(); + await transport.send(requestMessage, { + resumptionToken: 'event-0', + onRequestStreamEnd: streamEndSpy + }); + await vi.advanceTimersByTimeAsync(100); + + expect(streamEndSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalled(); + }); }); describe('prevent infinite recursion when server returns 401 after successful auth', () => { From abcf52119a30ecae0acca270d8ad32b2dd538dc9 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 06:23:21 +0000 Subject: [PATCH 05/10] review: settle exhaustion via try/finally at depth, track all pending reconnections - _scheduleReconnection now guarantees the caller settles on every no-reconnection-pending exit: exhaustion wraps onerror in try/finally, and a throwing custom scheduler settles before rethrowing. The caller-side catches in _handleSseStream become containment-only. - Replace the single _cancelReconnection slot with a Set so close() cancels every parked timer across concurrent chains (standby GET plus resumed per-request streams), not just the latest. - Document maxReconnectionDelay's second role as the stream-lifetime progress threshold. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 69 ++++++++++++++----- .../client/test/client/streamableHttp.test.ts | 47 ++++++++++++- 2 files changed, 96 insertions(+), 20 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 44efe3e60b..86e3f5a50b 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -99,6 +99,9 @@ export interface StartSSEOptions { export interface StreamableHTTPReconnectionOptions { /** * Maximum backoff time between reconnection attempts in milliseconds. + * Also serves as the stream-lifetime threshold for retry accounting: a + * stream that stays open at least this long counts as progress and resets + * the {@linkcode maxRetries} attempt count (see {@linkcode maxRetries}). * Default is 30000 (30 seconds). */ maxReconnectionDelay: number; @@ -335,7 +338,7 @@ export class StreamableHTTPClientTransport implements Transport { private _maxStepUpRetries: number; private _serverRetryMs?: number; // Server-provided retry delay from SSE retry field private readonly _reconnectionScheduler?: ReconnectionScheduler; - private _cancelReconnection?: () => void; + private _pendingReconnections = new Set<() => void>(); onclose?: () => void; onerror?: (error: Error) => void; @@ -678,17 +681,25 @@ export class StreamableHTTPClientTransport implements Transport { // Check if we've exceeded maximum retry attempts if (attemptCount >= maxRetries) { - this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`)); - // The per-request stream is now definitively gone. - options.onRequestStreamEnd?.(); + try { + this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`)); + } finally { + // The per-request stream is now definitively gone. Settlement + // must survive a throwing user `onerror` handler on every + // route into this branch (graceful close, mid-stream error, + // failed reconnect chain) — callers only contain propagation, + // they never settle. + options.onRequestStreamEnd?.(); + } return; } // Calculate next delay based on current attempt count const delay = this._getNextReconnectionDelay(attemptCount); + let cancelEntry: () => void; const reconnect = (): void => { - this._cancelReconnection = undefined; + this._pendingReconnections.delete(cancelEntry); // Honour BOTH the transport-wide abort and the per-request abort // (a listen subscription closed during the backoff delay): do not // resurrect a stream the caller already tore down. @@ -710,13 +721,32 @@ export class StreamableHTTPClientTransport implements Transport { }); }; - if (this._reconnectionScheduler) { - const cancel = this._reconnectionScheduler(reconnect, delay, attemptCount); - this._cancelReconnection = typeof cancel === 'function' ? cancel : undefined; - } else { - const handle = setTimeout(reconnect, delay); - this._cancelReconnection = () => clearTimeout(handle); + try { + if (this._reconnectionScheduler) { + const cancel = this._reconnectionScheduler(reconnect, delay, attemptCount); + cancelEntry = + typeof cancel === 'function' + ? cancel + : () => { + // No-op: the custom scheduler provided no cancel + // function; tracked so `reconnect` can still + // deregister the chain's pending entry. + }; + } else { + const handle = setTimeout(reconnect, delay); + cancelEntry = () => clearTimeout(handle); + } + } catch (error) { + // A throwing custom scheduler means no reconnection is pending — + // the stream is definitively gone. Settle the caller here (the + // only route that can still do it), then rethrow for reporting. + options.onRequestStreamEnd?.(); + throw error; } + // Track every pending reconnection — concurrent chains (the standby + // GET stream plus resumed per-request streams) each park a timer + // here, and close() must cancel all of them, not just the latest. + this._pendingReconnections.add(cancelEntry); } /** @@ -864,12 +894,13 @@ export class StreamableHTTPClientTransport implements Transport { // `ReconnectionScheduler` — is reachable from a graceful // close, and a throwing user callback must not fall into // the outer catch (which would surface a misleading - // disconnect error and schedule a second time). + // disconnect error and schedule a second time). Containment + // only: `_scheduleReconnection` itself guarantees the + // caller settles on every no-reconnection-pending exit. try { scheduleNext(); } catch (error) { this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); - onRequestStreamEnd?.(); } } else if (!isIntentionalAbort()) { // The per-request stream ended without reconnecting (no @@ -893,12 +924,12 @@ export class StreamableHTTPClientTransport implements Transport { const needsReconnect = canResume && !receivedResponse; if (needsReconnect && this._abortController && !isIntentionalAbort()) { // Use the exponential backoff reconnection strategy. Same - // accounting as the graceful-close path. + // accounting as the graceful-close path; containment only, + // settlement is guaranteed inside `_scheduleReconnection`. try { scheduleNext(); } catch (error) { this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); - onRequestStreamEnd?.(); } } else { // Non-deliberate stream error without reconnection: the @@ -974,9 +1005,13 @@ export class StreamableHTTPClientTransport implements Transport { async close(): Promise { try { - this._cancelReconnection?.(); + // Cancel EVERY pending reconnection — concurrent chains (standby + // GET + resumed per-request streams) can each have a parked timer. + for (const cancel of this._pendingReconnections) { + cancel(); + } } finally { - this._cancelReconnection = undefined; + this._pendingReconnections.clear(); this._abortController?.abort(); this.onclose?.(); } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 1981acd2a4..0468bbeff5 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2497,7 +2497,7 @@ describe('StreamableHTTPClientTransport', () => { ); // Verify no reconnection was scheduled - expect(transport['_cancelReconnection']).toBeUndefined(); + expect(transport['_pendingReconnections'].size).toBe(0); }); it('should schedule reconnection when maxRetries is greater than 0', async () => { @@ -2519,10 +2519,13 @@ describe('StreamableHTTPClientTransport', () => { // ASSERT - should schedule a reconnection, not report error yet expect(errorSpy).not.toHaveBeenCalled(); - expect(transport['_cancelReconnection']).toBeDefined(); + expect(transport['_pendingReconnections'].size).toBe(1); // Clean up the pending reconnection to avoid test pollution - transport['_cancelReconnection']?.(); + for (const cancel of transport['_pendingReconnections']) { + cancel(); + } + transport['_pendingReconnections'].clear(); }); }); @@ -2851,6 +2854,44 @@ describe('StreamableHTTPClientTransport', () => { expect(streamEndSpy).toHaveBeenCalledTimes(1); expect(errorSpy).toHaveBeenCalled(); }); + + it('close() cancels every pending reconnection timer, not just the latest', async () => { + // Two concurrent reconnect chains (standby GET + a resumed + // per-request stream) each park a timer; close() must cancel both. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 5000, + maxReconnectionDelay: 30000, + reconnectionDelayGrowFactor: 1, + maxRetries: 3 + } + }); + const fetchMock = globalThis.fetch as Mock; + // Every stream closes empty immediately, so each chain schedules + // a reconnection 5s out. + fetchMock.mockImplementation(async () => sseResponse(['id: keep-1\ndata: \n\n'])); + + const requestMessage: JSONRPCRequest = { + jsonrpc: '2.0', + method: 'long_running_tool', + id: 'request-1', + params: {} + }; + + await transport.start(); + await transport['_startOrAuthSse']({}); + await transport.send(requestMessage, { resumptionToken: 'event-0' }); + await vi.advanceTimersByTimeAsync(10); + + expect(transport['_pendingReconnections'].size).toBe(2); + expect(vi.getTimerCount()).toBe(2); + + await transport.close(); + + // Both parked timers were cancelled, not just the latest. + expect(transport['_pendingReconnections'].size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + }); }); describe('prevent infinite recursion when server returns 401 after successful auth', () => { From b0d2ee73b2c396fed72c9d9ed708aa43cdc33051 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 06:39:39 +0000 Subject: [PATCH 06/10] review: guard close() cancels per entry, register reconnection entry before scheduler runs - close() wraps each user-supplied cancel in try/catch, routing throws to onerror so shutdown neither skips remaining cancels nor rejects (the existing throwing-cancel test now pins the contained behavior). - The pending-reconnection entry is registered before the scheduler is invoked, so a synchronously-firing custom scheduler deregisters it instead of leaving a stale entry until close(). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 41 +++++++++++-------- .../client/test/client/streamableHttp.test.ts | 10 ++++- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 86e3f5a50b..928820d4ff 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -697,7 +697,13 @@ export class StreamableHTTPClientTransport implements Transport { // Calculate next delay based on current attempt count const delay = this._getNextReconnectionDelay(attemptCount); - let cancelEntry: () => void; + // The chain's registry entry. Registered before the scheduler is + // invoked so a synchronously-firing custom scheduler still + // deregisters it (otherwise the entry would be added after the fact + // and linger until close()); the real cancel behavior is filled in + // once the scheduler returns. + let cancelImpl: (() => void) | undefined; + const cancelEntry = (): void => cancelImpl?.(); const reconnect = (): void => { this._pendingReconnections.delete(cancelEntry); // Honour BOTH the transport-wide abort and the per-request abort @@ -721,32 +727,26 @@ export class StreamableHTTPClientTransport implements Transport { }); }; + // Track every pending reconnection — concurrent chains (the standby + // GET stream plus resumed per-request streams) each park a timer + // here, and close() must cancel all of them, not just the latest. + this._pendingReconnections.add(cancelEntry); try { if (this._reconnectionScheduler) { const cancel = this._reconnectionScheduler(reconnect, delay, attemptCount); - cancelEntry = - typeof cancel === 'function' - ? cancel - : () => { - // No-op: the custom scheduler provided no cancel - // function; tracked so `reconnect` can still - // deregister the chain's pending entry. - }; + cancelImpl = typeof cancel === 'function' ? cancel : undefined; } else { const handle = setTimeout(reconnect, delay); - cancelEntry = () => clearTimeout(handle); + cancelImpl = () => clearTimeout(handle); } } catch (error) { // A throwing custom scheduler means no reconnection is pending — - // the stream is definitively gone. Settle the caller here (the - // only route that can still do it), then rethrow for reporting. + // deregister the entry and settle the caller here (the only route + // that can still do it), then rethrow for reporting. + this._pendingReconnections.delete(cancelEntry); options.onRequestStreamEnd?.(); throw error; } - // Track every pending reconnection — concurrent chains (the standby - // GET stream plus resumed per-request streams) each park a timer - // here, and close() must cancel all of them, not just the latest. - this._pendingReconnections.add(cancelEntry); } /** @@ -1007,8 +1007,15 @@ export class StreamableHTTPClientTransport implements Transport { try { // Cancel EVERY pending reconnection — concurrent chains (standby // GET + resumed per-request streams) can each have a parked timer. + // Per-entry guard: a throwing user-supplied cancel (from a custom + // ReconnectionScheduler) must not skip the remaining cancels or + // escape the shutdown path. for (const cancel of this._pendingReconnections) { - cancel(); + try { + cancel(); + } catch (error) { + this.onerror?.(error instanceof Error ? error : new Error(String(error))); + } } } finally { this._pendingReconnections.clear(); diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 0468bbeff5..086412d28a 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -3085,7 +3085,10 @@ describe('StreamableHTTPClientTransport', () => { expect(onerror).not.toHaveBeenCalled(); }); - it('still aborts and fires onclose if the cancel function throws', async () => { + it('contains a throwing cancel function: close() completes, aborts, and fires onclose', async () => { + // A user-supplied cancel that throws must not escape the shutdown + // path or skip the remaining pending cancels — it is surfaced via + // onerror instead. transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { reconnectionOptions, reconnectionScheduler: () => () => { @@ -3094,12 +3097,15 @@ describe('StreamableHTTPClientTransport', () => { }); const onclose = vi.fn(); transport.onclose = onclose; + const onerror = vi.fn(); + transport.onerror = onerror; await transport.start(); triggerReconnection(transport); const abortController = transport['_abortController']; - await expect(transport.close()).rejects.toThrow('cancel failed'); + await expect(transport.close()).resolves.toBeUndefined(); + expect(onerror).toHaveBeenCalledWith(expect.objectContaining({ message: 'cancel failed' })); expect(abortController?.signal.aborted).toBe(true); expect(onclose).toHaveBeenCalledTimes(1); }); From 6b4029d970e0541984437b2adab3fc3953aa3ba7 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 06:53:01 +0000 Subject: [PATCH 07/10] review: measure stream lifetime with monotonic performance.now() Date.now() can step backwards (NTP correction, VM pause), which could misclassify a healthy long-lived stream as fruitless. performance.now() is monotonic, already used for durations in middleware.ts, and available in all target runtimes. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 928820d4ff..8d593e81a0 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -792,7 +792,7 @@ export class StreamableHTTPClientTransport implements Transport { // keeping repeated connect/idle-close cycles bounded by `maxRetries` // (#2682). let receivedMessage = false; - const streamOpenedAt = Date.now(); + const streamOpenedAt = performance.now(); // Single scheduling site for both the graceful-close and the // mid-stream-error paths below — the #2682 bug existed precisely @@ -810,7 +810,7 @@ export class StreamableHTTPClientTransport implements Transport { // looping forever (#2682). Priming events alone deliberately do // not reset the count — a server can send one and still close // immediately, which would re-arm exactly that loop. - const madeProgress = receivedMessage || Date.now() - streamOpenedAt >= this._reconnectionOptions.maxReconnectionDelay; + const madeProgress = receivedMessage || performance.now() - streamOpenedAt >= this._reconnectionOptions.maxReconnectionDelay; this._scheduleReconnection( { // A reconnected stream that ended before any event arrived From 2c27bd5e7072c25d0276d7201d055667d7dc6338 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 07:09:29 +0000 Subject: [PATCH 08/10] review: fix stale reconnection JSDoc, stop double-reporting resume-GET failures - _scheduleReconnection and _getNextReconnectionDelay doc blocks now describe the persistent per-chain attempt accounting instead of a nonexistent lastEventId param and 'for this specific stream'. - The resume-GET catch (and the 202-initialized standby-start sibling) no longer call onerror: _startOrAuthSse already reports every non-intentional rejection and deliberately stays silent on intentional aborts, so the outer report double-fired on genuine failures and surfaced a spurious AbortError on deliberate aborts. Settlement in the resume catch keeps its abort guard. - Test comment: Date.now -> performance.now. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 29 ++++++++++++++----- .../client/test/client/streamableHttp.test.ts | 2 +- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 8d593e81a0..d8f01b4249 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -651,7 +651,9 @@ export class StreamableHTTPClientTransport implements Transport { /** * Calculates the next reconnection delay using a backoff algorithm * - * @param attempt Current reconnection attempt count for the specific stream + * @param attempt Zero-indexed reconnection attempt number the delay is for + * (the attempt count persists across fruitless connect-then-close cycles — + * see `_scheduleReconnection`) * @returns Time to wait in milliseconds before next reconnection attempt */ private _getNextReconnectionDelay(attempt: number): number { @@ -672,8 +674,13 @@ export class StreamableHTTPClientTransport implements Transport { /** * Schedule a reconnection attempt using server-provided retry interval or backoff * - * @param lastEventId The ID of the last received event for resumability - * @param attemptCount Current reconnection attempt count for this specific stream + * @param options Stream options carried into the reconnected stream + * (resumption token, stream callbacks, per-request abort signal) + * @param attemptCount Zero-indexed number of this reconnection attempt + * within the current chain of fruitless attempts. It persists across + * connect-then-close cycles that make no progress (bounding them by + * `maxRetries`) and resets only when a stream delivers a message or stays + * open at least `maxReconnectionDelay` — see `_handleSseStream`. */ private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0): void { // Use provided options or default options @@ -1072,8 +1079,12 @@ export class StreamableHTTPClientTransport implements Transport { replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, requestSignal: options?.requestSignal, onRequestStreamEnd: options?.onRequestStreamEnd - }).catch(error => { - this.onerror?.(error as Error); + }).catch(() => { + // No onerror here: `_startOrAuthSse` already surfaced the + // failure for every non-intentional rejection and + // deliberately stays silent on intentional aborts — an + // unconditional report would double-fire on genuine + // failures and surface a spurious AbortError on aborts. // The resume GET failed outright (network rejection, HTTP // error, auth failure) before any stream existed, so none // of the downstream settlement paths (405/null-body, @@ -1237,8 +1248,12 @@ export class StreamableHTTPClientTransport implements Transport { // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isInitializedNotification(message)) { - // Start without a lastEventId since this is a fresh connection - this._startOrAuthSse({ resumptionToken: undefined }).catch(error => this.onerror?.(error)); + // Start without a lastEventId since this is a fresh connection. + // Rejections are already surfaced by `_startOrAuthSse` for + // every non-intentional failure (and deliberately suppressed + // on intentional aborts), so reporting here again would + // double-fire onerror or surface a spurious AbortError. + this._startOrAuthSse({ resumptionToken: undefined }).catch(() => {}); } return; } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 086412d28a..c9719b1b57 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2689,7 +2689,7 @@ describe('StreamableHTTPClientTransport', () => { await transport['_startOrAuthSse']({}); // Five cycles: each stream lives longer than maxReconnectionDelay - // (fake timers also drive Date.now), then idle-closes empty. + // (fake timers also drive performance.now), then idle-closes empty. for (let cycle = 0; cycle < 5; cycle++) { await vi.advanceTimersByTimeAsync(1500); controllers[cycle]!.close(); From c2f8d90353185a12035913dfffe976cc81ed9eea Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 07:26:48 +0000 Subject: [PATCH 09/10] review: enforce at-most-once per-request settlement at the send() entry point Two exactly-once violations: a user onRequestStreamEnd throwing at the 405/null-body settle sites (inside _startOrAuthSse's try) rejects the resume promise, whose catch settled a second time; and a custom scheduler that synchronously invokes reconnect() then throws could double-settle via the scheduler-throw catch. Wrap the callback once in send() (auth-retry recursion reuses the same wrapper), so every current and future settlement site can call it unconditionally; the fired flag is set before invoking so a throwing callback still counts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 37 +++++++++++++- .../client/test/client/streamableHttp.test.ts | 48 +++++++++++++++++++ 2 files changed, 84 insertions(+), 1 deletion(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index d8f01b4249..b79bcf6857 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -315,6 +315,34 @@ function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal { return controller.signal; } +/** + * Wrap a per-request stream-end callback so it fires at most once. + * + * The transport has several terminal settlement sites for one request — the + * 405 and null-body outcomes, reconnection exhaustion, a throwing custom + * scheduler, and the resumed-send failure path — and more than one of them + * can be reached for the same request (for example, a user callback that + * throws at the 405 site rejects the resume promise, whose catch settles + * again). Enforcing the exactly-once contract here, at the entry point, + * hardens every current and future settlement site at one stroke instead of + * requiring each site to infer whether another already ran. The flag is set + * BEFORE the callback is invoked so a throwing callback still counts as + * fired. + */ +function atMostOnce(callback?: () => void): (() => void) | undefined { + if (!callback) { + return undefined; + } + let fired = false; + return () => { + if (fired) { + return; + } + fired = true; + callback(); + }; +} + /** * Client transport for Streamable HTTP: this implements the MCP Streamable HTTP transport specification. * It will connect to a server using HTTP `POST` for sending messages and HTTP `GET` with Server-Sent Events @@ -1041,7 +1069,14 @@ export class StreamableHTTPClientTransport implements Transport { headers?: Readonly>; } ): Promise { - return this._send(message, options, false); + // Enforce the per-request settlement contract at the entry point: + // `onRequestStreamEnd` is wrapped to fire at most once no matter + // which terminal path (405/null-body, reconnection exhaustion, + // scheduler failure, resume failure) reaches it first — every + // downstream site can then call it unconditionally. Wrapped here + // rather than in `_send` so auth-retry recursion reuses the same + // wrapper instance. + return this._send(message, options && { ...options, onRequestStreamEnd: atMostOnce(options.onRequestStreamEnd) }, false); } private async _send( diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index c9719b1b57..8d35aac203 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2855,6 +2855,54 @@ describe('StreamableHTTPClientTransport', () => { expect(errorSpy).toHaveBeenCalled(); }); + it('settles at most once when the user onRequestStreamEnd throws at a terminal 405 resume', async () => { + // The 405 settle site runs inside _startOrAuthSse's try: a + // throwing user callback propagates to its catch, which reports + // via onerror and rejects the resume promise — whose own catch + // then reaches its settlement call. The at-most-once wrapper + // applied in send() keeps the exactly-once contract. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const streamEndSpy = vi.fn().mockImplementation(() => { + throw new Error('user callback exploded'); + }); + + const fetchMock = globalThis.fetch as Mock; + // The resume GET gets a 405: terminal non-resumable outcome. + fetchMock.mockResolvedValue({ + ok: false, + status: 405, + statusText: 'Method Not Allowed', + headers: new Headers(), + text: async () => '' + }); + + const requestMessage: JSONRPCRequest = { + jsonrpc: '2.0', + method: 'long_running_tool', + id: 'request-1', + params: {} + }; + + await transport.start(); + await transport.send(requestMessage, { + resumptionToken: 'event-0', + onRequestStreamEnd: streamEndSpy + }); + await vi.advanceTimersByTimeAsync(50); + + // Exactly one settlement attempt reached the user callback. + expect(streamEndSpy).toHaveBeenCalledTimes(1); + }); + it('close() cancels every pending reconnection timer, not just the latest', async () => { // Two concurrent reconnect chains (standby GET + a resumed // per-request stream) each park a timer; close() must cancel both. From f4a5e03afc68c00a95e13f1eb188e7583d3fee55 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 07:42:12 +0000 Subject: [PATCH 10/10] review: settle from the scheduler-throw catch only when nothing was dispatched MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A scheduler that synchronously invokes reconnect() before throwing leaves an attempt in flight that owns settlement; settling from the catch then prematurely ended the request (and the at-most-once guard would swallow the chain's later legitimate settlement). reconnect()'s first action deletes the registry entry, so the catch now settles only when its own delete succeeds — proof that no attempt was dispatched. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL --- packages/client/src/client/streamableHttp.ts | 18 +++++--- .../client/test/client/streamableHttp.test.ts | 45 +++++++++++++++++++ 2 files changed, 58 insertions(+), 5 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index b79bcf6857..9ec3b32a50 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -775,11 +775,19 @@ export class StreamableHTTPClientTransport implements Transport { cancelImpl = () => clearTimeout(handle); } } catch (error) { - // A throwing custom scheduler means no reconnection is pending — - // deregister the entry and settle the caller here (the only route - // that can still do it), then rethrow for reporting. - this._pendingReconnections.delete(cancelEntry); - options.onRequestStreamEnd?.(); + // A throwing custom scheduler settles the caller ONLY when it + // threw without dispatching. `reconnect()`'s first action is + // deleting this entry, so a successful delete here proves no + // attempt was dispatched — no reconnection is pending, the stream + // is definitively gone, and this is the only route that can still + // settle. If the scheduler synchronously invoked `reconnect()` + // before throwing, the in-flight attempt owns settlement (it + // reaches a terminal settlement site or ends by intentional + // abort) — settling here would prematurely end the request and + // swallow the chain's later legitimate settlement. + if (this._pendingReconnections.delete(cancelEntry)) { + options.onRequestStreamEnd?.(); + } throw error; } } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 8d35aac203..9282490eff 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2903,6 +2903,51 @@ describe('StreamableHTTPClientTransport', () => { expect(streamEndSpy).toHaveBeenCalledTimes(1); }); + it('does not settle from the scheduler-throw catch when the scheduler dispatched before throwing', async () => { + // A pathological scheduler that synchronously invokes reconnect() + // and THEN throws leaves an attempt in flight. The scheduler-throw + // catch must not settle the caller then — reconnect()'s first + // action deletes the registry entry, so the catch settles only + // when its own delete succeeds (i.e. nothing was dispatched). + // Settlement instead arrives later from the in-flight chain when + // it exhausts. + const order: string[] = []; + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1, + maxRetries: 1 + }, + reconnectionScheduler: reconnect => { + reconnect(); + throw new Error('scheduler exploded after dispatch'); + } + }); + transport.onerror = e => order.push(`error:${(e as Error).message}`); + const streamEndSpy = vi.fn().mockImplementation(() => order.push('settled')); + + const fetchMock = globalThis.fetch as Mock; + // Every GET opens fine and closes empty: the original stream + // schedules attempt 0 (sync-dispatched by the scheduler, which + // then throws); the dispatched stream closes empty and trips + // exhaustion at attempt 1 — the chain's own settlement. + fetchMock.mockImplementation(async () => sseResponse([])); + + await transport.start(); + await transport['_startOrAuthSse']({ onRequestStreamEnd: streamEndSpy }); + await vi.advanceTimersByTimeAsync(100); + + // The dispatched attempt ran (two GETs), and the caller settled + // exactly once — from exhaustion, strictly AFTER the max-retries + // report, not from the scheduler-throw catch. + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(streamEndSpy).toHaveBeenCalledTimes(1); + const exhaustedAt = order.indexOf('error:Maximum reconnection attempts (1) exceeded.'); + expect(exhaustedAt).toBeGreaterThanOrEqual(0); + expect(order.indexOf('settled')).toBeGreaterThan(exhaustedAt); + }); + it('close() cancels every pending reconnection timer, not just the latest', async () => { // Two concurrent reconnect chains (standby GET + a resumed // per-request stream) each park a timer; close() must cancel both.