import { describe, expect, it } from "@effect/vitest";import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";import { ErrorCode, LATEST_PROTOCOL_VERSION, type JSONRPCMessage,} from "@modelcontextprotocol/sdk/types.js";import { Effect, Option, Schema } from "effect";import { spawnSync } from "node:child_process";import { mkdtempSync, readFileSync, rmSync } from "node:fs";import { tmpdir } from "node:os";import { join, resolve } from "node:path"; const repoRoot = resolve(import.meta.dirname, "../../../..");const cliEntry = resolve(repoRoot, "apps/cli/src/main.ts");const testScope = resolve(repoRoot, "apps/local"); const decodeServerManifest = Schema.decodeUnknownOption( Schema.fromJsonString(Schema.Struct({ pid: Schema.optional(Schema.Number) })),); const manifestPid = (dataDir: string): number | undefined => Effect.runSync( Effect.try({ try: () => readFileSync(join(dataDir, "server-control", "server.json"), "utf8"), catch: () => undefined, }).pipe( Effect.map((text) => { const manifest = decodeServerManifest(text); return Option.isSome(manifest) ? manifest.value.pid : undefined; }), Effect.orElseSucceed(() => undefined), ), ); /** * Own the daemon this test bridges to, on its own port. * * `executor mcp` otherwise elects a daemon on the shared default port, and will * adopt one that is already listening there — including a daemon belonging to a * different `EXECUTOR_DATA_DIR`, whose bearer token this test does not have. A * dedicated port keeps the test hermetic and parallel-safe. `daemon run` falls * back to a free port if this one is taken, and writes the port it chose into * the manifest that `executor mcp` reads, so the exact number is not load-bearing. */const startDaemon = (dataDir: string): void => { const port = 20_000 + Math.floor(Math.random() * 20_000); const result = spawnSync( "bun", ["run", cliEntry, "daemon", "run", "--port", String(port), "--hostname", "127.0.0.1"], { env: { ...process.env, EXECUTOR_DATA_DIR: dataDir, EXECUTOR_SCOPE_DIR: testScope } }, ); expect(result.status, `daemon run failed: ${result.stderr?.toString() ?? ""}`).toBe(0);}; /** Stop the daemon started above; the manifest carries its pid. */const stopDaemon = (dataDir: string): Effect.Effect<void> => Effect.sync(() => manifestPid(dataDir)).pipe( Effect.flatMap((pid) => pid ? Effect.try({ try: () => process.kill(pid, "SIGTERM"), catch: () => undefined }).pipe( Effect.ignore, ) : Effect.void, ), Effect.ignore, ); const withDaemon = Effect.acquireRelease( Effect.sync(() => { const dataDir = mkdtempSync(join(tmpdir(), "executor-mcp-discover-test-")); startDaemon(dataDir); return dataDir; }), (dataDir) => stopDaemon(dataDir).pipe( Effect.ensuring(Effect.sync(() => rmSync(dataDir, { recursive: true, force: true }))), ),); const messageQueue = (transport: StdioClientTransport) => { const messages: Array<JSONRPCMessage> = []; const waiters: Array<(message: JSONRPCMessage) => void> = []; transport.onmessage = (message) => { const waiter = waiters.shift(); if (waiter) { waiter(message); } else { messages.push(message); } }; return { next: (): Promise<JSONRPCMessage> => { const message = messages.shift(); return message ? Promise.resolve(message) : new Promise((resolve) => waiters.push(resolve)); }, };}; describe("MCP stdio integration", () => { // The regression: a client that opens with an unsupported probe used to get // the transport's fatal `-32000 Server not initialized` on an HTTP 400, which // tore the bridge down before it could fall back to `initialize`. The whole // handshake must survive the probe on ONE connection, through to a tool call. it.effect( "unsupported discovery keeps the connection open for initialization and tool calls", () => Effect.gen(function* () { const dataDir = yield* withDaemon; const transport = new StdioClientTransport({ command: "bun", args: ["run", cliEntry, "mcp", "--scope", testScope], env: { ...process.env, EXECUTOR_DATA_DIR: dataDir }, }); const responses = messageQueue(transport); yield* Effect.acquireRelease( Effect.promise(() => transport.start()), () => Effect.promise(() => transport.close()), ); yield* Effect.promise(() => transport.send({ jsonrpc: "2.0", id: 1, method: "server/discover", params: { _meta: { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientInfo": { name: "discovery-test-client", version: "1.0.0", }, "io.modelcontextprotocol/clientCapabilities": {}, }, }, }), ); expect(yield* Effect.promise(() => responses.next())).toEqual({ jsonrpc: "2.0", id: 1, error: { code: ErrorCode.MethodNotFound, message: "Method not found", }, }); yield* Effect.promise(() => transport.send({ jsonrpc: "2.0", id: 2, method: "initialize", params: { protocolVersion: LATEST_PROTOCOL_VERSION, capabilities: {}, clientInfo: { name: "discovery-test-client", version: "1.0.0", }, }, }), ); const initialize = yield* Effect.promise(() => responses.next()); expect(initialize).toHaveProperty("result.protocolVersion"); yield* Effect.promise(() => transport.send({ jsonrpc: "2.0", method: "notifications/initialized", }), ); yield* Effect.promise(() => transport.send({ jsonrpc: "2.0", id: 3, method: "tools/list", params: {}, }), ); const listed = yield* Effect.promise(() => responses.next()); expect(listed).toHaveProperty( "result.tools", expect.arrayContaining([expect.objectContaining({ name: "execute" })]), ); yield* Effect.promise(() => transport.send({ jsonrpc: "2.0", id: 4, method: "tools/call", params: { name: "execute", arguments: { code: "return 2+2" }, }, }), ); const called = yield* Effect.promise(() => responses.next()); expect(called).toHaveProperty( "result.content", expect.arrayContaining([expect.objectContaining({ text: expect.stringContaining("4") })]), ); }).pipe(Effect.scoped), { timeout: 120_000 }, );});
import { describe, expect, it } from "@effect/vitest";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import {
ErrorCode,
LATEST_PROTOCOL_VERSION,
type JSONRPCMessage,
} from "@modelcontextprotocol/sdk/types.js";
import { Effect, Option, Schema } from "effect";
import { spawnSync } from "node:child_process";
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
const repoRoot = resolve(import.meta.dirname, "../../../..");
const cliEntry = resolve(repoRoot, "apps/cli/src/main.ts");
const testScope = resolve(repoRoot, "apps/local");
const decodeServerManifest = Schema.decodeUnknownOption(
Schema.fromJsonString(Schema.Struct({ pid: Schema.optional(Schema.Number) })),
);
const manifestPid = (dataDir: string): number | undefined =>
Effect.runSync(
Effect.try({
try: () => readFileSync(join(dataDir, "server-control", "server.json"), "utf8"),
catch: () => undefined,
}).pipe(
Effect.map((text) => {
const manifest = decodeServerManifest(text);
return Option.isSome(manifest) ? manifest.value.pid : undefined;
}),
Effect.orElseSucceed(() => undefined),
),
);
/**
* Own the daemon this test bridges to, on its own port.
*
* `executor mcp` otherwise elects a daemon on the shared default port, and will
* adopt one that is already listening there — including a daemon belonging to a
* different `EXECUTOR_DATA_DIR`, whose bearer token this test does not have. A
* dedicated port keeps the test hermetic and parallel-safe. `daemon run` falls
* back to a free port if this one is taken, and writes the port it chose into
* the manifest that `executor mcp` reads, so the exact number is not load-bearing.
*/
const startDaemon = (dataDir: string): void => {
const port = 20_000 + Math.floor(Math.random() * 20_000);
const result = spawnSync(
"bun",
["run", cliEntry, "daemon", "run", "--port", String(port), "--hostname", "127.0.0.1"],
{ env: { ...process.env, EXECUTOR_DATA_DIR: dataDir, EXECUTOR_SCOPE_DIR: testScope } },
);
expect(result.status, `daemon run failed: ${result.stderr?.toString() ?? ""}`).toBe(0);
};
/** Stop the daemon started above; the manifest carries its pid. */
const stopDaemon = (dataDir: string): Effect.Effect<void> =>
Effect.sync(() => manifestPid(dataDir)).pipe(
Effect.flatMap((pid) =>
pid
? Effect.try({ try: () => process.kill(pid, "SIGTERM"), catch: () => undefined }).pipe(
Effect.ignore,
)
: Effect.void,
),
Effect.ignore,
);
const withDaemon = Effect.acquireRelease(
Effect.sync(() => {
const dataDir = mkdtempSync(join(tmpdir(), "executor-mcp-discover-test-"));
startDaemon(dataDir);
return dataDir;
}),
(dataDir) =>
stopDaemon(dataDir).pipe(
Effect.ensuring(Effect.sync(() => rmSync(dataDir, { recursive: true, force: true }))),
),
);
const messageQueue = (transport: StdioClientTransport) => {
const messages: Array<JSONRPCMessage> = [];
const waiters: Array<(message: JSONRPCMessage) => void> = [];
transport.onmessage = (message) => {
const waiter = waiters.shift();
if (waiter) {
waiter(message);
} else {
messages.push(message);
}
};
return {
next: (): Promise<JSONRPCMessage> => {
const message = messages.shift();
return message ? Promise.resolve(message) : new Promise((resolve) => waiters.push(resolve));
},
};
};
describe("MCP stdio integration", () => {
// The regression: a client that opens with an unsupported probe used to get
// the transport's fatal `-32000 Server not initialized` on an HTTP 400, which
// tore the bridge down before it could fall back to `initialize`. The whole
// handshake must survive the probe on ONE connection, through to a tool call.
it.effect(
"unsupported discovery keeps the connection open for initialization and tool calls",
() =>
Effect.gen(function* () {
const dataDir = yield* withDaemon;
const transport = new StdioClientTransport({
command: "bun",
args: ["run", cliEntry, "mcp", "--scope", testScope],
env: { ...process.env, EXECUTOR_DATA_DIR: dataDir },
});
const responses = messageQueue(transport);
yield* Effect.acquireRelease(
Effect.promise(() => transport.start()),
() => Effect.promise(() => transport.close()),
);
yield* Effect.promise(() =>
transport.send({
jsonrpc: "2.0",
id: 1,
method: "server/discover",
params: {
_meta: {
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {
name: "discovery-test-client",
version: "1.0.0",
},
"io.modelcontextprotocol/clientCapabilities": {},
},
},
}),
);
expect(yield* Effect.promise(() => responses.next())).toEqual({
jsonrpc: "2.0",
id: 1,
error: {
code: ErrorCode.MethodNotFound,
message: "Method not found",
},
});
yield* Effect.promise(() =>
transport.send({
jsonrpc: "2.0",
id: 2,
method: "initialize",
params: {
protocolVersion: LATEST_PROTOCOL_VERSION,
capabilities: {},
clientInfo: {
name: "discovery-test-client",
version: "1.0.0",
},
},
}),
);
const initialize = yield* Effect.promise(() => responses.next());
expect(initialize).toHaveProperty("result.protocolVersion");
yield* Effect.promise(() =>
transport.send({
jsonrpc: "2.0",
method: "notifications/initialized",
}),
);
yield* Effect.promise(() =>
transport.send({
jsonrpc: "2.0",
id: 3,
method: "tools/list",
params: {},
}),
);
const listed = yield* Effect.promise(() => responses.next());
expect(listed).toHaveProperty(
"result.tools",
expect.arrayContaining([expect.objectContaining({ name: "execute" })]),
);
yield* Effect.promise(() =>
transport.send({
jsonrpc: "2.0",
id: 4,
method: "tools/call",
params: {
name: "execute",
arguments: { code: "return 2+2" },
},
}),
);
const called = yield* Effect.promise(() => responses.next());
expect(called).toHaveProperty(
"result.content",
expect.arrayContaining([expect.objectContaining({ text: expect.stringContaining("4") })]),
);
}).pipe(Effect.scoped),
{ timeout: 120_000 },
);
});