From 3b0be767c7a05cd85f0c38ce0abeca804c557cda Mon Sep 17 00:00:00 2001 From: Krishna Date: Mon, 5 Oct 2026 20:22:10 +0530 Subject: [PATCH 1/2] fix(cli): clean up MCP servers when their startup is interrupted InstanceState runs a lookup on the fiber of its first caller. When that caller is an HTTP request that the client abandons, the lookup is interrupted and ScopedCache kept the interruption forever, so MCP calls for the directory failed with 499 until the instance was disposed. MCP.state also registered its client finalizer only after every server connected, so servers that connected before the interruption were never closed, even by an instance dispose. Each abort and dispose cycle left them running. Expire interrupted InstanceState lookups immediately so the next caller closes the abandoned scope and retries, and register an MCP finalizer before connecting so already-connected servers are closed with it. Refs #14339 --- .changeset/mcp-interrupted-startup.md | 5 ++ .../opencode/src/effect/instance-state.ts | 6 +- .../src/kilocode/effect/instance-state.ts | 9 +++ packages/opencode/src/kilocode/mcp/cleanup.ts | 45 +++++++++++ packages/opencode/src/mcp/index.ts | 2 + .../test/kilocode/fixture/mcp-pid-stdio.ts | 22 +++++ .../kilocode/instance-state-interrupt.test.ts | 56 +++++++++++++ .../kilocode/mcp-interrupted-state.test.ts | 80 +++++++++++++++++++ 8 files changed, 224 insertions(+), 1 deletion(-) create mode 100644 .changeset/mcp-interrupted-startup.md create mode 100644 packages/opencode/src/kilocode/effect/instance-state.ts create mode 100644 packages/opencode/src/kilocode/mcp/cleanup.ts create mode 100644 packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts create mode 100644 packages/opencode/test/kilocode/instance-state-interrupt.test.ts create mode 100644 packages/opencode/test/kilocode/mcp-interrupted-state.test.ts diff --git a/.changeset/mcp-interrupted-startup.md b/.changeset/mcp-interrupted-startup.md new file mode 100644 index 00000000000..d743f57a320 --- /dev/null +++ b/.changeset/mcp-interrupted-startup.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": patch +--- + +Stop MCP servers from piling up when a request is cancelled while they are still starting. Servers that had already started are now shut down, and the next request starts MCP again instead of failing until the backend restarts. diff --git a/packages/opencode/src/effect/instance-state.ts b/packages/opencode/src/effect/instance-state.ts index 13b67187b4e..3151f26eaf5 100644 --- a/packages/opencode/src/effect/instance-state.ts +++ b/packages/opencode/src/effect/instance-state.ts @@ -1,5 +1,6 @@ import { Effect, ScopedCache, Scope } from "effect" import { capture, type InstanceContext } from "@/kilocode/instance" // kilocode_change +import * as KiloInstanceState from "@/kilocode/effect/instance-state" // kilocode_change import { InstanceRef, WorkspaceRef } from "./instance-ref" import { registerDisposer } from "./instance-registry" import { WorkspaceContext } from "@/control-plane/workspace-context" @@ -27,8 +28,11 @@ export const make = ( init: (ctx: InstanceContext) => Effect.Effect, ): Effect.Effect>, never, R | Scope.Scope> => Effect.gen(function* () { - const cache = yield* ScopedCache.make({ + // kilocode_change start - do not cache a lookup interrupted by a disconnected caller + const cache = yield* ScopedCache.makeWith({ capacity: Number.POSITIVE_INFINITY, + timeToLive: KiloInstanceState.ttl, + // kilocode_change end lookup: () => Effect.gen(function* () { return yield* init(yield* context) diff --git a/packages/opencode/src/kilocode/effect/instance-state.ts b/packages/opencode/src/kilocode/effect/instance-state.ts new file mode 100644 index 00000000000..332864f2aff --- /dev/null +++ b/packages/opencode/src/kilocode/effect/instance-state.ts @@ -0,0 +1,9 @@ +import { Duration, Exit } from "effect" + +/** + * Cache lifetime for an InstanceState lookup. The lookup runs on the fiber of the first + * caller, so a client that disconnects mid-lookup interrupts it. That interruption is not + * a result: expire it at once so the next caller closes the abandoned scope (running the + * finalizers it registered) and builds the state again. Real values and failures stay cached. + */ +export const ttl = (exit: Exit.Exit) => (Exit.hasInterrupts(exit) ? Duration.zero : Duration.infinity) diff --git a/packages/opencode/src/kilocode/mcp/cleanup.ts b/packages/opencode/src/kilocode/mcp/cleanup.ts new file mode 100644 index 00000000000..51eb00649fd --- /dev/null +++ b/packages/opencode/src/kilocode/mcp/cleanup.ts @@ -0,0 +1,45 @@ +import { Effect } from "effect" + +type Client = { + readonly transport?: unknown + close(): Promise +} + +/** + * Closes the clients the MCP state has connected so far. Registered before any server + * starts, so it also covers a state lookup that is interrupted part way through, when the + * upstream finalizer has not been registered yet and connected servers would otherwise + * keep running with no owner. When the lookup completes, the upstream finalizer runs + * first and empties `clients`, which makes this a no-op. + */ +export const release = ( + state: { clients: Record }, + descendants: (pid: number) => Effect.Effect, +) => + Effect.suspend(() => { + const clients = Object.values(state.clients) + state.clients = {} + return Effect.forEach( + clients, + (client) => + Effect.gen(function* () { + // Only stdio transports have a server process. + const transport = client.transport + if ( + typeof transport === "object" && + transport !== null && + "pid" in transport && + typeof transport.pid === "number" + ) { + for (const child of yield* descendants(transport.pid)) { + // The child may already have exited. + yield* Effect.try({ try: () => process.kill(child, "SIGTERM"), catch: () => undefined }).pipe( + Effect.ignore, + ) + } + } + yield* Effect.tryPromise(() => client.close()).pipe(Effect.ignore) + }), + { concurrency: "unbounded", discard: true }, + ) + }) diff --git a/packages/opencode/src/mcp/index.ts b/packages/opencode/src/mcp/index.ts index 4f1bafc78ea..e940cef83b7 100644 --- a/packages/opencode/src/mcp/index.ts +++ b/packages/opencode/src/mcp/index.ts @@ -34,6 +34,7 @@ import { McpOAuthPendingProvider, McpOAuthProvider, OAUTH_CALLBACK_PATH } from " import { McpOAuthCallback } from "./oauth-callback" import { McpAuth } from "./auth" import { probe } from "@/kilocode/mcp/sse-probe" // kilocode_change - normalize the optional GET stream probe +import * as KiloMcpCleanup from "@/kilocode/mcp/cleanup" // kilocode_change import { EventV2Bridge } from "@/event-v2-bridge" import { TuiEvent } from "@/server/tui-event" import { Cause, Effect, Exit, Layer, Context, Schema, Stream } from "effect" @@ -570,6 +571,7 @@ const layer = Layer.effect( defs: {}, instructions: {}, } + yield* Effect.addFinalizer(() => KiloMcpCleanup.release(s, descendants)) // kilocode_change - close servers if this lookup is interrupted yield* Effect.forEach( Object.entries(config), diff --git a/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts b/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts new file mode 100644 index 00000000000..4ec7ab2d718 --- /dev/null +++ b/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts @@ -0,0 +1,22 @@ +import { Server } from "@modelcontextprotocol/sdk/server/index.js" +import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js" +import { ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js" + +// Stdio MCP server that reports its pid. With --hang it never answers initialize. +const file = process.env.MCP_PID_FILE +if (!file) throw new Error("MCP_PID_FILE is required") + +if (process.argv.includes("--hang")) { + await Bun.write(file, String(process.pid)) + await new Promise(() => {}) +} + +const server = new Server({ name: "mcp-pid-stdio", version: "1.0.0" }, { capabilities: { tools: {} } }) + +server.setRequestHandler(ListToolsRequestSchema, async () => { + // Written once the client has connected and is listing tools. + await Bun.write(file, String(process.pid)) + return { tools: [{ name: "ping", description: "ping", inputSchema: { type: "object", properties: {} } }] } +}) + +await server.connect(new StdioServerTransport()) diff --git a/packages/opencode/test/kilocode/instance-state-interrupt.test.ts b/packages/opencode/test/kilocode/instance-state-interrupt.test.ts new file mode 100644 index 00000000000..c8c06c0e175 --- /dev/null +++ b/packages/opencode/test/kilocode/instance-state-interrupt.test.ts @@ -0,0 +1,56 @@ +import { expect } from "bun:test" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Deferred, Effect, Fiber, Layer } from "effect" +import { InstanceState } from "@/effect/instance-state" +import { provideInstanceEffect, testInstanceStoreLayer, tmpdirScoped } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +const it = testEffect(Layer.mergeAll(LayerNode.compile(CrossSpawnSpawner.node), testInstanceStoreLayer)) + +const access = (state: InstanceState.InstanceState, dir: string) => + InstanceState.get(state).pipe(provideInstanceEffect(dir)) + +it.live("InstanceState retries a lookup whose caller was interrupted", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped() + const started = yield* Deferred.make() + const released: number[] = [] + let n = 0 + const state = yield* InstanceState.make(() => + Effect.gen(function* () { + const id = ++n + yield* Effect.addFinalizer(() => Effect.sync(() => released.push(id))) + if (id === 1) { + yield* Deferred.succeed(started, undefined) + return yield* Effect.never + } + return id + }), + ) + + // A client that disconnects mid-lookup interrupts the request fiber running it. + const fiber = yield* access(state, dir).pipe(Effect.forkScoped) + yield* Deferred.await(started) + yield* Fiber.interrupt(fiber) + + // The next caller gets a fresh value instead of the cached interruption, and the + // abandoned lookup's resources are released. + expect(yield* access(state, dir)).toBe(2) + expect(released).toEqual([1]) + }), +) + +it.live("InstanceState still caches a lookup that failed", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped() + let n = 0 + const state = yield* InstanceState.make(() => Effect.suspend(() => Effect.fail(++n))) + + const a = yield* access(state, dir).pipe(Effect.flip) + const b = yield* access(state, dir).pipe(Effect.flip) + + expect(a).toBe(1) + expect(b).toBe(1) + }), +) diff --git a/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts b/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts new file mode 100644 index 00000000000..7515a14a4db --- /dev/null +++ b/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts @@ -0,0 +1,80 @@ +import path from "node:path" +import os from "node:os" +import fs from "node:fs/promises" +import { expect } from "bun:test" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Effect, Fiber } from "effect" +import { MCP } from "../../src/mcp/index" +import { pollWithTimeout, testEffect } from "../lib/effect" + +const it = testEffect(LayerNode.compile(MCP.node)) +const fixture = path.join(import.meta.dir, "fixture/mcp-pid-stdio.ts") +const dir = path.join(os.tmpdir(), `kilo-mcp-interrupt-${process.pid}`) +const ready = path.join(dir, "ready.pid") +const hanging = path.join(dir, "hanging.pid") + +const pid = (file: string) => + pollWithTimeout( + Effect.promise(async () => { + const handle = Bun.file(file) + return (await handle.exists()) ? Number(await handle.text()) : undefined + }), + `${path.basename(file)} was not written`, + ) + +const alive = (id: number) => { + try { + process.kill(id, 0) + return true + } catch { + return false + } +} + +it.instance( + "an interrupted MCP startup closes servers that already connected and retries", + () => + Effect.gen(function* () { + const mcp = yield* MCP.Service + + // A client that disconnects while MCP servers are still starting interrupts the lookup. + const fiber = yield* mcp.status().pipe(Effect.forkScoped) + const first = yield* pid(ready) + yield* pid(hanging) + yield* Effect.sleep("300 millis") + yield* Fiber.interrupt(fiber) + + const status = yield* mcp.status() + expect(status["ready"]?.status).toBe("connected") + expect(status["hanging"]?.status).toBe("failed") + + // The server that connected before the interruption is not left running. + yield* pollWithTimeout( + Effect.sync(() => (alive(first) ? undefined : true)), + "server from the interrupted startup is still running", + ) + }), + { + init: () => + Effect.promise(async () => { + await fs.rm(dir, { recursive: true, force: true }) + await fs.mkdir(dir, { recursive: true }) + }), + config: { + mcp: { + ready: { + type: "local", + command: [process.execPath, fixture], + environment: { MCP_PID_FILE: ready }, + }, + hanging: { + type: "local", + command: [process.execPath, fixture, "--hang"], + environment: { MCP_PID_FILE: hanging }, + timeout: 1500, + }, + }, + }, + }, + 30_000, +) From 6acd854af284c23d8a78b39ef624783048f16c5b Mon Sep 17 00:00:00 2001 From: Krishna Date: Tue, 6 Oct 2026 02:01:55 +0530 Subject: [PATCH 2/2] test(cli): wait for the stored MCP client instead of sleeping The interrupted-startup test slept 300 ms to make sure the ready client was in the MCP state before interrupting the lookup. The fixture now keeps sending tools/list_changed until the client re-lists tools, which only happens once the state holds it as connected, and writes a marker the test polls. The hanging server gets a longer timeout so it is still connecting when the interrupt lands. --- .../test/kilocode/fixture/mcp-pid-stdio.ts | 20 ++++++++++++++++--- .../kilocode/mcp-interrupted-state.test.ts | 11 ++++++++-- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts b/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts index 4ec7ab2d718..6e58e2b2e44 100644 --- a/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts +++ b/packages/opencode/test/kilocode/fixture/mcp-pid-stdio.ts @@ -11,11 +11,25 @@ if (process.argv.includes("--hang")) { await new Promise(() => {}) } -const server = new Server({ name: "mcp-pid-stdio", version: "1.0.0" }, { capabilities: { tools: {} } }) +const server = new Server( + { name: "mcp-pid-stdio", version: "1.0.0" }, + { capabilities: { tools: { listChanged: true } } }, +) +let timer: ReturnType | undefined +let listed = 0 server.setRequestHandler(ListToolsRequestSchema, async () => { - // Written once the client has connected and is listing tools. - await Bun.write(file, String(process.pid)) + listed++ + if (listed === 1) { + await Bun.write(file, String(process.pid)) + // The client only re-lists tools on a change notification once the MCP state holds it + // as a connected client, so keep announcing a change until that second request arrives. + timer = setInterval(() => void server.sendToolListChanged(), 20) + } + if (listed === 2) { + clearInterval(timer) + await Bun.write(`${file}.stored`, "") + } return { tools: [{ name: "ping", description: "ping", inputSchema: { type: "object", properties: {} } }] } }) diff --git a/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts b/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts index 7515a14a4db..a3794467b35 100644 --- a/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts +++ b/packages/opencode/test/kilocode/mcp-interrupted-state.test.ts @@ -41,7 +41,13 @@ it.instance( const fiber = yield* mcp.status().pipe(Effect.forkScoped) const first = yield* pid(ready) yield* pid(hanging) - yield* Effect.sleep("300 millis") + // The fixture writes this once the MCP state holds `ready` as a connected client. + yield* pollWithTimeout( + Effect.promise(() => Bun.file(`${ready}.stored`).exists()).pipe( + Effect.map((done) => (done ? true : undefined)), + ), + "ready client was never stored", + ) yield* Fiber.interrupt(fiber) const status = yield* mcp.status() @@ -71,7 +77,8 @@ it.instance( type: "local", command: [process.execPath, fixture, "--hang"], environment: { MCP_PID_FILE: hanging }, - timeout: 1500, + // Keeps the lookup in progress until the test interrupts it; also bounds the retry. + timeout: 5000, }, }, },