review: guard close() cancels per entry, register reconnection entry before scheduler runs

- close() wraps each user-supplied cancel in try/catch, routing throws
  to onerror so shutdown neither skips remaining cancels nor rejects
  (the existing throwing-cancel test now pins the contained behavior).
- The pending-reconnection entry is registered before the scheduler is
  invoked, so a synchronously-firing custom scheduler deregisters it
  instead of leaving a stale entry until close().

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

@@ -697,7 +697,13 @@ export class StreamableHTTPClientTransport implements Transport {
         // Calculate next delay based on current attempt count
         const delay = this._getNextReconnectionDelay(attemptCount);
 
-        let cancelEntry: () => void;
+        // The chain's registry entry. Registered before the scheduler is
+        // invoked so a synchronously-firing custom scheduler still
+        // deregisters it (otherwise the entry would be added after the fact
+        // and linger until close()); the real cancel behavior is filled in
+        // once the scheduler returns.
+        let cancelImpl: (() => void) | undefined;
+        const cancelEntry = (): void => cancelImpl?.();
         const reconnect = (): void => {
             this._pendingReconnections.delete(cancelEntry);
             // Honour BOTH the transport-wide abort and the per-request abort
@@ -721,32 +727,26 @@ export class StreamableHTTPClientTransport implements Transport {
             });
         };
 
+        // Track every pending reconnection — concurrent chains (the standby
+        // GET stream plus resumed per-request streams) each park a timer
+        // here, and close() must cancel all of them, not just the latest.
+        this._pendingReconnections.add(cancelEntry);
         try {
             if (this._reconnectionScheduler) {
                 const cancel = this._reconnectionScheduler(reconnect, delay, attemptCount);
-                cancelEntry =
-                    typeof cancel === 'function'
-                        ? cancel
-                        : () => {
-                              // No-op: the custom scheduler provided no cancel
-                              // function; tracked so `reconnect` can still
-                              // deregister the chain's pending entry.
-                          };
+                cancelImpl = typeof cancel === 'function' ? cancel : undefined;
             } else {
                 const handle = setTimeout(reconnect, delay);
-                cancelEntry = () => clearTimeout(handle);
+                cancelImpl = () => clearTimeout(handle);
             }
         } catch (error) {
             // A throwing custom scheduler means no reconnection is pending —
-            // the stream is definitively gone. Settle the caller here (the
-            // only route that can still do it), then rethrow for reporting.
+            // deregister the entry and settle the caller here (the only route
+            // that can still do it), then rethrow for reporting.
+            this._pendingReconnections.delete(cancelEntry);
             options.onRequestStreamEnd?.();
             throw error;
         }
-        // Track every pending reconnection — concurrent chains (the standby
-        // GET stream plus resumed per-request streams) each park a timer
-        // here, and close() must cancel all of them, not just the latest.
-        this._pendingReconnections.add(cancelEntry);
     }
 
     /**
@@ -1007,8 +1007,15 @@ export class StreamableHTTPClientTransport implements Transport {
         try {
             // Cancel EVERY pending reconnection — concurrent chains (standby
             // GET + resumed per-request streams) can each have a parked timer.
+            // Per-entry guard: a throwing user-supplied cancel (from a custom
+            // ReconnectionScheduler) must not skip the remaining cancels or
+            // escape the shutdown path.
             for (const cancel of this._pendingReconnections) {
-                cancel();
+                try {
+                    cancel();
+                } catch (error) {
+                    this.onerror?.(error instanceof Error ? error : new Error(String(error)));
+                }
             }
         } finally {
             this._pendingReconnections.clear();

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

@@ -3085,7 +3085,10 @@ describe('StreamableHTTPClientTransport', () => {
             expect(onerror).not.toHaveBeenCalled();
         });
 
-        it('still aborts and fires onclose if the cancel function throws', async () => {
+        it('contains a throwing cancel function: close() completes, aborts, and fires onclose', async () => {
+            // A user-supplied cancel that throws must not escape the shutdown
+            // path or skip the remaining pending cancels — it is surfaced via
+            // onerror instead.
             transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {
                 reconnectionOptions,
                 reconnectionScheduler: () => () => {
@@ -3094,12 +3097,15 @@ describe('StreamableHTTPClientTransport', () => {
             });
             const onclose = vi.fn();
             transport.onclose = onclose;
+            const onerror = vi.fn();
+            transport.onerror = onerror;
 
             await transport.start();
             triggerReconnection(transport);
             const abortController = transport['_abortController'];
 
-            await expect(transport.close()).rejects.toThrow('cancel failed');
+            await expect(transport.close()).resolves.toBeUndefined();
+            expect(onerror).toHaveBeenCalledWith(expect.objectContaining({ message: 'cancel failed' }));
             expect(abortController?.signal.aborted).toBe(true);
             expect(onclose).toHaveBeenCalledTimes(1);
         });