From c37f361c2bedf49f93aaf4c12756b605a66cd193 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:31:19 +0000 Subject: [PATCH 1/8] fix(client): make streamable HTTP auth awaits abortable by requestSignal token(), onUnauthorized 401 recovery, and step-up authorization were awaited with no path for the per-request or transport abort signal to reach them, so a hung auth flow parked send() forever. Race those awaits against the combined signal and offer it to onUnauthorized via the new optional UnauthorizedContext.signal field. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- .changeset/abortable-auth-awaits.md | 5 + packages/client/src/client/auth.ts | 8 + packages/client/src/client/streamableHttp.ts | 171 ++++++++++++++---- .../client/test/client/streamableHttp.test.ts | 100 ++++++++++ 4 files changed, 249 insertions(+), 35 deletions(-) create mode 100644 .changeset/abortable-auth-awaits.md diff --git a/.changeset/abortable-auth-awaits.md b/.changeset/abortable-auth-awaits.md new file mode 100644 index 0000000000..2bc7af5e60 --- /dev/null +++ b/.changeset/abortable-auth-awaits.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). diff --git a/packages/client/src/client/auth.ts b/packages/client/src/client/auth.ts index 9ebc6fd251..941b162db7 100644 --- a/packages/client/src/client/auth.ts +++ b/packages/client/src/client/auth.ts @@ -55,6 +55,14 @@ export interface UnauthorizedContext { serverUrl: URL; /** Fetch function configured with the transport's `requestInit`, for making auth requests. */ fetchFn: FetchLike; + /** + * Abort signal for the request (or transport) whose 401 triggered this + * recovery. The transport stops waiting for `onUnauthorized` when it + * aborts; cooperative implementations should pass it to their own fetches + * so the recovery work stops too. Optional — absent when the transport + * has no lifetime signal to offer. + */ + signal?: AbortSignal; } /** diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..9fe72eff76 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -303,6 +303,55 @@ function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal { return controller.signal; } +/** + * Normalize an aborted signal's `reason` to an `Error` (matching how the + * listen driver and `fetch` surface aborts) so a raced auth await rejects + * with something callers can inspect. + */ +function abortReasonError(signal: AbortSignal): Error { + const reason: unknown = signal.reason; + return reason instanceof Error ? reason : new Error(String(reason ?? 'Aborted')); +} + +/** + * Race a pending auth-chain await (`AuthProvider.token()`, `onUnauthorized`, + * step-up authorization) against the request/transport abort signal so an + * abort settles `send()` even when the underlying promise never does. The + * loser keeps running — the `AuthProvider` contract has no cancellation + * channel of its own beyond the optional `ctx.signal` — but the transport + * stops waiting on it, which is what `TransportSendOptions.requestSignal` + * promises. The abort listener is removed as soon as the promise settles so + * a long-lived transport signal does not accumulate closures per request. + */ +function raceWithSignal(promise: Promise, signal: AbortSignal | undefined): Promise { + if (signal === undefined) { + return promise; + } + if (signal.aborted) { + // Attach a no-op handler so the loser's eventual rejection (if any) + // does not escape as an unhandledRejection. + promise.catch(() => {}); + return Promise.reject(abortReasonError(signal)); + } + return new Promise((resolve, reject) => { + const onAbort = (): void => { + promise.catch(() => {}); + reject(abortReasonError(signal)); + }; + signal.addEventListener('abort', onAbort, { once: true }); + promise.then( + value => { + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error: unknown) => { + signal.removeEventListener('abort', onAbort); + reject(error instanceof Error ? error : new Error(String(error))); + } + ); + }); +} + /** * 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 @@ -432,12 +481,21 @@ export class StreamableHTTPClientTransport implements Transport { }); } - private async _commonHeaders(): Promise { + private async _commonHeaders(signal?: AbortSignal): Promise { const headers: RequestInit['headers'] & Record = {}; let token: string | undefined; try { - token = await this._authProvider?.token(); + // Raced against the per-request/transport abort so a hung + // `token()` (wedged refresh, slow broker) cannot park the send + // past its `requestSignal` (#2643). + token = this._authProvider === undefined ? undefined : await raceWithSignal(this._authProvider.token(), signal); } catch (error) { + // An abort is an intentional teardown, never an auth failure — + // leave it unstamped so the send-catch and the negotiation + // probe's classifier treat it as a plain abort. + if (signal?.aborted === true) { + throw error; + } // Auth-seam stamp: a throwing token() is an auth failure, never a // network failure. throw markAuthSeamEscape(error); @@ -527,9 +585,18 @@ export class StreamableHTTPClientTransport implements Transport { const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; try { + // Combined BEFORE header acquisition so the auth chain (token(), + // 401 recovery, step-up) is raced against the same abort as the + // GET itself. + const transportSignal = this._abortController?.signal; + const signal = + requestSignal !== undefined && transportSignal !== undefined + ? anySignal(transportSignal, requestSignal) + : (requestSignal ?? transportSignal); + // Try to open an initial SSE stream with GET to listen for server messages // This is optional according to the spec - server may not support it - const headers = await this._commonHeaders(); + const headers = await this._commonHeaders(signal); const userAccept = headers.get('accept'); const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'text/event-stream']; headers.set('accept', [...new Set(types)].join(', ')); @@ -539,11 +606,6 @@ export class StreamableHTTPClientTransport implements Transport { headers.set('last-event-id', resumptionToken); } - const transportSignal = this._abortController?.signal; - const signal = - requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, requestSignal) - : (requestSignal ?? transportSignal); const response = await (this._fetch ?? fetch)(this._url, { ...this._requestInit, method: 'GET', @@ -563,12 +625,25 @@ export class StreamableHTTPClientTransport implements Transport { if (this._authProvider.onUnauthorized && !isAuthRetry) { try { - await this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit - }); + // Raced against the per-request/transport abort so + // a hung 401 recovery cannot park the GET (#2643); + // the signal is also handed to the provider so a + // cooperative implementation can cancel its own work. + await raceWithSignal( + this._authProvider.onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), + signal + ); } catch (error) { + // An abort is an intentional teardown, not an auth + // failure — leave it unstamped. + if (signal?.aborted === true) { + throw error; + } // Auth-seam stamp: covers the SDK's OAuth flow and // custom onUnauthorized callbacks alike. throw markAuthSeamEscape(error); @@ -593,9 +668,14 @@ export class StreamableHTTPClientTransport implements Transport { const { resourceMetadataUrl, scope, error, errorDescription } = extractWWWAuthenticateParams(response); if (error === 'insufficient_scope') { const text = await response.text?.().catch(() => null); - const result = await this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries + // Raced against the per-request/transport abort so a + // hung step-up authorization cannot park the GET (#2643). + const result = await raceWithSignal( + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), + signal ); if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); @@ -962,7 +1042,18 @@ export class StreamableHTTPClientTransport implements Transport { return; } - const headers = await this._commonHeaders(); + // Per-request abort: when the caller supplies a request-scoped + // signal (the `subscriptions/listen` driver), aborting it cancels + // this POST and its SSE response stream without closing the + // transport. Combined BEFORE header acquisition so the auth chain + // (token(), 401 recovery, step-up) is raced against it too. + const transportSignal = this._abortController?.signal; + const signal = + options?.requestSignal !== undefined && transportSignal !== undefined + ? anySignal(transportSignal, options.requestSignal) + : (options?.requestSignal ?? transportSignal); + + const headers = await this._commonHeaders(signal); this._applyBodyDerivedHeaders(headers, message); // A new session starts "without a session ID attached" (2025-11-25 transports §Session Management). const isHandshake = Array.isArray(message) ? message.some(m => isInitializeRequest(m)) : isInitializeRequest(message); @@ -987,15 +1078,6 @@ export class StreamableHTTPClientTransport implements Transport { const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'application/json', 'text/event-stream']; headers.set('accept', [...new Set(types)].join(', ')); - // Per-request abort: when the caller supplies a request-scoped - // signal (the `subscriptions/listen` driver), aborting it cancels - // this POST and its SSE response stream without closing the - // transport. - const transportSignal = this._abortController?.signal; - const signal = - options?.requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, options.requestSignal) - : (options?.requestSignal ?? transportSignal); const init = { ...this._requestInit, method: 'POST', @@ -1025,12 +1107,25 @@ export class StreamableHTTPClientTransport implements Transport { if (this._authProvider.onUnauthorized && !isAuthRetry) { try { - await this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit - }); + // Raced against the per-request/transport abort so + // a hung 401 recovery cannot park the send (#2643); + // the signal is also handed to the provider so a + // cooperative implementation can cancel its own work. + await raceWithSignal( + this._authProvider.onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), + signal + ); } catch (error) { + // An abort is an intentional teardown, not an auth + // failure — leave it unstamped. + if (signal?.aborted === true) { + throw error; + } // Auth-seam stamp: covers the SDK's OAuth flow and // custom onUnauthorized callbacks alike. throw markAuthSeamEscape(error); @@ -1057,9 +1152,15 @@ export class StreamableHTTPClientTransport implements Transport { const { resourceMetadataUrl, scope, error, errorDescription } = extractWWWAuthenticateParams(response); if (error === 'insufficient_scope') { - const result = await this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries + // Raced against the per-request/transport abort so a + // hung step-up authorization (metadata discovery, token + // exchange) cannot park the send (#2643). + const result = await raceWithSignal( + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), + signal ); if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); @@ -1195,7 +1296,7 @@ export class StreamableHTTPClientTransport implements Transport { } try { - const headers = await this._commonHeaders(); + const headers = await this._commonHeaders(this._abortController?.signal); const init = { ...this._requestInit, diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..7ea2a8407c 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2736,4 +2736,104 @@ describe('StreamableHTTPClientTransport', () => { expect(onclose).toHaveBeenCalledTimes(1); }); }); + + describe('abortable auth awaits (requestSignal reaches the auth chain)', () => { + const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)); + const request: JSONRPCMessage = { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 1 }; + + const settledOrHung = async (p: Promise): Promise => { + const settled = p.then( + () => 'resolved', + e => e + ); + return Promise.race([settled, sleep(500).then(() => 'send() hung')]); + }; + + it('requestSignal abort settles a send parked in a hung token()', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + + const ac = new AbortController(); + const pending = hungTransport.send(request, { requestSignal: ac.signal }); + setTimeout(() => ac.abort(new Error('caller abort')), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect((outcome as Error).message).toContain('caller abort'); + // A per-request abort is intentional — no misleading onerror. + expect(onerror).not.toHaveBeenCalled(); + // The token() fetch never ran. + expect(globalThis.fetch).not.toHaveBeenCalled(); + await hungTransport.close(); + }); + + it('transport close() settles a send parked in a hung token()', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + await hungTransport.start(); + + const pending = hungTransport.send(request); + setTimeout(() => void hungTransport.close(), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect(outcome).not.toBe('send() hung'); + }); + + it('requestSignal abort settles a send parked in a hung onUnauthorized(); the signal is offered to the provider', async () => { + let seenSignal: AbortSignal | undefined; + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { + token: async () => 'stale-token', + onUnauthorized: ctx => { + seenSignal = ctx.signal; + return new Promise(() => {}); + } + } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 401, + statusText: 'Unauthorized', + headers: new Headers(), + text: async () => '' + }); + + const ac = new AbortController(); + const pending = hungTransport.send(request, { requestSignal: ac.signal }); + setTimeout(() => ac.abort(new Error('caller abort')), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect((outcome as Error).message).toContain('caller abort'); + expect(onerror).not.toHaveBeenCalled(); + // The provider was handed the combined signal so cooperative + // implementations can cancel their own recovery work. + expect(seenSignal).toBeDefined(); + expect(seenSignal?.aborted).toBe(true); + await hungTransport.close(); + }); + + it('a token() rejection (without an abort) is still surfaced as an auth failure, not swallowed by the race', async () => { + const failingTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { + token: async () => { + throw new UnauthorizedError('no credentials'); + } + } + }); + await failingTransport.start(); + await expect(failingTransport.send(request)).rejects.toThrow('no credentials'); + await failingTransport.close(); + }); + }); }); From bdacc42920f2c18085c38f23ff6fc96bf60b0421 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:47:50 +0000 Subject: [PATCH 2/8] fix(client): address automated review on abortable auth awaits - suppress onerror for transport-lifetime aborts too in _send's catch and guard the resumptionToken fire-and-forget like reconnect() does - reject already-aborted sends before invoking authProvider.token() - pass the raced await's rejection through verbatim so the auth-seam stamp (a property on the thrown value, Error or not) and cause chains survive; lock in with an identity-preservation test Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- packages/client/src/client/streamableHttp.ts | 44 ++++++++++++++++--- .../client/test/client/streamableHttp.test.ts | 25 +++++++++++ 2 files changed, 62 insertions(+), 7 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 9fe72eff76..48bae4aaa5 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -303,6 +303,15 @@ function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal { return controller.signal; } +/** + * `signal?.aborted === true` behind a function boundary: `aborted` flips + * asynchronously (across awaits), so an inline check after an earlier guard + * gets unsoundly narrowed to `false` by the type checker. + */ +function isAborted(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true; +} + /** * Normalize an aborted signal's `reason` to an `Error` (matching how the * listen driver and `fetch` surface aborts) so a raced auth await rejects @@ -346,7 +355,12 @@ function raceWithSignal(promise: Promise, signal: AbortSignal | undefined) }, (error: unknown) => { signal.removeEventListener('abort', onAbort); - reject(error instanceof Error ? error : new Error(String(error))); + // Verbatim: the loser's rejection must flow through + // identity-preserved — the auth-seam stamp is a property on + // the thrown object (Error or not), and rewrapping would strip + // it (and any `.cause` chain) on paths with no re-stamping + // catch, e.g. the step-up awaits. + reject(error); } ); }); @@ -482,6 +496,13 @@ export class StreamableHTTPClientTransport implements Transport { } private async _commonHeaders(signal?: AbortSignal): Promise { + // Already-aborted fast path BEFORE the provider is invoked: an + // aborted request must not start new auth work just to discard it + // (raceWithSignal's own fast-path runs after `token()` has been + // called, which is too late). + if (signal !== undefined && isAborted(signal)) { + throw abortReasonError(signal); + } const headers: RequestInit['headers'] & Record = {}; let token: string | undefined; try { @@ -493,7 +514,7 @@ export class StreamableHTTPClientTransport implements Transport { // An abort is an intentional teardown, never an auth failure — // leave it unstamped so the send-catch and the negotiation // probe's classifier treat it as a plain abort. - if (signal?.aborted === true) { + if (isAborted(signal)) { throw error; } // Auth-seam stamp: a throwing token() is an auth failure, never a @@ -1038,7 +1059,15 @@ export class StreamableHTTPClientTransport implements Transport { resumptionToken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, requestSignal: options?.requestSignal - }).catch(error => this.onerror?.(error)); + }).catch(error => { + // Same guard as `_scheduleReconnection`'s reconnect(): an + // abort of either signal during the resume (now reachable + // mid-auth-chain too) is intentional teardown, not an error. + if (this._abortController?.signal.aborted === true || options?.requestSignal?.aborted === true) { + return; + } + this.onerror?.(error); + }); return; } @@ -1262,13 +1291,14 @@ export class StreamableHTTPClientTransport implements Transport { await response.text?.().catch(() => {}); } } catch (error) { - // Intentional per-request abort BEFORE response headers (the - // `subscriptions/listen` driver aborting its `requestSignal`): - // fetch rejects with AbortError. Same guard as + // Intentional abort BEFORE response headers — the + // `subscriptions/listen` driver aborting its `requestSignal`, or + // `close()` aborting the transport signal while the send is in + // the auth chain or the fetch. Same guard as // `_handleSseStream`'s `isIntentionalAbort` — do not surface a // misleading onerror; still rethrow so `listen()`'s send-catch // settles the per-subscription state machine. - if (options?.requestSignal?.aborted !== true) { + if (options?.requestSignal?.aborted !== true && this._abortController?.signal.aborted !== true) { this.onerror?.(error as Error); } throw error; diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 7ea2a8407c..5f15decc1e 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2775,6 +2775,8 @@ describe('StreamableHTTPClientTransport', () => { const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { authProvider: { token: () => new Promise(() => {}) } }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; await hungTransport.start(); const pending = hungTransport.send(request); @@ -2783,6 +2785,29 @@ describe('StreamableHTTPClientTransport', () => { const outcome = await settledOrHung(pending); expect(outcome).toBeInstanceOf(Error); expect(outcome).not.toBe('send() hung'); + // A transport-lifetime abort is intentional teardown — same + // isIntentionalAbort discipline as the per-request abort. + expect(onerror).not.toHaveBeenCalled(); + }); + + it('a non-Error rejection from the raced auth await passes through identity-preserved (auth-seam stamp intact)', async () => { + // The auth-seam stamp is a property set on the thrown VALUE + // (Error or not); a rewrap inside the race would strip it on + // paths with no re-stamping catch (the step-up awaits). + const sentinel = { code: 'custom-auth-failure' }; + const failingTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => Promise.reject(sentinel) } + }); + await failingTransport.start(); + // The transport-lifetime signal engages the race; the rejection + // must surface as the very same object (markAuthSeamEscape stamps + // in place, preserving identity). + const outcome = await failingTransport.send(request).then( + () => 'resolved (unexpected)', + (e: unknown) => e + ); + expect(outcome).toBe(sentinel); + await failingTransport.close(); }); it('requestSignal abort settles a send parked in a hung onUnauthorized(); the signal is offered to the provider', async () => { From 2d4fbe7d21442b3d979ee86384527aa3f490b185 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:59:31 +0000 Subject: [PATCH 3/8] fix(client): finish the no-spurious-onerror discipline and harden raceWithSignal - guard terminateSession's catch and the 202/initialized standalone-GET fire-and-forget against transport-lifetime aborts, matching _send - normalize raceWithSignal's input via Promise.resolve so a plain-JS provider returning a bare value keeps working (the pre-race await tolerated it) and an executor throw cannot leak the abort listener - fold the pre-existing StartSSEOptions drop on the resume-via-send path: thread onresumptiontoken and onRequestStreamEnd like the fresh-POST path so resumed streams keep the token-persistence chain and report their terminal end Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- packages/client/src/client/streamableHttp.ts | 37 +++++++++++++++-- .../client/test/client/streamableHttp.test.ts | 40 +++++++++++++++++++ 2 files changed, 73 insertions(+), 4 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 48bae4aaa5..b35a755f4d 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -332,7 +332,14 @@ function abortReasonError(signal: AbortSignal): Error { * promises. The abort listener is removed as soon as the promise settles so * a long-lived transport signal does not accumulate closures per request. */ -function raceWithSignal(promise: Promise, signal: AbortSignal | undefined): Promise { +function raceWithSignal(value: Promise, signal: AbortSignal | undefined): Promise { + // Normalize first: a plain-JS provider may return a bare value (or a + // foreign thenable) where the types say Promise — the plain `await` these + // call sites used before the race tolerated that, so the race must too + // (`.then` on a bare string would TypeError and get misstamped as an + // auth failure). Identity-preserving for native promises, and it keeps + // the executor below throw-free so the abort listener cannot leak. + const promise = Promise.resolve(value); if (signal === undefined) { return promise; } @@ -1058,7 +1065,14 @@ export class StreamableHTTPClientTransport implements Transport { this._startOrAuthSse({ resumptionToken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, - requestSignal: options?.requestSignal + requestSignal: options?.requestSignal, + // Keep the caller's stream observers across the resume, + // matching the fresh-POST path below: without them a + // resume-via-send() dropped later resumption tokens (the + // persistence chain) and never reported the resumed + // stream's terminal end. + onresumptiontoken, + onRequestStreamEnd: options?.onRequestStreamEnd }).catch(error => { // Same guard as `_scheduleReconnection`'s reconnect(): an // abort of either signal during the resume (now reachable @@ -1242,7 +1256,16 @@ export class StreamableHTTPClientTransport implements Transport { // 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)); + this._startOrAuthSse({ resumptionToken: undefined }).catch(error => { + // A transport-lifetime abort during the GET (now + // reachable mid-auth-chain too) is intentional + // teardown, not an error. No per-request signal on + // this standalone-GET path. + if (this._abortController?.signal.aborted === true) { + return; + } + this.onerror?.(error); + }); } return; } @@ -1353,7 +1376,13 @@ export class StreamableHTTPClientTransport implements Transport { this._sessionId = undefined; } catch (error) { - this.onerror?.(error as Error); + // A transport-lifetime abort (close() during a DELETE parked in + // the auth chain or the fetch) is intentional teardown — same + // discipline as `_send`'s catch. No per-request signal on this + // path. Still rethrow so the caller observes the failure. + if (this._abortController?.signal.aborted !== true) { + this.onerror?.(error as Error); + } throw error; } } diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 5f15decc1e..25a8b48ff1 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2790,6 +2790,46 @@ describe('StreamableHTTPClientTransport', () => { expect(onerror).not.toHaveBeenCalled(); }); + it('transport close() settles terminateSession() parked in a hung token(), without onerror', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + // terminateSession early-returns without a session id. + hungTransport['_sessionId'] = 'session-1'; + + const pending = hungTransport.terminateSession(); + setTimeout(() => void hungTransport.close(), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect(outcome).not.toBe('send() hung'); + // Transport-lifetime abort is intentional teardown — no onerror. + expect(onerror).not.toHaveBeenCalled(); + }); + + it('a plain-JS provider returning a bare (non-thenable) token still works through the race', async () => { + const bareTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + // What a plain-JS caller can hand over despite the types: a + // synchronous return. The pre-race `await` tolerated it. + authProvider: { token: (() => 'bare-token') as unknown as () => Promise } + }); + await bareTransport.start(); + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + + await bareTransport.send({ jsonrpc: '2.0', method: 'notifications/x', params: {} }); + const [, init] = (globalThis.fetch as Mock).mock.calls[0] as [unknown, { headers: Headers }]; + expect(init.headers.get('authorization')).toBe('Bearer bare-token'); + await bareTransport.close(); + }); + it('a non-Error rejection from the raced auth await passes through identity-preserved (auth-seam stamp intact)', async () => { // The auth-seam stamp is a property set on the thrown VALUE // (Error or not); a rewrap inside the race would strip it on From f4e9e33bc13a1dd6374fbae186ebe6ad543d1ae6 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:14:29 +0000 Subject: [PATCH 4/8] =?UTF-8?q?fix(client):=20review=20round=20three=20?= =?UTF-8?q?=E2=80=94=20export=20UnauthorizedContext,=20dedupe=20onerror,?= =?UTF-8?q?=20extract=20=5FcombinedSignal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - export UnauthorizedContext from the client barrel and document ctx.signal forwarding in machine-auth.md and upgrade-to-v2.md - changeset now mentions the folded resume-observer fix - drop the duplicate onerror emission in the resume-via-send and 202/initialized catch blocks (_startOrAuthSse already emits before rethrowing); fire onRequestStreamEnd on a genuine resume failure so the observed per-request stream does not dead-end silently - extract _combinedSignal() for the duplicated transport+request signal computation (used by _send, _startOrAuthSse, terminateSession) - use isAborted() in the two onUnauthorized catch guards Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- .changeset/abortable-auth-awaits.md | 2 +- docs/clients/machine-auth.md | 2 +- docs/migration/upgrade-to-v2.md | 3 +- packages/client/src/client/streamableHttp.ts | 73 +++++++++++--------- packages/client/src/index.ts | 3 +- 5 files changed, 45 insertions(+), 38 deletions(-) diff --git a/.changeset/abortable-auth-awaits.md b/.changeset/abortable-auth-awaits.md index 2bc7af5e60..5780e12017 100644 --- a/.changeset/abortable-auth-awaits.md +++ b/.changeset/abortable-auth-awaits.md @@ -2,4 +2,4 @@ '@modelcontextprotocol/client': patch --- -Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). +Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). Also fixes resume-via-`send()`: the resumed GET now preserves the caller's `onresumptiontoken`/`onRequestStreamEnd` observers, so resumed streams keep the token-persistence chain, report their terminal end, and an outright resume failure no longer dead-ends silently. diff --git a/docs/clients/machine-auth.md b/docs/clients/machine-auth.md index 1b3c6614bb..ec8447617c 100644 --- a/docs/clients/machine-auth.md +++ b/docs/clients/machine-auth.md @@ -40,7 +40,7 @@ const authProvider: AuthProvider = { token: async () => getStoredToken() }; const transport = new StreamableHTTPClientTransport(new URL('https://api.example.com/mcp'), { authProvider }); ``` -The transport calls `token()` before every request and sets the `Authorization` header from whatever it returns. Without `onUnauthorized`, a 401 throws `UnauthorizedError`. Add `onUnauthorized(ctx)` to refresh the credential and the transport retries the request once. +The transport calls `token()` before every request and sets the `Authorization` header from whatever it returns. Without `onUnauthorized`, a 401 throws `UnauthorizedError`. Add `onUnauthorized(ctx)` to refresh the credential and the transport retries the request once. `ctx.signal` (when present) aborts once the caller or transport gives up on the request — forward it to your own fetches so the recovery work stops with it. ## Sign with a private key instead of a secret diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 19f4127733..81ce3d6d71 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1091,7 +1091,8 @@ The transport `authProvider` option is widened to `AuthProvider | OAuthClientPro **`AuthProvider`** is a new minimal interface — `{ token(): Promise; onUnauthorized?(ctx): Promise }` — for static-token / non-OAuth bearer auth. Transports call `token()` before every request and `onUnauthorized()` on 401 (then retry -once). Existing `OAuthClientProvider` implementations need no changes — transports adapt +once); `ctx.signal` (when present) aborts once the caller or transport gives up — forward +it to your own fetches. Existing `OAuthClientProvider` implementations need no changes — transports adapt them internally via the new `adaptOAuthProvider()` export. Also exported: `isOAuthClientProvider()` (type guard) and `handleOAuthUnauthorized()` (the standard OAuth `onUnauthorized` behavior, for composing your own adapter). diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index b35a755f4d..0e911986b9 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -502,6 +502,20 @@ export class StreamableHTTPClientTransport implements Transport { }); } + /** + * The abort signal governing one request: the transport-lifetime signal + * combined with the caller's per-request signal when both exist, + * whichever one exists otherwise. Computed BEFORE header acquisition so + * the auth chain (token(), 401 recovery, step-up) is raced against the + * same abort as the request itself. + */ + private _combinedSignal(requestSignal?: AbortSignal): AbortSignal | undefined { + const transportSignal = this._abortController?.signal; + return requestSignal !== undefined && transportSignal !== undefined + ? anySignal(transportSignal, requestSignal) + : (requestSignal ?? transportSignal); + } + private async _commonHeaders(signal?: AbortSignal): Promise { // Already-aborted fast path BEFORE the provider is invoked: an // aborted request must not start new auth work just to discard it @@ -613,14 +627,7 @@ export class StreamableHTTPClientTransport implements Transport { const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; try { - // Combined BEFORE header acquisition so the auth chain (token(), - // 401 recovery, step-up) is raced against the same abort as the - // GET itself. - const transportSignal = this._abortController?.signal; - const signal = - requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, requestSignal) - : (requestSignal ?? transportSignal); + const signal = this._combinedSignal(requestSignal); // Try to open an initial SSE stream with GET to listen for server messages // This is optional according to the spec - server may not support it @@ -669,7 +676,7 @@ export class StreamableHTTPClientTransport implements Transport { } catch (error) { // An abort is an intentional teardown, not an auth // failure — leave it unstamped. - if (signal?.aborted === true) { + if (isAborted(signal)) { throw error; } // Auth-seam stamp: covers the SDK's OAuth flow and @@ -1073,14 +1080,21 @@ export class StreamableHTTPClientTransport implements Transport { // stream's terminal end. onresumptiontoken, onRequestStreamEnd: options?.onRequestStreamEnd - }).catch(error => { - // Same guard as `_scheduleReconnection`'s reconnect(): an - // abort of either signal during the resume (now reachable - // mid-auth-chain too) is intentional teardown, not an error. + }).catch(() => { + // `_startOrAuthSse`'s own catch already routed the + // failure to onerror (suppressed for intentional aborts) + // before rethrowing — no second emission here. Same + // abort guard as `_scheduleReconnection`'s reconnect(): + // an abort of either signal during the resume (now + // reachable mid-auth-chain too) is intentional teardown. if (this._abortController?.signal.aborted === true || options?.requestSignal?.aborted === true) { return; } - this.onerror?.(error); + // An outright resume failure (network, non-401/403/405 + // HTTP error) is TERMINAL for the per-request stream the + // caller is observing — this is the only channel that + // reports it (the 405 case fires inside _startOrAuthSse). + options?.onRequestStreamEnd?.(); }); return; } @@ -1088,13 +1102,8 @@ export class StreamableHTTPClientTransport implements Transport { // Per-request abort: when the caller supplies a request-scoped // signal (the `subscriptions/listen` driver), aborting it cancels // this POST and its SSE response stream without closing the - // transport. Combined BEFORE header acquisition so the auth chain - // (token(), 401 recovery, step-up) is raced against it too. - const transportSignal = this._abortController?.signal; - const signal = - options?.requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, options.requestSignal) - : (options?.requestSignal ?? transportSignal); + // transport. + const signal = this._combinedSignal(options?.requestSignal); const headers = await this._commonHeaders(signal); this._applyBodyDerivedHeaders(headers, message); @@ -1166,7 +1175,7 @@ export class StreamableHTTPClientTransport implements Transport { } catch (error) { // An abort is an intentional teardown, not an auth // failure — leave it unstamped. - if (signal?.aborted === true) { + if (isAborted(signal)) { throw error; } // Auth-seam stamp: covers the SDK's OAuth flow and @@ -1255,17 +1264,13 @@ 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 => { - // A transport-lifetime abort during the GET (now - // reachable mid-auth-chain too) is intentional - // teardown, not an error. No per-request signal on - // this standalone-GET path. - if (this._abortController?.signal.aborted === true) { - return; - } - this.onerror?.(error); - }); + // Start without a lastEventId since this is a fresh + // connection. `_startOrAuthSse`'s own catch already + // routes failures to onerror (suppressed for intentional + // aborts) before rethrowing — emitting here again would + // double-report. Standalone GET: no per-request stream, + // so there is no terminal callback to fire either. + this._startOrAuthSse({ resumptionToken: undefined }).catch(() => {}); } return; } @@ -1349,7 +1354,7 @@ export class StreamableHTTPClientTransport implements Transport { } try { - const headers = await this._commonHeaders(this._abortController?.signal); + const headers = await this._commonHeaders(this._combinedSignal()); const init = { ...this._requestInit, diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index cfe9f88389..0d6f3c0e39 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -15,7 +15,8 @@ export type { OAuthClientInformationContext, OAuthClientProvider, OAuthDiscoveryState, - OAuthServerInfo + OAuthServerInfo, + UnauthorizedContext } from './client/auth'; export { assertSecureTokenEndpoint, From ce28a325ffd656d2de48048df640fda109ae8c80 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:37:07 +0000 Subject: [PATCH 5/8] =?UTF-8?q?fix(client):=20review=20round=20four=20?= =?UTF-8?q?=E2=80=94=20thunked=20race,=20full=20signal=20adoption,=20recon?= =?UTF-8?q?nect=20dedup,=20predicate=20extraction?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - raceWithSignal now takes a producer and runs its aborted fast-path BEFORE invoking it, so an abort landing ahead of any of the five auth stages (token(), 401 recovery x2, step-up x2) never starts side-effectful work; subsumes _commonHeaders' special-case gate - terminateSession uses _combinedSignal() for the DELETE fetch too - reconnect() no longer re-reports a failure _startOrAuthSse already routed to onerror; one report per failed attempt, retry kept - extract _isIntentionalAbort(requestSignal?) and use it at all seven guard sites (three hand-written variants unified) - tests: pre-aborted requestSignal never invokes token(); a failed reconnect attempt reports onerror exactly once Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- packages/client/src/client/streamableHttp.ts | 139 ++++++++++-------- .../client/test/client/streamableHttp.test.ts | 55 +++++++ 2 files changed, 136 insertions(+), 58 deletions(-) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 0e911986b9..3e9ba4f526 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -332,14 +332,27 @@ function abortReasonError(signal: AbortSignal): Error { * promises. The abort listener is removed as soon as the promise settles so * a long-lived transport signal does not accumulate closures per request. */ -function raceWithSignal(value: Promise, signal: AbortSignal | undefined): Promise { - // Normalize first: a plain-JS provider may return a bare value (or a - // foreign thenable) where the types say Promise — the plain `await` these - // call sites used before the race tolerated that, so the race must too +function raceWithSignal(produce: () => Promise, signal: AbortSignal | undefined): Promise { + // Aborted fast-path BEFORE the producer is invoked: an already-dead + // request must not start side-effectful auth work (a token refresh, an + // OAuth step-up that mutates discovery state or redirects the user) + // just to have its result discarded. + if (signal !== undefined && isAborted(signal)) { + return Promise.reject(abortReasonError(signal)); + } + // Normalize: a plain-JS provider may return a bare value (or a foreign + // thenable) where the types say Promise — the plain `await` these call + // sites used before the race tolerated that, so the race must too // (`.then` on a bare string would TypeError and get misstamped as an - // auth failure). Identity-preserving for native promises, and it keeps - // the executor below throw-free so the abort listener cannot leak. - const promise = Promise.resolve(value); + // auth failure). Identity-preserving for native promises. A synchronous + // throw from the producer becomes a rejection, matching how the plain + // `await` call sites surfaced it. + let promise: Promise; + try { + promise = Promise.resolve(produce()); + } catch (error) { + return Promise.reject(error); + } if (signal === undefined) { return promise; } @@ -516,21 +529,25 @@ export class StreamableHTTPClientTransport implements Transport { : (requestSignal ?? transportSignal); } + /** + * Whether teardown has been signalled for the transport (`close()`) or + * for the observed request (its `requestSignal`). Errors surfacing after + * either are intentional-abort fallout, not reportable failures — the + * guard-site mirror of {@linkcode _combinedSignal}. + */ + private _isIntentionalAbort(requestSignal?: AbortSignal): boolean { + return isAborted(this._abortController?.signal) || isAborted(requestSignal); + } + private async _commonHeaders(signal?: AbortSignal): Promise { - // Already-aborted fast path BEFORE the provider is invoked: an - // aborted request must not start new auth work just to discard it - // (raceWithSignal's own fast-path runs after `token()` has been - // called, which is too late). - if (signal !== undefined && isAborted(signal)) { - throw abortReasonError(signal); - } const headers: RequestInit['headers'] & Record = {}; + const provider = this._authProvider; let token: string | undefined; try { // Raced against the per-request/transport abort so a hung // `token()` (wedged refresh, slow broker) cannot park the send // past its `requestSignal` (#2643). - token = this._authProvider === undefined ? undefined : await raceWithSignal(this._authProvider.token(), signal); + token = provider === undefined ? undefined : await raceWithSignal(() => provider.token(), signal); } catch (error) { // An abort is an intentional teardown, never an auth failure — // leave it unstamped so the send-catch and the negotiation @@ -619,12 +636,6 @@ export class StreamableHTTPClientTransport implements Transport { private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise { const { resumptionToken, requestSignal } = options; - // Same guard as `_handleSseStream`: a resurrected listen stream (the - // POST-SSE → GET reconnect path threads `requestSignal` through - // `StartSSEOptions`) must honour the per-request abort exactly as the - // original POST did — both as a fetch signal and as a "do not surface - // onerror" gate. - const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; try { const signal = this._combinedSignal(requestSignal); @@ -658,19 +669,21 @@ export class StreamableHTTPClientTransport implements Transport { this._scope = computeScopeUnion(this._scope, scope); } - if (this._authProvider.onUnauthorized && !isAuthRetry) { + const onUnauthorized = this._authProvider.onUnauthorized?.bind(this._authProvider); + if (onUnauthorized && !isAuthRetry) { try { // Raced against the per-request/transport abort so // a hung 401 recovery cannot park the GET (#2643); // the signal is also handed to the provider so a // cooperative implementation can cancel its own work. await raceWithSignal( - this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit, - signal - }), + () => + onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), signal ); } catch (error) { @@ -706,10 +719,11 @@ export class StreamableHTTPClientTransport implements Transport { // Raced against the per-request/transport abort so a // hung step-up authorization cannot park the GET (#2643). const result = await raceWithSignal( - this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries - ), + () => + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), signal ); if (result !== 'AUTHORIZED') { @@ -744,7 +758,11 @@ export class StreamableHTTPClientTransport implements Transport { this._handleSseStream(response.body, options, true); } catch (error) { - if (!isIntentionalAbort()) { + // A resurrected listen stream (the POST-SSE → GET reconnect path + // threads `requestSignal` through `StartSSEOptions`) must honour + // the per-request abort exactly as the original POST did — both + // as a fetch signal and as a "do not surface onerror" gate. + if (!this._isIntentionalAbort(requestSignal)) { this.onerror?.(error as Error); } throw error; @@ -798,10 +816,12 @@ export class StreamableHTTPClientTransport implements Transport { // 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. - if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return; - this._startOrAuthSse(options).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)}`)); + if (this._isIntentionalAbort(options.requestSignal)) return; + this._startOrAuthSse(options).catch(() => { + if (this._isIntentionalAbort(options.requestSignal)) return; + // `_startOrAuthSse`'s own catch already routed the failure to + // onerror before rethrowing — re-emitting here would report + // every failed attempt twice. Just schedule the next try. try { this._scheduleReconnection(options, attemptCount + 1); } catch (scheduleError) { @@ -834,7 +854,7 @@ export class StreamableHTTPClientTransport implements Transport { // a clean shutdown: no misleading "SSE stream disconnected" onerror, // and no GET+Last-Event-ID reconnect that would resurrect a stream the // caller just tore down. - const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; + const isIntentionalAbort = (): boolean => this._isIntentionalAbort(requestSignal); let lastEventId: string | undefined; // Track whether we've received a priming event (event with ID) @@ -1087,7 +1107,7 @@ export class StreamableHTTPClientTransport implements Transport { // abort guard as `_scheduleReconnection`'s reconnect(): // an abort of either signal during the resume (now // reachable mid-auth-chain too) is intentional teardown. - if (this._abortController?.signal.aborted === true || options?.requestSignal?.aborted === true) { + if (this._isIntentionalAbort(options?.requestSignal)) { return; } // An outright resume failure (network, non-401/403/405 @@ -1157,19 +1177,21 @@ export class StreamableHTTPClientTransport implements Transport { this._scope = computeScopeUnion(this._scope, scope); } - if (this._authProvider.onUnauthorized && !isAuthRetry) { + const onUnauthorized = this._authProvider.onUnauthorized?.bind(this._authProvider); + if (onUnauthorized && !isAuthRetry) { try { // Raced against the per-request/transport abort so // a hung 401 recovery cannot park the send (#2643); // the signal is also handed to the provider so a // cooperative implementation can cancel its own work. await raceWithSignal( - this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit, - signal - }), + () => + onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), signal ); } catch (error) { @@ -1208,10 +1230,11 @@ export class StreamableHTTPClientTransport implements Transport { // hung step-up authorization (metadata discovery, token // exchange) cannot park the send (#2643). const result = await raceWithSignal( - this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries - ), + () => + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), signal ); if (result !== 'AUTHORIZED') { @@ -1322,11 +1345,10 @@ export class StreamableHTTPClientTransport implements Transport { // Intentional abort BEFORE response headers — the // `subscriptions/listen` driver aborting its `requestSignal`, or // `close()` aborting the transport signal while the send is in - // the auth chain or the fetch. Same guard as - // `_handleSseStream`'s `isIntentionalAbort` — do not surface a - // misleading onerror; still rethrow so `listen()`'s send-catch - // settles the per-subscription state machine. - if (options?.requestSignal?.aborted !== true && this._abortController?.signal.aborted !== true) { + // the auth chain or the fetch. Do not surface a misleading + // onerror; still rethrow so `listen()`'s send-catch settles the + // per-subscription state machine. + if (!this._isIntentionalAbort(options?.requestSignal)) { this.onerror?.(error as Error); } throw error; @@ -1354,13 +1376,14 @@ export class StreamableHTTPClientTransport implements Transport { } try { - const headers = await this._commonHeaders(this._combinedSignal()); + const signal = this._combinedSignal(); + const headers = await this._commonHeaders(signal); const init = { ...this._requestInit, method: 'DELETE', headers, - signal: this._abortController?.signal + signal }; const response = await (this._fetch ?? fetch)(this._url, init); @@ -1385,7 +1408,7 @@ export class StreamableHTTPClientTransport implements Transport { // the auth chain or the fetch) is intentional teardown — same // discipline as `_send`'s catch. No per-request signal on this // path. Still rethrow so the caller observes the failure. - if (this._abortController?.signal.aborted !== true) { + if (!this._isIntentionalAbort()) { this.onerror?.(error as Error); } throw error; diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 25a8b48ff1..f83a002b8a 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1246,6 +1246,44 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock.mock.calls[1]![1]?.method).toBe('GET'); }); + it('reports a failed reconnect attempt via onerror exactly once', async () => { + // _startOrAuthSse's own catch routes the failure to onerror before + // rethrowing; the reconnect scheduling must not report it again. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const failingStream = new ReadableStream({ + start(controller) { + controller.error(new Error('Network failure')); + } + }); + const fetchMock = globalThis.fetch as Mock; + // Initial GET stream drops, scheduling a reconnect… + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: failingStream + }); + // …whose GET then fails outright. + fetchMock.mockRejectedValueOnce(new Error('connection refused')); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(20); + + const refusedReports = errorSpy.mock.calls.filter(([e]) => String((e as Error).message).includes('connection refused')); + expect(refusedReports).toHaveLength(1); + }); + it('should NOT reconnect a POST-initiated stream that fails', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { @@ -2810,6 +2848,23 @@ describe('StreamableHTTPClientTransport', () => { expect(onerror).not.toHaveBeenCalled(); }); + it('an already-aborted requestSignal rejects send() before the provider is invoked', async () => { + // The aborted fast-path must run BEFORE the producer: a dead + // request must not start side-effectful auth work. + const token = vi.fn(() => Promise.resolve('t')); + const gatedTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token } + }); + await gatedTransport.start(); + + const ac = new AbortController(); + ac.abort(new Error('already dead')); + await expect(gatedTransport.send(request, { requestSignal: ac.signal })).rejects.toThrow('already dead'); + expect(token).not.toHaveBeenCalled(); + expect(globalThis.fetch).not.toHaveBeenCalled(); + await gatedTransport.close(); + }); + it('a plain-JS provider returning a bare (non-thenable) token still works through the race', async () => { const bareTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { // What a plain-JS caller can hand over despite the types: a From 123e3e9155f1b21dad2b4b9ee1ab8030f27bf83c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 06:42:55 +0000 Subject: [PATCH 6/8] test(e2e): pin one onerror per failed reconnect attempt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ce28a32 deliberately dropped the 'Failed to reconnect SSE stream:' re-wrap that duplicated each attempt's failure report; the reconnect-failure-onerror scenario pinned that duplicate's message. Each failed attempt still reaches onerror exactly once — now as the underlying failure itself — and the scenario asserts both that and the wrapper's absence. Retry budget and GET counts unchanged. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- test/e2e/scenarios/transport-http.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/test/e2e/scenarios/transport-http.test.ts b/test/e2e/scenarios/transport-http.test.ts index 7cda0f28aa..4a03d59941 100644 --- a/test/e2e/scenarios/transport-http.test.ts +++ b/test/e2e/scenarios/transport-http.test.ts @@ -1482,7 +1482,11 @@ verifies('client-transport:http:reconnect-failure-onerror', async (_args: TestAr await vi.waitFor(() => expect(transportErrors.filter(e => e.message === 'Maximum reconnection attempts (2) exceeded.')).toHaveLength(1) ); - expect(transportErrors.filter(e => e.message.startsWith('Failed to reconnect SSE stream:'))).toHaveLength(2); + // EXACTLY once per failed attempt: the underlying failure itself. + // The 'Failed to reconnect SSE stream:' re-wrap that used to + // accompany each report was a duplicate of the same failure. + expect(transportErrors.filter(e => e.message.startsWith('Failed to open SSE stream:'))).toHaveLength(2); + expect(transportErrors.filter(e => e.message.startsWith('Failed to reconnect SSE stream:'))).toHaveLength(0); expect(records.filter(r => r.method === 'GET')).toHaveLength(3); // The reconnection failure stays on onerror: an unrelated request issued afterwards still succeeds From 61062b0a50a07786b16a01df7ded459fe1d49334 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 07:01:28 +0000 Subject: [PATCH 7/8] fix(client): keep the prior resumption token across a pre-first-event drop; doc the reconnect emission contract - a reconnect scheduled before the resumed stream delivered any ID-bearing event now falls back to options.resumptionToken at both _scheduleReconnection sites, so Last-Event-ID is not silently lost (a token-less GET starts a fresh stream and the pending response never replays); pinned by a fake-timers test - changeset + upgrade-to-v2.md now state the reconnect emission contract: one onerror per failed attempt carrying the underlying error, wrapper gone, exhaustion message-text guarantee unchanged Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- .changeset/abortable-auth-awaits.md | 2 +- docs/migration/upgrade-to-v2.md | 7 +++- packages/client/src/client/streamableHttp.ts | 12 +++++- .../client/test/client/streamableHttp.test.ts | 40 +++++++++++++++++++ 4 files changed, 57 insertions(+), 4 deletions(-) diff --git a/.changeset/abortable-auth-awaits.md b/.changeset/abortable-auth-awaits.md index 5780e12017..f59c5ac053 100644 --- a/.changeset/abortable-auth-awaits.md +++ b/.changeset/abortable-auth-awaits.md @@ -2,4 +2,4 @@ '@modelcontextprotocol/client': patch --- -Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). Also fixes resume-via-`send()`: the resumed GET now preserves the caller's `onresumptiontoken`/`onRequestStreamEnd` observers, so resumed streams keep the token-persistence chain, report their terminal end, and an outright resume failure no longer dead-ends silently. +Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). Also fixes resume-via-`send()`: the resumed GET now preserves the caller's `onresumptiontoken`/`onRequestStreamEnd` observers, so resumed streams keep the token-persistence chain, report their terminal end, and an outright resume failure no longer dead-ends silently. Reconnect-attempt failures now reach `onerror` exactly once, as the underlying error — the `Failed to reconnect SSE stream: …` wrapper that duplicated each per-attempt report is gone (the `Maximum reconnection attempts (N) exceeded.` exhaustion message is unchanged). A reconnect scheduled before the resumed stream delivered any ID-bearing event now falls back to the stream's prior resumption token instead of silently dropping `Last-Event-ID`. diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 81ce3d6d71..8d52afd762 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1512,7 +1512,12 @@ rewrite required unless noted. 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. + class for this condition, so monitors that match the message text keep working. The + message-text guarantee is scoped to that exhaustion message: each failed attempt + before exhaustion now reaches `onerror` exactly once, as the underlying error itself — + the `Failed to reconnect SSE stream: …` wrapper that previously accompanied every + per-attempt report is gone, so monitors matching that wrapper text should match the + underlying error (or the exhaustion message) instead. - **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 3e9ba4f526..094a87c629 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -926,7 +926,12 @@ export class StreamableHTTPClientTransport implements Transport { if (needsReconnect && this._abortController && !isIntentionalAbort()) { this._scheduleReconnection( { - resumptionToken: lastEventId, + // Fall back to the token this stream was RESUMED + // from: a resumed stream that drops before its + // first ID-bearing event must not lose its + // Last-Event-ID (a token-less GET starts a fresh + // stream and the pending response never replays). + resumptionToken: lastEventId ?? options.resumptionToken, onresumptiontoken, replayMessageId, requestSignal, @@ -959,7 +964,10 @@ export class StreamableHTTPClientTransport implements Transport { try { this._scheduleReconnection( { - resumptionToken: lastEventId, + // Same fallback as the graceful-close path + // above: keep the token this stream was + // resumed from when it dies pre-first-event. + resumptionToken: lastEventId ?? options.resumptionToken, onresumptiontoken, replayMessageId, requestSignal, diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index f83a002b8a..76ca62fe0f 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1246,6 +1246,46 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock.mock.calls[1]![1]?.method).toBe('GET'); }); + it('a resumed stream dropping before its first ID-bearing event keeps the prior resumption token', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const fetchMock = globalThis.fetch as Mock; + // The resumed GET's stream errors before delivering any event… + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.error(new Error('dropped before first event')); + } + }) + }); + // …and the retry GET succeeds. + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream() + }); + + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'tok-1' }); + await vi.advanceTimersByTimeAsync(20); + + expect(fetchMock).toHaveBeenCalledTimes(2); + // The retry must NOT lose Last-Event-ID: no event arrived to + // supersede the token the dropped stream was resumed from. + const retryHeaders = (fetchMock.mock.calls[1]![1] as { headers: Headers }).headers; + expect(retryHeaders.get('last-event-id')).toBe('tok-1'); + }); + it('reports a failed reconnect attempt via onerror exactly once', async () => { // _startOrAuthSse's own catch routes the failure to onerror before // rethrowing; the reconnect scheduling must not report it again. From 53c393c4800db2165baf2b8961317f4a5a1e6a93 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 07:10:45 +0000 Subject: [PATCH 8/8] fix(client): notify onRequestStreamEnd when re-arming a reconnect throws MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A custom reconnectionScheduler that throws while scheduling attempt N+1 ends the retry chain — terminal for an observed per-request stream, exactly like budget exhaustion, but the catch fired only onerror so a listen() observer waited forever. Fire the stream-end callback too, matching the sibling terminal paths. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- packages/client/src/client/streamableHttp.ts | 6 ++++ .../client/test/client/streamableHttp.test.ts | 31 +++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 094a87c629..a24e080022 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -826,6 +826,12 @@ export class StreamableHTTPClientTransport implements Transport { this._scheduleReconnection(options, attemptCount + 1); } catch (scheduleError) { this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError))); + // A scheduler that throws while re-arming ends the retry + // chain — that is TERMINAL for an observed per-request + // stream, exactly like retry-budget exhaustion above, so + // the caller must be told or a listen() observer waits + // forever. + options.onRequestStreamEnd?.(); } }); }; diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index 76ca62fe0f..caa7436ca9 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -2699,6 +2699,37 @@ describe('StreamableHTTPClientTransport', () => { vi.useRealTimers(); }); + it('a scheduler that throws while re-arming fires onRequestStreamEnd (terminal for observed streams)', async () => { + let armed: (() => void) | undefined; + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: (cb, _delay, attemptCount) => { + if (attemptCount > 0) throw new Error('scheduler exploded'); + armed = cb; + return () => {}; + } + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const onRequestStreamEnd = vi.fn(); + (globalThis.fetch as Mock).mockRejectedValue(new Error('connection refused')); + await transport.start(); + + (transport as unknown as { _scheduleReconnection(opts: StartSSEOptions, attempt?: number): void })._scheduleReconnection( + { onRequestStreamEnd }, + 0 + ); + expect(armed).toBeDefined(); + // Attempt 0 runs, its GET fails, and re-arming attempt 1 throws. + armed!(); + await vi.advanceTimersByTimeAsync(0); + + // The retry chain is dead — that is terminal for an observed + // per-request stream, same as budget exhaustion. + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + expect(onerror.mock.calls.some(([e]) => (e as Error).message.includes('scheduler exploded'))).toBe(true); + }); + it('invokes the custom scheduler with reconnect, delay, and attemptCount', () => { const scheduler = vi.fn(); transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {