diff --git a/src/client/stdio.ts b/src/client/stdio.ts index 816a686fce..bff8724eb7 100644 --- a/src/client/stdio.ts +++ b/src/client/stdio.ts @@ -28,6 +28,11 @@ export type StdioServerParameters = { * How to handle stderr of the child process. This matches the semantics of Node's `child_process.spawn`. * * The default is "inherit", meaning messages to stderr will be printed to the parent process's stderr. + * + * When set to "pipe" or "overlapped", stderr is exposed on `StdioClientTransport.stderr`. + * The SDK drains that stream so an unread pipe cannot fill and deadlock the session. + * Attach a `data` listener (or `.pipe()` it) before `start()` / `Client.connect` to + * receive every chunk; without a listener the bytes are discarded. */ stderr?: IOType | Stream | number; @@ -172,6 +177,10 @@ export class StdioClientTransport implements Transport { if (this._stderrStream && this._process.stderr) { this._process.stderr.pipe(this._stderrStream); + // Flowing mode discards unread chunks so a chatty child cannot + // fill the PassThrough (16 KiB highWaterMark) and block on + // write(2). Listeners attached before start() still receive data. + this._stderrStream.resume(); } }); } @@ -182,6 +191,13 @@ export class StdioClientTransport implements Transport { * If stderr piping was requested, a PassThrough stream is returned _immediately_, allowing callers to * attach listeners before the start method is invoked. This prevents loss of any early * error output emitted by the child process. + * + * After `start()`, the SDK puts this stream in flowing mode so piping is safe without a + * consumer — unread stderr is drained and cannot deadlock the session. A listener (or + * `.pipe()` destination) attached before `start()` still receives every chunk. A late + * listener sees only data written after it attached; bytes already drained are not + * replayed. The paused-mode API (`read()` in a loop) is not supported once `start()` + * has put the stream in flowing mode — use `data` events or `.pipe()`. */ get stderr(): Stream | null { if (this._stderrStream) { diff --git a/test/client/stdio.test.ts b/test/client/stdio.test.ts index 577e0c946f..0e2461e42d 100644 --- a/test/client/stdio.test.ts +++ b/test/client/stdio.test.ts @@ -1,4 +1,5 @@ import { JSONRPCMessage } from '../../src/types.js'; +import { Client } from '../../src/client/index.js'; import { StdioClientTransport, StdioServerParameters } from '../../src/client/stdio.js'; // Configure default server parameters based on OS @@ -116,3 +117,119 @@ test('should fire onerror and close when ReadBuffer overflows', async () => { expect(error.message).toMatch(/ReadBuffer exceeded maximum size/); await closed; }); + +/** Unique markers so late-attach tests can tell drained startup bytes from post-request bytes. */ +const STDERR_STARTUP_MARKER = 'STDERR_STARTUP_MARKER'; +const STDERR_POST_REQUEST_MARKER = 'STDERR_POST_REQUEST_MARKER'; + +/** + * Minimal MCP server that floods stderr on every post-handshake request. + * Enough writes to fill a paused PassThrough (16 KiB) and the OS pipe (~64 KiB) + * so an undrained `stderr: 'pipe'` would block the child on write(2). + */ +function chattyStderrServerScript(): string { + return String.raw` + const { createInterface } = require('readline'); + const send = (o) => process.stdout.write(JSON.stringify(o) + '\n'); + process.stderr.write(${JSON.stringify(`${STDERR_STARTUP_MARKER}\n`)}); + createInterface({ input: process.stdin }).on('line', (line) => { + const m = JSON.parse(line); + if (m.method === 'initialize') { + return send({ + jsonrpc: '2.0', + id: m.id, + result: { + protocolVersion: '2025-06-18', + capabilities: { tools: {} }, + serverInfo: { name: 'chatty', version: '1.0.0' } + } + }); + } + if (m.method === 'notifications/initialized') return; + for (let i = 0; i < 256; i++) process.stderr.write('x'.repeat(512) + '\n'); + process.stderr.write(${JSON.stringify(`${STDERR_POST_REQUEST_MARKER}\n`)}); + send({ + jsonrpc: '2.0', + id: m.id, + result: { tools: [{ name: 'x', description: 'd', inputSchema: { type: 'object' } }] } + }); + }); + `; +} + +function chattyStderrTransport(): StdioClientTransport { + return new StdioClientTransport({ + command: process.execPath, + args: ['-e', chattyStderrServerScript()], + stderr: 'pipe' + }); +} + +/** stdout can settle before every flowing-mode stderr chunk is delivered. */ +async function waitForStderrMarker(captured: { text: string }, marker: string): Promise { + await vi.waitFor( + () => { + if (!captured.text.includes(marker)) { + throw new Error(`stderr has not yet included ${marker}`); + } + }, + { timeout: 3000, interval: 10 } + ); +} + +test('piped stderr without a reader does not deadlock listTools', async () => { + const transport = chattyStderrTransport(); + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toEqual([{ name: 'x', description: 'd', inputSchema: { type: 'object' } }]); + } finally { + await client.close(); + } +}, 8000); + +test('piped stderr listener attached before start still receives chunks', async () => { + const transport = chattyStderrTransport(); + const stderr = transport.stderr; + expect(stderr).not.toBeNull(); + const captured = { text: '' }; + stderr!.on('data', (chunk: Buffer) => { + captured.text += chunk.toString(); + }); + + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + await waitForStderrMarker(captured, STDERR_STARTUP_MARKER); + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toHaveLength(1); + await waitForStderrMarker(captured, STDERR_POST_REQUEST_MARKER); + } finally { + await client.close(); + } +}, 8000); + +test('late stderr listener does not deadlock and sees only post-attach chunks', async () => { + const transport = chattyStderrTransport(); + const client = new Client({ name: 'demo', version: '1.0.0' }); + try { + await client.connect(transport); + // Flowing-mode drain of startup stderr needs a turn to settle before we attach. + await new Promise(resolve => setImmediate(resolve)); + + const stderr = transport.stderr; + expect(stderr).not.toBeNull(); + const captured = { text: '' }; + stderr!.on('data', (chunk: Buffer) => { + captured.text += chunk.toString(); + }); + + const result = await client.listTools(undefined, { timeout: 5000 }); + expect(result.tools).toHaveLength(1); + await waitForStderrMarker(captured, STDERR_POST_REQUEST_MARKER); + expect(captured.text.includes(STDERR_STARTUP_MARKER)).toBe(false); + } finally { + await client.close(); + } +}, 8000);