fix(client): listen() honors RequestOptions.signal

listen(filter, options?) is typed RequestOptions but only options.timeout
was honored; options.signal was silently ignored. The auto-open path
forwards connect's options into listen(), so a connect-time signal was
also dropped.

Honor options.signal in the state machine: an already-aborted signal
rejects synchronously (mirrors request()); an abort while `opening`
rejects the pending listen() promise with the signal's reason and tears
the wire down; an abort while `open` closes the subscription. The abort
listener is removed by settle() once the subscription has closed.

Author: felixweinberger

packages/client/src/client/client.ts

@@ -1260,6 +1260,10 @@ export class Client extends Protocol<ClientContext> {
             );
         }
 
+        // Honor RequestOptions.signal exactly as request() does: an
+        // already-aborted signal rejects synchronously before any setup.
+        options?.signal?.throwIfAborted();
+
         if (this._onParkedNotification === undefined) {
             this._onParkedNotification = raw => this._listenFirstLook(raw);
         }
@@ -1281,6 +1285,7 @@ export class Client extends Protocol<ClientContext> {
         // leak the entry.
         let listenMessageId: number | undefined;
         let ackTimer: ReturnType<typeof setTimeout> | undefined;
+        let onCallerAbort: (() => void) | undefined;
         let resolveOpening!: (honored: SubscriptionFilter) => void;
         let rejectOpening!: (error: Error) => void;
         const opening = new Promise<SubscriptionFilter>((resolve, reject) => {
@@ -1303,6 +1308,9 @@ export class Client extends Protocol<ClientContext> {
                 return;
             }
             state = 'closed';
+            if (onCallerAbort !== undefined) {
+                options?.signal?.removeEventListener('abort', onCallerAbort);
+            }
             if (listenMessageId !== undefined) {
                 this._listenState.delete(listenMessageId);
             }
@@ -1347,6 +1355,22 @@ export class Client extends Protocol<ClientContext> {
             void wireTeardown().catch(() => {});
         }, ackTimeout);
 
+        // RequestOptions.signal aborts the subscription at any point in its
+        // lifecycle (mirrors request()'s cancel path). While `opening`, settle
+        // rejects the pending listen() promise with the signal's reason; while
+        // `open`, it transitions to `closed` and tears the wire down. The
+        // listener is removed by `settle()` once the subscription has closed.
+        if (options?.signal) {
+            const callerSignal = options.signal;
+            onCallerAbort = () => {
+                if (state === 'closed') return;
+                const reason = callerSignal.reason;
+                settle({ error: reason instanceof Error ? reason : new Error(String(reason ?? 'Aborted')) });
+                void wireTeardown().catch(() => {});
+            };
+            callerSignal.addEventListener('abort', onCallerAbort, { once: true });
+        }
+
         try {
             // The per-subscription state is registered BEFORE the request is
             // sent (`onBeforeSend`) so a synchronously-delivered ack (an

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

@@ -285,6 +285,66 @@ describe('Client.listen()', () => {
         await client.close();
     });
 
+    it('options.signal aborted while opening: listen() rejects fast with the signal reason', async () => {
+        const [clientTx, serverTx] = InMemoryTransport.createLinkedPair();
+        const written: JSONRPCMessage[] = [];
+        serverTx.onmessage = m => {
+            written.push(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: {},
+                        serverInfo: { name: 's', version: '1' }
+                    }
+                });
+            }
+            // No ack for subscriptions/listen — stays in `opening`.
+        };
+        await serverTx.start();
+        const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
+        await client.connect(clientTx);
+        const ac = new AbortController();
+        const t0 = Date.now();
+        const pending = client.listen({ toolsListChanged: true }, { signal: ac.signal });
+        ac.abort(new Error('caller-abort'));
+        const error = await pending.catch(e => e as Error);
+        expect((error as Error).message).toBe('caller-abort');
+        expect(Date.now() - t0).toBeLessThan(1000);
+        // wireTeardown sent notifications/cancelled referencing the listen id.
+        await flush();
+        const listenId = (written.find(m => (m as { method?: string }).method === 'subscriptions/listen') as { id: number | string }).id;
+        const cancelled = written.find(m => (m as { method?: string }).method === 'notifications/cancelled') as
+            | { params: { requestId: unknown } }
+            | undefined;
+        expect(cancelled?.params.requestId).toBe(listenId);
+        // No leaked state.
+        expect((client as unknown as { _listenState: Map<unknown, unknown> })._listenState.size).toBe(0);
+        expect((client as unknown as { _responseHandlers: Map<unknown, unknown> })._responseHandlers.size).toBe(0);
+        await client.close();
+    });
+
+    it('options.signal aborted while open: closes the subscription (notifications/cancelled sent)', async () => {
+        const { clientTx, written } = await scriptedModern();
+        const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } });
+        await client.connect(clientTx);
+        const ac = new AbortController();
+        const sub = await client.listen({ toolsListChanged: true }, { signal: ac.signal });
+        const listenId = (written.find(m => (m as { method?: string }).method === 'subscriptions/listen') as { id: number | string }).id;
+        written.length = 0;
+        ac.abort();
+        await flush();
+        expect(written).toEqual([{ jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: listenId } }]);
+        // close() after signal-abort is idempotent.
+        await sub.close();
+        expect(written).toHaveLength(1);
+        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' } });