refactor(core): extract connection-scoped state into a private Connection owner

packages/client/src/client/client.ts

@@ -518,6 +518,18 @@ export class Client extends Protocol<ClientContext> {
     private _autoOpenedSubscription?: McpSubscription;
     /** Backing store for {@linkcode getDiscoverResult}. Per-connection. */
     private _discoverResult?: DiscoverResult;
+    /**
+     * The protocol version to restore when resuming a server session on a new
+     * transport. Written when a connection establishes (the negotiated
+     * version must outlive the connection — a transport drop does not end the
+     * server session); cleared by {@linkcode _resetConnectionState}
+     * (close / fresh connect). The resume branches also fall back to the
+     * version the resuming transport itself carries
+     * ({@linkcode Transport.protocolVersion}), so resuming AFTER close() —
+     * where the consumer persisted sessionId + protocolVersion — still pushes
+     * the header.
+     */
+    private _sessionResumption?: { version: string };
 
     /**
      * Clears every per-connection field in one place. Called at the start of
@@ -527,6 +539,7 @@ export class Client extends Protocol<ClientContext> {
      */
     private _resetConnectionState(): void {
         this._negotiatedProtocolVersion = undefined;
+        this._sessionResumption = undefined;
         this._serverCapabilities = undefined;
         this._serverVersion = undefined;
         this._instructions = undefined;
@@ -945,10 +958,15 @@ export class Client extends Protocol<ClientContext> {
         // the originally negotiated version — from this instance, or from the
         // transport itself after close() wiped the instance's copy.
         if (transport.sessionId !== undefined) {
-            const version = this._negotiatedProtocolVersion ?? transport.protocolVersion;
-            if (version !== undefined) {
-                this._negotiatedProtocolVersion = version;
-                transport.setProtocolVersion?.(version);
+            // Resuming keeps the original negotiation: the version recorded
+            // when the session's connection established (or, after close(),
+            // carried by the resuming transport itself) seeds the new
+            // connection's era and is pushed onto the transport again.
+            const negotiatedProtocolVersion = this._sessionResumption?.version ?? transport.protocolVersion;
+            if (negotiatedProtocolVersion !== undefined) {
+                this._negotiatedProtocolVersion = negotiatedProtocolVersion;
+                this._sessionResumption = { version: negotiatedProtocolVersion };
+                transport.setProtocolVersion?.(negotiatedProtocolVersion);
             }
             return;
         }
@@ -1020,6 +1038,10 @@ export class Client extends Protocol<ClientContext> {
             // EXCHANGE is the legacy handshake by definition and completes on
             // that era.
             this._negotiatedProtocolVersion = result.protocolVersion;
+            // Recorded for later session-resuming reconnects: the negotiated
+            // version must outlive this connection (a transport drop does not
+            // end the server session), but not a close()/fresh handshake.
+            this._sessionResumption = { version: result.protocolVersion };
 
             // Set up list changed handlers now that we know server capabilities
             if (this._listChangedConfig) {
@@ -1046,10 +1068,16 @@ export class Client extends Protocol<ClientContext> {
         // this instance or, after close() wiped it, from the transport itself.
         if (transport.sessionId !== undefined) {
             await super.connect(transport);
-            const version = this._negotiatedProtocolVersion ?? transport.protocolVersion;
-            if (version !== undefined) {
-                this._negotiatedProtocolVersion = version;
-                transport.setProtocolVersion?.(version);
+            // Same restore as the legacy resume branch: recorded at session
+            // establishment, or carried by the resuming transport after a
+            // close().
+            const negotiatedProtocolVersion = this._sessionResumption?.version ?? transport.protocolVersion;
+            if (negotiatedProtocolVersion !== undefined) {
+                this._negotiatedProtocolVersion = negotiatedProtocolVersion;
+                this._sessionResumption = { version: negotiatedProtocolVersion };
+                if (transport.setProtocolVersion) {
+                    transport.setProtocolVersion(negotiatedProtocolVersion);
+                }
             }
             return;
         }
@@ -1090,6 +1118,7 @@ export class Client extends Protocol<ClientContext> {
         this._discoverResult = result.discover;
         // Modern selection: the same connection state the legacy handshake completion sets.
         this._negotiatedProtocolVersion = result.version;
+        this._sessionResumption = { version: result.version };
         // The single setProtocolVersion call site on this path, mirroring the legacy path after initialize.
         if (transport.setProtocolVersion) {
             transport.setProtocolVersion(result.version);
@@ -1210,6 +1239,7 @@ export class Client extends Protocol<ClientContext> {
         this._cache.setServerIdentity(this._deriveServerIdentity(transport));
         this._instructions = prior.instructions;
         this._negotiatedProtocolVersion = version;
+        this._sessionResumption = { version };
         transport.setProtocolVersion?.(version);
 
         // No auto-opened listen stream on this path (request-only workers).

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

@@ -20,6 +20,12 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { Client } from '../../src/client/client';
 
 const MODERN = '2026-07-28';
+
+/** White-box probe: pending response handlers on the client's live connection (0 when disconnected). */
+function responseHandlerCount(client: unknown): number {
+    const connection = (client as { _conn?: { responseHandlers: Map<unknown, unknown> } })._conn;
+    return connection?.responseHandlers.size ?? 0;
+}
 const flush = () => new Promise(r => setTimeout(r, 10));
 
 async function scriptedModern(onListen?: (id: number | string, filter: unknown, send: (m: JSONRPCMessage) => void) => void) {
@@ -374,7 +380,7 @@ describe('Client.listen()', () => {
         // registered before send threw; listen() never registers in
         // Protocol's `_responseHandlers` so there is nothing to leak there.
         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);
+        expect(responseHandlerCount(client)).toBe(0);
         clientTx.send = realSend;
         await client.close();
     });
@@ -739,7 +745,7 @@ describe('Client.listen()', () => {
         expect(error).toBeInstanceOf(Error);
         expect(Date.now() - t0).toBeLessThan(1000);
         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);
+        expect(responseHandlerCount(client)).toBe(0);
     });
 
     it("transport closes WHILE the subscription is open: closed resolves 'remote'; close() is a no-op", async () => {
@@ -991,7 +997,7 @@ describe('Client.listen() — ack timeout (fake timers)', () => {
         expect(cancelled).toMatchObject({ params: { requestId: 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);
+        expect(responseHandlerCount(client)).toBe(0);
         // Restore real timers before close to avoid hanging on transport timers.
         vi.useRealTimers();
         await client.close();

packages/client/test/client/sessionResume.test.ts

@@ -0,0 +1,111 @@
+/**
+ * Session resumption across a client close(): reconnecting with a transport
+ * that carries a sessionId (and the protocol version persisted alongside it)
+ * must push that version onto the transport so HTTP requests carry the
+ * required mcp-protocol-version header.
+ */
+import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal';
+import { isJSONRPCRequest } from '@modelcontextprotocol/core-internal';
+import { describe, expect, test } from 'vitest';
+
+import { Client } from '../../src/client/client';
+
+const LEGACY = '2025-11-25';
+
+/**
+ * A sessionful legacy server transport: answers initialize like a 2025
+ * server and assigns a session id with the initialize response, mirroring
+ * Streamable HTTP's mcp-session-id assignment.
+ */
+class SessionfulTransport {
+    onclose?: () => void;
+    onerror?: (error: Error) => void;
+    onmessage?: (message: JSONRPCMessage) => void;
+    sessionId?: string;
+    protocolVersion?: string;
+
+    sent: JSONRPCMessage[] = [];
+    setProtocolVersionCalls: string[] = [];
+
+    constructor(carried?: { sessionId: string; protocolVersion: string }) {
+        this.sessionId = carried?.sessionId;
+        this.protocolVersion = carried?.protocolVersion;
+    }
+
+    async start(): Promise<void> {}
+
+    async send(message: JSONRPCMessage): Promise<void> {
+        this.sent.push(message);
+        if (isJSONRPCRequest(message) && message.method === 'initialize') {
+            this.sessionId = 'session-123';
+            const reply: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                id: message.id,
+                result: {
+                    protocolVersion: LEGACY,
+                    capabilities: {},
+                    serverInfo: { name: 'sessionful-legacy-server', version: '1.0.0' }
+                }
+            };
+            queueMicrotask(() => this.onmessage?.(reply));
+        }
+    }
+
+    async close(): Promise<void> {
+        this.onclose?.();
+    }
+
+    setProtocolVersion(version: string): void {
+        this.setProtocolVersionCalls.push(version);
+        this.protocolVersion = version;
+    }
+}
+
+describe('r4: session resume after close()', () => {
+    // The resume branch seeds the new connection from the session-resumption
+    // record or, after close() cleared it, from the version the resuming
+    // transport itself carries — so setProtocolVersion is pushed either way.
+    // (It used to read the negotiated version, which close() had reset, and
+    // silently skip the push: the resumed session's HTTP requests went out
+    // without the required mcp-protocol-version header.)
+    test('reconnecting after close() with a carried sessionId + protocolVersion pushes the version onto the new transport', async () => {
+        const client = new Client({ name: 'resume-client', version: '1.0.0' });
+        const first = new SessionfulTransport();
+
+        await client.connect(first);
+        expect(first.setProtocolVersionCalls).toEqual([LEGACY]);
+        expect(first.sessionId).toBe('session-123');
+
+        await client.close();
+
+        // The consumer persisted the session id and negotiated version (the
+        // documented Streamable HTTP resumption flow) and reconnects with
+        // both carried on the new transport.
+        const resumed = new SessionfulTransport({ sessionId: 'session-123', protocolVersion: LEGACY });
+        await client.connect(resumed);
+
+        // No re-initialize on a resume...
+        expect(resumed.sent.filter(m => isJSONRPCRequest(m) && m.method === 'initialize')).toHaveLength(0);
+        // ...and the carried version is pushed to the transport.
+        expect(resumed.setProtocolVersionCalls).toEqual([LEGACY]);
+
+        await client.close();
+    });
+
+    test('resume on the same instance without close() re-pushes the negotiated version (pin)', async () => {
+        const client = new Client({ name: 'resume-client', version: '1.0.0' });
+        const first = new SessionfulTransport();
+
+        await client.connect(first);
+        // The transport drops (server side went away) — no client.close().
+        await first.close();
+
+        const resumed = new SessionfulTransport({ sessionId: 'session-123', protocolVersion: LEGACY });
+        await client.connect(resumed);
+
+        expect(resumed.sent.filter(m => isJSONRPCRequest(m) && m.method === 'initialize')).toHaveLength(0);
+        expect(resumed.setProtocolVersionCalls).toEqual([LEGACY]);
+
+        await client.close();
+    });
+});