refactor(core): clarify MCP config reload

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<never, void, never>()
     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 loadConfig = 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<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 = yield* loadConfig()
+    const configState = { 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<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 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
@@ -560,6 +573,65 @@ 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<void>(),
+      }
+      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)
+    })
+
+    const reloadConfig = Effect.fnUntraced(function* () {
+      const next = yield* loadConfig()
+      configState.timeout = next.timeout
+      const names = new Set([...configState.names, ...next.servers.keys()])
+      for (const name of names) {
+        const updated = next.servers.get(name)
+        if (!updated) {
+          yield* removeServer(name).pipe(locks.withLock(name))
+          continue
+        }
+        if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue
+        yield* replaceServer(name, updated).pipe(locks.withLock(name))
+      }
+      configState.names = new Set(next.servers.keys())
+    })
+
     // Disabled servers settle their startup immediately so queries never block on them.
     for (const [name, entry] of runtime) {
       if (entry.config.disabled) {
@@ -594,6 +666,15 @@ export const layer = (options?: Options) => Layer.effect(
       ),
     )
 
+    fork(
+      events.subscribe(Event.Updated).pipe(
+        Stream.runForEach(() =>
+          reloadConfig().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 +696,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<void>(),
-          }
-          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: { ...configState.timeout, ...config.timeout } }).pipe(
+          locks.withLock(name),
+        )
       }),
       connect: Effect.fn("MCP.connect")(function* (server) {
         const name = ServerName.make(server)
@@ -663,14 +720,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* () {

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"
@@ -149,49 +150,54 @@ function resourceServer(
   )
 }
 
-function resourceMcpLayer(
-  server: string | typeof ConfigMCP.Server.Type,
-  onFormCreated?: (form: Form.Info) => Effect.Effect<void>,
-  options?: MCP.Options,
-) {
+function resourceMcpLayer(input: {
+  server: string | typeof ConfigMCP.Server.Type
+  onFormCreated?: (form: Form.Info) => Effect.Effect<void>
+  options?: MCP.Options
+  entries?: Config.Interface["entries"]
+  subscribe?: EventV2.Interface["subscribe"]
+}) {
   const directory = AbsolutePath.make(import.meta.dir)
   const unusedIntegration = () => Effect.die("unused integration service")
-  return MCP.layer(options).pipe(
+  const entries =
+    input.entries ??
+    (() =>
+      Effect.succeed([
+        new Config.Document({
+          type: "document",
+          info: new Config.Info({
+            mcp: new ConfigMCP.Info({
+              servers: {
+                resources:
+                  typeof input.server === "string"
+                    ? new ConfigMCP.Remote({ type: "remote", url: input.server, oauth: false })
+                    : input.server,
+              },
+            }),
+          }),
+        }),
+      ]))
+  return MCP.layer(input.options).pipe(
     Layer.provideMerge(Form.layer),
     Layer.provide(
       Layer.mergeAll(
         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,
-                      },
-                    }),
-                  }),
-                }),
-              ]),
-          }),
+          Config.Service.of({ entries }),
         ),
         Layer.succeed(Location.Service, Location.Service.of(location({ directory }))),
         Layer.mock(EventV2.Service, {
-          subscribe: () => Stream.never,
+          subscribe: input.subscribe ?? (() => Stream.never),
           publish: (definition, data) => {
             const event = {
               id: EventV2.ID.create(),
               type: definition.type,
               data,
             } as EventV2.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))
+            if (event.type !== Form.Event.Created.type || !input.onFormCreated) return Effect.succeed(event)
+            return input
+              .onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form)
+              .pipe(Effect.as(event))
           },
         }),
         Layer.mock(Integration.Service, {
@@ -568,7 +574,7 @@ test("accepts empty MCP elicitations without creating forms", async () => {
           const result = yield* service.callTool({ server: "resources", name: "empty-elicitation" })
           expect(yield* forms.list()).toEqual([])
           return result
-        }).pipe(Effect.provide(resourceMcpLayer(server.url)))
+        }).pipe(Effect.provide(resourceMcpLayer({ server: server.url })))
 
         expect(result.structured).toEqual({ action: "accept", content: {} })
       }),
@@ -595,7 +601,12 @@ test("acknowledges completed MCP URL elicitations without returning internal con
           expect(yield* forms.state(form.id)).toEqual({ status: "answered", answer: { elicitation: true } })
           return result
         }).pipe(
-          Effect.provide(resourceMcpLayer(server.url, (form) => Deferred.succeed(created, form).pipe(Effect.asVoid))),
+          Effect.provide(
+            resourceMcpLayer({
+              server: server.url,
+              onFormCreated: (form) => Deferred.succeed(created, form).pipe(Effect.asVoid),
+            }),
+          ),
         )
 
         expect(result.structured).toEqual({ action: "accept" })
@@ -648,7 +659,10 @@ test("loads and reads MCP resources", async () => {
           expect(server.clientVersion()).toMatchObject({ name: "sdk", version: "1.2.3" })
         }).pipe(
           Effect.provide(
-            resourceMcpLayer(server.url, undefined, { clientInfo: { name: "sdk", version: "1.2.3" } }),
+            resourceMcpLayer({
+              server: server.url,
+              options: { clientInfo: { name: "sdk", version: "1.2.3" } },
+            }),
           ),
         )
       }),
@@ -711,13 +725,82 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => {
           expect(yield* service.remove("dynamic").pipe(Effect.flip)).toBeInstanceOf(MCP.NotFoundError)
         }).pipe(
           Effect.provide(
-            resourceMcpLayer(
-              new ConfigMCP.Local({
+            resourceMcpLayer({
+              server: new ConfigMCP.Local({
                 type: "local",
                 command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
                 disabled: true,
               }),
+            }),
+          ),
+        )
+      }),
+    ),
+  )
+})
+
+test("reconciles MCP servers when config updates", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const updates = yield* PubSub.unbounded<EventV2.Payload>()
+        const command = [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")]
+        let servers: Record<string, typeof ConfigMCP.Server.Type> = {
+          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<typeof Event.Updated>)
+
+        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({
+              server: new ConfigMCP.Local({ type: "local", command, disabled: true }),
+              entries,
+              subscribe,
+            }),
           ),
         )
       }),
@@ -756,13 +839,13 @@ test("serializes concurrent MCP lifecycle operations", async () => {
           expect((yield* service.tools()).length).toBeGreaterThan(0)
         }).pipe(
           Effect.provide(
-            resourceMcpLayer(
-              new ConfigMCP.Local({
+            resourceMcpLayer({
+              server: new ConfigMCP.Local({
                 type: "local",
                 command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")],
                 disabled: true,
               }),
-            ),
+            }),
           ),
         )
       }),