diff --git a/src/agents/sessions/bash-executor.test.ts b/src/agents/sessions/bash-executor.test.ts index 3962cdb69c0b..d40718afde3f 100644 --- a/src/agents/sessions/bash-executor.test.ts +++ b/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] ?? ""); diff --git a/src/agents/sessions/bash-executor.ts b/src/agents/sessions/bash-executor.ts index d15c6e51832c..e523a886157a 100644 --- a/src/agents/sessions/bash-executor.ts +++ b/src/agents/sessions/bash-executor.ts @@ -48,14 +48,16 @@ export async function executeBashWithOperations( operations: BashOperations, options?: BashExecutorOptions, ): Promise { - 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); }; diff --git a/src/agents/sessions/tools/bash-operations.ts b/src/agents/sessions/tools/bash-operations.ts index 66dd21393936..e1ac7b648b99 100644 --- a/src/agents/sessions/tools/bash-operations.ts +++ b/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; diff --git a/src/agents/sessions/tools/bash.test.ts b/src/agents/sessions/tools/bash.test.ts index e4162c3cb8c5..43f5929ca546 100644 --- a/src/agents/sessions/tools/bash.test.ts +++ b/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" }); + }); }); diff --git a/src/agents/sessions/tools/bash.ts b/src/agents/sessions/tools/bash.ts index 97468c849ba9..382c0d71066e 100644 --- a/src/agents/sessions/tools/bash.ts +++ b/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(); }; diff --git a/src/agents/sessions/tools/output-accumulator.test.ts b/src/agents/sessions/tools/output-accumulator.test.ts index 3084d8bce84b..0a1c75bd04d1 100644 --- a/src/agents/sessions/tools/output-accumulator.test.ts +++ b/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 }); + }); }); diff --git a/src/agents/sessions/tools/output-accumulator.ts b/src/agents/sessions/tools/output-accumulator.ts index 2ba7aac82a52..d1e1cbd7c469 100644 --- a/src/agents/sessions/tools/output-accumulator.ts +++ b/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(); 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 {