From 0d2b1660a83ef9ed120e89c1155736e3b9492895 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Thu, 16 Jul 2026 14:36:12 +0000 Subject: [PATCH 1/4] feat(core): add runtime MCP controls --- packages/core/src/mcp/index.ts | 33 ++++++++++++++++++++++- packages/core/test/fixture/mcp.ts | 2 ++ packages/core/test/mcp.test.ts | 45 +++++++++++++++++++++++++++++-- 3 files changed, 77 insertions(+), 3 deletions(-) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index c9d9a83f82c0..12909c03a2a2 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -127,6 +127,8 @@ const URL_ELICITATION_FIELD_KEY = "elicitation" export interface Interface { readonly servers: () => Effect.Effect + readonly connect: (server: ServerName | string) => Effect.Effect + readonly disconnect: (server: ServerName | string) => Effect.Effect readonly tools: () => Effect.Effect readonly callTool: (input: { readonly server: ServerName | string @@ -527,7 +529,7 @@ export const layer = Layer.effect( const match = Array.from(runtime).find(([, entry]) => entry.integrationID === integrationID) if (!match) return const [name, entry] = match - if (entry.config.disabled) return + if (entry.status.status === "disabled") return if (entry.scope) { yield* Scope.close(entry.scope, Exit.void) entry.scope = undefined @@ -562,6 +564,35 @@ export const layer = Layer.effect( }), ) }), + connect: Effect.fn("MCP.connect")(function* (server) { + const target = yield* requireServer(server) + if (target.entry.scope) { + const scope = target.entry.scope + target.entry.scope = undefined + target.entry.client = undefined + target.entry.tools = undefined + target.entry.prompts = undefined + yield* Scope.close(scope, Exit.void) + yield* events.publish(McpEvent.ToolsChanged, { server: target.name }).pipe(Effect.ignore) + yield* events.publish(McpEvent.ResourcesChanged, { server: target.name }).pipe(Effect.ignore) + yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) + } + yield* startServer(target.name, target.entry) + }), + disconnect: Effect.fn("MCP.disconnect")(function* (server) { + const target = yield* requireServer(server) + const scope = target.entry.scope + target.entry.scope = undefined + target.entry.client = undefined + target.entry.tools = undefined + target.entry.prompts = undefined + if (scope) yield* Scope.close(scope, Exit.void) + target.entry.status = { status: "disabled" } + yield* events.publish(McpEvent.ToolsChanged, { server: target.name }).pipe(Effect.ignore) + yield* events.publish(McpEvent.ResourcesChanged, { server: target.name }).pipe(Effect.ignore) + yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) + yield* events.publish(McpEvent.StatusChanged, { server: target.name }).pipe(Effect.ignore) + }), tools: Effect.fn("MCP.tools")(function* () { yield* whenAllReady return Array.from(runtime.values()) diff --git a/packages/core/test/fixture/mcp.ts b/packages/core/test/fixture/mcp.ts index b247a4d8b781..5503f81d598a 100644 --- a/packages/core/test/fixture/mcp.ts +++ b/packages/core/test/fixture/mcp.ts @@ -9,6 +9,8 @@ export const emptyMcpLayer = Layer.succeed( MCP.Service, MCP.Service.of({ servers: () => Effect.succeed([]), + connect: () => Effect.die("unused mcp.connect"), + disconnect: () => Effect.die("unused mcp.disconnect"), tools: () => Effect.succeed([]), callTool: () => Effect.die("unused mcp.callTool"), instructions: () => Effect.succeed([]), diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 22390b631645..76f543b80dce 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -146,7 +146,10 @@ function resourceServer( ) } -function resourceMcpLayer(url: string, onFormCreated?: (form: Form.Info) => Effect.Effect) { +function resourceMcpLayer( + server: string | typeof ConfigMCP.Server.Type, + onFormCreated?: (form: Form.Info) => Effect.Effect, +) { const directory = AbsolutePath.make(import.meta.dir) const unusedIntegration = () => Effect.die("unused integration service") return MCP.layer.pipe( @@ -162,7 +165,12 @@ function resourceMcpLayer(url: string, onFormCreated?: (form: Form.Info) => Effe type: "document", info: new Config.Info({ mcp: new ConfigMCP.Info({ - servers: { resources: new ConfigMCP.Remote({ type: "remote", url, oauth: false }) }, + servers: { + resources: + typeof server === "string" + ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) + : server, + }, }), }), }), @@ -630,6 +638,39 @@ test("loads and reads MCP resources", async () => { ) }) +test("disconnects and reconnects MCP servers at runtime", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + yield* Effect.gen(function* () { + const service = yield* MCP.Service + + expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) + yield* service.connect("resources") + expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + + yield* service.disconnect("resources") + expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) + expect(yield* service.tools()).toEqual([]) + + yield* service.connect("resources") + expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + }).pipe( + Effect.provide( + resourceMcpLayer( + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + disabled: true, + }), + ), + ), + ) + }), + ), + ) +}) + it.effect("advertises MCP output schemas to Code Mode", () => Effect.gen(function* () { const registry = yield* ToolRegistry.Service From 641caecf56bc91635155e09167b659f7b2dea174 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Thu, 16 Jul 2026 15:08:35 +0000 Subject: [PATCH 2/4] feat(core): add runtime MCP servers --- packages/core/src/mcp/index.ts | 115 ++++++++++++++++-------------- packages/core/test/fixture/mcp.ts | 1 + packages/core/test/mcp.test.ts | 43 +++++++++-- 3 files changed, 97 insertions(+), 62 deletions(-) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 12909c03a2a2..62601611fa02 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -127,6 +127,7 @@ const URL_ELICITATION_FIELD_KEY = "elicitation" export interface Interface { readonly servers: () => Effect.Effect + readonly add: (server: ServerName | string, config: typeof ConfigMCP.Server.Type) => Effect.Effect readonly connect: (server: ServerName | string) => Effect.Effect readonly disconnect: (server: ServerName | string) => Effect.Effect readonly tools: () => Effect.Effect @@ -185,14 +186,9 @@ export const layer = Layer.effect( // Register every remote server as an OAuth integration so credentials live in the global store // rather than in committed config. Servers that connect anonymously simply never use the method. - const registrations: Array<{ - readonly name: ServerName - readonly remote: typeof ConfigMCP.Remote.Type - readonly integrationID: Integration.ID - readonly methodID: Integration.MethodID - }> = [] - for (const [name, entry] of runtime) { - if (entry.config.type !== "remote" || entry.config.oauth === false) continue + const owned = new Set() + const register = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { + if (entry.config.type !== "remote" || entry.config.oauth === false) return const remote = entry.config // Key identity on name + url, not url alone: two configs for the same url under different names are // distinct logical servers that may hold different accounts, so they must not share a credential row. @@ -202,27 +198,24 @@ export const layer = Layer.effect( .update(name + "\u0000" + remote.url) .digest("hex") .slice(0, 16) - entry.integrationID = Integration.ID.make(suffix) - registrations.push({ - name, - remote, - integrationID: entry.integrationID, - methodID: Integration.MethodID.make(suffix), - }) - } - if (registrations.length > 0) - yield* integration.transform((draft) => { - for (const reg of registrations) { - draft.update(reg.integrationID, (ref) => { - ref.name = reg.name + const integrationID = Integration.ID.make(suffix) + entry.integrationID = integrationID + owned.add(integrationID) + const methodID = Integration.MethodID.make(suffix) + yield* integration + .transform((draft) => { + draft.update(integrationID, (ref) => { + ref.name = name }) draft.method.update({ - integrationID: reg.integrationID, - method: { id: reg.methodID, type: "oauth", label: reg.name }, - authorize: () => MCPOAuth.authorize({ name: reg.name, config: reg.remote, methodID: reg.methodID }), + integrationID, + method: { id: methodID, type: "oauth", label: name }, + authorize: () => MCPOAuth.authorize({ name, config: remote, methodID }), }) - } - }) + }) + .pipe(Scope.provide(root)) + }) + yield* Effect.forEach(runtime, ([name, entry]) => register(name, entry), { discard: true }) const requireServer = Effect.fnUntraced(function* (server: ServerName | string) { const name = ServerName.make(server) @@ -511,6 +504,18 @@ export const layer = Layer.effect( yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) }).pipe(Effect.ensuring(Deferred.succeed(entry.startup, undefined))) + const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { + const scope = entry.scope + entry.scope = undefined + entry.client = undefined + entry.tools = undefined + entry.prompts = undefined + if (scope) yield* Scope.close(scope, Exit.void) + yield* events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) + yield* events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) + yield* events.publish(Command.Event.Updated, {}).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) { @@ -523,21 +528,13 @@ export const layer = Layer.effect( // Bring a server online (or back to needs_auth) when its integration's credential changes, so an // OAuth login takes effect without a restart. Only fires for the integrations we registered. - const owned = new Set(registrations.map((reg) => reg.integrationID)) const reconnect = (integrationID: Integration.ID) => Effect.gen(function* () { const match = Array.from(runtime).find(([, entry]) => entry.integrationID === integrationID) if (!match) return const [name, entry] = match if (entry.status.status === "disabled") return - if (entry.scope) { - yield* Scope.close(entry.scope, Exit.void) - entry.scope = undefined - entry.client = undefined - entry.tools = undefined - entry.prompts = undefined - yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) - } + if (entry.scope) yield* stopServer(name, entry) yield* startServer(name, entry) }) fork( @@ -564,33 +561,41 @@ export const layer = Layer.effect( }), ) }), + add: Effect.fn("MCP.add")(function* (server, config) { + const name = ServerName.make(server) + const previous = runtime.get(name) + if (previous) { + yield* stopServer(name, previous) + if (previous.integrationID) { + const integrationID = previous.integrationID + owned.delete(integrationID) + yield* integration.transform((draft) => draft.remove(integrationID)).pipe(Scope.provide(root)) + } + } + const entry: ServerEntry = { + config: { ...config, timeout: { ...timeout, ...config.timeout } }, + status: { status: "pending" }, + startup: Deferred.makeUnsafe(), + } + runtime.set(name, entry) + yield* register(name, entry) + if (config.disabled) { + entry.status = { status: "disabled" } + Deferred.doneUnsafe(entry.startup, Exit.void) + yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + return + } + yield* startServer(name, entry) + }), connect: Effect.fn("MCP.connect")(function* (server) { const target = yield* requireServer(server) - if (target.entry.scope) { - const scope = target.entry.scope - target.entry.scope = undefined - target.entry.client = undefined - target.entry.tools = undefined - target.entry.prompts = undefined - yield* Scope.close(scope, Exit.void) - yield* events.publish(McpEvent.ToolsChanged, { server: target.name }).pipe(Effect.ignore) - yield* events.publish(McpEvent.ResourcesChanged, { server: target.name }).pipe(Effect.ignore) - yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) - } + if (target.entry.scope) yield* stopServer(target.name, target.entry) yield* startServer(target.name, target.entry) }), disconnect: Effect.fn("MCP.disconnect")(function* (server) { const target = yield* requireServer(server) - const scope = target.entry.scope - target.entry.scope = undefined - target.entry.client = undefined - target.entry.tools = undefined - target.entry.prompts = undefined - if (scope) yield* Scope.close(scope, Exit.void) + yield* stopServer(target.name, target.entry) target.entry.status = { status: "disabled" } - yield* events.publish(McpEvent.ToolsChanged, { server: target.name }).pipe(Effect.ignore) - yield* events.publish(McpEvent.ResourcesChanged, { server: target.name }).pipe(Effect.ignore) - yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) yield* events.publish(McpEvent.StatusChanged, { server: target.name }).pipe(Effect.ignore) }), tools: Effect.fn("MCP.tools")(function* () { diff --git a/packages/core/test/fixture/mcp.ts b/packages/core/test/fixture/mcp.ts index 5503f81d598a..9dfce59977f0 100644 --- a/packages/core/test/fixture/mcp.ts +++ b/packages/core/test/fixture/mcp.ts @@ -9,6 +9,7 @@ export const emptyMcpLayer = Layer.succeed( MCP.Service, MCP.Service.of({ servers: () => Effect.succeed([]), + add: () => Effect.die("unused mcp.add"), connect: () => Effect.die("unused mcp.connect"), disconnect: () => Effect.die("unused mcp.disconnect"), tools: () => Effect.succeed([]), diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 76f543b80dce..7acc29f9a2cd 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -638,7 +638,7 @@ test("loads and reads MCP resources", async () => { ) }) -test("disconnects and reconnects MCP servers at runtime", async () => { +test("adds, disconnects, and reconnects MCP servers at runtime", async () => { await Effect.runPromise( Effect.scoped( Effect.gen(function* () { @@ -646,15 +646,44 @@ test("disconnects and reconnects MCP servers at runtime", async () => { const service = yield* MCP.Service expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) - yield* service.connect("resources") - expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + yield* service.add( + "dynamic", + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + }), + ) + expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({ + status: "connected", + }) - yield* service.disconnect("resources") - expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) + yield* service.add( + "dynamic", + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + disabled: true, + }), + ) + expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({ + status: "disabled", + }) + expect(yield* service.tools()).toEqual([]) + + yield* service.connect("dynamic") + expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({ + status: "connected", + }) + yield* service.disconnect("dynamic") + expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({ + status: "disabled", + }) expect(yield* service.tools()).toEqual([]) - yield* service.connect("resources") - expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + yield* service.connect("dynamic") + expect((yield* service.servers()).find((server) => server.name === "dynamic")?.status).toEqual({ + status: "connected", + }) }).pipe( Effect.provide( resourceMcpLayer( From 2c5bdefa3851b8911426c0c7c5fb951176fa9bc8 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Fri, 17 Jul 2026 13:37:51 -0500 Subject: [PATCH 3/4] fix(core): serialize MCP lifecycle operations --- packages/core/src/mcp/index.ts | 164 ++++++++++++++++++++------------- packages/core/test/mcp.test.ts | 45 +++++++++ 2 files changed, 144 insertions(+), 65 deletions(-) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 62601611fa02..a252adabf93c 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -13,6 +13,7 @@ import { EventV2 } from "../event" import { Form } from "../form" import { Integration } from "../integration" import { IntegrationConnection } from "../integration/connection" +import { KeyedMutex } from "../effect/keyed-mutex" import { Location } from "../location" import { waitForAbort } from "../process" import { MCPClient } from "./client" @@ -173,6 +174,9 @@ export const layer = Layer.effect( ) // 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 ?? {})) { @@ -415,36 +419,44 @@ export const layer = Layer.effect( ), ) + // Runs a connection callback under the server lock, dropping it if the connection is no longer + // the entry's live client, so late SDK callbacks cannot commit obsolete state. + const whenLive = + (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => + (effect: Effect.Effect) => + fork( + Effect.suspend(() => (entry.client === connection ? effect : Effect.void)).pipe( + locks.withLock(name), + Effect.ignore, + ), + ) + const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => { - connection.onClose(() => { - // A reconnect closes the previous scope, but the SDK may fire this onclose after the new - // connection is already assigned; ignore the stale close so it can't null out the live client. - if (entry.client !== connection) return - entry.client = undefined - entry.tools = undefined - entry.prompts = undefined - entry.status = { status: "failed", error: "Connection closed" } - fork(events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)) - fork(events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)) - fork(events.publish(Command.Event.Updated, {}).pipe(Effect.ignore)) - fork(events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore)) - }) + const live = whenLive(name, entry, connection) + connection.onClose(() => + live( + Effect.gen(function* () { + entry.client = undefined + entry.tools = undefined + entry.prompts = undefined + entry.status = { status: "failed", error: "Connection closed" } + yield* events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) + yield* events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) + yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) + yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + }), + ), + ) connection.onLog((message) => fork(serverLog(name, message).pipe(Effect.ignore))) - connection.onToolsChanged(() => { - fork( + connection.onToolsChanged(() => + live( refreshTools(name, entry, connection).pipe( Effect.andThen(events.publish(McpEvent.ToolsChanged, { server: name })), - Effect.ignore, ), - ) - }) - connection.onPromptsChanged(() => { - fork(refreshPrompts(name, entry, connection).pipe(Effect.ignore)) - }) - connection.onResourcesChanged(() => { - if (entry.client !== connection) return - fork(events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore)) - }) + ), + ) + connection.onPromptsChanged(() => live(refreshPrompts(name, entry, connection))) + connection.onResourcesChanged(() => live(events.publish(McpEvent.ResourcesChanged, { server: name }))) } const serverLog = (server: ServerName, message: MCPClient.LogMessage) => { @@ -490,7 +502,7 @@ export const layer = Layer.effect( yield* events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) yield* events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) - fork(refreshPrompts(name, entry, result.value.connection).pipe(Effect.ignore)) + whenLive(name, entry, result.value.connection)(refreshPrompts(name, entry, result.value.connection)) return } yield* Scope.close(scope, Exit.void) @@ -506,11 +518,12 @@ export const layer = Layer.effect( const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { const scope = entry.scope + if (!scope) return entry.scope = undefined entry.client = undefined entry.tools = undefined entry.prompts = undefined - if (scope) yield* Scope.close(scope, Exit.void) + yield* Scope.close(scope, Exit.void) yield* events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) yield* events.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) yield* events.publish(Command.Event.Updated, {}).pipe(Effect.ignore) @@ -523,7 +536,7 @@ export const layer = Layer.effect( Deferred.doneUnsafe(entry.startup, Exit.void) continue } - fork(startServer(name, entry)) + fork(startServer(name, entry).pipe(locks.withLock(name))) } // Bring a server online (or back to needs_auth) when its integration's credential changes, so an @@ -532,10 +545,15 @@ export const layer = Layer.effect( Effect.gen(function* () { const match = Array.from(runtime).find(([, entry]) => entry.integrationID === integrationID) if (!match) return - const [name, entry] = match - if (entry.status.status === "disabled") return - if (entry.scope) yield* stopServer(name, entry) - yield* startServer(name, entry) + const name = match[0] + yield* Effect.gen(function* () { + // add() may have replaced the entry while we waited for the lock. + const entry = runtime.get(name) + if (!entry || entry.integrationID !== integrationID) return + if (entry.status.status === "disabled") return + yield* stopServer(name, entry) + yield* startServer(name, entry) + }).pipe(locks.withLock(name)) }) fork( events.subscribe(Integration.Event.ConnectionUpdated).pipe( @@ -545,10 +563,13 @@ export const layer = Layer.effect( ), ) - const whenAllReady = Effect.forEach(runtime.values(), (entry) => Deferred.await(entry.startup), { - concurrency: "unbounded", - discard: true, - }) + // 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), { + concurrency: "unbounded", + discard: true, + }), + ) return Service.of({ servers: Effect.fn("MCP.servers")(function* () { const entries = Array.from(runtime).toSorted(([a], [b]) => a.localeCompare(b)) @@ -563,40 +584,53 @@ export const layer = Layer.effect( }), add: Effect.fn("MCP.add")(function* (server, config) { const name = ServerName.make(server) - const previous = runtime.get(name) - if (previous) { - yield* stopServer(name, previous) - if (previous.integrationID) { - const integrationID = previous.integrationID - owned.delete(integrationID) - yield* integration.transform((draft) => draft.remove(integrationID)).pipe(Scope.provide(root)) + yield* Effect.gen(function* () { + const previous = runtime.get(name) + if (previous) { + yield* stopServer(name, previous) + if (previous.integrationID) { + const integrationID = previous.integrationID + owned.delete(integrationID) + yield* integration.transform((draft) => draft.remove(integrationID)).pipe(Scope.provide(root)) + } } - } - const entry: ServerEntry = { - config: { ...config, timeout: { ...timeout, ...config.timeout } }, - status: { status: "pending" }, - startup: Deferred.makeUnsafe(), - } - runtime.set(name, entry) - yield* register(name, entry) - if (config.disabled) { - entry.status = { status: "disabled" } - Deferred.doneUnsafe(entry.startup, Exit.void) - yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) - return - } - yield* startServer(name, entry) + 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)) }), connect: Effect.fn("MCP.connect")(function* (server) { - const target = yield* requireServer(server) - if (target.entry.scope) yield* stopServer(target.name, target.entry) - yield* startServer(target.name, target.entry) + const name = ServerName.make(server) + yield* Effect.gen(function* () { + const target = yield* requireServer(name) + yield* stopServer(name, target.entry) + yield* startServer(name, target.entry) + }).pipe(locks.withLock(name)) }), disconnect: Effect.fn("MCP.disconnect")(function* (server) { - const target = yield* requireServer(server) - yield* stopServer(target.name, target.entry) - target.entry.status = { status: "disabled" } - yield* events.publish(McpEvent.StatusChanged, { server: target.name }).pipe(Effect.ignore) + const name = ServerName.make(server) + yield* Effect.gen(function* () { + const target = yield* requireServer(name) + yield* stopServer(name, target.entry) + target.entry.status = { status: "disabled" } + yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + }).pipe(locks.withLock(name)) }), tools: Effect.fn("MCP.tools")(function* () { yield* whenAllReady diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 7acc29f9a2cd..b10db948ad4b 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -700,6 +700,51 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => { ) }) +test("serializes concurrent MCP lifecycle operations", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + yield* Effect.gen(function* () { + const service = yield* MCP.Service + + // Whatever order the racing operations land in, the resulting state must be consistent. + yield* Effect.all( + [ + service.connect("resources"), + service.connect("resources"), + service.disconnect("resources"), + service.connect("resources"), + ], + { concurrency: "unbounded", discard: true }, + ) + const status = (yield* service.servers()).find((server) => server.name === "resources")?.status + const tools = yield* service.tools() + expect(status?.status === "connected" || status?.status === "disabled").toBe(true) + if (status?.status === "disabled") expect(tools).toEqual([]) + if (status?.status === "connected") expect(tools.length).toBeGreaterThan(0) + + yield* service.disconnect("resources") + expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) + expect(yield* service.tools()).toEqual([]) + yield* service.connect("resources") + expect((yield* service.servers())[0]?.status).toEqual({ status: "connected" }) + expect((yield* service.tools()).length).toBeGreaterThan(0) + }).pipe( + Effect.provide( + resourceMcpLayer( + new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + disabled: true, + }), + ), + ), + ) + }), + ), + ) +}) + it.effect("advertises MCP output schemas to Code Mode", () => Effect.gen(function* () { const registry = yield* ToolRegistry.Service From da830c934eedd1907cc43e95d34f1f855a4493a2 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Fri, 17 Jul 2026 14:14:04 -0500 Subject: [PATCH 4/4] fix(core): dispose replaced MCP integration registrations --- packages/core/src/mcp/index.ts | 11 +++++------ packages/core/test/mcp.test.ts | 2 ++ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index a252adabf93c..d35061d7b17d 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -16,6 +16,7 @@ import { IntegrationConnection } from "../integration/connection" import { KeyedMutex } from "../effect/keyed-mutex" import { Location } from "../location" import { waitForAbort } from "../process" +import { State } from "../state" import { MCPClient } from "./client" import { MCPOAuth } from "./oauth" @@ -119,6 +120,7 @@ type ServerEntry = { prompts?: ReadonlyArray // Set when a remote server is registered as an OAuth integration; the credential lives in the global store. integrationID?: Integration.ID + registration?: State.Registration } // MCP elicitations are Location-scoped, not Session-scoped: the server cannot attribute them to a @@ -206,7 +208,7 @@ export const layer = Layer.effect( entry.integrationID = integrationID owned.add(integrationID) const methodID = Integration.MethodID.make(suffix) - yield* integration + entry.registration = yield* integration .transform((draft) => { draft.update(integrationID, (ref) => { ref.name = name @@ -588,11 +590,8 @@ export const layer = Layer.effect( const previous = runtime.get(name) if (previous) { yield* stopServer(name, previous) - if (previous.integrationID) { - const integrationID = previous.integrationID - owned.delete(integrationID) - yield* integration.transform((draft) => draft.remove(integrationID)).pipe(Scope.provide(root)) - } + if (previous.integrationID) owned.delete(previous.integrationID) + if (previous.registration) yield* previous.registration.dispose } const entry: ServerEntry = { config: { ...config, timeout: { ...timeout, ...config.timeout } }, diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index b10db948ad4b..9bc36bd211c5 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -646,6 +646,8 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => { const service = yield* MCP.Service expect((yield* service.servers())[0]?.status).toEqual({ status: "disabled" }) + expect(yield* service.connect("missing").pipe(Effect.flip)).toBeInstanceOf(MCP.NotFoundError) + expect(yield* service.disconnect("missing").pipe(Effect.flip)).toBeInstanceOf(MCP.NotFoundError) yield* service.add( "dynamic", new ConfigMCP.Local({