diff --git a/apps/server/src/mcp/AcpMcpOverAcpBridge.test.ts b/apps/server/src/mcp/AcpMcpOverAcpBridge.test.ts index e7e9ee78753c..33c8cc507f2f 100644 --- a/apps/server/src/mcp/AcpMcpOverAcpBridge.test.ts +++ b/apps/server/src/mcp/AcpMcpOverAcpBridge.test.ts @@ -1,6 +1,7 @@ // The bridge intentionally treats MCP JSON-RPC messages as opaque JSON. // @effect-diagnostics preferSchemaOverJson:off import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -117,4 +118,105 @@ describe("AcpMcpOverAcpBridge", () => { expect(failure.message).toContain("exceeds 8 MiB"); }), ); + + it.effect("caps concurrent connects at 16 and reuses a freed slot", () => + Effect.gen(function* () { + const firstEntered = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + let nextId = 0; + const bridge = yield* makeAcpMcpOverAcpBridge({ + endpoint: "http://127.0.0.1:1/mcp", + authorization: "Bearer bridge-test", + allocateConnectionId: Effect.suspend(() => { + const connectionId = `connection-${++nextId}`; + if (nextId !== 1) return Effect.succeed(connectionId); + return Deferred.succeed(firstEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirst)), + Effect.as(connectionId), + ); + }), + fetchImplementation: () => Promise.resolve(new Response(null, { status: 202 })), + }); + + expect(yield* bridge.connect({ serverId: "other" }).pipe(Effect.flip)).toMatchObject({ + _tag: "AcpMcpOverAcpError", + }); + + const first = yield* bridge + .connect({ serverId: "t3-code" }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(firstEntered); + const rest = yield* Effect.all( + Array.from({ length: 16 }, () => + bridge + .connect({ serverId: "t3-code" }) + .pipe(Effect.forkChild({ startImmediately: true })), + ), + { concurrency: "unbounded" }, + ); + yield* Deferred.succeed(releaseFirst, undefined); + + const outcomes = yield* Effect.forEach([first, ...rest], (fiber) => + Fiber.join(fiber).pipe(Effect.result), + ); + const successes = outcomes.filter((outcome) => outcome._tag === "Success"); + const failures = outcomes.filter((outcome) => outcome._tag === "Failure"); + expect(successes).toHaveLength(16); + expect(failures).toHaveLength(1); + expect(failures[0]).toMatchObject({ + failure: { + _tag: "AcpMcpOverAcpError", + message: "Too many MCP-over-ACP connections.", + }, + }); + + const connectionIds = successes.flatMap((outcome) => + outcome._tag === "Success" ? [outcome.success.connectionId] : [], + ); + expect(new Set(connectionIds).size).toBe(16); + yield* bridge.disconnect({ connectionId: connectionIds[0]! }); + const replacement = yield* bridge.connect({ serverId: "t3-code" }); + expect(connectionIds).not.toContain(replacement.connectionId); + expect(yield* bridge.connect({ serverId: "t3-code" }).pipe(Effect.flip)).toMatchObject({ + _tag: "AcpMcpOverAcpError", + message: "Too many MCP-over-ACP connections.", + }); + }).pipe(Effect.scoped), + ); + + it.effect("does not consume a slot when connection allocation is interrupted", () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const proceed = yield* Deferred.make(); + let nextId = 0; + const bridge = yield* makeAcpMcpOverAcpBridge({ + endpoint: "http://127.0.0.1:1/mcp", + authorization: "Bearer bridge-test", + allocateConnectionId: Effect.suspend(() => { + const connectionId = `connection-${++nextId}`; + if (nextId !== 1) return Effect.succeed(connectionId); + return Deferred.succeed(entered, undefined).pipe( + Effect.andThen(Deferred.await(proceed)), + Effect.as(connectionId), + ); + }), + fetchImplementation: () => Promise.resolve(new Response(null, { status: 202 })), + }); + + const hung = yield* bridge + .connect({ serverId: "t3-code" }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(entered); + yield* Fiber.interrupt(hung); + + const connected = yield* Effect.all( + Array.from({ length: 16 }, () => bridge.connect({ serverId: "t3-code" })), + ); + expect(connected).toHaveLength(16); + expect(yield* bridge.connect({ serverId: "t3-code" }).pipe(Effect.flip)).toMatchObject({ + _tag: "AcpMcpOverAcpError", + message: "Too many MCP-over-ACP connections.", + }); + }).pipe(Effect.scoped), + ); }); diff --git a/apps/server/src/mcp/AcpMcpOverAcpBridge.ts b/apps/server/src/mcp/AcpMcpOverAcpBridge.ts index a5185e2515ec..517c4e645d40 100644 --- a/apps/server/src/mcp/AcpMcpOverAcpBridge.ts +++ b/apps/server/src/mcp/AcpMcpOverAcpBridge.ts @@ -186,12 +186,14 @@ export const makeAcpMcpOverAcpBridge = Effect.fn("AcpMcpOverAcpBridge.make")(fun new AcpMcpOverAcpError(`Unknown ACP MCP server "${request.serverId}".`), ); } + const connectionId = yield* options.allocateConnectionId; + const mutex = yield* Semaphore.make(1); + // Check size and insert without yielding so concurrent connects cannot both observe a free slot. if (connections.size >= MAX_CONNECTIONS) { return yield* Effect.fail(new AcpMcpOverAcpError("Too many MCP-over-ACP connections.")); } - const connectionId = yield* options.allocateConnectionId; connections.set(connectionId, { - mutex: yield* Semaphore.make(1), + mutex, sessionId: null, protocolVersion: null, nextRequestId: 0,