diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 225919c27f..d313c0227f 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -2,8 +2,10 @@ export * as MCP from "./index" import { Mcp } from "@opencode-ai/schema/mcp" import { McpEvent } from "@opencode-ai/schema/mcp-event" +import { Event } from "@opencode-ai/schema/config" import { Command } from "@opencode-ai/schema/command" import { createHash } from "node:crypto" +import { isDeepStrictEqual } from "node:util" import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Stream } from "effect" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Config } from "../config" @@ -180,26 +182,37 @@ export const layer = (options?: Options) => Layer.effect( const fork = yield* FiberSet.makeRuntime() yield* Effect.addFinalizer((exit) => Scope.close(root, exit)) - const documents = (yield* config.entries()).filter((entry): entry is Config.Document => entry.type === "document") - // Global MCP timeout defaults, later config files overriding earlier ones. - const timeout = Object.assign( - {}, - ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])), - ) + const load = Effect.fnUntraced(function* () { + const documents = (yield* config.entries()).filter( + (entry): entry is Config.Document => entry.type === "document", + ) + // Global MCP timeout defaults, later config files overriding earlier ones. + const timeout = Object.assign( + {}, + ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])), + ) + const servers = new Map() + for (const entry of documents) { + for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) { + servers.set(ServerName.make(name), { ...server, timeout: { ...timeout, ...server.timeout } }) + } + } + return { timeout, servers } + }) + const initial = yield* load() + const configured = { names: new Set(initial.servers.keys()), timeout: initial.timeout } // Later config files win for duplicate server names; per-server timeout overrides globals. const runtime = new Map() // Serializes lifecycle operations per server. Anything taking this lock from a connection // callback must stay forked: lifecycle operations close scopes while holding it, firing onClose. const locks = KeyedMutex.makeUnsafe() const urlElicitations = new Map() - for (const entry of documents) { - for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) { - runtime.set(ServerName.make(name), { - config: { ...server, timeout: { ...timeout, ...server.timeout } }, - status: { status: "pending" }, - startup: Deferred.makeUnsafe(), - }) - } + for (const [name, server] of initial.servers) { + runtime.set(name, { + config: server, + status: { status: "pending" }, + startup: Deferred.makeUnsafe(), + }) } // Register every remote server as an OAuth integration so credentials live in the global store @@ -560,6 +573,49 @@ export const layer = (options?: Options) => Layer.effect( yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) }) + const disposeServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { + yield* stopServer(name, entry) + if (entry.integrationID) owned.delete(entry.integrationID) + if (entry.registration) yield* entry.registration.dispose + }) + + const replaceServer = Effect.fnUntraced(function* ( + name: ServerName, + config: typeof ConfigMCP.Server.Type, + ) { + const previous = runtime.get(name) + if (previous) yield* disposeServer(name, previous) + const entry: ServerEntry = { + config, + status: { status: "pending" }, + startup: Deferred.makeUnsafe(), + } + runtime.set(name, entry) + yield* Effect.gen(function* () { + yield* register(name, entry) + if (config.disabled) { + entry.status = { status: "disabled" } + yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + return + } + yield* startServer(name, entry) + }).pipe( + // Settle startup even when register fails or replacement is interrupted, so an entry that made it + // into runtime can never hang readers awaiting its startup. + Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))), + ) + }) + + const removeServer = Effect.fnUntraced(function* (name: ServerName) { + const entry = runtime.get(name) + if (!entry) return + yield* disposeServer(name, entry) + // Credentials are kept: they are keyed by name + url, so re-adding the same server + // reuses them without forcing re-auth, matching add()'s replacement semantics. + runtime.delete(name) + yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + }) + // Disabled servers settle their startup immediately so queries never block on them. for (const [name, entry] of runtime) { if (entry.config.disabled) { @@ -594,6 +650,29 @@ export const layer = (options?: Options) => Layer.effect( ), ) + fork( + events.subscribe(Event.Updated).pipe( + Stream.runForEach(() => + Effect.gen(function* () { + const next = yield* load() + configured.timeout = next.timeout + const names = new Set([...configured.names, ...next.servers.keys()]) + for (const name of names) { + const updated = next.servers.get(name) + if (updated) { + if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue + yield* replaceServer(name, updated).pipe(locks.withLock(name)) + continue + } + yield* removeServer(name).pipe(locks.withLock(name)) + } + configured.names = new Set(next.servers.keys()) + }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause }))), + ), + Effect.ignore, + ), + ) + // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run. const whenAllReady = Effect.suspend(() => Effect.forEach(Array.from(runtime.values()), (entry) => Deferred.await(entry.startup), { @@ -615,33 +694,9 @@ export const layer = (options?: Options) => Layer.effect( }), add: Effect.fn("MCP.add")(function* (server, config) { const name = ServerName.make(server) - yield* Effect.gen(function* () { - const previous = runtime.get(name) - if (previous) { - yield* stopServer(name, previous) - if (previous.integrationID) owned.delete(previous.integrationID) - if (previous.registration) yield* previous.registration.dispose - } - const entry: ServerEntry = { - config: { ...config, timeout: { ...timeout, ...config.timeout } }, - status: { status: "pending" }, - startup: Deferred.makeUnsafe(), - } - runtime.set(name, entry) - yield* Effect.gen(function* () { - yield* register(name, entry) - if (config.disabled) { - entry.status = { status: "disabled" } - yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) - return - } - yield* startServer(name, entry) - }).pipe( - // Settle startup even when register fails or add is interrupted, so an entry that made it - // into runtime can never hang readers awaiting its startup. - Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))), - ) - }).pipe(locks.withLock(name)) + yield* replaceServer(name, { ...config, timeout: { ...configured.timeout, ...config.timeout } }).pipe( + locks.withLock(name), + ) }), connect: Effect.fn("MCP.connect")(function* (server) { const name = ServerName.make(server) @@ -663,14 +718,8 @@ export const layer = (options?: Options) => Layer.effect( remove: Effect.fn("MCP.remove")(function* (server) { const name = ServerName.make(server) yield* Effect.gen(function* () { - const target = yield* requireServer(name) - yield* stopServer(name, target.entry) - if (target.entry.integrationID) owned.delete(target.entry.integrationID) - if (target.entry.registration) yield* target.entry.registration.dispose - // Credentials are kept: they are keyed by name + url, so re-adding the same server - // reuses them without forcing re-auth, matching add()'s replacement semantics. - runtime.delete(name) - yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + yield* requireServer(name) + yield* removeServer(name) }).pipe(locks.withLock(name)) }), tools: Effect.fn("MCP.tools")(function* () { diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 158ad32c67..55384e7783 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -1,5 +1,6 @@ import path from "node:path" import { describe, expect, test } from "bun:test" +import { Event } from "@opencode-ai/schema/config" import { Client } from "@modelcontextprotocol/sdk/client/index.js" import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js" import { Server } from "@modelcontextprotocol/sdk/server/index.js" @@ -28,7 +29,7 @@ import { SessionV2 } from "@opencode-ai/core/session" import { McpTool } from "@opencode-ai/core/tool/mcp" import { ToolRegistry } from "@opencode-ai/core/tool/registry" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" -import { Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect" +import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Stream } from "effect" import { Image } from "@opencode-ai/core/image" import { testEffect } from "./lib/effect" import { imagePassthrough } from "./lib/image" @@ -153,6 +154,10 @@ function resourceMcpLayer( server: string | typeof ConfigMCP.Server.Type, onFormCreated?: (form: Form.Info) => Effect.Effect, options?: MCP.Options, + overrides?: { + entries?: Config.Interface["entries"] + subscribe?: EventV2.Interface["subscribe"] + }, ) { const directory = AbsolutePath.make(import.meta.dir) const unusedIntegration = () => Effect.die("unused integration service") @@ -163,27 +168,29 @@ function resourceMcpLayer( Layer.succeed( Config.Service, Config.Service.of({ - entries: () => - Effect.succeed([ - new Config.Document({ - type: "document", - info: new Config.Info({ - mcp: new ConfigMCP.Info({ - servers: { - resources: - typeof server === "string" - ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) - : server, - }, + entries: + overrides?.entries ?? + (() => + Effect.succeed([ + new Config.Document({ + type: "document", + info: new Config.Info({ + mcp: new ConfigMCP.Info({ + servers: { + resources: + typeof server === "string" + ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) + : server, + }, + }), }), }), - }), - ]), + ])), }), ), Layer.succeed(Location.Service, Location.Service.of(location({ directory }))), Layer.mock(EventV2.Service, { - subscribe: () => Stream.never, + subscribe: overrides?.subscribe ?? (() => Stream.never), publish: (definition, data) => { const event = { id: EventV2.ID.create(), @@ -725,6 +732,76 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => { ) }) +test("reconciles MCP servers when config updates", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const updates = yield* PubSub.unbounded() + const command = [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")] + let servers: Record = { + configured: new ConfigMCP.Local({ type: "local", command }), + } + const entries = () => + Effect.sync(() => [ + new Config.Document({ + type: "document", + info: new Config.Info({ mcp: new ConfigMCP.Info({ servers }) }), + }), + ]) + const subscribe = (() => Stream.fromPubSub(updates)) as EventV2.Interface["subscribe"] + const update = () => + PubSub.publish(updates, { + id: EventV2.ID.create(), + created: DateTime.makeUnsafe(0), + type: Event.Updated.type, + data: {}, + } satisfies EventV2.Payload) + + yield* Effect.gen(function* () { + const service = yield* MCP.Service + yield* service.tools() + expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + + servers = { + configured: new ConfigMCP.Local({ type: "local", command, disabled: true }), + } + yield* update() + const disabled = yield* service.servers().pipe( + Effect.filterOrFail( + (items) => items[0]?.status.status === "disabled", + () => new Error("MCP config update was not applied"), + ), + Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }), + ) + expect(disabled.map((item) => String(item.name))).toEqual(["configured"]) + + servers = { + added: new ConfigMCP.Local({ type: "local", command, disabled: true }), + } + yield* update() + const replaced = yield* service.servers().pipe( + Effect.filterOrFail( + (items) => items.length === 1 && String(items[0]?.name) === "added", + () => new Error("MCP config replacement was not applied"), + ), + Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }), + ) + expect(replaced[0]?.status).toEqual({ status: "disabled" }) + }).pipe( + Effect.provide( + resourceMcpLayer( + new ConfigMCP.Local({ type: "local", command, disabled: true }), + undefined, + undefined, + { entries, subscribe }, + ), + ), + ) + }), + ), + ) +}) + test("serializes concurrent MCP lifecycle operations", async () => { await Effect.runPromise( Effect.scoped(