feat(mcp): PipeTransport, the stdio transport over a program started by the caller

StdioTransport and PipeTransport share the newline-delimited framing and
the stderr tail (transports/lines.ts). PipeTransport accepts pi-env's
RemoteProcess as its McpPipe, so p2 can run stdio MCP servers on
execution environments through the daemon's process operation.

packages/mcp/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Added
+
+- Added `PipeTransport`, MCP's stdio transport over a program started by the caller through an `McpPipe` (`write`, `closeStdin`, `close`, `exited`), for example `@earendil-works/pi-env`'s `openProcess()` on another machine. `StdioTransport` and `PipeTransport` share the newline-delimited framing and the stderr tail.
+
 ## [1.5.0] - 2026-11-02
 
 ### Changed

packages/mcp/README.md

@@ -2,7 +2,7 @@
 
 A small, standalone Model Context Protocol client. It does not depend on the official MCP SDK or other pi packages.
 
-The package provides a transport-neutral client core, stdio and Streamable HTTP transports, and an in-memory testing transport.
+The package provides a transport-neutral client core, stdio and Streamable HTTP transports, a stdio transport over a pipe to a program started elsewhere, and an in-memory testing transport.
 
 ## Usage
 
@@ -27,6 +27,22 @@ await client.close();
 
 For a remote server, use `new StreamableHttpTransport({ url, headers })`. Fetch can be injected for proxying or custom networking.
 
+### Stdio servers started elsewhere
+
+`PipeTransport` speaks MCP's stdio framing over a program that something else starts, for example on another machine. `open` starts the program, connects its stdout and stderr to the given handlers, and returns an `McpPipe` (`write`, `closeStdin`, `close` to kill, and an `exited` promise). `@earendil-works/pi-env`'s `openProcess()` returns one, without an adapter:
+
+```typescript
+import { openProcess } from "@earendil-works/pi-env";
+import { PipeTransport } from "@earendil-works/pi-mcp";
+
+const transport = new PipeTransport({
+	open: (handlers) => openProcess(connection, { cwd, argv, ...handlers }),
+});
+await client.connect(transport);
+```
+
+Closing the transport closes the program's stdin and kills it if it has not exited after `closeTimeoutMs` (default 2000). The transport closes when the program ends.
+
 ### Tools for an LLM
 
 `toLlmContent(result)` converts a `CallToolResult` to text and image content for a model, in the shape of `@earendil-works/pi-ai`'s `TextContent` and `ImageContent`. Text and images pass through, embedded text and image resources are unwrapped, and audio, resource links, and binary resources become short text placeholders. A result without content blocks but with `structuredContent` becomes its JSON.
@@ -123,6 +139,7 @@ An MCP transport owns framing and I/O. It delivers individual JSON-RPC messages
 - request cancellation
 - Streamable HTTP sessions, the server-to-client GET stream with reconnection, and resumption of dropped response streams with `Last-Event-ID`
 - stdio shutdown per the spec (close stdin, then SIGTERM, then SIGKILL), applied to the server's whole process group
+- stdio over a caller-provided pipe (`PipeTransport`)
 - server `ping` and `roots/list` requests
 - logging and tool-list-change notifications through the generic notification API
 - OAuth protected-resource and authorization-server discovery

packages/mcp/src/index.ts

@@ -55,6 +55,12 @@ export {
 	type ToolAnnotations,
 	type ToolExecution,
 } from "./protocol/types.ts";
+export {
+	type McpPipe,
+	type McpPipeHandlers,
+	PipeTransport,
+	type PipeTransportOptions,
+} from "./transports/pipe.ts";
 export { StdioTransport, type StdioTransportOptions } from "./transports/stdio.ts";
 export {
 	McpAuthRequiredError,

packages/mcp/src/transports/lines.ts

@@ -0,0 +1,77 @@
+import { type JsonRpcMessage, parseJsonRpcMessage } from "../protocol/jsonrpc.ts";
+import { DEFAULT_MAX_MESSAGE_BYTES } from "./transport.ts";
+
+const DEFAULT_MAX_STDERR_BYTES = 64 * 1024;
+
+export interface JsonRpcLineReaderOptions {
+	maxMessageBytes?: number;
+	maxStderrBytes?: number;
+	/** Every stderr chunk, decoded as UTF-8. */
+	onStderr?: (chunk: string) => void;
+	onMessage(message: JsonRpcMessage): void;
+	onError(error: unknown): void;
+}
+
+/**
+ * Reads a program's output as MCP's stdio framing: newline-delimited JSON-RPC messages on stdout, and a bounded
+ * tail of stderr for diagnostics.
+ */
+export class JsonRpcLineReader {
+	private readonly options: JsonRpcLineReaderOptions;
+	private stdoutBuffer = Buffer.alloc(0);
+	private stderrBuffer = Buffer.alloc(0);
+
+	constructor(options: JsonRpcLineReaderOptions) {
+		this.options = options;
+	}
+
+	/** The last `maxStderrBytes` of stderr. */
+	get stderr(): string {
+		return this.stderrBuffer.toString("utf8");
+	}
+
+	/** Takes the next stdout bytes and reports every complete message, or the error parsing it. */
+	readStdout(chunk: Uint8Array): void {
+		this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, chunk]);
+		const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES;
+		while (true) {
+			const newline = this.stdoutBuffer.indexOf(0x0a);
+			if (newline < 0) {
+				if (this.stdoutBuffer.length > maxMessageBytes) {
+					this.stdoutBuffer = Buffer.alloc(0);
+					this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
+				}
+				return;
+			}
+			const line = this.stdoutBuffer.subarray(0, newline);
+			this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1);
+			if (line.length > maxMessageBytes) {
+				this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
+				continue;
+			}
+			const text = line.toString("utf8").replace(/\r$/, "");
+			if (!text.trim()) continue;
+			try {
+				this.options.onMessage(parseJsonRpcMessage(JSON.parse(text)));
+			} catch (error) {
+				this.options.onError(error);
+			}
+		}
+	}
+
+	/** Takes the next stderr bytes into the tail and passes them to `onStderr`. */
+	readStderr(chunk: Uint8Array): void {
+		const maxStderrBytes = this.options.maxStderrBytes ?? DEFAULT_MAX_STDERR_BYTES;
+		this.stderrBuffer = Buffer.concat([this.stderrBuffer, chunk]);
+		if (this.stderrBuffer.length > maxStderrBytes) this.stderrBuffer = this.stderrBuffer.subarray(-maxStderrBytes);
+		this.options.onStderr?.(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString("utf8"));
+	}
+
+	/** Ends stdout: a message without its newline is reported as an error and dropped. */
+	end(): void {
+		if (this.stdoutBuffer.toString("utf8").trim()) {
+			this.options.onError(new Error("MCP stdio server closed with an incomplete JSON-RPC message"));
+		}
+		this.stdoutBuffer = Buffer.alloc(0);
+	}
+}

packages/mcp/src/transports/pipe.ts

@@ -0,0 +1,129 @@
+import { type JsonRpcMessage, McpConnectionClosedError } from "../protocol/jsonrpc.ts";
+import { JsonRpcLineReader } from "./lines.ts";
+import { type McpTransport, TransportEvents } from "./transport.ts";
+
+const DEFAULT_CLOSE_TIMEOUT_MS = 2_000;
+
+/** A running program's stdio, as `@earendil-works/pi-env`'s `RemoteProcess` provides it. */
+export interface McpPipe {
+	write(data: Uint8Array | string): void;
+	closeStdin(): void;
+	/** Kill the program (and what it started). The pipe owns cleaning up what the program started. */
+	close(): void;
+	/** Settles when the program has ended, whatever the reason; it must also settle after `close()`. */
+	readonly exited: Promise<unknown>;
+}
+
+/** Handlers `PipeTransportOptions.open` connects to the program's output. */
+export interface McpPipeHandlers {
+	onStdout(bytes: Uint8Array): void;
+	onStderr(bytes: Uint8Array): void;
+}
+
+export interface PipeTransportOptions {
+	/** Start the program; `onStdout`/`onStderr` receive its output from then on. */
+	open(handlers: McpPipeHandlers): Promise<McpPipe>;
+	maxMessageBytes?: number;
+	maxStderrBytes?: number;
+	onStderr?: (chunk: string) => void;
+	/** After closing stdin, how long the program may take to exit before `close()` kills it. Default 2000. */
+	closeTimeoutMs?: number;
+}
+
+/**
+ * MCP's stdio transport over a program that something else runs, for example on another machine through pi-env.
+ * The transport closes when the program ends; a rejected `exited` that `close()` did not cause is reported as an
+ * error first.
+ */
+export class PipeTransport extends TransportEvents implements McpTransport {
+	readonly options: Readonly<PipeTransportOptions>;
+	private readonly lines: JsonRpcLineReader;
+	private pipe: McpPipe | undefined;
+	private started = false;
+	private closed = false;
+	private ended = false;
+
+	constructor(options: PipeTransportOptions) {
+		super();
+		this.options = Object.freeze({ ...options });
+		this.lines = new JsonRpcLineReader({
+			maxMessageBytes: this.options.maxMessageBytes,
+			maxStderrBytes: this.options.maxStderrBytes,
+			onStderr: this.options.onStderr,
+			onMessage: (message) => this.emitMessage(message),
+			onError: (error) => this.emitError(error),
+		});
+	}
+
+	/** The last `maxStderrBytes` (default 64 KiB) of the program's stderr. */
+	get stderr(): string {
+		return this.lines.stderr;
+	}
+
+	async start(): Promise<void> {
+		if (this.started) throw new Error("MCP pipe transport already started");
+		if (this.closed) throw new McpConnectionClosedError();
+		this.started = true;
+		const pipe = await this.options.open({
+			onStdout: (bytes) => this.lines.readStdout(bytes),
+			onStderr: (bytes) => this.lines.readStderr(bytes),
+		});
+		if (this.closed) {
+			// close() ran while the program was starting and found nothing to shut down. Nobody else awaits
+			// `exited`, and a pipe may reject it after a kill.
+			pipe.exited.catch(() => {});
+			pipe.close();
+			throw new McpConnectionClosedError();
+		}
+		this.pipe = pipe;
+		pipe.exited.then(
+			() => this.exit(),
+			(error: unknown) => this.fail(error),
+		);
+	}
+
+	async send(message: JsonRpcMessage): Promise<void> {
+		if (!this.pipe || this.closed || this.ended) throw new McpConnectionClosedError();
+		this.pipe.write(`${JSON.stringify(message)}\n`);
+	}
+
+	async close(): Promise<void> {
+		if (this.closed) return;
+		this.closed = true;
+		const pipe = this.pipe;
+		if (!pipe) {
+			this.emitClose();
+			return;
+		}
+		if (this.ended) return;
+		// Shutdown per the spec: close stdin and let the program exit, then kill it.
+		pipe.closeStdin();
+		let timer: ReturnType<typeof setTimeout> | undefined;
+		const exitedInTime = await Promise.race([
+			pipe.exited.then(
+				() => true,
+				() => true,
+			),
+			new Promise<false>((resolve) => {
+				timer = setTimeout(() => resolve(false), this.options.closeTimeoutMs ?? DEFAULT_CLOSE_TIMEOUT_MS);
+			}),
+		]);
+		clearTimeout(timer);
+		if (exitedInTime) return;
+		pipe.close();
+		await pipe.exited.catch(() => {});
+	}
+
+	/** The program ended: report an unfinished message, then close. */
+	private exit(): void {
+		this.ended = true;
+		this.lines.end();
+		this.emitClose();
+	}
+
+	/** `exited` rejected: an error unless `close()` caused it, then the same as an exit. */
+	private fail(error: unknown): void {
+		if (!this.closed) this.emitError(error);
+		this.exit();
+	}
+}

packages/mcp/src/transports/stdio.ts

@@ -1,10 +1,10 @@
 import { type ChildProcess, spawn } from "node:child_process";
 import process from "node:process";
 import crossSpawn from "cross-spawn";
-import { type JsonRpcMessage, McpConnectionClosedError, parseJsonRpcMessage } from "../protocol/jsonrpc.ts";
-import { DEFAULT_MAX_MESSAGE_BYTES, type McpTransport, TransportEvents } from "./transport.ts";
+import { type JsonRpcMessage, McpConnectionClosedError } from "../protocol/jsonrpc.ts";
+import { JsonRpcLineReader } from "./lines.ts";
+import { type McpTransport, TransportEvents } from "./transport.ts";
 
-const DEFAULT_MAX_STDERR_BYTES = 64 * 1024;
 const DEFAULT_CLOSE_TIMEOUT_MS = 2_000;
 /** How long a server gets to exit on its own after stdin closes, before it is sent SIGTERM. */
 const STDIN_CLOSE_GRACE_MS = 500;
@@ -69,14 +69,20 @@ export interface StdioTransportOptions {
 export class StdioTransport extends TransportEvents implements McpTransport {
 	readonly options: Readonly<StdioTransportOptions>;
 	private child: ChildProcess | undefined;
-	private stdoutBuffer = Buffer.alloc(0);
-	private stderrBuffer = Buffer.alloc(0);
+	private readonly lines: JsonRpcLineReader;
 	private started = false;
 	private closed = false;
 
 	constructor(options: StdioTransportOptions) {
 		super();
 		this.options = Object.freeze({ ...options, args: options.args ? [...options.args] : undefined });
+		this.lines = new JsonRpcLineReader({
+			maxMessageBytes: this.options.maxMessageBytes,
+			maxStderrBytes: this.options.maxStderrBytes,
+			onStderr: this.options.onStderr,
+			onMessage: (message) => this.emitMessage(message),
+			onError: (error) => this.emitError(error),
+		});
 	}
 
 	get pid(): number | undefined {
@@ -84,7 +90,7 @@ export class StdioTransport extends TransportEvents implements McpTransport {
 	}
 
 	get stderr(): string {
-		return this.stderrBuffer.toString("utf8");
+		return this.lines.stderr;
 	}
 
 	async start(): Promise<void> {
@@ -108,19 +114,16 @@ export class StdioTransport extends TransportEvents implements McpTransport {
 			// Not on "exit": children of the server can outlive it while still holding its pipes.
 			child.once("close", () => liveProcessGroups.delete(pid));
 		}
-		child.stdout?.on("data", (chunk: Buffer | string) => this.handleStdout(chunk));
+		child.stdout?.on("data", (chunk: Buffer) => this.lines.readStdout(chunk));
 		child.stdout?.on("error", (error) => this.emitError(error));
 		child.stdin?.on("error", (error) => {
 			if (!this.closed) this.emitError(error);
 		});
-		child.stderr?.on("data", (chunk: Buffer | string) => this.handleStderr(chunk));
+		child.stderr?.on("data", (chunk: Buffer) => this.lines.readStderr(chunk));
 		child.stderr?.on("error", (error) => this.emitError(error));
 		child.on("close", () => {
 			this.child = undefined;
-			if (this.stdoutBuffer.toString("utf8").trim()) {
-				this.emitError(new Error("MCP stdio server closed with an incomplete JSON-RPC message"));
-			}
-			this.stdoutBuffer = Buffer.alloc(0);
+			this.lines.end();
 			this.emitClose();
 		});
 
@@ -180,40 +183,4 @@ export class StdioTransport extends TransportEvents implements McpTransport {
 			child.stdin?.end();
 		});
 	}
-
-	private handleStdout(chunk: Buffer | string): void {
-		this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)]);
-		const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES;
-		while (true) {
-			const newline = this.stdoutBuffer.indexOf(0x0a);
-			if (newline < 0) {
-				if (this.stdoutBuffer.length > maxMessageBytes) {
-					this.stdoutBuffer = Buffer.alloc(0);
-					this.emitError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
-				}
-				return;
-			}
-			const line = this.stdoutBuffer.subarray(0, newline);
-			this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1);
-			if (line.length > maxMessageBytes) {
-				this.emitError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`));
-				continue;
-			}
-			const text = line.toString("utf8").replace(/\r$/, "");
-			if (!text.trim()) continue;
-			try {
-				this.emitMessage(parseJsonRpcMessage(JSON.parse(text)));
-			} catch (error) {
-				this.emitError(error);
-			}
-		}
-	}
-
-	private handleStderr(chunk: Buffer | string): void {
-		const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-		const maxStderrBytes = this.options.maxStderrBytes ?? DEFAULT_MAX_STDERR_BYTES;
-		this.stderrBuffer = Buffer.concat([this.stderrBuffer, buffer]);
-		if (this.stderrBuffer.length > maxStderrBytes) this.stderrBuffer = this.stderrBuffer.subarray(-maxStderrBytes);
-		this.options.onStderr?.(buffer.toString("utf8"));
-	}
 }

packages/mcp/test/pipe.test.ts

@@ -0,0 +1,265 @@
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import {
+	type JsonRpcMessage,
+	McpClient,
+	McpConnectionClosedError,
+	type McpPipe,
+	type McpPipeHandlers,
+	PipeTransport,
+	type PipeTransportOptions,
+} from "../src/index.ts";
+
+const fixture = fileURLToPath(new URL("./fixtures/stdio-server.mjs", import.meta.url));
+
+interface FakeProgram {
+	transport: PipeTransport;
+	/** Calls of `closeStdin` and `close`, in order. */
+	calls: string[];
+	written: string[];
+	messages: JsonRpcMessage[];
+	errors: Error[];
+	closed: () => boolean;
+	stdout(text: string): void;
+	stderr(text: string): void;
+	exit(code?: number): void;
+	fail(error: Error): void;
+}
+
+/** A `PipeTransport` over an in-memory program that exits only when told to, or when killed. */
+async function startFake(options: Omit<PipeTransportOptions, "open"> = {}): Promise<FakeProgram> {
+	const exited = Promise.withResolvers<number>();
+	const calls: string[] = [];
+	const written: string[] = [];
+	let handlers: McpPipeHandlers | undefined;
+	const pipe: McpPipe = {
+		write: (data) => written.push(String(data)),
+		closeStdin: () => calls.push("closeStdin"),
+		close: () => {
+			calls.push("close");
+			exited.reject(new Error("aborted"));
+		},
+		exited: exited.promise,
+	};
+	const transport = new PipeTransport({
+		...options,
+		open: async (openHandlers) => {
+			handlers = openHandlers;
+			return pipe;
+		},
+	});
+	const messages: JsonRpcMessage[] = [];
+	const errors: Error[] = [];
+	let closed = false;
+	transport.onMessage((message) => messages.push(message));
+	transport.onError((error) => errors.push(error));
+	transport.onClose(() => {
+		closed = true;
+	});
+	await transport.start();
+	const encoder = new TextEncoder();
+	return {
+		transport,
+		calls,
+		written,
+		messages,
+		errors,
+		closed: () => closed,
+		stdout: (text) => handlers?.onStdout(encoder.encode(text)),
+		stderr: (text) => handlers?.onStderr(encoder.encode(text)),
+		exit: (code = 0) => exited.resolve(code),
+		fail: (error) => exited.reject(error),
+	};
+}
+
+/** Let the transport's reactions to `exited` run. */
+const settle = () => new Promise((resolve) => setTimeout(resolve, 0));
+
+const ping = (id: number) => ({ jsonrpc: "2.0", id, method: "ping" }) as const;
+
+afterEach(() => {
+	vi.useRealTimers();
+});
+
+describe("PipeTransport framing", () => {
+	it("joins messages split across chunks and splits several messages in one chunk", async () => {
+		const program = await startFake();
+		const [first, second, third] = [ping(1), ping(2), ping(3)].map((message) => JSON.stringify(message));
+		program.stdout(first.slice(0, 10));
+		program.stdout(`${first.slice(10)}\n${second}`);
+		program.stdout(`\n${third}\n`);
+		expect(program.messages).toEqual([ping(1), ping(2), ping(3)]);
+		expect(program.errors).toEqual([]);
+	});
+
+	it("strips a trailing carriage return and skips blank lines", async () => {
+		const program = await startFake();
+		program.stdout(`\n\r\n   \n${JSON.stringify(ping(1))}\r\n\n`);
+		expect(program.messages).toEqual([ping(1)]);
+		expect(program.errors).toEqual([]);
+	});
+
+	it("reports invalid messages and keeps reading", async () => {
+		const program = await startFake();
+		program.stdout(`not json\n{"jsonrpc":"1.0"}\n${JSON.stringify(ping(1))}\n`);
+		expect(program.messages).toEqual([ping(1)]);
+		expect(program.errors).toHaveLength(2);
+	});
+
+	it("drops messages over maxMessageBytes, complete or not", async () => {
+		const program = await startFake({ maxMessageBytes: 48 });
+		program.stdout(`${JSON.stringify({ jsonrpc: "2.0", method: "x".repeat(48) })}\n${JSON.stringify(ping(1))}\n`);
+		program.stdout("y".repeat(49));
+		program.stdout(`\n${JSON.stringify(ping(2))}\n`);
+		expect(program.messages).toEqual([ping(1), ping(2)]);
+		expect(program.errors.map((error) => error.message)).toEqual([
+			"MCP stdio message exceeds 48 bytes",
+			"MCP stdio message exceeds 48 bytes",
+		]);
+	});
+
+	it("keeps a bounded stderr tail and passes every chunk to onStderr", async () => {
+		const chunks: string[] = [];
+		const program = await startFake({ maxStderrBytes: 8, onStderr: (chunk) => chunks.push(chunk) });
+		program.stderr("starting\n");
+		program.stderr("ready\n");
+		expect(chunks).toEqual(["starting\n", "ready\n"]);
+		expect(program.transport.stderr).toBe("g\nready\n");
+	});
+});
+
+describe("PipeTransport lifecycle", () => {
+	it("writes each message as one line", async () => {
+		const program = await startFake();
+		await program.transport.send(ping(1));
+		expect(program.written).toEqual([`${JSON.stringify(ping(1))}\n`]);
+	});
+
+	it("reports an incomplete message when the program exits, then closes", async () => {
+		const program = await startFake();
+		program.stdout('{"jsonrpc":"2.0"');
+		program.exit(1);
+		await settle();
+		expect(program.errors.map((error) => error.message)).toEqual([
+			"MCP stdio server closed with an incomplete JSON-RPC message",
+		]);
+		expect(program.closed()).toBe(true);
+	});
+
+	it("reports a failed exit it did not cause, then closes", async () => {
+		const program = await startFake();
+		program.fail(new Error("connection lost"));
+		await settle();
+		expect(program.errors.map((error) => error.message)).toEqual(["connection lost"]);
+		expect(program.closed()).toBe(true);
+	});
+
+	it("rejects send after close and after the program ended", async () => {
+		const closedByUs = await startFake();
+		closedByUs.exit();
+		await closedByUs.transport.close();
+		await expect(closedByUs.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError);
+
+		const ended = await startFake();
+		ended.exit();
+		await settle();
+		await expect(ended.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError);
+		expect(ended.written).toEqual([]);
+	});
+
+	it("closes stdin and does not kill a program that exits in time", async () => {
+		const program = await startFake();
+		const closing = program.transport.close();
+		expect(program.calls).toEqual(["closeStdin"]);
+		await expect(program.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError);
+		expect(program.written).toEqual([]);
+		program.exit();
+		await closing;
+		expect(program.calls).toEqual(["closeStdin"]);
+		expect(program.closed()).toBe(true);
+		expect(program.errors).toEqual([]);
+	});
+
+	it("shuts down once when closed twice", async () => {
+		const program = await startFake();
+		const first = program.transport.close();
+		const second = program.transport.close();
+		program.exit();
+		await Promise.all([first, second]);
+		await program.transport.close();
+		expect(program.calls).toEqual(["closeStdin"]);
+		expect(program.closed()).toBe(true);
+	});
+
+	it("kills a program only after closeTimeoutMs", async () => {
+		vi.useFakeTimers();
+		const program = await startFake({ closeTimeoutMs: 1_000 });
+		const closing = program.transport.close();
+		await vi.advanceTimersByTimeAsync(999);
+		expect(program.calls).toEqual(["closeStdin"]);
+		await vi.advanceTimersByTimeAsync(1);
+		await closing;
+		expect(program.calls).toEqual(["closeStdin", "close"]);
+		expect(program.closed()).toBe(true);
+		// The rejection of `exited` that killing caused is not an error.
+		expect(program.errors).toEqual([]);
+	});
+
+	it("rejects start when open rejects", async () => {
+		const transport = new PipeTransport({ open: () => Promise.reject(new Error("not_supported")) });
+		await expect(transport.start()).rejects.toThrow("not_supported");
+	});
+
+	it("kills a program that finished starting after close", async () => {
+		const opened = Promise.withResolvers<McpPipe>();
+		const transport = new PipeTransport({ open: () => opened.promise });
+		const starting = transport.start();
+		await transport.close();
+		const exited = Promise.withResolvers<number>();
+		// Like pi-env's RemoteProcess: killing rejects `exited`, which must not become an unhandled rejection.
+		const close = vi.fn(() => exited.reject(new Error("aborted")));
+		opened.resolve({ write: () => {}, closeStdin: () => {}, close, exited: exited.promise });
+		await expect(starting).rejects.toBeInstanceOf(McpConnectionClosedError);
+		expect(close).toHaveBeenCalledOnce();
+		await settle();
+	});
+});
+
+/** Runs a program with `node:child_process`, standing in for pi-env's `openProcess`. */
+function spawnPipe(argv: readonly string[], handlers: McpPipeHandlers): Promise<McpPipe> {
+	const [command, ...args] = argv;
+	const child = spawn(command, args, { stdio: "pipe" });
+	child.stdout.on("data", handlers.onStdout);
+	child.stderr.on("data", handlers.onStderr);
+	// Writes racing the program's exit fail with EPIPE; `exited` reports the end.
+	child.stdin.on("error", () => {});
+	const exited = new Promise<number | null>((resolve) => child.once("close", resolve));
+	return new Promise((resolve, reject) => {
+		child.once("error", reject);
+		child.once("spawn", () =>
+			resolve({
+				write: (data) => child.stdin.write(data),
+				closeStdin: () => child.stdin.end(),
+				close: () => child.kill("SIGKILL"),
+				exited,
+			}),
+		);
+	});
+}
+
+describe("PipeTransport with a real program", () => {
+	it("runs initialize, tools/list and tools/call", async () => {
+		const transport = new PipeTransport({ open: (handlers) => spawnPipe([process.execPath, fixture], handlers) });
+		const client = new McpClient({ name: "pipe-test", version: "1.0.0" });
+		const result = await client.connect(transport);
+		expect(result.serverInfo).toEqual({ name: "stdio-fixture", version: "1.0.0" });
+		expect(await client.listTools()).toEqual([{ name: "echo", inputSchema: { type: "object" } }]);
+		expect(await client.callTool("echo", { text: "hello" })).toEqual({
+			content: [{ type: "text", text: "hello" }],
+		});
+		expect(transport.stderr).toContain("stdio fixture ready");
+		await client.close();
+		expect(client.connectionState).toBe("closed");
+	});
+});