fix(client): listen auto-open inherits only ack timeout; thread requestSignal through SSE reconnect; anySignal fallback removes sibling listener

- auto-open: forwarding connect()'s full RequestOptions into listen() bound a
  connect-scoped signal (e.g. AbortSignal.timeout for the handshake) to the
  subscription lifetime, silently tearing the auto-opened stream down when it
  fired after connect resolved. Forward only the ack timeout.
- _handleSseStream: the two _scheduleReconnection call sites rebuilt
  StartSSEOptions without requestSignal, so after one drop+reconnect of a
  listen stream the per-request abort guard stopped working.
- anySignal fallback (Node 20.0-20.2): {once:true} alone leaked the sibling
  listener on the transport-lifetime signal per _send() with a requestSignal.
  Remove both listeners when either input fires.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -909,8 +909,19 @@ export class Client extends Protocol<ClientContext> {
                 // connection is fully usable without a listen stream (the
                 // server may not support it, or refuse on capacity). Surface
                 // via onerror; the consumer can call listen() later.
+                //
+                // Forward ONLY the ack timeout from connect()'s options.
+                // listen() binds RequestOptions.signal to the SUBSCRIPTION
+                // lifetime, so a connect-scoped signal (e.g.
+                // `AbortSignal.timeout(30_000)` for the handshake) would tear
+                // the auto-opened stream down the moment it fires after
+                // connect has already resolved. Connect's signal governs the
+                // handshake only; the auto-opened subscription outlives it.
                 try {
-                    this._autoOpenedSubscription = await this.listen(filter, options);
+                    this._autoOpenedSubscription = await this.listen(
+                        filter,
+                        options?.timeout === undefined ? undefined : { timeout: options.timeout }
+                    );
                 } catch (error) {
                     this.onerror?.(error instanceof Error ? error : new Error(String(error)));
                 }

packages/client/src/client/streamableHttp.ts

@@ -188,9 +188,25 @@ function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal {
     const controller = new AbortController();
     if (a.aborted) return (controller.abort(a.reason), controller.signal);
     if (b.aborted) return (controller.abort(b.reason), controller.signal);
-    const forward = (source: AbortSignal) => () => controller.abort(source.reason);
-    a.addEventListener('abort', forward(a), { once: true });
-    b.addEventListener('abort', forward(b), { once: true });
+    // Standard polyfill shape: when EITHER input fires, remove the listener
+    // registered on the OTHER input too. `{once:true}` alone leaks the
+    // sibling listener — for `_send()`, `a` is the transport-lifetime signal,
+    // so every request-scoped `b` that aborts would otherwise leave one
+    // listener + closure pinned on `a` for the life of the transport.
+    const cleanup = (): void => {
+        a.removeEventListener('abort', onA);
+        b.removeEventListener('abort', onB);
+    };
+    function onA(): void {
+        cleanup();
+        controller.abort(a.reason);
+    }
+    function onB(): void {
+        cleanup();
+        controller.abort(b.reason);
+    }
+    a.addEventListener('abort', onA, { once: true });
+    b.addEventListener('abort', onB, { once: true });
     return controller.signal;
 }
 
@@ -501,7 +517,8 @@ export class StreamableHTTPClientTransport implements Transport {
                         {
                             resumptionToken: lastEventId,
                             onresumptiontoken,
-                            replayMessageId
+                            replayMessageId,
+                            requestSignal
                         },
                         0
                     );
@@ -527,7 +544,8 @@ export class StreamableHTTPClientTransport implements Transport {
                             {
                                 resumptionToken: lastEventId,
                                 onresumptiontoken,
-                                replayMessageId
+                                replayMessageId,
+                                requestSignal
                             },
                             0
                         );

packages/client/test/client/listen.test.ts

@@ -483,6 +483,39 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('connect-scoped signal does NOT bind to the auto-opened subscription lifetime', async () => {
+        // Regression: forwarding connect()'s full RequestOptions into the
+        // auto-open listen() call meant a connect-scoped signal — typically
+        // `AbortSignal.timeout(30_000)` for the handshake — was bound to the
+        // SUBSCRIPTION lifetime. When it fired after connect resolved, the
+        // auto-opened stream was silently torn down.
+        const { clientTx, written } = await scriptedModern();
+        const onChanged = () => {};
+        const client = new Client(
+            { name: 'c', version: '1' },
+            { versionNegotiation: { mode: 'auto' }, listChanged: { tools: { onChanged } } }
+        );
+        const errors: Error[] = [];
+        client.onerror = e => errors.push(e);
+        const connectScoped = new AbortController();
+        await client.connect(clientTx, { signal: connectScoped.signal });
+        expect(client.autoOpenedSubscription).toBeDefined();
+        written.length = 0;
+
+        // The connect-scoped signal fires AFTER connect resolved (as a
+        // handshake `AbortSignal.timeout` would).
+        connectScoped.abort();
+        await flush();
+
+        // The auto-opened subscription is still live: no wire teardown
+        // (`notifications/cancelled`) was sent, and the per-listen state
+        // entry is still registered.
+        expect(written.some(m => (m as JSONRPCNotification).method === 'notifications/cancelled')).toBe(false);
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(1);
+        expect(errors).toHaveLength(0);
+        await client.close();
+    });
+
     it('transport closes BEFORE the ack: listen() rejects fast', async () => {
         const { clientTx, serverTx } = await scriptedModernNoAck();
         const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });

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

@@ -1155,6 +1155,43 @@ describe('StreamableHTTPClientTransport', () => {
             expect(fetchMock).toHaveBeenCalledTimes(1);
         });
 
+        it('anySignal fallback removes the sibling listener (no leak on the transport-lifetime signal)', async () => {
+            // ARRANGE — force the manual fallback path (Node 20.0–20.2).
+            const nativeAny = AbortSignal.any;
+            (AbortSignal as { any?: unknown }).any = undefined;
+            try {
+                transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'));
+                const fetchMock = globalThis.fetch as Mock;
+                fetchMock.mockResolvedValue({ ok: true, status: 202, headers: new Headers() });
+                await transport.start();
+
+                const transportSignal = (transport as unknown as { _abortController: AbortController })._abortController.signal;
+                const addSpy = vi.spyOn(transportSignal, 'addEventListener');
+                const removeSpy = vi.spyOn(transportSignal, 'removeEventListener');
+
+                // ACT — N sends each with a fresh request-scoped signal that
+                // aborts after the send completes (the McpSubscription.close()
+                // pattern). Each send registers one fallback listener on the
+                // transport-lifetime signal; aborting the request-scoped
+                // signal must remove it.
+                for (let i = 0; i < 5; i++) {
+                    const requestAbort = new AbortController();
+                    await transport.send(
+                        { jsonrpc: '2.0', method: 'subscriptions/listen', id: `listen-${i}`, params: {} },
+                        { requestSignal: requestAbort.signal }
+                    );
+                    requestAbort.abort();
+                }
+
+                // ASSERT — every listener registered on the transport-lifetime
+                // signal was removed; nothing accrues per send().
+                expect(addSpy.mock.calls.length).toBeGreaterThan(0);
+                expect(removeSpy.mock.calls.length).toBe(addSpy.mock.calls.length);
+            } finally {
+                (AbortSignal as { any?: unknown }).any = nativeAny;
+            }
+        });
+
         it('should NOT reconnect a POST stream when error response was received', async () => {
             // ARRANGE
             transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {