diff --git a/.changeset/probe-close-handling.md b/.changeset/probe-close-handling.md new file mode 100644 index 0000000000..0fdcbb17b4 --- /dev/null +++ b/.changeset/probe-close-handling.md @@ -0,0 +1,10 @@ +--- +'@modelcontextprotocol/core-internal': patch +'@modelcontextprotocol/client': patch +--- + +Handle a connection that closes during the `server/discover` version-negotiation probe. + +On stdio, a server process that exits on the unrecognized probe is a legacy signal — symmetric with the stdio probe-timeout rule. Servers built on SDKs that terminate on any pre-`initialize` request (the official Rust SDK, rmcp, is the prominent example) previously connected fine under `mode: 'legacy'` but hard-failed under `mode: 'auto'`; `StdioClientTransport` now restarts itself (respawning the server from its retained parameters) and the client completes the plain `initialize` handshake there — the respawned process sees bytes identical to a plain legacy connect. A transport that was closed locally, or that cannot restart, rejects with a typed negotiation error instead: a deliberately terminated server is never resurrected. On HTTP-class transports a mid-probe close keeps surfacing as the same typed connect error as any other probe transport failure — an ambiguous drop is never era evidence. + +`StdioClientTransport` restart mechanics: child events are generation-scoped (a slow-dying child's late events cannot touch its successor), the read buffer is cleared on child exit, and with `stderr: 'pipe'` the stream spans restarts — it ends exactly once, on `close()`, not on a child's exit. diff --git a/docs/migration/support-2026-07-28.md b/docs/migration/support-2026-07-28.md index fdc221d930..822312b61f 100644 --- a/docs/migration/support-2026-07-28.md +++ b/docs/migration/support-2026-07-28.md @@ -78,8 +78,16 @@ error. Probe timeouts are **transport-aware**: on **stdio** a server that does n answer within `timeoutMs` is treated as legacy and the client falls back to `initialize` on the same stream (some legacy servers never respond to unknown pre-`initialize` requests at all); on **HTTP** a probe timeout rejects with `SdkError(RequestTimeout)` — -a dead HTTP server is never misreported as legacy. One browser-specific exception: an -opaque CORS/preflight `TypeError` during the probe falls back to the legacy era, because +a dead HTTP server is never misreported as legacy. A mid-probe **connection close** is +transport-aware the same way: on **stdio** a server process that exits on the +unrecognized probe is treated as legacy — the transport restarts itself (respawning +the server) and the client falls back to `initialize` there (stdio servers built on +SDKs that terminate on any pre-`initialize` request behave this way); on **HTTP** a +mid-probe close rejects with the same typed connect error as any probe transport +failure (`SdkError(EraNegotiationFailed)`) — an ambiguous drop is never misreported +as legacy. A transport closed locally during the probe is never restarted: `connect()` +rejects with the same typed error. One browser-specific exception: an 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. ```typescript diff --git a/docs/protocol-versions.md b/docs/protocol-versions.md index 6b06754ab5..dcdfe488d2 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 server process that exits on the unrecognized probe is a legacy server — stdio servers built on SDKs that terminate on any pre-`initialize` request behave exactly this way (the official Rust SDK, rmcp, is the prominent example) — so the transport restarts itself, respawning the server, and `connect()` falls back to `initialize` there; the respawned process sees the same bytes a `mode: 'legacy'` connect would have sent. On HTTP a mid-probe close is ambiguous — a proxy drop or a crash, never era evidence — so `connect()` rejects with the same typed `SdkError(EraNegotiationFailed)` as any probe transport failure. A transport closed locally during the probe is never restarted. + 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/packages/client/src/client/probeClassifier.ts b/packages/client/src/client/probeClassifier.ts index 1b78744964..42a51194c6 100644 --- a/packages/client/src/client/probeClassifier.ts +++ b/packages/client/src/client/probeClassifier.ts @@ -2,7 +2,8 @@ * Probe outcome classifier (pure module): maps the outcome of the connect-time * `server/discover` probe onto one of four verdicts — modern era, the * spec-mandated `-32022` corrective continuation, legacy fallback (the plain - * 2025 `initialize` handshake on the same connection), or a typed connect error. + * 2025 `initialize` handshake — on the same connection, or on a restarted + * transport when a stdio close consumed it), or a typed connect error. * * The classifier is deliberately conservative: anything it does not positively * recognize as modern resolves to the legacy fallback, and a network outage is a @@ -48,6 +49,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 }; @@ -82,8 +85,13 @@ export type ProbeVerdict = * arms a loop guard on the second rejection, throwing `error`. */ | { kind: 'corrective'; version: string; error: UnsupportedProtocolVersionError } - /** Definitive legacy signal or unrecognized shape: perform the plain legacy `initialize` handshake on the same connection. */ - | { kind: 'legacy' } + /** + * Definitive legacy signal or unrecognized shape: perform the plain legacy + * `initialize` handshake on the same connection. `restart` marks the one + * row where the same connection no longer exists (stdio close): the + * handshake needs the transport restarted first. + */ + | { kind: 'legacy'; restart?: true } /** Typed connect error — never converted to an era verdict. */ | { kind: 'error'; error: Error }; @@ -141,6 +149,21 @@ 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). The + // stream died with the process, so the fallback needs the + // transport restarted. + return { kind: 'legacy', restart: true }; + } + // On HTTP a mid-probe close is an ambiguous network condition + // (proxy drop, crash, redeploy) — the same typed connect error as + // any probe transport failure, never an era verdict. + 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 +276,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..86a465f70d 100644 --- a/packages/client/src/client/stdio.ts +++ b/packages/client/src/client/stdio.ts @@ -3,8 +3,8 @@ import process from 'node:process'; import type { Stream } from 'node:stream'; import { PassThrough } from 'node:stream'; -import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core-internal'; -import { ReadBuffer, SdkError, SdkErrorCode, serializeMessage } from '@modelcontextprotocol/core-internal'; +import type { JSONRPCMessage, RestartableTransport } from '@modelcontextprotocol/core-internal'; +import { ReadBuffer, SdkError, SdkErrorCode, serializeMessage, TRANSPORT_RESTART } from '@modelcontextprotocol/core-internal'; import spawn from 'cross-spawn'; export type StdioServerParameters = { @@ -98,11 +98,15 @@ export function getDefaultEnvironment(): Record { * * This transport is only available in Node.js environments. */ -export class StdioClientTransport implements Transport { +export class StdioClientTransport implements RestartableTransport { private _process?: ChildProcess; private _readBuffer: ReadBuffer; private _serverParams: StdioServerParameters; private _stderrStream: PassThrough | null = null; + /** Monotonic child generation; every child's event handlers check it so a dead child's late events cannot touch a successor. */ + private _generation = 0; + /** `true` after a local {@linkcode close} (caller shutdown or the transport's own error recovery); cleared by {@linkcode start}. */ + private _closedLocally = false; onclose?: () => void; onerror?: (error: Error) => void; @@ -118,6 +122,9 @@ export class StdioClientTransport implements Transport { /** * Starts the server process and prepares to communicate with it. + * + * After the child process exits, `start()` may be called again: a fresh + * process is spawned from the same {@linkcode StdioServerParameters}. */ async start(): Promise { if (this._process) { @@ -125,9 +132,16 @@ export class StdioClientTransport implements Transport { 'StdioClientTransport already started! If using Client class, note that connect() calls start() automatically.' ); } + this._closedLocally = false; + if (this._stderrStream?.writableEnded) { + // close() ended the previous life's aggregate stream; a caller + // start() after close() begins a new life with a fresh stream. + this._stderrStream = new PassThrough(); + } + const generation = ++this._generation; return new Promise((resolve, reject) => { - this._process = spawn(this._serverParams.command, this._serverParams.args ?? [], { + const proc = spawn(this._serverParams.command, this._serverParams.args ?? [], { // merge default env with server env because mcp server needs some env vars env: { ...getDefaultEnvironment(), @@ -138,26 +152,39 @@ export class StdioClientTransport implements Transport { windowsHide: process.platform === 'win32', cwd: this._serverParams.cwd }); + this._process = proc; - this._process.on('error', error => { + proc.on('error', error => { reject(error); - this.onerror?.(error); + if (this._generation === generation) { + this.onerror?.(error); + } }); - this._process.on('spawn', () => { + proc.on('spawn', () => { resolve(); }); - this._process.on('close', _code => { + proc.on('close', _code => { + if (this._generation !== generation) { + return; + } this._process = undefined; + // Drop any partial trailing line so a restarted transport starts with a clean stream. + this._readBuffer.clear(); this.onclose?.(); }); - this._process.stdin?.on('error', error => { - this.onerror?.(error); + proc.stdin?.on('error', error => { + if (this._generation === generation) { + this.onerror?.(error); + } }); - this._process.stdout?.on('data', chunk => { + proc.stdout?.on('data', chunk => { + if (this._generation !== generation) { + return; + } try { this._readBuffer.append(chunk); this.processReadBuffer(); @@ -167,22 +194,39 @@ export class StdioClientTransport implements Transport { } }); - this._process.stdout?.on('error', error => { - this.onerror?.(error); + proc.stdout?.on('error', error => { + if (this._generation === generation) { + this.onerror?.(error); + } }); - if (this._stderrStream && this._process.stderr) { - this._process.stderr.pipe(this._stderrStream); + if (this._stderrStream && proc.stderr) { + // end: false — the aggregate stream spans children; it ends in close(), never on a child's exit. + proc.stderr.pipe(this._stderrStream, { end: false }); } }); } + /** + * Version-negotiation close-fallback (internal, keyed by + * `TRANSPORT_RESTART`): respawn the server after the child died out from + * under the connection. Refuses after a local {@linkcode close} — a + * deliberately terminated server is never resurrected. + */ + async [TRANSPORT_RESTART](): Promise { + if (this._closedLocally) { + throw new Error('StdioClientTransport was closed locally; refusing to respawn the server process'); + } + await this.start(); + } + /** * The `stderr` stream of the child process, if {@linkcode StdioServerParameters.stderr} was set to `"pipe"` or `"overlapped"`. * * 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. + * error output emitted by the child process. The stream spans restarts — a child's exit + * never ends it; it ends exactly once, when {@linkcode close} is called. */ get stderr(): Stream | null { if (this._stderrStream) { @@ -217,8 +261,9 @@ export class StdioClientTransport implements Transport { } async close(): Promise { - if (this._process) { - const processToClose = this._process; + this._closedLocally = true; + const processToClose = this._process; + if (processToClose) { this._process = undefined; const closePromise = new Promise(resolve => { @@ -255,6 +300,14 @@ export class StdioClientTransport implements Transport { } this._readBuffer.clear(); + + if (this._stderrStream !== null && !this._stderrStream.writableEnded) { + // The aggregate stderr stream ends exactly once, here. Unpipe first: + // a SIGKILLed child may still flush after we gave up waiting, and a + // write to an ended stream would error. + processToClose?.stderr?.unpipe(this._stderrStream); + this._stderrStream.end(); + } } send(message: JSONRPCMessage): Promise { diff --git a/packages/client/src/client/versionNegotiation.ts b/packages/client/src/client/versionNegotiation.ts index 628d5d7bc1..e6e9026867 100644 --- a/packages/client/src/client/versionNegotiation.ts +++ b/packages/client/src/client/versionNegotiation.ts @@ -17,17 +17,19 @@ import { isJSONRPCErrorResponse, isJSONRPCResultResponse, isModernProtocolVersion, + isRestartableTransport, legacyProtocolVersions, modernProtocolVersions, SdkError, SdkErrorCode, SdkHttpError, - SUPPORTED_MODERN_PROTOCOL_VERSIONS + SUPPORTED_MODERN_PROTOCOL_VERSIONS, + TRANSPORT_RESTART } from '@modelcontextprotocol/core-internal'; 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 +73,11 @@ 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 transport split: on stdio a child that exits on the + * unrecognized probe is a legacy server — the transport restarts itself + * (respawning the process) and the fallback `initialize` runs there; on HTTP + * it rejects with the same typed connect error as a probe network 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. @@ -246,6 +252,12 @@ class ProbeWindow { * never collide with Protocol's numeric ids (e.g. on a shared stdio pipe). */ async exchange(buildRequest: (id: string) => JSONRPCRequest, timeoutMs: number): Promise { + if (this._closeDelivered) { + // The transport closed while no exchange was pending (e.g. in the + // corrective-continuation gap): every later exchange is a closed + // outcome, not a send failure racing the close event. + return { kind: 'closed' }; + } const id = `server-discover-probe-${++this._probeCounter}`; return new Promise(resolve => { let settled = false; @@ -264,17 +276,33 @@ class ProbeWindow { }); } - /** Detach the window's handlers, restoring any the caller pre-set, leaving the transport's own `start` untouched. */ - detach(): void { + /** Restore the caller's pre-set handlers, with `onclose` as given. */ + private _restoreHandlers(onclose: Transport['onclose']): void { this._pending = undefined; this._transport.onmessage = this._savedOnMessage; this._transport.onerror = this._savedOnError; - this._transport.onclose = this._closeDelivered ? undefined : this._savedOnClose; + this._transport.onclose = onclose; + } + + /** + * Detach the window's handlers, restoring any the caller pre-set, leaving + * the transport's own `start` untouched. Failure path: a mid-window close + * that was already forwarded stays spent — the caller's cleanup `close()` + * must not re-deliver it. + */ + detach(): void { + this._restoreHandlers(this._closeDelivered ? undefined : this._savedOnClose); } - /** Detach the handlers and arm the one-shot `start()` pass-through for the `Protocol.connect()` handover. */ + /** + * Detach the handlers and arm the one-shot `start()` pass-through for the + * `Protocol.connect()` handover. Success path: the connection is live — + * after the stdio close-fallback restart, a forwarded mid-window close is + * spent and the successor's eventual close is a new event, so the saved + * `onclose` is restored unconditionally. + */ release(): void { - this.detach(); + this._restoreHandlers(this._savedOnClose); const transport = this._transport; const originalStart = transport.start.bind(transport); let armed = true; @@ -339,7 +367,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 }; @@ -363,9 +393,11 @@ export type NegotiationResult = { era: 'modern'; version: string; discover: Disc /** * Run the negotiation probe state machine on a raw (not yet Protocol-connected) * transport. Resolves with the negotiated era; throws typed connect errors. On - * return the probe window has been released: the transport is started, - * handler-free, and ready for `Protocol.connect()` handover. On throw the - * window is detached and the transport's `start` is left untouched. + * return the probe window has been released: the transport is started — + * freshly restarted by its own internal restart capability when a stdio close + * consumed the probe connection — handler-free, and ready for + * `Protocol.connect()` handover. On throw the window is detached and the + * transport's `start` is left untouched. */ export async function negotiateEra( negotiation: Extract, @@ -421,11 +453,17 @@ export async function negotiateEra( continue; } case 'legacy': { + // The restart flavor carries close evidence (the server + // exited on the probe) — name it in the no-fallback + // diagnostics instead of implying the server answered. 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)` + verdict.restart + ? `Version negotiation failed: the connection closed during the server/discover probe ` + + `before the server offered pinned protocol version ${negotiation.version} (no fallback in pin mode)` + : `Version negotiation failed: the server did not offer pinned protocol version ${negotiation.version} ` + + `via server/discover (no fallback in pin mode)` ); } if (!negotiation.fallbackAvailable) { @@ -433,10 +471,38 @@ 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' + verdict.restart + ? 'Version negotiation failed: the connection closed during the server/discover probe ' + + '(a legacy signal on stdio) and this client supports no pre-2026-07-28 protocol version to fall back to' + : 'Version negotiation failed: the server gave no modern evidence and this client supports no ' + + 'pre-2026-07-28 protocol version to fall back to' ); } + if (verdict.restart) { + // The probe consumed the connection (the child exited): + // restart before the fallback handshake. Restartability + // is the transport's own capability — one without it, or + // whose restart refuses (a deliberate local close) or + // fails, is a typed negotiation failure. + const transport = deps.transport; + if (!isRestartableTransport(transport)) { + throw new SdkError( + SdkErrorCode.EraNegotiationFailed, + 'Version negotiation failed: the connection closed during the server/discover probe ' + + 'and this transport cannot restart for the legacy fallback' + ); + } + try { + await transport[TRANSPORT_RESTART](); + } catch (error) { + throw new SdkError( + SdkErrorCode.EraNegotiationFailed, + 'Version negotiation failed: the transport could not restart for the legacy fallback after ' + + `the connection closed during the probe: ${describeError(error)}`, + { cause: error } + ); + } + } 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..dfe6617f99 100644 --- a/packages/client/test/client/probeClassifier.test.ts +++ b/packages/client/test/client/probeClassifier.test.ts @@ -293,6 +293,43 @@ describe('row: timeout — transport-aware verdict', () => { }); }); +describe('row: connection closed mid-probe — transport-aware verdict', () => { + // The stdio counterpart of the timeout row: a child process that exits on + // the unrecognized probe instead of answering is legacy evidence — stdio + // servers built on SDKs that terminate on any pre-initialize request (the + // official Rust SDK, rmcp) behave exactly this way. The stream died with + // the process, so the verdict carries `restart`: the fallback initialize + // needs the transport restarted first. + test('stdio: close is a legacy signal requiring a transport restart — never a dead-end error', () => { + expect(classify({ kind: 'closed' }, { transportKind: 'stdio' })).toEqual({ kind: 'legacy', restart: true }); + }); + + test('stdio: the verdict is fallback-agnostic, like the timeout row — pin/modern-only conversion happens in the caller', () => { + expect(classify({ kind: 'closed' }, { transportKind: 'stdio', fallbackAvailable: false })).toEqual({ + kind: 'legacy', + restart: true + }); + }); + + test('HTTP: close mid-probe surfaces as the same typed connect error as any probe transport failure — never an era verdict', () => { + 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 as SdkError).message).toMatch(/Connection closed during the version negotiation probe/); + } + }); + + test('HTTP in a browser: still the typed error — the CORS leniency covers opaque fetch TypeErrors, not closes', () => { + const verdict = classify({ kind: 'closed' }, { transportKind: 'http', environment: 'browser' }); + expect(verdict.kind).toBe('error'); + if (verdict.kind === 'error') { + expect((verdict.error as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + } + }); +}); + 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/stdioRestart.test.ts b/packages/client/test/client/stdioRestart.test.ts new file mode 100644 index 0000000000..895ebd7f9e --- /dev/null +++ b/packages/client/test/client/stdioRestart.test.ts @@ -0,0 +1,302 @@ +/** + * StdioClientTransport restart contract — the transport owns its own + * restartability (internal `TRANSPORT_RESTART` capability, no public surface): + * + * - restart after the child died: respawn from the same parameters, events + * rebound to the new child, `send()` reaches it; + * - per-child event isolation: a slow-dying predecessor's events never touch + * the successor generation; + * - provenance: a local `close()` (caller shutdown or the transport's own + * error recovery) refuses to restart — a deliberately terminated server is + * never resurrected; + * - the read buffer is cleared on child exit — no partial trailing line leaks + * into the successor's stream; + * - with `stderr: 'pipe'`, the stream spans restarts: a child's exit never + * ends it; it ends exactly once, in `close()`. + */ +import type { Transport } from '@modelcontextprotocol/core-internal'; +import { isRestartableTransport, TRANSPORT_RESTART } from '@modelcontextprotocol/core-internal'; +import { describe, expect, test, vi } from 'vitest'; + +import { StdioClientTransport } from '../../src/client/stdio'; + +/** Line-delimited JSON-RPC child: answers `whoami` with its pid; `die` exits 1; `dieWithPartialLine` writes a truncated line, then exits. */ +const SERVER_SCRIPT = String.raw` + process.stderr.write('child-alive ' + process.pid + '\n'); + 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 === 'whoami' && message.id !== undefined) { + send({ jsonrpc: '2.0', id: message.id, result: { pid: process.pid } }); + } else if (message.method === 'die') { + process.exit(1); + } else if (message.method === 'dieWithPartialLine') { + process.stdout.write('{"jsonrpc":"2.0","id":9', () => process.exit(1)); + } + } + }); +`; + +const spawnTransport = (overrides?: { stderr?: 'pipe'; maxBufferSize?: number }): StdioClientTransport => + new StdioClientTransport({ command: process.execPath, args: ['-e', SERVER_SCRIPT], ...overrides }); + +/** Send `whoami` and await the child's pid reply (rebinds `onmessage`). */ +function whoami(transport: StdioClientTransport, id: number): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`whoami ${id} timed out`)), 5000); + transport.onmessage = message => { + if ('id' in message && message.id === id && 'result' in message) { + clearTimeout(timer); + resolve((message.result as { pid: number }).pid); + } + }; + transport.send({ jsonrpc: '2.0', id, method: 'whoami' }).catch((error: unknown) => { + clearTimeout(timer); + reject(error instanceof Error ? error : new Error(String(error))); + }); + }); +} + +test('carries the internal restart capability; a bare transport does not', () => { + expect(isRestartableTransport(spawnTransport())).toBe(true); + const bare: Transport = { + start: async () => {}, + send: async () => {}, + close: async () => {} + }; + expect(isRestartableTransport(bare)).toBe(false); +}); + +describe('restart after child death', () => { + test('respawns from the same parameters and rebinds: send() reaches the new child, its close is a new event', async () => { + const transport = spawnTransport(); + let closes = 0; + transport.onclose = () => { + closes++; + }; + + await transport.start(); + const firstPid = await whoami(transport, 1); + await transport.send({ jsonrpc: '2.0', method: 'die' }); + await vi.waitFor(() => expect(closes).toBe(1)); + expect(transport.pid).toBeNull(); + + await transport[TRANSPORT_RESTART](); + const secondPid = await whoami(transport, 2); + expect(secondPid).not.toBe(firstPid); + expect(transport.pid).toBe(secondPid); + + await transport.close(); + await vi.waitFor(() => expect(closes).toBe(2)); + }); + + test("the dead child's partial trailing line never leaks into the successor's stream", async () => { + const transport = spawnTransport(); + const errors: Error[] = []; + transport.onerror = error => { + errors.push(error); + }; + let closes = 0; + transport.onclose = () => { + closes++; + }; + + await transport.start(); + await transport.send({ jsonrpc: '2.0', method: 'dieWithPartialLine' }); + await vi.waitFor(() => expect(closes).toBe(1)); + + await transport[TRANSPORT_RESTART](); + const pid = await whoami(transport, 2); + expect(pid).toBe(transport.pid); + expect(errors).toEqual([]); + + await transport.close(); + }); +}); + +describe('per-child event isolation across generations', () => { + // The child answers `whoami`, exits promptly when its stdin ends, and + // spawns a grandchild that inherits (and so holds open) its stdio pipes: + // the transport's 'close' event for this child fires seconds AFTER the + // process itself died — deterministically after a successor was started. + const LINGERING_SERVER_SCRIPT = String.raw` + const { spawn } = require('child_process'); + spawn(process.execPath, ['-e', 'setTimeout(() => {}, 2500)'], { 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)); + `; + + test("a slow-dying predecessor's late close event cannot touch the successor", async () => { + const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', LINGERING_SERVER_SCRIPT] }); + + await transport.start(); + // close(): the child exits on stdin end, but its 'close' event pends on + // the grandchild's grip on the pipes — close() returns with the + // predecessor's close still in flight. + await transport.close(); + + await transport.start(); + let phantomCloses = 0; + transport.onclose = () => { + phantomCloses++; + }; + const successorPid = await whoami(transport, 1); + + // Let the predecessor's stale 'close' event land. + await new Promise(resolve => setTimeout(resolve, 1500)); + + expect(phantomCloses).toBe(0); + expect(transport.pid).toBe(successorPid); + // The successor's channel is untouched — still answering. + expect(await whoami(transport, 2)).toBe(successorPid); + + await transport.close(); + }, 15_000); +}); + +describe('restart provenance', () => { + test('a local close() refuses to restart — a deliberately terminated server is never resurrected', async () => { + const transport = spawnTransport(); + await transport.start(); + await transport.close(); + + await expect(transport[TRANSPORT_RESTART]()).rejects.toThrow(/closed locally/); + expect(transport.pid).toBeNull(); + }); + + test("the transport's own error-recovery close counts as local — restart refuses after a read-buffer overflow", async () => { + const transport = new StdioClientTransport({ + command: process.execPath, + args: ['-e', 'process.stdout.write(Buffer.alloc(200, 0x41)); setInterval(() => {}, 1000);'], + maxBufferSize: 100 + }); + const overflow = new Promise(resolve => { + transport.onerror = resolve; + }); + const closed = new Promise(resolve => { + transport.onclose = () => resolve(); + }); + + await transport.start(); + expect((await overflow).message).toMatch(/ReadBuffer exceeded maximum size/); + await closed; + + await expect(transport[TRANSPORT_RESTART]()).rejects.toThrow(/closed locally/); + }); + + test('a fresh caller start() after close() is a new life — restart works again after the next child death', async () => { + const transport = spawnTransport(); + await transport.start(); + await transport.close(); + + await transport.start(); + let closes = 0; + transport.onclose = () => { + closes++; + }; + await transport.send({ jsonrpc: '2.0', method: 'die' }); + await vi.waitFor(() => expect(closes).toBe(1)); + + await transport[TRANSPORT_RESTART](); + expect(transport.pid).not.toBeNull(); + await transport.close(); + }); +}); + +describe("stderr: 'pipe' contract", () => { + test("the stream spans restarts: a child's exit never ends it, and it carries every child's output", async () => { + const transport = spawnTransport({ stderr: 'pipe' }); + const chunks: string[] = []; + let ended = false; + transport.stderr?.on('data', chunk => chunks.push(String(chunk))); + transport.stderr?.on('end', () => { + ended = true; + }); + let closes = 0; + transport.onclose = () => { + closes++; + }; + + await transport.start(); + await vi.waitFor(() => expect(chunks.join('')).toMatch(/child-alive \d+/)); + await transport.send({ jsonrpc: '2.0', method: 'die' }); + await vi.waitFor(() => expect(closes).toBe(1)); + expect(ended).toBe(false); + + await transport[TRANSPORT_RESTART](); + await vi.waitFor(() => { + expect(chunks.join('').match(/child-alive \d+/g)?.length).toBe(2); + }); + const pids = [...chunks.join('').matchAll(/child-alive (\d+)/g)].map(m => m[1]); + expect(new Set(pids).size).toBe(2); + expect(ended).toBe(false); + + await transport.close(); + }); + + test('the stream ends exactly once, in close() — a second close() does not end it again', async () => { + const transport = spawnTransport({ stderr: 'pipe' }); + let ends = 0; + transport.stderr?.on('end', () => { + ends++; + }); + // Flowing mode so 'end' can fire. + transport.stderr?.on('data', () => {}); + + await transport.start(); + await transport.close(); + await vi.waitFor(() => expect(ends).toBe(1)); + + await transport.close(); + await new Promise(resolve => setTimeout(resolve, 50)); + expect(ends).toBe(1); + }); + + test("a caller start() after close() begins a fresh stream — the new child's output is observable, not silently dropped", async () => { + const transport = spawnTransport({ stderr: 'pipe' }); + const firstLife: string[] = []; + transport.stderr?.on('data', chunk => firstLife.push(String(chunk))); + await transport.start(); + // Drain the first child's banner so nothing stale can satisfy the second-life assertion. + await vi.waitFor(() => expect(firstLife.join('')).toMatch(/child-alive \d+/)); + await transport.close(); + + await transport.start(); + const pid = transport.pid; + const secondLife: string[] = []; + transport.stderr?.on('data', chunk => secondLife.push(String(chunk))); + await vi.waitFor(() => expect(secondLife.join('')).toContain(`child-alive ${pid}`)); + + await transport.close(); + }); +}); diff --git a/packages/client/test/client/versionNegotiation.test.ts b/packages/client/test/client/versionNegotiation.test.ts index 698bccd6ee..6484c1e001 100644 --- a/packages/client/test/client/versionNegotiation.test.ts +++ b/packages/client/test/client/versionNegotiation.test.ts @@ -8,12 +8,13 @@ * required 400) are exercised against real server transports in * test/integration/test/client/versionNegotiation.test.ts. */ -import type { JSONRPCMessage, JSONRPCRequest, Transport } from '@modelcontextprotocol/core-internal'; +import type { JSONRPCMessage, JSONRPCRequest, RestartableTransport, Transport } from '@modelcontextprotocol/core-internal'; import { isJSONRPCRequest, PROTOCOL_VERSION_META_KEY, SdkError, SdkErrorCode, + TRANSPORT_RESTART, UnsupportedProtocolVersionError } from '@modelcontextprotocol/core-internal'; import { describe, expect, test } from 'vitest'; @@ -438,6 +439,275 @@ 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) — and the transport restarts ITSELF + * (the internal TRANSPORT_RESTART capability; nothing outside the transport + * touches its lifecycle) before the fallback initialize runs. On HTTP-class + * transports a mid-probe close is ambiguous and rejects like any probe + * transport failure. + * ------------------------------------------------------------------------- */ + +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. Deliberately NOT + * restartable — restartability is a per-transport capability, not a + * property of being stdio-shaped. + */ + class RmcpShapedStdioTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + sessionId?: string; + + startCalls = 0; + sent: JSONRPCMessage[] = []; + setProtocolVersionCalls: string[] = []; + private _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._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 { + if (this._alive) { + this._alive = false; + this.onclose?.(); + } + } + + setProtocolVersion(version: string): void { + this.setProtocolVersionCalls.push(version); + } + } + + /** The restartable flavor: carries the transport-owned restart capability, like the real StdioClientTransport. */ + class RestartableRmcpTransport extends RmcpShapedStdioTransport implements RestartableTransport { + restartCalls = 0; + + async [TRANSPORT_RESTART](): Promise { + this.restartCalls++; + await this.start(); + } + } + + test('stdio: exit-on-probe restarts the transport (its own capability) and connects on the legacy era', async () => { + const transport = new RestartableRmcpTransport(); + 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"; the transport respawned itself. + expect(transport.restartCalls).toBe(1); + 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 restarted child. + expect(transport.setProtocolVersionCalls).toEqual(['2025-03-26']); + + // The probe-phase close was a real process exit and was delivered to + // the pre-set observer; the respawned process's close at session end is + // a NEW event and is delivered too. + 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 RestartableRmcpTransport(); + const autoClient = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + await autoClient.connect(autoTransport); + + const plainTransport = new RestartableRmcpTransport(); + 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 restarts. + expect(plainTransport.startCalls).toBe(1); + expect(plainTransport.restartCalls).toBe(0); + + await autoClient.close(); + await plainClient.close(); + }); + + test('stdio: a child that dies in the exchange gap (after answering -32022) still classifies as a close — restart, not a send failure', async () => { + // The close event lands while NO probe exchange is pending: the window + // must remember it, so the next exchange resolves closed instead of + // racing into a send-time NotConnected error. + class CorrectiveThenExitTransport extends RestartableRmcpTransport { + override async send(message: JSONRPCMessage): Promise { + if (isJSONRPCRequest(message) && message.method === 'server/discover') { + if (this.pid === null) { + throw new Error('Not connected'); + } + this.sent.push(message); + queueMicrotask(() => { + // Answer the probe, then die immediately: the reply + // settles the pending exchange first, so the close is + // delivered out-of-band (no pending). + this.onmessage?.({ + jsonrpc: '2.0', + id: message.id, + error: { code: -32_022, message: 'unsupported', data: { supported: [MODERN] } } + }); + void this.close(); + }); + return; + } + await super.send(message); + } + } + + const transport = new CorrectiveThenExitTransport(); + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); + + await client.connect(transport); + + expect(transport.restartCalls).toBe(1); + expect(client.getNegotiatedProtocolVersion()).toBe('2025-03-26'); + await client.close(); + }); + + test('stdio: pin mode fails loudly on exit-on-probe — no restart, no initialize, message names the close', async () => { + const transport = new RestartableRmcpTransport(); + 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 close-flavored diagnostic: the server never answered discover — + // the message must not imply it did. + expect((rejection as SdkError).message).toMatch(/connection closed during the server\/discover probe/); + expect(transport.restartCalls).toBe(0); + expect(transport.startCalls).toBe(1); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('stdio-shaped transport WITHOUT the restart capability: typed error — the probe layer never restarts from outside', async () => { + const transport = new RmcpShapedStdioTransport(); + 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(/cannot restart/); + expect(transport.startCalls).toBe(1); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('stdio: a restart the transport refuses (e.g. it was closed locally) is a typed negotiation error carrying the cause', async () => { + class RefusingRestartTransport extends RmcpShapedStdioTransport implements RestartableTransport { + async [TRANSPORT_RESTART](): Promise { + throw new Error('closed locally'); + } + } + const transport = new RefusingRestartTransport(); + 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('closed locally'); + expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); + }); + + test('stdio: a modern-only client gets the typed negotiation error — no restart, no initialize', async () => { + const transport = new RestartableRmcpTransport(); + const client = new Client( + { name: 'c', version: '0' }, + { versionNegotiation: { mode: 'auto' }, supportedProtocolVersions: [MODERN] } + ); + + await expect(client.connect(transport)).rejects.toSatisfy( + error => error instanceof SdkError && error.code === SdkErrorCode.EraNegotiationFailed + ); + expect(transport.restartCalls).toBe(0); + expect(transport.startCalls).toBe(1); + 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.setProtocolVersionCalls).toEqual([]); + }); +}); + /* ------------------------------------------------------------------------- * * -32022 corrective continuation — exactly once; loop guard on second * rejection. @@ -602,9 +872,10 @@ describe('pin mode', () => { expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false); }); - test('a failed negotiation leaves the transport start() untouched (no armed pass-through)', async () => { + test('a failed negotiation leaves the transport start() and close() untouched (no armed pass-through, no wrapper)', async () => { const transport = new ScriptedTransport(legacyServerScript); const originalStart = transport.start; + const originalClose = transport.close; const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: { pin: MODERN } } }); await expect(client.connect(transport)).rejects.toSatisfy( @@ -612,8 +883,11 @@ describe('pin mode', () => { ); // The probe window's one-shot start() pass-through must not stay armed - // on a transport the caller still owns after a failed connect. + // on a transport the caller still owns after a failed connect — and + // negotiation never wraps close() (by construction: nothing outside the + // transport touches its lifecycle methods). expect(transport.start).toBe(originalStart); + expect(transport.close).toBe(originalClose); expect(transport.onmessage).toBeUndefined(); }); }); @@ -801,9 +1075,14 @@ describe('probe window preserves pre-set transport handlers', () => { closed++; }; + const originalClose = transport.close; + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } }); await client.connect(transport); + // Negotiation never wraps close() — identity untouched after connect. + expect(transport.close).toBe(originalClose); + // The window restored the handler, so Protocol.connect chained it: // post-connect transport errors still reach the pre-set observer. const boom = new Error('post-connect transport error'); diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 226f6ab0bd..a97874f22d 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -177,3 +177,30 @@ export interface Transport { */ setSupportedProtocolVersions?: ((versions: string[]) => void) | undefined; } + +/** + * Internal capability marker (SDK-internal, on no public surface): keys a + * transport's restart method — see {@linkcode isRestartableTransport}. + * Registry symbol so the check survives dual-package (CJS/ESM) module twins. + */ +export const TRANSPORT_RESTART: unique symbol = Symbol.for('mcp.sdk.transportRestart') as never; + +/** + * A transport that owns its own restartability: it can re-establish the + * channel after the remote end died out from under a connection. Consumed by + * the client's version-negotiation close-fallback. + */ +export interface RestartableTransport extends Transport { + /** + * Re-establish the channel after the remote end died. Rejects when the + * transport cannot or must not restart — in particular after a local + * {@linkcode Transport.close | close()}, which is never era evidence and + * never resurrects the remote end. + */ + [TRANSPORT_RESTART](): Promise; +} + +/** `true` when `transport` carries the {@linkcode TRANSPORT_RESTART} capability. */ +export function isRestartableTransport(transport: Transport): transport is RestartableTransport { + return typeof (transport as Partial)[TRANSPORT_RESTART] === 'function'; +} diff --git a/test/integration/test/client/versionNegotiation.test.ts b/test/integration/test/client/versionNegotiation.test.ts index 2bd93eba93..1991b2723f 100644 --- a/test/integration/test/client/versionNegotiation.test.ts +++ b/test/integration/test/client/versionNegotiation.test.ts @@ -10,9 +10,12 @@ * * Plus: structural fallback hygiene (the auto client's post-probe traffic is * byte-identical to a plain legacy client's, zero 2026 headers), the typed - * connect errors for outage and HTTP timeout, and the stdio timeout fallback + * connect errors for outage and HTTP timeout, the stdio timeout fallback * (a silent legacy stdio server is detected by the probe timing out and the - * client falls back to initialize on the same pipe). + * client falls back to initialize on the same pipe), and the stdio close + * fallback (a legacy server that exits on the probe is respawned by the + * transport itself and initialized there; a locally closed transport is + * never respawned). */ import { randomUUID } from 'node:crypto'; import type { Server } from 'node:http'; @@ -24,7 +27,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 +289,151 @@ describe('stdio: silent legacy server (probe timeout fallback)', () => { } }, 15_000); }); + +describe('stdio: legacy server that exits on the probe (close-during-probe fallback)', () => { + // The other legacy-server stdio shape: servers built on SDKs that terminate + // the process on ANY pre-initialize request other than initialize itself + // (the official Rust SDK, rmcp, exits 1 with no JSON-RPC reply). Under + // mode: 'auto' the probe kills the child, so the fallback cannot run on the + // same pipe — the close classifies as a legacy signal, the transport + // restarts itself (respawning the server), and the plain initialize + // handshake runs there. The respawned process sees exactly what a + // mode: 'legacy' connect would have sent. + const EXIT_ON_PROBE_SERVER_SCRIPT = String.raw` + process.stderr.write('fixture-alive ' + process.pid + '\n'); + let buffer = ''; + let initialized = false; + 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' && message.id !== undefined) { + initialized = true; + send({ + jsonrpc: '2.0', + id: message.id, + result: { + protocolVersion: '2025-03-26', + capabilities: { tools: {} }, + serverInfo: { name: 'exit-on-probe-stdio-server', version: '1.0.0' } + } + }); + } else if (!initialized && message.id !== undefined) { + // The rmcp shape: die on any pre-initialize request, no reply. + process.exit(1); + } else if (message.method === 'tools/list' && message.id !== undefined) { + send({ jsonrpc: '2.0', id: message.id, result: { tools: [] } }); + } else if (message.id !== undefined) { + send({ jsonrpc: '2.0', id: message.id, error: { code: -32601, message: 'Method not found' } }); + } + } + }); + `; + + const spawnFixture = () => + new StdioClientTransport({ + command: process.execPath, + args: ['-e', EXIT_ON_PROBE_SERVER_SCRIPT] + }); + + it('auto mode: the probe kills the child, the transport respawns the server and the client connects on the legacy era', async () => { + const transport = spawnFixture(); + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' } }); + + try { + await client.connect(transport); + // A process that exits on the probe cannot be the one that answered + // initialize: the connection lives on a respawned child. + expect(transport.pid).not.toBeNull(); + expect(client.getNegotiatedProtocolVersion()).toBe('2025-03-26'); + expect(client.getServerVersion()?.name).toBe('exit-on-probe-stdio-server'); + // The respawned pipe is fully usable. + const tools = await client.listTools(); + expect(tools.tools).toEqual([]); + } finally { + await client.close(); + } + }, 15_000); + + it("stderr: 'pipe' spans the respawn — both children's stderr reaches the shared stream", async () => { + const transport = new StdioClientTransport({ + command: process.execPath, + args: ['-e', EXIT_ON_PROBE_SERVER_SCRIPT], + stderr: 'pipe' + }); + const collected: string[] = []; + transport.stderr?.on('data', chunk => collected.push(String(chunk))); + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' } }); + + try { + await client.connect(transport); + // Both the probe-killed child and its respawned successor announce + // themselves on stderr; the shared stream must carry both banners + // (it is not ended by the first child's exit). + await vi.waitFor(() => { + expect(collected.join('').match(/fixture-alive \d+/g)?.length).toBe(2); + }); + const pids = [...collected.join('').matchAll(/fixture-alive (\d+)/g)].map(m => m[1]); + expect(new Set(pids).size).toBe(2); + } finally { + await client.close(); + } + }, 15_000); + + it('a locally closed transport mid-probe is a typed error — the server is never respawned', async () => { + const transport = new StdioClientTransport({ + command: process.execPath, + args: [ + '-e', + String.raw`process.stderr.write('fixture-alive ' + process.pid + '\n'); process.stdin.on('end', () => process.exit(0)); process.stdin.resume();` + ], + stderr: 'pipe' + }); + const collected: string[] = []; + transport.stderr?.on('data', chunk => collected.push(String(chunk))); + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'auto' } }); + + const pending = client.connect(transport); + // Wait until the child is up (the probe is in flight), then shut the + // transport down from the caller side. + await vi.waitFor(() => expect(collected.join('')).toMatch(/fixture-alive \d+/)); + await transport.close(); + + const rejection = await pending.then( + () => {}, + (error: unknown) => error + ); + expect(rejection).toBeInstanceOf(SdkError); + expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed); + // Exactly one child ever existed — a deliberately terminated server is + // not resurrected. + expect(collected.join('').match(/fixture-alive \d+/g)?.length).toBe(1); + expect(transport.pid).toBeNull(); + }, 15_000); + + it("explicit mode: 'legacy' against the same server is untouched: no probe, one spawn, plain initialize", async () => { + const transport = spawnFixture(); + const client = new Client({ name: 'neg-client', version: '1.0.0' }, { versionNegotiation: { mode: 'legacy' } }); + + try { + // The fixture dies on ANY pre-initialize request — connecting at all + // proves nothing beyond the plain initialize handshake reached it. + await client.connect(transport); + expect(client.getNegotiatedProtocolVersion()).toBe('2025-03-26'); + const tools = await client.listTools(); + expect(tools.tools).toEqual([]); + } finally { + await client.close(); + } + }, 15_000); +});