diff --git a/docs/tools/mcp.md b/docs/tools/mcp.md index 09102fb37f9a..68af1c913f39 100644 --- a/docs/tools/mcp.md +++ b/docs/tools/mcp.md @@ -103,6 +103,8 @@ Run `openclaw mcp doctor --probe`. Doctor validates the saved definition Confirm the `command` resolves in the Gateway process environment and that `cwd` exists. Arguments belong in `args`, and an explicit `transport: "stdio"` requires a non-empty command. +For servers launched by OpenClaw's built-in MCP client, debug logs prefix stderr diagnostics with `bundle-mcp::`. Unicode characters survive split writes, and shutdown diagnostics are retained. Output without a newline is briefly buffered for up to 250 ms before being logged as progress fragments; this does not wait for the server to stop writing. A diagnostic exceeding the 8 KiB buffer retains its Unicode-safe tail with a `[stderr line truncated]` marker. + ### An HTTP server needs authorization Set `auth: "oauth"` plus any required `oauth` metadata, then: diff --git a/src/agents/mcp-client-lifecycle.ts b/src/agents/mcp-client-lifecycle.ts index 18fb1b46b67c..0ee3b5b0b927 100644 --- a/src/agents/mcp-client-lifecycle.ts +++ b/src/agents/mcp-client-lifecycle.ts @@ -110,33 +110,37 @@ export async function disposeMcpClient( session: LifecycleSession, timeoutMs = 5_000, ): Promise { - session.detachStderr?.(); - const closed = await settleWithin( - (async () => { - if (session.transportType === "streamable-http") { - await ignoreCloseFailure(() => session.transport.terminateSession?.()); - } - await ignoreCloseFailure(() => session.transport.close()); - await ignoreCloseFailure(() => session.client.close()); - })(), - timeoutMs, - ); - if (closed) { - return; - } + try { + const closed = await settleWithin( + (async () => { + if (session.transportType === "streamable-http") { + await ignoreCloseFailure(() => session.transport.terminateSession?.()); + } + await ignoreCloseFailure(() => session.transport.close()); + await ignoreCloseFailure(() => session.client.close()); + })(), + timeoutMs, + ); + if (closed) { + return; + } - // Closing an HTTP transport aborts a hung DELETE. Stdio owns a process - // group, so force it dead before disposal can report completion. - const { transport } = session; - const closeTransport = - session.transportType === "stdio" && transport instanceof OpenClawStdioClientTransport - ? () => transport.forceClose() - : () => transport.close(); - await settleWithin( - Promise.all([ - ignoreCloseFailure(closeTransport), - ignoreCloseFailure(() => session.client.close()), - ]), - timeoutMs, - ); + // Closing an HTTP transport aborts a hung DELETE. Stdio owns a process + // group, so force it dead before disposal can report completion. + const { transport } = session; + const closeTransport = + session.transportType === "stdio" && transport instanceof OpenClawStdioClientTransport + ? () => transport.forceClose() + : () => transport.close(); + await settleWithin( + Promise.all([ + ignoreCloseFailure(closeTransport), + ignoreCloseFailure(() => session.client.close()), + ]), + timeoutMs, + ); + } finally { + // Shutdown itself may emit the last diagnostic; detach only after it settles. + session.detachStderr?.(); + } } diff --git a/src/agents/mcp-stderr.test.ts b/src/agents/mcp-stderr.test.ts new file mode 100644 index 000000000000..8669ab7935fd --- /dev/null +++ b/src/agents/mcp-stderr.test.ts @@ -0,0 +1,169 @@ +import { once } from "node:events"; +import process from "node:process"; +import { PassThrough } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { disposeMcpClient } from "./mcp-client-lifecycle.js"; +import { OpenClawStdioClientTransport } from "./mcp-stdio-transport.js"; +import { resolveMcpTransport } from "./mcp-transport.js"; + +const { logDebug } = vi.hoisted(() => ({ logDebug: vi.fn() })); +vi.mock("../logger.js", () => ({ logDebug })); + +function createStderrProbe(args?: string[]) { + const resolved = resolveMcpTransport("probe", { command: process.execPath, args }); + if (!resolved || !(resolved.transport instanceof OpenClawStdioClientTransport)) { + throw new Error("Expected a stdio transport"); + } + const stderr = resolved.transport.stderr; + if (!(stderr instanceof PassThrough)) { + throw new Error("Expected a writable stderr test pipe"); + } + return { ...resolved, transport: resolved.transport, stderr }; +} + +describe("MCP stderr diagnostics", () => { + afterEach(() => { + vi.useRealTimers(); + logDebug.mockClear(); + }); + + it("joins a diagnostic split at every UTF-8 byte boundary", () => { + const probe = createStderrProbe(); + try { + for (const byte of Buffer.from("alpha 你好 😀 omega\r\n")) { + probe.stderr.write(Buffer.from([byte])); + } + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: alpha 你好 😀 omega"]]); + } finally { + probe.detachStderr?.(); + } + }); + + it("reports sustained newline-free progress without waiting for an idle gap", () => { + vi.useFakeTimers(); + const probe = createStderrProbe(); + try { + for (let index = 0; index < 3; index++) { + probe.stderr.write("loading "); + vi.advanceTimersByTime(index === 2 ? 50 : 100); + } + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: loading loading loading"]]); + } finally { + probe.detachStderr?.(); + } + expect(vi.getTimerCount()).toBe(0); + }); + + it("keeps incomplete UTF-8 bytes across a progress flush", () => { + vi.useFakeTimers(); + const probe = createStderrProbe(); + try { + const bytes = Buffer.from("loading 你"); + probe.stderr.write(bytes.subarray(0, -1)); + vi.advanceTimersByTime(250); + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: loading"]]); + probe.stderr.write(bytes.subarray(-1)); + probe.stderr.write("\n"); + expect(logDebug.mock.calls).toEqual([ + ["bundle-mcp:probe: loading"], + ["bundle-mcp:probe: 你"], + ]); + } finally { + probe.detachStderr?.(); + } + expect(vi.getTimerCount()).toBe(0); + }); + + it("emits CR progress frames without duplicating a split CRLF", () => { + const probe = createStderrProbe(); + try { + probe.stderr.write("start\rmiddle\r"); + probe.stderr.write("\nlast"); + probe.detachStderr?.(); + expect(logDebug.mock.calls).toEqual([ + ["bundle-mcp:probe: start"], + ["bundle-mcp:probe: middle"], + ["bundle-mcp:probe: last"], + ]); + } finally { + probe.detachStderr?.(); + } + }); + + it.each(["newline", "detach"])("marks a UTF-8-safe bounded tail on %s", (ending) => { + const probe = createStderrProbe(); + try { + probe.stderr.write(`xx😀${"y".repeat(4000)}`); + probe.stderr.write("y".repeat(4189)); + if (ending === "newline") { + probe.stderr.write("\n"); + } + probe.detachStderr?.(); + expect(logDebug.mock.calls).toEqual([ + [`bundle-mcp:probe: [stderr line truncated] ${"y".repeat(8189)}`], + ]); + } finally { + probe.detachStderr?.(); + } + }); + + it("flushes natural EOF once and releases its listeners", async () => { + const probe = createStderrProbe(); + try { + const ended = once(probe.stderr, "end"); + probe.stderr.end("fatal tail"); + await ended; + probe.detachStderr?.(); + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: fatal tail"]]); + for (const event of ["data", "end", "close"]) { + expect(probe.stderr.listenerCount(event)).toBe(0); + } + } finally { + probe.detachStderr?.(); + } + }); + + it("keeps diagnostics attached through forced disposal", async () => { + vi.useFakeTimers(); + const probe = createStderrProbe(); + const close = vi + .spyOn(probe.transport, "close") + .mockImplementation(() => new Promise(() => {})); + const forceClose = vi.spyOn(probe.transport, "forceClose").mockImplementation(async () => { + probe.stderr.write("forced shutdown tail"); + }); + try { + const disposing = disposeMcpClient({ ...probe, client: { close: async () => {} } }, 50); + await vi.advanceTimersByTimeAsync(50); + await disposing; + expect(forceClose).toHaveBeenCalledOnce(); + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: forced shutdown tail"]]); + expect(vi.getTimerCount()).toBe(0); + } finally { + close.mockRestore(); + forceClose.mockRestore(); + probe.detachStderr?.(); + } + }); + + it("retains a real child process's unterminated shutdown diagnostic", async () => { + const probe = createStderrProbe([ + "-e", + `process.stdin.resume(); + process.stdin.on("end", () => { + const message = Buffer.from("shutdown 你好"); + process.stderr.write(message.subarray(0, 10)); + setImmediate(() => process.stderr.end(message.subarray(10))); + });`, + ]); + try { + await probe.transport.start(); + await disposeMcpClient({ ...probe, client: { close: async () => {} } }); + expect(logDebug.mock.calls).toEqual([["bundle-mcp:probe: shutdown 你好"]]); + expect(probe.transport.pid).toBeNull(); + } finally { + await probe.transport.forceClose(); + probe.detachStderr?.(); + } + }); +}); diff --git a/src/agents/mcp-transport.ts b/src/agents/mcp-transport.ts index 9223e35faac7..59c9792eb386 100644 --- a/src/agents/mcp-transport.ts +++ b/src/agents/mcp-transport.ts @@ -4,11 +4,12 @@ * This module turns normalized MCP server config into stdio, SSE, or * streamable-HTTP SDK transports with OpenClaw auth, redirect, and logging rules. */ +import { StringDecoder } from "node:string_decoder"; 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"; import { logDebug } from "../logger.js"; +import { truncateUtf8Suffix } from "../utils/utf8-truncate.js"; import type { SessionMcpRequesterScope } from "./agent-bundle-mcp-types.js"; import { resolveMcpAuthProfileId, withMcpAuthProfileBearer } from "./mcp-auth-profile.js"; import { @@ -35,32 +36,63 @@ type ResolvedMcpTransport = { detachStderr?: () => void; }; +const MAX_MCP_STDERR_LINE_BYTES = 8 * 1024; + function attachStderrLogging(serverName: string, transport: OpenClawStdioClientTransport) { const stderr = transport.stderr; - if (!stderr || typeof stderr.on !== "function") { + if (!stderr) { return undefined; } + const decoder = new StringDecoder("utf8"); + let pending = ""; + let truncated = false; + let progressTimer: ReturnType | undefined; + const emit = (text: string) => { + const tail = truncateUtf8Suffix(text, MAX_MCP_STDERR_LINE_BYTES); + const message = `${truncated || tail !== text ? "[stderr line truncated] " : ""}${tail}`.trim(); + truncated = false; + if (message) { + logDebug(`bundle-mcp:${serverName}: ${message}`); + } + }; + const flushProgress = () => { + progressTimer = undefined; + const text = pending; + pending = ""; + emit(text); + }; const onData = (chunk: Buffer | string) => { - const message = - normalizeOptionalString(typeof chunk === "string" ? chunk : String(chunk)) ?? ""; - if (!message) { - return; + const decoded = decoder.write(typeof chunk === "string" ? Buffer.from(chunk) : chunk); + const lines = (pending + decoded).split(/[\r\n]/); + pending = lines.pop() ?? ""; + for (const line of lines) { + emit(line); } - for (const line of message.split(/\r?\n/)) { - const trimmed = line.trim(); - if (trimmed) { - logDebug(`bundle-mcp:${serverName}: ${trimmed}`); - } + const tail = truncateUtf8Suffix(pending, MAX_MCP_STDERR_LINE_BYTES); + truncated ||= tail !== pending; + pending = tail; + // No-newline progress must stay visible even under continuous writes. Flush + // complete characters within 250ms; only finalization ends the UTF-8 decoder. + if (pending && !progressTimer) { + progressTimer = setTimeout(flushProgress, 250); + progressTimer.unref(); + } else if (!pending) { + clearTimeout(progressTimer); + progressTimer = undefined; } }; + const finalize = () => { + stderr.off("data", onData); + stderr.off("end", finalize); + stderr.off("close", finalize); + clearTimeout(progressTimer); + pending += decoder.end(); + flushProgress(); + }; stderr.on("data", onData); - return () => { - if (typeof stderr.off === "function") { - stderr.off("data", onData); - } else if (typeof stderr.removeListener === "function") { - stderr.removeListener("data", onData); - } - }; + stderr.on("end", finalize); + stderr.on("close", finalize); + return finalize; } type SseEventSourceFetch = NonNullable<