fix(agents): give bash stdout and stderr independent decode lanes (#112325)

* fix(agents): give bash stdout and stderr independent decode lanes

stdout and stderr are independent pipes, but the local bash execution path
fed both into one onData callback sharing a single TextDecoder and one
streaming ANSI/OSC sanitizer. A multibyte UTF-8 character split across a
stdout read boundary was corrupted when stderr wrote between its bytes, and
an unterminated OSC on stdout swallowed subsequent stderr output. This
contradicts the documented invariant in shell-utils.ts ('Keep one ANSI
parser per process stream so control sequences can span callbacks').

Tag onData with an optional stream identifier and give each lane its own
TextDecoder and text-transform state; finish() flushes every lane. Untagged
callers keep the single shared lane for backward compatibility.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(agents): harden bash stream isolation

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/agents/sessions/bash-executor.test.ts

@@ -5,6 +5,21 @@ import { executeBashWithOperations } from "./bash-executor.js";
 import type { BashOperations } from "./tools/bash-operations.js";
 import { DEFAULT_MAX_BYTES } from "./tools/truncate.js";
 
+type OutputChunk = readonly [data: Buffer, stream?: "stdout" | "stderr"];
+
+const ESC = String.fromCharCode(27);
+
+function operationsForChunks(chunks: readonly OutputChunk[]): BashOperations {
+  return {
+    exec: async (_command, _cwd, options) => {
+      for (const [data, stream] of chunks) {
+        options.onData(data, stream);
+      }
+      return { exitCode: 0 };
+    },
+  };
+}
+
 describe("executeBashWithOperations", () => {
   it("stores truncated full output in an owner-only temp file", async () => {
     const sanitizedOutput = "secret output\n".repeat(9000);
@@ -76,6 +91,66 @@ describe("executeBashWithOperations", () => {
     expect(result.output).toBe("ABCDEFGH");
   });
 
+  it("preserves delivery order when tagged UTF-8 chunks interleave", async () => {
+    const chunks: string[] = [];
+    const operations = operationsForChunks([
+      [Buffer.from([0xe6, 0x97]), "stdout"], // leading bytes of 日
+      [Buffer.from("E"), "stderr"],
+      [Buffer.from([0xa5]), "stdout"],
+    ]);
+
+    const result = await executeBashWithOperations("printf output", "/tmp", operations, {
+      onChunk: (chunk) => chunks.push(chunk),
+    });
+
+    expect(chunks.join("")).toBe("E日");
+    expect(result.output).toBe("E日");
+  });
+
+  it.each([
+    {
+      name: "OSC on stdout",
+      pending: [Buffer.from(`${ESC}]0;unterminated`), "stdout"] as const,
+      visible: [Buffer.from("stderr visible\n"), "stderr"] as const,
+    },
+    {
+      name: "CSI on stderr",
+      pending: [Buffer.from(`${ESC}[31`), "stderr"] as const,
+      visible: [Buffer.from("stdout visible\n"), "stdout"] as const,
+    },
+  ])("does not let unterminated $name consume the other stream", async ({ pending, visible }) => {
+    const result = await executeBashWithOperations(
+      "printf output",
+      "/tmp",
+      operationsForChunks([pending, visible]),
+    );
+
+    expect(result.output).toBe(visible[0].toString("utf8"));
+  });
+
+  it("keeps one sanitizer lane for legacy untagged operations", async () => {
+    const operations = operationsForChunks([[Buffer.from(`${ESC}[`)], [Buffer.from("31mlegacy")]]);
+
+    const result = await executeBashWithOperations("printf output", "/tmp", operations);
+
+    expect(result.output).toBe("legacy");
+  });
+
+  it("flushes pending UTF-8 bytes from every tagged stream", async () => {
+    const chunks: string[] = [];
+    const operations = operationsForChunks([
+      [Buffer.from([0xe6, 0x97]), "stdout"],
+      [Buffer.from([0xe6, 0x97]), "stderr"],
+    ]);
+
+    const result = await executeBashWithOperations("printf output", "/tmp", operations, {
+      onChunk: (chunk) => chunks.push(chunk),
+    });
+
+    expect(chunks.join("")).toBe("��");
+    expect(result.output).toBe("��");
+  });
+
   it("stores sanitized split ANSI output in spilled full output files", async () => {
     const chunks = ["A\u001B]0;title", "\u0007B\n"];
     const repeatedChunks = Array.from({ length: 9000 }, (_, index) => chunks[index % 2] ?? "");

src/agents/sessions/bash-executor.ts

@@ -48,14 +48,16 @@ export async function executeBashWithOperations(
   operations: BashOperations,
   options?: BashExecutorOptions,
 ): Promise<BashResult> {
-  const sanitizeOutput = createStreamingBinaryOutputSanitizer();
   const output = new OutputAccumulator({
     tempFilePrefix: "openclaw-bash",
-    transformDecodedText: (text) => sanitizeOutput(text).replace(/\r/g, ""),
+    createTextTransform: () => {
+      const sanitizeOutput = createStreamingBinaryOutputSanitizer();
+      return (text) => sanitizeOutput(text).replace(/\r/g, "");
+    },
   });
 
-  const onData = (data: Buffer) => {
-    const text = output.append(data);
+  const onData = (data: Buffer, stream?: "stdout" | "stderr") => {
+    const text = output.append(data, stream);
     options?.onChunk?.(text);
   };
 

src/agents/sessions/tools/bash-operations.ts

@@ -1,12 +1,18 @@
 /**
  * Minimal shell execution interface injected into bash session tools.
  */
+
 export interface BashOperations {
   exec: (
     command: string,
     cwd: string,
     options: {
-      onData: (data: Buffer) => void;
+      /**
+       * stdout and stderr are independent pipes, so each needs its own decode
+       * state; tag chunks to keep them apart. Untagged chunks share one lane,
+       * preserving behavior for operations that cannot distinguish streams.
+       */
+      onData: (data: Buffer, stream?: "stdout" | "stderr") => void;
       signal?: AbortSignal;
       timeout?: number;
       env?: NodeJS.ProcessEnv;

src/agents/sessions/tools/bash.test.ts

@@ -83,4 +83,48 @@ describe("bash tool output lifecycle", () => {
 
     expect(result.content[0]).toEqual({ type: "text", text: "before\n" });
   });
+
+  it.runIf(process.platform !== "win32")(
+    "tags stdout and stderr from the local shell backend",
+    async () => {
+      const operations = createLocalBashOperations();
+      const chunks: Array<{ data: Buffer; stream?: "stdout" | "stderr" }> = [];
+
+      const result = await operations.exec("printf stdout; printf stderr >&2", process.cwd(), {
+        onData: (data, stream) => chunks.push({ data, stream }),
+      });
+
+      expect(result.exitCode).toBe(0);
+      expect(
+        Buffer.concat(
+          chunks.filter((chunk) => chunk.stream === "stdout").map((chunk) => chunk.data),
+        ).toString("utf8"),
+      ).toBe("stdout");
+      expect(
+        Buffer.concat(
+          chunks.filter((chunk) => chunk.stream === "stderr").map((chunk) => chunk.data),
+        ).toString("utf8"),
+      ).toBe("stderr");
+      expect(chunks.every((chunk) => chunk.stream !== undefined)).toBe(true);
+    },
+  );
+
+  it("decodes a split multi-byte character when the other stream interleaves", async () => {
+    // stdout and stderr are independent pipes: each needs its own decoder, or a
+    // character straddling a stdout read boundary is corrupted by an stderr write
+    // landing between its bytes.
+    const operations: BashOperations = {
+      exec: async (_command, _cwd, { onData }) => {
+        onData(Buffer.from([0xe6, 0x97]), "stdout"); // leading bytes of 日
+        onData(Buffer.from("E"), "stderr"); // interleaves mid-character
+        onData(Buffer.from([0xa5, 0x0a]), "stdout");
+        return { exitCode: 0 };
+      },
+    };
+    const tool = createBashTool(process.cwd(), { operations });
+
+    const result = await tool.execute("call-split-utf8", { command: "ignored" });
+
+    expect(result.content[0]).toEqual({ type: "text", text: "E日\n" });
+  });
 });

src/agents/sessions/tools/bash.ts

@@ -93,9 +93,10 @@ export function createLocalBashOperations(options?: { shellPath?: string }): Bas
             }
           }, timeoutMs);
         }
-        // Stream stdout and stderr.
-        child.stdout?.on("data", onData);
-        child.stderr?.on("data", onData);
+        // Stream stdout and stderr. Tag each pipe so downstream decode state
+        // stays per-stream; a pending sequence on one must not eat the other.
+        child.stdout?.on("data", (data: Buffer) => onData(data, "stdout"));
+        child.stderr?.on("data", (data: Buffer) => onData(data, "stderr"));
         // Handle abort signal by killing the entire process tree.
         const onAbort = () => {
           if (child.pid) {
@@ -374,11 +375,11 @@ export function createBashToolDefinition(
         onUpdate({ content: [], details: undefined });
       }
 
-      const handleData = (data: Buffer) => {
+      const handleData = (data: Buffer, stream?: "stdout" | "stderr") => {
         if (!acceptingOutput) {
           return;
         }
-        output.append(data);
+        output.append(data, stream);
         scheduleOutputUpdate();
       };
 

src/agents/sessions/tools/output-accumulator.test.ts

@@ -1,5 +1,5 @@
 // OutputAccumulator tests cover bounded UTF-8 tails and private spill files.
-import { rm, stat } from "node:fs/promises";
+import { readFile, rm, stat } from "node:fs/promises";
 import { describe, expect, it } from "vitest";
 import { OutputAccumulator } from "./output-accumulator.js";
 
@@ -42,4 +42,36 @@ describe("OutputAccumulator", () => {
     expect(snapshot.fullOutputPath).toBeDefined();
     await rm(snapshot.fullOutputPath!, { force: true });
   });
+
+  it("flushes pending bytes held by every stream lane", () => {
+    // Each lane decodes independently, so a truncated character left on one
+    // pipe must not stop the other pipe's tail from being flushed.
+    const accumulator = new OutputAccumulator();
+
+    accumulator.append(Buffer.from([0xe6, 0x97]), "stdout"); // leading bytes of 日
+    accumulator.append(Buffer.from([0xe6, 0x97]), "stderr");
+
+    const flushed = accumulator.finish();
+
+    expect(flushed).toBe("��");
+  });
+
+  it("spills tagged streams in decoded delivery order", async () => {
+    const accumulator = new OutputAccumulator({
+      maxBytes: 1,
+      maxLines: 10,
+      tempFilePrefix: "openclaw-output-test",
+    });
+
+    accumulator.append(Buffer.from([0xe6, 0x97]), "stdout"); // leading bytes of 日
+    accumulator.append(Buffer.from("E"), "stderr");
+    accumulator.append(Buffer.from([0xa5]), "stdout");
+    accumulator.finish();
+    const snapshot = accumulator.snapshot({ persistIfTruncated: true });
+    await accumulator.closeTempFile();
+
+    expect(snapshot.fullOutputPath).toBeDefined();
+    expect(await readFile(snapshot.fullOutputPath!, "utf8")).toBe("E日");
+    await rm(snapshot.fullOutputPath!, { force: true });
+  });
 });

src/agents/sessions/tools/output-accumulator.ts

@@ -16,7 +16,20 @@ interface OutputAccumulatorOptions {
   maxLines?: number;
   maxBytes?: number;
   tempFilePrefix?: string;
-  transformDecodedText?: (text: string) => string;
+  /**
+   * Builds the decoded-text transform. Called once per stream lane so stateful
+   * transforms (ANSI parsers) cannot consume another stream's pending sequence.
+   */
+  createTextTransform?: () => (text: string) => string;
+}
+
+type OutputStream = "stdout" | "stderr";
+
+/** Per-stream decode state. Streams are independent pipes and must not share it. */
+interface DecodeLane {
+  decoder: TextDecoder;
+  transform?: (text: string) => string;
+  spillDecoded: boolean;
 }
 
 interface OutputSnapshot {
@@ -41,8 +54,8 @@ export class OutputAccumulator {
   private readonly maxBytes: number;
   private readonly maxRollingBytes: number;
   private readonly tempFilePrefix: string;
-  private readonly transformDecodedText?: (text: string) => string;
-  private readonly decoder = new TextDecoder();
+  private readonly createTextTransform?: () => (text: string) => string;
+  private readonly lanes = new Map<OutputStream | undefined, DecodeLane>();
 
   private spillChunks: Buffer[] = [];
   private tailText = "";
@@ -64,22 +77,37 @@ export class OutputAccumulator {
     this.maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
     this.maxRollingBytes = Math.max(this.maxBytes * 2, 1);
     this.tempFilePrefix = options.tempFilePrefix ?? "openclaw-output";
-    this.transformDecodedText = options.transformDecodedText;
+    this.createTextTransform = options.createTextTransform;
   }
 
-  append(data: Buffer): string {
+  private lane(stream?: OutputStream): DecodeLane {
+    let lane = this.lanes.get(stream);
+    if (!lane) {
+      lane = {
+        decoder: new TextDecoder(),
+        transform: this.createTextTransform?.(),
+        // Tagged streams must spill decoded text because raw pipe bytes can
+        // interleave inside a UTF-8 character. Keep untagged raw spills stable.
+        spillDecoded: stream !== undefined || this.createTextTransform !== undefined,
+      };
+      this.lanes.set(stream, lane);
+    }
+    return lane;
+  }
+
+  append(data: Buffer, stream?: OutputStream): string {
     if (this.finished) {
       throw new Error("Cannot append to a finished output accumulator");
     }
 
     this.totalRawBytes += data.length;
-    const decodedText = this.decoder.decode(data, { stream: true });
-    const text = this.transformDecodedText?.(decodedText) ?? decodedText;
+    const lane = this.lane(stream);
+    const decodedText = lane.decoder.decode(data, { stream: true });
+    const text = lane.transform?.(decodedText) ?? decodedText;
     this.appendDecodedText(text);
 
-    // Transformed output must spill exactly what callers see so sanitization
-    // cannot be bypassed by reading the full-output file.
-    const spillChunk = this.transformDecodedText ? Buffer.from(text, "utf-8") : data;
+    // Decoded/transformed output must spill exactly what callers see.
+    const spillChunk = lane.spillDecoded ? Buffer.from(text, "utf-8") : data;
     if (this.tempFileStream || this.shouldUseTempFile()) {
       this.ensureTempFile();
     }
@@ -92,16 +120,24 @@ export class OutputAccumulator {
       return "";
     }
     this.finished = true;
-    const decodedText = this.decoder.decode();
-    const text = this.transformDecodedText?.(decodedText) ?? decodedText;
-    this.appendDecodedText(text);
-    if (this.transformDecodedText && text.length > 0) {
-      this.appendSpillChunk(Buffer.from(text, "utf-8"));
+    // Every lane holds its own pending bytes, so all of them must be flushed.
+    let flushed = "";
+    for (const lane of this.lanes.values()) {
+      const decodedText = lane.decoder.decode();
+      const text = lane.transform?.(decodedText) ?? decodedText;
+      if (text.length === 0) {
+        continue;
+      }
+      this.appendDecodedText(text);
+      if (lane.spillDecoded) {
+        this.appendSpillChunk(Buffer.from(text, "utf-8"));
+      }
+      flushed += text;
     }
     if (this.shouldUseTempFile()) {
       this.ensureTempFile();
     }
-    return text;
+    return flushed;
   }
 
   snapshot(options: { persistIfTruncated?: boolean } = {}): OutputSnapshot {