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 <qing.minlong@xydigit.com>

* test(opencode): satisfy pipe proof lint

Co-authored-by: qingminlong <qing.minlong@xydigit.com>

---------

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

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<typeof import("node:child_process")>();
+  childProcessMocks.spawn.mockImplementation((...args: Parameters<typeof actual.spawn>) => {
+    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<typeof import("openclaw/plugin-sdk/node-host")>();
@@ -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<void> {
+  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<void> {
+  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<void>((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<OpenClawPluginApi["registerSessionCatalog"]>[0] | undefined;
     let releaseSlow: ((value: unknown) => void) | undefined;

extensions/opencode/session-catalog.ts

@@ -244,6 +244,12 @@ async function runOpenCode(args: string[]): Promise<string> {
   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<string> {
     }
     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<number | null>((resolve, reject) => {
@@ -262,6 +270,9 @@ async function runOpenCode(args: string[]): Promise<string> {
   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)}`);