diff --git a/packages/harness/package.json b/packages/harness/package.json index b897a456..5ecde2bf 100644 --- a/packages/harness/package.json +++ b/packages/harness/package.json @@ -82,6 +82,7 @@ "@effect/platform-node": "4.0.5", "@effect/platform-node-shared": "4.0.5", "@effect/sql-sqlite-node": "4.0.5", + "@modelcontextprotocol/client": "2.3.1", "@npmcli/arborist": "^10.0.2", "@npmcli/config": "^11.0.1", "@platformatic/vfs": "^0.4.0", @@ -98,6 +99,7 @@ }, "devDependencies": { "@daytona/sdk": "^0.211.2", + "@modelcontextprotocol/server": "2.3.1", "@types/node": "catalog:", "@types/npm-package-arg": "^6.1.4", "@vercel/sandbox": "^3.3.0", diff --git a/packages/harness/src/mcp/client.ts b/packages/harness/src/mcp/client.ts new file mode 100644 index 00000000..abd08ea1 --- /dev/null +++ b/packages/harness/src/mcp/client.ts @@ -0,0 +1,484 @@ +import { pathToFileURL } from "node:url"; +import { + type CallToolResult, + Client, + type Implementation, + type OAuthClientProvider, + ProtocolError, + ProtocolErrorCode, + type ListResourcesResult, + type ListResourceTemplatesResult, + type LoggingMessageNotificationParams, + type ReadResourceResult, + SdkError, + SdkErrorCode, + SdkHttpError, + type ServerCapabilities, + StreamableHTTPClientTransport, + type Tool, + type Transport, + UnauthorizedError, +} from "@modelcontextprotocol/client"; +import { Cause, Clock, Data, Duration, Effect, Exit, FiberSet, Schedule } from "effect"; +import { hostPath } from "../host.ts"; +import { ConnectError, NeedsAuthError, RequestError, SessionExpiredError } from "./error.ts"; +import { McpStdio } from "./stdio.ts"; + +const DEFAULT_REQUEST_TIMEOUT = Duration.seconds(60); +const TERMINATE_TIMEOUT = Duration.seconds(1); + +export interface Stdio { + readonly _tag: "Stdio"; + readonly command: string; + readonly args: ReadonlyArray; + readonly env: Record; + readonly cwd: string; +} + +export interface Http { + readonly _tag: "Http"; + readonly url: URL; + readonly headers: Record; +} + +export type Endpoint = Stdio | Http; + +export interface Progress { + readonly progress: number; + readonly total?: number | undefined; + readonly message?: string | undefined; +} + +type Failed = RequestError | SessionExpiredError | NeedsAuthError; + +/** Handle over a connected MCP server that keeps the SDK `Client` out of the rest of the harness. */ +export interface Connection { + readonly serverInfo: Implementation | undefined; + readonly capabilities: ServerCapabilities | undefined; + readonly instructions: string | undefined; + /** Every tool, following pagination. Empty when the server has no tools capability. */ + readonly tools: () => Effect.Effect; + /** One page of resources, starting at `cursor`. Empty when the server has no resources capability. */ + readonly resources: (cursor?: string) => Effect.Effect; + /** One page of resource templates; empty when the server does not implement the method. */ + readonly resourceTemplates: (cursor?: string) => Effect.Effect; + readonly readResource: (uri: string) => Effect.Effect; + readonly callTool: (input: { + readonly name: string; + readonly args?: Record; + readonly onProgress?: (progress: Progress) => void; + }) => Effect.Effect; + /** Each handler has one consumer (the owner of this connection); setting it again replaces it. */ + readonly onToolsChanged: (handler: Effect.Effect) => void; + readonly onResourcesChanged: (handler: Effect.Effect) => void; + readonly onSessionExpired: (handler: Effect.Effect) => void; + /** + * The connection dropped (for example the server process exited). Runs right away when it already + * dropped before the handler was set. Never runs for a close caused by closing the owning scope. + */ + readonly onClose: (handler: (reason: string) => Effect.Effect) => void; +} + +export interface ConnectInput { + readonly server: string; + readonly endpoint: Endpoint; + /** Directories sent as roots; only when the session's sandbox is local, where these paths exist for the server. */ + readonly roots?: ReadonlyArray; + /** HTTP only. */ + readonly authProvider?: OAuthClientProvider; + /** `startup` bounds connecting and initializing, and defaults to `request`. */ + readonly timeout?: { readonly startup?: Duration.Input; readonly request?: Duration.Input }; + readonly clientInfo?: Implementation; + /** The server's `notifications/message` log entries, from the handshake on: startup failures log too. */ + readonly onLog?: (entry: LoggingMessageNotificationParams) => Effect.Effect; +} + +/** A raw SDK or transport failure, before it is mapped to one of the public errors. */ +class Failure extends Data.TaggedError("Mcp.Failure")<{ readonly error: unknown }> {} +const fail = (error: unknown) => new Failure({ error }); + +// Node's fetch reports network failures as `TypeError("fetch failed")` with the system code on `cause`. +const systemCode = (error: Error): string | undefined => { + const code = + (error as { readonly code?: unknown }).code ?? (error.cause as { readonly code?: unknown } | undefined)?.code; + return typeof code === "string" ? code : undefined; +}; + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +// The SDK reports a result that fails its schema as `Invalid result for : `, the issues +// being the validator's multi-line JSON list. One `path: message` per issue reads better. +const condense = (message: string) => { + const match = /^(Invalid result for [^:]+): (\[[\s\S]*\])$/.exec(message); + if (!match) return message; + try { + const issues: unknown = JSON.parse(match[2]!); + if (!Array.isArray(issues) || issues.length === 0) return message; + const parts = issues.map((issue) => { + if (!isRecord(issue)) return String(issue); + const path = Array.isArray(issue.path) && issue.path.length > 0 ? `${issue.path.join(".")}: ` : ""; + return `${path}${typeof issue.message === "string" ? issue.message : "invalid"}`; + }); + return `${match[1]}: ${parts.join("; ")}`; + } catch { + return message; + } +}; +const isInvalidResult = (error: unknown) => error instanceof SdkError && error.code === SdkErrorCode.InvalidResult; + +// HTTP statuses and network error codes live on error properties and are lost once flattened to a message. +const describe = (error: unknown) => { + if (!(error instanceof Error)) return String(error); + if (isInvalidResult(error)) return condense(error.message); + const detail = + error instanceof SdkHttpError + ? `HTTP ${error.status}` + : !(error instanceof SdkError) + ? systemCode(error) + : undefined; + return detail && !error.message.includes(detail) ? `${error.message} (${detail})` : error.message; +}; + +// Overloaded or restarting servers usually recover within a second. 501 means the server will never +// support the request. +const isTransientHttp = (error: unknown) => + error instanceof SdkHttpError && + (error.status === 408 || error.status === 429 || (error.status >= 500 && error.status !== 501)); +// Only a failure that carries a system code is a network one; any other `TypeError` (an invalid header +// value, say) would fail again, and its message may hold the value. +const isTransient = (error: unknown) => + isTransientHttp(error) || (error instanceof TypeError && systemCode(error) !== undefined); + +/** Connects one MCP server; closing the calling scope tears down the transport and any spawned process. */ +export const connect = Effect.fn("McpClient.connect")(function* (input: ConnectInput) { + const { server, endpoint } = input; + const requestTimeout = Duration.toMillis(input.timeout?.request ?? DEFAULT_REQUEST_TIMEOUT); + const startupTimeout = + input.timeout?.startup === undefined ? requestTimeout : Duration.toMillis(input.timeout.startup); + const runFork = yield* FiberSet.makeRuntime(); + + // Handlers the owner registers after connect; notifications can arrive before that. + const handlers = { + tools: undefined as Effect.Effect | undefined, + resources: undefined as Effect.Effect | undefined, + expired: Effect.void as Effect.Effect, + }; + // A list that changed before its handler was set, during the first listing say, is reported once + // the handler is set: the owner may be holding the list from before the change. + const missed = { tools: false, resources: false }; + const changed = (list: "tools" | "resources") => { + const handler = handlers[list]; + if (handler === undefined) missed[list] = true; + else runFork(handler); + }; + const listen = (list: "tools" | "resources", handler: Effect.Effect) => { + handlers[list] = handler; + if (!missed[list]) return; + missed[list] = false; + runFork(handler); + }; + // Every connect attempt builds its own client; only the one that connected reports a close. + const closed: { + shuttingDown: boolean; + active?: Client | undefined; + reason?: string | undefined; + handler?: ((reason: string) => Effect.Effect) | undefined; + } = { shuttingDown: false }; + const early = new Map(); + const report = (reason: string) => { + if (closed.reason !== undefined) return; + closed.reason = reason; + if (closed.handler) runFork(closed.handler(reason)); + }; + + // The SDK fails pending requests with a bare "Connection closed"; the transport error before it says why. + let lastError: string | undefined; + const explain = (error: unknown) => + error instanceof SdkError && error.code === SdkErrorCode.ConnectionClosed && lastError + ? `${describe(error)}: ${lastError}` + : describe(error); + // A server that wants (new) credentials. Without an auth provider a 401 is a configuration error. + const needsAuth = (error: unknown) => + error instanceof UnauthorizedError || + (input.authProvider !== undefined && error instanceof SdkHttpError && error.status === 401); + + // Two more attempts, 250 ms and 1 s later, while the failure is transient and the deadline has not passed. + const retrying = (what: string, effect: Effect.Effect, deadline = Infinity) => + Effect.retry( + effect, + Schedule.max([Schedule.exponential("250 millis", 4), Schedule.recurs(2)]).pipe( + Schedule.setInputType(), + Schedule.while(({ input, now, duration }) => + !isTransient(input.error) || now + Duration.toMillis(duration) >= deadline + ? Effect.succeed(false) + : Effect.logWarning(`retrying ${what}`, { server, error: describe(input.error) }).pipe( + Effect.as(true), + ), + ), + ), + ); + // Listing and reading are read-only: one more attempt, 250 ms later, after a transient HTTP error. + const retryingRead = (what: string, effect: Effect.Effect) => + Effect.retry( + effect, + Schedule.max([Schedule.spaced("250 millis"), Schedule.recurs(1)]).pipe( + Schedule.setInputType(), + Schedule.while(({ input }) => + isTransientHttp(input.error) + ? Effect.logWarning(`retrying ${what}`, { server, error: describe(input.error) }).pipe( + Effect.as(true), + ) + : Effect.succeed(false), + ), + ), + ); + + const http = (endpoint: Http, sessionId?: string) => + new StreamableHTTPClientTransport(endpoint.url, { + requestInit: { headers: endpoint.headers }, + ...(input.authProvider === undefined ? {} : { authProvider: input.authProvider }), + ...(sessionId === undefined ? {} : { sessionId }), + }); + // Ends the HTTP session a transport holds, bounded: the server expires it anyway. + const terminate = (transport: StreamableHTTPClientTransport) => + Effect.tryPromise({ try: () => transport.terminateSession(), catch: fail }).pipe( + Effect.timeoutOrElse({ + duration: TERMINATE_TIMEOUT, + orElse: () => Effect.fail(fail(`timed out after ${Duration.format(TERMINATE_TIMEOUT)}`)), + }), + Effect.tapError((failure) => + Effect.logWarning("failed to terminate mcp session", { server, error: describe(failure.error) }), + ), + Effect.ignore, + ); + + const initialize = Effect.fnUntraced(function* (transport: Transport, timeout: number) { + const client = new Client(input.clientInfo ?? { name: "codework", version: "unknown" }, { + capabilities: input.roots ? { roots: {} } : {}, + }); + // Background work such as the standalone SSE stream reports failures only here; aborts come from close. + client.onerror = (error) => { + if (error.name === "AbortError") return; + lastError = describe(error); + runFork(Effect.logWarning("mcp transport error", { server, error: lastError })); + }; + client.onclose = () => { + if (closed.shuttingDown) return; + const reason = lastError ? `connection closed: ${lastError}` : "connection closed"; + // Before connect returns, a close belongs to an attempt that may still fail and be retried. + if (closed.active === client) report(reason); + else if (!early.has(client)) early.set(client, reason); + }; + // Handled whether or not the server advertises `listChanged`, since some servers send them anyway. + client.setNotificationHandler("notifications/tools/list_changed", () => changed("tools")); + client.setNotificationHandler("notifications/resources/list_changed", () => changed("resources")); + client.setNotificationHandler("notifications/message", (notification) => { + if (input.onLog) runFork(input.onLog(notification.params)); + }); + const roots = input.roots; + if (roots) { + client.setRequestHandler("roots/list", () => ({ + roots: roots.map((directory) => ({ + uri: pathToFileURL(directory).href, + name: hostPath.basename(directory), + })), + })); + } + yield* Effect.tryPromise({ + try: (signal) => client.connect(transport, { timeout, signal }), + catch: fail, + }).pipe( + // The SDK times only the initialize request, not the initialized notification sent after it. + Effect.timeoutOrElse({ + duration: Duration.millis(timeout), + orElse: () => Effect.fail(fail(`mcp connect timed out after ${Duration.format(Duration.millis(timeout))}`)), + }), + Effect.onError(() => + Effect.gen(function* () { + // A failed attempt may already hold a session (unsupported version, failed notification). + // The SDK closes the transport itself when the handshake fails, which aborts the signal its + // DELETE would use, so the session is ended through a fresh, never-started transport. + if ( + endpoint._tag === "Http" && + transport instanceof StreamableHTTPClientTransport && + transport.sessionId !== undefined + ) + yield* terminate(http(endpoint, transport.sessionId)); + yield* Effect.promise(() => transport.close()).pipe(Effect.ignore); + }), + ), + ); + return client; + }); + + // The transport holds the Mcp-Session-Id the server minted, and the server answering it with 404, or + // with the 400 a freshly restarted single-session server emits, means it no longer knows this + // connection. Other 400s pass through untouched. + const session: { transport?: StreamableHTTPClientTransport; reported: boolean } = { reported: false }; + const failure = + (method: string) => + (error: unknown): Failed => { + if (needsAuth(error)) return new NeedsAuthError({ server, message: describe(error) }); + // The condensed message drops the validator's raw issue list; keep it for debugging. + if (isInvalidResult(error)) + runFork(Effect.logDebug("mcp invalid result", { server, method, error: (error as Error).message })); + const request = () => new RequestError({ server, method, message: explain(error) }); + if (!(error instanceof SdkHttpError) || session.transport?.sessionId === undefined) return request(); + const text = (error.data as { readonly text?: unknown }).text; + const expired = + error.status === 404 || + (error.status === 400 && typeof text === "string" && text.includes("Bad Request: Server not initialized")); + if (!expired) return request(); + if (!session.reported) { + session.reported = true; + runFork(handlers.expired); + } + return new SessionExpiredError({ server }); + }; + + let stderr = () => ""; + const exit = yield* Effect.gen(function* () { + if (endpoint._tag === "Stdio") { + const stdio = yield* McpStdio.make({ + server, + command: endpoint.command, + args: endpoint.args, + cwd: endpoint.cwd, + env: endpoint.env, + }); + stderr = stdio.stderr; + return yield* initialize(stdio.transport, startupTimeout); + } + // Every attempt opens a fresh transport and gets what is left of the startup timeout. + const deadline = (yield* Clock.currentTimeMillis) + startupTimeout; + const open = Effect.gen(function* () { + const remaining = Math.max(1, deadline - (yield* Clock.currentTimeMillis)); + session.transport = http(endpoint); + return yield* initialize(session.transport, remaining); + }); + return yield* retrying("mcp connect", open, deadline); + }).pipe(Effect.exit); + + if (Exit.isFailure(exit)) { + const squashed = Cause.squash(exit.cause); + const error = squashed instanceof Failure ? squashed.error : squashed; + if (needsAuth(error)) return yield* new NeedsAuthError({ server, message: describe(error) }); + // A server that fails to start usually says why on stderr. + const message = explain(error); + const tail = stderr().trim(); + return yield* new ConnectError({ + server, + message: tail && !message.includes(tail) ? `${message}\n${tail}` : message, + }); + } + + const client = exit.value; + closed.active = client; + // The server closed between the handshake finishing and this point. + const dropped = early.get(client); + early.clear(); + if (dropped !== undefined) report(dropped); + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + closed.shuttingDown = true; + // Close only aborts streams; the HTTP session lives on until the server expires it unless + // terminated explicitly. Terminate first: close aborts the signal the DELETE shares. + const transport = session.transport; + // Termination failures are logged below; onerror would report them a second time. + client.onerror = () => {}; + if (transport?.sessionId !== undefined && !session.reported) yield* terminate(transport); + yield* Effect.tryPromise(() => client.close()).pipe(Effect.ignore); + }), + ); + + // A success means transport errors from before the request no longer explain a later close. One + // that arrived meanwhile (a crash while other requests are pending) is kept. + const succeeded = (before: string | undefined) => + Effect.sync(() => { + if (lastError === before) lastError = undefined; + }); + const attempt = (run: (signal: AbortSignal) => Promise) => Effect.tryPromise({ try: run, catch: fail }); + // Maps the raw failure to a public error, traced as one span per request. + const settle = (method: string, effect: Effect.Effect) => + Effect.suspend(() => { + const before = lastError; + return effect.pipe( + Effect.mapError((error) => failure(method)(error.error)), + Effect.tap(succeeded(before)), + ); + }).pipe(Effect.withSpan(`McpClient.${method}`, { attributes: { "mcp.server": server } })); + const request = (method: string, run: (signal: AbortSignal) => Promise) => settle(method, attempt(run)); + // Tool calls are never retried: the server may already have run them. + const read = (method: string, run: (signal: AbortSignal) => Promise) => + settle(method, retryingRead(method, attempt(run))); + // Only a request given a progress handler asks for progress; each notification restarts its timeout, + // so long calls that report progress keep running. + const options = (signal: AbortSignal, onProgress?: (progress: Progress) => void) => + onProgress === undefined + ? { signal, timeout: requestTimeout } + : { signal, timeout: requestTimeout, resetTimeoutOnProgress: true, onprogress: onProgress }; + const capabilities = client.getServerCapabilities(); + + const connection: Connection = { + serverInfo: client.getServerVersion(), + capabilities, + instructions: client.getInstructions()?.trim() || undefined, + tools: () => + // Servers without the tools capability (prompts or resources only) do not answer tools/list. + capabilities?.tools === undefined + ? Effect.succeed([]) + : read("tools/list", (signal) => client.listTools(undefined, options(signal))).pipe( + Effect.map((result) => result.tools), + ), + resources: (cursor) => + capabilities?.resources === undefined + ? Effect.succeed({ resources: [] }) + : read("resources/list", (signal) => + client.request( + { method: "resources/list", params: cursor === undefined ? {} : { cursor } }, + options(signal), + ), + ), + resourceTemplates: (cursor) => + capabilities?.resources === undefined + ? Effect.succeed({ resourceTemplates: [] }) + : settle( + "resources/templates/list", + retryingRead( + "resources/templates/list", + attempt((signal) => + client.request( + { method: "resources/templates/list", params: cursor === undefined ? {} : { cursor } }, + options(signal), + ), + ), + ).pipe( + // Servers that do not implement templates have none. + Effect.catchIf( + (failure) => + failure.error instanceof ProtocolError && + failure.error.code === ProtocolErrorCode.MethodNotFound, + () => Effect.succeed({ resourceTemplates: [] }), + ), + ), + ), + readResource: (uri) => read("resources/read", (signal) => client.readResource({ uri }, options(signal))), + callTool: (call) => + request("tools/call", (signal) => + client.callTool({ name: call.name, arguments: call.args ?? {} }, options(signal, call.onProgress)), + ), + onToolsChanged: (handler) => listen("tools", handler), + onResourcesChanged: (handler) => listen("resources", handler), + onSessionExpired: (handler) => { + handlers.expired = handler; + }, + onClose: (handler) => { + closed.handler = handler; + if (closed.reason !== undefined) runFork(handler(closed.reason)); + }, + }; + return connection; +}); + +export * as McpClient from "./client.ts"; diff --git a/packages/harness/src/mcp/content.ts b/packages/harness/src/mcp/content.ts new file mode 100644 index 00000000..b3f2be69 --- /dev/null +++ b/packages/harness/src/mcp/content.ts @@ -0,0 +1,44 @@ +import type { Message } from "@codeworksh/aikit"; +import type { CallToolResult, ContentBlock } from "@modelcontextprotocol/client"; + +export type Content = Message.TextContent | Message.ImageContent; + +const fromBlock = (block: ContentBlock): Content => { + switch (block.type) { + case "text": + return { type: "text", text: block.text }; + case "image": + return { type: "image", data: block.data, mimeType: block.mimeType }; + case "audio": + return { type: "text", text: `[audio ${block.mimeType} omitted]` }; + case "resource_link": + return { type: "text", text: `${block.name}: ${block.uri}` }; + case "resource": { + const resource = block.resource; + if ("text" in resource) return { type: "text", text: resource.text }; + if (resource.mimeType?.startsWith("image/")) { + return { type: "image", data: resource.blob, mimeType: resource.mimeType }; + } + return { + type: "text", + text: `[binary resource ${resource.uri} (${resource.mimeType ?? "unknown type"}) omitted]`, + }; + } + default: + return { type: "text", text: `[unsupported MCP content ${(block as { type: string }).type}]` }; + } +}; + +/** + * A tool result as model content. A result without content blocks but with `structuredContent` + * becomes its JSON, since servers should, but do not always, mirror structured results as text. + */ +export const fromResult = (result: Pick): Content[] => { + const content = (result.content ?? []).map(fromBlock); + if (content.length === 0 && result.structuredContent !== undefined) { + content.push({ type: "text", text: JSON.stringify(result.structuredContent, null, 2) }); + } + return content; +}; + +export * as McpContent from "./content.ts"; diff --git a/packages/harness/src/mcp/error.ts b/packages/harness/src/mcp/error.ts new file mode 100644 index 00000000..a0b1aa6e --- /dev/null +++ b/packages/harness/src/mcp/error.ts @@ -0,0 +1,36 @@ +import { Schema } from "effect"; + +/** The server could not be started, reached or initialized. */ +export class ConnectError extends Schema.TaggedError()("Mcp.ConnectError", { + server: Schema.String, + message: Schema.String, +}) {} + +/** The server rejected the connection until the user signs in. */ +export class NeedsAuthError extends Schema.TaggedError()("Mcp.NeedsAuthError", { + server: Schema.String, + message: Schema.String, +}) {} + +/** A Streamable HTTP server no longer knows this connection's session; the caller reconnects. */ +export class SessionExpiredError extends Schema.TaggedError()("Mcp.SessionExpiredError", { + server: Schema.String, +}) { + override get message(): string { + return `mcp server session expired: ${this.server}`; + } +} + +/** A request on a connected server failed. */ +export class RequestError extends Schema.TaggedError()("Mcp.RequestError", { + server: Schema.String, + method: Schema.String, + message: Schema.String, +}) {} + +/** A failure inside the stdio transport, reported to the SDK through `onerror` or a rejected promise. */ +export class StdioError extends Schema.TaggedError()("Mcp.StdioError", { + message: Schema.String, +}) {} + +export * as McpError from "./error.ts"; diff --git a/packages/harness/src/mcp/log.ts b/packages/harness/src/mcp/log.ts new file mode 100644 index 00000000..09c954ae --- /dev/null +++ b/packages/harness/src/mcp/log.ts @@ -0,0 +1,57 @@ +import type { LoggingMessageNotificationParams } from "@modelcontextprotocol/client"; +import { DateTime, Effect, Semaphore } from "effect"; +import { fileSystem, hostPath } from "../host.ts"; + +const MAX_LOG_BYTES = 5 * 1024 * 1024; + +const formatData = (data: unknown): string => { + if (typeof data === "string") return data; + try { + return JSON.stringify(data) ?? String(data); + } catch { + return String(data); + } +}; + +/** One `notifications/message` from `server` as a log line; continuation lines are indented. */ +export const format = (server: string, params: LoggingMessageNotificationParams, now: DateTime.Utc): string => { + const logger = params.logger ? ` ${params.logger}:` : ""; + const text = formatData(params.data).replace(/\r?\n/g, "\n "); + return `${DateTime.formatIso(now)} [${server}] ${params.level}${logger} ${text}\n`; +}; + +/** + * Appends server log messages to one file, rotated to `.1` once it grows past 5 MB. Several + * processes may write to the same file, so the size is re-read before rotating. Write errors are + * ignored: logging must not break tools. + */ +export const make = Effect.fn("McpLog.make")(function* (path: string) { + const lock = yield* Semaphore.make(1); + let size: number | undefined; + const current = fileSystem.stat(path).pipe( + Effect.map((info) => Number(info.size)), + Effect.orElseSucceed(() => 0), + ); + + const write = (server: string, params: LoggingMessageNotificationParams) => + lock.withPermits(1)( + Effect.gen(function* () { + const line = format(server, params, yield* DateTime.now); + if (size === undefined) { + yield* fileSystem.makeDirectory(hostPath.dirname(path), { recursive: true }); + size = yield* current; + } + if (size > MAX_LOG_BYTES) { + // Another process may have rotated it already; check before renaming. + if ((yield* current) > MAX_LOG_BYTES) yield* fileSystem.rename(path, `${path}.1`); + size = yield* current; + } + yield* fileSystem.writeFileString(path, line, { flag: "a" }); + size += Buffer.byteLength(line); + }).pipe(Effect.ignore), + ); + + return { path, write }; +}); + +export * as McpLog from "./log.ts"; diff --git a/packages/harness/src/mcp/stdio.ts b/packages/harness/src/mcp/stdio.ts new file mode 100644 index 00000000..f69570d2 --- /dev/null +++ b/packages/harness/src/mcp/stdio.ts @@ -0,0 +1,258 @@ +import { type JSONRPCMessage, ReadBuffer, serializeMessage, type Transport } from "@modelcontextprotocol/client"; +import { Cause, Duration, Effect, Fiber, Option, Queue, Scope, Stream } from "effect"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { Process } from "../sandbox/utils/process.ts"; +import { StdioError } from "./error.ts"; + +/** How long a server gets to exit on its own after stdin closes, before it is terminated. */ +const CLOSE_GRACE = Duration.seconds(2); +/** Escalate SIGTERM to SIGKILL after this long. */ +const FORCE_KILL_AFTER = Duration.seconds(2); +const OUTGOING_CAPACITY = 64; +const MAX_FRAME_BYTES = 16 * 1024 * 1024; +// An unexpected exit is reported with its exit code and the end of stderr, which usually names the cause. +const EXIT_WAIT = Duration.seconds(1); +const STDERR_TAIL = 2_000; + +/** Process groups of running servers, terminated if the host exits without closing them. */ +const live = new Set(); +let exitHook = false; +const track = (pid: number) => { + if (process.platform === "win32") return () => {}; + if (!exitHook) { + exitHook = true; + process.once("exit", () => { + for (const group of live) { + try { + process.kill(-group, "SIGTERM"); + } catch { + // the group is already gone + } + } + }); + } + live.add(pid); + return () => { + live.delete(pid); + }; +}; + +export interface Options { + /** Server name; only used to attribute logs. */ + readonly server: string; + readonly command: string; + readonly args: ReadonlyArray; + readonly cwd: string; + /** Environment declared by the server config, merged over the host's. */ + readonly env: Record; +} + +/** + * MCP stdio transport for a server process on the host. + * + * The process is acquired in the calling scope: closing the scope kills it and its process group, + * whether or not the transport was closed. + */ +export const make = Effect.fn("McpStdio.make")(function* (options: Options) { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const scope = yield* Effect.scope; + const runPromise = Effect.runPromiseWith(yield* Effect.context()); + // Outgoing frames are queued rather than written to `handle.stdin` directly: the sink closes the + // stream it is run with, and stdin must stay open across the whole connection. + const outgoing = yield* Queue.bounded(OUTGOING_CAPACITY); + const buffer = new ReadBuffer({ maxBufferSize: MAX_FRAME_BYTES }); + const state: { + phase: "ready" | "starting" | "open" | "closed"; + handle?: ChildProcessSpawner.ChildProcessHandle | undefined; + stdoutOpen: boolean; + stderrOpen: boolean; + } = { phase: "ready", stdoutOpen: false, stderrOpen: false }; + let startup: Promise | undefined; + let closing: Promise | undefined; + let untrack = () => {}; + let outputStarted = false; + // Set when reading stdout failed (an oversized frame), which then explains the close instead of an exit. + let broken = false; + let stderr = ""; + + const stop = (handle: ChildProcessSpawner.ChildProcessHandle) => + Effect.timeoutOption(handle.exitCode, CLOSE_GRACE).pipe( + Effect.ignore, + Effect.andThen( + Effect.suspend(() => + Process.terminate(handle, { + forceKillAfter: FORCE_KILL_AFTER, + pipesOpen: state.stdoutOpen || state.stderrOpen, + graceful: true, + }), + ), + ), + Effect.ensuring(Effect.sync(() => untrack())), + ); + + const close = () => + (closing ??= runPromise( + Effect.gen(function* () { + state.phase = "closed"; + Queue.endUnsafe(outgoing); + if (startup) yield* Effect.promise(() => startup!.catch(() => undefined)); + const handle = state.handle; + if (!handle) { + // Closed before output was read (spawn failed, or closed while starting): nothing else reports it. + if (!outputStarted) transport.onclose?.(); + return; + } + state.handle = undefined; + yield* stop(handle); + }).pipe(Effect.ensuring(Queue.shutdown(outgoing)), Effect.ensuring(Effect.sync(() => buffer.clear()))), + )); + + const transport: Transport = { + start: () => { + if (state.phase !== "ready") + return Promise.reject(new StdioError({ message: "stdio transport already started" })); + state.phase = "starting"; + startup = runPromise( + Effect.gen(function* () { + const handle = yield* spawner + .spawn( + ChildProcess.make(options.command, [...options.args], { + cwd: options.cwd, + env: options.env, + extendEnv: true, + stdin: { stream: Stream.encodeText(Stream.fromQueue(outgoing)), endOnDone: true }, + stdout: "pipe", + stderr: "pipe", + forceKillAfter: FORCE_KILL_AFTER, + }), + ) + .pipe( + Effect.mapError((error) => { + if (error.reason._tag !== "NotFound") return error; + // The spawner checks the working directory before it spawns, with the same tag. + return error.reason.method === "spawn" + ? new StdioError({ message: `command not found: ${options.command}` }) + : new StdioError({ message: `working directory not found: ${options.cwd}` }); + }), + ); + state.handle = handle; + state.stdoutOpen = true; + state.stderrOpen = true; + untrack = track(Number(handle.pid)); + yield* Effect.addFinalizer(() => Effect.sync(() => untrack())); + if (state.phase === "closed") { + state.handle = undefined; + return yield* stop(handle); + } + state.phase = "open"; + yield* startOutput(handle); + }).pipe(Scope.provide(scope)), + ); + return startup; + }, + send: (message: JSONRPCMessage) => + state.phase !== "open" + ? Promise.reject(new StdioError({ message: "not connected" })) + : runPromise( + Queue.offer(outgoing, serializeMessage(message)).pipe( + Effect.flatMap((offered) => + offered ? Effect.void : Effect.fail(new StdioError({ message: "not connected" })), + ), + ), + ), + close, + }; + + const deliver = (chunk: Uint8Array) => + Effect.gen(function* () { + yield* Effect.try({ + try: () => buffer.append(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength)), + catch: () => new StdioError({ message: "mcp stdio frame exceeded 16 MiB" }), + }); + while (true) { + // `undefined` means the frame failed to parse: the buffer has already advanced past it, so + // keep draining. `null` means the buffer holds no complete frame yet. + const message = yield* Effect.try({ + try: () => buffer.readMessage(), + catch: (cause) => new StdioError({ message: cause instanceof Error ? cause.message : String(cause) }), + }).pipe( + Effect.catch((error) => + Effect.sync(() => { + transport.onerror?.(error); + return undefined; + }), + ), + ); + if (message === undefined) continue; + if (message === null) return; + transport.onmessage?.(message); + } + }); + + const startOutput = (handle: ChildProcessSpawner.ChildProcessHandle) => + Effect.gen(function* () { + outputStarted = true; + // Drain stderr into the debug log so chatty servers cannot stall on a full pipe. + const drain = yield* Effect.forkScoped( + handle.stderr.pipe( + Stream.decodeText(), + Stream.tap((output) => Effect.sync(() => (stderr = (stderr + output).slice(-STDERR_TAIL)))), + Stream.runForEach((output) => + output.trim() === "" + ? Effect.void + : Effect.logDebug("mcp server stderr", { server: options.server, output }), + ), + Effect.ignore, + // A descendant holding only stderr keeps the group alive, and the group id ours. + Effect.ensuring( + Effect.sync(() => { + state.stderrOpen = false; + }), + ), + ), + ); + + yield* Effect.forkScoped( + Stream.runForEach(handle.stdout, deliver).pipe( + Effect.tapCause((cause) => + Effect.sync(() => { + if (Cause.hasInterruptsOnly(cause)) return; + broken = true; + const error = Cause.squash(cause); + transport.onerror?.(error instanceof Error ? error : new StdioError({ message: String(error) })); + }), + ), + Effect.ignore, + // stdout ending means the server is gone. + Effect.ensuring( + Effect.gen(function* () { + state.stdoutOpen = false; + if (state.phase !== "closed") { + if (!broken) { + // stdout can end before the final stderr output has been read. A grandchild holding + // stderr open must not cost the exit code, so each wait is bounded separately. + yield* Fiber.await(drain).pipe(Effect.timeoutOption(EXIT_WAIT), Effect.ignore); + const code = yield* handle.exitCode.pipe( + Effect.timeoutOption(EXIT_WAIT), + Effect.orElseSucceed(() => Option.none()), + ); + const exited = Option.match(code, { + onNone: () => "mcp server process exited", + onSome: (code) => `mcp server process exited with code ${code}`, + }); + const output = stderr.trim(); + transport.onerror?.(new StdioError({ message: output ? `${exited}: ${output}` : exited })); + } + yield* Effect.promise(close); + } + transport.onclose?.(); + }), + ), + ), + ); + }); + + return { transport, stderr: () => stderr }; +}); + +export * as McpStdio from "./stdio.ts"; diff --git a/packages/harness/src/sandbox/shell/host.ts b/packages/harness/src/sandbox/shell/host.ts index dbaec367..07be7559 100644 --- a/packages/harness/src/sandbox/shell/host.ts +++ b/packages/harness/src/sandbox/shell/host.ts @@ -1,5 +1,6 @@ -import { Duration, Effect, Exit, Layer, Option, Stream } from "effect"; +import { Duration, Effect, Exit, Layer, Stream } from "effect"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { Process } from "../utils/process.ts"; import { type ExecChunk, type ExecResult, Shell, ShellError, type ShellOptions } from "./shell.ts"; /** @@ -41,15 +42,9 @@ export const layer = (options: Options = {}): Layer.Layer, opts?: ShellOptions) => Effect.gen(function* () { const spawned = yield* spawner.spawn( @@ -69,29 +64,7 @@ export const layer = (options: Options = {}): Layer.Layer Exit.isSuccess(exit) ? Effect.void - : Effect.gen(function* () { - const running = yield* spawned.isRunning.pipe(Effect.orElseSucceed(() => false)); - if (!running && openPipes === 0) return; - if (running) { - yield* spawned.kill({ killSignal: "SIGTERM" }).pipe( - Effect.timeoutOption(forceKillAfter), - Effect.flatMap( - Option.match({ - onSome: () => Effect.void, - onNone: () => spawned.kill({ killSignal: "SIGKILL" }), - }), - ), - Effect.ignore, - ); - } - yield* Effect.sync(() => { - try { - process.kill(-Number(spawned.pid), "SIGKILL"); - } catch { - // the group is already gone - } - }); - }), + : Effect.suspend(() => Process.terminate(spawned, { forceKillAfter, pipesOpen: openPipes > 0 })), ); return { exitCode: spawned.exitCode, ...output }; }); diff --git a/packages/harness/src/sandbox/utils/process.ts b/packages/harness/src/sandbox/utils/process.ts index 3f4e438a..4d4e50fb 100644 --- a/packages/harness/src/sandbox/utils/process.ts +++ b/packages/harness/src/sandbox/utils/process.ts @@ -1,5 +1,5 @@ import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"; -import { Effect, Layer } from "effect"; +import { type Duration, Effect, Layer, Option } from "effect"; import { ChildProcessSpawner } from "effect/process"; /** @@ -17,4 +17,54 @@ export const unsupported = Layer.succeed( ChildProcessSpawner.make(() => Effect.die(new Error("process execution is not supported by this sandbox"))), ); +const signalGroup = (handle: ChildProcessSpawner.ChildProcessHandle, signal: NodeJS.Signals) => + Effect.sync(() => { + try { + process.kill(-Number(handle.pid), signal); + } catch { + // the group is already gone + } + }); + +/** + * Stops a spawned process and its group. + * + * The spawner's own release sends SIGTERM and then waits for exit, so a process + * that ignores SIGTERM (and children inheriting that) would make a scope close + * wait forever. This SIGTERMs the leader, SIGKILLs it if it outlives + * `forceKillAfter`, then SIGKILLs whatever is left of the group — children can + * outlive a leader that already exited. + * + * `graceful` gives the rest of the group a SIGTERM first: when the leader has + * already exited it SIGTERMs the group (children that dropped their pipes + * included), and SIGKILLs it `forceKillAfter` later only while a pipe is still + * open. Otherwise an exited leader whose pipes are closed leaves the group alone. + */ +export const terminate = ( + handle: ChildProcessSpawner.ChildProcessHandle, + options: { readonly forceKillAfter: Duration.Input; readonly pipesOpen: boolean; readonly graceful?: boolean }, +) => + Effect.gen(function* () { + const running = yield* handle.isRunning.pipe(Effect.orElseSucceed(() => false)); + if (!running) { + if (options.graceful) { + yield* signalGroup(handle, "SIGTERM"); + if (!options.pipesOpen) return; + yield* Effect.sleep(options.forceKillAfter); + } else if (!options.pipesOpen) return; + return yield* signalGroup(handle, "SIGKILL"); + } + yield* handle.kill({ killSignal: "SIGTERM" }).pipe( + Effect.timeoutOption(options.forceKillAfter), + Effect.flatMap( + Option.match({ + onSome: () => Effect.void, + onNone: () => handle.kill({ killSignal: "SIGKILL" }), + }), + ), + Effect.ignore, + ); + yield* signalGroup(handle, "SIGKILL"); + }); + export * as Process from "./process.ts"; diff --git a/packages/harness/test/__artifacts__/mcp.client.json b/packages/harness/test/__artifacts__/mcp.client.json new file mode 100644 index 00000000..397353d6 --- /dev/null +++ b/packages/harness/test/__artifacts__/mcp.client.json @@ -0,0 +1,443 @@ +{ + "cancel": { + "after": "after cancel", + "cancelled": 1, + }, + "crash": { + "error": { + "message": "Connection closed: mcp server process exited with code 3: fatal: fixture crashed on purpose", + "tag": "Mcp.RequestError", + }, + "method": "tools/call", + "reason": "connection closed: mcp server process exited with code 3: fatal: fixture crashed on purpose", + }, + "failedConnectSession": { + "failedNotification": { + "error": { + "message": "Error POSTing to endpoint: notification failed (HTTP 500)", + "tag": "Mcp.ConnectError", + }, + "seen": [ + "POST initialize", + "POST notifications/initialized", + "DELETE", + "POST initialize", + "POST notifications/initialized", + "DELETE", + "POST initialize", + "POST notifications/initialized", + "DELETE", + ], + }, + "unsupported": { + "error": { + "message": "Server's protocol version is not supported: 1999-01-01", + "tag": "Mcp.ConnectError", + }, + "seen": [ + "POST initialize", + "DELETE", + ], + }, + }, + "frameLimit": { + "error": { + "message": "Connection closed: mcp stdio frame exceeded 16 MiB", + "tag": "Mcp.RequestError", + }, + "reason": "connection closed: mcp stdio frame exceeded 16 MiB", + }, + "framing": { + "after": "still alive", + "framed": "framed", + "toolsChanged": true, + }, + "http400": { + "notInitialized": { + "message": "mcp server session expired: raw", + "tag": "Mcp.SessionExpiredError", + }, + "other": { + "message": "Error POSTing to endpoint: {"jsonrpc":"2.0","error":{"code":-32000,"message":"Bad Request: something else"},"id":null} (HTTP 400)", + "tag": "Mcp.RequestError", + }, + }, + "httpClose": { + "deletes": 1, + "hangingDeleteBounded": true, + }, + "httpHappy": { + "echo": [ + { + "text": "over http", + "type": "text", + }, + ], + "serverInfo": { + "name": "http-fixture", + "version": "1.0.0", + }, + "tools": [ + "echo", + "mixed", + "structured", + "crash", + "progress", + "slow", + "hang", + "cancelled", + "roots", + "capabilities", + "spawn", + "frame", + "exit", + "huge", + "log", + "stderr-child", + "typed-missing", + "typed-mismatch", + ], + }, + "httpRetries": { + "call": { + "message": "Error POSTing to endpoint: overloaded (HTTP 503)", + "tag": "Mcp.RequestError", + }, + "callRequests": 1, + "listRequests": 2, + "tools": [ + "echo", + ], + }, + "initializedTimeout": { + "deleted": true, + "error": { + "message": "mcp connect timed out after ", + "tag": "Mcp.ConnectError", + }, + }, + "invalidEntry": { + "message": "Invalid result for tools/list: tools.1.inputSchema: Invalid input: expected object, received undefined", + "tag": "Mcp.RequestError", + }, + "listChangedWhileListing": { + "toolsChanged": true, + }, + "logFile": [ + " [fixture] info first line", + " second line", + " [fixture] warning db: {"query":"select 1","ms":3}", + "", + ], + "logRotation": { + "current": " [fixture] warning db: {"query":"select 1","ms":3} +", + "rotated": true, + }, + "missingCwd": { + "message": "working directory not found: /missing-dir", + "tag": "Mcp.ConnectError", + }, + "noToolsCapability": { + "printed": 0, + "tools": [], + }, + "onClose": { + "afterDeliberateClose": [], + "afterServerExit": "connection closed: mcp server process exited with code 0", + }, + "onLog": [ + { + "data": "ready", + "level": "info", + "logger": "startup", + }, + { + "data": "first line +second line", + "level": "info", + }, + { + "data": { + "ms": 3, + "query": "select 1", + }, + "level": "warning", + "logger": "db", + }, + ], + "outputSchema": { + "declared": [ + "typed-missing", + "typed-mismatch", + ], + "mismatch": { + "message": "Structured content does not match the tool's output schema: data/answer must be number", + "tag": "Mcp.RequestError", + }, + "missing": { + "message": "Tool typed-missing has an output schema but did not return structured content", + "tag": "Mcp.RequestError", + }, + }, + "processGroup": { + "killed": true, + }, + "progress": { + "progressed": "progress done", + "sawProgress": true, + "silent": { + "message": "Request timed out", + "tag": "Mcp.RequestError", + }, + }, + "refused": { + "error": { + "message": "fetch failed (ECONNREFUSED)", + "tag": "Mcp.ConnectError", + }, + }, + "resources": { + "first": { + "nextCursor": "page-2", + "resources": [ + { + "name": "one", + "uri": "fixture://one", + }, + ], + }, + "read": { + "contents": [ + { + "mimeType": "text/plain", + "text": "contents of fixture://one", + "uri": "fixture://one", + }, + ], + }, + "resourcesChanged": true, + "second": { + "resources": [ + { + "name": "changed", + "uri": "fixture://changed", + }, + ], + }, + "templates": { + "resourceTemplates": [], + }, + }, + "retry": { + "initializes": 2, + "tools": [ + "echo", + ], + }, + "retryCloseReason": { + "afterRegister": [], + "afterScope": [], + "initializes": 2, + "tools": [ + "echo", + ], + }, + "roots": { + "withRoots": { + "capabilities": { + "roots": {}, + }, + "roots": { + "roots": [ + { + "name": "project", + "uri": "file:///workspace/project", + }, + { + "name": "data dir", + "uri": "file:///srv/data%20dir", + }, + ], + }, + }, + "withoutRoots": { + "capabilities": {}, + }, + }, + "sessionExpired": { + "errors": [ + { + "message": "mcp server session expired: raw", + "tag": "Mcp.SessionExpiredError", + }, + { + "message": "mcp server session expired: raw", + "tag": "Mcp.SessionExpiredError", + }, + ], + "reported": 1, + "terminated": false, + }, + "spawnFailure": { + "message": "command not found: codework-mcp-missing-binary", + "tag": "Mcp.ConnectError", + }, + "stderrGrandchild": { + "killed": true, + }, + "stdioHappy": { + "echo": [ + { + "text": "hi", + "type": "text", + }, + ], + "instructions": "Use the echo tool.", + "mixed": [ + { + "text": "hello", + "type": "text", + }, + { + "data": "iVBORw0KGgo=", + "mimeType": "image/png", + "type": "image", + }, + { + "text": "embedded notes", + "type": "text", + }, + { + "data": "iVBORw0KGgo=", + "mimeType": "image/png", + "type": "image", + }, + { + "text": "[binary resource file:///data.bin (unknown type) omitted]", + "type": "text", + }, + { + "text": "[audio audio/wav omitted]", + "type": "text", + }, + { + "text": "readme: file:///README.md", + "type": "text", + }, + ], + "serverInfo": { + "name": "stdio-fixture", + "version": "1.0.0", + }, + "structured": [ + { + "text": "{ + "answer": 42, + "items": [ + "a", + "b" + ] +}", + "type": "text", + }, + ], + "tools": [ + "echo", + "mixed", + "structured", + "crash", + "progress", + "slow", + "hang", + "cancelled", + "roots", + "capabilities", + "spawn", + "frame", + "exit", + "huge", + "log", + "stderr-child", + "typed-missing", + "typed-mismatch", + ], + }, + "strictResources": { + "unnamed": { + "resources": { + "message": "Invalid result for resources/list: resources.0.name: Invalid input: expected string, received undefined", + "tag": "Mcp.RequestError", + }, + "templates": { + "message": "Invalid result for resources/templates/list: resourceTemplates.0.name: Invalid input: expected string, received undefined", + "tag": "Mcp.RequestError", + }, + }, + "valid": { + "resources": { + "resources": [ + { + "_meta": { + "origin": "fixture", + }, + "description": "A file", + "mimeType": "text/plain", + "name": "a", + "uri": "file:///a.txt", + }, + ], + }, + "templates": { + "resourceTemplates": [ + { + "description": "Any file", + "name": "files", + "uriTemplate": "file:///{path}", + }, + ], + }, + }, + }, + "stubborn": { + "boundedUnder8s": true, + "killed": true, + }, + "toolsPagination": { + "duplicate": { + "tools": [ + "t-start", + "t-again", + ], + }, + "emptyCursor": { + "error": { + "message": "tools/list: exceeded listMaxPages (64); server pagination did not terminate", + "tag": "Mcp.RequestError", + }, + }, + "nullCursor": { + "error": { + "message": "Invalid result for tools/list: nextCursor: Invalid input: expected string, received null", + "tag": "Mcp.RequestError", + }, + }, + }, + "unadvertisedListChanged": { + "listChanged": {}, + "toolsChanged": true, + }, + "unauthorized": { + "redirects": [ + "/authorize", + ], + "withProvider": { + "message": "Unauthorized", + "tag": "Mcp.NeedsAuthError", + }, + "without": { + "message": "Error POSTing to endpoint: unauthorized (HTTP 401)", + "tag": "Mcp.ConnectError", + }, + }, +} \ No newline at end of file diff --git a/packages/harness/test/fixtures/mcp/http.ts b/packages/harness/test/fixtures/mcp/http.ts new file mode 100644 index 00000000..ecb7cecb --- /dev/null +++ b/packages/harness/test/fixtures/mcp/http.ts @@ -0,0 +1,142 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import { Server, WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/server"; +import { Effect } from "effect"; +// @ts-expect-error plain .mjs fixture shared with the stdio server, which runs without a TS loader +import { install } from "./tools.mjs"; + +type Handler = (request: IncomingMessage, response: ServerResponse) => Promise; + +/** Serves `handler` on 127.0.0.1 at an ephemeral port for the calling scope; yields the `/mcp` URL. */ +export const listen = (handler: Handler) => + Effect.acquireRelease( + Effect.callback>((resume) => { + const server = createServer((request, response) => { + handler(request, response).catch((error: unknown) => { + if (!response.headersSent) response.writeHead(500); + response.end(String(error)); + }); + }); + server.listen(0, "127.0.0.1", () => resume(Effect.succeed(server))); + }), + (server) => + Effect.callback((resume) => { + server.closeAllConnections(); + server.close(() => resume(Effect.void)); + }), + ).pipe(Effect.map((server) => new URL(`http://127.0.0.1:${(server.address() as AddressInfo).port}/mcp`))); + +const readBody = async (request: IncomingMessage) => { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(chunk as Buffer); + return Buffer.concat(chunks); +}; + +/** A single-session streamable HTTP server running the shared fixture tools on the real SDK transport. */ +export const sdk = async (): Promise => { + const server = new Server({ name: "http-fixture", version: "1.0.0" }, { capabilities: { tools: {} } }); + install(server); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => "fixture-session" }); + await server.connect(transport); + return async (request, response) => { + const headers = new Headers(); + for (let index = 0; index < request.rawHeaders.length; index += 2) + headers.append(request.rawHeaders[index]!, request.rawHeaders[index + 1]!); + const body = request.method === "GET" || request.method === "HEAD" ? undefined : await readBody(request); + const answer = await transport.handleRequest( + new Request(new URL(request.url ?? "/", "http://127.0.0.1"), { + method: request.method ?? "GET", + headers, + ...(body === undefined ? {} : { body }), + }), + ); + response.writeHead(answer.status, Object.fromEntries(answer.headers)); + if (!answer.body) return void response.end(); + const reader = answer.body.getReader(); + response.on("close", () => void reader.cancel().catch(() => {})); + while (true) { + const chunk = await reader.read().catch(() => ({ done: true as const, value: undefined })); + if (chunk.done) break; + response.write(chunk.value); + } + response.end(); + }; +}; + +/** A scripted answer: a status with a text body, or no answer at all. */ +export type Reply = { readonly status: number; readonly body: string } | "hang"; + +/** Requests a hand-written server saw, as `METHOD rpc-method`. */ +export interface Raw { + readonly handler: Handler; + readonly seen: string[]; + /** Answer every request carrying a session id with 404, as a server that lost the session. */ + forget: boolean; +} + +/** A hand-written streamable HTTP server for raw status scenarios. */ +export const raw = ( + options: { + readonly failInitialize?: number; + readonly unauthorized?: boolean; + /** Protocol version to answer initialize with, instead of echoing the client's. */ + readonly protocolVersion?: string; + /** Overrides the answer to a request, keyed as in `seen`; `undefined` answers normally. */ + readonly script?: (request: string) => Reply | undefined; + } = {}, +): Raw => { + let failures = options.failInitialize ?? 0; + const state: Raw = { + seen: [], + forget: false, + handler: async (request, response) => { + // An OAuth-protected server: every MCP request is refused, and there is no discovery metadata. + if (options.unauthorized) { + state.seen.push(`${request.method} ${new URL(request.url ?? "/", "http://127.0.0.1").pathname}`); + await readBody(request); + if (request.url?.startsWith("/mcp") !== true) return void response.writeHead(404).end(); + response.writeHead(401, { "www-authenticate": 'Bearer error="invalid_token"' }); + return void response.end("unauthorized"); + } + const body = request.method === "POST" ? (JSON.parse((await readBody(request)).toString()) as Rpc) : undefined; + const key = body?.method ? `${request.method} ${body.method}` : `${request.method}`; + state.seen.push(key); + const scripted = options.script?.(key); + if (scripted === "hang") return; + if (scripted !== undefined) + return void response.writeHead(scripted.status, { "content-type": "text/plain" }).end(scripted.body); + if (state.forget && request.headers["mcp-session-id"] !== undefined) { + response.writeHead(404, { "content-type": "text/plain" }); + return void response.end("session not found"); + } + if (request.method === "GET") return void response.writeHead(405).end(); + if (request.method === "DELETE") return void response.writeHead(200).end(); + if (body === undefined || body.id === undefined) return void response.writeHead(202).end(); + if (body.method === "initialize" && failures > 0) { + failures--; + return void response.writeHead(503, { "content-type": "text/plain" }).end("overloaded"); + } + const result = + body.method === "initialize" + ? { + protocolVersion: options.protocolVersion ?? body.params?.protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: "raw-fixture", version: "1.0.0" }, + } + : body.method === "tools/list" + ? { tools: [{ name: "echo", inputSchema: { type: "object" } }] } + : body.method === "tools/call" + ? { content: [{ type: "text", text: "called" }] } + : {}; + response.writeHead(200, { "content-type": "application/json", "mcp-session-id": "raw-session" }); + response.end(JSON.stringify({ jsonrpc: "2.0", id: body.id, result })); + }, + }; + return state; +}; + +interface Rpc { + readonly id?: number | string; + readonly method?: string; + readonly params?: { readonly protocolVersion?: string }; +} diff --git a/packages/harness/test/fixtures/mcp/malformed.mjs b/packages/harness/test/fixtures/mcp/malformed.mjs new file mode 100644 index 00000000..b19c5f5b --- /dev/null +++ b/packages/harness/test/fixtures/mcp/malformed.mjs @@ -0,0 +1,53 @@ +// Hand-written stdio server sending list pages that stretch the spec; run as `node malformed.mjs`. +// MCP_MALFORMED picks the variant: "null" / "empty" (tools/list ends with nextCursor null / ""), +// "duplicate" (tools/list repeats its cursor), "invalid" (a tool without inputSchema), +// "resources" (well-formed entries with extra fields) or "unnamed" (resources and templates without names). +import { createInterface } from "node:readline"; + +const variant = process.env.MCP_MALFORMED; +const end = variant === "empty" ? "" : null; +const tool = (name) => ({ name, inputSchema: { type: "object" } }); + +const answer = (method, params) => { + switch (method) { + case "initialize": + return { + protocolVersion: params.protocolVersion, + capabilities: variant === "resources" || variant === "unnamed" ? { resources: {} } : { tools: {} }, + serverInfo: { name: "malformed-fixture", version: "1.0.0" }, + }; + case "tools/list": + if (variant === "invalid") return { tools: [tool("good"), { name: "bad" }] }; + if (variant === "duplicate") return { tools: [tool(`t-${params?.cursor ?? "start"}`)], nextCursor: "again" }; + return params?.cursor === "p2" + ? { tools: [tool("second")], nextCursor: end } + : { tools: [tool("first")], nextCursor: "p2" }; + case "resources/list": + return variant === "unnamed" + ? { resources: [{ uri: "file:///a.txt" }] } + : { + resources: [ + { + uri: "file:///a.txt", + name: "a", + description: "A file", + mimeType: "text/plain", + _meta: { origin: "fixture" }, + }, + ], + }; + case "resources/templates/list": + return variant === "unnamed" + ? { resourceTemplates: [{ uriTemplate: "file:///{path}" }] } + : { resourceTemplates: [{ uriTemplate: "file:///{path}", name: "files", description: "Any file" }] }; + default: + return {}; + } +}; + +for await (const line of createInterface({ input: process.stdin, crlfDelay: Infinity })) { + const message = JSON.parse(line); + if (message.id === undefined || message.method === undefined) continue; + const result = answer(message.method, message.params); + process.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", id: message.id, result })}\n`); +} diff --git a/packages/harness/test/fixtures/mcp/server.mjs b/packages/harness/test/fixtures/mcp/server.mjs new file mode 100644 index 00000000..1138d97f --- /dev/null +++ b/packages/harness/test/fixtures/mcp/server.mjs @@ -0,0 +1,37 @@ +// Stdio MCP fixture server built on the SDK; run as `node server.mjs`. +// MCP_FIXTURE picks the variant: unset (tools with listChanged), "plain" (tools without listChanged) +// or "resources" (resources only, paginated, without a templates/list handler). +import { Server } from "@modelcontextprotocol/server"; +import { StdioServerTransport } from "@modelcontextprotocol/server/stdio"; +import { install } from "./tools.mjs"; + +const variant = process.env.MCP_FIXTURE; +const capabilities = + variant === "resources" + ? { resources: { listChanged: true } } + : variant === "plain" + ? { tools: {} } + : { tools: { listChanged: true }, logging: {} }; +const server = new Server( + { name: "stdio-fixture", version: "1.0.0" }, + { capabilities, instructions: " Use the echo tool. \n" }, +); + +if (variant === "resources") { + const pages = { + "": { resources: [{ uri: "fixture://one", name: "one" }], nextCursor: "page-2" }, + "page-2": { resources: [{ uri: "fixture://changed", name: "changed" }] }, + }; + server.setRequestHandler("resources/list", (request) => pages[request.params?.cursor ?? ""]); + server.setRequestHandler("resources/read", async (request) => { + const uri = request.params.uri; + if (uri === "fixture://changed") await server.sendResourceListChanged(); + return { contents: [{ uri, mimeType: "text/plain", text: `contents of ${uri}` }] }; + }); +} else install(server); + +// A message sent right after the handshake, before the client has listed anything. +if (capabilities.logging) + server.oninitialized = () => void server.sendLoggingMessage({ level: "info", logger: "startup", data: "ready" }); + +await server.connect(new StdioServerTransport()); diff --git a/packages/harness/test/fixtures/mcp/stubborn.mjs b/packages/harness/test/fixtures/mcp/stubborn.mjs new file mode 100644 index 00000000..2d9b5219 --- /dev/null +++ b/packages/harness/test/fixtures/mcp/stubborn.mjs @@ -0,0 +1,23 @@ +// Answers initialize, then ignores stdin EOF and SIGTERM; so does the grandchild it spawns. +// The grandchild pid is reported through the initialize instructions. +import { spawn } from "node:child_process"; +import { createInterface } from "node:readline"; + +process.on("SIGTERM", () => {}); +const grandchild = spawn(process.execPath, ["-e", "process.on('SIGTERM', () => {}); setInterval(() => {}, 1000)"], { + stdio: "ignore", +}); +setInterval(() => {}, 1000); + +const lines = createInterface({ input: process.stdin, crlfDelay: Infinity }); +lines.on("line", (line) => { + const message = JSON.parse(line); + if (message.method !== "initialize") return; + const result = { + protocolVersion: message.params.protocolVersion, + capabilities: {}, + serverInfo: { name: "stubborn-fixture", version: "1.0.0" }, + instructions: JSON.stringify({ server: process.pid, grandchild: grandchild.pid }), + }; + process.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", id: message.id, result })}\n`); +}); diff --git a/packages/harness/test/fixtures/mcp/tools.mjs b/packages/harness/test/fixtures/mcp/tools.mjs new file mode 100644 index 00000000..2a404216 --- /dev/null +++ b/packages/harness/test/fixtures/mcp/tools.mjs @@ -0,0 +1,160 @@ +// Tools shared by the stdio fixture and the in-process streamable HTTP fixture. +import { spawn } from "node:child_process"; + +const PNG = "iVBORw0KGgo="; +const sleep = (ms, signal) => + new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + signal?.addEventListener("abort", () => { + clearTimeout(timer); + resolve(); + }); + }); +const text = (value) => ({ + content: [{ type: "text", text: typeof value === "string" ? value : JSON.stringify(value) }], +}); + +const tools = [ + "echo", + "mixed", + "structured", + "crash", + "progress", + "slow", + "hang", + "cancelled", + "roots", + "capabilities", + "spawn", + "frame", + "exit", + "huge", + "log", + "stderr-child", +] + .map((name) => ({ name, inputSchema: { type: "object" } })) + .concat( + ["typed-missing", "typed-mismatch"].map((name) => ({ + name, + inputSchema: { type: "object" }, + outputSchema: { type: "object", properties: { answer: { type: "number" } }, required: ["answer"] }, + })), + ); + +/** Installs the fixture tools on a low-level SDK `Server`. */ +export const install = (server) => { + const cancelled = []; + + let listed = false; + server.setRequestHandler("tools/list", async () => { + // MCP_FIXTURE_CHANGE_ON_LIST: the list changes while the first listing is answered. + if (process.env.MCP_FIXTURE_CHANGE_ON_LIST && !listed) await server.sendToolListChanged(); + listed = true; + return { tools }; + }); + server.setRequestHandler("tools/call", async (request, ctx) => { + const args = request.params.arguments ?? {}; + const signal = ctx.mcpReq.signal; + switch (request.params.name) { + case "echo": + return text(String(args.text)); + case "mixed": + return { + content: [ + { type: "text", text: "hello" }, + { type: "image", data: PNG, mimeType: "image/png" }, + { + type: "resource", + resource: { uri: "file:///notes.txt", mimeType: "text/plain", text: "embedded notes" }, + }, + { type: "resource", resource: { uri: "file:///pixel.png", mimeType: "image/png", blob: PNG } }, + { type: "resource", resource: { uri: "file:///data.bin", blob: "AAEC" } }, + { type: "audio", data: "AAAA", mimeType: "audio/wav" }, + { type: "resource_link", name: "readme", uri: "file:///README.md" }, + ], + }; + case "structured": + return { content: [], structuredContent: { answer: 42, items: ["a", "b"] } }; + case "crash": + process.stderr.write("fatal: fixture crashed on purpose\n"); + setTimeout(() => process.exit(3), 20); + return new Promise(() => {}); + case "progress": { + const token = request.params._meta?.progressToken; + const ms = Number(args.ms); + const every = Number(args.every); + for (let elapsed = 0, step = 1; elapsed < ms; elapsed += every, step++) { + await sleep(every, signal); + if (token !== undefined) + await ctx.mcpReq.notify({ + method: "notifications/progress", + params: { progressToken: token, progress: step }, + }); + } + return text("progress done"); + } + case "slow": + await sleep(Number(args.ms), signal); + return text("slow done"); + case "hang": { + // Acknowledges entering through progress and the cancellation through a log message. + const token = request.params._meta?.progressToken; + if (token !== undefined) + await ctx.mcpReq.notify({ + method: "notifications/progress", + params: { progressToken: token, progress: 1 }, + }); + await new Promise((resolve) => + signal.addEventListener("abort", () => { + cancelled.push(String(signal.reason?.message ?? signal.reason ?? "aborted")); + resolve(); + }), + ); + await server.sendLoggingMessage({ level: "info", data: "hang cancelled" }); + return text("hang aborted"); + } + case "cancelled": + return text({ cancelled: cancelled.length }); + case "roots": + return text(await server.listRoots()); + case "capabilities": + return text(server.getClientCapabilities() ?? {}); + case "spawn": { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + return text({ server: process.pid, grandchild: child.pid }); + } + case "frame": + // An unparsable line, then a CRLF-terminated notification the client must still see. + process.stdout.write("this is not json\n"); + process.stdout.write( + `${JSON.stringify({ jsonrpc: "2.0", method: "notifications/tools/list_changed" })}\r\n`, + ); + return text("framed"); + case "exit": + setTimeout(() => process.exit(0), 20); + return text("bye"); + case "huge": + // One 17 MiB line that never ends: more than the client's 16 MiB frame limit. + process.stdout.write("x".repeat(17 * 1024 * 1024)); + return new Promise(() => {}); + case "stderr-child": { + // The grandchild inherits only stderr; the server then exits on its own. + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + stdio: ["ignore", "ignore", "inherit"], + }); + setTimeout(() => process.exit(0), 50); + return text({ server: process.pid, grandchild: child.pid }); + } + case "log": + await server.sendLoggingMessage({ level: "info", data: "first line\nsecond line" }); + await server.sendLoggingMessage({ level: "warning", logger: "db", data: { query: "select 1", ms: 3 } }); + return text("logged"); + case "typed-missing": + return text("no structured content"); + case "typed-mismatch": + return { content: [{ type: "text", text: "wrong shape" }], structuredContent: { answer: "nope" } }; + default: + throw new Error(`unknown tool ${request.params.name}`); + } + }); +}; diff --git a/packages/harness/test/mcp.client.test.ts b/packages/harness/test/mcp.client.test.ts new file mode 100644 index 00000000..613bdd34 --- /dev/null +++ b/packages/harness/test/mcp.client.test.ts @@ -0,0 +1,957 @@ +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { LoggingMessageNotificationParams, OAuthClientProvider } from "@modelcontextprotocol/client"; +import { Deferred, Effect, Exit, Fiber, Option, Ref, Scope } from "effect"; +import { describe, expect, vi } from "vite-plus/test"; +import { McpClient } from "../src/mcp/client.ts"; +import { McpContent } from "../src/mcp/content.ts"; +import { McpLog } from "../src/mcp/log.ts"; +import { Process } from "../src/sandbox/utils/process.ts"; +import { listen, raw, sdk } from "./fixtures/mcp/http.ts"; +import { testEffect } from "./utils/effect.ts"; + +const { live: it } = testEffect(Process.host); + +const fixtures = fileURLToPath(new URL("./fixtures/mcp/", import.meta.url)); +const stdio = (fixture: string, env: Record = {}): McpClient.Stdio => ({ + _tag: "Stdio", + command: process.execPath, + args: [`${fixtures}${fixture}`], + env, + cwd: fixtures, +}); +const http = (url: URL): McpClient.Http => ({ _tag: "Http", url, headers: {} }); + +/** Each scenario's observable outcome, written to the artifact by the last test. */ +const summary: Record = {}; +/** Every key the scenarios record; the artifact test refuses to snapshot without all of them. */ +const scenarios = [ + "cancel", + "crash", + "failedConnectSession", + "frameLimit", + "framing", + "http400", + "httpClose", + "httpHappy", + "httpRetries", + "initializedTimeout", + "invalidEntry", + "logFile", + "logRotation", + "missingCwd", + "noToolsCapability", + "onClose", + "onLog", + "outputSchema", + "processGroup", + "progress", + "refused", + "resources", + "retry", + "retryCloseReason", + "roots", + "sessionExpired", + "spawnFailure", + "stderrGrandchild", + "stdioHappy", + "strictResources", + "stubborn", + "toolsPagination", + "unadvertisedListChanged", + "listChangedWhileListing", + "unauthorized", +]; + +const normalise = (message: string) => + message + .replaceAll(fixtures, "/") + .replaceAll(process.execPath, "") + .replace(/127\.0\.0\.1:\d+/g, "127.0.0.1:") + .replace(/\b\d+(\.\d+)?\s?ms\b/g, ""); + +const failure = (error: { readonly _tag: string; readonly message: string }) => ({ + tag: error._tag, + message: normalise(error.message), +}); + +const text = (result: { readonly content?: ReadonlyArray<{ readonly type: string; readonly text?: string }> }) => + result.content?.[0]?.text ?? ""; + +const alive = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +}; + +// Orphans are reaped by init shortly after their group dies. +const gone = (pids: ReadonlyArray) => + Effect.gen(function* () { + for (let attempt = 0; attempt < 50 && pids.some(alive); attempt++) yield* Effect.sleep("20 millis"); + return pids.filter(alive).length === 0; + }); + +// Independent of the client's own cleanup, so a regression there cannot leak processes past the test. +const reap = (pids: ReadonlyArray) => + Effect.sync(() => { + for (const pid of pids) + for (const target of [-pid, pid]) { + try { + process.kill(target, "SIGKILL"); + } catch { + // already gone + } + } + }); + +const logEntries: ReadonlyArray = [ + { level: "info", data: "first line\nsecond line" }, + { level: "warning", logger: "db", data: { query: "select 1", ms: 3 } }, +]; + +const tempDir = Effect.acquireRelease( + Effect.sync(() => mkdtempSync(join(tmpdir(), "codework-mcp-log-"))), + (dir) => Effect.sync(() => rmSync(dir, { recursive: true, force: true })), +); + +describe("mcp client", () => { + it( + "stdio: connects, lists tools and converts every content kind", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const tools = yield* connection.tools(); + const echo = yield* connection.callTool({ name: "echo", args: { text: "hi" } }); + const mixed = yield* connection.callTool({ name: "mixed" }); + const structured = yield* connection.callTool({ name: "structured" }); + summary.stdioHappy = { + serverInfo: connection.serverInfo, + instructions: connection.instructions, + tools: tools.map((tool) => tool.name), + echo: McpContent.fromResult(echo), + mixed: McpContent.fromResult(mixed), + structured: McpContent.fromResult(structured), + }; + expect(connection.instructions).toBe("Use the echo tool."); + expect(McpContent.fromResult(echo)).toEqual([{ type: "text", text: "hi" }]); + expect(McpContent.fromResult(mixed)).toEqual([ + { type: "text", text: "hello" }, + { type: "image", data: "iVBORw0KGgo=", mimeType: "image/png" }, + { type: "text", text: "embedded notes" }, + { type: "image", data: "iVBORw0KGgo=", mimeType: "image/png" }, + { type: "text", text: "[binary resource file:///data.bin (unknown type) omitted]" }, + { type: "text", text: "[audio audio/wav omitted]" }, + { type: "text", text: "readme: file:///README.md" }, + ]); + expect(McpContent.fromResult(structured)).toEqual([ + { type: "text", text: JSON.stringify({ answer: 42, items: ["a", "b"] }, null, 2) }, + ]); + }), + ); + + it( + "http: connects to a real streamable HTTP server and calls a tool", + Effect.gen(function* () { + const url = yield* listen(yield* Effect.promise(sdk)); + const connection = yield* McpClient.connect({ server: "http", endpoint: http(url) }); + const tools = yield* connection.tools(); + const echo = yield* connection.callTool({ name: "echo", args: { text: "over http" } }); + summary.httpHappy = { + serverInfo: connection.serverInfo, + tools: tools.map((tool) => tool.name), + echo: McpContent.fromResult(echo), + }; + expect(text(echo)).toBe("over http"); + }), + ); + + it( + "stdio: a crash mid-call fails the pending call with the exit code and stderr, and reports close", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const closed = yield* Deferred.make(); + connection.onClose((reason) => Deferred.succeed(closed, reason).pipe(Effect.asVoid)); + const error = yield* connection.callTool({ name: "crash" }).pipe(Effect.flip); + const reason = yield* Deferred.await(closed).pipe(Effect.timeout("2 seconds")); + summary.crash = { error: failure(error), method: "method" in error ? error.method : undefined, reason }; + expect(error._tag).toBe("Mcp.RequestError"); + expect(error.message).toContain("exited with code 3"); + expect(error.message).toContain("fatal: fixture crashed on purpose"); + expect(reason).toContain("connection closed"); + }), + ); + + it( + "stdio: closing the scope kills the server and its grandchild", + Effect.gen(function* () { + const outer = yield* Effect.scope; + const pids = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const spawned = JSON.parse(text(yield* connection.callTool({ name: "spawn" }))) as { + readonly server: number; + readonly grandchild: number; + }; + yield* Scope.addFinalizer(outer, reap([spawned.server, spawned.grandchild])); + expect(alive(spawned.server) && alive(spawned.grandchild)).toBe(true); + return [spawned.server, spawned.grandchild]; + }).pipe(Effect.scoped); + const killed = yield* gone(pids); + summary.processGroup = { killed }; + expect(killed).toBe(true); + }), + ); + + it( + "stdio: a server ignoring SIGTERM is still killed within a bounded time", + Effect.gen(function* () { + const outer = yield* Effect.scope; + const started = Date.now(); + const pids = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "stubborn", endpoint: stdio("stubborn.mjs") }); + const spawned = JSON.parse(connection.instructions ?? "{}") as { + readonly server: number; + readonly grandchild: number; + }; + yield* Scope.addFinalizer(outer, reap([spawned.server, spawned.grandchild])); + return [spawned.server, spawned.grandchild]; + }).pipe(Effect.scoped); + const elapsed = Date.now() - started; + const killed = yield* gone(pids); + summary.stubborn = { killed, boundedUnder8s: elapsed < 8_000 }; + expect(elapsed).toBeLessThan(8_000); + expect(killed).toBe(true); + }), + 15_000, + ); + + it( + "stdio: a grandchild holding only stderr is killed when the server exits, before scope close", + Effect.gen(function* () { + const outer = yield* Effect.scope; + const killed = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const closed = yield* Deferred.make(); + connection.onClose(() => Deferred.succeed(closed, undefined).pipe(Effect.asVoid)); + const spawned = JSON.parse(text(yield* connection.callTool({ name: "stderr-child" }))) as { + readonly server: number; + readonly grandchild: number; + }; + yield* Scope.addFinalizer(outer, reap([spawned.server, spawned.grandchild])); + // The server exits on its own; its grandchild keeps the stderr pipe open. The group is killed + // as soon as the connection notices the exit, while the connection's scope is still open: the + // spawner's own release at scope close would hide a missing kill. + yield* gone([spawned.server]); + yield* Deferred.await(closed).pipe(Effect.timeout("5 seconds")); + return yield* gone([spawned.grandchild]); + }).pipe(Effect.scoped); + summary.stderrGrandchild = { killed }; + expect(killed).toBe(true); + }), + ); + + it( + "stdio: skips an invalid JSON line and parses a CRLF-terminated message", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const changed = yield* Deferred.make(); + connection.onToolsChanged(Deferred.succeed(changed, undefined).pipe(Effect.asVoid)); + const framed = yield* connection.callTool({ name: "frame" }); + yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds")); + const after = yield* connection.callTool({ name: "echo", args: { text: "still alive" } }); + summary.framing = { framed: text(framed), toolsChanged: true, after: text(after) }; + expect(text(after)).toBe("still alive"); + }), + ); + + it( + "stdio: progress keeps a long call alive past the request timeout; silence times out", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "fixture", + endpoint: stdio("server.mjs"), + timeout: { request: "1 second" }, + }); + let updates = 0; + const progressed = yield* connection.callTool({ + name: "progress", + args: { ms: 2_000, every: 200 }, + onProgress: () => { + updates++; + }, + }); + const count = updates; + const silent = yield* connection.callTool({ name: "slow", args: { ms: 2_000 } }).pipe(Effect.flip); + summary.progress = { progressed: text(progressed), sawProgress: count > 0, silent: failure(silent) }; + expect(text(progressed)).toBe("progress done"); + expect(count).toBeGreaterThan(3); + expect(silent._tag).toBe("Mcp.RequestError"); + expect(silent.message).toContain("timed out"); + }), + ); + + it( + "stdio: interrupting a call cancels it on the server and keeps the connection usable", + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const acknowledged = yield* Deferred.make(); + const connection = yield* McpClient.connect({ + server: "fixture", + endpoint: stdio("server.mjs"), + onLog: (entry) => + entry.data === "hang cancelled" + ? Deferred.succeed(acknowledged, undefined).pipe(Effect.asVoid) + : Effect.void, + }); + const fiber = yield* connection + .callTool({ name: "hang", onProgress: () => void Deferred.doneUnsafe(entered, Effect.void) }) + .pipe(Effect.forkChild); + yield* Deferred.await(entered).pipe(Effect.timeout("2 seconds")); + yield* Fiber.interrupt(fiber); + yield* Deferred.await(acknowledged).pipe(Effect.timeout("2 seconds")); + const report = JSON.parse(text(yield* connection.callTool({ name: "cancelled" }))) as { + readonly cancelled: number; + }; + const echo = yield* connection.callTool({ name: "echo", args: { text: "after cancel" } }); + summary.cancel = { cancelled: report.cancelled, after: text(echo) }; + expect(report.cancelled).toBe(1); + expect(text(echo)).toBe("after cancel"); + }), + ); + + it( + "http: a transient 503 on initialize is retried", + Effect.gen(function* () { + const server = raw({ failInitialize: 1 }); + const url = yield* listen(server.handler); + const connection = yield* McpClient.connect({ server: "raw", endpoint: http(url) }); + const tools = yield* connection.tools(); + const initializes = server.seen.filter((entry) => entry === "POST initialize").length; + summary.retry = { initializes, tools: tools.map((tool) => tool.name) }; + expect(initializes).toBe(2); + }), + ); + + it( + "http: a 404 for a known session fails with SessionExpiredError and reports once", + Effect.gen(function* () { + const server = raw(); + const url = yield* listen(server.handler); + const expired = yield* Ref.make(0); + const outcome = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "raw", endpoint: http(url) }); + connection.onSessionExpired(Ref.update(expired, (count) => count + 1)); + server.forget = true; + const first = yield* connection.tools().pipe(Effect.flip); + const second = yield* connection.tools().pipe(Effect.flip); + return [failure(first), failure(second)]; + }).pipe(Effect.scoped); + yield* Effect.sleep("50 millis"); + const reported = yield* Ref.get(expired); + const terminated = server.seen.includes("DELETE"); + summary.sessionExpired = { errors: outcome, reported, terminated }; + expect(outcome.map((error) => error.tag)).toEqual(["Mcp.SessionExpiredError", "Mcp.SessionExpiredError"]); + expect(reported).toBe(1); + expect(terminated).toBe(false); + }), + ); + + it( + "stdio: roots are served as file URIs only when passed", + Effect.gen(function* () { + const withRoots = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "fixture", + endpoint: stdio("server.mjs"), + roots: ["/workspace/project", "/srv/data dir"], + }); + return { + roots: JSON.parse(text(yield* connection.callTool({ name: "roots" }))) as unknown, + capabilities: JSON.parse(text(yield* connection.callTool({ name: "capabilities" }))) as unknown, + }; + }).pipe(Effect.scoped); + const without = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + return JSON.parse(text(yield* connection.callTool({ name: "capabilities" }))) as Record; + }).pipe(Effect.scoped); + summary.roots = { withRoots, withoutRoots: { capabilities: without } }; + expect(withRoots.roots).toEqual({ + roots: [ + { uri: "file:///workspace/project", name: "project" }, + { uri: "file:///srv/data%20dir", name: "data dir" }, + ], + }); + expect(without).not.toHaveProperty("roots"); + }), + ); + + it( + "stdio: a missing command fails with ConnectError", + Effect.gen(function* () { + const error = yield* McpClient.connect({ + server: "missing", + endpoint: { ...stdio("server.mjs"), command: "codework-mcp-missing-binary" }, + timeout: { startup: "5 seconds" }, + }).pipe(Effect.flip); + summary.spawnFailure = failure(error); + expect(error._tag).toBe("Mcp.ConnectError"); + expect(error.message).toBe("command not found: codework-mcp-missing-binary"); + }), + ); + + it( + "onClose: silent on a deliberate scope close, immediate when the server already exited", + Effect.gen(function* () { + const deliberate = yield* Ref.make>([]); + yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + connection.onClose((reason) => Ref.update(deliberate, (reasons) => [...reasons, reason])); + yield* connection.tools(); + }).pipe(Effect.scoped); + yield* Effect.sleep("300 millis"); + const afterScope = yield* Ref.get(deliberate); + + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const spawned = JSON.parse(text(yield* connection.callTool({ name: "spawn" }))) as { readonly server: number }; + yield* connection.callTool({ name: "exit" }); + expect(yield* gone([spawned.server])).toBe(true); + // Requests fail once the transport is closed; then give the close a moment to settle. + for (let attempt = 0; attempt < 50; attempt++) { + if ((yield* connection.tools().pipe(Effect.exit))._tag === "Failure") break; + yield* Effect.sleep("20 millis"); + } + yield* Effect.sleep("300 millis"); + const late = yield* Deferred.make(); + connection.onClose((reason) => Deferred.succeed(late, reason).pipe(Effect.asVoid)); + const reason = yield* Deferred.await(late).pipe(Effect.timeout("500 millis")); + summary.onClose = { afterDeliberateClose: afterScope, afterServerExit: normalise(reason) }; + expect(afterScope).toEqual([]); + expect(reason).toContain("connection closed"); + }), + ); + + it( + "callTool: results are validated against the tool's output schema", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + // Listing caches the output schemas the SDK validates results against. + const tools = yield* connection.tools(); + const missing = yield* connection.callTool({ name: "typed-missing" }).pipe(Effect.flip); + const mismatch = yield* connection.callTool({ name: "typed-mismatch" }).pipe(Effect.flip); + summary.outputSchema = { + declared: tools.filter((tool) => tool.outputSchema !== undefined).map((tool) => tool.name), + missing: failure(missing), + mismatch: failure(mismatch), + }; + expect(missing._tag).toBe("Mcp.RequestError"); + expect(mismatch._tag).toBe("Mcp.RequestError"); + }), + ); + + it( + "http: a refused connection fails with ECONNREFUSED", + Effect.gen(function* () { + const port = yield* Effect.callback((resume) => { + const server = createServer(); + server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as AddressInfo; + server.close(() => resume(Effect.succeed(port))); + }); + }); + const error = yield* McpClient.connect({ + server: "refused", + endpoint: http(new URL(`http://127.0.0.1:${port}/mcp`)), + timeout: { startup: "5 seconds" }, + }).pipe(Effect.flip); + summary.refused = { error: failure(error) }; + expect(error._tag).toBe("Mcp.ConnectError"); + expect(error.message).toContain("ECONNREFUSED"); + }), + ); + + it( + "resources: paginated pages, missing templates handler, reads and list_changed", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "resources", + endpoint: stdio("server.mjs", { MCP_FIXTURE: "resources" }), + }); + const changed = yield* Deferred.make(); + connection.onResourcesChanged(Deferred.succeed(changed, undefined).pipe(Effect.asVoid)); + const templates = yield* connection.resourceTemplates(); + const first = yield* connection.resources(); + const second = yield* connection.resources(first.nextCursor); + const read = yield* connection.readResource("fixture://one"); + yield* connection.readResource("fixture://changed"); + yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds")); + summary.resources = { templates, first, second, read, resourcesChanged: true }; + expect(templates).toEqual({ resourceTemplates: [] }); + expect(first).toEqual({ resources: [{ uri: "fixture://one", name: "one" }], nextCursor: "page-2" }); + expect(second).toEqual({ resources: [{ uri: "fixture://changed", name: "changed" }] }); + expect(read.contents).toEqual([ + { uri: "fixture://one", mimeType: "text/plain", text: "contents of fixture://one" }, + ]); + }), + ); + + it( + "tools: a server without the tools capability lists none and prints nothing", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "resources", + endpoint: stdio("server.mjs", { MCP_FIXTURE: "resources" }), + }); + const write = vi.spyOn(process.stdout, "write"); + const debug = vi.spyOn(console, "debug"); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + write.mockRestore(); + debug.mockRestore(); + }), + ); + const tools = yield* connection.tools(); + // Restoring clears the recorded calls, so count first. + const printed = write.mock.calls.length + debug.mock.calls.length; + summary.noToolsCapability = { tools, printed }; + expect(tools).toEqual([]); + expect(printed).toBe(0); + }), + ); + + it( + "tools: a list_changed during the first listing reaches a handler set after it", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "fixture", + endpoint: stdio("server.mjs", { MCP_FIXTURE_CHANGE_ON_LIST: "1" }), + }); + yield* connection.tools(); + const changed = yield* Deferred.make(); + connection.onToolsChanged(Deferred.succeed(changed, undefined).pipe(Effect.asVoid)); + yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds")); + summary.listChangedWhileListing = { toolsChanged: true }; + }), + ); + + it( + "tools: list_changed from a server without tools.listChanged still reaches onToolsChanged", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "plain", + endpoint: stdio("server.mjs", { MCP_FIXTURE: "plain" }), + }); + const changed = yield* Deferred.make(); + connection.onToolsChanged(Deferred.succeed(changed, undefined).pipe(Effect.asVoid)); + yield* connection.callTool({ name: "frame" }); + yield* Deferred.await(changed).pipe(Effect.timeout("2 seconds")); + summary.unadvertisedListChanged = { listChanged: connection.capabilities?.tools, toolsChanged: true }; + expect(connection.capabilities?.tools).toEqual({}); + }), + ); + + it( + "stdio: a frame over 16 MiB fails the pending call and explains the close", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "fixture", endpoint: stdio("server.mjs") }); + const closed = yield* Deferred.make(); + connection.onClose((reason) => Deferred.succeed(closed, reason).pipe(Effect.asVoid)); + const error = yield* connection.callTool({ name: "huge" }).pipe(Effect.flip); + const reason = yield* Deferred.await(closed).pipe(Effect.timeout("5 seconds")); + summary.frameLimit = { error: failure(error), reason: normalise(reason) }; + expect(error._tag).toBe("Mcp.RequestError"); + expect(error.message).toContain("frame exceeded 16 MiB"); + expect(error.message).not.toContain("process exited"); + expect(reason).toContain("frame exceeded 16 MiB"); + }), + 15_000, + ); + + it( + "http: a 401 needs auth with an auth provider and is a ConnectError without one", + Effect.gen(function* () { + const server = raw({ unauthorized: true }); + const url = yield* listen(server.handler); + const redirects: string[] = []; + const authProvider: OAuthClientProvider = { + redirectUrl: "http://127.0.0.1/callback", + clientMetadata: { redirect_uris: ["http://127.0.0.1/callback"], client_name: "codework-test" }, + clientInformation: () => ({ client_id: "codework-test" }), + tokens: () => undefined, + saveTokens: () => {}, + redirectToAuthorization: (authorizationUrl) => { + redirects.push(authorizationUrl.pathname); + }, + saveCodeVerifier: () => {}, + codeVerifier: () => "verifier", + }; + const withProvider = yield* McpClient.connect({ server: "oauth", endpoint: http(url), authProvider }).pipe( + Effect.flip, + ); + const without = yield* McpClient.connect({ server: "oauth", endpoint: http(url) }).pipe(Effect.flip); + summary.unauthorized = { withProvider: failure(withProvider), redirects, without: failure(without) }; + expect(withProvider._tag).toBe("Mcp.NeedsAuthError"); + expect(without._tag).toBe("Mcp.ConnectError"); + expect(without.message).toContain("HTTP 401"); + }), + ); + + it( + "http: a retried connect leaves no stale close reason; a deliberate close stays silent", + Effect.gen(function* () { + const server = raw({ failInitialize: 1 }); + const url = yield* listen(server.handler); + const reasons = yield* Ref.make>([]); + const connected = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ server: "raw", endpoint: http(url) }); + connection.onClose((reason) => Ref.update(reasons, (all) => [...all, reason])); + yield* Effect.sleep("300 millis"); + const afterRegister = yield* Ref.get(reasons); + const tools = yield* connection.tools(); + return { afterRegister, tools: tools.map((tool) => tool.name) }; + }).pipe(Effect.scoped); + yield* Effect.sleep("300 millis"); + const afterScope = yield* Ref.get(reasons); + const initializes = server.seen.filter((entry) => entry === "POST initialize").length; + summary.retryCloseReason = { initializes, ...connected, afterScope }; + expect(initializes).toBe(2); + expect(connected.afterRegister).toEqual([]); + expect(afterScope).toEqual([]); + }), + ); + + it( + "stdio: a missing working directory fails with ConnectError", + Effect.gen(function* () { + const error = yield* McpClient.connect({ + server: "missing-cwd", + endpoint: { ...stdio("server.mjs"), cwd: `${fixtures}missing-dir` }, + timeout: { startup: "5 seconds" }, + }).pipe(Effect.flip); + summary.missingCwd = failure(error); + expect(error._tag).toBe("Mcp.ConnectError"); + expect(error.message).toBe(`working directory not found: ${fixtures}missing-dir`); + }), + ); + + it( + "tools/list: strict pagination for null, empty and repeated cursors", + Effect.gen(function* () { + const list = (variant: string) => + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "malformed", + endpoint: stdio("malformed.mjs", { MCP_MALFORMED: variant }), + timeout: { request: "5 seconds" }, + }); + const started = Date.now(); + const exit = yield* connection.tools().pipe(Effect.exit); + return { exit, elapsed: Date.now() - started }; + }).pipe(Effect.scoped); + const outcome = ({ + exit, + }: { + readonly exit: Exit.Exit< + ReadonlyArray<{ readonly name: string }>, + { readonly _tag: string; readonly message: string } + >; + }) => + Exit.isSuccess(exit) + ? { tools: exit.value.map((tool) => tool.name) } + : { error: Exit.findErrorOption(exit).pipe(Option.map(failure), Option.getOrUndefined) }; + const nullCursor = yield* list("null"); + const emptyCursor = yield* list("empty"); + const duplicate = yield* list("duplicate"); + summary.toolsPagination = { + nullCursor: outcome(nullCursor), + emptyCursor: outcome(emptyCursor), + duplicate: outcome(duplicate), + }; + expect(outcome(nullCursor).error).toEqual({ + tag: "Mcp.RequestError", + message: "Invalid result for tools/list: nextCursor: Invalid input: expected string, received null", + }); + // An empty cursor is followed as a real one, so the pages cycle until the SDK's cap. + expect(outcome(emptyCursor).error?.message).toContain("exceeded listMaxPages"); + // The SDK stops quietly at a repeated cursor and returns the pages it has. + expect(outcome(duplicate)).toEqual({ tools: ["t-start", "t-again"] }); + for (const run of [emptyCursor, duplicate]) expect(run.elapsed).toBeLessThan(5_000); + }), + ); + + it( + "resources: well-formed entries keep extra fields; entries without a name fail", + Effect.gen(function* () { + const valid = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "malformed", + endpoint: stdio("malformed.mjs", { MCP_MALFORMED: "resources" }), + }); + return { resources: yield* connection.resources(), templates: yield* connection.resourceTemplates() }; + }).pipe(Effect.scoped); + const unnamed = yield* Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "malformed", + endpoint: stdio("malformed.mjs", { MCP_MALFORMED: "unnamed" }), + }); + return { + resources: yield* connection.resources().pipe(Effect.flip), + templates: yield* connection.resourceTemplates().pipe(Effect.flip), + }; + }).pipe(Effect.scoped); + summary.strictResources = { + valid, + unnamed: { resources: failure(unnamed.resources), templates: failure(unnamed.templates) }, + }; + expect(valid.resources.resources).toEqual([ + { + uri: "file:///a.txt", + name: "a", + description: "A file", + mimeType: "text/plain", + _meta: { origin: "fixture" }, + }, + ]); + expect(valid.templates.resourceTemplates).toEqual([ + { uriTemplate: "file:///{path}", name: "files", description: "Any file" }, + ]); + expect(failure(unnamed.resources)).toEqual({ + tag: "Mcp.RequestError", + message: + "Invalid result for resources/list: resources.0.name: Invalid input: expected string, received undefined", + }); + expect(failure(unnamed.templates)).toEqual({ + tag: "Mcp.RequestError", + message: + "Invalid result for resources/templates/list: resourceTemplates.0.name: Invalid input: expected string, received undefined", + }); + }), + ); + + it( + "tools/list: an invalid entry fails with RequestError", + Effect.gen(function* () { + const connection = yield* McpClient.connect({ + server: "malformed", + endpoint: stdio("malformed.mjs", { MCP_MALFORMED: "invalid" }), + }); + const error = yield* connection.tools().pipe(Effect.flip); + summary.invalidEntry = failure(error); + expect(failure(error)).toEqual({ + tag: "Mcp.RequestError", + message: + "Invalid result for tools/list: tools.1.inputSchema: Invalid input: expected object, received undefined", + }); + }), + ); + + it( + "onLog: receives the server's notifications/message params, from the handshake on", + Effect.gen(function* () { + const entries = yield* Ref.make>([]); + const connection = yield* McpClient.connect({ + server: "fixture", + endpoint: stdio("server.mjs"), + onLog: (entry) => Ref.update(entries, (all) => [...all, entry]), + }); + yield* connection.callTool({ name: "log" }); + for (let attempt = 0; attempt < 50 && (yield* Ref.get(entries)).length < 3; attempt++) + yield* Effect.sleep("20 millis"); + const received = yield* Ref.get(entries); + summary.onLog = received; + // The handshake's own message arrives before any the client asked for. + expect(received).toEqual([{ level: "info", logger: "startup", data: "ready" }, ...logEntries]); + }), + ); + + it( + "McpLog: writes formatted lines with indented continuations", + Effect.gen(function* () { + const dir = yield* tempDir; + const log = yield* McpLog.make(join(dir, "logs", "mcp.log")); + for (const entry of logEntries) yield* log.write("fixture", entry); + const lines = readFileSync(log.path, "utf8").replace(/^\d{4}-\d\d-\d\dT[\d:.]+Z/gm, ""); + summary.logFile = lines.split("\n"); + expect(lines).toBe( + [ + " [fixture] info first line", + " second line", + ' [fixture] warning db: {"query":"select 1","ms":3}', + "", + ].join("\n"), + ); + }), + ); + + it( + "McpLog: rotates a file over 5 MB to .1", + Effect.gen(function* () { + const dir = yield* tempDir; + const path = join(dir, "mcp.log"); + const old = `${"x".repeat(5 * 1024 * 1024 + 16)}\n`; + writeFileSync(path, old); + const log = yield* McpLog.make(path); + yield* log.write("fixture", logEntries[1]!); + const rotated = readFileSync(`${path}.1`, "utf8") === old; + const current = readFileSync(path, "utf8").replace(/^\d{4}-\d\d-\d\dT[\d:.]+Z/gm, ""); + summary.logRotation = { rotated, current }; + expect(rotated).toBe(true); + expect(current).toBe(' [fixture] warning db: {"query":"select 1","ms":3}\n'); + }), + ); + + it( + "http: an unanswered initialized notification times the connect out and ends the session", + Effect.gen(function* () { + const server = raw({ + script: (request) => (request === "POST notifications/initialized" ? "hang" : undefined), + }); + const url = yield* listen(server.handler); + const started = Date.now(); + const error = yield* McpClient.connect({ + server: "raw", + endpoint: http(url), + timeout: { startup: "500 millis" }, + }).pipe(Effect.flip); + const elapsed = Date.now() - started; + summary.initializedTimeout = { error: failure(error), deleted: server.seen.includes("DELETE") }; + expect(error._tag).toBe("Mcp.ConnectError"); + expect(error.message).toContain("mcp connect timed out after"); + expect(elapsed).toBeLessThan(2_000); + }), + ); + + it( + "http: a failed connect attempt deletes the session it was given", + Effect.gen(function* () { + const unsupported = raw({ protocolVersion: "1999-01-01" }); + const failing = raw({ + script: (request) => + request === "POST notifications/initialized" ? { status: 500, body: "notification failed" } : undefined, + }); + const errors = []; + for (const server of [unsupported, failing]) { + const url = yield* listen(server.handler); + const error = yield* McpClient.connect({ + server: "raw", + endpoint: http(url), + timeout: { startup: "2 seconds" }, + }).pipe(Effect.flip); + errors.push({ error: failure(error), seen: [...server.seen].filter((entry) => entry !== "GET") }); + } + summary.failedConnectSession = { unsupported: errors[0], failedNotification: errors[1] }; + expect(errors.map(({ error, seen }) => ({ tag: error.tag, deleted: seen.includes("DELETE") }))).toEqual([ + { tag: "Mcp.ConnectError", deleted: true }, + { tag: "Mcp.ConnectError", deleted: true }, + ]); + }), + ); + + it( + "http: list requests retry a transient 503; tool calls never retry", + Effect.gen(function* () { + let listFailures = 1; + const server = raw({ + script: (request) => { + if (request === "POST tools/list" && listFailures-- > 0) return { status: 503, body: "overloaded" }; + // The call has run its side effect before the server fails the response. + if (request === "POST tools/call") return { status: 503, body: "overloaded" }; + return undefined; + }, + }); + const url = yield* listen(server.handler); + const connection = yield* McpClient.connect({ server: "raw", endpoint: http(url) }); + const tools = yield* connection.tools(); + const call = yield* connection.callTool({ name: "echo" }).pipe(Effect.flip); + const count = (key: string) => server.seen.filter((entry) => entry === key).length; + summary.httpRetries = { + tools: tools.map((tool) => tool.name), + listRequests: count("POST tools/list"), + call: failure(call), + callRequests: count("POST tools/call"), + }; + expect(count("POST tools/list")).toBe(2); + expect(call._tag).toBe("Mcp.RequestError"); + expect(count("POST tools/call")).toBe(1); + }), + ); + + it( + "http: closing the scope deletes the session once, and a hanging DELETE is bounded", + Effect.gen(function* () { + const normal = raw(); + const normalUrl = yield* listen(normal.handler); + yield* McpClient.connect({ server: "raw", endpoint: http(normalUrl) }).pipe(Effect.scoped); + const deletes = normal.seen.filter((entry) => entry === "DELETE").length; + + const hanging = raw({ script: (request) => (request === "DELETE" ? "hang" : undefined) }); + const hangingUrl = yield* listen(hanging.handler); + let closedIn = 0; + yield* Effect.gen(function* () { + yield* McpClient.connect({ server: "raw", endpoint: http(hangingUrl) }); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closedIn = Date.now(); + }), + ); + }).pipe(Effect.scoped); + // Finalizers run in reverse: the one above runs first, so this times the client's close. + const elapsed = Date.now() - closedIn; + summary.httpClose = { deletes, hangingDeleteBounded: elapsed < 1_500 }; + expect(deletes).toBe(1); + expect(hanging.seen).toContain("DELETE"); + expect(elapsed).toBeLessThan(1_500); + }), + ); + + it( + "http: a 400 'Server not initialized' expires the session; any other 400 is a request error", + Effect.gen(function* () { + const reply = (body: string) => { + let armed = false; + const server = raw({ + script: (request) => (armed && request === "POST tools/list" ? { status: 400, body } : undefined), + }); + return { server, arm: () => (armed = true) }; + }; + const run = (body: string) => + Effect.gen(function* () { + const { server, arm } = reply(body); + const url = yield* listen(server.handler); + const connection = yield* McpClient.connect({ server: "raw", endpoint: http(url) }); + arm(); + return failure(yield* connection.tools().pipe(Effect.flip)); + }).pipe(Effect.scoped); + const notInitialized = yield* run( + JSON.stringify({ + jsonrpc: "2.0", + error: { code: -32000, message: "Bad Request: Server not initialized" }, + id: null, + }), + ); + const other = yield* run( + JSON.stringify({ + jsonrpc: "2.0", + error: { code: -32000, message: "Bad Request: something else" }, + id: null, + }), + ); + summary.http400 = { notInitialized, other }; + expect(notInitialized.tag).toBe("Mcp.SessionExpiredError"); + expect(other.tag).toBe("Mcp.RequestError"); + }), + ); + + it( + "writes the scenario artifact", + Effect.gen(function* () { + // A filtered run must not snapshot a partial summary. + expect(Object.keys(summary).sort()).toEqual([...scenarios].sort()); + yield* Effect.promise(() => expect(summary).toMatchFileSnapshot("./__artifacts__/mcp.client.json")); + }), + ); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 35980607..423b4e50 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -608,6 +608,9 @@ importers: '@effect/sql-sqlite-node': specifier: 4.0.5 version: 4.0.5(effect@4.0.5) + '@modelcontextprotocol/client': + specifier: 2.3.1 + version: 2.3.1 '@npmcli/arborist': specifier: ^10.0.2 version: 10.0.4(supports-color@7.2.0) @@ -651,6 +654,9 @@ importers: '@daytona/sdk': specifier: ^0.211.2 version: 0.211.2(debug@4.4.3(supports-color@7.2.0))(supports-color@7.2.0) + '@modelcontextprotocol/server': + specifier: 2.3.1 + version: 2.3.1 '@types/node': specifier: 'catalog:' version: 26.6.5 @@ -2098,14 +2104,26 @@ packages: '@mixmark-io/domino@2.2.0': resolution: {integrity: sha512-Y28PR25bHXUg88kCV7nivXrP2Nj2RueZ3/l/jdx6J9f8J4nsEGcgX0Qe6lt7Pa+J79+kPiJU3LguR6O/6zrLOw==} + '@modelcontextprotocol/client@2.3.1': + resolution: {integrity: sha512-mIGZXpHsjnZ6lD+gD/WCMpR5k8yVQQ8nNFH1N0Srf7AvnwTUMYD6pvTY8ng2+He5FfV7YMZLmrjL7cLa/cc3dQ==} + engines: {node: '>=20'} + '@modelcontextprotocol/core@2.0.0': resolution: {integrity: sha512-pJCEwGG7Lfr/+PQp9ZTwKXNeO5wzbfKL7H3MYpCorM4oFBoQrdjnBgEoqG+RjhsvS1FKrDbKux+M1HhlnGWqcA==} engines: {node: '>=20'} + '@modelcontextprotocol/core@2.3.1': + resolution: {integrity: sha512-laVmIhPGpWi7rG9q+0UigblRN0Hws/Yjlb6qLFtVqryXVLkc4h+yN8z731POD2LnAGQlFe7/tXhUm7Bi0YB91g==} + engines: {node: '>=20'} + '@modelcontextprotocol/server@2.0.0': resolution: {integrity: sha512-YhHWdHfpFMQfd0prsEnxKeS3Qz3ytIGmsS0sth4KDjnacIT7hxk6hXHkJ9KysxlkvTM+WZAtQbbcUhdoP4Hvtw==} engines: {node: '>=20'} + '@modelcontextprotocol/server@2.3.1': + resolution: {integrity: sha512-e59MfWuSssj6DoQ96JCTqdW1AXT5/XtnTl0WWd6MU1Ka434fFUGGosCLJbGNySDhHGvxQWoRWK137FXYFkYNWw==} + engines: {node: '>=20'} + '@mongodb-js/zstd@7.0.0': resolution: {integrity: sha512-mQ2s0pYYiav+tzCDR05Zptem8Ey2v8s11lri5RKGhTtL4COVCvVCk5vtyRYNT+9L8qSfyOqqefF9UtnW8mC5jA==} engines: {node: '>= 20.19.0'} @@ -5810,6 +5828,10 @@ packages: resolution: {integrity: sha512-N9b0EvATDJfZj+Szg3m8WwqseQhei/mtTd6V0rCciGbI/yG24o3yBk0YVyfOZZsLVMZHWs++Zk3Aqd2y2EztnQ==} engines: {node: '>=22.12'} + eventsource@3.0.7: + resolution: {integrity: sha512-CRT1WTyuQoD771GW56XEZFQ/ZoSfWid1alKGDYMmkt2yl8UXrVR4pspqWNEcqKvVIzg6PAltWjxcSSPrboA4iA==} + engines: {node: '>=18.0.0'} + execa@5.1.1: resolution: {integrity: sha512-8uSpZZocAZRBAPIEINJj3Lo9HyGitllczc27Eh5YYojjMFMn8yHMDMaUHE2Jqfq05D/wucwI4JGURyXt1vchyg==} engines: {node: '>=10'} @@ -7369,6 +7391,10 @@ packages: resolution: {integrity: sha512-xhcb4yHu9sM/G7foGzoLtXYcC0zHEaOXXjRKhGup0fw78Nf2Tkiapv4EQyMzrbcmQPsllAI7DbFY2UT7PlI9Pg==} hasBin: true + pkce-challenge@5.0.1: + resolution: {integrity: sha512-wQ0b/W4Fr01qtpHlqSqspcj3EhBvimsdh0KlHhH8HRZnMsEa0ea2fTULOXOS9ccQr3om+GcGRk4e+isrZWV8qQ==} + engines: {node: '>=16.20.0'} + pkijs@3.4.1: resolution: {integrity: sha512-Oo/NZcSWccq8KyoG7gLE9fnltgHns+pNCjCAp/WmjsUySi+sX7y4z4Xqu4fVb42CDHzRPl33fjzT15V1wvcyhA==} engines: {node: '>=16.0.0'} @@ -10383,15 +10409,34 @@ snapshots: '@mixmark-io/domino@2.2.0': {} + '@modelcontextprotocol/client@2.3.1': + dependencies: + '@modelcontextprotocol/core': 2.3.1 + cross-spawn: 7.0.6 + eventsource: 3.0.7 + eventsource-parser: 3.1.1 + jose: 6.2.3 + pkce-challenge: 5.0.1 + zod: 4.6.5 + '@modelcontextprotocol/core@2.0.0': dependencies: zod: 4.6.5 + '@modelcontextprotocol/core@2.3.1': + dependencies: + zod: 4.6.5 + '@modelcontextprotocol/server@2.0.0': dependencies: '@modelcontextprotocol/core': 2.0.0 zod: 4.6.5 + '@modelcontextprotocol/server@2.3.1': + dependencies: + '@modelcontextprotocol/core': 2.3.1 + zod: 4.6.5 + '@mongodb-js/zstd@7.0.0': dependencies: node-addon-api: 8.9.2 @@ -14089,6 +14134,10 @@ snapshots: eventsource-parser@4.1.2: {} + eventsource@3.0.7: + dependencies: + eventsource-parser: 3.1.1 + execa@5.1.1: dependencies: cross-spawn: 7.0.6 @@ -16043,6 +16092,8 @@ snapshots: dependencies: pngjs: 7.0.0 + pkce-challenge@5.0.1: {} + pkijs@3.4.1: dependencies: '@noble/hashes': 1.8.0