diff --git a/src/agents/agent-bundle-mcp-runtime.test.ts b/src/agents/agent-bundle-mcp-runtime.test.ts index 6fef0f952a8d..de52ba217671 100644 --- a/src/agents/agent-bundle-mcp-runtime.test.ts +++ b/src/agents/agent-bundle-mcp-runtime.test.ts @@ -46,7 +46,10 @@ async function writeListToolsMcpServer(params: { inputSchema?: unknown; tools?: Array<{ name: string; description?: string; inputSchema?: unknown }>; capabilities?: Record; + pidPath?: string; notifyListChangedOnInitialized?: boolean; + notifyListChangedAfterFirstList?: boolean; + exitOnListCall?: number; listToolsMethodNotFound?: boolean; callToolIsError?: boolean; callToolJsonRpcError?: boolean; @@ -62,7 +65,10 @@ const delayMs = ${params.delayMs ?? 0}; const initializeDelayMs = ${params.initializeDelayMs ?? 0}; const hang = ${params.hang === true}; const capabilities = ${JSON.stringify(params.capabilities ?? { tools: {} })}; +const pidPath = ${JSON.stringify(params.pidPath)}; const notifyListChangedOnInitialized = ${params.notifyListChangedOnInitialized === true}; +const notifyListChangedAfterFirstList = ${params.notifyListChangedAfterFirstList === true}; +const exitOnListCall = ${params.exitOnListCall ?? 0}; const listToolsMethodNotFound = ${params.listToolsMethodNotFound === true}; const tools = ${JSON.stringify( params.tools ?? [ @@ -78,8 +84,12 @@ const callToolJsonRpcError = ${params.callToolJsonRpcError === true}; const resourceListJsonRpcError = ${params.resourceListJsonRpcError === true}; let buffer = ""; +let listCount = 0; let pendingTimer; let keepAlive; +if (pidPath) { + await fs.writeFile(pidPath, String(process.pid), "utf8"); +} function log(line) { void fs.appendFile(logPath, line + "\\n", "utf8").catch(() => {}); } @@ -116,6 +126,11 @@ function handle(message) { return; } if (message.method === "tools/list") { + listCount += 1; + if (listCount === exitOnListCall) { + log("exit tools/list " + listCount); + process.exit(1); + } if (listToolsMethodNotFound) { log("reject tools/list method not found"); send({ @@ -130,6 +145,7 @@ function handle(message) { keepAlive = setInterval(() => {}, 1000); return; } + const currentListCount = listCount; log("delay tools/list " + delayMs); pendingTimer = setTimeout(() => { send({ @@ -139,6 +155,10 @@ function handle(message) { tools, }, }); + if (notifyListChangedAfterFirstList && currentListCount === 1) { + log("notify tools/list_changed"); + send({ jsonrpc: "2.0", method: "notifications/tools/list_changed" }); + } }, delayMs); } if (message.method === "tools/call") { @@ -247,6 +267,29 @@ async function waitForPredicate( throw new Error(`Timed out waiting for ${description}`); } +async function waitForErrorMessage( + action: () => Promise, + expectedText: string, + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + let lastMessage = ""; + while (Date.now() < deadline) { + try { + await action(); + } catch (error) { + lastMessage = error instanceof Error ? error.message : String(error); + if (lastMessage.includes(expectedText)) { + return lastMessage; + } + } + await new Promise((resolve) => { + setTimeout(resolve, 10); + }); + } + throw new Error(`Timed out waiting for ${expectedText}; saw ${JSON.stringify(lastMessage)}`); +} + function makeRuntime( tools: Array<{ toolName: string; description: string }>, serverName = "bundleProbe", @@ -1036,6 +1079,92 @@ process.on("SIGINT", shutdown);`, } }); + it("fails fast with an attributable error after an MCP child process exits", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-child-exit-")); + const serverPath = path.join(tempDir, "server.mjs"); + const logPath = path.join(tempDir, "server.log"); + const pidPath = path.join(tempDir, "server.pid"); + await writeListToolsMcpServer({ filePath: serverPath, logPath, pidPath }); + + const runtime = await getOrCreateSessionMcpRuntime({ + sessionId: "session-child-exit", + sessionKey: "agent:test:session-child-exit", + workspaceDir: "/workspace", + cfg: { + mcp: { + servers: { + child: { command: process.execPath, args: [serverPath] }, + }, + }, + }, + }); + + try { + await expect(runtime.callTool("child", "slow_tool", {})).resolves.toMatchObject({ + isError: false, + }); + await waitForFileText(pidPath, "", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS); + const pid = Number.parseInt((await fs.readFile(pidPath, "utf8")).trim(), 10); + process.kill(pid); + + const message = await waitForErrorMessage( + () => runtime.callTool("child", "slow_tool", {}), + "is disconnected", + LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, + ); + expect(message).toBe('bundle-mcp server "child" is disconnected: mcp transport closed'); + } finally { + await runtime.dispose(); + await fs.rm(tempDir, { recursive: true, force: true }); + } + }); + + it("retires a reused MCP session that exits during catalog refresh", async () => { + const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-refresh-exit-")); + const serverPath = path.join(tempDir, "server.mjs"); + const logPath = path.join(tempDir, "server.log"); + await writeListToolsMcpServer({ + filePath: serverPath, + logPath, + capabilities: { tools: { listChanged: true } }, + notifyListChangedAfterFirstList: true, + exitOnListCall: 2, + }); + + const runtime = await getOrCreateSessionMcpRuntime({ + sessionId: "session-refresh-exit", + sessionKey: "agent:test:session-refresh-exit", + workspaceDir: "/workspace", + cfg: { + mcp: { + servers: { + child: { command: process.execPath, args: [serverPath] }, + }, + }, + }, + }); + + try { + expect((await runtime.getCatalog()).tools).toHaveLength(1); + await waitForFileText(logPath, "notify tools/list_changed", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS); + await waitForPredicate( + () => runtime.peekCatalog() === null, + "list_changed to invalidate the catalog", + LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, + ); + + const refreshedCatalog = await runtime.getCatalog(); + expect(refreshedCatalog.tools).toEqual([]); + expect(refreshedCatalog.diagnostics?.[0]?.serverName).toBe("child"); + await expect(runtime.callTool("child", "slow_tool", {})).rejects.toThrow( + 'bundle-mcp server "child" is not connected', + ); + } finally { + await runtime.dispose(); + await fs.rm(tempDir, { recursive: true, force: true }); + } + }); + it("does not cache a catalog invalidated while discovery is in flight", async () => { const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-inflight-invalidated-")); const serverPath = path.join(tempDir, "inflight-invalidated.mjs"); diff --git a/src/agents/agent-bundle-mcp-runtime.ts b/src/agents/agent-bundle-mcp-runtime.ts index 9c33db055864..4da471f5547c 100644 --- a/src/agents/agent-bundle-mcp-runtime.ts +++ b/src/agents/agent-bundle-mcp-runtime.ts @@ -45,6 +45,7 @@ type BundleMcpSession = { requestTimeoutMs: number; supportsParallelToolCalls: boolean; connected: boolean; + disconnectReason?: string; retiring: boolean; catalogUseCount: number; sharedAcrossCatalogGenerations: boolean; @@ -542,6 +543,17 @@ export function createSessionMcpRuntime(params: { throw createDisposedError(params.sessionId); } }; + const requireConnectedSession = (serverName: string): BundleMcpSession => { + const session = sessions.get(serverName); + if (!session || !session.connected) { + throw new Error( + session?.disconnectReason + ? `bundle-mcp server "${serverName}" is disconnected: ${session.disconnectReason}` + : `bundle-mcp server "${serverName}" is not connected`, + ); + } + return session; + }; const ensureSessionConnected = async ( session: BundleMcpSession, connectionTimeoutMs: number, @@ -640,6 +652,18 @@ export function createSessionMcpRuntime(params: { failIfDisposed(); let session = sessions.get(serverName); + while ( + session && + !session.retiring && + !session.connected && + !session.connectPromise + ) { + // A closed SDK client cannot reconnect cleanly on the same transport. + await retireSessionIfCurrent(serverName, session); + // Retirement yields while closing. Preserve any replacement that a + // newer catalog generation installed during that await. + session = sessions.get(serverName); + } if (session?.retiring) { session = undefined; } @@ -670,7 +694,7 @@ export function createSessionMcpRuntime(params: { }, }, ); - session = { + const createdSession: BundleMcpSession = { serverName, client, transport: resolved.transport, @@ -683,6 +707,14 @@ export function createSessionMcpRuntime(params: { sharedAcrossCatalogGenerations: false, detachStderr: resolved.detachStderr, }; + // The SDK exposes lifecycle hooks as callback properties. A close is + // terminal for this client/transport pair. + // oxlint-disable-next-line unicorn/prefer-add-event-listener -- MCP Client is not an EventTarget. + client.onclose = () => { + createdSession.connected = false; + createdSession.disconnectReason = "mcp transport closed"; + }; + session = createdSession; sessions.set(serverName, session); } @@ -693,11 +725,9 @@ export function createSessionMcpRuntime(params: { session.sharedAcrossCatalogGenerations = true; } session.catalogUseCount += 1; - let connectedForCatalog = false; try { failIfDisposed(); await ensureSessionConnected(session, resolved.connectionTimeoutMs); - connectedForCatalog = true; failIfDisposed(); const capabilities = summarizeServerCapabilities( session.client.getServerCapabilities(), @@ -782,9 +812,9 @@ export function createSessionMcpRuntime(params: { ]; const sharedWithNewerGeneration = session.sharedAcrossCatalogGenerations || session.catalogUseCount > 1; - if (!connectedForCatalog && !session.connected) { - // Timed-out connects can still leave the SDK client bound to a - // transport. Delete before async close so future catalogs start fresh. + if (!session.connected) { + // A close is terminal for every catalog generation sharing this + // session. The identity guard preserves any newer replacement. await retireSessionIfCurrent(serverName, session); } else if (!reusedSession && !sharedWithNewerGeneration) { // Catalog invalidation can overlap generations; an older failed @@ -894,10 +924,7 @@ export function createSessionMcpRuntime(params: { async callTool(serverName, toolName, input) { failIfDisposed(); await getCatalog(); - const session = sessions.get(serverName); - if (!session) { - throw new Error(`bundle-mcp server "${serverName}" is not connected`); - } + const session = requireConnectedSession(serverName); return await runGuardedServerRequest( serverName, async () => @@ -914,10 +941,7 @@ export function createSessionMcpRuntime(params: { async listResources(serverName) { failIfDisposed(); await getCatalog(); - const session = sessions.get(serverName); - if (!session) { - throw new Error(`bundle-mcp server "${serverName}" is not connected`); - } + const session = requireConnectedSession(serverName); return await runGuardedServerRequest(serverName, async () => listAllResources(session.client, session.requestTimeoutMs), ); @@ -925,10 +949,7 @@ export function createSessionMcpRuntime(params: { async readResource(serverName, uri) { failIfDisposed(); await getCatalog(); - const session = sessions.get(serverName); - if (!session) { - throw new Error(`bundle-mcp server "${serverName}" is not connected`); - } + const session = requireConnectedSession(serverName); return await runGuardedServerRequest( serverName, async () => @@ -938,10 +959,7 @@ export function createSessionMcpRuntime(params: { async listPrompts(serverName) { failIfDisposed(); await getCatalog(); - const session = sessions.get(serverName); - if (!session) { - throw new Error(`bundle-mcp server "${serverName}" is not connected`); - } + const session = requireConnectedSession(serverName); return await runGuardedServerRequest(serverName, async () => listAllPrompts(session.client, session.requestTimeoutMs), ); @@ -949,10 +967,7 @@ export function createSessionMcpRuntime(params: { async getPrompt(serverName, name, args) { failIfDisposed(); await getCatalog(); - const session = sessions.get(serverName); - if (!session) { - throw new Error(`bundle-mcp server "${serverName}" is not connected`); - } + const session = requireConnectedSession(serverName); return await runGuardedServerRequest( serverName, async () => diff --git a/test/scripts/lint-suppressions.test.ts b/test/scripts/lint-suppressions.test.ts index 005e9c0bcc64..fa9e7f050823 100644 --- a/test/scripts/lint-suppressions.test.ts +++ b/test/scripts/lint-suppressions.test.ts @@ -193,6 +193,7 @@ describe("production lint suppressions", () => { "extensions/feishu/src/bitable.ts|typescript/no-unnecessary-type-parameters|1", "extensions/matrix/src/onboarding.test-harness.ts|typescript/no-unnecessary-type-parameters|1", "extensions/slack/src/monitor/provider-support.ts|typescript/no-unnecessary-type-parameters|1", + "src/agents/agent-bundle-mcp-runtime.ts|unicorn/prefer-add-event-listener|1", "src/channels/plugins/channel-runtime-surface.types.ts|typescript/no-unnecessary-type-parameters|1", "src/channels/plugins/contracts/test-helpers.ts|typescript/no-unnecessary-type-parameters|1", "src/channels/plugins/types.plugin.ts|typescript/no-explicit-any|1",