From 1cf8ea446df40b1cd2228889f80928f8353b0fcd Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 17 Aug 2026 21:37:12 -0700 Subject: [PATCH] fix(mcp): prevent stale catalogs and silent tool failures (#125564) * fix(mcp): harden lifecycle and result integrity * fix(mcp): bound catalog invalidation replay * test(mcp): stabilize process cleanup proof --- config/assertion-safety-baseline.txt | 2 +- src/agents/agent-bundle-mcp-materialize.ts | 34 +- src/agents/agent-bundle-mcp-runtime.test.ts | 188 +++++++--- src/agents/agent-bundle-mcp-runtime.ts | 133 +++---- src/agents/mcp-client-lifecycle.ts | 10 +- src/agents/mcp-content.ts | 33 +- src/agents/mcp-error.ts | 23 ++ src/agents/mcp-http-transport.test.ts | 166 +++++++++ src/agents/mcp-http-transport.ts | 189 ++++++++++ .../mcp-stdio-transport.process.test.ts | 55 +++ src/agents/mcp-stdio-transport.test.ts | 18 + src/agents/mcp-stdio-transport.ts | 27 +- src/agents/mcp-tool-metadata.ts | 59 +++ src/agents/mcp-transport.ts | 14 +- src/agents/node-plugin-tools.test.ts | 33 +- src/agents/node-plugin-tools.ts | 11 +- src/node-host/invoke.mcp.test.ts | 9 +- src/node-host/invoke.ts | 34 +- src/node-host/mcp.recovery.test.ts | 132 ++++++- src/node-host/mcp.test.ts | 28 +- src/node-host/mcp.ts | 103 ++++-- .../runtime/gateway-node-mcp.fixture.mjs | 223 ++++++++++-- .../gateway-node-mcp.stress.e2e.test.ts | 336 ++++++++++++++++++ .../gateway-node-mcp.test-support.test.ts | 53 +++ .../runtime/gateway-node-mcp.test-support.ts | 90 +++-- test/scripts/lint-suppressions.test.ts | 1 + 26 files changed, 1699 insertions(+), 305 deletions(-) create mode 100644 src/agents/mcp-error.ts create mode 100644 src/agents/mcp-http-transport.test.ts create mode 100644 src/agents/mcp-http-transport.ts create mode 100644 src/agents/mcp-stdio-transport.process.test.ts create mode 100644 src/agents/mcp-tool-metadata.ts create mode 100644 test/e2e/qa-lab/runtime/gateway-node-mcp.stress.e2e.test.ts create mode 100644 test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.test.ts diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index b33754753e28..5e909773e0de 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -1683,7 +1683,7 @@ src/agents/agent-bundle-lsp-runtime.ts 4 src/agents/agent-bundle-mcp-combined.ts 2 src/agents/agent-bundle-mcp-manager-api.ts 2 src/agents/agent-bundle-mcp-manager-lifecycle.ts 1 -src/agents/agent-bundle-mcp-materialize.ts 4 +src/agents/agent-bundle-mcp-materialize.ts 3 src/agents/agent-bundle-mcp-runtime-shared.ts 2 src/agents/agent-bundle-mcp-runtime.ts 9 src/agents/agent-command-restart-recovery.ts 4 diff --git a/src/agents/agent-bundle-mcp-materialize.ts b/src/agents/agent-bundle-mcp-materialize.ts index 7c7ba7136b3b..edb8035ea1af 100644 --- a/src/agents/agent-bundle-mcp-materialize.ts +++ b/src/agents/agent-bundle-mcp-materialize.ts @@ -20,7 +20,7 @@ import type { McpToolCatalog, SessionMcpRuntime, } from "./agent-bundle-mcp-types.js"; -import { projectMcpCallToolResultContent } from "./mcp-content.js"; +import { projectMcpCallToolResult } from "./mcp-content.js"; import { isMcpToolAllowed } from "./mcp-tool-filter.js"; import { buildMcpAppCanvasPayload, fetchMcpAppView } from "./mcp-ui-resource.js"; import type { AgentToolResult } from "./runtime/index.js"; @@ -100,38 +100,10 @@ function toAgentToolResult(params: { toolName: string; result: CallToolResult; }): AgentToolResult { - const content = projectMcpCallToolResultContent(params.result); - const normalizedContent: AgentToolResult["content"] = - content.length > 0 - ? content - : ([ - { - type: "text", - text: JSON.stringify( - { - status: params.result.isError === true ? "error" : "ok", - server: params.serverName, - tool: params.toolName, - }, - null, - 2, - ), - }, - ] as AgentToolResult["content"]); - const details: Record = { + return projectMcpCallToolResult(params.result, { mcpServer: params.serverName, mcpTool: params.toolName, - }; - if (params.result.structuredContent !== undefined) { - details.structuredContent = params.result.structuredContent; - } - if (params.result.isError === true) { - details.status = "error"; - } - return { - content: normalizedContent, - details, - }; + }); } function toJsonAgentToolResult(params: { diff --git a/src/agents/agent-bundle-mcp-runtime.test.ts b/src/agents/agent-bundle-mcp-runtime.test.ts index e7562f57c8c7..2cd4f83d27b6 100644 --- a/src/agents/agent-bundle-mcp-runtime.test.ts +++ b/src/agents/agent-bundle-mcp-runtime.test.ts @@ -93,6 +93,8 @@ async function writeListToolsMcpServer(params: { name: string; description?: string; inputSchema?: unknown; + outputSchema?: unknown; + execution?: { taskSupport?: "forbidden" | "optional" | "required" }; _meta?: Record; }>; capabilities?: Record; @@ -101,6 +103,7 @@ async function writeListToolsMcpServer(params: { hangToolCallsUntilRestartMarkerPath?: string; notifyListChangedOnInitialized?: boolean; notifyListChangedAfterFirstList?: boolean; + notifyListChangedBeforeEveryListResponse?: boolean; exitOnListCall?: number; listToolsMethodNotFound?: boolean; listToolsJsonRpcErrorMessage?: string; @@ -137,6 +140,7 @@ const hangToolCallsUntilRestartMarkerPath = ${JSON.stringify( )}; const notifyListChangedOnInitialized = ${params.notifyListChangedOnInitialized === true}; const notifyListChangedAfterFirstList = ${params.notifyListChangedAfterFirstList === true}; +const notifyListChangedBeforeEveryListResponse = ${params.notifyListChangedBeforeEveryListResponse === true}; const exitOnListCall = ${params.exitOnListCall ?? 0}; const listToolsMethodNotFound = ${params.listToolsMethodNotFound === true}; const listToolsJsonRpcErrorMessage = ${JSON.stringify(params.listToolsJsonRpcErrorMessage)}; @@ -257,6 +261,10 @@ function handle(message) { const toolPageCursor = toolPageCursors?.[currentListCount - 1]; log("delay tools/list " + delayMs); const sendListResponse = () => { + if (notifyListChangedBeforeEveryListResponse) { + log("notify tools/list_changed before response"); + send({ jsonrpc: "2.0", method: "notifications/tools/list_changed" }); + } send({ jsonrpc: "2.0", id: message.id, @@ -1180,7 +1188,7 @@ describe("session MCP runtime", () => { const diagnostic = catalog.diagnostics?.[0]; expect(diagnostic?.serverName).toBe("diagnostic"); expect(diagnostic?.message).toContain("Authorization: Bearer "); - expect(diagnostic?.message).toContain("…"); + expect(diagnostic?.message).toContain("***"); expect(diagnostic?.message).not.toContain(secret); } finally { await runtime.dispose(); @@ -1754,7 +1762,7 @@ process.on("SIGINT", shutdown);`, LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, ); await expect(runtime.callTool("child", "slow_tool", {})).rejects.toThrow("is not connected"); - await waitForFileTextCount(logPath, "recv tools/list", 2, LIST_TOOLS_SERVER_LOG_TIMEOUT_MS); + await waitForFileTextCount(logPath, "recv tools/list", 2, LIST_TOOLS_TEST_DEADLINE_MS); await expect( withTestTimeout( runtime.callTool("healthy", "slow_tool", {}), @@ -1772,13 +1780,9 @@ process.on("SIGINT", shutdown);`, } }, "child server to reconnect", - LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, - ); - const replacementPid = await waitForChangedPid( - pidPath, - pid, - LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, + LIST_TOOLS_TEST_DEADLINE_MS, ); + const replacementPid = await waitForChangedPid(pidPath, pid, LIST_TOOLS_TEST_DEADLINE_MS); expect(replacementPid).not.toBe(pid); } finally { await runtime.dispose(); @@ -1821,11 +1825,10 @@ process.on("SIGINT", shutdown);`, ); const refreshedCatalog = await runtime.getCatalog(); - expect(refreshedCatalog.tools).toEqual([]); - expect(refreshedCatalog.diagnostics?.[0]?.serverName).toBe("child"); - // The refresh reports the exited server, but the runtime does not stay stuck on it: - // the closed transport invalidated the catalog, so the next request rebuilds against - // a fresh child instead of failing with "is not connected" indefinitely. + expect(refreshedCatalog.tools.map((tool) => tool.toolName)).toEqual(["slow_tool"]); + expect(refreshedCatalog.diagnostics ?? []).toEqual([]); + // The failed refresh is retired before catalog loading returns, so callers + // see only the replacement generation and never receive its stale diagnostic. await expect(runtime.callTool("child", "slow_tool", {})).resolves.toMatchObject({ isError: false, }); @@ -1934,9 +1937,9 @@ process.on("SIGINT", shutdown);`, try { const firstCatalog = await runtime.getCatalog(); - expect(firstCatalog.tools.map((tool) => tool.toolName)).toEqual(["old_tool"]); + expect(firstCatalog.tools.map((tool) => tool.toolName)).toEqual(["new_tool"]); await waitForFileText(logPath, "sent tools/list_changed", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS); - expect(runtime.peekCatalog()).toBeNull(); + expect(runtime.peekCatalog()?.tools.map((tool) => tool.toolName)).toEqual(["new_tool"]); const secondCatalog = await runtime.getCatalog(); expect(secondCatalog.tools.map((tool) => tool.toolName)).toEqual(["new_tool"]); @@ -1947,6 +1950,57 @@ process.on("SIGINT", shutdown);`, } }); + it("bounds catalog replay when a server invalidates every tools/list response", async () => { + const tempDir = tempDirTracker.make("bundle-mcp-continuous-invalidation-"); + const noisyServerPath = path.join(tempDir, "noisy-server.mjs"); + const noisyLogPath = path.join(tempDir, "noisy-server.log"); + const healthyServerPath = path.join(tempDir, "healthy-server.mjs"); + const healthyLogPath = path.join(tempDir, "healthy-server.log"); + await writeListToolsMcpServer({ + filePath: noisyServerPath, + logPath: noisyLogPath, + capabilities: { tools: { listChanged: true } }, + tools: [{ name: "noisy_tool", inputSchema: { type: "object", properties: {} } }], + notifyListChangedBeforeEveryListResponse: true, + }); + await writeListToolsMcpServer({ + filePath: healthyServerPath, + logPath: healthyLogPath, + tools: [{ name: "healthy_tool", inputSchema: { type: "object", properties: {} } }], + }); + + const runtime = await getOrCreateSessionMcpRuntime({ + sessionId: "session-continuous-invalidation", + sessionKey: "agent:test:session-continuous-invalidation", + workspaceDir: "/workspace", + cfg: { + mcp: { + servers: { + noisy: { command: process.execPath, args: [noisyServerPath] }, + healthy: { command: process.execPath, args: [healthyServerPath] }, + }, + }, + }, + }); + + try { + const catalog = await withTestTimeout( + runtime.getCatalog(), + LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, + "continuous tools/list invalidation blocked the catalog", + ); + + expect(catalog.tools.map((tool) => tool.toolName).toSorted()).toEqual([ + "healthy_tool", + "noisy_tool", + ]); + const noisyLog = await fs.readFile(noisyLogPath, "utf8"); + expect(noisyLog.match(/tools\/list cursor/g)).toHaveLength(2); + } finally { + await runtime.dispose(); + } + }); + it.each([ { name: "resource-only servers reporting method not found", @@ -2281,7 +2335,7 @@ process.on("SIGINT", shutdown);`, hanging: { command: process.execPath, args: [serverPath], - requestTimeoutMs: 25, + requestTimeoutMs: 500, }, }, }, @@ -2312,9 +2366,9 @@ process.on("SIGINT", shutdown);`, } }, "timed-out server to recover without stale backoff", - LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, + LIST_TOOLS_TEST_DEADLINE_MS, ); - expect(await waitForChangedPid(pidPath, pid, LIST_TOOLS_SERVER_LOG_TIMEOUT_MS)).not.toBe(pid); + expect(await waitForChangedPid(pidPath, pid, LIST_TOOLS_TEST_DEADLINE_MS)).not.toBe(pid); } finally { await runtime.dispose(); await fs.rm(tempDir, { recursive: true, force: true }); @@ -2348,6 +2402,55 @@ process.on("SIGINT", shutdown);`, } }); + it("retains paginated output metadata and hides required-task tools", async () => { + const tempDir = tempDirTracker.make("bundle-mcp-tool-metadata-pages-"); + const serverPath = path.join(tempDir, "tool-pages.mjs"); + const logPath = path.join(tempDir, "server.log"); + await writeListToolsMcpServer({ + filePath: serverPath, + logPath, + toolPageCursors: ["page-2", null], + tools: [ + { + name: "structured", + inputSchema: { type: "object", properties: {} }, + outputSchema: { + type: "object", + properties: { count: { type: "number" } }, + required: ["count"], + additionalProperties: false, + }, + }, + { + name: "task_only", + inputSchema: { type: "object", properties: {} }, + execution: { taskSupport: "required" }, + }, + ], + callToolResult: { + content: [{ type: "text", text: "invalid" }], + structuredContent: { count: "not-a-number" }, + }, + }); + const runtime = createSessionMcpRuntime({ + sessionId: "session-tool-metadata-pages", + workspaceDir: "/workspace", + cfg: { + mcp: { servers: { paged: { command: process.execPath, args: [serverPath] } } }, + }, + }); + + try { + const catalog = await runtime.getCatalog(); + expect(catalog.tools.map((tool) => tool.toolName)).toEqual(["structured-1", "structured-2"]); + await expect(runtime.callTool("paged", "structured-1", {})).rejects.toThrow( + "does not match the tool's output schema", + ); + } finally { + await runtime.dispose(); + } + }); + it("isolates a cyclic tool catalog while a healthy bundle MCP sibling survives", async () => { const tempDir = tempDirTracker.make("bundle-mcp-tool-cycle-"); const loopingPath = path.join(tempDir, "looping.mjs"); @@ -5669,15 +5772,15 @@ function log(line) { function send(message) { process.stdout.write(JSON.stringify(message) + "\\n"); } -async function isFirstConnect() { +async function claimFirstConnect() { try { - const handle = await fs.open(markerPath, "wx"); - await handle.close(); + await fs.writeFile(markerPath, String(process.pid), { flag: "wx" }); return true; } catch { return false; } } +const firstConnect = await claimFirstConnect(); async function handle(message) { if (!message || typeof message !== "object") { return; @@ -5693,7 +5796,7 @@ async function handle(message) { serverInfo: { name: "timeout-slow", version: "1.0.0" }, }, }; - if (await isFirstConnect()) { + if (firstConnect) { log("slow first initialize"); return; } @@ -5749,7 +5852,7 @@ process.on("SIGINT", shutdown);`, slow: { command: process.execPath, args: [slowServerPath], - connectionTimeoutMs: 150, + connectionTimeoutMs: 1_000, }, }, }, @@ -5758,6 +5861,8 @@ process.on("SIGINT", shutdown);`, try { const firstCatalog = runtime.getCatalog(); + await waitForFileText(firstConnectMarkerPath, "", LIST_TOOLS_SERVER_LOG_TIMEOUT_MS); + const firstSlowPid = Number(await fs.readFile(firstConnectMarkerPath, "utf8")); await waitForFileText( triggerLogPath, "sent initial tools/list_changed", @@ -5770,23 +5875,23 @@ process.on("SIGINT", shutdown);`, secondCatalogPromise, ]); - const firstSlowDiagnostic = firstCatalogResult.diagnostics?.find( - (diag) => diag.serverName === "slow", - ); - expect(firstSlowDiagnostic?.message).toContain("timed out"); - expect(firstCatalogResult.servers.slow).toBeUndefined(); + // The notification refresh is serialized after the timed-out first generation, + // so callers now receive the newest clean catalog rather than its stale diagnostic. + expect(firstCatalogResult.servers.slow).toBeDefined(); expect(secondCatalog.servers.trigger).toBeDefined(); - const secondSlowDiagnostic = secondCatalog.diagnostics?.find( - (diag) => diag.serverName === "slow", + expect(secondCatalog.servers.slow).toBeDefined(); + await waitForPredicate( + () => { + try { + process.kill(firstSlowPid, 0); + return false; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ESRCH"; + } + }, + "timed-out first MCP generation to exit", + LIST_TOOLS_SERVER_LOG_TIMEOUT_MS, ); - // A loaded runner can let generation one retire the timed-out client before - // generation two adopts it. Both the shared timeout and fast replacement are valid. - if (secondSlowDiagnostic) { - expect(secondSlowDiagnostic.message).toContain("timed out"); - expect(secondCatalog.servers.slow).toBeUndefined(); - } else { - expect(secondCatalog.servers.slow).toBeDefined(); - } await expect(runtime.callTool("trigger", "poke", {})).resolves.toMatchObject({ content: [{ type: "text", text: "poked" }], isError: false, @@ -5822,8 +5927,8 @@ process.on("SIGINT", shutdown);`, ); it( - "does not dispose sessions shared with a newer catalog generation", - { timeout: LIST_TOOLS_TEST_DEADLINE_MS }, + "serializes invalidated catalog generations on one session", + { timeout: LIST_TOOLS_TEST_DEADLINE_MS * 2 }, async () => { const tempDir = makeTempDir(tempDirs, "bundle-mcp-overlap-generation-"); const serverPath = path.join(tempDir, "overlap-server.mjs"); @@ -5945,7 +6050,8 @@ process.on("SIGINT", shutdown);`, const secondCatalog = await runtime.getCatalog(); const firstCatalogResult = await firstCatalog; - expect(firstCatalogResult.diagnostics?.[0]?.serverName).toBe("overlap"); + expect(firstCatalogResult.diagnostics ?? []).toEqual([]); + expect(firstCatalogResult.tools.map((tool) => tool.toolName)).toEqual(["ok_tool"]); expect(secondCatalog.diagnostics ?? []).toEqual([]); expect(secondCatalog.tools.map((tool) => tool.toolName)).toEqual(["ok_tool"]); diff --git a/src/agents/agent-bundle-mcp-runtime.ts b/src/agents/agent-bundle-mcp-runtime.ts index d2c8bff1731e..6dde5e909d01 100644 --- a/src/agents/agent-bundle-mcp-runtime.ts +++ b/src/agents/agent-bundle-mcp-runtime.ts @@ -3,18 +3,18 @@ import { Client, type ClientOptions } from "@modelcontextprotocol/sdk/client/ind import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { ErrorCode, + ListToolsResultSchema, McpError, type CallToolResult, type ClientCapabilities, type ServerCapabilities, + type Tool, } from "@modelcontextprotocol/sdk/types.js"; -import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import type { SessionToolOverrides } from "../config/sessions/types.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { logWarn } from "../logger.js"; -import { redactToolPayloadText } from "../logging/redact.js"; import type { PluginManifestRegistry } from "../plugins/manifest-registry.js"; import { runTasksWithConcurrency } from "../utils/run-with-concurrency.js"; import { mergeMcpToolCatalogs } from "./agent-bundle-mcp-combined.js"; @@ -66,10 +66,12 @@ import { applyMcpConnectionOverride, type McpServerConnectionResolved, } from "./mcp-connection-resolver.js"; +import { redactMcpDiagnosticError } from "./mcp-error.js"; import { createMcpJsonSchemaValidator } from "./mcp-json-schema-validator.js"; import { sanitizeMcpMetadataText } from "./mcp-metadata.js"; import { collectMcpPaginatedItems } from "./mcp-pagination.js"; import { isMcpToolAllowed, normalizeMcpToolFilter } from "./mcp-tool-filter.js"; +import { createMcpToolCatalogMetadata, type McpToolCatalogMetadata } from "./mcp-tool-metadata.js"; import { resolveMcpTransport } from "./mcp-transport.js"; type BundleMcpSession = { @@ -82,13 +84,12 @@ type BundleMcpSession = { connected: boolean; disconnectReason?: string; retiring: boolean; - catalogUseCount: number; - sharedAcrossCatalogGenerations: boolean; connectPromise?: Promise; detachStderr?: () => void; + toolMetadata?: McpToolCatalogMetadata; }; -type ListedTool = Awaited>["tools"][number]; +type ListedTool = Tool; const MCP_APPS_CLIENT_EXTENSION = "io.modelcontextprotocol/ui"; const MCP_APP_RESOURCE_MIME_TYPE = "text/html;profile=mcp-app"; const BUNDLE_MCP_FAILURE_THRESHOLD = 3; @@ -123,12 +124,13 @@ type McpServerBackoffState = { export { createMcpJsonSchemaValidator as createBundleMcpJsonSchemaValidator }; -function redactMcpDiagnosticError(error: unknown): string { - return redactToolPayloadText(redactSensitiveUrlLikeString(String(error))); -} - -async function listAllTools(client: Client, timeoutMs: number, signal: AbortSignal) { - return await collectMcpPaginatedItems({ +async function listAllTools( + client: Client, + timeoutMs: number, + signal: AbortSignal, + schemaValidator = createMcpJsonSchemaValidator(), +) { + const tools = await collectMcpPaginatedItems({ label: "MCP tool listing", itemLabel: "tools", timeoutMs, @@ -144,17 +146,26 @@ async function listAllTools(client: Client, timeoutMs: number, signal: AbortSign onAbort(); } try { - const page = await client.listTools(cursor === undefined ? undefined : { cursor }, { - timeout: requestTimeoutMs, - maxTotalTimeout: requestTimeoutMs, - signal: requestController.signal, - }); + const page = await client.request( + { method: "tools/list", params: cursor === undefined ? undefined : { cursor } }, + ListToolsResultSchema, + { + timeout: requestTimeoutMs, + maxTotalTimeout: requestTimeoutMs, + signal: requestController.signal, + }, + ); return { items: page.tools, nextCursor: page.nextCursor, serializedValue: page }; } finally { requestSignal.removeEventListener("abort", onAbort); } }, }); + const metadata = createMcpToolCatalogMetadata(tools, schemaValidator); + return { + tools: tools.filter((tool) => !metadata.isRequiredTaskTool(tool.name)), + metadata, + }; } function isMcpMethodNotFoundError(error: unknown): boolean { @@ -165,22 +176,6 @@ function isMcpMethodNotFoundError(error: unknown): boolean { return message.includes("-32601") || /\b(?:method not found|unknown method)\b/i.test(message); } -async function listAllToolsBestEffort(params: { - client: Client; - timeoutMs: number; - signal: AbortSignal; - suppressUnsupported: boolean; -}): Promise { - try { - return await listAllTools(params.client, params.timeoutMs, params.signal); - } catch (error) { - if (params.suppressUnsupported && isMcpMethodNotFoundError(error)) { - return []; - } - throw error; - } -} - function hasConfiguredMcpRequestTimeout(rawServer: unknown): boolean { if (!rawServer || typeof rawServer !== "object") { return false; @@ -316,7 +311,6 @@ export function createSessionMcpRuntime(params: { catalogInvalidationGeneration += 1; catalog = null; catalogRetryAfterMs = undefined; - catalogInFlight = undefined; }; const scheduleCatalogServerRetry = (serverName: string, message: string) => { const currentCatalog = catalog; @@ -689,6 +683,7 @@ export function createSessionMcpRuntime(params: { session = undefined; } const reusedSession = Boolean(session); + const schemaValidator = createMcpJsonSchemaValidator(); if (!session) { const client = new Client( { @@ -697,7 +692,7 @@ export function createSessionMcpRuntime(params: { }, { ...buildMcpClientOptions(mcpAppsEnabled), - jsonSchemaValidator: createMcpJsonSchemaValidator(), + jsonSchemaValidator: schemaValidator, listChanged: { tools: { autoRefresh: false, @@ -723,8 +718,6 @@ export function createSessionMcpRuntime(params: { supportsParallelToolCalls: resolved.supportsParallelToolCalls, connected: false, retiring: false, - catalogUseCount: 0, - sharedAcrossCatalogGenerations: false, detachStderr: resolved.detachStderr, }; // The SDK exposes lifecycle hooks as callback properties. A close is @@ -750,13 +743,6 @@ export function createSessionMcpRuntime(params: { sessions.set(serverName, session); } - if (session.catalogUseCount === 0) { - session.sharedAcrossCatalogGenerations = false; - } - if (reusedSession && session.catalogUseCount > 0) { - session.sharedAcrossCatalogGenerations = true; - } - session.catalogUseCount += 1; try { failIfDisposed(); await ensureSessionConnected(session, resolved.connectionTimeoutMs); @@ -764,14 +750,27 @@ export function createSessionMcpRuntime(params: { const capabilities = summarizeServerCapabilities( session.client.getServerCapabilities(), ); - const listedTools = await listAllToolsBestEffort({ - client: session.client, - timeoutMs: getCatalogListTimeoutMs(rawServer, resolved.requestTimeoutMs), - signal: lifecycleAbortController.signal, - suppressUnsupported: Boolean( - !capabilities.tools && (capabilities.resources || capabilities.prompts), - ), - }); + let listedTools: ListedTool[]; + try { + const listed = await listAllTools( + session.client, + getCatalogListTimeoutMs(rawServer, resolved.requestTimeoutMs), + lifecycleAbortController.signal, + schemaValidator, + ); + listedTools = listed.tools; + session.toolMetadata = listed.metadata; + } catch (error) { + if ( + !capabilities.tools && + (capabilities.resources || capabilities.prompts) && + isMcpMethodNotFoundError(error) + ) { + listedTools = []; + } else { + throw error; + } + } failIfDisposed(); const toolFilter = normalizeMcpToolFilter( isRecord(rawServer) ? rawServer.toolFilter : undefined, @@ -865,15 +864,13 @@ export function createSessionMcpRuntime(params: { message, }, ]; - const sharedWithNewerGeneration = - session.sharedAcrossCatalogGenerations || session.catalogUseCount > 1; 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 - // generation must not dispose a session a newer one already reused. + } else if (!reusedSession && catalogInvalidationGeneration === catalogGeneration) { + // An isolated startup failure gets a fresh process on retry. When a + // notification superseded this list, the queued generation reuses it. await retireSessionIfCurrent(serverName, session); } failIfDisposed(); @@ -883,11 +880,6 @@ export function createSessionMcpRuntime(params: { toolEntries: [], diagnostics: diags, } as ServerResult; - } finally { - session.catalogUseCount -= 1; - if (session.catalogUseCount === 0) { - session.sharedAcrossCatalogGenerations = false; - } } }, ); @@ -960,7 +952,14 @@ export function createSessionMcpRuntime(params: { return catalog; } if (!catalog) { - return loadCatalog(); + await loadCatalog(); + if (catalog) { + return catalog; + } + // Replay one in-flight invalidation before accepting the latest completed + // snapshot. A server that invalidates every list must not block its siblings. + const replayedCatalog = await loadCatalog(); + return catalog ?? replayedCatalog; } const staleCatalog = catalog; @@ -1021,18 +1020,24 @@ export function createSessionMcpRuntime(params: { }, async callTool(serverName, toolName, input) { const session = await getActiveSession(serverName); - return (await runGuardedMcpRequest(serverName, session, (signal) => + const result = (await runGuardedMcpRequest(serverName, session, (signal) => session.client.callTool( { name: toolName, arguments: isRecord(input) ? input : {} }, undefined, { timeout: session.requestTimeoutMs, signal }, ), )) as CallToolResult; + session.toolMetadata?.validateResult(toolName, result); + return result; }, async listTools(serverName, requestParams) { const session = await getActiveSession(serverName); return await runGuardedMcpRequest(serverName, session, (signal) => - session.client.listTools(requestParams, { timeout: session.requestTimeoutMs, signal }), + session.client.request( + { method: "tools/list", params: requestParams }, + ListToolsResultSchema, + { timeout: session.requestTimeoutMs, signal }, + ), ); }, async listResources(serverName, options) { diff --git a/src/agents/mcp-client-lifecycle.ts b/src/agents/mcp-client-lifecycle.ts index f2ba99735e57..18fb1b46b67c 100644 --- a/src/agents/mcp-client-lifecycle.ts +++ b/src/agents/mcp-client-lifecycle.ts @@ -1,11 +1,9 @@ import type { Client } from "@modelcontextprotocol/sdk/client/index.js"; -import { - StreamableHTTPClientTransport, - StreamableHTTPError, -} from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { StreamableHTTPError } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { ErrorCode } from "@modelcontextprotocol/sdk/types.js"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { OpenClawStreamableHTTPClientTransport } from "./mcp-http-transport.js"; import { OpenClawStdioClientTransport } from "./mcp-stdio-transport.js"; type LifecycleSession = { @@ -24,7 +22,7 @@ export function isStatefulMcpHttpSessionExpired( ): boolean { return ( session.transportType === "streamable-http" && - session.transport instanceof StreamableHTTPClientTransport && + session.transport instanceof OpenClawStreamableHTTPClientTransport && session.transport.sessionId !== undefined && error instanceof StreamableHTTPError && error.code === 404 @@ -67,7 +65,7 @@ export async function connectMcpClient(params: { transportType: params.transport instanceof OpenClawStdioClientTransport ? "stdio" - : params.transport instanceof StreamableHTTPClientTransport + : params.transport instanceof OpenClawStreamableHTTPClientTransport ? "streamable-http" : "sse", }, diff --git a/src/agents/mcp-content.ts b/src/agents/mcp-content.ts index 415f9a7a5c8d..ef762bf7bce3 100644 --- a/src/agents/mcp-content.ts +++ b/src/agents/mcp-content.ts @@ -12,7 +12,7 @@ function stringifyMcpContent(value: unknown): string { } /** Converts untrusted MCP content into the agent text/image contract. */ -export function mcpContentBlockToAgentContent(block: unknown): McpAgentContentBlock { +function mcpContentBlockToAgentContent(block: unknown): McpAgentContentBlock { if (!isRecord(block)) { return { type: "text", text: stringifyMcpContent(block) }; } @@ -55,7 +55,7 @@ export function mcpContentBlockToAgentContent(block: unknown): McpAgentContentBl return { type: "text", text: stringifyMcpContent(block) }; } -export function projectMcpCallToolResultContent(result: { +function projectMcpCallToolResultContent(result: { content?: unknown; structuredContent?: unknown; }): AgentToolResult["content"] { @@ -73,3 +73,32 @@ export function projectMcpCallToolResultContent(result: { } return sourceContent.map(mcpContentBlockToAgentContent); } + +/** Projects a raw MCP CallToolResult exactly once at the model boundary. */ +export function projectMcpCallToolResult( + result: { content?: unknown; structuredContent?: unknown; isError?: unknown }, + details: Record = {}, +): AgentToolResult { + const isError = result.isError === true; + const content = projectMcpCallToolResultContent(result); + return { + content: + content.length > 0 + ? content + : [ + { + type: "text", + text: isError + ? "MCP tool failed without returning content." + : "MCP tool completed without returning content.", + }, + ], + details: { + ...details, + ...(result.structuredContent !== undefined + ? { structuredContent: result.structuredContent } + : {}), + ...(isError ? { status: "error" } : {}), + }, + }; +} diff --git a/src/agents/mcp-error.ts b/src/agents/mcp-error.ts new file mode 100644 index 000000000000..e5bf0c2800c1 --- /dev/null +++ b/src/agents/mcp-error.ts @@ -0,0 +1,23 @@ +import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url"; +import { formatErrorMessage } from "../infra/errors.js"; +import { redactToolPayloadText } from "../logging/redact.js"; + +const STREAMABLE_RESPONSE_BODY_MARKER = "Error POSTing to endpoint:"; +const LEGACY_RESPONSE_BODY_RE = /Error POSTing to endpoint \(HTTP \d+\):/; + +/** Redacts MCP diagnostics, including response bodies the SDK includes in thrown errors. */ +export function redactMcpDiagnosticError(error: unknown): string { + let message = formatErrorMessage(error); + const streamableIndex = message.indexOf(STREAMABLE_RESPONSE_BODY_MARKER); + const legacyMatch = LEGACY_RESPONSE_BODY_RE.exec(message); + const prefixEnd = + streamableIndex >= 0 + ? streamableIndex + STREAMABLE_RESPONSE_BODY_MARKER.length + : legacyMatch + ? legacyMatch.index + legacyMatch[0].length + : undefined; + if (prefixEnd !== undefined) { + message = `${message.slice(0, prefixEnd)} [redacted response body]`; + } + return redactToolPayloadText(redactSensitiveUrlLikeString(message)); +} diff --git a/src/agents/mcp-http-transport.test.ts b/src/agents/mcp-http-transport.test.ts new file mode 100644 index 000000000000..b5b483f2c37f --- /dev/null +++ b/src/agents/mcp-http-transport.test.ts @@ -0,0 +1,166 @@ +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { describe, expect, it, vi } from "vitest"; +import { disposeMcpClient } from "./mcp-client-lifecycle.js"; +import { redactMcpDiagnosticError } from "./mcp-error.js"; +import { + OpenClawSSEClientTransport, + OpenClawStreamableHTTPClientTransport, +} from "./mcp-http-transport.js"; + +function jsonResponse(value: unknown, init?: ResponseInit): Response { + const headers = new Headers(init?.headers); + headers.set("content-type", "application/json"); + return new Response(JSON.stringify(value), { + ...init, + headers, + }); +} + +function initializedFetch(params: { + onGet: () => Promise | Response; + onDelete?: (init: RequestInit) => void; +}) { + return vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + if (init?.method === "DELETE") { + params.onDelete?.(init); + return new Response(null, { status: 204 }); + } + if (init?.method === "GET") { + return await params.onGet(); + } + if (typeof init?.body !== "string") { + throw new Error("expected serialized JSON-RPC request body"); + } + const message = JSON.parse(init.body) as { id?: number; method?: string }; + if (message.method === "initialize") { + return jsonResponse( + { + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-06-18", + capabilities: { tools: { listChanged: true } }, + serverInfo: { name: "fixture", version: "1" }, + }, + }, + { headers: { "mcp-session-id": "session-1" } }, + ); + } + return new Response(null, { status: 202 }); + }); +} + +describe("OpenClaw MCP HTTP lifecycle adapters", () => { + it.each([ + "Streamable HTTP error: Error POSTing to endpoint: bearer=body-secret", + "Error POSTing to endpoint (HTTP 500): bearer=body-secret", + ])("redacts an HTTP response body from %s", (message) => { + const redacted = redactMcpDiagnosticError(new Error(message)); + expect(redacted).not.toContain("body-secret"); + expect(redacted).toContain("[redacted response body]"); + }); + + it("turns legacy SSE HTTP 204 into owner-visible closure", async () => { + const transport = new OpenClawSSEClientTransport(new URL("http://mcp.invalid/sse"), { + eventSourceInit: { + fetch: async () => new Response(null, { status: 204, statusText: "No Content" }), + }, + }); + const onclose = vi.fn(); + // MCP transports expose callback properties rather than EventTarget listeners. + // oxlint-disable-next-line unicorn/prefer-add-event-listener + transport.onclose = onclose; + + await expect(transport.start()).rejects.toThrow(); + await vi.waitFor(() => expect(onclose).toHaveBeenCalledOnce()); + }); + + it("closes after Streamable notification retry exhaustion", async () => { + let getCount = 0; + const fetchMock = initializedFetch({ + onGet: () => { + getCount += 1; + return getCount === 1 + ? new Response(new ReadableStream({ start: (controller) => controller.close() }), { + headers: { "content-type": "text/event-stream" }, + }) + : new Response(null, { status: 503, statusText: "Unavailable" }); + }, + }); + const transport = new OpenClawStreamableHTTPClientTransport(new URL("http://mcp.invalid/mcp"), { + fetch: fetchMock, + reconnectionOptions: { + initialReconnectionDelay: 1, + maxReconnectionDelay: 1, + reconnectionDelayGrowFactor: 1, + maxRetries: 2, + }, + }); + const client = new Client({ name: "test", version: "1" }); + const onclose = vi.fn(); + // MCP clients expose callback properties rather than EventTarget listeners. + // oxlint-disable-next-line unicorn/prefer-add-event-listener + client.onclose = onclose; + + await client.connect(transport); + await vi.waitFor(() => expect(onclose).toHaveBeenCalledOnce()); + expect(fetchMock.mock.calls.filter((call) => call[1]?.method === "GET")).toHaveLength(3); + }); + + it("sends stateful DELETE after failed initialization closed the SDK transport", async () => { + const deleteRequests: RequestInit[] = []; + const fetchMock = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + if (init?.method === "DELETE") { + deleteRequests.push(init); + return new Response(null, { status: 204 }); + } + return new Response("initialize failed", { + status: 500, + headers: { "mcp-session-id": "allocated-before-failure" }, + }); + }); + const transport = new OpenClawStreamableHTTPClientTransport(new URL("http://mcp.invalid/mcp"), { + fetch: fetchMock, + }); + const client = new Client({ name: "test", version: "1" }); + + await expect(client.connect(transport)).rejects.toThrow("initialize failed"); + await disposeMcpClient({ client, transport, transportType: "streamable-http" }); + + expect(deleteRequests).toHaveLength(1); + expect(new Headers(deleteRequests[0]?.headers).get("mcp-session-id")).toBe( + "allocated-before-failure", + ); + expect(deleteRequests[0]?.signal?.aborted).toBe(false); + }); + + it("does not fetch another notification stream after close returns", async () => { + let getCount = 0; + const fetchMock = initializedFetch({ + onGet: () => { + getCount += 1; + return new Response(new ReadableStream({ start: (controller) => controller.close() }), { + headers: { "content-type": "text/event-stream" }, + }); + }, + }); + const transport = new OpenClawStreamableHTTPClientTransport(new URL("http://mcp.invalid/mcp"), { + fetch: fetchMock, + reconnectionOptions: { + initialReconnectionDelay: 20, + maxReconnectionDelay: 20, + reconnectionDelayGrowFactor: 1, + maxRetries: 2, + }, + }); + const client = new Client({ name: "test", version: "1" }); + await client.connect(transport); + await vi.waitFor(() => expect(getCount).toBe(1)); + + await client.close(); + await new Promise((resolve) => { + setTimeout(resolve, 80); + }); + expect(getCount).toBe(1); + }); +}); diff --git a/src/agents/mcp-http-transport.ts b/src/agents/mcp-http-transport.ts new file mode 100644 index 000000000000..807745597309 --- /dev/null +++ b/src/agents/mcp-http-transport.ts @@ -0,0 +1,189 @@ +import { + SSEClientTransport, + SseError, + type SSEClientTransportOptions, +} from "@modelcontextprotocol/sdk/client/sse.js"; +import { + StreamableHTTPClientTransport, + StreamableHTTPError, + type StreamableHTTPClientTransportOptions, +} from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import type { FetchLike, Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; +import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"; + +const STREAM_RETRY_EXHAUSTED_RE = /^Maximum reconnection attempts \(\d+\) exceeded\.$/; +const SESSION_TERMINATION_TIMEOUT_MS = 5_000; + +abstract class OpenClawMcpHttpTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + protected closed = false; + private closeEmitted = false; + + protected emitClose(): void { + if (this.closeEmitted) { + return; + } + this.closeEmitted = true; + this.onclose?.(); + } + + protected emitError(error: Error): void { + if (!this.closed) { + this.onerror?.(error); + } + } + + abstract start(): Promise; + abstract close(): Promise; + abstract send(message: JSONRPCMessage): Promise; +} + +/** Converts legacy SSE terminal HTTP failures into the lifecycle close the SDK omits. */ +export class OpenClawSSEClientTransport extends OpenClawMcpHttpTransport { + private readonly transport: SSEClientTransport; + + constructor(url: URL, options?: SSEClientTransportOptions) { + super(); + this.transport = new SSEClientTransport(url, options); + } + + async start(): Promise { + // The SDK transport exposes callback properties rather than EventTarget listeners. + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onmessage = (message) => this.onmessage?.(message); + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onclose = () => this.emitClose(); + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onerror = (error) => { + this.emitError(error); + if (error instanceof SseError && error.code === 204) { + void this.close(); + } + }; + await this.transport.start(); + } + + async close(): Promise { + if (this.closed) { + return; + } + this.closed = true; + await this.transport.close(); + this.emitClose(); + } + + async send(message: JSONRPCMessage): Promise { + await this.transport.send(message); + } + + setProtocolVersion(version: string): void { + this.transport.setProtocolVersion(version); + } +} + +type OpenClawStreamableHttpOptions = StreamableHTTPClientTransportOptions & { + fetch?: FetchLike; + requestInit?: RequestInit; +}; + +/** Owns Streamable HTTP notification recovery and stateful cleanup around SDK 1.30.0. */ +export class OpenClawStreamableHTTPClientTransport extends OpenClawMcpHttpTransport { + private readonly transport: StreamableHTTPClientTransport; + private readonly url: URL; + private readonly cleanupFetch: FetchLike; + private readonly requestInit?: RequestInit; + private terminatedSessionId?: string; + + constructor(url: URL, options: OpenClawStreamableHttpOptions = {}) { + super(); + this.url = url; + this.cleanupFetch = options.fetch ?? fetch; + this.requestInit = options.requestInit; + const runtimeFetch: FetchLike = async (input, init) => { + if (this.closed) { + throw new Error("MCP Streamable HTTP transport is closed"); + } + return await this.cleanupFetch(input, init); + }; + this.transport = new StreamableHTTPClientTransport(url, { + ...options, + fetch: runtimeFetch, + }); + } + + get sessionId(): string | undefined { + return this.transport.sessionId; + } + + get protocolVersion(): string | undefined { + return this.transport.protocolVersion; + } + + async start(): Promise { + // The SDK transport exposes callback properties rather than EventTarget listeners. + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onmessage = (message) => this.onmessage?.(message); + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onclose = () => this.emitClose(); + // oxlint-disable-next-line unicorn/prefer-add-event-listener + this.transport.onerror = (error) => { + if (this.closed) { + // SDK reconnect callbacks can finish after close() cleared their old timer. + // Defer a second close so any timer armed later in that callback is cancelled. + setTimeout(() => void this.transport.close(), 0).unref?.(); + return; + } + this.emitError(error); + if (STREAM_RETRY_EXHAUSTED_RE.test(error.message)) { + void this.close(); + } + }; + await this.transport.start(); + } + + async close(): Promise { + if (this.closed) { + return; + } + this.closed = true; + await this.transport.close(); + this.emitClose(); + } + + async send(message: JSONRPCMessage, options?: Parameters[1]): Promise { + await this.transport.send(message, options); + } + + setProtocolVersion(version: string): void { + this.transport.setProtocolVersion(version); + } + + /** Uses a fresh request signal because failed initialization makes the SDK's signal unusable. */ + async terminateSession(): Promise { + const sessionId = this.sessionId; + if (!sessionId || sessionId === this.terminatedSessionId) { + return; + } + const headers = new Headers(this.requestInit?.headers); + headers.set("mcp-session-id", sessionId); + if (this.protocolVersion) { + headers.set("mcp-protocol-version", this.protocolVersion); + } + const response = await this.cleanupFetch(this.url, { + ...this.requestInit, + method: "DELETE", + headers, + signal: AbortSignal.timeout(SESSION_TERMINATION_TIMEOUT_MS), + }); + await response.body?.cancel(); + if (!response.ok && response.status !== 405) { + throw new StreamableHTTPError( + response.status, + `Failed to terminate session: ${response.statusText}`, + ); + } + this.terminatedSessionId = sessionId; + } +} diff --git a/src/agents/mcp-stdio-transport.process.test.ts b/src/agents/mcp-stdio-transport.process.test.ts new file mode 100644 index 000000000000..3b983c99a67a --- /dev/null +++ b/src/agents/mcp-stdio-transport.process.test.ts @@ -0,0 +1,55 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { isPidAlive } from "../shared/pid-alive.js"; +import { OpenClawStdioClientTransport } from "./mcp-stdio-transport.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +describe.skipIf(process.platform === "win32")("OpenClaw stdio process-group ownership", () => { + it( + "kills same-group descendants after the leader exits spontaneously", + { timeout: 10_000 }, + async () => { + const root = tempDirs.make("mcp-stdio-descendant-"); + const serverPath = path.join(root, "leader.mjs"); + const descendantPidPath = path.join(root, "descendant.pid"); + const exitMarkerPath = path.join(root, "exit.marker"); + await fs.writeFile( + serverPath, + `import {spawn} from "node:child_process"; import fs from "node:fs"; const child=spawn(process.execPath,["-e","setInterval(()=>{},1000)"],{stdio:"ignore"}); fs.writeFileSync(${JSON.stringify(descendantPidPath)},String(child.pid)); const timer=setInterval(()=>{if(fs.existsSync(${JSON.stringify(exitMarkerPath)})){clearInterval(timer);process.exit(1)}},10);`, + "utf8", + ); + const transport = new OpenClawStdioClientTransport({ + command: process.execPath, + args: [serverPath], + stderr: "ignore", + }); + const closed = new Promise((resolve) => { + // MCP transports expose callback properties rather than EventTarget listeners. + // oxlint-disable-next-line unicorn/prefer-add-event-listener + transport.onclose = resolve; + }); + let descendantPid = 0; + try { + await transport.start(); + await vi.waitFor(async () => { + descendantPid = Number(await fs.readFile(descendantPidPath, "utf8")); + expect(isPidAlive(descendantPid)).toBe(true); + }); + await fs.writeFile(exitMarkerPath, "exit", "utf8"); + await closed; + await vi.waitFor(() => expect(isPidAlive(descendantPid)).toBe(false)); + + await transport.close(); + expect(isPidAlive(descendantPid)).toBe(false); + } finally { + await transport.forceClose(); + if (descendantPid && isPidAlive(descendantPid)) { + process.kill(descendantPid, "SIGKILL"); + } + } + }, + ); +}); diff --git a/src/agents/mcp-stdio-transport.test.ts b/src/agents/mcp-stdio-transport.test.ts index 7d72dc14931e..aa93535cd2aa 100644 --- a/src/agents/mcp-stdio-transport.test.ts +++ b/src/agents/mcp-stdio-transport.test.ts @@ -186,6 +186,24 @@ describe("OpenClawStdioClientTransport", () => { expect(killProcessTreeMock).not.toHaveBeenCalled(); }); + it("immediately kills the retained process group after the stdio leader exits", async () => { + vi.useFakeTimers(); + const child = new MockChildProcess(); + spawnMock.mockReturnValue(child); + const transport = new OpenClawStdioClientTransport({ command: "npx" }); + const started = transport.start(); + child.emit("spawn"); + await started; + + child.exitCode = 1; + child.emit("close", 1); + const closing = transport.close(); + + expect(signalProcessTreeMock).toHaveBeenCalledWith(4321, "SIGKILL", { detached: true }); + await vi.advanceTimersByTimeAsync(500); + await closing; + }); + it("sends and receives JSON-RPC messages over stdio", async () => { const child = new MockChildProcess(); spawnMock.mockReturnValue(child); diff --git a/src/agents/mcp-stdio-transport.ts b/src/agents/mcp-stdio-transport.ts index 535aa1fe971b..9045b1139082 100644 --- a/src/agents/mcp-stdio-transport.ts +++ b/src/agents/mcp-stdio-transport.ts @@ -41,6 +41,7 @@ export class OpenClawStdioClientTransport implements Transport { private readonly stderrStream: PassThrough | null = null; private process?: ChildProcess; private closingProcess?: ChildProcess; + private ownedProcessGroupId?: number; constructor(private readonly serverParams: OpenClawStdioServerParameters) { if (serverParams.stderr === "pipe" || serverParams.stderr === "overlapped") { @@ -83,6 +84,11 @@ export class OpenClawStdioClientTransport implements Transport { windowsHide: process.platform === "win32", }); this.process = child; + if (process.platform !== "win32" && child.pid) { + // Detached spawn makes the leader PID the durable PGID. Keep it after + // the leader handle exits so descendants remain owned until disposal. + this.ownedProcessGroupId = child.pid; + } child.on("error", (error: Error) => { reject(error); @@ -90,7 +96,16 @@ export class OpenClawStdioClientTransport implements Transport { }); child.on("spawn", () => resolve()); child.on("close", () => { - this.process = undefined; + const exitedUnexpectedly = this.process === child && this.closingProcess !== child; + if (this.process === child) { + this.process = undefined; + } + if (exitedUnexpectedly && child.pid && this.ownedProcessGroupId === child.pid) { + // The leader still owns this PGID at close notification time. Kill any + // descendants now so a retained numeric PGID can never outlive ownership. + signalProcessTree(child.pid, "SIGKILL", { detached: true }); + this.ownedProcessGroupId = undefined; + } this.onclose?.(); }); child.stdin?.on("error", (error: Error) => this.onerror?.(error)); @@ -135,6 +150,7 @@ export class OpenClawStdioClientTransport implements Transport { async close(): Promise { const processToClose = this.process ?? this.closingProcess; + const ownedProcessGroupId = this.ownedProcessGroupId; this.process = undefined; this.closingProcess = processToClose; if (processToClose) { @@ -163,11 +179,15 @@ export class OpenClawStdioClientTransport implements Transport { if (this.closingProcess === processToClose) { this.closingProcess = undefined; } + if (this.ownedProcessGroupId === ownedProcessGroupId) { + this.ownedProcessGroupId = undefined; + } this.readBuffer.clear(); } async forceClose(): Promise { const processToClose = this.process ?? this.closingProcess; + const ownedProcessGroupId = this.ownedProcessGroupId; this.process = undefined; if (processToClose?.pid && processToClose.exitCode === null) { const closePromise = new Promise((resolve) => { @@ -175,10 +195,15 @@ export class OpenClawStdioClientTransport implements Transport { }); signalProcessTree(processToClose.pid, "SIGKILL", { detached: true }); await Promise.race([closePromise, delay(SIGKILL_REAP_TIMEOUT_MS)]); + } else if (ownedProcessGroupId) { + signalProcessTree(ownedProcessGroupId, "SIGKILL", { detached: true }); } if (this.closingProcess === processToClose) { this.closingProcess = undefined; } + if (this.ownedProcessGroupId === ownedProcessGroupId) { + this.ownedProcessGroupId = undefined; + } this.readBuffer.clear(); } diff --git a/src/agents/mcp-tool-metadata.ts b/src/agents/mcp-tool-metadata.ts new file mode 100644 index 000000000000..02fd42d85755 --- /dev/null +++ b/src/agents/mcp-tool-metadata.ts @@ -0,0 +1,59 @@ +import { + ErrorCode, + McpError, + type CallToolResult, + type Tool, +} from "@modelcontextprotocol/sdk/types.js"; +import type { + JsonSchemaValidator, + jsonSchemaValidator, +} from "@modelcontextprotocol/sdk/validation/types.js"; + +type ToolOutputValidator = JsonSchemaValidator; + +export type McpToolCatalogMetadata = { + isRequiredTaskTool(toolName: string): boolean; + validateResult(toolName: string, result: CallToolResult): void; +}; + +/** Owns complete tool metadata after all list pages have been merged. */ +export function createMcpToolCatalogMetadata( + tools: readonly Tool[], + schemaValidator: jsonSchemaValidator, +): McpToolCatalogMetadata { + const outputValidators = new Map(); + const requiredTaskTools = new Set(); + for (const tool of tools) { + if (tool.outputSchema) { + outputValidators.set(tool.name, schemaValidator.getValidator(tool.outputSchema)); + } + if (tool.execution?.taskSupport === "required") { + requiredTaskTools.add(tool.name); + } + } + return { + isRequiredTaskTool: (toolName) => requiredTaskTools.has(toolName), + validateResult(toolName, result) { + const validator = outputValidators.get(toolName); + if (!validator) { + return; + } + if (result.structuredContent === undefined && result.isError !== true) { + throw new McpError( + ErrorCode.InvalidRequest, + `Tool ${toolName} has an output schema but did not return structured content`, + ); + } + if (result.structuredContent === undefined) { + return; + } + const validation = validator(result.structuredContent); + if (!validation.valid) { + throw new McpError( + ErrorCode.InvalidParams, + `Structured content does not match the tool's output schema: ${validation.errorMessage}`, + ); + } + }, + }; +} diff --git a/src/agents/mcp-transport.ts b/src/agents/mcp-transport.ts index ed176808a0b6..9223e35faac7 100644 --- a/src/agents/mcp-transport.ts +++ b/src/agents/mcp-transport.ts @@ -4,11 +4,7 @@ * This module turns normalized MCP server config into stdio, SSE, or * streamable-HTTP SDK transports with OpenClaw auth, redirect, and logging rules. */ -import { - SSEClientTransport, - type SSEClientTransportOptions, -} from "@modelcontextprotocol/sdk/client/sse.js"; -import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import type { SSEClientTransportOptions } from "@modelcontextprotocol/sdk/client/sse.js"; import type { FetchLike, Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import type { OpenClawConfig } from "../config/types.openclaw.js"; @@ -20,6 +16,10 @@ import { withoutMcpAuthorizationHeader, withSameOriginMcpHttpHeaders, } from "./mcp-http-fetch.js"; +import { + OpenClawSSEClientTransport, + OpenClawStreamableHTTPClientTransport, +} from "./mcp-http-transport.js"; import { withMcpOAuthBearer } from "./mcp-oauth-fetch.js"; import { operatorMcpOAuthIdentity, requesterMcpOAuthIdentity } from "./mcp-oauth-identity.js"; import { OpenClawStdioClientTransport } from "./mcp-stdio-transport.js"; @@ -173,7 +173,7 @@ export function resolveMcpTransport( : baseFetch; if (resolved.transportType === "streamable-http") { return { - transport: new StreamableHTTPClientTransport(new URL(resolved.url), { + transport: new OpenClawStreamableHTTPClientTransport(new URL(resolved.url), { requestInit: resolved.auth === "oauth" || !headers ? undefined : { headers }, fetch: httpFetch, }), @@ -187,7 +187,7 @@ export function resolveMcpTransport( const sseHeaders: Record = { ...headers }; const hasHeaders = Object.keys(sseHeaders).length > 0; return { - transport: new SSEClientTransport(new URL(resolved.url), { + transport: new OpenClawSSEClientTransport(new URL(resolved.url), { requestInit: resolved.auth === "oauth" || !hasHeaders ? undefined : { headers: sseHeaders }, fetch: httpFetch, eventSourceInit: { diff --git a/src/agents/node-plugin-tools.test.ts b/src/agents/node-plugin-tools.test.ts index 8a50993cfdb6..f8f169642422 100644 --- a/src/agents/node-plugin-tools.test.ts +++ b/src/agents/node-plugin-tools.test.ts @@ -36,7 +36,7 @@ function replaceNodePluginTools( function createCodeModeHarness(tools: AnyAgentTool[]) { const catalogRef = createToolSearchCatalogRef(); - const config = { tools: { codeMode: true } } as never; + const config = { tools: { codeMode: { enabled: true, timeoutMs: 120_000 } } } as never; const ctx = { config, runtimeConfig: config, @@ -366,7 +366,7 @@ describe("createNodePluginTools", () => { `, ); - expect(details.status).toBe("completed"); + expect(details.status, JSON.stringify(details)).toBe("completed"); expect(details.value).toEqual({ api: expect.stringContaining("query: string;"), called: { @@ -392,6 +392,33 @@ describe("createNodePluginTools", () => { ); }); + it("makes empty MCP application results visible", async () => { + replaceNodePluginTools({ + nodeId: "node-1", + tools: [ + { + pluginId: "node-mcp", + name: "docs_fail", + description: "Fail without content", + parameters: { type: "object", properties: {} }, + command: "mcp.tools.call.v1", + mcp: { server: "docs", tool: "fail" }, + }, + ], + }); + vi.mocked(callGatewayTool).mockResolvedValueOnce({ + payload: { content: [], isError: true }, + }); + + const tool = expectDefined(createNodePluginTools({})[0], "node MCP tool"); + const result = await tool.execute("empty-error", {}); + + expect(result.content).toEqual([ + { type: "text", text: "MCP tool failed without returning content." }, + ]); + expect(isToolResultError(result)).toBe(true); + }); + it("disambiguates gateway-node and node-node MCP server collisions", async () => { for (const [nodeId, displayName, serverName, toolName] of [ ["node-a", "Node A", "tickets", "search_a"], @@ -434,7 +461,7 @@ describe("createNodePluginTools", () => { `, ); - expect(details.status).toBe("completed"); + expect(details.status, JSON.stringify(details)).toBe("completed"); expect(details.value).toEqual({ files: [ "mcp/index.d.ts", diff --git a/src/agents/node-plugin-tools.ts b/src/agents/node-plugin-tools.ts index e29d35eee2bd..eec9aaa8021a 100644 --- a/src/agents/node-plugin-tools.ts +++ b/src/agents/node-plugin-tools.ts @@ -9,7 +9,7 @@ import { import { setPluginToolMeta } from "../plugins/tools.js"; import { sanitizeServerName } from "./agent-bundle-mcp-names.js"; import { compileGlobPatterns, matchesAnyGlobPattern } from "./glob-pattern.js"; -import { projectMcpCallToolResultContent } from "./mcp-content.js"; +import { projectMcpCallToolResult } from "./mcp-content.js"; import type { AgentToolResult } from "./runtime/index.js"; import { DEFAULT_PLUGIN_TOOLS_ALLOWLIST_ENTRY, normalizeToolPolicyName } from "./tool-policy.js"; import { jsonResult } from "./tools/common.js"; @@ -37,14 +37,7 @@ function mapMcpPayloadToAgentToolResult(payload: unknown): AgentToolResult { diff --git a/src/node-host/invoke.mcp.test.ts b/src/node-host/invoke.mcp.test.ts index fe549833043f..c7cf0b8c361c 100644 --- a/src/node-host/invoke.mcp.test.ts +++ b/src/node-host/invoke.mcp.test.ts @@ -36,7 +36,7 @@ function managerWith(callMcpTool: NodeHostMcpManager["callMcpTool"]): NodeHostMc } describe("mcp.tools.call.v1", () => { - it("dispatches validated params and preserves text/image content", async () => { + it("dispatches validated params and preserves raw MCP content for one final projection", async () => { const callMcpTool = vi.fn().mockResolvedValue({ content: [ { type: "text", text: "pong" }, @@ -67,7 +67,12 @@ describe("mcp.tools.call.v1", () => { content: [ { type: "text", text: "pong" }, { type: "image", data: "aW1hZ2U=", mimeType: "image/png" }, - { type: "text", text: "[Report] https://example.com/report" }, + { + type: "resource_link", + uri: "https://example.com/report", + name: "report", + title: "Report", + }, ], structuredContent: { ok: true }, }); diff --git a/src/node-host/invoke.ts b/src/node-host/invoke.ts index 2e7297b68725..ca8da6ed2d02 100644 --- a/src/node-host/invoke.ts +++ b/src/node-host/invoke.ts @@ -5,7 +5,6 @@ import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization"; import { sliceUtf16Safe, truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; -import { mcpContentBlockToAgentContent } from "../agents/mcp-content.js"; import { analyzeArgvCommand, createExecApprovalPolicySnapshot, @@ -35,6 +34,7 @@ import { sanitizeHostExecEnv, sanitizeSystemRunEnvOverrides, } from "../infra/host-env-security.js"; +import { jsonUtf8Bytes } from "../infra/json-utf8-bytes.js"; import { NODE_AGENT_CLI_CLAUDE_RUN_COMMAND, NODE_DEVICE_APPS_COMMAND, @@ -953,17 +953,7 @@ function decodeMcpToolsCallParams(raw?: string | null): McpToolsCallParams { }; } -type McpInvokeContentBlock = - | { type: "text"; text: string } - | { type: "image"; data: string; mimeType: string }; - -function normalizeMcpContentBlock(block: unknown): McpInvokeContentBlock | null { - return isRecord(block) ? mcpContentBlockToAgentContent(block) : null; -} - -function serializedJsonBytes(value: unknown): number { - return Buffer.byteLength(JSON.stringify(value)); -} +type McpInvokeContentBlock = Record; /** Keeps MCP text/image content while bounding text sent through node.invoke. */ function boundMcpToolResultPayload(result: { @@ -975,11 +965,11 @@ function boundMcpToolResultPayload(result: { structuredContent?: Record; isError?: true; } { - const normalizedBlocks = result.content - .map(normalizeMcpContentBlock) - .filter((block): block is McpInvokeContentBlock => block !== null); + const normalizedBlocks = result.content.filter(isRecord); const totalTextBytes = normalizedBlocks.reduce( - (total, block) => total + (block.type === "text" ? Buffer.byteLength(block.text) : 0), + (total, block) => + total + + (block.type === "text" && typeof block.text === "string" ? Buffer.byteLength(block.text) : 0), 0, ); let remainingTextBytes = @@ -989,7 +979,7 @@ function boundMcpToolResultPayload(result: { let markedTruncated = false; const textBoundedContent: McpInvokeContentBlock[] = []; for (const block of normalizedBlocks) { - if (block.type === "image") { + if (block.type !== "text" || typeof block.text !== "string") { textBoundedContent.push(block); continue; } @@ -1017,15 +1007,13 @@ function boundMcpToolResultPayload(result: { } } const payloadMarker = { type: "text" as const, text: MCP_PAYLOAD_TRUNCATION_MARKER }; - const reservedMarkerBytes = serializedJsonBytes(payloadMarker) + 1; + const reservedMarkerBytes = jsonUtf8Bytes(payloadMarker) + 1; const isError = result.isError === true; - let usedBytes = Buffer.byteLength( - JSON.stringify({ content: [], ...(isError ? { isError } : {}) }), - ); + let usedBytes = jsonUtf8Bytes({ content: [], ...(isError ? { isError } : {}) }); let payloadTruncated = false; const content: McpInvokeContentBlock[] = []; for (const block of textBoundedContent) { - const blockBytes = serializedJsonBytes(block) + (content.length > 0 ? 1 : 0); + const blockBytes = jsonUtf8Bytes(block) + (content.length > 0 ? 1 : 0); if (usedBytes + blockBytes + reservedMarkerBytes > MCP_INVOKE_PAYLOAD_MAX_BYTES) { payloadTruncated = true; continue; @@ -1036,7 +1024,7 @@ function boundMcpToolResultPayload(result: { let structuredContent: Record | undefined; if (result.structuredContent) { const structuredBytes = - Buffer.byteLength(',"structuredContent":') + serializedJsonBytes(result.structuredContent); + Buffer.byteLength(',"structuredContent":') + jsonUtf8Bytes(result.structuredContent); if (usedBytes + structuredBytes + reservedMarkerBytes <= MCP_INVOKE_PAYLOAD_MAX_BYTES) { structuredContent = result.structuredContent; } else { diff --git a/src/node-host/mcp.recovery.test.ts b/src/node-host/mcp.recovery.test.ts index 156fc6e6594e..67be86ac374b 100644 --- a/src/node-host/mcp.recovery.test.ts +++ b/src/node-host/mcp.recovery.test.ts @@ -1,11 +1,9 @@ /** Behavior tests for live node-host MCP catalog and connection recovery. */ -import { - StreamableHTTPClientTransport, - StreamableHTTPError, -} from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { StreamableHTTPError } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { CallToolResult, Tool } from "@modelcontextprotocol/sdk/types.js"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { OpenClawStreamableHTTPClientTransport } from "../agents/mcp-http-transport.js"; import { startNodeHostMcpManager } from "./mcp.js"; function tool(name: string, inputSchema: Tool["inputSchema"] = { type: "object" }): Tool { @@ -15,7 +13,7 @@ function tool(name: string, inputSchema: Tool["inputSchema"] = { type: "object" function createClient(params: { tools?: () => Tool[]; connect?: () => Promise; - list?: () => Promise<{ tools: Tool[] }>; + list?: (input?: { cursor?: string }) => Promise<{ tools: Tool[]; nextCursor?: string }>; call?: () => Promise; }) { const call = @@ -24,7 +22,9 @@ function createClient(params: { return { onclose: undefined as (() => void) | undefined, connect: vi.fn(params.connect ?? (async () => {})), - listTools: vi.fn(params.list ?? (async () => ({ tools: params.tools?.() ?? [] }))), + request: vi.fn(async (request: { method: "tools/list"; params?: { cursor?: string } }) => + params.list ? await params.list(request.params) : { tools: params.tools?.() ?? [] }, + ), callTool: vi.fn(call), close: vi.fn(async () => {}), }; @@ -38,7 +38,7 @@ const stdioTransport = { }; function httpTransport(sessionId?: string) { - const transport = new StreamableHTTPClientTransport( + const transport = new OpenClawStreamableHTTPClientTransport( new URL("http://127.0.0.1:1/mcp"), sessionId ? { sessionId } : undefined, ); @@ -57,6 +57,116 @@ afterEach(() => { }); describe("node host MCP live lifecycle", () => { + it("serializes a startup notification refresh after the initial tool list", async () => { + let notifyToolsChanged: (() => void) | undefined; + let activeLists = 0; + let maxActiveLists = 0; + const pending: Array<(value: { tools: Tool[] }) => void> = []; + const client = createClient({ + list: async () => { + activeLists += 1; + maxActiveLists = Math.max(maxActiveLists, activeLists); + try { + return await new Promise<{ tools: Tool[] }>((resolve) => { + pending.push(resolve); + }); + } finally { + activeLists -= 1; + } + }, + }); + const starting = startNodeHostMcpManager( + { docs: { command: "docs" } }, + { + createClient: (_serverName, options) => { + notifyToolsChanged = options.onToolsChanged; + return client; + }, + resolveTransport: () => stdioTransport, + warn: vi.fn(), + }, + ); + + await vi.waitFor(() => expect(pending).toHaveLength(1)); + notifyToolsChanged?.(); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(client.request).toHaveBeenCalledOnce(); + expect(maxActiveLists).toBe(1); + + pending[0]?.({ tools: [tool("stale")] }); + await vi.waitFor(() => expect(pending).toHaveLength(2)); + pending[1]?.({ tools: [tool("fresh")] }); + + const manager = await starting; + expect(maxActiveLists).toBe(1); + expect(manager.descriptors.map((descriptor) => descriptor.mcp?.tool)).toEqual(["fresh"]); + await manager.close(); + }); + + it("owns paginated output metadata and hides required-task tools", async () => { + const outputSchema = { + type: "object" as const, + properties: { count: { type: "number" as const } }, + required: ["count"], + additionalProperties: false, + }; + const client = createClient({ + list: async (input) => + input?.cursor === "page-2" + ? { tools: [tool("ordinary")] } + : { + tools: [ + { ...tool("structured"), outputSchema }, + { + ...tool("task_only"), + execution: { taskSupport: "required" as const }, + }, + ], + nextCursor: "page-2", + }, + call: async () => ({ + content: [{ type: "text", text: "invalid" }], + structuredContent: { count: "not-a-number" }, + }), + }); + const manager = await startNodeHostMcpManager( + { docs: { command: "docs" } }, + { createClient: () => client, resolveTransport: () => stdioTransport, warn: vi.fn() }, + ); + + expect( + manager.descriptors + .map((descriptor) => descriptor.mcp?.tool) + .toSorted((left, right) => (left ?? "").localeCompare(right ?? "")), + ).toEqual(["ordinary", "structured"]); + await expect(manager.callMcpTool({ server: "docs", tool: "structured" })).rejects.toMatchObject( + { code: "MCP_TOOL_ERROR" }, + ); + await manager.close(); + }); + + it("redacts Streamable HTTP response bodies from node diagnostics", async () => { + const client = createClient({ + tools: () => [tool("fail")], + call: async () => { + throw new StreamableHTTPError(500, "Error POSTing to endpoint: bearer=body-secret"); + }, + }); + const manager = await startNodeHostMcpManager( + { docs: { command: "docs" } }, + { createClient: () => client, resolveTransport: () => stdioTransport, warn: vi.fn() }, + ); + + const error = await manager + .callMcpTool({ server: "docs", tool: "fail" }) + .catch((caught: unknown) => caught); + expect(String(error)).not.toContain("body-secret"); + expect(String(error)).toContain("[redacted response body]"); + await manager.close(); + }); + it("refreshes additions, removals, and schemas without replacing descriptor authority", async () => { let listed = [tool("before")]; let notifyToolsChanged: (() => void) | undefined; @@ -93,7 +203,7 @@ describe("node host MCP live lifecycle", () => { expect(onDescriptorsChanged).toHaveBeenCalledOnce(); notifyToolsChanged?.(); - await vi.waitFor(() => expect(client.listTools).toHaveBeenCalledTimes(3)); + await vi.waitFor(() => expect(client.request).toHaveBeenCalledTimes(3)); expect(onDescriptorsChanged).toHaveBeenCalledOnce(); await manager.close(); }); @@ -138,9 +248,9 @@ describe("node host MCP live lifecycle", () => { for (let index = 0; index < 20; index += 1) { notifyToolsChanged?.(); } - expect(client.listTools).toHaveBeenCalledTimes(2); + expect(client.request).toHaveBeenCalledTimes(2); pending.shift()?.({ tools: [tool("middle")] }); - await vi.waitFor(() => expect(client.listTools).toHaveBeenCalledTimes(3)); + await vi.waitFor(() => expect(client.request).toHaveBeenCalledTimes(3)); expect(maxActiveLists).toBe(1); pending.shift()?.({ tools: [tool("final")] }); await vi.waitFor(() => @@ -184,7 +294,7 @@ describe("node host MCP live lifecycle", () => { ); notifyToolsChanged?.(); - await vi.waitFor(() => expect(stale.listTools).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(stale.request).toHaveBeenCalledTimes(2)); stale.onclose?.(); expect(manager.descriptors).toEqual([]); await vi.waitFor(() => diff --git a/src/node-host/mcp.test.ts b/src/node-host/mcp.test.ts index 0c01fd9ac9cc..d39a57d47738 100644 --- a/src/node-host/mcp.test.ts +++ b/src/node-host/mcp.test.ts @@ -32,8 +32,13 @@ function createClient(params?: { throw params.connectError; } }), - listTools: vi.fn(async (input?: { cursor?: string }, options?: { timeout?: number }) => - params?.list ? await params.list(input, options) : { tools: params?.tools ?? [] }, + request: vi.fn( + async ( + request: { method: "tools/list"; params?: { cursor?: string } }, + _schema: unknown, + options?: { timeout?: number }, + ) => + params?.list ? await params.list(request.params, options) : { tools: params?.tools ?? [] }, ), callTool: vi.fn( async ( @@ -377,8 +382,11 @@ describe("node host MCP manager", () => { { createClient: () => client, resolveTransport: () => transport, warn: vi.fn() }, ); - expect(client.listTools).toHaveBeenCalledTimes(2); - expect(client.listTools.mock.calls.map((call) => call[0])).toEqual([undefined, { cursor: "" }]); + expect(client.request).toHaveBeenCalledTimes(2); + expect(client.request.mock.calls.map((call) => call[0].params)).toEqual([ + undefined, + { cursor: "" }, + ]); expect(manager.descriptors.map((descriptor) => descriptor.mcp?.tool)).toEqual([ "first", "second", @@ -402,7 +410,7 @@ describe("node host MCP manager", () => { }, ); - expect(looping.listTools).toHaveBeenCalledTimes(2); + expect(looping.request).toHaveBeenCalledTimes(2); expect(looping.close).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledWith(expect.stringContaining("repeated pagination cursor")); @@ -435,7 +443,7 @@ describe("node host MCP manager", () => { }, ); - expect(endless.listTools).toHaveBeenCalledTimes(128); + expect(endless.request).toHaveBeenCalledTimes(128); expect(endless.close).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledWith(expect.stringContaining("exceeded 128 pages")); @@ -472,8 +480,8 @@ describe("node host MCP manager", () => { await vi.advanceTimersByTimeAsync(50); const manager = await starting; - expect(slow.listTools).toHaveBeenCalledTimes(2); - expect(slow.listTools.mock.calls.map((call) => call[1]?.timeout)).toEqual([50, 50]); + expect(slow.request).toHaveBeenCalledTimes(2); + expect(slow.request.mock.calls.map((call) => call[2]?.timeout)).toEqual([50, 50]); expect(slow.close).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledWith(expect.stringContaining("timed out after 50ms")); @@ -512,7 +520,7 @@ describe("node host MCP manager", () => { }, ); - expect(oversized.listTools).toHaveBeenCalledTimes(2); + expect(oversized.request).toHaveBeenCalledTimes(2); expect(oversized.close).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledOnce(); expect(warn).toHaveBeenCalledWith(expect.stringMatching(/listing exceeded \d+ bytes/u)); @@ -561,7 +569,7 @@ describe("node host MCP manager", () => { warn, }, ); - await vi.waitFor(() => expect(client.listTools).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(client.request).toHaveBeenCalledOnce()); controller.abort(); const manager = await starting; diff --git a/src/node-host/mcp.ts b/src/node-host/mcp.ts index e1a52b3c1b68..5a04cc559f9d 100644 --- a/src/node-host/mcp.ts +++ b/src/node-host/mcp.ts @@ -2,8 +2,13 @@ import { isDeepStrictEqual } from "node:util"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; -import { ErrorCode, type CallToolResult, type Tool } from "@modelcontextprotocol/sdk/types.js"; -import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url"; +import { + ErrorCode, + ListToolsResultSchema, + type CallToolResult, + type ListToolsResult, + type Tool, +} from "@modelcontextprotocol/sdk/types.js"; import { clampPositiveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; @@ -13,15 +18,19 @@ import { disposeMcpClient, isStatefulMcpHttpSessionExpired, } from "../agents/mcp-client-lifecycle.js"; +import { redactMcpDiagnosticError } from "../agents/mcp-error.js"; import { createMcpJsonSchemaValidator } from "../agents/mcp-json-schema-validator.js"; import { sanitizeMcpMetadataText } from "../agents/mcp-metadata.js"; import { collectMcpPaginatedItems } from "../agents/mcp-pagination.js"; import { isMcpToolAllowed } from "../agents/mcp-tool-filter.js"; +import { + createMcpToolCatalogMetadata, + type McpToolCatalogMetadata, +} from "../agents/mcp-tool-metadata.js"; import { resolveMcpRequestTimeoutMs } from "../agents/mcp-transport-config.js"; import { resolveMcpTransport } from "../agents/mcp-transport.js"; import { normalizeConfiguredMcpServers } from "../config/mcp-config-normalize.js"; import type { McpServerConfig } from "../config/types.mcp.js"; -import { toErrorObject } from "../infra/errors.js"; import { NODE_MCP_TOOL_CALL_TIMEOUT_MS, NODE_MCP_TOOLS_CALL_COMMAND, @@ -50,10 +59,11 @@ type NodeHostMcpClient = { transport: Transport, options: { signal: AbortSignal; timeout: number; maxTotalTimeout: number }, ): Promise; - listTools( - params?: { cursor?: string }, + request( + request: { method: "tools/list"; params?: { cursor?: string } }, + resultSchema: typeof ListToolsResultSchema, options?: { timeout?: number; maxTotalTimeout?: number; signal?: AbortSignal }, - ): Promise<{ tools: Tool[]; nextCursor?: string }>; + ): Promise; callTool( params: { name: string; arguments?: Record }, resultSchema?: undefined, @@ -75,6 +85,7 @@ type NodeHostMcpSession = NodeHostMcpTransport & { connected: boolean; toolCallTimeoutMs: number; abortController: AbortController; + toolMetadata?: McpToolCatalogMetadata; }; type NodeHostMcpServerState = { @@ -83,6 +94,7 @@ type NodeHostMcpServerState = { current?: NodeHostMcpSession; listedTools: Tool[]; work: Promise; + catalogWork: Promise; refreshQueued: boolean; retryDelayMs: number; retryTimer?: ReturnType; @@ -129,10 +141,7 @@ function defaultWarn(message: string): void { } function formatMcpError(error: unknown): string { - return truncateUtf16Safe( - redactSensitiveUrlLikeString(toErrorObject(error, "MCP request failed").message), - NODE_MCP_ERROR_MAX_CHARS, - ); + return truncateUtf16Safe(redactMcpDiagnosticError(error), NODE_MCP_ERROR_MAX_CHARS); } function sanitizeDescriptorFragment(raw: string, fallback: string): string { @@ -222,8 +231,8 @@ async function listAllTools( timeoutMs: number, shouldInclude: (toolName: string) => boolean, signal?: AbortSignal, -): Promise { - return await collectMcpPaginatedItems({ +): Promise<{ tools: Tool[]; metadata: McpToolCatalogMetadata }> { + const tools = await collectMcpPaginatedItems({ label: "MCP tool listing", itemLabel: "tools", timeoutMs, @@ -232,11 +241,15 @@ async function listAllTools( maxBytes: NODE_MCP_MAX_CATALOG_BYTES, signal, loadPage: async ({ cursor, requestTimeoutMs, signal: requestSignal }) => { - const page = await client.listTools(cursor === undefined ? undefined : { cursor }, { - timeout: requestTimeoutMs, - maxTotalTimeout: requestTimeoutMs, - signal: requestSignal, - }); + const page = await client.request( + { method: "tools/list", params: cursor === undefined ? undefined : { cursor } }, + ListToolsResultSchema, + { + timeout: requestTimeoutMs, + maxTotalTimeout: requestTimeoutMs, + signal: requestSignal, + }, + ); return { items: page.tools, nextCursor: page.nextCursor, serializedValue: page }; }, mapItem: (tool) => { @@ -247,6 +260,11 @@ async function listAllTools( return { ...tool, name: toolName }; }, }); + const metadata = createMcpToolCatalogMetadata(tools, createMcpJsonSchemaValidator()); + return { + tools: tools.filter((tool) => !metadata.isRequiredTaskTool(tool.name)), + metadata, + }; } function disposeNodeHostMcpSession(session: NodeHostMcpSession): Promise { @@ -290,6 +308,7 @@ export async function startNodeHostMcpManager( config, listedTools: [], work: Promise.resolve(), + catalogWork: Promise.resolve(), refreshQueued: false, retryDelayMs: NODE_MCP_RETRY_INITIAL_MS, }, @@ -338,6 +357,15 @@ export async function startNodeHostMcpManager( state.work = state.work.then(task, task).catch(() => {}); }; + const enqueueCatalogWork = ( + state: NodeHostMcpServerState, + task: () => Promise, + ): Promise => { + const work = state.catalogWork.then(task, task); + state.catalogWork = work.catch(() => {}); + return work; + }; + const scheduleRetry = (state: NodeHostMcpServerState): void => { if ( closed || @@ -408,18 +436,24 @@ export async function startNodeHostMcpManager( } session.connected = true; const listSignal = AbortSignal.any([signal, session.abortController.signal]); - const tools = await listAllTools( - client, - resolved.requestTimeoutMs, - (toolName) => isMcpToolAllowed(state.config.toolFilter, toolName), - listSignal, - ); - if (closed || state.current !== session) { - return; - } - state.listedTools = tools; - state.retryDelayMs = NODE_MCP_RETRY_INITIAL_MS; - rebuildDescriptors(); + await enqueueCatalogWork(state, async () => { + const next = await listAllTools( + client, + createdSession.requestTimeoutMs, + (toolName) => isMcpToolAllowed(state.config.toolFilter, toolName), + listSignal, + ); + if (closed || state.current !== createdSession) { + return; + } + createdSession.toolMetadata = next.metadata; + state.listedTools = next.tools; + state.retryDelayMs = NODE_MCP_RETRY_INITIAL_MS; + rebuildDescriptors(); + }); + // A notification received during startup queues behind the initial list. + // Wait for that refresh so the manager never publishes the older snapshot. + await state.catalogWork; } catch (error) { const lostOwnership = session !== undefined && state.current !== session; if (session && state.current === session) { @@ -456,7 +490,7 @@ export async function startNodeHostMcpManager( return; } try { - const tools = await listAllTools( + const next = await listAllTools( session.client, session.requestTimeoutMs, (toolName) => isMcpToolAllowed(state.config.toolFilter, toolName), @@ -465,7 +499,8 @@ export async function startNodeHostMcpManager( if (closed || state.current !== session) { return; } - state.listedTools = tools; + session.toolMetadata = next.metadata; + state.listedTools = next.tools; rebuildDescriptors(); } catch (error) { if (closed || lifecycleSignal.aborted || state.current !== session) { @@ -493,7 +528,7 @@ export async function startNodeHostMcpManager( state.refreshQueued = true; enqueueWork(state, async () => { state.refreshQueued = false; - await refresh(state, session); + await enqueueCatalogWork(state, () => refresh(state, session)); }); } @@ -547,7 +582,7 @@ export async function startNodeHostMcpManager( const requestedTimeoutMs = clampPositiveTimerTimeoutMs(params.timeoutMs) ?? NODE_MCP_TOOL_CALL_TIMEOUT_MS; try { - return await session.client.callTool( + const result = await session.client.callTool( { name: params.tool, arguments: params.arguments ?? {} }, undefined, { @@ -555,6 +590,8 @@ export async function startNodeHostMcpManager( ...(params.signal ? { signal: params.signal } : {}), }, ); + session.toolMetadata?.validateResult(params.tool, result); + return result; } catch (error) { const sessionExpired = isStatefulMcpHttpSessionExpired(session, error); if (sessionExpired && invalidateCurrent(state, session)) { diff --git a/test/e2e/qa-lab/runtime/gateway-node-mcp.fixture.mjs b/test/e2e/qa-lab/runtime/gateway-node-mcp.fixture.mjs index 456c984f6cb0..ce74bbf70a37 100644 --- a/test/e2e/qa-lab/runtime/gateway-node-mcp.fixture.mjs +++ b/test/e2e/qa-lab/runtime/gateway-node-mcp.fixture.mjs @@ -1,5 +1,7 @@ #!/usr/bin/env node +import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; +import fs from "node:fs"; import { createMcpExpressApp } from "@modelcontextprotocol/sdk/server/express.js"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; @@ -15,7 +17,58 @@ function readOption(name) { return index >= 0 ? process.argv[index + 1] : undefined; } -function createProbeServer(label, catalogState = { rotated: false }) { +function buildProbeResult({ label, marker, generation, expiryCalls }) { + if (marker === "expiry-stats") { + return { + content: [ + { + type: "text", + text: JSON.stringify({ label, marker, pid: process.pid, expiryCalls: expiryCalls ?? 0 }), + }, + ], + }; + } + if (marker === "empty-error") { + return { content: [], isError: true }; + } + if (marker === "rich-result") { + return { + content: [ + { type: "text", text: "mirrored" }, + { type: "resource_link", uri: "memo://report", name: "report", title: "Report" }, + { type: "resource", resource: { uri: "memo://one", text: "memo body" } }, + { type: "audio", data: "AAAA", mimeType: "audio/mpeg" }, + { type: "image", data: "aW1hZ2U=", mimeType: "image/png" }, + ], + structuredContent: { label, marker, rich: true }, + }; + } + const response = { + content: [ + { + type: "text", + text: JSON.stringify({ + label, + marker, + pid: process.pid, + ...(generation === undefined ? {} : { generation }), + }), + }, + ], + }; + return marker.startsWith("error-") + ? { + ...response, + structuredContent: { label, marker, retryable: true }, + isError: true, + } + : response; +} + +function createProbeServer(label, catalogState = { rotated: false }, control = {}) { + catalogState.generations ??= {}; + const generation = (catalogState.generations[label] ?? 0) + 1; + catalogState.generations[label] = generation; const server = new McpServer({ name: `openclaw-mcp-parity-${label}`, version: "1.0.0" }); const initialToolConfig = { description: `MCP parity probe for ${label}`, @@ -25,12 +78,13 @@ function createProbeServer(label, catalogState = { rotated: false }) { description: `Rotated MCP parity probe for ${label}`, inputSchema: { marker: z.string(), revision: z.string().optional() }, }; - const result = (resultLabel, marker) => ({ - content: [ - { type: "text", text: JSON.stringify({ label: resultLabel, marker, pid: process.pid }) }, - ], - }); const runProbe = async ({ marker }) => { + if (marker === "break-notifications") { + control.breakNotifications?.(); + } + if (marker === "crash-generation") { + control.crashGeneration?.(); + } if (marker === "rotate-remove" && !catalogState.rotated) { catalogState.rotated = true; registeredProbe.update({ @@ -38,20 +92,18 @@ function createProbeServer(label, catalogState = { rotated: false }) { paramsSchema: rotatedToolConfig.inputSchema, }); } - const response = result(label, marker); - return marker.startsWith("error-") - ? { - ...response, - structuredContent: { label, marker, retryable: true }, - isError: true, - } - : response; + return buildProbeResult({ + label, + marker, + generation, + expiryCalls: catalogState.expiryCalls, + }); }; const registeredProbe = catalogState.rotated ? server.registerTool("parity_rotated", rotatedToolConfig, runProbe) : server.registerTool("parity_probe", initialToolConfig, runProbe); server.registerTool("parity_hidden", initialToolConfig, async ({ marker }) => - result(`${label}-hidden`, marker), + buildProbeResult({ label: `${label}-hidden`, marker, generation }), ); return server; } @@ -70,11 +122,111 @@ function installSignalShutdown(shutdown) { async function runStdio() { const label = readOption("--label")?.trim() || "stdio"; - const server = createProbeServer(label); + if (process.env.MCP_STRESS_STARTUP_INVERSION === "1") { + await runStressStdio(label); + return; + } + const eventPath = process.env.MCP_STRESS_EVENT_PATH; + let descendant; + if (eventPath) { + descendant = spawn(process.execPath, ["-e", "setInterval(()=>{},1000)"], { stdio: "ignore" }); + fs.appendFileSync( + eventPath, + `${JSON.stringify({ leaderPid: process.pid, descendantPid: descendant.pid })}\n`, + ); + } + const server = createProbeServer(label, undefined, { + crashGeneration: () => setTimeout(() => process.exit(1), 25), + }); installSignalShutdown(async () => await server.close()); await server.connect(new StdioServerTransport()); } +async function runStressStdio(label) { + const eventPath = process.env.MCP_STRESS_EVENT_PATH; + const descendant = spawn(process.execPath, ["-e", "setInterval(()=>{},1000)"], { + stdio: "ignore", + }); + if (eventPath) { + fs.appendFileSync( + eventPath, + `${JSON.stringify({ leaderPid: process.pid, descendantPid: descendant.pid })}\n`, + ); + } + let buffer = ""; + let listCount = 0; + const send = (message) => process.stdout.write(`${JSON.stringify(message)}\n`); + const tools = (name) => [ + { + name, + description: `MCP stress probe for ${label}`, + inputSchema: { + type: "object", + properties: { marker: { type: "string" } }, + required: ["marker"], + }, + }, + ]; + const handle = (message) => { + if (message.method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: message.params?.protocolVersion ?? "2025-06-18", + capabilities: { tools: { listChanged: true } }, + serverInfo: { name: "stress-stdio", version: "1" }, + }, + }); + return; + } + if (message.method === "notifications/initialized") { + return; + } + if (message.method === "tools/list") { + listCount += 1; + if (listCount === 1) { + send({ jsonrpc: "2.0", method: "notifications/tools/list_changed" }); + } + const response = { + jsonrpc: "2.0", + id: message.id, + result: { tools: tools(listCount === 1 ? "parity_stale" : "parity_probe") }, + }; + setTimeout(() => send(response), listCount === 1 ? 125 : 0); + return; + } + if (message.method !== "tools/call") { + return; + } + const marker = message.params?.arguments?.marker ?? ""; + const result = buildProbeResult({ label, marker }); + send({ jsonrpc: "2.0", id: message.id, result }); + if (marker === "crash-generation") { + setTimeout(() => process.exit(1), 25); + } + }; + process.stdin.setEncoding("utf8"); + process.stdin.on("data", (chunk) => { + buffer += chunk; + while (true) { + const newline = buffer.indexOf("\n"); + if (newline < 0) { + return; + } + const line = buffer.slice(0, newline).replace(/\r$/, ""); + buffer = buffer.slice(newline + 1); + if (line.trim()) { + handle(JSON.parse(line)); + } + } + }); + const stop = () => process.exit(0); + process.stdin.on("end", stop); + process.on("SIGTERM", stop); + process.on("SIGINT", stop); +} + async function runHttp() { const labelPrefix = readOption("--label-prefix")?.trim(); if (!labelPrefix) { @@ -83,7 +235,9 @@ async function runHttp() { const app = createMcpExpressApp(); const sessions = new Map(); const records = new Set(); - const catalogState = { rotated: false }; + const catalogState = { rotated: false, expiryCalls: 0 }; + let failStreamableGets = 0; + let terminalSseOnce = false; const route = (handler) => (req, res, next) => void handler(req, res).catch(next); const rpcError = (res, code, message) => res.status(code === -32603 ? 500 : 400).json({ @@ -109,15 +263,16 @@ async function runHttp() { async function handleStreamableRequest(req, res) { try { + if (req.method === "GET" && failStreamableGets > 0) { + failStreamableGets -= 1; + res.status(503).send("notification stream unavailable"); + return; + } const sessionId = req.headers["mcp-session-id"]; let transport; if (typeof sessionId === "string") { - const record = sessions.get(sessionId); - if (!(record?.transport instanceof StreamableHTTPServerTransport)) { - rpcError(res, -32000, "Unknown Streamable HTTP session"); - return; - } if (req.body?.params?.arguments?.marker === "expire-session") { + catalogState.expiryCalls += 1; sessions.delete(sessionId); res.status(404).json({ jsonrpc: "2.0", @@ -126,6 +281,11 @@ async function runHttp() { }); return; } + const record = sessions.get(sessionId); + if (!(record?.transport instanceof StreamableHTTPServerTransport)) { + rpcError(res, -32000, "Unknown Streamable HTTP session"); + return; + } transport = record.transport; } else if (req.method === "POST" && isInitializeRequest(req.body)) { const createdTransport = new StreamableHTTPServerTransport({ @@ -134,7 +294,12 @@ async function runHttp() { sessions.set(createdSessionId, record); }, }); - const server = createProbeServer(`${labelPrefix}-streamable-http`, catalogState); + const server = createProbeServer(`${labelPrefix}-streamable-http`, catalogState, { + breakNotifications: () => { + failStreamableGets = 2; + setTimeout(() => createdTransport.closeStandaloneSSEStream(), 25); + }, + }); const record = track(server, createdTransport); transport = createdTransport; await server.connect(createdTransport); @@ -154,8 +319,18 @@ async function runHttp() { app.all("/mcp", route(handleStreamableRequest)); async function handleSseConnect(_req, res) { + if (terminalSseOnce) { + terminalSseOnce = false; + res.status(204).end(); + return; + } const transport = new SSEServerTransport("/messages", res); - const server = createProbeServer(`${labelPrefix}-sse`, catalogState); + const server = createProbeServer(`${labelPrefix}-sse`, catalogState, { + breakNotifications: () => { + terminalSseOnce = true; + setTimeout(() => void transport.close().catch(() => {}), 25); + }, + }); const record = track(server, transport); sessions.set(transport.sessionId, record); await server.connect(transport); diff --git a/test/e2e/qa-lab/runtime/gateway-node-mcp.stress.e2e.test.ts b/test/e2e/qa-lab/runtime/gateway-node-mcp.stress.e2e.test.ts new file mode 100644 index 000000000000..50556f4aa6c8 --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-node-mcp.stress.e2e.test.ts @@ -0,0 +1,336 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js"; +import type { NodePluginToolDescriptor } from "../../../../packages/gateway-protocol/src/schema/nodes.js"; +import { createSessionMcpRuntime } from "../../../../src/agents/agent-bundle-mcp-runtime.js"; +import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; +import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js"; +import { + NODE_MCP_COMMAND, + TEST_TIMEOUT_MS, + WAIT_OPTIONS, + approvePairing, + createChildEnv, + createMcpServers, + invokeNodeMcp, + invokeNodeMcpPayload, + parseNodeMcpTextRecord, + processIsAlive, + startHttpFixture, + startNodeProcess, + stopChild, + waitForNode, + waitForProcessExit, + type CapturedChild, + type GatewayHandle, + type HttpFixture, +} from "./gateway-node-mcp.test-support.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +function descriptorFor( + descriptors: readonly NodePluginToolDescriptor[], + server: string, +): NodePluginToolDescriptor { + const descriptor = descriptors.find((entry) => entry.mcp?.server === server); + if (!descriptor) { + throw new Error(`missing ${server} MCP descriptor`); + } + return descriptor; +} + +async function readProcessRecords(filePath: string) { + const text = await fs.readFile(filePath, "utf8"); + return text + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as { leaderPid: number; descendantPid: number }); +} + +describe("Gateway/node MCP real-process stress", () => { + it( + "serializes catalogs, recovers terminal streams, fences expiry, and reaps crash generations", + { timeout: TEST_TIMEOUT_MS }, + async () => { + const repoRoot = process.cwd(); + const root = tempDirs.make("openclaw-gateway-node-mcp-stress-"); + const at = (...parts: string[]) => path.join(root, ...parts); + const nodeHome = at("node", "home"); + const nodeStateDir = at("node", "state"); + const nodeConfigPath = at("node", "openclaw.json"); + const nodeTempDir = at("node", "tmp"); + const sessionHome = at("session", "home"); + const sessionTempDir = at("session", "tmp"); + const sessionWorkspace = at("session", "workspace"); + const nodeEvents = at("node-generations.jsonl"); + const sessionEvents = at("session-generations.jsonl"); + const fixturePath = path.join( + repoRoot, + "test/e2e/qa-lab/runtime/gateway-node-mcp.fixture.mjs", + ); + await Promise.all( + [nodeHome, nodeStateDir, nodeTempDir, sessionHome, sessionTempDir, sessionWorkspace].map( + (dir) => fs.mkdir(dir, { recursive: true }), + ), + ); + + let fixture: HttpFixture | undefined; + let gateway: GatewayHandle | undefined; + let node: CapturedChild | undefined; + let sessionRuntime: ReturnType | undefined; + try { + const fixtureEnv = createChildEnv({ home: nodeHome, tempDir: nodeTempDir }); + fixture = await startHttpFixture({ fixturePath, labelPrefix: "node", env: fixtureEnv }); + const nodeStdioEnv = createChildEnv({ + home: nodeHome, + tempDir: nodeTempDir, + extra: { + MCP_STRESS_STARTUP_INVERSION: "1", + MCP_STRESS_EVENT_PATH: nodeEvents, + }, + }); + const nodeServers = createMcpServers({ + placement: "node", + fixture, + stdioEnv: nodeStdioEnv, + fixturePath, + repoRoot, + }); + const nodeConfig: OpenClawConfig = { + gateway: { mode: "local" }, + plugins: { enabled: false }, + nodeHost: { mcp: { servers: nodeServers }, skills: { enabled: false } }, + }; + await fs.writeFile(nodeConfigPath, `${JSON.stringify(nodeConfig, null, 2)}\n`, "utf8"); + + gateway = await startQaGatewayChild({ + repoRoot, + command: { + executablePath: process.execPath, + argsPrefix: ["dist/index.js"], + cwd: repoRoot, + usePackagedPlugins: true, + }, + transportBaseUrl: "http://127.0.0.1", + controlUiEnabled: false, + runtimeEnvPatch: { + OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", + OPENCLAW_SKIP_CHANNELS: "1", + OPENCLAW_SKIP_PROVIDERS: "1", + OPENCLAW_TEST_MINIMAL_GATEWAY: "1", + }, + mutateConfig: (cfg) => { + const { plugins: _plugins, ...withoutPlugins } = cfg; + return { + ...withoutPlugins, + gateway: { + ...cfg.gateway, + nodes: { + ...cfg.gateway?.nodes, + commands: { allow: [NODE_MCP_COMMAND] }, + pairing: { ...cfg.gateway?.nodes?.pairing, autoApproveLocal: false }, + }, + }, + }; + }, + }); + const nodeEnv = createChildEnv({ + home: nodeHome, + tempDir: nodeTempDir, + extra: { + OPENCLAW_HOME: nodeHome, + OPENCLAW_STATE_DIR: nodeStateDir, + OPENCLAW_CONFIG_PATH: nodeConfigPath, + OPENCLAW_GATEWAY_TOKEN: gateway.token, + OPENCLAW_ALLOW_INSECURE_PRIVATE_WS: "1", + OPENCLAW_DISABLE_BUNDLED_PLUGINS: "1", + OPENCLAW_SKIP_CHANNELS: "1", + OPENCLAW_SKIP_PROVIDERS: "1", + }, + }); + const gatewayPort = Number(new URL(gateway.baseUrl).port); + node = startNodeProcess(gatewayPort, nodeEnv); + const nodeId = await approvePairing(gateway, "device"); + await stopChild(node); + node = startNodeProcess(gatewayPort, nodeEnv); + await approvePairing(gateway, "node", nodeId); + + let descriptors = (await waitForNode(gateway, nodeId, 3)).nodePluginTools ?? []; + expect(descriptorFor(descriptors, "stdio").mcp?.tool).toBe("parity_probe"); + expect(descriptors.some((entry) => entry.mcp?.tool === "parity_stale")).toBe(false); + + const sessionStdioEnv = createChildEnv({ + home: sessionHome, + tempDir: sessionTempDir, + extra: { + MCP_STRESS_STARTUP_INVERSION: "1", + MCP_STRESS_EVENT_PATH: sessionEvents, + }, + }); + const sessionServers = createMcpServers({ + placement: "session", + fixture, + stdioEnv: sessionStdioEnv, + fixturePath, + repoRoot, + }); + const sessionStdio = sessionServers.stdio; + if (!sessionStdio) { + throw new Error("session stdio MCP server config was not created"); + } + sessionRuntime = createSessionMcpRuntime({ + sessionId: `stress-${randomUUID()}`, + workspaceDir: sessionWorkspace, + cfg: { + plugins: { enabled: false }, + mcp: { servers: { stdio: sessionStdio } }, + }, + }); + await sessionRuntime.getCatalog(); + + const stdioDescriptor = descriptorFor(descriptors, "stdio"); + const nodeRich = await invokeNodeMcpPayload({ + gateway, + nodeId, + descriptor: stdioDescriptor, + marker: "rich-result", + }); + const sessionRich = await sessionRuntime.callTool("stdio", "parity_probe", { + marker: "rich-result", + }); + expect(nodeRich).toMatchObject({ + payload: { + content: sessionRich.content.map((block) => ({ type: block.type })), + structuredContent: { marker: "rich-result", rich: true }, + }, + }); + expect(sessionRich.structuredContent).toMatchObject({ marker: "rich-result", rich: true }); + + const nodeEmpty = await invokeNodeMcpPayload({ + gateway, + nodeId, + descriptor: stdioDescriptor, + marker: "empty-error", + }); + const sessionEmpty = await sessionRuntime.callTool("stdio", "parity_probe", { + marker: "empty-error", + }); + expect(nodeEmpty).toMatchObject({ payload: { content: [], isError: true } }); + expect(sessionEmpty).toMatchObject({ content: [], isError: true }); + await sessionRuntime.dispose(); + sessionRuntime = undefined; + + for (const server of ["sse", "streamableHttp"] as const) { + descriptors = (await waitForNode(gateway, nodeId, 3)).nodePluginTools ?? []; + const descriptor = descriptorFor(descriptors, server); + const before = parseNodeMcpTextRecord( + await invokeNodeMcpPayload({ + gateway, + nodeId, + descriptor, + marker: `before-${server}`, + }), + ); + await invokeNodeMcp({ + gateway, + nodeId, + descriptor, + marker: "break-notifications", + }); + await vi.waitFor(async () => { + const current = (await waitForNode(gateway!, nodeId, 3)).nodePluginTools ?? []; + const after = parseNodeMcpTextRecord( + await invokeNodeMcpPayload({ + gateway: gateway!, + nodeId, + descriptor: descriptorFor(current, server), + marker: `after-${server}`, + }), + ); + expect(Number(after.generation)).toBeGreaterThan(Number(before.generation)); + }, WAIT_OPTIONS); + } + + for (let generation = 0; generation < 3; generation += 1) { + descriptors = (await waitForNode(gateway, nodeId, 3)).nodePluginTools ?? []; + const result = await invokeNodeMcp({ + gateway, + nodeId, + descriptor: descriptorFor(descriptors, "stdio"), + marker: "crash-generation", + }); + await vi.waitFor(async () => { + const records = await readProcessRecords(nodeEvents); + const record = records.find((entry) => entry.leaderPid === result.pid); + expect(record).toBeDefined(); + expect(processIsAlive(record?.leaderPid ?? 0)).toBe(false); + expect(processIsAlive(record?.descendantPid ?? 0)).toBe(false); + }, WAIT_OPTIONS); + await vi.waitFor(async () => { + const current = (await waitForNode(gateway!, nodeId, 3)).nodePluginTools ?? []; + const recovered = await invokeNodeMcp({ + gateway: gateway!, + nodeId, + descriptor: descriptorFor(current, "stdio"), + marker: `recovered-${generation}`, + }); + expect(recovered.pid).not.toBe(result.pid); + }, WAIT_OPTIONS); + } + + descriptors = (await waitForNode(gateway, nodeId, 3)).nodePluginTools ?? []; + const streamable = descriptorFor(descriptors, "streamableHttp"); + const expired = await Promise.allSettled([ + invokeNodeMcpPayload({ + gateway, + nodeId, + descriptor: streamable, + marker: "expire-session", + }), + invokeNodeMcpPayload({ + gateway, + nodeId, + descriptor: streamable, + marker: "expire-session", + }), + ]); + expect(expired.every((result) => result.status === "rejected")).toBe(true); + await vi.waitFor(async () => { + const current = (await waitForNode(gateway!, nodeId, 3)).nodePluginTools ?? []; + const stats = parseNodeMcpTextRecord( + await invokeNodeMcpPayload({ + gateway: gateway!, + nodeId, + descriptor: descriptorFor(current, "streamableHttp"), + marker: "expiry-stats", + }), + ); + expect(stats.expiryCalls).toBe(2); + }, WAIT_OPTIONS); + } finally { + await Promise.allSettled([ + ...(sessionRuntime ? [sessionRuntime.dispose()] : []), + ...(node ? [stopChild(node)] : []), + ]); + await Promise.allSettled([ + ...(gateway ? [Promise.resolve(gateway.stop())] : []), + ...(fixture ? [stopChild(fixture)] : []), + ]); + for (const eventPath of [nodeEvents, sessionEvents]) { + const records = await readProcessRecords(eventPath).catch(() => []); + for (const record of records) { + for (const pid of [record.leaderPid, record.descendantPid]) { + if (processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + await waitForProcessExit(pid).catch(() => {}); + } + } + } + } + } + }, + ); +}); diff --git a/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.test.ts b/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.test.ts new file mode 100644 index 000000000000..fc9ed5b62439 --- /dev/null +++ b/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.test.ts @@ -0,0 +1,53 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js"; +import { + createChildEnv, + parseNodeMcpTextRecord, + processIsAlive, + startHttpFixture, +} from "./gateway-node-mcp.test-support.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +describe("gateway node MCP fixture ownership", () => { + it.each([ + ["direct", (payload: object) => payload], + ["node.invoke", (payload: object) => ({ ok: true, payload })], + ])("parses %s MCP text records", (_label, wrap) => { + const fact = { label: "node-stdio", marker: "ready", pid: 42 }; + expect( + parseNodeMcpTextRecord(wrap({ content: [{ type: "text", text: JSON.stringify(fact) }] })), + ).toEqual(fact); + }); + + it("kills a spawned fixture when readiness validation fails", async () => { + const root = tempDirs.make("mcp-fixture-startup-failure-"); + const fixturePath = path.join(root, "invalid-fixture.mjs"); + const pidPath = path.join(root, "fixture.pid"); + await fs.writeFile( + fixturePath, + `import fs from "node:fs"; fs.writeFileSync(${JSON.stringify(pidPath)}, String(process.pid)); console.log(JSON.stringify({type:"wrong"})); setInterval(() => {}, 1000);`, + "utf8", + ); + + await expect( + startHttpFixture({ + fixturePath, + labelPrefix: "node", + env: createChildEnv({ home: root, tempDir: os.tmpdir() }), + }), + ).rejects.toThrow("invalid readiness"); + const pid = Number(await fs.readFile(pidPath, "utf8")); + try { + await vi.waitFor(() => expect(processIsAlive(pid)).toBe(false), { timeout: 1_000 }); + } finally { + if (processIsAlive(pid)) { + process.kill(pid, "SIGKILL"); + } + } + expect(processIsAlive(pid)).toBe(false); + }); +}); diff --git a/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.ts b/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.ts index 7987f1fa17f8..8a2809cb1369 100644 --- a/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.ts +++ b/test/e2e/qa-lab/runtime/gateway-node-mcp.test-support.ts @@ -111,41 +111,50 @@ export async function startHttpFixture(params: { stdio: ["ignore", "pipe", "pipe"], }), ); - const pid = captured.child.pid; - if (pid === undefined) { - throw new Error("HTTP MCP fixture did not start"); + let transferred = false; + let lines: ReturnType | undefined; + try { + const pid = captured.child.pid; + if (pid === undefined) { + throw new Error("HTTP MCP fixture did not start"); + } + if (!captured.child.stdout) { + throw new Error("HTTP MCP fixture stdout was not piped"); + } + lines = createInterface({ input: captured.child.stdout }); + const line = await Promise.race([ + new Promise((resolve) => { + lines?.once("line", resolve); + }), + captured.exited.then(() => { + throw new Error(`HTTP MCP fixture exited before readiness:\n${captured.logs()}`); + }), + delay(WAIT_TIMEOUT_MS, undefined, { ref: false }).then(() => { + throw new Error(`HTTP MCP fixture readiness timed out:\n${captured.logs()}`); + }), + ]); + const value: unknown = JSON.parse(line); + if ( + !isRecord(value) || + value.type !== FIXTURE_READY_TYPE || + !isRecord(value.urls) || + typeof value.urls.streamableHttp !== "string" || + typeof value.urls.sse !== "string" + ) { + throw new Error(`HTTP MCP fixture returned invalid readiness: ${line}`); + } + transferred = true; + return { + ...captured, + pid, + urls: { streamableHttp: value.urls.streamableHttp, sse: value.urls.sse }, + }; + } finally { + lines?.close(); + if (!transferred) { + await stopChild(captured); + } } - if (!captured.child.stdout) { - throw new Error("HTTP MCP fixture stdout was not piped"); - } - const lines = createInterface({ input: captured.child.stdout }); - const line = await Promise.race([ - new Promise((resolve) => { - lines.once("line", resolve); - }), - captured.exited.then(() => { - throw new Error(`HTTP MCP fixture exited before readiness:\n${captured.logs()}`); - }), - delay(WAIT_TIMEOUT_MS, undefined, { ref: false }).then(() => { - throw new Error(`HTTP MCP fixture readiness timed out:\n${captured.logs()}`); - }), - ]); - lines.close(); - const value: unknown = JSON.parse(line); - if ( - !isRecord(value) || - value.type !== FIXTURE_READY_TYPE || - !isRecord(value.urls) || - typeof value.urls.streamableHttp !== "string" || - typeof value.urls.sse !== "string" - ) { - throw new Error(`HTTP MCP fixture returned invalid readiness: ${line}`); - } - return { - ...captured, - pid, - urls: { streamableHttp: value.urls.streamableHttp, sse: value.urls.sse }, - }; } export function startNodeProcess(gatewayPort: number, nodeEnv: NodeJS.ProcessEnv): CapturedChild { @@ -262,7 +271,7 @@ export async function waitForNode( return node; } -export function parseProbeResult(value: unknown): ProbeResult { +export function parseNodeMcpTextRecord(value: unknown): Record { const payload = isRecord(value) && isRecord(value.payload) ? value.payload : value; if (!isRecord(payload) || !Array.isArray(payload.content)) { throw new Error(`MCP result omitted content: ${JSON.stringify(value)}`); @@ -274,13 +283,20 @@ export function parseProbeResult(value: unknown): ProbeResult { throw new Error(`MCP result omitted text: ${JSON.stringify(value)}`); } const parsed: unknown = JSON.parse(text.text); + if (!isRecord(parsed)) { + throw new Error(`MCP text was not an object: ${text.text}`); + } + return parsed; +} + +export function parseProbeResult(value: unknown): ProbeResult { + const parsed = parseNodeMcpTextRecord(value); if ( - !isRecord(parsed) || typeof parsed.label !== "string" || typeof parsed.marker !== "string" || typeof parsed.pid !== "number" ) { - throw new Error(`MCP result had invalid probe data: ${text.text}`); + throw new Error(`MCP result had invalid probe data: ${JSON.stringify(parsed)}`); } return { label: parsed.label, marker: parsed.marker, pid: parsed.pid }; } diff --git a/test/scripts/lint-suppressions.test.ts b/test/scripts/lint-suppressions.test.ts index 13e579872056..06302c592901 100644 --- a/test/scripts/lint-suppressions.test.ts +++ b/test/scripts/lint-suppressions.test.ts @@ -200,6 +200,7 @@ describe("production lint suppressions", () => { "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/agents/agent-tools.abort.ts|typescript/prefer-promise-reject-errors|1", + "src/agents/mcp-http-transport.ts|unicorn/prefer-add-event-listener|6", "src/agents/sessions/session-manager-entries.ts|unicorn/prefer-structured-clone|1", "src/audit/audit-event-writer.ts|unicorn/require-post-message-target-origin|2", "src/channels/plugins/channel-runtime-surface.types.ts|typescript/no-unnecessary-type-parameters|1",