fix(mcp): harden concurrent writes and diagnostics (#113026)

src/agents/agent-bundle-mcp-runtime.test.ts

@@ -65,6 +65,7 @@ async function writeListToolsMcpServer(params: {
   notifyListChangedAfterFirstList?: boolean;
   exitOnListCall?: number;
   listToolsMethodNotFound?: boolean;
+  listToolsJsonRpcErrorMessage?: string;
   callToolIsError?: boolean;
   callToolJsonRpcError?: boolean;
   resourceListJsonRpcError?: boolean;
@@ -86,6 +87,7 @@ const notifyListChangedOnInitialized = ${params.notifyListChangedOnInitialized =
 const notifyListChangedAfterFirstList = ${params.notifyListChangedAfterFirstList === true};
 const exitOnListCall = ${params.exitOnListCall ?? 0};
 const listToolsMethodNotFound = ${params.listToolsMethodNotFound === true};
+const listToolsJsonRpcErrorMessage = ${JSON.stringify(params.listToolsJsonRpcErrorMessage)};
 const tools = ${JSON.stringify(
       params.tools ?? [
         {
@@ -163,6 +165,15 @@ function handle(message) {
       });
       return;
     }
+    if (listToolsJsonRpcErrorMessage) {
+      log("reject tools/list with configured error");
+      send({
+        jsonrpc: "2.0",
+        id: message.id,
+        error: { code: -32000, message: listToolsJsonRpcErrorMessage },
+      });
+      return;
+    }
     if (hang) {
       log("hang tools/list");
       keepAlive = setInterval(() => {}, 1000);
@@ -990,6 +1001,46 @@ describe("session MCP runtime", () => {
     }
   });
 
+  it("redacts credentials from MCP catalog diagnostics", async () => {
+    const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-diagnostic-redaction-"));
+    const serverPath = path.join(tempDir, "diagnostic-redaction.mjs");
+    const logPath = path.join(tempDir, "server.log");
+    const secret = "test-diagnostic-token";
+    await writeListToolsMcpServer({
+      filePath: serverPath,
+      logPath,
+      listToolsJsonRpcErrorMessage: `Authorization: Bearer ${secret}`,
+    });
+
+    const runtime = await getOrCreateSessionMcpRuntime({
+      sessionId: "session-diagnostic-redaction",
+      sessionKey: "agent:test:session-diagnostic-redaction",
+      workspaceDir: "/workspace",
+      cfg: {
+        mcp: {
+          servers: {
+            diagnostic: {
+              command: process.execPath,
+              args: [serverPath],
+            },
+          },
+        },
+      },
+    });
+
+    try {
+      const catalog = await runtime.getCatalog();
+      const diagnostic = catalog.diagnostics?.[0];
+      expect(diagnostic?.serverName).toBe("diagnostic");
+      expect(diagnostic?.message).toContain("Authorization: Bearer ");
+      expect(diagnostic?.message).toContain("…");
+      expect(diagnostic?.message).not.toContain(secret);
+    } finally {
+      await runtime.dispose();
+      await fs.rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
   it("filters listed MCP tools with per-server include and exclude rules", async () => {
     const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "bundle-mcp-tool-filter-"));
     const serverPath = path.join(tempDir, "tool-filter.mjs");

src/agents/agent-bundle-mcp-runtime.ts

@@ -12,6 +12,7 @@ import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensit
 import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
 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";
@@ -158,8 +159,8 @@ async function connectWithTimeout(
   }
 }
 
-function redactErrorUrls(error: unknown): string {
-  return redactSensitiveUrlLikeString(String(error));
+function redactMcpDiagnosticError(error: unknown): string {
+  return redactToolPayloadText(redactSensitiveUrlLikeString(String(error)));
 }
 
 async function listAllTools(client: Client, timeoutMs: number) {
@@ -627,7 +628,7 @@ export function createSessionMcpRuntime(params: {
                         onChanged: (error) => {
                           if (error) {
                             logWarn(
-                              `bundle-mcp: failed to refresh changed tool list for server "${serverName}": ${redactErrorUrls(error)}`,
+                              `bundle-mcp: failed to refresh changed tool list for server "${serverName}": ${redactMcpDiagnosticError(error)}`,
                             );
                           }
                           catalogInvalidationGeneration += 1;
@@ -752,7 +753,7 @@ export function createSessionMcpRuntime(params: {
                   diagnostics: [] as McpToolCatalogDiagnostic[],
                 };
               } catch (error) {
-                const message = redactErrorUrls(error);
+                const message = redactMcpDiagnosticError(error);
                 if (!disposed) {
                   const action = reusedSession ? "refresh" : "start";
                   logWarn(

src/gateway/server-methods/config-write-flow.test.ts

@@ -0,0 +1,51 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { OpenClawConfig } from "../../config/types.openclaw.js";
+
+const configMocks = vi.hoisted(() => ({
+  replaceConfigFile: vi.fn(),
+  resolveConfigSnapshotHash: vi.fn(),
+}));
+
+vi.mock("../../config/config.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../../config/config.js")>();
+  return {
+    ...actual,
+    replaceConfigFile: configMocks.replaceConfigFile,
+    resolveConfigSnapshotHash: configMocks.resolveConfigSnapshotHash,
+  };
+});
+
+import { commitGatewayConfigWrite } from "./config-write-flow.js";
+
+describe("commitGatewayConfigWrite", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    configMocks.resolveConfigSnapshotHash.mockReturnValue("missing-config-revision");
+    configMocks.replaceConfigFile.mockResolvedValue({
+      nextConfig: {},
+      persistedHash: "persisted-hash",
+    });
+  });
+
+  it("carries a missing file revision into the lock-time compare-and-swap", async () => {
+    const snapshot = {
+      path: "/tmp/openclaw.json",
+      exists: false,
+      raw: null,
+      hash: "missing-config-revision",
+    };
+
+    await commitGatewayConfigWrite({
+      snapshot: snapshot as never,
+      writeOptions: {},
+      nextConfig: {} satisfies OpenClawConfig,
+    });
+
+    expect(configMocks.replaceConfigFile).toHaveBeenCalledWith(
+      expect.objectContaining({
+        baseHash: "missing-config-revision",
+        nextConfig: {},
+      }),
+    );
+  });
+});

src/gateway/server-methods/config-write-flow.ts

@@ -5,6 +5,7 @@ import {
   createConfigIO,
   readConfigFileSnapshotForWrite,
   replaceConfigFile,
+  resolveConfigSnapshotHash,
 } from "../../config/config.js";
 import { extractDeliveryInfo } from "../../config/sessions.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
@@ -244,6 +245,9 @@ export async function commitGatewayConfigWrite(params: {
 }> {
   const result = await replaceConfigFile({
     nextConfig: params.nextConfig,
+    // The early RPC hash check is only advisory until this lock-time CAS. Without
+    // it, concurrent writers can both succeed and overwrite each other's config.
+    baseHash: resolveConfigSnapshotHash(params.snapshot) ?? undefined,
     writeOptions: {
       ...params.writeOptions,
       auditOrigin: "config-rpc",

src/gateway/server-methods/config.ts

@@ -28,6 +28,7 @@ import {
 import { createMergePatch, projectSourceOntoRuntimeShape } from "../../config/io.write-prepare.js";
 import { formatConfigIssueLines } from "../../config/issue-format.js";
 import { applyMergePatch, isMergePatchObjectKeyAllowed } from "../../config/merge-patch.js";
+import { ConfigMutationConflictError } from "../../config/mutation-conflict.js";
 import { normalizeConfigPatchReplacePaths } from "../../config/patch-replace-paths.js";
 import { redactConfigObject, restoreRedactedValues } from "../../config/redact-snapshot.js";
 import { loadGatewayRuntimeConfigSchema } from "../../config/runtime-schema.js";
@@ -660,6 +661,24 @@ function loadSchemaWithPlugins(): ConfigSchemaResponse {
   return response;
 }
 
+async function commitGatewayConfigWriteOrRespond(
+  params: Parameters<typeof commitGatewayConfigWrite>[0] & { respond: RespondFn },
+): Promise<Awaited<ReturnType<typeof commitGatewayConfigWrite>> | null> {
+  try {
+    return await commitGatewayConfigWrite(params);
+  } catch (error) {
+    if (!(error instanceof ConfigMutationConflictError)) {
+      throw error;
+    }
+    params.respond(
+      false,
+      undefined,
+      errorShape(ErrorCodes.INVALID_REQUEST, `${error.message}; re-run config.get and retry`),
+    );
+    return null;
+  }
+}
+
 export const configHandlers: GatewayRequestHandlers = {
   "config.get": async ({ params, respond }) => {
     if (!assertValidParams(params, validateConfigGetParams, "config.get", respond)) {
@@ -728,12 +747,16 @@ export const configHandlers: GatewayRequestHandlers = {
     if (!preparedSecretsSnapshot) {
       return;
     }
-    const writeResult = await commitGatewayConfigWrite({
+    const writeResult = await commitGatewayConfigWriteOrRespond({
       snapshot,
       writeOptions,
       nextConfig: parsed.writeConfig,
       context,
+      respond,
     });
+    if (!writeResult) {
+      return;
+    }
     clearConfigSchemaResponseCache();
     respond(
       true,
@@ -905,13 +928,17 @@ export const configHandlers: GatewayRequestHandlers = {
       nextConfig: validated.config,
       preparedSecretsSnapshot,
     });
-    const writeResult = await commitGatewayConfigWrite({
+    const writeResult = await commitGatewayConfigWriteOrRespond({
       snapshot,
       writeOptions,
       nextConfig: writeConfig,
       context,
       disconnectSharedAuthClients,
+      respond,
     });
+    if (!writeResult) {
+      return;
+    }
     await respondWithConfigRestartWrite({
       requestParams: params,
       kind: "config-patch",
@@ -957,13 +984,17 @@ export const configHandlers: GatewayRequestHandlers = {
       nextConfig: parsed.config,
       preparedSecretsSnapshot,
     });
-    const writeResult = await commitGatewayConfigWrite({
+    const writeResult = await commitGatewayConfigWriteOrRespond({
       snapshot,
       writeOptions,
       nextConfig: parsed.writeConfig,
       context,
       disconnectSharedAuthClients,
+      respond,
     });
+    if (!writeResult) {
+      return;
+    }
     await respondWithConfigRestartWrite({
       requestParams: params,
       kind: "config-apply",

src/gateway/server.config-patch.test.ts

@@ -568,6 +568,42 @@ describe("gateway config methods", () => {
     }
   });
 
+  it("rejects concurrent config.patch writes that share a stale base hash", async () => {
+    const original = await getCurrentConfigObject();
+    const names = Array.from({ length: 8 }, (_, index) => `concurrent-mcp-${index}`);
+
+    try {
+      const results = await Promise.all(
+        names.map((name, index) =>
+          rpcReq<{ ok?: boolean; error?: { message?: string } }>(requireWs(), "config.patch", {
+            raw: JSON.stringify({
+              mcp: {
+                servers: {
+                  [name]: { command: "node", args: [`server-${index}.mjs`] },
+                },
+              },
+            }),
+            baseHash: original.hash,
+          }),
+        ),
+      );
+
+      expect(results.filter((result) => result.ok).length).toBe(1);
+      const failures = results.filter((result) => !result.ok);
+      expect(failures).toHaveLength(names.length - 1);
+      for (const failure of failures) {
+        expect(failure.error?.message).toContain("config changed since last load");
+      }
+
+      const after = await getCurrentConfigObject();
+      const mcp = requireConfigObject(after.config.mcp, "mcp");
+      const servers = requireConfigObject(mcp.servers, "mcp.servers");
+      expect(names.filter((name) => Object.hasOwn(servers, name))).toHaveLength(1);
+    } finally {
+      await restoreConfigFileForTest(original);
+    }
+  });
+
   it("does not reject config.set for unresolved auth-profile refs outside submitted config", async () => {
     const missingEnvVar = `OPENCLAW_MISSING_AUTH_PROFILE_REF_${Date.now()}`;
     await writeUnresolvedAuthProfileTokenRef(missingEnvVar);

src/mcp/plugin-tools-handlers.ts

@@ -2,10 +2,12 @@ import { randomUUID } from "node:crypto";
 import { isRecord } from "@openclaw/normalization-core/record-coerce";
 // Plugin MCP tool handlers route plugin tool calls through the active runtime.
 import {
+  consumeAdjustedParamsForToolCall,
   isToolWrappedWithBeforeToolCallHook,
   rewrapToolWithBeforeToolCallHook,
   wrapToolWithBeforeToolCallHook,
 } from "../agents/agent-tools.before-tool-call.js";
+import { BEFORE_TOOL_CALL_HOOK_CONTEXT } from "../agents/before-tool-call-metadata.js";
 import type { AnyAgentTool } from "../agents/tools/common.js";
 import { formatErrorMessage } from "../infra/errors.js";
 import { coerceChatContentText } from "../shared/chat-content.js";
@@ -52,6 +54,11 @@ function resolveJsonSchemaForTool(tool: AnyAgentTool): Record<string, unknown> {
   return { type: "object", properties: {} };
 }
 
+function resolveBeforeToolCallRunId(tool: AnyAgentTool): string | undefined {
+  const context = (tool as unknown as Record<symbol, unknown>)[BEFORE_TOOL_CALL_HOOK_CONTEXT];
+  return isRecord(context) && typeof context.runId === "string" ? context.runId : undefined;
+}
+
 export function createPluginToolsMcpHandlers(tools: AnyAgentTool[]) {
   const wrappedTools = tools.map((tool) => {
     if (isToolWrappedWithBeforeToolCallHook(tool)) {
@@ -61,9 +68,9 @@ export function createPluginToolsMcpHandlers(tools: AnyAgentTool[]) {
     // as the agent and HTTP tool execution paths.
     return wrapToolWithBeforeToolCallHook(tool, undefined, { approvalMode: "report" });
   });
-  const toolMap = new Map<string, AnyAgentTool>();
+  const toolMap = new Map<string, { tool: AnyAgentTool; runId: string | undefined }>();
   for (const tool of wrappedTools) {
-    toolMap.set(tool.name, tool);
+    toolMap.set(tool.name, { tool, runId: resolveBeforeToolCallRunId(tool) });
   }
 
   return {
@@ -75,15 +82,16 @@ export function createPluginToolsMcpHandlers(tools: AnyAgentTool[]) {
       })),
     }),
     callTool: async (params: CallPluginToolParams, signal?: AbortSignal) => {
-      const tool = toolMap.get(params.name);
-      if (!tool) {
+      const entry = toolMap.get(params.name);
+      if (!entry) {
         return {
           content: [{ type: "text", text: `Unknown tool: ${params.name}` }],
           isError: true,
         };
       }
+      const toolCallId = `mcp-${randomUUID()}`;
       try {
-        const result = await tool.execute(`mcp-${randomUUID()}`, params.arguments ?? {}, signal);
+        const result = await entry.tool.execute(toolCallId, params.arguments ?? {}, signal);
         const rawContent =
           result && typeof result === "object" && "content" in result
             ? (result as { content?: unknown }).content
@@ -98,6 +106,10 @@ export function createPluginToolsMcpHandlers(tools: AnyAgentTool[]) {
           content: [{ type: "text", text: `Tool error: ${formatErrorMessage(err)}` }],
           isError: true,
         };
+      } finally {
+        // Direct MCP calls have no agent event consumer, so release the cloned
+        // hook arguments as soon as the tool reaches a terminal state.
+        consumeAdjustedParamsForToolCall(toolCallId, entry.runId);
       }
     },
   };

src/mcp/plugin-tools-serve.test.ts

@@ -4,6 +4,7 @@ import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
 import { CallToolResultSchema } from "@modelcontextprotocol/sdk/types.js";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import {
+  consumeAdjustedParamsForToolCall,
   type HookContext,
   wrapToolWithBeforeToolCallHook,
 } from "../agents/agent-tools.before-tool-call.js";
@@ -256,6 +257,7 @@ describe("plugin tools MCP server", () => {
     expect(new Set(toolCallIds).size).toBe(toolCallIds.length);
     for (const toolCallId of toolCallIds) {
       expect(consumeTrackedToolExecutionStarted(toolCallId)).toBeUndefined();
+      expect(consumeAdjustedParamsForToolCall(toolCallId)).toBeUndefined();
     }
   });
 
@@ -392,6 +394,39 @@ describe("plugin tools MCP server", () => {
     expect(failed.content).toEqual([{ type: "text", text: "Tool error: boom" }]);
   });
 
+  it("releases run-scoped adjusted arguments after a pre-wrapped direct MCP call", async () => {
+    const runId = "run-direct-mcp";
+    const execute = vi.fn().mockResolvedValue({ content: "Stored." });
+    initializeGlobalHookRunner(
+      createMockPluginRegistry([
+        {
+          hookName: "before_tool_call",
+          handler: async () => ({ params: { text: "adjusted" } }),
+        },
+      ]),
+    );
+    const tool = wrapToolWithBeforeToolCallHook(
+      {
+        name: "memory_store",
+        description: "Store memory",
+        parameters: { type: "object", properties: {} },
+        execute,
+      } as unknown as AnyAgentTool,
+      { runId, sessionKey: "session-direct-mcp" },
+    );
+
+    const handlers = createPluginToolsMcpHandlers([tool]);
+    await handlers.callTool({
+      name: "memory_store",
+      arguments: { text: "original" },
+    });
+
+    const executeCall = requireFirstMockCall(execute.mock.calls, "plugin tool execute");
+    const toolCallId = String(executeCall[0]);
+    expect(executeCall[1]).toEqual({ text: "adjusted" });
+    expect(consumeAdjustedParamsForToolCall(toolCallId, runId)).toBeUndefined();
+  });
+
   it("reports approval requirements without opening plugin approvals on the MCP bridge", async () => {
     let hookCalls = 0;
     const onResolution = vi.fn();

ui/src/components/mcp-servers-card.test.ts

@@ -306,6 +306,26 @@ describe("openclaw-mcp-servers-card", () => {
       transport: "stdio" as const,
       target: 'npx some-mcp-server "unfinished',
     },
+    {
+      label: "HTTP URL without a host",
+      transport: "streamable-http" as const,
+      target: "http://",
+    },
+    {
+      label: "HTTP URL with whitespace in the host",
+      transport: "streamable-http" as const,
+      target: "https://exa mple.com/mcp",
+    },
+    {
+      label: "SSE URL with malformed IPv6",
+      transport: "sse" as const,
+      target: "https://[::1/mcp",
+    },
+    {
+      label: "HTTP URL with a nonnumeric port",
+      transport: "streamable-http" as const,
+      target: "https://example.com:bad",
+    },
   ])("rejects a mismatched or malformed $label", async ({ transport, target }) => {
     const { card, harness } = await mountCard();
 

ui/src/lib/config/mcp-servers.ts

@@ -101,7 +101,12 @@ export function parseMcpTarget(
   transport: McpServerTransport,
 ): Record<string, unknown> | null {
   if (transport !== "stdio") {
-    return /^https?:\/\//i.test(target) ? { url: target, transport } : null;
+    try {
+      const protocol = new URL(target).protocol;
+      return protocol === "http:" || protocol === "https:" ? { url: target, transport } : null;
+    } catch {
+      return null;
+    }
   }
   if (/^https?:\/\//i.test(target)) {
     return null;