From 6cecbed34ae5a53f1d9a52eedf54ca2ec0fd5a65 Mon Sep 17 00:00:00 2001 From: Omid Saffari Date: Sat, 1 Aug 2026 12:04:50 +0400 Subject: [PATCH] fix(core): fence runtime MCP tool reconciliation --- packages/core/src/tool/mcp.ts | 14 +- .../core/test/mcp-tool-reconciliation.test.ts | 123 ++++++++++++++++++ packages/server/src/handlers/mcp.ts | 9 ++ 3 files changed, 143 insertions(+), 3 deletions(-) create mode 100644 packages/core/test/mcp-tool-reconciliation.test.ts diff --git a/packages/core/src/tool/mcp.ts b/packages/core/src/tool/mcp.ts index d82e45ae33e0..abbc78af8164 100644 --- a/packages/core/src/tool/mcp.ts +++ b/packages/core/src/tool/mcp.ts @@ -2,7 +2,7 @@ export * as McpTool from "./mcp" import { ToolFailure } from "@opencode-ai/ai" import { McpEvent } from "@opencode-ai/schema/mcp-event" -import { Effect, Exit, type JsonSchema, Layer, Scope, Semaphore, Stream } from "effect" +import { Context, Effect, Exit, type JsonSchema, Layer, Scope, Semaphore, Stream } from "effect" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Bus } from "../bus" @@ -16,7 +16,14 @@ import { Tool } from "../tool" export const namespace = (server: string) => server.replace(/[^a-zA-Z0-9_-]/g, "_") export const name = (server: string, tool: string) => `${namespace(server)}_${tool.replace(/[^a-zA-Z0-9_-]/g, "_")}` -export const layer = Layer.effectDiscard( +export interface Interface { + readonly reconcile: Effect.Effect +} + +export class Service extends Context.Service()("@opencode/McpTool") {} + +export const layer = Layer.effect( + Service, Effect.gen(function* () { const mcp = yield* MCP.Service const tools = yield* Tool.Service @@ -118,11 +125,12 @@ export const layer = Layer.effectDiscard( Stream.runForEach(() => reconcile), Effect.forkScoped({ startImmediately: true }), ) + return Service.of({ reconcile }) }), ) export const node = makeLocationNode({ - name: "mcp-tools", + service: Service, layer, deps: [Tool.node, MCP.node, Bus.node, Permission.node], }) diff --git a/packages/core/test/mcp-tool-reconciliation.test.ts b/packages/core/test/mcp-tool-reconciliation.test.ts new file mode 100644 index 000000000000..972b13b1545b --- /dev/null +++ b/packages/core/test/mcp-tool-reconciliation.test.ts @@ -0,0 +1,123 @@ +import { expect, test } from "bun:test" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { Bus } from "@opencode-ai/core/bus" +import { Image } from "@opencode-ai/core/image" +import { MCP } from "@opencode-ai/core/mcp/index" +import { Permission } from "@opencode-ai/core/permission" +import { McpTool } from "@opencode-ai/core/tool/mcp" +import { Tool } from "@opencode-ai/core/tool" +import { Event } from "@opencode-ai/schema/event" +import { McpEvent } from "@opencode-ai/schema/mcp-event" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { DateTime, Deferred, Effect, Fiber, Layer, PubSub, Ref, Stream } from "effect" +import { imagePassthrough } from "./lib/image" + +test("fences asynchronous MCP tool additions and removals", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const initialRead = yield* Deferred.make() + const addStarted = yield* Deferred.make() + const releaseAdd = yield* Deferred.make() + const removeStarted = yield* Deferred.make() + const releaseRemove = yield* Deferred.make() + const gates = [ + { started: addStarted, release: releaseAdd }, + { started: removeStarted, release: releaseRemove }, + ] + const updates = yield* PubSub.unbounded() + const reads = yield* Ref.make(0) + const catalog = yield* Ref.make>([]) + + function subscribe(): Stream.Stream + function subscribe(definition: D): Stream.Stream> + function subscribe( + definitions: D, + ): Stream.Stream> + function subscribe(): Stream.Stream { + return Stream.fromPubSub(updates) + } + + const layer = AppNodeBuilder.build(LayerNode.group([Tool.node, McpTool.node]), [ + [ + MCP.node, + Layer.mock(MCP.Service, { + tools: () => + Effect.gen(function* () { + const read = yield* Ref.updateAndGet(reads, (count) => count + 1) + if (read === 1) { + const current = yield* Ref.get(catalog) + yield* Deferred.succeed(initialRead, undefined) + return current + } + if (read % 2 === 0) { + const gate = gates[read / 2 - 1] + if (gate) { + yield* Deferred.succeed(gate.started, undefined) + yield* Deferred.await(gate.release) + } + } + return yield* Ref.get(catalog) + }), + }), + ], + [Bus.node, Layer.mock(Bus.Service, { subscribe })], + [Permission.node, Layer.mock(Permission.Service, {})], + [Image.node, imagePassthrough], + ]) + + yield* Effect.gen(function* () { + const registry = yield* Tool.Service + const adapter = yield* McpTool.Service + yield* Deferred.await(initialRead) + yield* Ref.set(catalog, [ + new MCP.Tool({ + server: MCP.ServerName.make("voice"), + name: "list_open_tabs", + inputSchema: { type: "object", properties: {} }, + }), + ]) + + yield* PubSub.publish(updates, { + id: Event.ID.create(), + created: yield* DateTime.now, + type: McpEvent.ToolsChanged.type, + data: { server: "voice" }, + }) + yield* Deferred.await(addStarted) + + const staleAddition = yield* registry.snapshot() + expect(staleAddition.codeModeCatalog?.some((entry) => entry.path === "voice.list_open_tabs")).toBe(false) + + const addFence = yield* Effect.forkChild(adapter.reconcile, { startImmediately: true }) + expect(addFence.pollUnsafe()).toBeUndefined() + yield* Deferred.succeed(releaseAdd, undefined) + yield* Fiber.join(addFence) + + const added = yield* registry.snapshot() + expect(added.codeModeCatalog?.some((entry) => entry.path === "voice.list_open_tabs")).toBe(true) + + yield* Ref.set(catalog, []) + yield* PubSub.publish(updates, { + id: Event.ID.create(), + created: yield* DateTime.now, + type: McpEvent.ToolsChanged.type, + data: { server: "voice" }, + }) + yield* Deferred.await(removeStarted) + + const staleRemoval = yield* registry.snapshot() + expect(staleRemoval.codeModeCatalog?.some((entry) => entry.path === "voice.list_open_tabs")).toBe(true) + + const removeFence = yield* Effect.forkChild(adapter.reconcile, { startImmediately: true }) + expect(removeFence.pollUnsafe()).toBeUndefined() + yield* Deferred.succeed(releaseRemove, undefined) + yield* Fiber.join(removeFence) + + const removed = yield* registry.snapshot() + expect(removed.codeModeCatalog?.some((entry) => entry.path === "voice.list_open_tabs")).toBe(false) + }).pipe(Effect.provide(layer)) + }), + ), + ) +}) diff --git a/packages/server/src/handlers/mcp.ts b/packages/server/src/handlers/mcp.ts index 0bb206997c22..b38f150c5783 100644 --- a/packages/server/src/handlers/mcp.ts +++ b/packages/server/src/handlers/mcp.ts @@ -1,4 +1,5 @@ import { MCP } from "@opencode-ai/core/mcp/index" +import { McpTool } from "@opencode-ai/core/tool/mcp" import { McpServerNotFoundError } from "@opencode-ai/protocol/errors" import { Effect } from "effect" import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" @@ -30,7 +31,9 @@ export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) => "mcp.add", Effect.fn(function* (ctx) { const service = yield* MCP.Service + const tools = yield* McpTool.Service yield* service.add(ctx.params.server, ctx.payload.config) + yield* tools.reconcile return HttpApiSchema.NoContent.make() }), ) @@ -38,7 +41,9 @@ export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) => "mcp.remove", Effect.fn(function* (ctx) { const service = yield* MCP.Service + const tools = yield* McpTool.Service yield* notFound(service.remove(ctx.params.server)) + yield* tools.reconcile return HttpApiSchema.NoContent.make() }), ) @@ -46,7 +51,9 @@ export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) => "mcp.connect", Effect.fn(function* (ctx) { const service = yield* MCP.Service + const tools = yield* McpTool.Service yield* notFound(service.connect(ctx.params.server)) + yield* tools.reconcile return HttpApiSchema.NoContent.make() }), ) @@ -54,7 +61,9 @@ export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) => "mcp.disconnect", Effect.fn(function* (ctx) { const service = yield* MCP.Service + const tools = yield* McpTool.Service yield* notFound(service.disconnect(ctx.params.server)) + yield* tools.reconcile return HttpApiSchema.NoContent.make() }), )