From b23ae55bd1483f40fec546ba323dfc954285b025 Mon Sep 17 00:00:00 2001 From: shum Date: Sat, 24 Oct 2026 14:27:06 +0000 Subject: [PATCH] mcp: fix daemon disconnects, stop and idle calls --- packages/simplex-chat-mcp/src/daemon.ts | 38 ++-- packages/simplex-chat-mcp/src/files.ts | 6 +- .../simplex-chat-mcp/test/unit/daemon.test.ts | 173 +++++++++++++++--- .../simplex-chat-mcp/test/unit/files.test.ts | 6 + 4 files changed, 185 insertions(+), 38 deletions(-) diff --git a/packages/simplex-chat-mcp/src/daemon.ts b/packages/simplex-chat-mcp/src/daemon.ts index ae68000fb8..902be335eb 100644 --- a/packages/simplex-chat-mcp/src/daemon.ts +++ b/packages/simplex-chat-mcp/src/daemon.ts @@ -56,6 +56,7 @@ interface Entry { opening: Promise agent?: AgentLike attached: number + calls: number idle?: NodeJS.Timeout } @@ -73,11 +74,11 @@ export class Daemon { this.nextNumber = Math.max(0, ...sessionNumbers(deps.root)) + 1 } - static async start(deps: DaemonDeps, config: Config): Promise { + static async start(deps: DaemonDeps, config: () => Config): Promise { acquireLock(deps.root) try { const daemon = new Daemon(deps) - daemon.hub = await deps.openHub(config, daemon.sessions(), (action, signal) => daemon.perform(action, signal)) + daemon.hub = await deps.openHub(config(), daemon.sessions(), (action, signal) => daemon.perform(action, signal)) try { daemon.server = await listen(socketPath(deps.root), deps.token, daemon.routes()) } catch (error) { @@ -129,15 +130,12 @@ export class Daemon { releaseLock(this.deps.root) log("Stopped.") const stopped = (): void => { + this.server.closeAllConnections() closed.then(() => this.deps.onStopped(0)) } - // the /stop reply is sent after this promise resolves; its connection closes only once the reply is written - if (reply === undefined) stopped() - else - reply.once("finish", () => { - this.server.closeIdleConnections() - stopped() - }) + // the /stop reply is sent after this promise resolves; "close" also fires when the client leaves before it is written + if (reply === undefined || reply.destroyed) stopped() + else reply.once("close", stopped) } private agents(): AgentLike[] { @@ -187,19 +185,26 @@ export class Daemon { } private async onCall({id, tool, input}: CallRequest, res: ServerResponse): Promise { - const agent = await this.entry(id).opening + const entry = this.entry(id) + const agent = await entry.opening if (!this.hold(res)) return const controller = new AbortController() res.once("close", () => { if (!res.writableFinished) controller.abort() }) const write = startLines(res) - const result = await agent.call(tool, input, {signal: controller.signal, progress: progress => write({progress})}) - if (!controller.signal.aborted) res.end(`${JSON.stringify({result: result ?? null})}\n`) + entry.calls++ + try { + const result = await agent.call(tool, input, {signal: controller.signal, progress: progress => write({progress})}) + if (!controller.signal.aborted) res.end(`${JSON.stringify({result: result ?? null})}\n`) + } finally { + entry.calls-- + if (entry.attached === 0 && entry.calls === 0) this.scheduleIdle(id, entry) + } } private hold(res: ServerResponse): boolean { - if (this.stopping !== undefined) { + if (this.stopping !== undefined || res.destroyed) { res.destroy() return false } @@ -210,10 +215,11 @@ export class Daemon { // the stop timer runs only while no entries exist, so a new entry cancels it private entry(id: string, request?: SessionRequest): Entry { + if (this.stopping !== undefined) throw new SocketError("The SimpleX daemon is stopping.", 503) const existing = this.entries.get(id) if (existing !== undefined) return existing clearTimeout(this.stopTimer) - const entry: Entry = {attached: 0, opening: this.openAgent(id, request)} + const entry: Entry = {attached: 0, calls: 0, opening: this.openAgent(id, request)} this.entries.set(id, entry) this.scheduleIdle(id, entry) entry.opening.then( @@ -247,7 +253,7 @@ export class Daemon { if (this.stopping !== undefined) return entry.idle = setTimeout(() => { const {agent} = entry - if (agent === undefined || this.hub.actions.ofSession(agent.number).length > 0) return this.scheduleIdle(id, entry) + if (agent === undefined || entry.calls > 0 || this.hub.actions.ofSession(agent.number).length > 0) return this.scheduleIdle(id, entry) this.entries.delete(id) this.closeAgent(id, agent) if (this.entries.size === 0) this.scheduleStop() @@ -282,7 +288,7 @@ export async function runDaemon(env: NodeJS.ProcessEnv = process.env): Promise readConfig(env)) } catch (error) { if (error instanceof LockedError) return ready("running") ready(`error: ${errorText(error)}`) diff --git a/packages/simplex-chat-mcp/src/files.ts b/packages/simplex-chat-mcp/src/files.ts index 8a67ab630c..4c11f2dfaf 100644 --- a/packages/simplex-chat-mcp/src/files.ts +++ b/packages/simplex-chat-mcp/src/files.ts @@ -86,7 +86,11 @@ export function acquireLock(dir: string): void { } export function releaseLock(dir: string): void { - unlinkSync(join(dir, "lock")) + try { + unlinkSync(join(dir, "lock")) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error + } } export function serverToken(dir: string): string { diff --git a/packages/simplex-chat-mcp/test/unit/daemon.test.ts b/packages/simplex-chat-mcp/test/unit/daemon.test.ts index 51ea663bbd..01241c2132 100644 --- a/packages/simplex-chat-mcp/test/unit/daemon.test.ts +++ b/packages/simplex-chat-mcp/test/unit/daemon.test.ts @@ -1,4 +1,7 @@ +import {once} from "events" import {closeSync, existsSync, mkdirSync, mkdtempSync, openSync, readFileSync, rmSync, writeFileSync} from "fs" +import {request} from "http" +import {Socket} from "net" import {tmpdir} from "os" import {join} from "path" import {Actions} from "../../src/actions" @@ -12,6 +15,7 @@ const id1 = "11111111-1111-4111-8111-111111111111" const id2 = "22222222-2222-4222-8222-222222222222" const token = "t" const minutes = 60_000 +const readDefaultConfig = () => defaultConfig const fakeTimers: Parameters[0] = {doNotFake: ["nextTick", "setImmediate", "queueMicrotask"]} interface FakeAgent { @@ -33,8 +37,10 @@ function deps(root: string) { const actions: {session: number}[] = [] const hubStops: number[] = [] const events: string[] = [] - const gates: {open?: Promise; close?: Promise} = {} + const gates: {open?: Promise; close?: Promise; call?: Promise; hubStop?: Promise} = {} const opening = deferred() + const calling = deferred() + const aborted = deferred() const hubStopping = deferred() const stopped = deferred() const onStopped = jest.fn(stopped.resolve) @@ -50,6 +56,7 @@ function deps(root: string) { stop: async () => { hubStops.push(agents.filter(({closed}) => closed).length) hubStopping.resolve() + await gates.hubStop return 0 }, } @@ -82,8 +89,12 @@ function deps(root: string) { }, call: async (tool: string, _input: unknown, call: {signal: AbortSignal; progress: (m: string) => void}) => { agent.calls.push(tool) + calling.resolve() call.progress("waiting") - if (tool === "receive") await new Promise(resolve => call.signal.addEventListener("abort", resolve)) + if (tool === "receive") { + await new Promise(resolve => call.signal.addEventListener("abort", resolve)) + aborted.resolve() + } else await gates.call return call.signal.aborted ? undefined : {content: [{type: "text", text: "{}"}], structuredContent: {}} }, } @@ -95,7 +106,7 @@ function deps(root: string) { gates[name] = promise return resolve } - return {d, agents, configs, actions, hubStops, events, onStopped, stoppedCode: stopped.promise, opening: opening.promise, hubStopping: hubStopping.promise, gate} + return {d, agents, configs, actions, hubStops, events, onStopped, stoppedCode: stopped.promise, opening: opening.promise, calling: calling.promise, aborted: aborted.promise, hubStopping: hubStopping.promise, gate} } let root: string @@ -125,6 +136,10 @@ function attach(id: string, signal: AbortSignal) { return {attached, firstLine: first.promise, lines} } +function serverConnection(daemon: Daemon): Promise { + return once(daemon["server"], "connection").then(([socket]) => socket as Socket) +} + function storeSession(id: string, number: number, name: string) { const folder = sessionFolder(root, id) createDataFolder(folder) @@ -134,7 +149,7 @@ function storeSession(id: string, number: number, name: string) { describe("daemon", () => { it("registers sessions with stable numbers and renames on a new title", async () => { const {d, agents, configs} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) const first = await register(id1) expect(first).toEqual({number: 1, displayName: "SimpleX MCP - app", version: "0.2.0", protocol: protocolVersion, owner: {owners: [], link: "https://simplex.chat/g#link"}}) expect(await post("/session", {id: id2, name: "docs", project: "/p/app", config: {...defaultConfig, mainGroup: "Agents"}})).toMatchObject({number: 2, displayName: "SimpleX MCP - docs"}) @@ -146,18 +161,18 @@ describe("daemon", () => { it("opens one agent for concurrent registrations", async () => { const {d, agents} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await Promise.all([1, 2, 3].map(() => register(id1))) expect(agents).toHaveLength(1) }) it("holds the lock while running and refuses a second daemon in the same folder", async () => { const {d} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) expect(readFileSync(lockFile(), "utf8")).toBe(`${process.pid}\n`) writeFileSync(lockFile(), `${process.ppid}\n`) const openHub = jest.fn(d.openHub) - await expect(Daemon.start({...d, openHub}, defaultConfig)).rejects.toThrow(LockedError) + await expect(Daemon.start({...d, openHub}, readDefaultConfig)).rejects.toThrow(LockedError) expect(openHub).not.toHaveBeenCalled() writeFileSync(lockFile(), `${process.pid}\n`) }) @@ -165,14 +180,14 @@ describe("daemon", () => { it("stops the hub and releases the lock when it cannot listen", async () => { const {d, hubStops} = deps(root) mkdirSync(socketPath(root)) - await expect(Daemon.start(d, defaultConfig)).rejects.toThrow() + await expect(Daemon.start(d, readDefaultConfig)).rejects.toThrow() expect(hubStops).toHaveLength(1) expect(existsSync(lockFile())).toBe(false) }) it("streams progress and the result, and cancels the call when the client closes", async () => { - const {d, agents} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + const {d, agents, aborted} = deps(root) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) const lines: unknown[] = [] await streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, line => lines.push(line)) @@ -180,12 +195,13 @@ describe("daemon", () => { const controller = new AbortController() const receiving = streamLines(socketPath(root), token, "/call", {id: id1, tool: "receive", input: {chat: "@1"}}, () => controller.abort(), controller.signal) await expect(receiving).rejects.toThrow() + await aborted expect(agents[0].calls).toEqual(["list", "receive"]) }) it("refuses a call or an attachment of an unknown session", async () => { const {d, agents} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) const unknown = "Unknown session; restart the Claude Code session." await expect(streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, () => undefined)).rejects.toThrow(unknown) await expect(streamLines(socketPath(root), token, "/attach", {id: id1}, () => undefined)).rejects.toMatchObject({message: unknown, status: 409}) @@ -194,7 +210,7 @@ describe("daemon", () => { it("writes the versions first on attach and counts attachments", async () => { const {d} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) const controller = new AbortController() const {attached, firstLine, lines} = attach(id1, controller.signal) @@ -208,7 +224,7 @@ describe("daemon", () => { it("closes an idle agent without actions, and then stops", async () => { jest.useFakeTimers(fakeTimers) const {d, agents, actions, onStopped, stoppedCode} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) const controller = new AbortController() const {attached, firstLine} = attach(id1, controller.signal) @@ -233,7 +249,7 @@ describe("daemon", () => { it("closes a registered agent that never attaches", async () => { jest.useFakeTimers(fakeTimers) const {d, agents} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) await jest.advanceTimersByTimeAsync(10 * minutes - 1) expect(agents[0].closed).toBe(false) @@ -244,7 +260,7 @@ describe("daemon", () => { it("stops 10 minutes after start when no session registers", async () => { jest.useFakeTimers(fakeTimers) const {d, stoppedCode} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await jest.advanceTimersByTimeAsync(10 * minutes) expect(await stoppedCode).toBe(0) daemon = undefined @@ -253,7 +269,7 @@ describe("daemon", () => { it("keeps running when a session registers during the idle stop wait", async () => { jest.useFakeTimers(fakeTimers) const {d, agents, onStopped} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await jest.advanceTimersByTimeAsync(5 * minutes) await register(id1) await jest.advanceTimersByTimeAsync(9 * minutes) @@ -264,7 +280,7 @@ describe("daemon", () => { it("keeps running after a refused call of an unknown session, and stops when idle", async () => { jest.useFakeTimers(fakeTimers) const {d, stoppedCode} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await expect(streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, () => undefined)).rejects.toThrow() await jest.advanceTimersByTimeAsync(10 * minutes) expect(await stoppedCode).toBe(0) @@ -275,7 +291,7 @@ describe("daemon", () => { storeSession(id1, 1, "app") jest.useFakeTimers(fakeTimers) const {d, agents} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, () => undefined) expect(agents).toMatchObject([{number: 1, displayName: "SimpleX MCP - app", calls: ["list"]}]) expect(await register(id2)).toMatchObject({number: 2}) @@ -286,7 +302,7 @@ describe("daemon", () => { it("opens a session again only after its idle close completes", async () => { jest.useFakeTimers(fakeTimers) const {d, events, gate} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) const openClose = gate("close") await jest.advanceTimersByTimeAsync(10 * minutes) @@ -301,7 +317,7 @@ describe("daemon", () => { storeSession(id1, 1, "app") const {d, agents, opening, hubStopping, gate, stoppedCode} = deps(root) const finishOpen = gate("open") - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) const attached = streamLines(socketPath(root), token, "/attach", {id: id1}, () => undefined) await opening const stopping = daemon.stop() @@ -313,9 +329,117 @@ describe("daemon", () => { expect(await stoppedCode).toBe(0) }) + it("does not count an attachment whose client left while the session opened", async () => { + jest.useFakeTimers(fakeTimers) + storeSession(id1, 1, "app") + const {d, agents, opening, gate} = deps(root) + const finishOpen = gate("open") + daemon = await Daemon.start(d, readDefaultConfig) + const connection = serverConnection(daemon) + const controller = new AbortController() + const attached = streamLines(socketPath(root), token, "/attach", {id: id1}, () => undefined, controller.signal).catch(() => undefined) + await opening + controller.abort() + await once(await connection, "close") + finishOpen() + await attached + expect(await requestJson(socketPath(root), token, "GET", "/status")).toMatchObject({sessions: [{number: 1, attached: 0}]}) + await jest.advanceTimersByTimeAsync(10 * minutes) + expect(agents[0].closed).toBe(true) + }) + + it("does not run a call whose client left while the session opened", async () => { + storeSession(id1, 1, "app") + const {d, agents, opening, gate, stoppedCode} = deps(root) + const finishOpen = gate("open") + daemon = await Daemon.start(d, readDefaultConfig) + const connection = serverConnection(daemon) + const controller = new AbortController() + const called = streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, () => undefined, controller.signal).catch(() => undefined) + await opening + controller.abort() + await once(await connection, "close") + finishOpen() + await called + await daemon.stop() + expect(agents[0].calls).toEqual([]) + expect(await stoppedCode).toBe(0) + }) + + it("keeps an unattached session open during a call and closes it 10 minutes after", async () => { + jest.useFakeTimers(fakeTimers) + const {d, agents, calling, gate} = deps(root) + daemon = await Daemon.start(d, readDefaultConfig) + await register(id1) + const finishCall = gate("call") + const called = streamLines(socketPath(root), token, "/call", {id: id1, tool: "list", input: {}}, () => undefined) + await calling + await jest.advanceTimersByTimeAsync(25 * minutes) + expect(agents[0].closed).toBe(false) + expect(await requestJson(socketPath(root), token, "GET", "/status")).toMatchObject({sessions: [{number: 1, attached: 0}]}) + finishCall() + await called + await jest.advanceTimersByTimeAsync(10 * minutes - 1) + expect(agents[0].closed).toBe(false) + await jest.advanceTimersByTimeAsync(1) + expect(agents[0].closed).toBe(true) + }) + + it("refuses a registration that arrives while the daemon stops", async () => { + const {d, agents, gate, hubStopping, stoppedCode} = deps(root) + const finishHubStop = gate("hubStop") + daemon = await Daemon.start(d, readDefaultConfig) + const requested = once(daemon["server"], "request") + const req = request({socketPath: socketPath(root), method: "POST", path: "/session", headers: {authorization: `Bearer ${token}`}}) + const response = new Promise<{status: number | undefined; text: string}>((resolve, reject) => { + req.on("response", res => { + let text = "" + res.setEncoding("utf8") + res.on("data", chunk => (text += chunk)) + res.on("end", () => resolve({status: res.statusCode, text})) + }) + req.on("error", reject) + }) + req.flushHeaders() + await requested + const stopping = daemon.stop() + await hubStopping + req.end(JSON.stringify({id: id1, project: "/p/app", config: defaultConfig})) + expect(await response).toEqual({status: 503, text: JSON.stringify({error: "The SimpleX daemon is stopping."})}) + finishHubStop() + await stopping + expect(agents).toHaveLength(0) + expect(await stoppedCode).toBe(0) + }) + + it("stops when the /stop client leaves before the reply", async () => { + const {d, hubStopping, gate, stoppedCode} = deps(root) + const finishHubStop = gate("hubStop") + daemon = await Daemon.start(d, readDefaultConfig) + const connection = serverConnection(daemon) + const controller = new AbortController() + const stopRequest = streamLines(socketPath(root), token, "/stop", undefined, () => undefined, controller.signal).catch(() => undefined) + await hubStopping + controller.abort() + await once(await connection, "close") + finishHubStop() + await stopRequest + expect(await stoppedCode).toBe(0) + daemon = undefined + }) + + it("stops when its lock file is already gone", async () => { + const {d, stoppedCode} = deps(root) + daemon = await Daemon.start(d, readDefaultConfig) + rmSync(lockFile()) + await daemon.stop() + expect(await stoppedCode).toBe(0) + daemon = undefined + }) + it("reports status, resets owners and stops on request", async () => { const {d, agents, hubStops, stoppedCode} = deps(root) - daemon = await Daemon.start(d, defaultConfig) + daemon = await Daemon.start(d, readDefaultConfig) await register(id1) expect(await requestJson(socketPath(root), token, "GET", "/status")).toEqual({ version: "0.2.0", @@ -343,6 +467,13 @@ describe("runDaemon", () => { expect(readFileSync(lockFile(), "utf8")).toBe(`${process.ppid}\n`) }) + it("reports running before it reads the config", async () => { + writeFileSync(lockFile(), `${process.ppid}\n`) + const write = jest.spyOn(process.stdout, "write").mockImplementation(() => true) + await runDaemon({SIMPLEX_MCP_HOME: root, CLAUDE_PLUGIN_OPTION_APPROVE: "always"}) + expect(write.mock.calls).toEqual([["running\n"]]) + }) + it("reports to the ready descriptor when given", async () => { writeFileSync(lockFile(), `${process.ppid}\n`) const readyFile = join(root, "ready") diff --git a/packages/simplex-chat-mcp/test/unit/files.test.ts b/packages/simplex-chat-mcp/test/unit/files.test.ts index 109fa281f3..7d9323b83a 100644 --- a/packages/simplex-chat-mcp/test/unit/files.test.ts +++ b/packages/simplex-chat-mcp/test/unit/files.test.ts @@ -25,6 +25,12 @@ describe("lock", () => { expect(existsSync(join(dir, "lock"))).toBe(false) }) + it("tolerates a lock that is already removed, and reports other release errors", () => { + releaseLock(dir) + mkdirSync(join(dir, "lock")) + expect(() => releaseLock(dir)).toThrow(expect.objectContaining({code: expect.stringMatching(/^(EISDIR|EPERM)$/)})) + }) + it("refuses a lock of a running process", () => { const runningPid = process.ppid writeFileSync(join(dir, "lock"), `${runningPid}\n`)