From bc676068deaf44f20363809654ac070d5a8d799e Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Fri, 17 Jul 2026 17:47:05 +0000 Subject: [PATCH] fix(client): handle connection close during the version negotiation probe A stdio server that exits on the unrecognized server/discover probe (the rmcp shape: terminate on any pre-initialize request) connected fine under mode 'legacy' but hard-failed under mode 'auto'. Classify that close as a legacy signal, symmetric with the stdio probe-timeout rule. StdioClientTransport owns the restart: an internal capability respawns the server after the child died, with generation-scoped child event handlers, the read buffer cleared on child exit, and refusal after a local close(); with stderr 'pipe' the aggregate stream spans restarts and ends exactly once, in close(). The probe layer checks for the capability and asks the transport to restart before running the plain initialize fallback; a transport without it, or whose restart refuses or fails, rejects with the existing typed negotiation error. On HTTP a mid-probe close keeps surfacing as the same typed connect error as any probe transport failure. --- .changeset/probe-close-handling.md | 10 + docs/migration/support-2026-07-28.md | 12 +- docs/protocol-versions.md | 2 + packages/client/src/client/probeClassifier.ts | 32 +- packages/client/src/client/stdio.ts | 89 ++++-- .../client/src/client/versionNegotiation.ts | 98 +++++- .../test/client/probeClassifier.test.ts | 37 +++ .../client/test/client/stdioRestart.test.ts | 302 ++++++++++++++++++ .../test/client/versionNegotiation.test.ts | 285 ++++++++++++++++- .../core-internal/src/shared/transport.ts | 27 ++ .../test/client/versionNegotiation.test.ts | 157 ++++++++- 11 files changed, 1005 insertions(+), 46 deletions(-) create mode 100644 .changeset/probe-close-handling.md create mode 100644 packages/client/test/client/stdioRestart.test.ts 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); +});