From 552fd224cf0c150ce319b11220930e475fb74f8c Mon Sep 17 00:00:00 2001 From: qingminlong Date: Thu, 16 Jul 2026 19:57:28 +0800 Subject: [PATCH] fix(opencode): reject output stream failures in session catalog (#108218) * fix(opencode): reject output stream failures in session catalog * test(opencode): format stream failure regression * test(opencode): prove output pipe failure cleanup Co-authored-by: qingminlong * test(opencode): satisfy pipe proof lint Co-authored-by: qingminlong --------- Co-authored-by: Peter Steinberger --- extensions/opencode/session-catalog.test.ts | 97 +++++++++++++++++++++ extensions/opencode/session-catalog.ts | 11 +++ 2 files changed, 108 insertions(+) diff --git a/extensions/opencode/session-catalog.test.ts b/extensions/opencode/session-catalog.test.ts index 2e859103474..785d9821e85 100644 --- a/extensions/opencode/session-catalog.test.ts +++ b/extensions/opencode/session-catalog.test.ts @@ -1,3 +1,5 @@ +import type { ChildProcess } from "node:child_process"; +import { once } from "node:events"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -7,6 +9,20 @@ import { afterEach, describe, expect, it, vi } from "vitest"; const nodeHostMocks = vi.hoisted(() => ({ runNodePtyCommand: vi.fn(async () => ({ exitCode: 0 })), })); +const childProcessMocks = vi.hoisted(() => ({ + children: [] as ChildProcess[], + spawn: vi.fn(), +})); + +vi.mock("node:child_process", async (importOriginal) => { + const actual = await importOriginal(); + childProcessMocks.spawn.mockImplementation((...args: Parameters) => { + const child = actual.spawn(...args); + childProcessMocks.children.push(child); + return child; + }); + return { ...actual, spawn: childProcessMocks.spawn }; +}); vi.mock("openclaw/plugin-sdk/node-host", async (importOriginal) => { const actual = await importOriginal(); @@ -45,6 +61,7 @@ import { const temporaryDirectories: string[] = []; const originalPath = process.env.PATH; +const originalPathExt = process.env.PATHEXT; const originalUnrelatedEnv = process.env.CATALOG_UNRELATED_ENV; function captureOpenCodeSessionRegistrations(pluginConfig: unknown = {}) { @@ -134,9 +151,58 @@ if (args[0] === "--pure" && args[1] === "db" && args.includes("--format") && arg return directory; } +async function installHangingOpenCode(): Promise { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-opencode-stream-")); + temporaryDirectories.push(directory); + const executableName = process.platform === "win32" ? "opencode.js" : "opencode"; + await fs.writeFile( + path.join(directory, executableName), + `${process.platform === "win32" ? "" : "#!/usr/bin/env node\n"}setTimeout(() => process.stdout.write("ready\\n"), 50); +setInterval(() => {}, 1_000); +`, + ); + if (process.platform !== "win32") { + await fs.chmod(path.join(directory, executableName), 0o755); + } + process.env.PATH = `${directory}${path.delimiter}${originalPath ?? ""}`; + if (process.platform === "win32") { + // The production resolver converts a PATHEXT-resolved .js command into + // process.execPath plus the script path, so this remains a direct real-child spawn. + process.env.PATHEXT = `.JS;${originalPathExt ?? ".EXE;.CMD;.BAT;.COM"}`; + } +} + +function isProcessRunning(pid: number | undefined): boolean { + if (!pid) { + return false; + } + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function stopChild(child: ChildProcess | undefined): Promise { + if (!child || !isProcessRunning(child.pid)) { + return; + } + const closed = once(child, "close"); + child.kill("SIGKILL"); + await closed; +} + afterEach(async () => { nodeHostMocks.runNodePtyCommand.mockClear(); + childProcessMocks.spawn.mockClear(); + await Promise.all(childProcessMocks.children.splice(0).map((child) => stopChild(child))); process.env.PATH = originalPath; + if (originalPathExt === undefined) { + delete process.env.PATHEXT; + } else { + process.env.PATHEXT = originalPathExt; + } if (originalUnrelatedEnv === undefined) { delete process.env.CATALOG_UNRELATED_ENV; } else { @@ -548,6 +614,37 @@ describe("OpenCode session catalog", () => { ); }); + it.each(["stdout", "stderr"] as const)( + "rejects and reaps the real OpenCode child when its %s pipe fails", + async (streamName) => { + await installHangingOpenCode(); + const uncaughtException = vi.fn(); + process.on("uncaughtExceptionMonitor", uncaughtException); + let child: ChildProcess | undefined; + try { + const listing = listLocalOpenCodeSessionPage({ limit: 20 }); + await vi.waitFor(() => expect(childProcessMocks.spawn).toHaveBeenCalledTimes(1)); + child = childProcessMocks.children[0]; + expect(child?.pid).toBeTypeOf("number"); + await once(child!.stdout!, "data"); + + child![streamName]!.destroy(new Error(`${streamName} EPIPE`)); + + await expect(listing).rejects.toThrow( + `OpenCode ${streamName} stream failed: ${streamName} EPIPE`, + ); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(uncaughtException).not.toHaveBeenCalled(); + expect(isProcessRunning(child!.pid)).toBe(false); + } finally { + process.off("uncaughtExceptionMonitor", uncaughtException); + await stopChild(child); + } + }, + ); + it("fans out paired-node listing instead of blocking later hosts", async () => { let provider: Parameters[0] | undefined; let releaseSlow: ((value: unknown) => void) | undefined; diff --git a/extensions/opencode/session-catalog.ts b/extensions/opencode/session-catalog.ts index 4c7b0c6c4bc..f748460018b 100644 --- a/extensions/opencode/session-catalog.ts +++ b/extensions/opencode/session-catalog.ts @@ -244,6 +244,12 @@ async function runOpenCode(args: string[]): Promise { let overflow = false; const timeout = setTimeout(() => child.kill("SIGKILL"), CLI_TIMEOUT_MS); timeout.unref?.(); + let outputError: Error | undefined; + const failFromOutputError = (source: "stdout" | "stderr", error: unknown) => { + const message = error instanceof Error ? error.message : String(error); + outputError ??= new Error(`OpenCode ${source} stream failed: ${message}`, { cause: error }); + child.kill("SIGKILL"); + }; const collect = (target: Buffer[], chunk: Buffer) => { bytes += chunk.length; if (bytes > MAX_CLI_OUTPUT_BYTES) { @@ -253,6 +259,8 @@ async function runOpenCode(args: string[]): Promise { } target.push(chunk); }; + child.stdout.once("error", (error) => failFromOutputError("stdout", error)); + child.stderr.once("error", (error) => failFromOutputError("stderr", error)); child.stdout.on("data", (chunk: Buffer) => collect(stdout, chunk)); child.stderr.on("data", (chunk: Buffer) => collect(stderr, chunk)); const exitCode = await new Promise((resolve, reject) => { @@ -262,6 +270,9 @@ async function runOpenCode(args: string[]): Promise { if (overflow) { throw new Error("OpenCode session output exceeded the safety limit"); } + if (outputError) { + throw outputError; + } if (exitCode !== 0) { const detail = Buffer.concat(stderr).toString("utf8").trim(); throw new Error(detail || `OpenCode exited with code ${String(exitCode)}`);