fix(core): reload changed MCP config (#41204)

Co-authored-by: Aiden Cline <rekram1-node@users.noreply.github.com>

packages/core/src/mcp/index.ts

@@ -3,10 +3,11 @@ export * as MCP from "./index"
 import { Mcp } from "@opencode-ai/schema/mcp"
 import { McpEvent } from "@opencode-ai/schema/mcp-event"
 import { Command } from "@opencode-ai/schema/command"
-import { Document } from "@opencode-ai/schema/config"
+import { Document, Event, type Entry } from "@opencode-ai/schema/config"
 import { ConfigMCP } from "@opencode-ai/schema/config/mcp"
 import { createHash } from "node:crypto"
-import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Stream } from "effect"
+import { isDeepStrictEqual } from "node:util"
+import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Semaphore, Stream } from "effect"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { Config } from "../config"
 import { Credential } from "../credential"
@@ -180,26 +181,36 @@ export const layer = (options?: Options) =>
       const fork = yield* FiberSet.makeRuntime<never, void, never>()
       yield* Effect.addFinalizer((exit) => Scope.close(root, exit))
 
-      const documents = (yield* config.entries()).filter((entry): entry is 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 loadConfig = (entries: readonly Entry[]) => {
+        const documents = entries.filter((entry): entry is 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<ServerName, typeof ConfigMCP.Server.Type>()
+        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 = loadConfig(yield* config.entries())
+      const configState = { servers: initial.servers, timeout: initial.timeout }
       // Later config files win for duplicate server names; per-server timeout overrides globals.
       const runtime = new Map<ServerName, ServerEntry>()
       // 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<ServerName>()
+      const reloadLock = Semaphore.makeUnsafe(1)
       const urlElicitations = new Map<string, Form.ID>()
-      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<void>(),
-          })
-        }
+      for (const [name, server] of initial.servers) {
+        runtime.set(name, {
+          config: server,
+          status: { status: "pending" },
+          startup: Deferred.makeUnsafe<void>(),
+        })
       }
 
       // Register every remote server as an OAuth integration so credentials live in the global store
@@ -552,6 +563,68 @@ export const layer = (options?: Options) =>
         yield* bus.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,
+        serverConfig: typeof ConfigMCP.Server.Type,
+      ) {
+        const previous = runtime.get(name)
+        if (previous) yield* disposeServer(name, previous)
+        const entry: ServerEntry = {
+          config: serverConfig,
+          status: { status: "pending" },
+          startup: Deferred.makeUnsafe<void>(),
+        }
+        runtime.set(name, entry)
+        yield* Effect.gen(function* () {
+          yield* register(name, entry)
+          if (serverConfig.disabled) {
+            entry.status = { status: "disabled" }
+            yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
+            return
+          }
+          yield* startServer(name, entry)
+        }).pipe(
+          // Settle startup even when registration fails or replacement is interrupted, so readers cannot hang.
+          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 keyed by name + URL and intentionally survive removal for a later re-add.
+        runtime.delete(name)
+        yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
+      })
+
+      const reloadConfig = Effect.fnUntraced(function* () {
+        yield* reloadLock.withPermit(
+          Effect.gen(function* () {
+            const next = loadConfig(yield* config.entries())
+            const names = new Set([...configState.servers.keys(), ...next.servers.keys()])
+            for (const name of names) {
+              const previous = configState.servers.get(name)
+              const updated = next.servers.get(name)
+              if (isDeepStrictEqual(previous, updated)) continue
+              if (!updated) {
+                yield* removeServer(name).pipe(locks.withLock(name))
+                continue
+              }
+              yield* replaceServer(name, updated).pipe(locks.withLock(name))
+            }
+            configState.servers = next.servers
+            configState.timeout = next.timeout
+          }),
+        )
+      })
+
       // Disabled servers settle their startup immediately so queries never block on them.
       for (const [name, entry] of runtime) {
         if (entry.config.disabled) {
@@ -585,6 +658,16 @@ export const layer = (options?: Options) =>
           Effect.ignore,
         ),
       )
+      yield* bus.subscribe(Event.Updated).pipe(
+        Stream.runForEach(() =>
+          reloadConfig().pipe(
+            Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause })),
+          ),
+        ),
+        Effect.forkScoped({ startImmediately: true }),
+      )
+      // Close the gap between the initial snapshot and the live subscription becoming active.
+      yield* reloadConfig()
 
       // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run.
       const whenAllReady = Effect.suspend(() =>
@@ -601,33 +684,9 @@ export const layer = (options?: Options) =>
         }),
         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<void>(),
-            }
-            runtime.set(name, entry)
-            yield* Effect.gen(function* () {
-              yield* register(name, entry)
-              if (config.disabled) {
-                entry.status = { status: "disabled" }
-                yield* bus.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: { ...configState.timeout, ...config.timeout } }).pipe(
+            locks.withLock(name),
+          )
         }),
         connect: Effect.fn("MCP.connect")(function* (server) {
           const name = ServerName.make(server)
@@ -649,14 +708,8 @@ export const layer = (options?: Options) =>
         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* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)
+            yield* requireServer(name)
+            yield* removeServer(name)
           }).pipe(locks.withLock(name))
         }),
         tools: Effect.fn("MCP.tools")(function* () {

packages/core/test/mcp.test.ts

@@ -12,14 +12,14 @@ import {
   ListToolsRequestSchema,
   ReadResourceRequestSchema,
 } from "@modelcontextprotocol/sdk/types.js"
-import { Document, Info } from "@opencode-ai/schema/config"
+import { Document, Event, Info } from "@opencode-ai/schema/config"
 import { ConfigMCP } from "@opencode-ai/schema/config/mcp"
 import { Config } from "@opencode-ai/core/config"
 import { Credential } from "@opencode-ai/core/credential"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { LayerNode } from "@opencode-ai/util/effect/layer-node"
 import { Bus } from "@opencode-ai/core/bus"
-import { Event } from "@opencode-ai/schema/event"
+import { ID, type Payload } from "@opencode-ai/schema/event"
 import { Form } from "@opencode-ai/core/form"
 import { Integration } from "@opencode-ai/core/integration"
 import { Location } from "@opencode-ai/core/location"
@@ -30,7 +30,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
 import { Session } from "@opencode-ai/core/session"
 import { McpTool } from "@opencode-ai/core/tool/mcp"
 import { Tool } from "@opencode-ai/core/tool"
-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"
@@ -67,6 +67,8 @@ function resourceServer(
         ] as Array<{ uri: string; text: string; mimeType?: string } | { uri: string; blob: string; mimeType?: string }>,
         resourceLists: 0,
         templateLists: 0,
+        toolLists: 0,
+        initializations: 0,
       }
       const protocol = new Server(
         { name: "mcp-resources", version: "1.0.0" },
@@ -77,15 +79,16 @@ function resourceServer(
           },
         },
       )
-      protocol.setRequestHandler(ListToolsRequestSchema, () =>
-        Promise.resolve({
+      protocol.setRequestHandler(ListToolsRequestSchema, () => {
+        state.toolLists += 1
+        return Promise.resolve({
           tools: input.emptyElicitation
             ? [{ name: "empty-elicitation", inputSchema: { type: "object" as const, properties: {} } }]
             : input.urlElicitation
               ? [{ name: "url-elicitation", inputSchema: { type: "object" as const, properties: {} } }]
               : [],
-        }),
-      )
+        })
+      })
       if (input.emptyElicitation) {
         protocol.setRequestHandler(CallToolRequestSchema, async () => {
           const result = await protocol.elicitInput({
@@ -133,7 +136,13 @@ function resourceServer(
       await protocol.connect(transport)
       const http = Bun.serve({
         port: 0,
-        fetch: (request) => transport.handleRequest(request),
+        fetch: async (request) => {
+          const body: unknown = request.method === "POST" ? await request.clone().json() : undefined
+          if (typeof body === "object" && body !== null && "method" in body && body.method === "initialize") {
+            state.initializations += 1
+          }
+          return transport.handleRequest(request)
+        },
       })
       return {
         state,
@@ -155,6 +164,10 @@ function resourceMcpLayer(
   server: string | typeof ConfigMCP.Server.Type,
   onFormCreated?: (form: Form.Info) => Effect.Effect<void>,
   options?: MCP.Options,
+  overrides?: {
+    entries?: Config.Interface["entries"]
+    subscribe?: Bus.Interface["subscribe"]
+  },
 ) {
   const directory = AbsolutePath.make(import.meta.dir)
   const unusedIntegration = () => Effect.die("unused integration service")
@@ -162,30 +175,35 @@ function resourceMcpLayer(
     Layer.provideMerge(Form.layer),
     Layer.provide(
       Layer.mergeAll(
-        Config.testLayer([
-          new Document({
-            type: "document",
-            info: new Info({
-              mcp: new ConfigMCP.Info({
-                servers: {
-                  resources:
-                    typeof server === "string"
-                      ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false })
-                      : server,
-                },
+        overrides?.entries
+          ? Layer.succeed(
+              Config.Service,
+              Config.Service.of({ entries: overrides.entries, changes: () => Stream.never }),
+            )
+          : Config.testLayer([
+              new Document({
+                type: "document",
+                info: new 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(Bus.Service, {
-          subscribe: () => Stream.never,
+          subscribe: overrides?.subscribe ?? (() => Stream.never),
           publish: (definition, data) => {
             const event = {
-              id: Event.ID.create(),
+              id: ID.create(),
               type: definition.type,
               data,
-            } as Event.Payload<typeof definition>
+            } as Payload<typeof definition>
             if (event.type !== Form.Event.Created.type || !onFormCreated) return Effect.succeed(event)
             return onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form).pipe(Effect.as(event))
           },
@@ -785,6 +803,123 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => {
   )
 })
 
+test("reconciles only changed MCP server config", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const server = yield* resourceServer()
+        const updates = yield* PubSub.unbounded<Payload>()
+        const resources = (codemode?: boolean) =>
+          new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, codemode })
+        const added = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })
+        const dynamic = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })
+        const document = (servers: Record<string, typeof ConfigMCP.Server.Type>, username?: string) =>
+          new Document({
+            type: "document",
+            info: new Info({
+              username,
+              mcp: new ConfigMCP.Info({ servers }),
+            }),
+          })
+        let entries = [document({ resources: resources() })]
+        const publishUpdate = () =>
+          PubSub.publish(updates, {
+            id: ID.create(),
+            created: DateTime.makeUnsafe(0),
+            type: Event.Updated.type,
+            data: {},
+          } satisfies Payload<typeof Event.Updated>)
+
+        yield* Effect.gen(function* () {
+          const service = yield* MCP.Service
+          yield* service.tools()
+          expect(server.state.toolLists).toBe(1)
+          expect(server.state.initializations).toBe(1)
+
+          yield* service.add("dynamic", dynamic)
+          entries = [document({ resources: resources() }, "unrelated")]
+          yield* publishUpdate()
+          entries = [document({ resources: resources(), added }, "unrelated")]
+          yield* publishUpdate()
+          const appended = yield* service.servers().pipe(
+            Effect.filterOrFail(
+              (items) => items.some((item) => item.name === "added"),
+              () => new Error("MCP config addition was not applied"),
+            ),
+            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
+          )
+          expect(appended.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic", "resources"])
+          expect(server.state.toolLists).toBe(1)
+          expect(server.state.initializations).toBe(1)
+
+          entries = [
+            document(
+              {
+                resources: resources(false),
+                added,
+              },
+              "unrelated",
+            ),
+          ]
+          yield* publishUpdate()
+          yield* Effect.sync(() => server.state.initializations).pipe(
+            Effect.filterOrFail(
+              (count) => count === 2,
+              () => new Error("MCP config change did not reconnect the server"),
+            ),
+            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
+          )
+
+          entries = [document({ added }, "unrelated")]
+          yield* publishUpdate()
+          const removed = yield* service.servers().pipe(
+            Effect.filterOrFail(
+              (items) => !items.some((item) => item.name === "resources"),
+              () => new Error("MCP config removal was not applied"),
+            ),
+            Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }),
+          )
+          expect(removed.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic"])
+        }).pipe(
+          Effect.provide(
+            resourceMcpLayer(resources(), undefined, undefined, {
+              entries: () => Effect.sync(() => entries),
+              subscribe: (() => Stream.fromPubSub(updates)) as Bus.Interface["subscribe"],
+            }),
+          ),
+        )
+      }),
+    ),
+  )
+})
+
+test("reconciles MCP config changed during startup", async () => {
+  const server = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })
+  let reads = 0
+  const entries = () =>
+    Effect.sync(() => {
+      reads += 1
+      return [
+        new Document({
+          type: "document",
+          info: new Info({
+            mcp: new ConfigMCP.Info({ servers: reads === 1 ? { initial: server } : { initial: server, added: server } }),
+          }),
+        }),
+      ]
+    })
+
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const service = yield* MCP.Service
+        expect((yield* service.servers()).map((item) => String(item.name))).toEqual(["added", "initial"])
+        expect(reads).toBeGreaterThanOrEqual(2)
+      }).pipe(Effect.provide(resourceMcpLayer(server, undefined, undefined, { entries }))),
+    ),
+  )
+})
+
 test("serializes concurrent MCP lifecycle operations", async () => {
   await Effect.runPromise(
     Effect.scoped(