From 27415497d32a37a650efd37ef09710e0cb175c31 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 20 Jul 2026 15:02:06 +0000 Subject: [PATCH] fix(client): handle connection close during the version negotiation probe A stdio child that exits on the unrecognized server/discover probe is a legacy signal, symmetric with the stdio probe-timeout rule: respawn the child once via the transport's public close()/start() and run the plain initialize handshake there. The respawn is gated on a close-provenance stamp (internal well-known symbol, implemented by StdioClientTransport): a transport the caller closed mid-probe, or one that does not report provenance, rejects with the typed connect error instead, and HTTP-class mid-probe closes keep rejecting typed. StdioClientTransport.start() clears the read buffer, and close() detaches the child's stderr pipe and ends the aggregate stream once teardown completes, so a dead predecessor cannot corrupt the next child's first frame, stderr readers of a closed transport always see end, and a restarted life begins on a fresh stream. --- .changeset/probe-close-handling.md | 5 + docs/migration/support-2026-07-28.md | 6 + docs/protocol-versions.md | 2 + examples/cli-client/host/host.ts | 9 +- packages/client/src/client/probeClassifier.ts | 17 +- packages/client/src/client/stdio.ts | 30 ++ .../client/src/client/versionNegotiation.ts | 101 +++++- .../test/client/probeClassifier.test.ts | 16 + packages/client/test/client/stdio.test.ts | 137 ++++++++ .../test/client/versionNegotiation.test.ts | 298 +++++++++++++++++- .../test/client/versionNegotiation.test.ts | 112 ++++++- 11 files changed, 721 insertions(+), 12 deletions(-) create mode 100644 .changeset/probe-close-handling.md diff --git a/.changeset/probe-close-handling.md b/.changeset/probe-close-handling.md new file mode 100644 index 0000000000..d957303218 --- /dev/null +++ b/.changeset/probe-close-handling.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Handle a connection that closes during the `server/discover` version-negotiation probe. On stdio, a child that exits on the unrecognized probe is a legacy signal (symmetric with the stdio probe-timeout rule): under `versionNegotiation: { mode: 'auto' }` the client now respawns it once via the transport's own `close()`/`start()` and completes the plain `initialize` handshake there (the exit must close the child's stdio pipes to register as a close; an exit hidden behind a helper process holding them open falls to the probe-timeout path instead) — stdio servers built on SDKs that terminate on any pre-`initialize` request (the official Rust SDK, rmcp) previously hard-failed under `'auto'`. The respawn happens only for transports that positively report close provenance (an internal well-known-symbol stamp implemented by `StdioClientTransport`): a transport the caller closed mid-probe, or a custom transport that does not track provenance, rejects with the typed connect error instead — and on HTTP-class transports a mid-probe close keeps rejecting with the typed connect error. `StdioClientTransport.start()` now also clears the read buffer, and `close()` detaches the child's stderr pipe and ends the aggregate stream once teardown completes, so a dead predecessor cannot corrupt the next child's first frame, `stderr` readers of a closed transport always see `end`, and a restarted life begins on a fresh stream (re-read the `stderr` getter). diff --git a/docs/migration/support-2026-07-28.md b/docs/migration/support-2026-07-28.md index fdc221d930..1ffefb34ad 100644 --- a/docs/migration/support-2026-07-28.md +++ b/docs/migration/support-2026-07-28.md @@ -82,6 +82,12 @@ a dead HTTP server is never misreported as legacy. One browser-specific exceptio opaque CORS/preflight `TypeError` during the probe falls back to the legacy era, because deployed 2025 servers commonly have CORS allow-lists that predate the 2026 headers. +One behavior change to the transport itself: with `stderr: 'pipe'`, +`StdioClientTransport.close()` now ends the aggregate stderr stream when its teardown +completes (previously the stream ended only when the child's own stderr closed — never, +if that pipe outlived `close()`), and a `start()` after `close()` begins a fresh +stream — re-read the `stderr` getter. + ```typescript versionNegotiation: { mode: 'auto', diff --git a/docs/protocol-versions.md b/docs/protocol-versions.md index 6b06754ab5..05b25225f3 100644 --- a/docs/protocol-versions.md +++ b/docs/protocol-versions.md @@ -101,6 +101,8 @@ const cli = new Client( A probe timeout is transport-aware. On stdio a silent server is a legacy server, so `connect()` falls back to `initialize` on the same stream; on HTTP silence is an outage, so `connect()` rejects with `SdkError(RequestTimeout)` instead of misreporting a dead server as legacy. One browser exception: an opaque CORS `TypeError` during the probe falls back to the legacy era, because deployed 2025 servers commonly have allow-lists that predate the 2026 headers. +A connection that closes mid-probe follows the same transport split. On stdio a child that exits on the unrecognized probe is a legacy server — stdio servers built on SDKs that terminate on any pre-`initialize` request (the official Rust SDK, rmcp) behave exactly this way — so under `'auto'` the client respawns the child once and falls back to `initialize` there (the exit must close the child's stdio pipes to register as a close — an exit hidden behind a helper process holding them open falls to the probe-timeout path instead): pre-connect `stderr` subscribers should re-read that getter after `connect()` resolves, and a pre-set `onclose` observer will see the probe child's close. The respawn is limited to transports that positively report close provenance (the SDK's own `StdioClientTransport` does): a transport the caller closed during the probe, or a custom stdio-shaped transport that does not track it, is never respawned — and on HTTP a mid-probe close is ambiguous, never era evidence — `connect()` rejects with the same typed `SdkError(EraNegotiationFailed)` as any probe transport failure. + The client's `supportedProtocolVersions` option shapes the probe: its 2026+ entries are the versions the probe offers, and the legacy fallback stays available only while the list keeps a pre-2026 entry. A list with no pre-2026 entry removes the fallback — against a 2025-only server, `connect()` rejects with `SdkError(EraNegotiationFailed)`. ::: warning diff --git a/examples/cli-client/host/host.ts b/examples/cli-client/host/host.ts index 49165c9554..181ae99753 100644 --- a/examples/cli-client/host/host.ts +++ b/examples/cli-client/host/host.ts @@ -513,11 +513,16 @@ export class McpHost { cwd: entry.cwd, stderr: 'pipe' }); - transport.stderr?.on('data', (chunk: Buffer) => { + const logStderr = (chunk: Buffer) => { const line = String(chunk).trim(); if (line) this.ui.serverLog(name, 'stderr', line); - }); + }; + const preConnectStderr = transport.stderr; + preConnectStderr?.on('data', logStderr); await client.connect(transport); + // Under 'auto', a legacy fallback may have respawned the child with a + // fresh stderr stream — re-read the getter and subscribe the new life. + if (transport.stderr !== preConnectStderr) transport.stderr?.on('data', logStderr); } try { diff --git a/packages/client/src/client/probeClassifier.ts b/packages/client/src/client/probeClassifier.ts index 1b78744964..ad7e8cf32f 100644 --- a/packages/client/src/client/probeClassifier.ts +++ b/packages/client/src/client/probeClassifier.ts @@ -48,6 +48,8 @@ export type ProbeOutcome = | { kind: 'network-error'; error: unknown } /** The transport's auth flow challenged during the probe send (`UnauthorizedError`). */ | { kind: 'auth-required'; error: Error } + /** The transport reported close while the probe awaited its reply. */ + | { kind: 'closed' } /** No response arrived within the probe timeout. */ | { kind: 'timeout'; timeoutMs: number }; @@ -141,6 +143,18 @@ export function classifyProbeOutcome(outcome: ProbeOutcome, context: ProbeClassi // handshake an auth-gated modern server as legacy. return { kind: 'error', error: outcome.error }; } + case 'closed': { + if (context.transportKind === 'stdio') { + // A stdio child that exits on the unrecognized probe instead of + // answering is a legacy signal — the same backward-compatibility + // rule as the timeout row below (SDKs like the official Rust one + // terminate the server on ANY pre-initialize request). + return { kind: 'legacy' }; + } + // On HTTP a mid-probe close is an ambiguous network condition — the + // same typed connect error as any probe transport failure. + return classifyNetworkError(new Error('Connection closed during the version negotiation probe'), context); + } case 'timeout': { if (context.transportKind === 'stdio') { // Per the stdio transport's backward-compatibility rule, a probe @@ -253,7 +267,8 @@ function isOpaqueFetchTypeError(error: unknown): boolean { return error instanceof TypeError || (error instanceof Error && error.name === 'TypeError'); } -function describeError(error: unknown): string { +/** Human-readable rendering of an unknown thrown value for probe diagnostics. */ +export function describeError(error: unknown): string { return error instanceof Error ? error.message : String(error); } diff --git a/packages/client/src/client/stdio.ts b/packages/client/src/client/stdio.ts index 29b0027eae..c9e60d8d30 100644 --- a/packages/client/src/client/stdio.ts +++ b/packages/client/src/client/stdio.ts @@ -7,6 +7,8 @@ import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core-inter import { ReadBuffer, SdkError, SdkErrorCode, serializeMessage } from '@modelcontextprotocol/core-internal'; import spawn from 'cross-spawn'; +import { CLOSED_BY_CALLER } from './versionNegotiation'; + export type StdioServerParameters = { /** * The executable to run to start the server. @@ -126,6 +128,16 @@ export class StdioClientTransport implements Transport { ); } + // Close-provenance stamp (see CLOSED_BY_CALLER): a fresh life has not been closed locally. + (this as { [CLOSED_BY_CALLER]?: boolean })[CLOSED_BY_CALLER] = false; + // A dead predecessor's late flush must not prepend to the new child's first frame. + this._readBuffer.clear(); + if (this._stderrStream?.writableEnded) { + // stderr: 'pipe' — the previous life's stream was ended (by its natural death, or by + // close() completing teardown): this life pipes into a fresh one. Re-read the getter. + this._stderrStream = new PassThrough(); + } + return new Promise((resolve, reject) => { this._process = spawn(this._serverParams.command, this._serverParams.args ?? [], { // merge default env with server env because mcp server needs some env vars @@ -183,6 +195,9 @@ export class StdioClientTransport implements Transport { * If `stderr` piping was requested, a `PassThrough` stream is returned _immediately_, allowing callers to * attach listeners before the `start` method is invoked. This prevents loss of any early * error output emitted by the child process. + * + * Under `versionNegotiation: { mode: 'auto' }`, a legacy fallback may respawn the child with a + * fresh stream — re-read this getter after `connect()` resolves. */ get stderr(): Stream | null { if (this._stderrStream) { @@ -217,6 +232,8 @@ export class StdioClientTransport implements Transport { } async close(): Promise { + // Close-provenance stamp (see CLOSED_BY_CALLER): a local close is never respawned. + (this as { [CLOSED_BY_CALLER]?: boolean })[CLOSED_BY_CALLER] = true; if (this._process) { const processToClose = this._process; this._process = undefined; @@ -252,6 +269,19 @@ export class StdioClientTransport implements Transport { // ignore } } + + // Teardown is done: a predecessor whose stderr pipe outlives close() + // (a held-open pipe, a SIGKILL still landing) must not end the stream + // a NEXT life pipes into after start()'s recreation check. Detached + // here — not at the top — so shutdown-time stderr (the child's own + // signal/EOF diagnostics) still reaches subscribers during the waits above. + processToClose.stderr?.unpipe(); + // ...and end the aggregate deterministically: after the unpipe the + // predecessor's own end can no longer arrive, so without this a + // stream whose child outlived close() would NEVER end — hanging + // finished()/for-await readers. A stream the graceful path already + // ended is unaffected; start() begins the next life on a fresh one. + this._stderrStream?.end(); } this._readBuffer.clear(); diff --git a/packages/client/src/client/versionNegotiation.ts b/packages/client/src/client/versionNegotiation.ts index 628d5d7bc1..3b62c73a6f 100644 --- a/packages/client/src/client/versionNegotiation.ts +++ b/packages/client/src/client/versionNegotiation.ts @@ -27,7 +27,7 @@ import { import { UnauthorizedError } from './auth'; import type { ProbeEnvironment, ProbeOutcome, ProbeTransportKind, ProbeVerdict } from './probeClassifier'; -import { classifyProbeOutcome } from './probeClassifier'; +import { classifyProbeOutcome, describeError } from './probeClassifier'; /** * Probe policy for `'auto'` and pinned negotiation modes. @@ -71,7 +71,16 @@ export interface VersionNegotiationProbeOptions { * outcome is definitive modern evidence. Network outage rejects with a typed * connect error; a probe timeout falls back to `initialize` on stdio (a silent * server on a local pipe is a legacy server) and rejects with a typed timeout - * error on HTTP (silence there is an outage). + * error on HTTP (silence there is an outage). A connection closed mid-probe + * follows the same split: on stdio a child that exits on the unrecognized + * probe is a legacy server — the client respawns it once (`close()` + + * `start()`; only for transports that positively report close provenance, + * as the SDK's `StdioClientTransport` does — others reject with the typed + * connect error) and runs `initialize` on the fresh child. The exit counts + * as a close only once it closes the child's stdio pipes; an exit hidden + * behind a helper process holding them open falls to the probe-timeout + * path instead. On HTTP a mid-probe close rejects with the same typed + * connect error as any probe transport failure. * - `{ pin: '' }` — modern era at exactly the pinned revision: the * connect-time `server/discover` must offer it. No fallback — anything else * fails loudly with a typed error. @@ -158,6 +167,22 @@ export function detectProbeTransportKind(transport: Transport): ProbeTransportKi return 'stderr' in transport && 'pid' in transport ? 'stdio' : 'http'; } +/** + * Close-provenance contract: a transport opts into the legacy-fallback respawn + * by stamping `this[CLOSED_BY_CALLER] = false` in `start()` and `= true` at the + * top of `close()` — negotiation respawns only on a `false` stamp; a `true` + * stamp (local close) or an absent stamp (no provenance tracked) is never + * respawned. `Symbol.for`: one shared registry key across bundled SDK copies, + * no import needed to interoperate. Internal — not part of the public surface. + */ +export const CLOSED_BY_CALLER: unique symbol = Symbol.for('mcp.closedByCaller'); + +/** Typed read of the {@linkcode CLOSED_BY_CALLER} stamp — `undefined` when the transport does not implement the contract. */ +function closeProvenance(transport: Transport): boolean | undefined { + const value = (transport as { [CLOSED_BY_CALLER]?: unknown })[CLOSED_BY_CALLER]; + return typeof value === 'boolean' ? value : undefined; +} + /** Raw reply from one probe exchange, before normalization. */ type RawProbeReply = | { kind: 'response'; result?: unknown; error?: { code: number; message: string; data?: unknown } } @@ -264,6 +289,16 @@ class ProbeWindow { }); } + /** + * The legacy-fallback respawn opened a fresh connection: the forwarded + * mid-window close belonged to the previous life, so the successor's + * eventual close is a NEW event — detach()'s spent-close guard must not + * swallow it for the caller's pre-set observer. + */ + noteRespawn(): void { + this._closeDelivered = false; + } + /** Detach the window's handlers, restoring any the caller pre-set, leaving the transport's own `start` untouched. */ detach(): void { this._pending = undefined; @@ -339,7 +374,9 @@ function normalizeReply(reply: RawProbeReply, timeoutMs: number): ProbeOutcome { return { kind: 'network-error', error }; } case 'closed': { - return { kind: 'network-error', error: new Error('Connection closed during the version negotiation probe') }; + // Not folded into network-error: the classifier's closed row is + // transport-aware (stdio legacy signal vs typed error). + return { kind: 'closed' }; } case 'timeout': { return { kind: 'timeout', timeoutMs }; @@ -421,11 +458,24 @@ export async function negotiateEra( continue; } case 'legacy': { + // A closed outcome carries its own cause — the pin and + // modern-only diagnostics must name the close (the caller's + // own close when provenance says so) instead of implying a + // server/discover verdict that never happened. + const closedCause = + outcome.kind === 'closed' + ? closeProvenance(deps.transport) === true + ? 'the transport was closed during the server/discover probe' + : 'the connection closed during the server/discover probe' + : undefined; if (negotiation.kind === 'pin') { throw new SdkError( SdkErrorCode.EraNegotiationFailed, - `Version negotiation failed: the server did not offer pinned protocol version ${negotiation.version} ` + - `via server/discover (no fallback in pin mode)` + closedCause === undefined + ? `Version negotiation failed: the server did not offer pinned protocol version ${negotiation.version} ` + + `via server/discover (no fallback in pin mode)` + : `Version negotiation failed: ${closedCause} before the server offered ` + + `pinned protocol version ${negotiation.version} (no fallback in pin mode)` ); } if (!negotiation.fallbackAvailable) { @@ -433,10 +483,47 @@ export async function negotiateEra( // unavailable and must never carry a 2026-era version string. throw new SdkError( SdkErrorCode.EraNegotiationFailed, - 'Version negotiation failed: the server gave no modern evidence and this client supports no ' + - 'pre-2026-07-28 protocol version to fall back to' + closedCause === undefined + ? 'Version negotiation failed: the server gave no modern evidence and this client supports no ' + + 'pre-2026-07-28 protocol version to fall back to' + : `Version negotiation failed: ${closedCause} and this client supports no ` + + 'pre-2026-07-28 protocol version to fall back to' ); } + if (outcome.kind === 'closed') { + // The probe consumed the connection (the child exited on + // the unrecognized request): respawn via the public + // lifecycle, then run the fallback handshake there — + // but only on positive provenance that the close was + // NOT local. The read happens BEFORE our own close() + // below (the only close this path issues; start() + // re-stamps false), so `true` always means the caller + // closed mid-probe: never override that kill decision. + // An absent stamp (custom/wrapper transports) fails + // safe: never respawn a transport that cannot + // distinguish its own close() from a child death. + const provenance = closeProvenance(deps.transport); + if (provenance !== false) { + throw new SdkError( + SdkErrorCode.EraNegotiationFailed, + provenance === true + ? 'Version negotiation failed: the transport was closed during the server/discover probe' + : 'Version negotiation failed: the connection closed during the server/discover probe and ' + + 'this transport does not report close provenance, so the legacy-fallback respawn is unavailable' + ); + } + try { + await deps.transport.close(); + await deps.transport.start(); + } catch (error) { + throw new SdkError( + SdkErrorCode.EraNegotiationFailed, + `Version negotiation failed: the transport could not restart for the legacy fallback: ${describeError(error)}`, + { cause: error } + ); + } + window.noteRespawn(); + } return { era: 'legacy' }; } case 'error': { diff --git a/packages/client/test/client/probeClassifier.test.ts b/packages/client/test/client/probeClassifier.test.ts index 74d0c3248f..ac93d60340 100644 --- a/packages/client/test/client/probeClassifier.test.ts +++ b/packages/client/test/client/probeClassifier.test.ts @@ -293,6 +293,22 @@ describe('row: timeout — transport-aware verdict', () => { }); }); +describe('row: closed — transport-aware verdict, symmetric with the timeout row', () => { + test('stdio: a child that exits on the unrecognized probe is a legacy signal → legacy fallback', () => { + expect(classify({ kind: 'closed' }, { transportKind: 'stdio' })).toEqual({ kind: 'legacy' }); + }); + + test('HTTP: a mid-probe close is ambiguous — the same typed connect error as any probe network failure', () => { + const verdict = classify({ kind: 'closed' }, { transportKind: 'http' }); + expect(verdict.kind).toBe('error'); + if (verdict.kind === 'error') { + expect(verdict.error).toBeInstanceOf(SdkError); + expect((verdict.error as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect(verdict.error.message).toContain('Connection closed during the version negotiation probe'); + } + }); +}); + describe('row: browser opaque CORS/preflight TypeError, PROBE PHASE ONLY → legacy fallback (F-7)', () => { test('browser environment + bare TypeError → legacy', () => { expect(classify({ kind: 'network-error', error: new TypeError('Failed to fetch') }, { environment: 'browser' })).toEqual({ diff --git a/packages/client/test/client/stdio.test.ts b/packages/client/test/client/stdio.test.ts index 594ad6dc63..cee73fc19b 100644 --- a/packages/client/test/client/stdio.test.ts +++ b/packages/client/test/client/stdio.test.ts @@ -99,6 +99,143 @@ test('should respect custom maxBufferSize option', async () => { await closed; }); +test("a dead child's late stdout flush landing between close() and start() cannot poison the next life's first frame", async () => { + // The grandchild inherits (and so holds open) the child's stdout pipe and + // writes a stale partial frame into it AFTER close() returned: close() + // races its 2s teardown wait, sees the child already exited, and returns + // with the pipe still held, so the flush lands in the shared read buffer + // between close() and the next start(). start() clears the buffer, so the + // next life's first JSON-RPC line cannot parse as . + const FLUSHING_SERVER_SCRIPT = String.raw` + const { spawn } = require('child_process'); + spawn(process.execPath, ['-e', "setTimeout(() => process.stdout.write('stale:'), 2400); setTimeout(() => {}, 4000);"], { + stdio: 'inherit' + }); + let buffer = ''; + process.stdin.on('data', chunk => { + buffer += chunk.toString(); + let index; + while ((index = buffer.indexOf('\n')) !== -1) { + const line = buffer.slice(0, index); + buffer = buffer.slice(index + 1); + if (line.trim() === '') continue; + let message; + try { + message = JSON.parse(line); + } catch { + continue; + } + if (message.method === 'whoami' && message.id !== undefined) { + process.stdout.write(JSON.stringify({ jsonrpc: '2.0', id: message.id, result: { pid: process.pid } }) + '\n'); + } + } + }); + process.stdin.on('end', () => process.exit(0)); + `; + const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', FLUSHING_SERVER_SCRIPT] }); + + await transport.start(); + await transport.close(); + + // Let the predecessor's stale flush land in the close()->start() window. + await new Promise(resolve => setTimeout(resolve, 700)); + + await transport.start(); + const reply = await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('whoami timed out — stale residue poisoned the frame')), 5000); + transport.onmessage = message => { + clearTimeout(timer); + resolve(message); + }; + transport.send({ jsonrpc: '2.0', id: 1, method: 'whoami' }).catch(reject); + }); + expect(reply).toMatchObject({ id: 1, result: { pid: transport.pid } }); + + await transport.close(); +}, 15_000); + +test('shutdown-time stderr written while close() tears the child down still reaches subscribers', async () => { + // The child logs a diagnostic on stdin EOF — mid-close() output. close() + // must not detach the stderr pipe before the teardown delivers it. + const transport = new StdioClientTransport({ + command: process.execPath, + args: [ + '-e', + "process.stdin.on('end', () => { process.stderr.write('shutdown-diag\\n'); process.exit(0); }); process.stdin.resume();" + ], + stderr: 'pipe' + }); + const chunks: string[] = []; + transport.stderr?.on('data', chunk => chunks.push(String(chunk))); + + await transport.start(); + await transport.close(); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(chunks.join('')).toContain('shutdown-diag'); +}, 10_000); + +test("close() ends the aggregate stderr stream even when the predecessor's pipe outlives it", async () => { + // A grandchild inherits (and so holds open) the child's stderr pipe past + // close(), so the predecessor's own 'end' can only land after close() + // unpiped it — i.e. never. Without close() ending the aggregate itself, + // stream.finished()/for-await readers of a closed transport would hang. + const HOLDING_SCRIPT = String.raw` + const { spawn } = require('child_process'); + spawn(process.execPath, ['-e', 'setTimeout(() => {}, 3000)'], { stdio: 'inherit' }); + process.stdin.on('end', () => process.exit(0)); + process.stdin.resume(); + `; + const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', HOLDING_SCRIPT], stderr: 'pipe' }); + let ended = false; + transport.stderr?.on('end', () => { + ended = true; + }); + transport.stderr?.on('data', () => {}); // flowing mode so 'end' can fire + + await transport.start(); + await transport.close(); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(ended).toBe(true); +}, 15_000); + +test("a dead predecessor whose stderr pipe outlives close() cannot touch the next life's stderr stream", async () => { + // The grandchild holds the child's stderr pipe open past close(); close() + // unpipes the predecessor and ends the aggregate, so the next start() + // begins a fresh stream the predecessor cannot touch — the second life's + // output past the predecessor's teardown must still arrive. + const STDERR_HOLDING_SCRIPT = String.raw` + const { spawn } = require('child_process'); + spawn(process.execPath, ['-e', 'setTimeout(() => {}, 3000)'], { stdio: 'inherit' }); + process.stderr.write('alive ' + process.pid + '\n'); + setTimeout(() => process.stderr.write('late ' + process.pid + '\n'), 1500); + process.stdin.on('end', () => process.exit(0)); + process.stdin.resume(); + `; + const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', STDERR_HOLDING_SCRIPT], stderr: 'pipe' }); + const chunks: string[] = []; + const collect = (chunk: unknown) => chunks.push(String(chunk)); + const firstStream = transport.stderr; + firstStream?.on('data', collect); + + await transport.start(); + // close() returns via its 2s race: the child exited on stdin end, but the + // grandchild's grip keeps the stderr pipe (and the child 'close' event) pending. + await transport.close(); + await transport.start(); + const secondPid = transport.pid; + // Life 2 may be on a fresh stream or the kept one — read the getter again. + if (transport.stderr !== firstStream) { + transport.stderr?.on('data', collect); + } + + // Life 1's grandchild exits at ~3s, delivering the predecessor's stderr + // 'end'; life 2's late write lands after it (~3.6s) and must survive. + await new Promise(resolve => setTimeout(resolve, 2500)); + expect(chunks.join('')).toContain(`late ${secondPid}`); + + await transport.close(); +}, 15_000); + test('should fire onerror and close when ReadBuffer overflows', async () => { const client = new StdioClientTransport({ command: 'node', diff --git a/packages/client/test/client/versionNegotiation.test.ts b/packages/client/test/client/versionNegotiation.test.ts index 698bccd6ee..6ab1f45aaf 100644 --- a/packages/client/test/client/versionNegotiation.test.ts +++ b/packages/client/test/client/versionNegotiation.test.ts @@ -22,7 +22,7 @@ import { UnauthorizedError } from '../../src/client/auth'; import { Client } from '../../src/client/client'; import type { StreamableHTTPClientTransportOptions } from '../../src/client/streamableHttp'; import type { StdioServerParameters } from '../../src/client/stdio'; -import { resolveVersionNegotiation } from '../../src/client/versionNegotiation'; +import { CLOSED_BY_CALLER, resolveVersionNegotiation } from '../../src/client/versionNegotiation'; const MODERN = '2026-07-28'; @@ -438,6 +438,302 @@ describe('probe timeout policy (transport-aware)', () => { }); }); +/* ------------------------------------------------------------------------- * + * Probe close policy: transport-aware, symmetric with the timeout policy. On + * stdio a child that exits on the unrecognized probe is a legacy server (the + * shape of stdio servers built on SDKs that terminate on any pre-initialize + * request — the official Rust SDK, rmcp) — the client respawns it once via the + * public lifecycle (close() + start()) and runs the fallback initialize there. + * A caller's own close() mid-probe is never overridden: the transport flags it + * and negotiation rejects typed instead of respawning. On HTTP-class + * transports a mid-probe close stays a typed connect error. + * ------------------------------------------------------------------------- */ + +describe('probe close policy (transport-aware)', () => { + /** + * A stdio-shaped transport whose scripted "child" exits (close, no reply) + * on any pre-initialize request other than initialize. Mirrors the real + * StdioClientTransport lifecycle contract: start() after close() respawns, + * and the CLOSED_BY_CALLER provenance stamp is set in close() / cleared in + * start(). + */ + class RmcpShapedStdioTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + sessionId?: string; + + startCalls = 0; + sent: JSONRPCMessage[] = []; + setProtocolVersionCalls: string[] = []; + [CLOSED_BY_CALLER]?: boolean; + protected _alive = false; + private _initialized = false; + + get stderr(): null { + return null; + } + get pid(): number | null { + return this._alive ? 4242 + this.startCalls : null; + } + + async start(): Promise { + if (this._alive) { + throw new Error('RmcpShapedStdioTransport already started!'); + } + this[CLOSED_BY_CALLER] = false; + this._alive = true; + this._initialized = false; + this.startCalls++; + } + + async send(message: JSONRPCMessage): Promise { + if (!this._alive) { + throw new Error('Not connected'); + } + this.sent.push(message); + queueMicrotask(() => { + if (!this._alive || !isJSONRPCRequest(message)) return; + if (message.method === 'initialize') { + this._initialized = true; + this.onmessage?.({ + jsonrpc: '2.0', + id: message.id, + result: { + protocolVersion: '2025-03-26', + capabilities: {}, + serverInfo: { name: 'rmcp-shaped-server', version: '1.0.0' } + } + }); + } else if (!this._initialized) { + // The rmcp shape: any other pre-initialize request kills the + // child — no reply, just the close. + this._alive = false; + this.onclose?.(); + } + }); + } + + async close(): Promise { + this[CLOSED_BY_CALLER] = true; + if (this._alive) { + this._alive = false; + this.onclose?.(); + } + } + + setProtocolVersion(version: string): void { + this.setProtocolVersionCalls.push(version); + } + } + + test('stdio: exit-on-probe respawns once via the public lifecycle and connects on the legacy era', async () => { + const transport = new RmcpShapedStdioTransport(); + let presetCloses = 0; + transport.onclose = () => { + presetCloses++; + }; + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + await client.connect(transport); + + // The probe consumed the first "child"; negotiation respawned it (one extra start), exactly once. + expect(transport.startCalls).toBe(2); + const sent = requests(transport.sent); + expect(sent.filter(r => r.method === 'server/discover')).toHaveLength(1); + expect(sent.some(r => r.method === 'initialize')).toBe(true); + expect(client.getNegotiatedProtocolVersion()).toBe('2025-03-26'); + expect(client.getServerVersion()?.name).toBe('rmcp-shaped-server'); + // Stamped exactly once, by the fallback handshake on the respawned child. + expect(transport.setProtocolVersionCalls).toEqual(['2025-03-26']); + + // The probe child's close was delivered to the pre-set observer (a real + // process exit), and the observer keeps seeing the successor's close. + expect(presetCloses).toBe(1); + await client.close(); + expect(presetCloses).toBe(2); + }); + + test('stdio: post-respawn traffic is byte-identical to a plain mode:legacy connect against the same server shape', async () => { + const autoTransport = new RmcpShapedStdioTransport(); + const autoClient = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + await autoClient.connect(autoTransport); + + const plainTransport = new RmcpShapedStdioTransport(); + const plainClient = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'legacy' } }); + await plainClient.connect(plainTransport); + + // Drop the probe: the respawned child sees exactly the bytes a plain + // legacy connect sends (same initialize id and body). + expect(JSON.stringify(autoTransport.sent.slice(1))).toBe(JSON.stringify(plainTransport.sent)); + // Regression pins: legacy mode never probes, never respawns. + expect(plainTransport.startCalls).toBe(1); + expect(requests(plainTransport.sent)[0]!.method).toBe('initialize'); + + await autoClient.close(); + await plainClient.close(); + }); + + test('stdio: a caller close while the probe reply is pending is never overridden — typed error, no respawn', async () => { + // A silent stdio child: never answers the probe, stays alive — the + // caller cancels while the reply is pending. The caller's close() sets + // the flag BEFORE the close verdict reaches negotiation; negotiation's + // own respawn close() runs only after that read (and its start() resets + // the flag), so the flag being true always means the caller closed. + class SilentStdioTransport extends RmcpShapedStdioTransport { + override async send(message: JSONRPCMessage): Promise { + if (!this._alive) throw new Error('Not connected'); + this.sent.push(message); // swallow: no reply, no exit + } + } + const transport = new SilentStdioTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + const pending = client.connect(transport); + pending.catch(() => undefined); + // Let the probe send settle, then cancel like a caller would. + await new Promise(resolve => setTimeout(resolve, 0)); + await transport.close(); + + const rejection = await pending.then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect((rejection as SdkError).message).toMatch(/closed during the server\/discover probe/); + expect(transport.startCalls).toBe(1); + expect(requests(transport.sent).some(r => r.method === 'initialize')).toBe(false); + }); + + test('stdio: a caller close racing the probe write stays a typed error, no respawn', async () => { + // The other caller-close flavor: the close lands before the probe write + // completes, so the send itself rejects (the real transport throws + // NotConnected once its process handle is gone). + class CloseRacingSendTransport extends RmcpShapedStdioTransport { + override async send(message: JSONRPCMessage): Promise { + this.sent.push(message); + await this.close(); + throw new Error('write EPIPE'); + } + } + const transport = new CloseRacingSendTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + const rejection = await client.connect(transport).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect(transport.startCalls).toBe(1); + expect(requests(transport.sent).some(r => r.method === 'initialize')).toBe(false); + }); + + test('stdio: a transport without the close-provenance stamp is never respawned — typed error on exit-on-probe', async () => { + // A custom stdio-shaped transport implementing only the documented + // Transport surface: no CLOSED_BY_CALLER stamp. Negotiation must fail + // safe — it cannot distinguish this close from a caller's kill. + class NoProvenanceTransport extends RmcpShapedStdioTransport { + override async start(): Promise { + await super.start(); + delete this[CLOSED_BY_CALLER]; + } + override async close(): Promise { + await super.close(); + delete this[CLOSED_BY_CALLER]; + } + } + const transport = new NoProvenanceTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + const rejection = await client.connect(transport).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect((rejection as SdkError).message).toMatch(/does not report close provenance/); + // No respawn attempted: one start, no initialize. + expect(transport.startCalls).toBe(1); + expect(requests(transport.sent).some(r => r.method === 'initialize')).toBe(false); + }); + + test('stdio: pin mode fails loudly on exit-on-probe — no respawn, no initialize, message names the close', async () => { + const transport = new RmcpShapedStdioTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: { pin: MODERN } } }); + + const rejection = await client.connect(transport).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + // The server never answered discover — the diagnostic must name the + // close, not imply a server verdict. + expect((rejection as SdkError).message).toMatch(/connection closed during the server\/discover probe/); + expect(transport.startCalls).toBe(1); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('stdio: a modern-only client gets the close-flavored typed error — no respawn, no initialize', async () => { + const transport = new RmcpShapedStdioTransport(); + const client = new Client( + { name: 'c', version: '0' }, + { versionNegotiation: { mode: 'auto' }, supportedProtocolVersions: [MODERN] } + ); + + const rejection = await client.connect(transport).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect((rejection as SdkError).message).toMatch(/connection closed during the server\/discover probe/); + expect(transport.startCalls).toBe(1); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('stdio: a respawn whose start() fails rejects typed with the cause — no raw transport error escapes', async () => { + class OneShotStartTransport extends RmcpShapedStdioTransport { + override async start(): Promise { + if (this.startCalls > 0) throw new Error('spawn ENOENT'); + await super.start(); + } + } + const transport = new OneShotStartTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + const rejection = await client.connect(transport).then( + () => undefined, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect(((rejection as SdkError).data as { cause?: Error }).cause?.message).toBe('spawn ENOENT'); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('HTTP-class transport: close mid-probe rejects like any probe transport failure — never a legacy verdict', async () => { + const script: Script = (message, t) => { + if (!isJSONRPCRequest(message)) return; + if (message.method === 'server/discover') { + t.onclose?.(); + return; + } + legacyServerScript(message, t); + }; + const transport = new ScriptedTransport(script); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + await expect(client.connect(transport)).rejects.toSatisfy( + error => error instanceof SdkError && error.code === SdkErrorCode.EraNegotiationFailed + ); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + expect(transport.startCalls).toBe(1); + }); +}); + /* ------------------------------------------------------------------------- * * -32022 corrective continuation — exactly once; loop guard on second * rejection. diff --git a/test/integration/test/client/versionNegotiation.test.ts b/test/integration/test/client/versionNegotiation.test.ts index 2bd93eba93..d140f1bfd7 100644 --- a/test/integration/test/client/versionNegotiation.test.ts +++ b/test/integration/test/client/versionNegotiation.test.ts @@ -24,7 +24,7 @@ import { SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; import { NodeStreamableHTTPServerTransport } from '@modelcontextprotocol/node'; import { McpServer } from '@modelcontextprotocol/server'; import { listenOnRandomPort } from '@modelcontextprotocol/test-helpers'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import * as z from 'zod/v4'; /** A fetch wrapper recording every request our client puts on the wire (URL, headers, body) and the raw response (status, body). */ @@ -286,3 +286,113 @@ describe('stdio: silent legacy server (probe timeout fallback)', () => { } }, 15_000); }); + +describe('stdio: rmcp-shaped legacy server (exit-on-probe fallback)', () => { + // A real child in the shape of stdio servers built on SDKs that terminate + // on ANY pre-initialize request (the official Rust SDK, rmcp): it exits 1 + // when the server/discover probe arrives, but is a perfectly good legacy + // server after a plain initialize. Banner on stderr identifies each life. + const RMCP_SHAPED_SERVER_SCRIPT = String.raw` + process.stderr.write('child-alive ' + process.pid + '\n'); + let initialized = false; + let buffer = ''; + const send = obj => process.stdout.write(JSON.stringify(obj) + '\n'); + process.stdin.on('data', chunk => { + buffer += chunk.toString(); + let index; + while ((index = buffer.indexOf('\n')) !== -1) { + const line = buffer.slice(0, index); + buffer = buffer.slice(index + 1); + if (line.trim() === '') continue; + let message; + try { + message = JSON.parse(line); + } catch { + continue; + } + if (message.method === 'initialize') { + initialized = true; + send({ + jsonrpc: '2.0', + id: message.id, + result: { + protocolVersion: message.params.protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: 'rmcp-shaped-server', version: '1.0.0' } + } + }); + } else if (!initialized && message.id !== undefined) { + process.exit(1); + } else if (message.method === 'tools/call') { + send({ + jsonrpc: '2.0', + id: message.id, + result: { content: [{ type: 'text', text: message.params.arguments.text }] } + }); + } + } + }); + `; + + const spawnRmcp = () => + new StdioClientTransport({ command: process.execPath, args: ['-e', RMCP_SHAPED_SERVER_SCRIPT], stderr: 'pipe' }); + + it('auto mode: the probe kills the child, the client respawns it once and completes the legacy handshake — tools/call works', async () => { + const transport = spawnRmcp(); + // Pre-connect stderr subscription sees the FIRST life only. + const firstLife: string[] = []; + transport.stderr?.on('data', chunk => firstLife.push(String(chunk))); + + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' } }); + try { + await client.connect(transport); + + expect(client.getServerVersion()?.name).toBe('rmcp-shaped-server'); + const result = await client.callTool({ name: 'echo', arguments: { text: 'respawned' } }); + expect(result.content).toEqual([{ type: 'text', text: 'respawned' }]); + + // Two spawns: the probe consumed life 1; the documented sharp edge — + // re-read the stderr getter after connect() — yields life 2's fresh stream. + const secondLife: string[] = []; + transport.stderr?.on('data', chunk => secondLife.push(String(chunk))); + await vi.waitFor(() => { + expect(firstLife.join('')).toMatch(/child-alive \d+/); + expect(secondLife.join('')).toMatch(/child-alive \d+/); + }); + const firstPid = Number(/child-alive (\d+)/.exec(firstLife.join(''))![1]); + const secondPid = Number(/child-alive (\d+)/.exec(secondLife.join(''))![1]); + expect(secondPid).not.toBe(firstPid); + expect(transport.pid).toBe(secondPid); + } finally { + await client.close(); + } + }, 15_000); + + it("a caller close() mid-probe is never overridden: typed error, the child stays dead — the caller's kill decision wins", async () => { + // A child that ignores the probe entirely, so the reply is still + // pending when the caller cancels. + const SILENT_SCRIPT = String.raw` + process.stdin.resume(); + `; + const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', SILENT_SCRIPT] }); + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' } }); + + const pending = client.connect(transport); + pending.catch(() => {}); + await vi.waitFor(() => expect(transport.pid).not.toBeNull()); + await new Promise(resolve => setTimeout(resolve, 200)); + + await transport.close(); + + const rejection = await pending.then( + () => {}, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + expect((rejection as SdkError).message).toMatch(/closed during the server\/discover probe/); + // No respawn, now or later. + await new Promise(resolve => setTimeout(resolve, 200)); + expect(transport.pid).toBeNull(); + }, 15_000); +});