fix(client): listen polish — derived ack-wait signal; envelope on listen-cancel; intentional-abort guards across reconnect path; typed result-during-opening

connect()'s auto-open now derives a one-shot AbortController bound to
connect's signal only for the listen() ack-wait duration (listener
removed in finally; already-aborted handled), so an aborted connect
rejects fast instead of blocking for the ack timeout, while the
auto-opened subscription still outlives connect's signal.

wireTeardown's notifications/cancelled now spreads
_outboundMetaEnvelope() into params._meta — the listen-path cancel was
the only modern outbound bypassing the auto-envelope.

streamableHttp: the intentional per-request abort guard now also covers
_send()'s catch (abort before headers), _startOrAuthSse()'s GET fetch
signal + catch, and _scheduleReconnection()'s timer-fire / catch — a
closed listen subscription on a flaky network can no longer be
resurrected as a GET stream nor surface a misleading onerror.

parked.terminated 'response' surfaces a typed SdkErrorCode.InvalidResult
('server answered subscriptions/listen with a result; expected the
acknowledged notification') instead of a generic stream-ended message.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -927,20 +927,37 @@ export class Client extends Protocol<ClientContext> {
                 // 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.
+                // lifetime, so connect()'s signal must NOT be forwarded
+                // verbatim — a connect-scoped `AbortSignal.timeout(30_000)`
+                // would silently tear the auto-opened stream down the moment
+                // it fires after connect has resolved. But connect()'s signal
+                // MUST still cancel the in-connect ack WAIT (otherwise an
+                // aborted connect blocks here for the full ack timeout).
+                // Derived one-shot: bound to connect()'s signal only for the
+                // duration of the listen() await; the listener is removed in
+                // `finally` so the auto-opened subscription outlives connect's
+                // signal.
+                const ackAbort = new AbortController();
+                const onConnectAbort = (): void => ackAbort.abort(options?.signal?.reason);
+                // Handle the already-aborted case (aborted between the
+                // discover leg resolving and now): the listener never fires
+                // for a past event.
+                if (options?.signal?.aborted) onConnectAbort();
+                options?.signal?.addEventListener('abort', onConnectAbort);
                 try {
-                    this._autoOpenedSubscription = await this.listen(
-                        filter,
-                        options?.timeout === undefined ? undefined : { timeout: options.timeout }
-                    );
+                    this._autoOpenedSubscription = await this.listen(filter, {
+                        timeout: options?.timeout,
+                        signal: ackAbort.signal
+                    });
                 } catch (error) {
+                    // Connect-signal abort during the ack wait propagates as a
+                    // connect() rejection (caller asked to abort connect); a
+                    // server-side refusal stays a soft onerror.
+                    if (options?.signal?.aborted) throw error;
                     this.onerror?.(error instanceof Error ? error : new Error(String(error)));
+                } finally {
+                    options?.signal?.removeEventListener('abort', onConnectAbort);
                 }
             }
         }
@@ -1404,8 +1421,16 @@ export class Client extends Protocol<ClientContext> {
             requestAbort.abort();
             const id = listenMessageId;
             if (id !== undefined) {
+                // Carry the same modern auto-envelope as every other outbound
+                // (request()'s cancel and Protocol.notification() both go via
+                // `_envelopeOutbound`); the listen path was the only outbound
+                // bypassing it.
                 await this.transport
-                    ?.send({ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: id } })
+                    ?.send({
+                        jsonrpc: '2.0',
+                        method: 'notifications/cancelled',
+                        params: { _meta: { ...this._outboundMetaEnvelope() }, requestId: id }
+                    })
                     .catch(() => {});
             }
         };
@@ -1485,9 +1510,22 @@ export class Client extends Protocol<ClientContext> {
             // Pre-ack capacity / params rejection arrives as a JSON-RPC error
             // for the listen id; transport close is delivered the same way.
             // A 'response' (the spec defines listen as never receiving a
-            // result) is treated as the server having ended the stream.
+            // result) surfaces as a typed protocol-shape error so a server
+            // bug — answering listen with a JSON-RPC result instead of the
+            // acknowledged notification — is diagnosable, not a 60s ack
+            // timeout.
             void parked.terminated.then(({ reason, error }) => {
                 if (reason === 'unparked') return;
+                if (reason === 'response') {
+                    settle({
+                        cause: 'remote',
+                        error: new SdkError(
+                            SdkErrorCode.InvalidResult,
+                            'server answered subscriptions/listen with a result; expected the acknowledged notification'
+                        )
+                    });
+                    return;
+                }
                 settle({ cause: 'remote', error: error ?? new Error('subscriptions/listen: stream ended') });
             });
             parked.sent.catch(error => {

packages/client/src/client/streamableHttp.ts

@@ -300,7 +300,13 @@ export class StreamableHTTPClientTransport implements Transport {
     }
 
     private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false): Promise<void> {
-        const { resumptionToken } = options;
+        const { resumptionToken, requestSignal } = options;
+        // Same guard as `_handleSseStream`: a resurrected listen stream (the
+        // POST-SSE → GET reconnect path threads `requestSignal` through
+        // `StartSSEOptions`) must honour the per-request abort exactly as the
+        // original POST did — both as a fetch signal and as a "do not surface
+        // onerror" gate.
+        const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true;
 
         try {
             // Try to open an initial SSE stream with GET to listen for server messages
@@ -315,11 +321,16 @@ export class StreamableHTTPClientTransport implements Transport {
                 headers.set('last-event-id', resumptionToken);
             }
 
+            const transportSignal = this._abortController?.signal;
+            const signal =
+                requestSignal !== undefined && transportSignal !== undefined
+                    ? anySignal(transportSignal, requestSignal)
+                    : (requestSignal ?? transportSignal);
             const response = await (this._fetch ?? fetch)(this._url, {
                 ...this._requestInit,
                 method: 'GET',
                 headers,
-                signal: this._abortController?.signal
+                signal
             });
 
             if (!response.ok) {
@@ -366,7 +377,9 @@ export class StreamableHTTPClientTransport implements Transport {
 
             this._handleSseStream(response.body, options, true);
         } catch (error) {
-            this.onerror?.(error as Error);
+            if (!isIntentionalAbort()) {
+                this.onerror?.(error as Error);
+            }
             throw error;
         }
     }
@@ -413,8 +426,12 @@ export class StreamableHTTPClientTransport implements Transport {
 
         const reconnect = (): void => {
             this._cancelReconnection = undefined;
-            if (this._abortController?.signal.aborted) return;
+            // Honour BOTH the transport-wide abort and the per-request abort
+            // (a listen subscription closed during the backoff delay): do not
+            // resurrect a stream the caller already tore down.
+            if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return;
             this._startOrAuthSse(options).catch(error => {
+                if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return;
                 this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`));
                 try {
                     this._scheduleReconnection(options, attemptCount + 1);
@@ -780,7 +797,14 @@ export class StreamableHTTPClientTransport implements Transport {
                 await response.text?.().catch(() => {});
             }
         } catch (error) {
-            this.onerror?.(error as Error);
+            // Intentional per-request abort BEFORE response headers (the
+            // `subscriptions/listen` driver aborting its `requestSignal`):
+            // fetch rejects with AbortError. Same guard as
+            // `_handleSseStream`'s `isIntentionalAbort` — do not surface a
+            // misleading onerror; still rethrow so `parked.sent` settles.
+            if (options?.requestSignal?.aborted !== true) {
+                this.onerror?.(error as Error);
+            }
             throw error;
         }
     }

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

@@ -7,7 +7,14 @@
  * connection.
  */
 import type { JSONRPCMessage, JSONRPCNotification } from '@modelcontextprotocol/core';
-import { InMemoryTransport, LATEST_PROTOCOL_VERSION, SdkError, SdkErrorCode, SUBSCRIPTION_ID_META_KEY } from '@modelcontextprotocol/core';
+import {
+    InMemoryTransport,
+    LATEST_PROTOCOL_VERSION,
+    PROTOCOL_VERSION_META_KEY,
+    SdkError,
+    SdkErrorCode,
+    SUBSCRIPTION_ID_META_KEY
+} from '@modelcontextprotocol/core';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { Client } from '../../src/client/client.js';
@@ -143,7 +150,13 @@ describe('Client.listen()', () => {
         const listenId = (written.find(m => (m as { method?: string }).method === 'subscriptions/listen') as { id: number | string }).id;
         written.length = 0;
         await sub.close();
-        expect(written).toEqual([{ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: listenId } }]);
+        expect(written).toHaveLength(1);
+        const cancel = written[0] as unknown as { method: string; params: { requestId: unknown; _meta?: Record<string, unknown> } };
+        expect(cancel.method).toBe('notifications/cancelled');
+        expect(cancel.params.requestId).toBe(listenId);
+        // The listen-path cancel carries the same modern auto-envelope as
+        // every other outbound (request()'s cancel, Protocol.notification()).
+        expect(cancel.params._meta?.[PROTOCOL_VERSION_META_KEY]).toBe(MODERN);
         // Idempotent.
         await sub.close();
         expect(written).toHaveLength(1);
@@ -422,7 +435,9 @@ describe('Client.listen()', () => {
         written.length = 0;
         ac.abort();
         await flush();
-        expect(written).toEqual([{ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: listenId } }]);
+        expect(written).toHaveLength(1);
+        expect((written[0] as JSONRPCNotification).method).toBe('notifications/cancelled');
+        expect((written[0] as unknown as { params: { requestId: unknown } }).params.requestId).toBe(listenId);
         // Caller-signal abort is consumer-initiated → 'local'.
         await expect(sub.closed).resolves.toBe('local');
         // close() after signal-abort is idempotent.
@@ -575,6 +590,66 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('connect-scoped signal aborted DURING the auto-open ack wait: connect rejects fast (no 60s hang)', async () => {
+        // Regression: forwarding only {timeout} into the auto-open listen()
+        // meant connect()'s signal could not cancel the in-connect ack wait —
+        // an aborted connect blocked here for the full ack timeout.
+        const { clientTx } = await scriptedModernNoAck();
+        const onChanged = () => {};
+        const client = new Client(
+            { name: 'c', version: '1' },
+            { versionNegotiation: { mode: 'auto' }, listChanged: { tools: { onChanged } } }
+        );
+        const connectScoped = new AbortController();
+        const t0 = Date.now();
+        const pending = client.connect(clientTx, { signal: connectScoped.signal });
+        // discover resolves; connect is now awaiting the auto-open ack.
+        await flush();
+        connectScoped.abort(new Error('connect-abort'));
+        const error = await pending.catch(e => e as Error);
+        expect(error).toBeInstanceOf(Error);
+        expect(Date.now() - t0).toBeLessThan(1000);
+        // No leaked per-listen state on the aborted connect.
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(0);
+        await client.close();
+    });
+
+    it('server answers listen with a JSON-RPC RESULT during opening: rejects with a typed InvalidResult (not 60s)', async () => {
+        const [clientTx, serverTx] = InMemoryTransport.createLinkedPair();
+        serverTx.onmessage = m => {
+            const req = m as { id?: number | string; method?: string };
+            if (req.method === 'server/discover' && req.id !== undefined) {
+                void serverTx.send({
+                    jsonrpc: '2.0',
+                    id: req.id,
+                    result: {
+                        resultType: 'complete',
+                        supportedVersions: [MODERN],
+                        capabilities: { tools: { listChanged: true } },
+                        serverInfo: { name: 's', version: '1' }
+                    }
+                });
+            }
+            if (req.method === 'subscriptions/listen' && req.id !== undefined) {
+                // Buggy server: answers with a result instead of the
+                // acknowledged notification. Spec defines listen as never
+                // receiving a result.
+                void serverTx.send({ jsonrpc: '2.0', id: req.id, result: {} });
+            }
+        };
+        await serverTx.start();
+        const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
+        await client.connect(clientTx);
+        const t0 = Date.now();
+        const error = await client.listen({ toolsListChanged: true }).catch(e => e as SdkError);
+        expect(error).toBeInstanceOf(SdkError);
+        expect((error as SdkError).code).toBe(SdkErrorCode.InvalidResult);
+        expect((error as SdkError).message).toContain('expected the acknowledged notification');
+        expect(Date.now() - t0).toBeLessThan(1000);
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(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' } });
@@ -624,7 +699,9 @@ describe('Client.listen()', () => {
         written.length = 0;
         await a.close();
         // Only `a`'s id is cancelled; `b` stays open.
-        expect(written).toEqual([{ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: ids[0] } }]);
+        expect(written).toHaveLength(1);
+        expect((written[0] as JSONRPCNotification).method).toBe('notifications/cancelled');
+        expect((written[0] as unknown as { params: { requestId: unknown } }).params.requestId).toBe(ids[0]);
         expect(listenState.size).toBe(1);
         await b.close();
         expect(listenState.size).toBe(0);

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

@@ -1155,6 +1155,40 @@ describe('StreamableHTTPClientTransport', () => {
             expect(fetchMock).toHaveBeenCalledTimes(1);
         });
 
+        it('per-request requestSignal abort BEFORE response headers: no misleading onerror; send() still rejects', async () => {
+            // ARRANGE — fetch is in flight (pending promise) when the
+            // requestSignal aborts; fetch rejects with AbortError before the
+            // SSE stream handler ever runs. _send's catch must apply the same
+            // intentional-abort guard as _handleSseStream.
+            transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'));
+            const errorSpy = vi.fn();
+            transport.onerror = errorSpy;
+            const fetchMock = globalThis.fetch as Mock;
+            fetchMock.mockImplementationOnce(
+                (_url, init: RequestInit) =>
+                    new Promise((_resolve, reject) => {
+                        init.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true });
+                    })
+            );
+
+            const requestAbort = new AbortController();
+            await transport.start();
+            const sent = transport.send(
+                { jsonrpc: '2.0', method: 'subscriptions/listen', id: 'listen-1', params: {} },
+                { requestSignal: requestAbort.signal }
+            );
+            // Let _send reach the in-flight fetch.
+            await vi.advanceTimersByTimeAsync(0);
+            expect(fetchMock).toHaveBeenCalledTimes(1);
+
+            // ACT — abort before headers.
+            requestAbort.abort(new Error('intentional'));
+
+            // ASSERT — send() rejects (so parked.sent settles), but no onerror.
+            await expect(sent).rejects.toThrow();
+            expect(errorSpy).not.toHaveBeenCalled();
+        });
+
         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;