fix(client): guard listen() connectivity before park; auto-open failure surfaces via onerror

Two related hardening fixes:

- listen() now checks this.transport before any setup. _parkRequest throws
  NotConnected synchronously, and it was called inside the ack-promise
  executor AFTER the ack timer was armed — so a not-connected listen() leaked
  a timer whose callback then dereferenced an unassigned park handle. The
  guard returns the rejection as the listen() promise with no setup started.

- ClientOptions.listChanged auto-open during _connectNegotiated no longer
  fails connect when listen() rejects (server refuses on capacity, does not
  support listen, etc.). The modern connection is fully usable without a
  listen stream; the failure surfaces via onerror and autoOpenedSubscription
  stays undefined.

Tests added for both.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -799,7 +799,15 @@ export class Client extends Protocol<ClientContext> {
                 ...(config.resources && { resourcesListChanged: true as const })
             };
             if (Object.keys(filter).length > 0) {
-                this._autoOpenedSubscription = await this.listen(filter, options);
+                // A failed auto-open MUST NOT fail connect: the modern
+                // 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.
+                try {
+                    this._autoOpenedSubscription = await this.listen(filter, options);
+                } catch (error) {
+                    this.onerror?.(error instanceof Error ? error : new Error(String(error)));
+                }
             }
         }
     }
@@ -1147,13 +1155,20 @@ export class Client extends Protocol<ClientContext> {
                 { method: 'subscriptions/listen', protocolVersion: negotiated }
             );
         }
+        // Connectivity is checked here so the rejection is delivered as the
+        // returned promise (no setup or ack timer is started) — `_parkRequest`
+        // would otherwise throw NotConnected from inside the executor below
+        // after the timer is armed.
+        if (this.transport === undefined) {
+            throw new SdkError(SdkErrorCode.NotConnected, 'Not connected');
+        }
 
         if (this._onParkedNotification === undefined) {
             this._onParkedNotification = raw => this._listenFirstLook(raw);
         }
 
         const requestAbort = new AbortController();
-        const transportKind = this.transport === undefined ? 'http' : detectProbeTransportKind(this.transport);
+        const transportKind = detectProbeTransportKind(this.transport);
 
         let closed = false;
         let parked!: ReturnType<typeof this._parkRequest>;

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

@@ -166,6 +166,19 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('rejects with NotConnected (as a rejected promise, no setup) when no transport is connected', async () => {
+        const { clientTx } = await scriptedModern();
+        const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
+        await client.connect(clientTx);
+        await client.close();
+        // listen() is async, so a pre-send guard throw is delivered as the
+        // returned promise's rejection (no ack timer started, no park state).
+        const pending = client.listen({ toolsListChanged: true });
+        const error = await pending.catch(e => e as SdkError);
+        expect(error).toBeInstanceOf(SdkError);
+        expect((error as SdkError).code).toBe(SdkErrorCode.NotConnected);
+    });
+
     it('ClientOptions.listChanged auto-opens a listen stream on a modern connection (filter derived from sub-options)', async () => {
         const filters: unknown[] = [];
         const { clientTx } = await scriptedModern((_id, filter) => filters.push(filter));
@@ -181,4 +194,41 @@ describe('Client.listen()', () => {
         await client.autoOpenedSubscription!.close();
         await client.close();
     });
+
+    it('a failed auto-open surfaces via onerror and does NOT fail connect', 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) {
+                // Server refuses listen (capacity guard / not supported).
+                void serverTx.send({ jsonrpc: '2.0', id: req.id, error: { code: -32_603, message: 'Subscription limit reached' } });
+            }
+        };
+        await serverTx.start();
+        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);
+        // connect MUST resolve: the modern connection is usable without listen.
+        await client.connect(clientTx);
+        expect(client.autoOpenedSubscription).toBeUndefined();
+        expect(errors).toHaveLength(1);
+        expect((errors[0] as { code?: number }).code).toBe(-32_603);
+        await client.close();
+    });
 });