fix(client): fire onRequestStreamEnd on the GET-resume 405/null-body terminal; skip auto-open when handler registration throws; await transport close on connect-abort during the auto-open ack wait; document the protected-override contract

Author: felixweinberger

.changeset/subscriptions-listen-client.md

@@ -3,4 +3,4 @@
 '@modelcontextprotocol/client': minor
 ---
 
-`Client.listen(filter)` opens a `subscriptions/listen` stream on a 2026-07-28-era connection, resolving once the server's acknowledged notification arrives with an `McpSubscription { honoredFilter, close(), closed }`. `closed` is a `Promise<'local' | 'remote'>` that resolves exactly once when the subscription terminates (`'local'` = you called `close()`; `'remote'` = the server cancelled, the stream ended, or the transport dropped — re-listen if you still want events) and never rejects. Change notifications delivered on the stream dispatch to the existing `setNotificationHandler` registrations — the same handlers the 2025-era unsolicited notifications fire on a legacy connection — so `listen()` is era-transparent for consumers that already register those. `close()` aborts the listen request's stream (where the transport supports it) and sends `notifications/cancelled` referencing the listen id — both, on every transport; no automatic re-listen. On a 2025-era connection `listen()` throws a typed `MethodNotSupportedByProtocolVersion` steering to `resources/subscribe` and `ClientOptions.listChanged`. `ClientOptions.listChanged` now auto-opens a listen stream on a modern connection — the filter is the intersection of the configured sub-options and the server-advertised `listChanged` capabilities; auto-open is skipped (`client.autoOpenedSubscription` stays `undefined`) when that intersection is empty; otherwise the auto-opened subscription is exposed at `client.autoOpenedSubscription`. `TransportSendOptions` gains `requestSignal` for per-request abort on the Streamable HTTP transport.
+`Client.listen(filter)` opens a `subscriptions/listen` stream on a 2026-07-28-era connection, resolving once the server's acknowledged notification arrives with an `McpSubscription { honoredFilter, close(), closed }`. `closed` is a `Promise<'local' | 'remote'>` that resolves exactly once when the subscription terminates (`'local'` = you called `close()`; `'remote'` = the server cancelled, the stream ended, or the transport dropped — re-listen if you still want events) and never rejects. Change notifications delivered on the stream dispatch to the existing `setNotificationHandler` registrations — the same handlers the 2025-era unsolicited notifications fire on a legacy connection — so `listen()` is era-transparent for consumers that already register those. `close()` aborts the listen request's stream (where the transport supports it) and sends `notifications/cancelled` referencing the listen id — both, on every transport; no automatic re-listen. On a 2025-era connection `listen()` throws a typed `MethodNotSupportedByProtocolVersion` steering to `resources/subscribe` and `ClientOptions.listChanged`. `ClientOptions.listChanged` now auto-opens a listen stream on a modern connection — the filter is the intersection of the configured sub-options and the server-advertised `listChanged` capabilities; auto-open is skipped (`client.autoOpenedSubscription` stays `undefined`) when that intersection is empty; otherwise the auto-opened subscription is exposed at `client.autoOpenedSubscription`. `TransportSendOptions` gains `requestSignal` (per-request abort) and `onRequestStreamEnd` (fires when a per-request response stream ends or errors for any non-deliberate reason) on the Streamable HTTP transport.

packages/client/src/client/client.ts

@@ -932,17 +932,24 @@ export class Client extends Protocol<ClientContext> {
             // throw on misconfiguration; the modern connection IS established
             // at this point and is fully usable without listChanged handlers,
             // so a misconfiguration surfaces via onerror and connect resolves
-            // (matching the auto-open soft-fail posture).
+            // (matching the auto-open soft-fail posture). When registration
+            // fails the auto-open is SKIPPED — opening a listen stream for
+            // types whose handler never registered would consume a server
+            // slot to deliver notifications nothing handles.
+            let handlersRegistered = true;
             try {
                 this._setupListChangedHandlers(effective);
             } catch (error) {
+                handlersRegistered = false;
                 this.onerror?.(error instanceof Error ? error : new Error(String(error)));
             }
-            const filter: SubscriptionFilter = {
-                ...(effective.tools && { toolsListChanged: true as const }),
-                ...(effective.prompts && { promptsListChanged: true as const }),
-                ...(effective.resources && { resourcesListChanged: true as const })
-            };
+            const filter: SubscriptionFilter = handlersRegistered
+                ? {
+                      ...(effective.tools && { toolsListChanged: true as const }),
+                      ...(effective.prompts && { promptsListChanged: true as const }),
+                      ...(effective.resources && { resourcesListChanged: true as const })
+                  }
+                : {};
             if (Object.keys(filter).length > 0) {
                 // A failed auto-open MUST NOT fail connect: the modern
                 // connection is fully usable without a listen stream (the
@@ -980,7 +987,7 @@ export class Client extends Protocol<ClientContext> {
                     // half-open connection. A server-side refusal stays a
                     // soft onerror (connect succeeds, no listen stream).
                     if (options?.signal?.aborted) {
-                        void this.close();
+                        await this.close().catch(() => {});
                         throw error;
                     }
                     this.onerror?.(error instanceof Error ? error : new Error(String(error)));

packages/client/src/client/streamableHttp.ts

@@ -375,6 +375,15 @@ export class StreamableHTTPClientTransport implements Transport {
                 // 405 indicates that the server does not offer an SSE stream at GET endpoint
                 // This is an expected case that should not trigger an error
                 if (response.status === 405) {
+                    // A 405 on the standalone-GET path is benign (the caller
+                    // never had a per-request stream). On the POST→GET resume
+                    // path it is a TERMINAL non-resumable outcome for a
+                    // per-request stream the caller is observing — fire the
+                    // stream-end callback so the caller can settle (otherwise
+                    // a resumed listen subscription dead-ends silently). The
+                    // standalone-GET callers never pass `onRequestStreamEnd`,
+                    // so this is a no-op for them.
+                    options.onRequestStreamEnd?.();
                     return;
                 }
 
@@ -463,6 +472,11 @@ export class StreamableHTTPClientTransport implements Transport {
 
     private _handleSseStream(stream: ReadableStream<Uint8Array> | null, options: StartSSEOptions, isReconnectable: boolean): void {
         if (!stream) {
+            // A null body on a per-request stream (or its GET resume) is the
+            // same terminal non-resumable outcome as a 405 — fire the
+            // stream-end callback so the caller can settle. No-op for
+            // standalone-GET callers (they never pass `onRequestStreamEnd`).
+            options.onRequestStreamEnd?.();
             return;
         }
         const { onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd } = options;

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

@@ -553,6 +553,31 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('a misconfigured listChanged handler surfaces via onerror and SKIPS auto-open (no wire write)', async () => {
+        // Regression: when handler registration threw (the soft-fail catch),
+        // the auto-open filter was still built from the same `effective`,
+        // opening a listen stream for types whose handler never registered —
+        // delivered notifications dropped on the floor while consuming a
+        // server slot. Now a registration failure skips auto-open entirely.
+        const { clientTx, written } = await scriptedModernNoAck();
+        const onChanged = () => {};
+        const client = new Client(
+            { name: 'c', version: '1' },
+            { versionNegotiation: { mode: 'auto' }, listChanged: { tools: { onChanged, debounceMs: -1 } } }
+        );
+        const errors: Error[] = [];
+        client.onerror = e => errors.push(e);
+        // connect MUST resolve: the modern connection is usable without listen.
+        await client.connect(clientTx);
+        expect(errors).toHaveLength(1);
+        expect(errors[0]!.message).toContain('Invalid tools listChanged options');
+        // Auto-open SKIPPED: no listen request hit the wire, no subscription.
+        expect(client.autoOpenedSubscription).toBeUndefined();
+        expect(written.some(m => (m as { method?: string }).method === 'subscriptions/listen')).toBe(false);
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(0);
+        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
@@ -591,6 +616,7 @@ describe('Client.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 closeSpy = vi.spyOn(clientTx, 'close');
         const onChanged = () => {};
         const client = new Client(
             { name: 'c', version: '1' },
@@ -607,6 +633,11 @@ describe('Client.listen()', () => {
         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);
+        // A connect() rejection MUST NOT leave a half-open connection: the
+        // transport was closed before rethrowing (b142b80ea regression assertion).
+        await flush();
+        expect(closeSpy).toHaveBeenCalled();
+        expect(client.transport).toBeUndefined();
         await client.close();
     });