fix(client): abort legacy SSE reconnect chain when the originating request settles

On a legacy (2025-11-25) Streamable HTTP session, a request that settled
via timeout or caller abort POSTed notifications/cancelled but left the
transport's request-scoped SSE reconnect chain (GET + Last-Event-ID
resumption) running: the per-request AbortController was only created
when stream-close IS the cancel signal (modern era), so the transport's
requestSignal guards never fired, each successful resume reset the
retry counter, and a late resumed response surfaced as 'Received a
response for an unknown message ID' (#2615).

Create the request-scoped AbortController for every per-request-stream
transport regardless of era, and abort it when the request settles. On
the legacy era this is purely local teardown alongside the (unchanged)
notifications/cancelled POST; the modern-era stream-close-cancels path
and single-channel transports (stdio/in-memory) are byte-identical to
before.

Author: claude

.changeset/legacy-sse-reconnect-after-timeout.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/client': 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 alongside the `notifications/cancelled` POST when the request settles — the wire-visible cancellation behavior is unchanged for every (era × transport) combination.

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

@@ -4,6 +4,7 @@ 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,6 +1465,57 @@ describe('StreamableHTTPClientTransport', () => {
             expect(fetchMock).toHaveBeenCalledTimes(1);
         });
 
+        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<Uint8Array>({
+                start(controller) {
+                    controller.enqueue(new TextEncoder().encode('id: ev-1\ndata: \n\n'));
+                    controller.close();
+                }
+            });
+            const fetchMock = globalThis.fetch as Mock;
+            fetchMock.mockResolvedValueOnce({
+                ok: true,
+                status: 200,
+                headers: new Headers({ 'content-type': 'text/event-stream' }),
+                body: primedClosingStream
+            });
+
+            const requestAbort = new AbortController();
+            await transport.start();
+            await transport.send(
+                { 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(fetchMock).toHaveBeenCalledTimes(1);
+
+            // ACT — the request settles (timeout/cancel) before the reconnect fires.
+            requestAbort.abort();
+            await vi.advanceTimersByTimeAsync(100);
+
+            // 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 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
@@ -2737,3 +2789,160 @@ describe('StreamableHTTPClientTransport', () => {
         });
     });
 });
+
+/**
+ * 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<Uint8Array>({
+            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();
+    });
+});

packages/core-internal/src/shared/protocol.ts

@@ -1413,9 +1413,18 @@ export abstract class Protocol<ContextT extends BaseContext> {
             // 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).
+            const requestAbort = this._transport.hasPerRequestStream === true ? new AbortController() : undefined;
 
             const messageId = this._requestMessageId++;
             cleanupMessageId = messageId;
@@ -1450,7 +1459,7 @@ export abstract class Protocol<ContextT extends BaseContext> {
                 }
                 this._progressHandlers.delete(messageId);
 
-                if (requestAbort === undefined) {
+                if (!streamCloseCancels) {
                     this._transport
                         ?.send(
                             this._envelopeOutbound({
@@ -1464,14 +1473,18 @@ export abstract class Protocol<ContextT extends BaseContext> {
                             { relatedRequestId, resumptionToken, onresumptiontoken }
                         )
                         .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();
                 }
+                // 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));

packages/core-internal/test/shared/protocol.test.ts

@@ -881,7 +881,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 +890,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 () => {