diff --git a/packages/harness/src/effect/harness.ts b/packages/harness/src/effect/harness.ts index a9dd1878..173d7d83 100644 --- a/packages/harness/src/effect/harness.ts +++ b/packages/harness/src/effect/harness.ts @@ -13,6 +13,7 @@ import { anchor } from "../plugin/loader.ts"; import { PluginSource } from "../plugin/source.ts"; import { PluginStore } from "../plugin/store.ts"; import { RunnerExecute } from "../runner/execute.ts"; +import { Mcp } from "../mcp/mcp.ts"; import { LLM } from "../runner/llm.ts"; import { Loop } from "../runner/loop.ts"; import { SandboxController } from "../sandbox/control.ts"; @@ -21,6 +22,7 @@ import { MemorySandboxDriver } from "../sandbox/drivers/memory.ts"; import { SqldbSandboxDriver } from "../sandbox/drivers/sqldb.ts"; import { SandboxDriverLoader } from "../sandbox/loader.ts"; import { SandboxDriverRegistry } from "../sandbox/registry.ts"; +import { Process } from "../sandbox/utils/process.ts"; import { SessionLive } from "../session/live.ts"; import { SessionRuntime } from "../session/runtime.ts"; import type { Info as SettingsInfo } from "../settings/schema.ts"; @@ -253,6 +255,7 @@ export const layer = (options: Options = {}) => return Control.layer.pipe( Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(loop))), Layer.provideMerge(State.layer({}, pool, references, followStore, rebuild, viewOf)), + Layer.provideMerge(Mcp.layer({ hostCwd, home: paths.home }).pipe(Layer.provide(Process.host))), Layer.provideMerge(Settings.layer(settingsOptions)), Layer.provideMerge(SessionRuntime.layer), Layer.provideMerge(sandboxes), diff --git a/packages/harness/src/mcp/client.ts b/packages/harness/src/mcp/client.ts index abd08ea1..7ce84607 100644 --- a/packages/harness/src/mcp/client.ts +++ b/packages/harness/src/mcp/client.ts @@ -77,6 +77,8 @@ export interface Connection { * 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; + /** Why the connection dropped, as soon as it has, even before `onClose` handlers run. */ + readonly dropped: () => string | undefined; } export interface ConnectInput { @@ -477,6 +479,7 @@ export const connect = Effect.fn("McpClient.connect")(function* (input: ConnectI closed.handler = handler; if (closed.reason !== undefined) runFork(handler(closed.reason)); }, + dropped: () => closed.reason, }; return connection; }); diff --git a/packages/harness/src/mcp/config.ts b/packages/harness/src/mcp/config.ts new file mode 100644 index 00000000..fad45097 --- /dev/null +++ b/packages/harness/src/mcp/config.ts @@ -0,0 +1,153 @@ +/** + * MCP server configuration: `mcp.json` in the user config directory and in the project's + * `.codework/`, in the `mcpServers` shape other MCP clients use, so an existing entry can be copied + * over. A project entry replaces a user entry with the same name. + * + * A project entry without `command`, `url` or `type` overrides only `enabled`, `exposure` and + * `toolExposure` of the user server with the same name, for example to turn it off in one project: + * `{ "mcpServers": { "internal-tools": { "enabled": false } } }`. The rest of the user entry is kept, + * including credentials the project could not set itself. + * + * Unlike `settings.jsonc`, a broken file or entry does not fail the read: it is reported and + * skipped, so one bad server never keeps the others from connecting. + */ + +import type { ServerConfig } from "@codeworksh/plugin/plugin/mcp"; +import type { McpServer, McpSettings } from "@codeworksh/plugin/settings"; +import { Effect, Predicate, Result } from "effect"; +import { Global } from "../global.ts"; +import { fileSystem, hostPath } from "../host.ts"; +import * as Jsonc from "../util/jsonc.ts"; + +const SERVER_NAME = /^[A-Za-z0-9_-]+$/; +const EXPOSURES: ReadonlyArray = ["direct", "hidden"]; +const OVERRIDE_KEYS: ReadonlyArray = ["enabled", "exposure", "toolExposure"]; + +/** Names that differ only in `-` and `_` would give their tools the same names. */ +export const namespace = (server: string) => `mcp__${server.replace(/-/g, "_")}`; + +const isRecord = (value: unknown): value is Record => + Predicate.isObject(value) && !Array.isArray(value); +const isStringRecord = (value: unknown): value is Record => + isRecord(value) && Object.values(value).every(Predicate.isString); + +/** Whether a project entry overrides a user server instead of defining one. */ +const isOverride = (value: Record) => + value.command === undefined && value.url === undefined && value.type === undefined; + +/** One `mcpServers` entry, checked. Returns the config, or what is wrong with it. */ +export const validate = (name: string, value: unknown): ServerConfig | string => { + if (!SERVER_NAME.test(name)) return `invalid server name "${name}" (use letters, digits, "_" and "-")`; + if (!isRecord(value)) return `server "${name}" must be an object`; + const { type, exposure, enabled, timeout, toolExposure, description } = value; + const exposures = EXPOSURES.map((one) => `"${one}"`).join(", "); + if (exposure !== undefined && !EXPOSURES.includes(exposure as string)) + return `server "${name}": exposure must be one of ${exposures}`; + if (toolExposure !== undefined) { + if (!isRecord(toolExposure)) return `server "${name}": toolExposure must map tool names to exposures`; + for (const [tool, one] of Object.entries(toolExposure)) { + if (!EXPOSURES.includes(one as string)) + return `server "${name}": toolExposure "${tool}" must be one of ${exposures}`; + } + } + if (enabled !== undefined && !Predicate.isBoolean(enabled)) return `server "${name}": enabled must be a boolean`; + if (description !== undefined && !Predicate.isString(description)) + return `server "${name}": description must be a string`; + if (timeout !== undefined && !(Predicate.isNumber(timeout) && timeout > 0)) + return `server "${name}": timeout must be a positive number of seconds`; + // Until sign-in exists, an entry relying on it could only fail; say so instead of ignoring the key. + for (const key of ["auth", "oauth"]) + if (value[key] !== undefined) return `server "${name}": "${key}" is not supported yet`; + if (type === "sse") return `server "${name}": legacy SSE transport is not supported; use the streamable HTTP URL`; + + if (Predicate.isString(value.url) && (type === undefined || type === "http" || type === "streamable-http")) { + if (!URL.canParse(value.url) || !/^https?:$/.test(new URL(value.url).protocol)) + return `server "${name}": url must be an http or https URL`; + if (value.headers !== undefined && !isStringRecord(value.headers)) + return `server "${name}": headers must map names to strings`; + return value as unknown as ServerConfig; + } + if (Predicate.isString(value.command) && (type === undefined || type === "stdio")) { + if (value.args !== undefined && !(Array.isArray(value.args) && value.args.every(Predicate.isString))) + return `server "${name}": args must be an array of strings`; + if (value.env !== undefined && !isStringRecord(value.env)) + return `server "${name}": env must map names to strings`; + if (value.cwd !== undefined && !Predicate.isString(value.cwd)) return `server "${name}": cwd must be a string`; + return value as unknown as ServerConfig; + } + return `server "${name}" needs either "command" (stdio) or "url" (streamable HTTP)`; +}; + +/** The server whose name differs from `name` only in `-` and `_`, if there is one. */ +export const clash = (names: Iterable, name: string) => { + for (const other of names) if (other !== name && namespace(other) === namespace(name)) return other; + return undefined; +}; + +interface State { + readonly servers: Map; + readonly errors: string[]; +} + +const read = Effect.fnUntraced(function* (path: string, scope: "user" | "project", state: State) { + const source = yield* Effect.result(fileSystem.readFileString(path)); + if (Result.isFailure(source)) { + const reason = source.failure.reason._tag; + if (reason !== "NotFound") state.errors.push(`${path}: cannot be read: ${reason}`); + return; + } + const document = Jsonc.parse(source.success); + if ("error" in document) { + state.errors.push(`${path}: ${document.error}`); + return; + } + const parsed = document.value; + if (!isRecord(parsed) || (parsed.mcpServers !== undefined && !isRecord(parsed.mcpServers))) { + state.errors.push(`${path}: expected an object with an "mcpServers" object`); + return; + } + const { servers, errors: problems } = state; + for (const [name, value] of Object.entries(parsed.mcpServers ?? {})) { + if (scope === "project" && isRecord(value) && isOverride(value)) { + const base = servers.get(name); + const extra = Object.keys(value).filter((key) => !OVERRIDE_KEYS.includes(key)); + if (!base) problems.push(`${path}: server "${name}" needs "command" or "url", or a user server to override`); + else if (extra.length > 0) + problems.push(`${path}: server "${name}": an override can only set ${OVERRIDE_KEYS.join(", ")}`); + else { + const config = validate(name, { ...base.config, ...value }); + if (Predicate.isString(config)) problems.push(`${path}: ${config}`); + else servers.set(name, { ...base, config, override: path }); + } + continue; + } + const config = validate(name, value); + if (Predicate.isString(config)) { + problems.push(`${path}: ${config}`); + continue; + } + const other = clash(servers.keys(), name); + if (other !== undefined) { + problems.push(`${path}: server "${name}" conflicts with "${other}"`); + continue; + } + servers.set(name, { name, config, file: path }); + } +}); + +/** + * Both layers. `userDir` is where the user `settings.jsonc` is read from; `root` is the session's + * project root, absent for a session with none. + */ +export const load = Effect.fn("Mcp.loadConfig")(function* (input: { + readonly userDir: string; + readonly root?: string | undefined; +}) { + const state: State = { servers: new Map(), errors: [] }; + yield* read(hostPath.join(input.userDir, "mcp.json"), "user", state); + if (input.root !== undefined) + yield* read(hostPath.join(input.root, Global.appConfigDir, "mcp.json"), "project", state); + return { servers: [...state.servers.values()], errors: state.errors } satisfies McpSettings; +}); + +export * as McpConfig from "./config.ts"; diff --git a/packages/harness/src/mcp/mcp.ts b/packages/harness/src/mcp/mcp.ts new file mode 100644 index 00000000..0fc27170 --- /dev/null +++ b/packages/harness/src/mcp/mcp.ts @@ -0,0 +1,497 @@ +/** + * The MCP servers each session runs, kept connected across its exchanges. + * + * Every exchange hands over the session's servers as its `mcp` plugins left them ({@link + * Interface.reconcile}); servers whose connection settings changed restart, removed ones close, new + * ones connect in the background. The tools an exchange registers are the ones the servers offered + * at that moment, frozen with the rest of the exchange; the connection behind a tool is looked up + * when it is called: + * + * - A connection that drops fails the calls running on it, and the next call reconnects. + * - A Streamable HTTP session the server forgot is replaced, and calls that hit it run once more on + * the new one. The old connection closes when its last call finishes, never under one. + * - A tool call is never retried after the server may have run it. + * - A changed tool list takes effect at the next exchange; a call to a withdrawn tool says so. + * + * Each connection lives in its own scope under the service's, so closing it, or shutting the + * service down, stops the server process and ends the HTTP session. + */ + +import type { ServerConfig } from "@codeworksh/plugin/plugin/mcp"; +import type { CallToolResult, Tool } from "@modelcontextprotocol/client"; +import { + Cause, + Context, + Deferred, + Duration, + Effect, + Exit, + Fiber, + FiberSet, + Layer, + Schema, + Scope, + Semaphore, +} from "effect"; +import type { ChildProcessSpawner } from "effect/process"; +import { NeedsAuthError } from "./error.ts"; +import type { ID as SessionId } from "../session/schema.ts"; +import type { RegisteredTool } from "../tool/tool.ts"; +import { McpClient } from "./client.ts"; +import { McpLog } from "./log.ts"; +import { McpTool } from "./tool.ts"; +import { McpValue } from "./value.ts"; +import { hostPath } from "../host.ts"; +import { expandTilde } from "../util/home.ts"; + +/** An exchange waits this long for the servers it started whose tools are declared to the model. */ +const STARTUP_WAIT = Duration.seconds(10); +const DEFAULT_TIMEOUT_SECONDS = 60; + +export type State = "connecting" | "connected" | "disconnected" | "failed" | "needs-auth" | "disabled"; + +export interface Status { + readonly server: string; + readonly state: State; + readonly error?: string; + /** Tools the server offered when last listed. */ + readonly tools: number; +} + +export interface Input { + readonly sessionId: SessionId; + readonly servers: ReadonlyArray; + /** + * Where stdio servers start (a server's own relative `cwd` resolves against it), and the root + * sent to servers: the session's directory when its sandbox is the host's own filesystem. Absent + * otherwise, since that directory does not exist here; servers then start in the session's host + * directory and are sent no roots. + */ + readonly directory?: string | undefined; + readonly hostDir?: string | undefined; + /** Configuration problems from `mcp.json`, reported once per session. */ + readonly errors?: ReadonlyArray; +} + +export interface Interface { + /** Settle a session's servers for its next exchange, and return the tools that exchange declares. */ + readonly reconcile: (input: Input) => Effect.Effect>; + readonly status: (sessionId: SessionId) => Effect.Effect>; +} + +export class Service extends Context.Service()("@codeworksh/harness/mcp/mcp/Service") {} + +interface Live { + readonly connection: McpClient.Connection; + readonly scope: Scope.Closeable; + inflight: number; + retired: boolean; + /** Bumped by every `tools/list_changed`, so only the latest listing is kept. */ + refreshes: number; +} + +interface Server { + readonly name: string; + config: ServerConfig; + /** Everything a connection is made from; a change restarts the server. */ + readonly key: string; + readonly cwd: string; + readonly roots: ReadonlyArray | undefined; + state: State; + error: string | undefined; + tools: ReadonlyArray; + live: Live | undefined; + /** The connect attempt in progress. Callers share it; closing the server interrupts it. */ + attempt: Fiber.Fiber | undefined; + /** Settles when the first background attempt finishes. */ + ready: Deferred.Deferred | undefined; + closed: boolean; + readonly lock: Semaphore.Semaphore; +} + +interface Session { + readonly servers: Map; + readonly names: ReturnType; + readonly lock: Semaphore.Semaphore; + readonly reported: Set; +} + +/** The parts of a config that decide the connection; exposure and description do not. */ +const connectionKey = (config: ServerConfig, cwd: string, roots: ReadonlyArray | undefined) => { + const { exposure: _exposure, toolExposure: _tools, description: _description, enabled, ...rest } = config; + return JSON.stringify({ rest, enabled: enabled !== false, cwd, roots }); +}; + +const hasDirectTools = (config: ServerConfig) => + (config.exposure ?? "direct") === "direct" || + Object.values(config.toolExposure ?? {}).some((exposure) => exposure === "direct"); + +export const layer = (options: { readonly hostCwd: string; readonly home: string }) => + Layer.effect( + Service, + Effect.gen(function* () { + const root = yield* Effect.scope; + // Connections close in parallel on shutdown, so one slow server's stop grace does not hold up + // the rest. Made before the fibers below, so connect attempts are interrupted before it closes. + const servers = yield* Scope.fork(root, "parallel"); + const fork = yield* FiberSet.makeRuntime(); + const forkAttempt = yield* FiberSet.makeRuntime(); + const context = yield* Effect.context(); + const log = yield* McpLog.make(hostPath.join(options.home, "mcp.log")); + const sessions = new Map(); + // Shared remote endpoints should not see a burst of concurrent startups. + const endpoints = new Map(); + const endpointLock = (url: string) => { + let lock = endpoints.get(url); + if (lock === undefined) endpoints.set(url, (lock = Semaphore.makeUnsafe(1))); + return lock; + }; + + const release = (live: Live) => + live.retired && live.inflight === 0 ? Scope.close(live.scope, Exit.void) : Effect.void; + + /** Stop handing out this connection; it closes once no call runs on it. */ + const retire = (server: Server, live: Live) => + Effect.suspend(() => { + live.retired = true; + if (server.live === live) server.live = undefined; + return release(live); + }); + + const endpoint = Effect.fnUntraced(function* (server: Server) { + const config = server.config; + const what = `MCP server "${server.name}"`; + if ("url" in config) + return { + _tag: "Http", + url: new URL(config.url), + headers: yield* McpValue.resolveAll(config.headers ?? {}, "header", `${what} header`), + } satisfies McpClient.Http; + return { + _tag: "Stdio", + // One executable and its arguments, not a command line: only a leading `~/` is expanded. + command: expandTilde(config.command, hostPath), + args: (config.args ?? []).map((arg) => expandTilde(arg, hostPath)), + env: yield* McpValue.resolveAll(config.env ?? {}, "env", `${what} env`), + cwd: hostPath.resolve(server.cwd, expandTilde(config.cwd ?? ".", hostPath)), + } satisfies McpClient.Stdio; + }); + + const watch = (server: Server, live: Live) => { + // Callbacks run in the connection's own scope, and retiring closes that scope; run them + // in the service's instead, or closing would wait on the callback doing the closing. + const current = (effect: Effect.Effect) => + Effect.sync(() => { + fork( + Effect.suspend(() => (server.live === live ? effect : Effect.void)).pipe( + server.lock.withPermits(1), + ), + ); + }); + live.connection.onClose((reason) => + current( + Effect.gen(function* () { + server.state = "disconnected"; + server.error = reason; + yield* retire(server, live); + }), + ), + ); + // Calls replace an expired session themselves; this covers a background request seeing it first. + live.connection.onSessionExpired(current(retire(server, live))); + // Listed without the server's lock, which calls and closing need: the listing can take a while. + // Listings can finish out of order; only the latest one counts. + live.connection.onToolsChanged( + Effect.sync(() => { + const refresh = ++live.refreshes; + const latest = () => server.live === live && live.refreshes === refresh; + fork( + live.connection.tools().pipe( + Effect.match({ + onSuccess: (tools) => { + if (latest()) server.tools = tools; + }, + onFailure: (error) => { + if (latest()) server.error = `failed to refresh tools: ${error.message}`; + }, + }), + ), + ); + }), + ); + }; + + /** One connect attempt, run apart from any caller so that closing the server can interrupt it. */ + const attempt = (server: Server): Effect.Effect => + Effect.gen(function* () { + const scope = yield* Scope.fork(servers, "sequential"); + const config = server.config; + const connect = Effect.gen(function* () { + const target = yield* endpoint(server); + const connection = yield* McpClient.connect({ + server: server.name, + endpoint: target, + ...(server.roots === undefined ? {} : { roots: server.roots }), + timeout: { request: Duration.seconds(config.timeout ?? DEFAULT_TIMEOUT_SECONDS) }, + // Written on the service's fibers: a failed handshake closes the connection's, and + // its last messages usually say why. + onLog: (entry) => Effect.sync(() => fork(log.write(server.name, entry))), + }).pipe(Scope.provide(scope)); + const tools = yield* connection.tools(); + return { connection, tools }; + }); + const exit = yield* ("url" in config ? endpointLock(config.url).withPermit(connect) : connect).pipe( + Effect.onInterrupt(() => Scope.close(scope, Exit.void)), + Effect.exit, + ); + if (Exit.isFailure(exit)) { + yield* Scope.close(scope, Exit.void); + const error = Cause.squash(exit.cause); + const message = + typeof error === "string" ? error : error instanceof Error ? error.message : String(error); + server.state = Schema.is(NeedsAuthError)(error) ? "needs-auth" : "failed"; + server.error = message; + yield* Effect.logWarning("mcp connect failed", { server: server.name, error: message }); + return yield* Effect.fail(message); + } + const live: Live = { + connection: exit.value.connection, + scope, + inflight: 0, + retired: false, + refreshes: 0, + }; + server.live = live; + server.tools = exit.value.tools; + server.state = "connected"; + server.error = undefined; + watch(server, live); + yield* Effect.logInfo("mcp connected", { server: server.name, tools: server.tools.length }); + return live; + }).pipe( + Effect.ensuring( + Effect.sync(() => { + server.attempt = undefined; + }), + ), + Effect.provide(context), + ); + + /** The server's current connection, connecting first when it has none. */ + const open = (server: Server): Effect.Effect => + Effect.gen(function* () { + const next = yield* Effect.gen(function* () { + if (server.closed) return yield* Effect.fail(`MCP server "${server.name}" is shut down`); + const current = server.live; + // A drop is known here before its `onClose` handler has run. + const reason = current?.connection.dropped(); + if (current !== undefined && reason === undefined) return current; + if (current !== undefined) { + server.state = "disconnected"; + server.error = reason; + yield* retire(server, current); + } + if (server.attempt === undefined) { + server.state = "connecting"; + // Started after the assignment: an attempt that fails at once would otherwise clear + // `attempt` before it is set, and the finished fiber would stand in for every retry. + server.attempt = forkAttempt(Effect.andThen(Effect.yieldNow, attempt(server))); + } + return server.attempt; + }).pipe(server.lock.withPermits(1)); + if (!Fiber.isFiber(next)) return next; + // An interrupted caller leaves the attempt running for whoever needs the server next. + const exit = yield* Fiber.await(next); + if (Exit.isSuccess(exit)) return exit.value; + return yield* Cause.hasInterruptsOnly(exit.cause) + ? Effect.fail(`MCP server "${server.name}" is shut down`) + : Effect.failCause(exit.cause); + }); + + /** Stop the server: interrupt a connect in progress, and retire the connection it has. */ + const close = (server: Server) => + Effect.gen(function* () { + server.closed = true; + const pending = server.attempt; + if (pending !== undefined) yield* Fiber.interrupt(pending); + yield* Effect.suspend(() => + server.live === undefined ? Effect.void : retire(server, server.live), + ).pipe(server.lock.withPermits(1)); + }); + + /** One call of one of the server's tools, on whichever connection is current when it runs. */ + const call = + (session: Session, name: string, tool: string) => + (input: McpTool.Call): Effect.Effect => + Effect.gen(function* () { + for (let attempt = 1; ; attempt++) { + const server = session.servers.get(name); + if (server === undefined || server.closed || server.config.enabled === false) + return yield* Effect.fail(`MCP server "${name}" is disabled.`); + if ( + McpTool.exposure(server.config, tool) === "hidden" || + (server.live !== undefined && !server.tools.some((offered) => offered.name === tool)) + ) + return yield* Effect.fail(`MCP tool "${name}/${tool}" is no longer available.`); + const live = yield* open(server).pipe( + Effect.mapError((error) => `MCP server "${name}" is not connected: ${error}`), + ); + live.inflight += 1; + const exit = yield* live.connection + .callTool({ name: tool, args: input.args, onProgress: input.onProgress }) + .pipe( + Effect.exit, + Effect.ensuring( + Effect.suspend(() => { + live.inflight -= 1; + return release(live); + }), + ), + ); + if (Exit.isSuccess(exit)) return exit.value; + const error = Cause.findErrorOption(exit.cause); + if (error._tag === "None") return yield* Effect.failCause(exit.cause as Cause.Cause); + const failure = error.value; + // The server no longer knows the session, so it did not run the call. + if (failure._tag === "Mcp.SessionExpiredError" && attempt === 1) { + yield* retire(server, live).pipe(server.lock.withPermits(1)); + continue; + } + if (failure._tag === "Mcp.NeedsAuthError") { + yield* Effect.gen(function* () { + server.state = "needs-auth"; + server.error = failure.message; + yield* retire(server, live); + }).pipe(server.lock.withPermits(1)); + return yield* Effect.fail(`MCP server "${name}" needs sign-in: ${failure.message}`); + } + return yield* Effect.fail(`MCP tool "${name}/${tool}" failed: ${failure.message}`); + } + }); + + /** Connect in the background; `ready` settles when the attempt does. */ + const start = (server: Server) => { + if (server.config.enabled === false) { + server.state = "disabled"; + return; + } + const ready = Deferred.makeUnsafe(); + server.ready = ready; + fork(open(server).pipe(Effect.ignore, Effect.ensuring(Deferred.succeed(ready, undefined)))); + }; + + const sessionOf = (sessionId: SessionId) => { + let session = sessions.get(sessionId); + if (session === undefined) { + session = { + servers: new Map(), + names: McpTool.names(), + lock: Semaphore.makeUnsafe(1), + reported: new Set(), + }; + sessions.set(sessionId, session); + } + return session; + }; + + const reconcile = Effect.fn("Mcp.reconcile")(function* (input: Input) { + const session = sessionOf(input.sessionId); + for (const error of input.errors ?? []) { + if (session.reported.has(error)) continue; + session.reported.add(error); + yield* Effect.logWarning(`mcp config: ${error}`); + } + const cwd = input.directory ?? input.hostDir ?? options.hostCwd; + const roots = input.directory === undefined ? undefined : [input.directory]; + const wanted = new Map(input.servers); + const started: Server[] = []; + yield* Effect.gen(function* () { + for (const [name, server] of session.servers) { + const config = wanted.get(name); + // Exposure and description apply to the next exchange without reconnecting. + if (config !== undefined && connectionKey(config, cwd, roots) === server.key) { + server.config = config; + // A server that could not connect declares no tools, so no call would retry it. + // Retry in the background; waiting for it would hold up every prompt while it is down. + if (server.state === "failed" && server.attempt === undefined) start(server); + continue; + } + session.servers.delete(name); + // Stopping can take seconds; the prompt does not wait for it. + fork(close(server)); + } + for (const [name, config] of wanted) { + if (session.servers.has(name)) continue; + const server: Server = { + name, + config, + key: connectionKey(config, cwd, roots), + cwd, + roots, + state: "connecting", + error: undefined, + tools: [], + live: undefined, + attempt: undefined, + ready: undefined, + closed: false, + lock: Semaphore.makeUnsafe(1), + }; + session.servers.set(name, server); + start(server); + started.push(server); + } + }).pipe(session.lock.withPermits(1)); + + // Wait for servers this exchange started (all of them at first, then new or restarted ones) so + // their tools are declared in its first request, but never indefinitely: one hanging server + // must not hold up the prompt. + const pending = started.filter( + (server) => server.config.enabled !== false && hasDirectTools(server.config), + ); + if (pending.length > 0) { + const finished = yield* Effect.forEach(pending, (server) => + server.ready === undefined ? Effect.void : Deferred.await(server.ready), + ).pipe(Effect.timeoutOption(STARTUP_WAIT)); + if (finished._tag === "None") + yield* Effect.logInfo("mcp servers are still connecting; their tools appear once connected"); + } + + const tools: RegisteredTool[] = []; + for (const server of session.servers.values()) { + if (server.config.enabled === false || server.tools.length === 0) continue; + const names = session.names( + server.name, + server.tools.map((tool) => tool.name), + ); + for (const tool of server.tools) { + if (McpTool.exposure(server.config, tool.name) !== "direct") continue; + tools.push( + McpTool.make({ + server: server.name, + tool, + name: names.get(tool.name)!, + call: call(session, server.name, tool.name), + }), + ); + } + } + return tools; + }); + + return Service.of({ + reconcile, + status: (sessionId) => + Effect.sync(() => + [...(sessions.get(sessionId)?.servers.values() ?? [])].map((server) => ({ + server: server.name, + state: server.state, + ...(server.error === undefined ? {} : { error: server.error }), + tools: server.tools.length, + })), + ), + }); + }), + ); + +export * as Mcp from "./mcp.ts"; diff --git a/packages/harness/src/mcp/tool.ts b/packages/harness/src/mcp/tool.ts new file mode 100644 index 00000000..41e3561b --- /dev/null +++ b/packages/harness/src/mcp/tool.ts @@ -0,0 +1,177 @@ +/** + * MCP tools as harness tools: their names, which of them the model sees, and how a call's result + * reaches it. + */ + +import { normalizeOptionalNulls } from "@codeworksh/aikit"; +import type { Exposure, ServerConfig } from "@codeworksh/plugin/plugin/mcp"; +import type { CallToolResult, Tool as McpTool } from "@modelcontextprotocol/client"; +import { Effect, Schema } from "effect"; +import { ToolProgress } from "../tool/progress.ts"; +import * as Tool from "../tool/tool.ts"; +import { Hash } from "../util/hash.ts"; +import { McpContent } from "./content.ts"; +import type { Progress } from "./client.ts"; + +/** Provider tool names are limited to 64 characters of `[A-Za-z0-9_-]`. */ +const MAX_NAME_LENGTH = 64; + +/** + * `mcp____`, with everything but `[A-Za-z0-9_]` replaced by `_`. A name that is too + * long, or that `taken` reports as used by another MCP tool, ends in a hash of the server and tool + * instead: sanitizing can map two tools to one name (`a-b` and `a_b`). A server can also offer a tool + * already named like that hash, so the hash grows until the name is free. + */ +export const name = (server: string, tool: string, taken: (candidate: string) => boolean = () => false) => { + const plain = `mcp__${server}__${tool}`.replace(/[^A-Za-z0-9_]/g, "_"); + if (plain.length <= MAX_NAME_LENGTH && !taken(plain)) return plain; + const hash = Hash.sha256(`${server}\0${tool}`); + let candidate = plain; + // The executor refuses two tools of one name, so the hash grows, while `mcp__` still fits before it. + for (let length = 8; length <= MAX_NAME_LENGTH - "mcp__".length - 1; length += 8) { + candidate = `${plain.slice(0, MAX_NAME_LENGTH - length - 1)}_${hash.slice(0, length)}`; + if (!taken(candidate)) return candidate; + } + return candidate; +}; + +/** + * The tool names of one session. A name, once given to a server's tool, stays with it for the + * session, so a reconnect or a changed tool list never moves a name to a different tool. + */ +export const names = () => { + const owners = new Map(); + /** Names for `tools`, every tool of `server` the server offers right now. */ + return (server: string, tools: ReadonlyArray): ReadonlyMap => { + const assigned = new Map(); + const used = new Set(); + // Every tool whose sanitized name collides gets the suffix, so which one would keep the plain + // name does not depend on the order of the list. + const plain = [...new Set(tools)].map((tool) => name(server, tool)); + for (const tool of new Set(tools)) { + const owner = `${server}\0${tool}`; + const given = name(server, tool, (candidate) => { + const existing = owners.get(candidate); + return ( + (existing !== undefined && existing !== owner) || + used.has(candidate) || + plain.indexOf(candidate) !== plain.lastIndexOf(candidate) + ); + }); + owners.set(given, owner); + used.add(given); + assigned.set(tool, given); + } + return assigned; + }; +}; + +const pattern = (glob: string) => + new RegExp( + `^${glob + .split("*") + .map((part) => part.replace(/[.+?^${}()|[\]\\]/g, "\\$&")) + .join(".*")}$`, + ); + +/** One tool's exposure: its `toolExposure` entry (exact name first, then patterns), else the server's. */ +export const exposure = (config: ServerConfig, tool: string): Exposure => { + const overrides = config.toolExposure ?? {}; + const exact = overrides[tool]; + if (exact !== undefined) return exact; + for (const [glob, value] of Object.entries(overrides)) { + if (glob.includes("*") && pattern(glob).test(tool)) return value; + } + return config.exposure ?? "direct"; +}; + +/** What the model reads, kept the same for a result and for a failure. */ +const Outcome = Schema.Struct({ + content: Schema.Array(Schema.Unknown), + server: Schema.String, + tool: Schema.String, + structuredContent: Schema.optional(Schema.Unknown), +}); +type Outcome = typeof Outcome.Type; + +/** Tool input schemas must be objects; servers may omit `type`, and some providers want `properties`. */ +const parameters = (schema: McpTool["inputSchema"]): Record => ({ + ...schema, + type: schema.type ?? "object", + ...(schema.properties === undefined ? { properties: {} } : {}), +}); + +const withoutOptionalNulls = (args: Record, schema: Record) => { + const normalized = structuredClone(args); + normalizeOptionalNulls(normalized, schema); + return normalized; +}; + +export interface Call { + readonly args: Record; + readonly onProgress: (progress: Progress) => void; +} + +/** + * A server tool as a harness tool. `call` resolves the server's current connection when it runs, so + * the definition can outlive the connection it was listed on. A failed call and a result the server + * flags `isError` both reach the model as an error result. + */ +export const make = (input: { + readonly server: string; + readonly tool: McpTool; + readonly name: string; + readonly call: (call: Call) => Effect.Effect; +}) => { + const { server, tool } = input; + const title = tool.title ?? tool.annotations?.title; + const failed = (message: string): Outcome => ({ + content: [{ type: "text", text: message }], + server, + tool: tool.name, + }); + const schema = parameters(tool.inputSchema); + return Tool.register( + Tool.make({ + name: input.name, + label: `${server}/${tool.name}`, + description: tool.description?.trim() || title || `MCP tool ${tool.name} from server ${server}`, + parameters: Tool.fromJsonSchema(schema), + success: Outcome, + failure: Outcome, + encodeContent: (outcome) => outcome.content as McpContent.Content[], + encodeFailureContent: (outcome) => outcome.content as McpContent.Content[], + encodeDetails: ({ content: _content, ...details }) => details, + // The arguments as the model wrote them, minus the `null`s strict sampling writes for omitted + // optional ones: validation only checks them, and the server, not a translation of its schema, + // decides what they mean. + handler: (_params, ctx) => + Effect.gen(function* () { + const progress = yield* ToolProgress; + const report = Effect.runForkWith(yield* Effect.context()); + const result = yield* input + .call({ + args: withoutOptionalNulls(ctx.rawArgs, schema), + onProgress: (update) => { + const total = update.total === undefined ? "" : `/${update.total}`; + const text = update.message ?? `Progress ${update.progress}${total}`; + report(progress.report({ content: [{ type: "text", text }] })); + }, + }) + .pipe(Effect.mapError(failed)); + const content = McpContent.fromResult(result); + if (result.isError && !content.some((block) => block.type === "text" && block.text !== "")) + content.push({ type: "text", text: `MCP tool ${server}/${tool.name} returned an error` }); + const outcome: Outcome = { + content, + server, + tool: tool.name, + ...(result.structuredContent === undefined ? {} : { structuredContent: result.structuredContent }), + }; + return result.isError ? yield* Effect.fail(outcome) : outcome; + }), + }), + ); +}; + +export * as McpTool from "./tool.ts"; diff --git a/packages/harness/src/mcp/value.ts b/packages/harness/src/mcp/value.ts new file mode 100644 index 00000000..5dbf85ea --- /dev/null +++ b/packages/harness/src/mcp/value.ts @@ -0,0 +1,156 @@ +/** + * Values of a server's `env` and `headers`, resolved when the server connects: + * + * - `!command` runs the rest through `/bin/sh` and uses its trimmed stdout, once per process: later + * and concurrent connects share it. A command that fails runs again next time, since the host + * outlives sessions. + * - Otherwise `$NAME` and `${NAME}` are replaced by host environment variables; `$$` is a literal + * `$` and `$!` a literal `!`. Anything else is literal. + * + * A header holding a line break or NUL, or a variable holding NUL, is refused without being shown: it + * cannot be sent or set, and the error the runtime would raise quotes the value, often a secret. A + * variable may span lines (a PEM key, say). + */ + +import { Config, Duration, Effect, Fiber, Option, Stream } from "effect"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; + +const COMMAND_TIMEOUT = Duration.seconds(10); +const NAME = /^[A-Za-z_][A-Za-z0-9_]*$/; +const NAME_PREFIX = /^[A-Za-z_][A-Za-z0-9_]*/; +const COMMAND_KILL_AFTER = Duration.seconds(2); +const UNSAFE = { + header: { pattern: /[\r\n\0]/, what: "a line break or NUL character" }, + env: { pattern: /\0/, what: "a NUL character" }, +} as const; +const commands = new Map>(); + +type Part = { readonly type: "literal"; readonly value: string } | { readonly type: "env"; readonly name: string }; + +const template = (value: string): ReadonlyArray => { + const parts: Part[] = []; + const literal = (text: string) => { + if (!text) return; + const last = parts.at(-1); + if (last?.type === "literal") parts[parts.length - 1] = { type: "literal", value: last.value + text }; + else parts.push({ type: "literal", value: text }); + }; + let index = 0; + while (index < value.length) { + const dollar = value.indexOf("$", index); + if (dollar < 0) { + literal(value.slice(index)); + break; + } + literal(value.slice(index, dollar)); + const next = value[dollar + 1]; + if (next === "$" || next === "!") { + literal(next); + index = dollar + 2; + continue; + } + if (next === "{") { + const end = value.indexOf("}", dollar + 2); + if (end < 0) { + literal("$"); + index = dollar + 1; + continue; + } + const name = value.slice(dollar + 2, end); + if (NAME.test(name)) parts.push({ type: "env", name }); + else literal(value.slice(dollar, end + 1)); + index = end + 1; + continue; + } + const match = NAME_PREFIX.exec(value.slice(dollar + 1)); + if (match) { + parts.push({ type: "env", name: match[0] }); + index = dollar + 1 + match[0].length; + continue; + } + literal("$"); + index = dollar + 1; + } + return parts; +}; + +/** stdout of `command`, or nothing when it fails, exits non-zero, prints nothing or times out. */ +const run = (command: string) => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const handle = yield* spawner.spawn( + // A helper that ignores SIGTERM must not outlast the timeout. + ChildProcess.make("/bin/sh", ["-c", command], { + stdin: "ignore", + stderr: "ignore", + forceKillAfter: COMMAND_KILL_AFTER, + }), + ); + const [output, code] = yield* Effect.all([Stream.mkString(Stream.decodeText(handle.stdout)), handle.exitCode], { + concurrency: 2, + }); + const trimmed = output.trim(); + return code === 0 && trimmed ? trimmed : undefined; + }).pipe( + Effect.scoped, + Effect.timeoutOption(COMMAND_TIMEOUT), + Effect.map((value) => (value._tag === "Some" ? value.value : undefined)), + Effect.orElseSucceed(() => undefined), + ); + +const command = Effect.fnUntraced(function* (line: string, what: string) { + let running = commands.get(line); + if (running === undefined) { + // Detached, so a caller giving up does not stop it for the others waiting on it. A failure is + // forgotten when it happens, whether or not anyone still waits. + const forget = (output: string | undefined) => + Effect.sync(() => { + if (output === undefined && commands.get(line) === running) commands.delete(line); + }); + running = yield* Effect.forkDetach(Effect.andThen(Effect.yieldNow, run(line).pipe(Effect.tap(forget)))); + commands.set(line, running); + } + const output = yield* Fiber.join(running); + if (output === undefined) return yield* Effect.fail(`Failed to resolve ${what} from shell command: ${line}`); + return output; +}); + +const variables = Effect.fnUntraced(function* (value: string, what: string) { + const parts = template(value); + const names = [...new Set(parts.flatMap((part) => (part.type === "env" ? [part.name] : [])))]; + // An empty variable counts as unset. + const env = new Map(); + for (const name of names) { + const found = yield* Config.option(Config.String(name)).pipe(Effect.orElseSucceed(() => Option.none())); + if (Option.isSome(found) && found.value !== "") env.set(name, found.value); + } + const missing = names.filter((name) => !env.has(name)); + if (missing.length > 0) + return yield* Effect.fail( + `Failed to resolve ${what} from environment variable${missing.length === 1 ? "" : "s"}: ${missing.join(", ")}`, + ); + return parts.map((part) => (part.type === "literal" ? part.value : env.get(part.name))).join(""); +}); + +/** + * One value, used as `kind`. `what` names it in the failure, for example + * `MCP server "docs" header "Authorization"`. + */ +export const resolve = Effect.fnUntraced(function* (value: string, kind: keyof typeof UNSAFE, what: string) { + const resolved = yield* value.startsWith("!") ? command(value.slice(1), what) : variables(value, what); + if (UNSAFE[kind].pattern.test(resolved)) return yield* Effect.fail(`${what} contains ${UNSAFE[kind].what}`); + return resolved; +}); + +/** Every value of a record, failing on the first that does not resolve. */ +export const resolveAll = Effect.fnUntraced(function* ( + values: Readonly>, + kind: keyof typeof UNSAFE, + what: string, +) { + const resolved: Record = {}; + for (const [key, value] of Object.entries(values)) resolved[key] = yield* resolve(value, kind, `${what} "${key}"`); + return resolved; +}); + +export * as McpValue from "./value.ts"; diff --git a/packages/harness/src/plugin/host.ts b/packages/harness/src/plugin/host.ts index f57a0153..c81770d5 100644 --- a/packages/harness/src/plugin/host.ts +++ b/packages/harness/src/plugin/host.ts @@ -1,4 +1,6 @@ +import type { ServerConfig } from "@codeworksh/plugin/plugin/mcp"; import { Cause, Effect, Schema } from "effect"; +import type { RegisteredTool } from "../tool/tool.ts"; import type { Prepared } from "./catalog.ts"; import type { SharedPluginContext } from "./context.ts"; import { make } from "./registry.ts"; @@ -12,11 +14,21 @@ export class SetupError extends Schema.TaggedError()("Plugin.SetupEr export const run = Effect.fn("PluginHost.run")(function* ( selection: ReadonlyArray, input: Omit, + /** The tools of the MCP servers the `mcp` domain settled on, added before the `tool` domain runs. */ + mcp: (servers: ReadonlyArray) => Effect.Effect>, ) { - const buckets = make(); + const buckets = make(input.settings.mcp.servers); const ctx = Object.freeze({ ...input, plugin: buckets.registry }); + let settled = false; + const settle = Effect.gen(function* () { + if (settled) return; + settled = true; + const tools = yield* mcp(buckets.settle()); + for (const tool of tools) buckets.registry.tools.add(tool); + }); const setup = Effect.gen(function* () { for (const { plugin, options } of selection) { + if (plugin.kind !== "mcp") yield* settle; yield* Effect.suspend(() => { const result = plugin.setup(ctx, options); if (Effect.isEffect(result)) @@ -56,6 +68,7 @@ export const run = Effect.fn("PluginHost.run")(function* ( }), ); } + yield* settle; return yield* Effect.try({ try: () => buckets.freeze(input.location.directory), catch: (cause) => diff --git a/packages/harness/src/plugin/mcp/registry.ts b/packages/harness/src/plugin/mcp/registry.ts new file mode 100644 index 00000000..d2329fd1 --- /dev/null +++ b/packages/harness/src/plugin/mcp/registry.ts @@ -0,0 +1,69 @@ +import type { McpRegistry, ServerConfig, ServerPatch } from "@codeworksh/plugin/plugin/mcp"; +import type { McpServer } from "@codeworksh/plugin/settings"; +import { Predicate } from "effect"; +import { McpConfig } from "../../mcp/config.ts"; + +const PATCH_KEYS: ReadonlyArray = ["enabled", "exposure", "toolExposure", "timeout", "description"]; + +/** A copy no one else holds, frozen all the way down, so a settled server cannot change behind the service. */ +const frozen = (value: A): A => { + const freeze = (node: unknown) => { + if (!Predicate.isObject(node) || Object.isFrozen(node)) return; + for (const child of Object.values(node)) freeze(child); + Object.freeze(node); + }; + const copy = structuredClone(value); + freeze(copy); + return copy; +}; + +/** + * The session's MCP servers for one exchange, seeded from `mcp.json`. A plugin written in plain + * JavaScript can hand in anything, so every write is checked here with the rules `mcp.json` follows, + * and a bad one fails that plugin's `setup` rather than a connection later. + */ +export const make = (seed: ReadonlyArray) => { + let open = true; + const servers = new Map(seed.map((server) => [server.name, frozen(server.config)])); + const writable = () => { + if (!open) throw new Error("mcp registry is closed: servers are settled before the tool domain runs"); + }; + const checked = (name: string, config: unknown) => { + const result = McpConfig.validate(name, config); + if (Predicate.isString(result)) throw new Error(result); + return frozen(result); + }; + const registry: McpRegistry = Object.freeze({ + list: () => Object.freeze([...servers].map((entry) => Object.freeze(entry))), + get: (name: string) => servers.get(name), + set: (name: string, config: ServerConfig) => { + writable(); + const other = McpConfig.clash(servers.keys(), name); + if (other !== undefined) throw new Error(`MCP server "${name}" conflicts with "${other}"`); + servers.set(name, checked(name, config)); + }, + update: (name: string, patch: ServerPatch) => { + writable(); + const current = servers.get(name); + if (current === undefined) throw new Error(`Unknown MCP server: ${name}`); + const extra = Predicate.isObject(patch) ? Object.keys(patch).filter((key) => !PATCH_KEYS.includes(key)) : []; + if (extra.length > 0) + throw new Error( + `MCP server "${name}": an update can only set ${PATCH_KEYS.join(", ")}; use set to replace it`, + ); + servers.set(name, checked(name, { ...current, ...patch })); + }, + remove: (name: string) => { + writable(); + servers.delete(name); + }, + }); + return { + registry, + /** Close the bucket and return the servers it settled on. */ + settle: () => { + open = false; + return Object.freeze([...servers]); + }, + }; +}; diff --git a/packages/harness/src/plugin/registry.ts b/packages/harness/src/plugin/registry.ts index e7bc56da..33d61f47 100644 --- a/packages/harness/src/plugin/registry.ts +++ b/packages/harness/src/plugin/registry.ts @@ -1,5 +1,7 @@ import type { PluginRegistry } from "@codeworksh/plugin/plugin"; +import type { McpServer } from "@codeworksh/plugin/settings"; import { make as makeCatalog } from "../tool/registry.ts"; +import { make as makeMcp } from "./mcp/registry.ts"; import { make as makePrompt } from "./prompt/registry.ts"; import { render } from "./prompt/render.ts"; import { make as makeTools } from "./tool/registry.ts"; @@ -7,15 +9,20 @@ import { make as makeTools } from "./tool/registry.ts"; /** The buckets a plugin writes into during `setup`. Declared in `@codeworksh/plugin`. */ export type { PluginRegistry }; -export const make = () => { +/** `servers` seeds the MCP bucket: the session's `mcp.json` servers, before any plugin edits them. */ +export const make = (servers: ReadonlyArray) => { + const mcp = makeMcp(servers); const tools = makeTools(); const prompt = makePrompt(); const close = () => { + mcp.settle(); tools.close(); prompt.close(); }; return { - registry: Object.freeze({ tools: tools.registry, prompt: prompt.registry }), + registry: Object.freeze({ mcp: mcp.registry, tools: tools.registry, prompt: prompt.registry }), + /** Close the MCP bucket and return the servers the `mcp` domain settled on. */ + settle: mcp.settle, close, freeze: (directory: string) => { close(); diff --git a/packages/harness/src/settings/settings.ts b/packages/harness/src/settings/settings.ts index 0465dee4..573b111f 100644 --- a/packages/harness/src/settings/settings.ts +++ b/packages/harness/src/settings/settings.ts @@ -38,10 +38,11 @@ */ import { Context, Effect, Layer, Predicate, Result, Schema, SchemaIssue } from "effect"; -import { type ParseError, parse as parseJsonc, printParseErrorCode } from "jsonc-parser"; import { Global } from "../global.ts"; +import { McpConfig } from "../mcp/config.ts"; import { fileSystem, hostPath } from "../host.ts"; import { expandTilde } from "../util/home.ts"; +import * as Jsonc from "../util/jsonc.ts"; import { merge, normalize } from "./merge.ts"; import { defaults, Patch, type Declared, type DeclaredModel, type Info, type PluginEntry } from "./schema.ts"; @@ -159,12 +160,6 @@ export class SettingsError extends Schema.TaggedError()("Settings } } -/** `line:column`, 1-based, for a parser offset into the source. */ -const at = (source: string, offset: number): string => { - const before = source.slice(0, offset).split("\n"); - return `${before.length}:${(before.at(-1)?.length ?? 0) + 1}`; -}; - /** * Decode failures name the field, and explain it only with a message a schema check wrote. Effect's * defaults quote the rejected value, and in a settings file that value can be a secret, so any @@ -178,19 +173,10 @@ const decodeFormatter = SchemaIssue.makeFormatterStandardSchemaV1({ }); export const parse = Effect.fn("Settings.parse")(function* (path: string, source: string) { - // JSONC: a settings file is written by hand, so comments and a trailing comma are part of - // the format rather than mistakes. The parser reports offsets, which become `line:column` - // below; Schema validates the normalized value after that. - const errors: ParseError[] = []; - const json: unknown = parseJsonc(source, errors, { allowTrailingComma: true, disallowComments: false }); - const first = errors[0]; - if (first !== undefined) { - return yield* new SettingsError({ - path, - reason: "parse", - detail: `${printParseErrorCode(first.error)} at ${at(source, first.offset)}`, - }); - } + // Syntax first; Schema validates the normalized value after that. + const parsed = Jsonc.parse(source); + if ("error" in parsed) return yield* new SettingsError({ path, reason: "parse", detail: parsed.error }); + const json = parsed.value; // Normalization reads a `null` as "absent", which is right for a settings patch and wrong // inside a plugin's `options`: that block is opaque, and `{ "endpoint": null }` is a value // its plugin may need. The `plugins` array is therefore decoded exactly as written. @@ -310,7 +296,10 @@ export const load = Effect.fn("Settings.load")(function* (options: Options & { r return yield* error; } } - return { ...settings, plugins: declared.map((one) => one.entry), declared, models }; + // `mcp.json` sits next to each settings file, but a broken one is reported rather than failing + // the read: one bad server must not keep a session from running. + const mcp = yield* McpConfig.load({ userDir: options.userConfigDir ?? options.home, root }); + return { ...settings, plugins: declared.map((one) => one.entry), declared, models, mcp }; }); /** diff --git a/packages/harness/src/state/state.ts b/packages/harness/src/state/state.ts index 6b460f39..a801ee58 100644 --- a/packages/harness/src/state/state.ts +++ b/packages/harness/src/state/state.ts @@ -22,6 +22,7 @@ import { select, type Origin, type PluginRef, type Pool } from "../plugin/catalo import { makeEvents, type PromptResolver } from "../plugin/context.ts"; import { run as setup } from "../plugin/host.ts"; import type { Plugin } from "../plugin/plugin.ts"; +import { Mcp } from "../mcp/mcp.ts"; import { LLM } from "../runner/llm.ts"; import type { Runner } from "../runner/run.ts"; import { SandboxIO } from "../sandbox/io.ts"; @@ -251,6 +252,7 @@ export const layer = ( const eventService = yield* Event.Service; const events = makeEvents(eventService); const global = yield* Global.Service; + const mcp = yield* Mcp.Service; /* * One load pass at a time, for the whole process. * @@ -457,19 +459,33 @@ export const layer = ( }); } - const contributions = yield* setup(chosen.selection, { - sessionId, - sandbox, - paths: global, - location, - settings: loadedSettings, - model: resolvedModel, - events, - config: { - ...(promptCustom === undefined ? {} : { promptCustom: resolver(promptCustom) }), - ...(promptSystemAppend === undefined ? {} : { promptSystemAppend: resolver(promptSystemAppend) }), + const contributions = yield* setup( + chosen.selection, + { + sessionId, + sandbox, + paths: global, + location, + settings: loadedSettings, + model: resolvedModel, + events, + config: { + ...(promptCustom === undefined ? {} : { promptCustom: resolver(promptCustom) }), + ...(promptSystemAppend === undefined + ? {} + : { promptSystemAppend: resolver(promptSystemAppend) }), + }, }, - }).pipe(Effect.mapError((cause) => new SnapshotError({ sessionId, reason: cause.message, cause }))); + (servers) => + mcp.reconcile({ + sessionId, + servers, + // MCP servers run on the host, so only a host sandbox's directory exists for them. + directory: sandbox.kind === "local" ? location.directory : undefined, + hostDir, + errors: loadedSettings.mcp.errors, + }), + ).pipe(Effect.mapError((cause) => new SnapshotError({ sessionId, reason: cause.message, cause }))); return { sessionId, diff --git a/packages/harness/src/tool/executor.ts b/packages/harness/src/tool/executor.ts index 21a2c843..9c881617 100644 --- a/packages/harness/src/tool/executor.ts +++ b/packages/harness/src/tool/executor.ts @@ -25,7 +25,14 @@ import { isAikitToolCallTerminalPart } from "../schema.ts"; import { errorMessage } from "../util/error.ts"; import { ToolExecutionError } from "./error.ts"; import { ToolProgress, type ToolProgressPartial } from "./progress.ts"; -import { type AnyToolDef, type ModelContent, type RegisteredTool, toAikitTool, type ToolCallContext } from "./tool.ts"; +import { + type AnyToolDef, + type ModelContent, + type RegisteredTool, + isFromJsonSchema, + toAikitTool, + type ToolCallContext, +} from "./tool.ts"; /** * `ToolExecutor` — the uniform pipeline run for every tool call: @@ -284,8 +291,10 @@ export const make = (tools: ReadonlyArray): E // does not allow are dropped first. const decoders = new Map( wire.map((tool) => { + const parameters = impls.get(tool.name)!.tool.definition.parameters; const decode = Schema.decodeUnknownEffect( - Schema.toCodecJson(asCodec(impls.get(tool.name)!.tool.definition.parameters)), + Schema.toCodecJson(asCodec(parameters)), + isFromJsonSchema(parameters) ? { onExcessProperty: "error" } : undefined, ); const parse = (args: unknown) => { const normalized = structuredClone(args); diff --git a/packages/harness/src/util/jsonc.ts b/packages/harness/src/util/jsonc.ts new file mode 100644 index 00000000..15dd4d33 --- /dev/null +++ b/packages/harness/src/util/jsonc.ts @@ -0,0 +1,20 @@ +import { type ParseError, parse as parseJsonc, printParseErrorCode } from "jsonc-parser"; + +/** `line:column`, 1-based, for a parser offset into the source. */ +const at = (source: string, offset: number): string => { + const before = source.slice(0, offset).split("\n"); + return `${before.length}:${(before.at(-1)?.length ?? 0) + 1}`; +}; + +/** + * A hand-written config file: comments and a trailing comma are part of the format rather than + * mistakes. A syntax error is reported once, as the parser's reason at `line:column`. + */ +export const parse = (source: string): { readonly value: unknown } | { readonly error: string } => { + const errors: ParseError[] = []; + const value: unknown = parseJsonc(source, errors, { allowTrailingComma: true, disallowComments: false }); + const first = errors[0]; + return first === undefined + ? { value } + : { error: `${printParseErrorCode(first.error)} at ${at(source, first.offset)}` }; +}; diff --git a/packages/harness/test/__artifacts__/mcp.session.json b/packages/harness/test/__artifacts__/mcp.session.json new file mode 100644 index 00000000..5f44bea5 --- /dev/null +++ b/packages/harness/test/__artifacts__/mcp.session.json @@ -0,0 +1,173 @@ +{ + "requests": 6, + "tools": [ + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__bye" + ], + "echoTool": { + "name": "mcp__e2e__echo", + "description": "Echo text back, prefixed with the configured greeting.", + "parameters": { + "type": "object", + "properties": { + "text": { + "type": "string", + "description": "What to echo." + } + }, + "required": [ + "text" + ] + } + }, + "plugins": { + "mcpDomain": [ + [ + [ + "e2e", + true + ], + [ + "off", + false + ], + [ + "my-srv", + false + ], + [ + "badheader", + true + ] + ], + [ + [ + "e2e", + true + ], + [ + "off", + false + ], + [ + "my-srv", + false + ], + [ + "badheader", + true + ] + ] + ], + "toolDomain": [ + "mcp registry is closed: servers are settled before the tool domain runs", + "mcp registry is closed: servers are settled before the tool domain runs" + ] + }, + "calls": { + "echo": { + "status": "completed", + "isError": false, + "content": [ + "hello: hi" + ] + }, + "fail": { + "status": "error", + "isError": true, + "content": [ + "the fixture refused" + ] + }, + "structured": { + "status": "completed", + "isError": false, + "content": [ + "{\n \"version\": 2\n}" + ] + }, + "nulls": { + "status": "completed", + "isError": false, + "content": [ + "{\"word\":\"abc\"}" + ] + }, + "pid-before": { + "status": "completed", + "isError": false, + "content": [ + "" + ] + }, + "crash": { + "status": "error", + "isError": true, + "content": [ + "MCP tool \"e2e/crash\" failed: Connection closed: mcp server process exited with code 3: fatal: e2e fixture crashed on purpose" + ] + }, + "pid-after": { + "status": "completed", + "isError": false, + "content": [ + "" + ] + }, + "invalid": { + "status": "error", + "isError": true, + "content": [ + "Invalid arguments for mcp__e2e__echo: Missing key\n at [\"text\"]" + ] + }, + "pattern": { + "status": "error", + "isError": true, + "content": [ + "Invalid arguments for mcp__e2e__args: Expected a string matching the RegExp ^[a-z]+$\n at [\"word\"]" + ] + }, + "excess": { + "status": "error", + "isError": true, + "content": [ + "Invalid arguments for mcp__e2e__args: Expected no excess property\n at [\"extra\"]" + ] + } + }, + "status": [ + { + "server": "e2e", + "state": "connected", + "tools": 8 + }, + { + "server": "off", + "state": "disabled", + "tools": 0 + }, + { + "server": "my-srv", + "state": "disabled", + "tools": 0 + }, + { + "server": "badheader", + "state": "failed", + "error": "MCP server \"badheader\" header \"Authorization\" contains a line break or NUL character", + "tools": 0 + } + ], + "configErrors": [ + "/home/mcp.json: server \"my_srv\" conflicts with \"my-srv\"", + "/home/mcp.json: server \"broken\": url must be an http or https URL", + "/home/mcp.json: server \"signin\": \"oauth\" is not supported yet", + "/project/.codework/mcp.json: server \"ghost\" needs \"command\" or \"url\", or a user server to override" + ] +} diff --git a/packages/harness/test/__artifacts__/mcp.session.lifecycle.json b/packages/harness/test/__artifacts__/mcp.session.lifecycle.json new file mode 100644 index 00000000..0c74d0ca --- /dev/null +++ b/packages/harness/test/__artifacts__/mcp.session.lifecycle.json @@ -0,0 +1,141 @@ +{ + "calls": { + "pid-1": { + "isError": false, + "content": [ + "" + ] + }, + "remote-1": { + "isError": false, + "content": [ + "called" + ] + }, + "remote-retried": { + "isError": false, + "content": [ + "called" + ] + }, + "bye": { + "isError": false, + "content": [ + "bye" + ] + }, + "pid-2": { + "isError": false, + "content": [ + "" + ] + }, + "pid-3": { + "isError": false, + "content": [ + "" + ] + }, + "remote-expired-twice": { + "isError": true, + "content": [ + "MCP tool \"remote/echo\" failed: mcp server session expired: remote" + ] + }, + "pid-4": { + "isError": false, + "content": [ + "" + ] + }, + "echo-changed": { + "isError": false, + "content": [ + "changed: x" + ] + } + }, + "declared": [ + [ + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__secret", + "mcp__e2e__bye", + "mcp__remote__echo" + ], + [ + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__secret", + "mcp__e2e__bye", + "mcp__remote__echo" + ], + [ + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__secret", + "mcp__e2e__bye", + "mcp__remote__echo" + ], + [ + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__secret", + "mcp__remote__echo" + ], + [ + "mcp__remote__echo", + "mcp__e2e__echo", + "mcp__e2e__args", + "mcp__e2e__fail", + "mcp__e2e__pid", + "mcp__e2e__crash", + "mcp__e2e__my_tool_v2", + "mcp__e2e__secret" + ] + ], + "remote": [ + "POST initialize", + "GET", + "POST tools/list", + "POST tools/call", + "POST tools/call", + "POST initialize", + "GET", + "POST tools/list", + "POST tools/call", + "POST tools/call", + "POST initialize", + "GET", + "POST tools/list", + "POST tools/call" + ], + "status": [ + { + "server": "remote", + "state": "connected", + "tools": 1 + }, + { + "server": "e2e", + "state": "connected", + "tools": 8 + } + ] +} diff --git a/packages/harness/test/fixtures/mcp/session.mjs b/packages/harness/test/fixtures/mcp/session.mjs new file mode 100644 index 00000000..845eb900 --- /dev/null +++ b/packages/harness/test/fixtures/mcp/session.mjs @@ -0,0 +1,57 @@ +// Stdio MCP server for the session test; run as `node session.mjs`. +import { Server } from "@modelcontextprotocol/server"; +import { StdioServerTransport } from "@modelcontextprotocol/server/stdio"; + +const object = (properties = {}, required = []) => ({ type: "object", properties, required }); +const text = (value) => ({ content: [{ type: "text", text: String(value) }] }); + +const server = new Server({ name: "session-fixture", version: "1.0.0" }, { capabilities: { tools: {} } }); +server.setRequestHandler("tools/list", () => ({ + tools: [ + { + name: "echo", + description: "Echo text back, prefixed with the configured greeting.", + inputSchema: object({ text: { type: "string", description: "What to echo." } }, ["text"]), + }, + { + name: "args", + description: "The arguments as received.", + inputSchema: { + ...object({ word: { type: "string", pattern: "^[a-z]+$" }, limit: { type: "integer" } }, ["word"]), + additionalProperties: false, + }, + }, + { name: "fail", description: "Answer with an error result.", inputSchema: object() }, + { name: "pid", description: "The server's process id.", inputSchema: object() }, + { name: "crash", description: "Exit while the call is running.", inputSchema: object() }, + { name: "my-tool.v2", inputSchema: { type: "object" } }, + { name: "secret", description: "Never declared.", inputSchema: object() }, + { name: "bye", description: "Answer, then exit while idle.", inputSchema: object() }, + ], +})); +server.setRequestHandler("tools/call", (request) => { + const args = request.params.arguments ?? {}; + switch (request.params.name) { + case "echo": + return text(`${process.env.E2E_GREETING}: ${String(args.text)}`); + case "args": + return text(JSON.stringify(args)); + case "fail": + return { isError: true, content: [{ type: "text", text: "the fixture refused" }] }; + case "pid": + return text(process.pid); + case "crash": + process.stderr.write("fatal: e2e fixture crashed on purpose\n"); + setTimeout(() => process.exit(3), 20); + return new Promise(() => {}); + case "bye": + setTimeout(() => process.exit(0), 20); + return text("bye"); + case "my-tool.v2": + return { content: [], structuredContent: { version: 2 } }; + default: + throw new Error(`unknown tool ${request.params.name}`); + } +}); + +await server.connect(new StdioServerTransport()); diff --git a/packages/harness/test/fixtures/runner.cycle.spec.ts b/packages/harness/test/fixtures/runner.cycle.spec.ts index 174ec4dc..c0439425 100644 --- a/packages/harness/test/fixtures/runner.cycle.spec.ts +++ b/packages/harness/test/fixtures/runner.cycle.spec.ts @@ -29,6 +29,8 @@ import { Session } from "../../src/session/session.ts"; import { builtins } from "../../src/plugin/builtin.ts"; import { State } from "../../src/state/state.ts"; import { pooled } from "./pool.ts"; +import { Mcp } from "../../src/mcp/mcp.ts"; +import { Process } from "../../src/sandbox/utils/process.ts"; const database = Database.layer(":memory:"); const vercel = VercelSandboxDriver.make(); @@ -42,6 +44,7 @@ const runtime = (root: string, custom: string) => Layer.provideMerge( ((seeded) => State.layer({}, seeded.ref, seeded.references, seeded.follow, seeded.rebuild))(pooled(builtins)), ), + Layer.provideMerge(Mcp.layer({ hostCwd: "/", home: "/unused/home" }).pipe(Layer.provide(Process.host))), Layer.provideMerge(SessionRuntime.layer), Layer.provideMerge( Settings.layer({ userConfigDir: custom }).pipe( diff --git a/packages/harness/test/mcp.session.test.ts b/packages/harness/test/mcp.session.test.ts new file mode 100644 index 00000000..4782bc2c --- /dev/null +++ b/packages/harness/test/mcp.session.test.ts @@ -0,0 +1,332 @@ +import "./utils/env.ts"; +import { createAssistantMessageEventStream, type Message } from "@codeworksh/aikit"; +import { Effect } from "effect"; +import fs from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, it } from "vite-plus/test"; +import { ContextCodec } from "../src/context/codec.ts"; +import { Harness } from "../src/effect/harness.ts"; +import { Session } from "../src/effect/session.ts"; +import { Mcp } from "../src/mcp/mcp.ts"; +import { defaultPromptPlugin } from "../src/plugin/builtin/prompt/default.ts"; +import { define } from "../src/plugin/plugin.ts"; +import type { LLM } from "../src/runner/llm.ts"; +import { Settings } from "../src/settings/settings.ts"; +import { assistant } from "./fixtures/llm.ts"; +import { listen, raw } from "./fixtures/mcp/http.ts"; +import { tmpdir } from "./fixtures/tempdir.ts"; +import { pendingCall } from "./tools.fixture.ts"; + +const fixture = fileURLToPath(new URL("./fixtures/mcp/session.mjs", import.meta.url)); +const stdio = (extra: Record = {}) => ({ + command: process.execPath, + args: [fixture], + env: { E2E_GREETING: "${MCP_E2E_GREETING}" }, + ...extra, +}); + +/** One reply per model request, in order; an empty one ends the exchange. */ +const scripted = + (replies: ReadonlyArray>, requests: LLM.Input[]): LLM.Open => + (input) => + Effect.sync(() => { + const index = requests.push(input); + const calls = replies[index - 1] ?? []; + const message = assistant(input, index, calls.length > 0 ? { stopReason: "toolUse", parts: [...calls] } : {}); + const stream = createAssistantMessageEventStream(); + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: calls.length > 0 ? "toolUse" : "stop", message }); + return stream; + }); + +const write = async (file: string, value: unknown) => { + await fs.mkdir(join(file, ".."), { recursive: true }); + await fs.writeFile(file, JSON.stringify(value, null, 2)); +}; + +it("mcp: config layers, plugin bucket, direct tools, a crash mid-exchange and the next call reconnecting", async () => { + await using dir = await tmpdir(); + const home = join(dir.path, "home"); + const project = join(dir.path, "project"); + const work = join(dir.path, "work"); + await fs.mkdir(work, { recursive: true }); + process.env.MCP_E2E_GREETING = "hello"; + // A `!command` printing a notice after the token, say: never sent, and never shown. + process.env.MCP_E2E_BAD_HEADER = "Bearer sk-e2e-secret\nnotice"; + // Unlike a header, a variable may span lines. + process.env.MCP_E2E_PEM = "-----BEGIN KEY-----\nabc\n-----END KEY-----"; + + await write(join(home, "mcp.json"), { + mcpServers: { + e2e: stdio({ + toolExposure: { secret: "hidden" }, + env: { E2E_GREETING: "${MCP_E2E_GREETING}", E2E_PEM: "${MCP_E2E_PEM}" }, + }), + off: stdio(), + "my-srv": stdio({ enabled: false }), + my_srv: stdio(), + broken: { url: "ftp://example.com/mcp" }, + signin: { url: "https://example.com/mcp", oauth: { clientId: "e2e" } }, + badheader: { url: "http://127.0.0.1:9/mcp", headers: { Authorization: "${MCP_E2E_BAD_HEADER}" } }, + }, + }); + // The project turns a user server off without restating it, and cannot add an override with no base. + await write(join(project, ".codework", "mcp.json"), { + mcpServers: { off: { enabled: false }, ghost: { enabled: false } }, + }); + + const observed: { readonly mcpDomain: unknown[]; readonly toolDomain: string[] } = { mcpDomain: [], toolDomain: [] }; + const observer = define({ + id: "test.mcp.observer", + kind: "mcp", + setup: (ctx) => { + observed.mcpDomain.push(ctx.plugin.mcp.list().map(([name, config]) => [name, config.enabled ?? true])); + ctx.plugin.mcp.update("e2e", { description: "The E2E fixture." }); + }, + }); + const latecomer = define({ + id: "test.mcp.latecomer", + kind: "tool", + setup: (ctx) => { + try { + ctx.plugin.mcp.remove("e2e"); + } catch (error) { + observed.toolDomain.push((error as Error).message); + } + }, + }); + + const requests: LLM.Input[] = []; + const llm = scripted( + [ + [ + pendingCall("mcp__e2e__echo", { text: "hi" }, "echo"), + pendingCall("mcp__e2e__fail", {}, "fail"), + pendingCall("mcp__e2e__my_tool_v2", {}, "structured"), + // Strict sampling writes `null` for an omitted optional argument; the server never sees it. + pendingCall("mcp__e2e__args", { word: "abc", limit: null }, "nulls"), + ], + [pendingCall("mcp__e2e__pid", {}, "pid-before"), pendingCall("mcp__e2e__crash", {}, "crash")], + // Same exchange: the server is gone, so this call reconnects first. + [pendingCall("mcp__e2e__pid", {}, "pid-after")], + [], + // Next exchange: arguments the server's schema rejects never reach it. + [ + pendingCall("mcp__e2e__echo", {}, "invalid"), + pendingCall("mcp__e2e__args", { word: "ABC" }, "pattern"), + pendingCall("mcp__e2e__args", { word: "abc", extra: true }, "excess"), + ], + [], + ], + requests, + ); + + const result = await Effect.gen(function* () { + const session = yield* Session.create({ + directory: work, + hostDir: project, + model: { provider: "openai", id: "gpt-5.6-luna" }, + }); + yield* session.run("Use the fixture."); + yield* session.run("Once more."); + const path = yield* session.path(); + const messages = yield* Effect.forEach(path, ContextCodec.decodeMessage); + const status = yield* Mcp.Service.use((mcp) => mcp.status(session.id)); + const settings = yield* Settings.Service.use((loader) => loader.load(project)); + return { messages, status, errors: settings.mcp.errors }; + }).pipe( + Effect.provide( + Harness.layer({ + home, + hostCwd: dir.path, + database: ":memory:", + plugins: [observer, latecomer, defaultPromptPlugin], + llm, + }), + ), + Effect.scoped, + Effect.runPromise, + ); + + const calls = new Map( + result.messages + .flatMap((message) => message.parts) + .flatMap((part) => (part.type === "toolCall" && "result" in part ? [[part.callID, part] as const] : [])), + ); + const outcome = (id: string) => { + const part = calls.get(id); + if (part === undefined) return undefined; + return { + status: part.status, + isError: part.result.isError, + content: part.result.content.map((block) => (block.type === "text" ? block.text : `[${block.type}]`)), + }; + }; + + const pidBefore = outcome("pid-before")?.content[0]; + const pidAfter = outcome("pid-after")?.content[0]; + expect(pidBefore).toMatch(/^\d+$/); + expect(pidAfter).toMatch(/^\d+$/); + expect(pidAfter).not.toBe(pidBefore); + + const firstTools = requests[0]?.context.tools ?? []; + expect(firstTools.map((tool) => tool.name)).not.toContain("mcp__e2e__secret"); + // The tool set is frozen per exchange and stable across them, so every request declares the same tools. + for (const request of requests) expect(request.context.tools).toEqual(firstTools); + // The server's JSON Schema reaches the provider as written. + expect(firstTools.find((tool) => tool.name === "mcp__e2e__echo")?.parameters).toEqual({ + type: "object", + properties: { text: { type: "string", description: "What to echo." } }, + required: ["text"], + }); + + const normalise = (value: unknown) => + JSON.parse( + JSON.stringify(value) + .replaceAll(dir.path, "") + .replaceAll(process.execPath, "") + .replaceAll(`"${pidBefore}"`, '""') + .replaceAll(`"${pidAfter}"`, '""'), + ); + const artifact = normalise({ + requests: requests.length, + tools: firstTools.map((tool) => tool.name), + echoTool: firstTools.find((tool) => tool.name === "mcp__e2e__echo"), + plugins: observed, + calls: Object.fromEntries( + [ + "echo", + "fail", + "structured", + "nulls", + "pid-before", + "crash", + "pid-after", + "invalid", + "pattern", + "excess", + ].map((id) => [id, outcome(id)]), + ), + status: result.status, + configErrors: result.errors, + }); + expect(JSON.stringify(result)).not.toContain("sk-e2e-secret"); + await expect(`${JSON.stringify(artifact, null, "\t")}\n`).toMatchFileSnapshot("./__artifacts__/mcp.session.json"); +}, 60_000); + +it("mcp: a server exiting while idle, reconfiguration, and an expired HTTP session", async () => { + await using dir = await tmpdir(); + const home = join(dir.path, "home"); + const work = join(dir.path, "work"); + await fs.mkdir(work, { recursive: true }); + process.env.MCP_E2E_GREETING = "hello"; + + // Expires the HTTP session on the next `expire` tool calls: the server answers 404 to the session. + let expire = 0; + const remote = raw({ + script: (request) => { + if (request !== "POST tools/call" || expire === 0) return undefined; + expire -= 1; + return { status: 404, body: "session not found" }; + }, + }); + const configure = (url: URL, e2e: Record) => + write(join(home, "mcp.json"), { mcpServers: { e2e: stdio(e2e), remote: { url: url.href } } }); + + const runs: ReadonlyArray> = [ + [pendingCall("mcp__e2e__pid", {}, "pid-1"), pendingCall("mcp__remote__echo", {}, "remote-1")], + [pendingCall("mcp__remote__echo", {}, "remote-retried"), pendingCall("mcp__e2e__bye", {}, "bye")], + // The server exited while idle; the next call starts it again. + [pendingCall("mcp__e2e__pid", {}, "pid-2")], + // Only exposure changed: the same process, with `bye` no longer declared. + [pendingCall("mcp__e2e__pid", {}, "pid-3"), pendingCall("mcp__remote__echo", {}, "remote-expired-twice")], + // The environment changed: a new process. + [pendingCall("mcp__e2e__pid", {}, "pid-4"), pendingCall("mcp__e2e__echo", { text: "x" }, "echo-changed")], + ]; + const requests: LLM.Input[] = []; + const llm = scripted( + runs.flatMap((calls) => [calls, []]), + requests, + ); + + const result = await Effect.gen(function* () { + const url = yield* listen(remote.handler); + yield* Effect.promise(() => configure(url, {})); + const session = yield* Session.create({ directory: work, model: { provider: "openai", id: "gpt-5.6-luna" } }); + const status = Mcp.Service.use((mcp) => mcp.status(session.id)); + const state = (server: string) => + status.pipe(Effect.map((all) => all.find((one) => one.server === server)?.state)); + + yield* session.run("one"); + expire = 1; + yield* session.run("two"); + // Wait for the idle exit to be noticed, so the next call meets a dropped connection. + for (let tries = 0; tries < 250 && (yield* state("e2e")) !== "disconnected"; tries++) + yield* Effect.sleep("20 millis"); + const afterBye = yield* state("e2e"); + yield* session.run("three"); + yield* Effect.promise(() => configure(url, { toolExposure: { bye: "hidden" } })); + expire = 2; + yield* session.run("four"); + yield* Effect.promise(() => + configure(url, { toolExposure: { bye: "hidden" }, env: { E2E_GREETING: "changed" } }), + ); + yield* session.run("five"); + + const messages = yield* Effect.forEach(yield* session.path(), ContextCodec.decodeMessage); + return { messages, afterBye, status: yield* status }; + }).pipe( + Effect.provide( + Harness.layer({ home, hostCwd: dir.path, database: ":memory:", plugins: [defaultPromptPlugin], llm }), + ), + Effect.scoped, + Effect.runPromise, + ); + + const texts = new Map( + result.messages + .flatMap((message) => message.parts) + .flatMap((part) => + part.type === "toolCall" && "result" in part + ? [ + [ + part.callID, + { + isError: part.result.isError, + content: part.result.content.map((block) => (block.type === "text" ? block.text : "")), + }, + ] as const, + ] + : [], + ), + ); + const pid = (id: string) => texts.get(id)?.content[0]; + expect(result.afterBye).toBe("disconnected"); + expect(pid("pid-2")).not.toBe(pid("pid-1")); + expect(pid("pid-3")).toBe(pid("pid-2")); + expect(pid("pid-4")).not.toBe(pid("pid-3")); + + // The first request of each exchange: exposure changes apply at the next exchange. + const declared = runs.map((_, index) => requests[index * 2]?.context.tools?.map((tool) => tool.name) ?? []); + expect(declared[2]).toContain("mcp__e2e__bye"); + expect(declared[3]).not.toContain("mcp__e2e__bye"); + + const pids = [...new Set(["pid-1", "pid-2", "pid-3", "pid-4"].map(pid))]; + let serialized = JSON.stringify({ + calls: Object.fromEntries(texts), + declared, + // One expiry: a second initialize and the call once more. Two in a row: the call fails. + remote: remote.seen.filter((request) => request !== "POST notifications/initialized"), + status: result.status, + }); + pids.forEach((value, index) => { + serialized = serialized.replaceAll(`"${value}"`, `""`); + }); + const artifact = JSON.parse( + serialized.replaceAll(dir.path, "").replace(/127\.0\.0\.1:\d+/g, "127.0.0.1:"), + ); + await expect(`${JSON.stringify(artifact, null, "\t")}\n`).toMatchFileSnapshot( + "./__artifacts__/mcp.session.lifecycle.json", + ); +}, 60_000); diff --git a/packages/harness/test/plugin.host.test.ts b/packages/harness/test/plugin.host.test.ts index a176dc1e..e1d8f56b 100644 --- a/packages/harness/test/plugin.host.test.ts +++ b/packages/harness/test/plugin.host.test.ts @@ -27,7 +27,7 @@ const echo = (value: string) => describe("plugin domains and exchange host", () => { it("replaces tool and hooks together, patches prose, and closes retained buckets", async () => { - const buckets = make(); + const buckets = make([]); const tools = buckets.registry.tools; let stale = 0; tools.add(echo("first"), { diff --git a/packages/harness/test/runner.loop.test.ts b/packages/harness/test/runner.loop.test.ts index 09ae6c5c..c6336bf2 100644 --- a/packages/harness/test/runner.loop.test.ts +++ b/packages/harness/test/runner.loop.test.ts @@ -36,6 +36,8 @@ import * as Tool from "../src/tool/tool.ts"; import { assistant, immediateOpen } from "./fixtures/llm.ts"; import { seedSpace } from "./fixtures/space.ts"; import { testEffect } from "./utils/effect.ts"; +import { Mcp } from "../src/mcp/mcp.ts"; +import { Process } from "../src/sandbox/utils/process.ts"; const runtime = ( options: { @@ -61,6 +63,7 @@ const runtime = ( pooled(options.plugins ?? plugins), ), ), + Layer.provideMerge(Mcp.layer({ hostCwd: "/", home: "/unused/home" }).pipe(Layer.provide(Process.host))), Layer.provideMerge(SessionRuntime.layer), Layer.provideMerge( Layer.succeed(Settings.Service, { diff --git a/packages/plugin/package.json b/packages/plugin/package.json index 319d976f..bea43e25 100644 --- a/packages/plugin/package.json +++ b/packages/plugin/package.json @@ -53,6 +53,12 @@ "import": "./dist/pack/plugin/prompt.mjs", "default": "./dist/pack/plugin/prompt.mjs" }, + "./plugin/mcp": { + "development": "./src/plugin/mcp.ts", + "types": "./dist/pack/plugin/mcp.d.mts", + "import": "./dist/pack/plugin/mcp.mjs", + "default": "./dist/pack/plugin/mcp.mjs" + }, "./plugin/section": { "development": "./src/plugin/section.ts", "types": "./dist/pack/plugin/section.d.mts", diff --git a/packages/plugin/src/plugin.ts b/packages/plugin/src/plugin.ts index 4c285b27..70df15a6 100644 --- a/packages/plugin/src/plugin.ts +++ b/packages/plugin/src/plugin.ts @@ -8,6 +8,7 @@ */ export type { Config, Events, Paths, PromptResolver, SharedPluginContext } from "./plugin/context.ts"; export { define, domains, type Mount, type Plugin, type PluginKind, rank } from "./plugin/plugin.ts"; +export * as Mcp from "./plugin/mcp.ts"; export * as Prompt from "./plugin/prompt.ts"; export * as Section from "./plugin/section.ts"; export type { PluginOptions, PluginPatch, PluginRef, PluginSpec } from "./plugin/ref.ts"; diff --git a/packages/plugin/src/plugin/mcp.ts b/packages/plugin/src/plugin/mcp.ts new file mode 100644 index 00000000..742d8290 --- /dev/null +++ b/packages/plugin/src/plugin/mcp.ts @@ -0,0 +1,69 @@ +/** + * MCP servers a session runs, as plugins see and edit them. + * + * The harness seeds the bucket from `mcp.json` (the user file, then the project's), `mcp` plugins + * edit it, and it freezes before the `tool` domain runs: the servers are settled before any tool + * exists. Only configuration lives here. Whether a server is connected changes on its own, at any + * moment, so it is never part of what a plugin reads during setup. + */ + +/** + * How a server's tools reach the model. + * + * - `direct`: declared to the model like any other tool. + * - `hidden`: never registered. + */ +export type Exposure = "direct" | "hidden"; + +interface Base { + /** Default: `direct`. */ + readonly exposure?: Exposure; + /** + * Exposure of single tools, overriding `exposure`. Keys are tool names as the server offers them, + * or patterns where `*` matches any characters. An exact name wins over patterns; among patterns + * the first match wins. `"exposure": "hidden"` with a few entries here exposes only those tools. + */ + readonly toolExposure?: Readonly>; + /** `false` keeps the entry without connecting. Default: `true`. */ + readonly enabled?: boolean; + /** Per-request timeout in seconds. Progress notifications from the server reset it. Default: 60. */ + readonly timeout?: number; + /** What the server offers, in a sentence. */ + readonly description?: string; +} + +/** + * A server the harness starts as a process on the host. `env` values may name environment + * variables (`${NAME}`) or be a command whose output is the value (`!command`). + */ +export interface StdioServer extends Base { + readonly type?: "stdio"; + /** One executable, not a shell command line. */ + readonly command: string; + readonly args?: ReadonlyArray; + readonly env?: Readonly>; + /** Relative to the session's directory. */ + readonly cwd?: string; +} + +/** A Streamable HTTP server. `headers` values resolve like a stdio server's `env`. */ +export interface HttpServer extends Base { + readonly type?: "http" | "streamable-http"; + readonly url: string; + readonly headers?: Readonly>; +} + +export type ServerConfig = StdioServer | HttpServer; + +/** The settings a plugin can change on a server without restating where it runs. */ +export type ServerPatch = Pick; + +export interface McpRegistry { + readonly list: () => ReadonlyArray; + readonly get: (name: string) => ServerConfig | undefined; + /** Add a server or replace one. Throws on an invalid name or config. */ + readonly set: (name: string, config: ServerConfig) => void; + /** Throws when no server has this name. */ + readonly update: (name: string, patch: ServerPatch) => void; + readonly remove: (name: string) => void; +} diff --git a/packages/plugin/src/plugin/plugin.ts b/packages/plugin/src/plugin/plugin.ts index 8c99ebbd..b36bbfe4 100644 --- a/packages/plugin/src/plugin/plugin.ts +++ b/packages/plugin/src/plugin/plugin.ts @@ -19,10 +19,13 @@ import type { PluginOptions } from "./ref.ts"; * composition actually happens -- a plugin patching another's tool, or appending to the prompt a * previous one rendered, is written next to it and deliberately after it. * + * `mcp` runs first: the MCP servers a session runs are settled before any tool exists, because + * their tools join the tool registry before the `tool` domain starts. + * * Kept as data so a later release can add a domain, or make the order configurable, without * touching the loader: `{ tool: 1, skill: 2, prompt: 3 }` needs no code change here. */ -export const domains = { tool: 1, prompt: 2 } as const; +export const domains = { mcp: 1, tool: 2, prompt: 3 } as const; export type PluginKind = keyof typeof domains; @@ -33,9 +36,9 @@ export type Mount = SandboxIO.Provides | LocationService; export interface Plugin { readonly id: string; /** - * The domain this plugin extends, which decides when it runs: every `tool` plugin is set up - * before any `prompt` plugin, whichever settings layer each came from, so a prompt plugin - * always sees the complete tool set. Plugins within one domain run in the order they were + * The domain this plugin extends, which decides when it runs: every `mcp` plugin is set up + * before any `tool` plugin, and every `tool` plugin before any `prompt` plugin, whichever + * settings layer each came from, so a prompt plugin always sees the complete tool set. Plugins within one domain run in the order they were * written. */ readonly kind: PluginKind; diff --git a/packages/plugin/src/plugin/registry.ts b/packages/plugin/src/plugin/registry.ts index 0cef377c..ece204f9 100644 --- a/packages/plugin/src/plugin/registry.ts +++ b/packages/plugin/src/plugin/registry.ts @@ -1,3 +1,4 @@ +import type { McpRegistry } from "./mcp.ts"; import type { PromptRegistry } from "./prompt.ts"; import type { ToolRegistry } from "./tool.ts"; @@ -9,6 +10,8 @@ import type { ToolRegistry } from "./tool.ts"; * effective tool catalog and the final system prompt — is the harness's job, not the contract's. */ export interface PluginRegistry { + /** Writable during the `mcp` domain only; later domains read the servers it settled. */ + readonly mcp: McpRegistry; readonly tools: ToolRegistry; readonly prompt: PromptRegistry; } diff --git a/packages/plugin/src/settings.ts b/packages/plugin/src/settings.ts index b16df1c2..dd26b16a 100644 --- a/packages/plugin/src/settings.ts +++ b/packages/plugin/src/settings.ts @@ -1,5 +1,6 @@ import { Model as AikitModel, type Protocol } from "@codeworksh/aikit"; import { Effect, Schema, type SchemaAST, SchemaGetter, SchemaIssue } from "effect"; +import type { ServerConfig } from "./plugin/mcp.ts"; import { isAikitModelInfo, NonNegativeInt, validateAikitModelInfo } from "./schema.ts"; export const ThinkingLevel = Schema.Literals(Object.values(AikitModel.ThinkingLevelEnum)); @@ -203,6 +204,22 @@ export interface DeclaredModel { readonly file: string; } +/** One server from an `mcp.json`, and the file that defined it. */ +export interface McpServer { + readonly name: string; + readonly config: ServerConfig; + readonly file: string; + /** The project `mcp.json` whose entry overrides this user server's `enabled` or exposure. */ + readonly override?: string; +} + +/** The `mcp.json` layers: the user file, then the project's, whose entries replace same-named ones. */ +export interface McpSettings { + readonly servers: ReadonlyArray; + /** Files and entries that could not be used. Each is skipped; everything else still loads. */ + readonly errors: ReadonlyArray; +} + export interface Info { /** * Plugin entries added to the harness selection, in order: every layer's entries, lowest @@ -226,6 +243,7 @@ export interface Info { readonly toolExecution: ToolExecution; }; readonly retry: RetryPolicy; + readonly mcp: McpSettings; } /** Let aikit supply model-aware generation defaults. */ @@ -241,4 +259,5 @@ export const defaults: Info = { options: { timeoutMs: 3_600_000, idleTimeoutMs: 300_000, maxRetries: 0 }, }, retry: { maxRetries: 3, baseDelayMs: 2_000, maxDelayMs: 60_000 }, + mcp: { servers: [], errors: [] }, }; diff --git a/packages/plugin/src/tool.ts b/packages/plugin/src/tool.ts index a1b4f437..4457a717 100644 --- a/packages/plugin/src/tool.ts +++ b/packages/plugin/src/tool.ts @@ -1,5 +1,5 @@ import type { Message } from "@codeworksh/aikit"; -import { Effect, type Layer, Schema, SchemaAST } from "effect"; +import { Effect, JsonSchema, type Layer, Schema, SchemaAST, SchemaRepresentation } from "effect"; import type { TSchema } from "typebox"; import { ToolExecutionError } from "./tool/error.ts"; import type { ToolProgress } from "./tool/progress.ts"; @@ -233,12 +233,47 @@ export const provide = < ToolProgress >); +const providerJsonSchema = "codework/providerJsonSchema"; + +/** + * Parameters from a JSON Schema someone else wrote, such as an MCP server's `inputSchema`. The + * provider receives it exactly as written; arguments are validated against its Effect translation, + * `pattern`s included: the user chose the server. A schema the translation cannot express (an + * external `$ref`, a pattern invalid in Unicode mode) validates nothing, since rejecting arguments + * the schema allows would be worse than letting the server check them. + */ +export const fromJsonSchema = (schema: Record): Schema.Top => { + const validator = (() => { + try { + const draft07 = + (typeof schema.$schema === "string" && schema.$schema.includes("draft-07")) || "definitions" in schema; + return SchemaRepresentation.fromJsonSchemaDocument( + draft07 ? JsonSchema.fromSchemaDraft07(schema) : JsonSchema.fromSchemaDraft2020_12(schema), + { patterns: "apply" }, + ); + } catch { + return Schema.Unknown; + } + })(); + return validator.annotate({ [providerJsonSchema]: schema }); +}; + +/** + * Whether `schema` came from {@link fromJsonSchema}. Its arguments are checked as JSON Schema checks + * them: a key a closed object does not declare is an error, not dropped, since the tool receives the + * arguments as written. + */ +export const isFromJsonSchema = (schema: Schema.Top) => + SchemaAST.resolveAt>(providerJsonSchema)(schema.ast) !== undefined; + /** * Derive a provider-safe JSON Schema object from an Effect Schema. This is the * single boundary that crosses into the provider wire; if a provider rejects a * construct, fix it here. Draft-07 inlines the root (no top-level `$ref`). */ export const toProviderJsonSchema = (schema: Schema.Top): Record => { + const written = SchemaAST.resolveAt>(providerJsonSchema)(schema.ast); + if (written !== undefined) return written; // Effect reads `Struct({})` as TS `{}` (any non-null) and emits `{ not: { type: "null" } }`; // providers require an object, so emit one shaped like a non-empty struct's. const encoded = SchemaAST.toEncoded(schema.ast); diff --git a/packages/plugin/vite.config.ts b/packages/plugin/vite.config.ts index e1bbb3e0..d0454f7c 100644 --- a/packages/plugin/vite.config.ts +++ b/packages/plugin/vite.config.ts @@ -28,6 +28,7 @@ export default defineConfig({ "src/index.ts", "src/location.ts", "src/plugin.ts", + "src/plugin/mcp.ts", "src/plugin/prompt.ts", "src/plugin/section.ts", "src/plugin/tool.ts",