fix(client): per-request abort in StreamableHTTP is a clean shutdown — no onerror, no reconnect

McpSubscription.close() aborts the listen POST via requestSignal, but
_handleSseStream only knew about the transport-level abort: every close()
fired a misleading 'SSE stream disconnected' onerror, and (when the server
had stamped SSE event ids for resumability) scheduled a GET+Last-Event-ID
reconnect that resurrected the subscription the caller just tore down.

Thread the per-request requestSignal into _handleSseStream via
StartSSEOptions; when EITHER the transport signal or the per-request
signal is aborted, skip onerror and skip reconnect (mirrors the existing
transport-level reconnect guard).

Author: felixweinberger

packages/client/src/client/streamableHttp.ts

@@ -50,6 +50,15 @@ export interface StartSSEOptions {
      * so that the response can be associated with the new resumed request.
      */
     replayMessageId?: string | number;
+
+    /**
+     * The per-request abort signal supplied by the caller via
+     * `TransportSendOptions.requestSignal`. When this signal is aborted the
+     * originating POST and its SSE response stream are torn down
+     * intentionally — `_handleSseStream` treats it exactly like the
+     * transport-level abort: no `onerror`, no reconnect.
+     */
+    requestSignal?: AbortSignal;
 }
 
 /**
@@ -391,7 +400,13 @@ export class StreamableHTTPClientTransport implements Transport {
         if (!stream) {
             return;
         }
-        const { onresumptiontoken, replayMessageId } = options;
+        const { onresumptiontoken, replayMessageId, requestSignal } = 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,
+        // and no GET+Last-Event-ID reconnect that would resurrect a stream the
+        // caller just tore down.
+        const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true;
 
         let lastEventId: string | undefined;
         // Track whether we've received a priming event (event with ID)
@@ -460,7 +475,7 @@ export class StreamableHTTPClientTransport implements Transport {
                 // 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 && !this._abortController.signal.aborted) {
+                if (needsReconnect && this._abortController && !isIntentionalAbort()) {
                     this._scheduleReconnection(
                         {
                             resumptionToken: lastEventId,
@@ -471,6 +486,11 @@ export class StreamableHTTPClientTransport implements Transport {
                     );
                 }
             } catch (error) {
+                if (isIntentionalAbort()) {
+                    // The reader threw because we aborted it. Not an error; do
+                    // not surface onerror, do not reconnect.
+                    return;
+                }
                 // Handle stream errors - likely a network disconnect
                 this.onerror?.(new Error(`SSE stream disconnected: ${error}`));
 
@@ -479,7 +499,7 @@ export class StreamableHTTPClientTransport implements Transport {
                 // 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 && !this._abortController.signal.aborted) {
+                if (needsReconnect && this._abortController && !isIntentionalAbort()) {
                     // Use the exponential backoff reconnection strategy
                     try {
                         this._scheduleReconnection(
@@ -699,7 +719,7 @@ export class StreamableHTTPClientTransport implements Transport {
                     // Handle SSE stream responses for requests
                     // We use the same handler as standalone streams, which now supports
                     // reconnection with the last event ID
-                    this._handleSseStream(response.body, { onresumptiontoken }, false);
+                    this._handleSseStream(response.body, { onresumptiontoken, requestSignal: options?.requestSignal }, false);
                 } else if (contentType?.includes('application/json')) {
                     // For non-streaming servers, we might get direct JSON responses
                     const data = await response.json();

packages/client/test/client/streamableHttp.test.ts

@@ -1102,6 +1102,59 @@ describe('StreamableHTTPClientTransport', () => {
             expect(fetchMock.mock.calls[0]![1]?.method).toBe('POST');
         });
 
+        it('per-request requestSignal abort: no onerror, no reconnect (McpSubscription.close())', async () => {
+            // ARRANGE — a POST stream that has been primed with an SSE event id
+            // (server-side resumability), so without the per-request abort
+            // guard the transport WOULD schedule a GET+Last-Event-ID reconnect.
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 10,
+                    maxRetries: 1,
+                    maxReconnectionDelay: 1000,
+                    reconnectionDelayGrowFactor: 1
+                }
+            });
+            const errorSpy = vi.fn();
+            transport.onerror = errorSpy;
+
+            let streamController!: ReadableStreamDefaultController<Uint8Array>;
+            const primedStream = new ReadableStream<Uint8Array>({
+                start(controller) {
+                    streamController = controller;
+                    // Priming event with an id — would arm POST-stream resumability.
+                    controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n'));
+                }
+            });
+            const fetchMock = globalThis.fetch as Mock;
+            fetchMock.mockImplementationOnce((_url, init: RequestInit) => {
+                // Propagate abort to the stream the way fetch does.
+                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: primedStream
+                });
+            });
+
+            const requestAbort = new AbortController();
+            await transport.start();
+            await transport.send(
+                { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen-1', params: {} },
+                { requestSignal: requestAbort.signal }
+            );
+            await vi.advanceTimersByTimeAsync(5);
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+
+            // ACT — McpSubscription.close() aborts the per-request signal.
+            requestAbort.abort();
+            await vi.advanceTimersByTimeAsync(50);
+
+            // ASSERT — intentional per-request abort: no onerror, no reconnect.
+            expect(errorSpy).not.toHaveBeenCalled();
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+        });
+
         it('should NOT reconnect a POST stream when error response was received', async () => {
             // ARRANGE
             transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {