fix(client): probe stdio servers on a disposable sibling process (#2514)

This commit is contained in:
Felix Weinberger 2026-07-21 13:07:12 +01:00 • committed by GitHub
commit 6fe196310d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1119 additions and 66 deletions

View file

@ -62,7 +62,7 @@ describe('row: DiscoverResult with version overlap → modern, select from suppo
});
});
describe('row: DiscoverResult with NO overlap → initialize on the same connection, else typed error with synthesized data', () => {
describe('row: DiscoverResult with NO overlap → legacy initialize fallback, else typed error with synthesized data', () => {
test('fallback possible → legacy (era selection on a dual-era server)', () => {
const verdict = classify({ kind: 'result', result: discoverResult(['2027-12-31']) });
expect(verdict).toEqual({ kind: 'legacy' });
@ -288,11 +288,27 @@ describe('row: timeout — transport-aware verdict', () => {
}
});
test('stdio: timeout is a legacy-server signal → fall back to initialize on the same stream', () => {
test('stdio: timeout is a legacy-server signal → legacy initialize fallback', () => {
expect(classify({ kind: 'timeout', timeoutMs: 5_000 }, { transportKind: 'stdio' })).toEqual({ kind: 'legacy' });
});
});
describe('row: closed — transport-aware verdict, symmetric with the timeout row', () => {
test('stdio: a child that exits on the unrecognized probe is a legacy signal → legacy verdict', () => {
expect(classify({ kind: 'closed' }, { transportKind: 'stdio' })).toEqual({ kind: 'legacy' });
});
test('HTTP: a mid-probe close is ambiguous — the same typed connect error as any probe network failure', () => {
const verdict = classify({ kind: 'closed' }, { transportKind: 'http' });
expect(verdict.kind).toBe('error');
if (verdict.kind === 'error') {
expect(verdict.error).toBeInstanceOf(SdkError);
expect((verdict.error as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed);
expect(verdict.error.message).toContain('Connection closed during the version negotiation probe');
}
});
});
describe('row: browser opaque CORS/preflight TypeError, PROBE PHASE ONLY → legacy fallback (F-7)', () => {
test('browser environment + bare TypeError → legacy', () => {
expect(classify({ kind: 'network-error', error: new TypeError('Failed to fetch') }, { environment: 'browser' })).toEqual({

View file

@ -89,7 +89,7 @@ const CORPUS: CorpusRow[] = [
},
// --- T9 edge 3: probe success but no version overlap → initialize on the SAME connection.
{
name: 'T9: DiscoverResult with no mutual version + fallback available → legacy (initialize on the same connection)',
name: 'T9: DiscoverResult with no mutual version + fallback available → legacy (initialize fallback)',
outcome: {
kind: 'result',
result: { supportedVersions: ['2027-01-01'], capabilities: {} }

View file

@ -1,3 +1,6 @@
import { existsSync } from 'node:fs';
import { tmpdir } from 'node:os';
import type { JSONRPCMessage } from '@modelcontextprotocol/core-internal';
import type { StdioServerParameters } from '../../src/client/stdio';
@ -118,3 +121,32 @@ test('should fire onerror and close when ReadBuffer overflows', async () => {
expect(error.message).toMatch(/ReadBuffer exceeded maximum size/);
await closed;
});
test('_dispose releases the parent-side pipe handles even when a helper process holds the child stdio', async () => {
// The rmcp-holding anatomy: the child exits, but a helper it spawned with
// stdio: 'inherit' keeps the pipe write ends open. Awaiting 'exit' settles
// disposal promptly — but without destroying the PARENT-side handles, the
// flowing stdout read handle stays ref'd until the helper exits, pinning
// the host's event loop (indefinitely for a daemon helper).
const readyFile = `${tmpdir()}/mcp-dispose-ready-${process.pid}-${Date.now()}`;
const HOLDING_SCRIPT = String.raw`
const { spawn } = require('child_process');
spawn(process.execPath, ['-e', 'setTimeout(() => {}, 3000)'], { stdio: 'inherit' });
require('fs').writeFileSync(${JSON.stringify(readyFile)}, 'ready');
process.stdin.on('end', () => process.exit(0));
process.stdin.resume();
`;
const transport = new StdioClientTransport({ command: process.execPath, args: ['-e', HOLDING_SCRIPT] });
await transport.start();
// Wait until the grandchild actually exists and holds the inherited pipes
// — disposing earlier would kill the child before its script even runs.
while (!existsSync(readyFile)) await new Promise(resolve => setTimeout(resolve, 25));
const proc = (transport as unknown as { _process: import('node:child_process').ChildProcess })._process;
await (transport as unknown as { _dispose: () => Promise<void> })._dispose();
// The child is confirmed gone AND the parent-side handles are released —
// destroyed flags are the deterministic proxy for "nothing pins the loop".
expect(proc.stdout?.destroyed).toBe(true);
expect(proc.stdin?.destroyed).toBe(true);
}, 10_000);

View file

@ -438,6 +438,342 @@ describe('probe timeout policy (transport-aware)', () => {
});
});
/* ------------------------------------------------------------------------- *
* Probe close policy. The SDK's stdio transport probes on a DISPOSABLE
* SIBLING spawned from the same `_serverParams` — servers built on SDKs that
* terminate on any pre-initialize request (the official Rust SDK, rmcp) exit
* when the probe arrives, so the probe must not spend the caller's one child
* life. The session transport starts exactly once, after the era is known.
* Stdio-shaped transports without readable params probe in place, where a
* mid-probe close stays a typed error; HTTP closes stay typed errors.
* ------------------------------------------------------------------------- */
describe('stdio sibling probe (disposable probe transport)', () => {
/**
* An SDK-shaped stdio transport: retains `_serverParams` (the sibling
* seam), and scripts its "child" from `params.behavior` — 'rmcp' exits on
* any pre-initialize request, 'modern' answers server/discover, 'silent'
* never replies. Mirrors StdioClientTransport's surface, including the
* internal `_dispose` reaper (recorded, for the leak pins).
*/
class FakeStdioTransport implements Transport {
static instances: FakeStdioTransport[] = [];
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
sessionId?: string;
readonly _serverParams: Record<string, unknown>;
startCalls = 0;
sent: JSONRPCMessage[] = [];
setProtocolVersionCalls: string[] = [];
disposed = false;
private _alive = false;
private _initialized = false;
constructor(params: Record<string, unknown>) {
this._serverParams = params;
FakeStdioTransport.instances.push(this);
}
get stderr(): null {
return null;
}
get pid(): number | null {
return this._alive ? 100 + FakeStdioTransport.instances.indexOf(this) : null;
}
async start(): Promise<void> {
if (this._alive) throw new Error('FakeStdioTransport already started!');
if (this._serverParams.behavior === 'spawnfail') throw new Error('spawn ENOENT');
this._alive = true;
this._initialized = false;
this.startCalls++;
}
async send(message: JSONRPCMessage): Promise<void> {
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: 'fake-stdio-server', version: '1.0.0' }
}
});
} else if (message.method === 'server/discover' && this._serverParams.behavior === 'modern') {
this.onmessage?.({ jsonrpc: '2.0', id: message.id, result: discoverResult([MODERN]) });
} else if (!this._initialized && this._serverParams.behavior === 'rmcp') {
// The rmcp shape: any other pre-initialize request kills the
// child — no reply, just the close.
this._alive = false;
this.onclose?.();
}
// 'silent': no reply at all.
});
}
async close(): Promise<void> {
if (this._alive) {
this._alive = false;
this.onclose?.();
}
}
/** Test hook: observe the moment disposal begins (reset after use). */
static onDisposeStart: (() => void) | undefined;
// Called structurally by the sibling reaper, like the real transport's.
private async _dispose(): Promise<void> {
FakeStdioTransport.onDisposeStart?.();
this.disposed = true;
this._alive = false;
}
setProtocolVersion(version: string): void {
this.setProtocolVersionCalls.push(version);
}
}
const freshInstances = () => {
FakeStdioTransport.instances = [];
};
test('rmcp exit-on-probe: the sibling spends itself, the session connects legacy on its first and only spawn', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'rmcp' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await client.connect(session);
expect(FakeStdioTransport.instances).toHaveLength(2);
const sibling = FakeStdioTransport.instances[1]!;
// The sibling carries the same params with stderr discarded, took the
// probe (and only the probe), and was reaped.
expect(sibling._serverParams).toMatchObject({ command: 'srv', stderr: 'ignore' });
expect(requests(sibling.sent).map(r => r.method)).toEqual(['server/discover']);
expect(sibling.disposed).toBe(true);
// The session transport: one spawn, no probe bytes, plain legacy connect.
expect(session.startCalls).toBe(1);
expect(requests(session.sent).map(r => r.method)).toEqual(['initialize']);
expect(client.getNegotiatedProtocolVersion()).toBe('2025-03-26');
expect(session.setProtocolVersionCalls).toEqual(['2025-03-26']);
await client.close();
});
test("the session's traffic is byte-identical to a plain mode:'legacy' connect", async () => {
freshInstances();
const autoSession = new FakeStdioTransport({ command: 'srv', behavior: 'rmcp' });
const autoClient = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await autoClient.connect(autoSession);
const plainSession = new FakeStdioTransport({ command: 'srv', behavior: 'rmcp' });
const plainClient = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'legacy' } });
await plainClient.connect(plainSession);
expect(JSON.stringify(autoSession.sent)).toBe(JSON.stringify(plainSession.sent));
// Explicit legacy mode never probes and never spawns a sibling.
expect(FakeStdioTransport.instances).toHaveLength(3);
expect(plainSession.startCalls).toBe(1);
await autoClient.close();
await plainClient.close();
});
test('modern server: the verdict is adopted verbatim — the session wire carries NO server/discover and no initialize', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'modern' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await client.connect(session);
expect(FakeStdioTransport.instances).toHaveLength(2);
expect(FakeStdioTransport.instances[1]!.disposed).toBe(true);
expect(client.getNegotiatedProtocolVersion()).toBe(MODERN);
expect(client.getServerVersion()?.name).toBe('scripted-modern-server');
expect(session.startCalls).toBe(1);
// Byte-trace: the sibling probed; the session sent nothing at connect.
expect(requests(session.sent)).toHaveLength(0);
expect(session.setProtocolVersionCalls).toEqual([MODERN]);
await client.close();
});
test('caller close() mid-probe aborts promptly: typed error, the session transport is never started', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'silent' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto', probe: { timeoutMs: 30_000 } } });
const pending = client.connect(session);
pending.catch(() => {});
await new Promise(resolve => setTimeout(resolve, 0));
await session.close();
const rejection = await pending.then(
() => undefined,
(error: unknown) => error
);
expect(rejection).toBeInstanceOf(SdkError);
expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed);
expect((rejection as SdkError).message).toMatch(/transport was closed during the server\/discover probe/);
expect(session.startCalls).toBe(0);
expect(FakeStdioTransport.instances[1]!.disposed).toBe(true);
// The wrapper restored the caller's close by identity.
expect(session.close).toBe(FakeStdioTransport.prototype.close);
});
test('pin mode: exit-on-probe rejects naming the close — session never started', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'rmcp' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: { pin: MODERN } } });
const rejection = await client.connect(session).then(
() => undefined,
(error: unknown) => error
);
expect(rejection).toBeInstanceOf(SdkError);
expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed);
expect((rejection as SdkError).message).toMatch(
/connection closed during the server\/discover probe before the server offered pinned/
);
expect(session.startCalls).toBe(0);
expect(FakeStdioTransport.instances[1]!.disposed).toBe(true);
});
test('modern-only client: exit-on-probe rejects naming the close — session never started', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'rmcp' });
const client = new Client(
{ name: 'c', version: '0' },
{ versionNegotiation: { mode: 'auto' }, supportedProtocolVersions: [MODERN] }
);
const rejection = await client.connect(session).then(
() => undefined,
(error: unknown) => error
);
expect(rejection).toBeInstanceOf(SdkError);
expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed);
expect((rejection as SdkError).message).toMatch(/connection closed during the server\/discover probe and this client supports no/);
expect(session.startCalls).toBe(0);
});
test('a sibling that cannot spawn propagates the raw spawn error (the session would fail identically)', async () => {
freshInstances();
const session = new FakeStdioTransport({ command: 'missing', behavior: 'spawnfail' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await expect(client.connect(session)).rejects.toThrow(/spawn ENOENT/);
expect(session.startCalls).toBe(0);
});
test('a SUBCLASS of the SDK stdio transport probes in place: typed error naming the base-class requirement, no sibling spawned', async () => {
// A subclass cannot be faithfully cloned by re-invoking its constructor
// with the retained params alone — the gate is exact-class (only the
// base prototype owns the internal reaper), so subclasses keep the
// in-place behavior.
freshInstances();
class SubclassedStdioTransport extends FakeStdioTransport {}
const transport = new SubclassedStdioTransport({ command: 'srv', behavior: 'rmcp' });
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(/sibling probe requires the SDK's base StdioClientTransport/);
// No sibling was constructed: the only instance is the caller's own,
// and it took the probe itself (in place).
expect(FakeStdioTransport.instances).toHaveLength(1);
expect(requests(transport.sent).map(r => r.method)).toEqual(['server/discover']);
});
test('a caller close() landing during sibling DISPOSAL still aborts: the session transport is never started', async () => {
// The close watch stays armed through disposal (the finally disposes
// BEFORE restoring close, and the abort is re-checked after it), so a
// shutdown racing the reap cannot be followed by a session spawn.
freshInstances();
const session = new FakeStdioTransport({ command: 'srv', behavior: 'modern' });
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
let closedDuringDisposal = false;
FakeStdioTransport.onDisposeStart = () => {
if (!closedDuringDisposal) {
closedDuringDisposal = true;
void session.close();
}
};
try {
const rejection = await client.connect(session).then(
() => undefined,
(error: unknown) => error
);
expect(closedDuringDisposal).toBe(true);
expect(rejection).toBeInstanceOf(SdkError);
expect((rejection as SdkError).code).toBe(SdkErrorCode.EraNegotiationFailed);
expect((rejection as SdkError).message).toMatch(/transport was closed during the server\/discover probe/);
expect(session.startCalls).toBe(0);
} finally {
FakeStdioTransport.onDisposeStart = undefined;
}
});
test('a stdio-shaped transport WITHOUT readable spawn params probes in place: a mid-probe close stays a typed error', async () => {
// Foreign transports have no sibling seam — exactly main's behavior.
class ForeignStdioTransport extends ScriptedTransport {
get stderr(): null {
return null;
}
get pid(): number {
return 4242;
}
}
const transport = new ForeignStdioTransport((message, t) => {
if (!isJSONRPCRequest(message)) return;
if (message.method === 'server/discover') {
t.onclose?.();
}
});
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(/connection closed during the server\/discover probe/);
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, never a sibling', async () => {
const script: Script = (message, t) => {
if (!isJSONRPCRequest(message)) return;
if (message.method === 'server/discover') {
t.onclose?.();
return;
}
legacyServerScript(message, t);
};
const transport = new ScriptedTransport(script);
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await expect(client.connect(transport)).rejects.toSatisfy(
error => error instanceof SdkError && error.code === SdkErrorCode.EraNegotiationFailed
);
expect(transport.sent.some(m => 'method' in m && m.method === 'initialize')).toBe(false);
expect(transport.startCalls).toBe(1);
});
});
/* ------------------------------------------------------------------------- *
* -32022 corrective continuation — exactly once; loop guard on second
* rejection.
@ -869,6 +1205,114 @@ describe('probe window preserves pre-set transport handlers', () => {
await client.connect(transport).catch(() => undefined);
expect(closes).toBe(1);
// The spent-close guard was disarmed once the cleanup close settled
// (ScriptedTransport.close() re-fires onclose, like the real HTTP
// transport — the guard consumed exactly that re-delivery). A LATER
// close is genuine and must reach the pre-set observer directly.
transport.onclose?.();
expect(closes).toBe(2);
});
test('a negotiation that SUCCEEDS after a same-tick reply-then-close keeps the observer armed for the live session', async () => {
// The success-path corner: the reply settles the exchange, then the
// close arrives in the same tick — forwarded with nothing pending, so
// negotiation still classifies the reply and succeeds. No cleanup
// close ever runs on success, so no re-delivery is coming: the guard
// must be disarmed before Protocol chains the handler slot, or the
// session's first genuine close would be swallowed.
const script: Script = (message, t) => {
if (!isJSONRPCRequest(message)) return;
if (message.method === 'server/discover') {
t.reply({ jsonrpc: '2.0', id: message.id, error: { code: -32_601, message: 'Method not found' } });
t.onclose?.();
return;
}
legacyServerScript(message, t);
};
const transport = new ScriptedTransport(script);
let closes = 0;
transport.onclose = () => {
closes++;
};
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
await client.connect(transport);
expect(client.getNegotiatedProtocolVersion()).toBe('2025-11-25');
// The mid-window close was forwarded exactly once.
expect(closes).toBe(1);
// The session's own close is genuine and must reach the observer.
await client.close();
expect(closes).toBe(2);
});
test("stdio-semantics transport (close() never re-fires onclose): a restarted life's genuine close reaches the pre-set observer after a failed negotiation", async () => {
// Real stdio semantics: onclose fires only from the child's own close
// event — the cleanup close() on a dead transport re-delivers NOTHING,
// so the spent-close guard cannot rely on that re-delivery to consume
// its skip. It is disarmed from the cleanup site instead, once the
// cleanup close has settled. Ordering note: a naive self-restoring
// wrapper would NOT fix this — with no re-delivery ever coming, the
// wrapper's FIRST invocation would already be the next genuine close,
// swallowed even as it restores the handler.
class NoRefireStdioShapedTransport implements Transport {
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
sessionId?: string;
startCalls = 0;
sent: JSONRPCMessage[] = [];
private _alive = false;
get stderr(): null {
return null;
}
get pid(): number | null {
return this._alive ? 4242 : null;
}
async start(): Promise<void> {
this._alive = true;
this.startCalls++;
}
async send(message: JSONRPCMessage): Promise<void> {
if (!this._alive) throw new Error('Not connected');
this.sent.push(message);
queueMicrotask(() => {
if (!this._alive || !isJSONRPCRequest(message)) return;
// Exits on the probe, like an rmcp child: the close event.
this.simulateChildClose();
});
}
async close(): Promise<void> {
// stdio semantics: close() tears down but never re-fires
// onclose — the child's close event already delivered it.
this._alive = false;
}
/** The child's own close event — the only onclose source, like the real transport. */
simulateChildClose(): void {
this._alive = false;
this.onclose?.();
}
}
const transport = new NoRefireStdioShapedTransport();
let closes = 0;
transport.onclose = () => {
closes++;
};
const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto' } });
// No _serverParams: probes in place; the mid-probe close is a typed error.
await expect(client.connect(transport)).rejects.toThrow(/connection closed during the server\/discover probe/);
// The child's death mid-probe was forwarded to the observer exactly once
// (the cleanup close, per stdio semantics, re-delivered nothing).
expect(closes).toBe(1);
// The caller restarts the transport; the new life's death is a GENUINE
// close and must reach the observer — an armed skip would swallow it.
await transport.start();
transport.simulateChildClose();
expect(closes).toBe(2);
});
test('transport errors DURING the probe window are forwarded to the pre-set handler', async () => {