fix(mcp): preserve streamed stderr diagnostics (#110652)

Co-authored-by: Peter Steinberger <steipete@gmail.com>

docs/tools/mcp.md

@@ -103,6 +103,8 @@ Run `openclaw mcp doctor <name> --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:<name>:`. 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:

src/agents/mcp-client-lifecycle.ts

@@ -110,33 +110,37 @@ export async function disposeMcpClient(
   session: LifecycleSession,
   timeoutMs = 5_000,
 ): Promise<void> {
-  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?.();
+  }
 }

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

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<typeof setTimeout> | 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<