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

@@ -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

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<unknown> {
-  const content = projectMcpCallToolResultContent(params.result);
-  const normalizedContent: AgentToolResult<unknown>["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<unknown>["content"]);
-  const details: Record<string, unknown> = {
+  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: {

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<string, unknown>;
   }>;
   capabilities?: Record<string, unknown>;
@@ -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"]);
 

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<void>;
   detachStderr?: () => void;
+  toolMetadata?: McpToolCatalogMetadata;
 };
 
-type ListedTool = Awaited<ReturnType<Client["listTools"]>>["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<ListedTool[]> {
-  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) {

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",
         },

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<unknown>["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<string, unknown> = {},
+): AgentToolResult<unknown> {
+  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" } : {}),
+    },
+  };
+}

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));
+}

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> | 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<void>((resolve) => {
+      setTimeout(resolve, 80);
+    });
+    expect(getCount).toBe(1);
+  });
+});

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<void>;
+  abstract close(): Promise<void>;
+  abstract send(message: JSONRPCMessage): Promise<void>;
+}
+
+/** 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<void> {
+    // 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<void> {
+    if (this.closed) {
+      return;
+    }
+    this.closed = true;
+    await this.transport.close();
+    this.emitClose();
+  }
+
+  async send(message: JSONRPCMessage): Promise<void> {
+    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<void> {
+    // 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<void> {
+    if (this.closed) {
+      return;
+    }
+    this.closed = true;
+    await this.transport.close();
+    this.emitClose();
+  }
+
+  async send(message: JSONRPCMessage, options?: Parameters<Transport["send"]>[1]): Promise<void> {
+    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<void> {
+    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;
+  }
+}

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<void>((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");
+        }
+      }
+    },
+  );
+});

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);

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<void> {
     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<void> {
     const processToClose = this.process ?? this.closingProcess;
+    const ownedProcessGroupId = this.ownedProcessGroupId;
     this.process = undefined;
     if (processToClose?.pid && processToClose.exitCode === null) {
       const closePromise = new Promise<void>((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();
   }
 

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<unknown>;
+
+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<string, ToolOutputValidator>();
+  const requiredTaskTools = new Set<string>();
+  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}`,
+        );
+      }
+    },
+  };
+}

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<string, string> = { ...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: {

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",

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<unkno
   if (!isRecord(payload)) {
     return jsonResult(payload);
   }
-  const content = projectMcpCallToolResultContent({
-    content: payload.content,
-    structuredContent: payload.structuredContent,
-  });
-  return {
-    content,
-    details: payload.isError === true ? { ...payload, status: "error" } : payload,
-  };
+  return projectMcpCallToolResult(payload, payload);
 }
 
 function normalizePolicyNames(values: readonly string[] | undefined): Set<string> {

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<NodeHostMcpManager["callMcpTool"]>().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 },
     });

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<string, unknown>;
 
 /** Keeps MCP text/image content while bounding text sent through node.invoke. */
 function boundMcpToolResultPayload(result: {
@@ -975,11 +965,11 @@ function boundMcpToolResultPayload(result: {
   structuredContent?: Record<string, unknown>;
   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<number>(
-    (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<string, unknown> | 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 {

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<void>;
-  list?: () => Promise<{ tools: Tool[] }>;
+  list?: (input?: { cursor?: string }) => Promise<{ tools: Tool[]; nextCursor?: string }>;
   call?: () => Promise<CallToolResult>;
 }) {
   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<void>((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(() =>

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;

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<void>;
-  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<ListToolsResult>;
   callTool(
     params: { name: string; arguments?: Record<string, unknown> },
     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<void>;
+  catalogWork: Promise<void>;
   refreshQueued: boolean;
   retryDelayMs: number;
   retryTimer?: ReturnType<typeof setTimeout>;
@@ -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<Tool[]> {
-  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<void> {
@@ -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<void>,
+  ): Promise<void> => {
+    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)) {

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);

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<typeof createSessionMcpRuntime> | 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(() => {});
+              }
+            }
+          }
+        }
+      }
+    },
+  );
+});

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);
+  });
+});

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<typeof createInterface> | 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<string>((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<string>((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<string, unknown> {
   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 };
 }

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",