diff --git a/.changeset/legacy-sse-reconnect-after-timeout.md b/.changeset/legacy-sse-reconnect-after-timeout.md new file mode 100644 index 0000000000..1992c14225 --- /dev/null +++ b/.changeset/legacy-sse-reconnect-after-timeout.md @@ -0,0 +1,8 @@ +--- +'@modelcontextprotocol/client': patch +'@modelcontextprotocol/server': patch +--- + +Stop the legacy-era (2025-11-25) Streamable HTTP SSE reconnect chain when its originating request settles. Previously, when a request on a legacy session timed out or was aborted, the client POSTed `notifications/cancelled` but the transport kept resuming the request's SSE stream via GET + Last-Event-ID indefinitely (each successful resume reset the retry counter), and a late resumed response surfaced as "Received a response for an unknown message ID". The protocol layer now threads a request-scoped abort signal to per-request-stream transports on the legacy era too, and aborts it on EVERY settlement path — cancel/timeout, `maxTotalTimeout` exceeded, and successful completion (which also releases the per-request abort listener otherwise leaked on Node 20.0–20.2) — the wire cancellation mechanism per era is unchanged (legacy connections still POST `notifications/cancelled`; modern 2026-07-28 connections still cancel via per-request stream close), and `maxTotalTimeout` settlements — which previously settled without any wire cancel signal at either era — now route through the request's cancel path and emit the era's signal (the `notifications/cancelled` POST on legacy connections and modern single-channel transports, the stream-close cancel on modern per-request-stream connections) while the caller still sees the original maxTotalTimeout error. This lives in the shared `Protocol` base, so server-initiated requests (`createMessage`, `elicitInput`) gain the same maxTotalTimeout cancellation signal. Additionally, the `notifications/cancelled` send no longer inherits the original request's `resumptionToken`, which on Streamable HTTP short-circuited the send into an SSE resume and silently swallowed the cancellation. For a request re-issued with `resumptionToken` on the SDK's own Streamable HTTP transport that POST is best-effort: the re-issued request's JSON-RPC id never reached the server (the send resumed via GET instead of POSTing), so the server cannot correlate the cancellation and such requests are only torn down locally; custom per-request-stream transports that POST the re-issued body normally give the server a correlatable id. (Same asymmetry inbound, pre-existing: progress notifications replayed on a resumed stream carry the original request's `progressToken`, so `onprogress`/`resetTimeoutOnProgress` do not survive a `resumptionToken` re-issue on this transport.) The client transport's `onerror` contract is also tightened: each failed SSE reconnect leg now reports exactly once with the underlying error (the `"Failed to reconnect SSE stream:"` wrapper message is gone), deliberate teardown (`close()` landing mid-POST/mid-GET/mid-DELETE, or a settled request's signal aborting its resume) no longer surfaces an `AbortError` through `onerror`, and `onRequestStreamEnd` now fires when a `resumptionToken` resume fails to open (a terminal outcome that previously reported only through `onerror`). + +Two hardening details in the same machinery: `close()` now disarms every pending reconnect chain even when a custom `ReconnectionScheduler` cancel throws (the first error still propagates, after all chains are disarmed) and releases each chain's settlement listener, so a user-supplied cancel runs at most once across `close()` and the request's own settlement; and each request chain's reconnect legs now reuse one composed transport+request abort signal instead of composing a fresh one per leg (on Node 20.0-20.2, where `AbortSignal.any` is unavailable, per-leg composition stranded one abort-listener pair per completed resume leg on both input signals until the request settled). Two adjacent lifecycle/boundary fixes from the same review sweep: `maxTotalTimeout: 0` is now honored as the strictest budget (rejecting on the first progress-driven timeout reset) instead of being silently disabled by a falsy check; and on the legacy HTTP+SSE transport (`SSEClientTransport`), a `close()` that lands while a mid-session 401 token refresh (`onUnauthorized`) is pending no longer lets the recovery continuation open a new EventSource that nothing can tear down — the continuation now rejects with `UnauthorizedError('Transport closed during re-authentication')`. diff --git a/docs/advanced/custom-transports.md b/docs/advanced/custom-transports.md index 242d55937e..ee1e95dcad 100644 --- a/docs/advanced/custom-transports.md +++ b/docs/advanced/custom-transports.md @@ -1,6 +1,7 @@ --- shape: how-to --- + # Custom transports A **transport** moves `JSONRPCMessage` values in both directions over a channel the SDK knows nothing about. Implement the `Transport` interface and `connect()` accepts it like a built-in one. @@ -180,7 +181,7 @@ async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise { if (event.code === 401 && this._authProvider) { @@ -219,13 +224,36 @@ export class SSEClientTransport implements Transport { this._eventSource?.close(); this._authProvider.onUnauthorized({ response, serverUrl: this._url, fetchFn: this._fetchWithInit }).then( // onUnauthorized succeeded → retry fresh. Its onerror handles its own onerror?.() + reject. - () => this._startOrAuth().then(resolve, reject), + () => { + // Deferred continuation after an arbitrarily + // long refresh await: a close() that landed in + // the meantime must not be undone by opening a + // brand-new EventSource — the ES wrapper fetch + // never carries the transport-lifetime signal, + // and close() already ran against the old + // instance, so nothing could ever tear the + // resurrected stream down. + if (this._abortController?.signal.aborted === true) { + reject(new UnauthorizedError('Transport closed during re-authentication')); + return; + } + this._startOrAuth().then(resolve, reject); + }, // onUnauthorized failed → not yet reported. Auth-seam // stamp: covers the SDK's OAuth flow and custom // callbacks alike. (error: unknown) => { markAuthSeamEscape(error); - this.onerror?.(error as Error); + // Mirror the success arm's closed-state guard: + // a refresh that rejects AFTER close() (token + // endpoint unreachable at shutdown, abandoned + // interactive flow) must not surface a + // spurious auth error through onerror + // post-shutdown. Still reject so a pending + // start() settles. + if (this._abortController?.signal.aborted !== true) { + this.onerror?.(error as Error); + } reject(error); } ); @@ -338,9 +366,15 @@ export class SSEClientTransport implements Transport { } async close(): Promise { - this._abortController?.abort(); - this._eventSource?.close(); - this.onclose?.(); + try { + this._abortController?.abort(); + this._eventSource?.close(); + } finally { + // onclose is the ONLY trigger for Protocol._onclose (which settles + // every pending request with ConnectionClosed) — it must fire even + // if the EventSource teardown throws. + this.onclose?.(); + } } async send(message: JSONRPCMessage): Promise { @@ -407,7 +441,13 @@ export class SSEClientTransport implements Transport { // Release connection - POST responses don't have content we need await response.text?.().catch(() => {}); } catch (error) { - this.onerror?.(error as Error); + // The POST runs on the transport-lifetime signal alone, so a + // close() landing mid-flight rejects with an intentional + // AbortError — a clean shutdown, not a transport error. Still + // rethrow so callers see the failure. + if (this._abortController?.signal.aborted !== true) { + this.onerror?.(error as Error); + } throw error; } } diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..cfb3a9c4ef 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -71,6 +71,12 @@ export interface StartSSEOptions { /** * Override Message ID to associate with the replay message * so that the response can be associated with the new resumed request. + * + * Only JSON-RPC RESPONSES are remapped. Notifications replayed on the + * resumed stream (e.g. `notifications/progress`) pass through verbatim, + * still carrying the original request's identifiers — the transport never + * learns the original wire id (callers persist SSE event ids, not message + * ids), so it has nothing to remap `params.progressToken` with. */ replayMessageId?: string | number; @@ -93,6 +99,23 @@ export interface StartSSEOptions { onRequestStreamEnd?: () => void; } +/** + * Internal extension of {@linkcode StartSSEOptions}: the composed + * transport+request abort signal for a request chain, built once by the + * chain's first leg and reused verbatim by every rebuilt reconnect leg. + * Reuse matters on Node 20.0–20.2, where the `anySignal` fallback removes + * its listener pair only when one of its inputs fires — a fresh composite + * per leg would strand one closure pair per gracefully-completed leg on + * BOTH the transport signal and the request signal until the request + * settles (`MaxListenersExceededWarning` after ~11 polling cycles). The + * native `AbortSignal.any` path benefits too: one composite allocation per + * chain instead of one per leg. + */ +type SseLegOptions = StartSSEOptions & { + /** The composed fetch signal shared by every leg of this request chain. */ + fetchSignal?: AbortSignal; +}; + /** * Configuration options for reconnection behavior of the {@linkcode StreamableHTTPClientTransport}. */ @@ -133,9 +156,10 @@ export interface StreamableHTTPReconnectionOptions { * @param reconnect - Call this to perform the reconnection attempt. * @param delay - Suggested delay in milliseconds (from backoff calculation). * @param attemptCount - Zero-indexed retry attempt number. - * @returns An optional cancel function. If returned, it will be called on - * {@linkcode StreamableHTTPClientTransport.close | transport.close()} to abort the - * pending reconnection. + * @returns An optional cancel function. If returned, it is called AT MOST once — + * when the pending reconnection is disarmed early, either by + * {@linkcode StreamableHTTPClientTransport.close | transport.close()} or by the + * originating request settling first. Never called after the attempt fires. * * @example * ```ts source="./streamableHttp.examples.ts#ReconnectionScheduler_basicUsage" @@ -326,7 +350,24 @@ 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; + /** + * Disarm bookkeeping for EVERY pending scheduled reconnection attempt. + * The transport can own several concurrent reconnect chains at once — the + * standalone notification GET plus one per in-flight request on a legacy + * session — so this must be a set, not a single slot: a slot would be + * overwritten by each schedule and `close()` could disarm only the + * last-written chain, leaving armed timers (or un-cancelled custom + * scheduler tasks) behind. + * + * Each entry carries BOTH halves of a chain's teardown: `cancel` (the + * user-supplied or default timer cancel) and `release` (drops the chain's + * settlement listener from its `requestSignal`). `close()` must invoke + * both — releasing the listener is what keeps the request's own + * settlement (onclose → `Protocol._onclose` → the request funnel's + * `.finally()` aborting `requestSignal`) from invoking the user's cancel + * a second time. + */ + private readonly _pendingReconnectCancels = new Set<{ cancel: () => void; release: () => void }>(); onclose?: () => void; onerror?: (error: Error) => void; @@ -334,9 +375,12 @@ export class StreamableHTTPClientTransport implements Transport { /** * Streamable HTTP opens one POST (and SSE response stream) per outbound - * request and honors `TransportSendOptions.requestSignal`. On a 2026-era - * connection the protocol layer aborts that per-request stream as the - * spec cancellation signal instead of POSTing `notifications/cancelled`. + * request and honors `TransportSendOptions.requestSignal`. The protocol + * layer threads `requestSignal` into every outbound request and aborts it + * when the request settles — on a 2026-era connection that abort IS the + * spec cancellation signal (no `notifications/cancelled` is sent); on a + * 2025-era connection it is purely local teardown (it stops the request's + * SSE reconnect chain) accompanying the `notifications/cancelled` POST. */ readonly hasPerRequestStream = true; @@ -517,7 +561,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: SseLegOptions, 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 @@ -526,6 +570,19 @@ export class StreamableHTTPClientTransport implements Transport { // onerror" gate. const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; + // Compose the fetch signal ONCE per request chain: the first leg + // builds it, every rebuilt leg reuses it via `fetchSignal` (see + // `SseLegOptions` — on the Node 20.0-20.2 `anySignal` fallback a + // fresh composite per leg would leak one listener pair per completed + // leg on both input signals). + const transportSignal = this._abortController?.signal; + const signal = + options.fetchSignal ?? + (requestSignal !== undefined && transportSignal !== undefined + ? anySignal(transportSignal, requestSignal) + : (requestSignal ?? transportSignal)); + const legOptions: SseLegOptions = options.fetchSignal === signal ? options : { ...options, fetchSignal: signal }; + try { // 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 @@ -539,11 +596,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', @@ -575,7 +627,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(legOptions, true, stepUpRetries); } await response.text?.().catch(() => {}); if (isAuthRetry) { @@ -600,7 +652,7 @@ export class StreamableHTTPClientTransport implements Transport { if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); } - return this._startOrAuthSse(options, isAuthRetry, stepUpRetries + 1); + return this._startOrAuthSse(legOptions, isAuthRetry, stepUpRetries + 1); } } @@ -616,8 +668,11 @@ export class StreamableHTTPClientTransport implements Transport { // stream-end callback so the caller can settle (otherwise // a resumed listen subscription dead-ends silently). The // standalone-GET callers never pass `onRequestStreamEnd`, - // so this is a no-op for them. - options.onRequestStreamEnd?.(); + // so this is a no-op for them. Guarded: a throwing + // callback would otherwise land in this function's catch + // (misattributed onerror + rethrow) and be re-invoked by + // the resumptionToken short-circuit's own catch. + this._fireRequestStreamEnd(options.onRequestStreamEnd); return; } @@ -627,7 +682,7 @@ export class StreamableHTTPClientTransport implements Transport { }); } - this._handleSseStream(response.body, options, true); + this._handleSseStream(response.body, legOptions, true); } catch (error) { if (!isIntentionalAbort()) { this.onerror?.(error as Error); @@ -657,13 +712,31 @@ export class StreamableHTTPClientTransport implements Transport { return Math.min(initialDelay * Math.pow(growFactor, attempt), maxDelay); } + /** + * Invoke a caller-supplied `onRequestStreamEnd` under a guard. The + * callback fires from floating promise chains (the reconnect closure, the + * resumptionToken short-circuit's catch) and fire-and-forget stream + * processors — contexts where a synchronous throw would either become an + * unhandledRejection (process exit by default) or land in an unrelated + * catch and corrupt its settlement logic. Failures are routed through + * `onerror` instead; the callback is a completion signal, not a caller to + * reject. + */ + private _fireRequestStreamEnd(onRequestStreamEnd?: () => void): void { + try { + onRequestStreamEnd?.(); + } catch (error) { + this.onerror?.(error instanceof Error ? error : new Error(String(error))); + } + } + /** * 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 */ - private _scheduleReconnection(options: StartSSEOptions, attemptCount = 0): void { + private _scheduleReconnection(options: SseLegOptions, attemptCount = 0): void { // Use provided options or default options const maxRetries = this._reconnectionOptions.maxRetries; @@ -671,49 +744,111 @@ export class StreamableHTTPClientTransport implements Transport { if (attemptCount >= maxRetries) { this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`)); // The per-request stream is now definitively gone. - options.onRequestStreamEnd?.(); + this._fireRequestStreamEnd(options.onRequestStreamEnd); return; } // Calculate next delay based on current attempt count const delay = this._getNextReconnectionDelay(attemptCount); + // Per-chain cancel bookkeeping: each scheduled attempt registers its + // own cancel in `_pendingReconnectCancels` and removes it when the + // attempt fires or is disarmed. `listenerCleanup` releases the + // settlement listener below in the same motion. + let entry: { cancel: () => void; release: () => void } | undefined; + let fired = false; + const listenerCleanup = new AbortController(); + const disarmBookkeeping = (): void => { + if (entry !== undefined) { + this._pendingReconnectCancels.delete(entry); + entry = undefined; + } + listenerCleanup.abort(); + }; + const reconnect = (): void => { - this._cancelReconnection = undefined; + fired = true; + disarmBookkeeping(); // 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 => { + this._startOrAuthSse(options).catch(() => { 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)}`)); + // No onerror here: `_startOrAuthSse`'s own catch already + // reported the genuine failure once before rethrowing (and + // stayed silent on an intentional abort — caught above). + // Reporting again would double-fire onerror for every failed + // reconnect leg. Just schedule the next attempt. try { this._scheduleReconnection(options, attemptCount + 1); } catch (scheduleError) { this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError))); + // The chain is terminally dead (no timer was armed) — + // settle the caller, mirroring the maxRetries-exhaustion + // branch. No double-fire is possible: that branch returns + // before the scheduler runs, so _scheduleReconnection can + // never both fire the callback and throw. Guarded: this + // runs inside a floating promise chain, so a throwing + // callback would otherwise become an unhandledRejection. + this._fireRequestStreamEnd(options.onRequestStreamEnd); } }); }; + let cancelPending: () => void; if (this._reconnectionScheduler) { const cancel = this._reconnectionScheduler(reconnect, delay, attemptCount); - this._cancelReconnection = typeof cancel === 'function' ? cancel : undefined; + cancelPending = typeof cancel === 'function' ? cancel : () => {}; } else { const handle = setTimeout(reconnect, delay); - this._cancelReconnection = () => clearTimeout(handle); + cancelPending = () => clearTimeout(handle); + } + // A custom scheduler may invoke `reconnect` synchronously; do not + // register bookkeeping for an attempt that already fired. + if (!fired) { + entry = { cancel: cancelPending, release: () => listenerCleanup.abort() }; + this._pendingReconnectCancels.add(entry); + // A settled request disarms its own pending attempt immediately + // instead of leaving an armed timer to bail at fire time (the + // request-scoped signal is aborted on every settlement path). + options.requestSignal?.addEventListener( + 'abort', + () => { + // catch: an exception thrown inside an AbortSignal 'abort' + // listener is not delivered to the abort() caller — Node + // reports it as an uncaughtException (process exit by + // default), and the funnel aborts requestSignal on EVERY + // settlement, so a throwing user-supplied cancel would + // kill the process on the common path. Route it through + // onerror instead (close() has a caller to reject, so its + // sibling loop rethrows; this listener does not). + // finally: the throw must not leave a stale Set entry + // (which close() would re-invoke) or an armed settlement + // listener behind. + try { + cancelPending(); + } catch (error) { + this.onerror?.(error instanceof Error ? error : new Error(String(error))); + } finally { + disarmBookkeeping(); + } + }, + { once: true, signal: listenerCleanup.signal } + ); } } - private _handleSseStream(stream: ReadableStream | null, options: StartSSEOptions, isReconnectable: boolean): void { + private _handleSseStream(stream: ReadableStream | null, options: SseLegOptions, isReconnectable: boolean): 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 // stream-end callback so the caller can settle. No-op for // standalone-GET callers (they never pass `onRequestStreamEnd`). - options.onRequestStreamEnd?.(); + this._fireRequestStreamEnd(options.onRequestStreamEnd); return; } - const { onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd } = options; + const { onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd, fetchSignal } = options; // An intentional abort — transport-wide close OR a per-request abort // (McpSubscription.close() aborting its `requestSignal`) — must read as // a clean shutdown: no misleading "SSE stream disconnected" onerror, @@ -728,6 +863,10 @@ 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; + // A throwing caller-supplied stream-end callback must not escape the + // fire-and-forget processStream() promise as an unhandledRejection — + // route it through onerror instead. + const fireStreamEnd = (): void => this._fireRequestStreamEnd(onRequestStreamEnd); const processStream = async () => { // this is the closest we can get to trying to catch network errors // if something happens reader will throw @@ -756,7 +895,18 @@ export class StreamableHTTPClientTransport implements Transport { lastEventId = event.id; // Mark that we've received a priming event - stream is now resumable hasPrimingEvent = true; - onresumptiontoken?.(event.id); + // Guard the caller's persistence hook: a throw (e.g. + // QuotaExceededError from the storage write this + // callback is documented for) must not exit the read + // loop — that would misattribute the failure as "SSE + // stream disconnected", lose this event's payload + // unreplayably (its id was already consumed), and + // re-enter reconnection at attempt 0 on every resume. + try { + onresumptiontoken?.(event.id); + } catch (callbackError) { + this.onerror?.(callbackError instanceof Error ? callbackError : new Error(String(callbackError))); + } } // Skip events with no data (priming events, keep-alives) @@ -781,30 +931,6 @@ export class StreamableHTTPClientTransport implements Transport { } } } - - // Handle graceful server-side disconnect - // Server may close connection after sending event ID and retry field - // Reconnect if: already reconnectable (GET stream) OR received a priming event (POST stream with event ID) - // BUT don't reconnect if we already received a response - the request is complete - const canResume = isReconnectable || hasPrimingEvent; - const needsReconnect = canResume && !receivedResponse; - if (needsReconnect && this._abortController && !isIntentionalAbort()) { - this._scheduleReconnection( - { - resumptionToken: lastEventId, - onresumptiontoken, - replayMessageId, - requestSignal, - onRequestStreamEnd - }, - 0 - ); - } else if (!isIntentionalAbort()) { - // The per-request stream ended without reconnecting (no - // priming event for a POST stream, or response already - // received). Not a deliberate abort — notify the caller. - onRequestStreamEnd?.(); - } } catch (error) { if (isIntentionalAbort()) { // The reader threw because we aborted it. Not an error; do @@ -824,23 +950,79 @@ export class StreamableHTTPClientTransport implements Transport { try { this._scheduleReconnection( { - resumptionToken: lastEventId, + // Same fallback as the graceful-close path + // below: never rebuild a resume without its + // token. + resumptionToken: lastEventId ?? options.resumptionToken, onresumptiontoken, replayMessageId, requestSignal, - onRequestStreamEnd + onRequestStreamEnd, + fetchSignal }, 0 ); - } catch (error) { - this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); - onRequestStreamEnd?.(); + } catch (scheduleError) { + // Report the raw scheduler error — same shape as the + // graceful-close tail's schedule catch and the + // reschedule catch, so monitors see one consistent + // identity for this failure class regardless of how + // the previous leg ended. + this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError))); + fireStreamEnd(); } } else { // Non-deliberate stream error without reconnection: the // per-request stream is gone — notify the caller. - onRequestStreamEnd?.(); + fireStreamEnd(); + } + return; + } + + // Handle graceful server-side disconnect. The settlement tail + // runs OUTSIDE the read-loop try: a synchronous throw from a + // user-supplied scheduler or stream-end callback must not land in + // the catch above, where it would be mislabeled "SSE stream + // disconnected" and re-drive this tail a second time (two onerror + // reports for one scheduler failure; a throwing callback invoked + // twice, with the second throw escaping the fire-and-forget + // processStream() as an unhandledRejection). + // Server may close connection after sending event ID and retry field + // Reconnect if: already reconnectable (GET stream) OR received a priming event (POST stream with event ID) + // BUT don't reconnect if we already received a response - the request is complete + const canResume = isReconnectable || hasPrimingEvent; + const needsReconnect = canResume && !receivedResponse; + if (needsReconnect && this._abortController && !isIntentionalAbort()) { + try { + this._scheduleReconnection( + { + // Fall back to the token this leg was opened with: + // a resume leg that drops before delivering its + // first event has no `lastEventId`, and rebuilding + // without a token would degrade the resume into a + // token-less standalone GET (dead-ends on 405 + // servers; loses the replay position otherwise). + resumptionToken: lastEventId ?? options.resumptionToken, + onresumptiontoken, + replayMessageId, + requestSignal, + onRequestStreamEnd, + fetchSignal + }, + 0 + ); + } catch (scheduleError) { + // First-schedule scheduler throw: report the raw error + // exactly once, then settle the caller — mirroring the + // reschedule-throw handling inside _scheduleReconnection. + this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError))); + fireStreamEnd(); } + } else if (!isIntentionalAbort()) { + // The per-request stream ended without reconnecting (no + // priming event for a POST stream, or response already + // received). Not a deliberate abort — notify the caller. + fireStreamEnd(); } }; processStream(); @@ -910,9 +1092,35 @@ export class StreamableHTTPClientTransport implements Transport { async close(): Promise { try { - this._cancelReconnection?.(); + // Disarm EVERY pending scheduled reconnection — the transport can + // own several concurrent chains, each with its own pending + // attempt. Per-entry try/finally: one throwing cancel must not + // skip the remaining chains' cancels, and each chain's settlement + // listener is released here so the request's own settlement + // (onclose → `Protocol._onclose` → the request funnel's + // `.finally()` aborting `requestSignal`) cannot invoke the user's + // cancel a second time. The FIRST cancel error still propagates + // to the caller — after every chain is disarmed; the outer + // finally guarantees the abort and onclose regardless. + let firstError: unknown; + let hasError = false; + for (const entry of this._pendingReconnectCancels) { + try { + entry.cancel(); + } catch (error) { + if (!hasError) { + hasError = true; + firstError = error; + } + } finally { + entry.release(); + } + } + if (hasError) { + throw firstError; + } } finally { - this._cancelReconnection = undefined; + this._pendingReconnectCancels.clear(); this._abortController?.abort(); this.onclose?.(); } @@ -954,11 +1162,48 @@ export class StreamableHTTPClientTransport implements Transport { // same per-request abort as the original POST — modern-era // cancel-via-stream-close routes through `requestSignal`, and // without it a resumed long-running request would not cancel. + // `onresumptiontoken` / `onRequestStreamEnd` are forwarded like + // every other `_startOrAuthSse` call site, so a resumed request + // keeps reporting newer event IDs to the caller's persistence + // hook and still fires the stream-end callback on a terminal + // non-resumable outcome. + // + // Known limitation: `onprogress` / `resetTimeoutOnProgress` do + // NOT survive a resumptionToken re-issue on this transport. + // The re-issued request's fresh progressToken never reaches + // the wire (this path resumes via GET instead of POSTing), + // and replayed `notifications/progress` carry the ORIGINAL + // request's token — see the `replayMessageId` JSDoc: only + // responses are remapped. this._startOrAuthSse({ resumptionToken, + onresumptiontoken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, - requestSignal: options?.requestSignal - }).catch(error => this.onerror?.(error)); + requestSignal: options?.requestSignal, + onRequestStreamEnd: options?.onRequestStreamEnd + }).catch(() => { + // Swallow the rethrow: `_startOrAuthSse`'s own catch already + // surfaced genuine failures via `onerror` before rethrowing + // (and deliberately stayed silent on an intentional abort — + // transport close or a settled request's `requestSignal`). + // Reporting here would double-fire `onerror` for real + // failures and turn a clean per-request teardown into a + // spurious `AbortError`. + // + // A genuine open failure IS terminal for the resumed + // stream (an initial-open failure never enters the + // reconnect loop) and this send() already resolved + // fire-and-forget — fire the stream-end callback so the + // caller can settle, mirroring `_scheduleReconnection`'s + // maxRetries-exhaustion branch. Not on intentional aborts: + // the contract excludes deliberate teardown. + if (options?.requestSignal?.aborted !== true && this._abortController?.signal.aborted !== true) { + // Guarded: this catch is a floating promise chain, so + // a throwing callback would otherwise become an + // unhandledRejection. + this._fireRequestStreamEnd(options?.onRequestStreamEnd); + } + }); return; } @@ -1111,8 +1356,14 @@ 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. + // Swallow the rethrow: `_startOrAuthSse`'s own catch already + // surfaced genuine failures via `onerror` before rethrowing + // (and deliberately stayed silent on an intentional abort — + // transport close). Reporting here would double-fire + // `onerror` for real failures and turn a clean shutdown + // into a spurious `AbortError`. + this._startOrAuthSse({ resumptionToken: undefined }).catch(() => {}); } return; } @@ -1136,7 +1387,11 @@ export class StreamableHTTPClientTransport implements Transport { { onresumptiontoken, requestSignal: options?.requestSignal, - onRequestStreamEnd: options?.onRequestStreamEnd + onRequestStreamEnd: options?.onRequestStreamEnd, + // Reuse the POST's composed signal for every + // reconnect leg of this response stream (see + // `SseLegOptions`). + fetchSignal: signal }, false ); @@ -1161,13 +1416,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 - // `_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) { + // Intentional abort BEFORE response headers — a per-request abort + // (the `subscriptions/listen` driver aborting its `requestSignal`) + // or a transport-wide close() landing mid-POST: fetch rejects with + // AbortError. Same guard as `_handleSseStream`'s + // `isIntentionalAbort` (BOTH signal halves) — do not surface a + // misleading onerror; still rethrow so `listen()`'s send-catch and + // the protocol layer settle their state machines. + if (options?.requestSignal?.aborted !== true && this._abortController?.signal.aborted !== true) { this.onerror?.(error as Error); } throw error; @@ -1222,7 +1478,14 @@ export class StreamableHTTPClientTransport implements Transport { this._sessionId = undefined; } catch (error) { - this.onerror?.(error as Error); + // Same guard as the POST path: the DELETE runs on the + // transport-lifetime signal alone, so close() landing mid-flight + // (or terminateSession() called after close()) rejects with an + // intentional AbortError — a clean shutdown, not a transport + // error. Still rethrow so the caller sees the failure. + if (this._abortController?.signal.aborted !== true) { + this.onerror?.(error as Error); + } throw error; } } diff --git a/packages/client/src/client/versionNegotiation.ts b/packages/client/src/client/versionNegotiation.ts index 9b89d962ba..b4687a253f 100644 --- a/packages/client/src/client/versionNegotiation.ts +++ b/packages/client/src/client/versionNegotiation.ts @@ -260,6 +260,17 @@ class ProbeWindow { const id = `server-discover-probe-${++this._probeCounter}`; return new Promise(resolve => { let settled = false; + // Request-scoped abort for this probe exchange, mirroring the + // signal the protocol layer threads for every real request: + // settling the probe — the timeout settle included — must tear + // its POST (and, over Streamable HTTP, any primed per-request + // SSE reconnect chain) down on the wire, not merely stop + // listening. Otherwise a timed-out probe under + // `probe.maxRetries >= 1` leaves an unabortable stale exchange + // whose late response leaks into the live session — the #2615 + // failure shape, one layer down. No-op on transports that ignore + // `requestSignal` (stdio). + const exchangeAbort = new AbortController(); const settle = (reply: RawProbeReply) => { if (settled) return; settled = true; @@ -267,11 +278,14 @@ class ProbeWindow { if (this._pending?.id === id) { this._pending = undefined; } + exchangeAbort.abort(); resolve(reply); }; const timer = setTimeout(() => settle({ kind: 'timeout' }), timeoutMs); this._pending = { id, resolve: settle }; - this._transport.send(buildRequest(id)).catch((error: unknown) => settle({ kind: 'send-error', error })); + this._transport + .send(buildRequest(id), { requestSignal: exchangeAbort.signal }) + .catch((error: unknown) => settle({ kind: 'send-error', error })); }); } diff --git a/packages/client/test/client/sse.test.ts b/packages/client/test/client/sse.test.ts index a0d4e7b6f9..2be88d4db7 100644 --- a/packages/client/test/client/sse.test.ts +++ b/packages/client/test/client/sse.test.ts @@ -246,6 +246,83 @@ describe('SSEClientTransport', () => { await expect(transport.send(testMessage)).rejects.toThrow(/500/); }); + + it('close() while a POST is in flight surfaces no spurious onerror', async () => { + // Create a server whose POST endpoint never responds, so the send + // stays in flight until the transport-lifetime signal aborts it. + await resourceServer.close(); + + resourceServer = createServer((req, res) => { + if (req.method === 'GET') { + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive' + }); + res.write('event: endpoint\n'); + res.write(`data: ${resourceBaseUrl.href}\n\n`); + } + // POST: never respond. + }); + + resourceBaseUrl = await listenOnRandomPort(resourceServer); + + transport = new SSEClientTransport(resourceBaseUrl); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + await transport.start(); + + let sendError: unknown; + const pending = transport.send({ jsonrpc: '2.0', id: 'test-1', method: 'test', params: {} }).catch(error => { + sendError = error; + }); + await new Promise(resolve => setTimeout(resolve, 50)); + + // ACT — deliberate shutdown while the POST is in flight. + await transport.close(); + await pending; + + // ASSERT — the rethrow is kept (send() rejects so callers + // settle), but no onerror fires for a deliberate shutdown. + expect((sendError as Error).name).toBe('AbortError'); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('close() still fires onclose when the EventSource teardown throws', async () => { + // onclose is the ONLY trigger for Protocol._onclose (which settles + // every pending request with ConnectionClosed) — a throwing + // eventSource.close() must not strand the protocol layer. + transport = new SSEClientTransport(resourceBaseUrl); + const onclose = vi.fn(); + transport.onclose = onclose; + transport['_eventSource'] = { + close: () => { + throw new Error('es close failed'); + } + } as unknown as (typeof transport)['_eventSource']; + + await expect(transport.close()).rejects.toThrow('es close failed'); + expect(onclose).toHaveBeenCalledTimes(1); + + // Reset the fake so the afterEach close() doesn't throw again. + transport['_eventSource'] = undefined; + }); + + it('keeps a single transport-lifetime AbortController across _startOrAuth invocations', async () => { + // _startOrAuth also runs on the mid-session 401 recovery path; + // replacing the controller there would orphan the signal captured + // by any POST already in flight — close() could no longer cancel + // it, and _send's intentional-abort guard would consult the wrong + // controller. + transport = new SSEClientTransport(resourceBaseUrl); + await transport.start(); + const controller = transport['_abortController']; + expect(controller).toBeDefined(); + + // Simulate the 401 recovery path re-invoking _startOrAuth. + await transport['_startOrAuth'](); + expect(transport['_abortController']).toBe(controller); + }); }); describe('header handling', () => { @@ -1685,6 +1762,106 @@ describe('SSEClientTransport', () => { expect(getAttempt).toBe(3); }); + it('close() during a pending onUnauthorized does not resurrect the EventSource', async () => { + // Regression: the 401-recovery success continuation ran + // _startOrAuth() after an arbitrarily long onUnauthorized await + // with no closed-state check. A close() landing while the token + // refresh was pending let _startOrAuth open a brand-new + // EventSource that nothing could ever tear down — the ES wrapper + // fetch never carries the transport-lifetime signal, and close() + // had already run against the old instance. + await resourceServer.close(); + + let getAttempts = 0; + resourceServer = createServer((req, res) => { + if (req.method === 'GET') { + getAttempts++; + res.writeHead(401).end(); + } + }); + resourceBaseUrl = await listenOnRandomPort(resourceServer); + + let releaseRefresh!: () => void; + const refreshPending = new Promise(resolve => { + releaseRefresh = resolve; + }); + let refreshStarted!: () => void; + const refreshStartedPromise = new Promise(resolve => { + refreshStarted = resolve; + }); + const authProvider: AuthProvider = { + token: vi.fn(async () => 'token'), + onUnauthorized: vi.fn(async () => { + refreshStarted(); + await refreshPending; + }) + }; + transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); + + const startPromise = transport.start(); + const startRejection = expect(startPromise).rejects.toThrow('Transport closed during re-authentication'); + // Wait until the 401 recovery is mid-refresh, then close the + // transport while onUnauthorized is still pending. + await refreshStartedPromise; + const attemptsAtClose = getAttempts; + await transport.close(); + + // The refresh resolves AFTER close(): the continuation must bail + // instead of opening a new EventSource. + releaseRefresh(); + await startRejection; + // Give a resurrected EventSource ample time to hit the server. + await new Promise(resolve => setTimeout(resolve, 100)); + expect(getAttempts).toBe(attemptsAtClose); + }); + + it('a refresh that rejects after close() does not surface onerror (deliberate teardown)', async () => { + // Regression: the FAILURE arm of the 401-recovery continuation + // called this.onerror unconditionally, while the success arm got + // a closed-state guard — so a token refresh that rejects after a + // clean close() (endpoint unreachable at shutdown, abandoned + // interactive flow) surfaced a spurious auth error through + // onerror arbitrarily long after the transport was torn down. + await resourceServer.close(); + + resourceServer = createServer((req, res) => { + if (req.method === 'GET') { + res.writeHead(401).end(); + } + }); + resourceBaseUrl = await listenOnRandomPort(resourceServer); + + let rejectRefresh!: (error: Error) => void; + const refreshPending = new Promise((_resolve, reject) => { + rejectRefresh = reject; + }); + let refreshStarted!: () => void; + const refreshStartedPromise = new Promise(resolve => { + refreshStarted = resolve; + }); + const authProvider: AuthProvider = { + token: vi.fn(async () => 'token'), + onUnauthorized: vi.fn(async () => { + refreshStarted(); + await refreshPending; + }) + }; + transport = new SSEClientTransport(resourceBaseUrl, { authProvider }); + const onerror = vi.fn(); + transport.onerror = onerror; + + const startPromise = transport.start(); + const startRejection = expect(startPromise).rejects.toThrow('refresh failed'); + await refreshStartedPromise; + await transport.close(); + + // The refresh rejects AFTER close(): still settles start(), but + // must not report post-shutdown noise through onerror. + rejectRefresh(new Error('refresh failed')); + await startRejection; + expect(onerror).not.toHaveBeenCalled(); + }); + it('retry failure during SSE connect fires onerror exactly once', async () => { // Regression: when the retry EventSource rejected, its onerror fired inside, then // the outer .then() rejection handler fired onerror AGAIN for the same error. diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..bf6fb0b298 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1,9 +1,12 @@ +import { getEventListeners } from 'node:events'; + import type { JSONRPCMessage, JSONRPCRequest, OAuthTokens } from '@modelcontextprotocol/core-internal'; import { OAuthError, OAuthErrorCode, SdkErrorCode, SdkHttpError } from '@modelcontextprotocol/core-internal'; import type { Mock, Mocked } from 'vitest'; import type { OAuthClientProvider } from '../../src/client/auth'; import { UnauthorizedError } from '../../src/client/auth'; +import { Client } from '../../src/client/client'; import type { ReconnectionScheduler, StartSSEOptions, StreamableHTTPReconnectionOptions } from '../../src/client/streamableHttp'; import { StreamableHTTPClientTransport } from '../../src/client/streamableHttp'; @@ -1464,263 +1467,807 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock).toHaveBeenCalledTimes(1); }); - it('onRequestStreamEnd fires when the per-request POST stream ends gracefully without reconnecting', async () => { - // ARRANGE — a POST stream with NO priming event id (so the - // graceful-close path does NOT schedule a reconnect): the - // per-request stream simply ends. - transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); - let streamController!: ReadableStreamDefaultController; - const unprimedStream = new ReadableStream({ + it('per-request requestSignal abort while a reconnect is scheduled: the pending reconnect never fires (#2615)', async () => { + // ARRANGE — a POST stream that is primed (SSE event id) and then + // closes gracefully WITHOUT delivering the response, so the + // transport schedules a GET+Last-Event-ID reconnect. The abort + // lands in the window between "reconnect scheduled" and "reconnect + // fires" — the shape a request timeout produces. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 5, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const primedClosingStream = new ReadableStream({ start(controller) { - streamController = controller; - // An ack frame with no SSE event id — does NOT arm POST-stream resumability. - controller.enqueue( - new TextEncoder().encode( - 'data: {"jsonrpc":"2.0","method":"notifications/subscriptions/acknowledged","params":{}}\n\n' - ) - ); + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + controller.close(); } }); const fetchMock = globalThis.fetch as Mock; - fetchMock.mockImplementationOnce(() => - Promise.resolve({ - ok: true, - status: 200, - headers: new Headers({ 'content-type': 'text/event-stream' }), - body: unprimedStream - }) - ); + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: primedClosingStream + }); const requestAbort = new AbortController(); - const onStreamEnd = vi.fn(); await transport.start(); await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { requestSignal: requestAbort.signal } ); + // Let the stream close and the reconnect get scheduled (delay 10ms). await vi.advanceTimersByTimeAsync(5); - expect(onStreamEnd).not.toHaveBeenCalled(); + expect(fetchMock).toHaveBeenCalledTimes(1); - // ACT — server gracefully closes the SSE stream. - streamController.close(); - await vi.advanceTimersByTimeAsync(5); + // ACT — the request settles (timeout/cancel) before the reconnect fires. + requestAbort.abort(); + await vi.advanceTimersByTimeAsync(100); - // ASSERT — non-deliberate stream end without reconnecting: - // onRequestStreamEnd fired exactly once; no further fetches. - expect(onStreamEnd).toHaveBeenCalledTimes(1); + // ASSERT — the scheduled reconnect saw the aborted requestSignal + // and bailed: no GET resume, no onerror. expect(fetchMock).toHaveBeenCalledTimes(1); + expect(errorSpy).not.toHaveBeenCalled(); }); - it('onRequestStreamEnd does NOT fire on a deliberate per-request abort', async () => { - // Same shape as the no-onerror/no-reconnect test, but assert the - // stream-end callback is NEVER invoked when `requestSignal` was the - // abort source. + it('resumptionToken send: requestSignal abort while the resume GET is in flight surfaces no spurious onerror (#2615)', async () => { + // ARRANGE — a request re-issued with options.resumptionToken + // short-circuits send() into a GET+Last-Event-ID resume. The + // server hangs on that GET (the exact scenario request timeouts + // exist for), so the settlement abort lands MID-FETCH — not in + // the scheduled-reconnect window the sibling test covers. transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); - let streamController!: ReadableStreamDefaultController; - const stream = new ReadableStream({ - start(controller) { - streamController = controller; - } - }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const fetchMock = globalThis.fetch as Mock; - fetchMock.mockImplementationOnce((_url, init: RequestInit) => { - init.signal?.addEventListener('abort', () => streamController.error(init.signal?.reason), { once: true }); - return Promise.resolve({ - ok: true, - status: 200, - headers: new Headers({ 'content-type': 'text/event-stream' }), - body: stream - }); - }); + fetchMock.mockImplementation( + (_url, init: RequestInit) => + new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('The operation was aborted', 'AbortError'))); + }) + ); const requestAbort = new AbortController(); const onStreamEnd = vi.fn(); await transport.start(); await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { resumptionToken: 'evt-42', requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } ); await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(1); - // ACT — deliberate per-request abort. + // ACT — the request settles (timeout/cancel) while the resume GET + // is still in flight; the fetch rejects with an AbortError. requestAbort.abort(); - await vi.advanceTimersByTimeAsync(50); + await vi.advanceTimersByTimeAsync(100); - // ASSERT — deliberate abort: onRequestStreamEnd never fires. + // ASSERT — a deliberate per-request teardown is a clean shutdown: + // no transport error, and no stream-end callback (the contract + // excludes deliberate requestSignal aborts). + expect(errorSpy).not.toHaveBeenCalled(); expect(onStreamEnd).not.toHaveBeenCalled(); }); - it('onRequestStreamEnd fires when reconnection attempts are exhausted (maxRetries reached)', async () => { - // ARRANGE — a primed POST stream (so a non-deliberate close - // schedules a GET resume); every GET resume fails; maxRetries 1 - // means the second schedule hits the exhausted branch. + it('resumptionToken send: a genuine resume GET failure surfaces onerror exactly once (no double-report)', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const failure = new TypeError('fetch failed'); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockRejectedValueOnce(failure); + + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { resumptionToken: 'evt-42', onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + + // _startOrAuthSse's own catch reports the failure; the send() + // short-circuit must not report it a second time. + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith(failure); + + // A genuine open failure is TERMINAL for the resumed stream (no + // reconnect is ever scheduled for an initial-open failure) and + // send() already resolved fire-and-forget — the stream-end + // callback is the caller's only per-request settlement signal. + expect(onStreamEnd).toHaveBeenCalledTimes(1); + }); + + it('a ReconnectionScheduler that throws on reschedule still settles the caller via onRequestStreamEnd', async () => { + let schedulerCalls = 0; + const scheduler: ReconnectionScheduler = reconnect => { + schedulerCalls++; + if (schedulerCalls === 1) { + const handle = setTimeout(reconnect, 1); + return () => clearTimeout(handle); + } + // The reschedule after a failed leg: platform denies the task. + throw new Error('platform denied background task'); + }; transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { reconnectionOptions: { - initialReconnectionDelay: 5, - maxRetries: 1, + initialReconnectionDelay: 10, + maxRetries: 5, maxReconnectionDelay: 1000, reconnectionDelayGrowFactor: 1 - } + }, + reconnectionScheduler: scheduler }); const errorSpy = vi.fn(); transport.onerror = errorSpy; - let streamController!: ReadableStreamDefaultController; - const primedStream = new ReadableStream({ - start(controller) { - streamController = controller; - controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); - } - }); const fetchMock = globalThis.fetch as Mock; + // POST: primed SSE stream, graceful close without a response. fetchMock.mockResolvedValueOnce({ ok: true, status: 200, headers: new Headers({ 'content-type': 'text/event-stream' }), - body: primedStream + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + controller.close(); + } + }) }); - // The GET resume fails with a 5xx → reconnect catch reschedules → exhausted. - fetchMock.mockResolvedValue({ ok: false, status: 503, statusText: 'unavailable', headers: new Headers() }); + // The reconnect leg fails genuinely, forcing a reschedule. + fetchMock.mockRejectedValueOnce(new TypeError('fetch failed')); const onStreamEnd = vi.fn(); await transport.start(); await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { onRequestStreamEnd: onStreamEnd } ); - await vi.advanceTimersByTimeAsync(5); - expect(onStreamEnd).not.toHaveBeenCalled(); - - // ACT — server closes the primed POST stream non-deliberately. - streamController.close(); - await vi.advanceTimersByTimeAsync(100); + await vi.advanceTimersByTimeAsync(10); - // ASSERT — exhausted: onRequestStreamEnd fired exactly once (the - // max-retries branch); the exhausted onerror surfaced. + // The failed leg reported once, the scheduler throw reported once, + // and — because no further attempt could be armed — the chain is + // terminally dead, so the caller was settled. + expect(schedulerCalls).toBe(2); + expect(errorSpy).toHaveBeenCalledWith(new TypeError('fetch failed')); + expect(errorSpy).toHaveBeenCalledWith(new Error('platform denied background task')); expect(onStreamEnd).toHaveBeenCalledTimes(1); - expect(errorSpy).toHaveBeenCalledWith( - expect.objectContaining({ message: expect.stringContaining('Maximum reconnection attempts') }) - ); }); - it('onRequestStreamEnd fires when the per-request POST stream ERRORS without reconnecting', async () => { - // ARRANGE — a POST stream with NO priming event id; the body - // errors (network drop). The error-branch `else` (no reconnect, - // not intentional-abort) must fire onRequestStreamEnd. - transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); - const failingStream = new ReadableStream({ - start(controller) { - controller.enqueue( - new TextEncoder().encode( - 'data: {"jsonrpc":"2.0","method":"notifications/subscriptions/acknowledged","params":{}}\n\n' - ) - ); - queueMicrotask(() => controller.error(new Error('network drop'))); + it('close() disarms EVERY pending scheduled reconnect, not just the last-scheduled chain', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10_000, + maxRetries: 2, + maxReconnectionDelay: 30_000, + reconnectionDelayGrowFactor: 1 } }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + // Every POST returns a primed SSE stream that closes gracefully + // without a response — each send spawns its own reconnect chain. const fetchMock = globalThis.fetch as Mock; - fetchMock.mockResolvedValueOnce({ + fetchMock.mockImplementation(async () => ({ ok: true, status: 200, headers: new Headers({ 'content-type': 'text/event-stream' }), - body: failingStream - }); + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + controller.close(); + } + }) + })); - const onStreamEnd = vi.fn(); await transport.start(); - await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } - ); - await vi.advanceTimersByTimeAsync(50); + await transport.send({ jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }); + await transport.send({ jsonrpc: '2.0', method: 'long_running_tool', id: 'request-2', params: {} }); + await vi.advanceTimersByTimeAsync(5); - // ASSERT — error-branch fired exactly once; no reconnection - // attempted (POST stream wasn't primed). - expect(onStreamEnd).toHaveBeenCalledTimes(1); - expect(fetchMock).toHaveBeenCalledTimes(1); + // Two chains, two pending scheduled attempts. + expect(vi.getTimerCount()).toBe(2); + + // ACT — close must disarm BOTH, not only the last-written one. + await transport.close(); + + expect(vi.getTimerCount()).toBe(0); + expect(errorSpy).not.toHaveBeenCalled(); }); - it('onRequestStreamEnd does NOT fire on transport.close()', async () => { - // The transport-wide abort is the OTHER deliberate teardown - // (`isIntentionalAbort()` checks both signals): a per-request - // stream-end callback must not fire when close() tore the stream - // down — `_onclose` is the settle path for that. - transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); - let streamController!: ReadableStreamDefaultController; - const stream = new ReadableStream({ - start(controller) { - streamController = controller; - controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + it('a settled request disarms its own pending scheduled reconnect immediately', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10_000, + maxRetries: 2, + maxReconnectionDelay: 30_000, + reconnectionDelayGrowFactor: 1 } }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const fetchMock = globalThis.fetch as Mock; - fetchMock.mockImplementationOnce((_url, init: RequestInit) => { - init.signal?.addEventListener('abort', () => streamController.error(init.signal?.reason), { once: true }); - return Promise.resolve({ - ok: true, - status: 200, - headers: new Headers({ 'content-type': 'text/event-stream' }), - body: stream - }); + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + controller.close(); + } + }) }); - const onStreamEnd = vi.fn(); + const requestAbort = new AbortController(); await transport.start(); await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { requestSignal: requestAbort.signal } ); await vi.advanceTimersByTimeAsync(5); + expect(vi.getTimerCount()).toBe(1); - // ACT — transport-wide close. - await transport.close(); - await vi.advanceTimersByTimeAsync(50); + // ACT — the request settles during the backoff window: the armed + // timer is released immediately, not left to bail at fire time. + requestAbort.abort(); + expect(vi.getTimerCount()).toBe(0); - // ASSERT — deliberate transport close: onRequestStreamEnd never fires. - expect(onStreamEnd).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(60_000); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(errorSpy).not.toHaveBeenCalled(); }); - it('onRequestStreamEnd fires when a primed POST→GET resume hits 405 (non-resumable terminal)', async () => { - // R1 regression: against a server that stamps SSE event ids on the - // listen POST stream but returns 405 on the GET resume, - // `_startOrAuthSse` resolved without a stream and nothing fired — - // the subscription dead-ended silently. The 405 is now a terminal - // per-request stream-end. ALSO asserts the GET resume carried the - // per-request `requestSignal` (the close-after-reconnect path). + it('resume leg that drops before its first event reschedules with the ORIGINAL resumption token', async () => { transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { reconnectionOptions: { - initialReconnectionDelay: 5, - maxRetries: 3, + initialReconnectionDelay: 10, + maxRetries: 2, maxReconnectionDelay: 1000, reconnectionDelayGrowFactor: 1 } }); - let streamController!: ReadableStreamDefaultController; - const primedStream = new ReadableStream({ - start(controller) { - streamController = controller; - controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); - } - }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + const fetchMock = globalThis.fetch as Mock; - let getSignal: AbortSignal | null | undefined; + // GET#1 (the resume): closes gracefully with ZERO events, so the + // leg produces no lastEventId of its own. fetchMock.mockResolvedValueOnce({ ok: true, status: 200, headers: new Headers({ 'content-type': 'text/event-stream' }), - body: primedStream - }); - fetchMock.mockImplementationOnce((_url, init: RequestInit) => { - getSignal = init.signal; - return Promise.resolve({ ok: false, status: 405, headers: new Headers() }); + body: new ReadableStream({ + start(controller) { + controller.close(); + } + }) }); + // GET#2 (the rebuilt leg): hangs, keeping the count deterministic. + fetchMock.mockImplementation(() => new Promise(() => {})); - const requestAbort = new AbortController(); - const onStreamEnd = vi.fn(); await transport.start(); await transport.send( - { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, - { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { resumptionToken: 'evt-42' } + ); + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // Fire the scheduled rebuild. + await vi.advanceTimersByTimeAsync(15); + expect(fetchMock).toHaveBeenCalledTimes(2); + + // The rebuilt GET must still carry the token this resume was + // opened with — without the fallback it degrades into a + // token-less standalone GET and loses the replay position. + const secondInit = fetchMock.mock.calls[1]?.[1] as RequestInit; + expect((secondInit.headers as Headers).get('last-event-id')).toBe('evt-42'); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('resumptionToken send: forwards onresumptiontoken so the resumed stream reports newer event IDs', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + // The resumed GET replays a newer event carrying the response. + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('id: evt-43\ndata: {"jsonrpc":"2.0","id":"request-1","result":{}}\n\n') + ); + controller.close(); + } + }) + }); + + const tokens: string[] = []; + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { resumptionToken: 'evt-42', onresumptiontoken: token => tokens.push(token) } + ); + await vi.advanceTimersByTimeAsync(5); + + // The caller's persistence hook saw the newer event ID. + expect(tokens).toEqual(['evt-43']); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('resumptionToken send: forwards onRequestStreamEnd so a terminal non-resumable outcome settles the caller', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + // 405 on the resume GET: terminal, non-resumable — the stream-end + // callback must fire so the caller can settle. + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 405, + statusText: 'Method Not Allowed', + headers: new Headers(), + text: async () => '' + }); + + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }, + { resumptionToken: 'evt-42', onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + + expect(onStreamEnd).toHaveBeenCalledTimes(1); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('202-initialized standalone GET: a genuine failure surfaces onerror exactly once (no double-report)', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const failure = new TypeError('fetch failed'); + const fetchMock = globalThis.fetch as Mock; + // The notifications/initialized POST is 202-accepted... + fetchMock.mockResolvedValueOnce({ ok: true, status: 202, headers: new Headers(), text: async () => '' }); + // ...and the standalone GET it triggers fails genuinely. + fetchMock.mockRejectedValueOnce(failure); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(2); + + // _startOrAuthSse's own catch reports the failure; the 202 branch + // must not report it a second time. + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith(failure); + }); + + it('202-initialized standalone GET: transport.close() while the GET is in flight surfaces no spurious onerror', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ ok: true, status: 202, headers: new Headers(), text: async () => '' }); + // The standalone GET hangs until its signal aborts. + fetchMock.mockImplementationOnce( + (_url, init: RequestInit) => + new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('The operation was aborted', 'AbortError'))); + }) + ); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(2); + + // ACT — deliberate shutdown while the standalone GET is in flight. + await transport.close(); + await vi.advanceTimersByTimeAsync(5); + + // ASSERT — a clean shutdown, not a transport error. + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('notification POST: transport.close() while the POST is in flight surfaces no spurious onerror', async () => { + // Notification sends carry no requestSignal, so the POST's fetch + // signal is the transport-lifetime signal alone — close() landing + // mid-POST must read as a clean shutdown, not a transport error. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementationOnce( + (_url, init: RequestInit) => + new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('The operation was aborted', 'AbortError'))); + }) + ); + + await transport.start(); + let sendError: unknown; + const pending = transport.send({ jsonrpc: '2.0', method: 'notifications/roots/list_changed' }).catch(error => { + sendError = error; + }); + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // ACT — deliberate shutdown while the POST is in flight. + await transport.close(); + await pending; + + // ASSERT — the rethrow is kept (send() rejects so callers settle), + // but no onerror fires for a deliberate shutdown. + expect(sendError).toBeInstanceOf(DOMException); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('terminateSession: transport.close() while the DELETE is in flight surfaces no spurious onerror', async () => { + // The session-termination DELETE runs on the transport-lifetime + // signal alone — close() landing mid-flight must read as a clean + // shutdown, not a transport error. (A sessionId is required or + // terminateSession() no-ops without fetching.) + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { sessionId: 'session-1' }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementationOnce( + (_url, init: RequestInit) => + new Promise((_resolve, reject) => { + init.signal?.addEventListener('abort', () => reject(new DOMException('The operation was aborted', 'AbortError'))); + }) + ); + + await transport.start(); + let terminateError: unknown; + const pending = transport.terminateSession().catch(error => { + terminateError = error; + }); + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // ACT — deliberate shutdown while the DELETE is in flight. + await transport.close(); + await pending; + + // ASSERT — the rethrow is kept (terminateSession() rejects so the + // caller sees the outcome), but no onerror fires. + expect(terminateError).toBeInstanceOf(DOMException); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('failed reconnect leg surfaces onerror exactly once (no double-report), then retries', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 5, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const failure = new TypeError('fetch failed'); + const fetchMock = globalThis.fetch as Mock; + // POST stream: primed (SSE event id), then a graceful close + // without the response — schedules a GET reconnect at +10ms. + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + controller.close(); + } + }) + }); + // The reconnect GET fails genuinely. + fetchMock.mockRejectedValueOnce(failure); + // The retried leg after it hangs, so the count stays deterministic. + fetchMock.mockImplementation(() => new Promise(() => {})); + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'long_running_tool', id: 'request-1', params: {} }); + // Let the stream close and the reconnect get scheduled... + await vi.advanceTimersByTimeAsync(5); + expect(fetchMock).toHaveBeenCalledTimes(1); + // ...then fire the reconnect leg and let it fail. + await vi.advanceTimersByTimeAsync(10); + expect(fetchMock).toHaveBeenCalledTimes(2); + + // _startOrAuthSse's own catch is the single reporting site — the + // reconnect scheduler must not report the same failure again. + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith(failure); + + // The retry itself still happens (attempt 2 fires the next GET). + await vi.advanceTimersByTimeAsync(15); + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(errorSpy).toHaveBeenCalledTimes(1); + }); + + it('onRequestStreamEnd fires when the per-request POST stream ends gracefully without reconnecting', async () => { + // ARRANGE — a POST stream with NO priming event id (so the + // graceful-close path does NOT schedule a reconnect): the + // per-request stream simply ends. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + let streamController!: ReadableStreamDefaultController; + const unprimedStream = new ReadableStream({ + start(controller) { + streamController = controller; + // An ack frame with no SSE event id — does NOT arm POST-stream resumability. + controller.enqueue( + new TextEncoder().encode( + 'data: {"jsonrpc":"2.0","method":"notifications/subscriptions/acknowledged","params":{}}\n\n' + ) + ); + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementationOnce(() => + Promise.resolve({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: unprimedStream + }) + ); + + const requestAbort = new AbortController(); + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + expect(onStreamEnd).not.toHaveBeenCalled(); + + // ACT — server gracefully closes the SSE stream. + streamController.close(); + await vi.advanceTimersByTimeAsync(5); + + // ASSERT — non-deliberate stream end without reconnecting: + // onRequestStreamEnd fired exactly once; no further fetches. + expect(onStreamEnd).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('onRequestStreamEnd does NOT fire on a deliberate per-request abort', async () => { + // Same shape as the no-onerror/no-reconnect test, but assert the + // stream-end callback is NEVER invoked when `requestSignal` was the + // abort source. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + let streamController!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(controller) { + streamController = controller; + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementationOnce((_url, init: RequestInit) => { + init.signal?.addEventListener('abort', () => streamController.error(init.signal?.reason), { once: true }); + return Promise.resolve({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: stream + }); + }); + + const requestAbort = new AbortController(); + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + + // ACT — deliberate per-request abort. + requestAbort.abort(); + await vi.advanceTimersByTimeAsync(50); + + // ASSERT — deliberate abort: onRequestStreamEnd never fires. + expect(onStreamEnd).not.toHaveBeenCalled(); + }); + + it('onRequestStreamEnd fires when reconnection attempts are exhausted (maxRetries reached)', async () => { + // ARRANGE — a primed POST stream (so a non-deliberate close + // schedules a GET resume); every GET resume fails; maxRetries 1 + // means the second schedule hits the exhausted branch. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 5, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + let streamController!: ReadableStreamDefaultController; + const primedStream = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: primedStream + }); + // The GET resume fails with a 5xx → reconnect catch reschedules → exhausted. + fetchMock.mockResolvedValue({ ok: false, status: 503, statusText: 'unavailable', headers: new Headers() }); + + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + expect(onStreamEnd).not.toHaveBeenCalled(); + + // ACT — server closes the primed POST stream non-deliberately. + streamController.close(); + await vi.advanceTimersByTimeAsync(100); + + // ASSERT — exhausted: onRequestStreamEnd fired exactly once (the + // max-retries branch); the exhausted onerror surfaced. + expect(onStreamEnd).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining('Maximum reconnection attempts') }) + ); + }); + + it('onRequestStreamEnd fires when the per-request POST stream ERRORS without reconnecting', async () => { + // ARRANGE — a POST stream with NO priming event id; the body + // errors (network drop). The error-branch `else` (no reconnect, + // not intentional-abort) must fire onRequestStreamEnd. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + const failingStream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"jsonrpc":"2.0","method":"notifications/subscriptions/acknowledged","params":{}}\n\n' + ) + ); + queueMicrotask(() => controller.error(new Error('network drop'))); + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: failingStream + }); + + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(50); + + // ASSERT — error-branch fired exactly once; no reconnection + // attempted (POST stream wasn't primed). + expect(onStreamEnd).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('onRequestStreamEnd does NOT fire on transport.close()', async () => { + // The transport-wide abort is the OTHER deliberate teardown + // (`isIntentionalAbort()` checks both signals): a per-request + // stream-end callback must not fire when close() tore the stream + // down — `_onclose` is the settle path for that. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp')); + let streamController!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + } + }); + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementationOnce((_url, init: RequestInit) => { + init.signal?.addEventListener('abort', () => streamController.error(init.signal?.reason), { once: true }); + return Promise.resolve({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: stream + }); + }); + + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: new AbortController().signal, onRequestStreamEnd: onStreamEnd } + ); + await vi.advanceTimersByTimeAsync(5); + + // ACT — transport-wide close. + await transport.close(); + await vi.advanceTimersByTimeAsync(50); + + // ASSERT — deliberate transport close: onRequestStreamEnd never fires. + expect(onStreamEnd).not.toHaveBeenCalled(); + }); + + it('onRequestStreamEnd fires when a primed POST→GET resume hits 405 (non-resumable terminal)', async () => { + // R1 regression: against a server that stamps SSE event ids on the + // listen POST stream but returns 405 on the GET resume, + // `_startOrAuthSse` resolved without a stream and nothing fired — + // the subscription dead-ended silently. The 405 is now a terminal + // per-request stream-end. ALSO asserts the GET resume carried the + // per-request `requestSignal` (the close-after-reconnect path). + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 5, + maxRetries: 3, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + let streamController!: ReadableStreamDefaultController; + const primedStream = new ReadableStream({ + start(controller) { + streamController = controller; + controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n')); + } + }); + const fetchMock = globalThis.fetch as Mock; + let getSignal: AbortSignal | null | undefined; + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: primedStream + }); + fetchMock.mockImplementationOnce((_url, init: RequestInit) => { + getSignal = init.signal; + return Promise.resolve({ ok: false, status: 405, headers: new Headers() }); + }); + + const requestAbort = new AbortController(); + const onStreamEnd = vi.fn(); + await transport.start(); + await transport.send( + { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen:0', params: {} }, + { requestSignal: requestAbort.signal, onRequestStreamEnd: onStreamEnd } ); await vi.advanceTimersByTimeAsync(5); @@ -2497,7 +3044,7 @@ describe('StreamableHTTPClientTransport', () => { ); // Verify no reconnection was scheduled - expect(transport['_cancelReconnection']).toBeUndefined(); + expect(transport['_pendingReconnectCancels'].size).toBe(0); }); it('should schedule reconnection when maxRetries is greater than 0', async () => { @@ -2519,10 +3066,12 @@ describe('StreamableHTTPClientTransport', () => { // ASSERT - should schedule a reconnection, not report error yet expect(errorSpy).not.toHaveBeenCalled(); - expect(transport['_cancelReconnection']).toBeDefined(); + expect(transport['_pendingReconnectCancels'].size).toBe(1); // Clean up the pending reconnection to avoid test pollution - transport['_cancelReconnection']?.(); + for (const entry of transport['_pendingReconnectCancels']) { + entry.cancel(); + } }); }); @@ -2735,5 +3284,661 @@ describe('StreamableHTTPClientTransport', () => { expect(abortController?.signal.aborted).toBe(true); expect(onclose).toHaveBeenCalledTimes(1); }); + + it('a throwing cancel does not skip cancelling sibling chains on close()', async () => { + const cancel1 = vi.fn(() => { + throw new Error('cancel 1 failed'); + }); + const cancel2 = vi.fn(); + const cancels = [cancel1, cancel2]; + let scheduleCount = 0; + const scheduler: ReconnectionScheduler = vi.fn(() => cancels[scheduleCount++]); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + const onclose = vi.fn(); + transport.onclose = onclose; + + await transport.start(); + // Two independent pending chains, each with its own cancel. + triggerReconnection(transport); + triggerReconnection(transport); + const abortController = transport['_abortController']; + + // The FIRST cancel error still propagates, but only after every + // sibling chain has been disarmed too. + await expect(transport.close()).rejects.toThrow('cancel 1 failed'); + expect(cancel1).toHaveBeenCalledTimes(1); + expect(cancel2).toHaveBeenCalledTimes(1); + expect(abortController?.signal.aborted).toBe(true); + expect(onclose).toHaveBeenCalledTimes(1); + expect(transport['_pendingReconnectCancels'].size).toBe(0); + }); + + it("runs a pending chain's cancel at most once across close() and request settlement", async () => { + const cancel = vi.fn(); + const scheduler: ReconnectionScheduler = vi.fn(() => cancel); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + + await transport.start(); + // A per-request chain in its backoff window: the settlement + // listener is armed on the request-scoped signal. + const requestAbort = new AbortController(); + (transport as unknown as { _scheduleReconnection(opts: StartSSEOptions, attempt?: number): void })._scheduleReconnection( + { requestSignal: requestAbort.signal }, + 0 + ); + + await transport.close(); + expect(cancel).toHaveBeenCalledTimes(1); + + // close() -> onclose -> Protocol._onclose rejects the pending + // request -> its .finally() aborts the request-scoped signal. + // The chain's settlement listener must have been released by + // close(), so the cancel does NOT run a second time. + requestAbort.abort(); + expect(cancel).toHaveBeenCalledTimes(1); + }); + + it('a throwing cancel at request settlement reports through onerror instead of escaping the abort dispatch', async () => { + // Regression: the settlement listener invoked the user-supplied + // cancel with try/finally but no catch. An exception thrown in an + // AbortSignal 'abort' listener is NOT delivered to the abort() + // caller — abort() returns normally and Node reports the + // exception as an uncaughtException, terminating the process by + // default. The protocol funnel aborts requestSignal on EVERY + // settlement (success, timeout, caller abort, maxTotalTimeout), + // so a throwing custom-scheduler cancel would kill the process on + // the common path. + const cancelError = new Error('cancel failed at settlement'); + const scheduler: ReconnectionScheduler = vi.fn(() => () => { + throw cancelError; + }); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + const onerror = vi.fn(); + transport.onerror = onerror; + + await transport.start(); + const requestAbort = new AbortController(); + (transport as unknown as { _scheduleReconnection(opts: StartSSEOptions, attempt?: number): void })._scheduleReconnection( + { requestSignal: requestAbort.signal }, + 0 + ); + + // The request settles: the funnel aborts its request-scoped + // signal. The throwing cancel must be routed to onerror, not + // escape the EventTarget dispatch. + requestAbort.abort(); + + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledWith(cancelError); + // The finally disarm still ran: no stale Set entry survives. + expect(transport['_pendingReconnectCancels'].size).toBe(0); + }); + + it('a first-schedule scheduler throw on graceful close reports the raw error exactly once', async () => { + // Regression: the graceful-close settlement tail ran inside the + // same try as the stream read loop. A scheduler that throws on + // the FIRST schedule of a gap landed in the generic catch, got + // mislabeled "SSE stream disconnected", and the catch re-drove + // the tail: two onerror reports for one scheduler failure. + const scheduleError = new Error('platform denied background task'); + const scheduler: ReconnectionScheduler = vi.fn(() => { + throw scheduleError; + }); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const onRequestStreamEnd = vi.fn(); + await transport.start(); + + // A primed POST stream that closes gracefully without a response: + // the graceful tail schedules the first reconnect attempt. + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('id: evt-1\ndata: \n\n')); + controller.close(); + } + }); + transport['_handleSseStream'](stream, { onRequestStreamEnd }, false); + await vi.advanceTimersByTimeAsync(50); + + expect(scheduler).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledWith(scheduleError); + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + }); + + it('a first-schedule scheduler throw on the ERROR path reports the raw error, matching the graceful path', async () => { + // Regression: the error-path first-schedule catch wrapped the + // scheduler throw as `Failed to reconnect: `, discarding + // the original error object, while the graceful-close tail and + // the reschedule catch both report the raw error — inconsistent + // shape for the same failure class. + const scheduleError = new Error('platform denied background task'); + const scheduler: ReconnectionScheduler = vi.fn(() => { + throw scheduleError; + }); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const onRequestStreamEnd = vi.fn(); + await transport.start(); + + // A reconnectable stream that ERRORS mid-read: the catch path + // schedules the first reconnect attempt, and the scheduler throws. + const stream = new ReadableStream({ + start(controller) { + controller.error(new Error('network dropped')); + } + }); + transport['_handleSseStream'](stream, { onRequestStreamEnd }, true); + await vi.advanceTimersByTimeAsync(50); + + expect(scheduler).toHaveBeenCalledTimes(1); + // The stream error reports once ("SSE stream disconnected"), and + // the scheduler failure reports RAW — same identity, no wrapper. + expect(onerror.mock.calls.map(call => call[0])).toEqual(expect.arrayContaining([scheduleError])); + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + }); + + it('a throwing onRequestStreamEnd on graceful close is invoked once and never escapes processStream', async () => { + // Regression: with the settlement tail inside the read-loop try, + // a throwing caller-supplied onRequestStreamEnd was caught, the + // catch re-invoked the SAME callback, and its second throw + // escaped the fire-and-forget processStream() promise as an + // unhandledRejection. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const callbackError = new Error('stream end failed'); + const onRequestStreamEnd = vi.fn(() => { + throw callbackError; + }); + await transport.start(); + + // A POST stream with no priming event ends gracefully: the tail + // takes the no-reconnect branch and fires the stream-end callback. + const stream = new ReadableStream({ + start(controller) { + controller.close(); + } + }); + transport['_handleSseStream'](stream, { onRequestStreamEnd }, false); + await vi.advanceTimersByTimeAsync(50); + + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledWith(callbackError); + }); + + it('a throwing onRequestStreamEnd in the reschedule-throw branch routes to onerror, not an unhandledRejection', async () => { + // Regression: the reconnect closure's scheduleError branch fires + // the caller's stream-end callback inside a floating promise + // chain (`_startOrAuthSse(options).catch(...)`) — a throwing + // callback rejected that chain with nothing attached, + // terminating the process by default. + const scheduleError = new Error('platform denied background task'); + let scheduleCalls = 0; + let capturedReconnect: (() => void) | undefined; + const scheduler: ReconnectionScheduler = vi.fn(reconnect => { + scheduleCalls++; + if (scheduleCalls > 1) { + throw scheduleError; + } + capturedReconnect = reconnect; + return () => {}; + }); + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: scheduler + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const callbackError = new Error('stream end failed'); + const onRequestStreamEnd = vi.fn(() => { + throw callbackError; + }); + await transport.start(); + + // First schedule arms; the leg's fetch then fails, so the catch + // reschedules — and the second schedule throws. + (globalThis.fetch as Mock).mockRejectedValue(new Error('network down')); + (transport as unknown as { _scheduleReconnection(opts: StartSSEOptions, attempt?: number): void })._scheduleReconnection( + { onRequestStreamEnd }, + 0 + ); + capturedReconnect!(); + await vi.advanceTimersByTimeAsync(50); + + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + // onerror: the leg's open failure, the scheduler error, and the + // contained callback error — nothing escaped the floating chain. + expect(onerror.mock.calls.map(call => call[0])).toEqual(expect.arrayContaining([scheduleError, callbackError])); + }); + + it('a throwing onRequestStreamEnd in the resumptionToken short-circuit routes to onerror, not an unhandledRejection', async () => { + // Regression: the short-circuit's catch fires the stream-end + // callback inside a floating promise chain; a throwing callback + // became an unhandledRejection. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const callbackError = new Error('stream end failed'); + const onRequestStreamEnd = vi.fn(() => { + throw callbackError; + }); + await transport.start(); + + // A genuine open failure (not an abort) on the resumed GET fires + // the stream-end callback from the short-circuit's catch. + (globalThis.fetch as Mock).mockRejectedValue(new Error('network down')); + await transport.send({ jsonrpc: '2.0', method: 'ping', id: 'r-1' }, { resumptionToken: 'evt-0', onRequestStreamEnd }); + await vi.advanceTimersByTimeAsync(50); + + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + expect(onerror.mock.calls.map(call => call[0])).toEqual(expect.arrayContaining([callbackError])); + }); + + it('a throwing onresumptiontoken does not lose the event or misattribute a stream disconnect', async () => { + // Regression: the caller's persistence hook was invoked bare in + // the read loop. A throw (e.g. QuotaExceededError from the + // documented storage write) exited into the generic catch: a + // misattributed "SSE stream disconnected" onerror, the + // response-bearing event's payload lost unreplayably, and a + // reconnect chain re-entered at attempt 0 with the request never + // marked complete. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const onmessage = vi.fn(); + transport.onmessage = onmessage; + const quotaError = new Error('QuotaExceededError: storage full'); + const onresumptiontoken = vi.fn(() => { + throw quotaError; + }); + await transport.start(); + + // One event carrying BOTH the id (hook throws) and the response. + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('id: evt-1\ndata: {"jsonrpc":"2.0","id":"r-1","result":{}}\n\n')); + controller.close(); + } + }); + transport['_handleSseStream'](stream, { onresumptiontoken }, false); + await vi.advanceTimersByTimeAsync(50); + + // The hook's failure is reported once, correctly attributed… + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledWith(quotaError); + // …the response still dispatches… + expect(onmessage).toHaveBeenCalledTimes(1); + // …and no bogus reconnect chain starts (response received). + expect(transport['_pendingReconnectCancels'].size).toBe(0); + expect(globalThis.fetch).not.toHaveBeenCalled(); + }); + }); +}); + +/** + * End-to-end regression for #2615: on a legacy (2025-11-25) session, the + * transport's request-scoped SSE reconnect chain (GET + Last-Event-ID + * resumption) must stop once the originating request settles via timeout. + * Before the fix, the chain kept resuming forever (every successful resume + * resets the retry counter), and a late resumed GET carrying the original + * JSON-RPC response surfaced as "Received a response for an unknown message + * ID". + */ +describe('legacy era (2025-11-25): request timeout stops the SSE reconnect chain (#2615)', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(globalThis, 'fetch'); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.clearAllMocks(); + }); + + const encoder = new TextEncoder(); + const sseResponse = (chunks: string[]) => ({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(chunk)); + } + controller.close(); + } + }) + }); + const jsonResponse = (message: JSONRPCMessage) => ({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'application/json' }), + json: async () => message, + text: async () => JSON.stringify(message) + }); + const accepted = () => ({ ok: true, status: 202, headers: new Headers(), text: async () => '' }); + const methodNotAllowed = () => ({ + ok: false, + status: 405, + statusText: 'Method Not Allowed', + headers: new Headers(), + text: async () => '' + }); + + it('stops resuming once the request times out; the late response never surfaces as an unknown message ID', async () => { + let pingId: string | number | undefined; + let eventSeq = 0; + let settled = false; + let resumesAfterSettle = 0; + const cancelledPosts: JSONRPCMessage[] = []; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementation(async (_url, init: RequestInit) => { + if (init.method === 'GET') { + const lastEventId = (init.headers as Headers).get('last-event-id'); + // Standalone notification stream: not offered by this server. + if (lastEventId === null) { + return methodNotAllowed(); + } + // Request-scoped resume. Once the request has settled, hand + // back the late original response — before the fix this is + // the resumed GET that surfaced "unknown message ID". + if (settled) { + resumesAfterSettle++; + return sseResponse([`id: evt-${++eventSeq}\ndata: {"jsonrpc":"2.0","id":${JSON.stringify(pingId)},"result":{}}\n\n`]); + } + // Keep the chain alive: a priming event id, then a graceful + // close without the response (the server expects the client + // to resume via GET + Last-Event-ID). + return sseResponse([`id: evt-${++eventSeq}\ndata: \n\n`]); + } + const message = JSON.parse(init.body as string) as JSONRPCMessage; + if ('method' in message) { + if (message.method === 'initialize' && 'id' in message) { + return jsonResponse({ + jsonrpc: '2.0', + id: message.id, + result: { + protocolVersion: '2025-11-25', + capabilities: {}, + serverInfo: { name: 'legacy-server', version: '1.0.0' } + } + }); + } + if (message.method === 'notifications/cancelled') { + cancelledPosts.push(message); + return accepted(); + } + if (message.method === 'ping' && 'id' in message) { + pingId = message.id; + // SSE response: retry hint + priming event id, then a + // graceful close without the response. + return sseResponse([`retry: 10\nid: evt-${++eventSeq}\ndata: \n\n`]); + } + } + return accepted(); + }); + + const transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 2, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const client = new Client({ name: 'test-client', version: '1.0.0' }); + const errors: Error[] = []; + client.onerror = error => errors.push(error); + + await client.connect(transport); + + const resumeGetCount = () => + fetchMock.mock.calls.filter(call => call[1]?.method === 'GET' && (call[1].headers as Headers).get('last-event-id') !== null) + .length; + + let settledError: unknown; + const pending = client.ping({ timeout: 100 }).catch(error => { + settled = true; + settledError = error; + }); + + // Let the reconnect chain run a few resume cycles before the timeout. + await vi.advanceTimersByTimeAsync(50); + expect(resumeGetCount()).toBeGreaterThan(0); + expect(settled).toBe(false); + + // Cross the request timeout. + await vi.advanceTimersByTimeAsync(100); + await pending; + expect(settled).toBe(true); + expect(String(settledError)).toContain('Request timed out'); + + // The legacy wire cancel signal is unchanged: exactly one + // notifications/cancelled POST. + expect(cancelledPosts).toHaveLength(1); + + // Give an orphaned chain ample time to keep resuming (before the fix + // it reconnected forever — each successful resume resets the retry + // counter, so maxRetries never binds). + await vi.advanceTimersByTimeAsync(2000); + + // THE KEY ASSERTIONS: no resumed GET after the request settled, and + // the late response never surfaced as an unknown message ID. + expect(resumesAfterSettle).toBe(0); + expect(errors.map(e => e.message)).not.toContainEqual(expect.stringContaining('unknown message ID')); + + await client.close(); + }); + + it('cancellation POST still hits the wire when the original request was issued with a resumptionToken', async () => { + // Regression for the cancel-path resumption leak: transport.send() + // with a resumptionToken short-circuits into a GET+Last-Event-ID + // resume WITHOUT posting the message. If the request's own + // resumptionToken were forwarded into the notifications/cancelled + // send, the cancellation would be silently swallowed (no POST) and a + // fresh SSE reconnect chain — without the request-scoped abort signal + // — would be spawned in its place. + let eventSeq = 0; + const cancelledPosts: JSONRPCMessage[] = []; + + const fetchMock = globalThis.fetch as Mock; + fetchMock.mockImplementation(async (_url, init: RequestInit) => { + if (init.method === 'GET') { + const lastEventId = (init.headers as Headers).get('last-event-id'); + // Standalone notification stream: not offered by this server. + if (lastEventId === null) { + return methodNotAllowed(); + } + // Request-scoped resume: a priming event id, then a graceful + // close without the response, so the chain keeps resuming + // until the client tears it down. + return sseResponse([`retry: 10\nid: evt-${++eventSeq}\ndata: \n\n`]); + } + const message = JSON.parse(init.body as string) as JSONRPCMessage; + if ('method' in message) { + if (message.method === 'initialize' && 'id' in message) { + return jsonResponse({ + jsonrpc: '2.0', + id: message.id, + result: { + protocolVersion: '2025-11-25', + capabilities: {}, + serverInfo: { name: 'legacy-server', version: '1.0.0' } + } + }); + } + if (message.method === 'notifications/cancelled') { + cancelledPosts.push(message); + return accepted(); + } + } + return accepted(); + }); + + const transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 2, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const client = new Client({ name: 'test-client', version: '1.0.0' }); + const errors: Error[] = []; + client.onerror = error => errors.push(error); + + await client.connect(transport); + + const resumeGetCount = () => + fetchMock.mock.calls.filter(call => call[1]?.method === 'GET' && (call[1].headers as Headers).get('last-event-id') !== null) + .length; + + // Issue the request WITH a resumption token: the transport resumes the + // request's stream via GET + Last-Event-ID instead of POSTing it. + let settled = false; + const pending = client.ping({ timeout: 100, resumptionToken: 'evt-0' }).catch(() => { + settled = true; + }); + + await vi.advanceTimersByTimeAsync(50); + expect(resumeGetCount()).toBeGreaterThan(0); + expect(settled).toBe(false); + + // Cross the request timeout. + await vi.advanceTimersByTimeAsync(100); + await pending; + expect(settled).toBe(true); + + // THE KEY ASSERTION: the cancellation actually reached the wire as a + // POST — it was not swallowed into another GET resume. Note this is + // best-effort on the SDK's own transport: the POST carries the + // re-issued request's fresh JSON-RPC id, which never reached the + // server here (the send itself short-circuited into a GET resume), so + // the server cannot correlate it and a resumed request can only be + // torn down locally (the requestSignal abort). The POST is kept + // because custom per-request-stream transports that POST the + // re-issued body normally DO give the server a correlatable id. + expect(cancelledPosts).toHaveLength(1); + + // And no fresh (unguarded) reconnect chain was spawned by the + // cancellation send: once the request settled, the resume GET count + // stays flat. + const resumesAtSettle = resumeGetCount(); + await vi.advanceTimersByTimeAsync(2000); + expect(resumeGetCount()).toBe(resumesAtSettle); + + // The settlement abort landed on a deliberately-resumed request: a + // clean teardown, so no error — spurious AbortError included — may + // surface through client.onerror. + expect(errors).toEqual([]); + + await client.close(); + }); +}); + +/** + * Regression for the Node 20.0-20.2 `anySignal` fallback: it removes its + * listener pair only when one of its input signals fires, so composing a + * FRESH transport+request signal per SSE reconnect leg strands one closure + * pair per gracefully-completed leg on BOTH input signals until the request + * settles (MaxListenersExceededWarning after ~11 polling cycles). The + * composite must be built once per request chain and reused by every rebuilt + * leg. + */ +describe('anySignal fallback (Node 20.0-20.2): reconnect legs reuse one composite per request chain', () => { + const originalAny = AbortSignal.any; + + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(globalThis, 'fetch'); + // Simulate Node 20.0-20.2, where `AbortSignal.any` is unavailable and + // the manual fallback combinator must be used. + (AbortSignal as { any?: typeof AbortSignal.any }).any = undefined; + }); + + afterEach(() => { + (AbortSignal as { any?: typeof AbortSignal.any }).any = originalAny; + vi.useRealTimers(); + vi.clearAllMocks(); + }); + + it('does not accrue abort listeners on the request or transport signal across resume legs', async () => { + const encoder = new TextEncoder(); + let eventSeq = 0; + 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) { + // A priming event id, then a graceful close without a + // response: the documented SSE polling pattern — each leg + // completes with NEITHER input signal aborting. + controller.enqueue(encoder.encode(`id: evt-${++eventSeq}\ndata: \n\n`)); + controller.close(); + } + }) + })); + + const transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 2, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + await transport.start(); + + const requestAbort = new AbortController(); + await transport['_startOrAuthSse']({ resumptionToken: 'evt-0', requestSignal: requestAbort.signal }); + + // Let the chain run several full polling cycles (stream close + + // backoff + resume GET). + for (let cycle = 0; cycle < 5; cycle++) { + await vi.advanceTimersByTimeAsync(20); + } + expect(fetchMock.mock.calls.length).toBeGreaterThanOrEqual(4); + + const transportSignal = transport['_abortController']!.signal; + // One composite per request chain: exactly one fallback listener pair + // total, no matter how many legs have completed. (Between legs the + // pending attempt's settlement listener is also disarmed, so the + // fallback's listener is the only one left on the request signal.) + const requestListeners = getEventListeners(requestAbort.signal, 'abort'); + const transportListeners = getEventListeners(transportSignal, 'abort'); + expect(requestListeners.length).toBeLessThanOrEqual(2); + expect(transportListeners.length).toBeLessThanOrEqual(2); + + await transport.close(); }); }); diff --git a/packages/client/test/client/versionNegotiation.test.ts b/packages/client/test/client/versionNegotiation.test.ts index 60c9a2aaba..7bb1b59c11 100644 --- a/packages/client/test/client/versionNegotiation.test.ts +++ b/packages/client/test/client/versionNegotiation.test.ts @@ -159,6 +159,59 @@ const requests = (sent: JSONRPCMessage[]): JSONRPCRequest[] => sent.filter(isJSO * Probe mechanics (T9) + modern resolution. * ------------------------------------------------------------------------- */ +describe('probe exchange request-scoped abort (#2615 shape, one layer down)', () => { + test('a timed-out probe aborts its requestSignal; the retry sends with a fresh one', async () => { + // Regression: the probe was sent with no send options, so a timed-out + // exchange's settle() tore down nothing on the wire — over Streamable + // HTTP with probe.maxRetries >= 1, the stale probe's POST (and any + // primed GET+Last-Event-ID reconnect chain) stayed alive and its late + // response leaked into the live session. + const sendSignals: Array = []; + let probeSeq = 0; + class CapturingTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + sessionId?: string; + async start(): Promise {} + async close(): Promise { + this.onclose?.(); + } + setProtocolVersion(): void {} + async send(message: JSONRPCMessage, options?: { requestSignal?: AbortSignal }): Promise { + if (isJSONRPCRequest(message) && message.method === 'server/discover') { + sendSignals.push(options?.requestSignal); + probeSeq++; + if (probeSeq >= 2) { + const id = message.id; + queueMicrotask(() => this.onmessage?.({ jsonrpc: '2.0', id, result: discoverResult([MODERN]) })); + } + // First probe: never answered — times out, then retries. + } + } + } + + const transport = new CapturingTransport(); + const client = new Client( + { name: 'c', version: '0' }, + { versionNegotiation: { mode: 'auto', probe: { timeoutMs: 30, maxRetries: 1 } } } + ); + await client.connect(transport); + + expect(sendSignals).toHaveLength(2); + const [first, second] = sendSignals; + // Every probe exchange carries its own request-scoped signal… + expect(first).toBeDefined(); + expect(second).toBeDefined(); + expect(second).not.toBe(first); + // …and the timeout settle aborted the stale exchange on the wire, so + // its transport-level guards (fetch abort, reconnect disarm) engage. + expect(first!.aborted).toBe(true); + + await client.close(); + }); +}); + describe('auto mode against a modern server', () => { test('probe-first with a string id, no initialize, setProtocolVersion exactly once after era resolution', async () => { const transport = new ScriptedTransport(modernServerScript()); diff --git a/packages/core-internal/src/shared/protocol.ts b/packages/core-internal/src/shared/protocol.ts index 0a19770082..36f00327ba 100644 --- a/packages/core-internal/src/shared/protocol.ts +++ b/packages/core-internal/src/shared/protocol.ts @@ -97,10 +97,29 @@ export const DEFAULT_REQUEST_TIMEOUT_MSEC = 60_000; /** * Options that can be given per request. + * + * The {@linkcode TransportSendOptions} members this type absorbs are not all + * caller-controllable on the `request()` path — the protocol layer owns the + * per-request stream lifecycle there: + * + * - `requestSignal` is OVERWRITTEN with the protocol layer's request-scoped + * signal (aborted when the request settles); a caller-supplied value is + * ignored. Cancel via {@linkcode RequestOptions.signal | signal} instead. + * - `onRequestStreamEnd` is NOT forwarded to the transport. To observe the + * per-request stream's lifecycle directly, call `transport.send()` yourself. + * - `resumptionToken` / `onresumptiontoken` / `relatedRequestId` / `headers` + * are forwarded as documented. */ export type RequestOptions = { /** * If set, requests progress notifications from the remote end (if supported). When progress notifications are received, this callback will be invoked. + * + * Does not survive a `resumptionToken` re-issue on the SDK's Streamable + * HTTP transport: the resumed send never POSTs the fresh progress token, + * and progress notifications replayed on the resumed stream carry the + * original request's token, so this callback (and + * {@linkcode RequestOptions.resetTimeoutOnProgress | resetTimeoutOnProgress}) + * will not fire for the resumed request. */ onprogress?: ProgressCallback; @@ -124,8 +143,23 @@ export type RequestOptions = { resetTimeoutOnProgress?: boolean; /** - * Maximum total time (in milliseconds) to wait for a response. - * If exceeded, an {@linkcode SdkError} with code {@linkcode SdkErrorCode.RequestTimeout} will be raised, regardless of progress notifications. + * Maximum total time (in milliseconds) to wait for a response across + * progress-driven timeout resets. + * + * The budget is event-gated, not timer-enforced: it is checked when a + * progress notification arrives, and only when the request was issued + * with {@linkcode RequestOptions.resetTimeoutOnProgress | resetTimeoutOnProgress} + * `: true` and an {@linkcode RequestOptions.onprogress | onprogress} + * handler — the combination that makes per-leg timeout resets possible + * (without resets, the per-leg {@linkcode RequestOptions.timeout | timeout} + * already bounds the request on its own, and this option has no effect). + * When a progress notification lands after the budget has elapsed, the + * request rejects with an {@linkcode SdkError} with code + * {@linkcode SdkErrorCode.RequestTimeout} (`Maximum total timeout + * exceeded`). If the remote side stops sending progress near the budget + * boundary, settlement falls to the current per-leg timer instead, so the + * total wait can exceed the budget by up to `timeout` ms. + * * If not specified, there is no maximum total timeout. * * For multi-round-trip requests fulfilled by the auto-fulfilment driver @@ -519,7 +553,14 @@ type TimeoutInfo = { timeout: number; maxTotalTimeout?: number; resetTimeoutOnProgress: boolean; - onTimeout: () => void; + /** + * Settles the request through its cancel path. Called with no argument by + * the per-leg `setTimeout` (plain "Request timed out"); called with the + * `maxTotalTimeout` error by `_onprogress` so that settlement takes the + * same cancel path — emitting the era's wire cancel signal — while the + * caller still sees the original maxTotalTimeout error. + */ + onTimeout: (error?: Error) => void; }; /* @@ -724,7 +765,11 @@ export abstract class Protocol { } private async _oncancel(notification: CancelledNotification): Promise { - if (!notification.params.requestId) { + // Explicit undefined check: `0` is a legitimate JSON-RPC request id — + // it is every peer's FIRST outbound request id (`_requestMessageId` + // starts at 0) — and a falsy guard would silently drop its + // cancellation. + if (notification.params.requestId === undefined) { return; } // Handle request cancellation @@ -736,11 +781,15 @@ export abstract class Protocol { messageId: number, timeout: number, maxTotalTimeout: number | undefined, - onTimeout: () => void, + onTimeout: (error?: Error) => void, resetTimeoutOnProgress: boolean = false ) { this._timeoutInfo.set(messageId, { - timeoutId: setTimeout(onTimeout, timeout), + // Wrapped so the timer fires the handler with NO arguments: + // Gecko (Firefox) invokes setTimeout callbacks with an extra + // "lateness" Number argument, which must not be mistaken for the + // handler's optional error parameter. + timeoutId: setTimeout(() => onTimeout(), timeout), startTime: Date.now(), timeout, maxTotalTimeout, @@ -754,7 +803,16 @@ export abstract class Protocol { if (!info) return false; const totalElapsed = Date.now() - info.startTime; - if (info.maxTotalTimeout && totalElapsed >= info.maxTotalTimeout) { + // `!== undefined`, not truthiness: `maxTotalTimeout: 0` is the + // STRICTEST budget (rejects on the first check) — a falsy gate would + // silently disable it instead. + if (info.maxTotalTimeout !== undefined && totalElapsed >= info.maxTotalTimeout) { + // Disarm the still-armed per-leg timer BEFORE dropping the map + // entry: once the entry is gone, `_cleanupTimeout` (the funnel's + // `.finally()` cleanup) can no longer reach the timer, and an + // orphaned leg timer would fire cancel() again after this + // settlement. + clearTimeout(info.timeoutId); this._timeoutInfo.delete(messageId); throw new SdkError(SdkErrorCode.RequestTimeout, 'Maximum total timeout exceeded', { maxTotalTimeout: info.maxTotalTimeout, @@ -763,7 +821,9 @@ export abstract class Protocol { } clearTimeout(info.timeoutId); - info.timeoutId = setTimeout(info.onTimeout, info.timeout); + // Wrapped for the same reason as `_setupTimeout`: Gecko passes a + // lateness Number to setTimeout callbacks. + info.timeoutId = setTimeout(() => info.onTimeout(), info.timeout); return true; } @@ -1176,11 +1236,18 @@ export abstract class Protocol { try { this._resetTimeout(messageId); } catch (error) { - // Clean up if maxTotalTimeout was exceeded - this._responseHandlers.delete(messageId); - this._progressHandlers.delete(messageId); - this._cleanupTimeout(messageId); - responseHandler(error as Error); + // maxTotalTimeout exceeded. Settle through the request's + // cancel path (stored as `onTimeout`) rather than the response + // handler directly, so this settlement emits the same wire + // cancel signal as a plain timeout on the same session — the + // `notifications/cancelled` POST on a legacy (2025-11-25) + // connection, the stream-close abort alone on a modern + // (2026-07-28) one. cancel() rejects with an SdkError reason + // unchanged, so the caller still sees the original + // maxTotalTimeout error; handler/timeout cleanup runs in the + // request funnel's `.finally()`, exactly as for a plain + // timeout settlement. + timeoutInfo.onTimeout(error as Error); return; } } @@ -1372,6 +1439,7 @@ export abstract class Protocol { let onAbort: (() => void) | undefined; let cleanupMessageId: number | undefined; + let requestAbort: AbortController | undefined; // Send the request return new Promise>((resolve, reject) => { @@ -1413,9 +1481,18 @@ export abstract class Protocol { // POSTing `notifications/cancelled`. Every other (era × transport) // combination — legacy era on any transport, modern era on stdio / // in-memory — keeps today's `notifications/cancelled` POST path - // unchanged. + // unchanged (the legacy era on a per-request-stream transport + // additionally aborts `requestSignal` locally; see below). const streamCloseCancels = codec.era === MODERN_WIRE_REVISION && this._transport.hasPerRequestStream === true; - const requestAbort = streamCloseCancels ? new AbortController() : undefined; + // The per-request AbortController exists on EVERY per-request-stream + // transport, not just when stream-close is the spec cancel signal. + // On the legacy era the `notifications/cancelled` POST below stays + // the wire signal, but the transport still owns a per-request SSE + // reconnect chain (GET + Last-Event-ID resumption) that nothing + // else tears down: without this signal, a request that settles via + // timeout or caller abort leaves orphaned reconnects running until + // the late response surfaces as "unknown message ID" (#2615). + requestAbort = this._transport.hasPerRequestStream === true ? new AbortController() : undefined; const messageId = this._requestMessageId++; cleanupMessageId = messageId; @@ -1442,16 +1519,31 @@ export abstract class Protocol { // built above. const outbound = this._envelopeOutbound(jsonrpcRequest); - let responseReceived = false; + // `true` once the request has settled through EITHER channel — + // the response handler or cancel(). Guarding cancel() on it (and + // having cancel() set it) makes cancellation idempotent: a late + // per-leg timer or any future duplicate cancel path becomes a + // no-op instead of re-running the body and POSTing a second + // `notifications/cancelled` for an already-settled request. + let settled = false; const cancel = (reason: unknown) => { - if (responseReceived) { + if (settled) { return; } + settled = true; this._progressHandlers.delete(messageId); - if (requestAbort === undefined) { - this._transport + if (!streamCloseCancels) { + // Capture the transport identity at send time: the catch + // below must key on whether THIS connection is still the + // live one when the rejection lands. A bare + // `this._transport !== undefined` re-arms the report after + // a close-then-reconnect and would resurface the + // deliberately-aborted POST's AbortError on the brand-new + // connection (same idiom as _onrequest's capturedTransport). + const sendTransport = this._transport; + sendTransport ?.send( this._envelopeOutbound({ jsonrpc: '2.0', @@ -1461,17 +1553,41 @@ export abstract class Protocol { reason: String(reason) } }), - { relatedRequestId, resumptionToken, onresumptiontoken } + // Deliberately NOT forwarding the original request's + // resumptionToken/onresumptiontoken: a notification is + // not a resumable request, and on Streamable HTTP a + // resumption token short-circuits send() into a + // GET+Last-Event-ID resume WITHOUT posting the message + // — the cancellation would be silently swallowed and a + // fresh, unguarded SSE reconnect chain spawned in its + // place. + { relatedRequestId } ) - .catch(error => this._onerror(new Error(`Failed to send cancellation: ${error}`))); - } else { - // Modern-era per-request-stream transport: aborting the - // request's underlying stream IS the spec cancel signal. - // The transport already swallows the resulting AbortError - // (no spurious `onerror`); a post-abort send() rejection - // re-hits an already-settled promise below and is a no-op. - requestAbort.abort(); + .catch(error => { + // A deliberate transport close aborts an in-flight + // cancellation POST: the transport's own catch stays + // silent (intentional-abort guard) and rethrows, and + // by the time the rejection lands here `_onclose` has + // already cleared `_transport`. Re-reporting would + // resurface an AbortError for a clean shutdown — + // report only failures on the connection the POST + // was actually sent on. + if (this._transport === sendTransport) { + this._onerror(new Error(`Failed to send cancellation: ${error}`)); + } + }); } + // Aborting the request-scoped signal is either the spec cancel + // signal itself (modern era: closing the per-request stream IS + // the cancellation, so no `notifications/cancelled` above) or a + // purely local teardown alongside the POST (legacy era: it + // stops the transport's SSE reconnect chain for this request — + // #2615). The transport already swallows the resulting + // AbortError (no spurious `onerror`); a post-abort send() + // rejection re-hits an already-settled promise below and is a + // no-op. The cancelled POST above does not carry this signal, + // so aborting here cannot cut off that send. + requestAbort?.abort(); // Wrap the reason in an SdkError if it isn't already const error = reason instanceof SdkError ? reason : new SdkError(SdkErrorCode.RequestTimeout, String(reason)); @@ -1482,7 +1598,7 @@ export abstract class Protocol { if (options?.signal?.aborted) { return; } - responseReceived = true; + settled = true; if (response instanceof Error) { return reject(response); @@ -1552,7 +1668,12 @@ export abstract class Protocol { options?.signal?.addEventListener('abort', onAbort, { once: true }); const timeout = options?.timeout ?? DEFAULT_REQUEST_TIMEOUT_MSEC; - const timeoutHandler = () => cancel(new SdkError(SdkErrorCode.RequestTimeout, 'Request timed out', { timeout })); + // `instanceof Error` rather than `??`: a timer implementation that + // invokes its callback with a non-Error argument (Gecko passes a + // lateness Number) must still produce the plain timeout error, not + // reject the request with that stray value as its message. + const timeoutHandler = (error?: Error) => + cancel(error instanceof Error ? error : new SdkError(SdkErrorCode.RequestTimeout, 'Request timed out', { timeout })); this._setupTimeout(messageId, timeout, options?.maxTotalTimeout, timeoutHandler, options?.resetTimeoutOnProgress ?? false); @@ -1574,6 +1695,19 @@ export abstract class Protocol { this._responseHandlers.delete(cleanupMessageId); this._cleanupTimeout(cleanupMessageId); } + // Release the request-scoped signal on EVERY settlement path, not + // just cancel(): a successful completion (and a send() failure) + // never goes through cancel() and would otherwise pin one abort + // listener per request on the transport-lifetime signal via the + // Node 20.0–20.2 anySignal fallback — and leave the transport's + // per-request SSE reconnect chain orphaned (#2615) on any + // settlement cancel() missed. Idempotent after the cancel() abort + // above (timeouts, caller aborts, and maxTotalTimeout all route + // through cancel()); a no-op for single-channel transports + // (requestAbort is undefined). Aborting after a completed response + // is pure local teardown — the transport treats it as intentional + // (no onerror, no reconnect) and nothing is put on the wire. + requestAbort?.abort(); }); } @@ -1611,7 +1745,12 @@ export abstract class Protocol { const debouncedMethods = this._options?.debouncedNotificationMethods ?? []; // A notification can only be debounced if it's in the list AND it's "simple" // (i.e., has no parameters and no related request ID that could be lost). - const canDebounce = debouncedMethods.includes(notification.method) && !notification.params && !options?.relatedRequestId; + // `=== undefined` rather than falsiness on the id: `0` is a + // legitimate relatedRequestId (every peer's FIRST outbound request + // id), and a related notification must never be debounced/coalesced + // away from its request association. + const canDebounce = + debouncedMethods.includes(notification.method) && !notification.params && options?.relatedRequestId === undefined; if (canDebounce) { // If a notification of this type is already scheduled, do nothing. @@ -1634,8 +1773,21 @@ export abstract class Protocol { } // Send the notification, but don't await it here to avoid blocking. - // Handle potential errors with a .catch(). - this._transport?.send(jsonrpcNotification, options).catch(error => this._onerror(error)); + // Handle potential errors with a .catch(). Capture the + // transport identity at send time (same idiom as cancel()'s + // notifications/cancelled POST): a deliberate close() aborts + // the in-flight coalesced POST — the transport's own catch + // stays silent (intentional-abort guard) and rethrows, and by + // the time the rejection lands here `_onclose` has already + // cleared `_transport`. Re-reporting would resurface an + // AbortError through onerror for a clean shutdown — report + // only failures on the connection the POST was sent on. + const sendTransport = this._transport; + sendTransport.send(jsonrpcNotification, options).catch(error => { + if (this._transport === sendTransport) { + this._onerror(error); + } + }); }); // Return immediately. diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 226f6ab0bd..d2d83bc8ef 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -130,11 +130,15 @@ export interface Transport { * `true` when this transport opens one underlying request per outbound * JSON-RPC request (the Streamable HTTP POST-per-request model) and * therefore honors {@linkcode TransportSendOptions.requestSignal}. The - * 2026-07-28 spec makes closing that per-request stream the cancellation - * signal — the protocol layer aborts `requestSignal` instead of POSTing - * `notifications/cancelled` when this flag is set on a 2026-era - * connection. Transports that share a single channel (stdio, in-memory) - * leave it `undefined`. + * protocol layer threads a request-scoped `requestSignal` into every + * outbound request on such transports and aborts it when the request + * settles (response, error, timeout, or caller abort). On a 2026-07-28 + * connection that abort IS the spec cancellation signal — the 2026-07-28 + * spec makes closing the per-request stream the cancellation, so no + * `notifications/cancelled` is sent. On a 2025-era connection it is + * purely local teardown (it stops the request's SSE reconnect chain) + * accompanying the `notifications/cancelled` POST. Transports that share + * a single channel (stdio, in-memory) leave it `undefined`. */ readonly hasPerRequestStream?: boolean; diff --git a/packages/core-internal/test/shared/protocol.test.ts b/packages/core-internal/test/shared/protocol.test.ts index 2ecdc40adc..39767b404e 100644 --- a/packages/core-internal/test/shared/protocol.test.ts +++ b/packages/core-internal/test/shared/protocol.test.ts @@ -492,6 +492,42 @@ describe('protocol tests', () => { expect(onProgressMock).toHaveBeenCalledTimes(1); }); + test('maxTotalTimeout: 0 is the strictest budget — rejects on the first progress check, not disabled', async () => { + await protocol.connect(transport); + const request = { method: 'example', params: {} }; + const mockSchema: ZodType<{ result: string }> = z.object({ + result: z.string() + }); + const onProgressMock = vi.fn(); + const requestPromise = testRequest(protocol, request, mockSchema, { + timeout: 1000, + maxTotalTimeout: 0, + resetTimeoutOnProgress: true, + onprogress: onProgressMock + }); + + // The budget is already exhausted, so the FIRST progress + // notification's reset check must reject. Before the fix the + // falsy gate (`info.maxTotalTimeout && ...`) silently disabled a + // 0 budget entirely, leaving the request indefinitely + // progress-extendable — the strictest value was the only one that + // never rejected. + vi.advanceTimersByTime(10); + if (transport.onmessage) { + transport.onmessage({ + jsonrpc: '2.0', + method: 'notifications/progress', + params: { + progressToken: 0, + progress: 25, + total: 100 + } + }); + } + await expect(requestPromise).rejects.toThrow('Maximum total timeout exceeded'); + expect(onProgressMock).not.toHaveBeenCalled(); + }); + test('should timeout if no progress received within timeout period', async () => { await protocol.connect(transport); const request = { method: 'example', params: {} }; @@ -656,6 +692,24 @@ describe('protocol tests', () => { expect(sendSpy).toHaveBeenCalledWith(expect.any(Object), { relatedRequestId: 'req-2' }); }); + it('should NOT debounce a notification whose relatedRequestId is 0 (falsy but legitimate first-request id)', async () => { + // `_requestMessageId` starts at 0, so a notification related to a + // peer's first request carries relatedRequestId 0. A falsy gate + // wrongly debounced it — coalescing synchronous sends away and + // dropping the request association from the send options. + protocol = new TestProtocolImpl({ debouncedNotificationMethods: ['test/debounced_with_options'] }); + await protocol.connect(transport); + + // ACT — synchronous back-to-back sends, like the coalescing tests. + protocol.notification({ method: 'test/debounced_with_options' }, { relatedRequestId: 0 }); + protocol.notification({ method: 'test/debounced_with_options' }, { relatedRequestId: 0 }); + await flushMicrotasks(); + + // ASSERT — both sends hit the wire, each keeping its association. + expect(sendSpy).toHaveBeenCalledTimes(2); + expect(sendSpy).toHaveBeenCalledWith(expect.any(Object), { relatedRequestId: 0 }); + }); + it('should clear pending debounced notifications on connection close', async () => { // ARRANGE protocol = new TestProtocolImpl({ debouncedNotificationMethods: ['test/debounced'] }); @@ -676,6 +730,67 @@ describe('protocol tests', () => { expect(sendSpy).not.toHaveBeenCalled(); }); + it('debounced send rejection after close() does not surface onerror (deliberate teardown)', async () => { + // Regression: the coalesced POST's fire-and-forget catch reported + // unconditionally. A close() landing while the POST was in flight + // resurfaced the deliberate-teardown AbortError through + // Protocol.onerror — the transport-level guards suppress their + // own report and rethrow, so the protocol layer must key the + // report on whether THIS connection is still the live one (same + // capture-and-compare idiom as cancel()'s notifications/cancelled + // POST). + protocol = new TestProtocolImpl({ debouncedNotificationMethods: ['test/debounced'] }); + await protocol.connect(transport); + const onerror = vi.fn(); + protocol.onerror = onerror; + + let rejectSend!: (error: Error) => void; + sendSpy.mockImplementation( + () => + new Promise((_resolve, reject) => { + rejectSend = reject; + }) + ); + + protocol.notification({ method: 'test/debounced' }); + await flushMicrotasks(); + expect(sendSpy).toHaveBeenCalledTimes(1); + + // close() clears _transport; the aborted POST's rejection lands + // afterwards. + await protocol.close(); + rejectSend(new Error('This operation was aborted')); + await flushMicrotasks(); + + expect(onerror).not.toHaveBeenCalled(); + }); + + it('debounced send rejection on a live connection still reports through onerror', async () => { + protocol = new TestProtocolImpl({ debouncedNotificationMethods: ['test/debounced'] }); + await protocol.connect(transport); + const onerror = vi.fn(); + protocol.onerror = onerror; + + const sendError = new Error('network down'); + let rejectSend!: (error: Error) => void; + sendSpy.mockImplementation( + () => + new Promise((_resolve, reject) => { + rejectSend = reject; + }) + ); + + protocol.notification({ method: 'test/debounced' }); + await flushMicrotasks(); + expect(sendSpy).toHaveBeenCalledTimes(1); + + rejectSend(sendError); + await flushMicrotasks(); + + expect(onerror).toHaveBeenCalledTimes(1); + expect(onerror).toHaveBeenCalledWith(sendError); + }); + it('should debounce multiple synchronous calls when params property is omitted', async () => { // ARRANGE protocol = new TestProtocolImpl({ debouncedNotificationMethods: ['test/debounced'] }); @@ -819,6 +934,32 @@ describe('protocol tests', () => { // Verify the request was aborted expect(wasAborted).toBe(true); }); + + test("aborts the request handler for request id 0 (falsy but legitimate — every peer's FIRST request id)", async () => { + // `_requestMessageId` starts at 0, so a peer cancelling its first + // outbound request sends `requestId: 0`. A falsy guard in + // _oncancel silently dropped exactly that cancellation. + await protocol.connect(transport); + + let wasAborted = false; + protocol.setRequestHandler('ping', async (_request, ctx) => { + await new Promise(resolve => setTimeout(resolve, 100)); + wasAborted = ctx.mcpReq.signal.aborted; + return {}; + }); + + transport.onmessage?.({ jsonrpc: '2.0', id: 0, method: 'ping', params: {} }); + await new Promise(resolve => setTimeout(resolve, 10)); + + transport.onmessage?.({ + jsonrpc: '2.0', + method: 'notifications/cancelled', + params: { requestId: 0, reason: 'User cancelled' } + }); + + await new Promise(resolve => setTimeout(resolve, 150)); + expect(wasAborted).toBe(true); + }); }); // Spec basic/patterns/cancellation §Transport-Specific (2026-07-28): on a @@ -881,7 +1022,7 @@ describe('protocol tests', () => { expect(cancelledSent(sent)).toHaveLength(1); }); - test('legacy era + per-request-stream transport: behavior unchanged — POSTs notifications/cancelled, no requestSignal', async () => { + test('legacy era + per-request-stream transport: POSTs notifications/cancelled AND aborts the requestSignal (#2615)', async () => { const tx = new PerRequestStreamTransport(); const proto = createTestProtocol(); await proto.connect(tx); @@ -890,13 +1031,64 @@ describe('protocol tests', () => { const ac = new AbortController(); const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { signal: ac.signal }); - // Legacy path is byte-identical to before: no requestSignal threaded. - expect(tx.lastRequestSignal).toBeUndefined(); + // The requestSignal is threaded on the legacy era too — it is not + // the spec cancel signal there (the POST below is), but the + // transport needs it to tear down the request's SSE reconnect + // chain (GET + Last-Event-ID resumption) when the request settles. + const requestSignal = tx.lastRequestSignal; + expect(requestSignal).toBeInstanceOf(AbortSignal); + expect(requestSignal?.aborted).toBe(false); ac.abort('user cancel'); await expect(pending).rejects.toThrow(); + // The wire signal is unchanged (spec cancel = notifications/cancelled)… + expect(cancelledSent(tx.sent)).toHaveLength(1); + // …and the request-scoped abort additionally stops any reconnect + // chain the transport still owns for this request (#2615). + expect(requestSignal?.aborted).toBe(true); + }); + + test('legacy era + per-request-stream transport: timeout POSTs notifications/cancelled AND aborts the requestSignal (#2615)', async () => { + const tx = new PerRequestStreamTransport(); + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { timeout: 0 }); + const requestSignal = tx.lastRequestSignal; + expect(requestSignal).toBeInstanceOf(AbortSignal); + + await expect(pending).rejects.toThrow('Request timed out'); + expect(cancelledSent(tx.sent)).toHaveLength(1); + expect(requestSignal?.aborted).toBe(true); + }); + + test('legacy era + single-channel transport (no hasPerRequestStream): POSTs notifications/cancelled, no requestSignal', async () => { + // stdio / in-memory shape: hasPerRequestStream is undefined. + const sent: JSONRPCMessage[] = []; + let sawRequestSignal: AbortSignal | undefined; + const tx = new MockTransport(); + tx.send = async (m: JSONRPCMessage, opts?: TransportSendOptions) => { + sent.push(m); + if (opts?.requestSignal !== undefined) { + sawRequestSignal = opts.requestSignal; + } + }; + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const ac = new AbortController(); + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { signal: ac.signal }); + ac.abort('user cancel'); + await expect(pending).rejects.toThrow(); + + // No per-request stream to tear down — the legacy single-channel + // path stays byte-identical: cancelled POST only, no requestSignal. + expect(cancelledSent(sent)).toHaveLength(1); + expect(sawRequestSignal).toBeUndefined(); }); test('modern era + per-request-stream transport: timeout aborts the stream, NO notifications/cancelled', async () => { @@ -911,6 +1103,241 @@ describe('protocol tests', () => { expect(tx.lastRequestSignal?.aborted).toBe(true); expect(cancelledSent(tx.sent)).toHaveLength(0); }); + + test.each(['2025-11-25', '2026-07-28'])( + '%s era + per-request-stream transport: maxTotalTimeout settlement aborts the requestSignal (#2615)', + async era => { + const tx = new PerRequestStreamTransport(); + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, era); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { + timeout: 200, + maxTotalTimeout: 1, + resetTimeoutOnProgress: true, + onprogress: () => {} + }); + const requestSignal = tx.lastRequestSignal; + expect(requestSignal).toBeInstanceOf(AbortSignal); + expect(requestSignal?.aborted).toBe(false); + + // Cross the total budget, then deliver a progress notification: + // the maxTotalTimeout check fires on the timeout-reset path + // inside _onprogress, which routes the settlement through the + // request's cancel path — so this settlement emits the same + // wire cancel signal as a plain timeout, while the caller + // still sees the original maxTotalTimeout error. + await new Promise(resolve => setTimeout(resolve, 5)); + tx.onmessage?.({ + jsonrpc: '2.0', + method: 'notifications/progress', + params: { progressToken: 0, progress: 1 } + }); + + await expect(pending).rejects.toThrow('Maximum total timeout exceeded'); + expect(requestSignal?.aborted).toBe(true); + + // The wire cancel signal matches the era, exactly like a plain + // timeout settlement on the same session: legacy POSTs + // notifications/cancelled (carrying the request's id); modern + // cancels via the stream-close abort alone. + const cancelled = cancelledSent(tx.sent); + if (era === '2025-11-25') { + expect(cancelled).toHaveLength(1); + expect((cancelled[0] as { params?: { requestId?: unknown } }).params?.requestId).toBe(0); + } else { + expect(cancelled).toHaveLength(0); + } + + // The settlement must fully disarm the per-leg timer: cross + // the leg `timeout` (200ms) and re-assert nothing else went on + // the wire. Before the fix the over-budget branch orphaned the + // armed leg timer (deleted the map entry without clearTimeout, + // and cancel() never marked the request settled), so the late + // timer re-ran cancel() and POSTed a SECOND + // notifications/cancelled for the same requestId on legacy. + await new Promise(resolve => setTimeout(resolve, 250)); + expect(cancelledSent(tx.sent)).toHaveLength(era === '2025-11-25' ? 1 : 0); + } + ); + + test('per-leg timeout under a Gecko-style timer (callback invoked with a lateness argument): plain timeout error, not "0.42"', async () => { + // Firefox/Gecko invokes setTimeout callbacks with an extra Number + // ("lateness") argument. It must never be mistaken for the timeout + // handler's optional error parameter — that would reject the + // request with the stray number as its message and POST a + // notifications/cancelled whose reason is that number. + const realSetTimeout = globalThis.setTimeout.bind(globalThis); + const setTimeoutSpy = vi + .spyOn(globalThis, 'setTimeout') + .mockImplementation(((fn: (...args: unknown[]) => void, ms?: number) => + realSetTimeout(() => fn(0.42), ms)) as unknown as typeof setTimeout); + try { + const tx = new PerRequestStreamTransport(); + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { timeout: 1 }); + await expect(pending).rejects.toThrow('Request timed out'); + + const cancelled = cancelledSent(tx.sent); + expect(cancelled).toHaveLength(1); + expect((cancelled[0] as { params?: { reason?: string } }).params?.reason).toContain('Request timed out'); + } finally { + setTimeoutSpy.mockRestore(); + } + }); + + test('close() landing while the cancellation POST is in flight: no spurious "Failed to send cancellation" onerror', async () => { + let rejectCancelSend: ((error: unknown) => void) | undefined; + const tx = new PerRequestStreamTransport(); + const baseSend = tx.send.bind(tx); + tx.send = async (message: JSONRPCMessage, opts?: TransportSendOptions) => { + await baseSend(message, opts); + if ('method' in message && message.method === 'notifications/cancelled') { + // The cancellation POST stays in flight until close() + // aborts it (the transport rethrows the AbortError after + // its own intentional-abort guard stays silent). + return new Promise((_resolve, reject) => { + rejectCancelSend = reject; + }); + } + }; + const proto = createTestProtocol(); + const errors: Error[] = []; + proto.onerror = error => void errors.push(error); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { timeout: 0 }); + await expect(pending).rejects.toThrow('Request timed out'); + expect(rejectCancelSend).toBeDefined(); + + // Deliberate shutdown; the aborted POST's rejection lands after + // _onclose has already cleared the transport. + await proto.close(); + rejectCancelSend?.(new DOMException('The operation was aborted', 'AbortError')); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(errors.map(e => e.message).filter(m => m.includes('Failed to send cancellation'))).toHaveLength(0); + }); + + test('cancellation POST aborted by close(): stays silent even after an immediate reconnect', async () => { + // The close-then-reconnect recovery pattern re-attaches a + // transport before the aborted POST's rejection lands — the guard + // must key on the connection the POST was SENT on, not on whether + // any transport happens to be attached at rejection time. + let rejectCancelSend: ((error: unknown) => void) | undefined; + const tx = new PerRequestStreamTransport(); + const baseSend = tx.send.bind(tx); + tx.send = async (message: JSONRPCMessage, opts?: TransportSendOptions) => { + await baseSend(message, opts); + if ('method' in message && message.method === 'notifications/cancelled') { + return new Promise((_resolve, reject) => { + rejectCancelSend = reject; + }); + } + }; + const proto = createTestProtocol(); + const errors: Error[] = []; + proto.onerror = error => void errors.push(error); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { timeout: 0 }); + await expect(pending).rejects.toThrow('Request timed out'); + expect(rejectCancelSend).toBeDefined(); + + // Deliberate shutdown, then an IMMEDIATE reconnect on the same + // instance — before the aborted POST's rejection lands. + await proto.close(); + await proto.connect(new PerRequestStreamTransport()); + rejectCancelSend?.(new DOMException('The operation was aborted', 'AbortError')); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(errors.map(e => e.message).filter(m => m.includes('Failed to send cancellation'))).toHaveLength(0); + }); + + test('genuine cancellation-send failure on a live connection still reports through onerror', async () => { + const tx = new PerRequestStreamTransport(); + const baseSend = tx.send.bind(tx); + tx.send = async (message: JSONRPCMessage, opts?: TransportSendOptions) => { + await baseSend(message, opts); + if ('method' in message && message.method === 'notifications/cancelled') { + throw new TypeError('fetch failed'); + } + }; + const proto = createTestProtocol(); + const errors: Error[] = []; + proto.onerror = error => void errors.push(error); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { timeout: 0 }); + await expect(pending).rejects.toThrow('Request timed out'); + await new Promise(resolve => setTimeout(resolve, 10)); + + expect(errors.map(e => e.message).filter(m => m.includes('Failed to send cancellation'))).toHaveLength(1); + }); + + test('per-request-stream transport: successful completion releases (aborts) the request-scoped signal', async () => { + const tx = new PerRequestStreamTransport(); + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({})); + const requestSignal = tx.lastRequestSignal; + expect(requestSignal).toBeInstanceOf(AbortSignal); + expect(requestSignal?.aborted).toBe(false); + + tx.onmessage?.({ jsonrpc: '2.0', id: 0, result: {} }); + await expect(pending).resolves.toEqual({}); + + // A completed request sends no wire cancel of any kind… + expect(cancelledSent(tx.sent)).toHaveLength(0); + // …but the request-scoped signal is released (aborted) so the + // transport can drop its per-request state — otherwise every + // successful request leaks an abort listener on the transport- + // lifetime signal via the Node 20.0–20.2 anySignal fallback. + expect(requestSignal?.aborted).toBe(true); + }); + + test('legacy era: cancellation POST does not inherit the original request resumptionToken', async () => { + // On Streamable HTTP, transport.send() with a resumptionToken + // short-circuits into a GET+Last-Event-ID resume WITHOUT posting + // the message. A notification is not a resumable request, so the + // cancellation send must never carry the original request's + // resumption options — forwarding them silently swallows the + // cancellation and spawns a fresh, unguarded SSE reconnect chain. + class RecordingTransport extends MockTransport { + readonly hasPerRequestStream = true; + calls: { message: JSONRPCMessage; options?: TransportSendOptions }[] = []; + override async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise { + this.calls.push({ message, options }); + } + } + const tx = new RecordingTransport(); + const proto = createTestProtocol(); + await proto.connect(tx); + setNegotiatedProtocolVersion(proto, '2025-11-25'); + + const ac = new AbortController(); + const pending = testRequest(proto, { method: 'example', params: {} }, z.object({}), { + signal: ac.signal, + resumptionToken: 'evt-42', + onresumptiontoken: () => {} + }); + ac.abort('user cancel'); + await expect(pending).rejects.toThrow(); + + const cancelled = tx.calls.find(c => 'method' in c.message && c.message.method === 'notifications/cancelled'); + expect(cancelled).toBeDefined(); + expect(cancelled?.options?.resumptionToken).toBeUndefined(); + expect(cancelled?.options?.onresumptiontoken).toBeUndefined(); + }); }); }); diff --git a/test/e2e/scenarios/transport-http.test.ts b/test/e2e/scenarios/transport-http.test.ts index 7cda0f28aa..ab7f627b7e 100644 --- a/test/e2e/scenarios/transport-http.test.ts +++ b/test/e2e/scenarios/transport-http.test.ts @@ -1482,7 +1482,12 @@ 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); + // Each failed reconnect leg reports exactly once — the SSE open + // failure itself, from the transport's single reporting site; the + // scheduler no longer wraps the same failure in a duplicate + // "Failed to reconnect SSE stream" onerror. + 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