fix(client): avoid stale-session 404 race and stop retry loop on session-bound 404

Author: Maverick-666

packages/client/src/client/streamableHttp.ts

@@ -25,6 +25,23 @@ const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOp
     maxRetries: 2
 };
 
+const SESSION_BOUND_404_ERROR = Symbol('sessionBound404Error');
+
+type SessionBound404Error = Error & { [SESSION_BOUND_404_ERROR]?: true };
+
+function markSessionBound404Error(error: Error): Error {
+    (error as SessionBound404Error)[SESSION_BOUND_404_ERROR] = true;
+    return error;
+}
+
+function isSessionBound404Error(error: unknown): boolean {
+    return Boolean(
+        error &&
+            typeof error === 'object' &&
+            (error as SessionBound404Error)[SESSION_BOUND_404_ERROR] === true
+    );
+}
+
 /**
  * Options for starting or authenticating an SSE connection
  */
@@ -237,7 +254,7 @@ export class StreamableHTTPClientTransport implements Transport {
             // Try to open an initial SSE stream with GET to listen for server messages
             // This is optional according to the spec - server may not support it
             const headers = await this._commonHeaders();
-            const sentWithSession = headers.has('mcp-session-id');
+            const sentSessionId = headers.get('mcp-session-id');
             const userAccept = headers.get('accept');
             const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'text/event-stream'];
             headers.set('accept', [...new Set(types)].join(', '));
@@ -255,7 +272,9 @@ export class StreamableHTTPClientTransport implements Transport {
             });
 
             if (!response.ok) {
-                if (response.status === 404 && sentWithSession) {
+                const shouldClearSessionFor404 =
+                    response.status === 404 && sentSessionId !== null && this._sessionId === sentSessionId;
+                if (shouldClearSessionFor404) {
                     this._sessionId = undefined;
                 }
 
@@ -293,10 +312,15 @@ export class StreamableHTTPClientTransport implements Transport {
                     return;
                 }
 
-                throw new SdkError(SdkErrorCode.ClientHttpFailedToOpenStream, `Failed to open SSE stream: ${response.statusText}`, {
+                const error = new SdkError(
+                    SdkErrorCode.ClientHttpFailedToOpenStream,
+                    `Failed to open SSE stream: ${response.statusText}`,
+                    {
                     status: response.status,
                     statusText: response.statusText
-                });
+                    }
+                );
+                throw shouldClearSessionFor404 ? markSessionBound404Error(error) : error;
             }
 
             this._handleSseStream(response.body, options, true);
@@ -350,7 +374,11 @@ export class StreamableHTTPClientTransport implements Transport {
             this._cancelReconnection = undefined;
             if (this._abortController?.signal.aborted) return;
             this._startOrAuthSse(options).catch(error => {
-                this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`));
+                const reconnectError = error instanceof Error ? error : new Error(String(error));
+                this.onerror?.(new Error(`Failed to reconnect SSE stream: ${reconnectError.message}`));
+                if (isSessionBound404Error(reconnectError)) {
+                    return;
+                }
                 try {
                     this._scheduleReconnection(options, attemptCount + 1);
                 } catch (scheduleError) {
@@ -544,7 +572,7 @@ export class StreamableHTTPClientTransport implements Transport {
             }
 
             const headers = await this._commonHeaders();
-            const sentWithSession = headers.has('mcp-session-id');
+            const sentSessionId = headers.get('mcp-session-id');
             headers.set('content-type', 'application/json');
             const userAccept = headers.get('accept');
             const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'application/json', 'text/event-stream'];
@@ -567,7 +595,7 @@ export class StreamableHTTPClientTransport implements Transport {
             }
 
             if (!response.ok) {
-                if (response.status === 404 && sentWithSession) {
+                if (response.status === 404 && sentSessionId !== null && this._sessionId === sentSessionId) {
                     this._sessionId = undefined;
                 }
 

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

@@ -303,6 +303,49 @@ describe('StreamableHTTPClientTransport', () => {
         expect(lastCall[1].headers.get('mcp-session-id')).toBeNull();
     });
 
+    it('should not clear a newer session ID when a stale session-bound POST request returns 404', async () => {
+        transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+            sessionId: 'stale-session-A'
+        });
+
+        const message: JSONRPCMessage = {
+            jsonrpc: '2.0',
+            method: 'tools/list',
+            params: {},
+            id: 'test-id'
+        };
+
+        let resolveFetch!: (value: unknown) => void;
+        const deferredFetch = new Promise(resolve => {
+            resolveFetch = resolve;
+        });
+
+        (globalThis.fetch as Mock).mockImplementationOnce(() => {
+            // Simulate another in-flight request establishing a fresh session while this request is pending.
+            (transport as unknown as { _sessionId?: string })._sessionId = 'fresh-session-B';
+            return deferredFetch;
+        });
+
+        const sendPromise = transport.send(message);
+
+        resolveFetch({
+            ok: false,
+            status: 404,
+            statusText: 'Not Found',
+            text: () => Promise.resolve('Session not found'),
+            headers: new Headers()
+        });
+
+        await expect(sendPromise).rejects.toMatchObject({
+            code: SdkErrorCode.ClientHttpNotImplemented,
+            data: expect.objectContaining({
+                status: 404
+            })
+        });
+
+        expect(transport.sessionId).toBe('fresh-session-B');
+    });
+
     it('should handle non-streaming JSON response', async () => {
         const message: JSONRPCMessage = {
             jsonrpc: '2.0',
@@ -395,6 +438,46 @@ describe('StreamableHTTPClientTransport', () => {
         expect(getCall[1].headers.get('mcp-session-id')).toBe('stale-session-id');
     });
 
+    it('should not clear a newer session ID when a stale session-bound GET request returns 404', async () => {
+        transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+            sessionId: 'stale-session-A'
+        });
+        await transport.start();
+
+        let resolveFetch!: (value: unknown) => void;
+        const deferredFetch = new Promise(resolve => {
+            resolveFetch = resolve;
+        });
+
+        (globalThis.fetch as Mock).mockImplementationOnce(() => {
+            // Simulate another in-flight request establishing a fresh session while this request is pending.
+            (transport as unknown as { _sessionId?: string })._sessionId = 'fresh-session-B';
+            return deferredFetch;
+        });
+
+        const startPromise = (
+            transport as unknown as { _startOrAuthSse: (opts: StartSSEOptions) => Promise<void> }
+        )._startOrAuthSse({});
+
+        resolveFetch({
+            ok: false,
+            status: 404,
+            statusText: 'Not Found',
+            text: () => Promise.resolve('Session not found'),
+            headers: new Headers()
+        });
+
+        await expect(startPromise).rejects.toMatchObject({
+            code: SdkErrorCode.ClientHttpFailedToOpenStream,
+            data: expect.objectContaining({
+                status: 404,
+                statusText: 'Not Found'
+            })
+        });
+
+        expect(transport.sessionId).toBe('fresh-session-B');
+    });
+
     it('should handle successful initial GET connection for SSE', async () => {
         // Set up readable stream for SSE events
         const encoder = new TextEncoder();
@@ -1022,6 +1105,40 @@ describe('StreamableHTTPClientTransport', () => {
             expect(fetchMock.mock.calls[1]![1]?.method).toBe('GET');
         });
 
+        it('should stop retrying GET reconnection after a session-bound 404 clears the stale session', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                sessionId: 'stale-session-id',
+                reconnectionOptions: {
+                    initialReconnectionDelay: 10,
+                    maxRetries: 3,
+                    maxReconnectionDelay: 1000,
+                    reconnectionDelayGrowFactor: 1
+                }
+            });
+            await transport.start();
+
+            const fetchMock = globalThis.fetch as Mock;
+            fetchMock.mockResolvedValue({
+                ok: false,
+                status: 404,
+                statusText: 'Not Found',
+                headers: new Headers(),
+                text: () => Promise.resolve('Session not found')
+            });
+
+            (
+                transport as unknown as {
+                    _scheduleReconnection: (opts: StartSSEOptions, attemptCount?: number) => void;
+                }
+            )._scheduleReconnection({}, 0);
+
+            await vi.advanceTimersByTimeAsync(20);
+            await vi.advanceTimersByTimeAsync(100);
+
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+            expect(transport.sessionId).toBeUndefined();
+        });
+
         it('should NOT reconnect a POST-initiated stream that fails', async () => {
             // ARRANGE
             transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {