review: thread stream callbacks through resumed send(), guard graceful-close scheduling

- The send()-resume branch now forwards onresumptiontoken and
  onRequestStreamEnd to the resumed GET, so new event IDs keep reaching
  the caller's persistence hook and the pending request settles when
  reconnection attempts are exhausted instead of hanging to timeout.
- Guard the graceful-close scheduleNext() call the same way as the
  error branch: exhaustion now runs user callbacks from graceful close,
  and a throwing handler must not fall into the outer catch and
  double-fire or surface a misleading disconnect error.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Eqc26ABfbhTimyUUszsUxL

Author: claude

packages/client/src/client/streamableHttp.ts

@@ -858,7 +858,19 @@ export class StreamableHTTPClientTransport implements Transport {
                 const canResume = isReconnectable || hasPrimingEvent;
                 const needsReconnect = canResume && !receivedResponse;
                 if (needsReconnect && this._abortController && !isIntentionalAbort()) {
-                    scheduleNext();
+                    // Same guard as the error path below. With the fruitless
+                    // accounting, exhaustion — which synchronously invokes
+                    // `onerror`/`onRequestStreamEnd` and any custom
+                    // `ReconnectionScheduler` — is reachable from a graceful
+                    // close, and a throwing user callback must not fall into
+                    // the outer catch (which would surface a misleading
+                    // disconnect error and schedule a second time).
+                    try {
+                        scheduleNext();
+                    } catch (error) {
+                        this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`));
+                        onRequestStreamEnd?.();
+                    }
                 } else if (!isIntentionalAbort()) {
                     // The per-request stream ended without reconnecting (no
                     // priming event for a POST stream, or response already
@@ -1006,10 +1018,18 @@ 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.
+                // Thread the caller's stream callbacks through as well:
+                // `onresumptiontoken` so new event IDs on the resumed stream
+                // keep reaching the caller's persistence hook, and
+                // `onRequestStreamEnd` so the pending request settles instead
+                // of hanging when the resumed stream ends for good (e.g.
+                // reconnection attempts are exhausted).
                 this._startOrAuthSse({
                     resumptionToken,
+                    onresumptiontoken,
                     replayMessageId: isJSONRPCRequest(message) ? message.id : undefined,
-                    requestSignal: options?.requestSignal
+                    requestSignal: options?.requestSignal,
+                    onRequestStreamEnd: options?.onRequestStreamEnd
                 }).catch(error => this.onerror?.(error));
                 return;
             }

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

@@ -2727,6 +2727,91 @@ describe('StreamableHTTPClientTransport', () => {
             // Before the fix this was null: the empty second stream dropped the token.
             expect(thirdCallHeaders.get('last-event-id')).toBe('event-1');
         });
+
+        it('threads the stream callbacks through a resumed send(), settling the caller on exhaustion', async () => {
+            // A send() with a resumptionToken resumes via GET. New event IDs on
+            // the resumed stream must reach the caller's onresumptiontoken, and
+            // when reconnection attempts are exhausted the caller's
+            // onRequestStreamEnd must fire so the pending request settles
+            // instead of hanging until its timeout.
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 10,
+                    maxReconnectionDelay: 1000,
+                    reconnectionDelayGrowFactor: 1,
+                    maxRetries: 1
+                }
+            });
+            const errorSpy = vi.fn();
+            transport.onerror = errorSpy;
+            const tokenSpy = vi.fn();
+            const streamEndSpy = vi.fn();
+
+            const fetchMock = globalThis.fetch as Mock;
+            // Resumed stream delivers a fresh priming event then idle-closes;
+            // every stream after that closes empty until exhaustion.
+            const bodies: string[][] = [['id: event-next\ndata: \n\n']];
+            fetchMock.mockImplementation(async () => sseResponse(bodies.shift() ?? []));
+
+            const requestMessage: JSONRPCRequest = {
+                jsonrpc: '2.0',
+                method: 'long_running_tool',
+                id: 'request-1',
+                params: {}
+            };
+
+            await transport.start();
+            await transport.send(requestMessage, {
+                resumptionToken: 'event-0',
+                onresumptiontoken: tokenSpy,
+                onRequestStreamEnd: streamEndSpy
+            });
+            await vi.advanceTimersByTimeAsync(200);
+
+            // The resume goes out as a GET with the caller's token.
+            expect(fetchMock.mock.calls[0]![1]?.method).toBe('GET');
+            expect((fetchMock.mock.calls[0]![1]?.headers as Headers).get('last-event-id')).toBe('event-0');
+            // The fresh priming event reached the caller's persistence hook.
+            expect(tokenSpy).toHaveBeenCalledWith('event-next');
+            // Exhaustion settled the caller instead of leaving it hanging.
+            expect(streamEndSpy).toHaveBeenCalledTimes(1);
+            expect(errorSpy).toHaveBeenCalledWith(
+                expect.objectContaining({
+                    message: 'Maximum reconnection attempts (1) exceeded.'
+                })
+            );
+        });
+
+        it('does not double-fire exhaustion callbacks when a user onerror handler throws on graceful close', async () => {
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
+                reconnectionOptions: {
+                    initialReconnectionDelay: 10,
+                    maxReconnectionDelay: 1000,
+                    reconnectionDelayGrowFactor: 1,
+                    maxRetries: 0 // exhaustion trips on the first graceful close
+                }
+            });
+            const errorSpy = vi.fn().mockImplementationOnce(() => {
+                throw new Error('user handler exploded');
+            });
+            transport.onerror = errorSpy;
+            const streamEndSpy = vi.fn();
+
+            const fetchMock = globalThis.fetch as Mock;
+            fetchMock.mockImplementation(async () => sseResponse([]));
+
+            await transport.start();
+            await transport['_startOrAuthSse']({ onRequestStreamEnd: streamEndSpy });
+            await vi.advanceTimersByTimeAsync(100);
+
+            // The throwing handler is contained by the graceful branch's guard:
+            // the caller still settles exactly once, no reconnect is scheduled,
+            // and no misleading 'SSE stream disconnected' error is emitted.
+            expect(streamEndSpy).toHaveBeenCalledTimes(1);
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+            const messages = errorSpy.mock.calls.map(args => (args[0] as Error).message);
+            expect(messages.some(m => m.includes('SSE stream disconnected'))).toBe(false);
+        });
     });
 
     describe('prevent infinite recursion when server returns 401 after successful auth', () => {