History

Make executor mcp attach to daemon

Rhys Sullivan committed 5 months ago

Commit aefc53f1 parent 49e16755 files · +478 −34Browse files

Unified · Split

5 files
  1. apps/cli/package.json +1 −0
  2. apps/cli/src/main.ts +160 −34
  3. bun.lock +1 −0
  4. e2e/local/cli-mcp-daemon-attach.test.ts +168 −0
  5. packages/hosts/mcp/src/stdio-integration.test.ts +148 −0
apps/cli/package.json+1−0
@@ -30,6 +30,7 @@ · 29 unmodified lines
3030 "@executor-js/runtime-quickjs": "workspace:*",
3131 "@executor-js/sdk": "workspace:*",
3232 "@jitl/quickjs-wasmfile-release-sync": "catalog:",
33+ "@modelcontextprotocol/sdk": "^1.29.0",
3334 "@sentry/bun": "^10.57.0",
3435 "effect": "catalog:",
3536 "quickjs-emscripten": "catalog:"
apps/cli/src/main.ts+160−34
@@ -64,6 +64,9 @@ import type { PlatformError } from "effect/PlatformError"; · 63 unmodified lines
6464 import * as Effect from "effect/Effect";
6565 import * as Option from "effect/Option";
6666 import * as Cause from "effect/Cause";
67+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
68+import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
69+import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js";
6770
6871 import { ExecutorApi } from "@executor-js/api";
6972 import {
@@ -1122,9 +1125,123 @@ const withStdoutReroutedToStderr = async <A>(body: () => Promise<A>): Promise<A> · 1052 unmodified lines
11221125 }
11231126 };
11241127
1128+const mcpUrlForActiveLocalServer = (
1129+ connection: ExecutorServerConnection,
1130+ elicitationMode: "browser" | "model",
1131+): URL => {
1132+ const url = new URL("/mcp", connection.origin);
1133+ if (elicitationMode === "browser") {
1134+ url.searchParams.set("elicitation_mode", "browser");
1135+ }
1136+ return url;
1137+};
1138+
1139+const closeMcpBridgeTransport = async (close: () => Promise<void>): Promise<void> => {
1140+ await close();
1141+};
1142+
1143+const isAbortDuringMcpBridgeClose = (error: Error): boolean =>
1144+ error.name === "AbortError" || error.message.toLowerCase().includes("aborted");
1145+
1146+const runMcpHttpBridge = async (input: {
1147+ readonly manifest: ExecutorLocalServerManifest;
1148+ readonly elicitationMode: "browser" | "model";
1149+}): Promise<void> => {
1150+ const stdio = new StdioServerTransport();
1151+ const authorization = getExecutorServerAuthorizationHeader(input.manifest.connection);
1152+ const http = new StreamableHTTPClientTransport(
1153+ mcpUrlForActiveLocalServer(input.manifest.connection, input.elicitationMode),
1154+ authorization ? { requestInit: { headers: { Authorization: authorization } } } : undefined,
1155+ );
1156+
1157+ let finished = false;
1158+ let closing = false;
1159+ let closePromise: Promise<void> | null = null;
1160+ let resolveExit: () => void = () => {};
1161+
1162+ const waitForExit = new Promise<void>((resolve) => {
1163+ resolveExit = resolve;
1164+ });
1165+
1166+ const finish = () => {
1167+ if (finished) return;
1168+ finished = true;
1169+ process.off("SIGINT", shutdown);
1170+ process.off("SIGTERM", shutdown);
1171+ process.stdin.off("end", shutdown);
1172+ process.stdin.off("close", shutdown);
1173+ resolveExit();
1174+ };
1175+
1176+ const closeBoth = (): Promise<void> => {
1177+ if (!closePromise) {
1178+ closing = true;
1179+ closePromise = Promise.resolve()
1180+ .then(() =>
1181+ Promise.allSettled([
1182+ closeMcpBridgeTransport(() => stdio.close()),
1183+ closeMcpBridgeTransport(() => http.close()),
1184+ ]),
1185+ )
1186+ .then(() => undefined);
1187+ }
1188+ return closePromise;
1189+ };
1190+
1191+ function shutdown() {
1192+ finish();
1193+ void closeBoth();
1194+ }
1195+
1196+ const reportError = (context: string, cause: unknown) => {
1197+ const error = toError(cause);
1198+ if (closing && isAbortDuringMcpBridgeClose(error)) return;
1199+ console.error(`Executor MCP bridge ${context}: ${error.message}`);
1200+ };
1201+
1202+ const forwardMessage =
1203+ (send: (message: JSONRPCMessage) => Promise<void>, context: string) =>
1204+ (message: JSONRPCMessage) => {
1205+ void send(message).then(undefined, (cause: unknown) => {
1206+ reportError(context, cause);
1207+ shutdown();
1208+ });
1209+ };
1210+
1211+ process.once("SIGINT", shutdown);
1212+ process.once("SIGTERM", shutdown);
1213+ process.stdin.once("end", shutdown);
1214+ process.stdin.once("close", shutdown);
1215+
1216+ stdio.onclose = shutdown;
1217+ http.onclose = shutdown;
1218+ stdio.onerror = (error) => reportError("stdio transport error", error);
1219+ http.onerror = (error) => reportError("daemon transport error", error);
1220+ stdio.onmessage = forwardMessage((message) => http.send(message), "failed to send to daemon");
1221+ http.onmessage = forwardMessage((message) => stdio.send(message), "failed to send to stdio");
1222+
1223+ try {
1224+ await http.start();
1225+ await stdio.start();
1226+ await waitForExit;
1227+ } finally {
1228+ finish();
1229+ await closeBoth();
1230+ }
1231+};
1232+
11251233 const runStdioMcpSession = (input: { readonly elicitationMode: "browser" | "model" }) =>
11261234 Effect.gen(function* () {
1235+ const active = yield* readActiveLocalServerManifest();
1236+ if (active) {
1237+ yield* Effect.promise(() =>
1238+ runMcpHttpBridge({ manifest: active, elicitationMode: input.elicitationMode }),
1239+ );
1240+ return;
1241+ }
1242+
11271243 const startupLock = yield* acquireLocalServerStartLock();
1244+ let activeAfterLock: ExecutorLocalServerManifest | null = null;
11281245 let web: Awaited<
11291246 ReturnType<
11301247 typeof withStdoutReroutedToStderr<{
@@ -1137,44 +1254,53 @@ const runStdioMcpSession = (input: { readonly elicitationMode: "browser" | "mode · 6 unmodified lines
11371254 > | null = null;
11381255
11391256 try {
1140− yield* assertNoOtherActiveLocalServer();
1141− web = yield* Effect.promise(() =>
1142− withStdoutReroutedToStderr(async () => {
1143− const host = "127.0.0.1";
1144− const port = await Effect.runPromise(
1145− chooseDaemonPort({ preferredPort: DEFAULT_PORT, hostname: host }),
1146− );
1147− const baseUrl = `http://localhost:${port}`;
1148− const restoreWebBaseUrl = installDefaultExecutorWebBaseUrl(baseUrl);
1149−
1150− try {
1151− const executor = await getExecutor();
1152− const server = await startServer({
1153− port,
1154− hostname: host,
1155− embeddedWebUI,
1156− });
1157− const serverBaseUrl = `http://localhost:${server.port}`;
1158− return { executor, server, baseUrl: serverBaseUrl, restoreWebBaseUrl };
1159− } catch (cause) {
1160− restoreWebBaseUrl();
1161− throw cause;
1162− }
1163− }),
1164− );
1165− yield* publishLocalServerManifest({
1166− kind: "foreground",
1167− connection: normalizeExecutorServerConnection({
1168− kind: "http",
1169− origin: web.baseUrl,
1170− displayName: "CLI MCP",
1171− auth: { kind: "bearer", token: web.server.authToken },
1172− }),
1173− });
1257+ activeAfterLock = yield* readActiveLocalServerManifest();
1258+ if (!activeAfterLock) {
1259+ web = yield* Effect.promise(() =>
1260+ withStdoutReroutedToStderr(async () => {
1261+ const host = "127.0.0.1";
1262+ const port = await Effect.runPromise(
1263+ chooseDaemonPort({ preferredPort: DEFAULT_PORT, hostname: host }),
1264+ );
1265+ const baseUrl = `http://localhost:${port}`;
1266+ const restoreWebBaseUrl = installDefaultExecutorWebBaseUrl(baseUrl);
1267+
1268+ try {
1269+ const executor = await getExecutor();
1270+ const server = await startServer({
1271+ port,
1272+ hostname: host,
1273+ embeddedWebUI,
1274+ });
1275+ const serverBaseUrl = `http://localhost:${server.port}`;
1276+ return { executor, server, baseUrl: serverBaseUrl, restoreWebBaseUrl };
1277+ } catch (cause) {
1278+ restoreWebBaseUrl();
1279+ throw cause;
1280+ }
1281+ }),
1282+ );
1283+ yield* publishLocalServerManifest({
1284+ kind: "foreground",
1285+ connection: normalizeExecutorServerConnection({
1286+ kind: "http",
1287+ origin: web.baseUrl,
1288+ displayName: "CLI MCP",
1289+ auth: { kind: "bearer", token: web.server.authToken },
1290+ }),
1291+ });
1292+ }
11741293 } finally {
11751294 yield* releaseLocalServerStartLock(startupLock).pipe(Effect.ignore);
11761295 }
11771296
1297+ if (activeAfterLock) {
1298+ yield* Effect.promise(() =>
1299+ runMcpHttpBridge({ manifest: activeAfterLock, elicitationMode: input.elicitationMode }),
1300+ );
1301+ return;
1302+ }
1303+
11781304 if (!web) return yield* Effect.fail(new Error("Failed to start local Executor MCP server."));
11791305
11801306 try {
bun.lock+1−0
@@ -42,6 +42,7 @@ · 41 unmodified lines
4242 "@executor-js/runtime-quickjs": "workspace:*",
4343 "@executor-js/sdk": "workspace:*",
4444 "@jitl/quickjs-wasmfile-release-sync": "catalog:",
45+ "@modelcontextprotocol/sdk": "^1.29.0",
4546 "@sentry/bun": "^10.57.0",
4647 "effect": "catalog:",
4748 "quickjs-emscripten": "catalog:",
e2e/local/cli-mcp-daemon-attach.test.ts+168−0
@@ -0,0 +1,168 @@
1+// Local CLI: MCP clients commonly install `executor mcp` as a stdio command.
2+// When a durable local daemon is already running, that command should attach to
3+// the daemon's HTTP MCP endpoint instead of starting a second local server and
4+// tripping the data-dir singleton guard.
5+import { expect } from "@effect/vitest";
6+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
7+import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
8+import { Effect } from "effect";
9+import { mkdtempSync, rmSync } from "node:fs";
10+import { tmpdir } from "node:os";
11+import { join } from "node:path";
12+import { fileURLToPath } from "node:url";
13+import type { Subprocess } from "bun";
14+
15+import { scenario } from "../src/scenario";
16+
17+const repoRoot = fileURLToPath(new URL("../../", import.meta.url));
18+const testScope = join(repoRoot, "apps/local");
19+const readyTimeoutMs = 60_000;
20+
21+const waitForDaemonReady = (
22+ proc: Subprocess<"ignore", "pipe", "pipe">,
23+): Promise<{ readonly port: number; readonly stderr: () => string }> =>
24+ // oxlint-disable-next-line executor/no-promise-reject -- boundary: local e2e watches a real daemon process
25+ new Promise((resolveReady, rejectReady) => {
26+ let stdoutBuffer = "";
27+ let stderrBuffer = "";
28+ let settled = false;
29+ const decoder = new TextDecoder();
30+ const stdout = proc.stdout.getReader();
31+ const stderr = proc.stderr.getReader();
32+
33+ const deadline = setTimeout(() => {
34+ if (settled) return;
35+ settled = true;
36+ // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: local e2e failure includes captured daemon stderr
37+ rejectReady(new Error(`daemon did not announce ready: ${stderrBuffer}`));
38+ }, readyTimeoutMs);
39+
40+ void (async () => {
41+ while (true) {
42+ const { value, done } = await stderr.read();
43+ if (done) return;
44+ stderrBuffer += decoder.decode(value);
45+ }
46+ })();
47+
48+ void (async () => {
49+ while (true) {
50+ const { value, done } = await stdout.read();
51+ if (done) {
52+ if (!settled) {
53+ settled = true;
54+ clearTimeout(deadline);
55+ // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: local e2e failure includes captured daemon stderr
56+ rejectReady(new Error(`daemon stdout closed before ready: ${stderrBuffer}`));
57+ }
58+ return;
59+ }
60+
61+ stdoutBuffer += decoder.decode(value);
62+ const match = /Daemon ready on http:\/\/(?:\[[^\]]+\]|[^:\s]+):(\d+)/.exec(stdoutBuffer);
63+ if (match) {
64+ settled = true;
65+ clearTimeout(deadline);
66+ resolveReady({ port: Number(match[1]), stderr: () => stderrBuffer });
67+ return;
68+ }
69+ }
70+ })();
71+ });
72+
73+const stopDaemonProcess = async (
74+ proc: Subprocess<"ignore", "pipe", "pipe">,
75+ exitCode: () => number | null,
76+): Promise<void> => {
77+ if (exitCode() !== null) return;
78+ proc.kill("SIGTERM");
79+ await Promise.race([proc.exited, Bun.sleep(3000)]);
80+ if (exitCode() === null) proc.kill("SIGKILL");
81+};
82+
83+const withTempRoot = Effect.acquireRelease(
84+ Effect.sync(() => mkdtempSync(join(tmpdir(), "executor-local-mcp-daemon-attach-"))),
85+ (root) => Effect.sync(() => rmSync(root, { recursive: true, force: true })),
86+);
87+
88+const startForegroundDaemon = (dataDir: string) =>
89+ Effect.acquireRelease(
90+ Effect.gen(function* () {
91+ const proc = Bun.spawn(
92+ [
93+ "bun",
94+ "run",
95+ "dev:cli",
96+ "daemon",
97+ "run",
98+ "--foreground",
99+ "--port",
100+ "0",
101+ "--hostname",
102+ "127.0.0.1",
103+ "--scope",
104+ testScope,
105+ ],
106+ {
107+ cwd: repoRoot,
108+ env: { ...process.env, EXECUTOR_DATA_DIR: dataDir },
109+ stdin: "ignore",
110+ stdout: "pipe",
111+ stderr: "pipe",
112+ },
113+ );
114+ let exitCode: number | null = null;
115+ void proc.exited.then((code) => {
116+ exitCode = code;
117+ });
118+
119+ const ready = yield* Effect.promise(() => waitForDaemonReady(proc)).pipe(
120+ Effect.tapError(() => Effect.promise(() => stopDaemonProcess(proc, () => exitCode))),
121+ );
122+ return { proc, port: ready.port, stderr: ready.stderr, exitCode: () => exitCode };
123+ }),
124+ ({ proc, exitCode }) => Effect.promise(() => stopDaemonProcess(proc, exitCode)),
125+ );
126+
127+scenario(
128+ "Local CLI MCP · stdio attaches to the active daemon instead of starting a competing server",
129+ { timeout: 120_000 },
130+ Effect.gen(function* () {
131+ const root = yield* withTempRoot;
132+ const dataDir = join(root, "data");
133+ const daemon = yield* startForegroundDaemon(dataDir);
134+
135+ const transport = new StdioClientTransport({
136+ command: "bun",
137+ args: ["run", "dev:cli", "mcp", "--scope", testScope],
138+ cwd: repoRoot,
139+ env: { ...process.env, EXECUTOR_DATA_DIR: dataDir },
140+ stderr: "pipe",
141+ });
142+ const client = new Client({ name: "e2e-local-mcp-daemon-attach", version: "1.0.0" });
143+
144+ yield* Effect.acquireRelease(
145+ Effect.promise(() => client.connect(transport)),
146+ () => Effect.promise(() => transport.close()),
147+ );
148+
149+ const { tools } = yield* Effect.promise(() => client.listTools());
150+ expect(
151+ tools.map((tool) => tool.name),
152+ "daemon-backed stdio MCP lists tools",
153+ ).toContain("execute");
154+
155+ const result = yield* Effect.promise(() =>
156+ client.callTool({
157+ name: "execute",
158+ arguments: { code: "return 2 + 2" },
159+ }),
160+ );
161+
162+ const text = (result.content as Array<{ type: string; text: string }>)[0]?.text;
163+ expect(text, "execute ran through the daemon-backed MCP bridge").toContain("4");
164+ expect(result.isError, "execute should not report a tool error").toBeFalsy();
165+ expect(daemon.exitCode(), daemon.stderr()).toBeNull();
166+ expect(daemon.port, "daemon announced an OS-assigned port").toBeGreaterThan(0);
167+ }).pipe(Effect.scoped),
168+);
packages/hosts/mcp/src/stdio-integration.test.ts+148−0
@@ -2,13 +2,122 @@ import { describe, expect, it } from "@effect/vitest"; · 1 unmodified line
22 import { Client } from "@modelcontextprotocol/sdk/client/index.js";
33 import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
44 import { Effect } from "effect";
5+import { spawn, type ChildProcessByStdio } from "node:child_process";
56 import { mkdtempSync } from "node:fs";
67 import { tmpdir } from "node:os";
78 import { join, resolve } from "node:path";
9+import type { Readable } from "node:stream";
810
911 const repoRoot = resolve(import.meta.dirname, "../../../..");
1012 const cliEntry = resolve(repoRoot, "apps/cli/src/main.ts");
1113 const testScope = resolve(repoRoot, "apps/local");
14+const readyTimeoutMs = 30_000;
15+type DaemonProcess = ChildProcessByStdio<null, Readable, Readable>;
16+
17+const waitForDaemonReady = (
18+ proc: DaemonProcess,
19+): Promise<{ readonly port: number; readonly stderr: () => string }> =>
20+ // oxlint-disable-next-line executor/no-promise-reject -- boundary: integration test watches a real daemon process
21+ new Promise((resolveReady, rejectReady) => {
22+ let stdoutBuffer = "";
23+ let stderrBuffer = "";
24+ let settled = false;
25+
26+ const cleanup = () => {
27+ clearTimeout(deadline);
28+ proc.stdout.off("data", onStdout);
29+ proc.stderr.off("data", onStderr);
30+ proc.off("error", onError);
31+ proc.off("close", onClose);
32+ };
33+
34+ const fail = (cause: Error) => {
35+ if (settled) return;
36+ settled = true;
37+ cleanup();
38+ // oxlint-disable-next-line executor/no-promise-reject -- boundary: adapts child_process callbacks into the Promise waited by the Effect test helper
39+ rejectReady(cause);
40+ };
41+
42+ const deadline = setTimeout(() => {
43+ // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: integration test failure includes captured daemon stderr
44+ fail(new Error(`daemon did not announce ready: ${stderrBuffer}`));
45+ }, readyTimeoutMs);
46+
47+ const onStderr = (chunk: Buffer | string) => {
48+ stderrBuffer += chunk.toString();
49+ };
50+
51+ const onStdout = (chunk: Buffer | string) => {
52+ stdoutBuffer += chunk.toString();
53+ const match = /Daemon ready on http:\/\/(?:\[[^\]]+\]|[^:\s]+):(\d+)/.exec(stdoutBuffer);
54+ if (match) {
55+ settled = true;
56+ cleanup();
57+ resolveReady({ port: Number(match[1]), stderr: () => stderrBuffer });
58+ }
59+ };
60+
61+ const onError = (error: Error) => fail(error);
62+
63+ const onClose = () => {
64+ // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: integration test failure includes captured daemon stderr
65+ fail(new Error(`daemon stdout closed before ready: ${stderrBuffer}`));
66+ };
67+
68+ proc.stdout.on("data", onStdout);
69+ proc.stderr.on("data", onStderr);
70+ proc.once("error", onError);
71+ proc.once("close", onClose);
72+ });
73+
74+const stopDaemonProcess = async (
75+ proc: DaemonProcess,
76+ exitCode: () => number | null,
77+): Promise<void> => {
78+ if (exitCode() !== null) return;
79+ proc.kill("SIGTERM");
80+ await Promise.race([
81+ new Promise<void>((resolve) => {
82+ proc.once("close", () => resolve());
83+ }),
84+ new Promise<void>((resolve) => setTimeout(resolve, 3000)),
85+ ]);
86+ if (exitCode() === null) proc.kill("SIGKILL");
87+};
88+
89+const startForegroundDaemon = (dataDir: string) =>
90+ Effect.acquireRelease(
91+ Effect.gen(function* () {
92+ const proc = spawn(
93+ "bun",
94+ [
95+ "run",
96+ cliEntry,
97+ "daemon",
98+ "run",
99+ "--foreground",
100+ "--port",
101+ "0",
102+ "--hostname",
103+ "127.0.0.1",
104+ "--scope",
105+ testScope,
106+ ],
107+ { env: { ...process.env, EXECUTOR_DATA_DIR: dataDir }, stdio: ["ignore", "pipe", "pipe"] },
108+ );
109+ let exitCode: number | null = null;
110+ proc.once("close", (code) => {
111+ exitCode = code;
112+ });
113+
114+ const ready = yield* Effect.promise(() => waitForDaemonReady(proc)).pipe(
115+ Effect.tapError(() => Effect.promise(() => stopDaemonProcess(proc, () => exitCode))),
116+ );
117+ return { proc, port: ready.port, stderr: ready.stderr, exitCode: () => exitCode };
118+ }),
119+ ({ proc, exitCode }) => Effect.promise(() => stopDaemonProcess(proc, exitCode)),
120+ );
12121
13122 describe("MCP stdio integration", () => {
14123 it.effect(
@@ -48,4 +157,43 @@ describe("MCP stdio integration", () => { · 33 unmodified lines
48157 }).pipe(Effect.scoped),
49158 { timeout: 30_000 },
50159 );
160+
161+ it.effect(
162+ "attaches stdio MCP to an active local daemon",
163+ () =>
164+ Effect.gen(function* () {
165+ const dataDir = mkdtempSync(join(tmpdir(), "executor-mcp-daemon-test-"));
166+ const daemon = yield* startForegroundDaemon(dataDir);
167+
168+ const transport = new StdioClientTransport({
169+ command: "bun",
170+ args: ["run", cliEntry, "mcp", "--scope", testScope],
171+ env: { ...process.env, EXECUTOR_DATA_DIR: dataDir },
172+ });
173+
174+ const client = new Client({ name: "test-client", version: "1.0.0" }, { capabilities: {} });
175+
176+ yield* Effect.acquireRelease(
177+ Effect.promise(() => client.connect(transport)),
178+ () => Effect.promise(() => transport.close()),
179+ );
180+
181+ const { tools } = yield* Effect.promise(() => client.listTools());
182+ expect(tools.map((t) => t.name)).toContain("execute");
183+
184+ const result = yield* Effect.promise(() =>
185+ client.callTool({
186+ name: "execute",
187+ arguments: { code: "return 2+2" },
188+ }),
189+ );
190+
191+ const text = (result.content as Array<{ type: string; text: string }>)[0]?.text;
192+ expect(text).toContain("4");
193+ expect(result.isError).toBeFalsy();
194+ expect(daemon.exitCode(), daemon.stderr()).toBeNull();
195+ expect(daemon.port).toBeGreaterThan(0);
196+ }).pipe(Effect.scoped),
197+ { timeout: 45_000 },
198+ );
51199 });

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t