fix(imessage): handle stdout/stderr stream errors in the RPC client child (#101084)

* fix(imessage): handle stdout/stderr stream errors in the RPC client child

A dead imsg RPC helper can emit an async error on any of its stdio streams.
On a raw stream an unhandled 'error' event throws and surfaces as an
uncaughtException, crashing the gateway. #75438 added this guard for stdin
but left stdout/stderr — on the same long-lived child — unguarded.

Route stdout/stderr stream errors through the existing failAll path via a
shared failFromStreamError helper, mirroring the stdin handler. Add a
regression test that emits 'error' on each stream and asserts the child does
not throw and in-flight requests reject cleanly.

* fix(imessage): terminate failed RPC transports

* test(imessage): exercise real stream failure

---------

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

extensions/imessage/src/client.test.ts

@@ -0,0 +1,115 @@
+// iMessage tests cover the RPC client child-process stream error handling.
+import { EventEmitter } from "node:events";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const spawnMock = vi.hoisted(() => vi.fn());
+
+vi.mock("node:child_process", () => ({
+  spawn: spawnMock,
+}));
+
+// A dead imsg helper can emit an async `error` on any of its stdio streams. On
+// a raw EventEmitter an unhandled `error` throws synchronously, which in the
+// real gateway surfaces as an uncaughtException and crashes the process (#75438
+// covered stdin only). The mock child mirrors that stdio shape so we can assert
+// each stream's `error` is caught and routed to failAll.
+type MockChild = EventEmitter & {
+  stdout: EventEmitter;
+  stderr: EventEmitter;
+  stdin: EventEmitter & {
+    write: (line: string, cb?: (err?: Error | null) => void) => boolean;
+    end: () => void;
+  };
+  killed: boolean;
+  kill: ReturnType<typeof vi.fn>;
+};
+
+function createMockChild(): MockChild {
+  const child = new EventEmitter() as MockChild;
+  child.stdout = new EventEmitter();
+  child.stderr = new EventEmitter();
+  const stdin = new EventEmitter() as MockChild["stdin"];
+  // Resolve every write cleanly so the pending request only settles via the
+  // stream error path under test.
+  stdin.write = (_line, cb) => {
+    cb?.(null);
+    return true;
+  };
+  stdin.end = () => {};
+  child.stdin = stdin;
+  child.killed = false;
+  child.kill = vi.fn(() => {
+    child.killed = true;
+    return true;
+  });
+  return child;
+}
+
+describe("IMessageRpcClient child stream error handling", () => {
+  let child: MockChild;
+
+  beforeEach(() => {
+    // start() refuses to spawn under a test env; clear the markers so the real
+    // spawn/listener wiring runs against the mock child.
+    vi.stubEnv("NODE_ENV", "development");
+    vi.stubEnv("VITEST", "");
+    child = createMockChild();
+    spawnMock.mockReset().mockReturnValue(child);
+  });
+
+  afterEach(() => {
+    vi.unstubAllEnvs();
+    vi.resetModules();
+  });
+
+  it.each(["stdout", "stderr", "stdin"] as const)(
+    "catches a %s stream error and rejects in-flight requests instead of crashing",
+    async (streamName) => {
+      const { IMessageRpcClient } = await import("./client.js");
+      const client = new IMessageRpcClient({ cliPath: "imsg" });
+      await client.start();
+
+      const pending = client.request("ping", {}, { timeoutMs: 0 });
+      // Keep the rejection from surfacing as an unhandled rejection before we
+      // assert on it.
+      pending.catch(() => {});
+
+      const streamError = new Error(`${streamName} broke`);
+      expect(() => child[streamName].emit("error", streamError)).not.toThrow();
+
+      await expect(pending).rejects.toThrow(`${streamName} broke`);
+      await expect(client.waitForClose()).resolves.toBeUndefined();
+      expect(child.kill).toHaveBeenCalledOnce();
+      expect(child.kill).toHaveBeenCalledWith("SIGTERM");
+
+      await client.stop();
+    },
+  );
+
+  it("settles the client after a real child stdout stream failure", async () => {
+    const childProcess =
+      await vi.importActual<typeof import("node:child_process")>("node:child_process");
+    const realChild = childProcess.spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
+      stdio: ["pipe", "pipe", "pipe"],
+    });
+    spawnMock.mockReturnValueOnce(realChild);
+    const { IMessageRpcClient } = await import("./client.js");
+    const client = new IMessageRpcClient({ cliPath: "imsg" });
+    await client.start();
+
+    try {
+      const pending = client.request("ping", {}, { timeoutMs: 0 });
+      pending.catch(() => {});
+      realChild.stdout.destroy(new Error("real stdout failure"));
+
+      await expect(pending).rejects.toThrow("real stdout failure");
+      await expect(client.waitForClose()).resolves.toBeUndefined();
+      expect(realChild.killed).toBe(true);
+    } finally {
+      if (!realChild.killed) {
+        realChild.kill("SIGTERM");
+      }
+      await client.stop();
+    }
+  });
+});

extensions/imessage/src/client.ts

@@ -118,23 +118,29 @@ export class IMessageRpcClient {
       }
     });
 
-    child.on("error", (err) => {
-      this.failAll(err instanceof Error ? err : new Error(String(err)));
-      this.closedResolve?.();
-    });
-
-    // Without this listener, async EPIPE from a dead child crashes the
-    // gateway via uncaughtException. (#75438)
-    child.stdin.on("error", (err) => {
-      this.failAll(err instanceof Error ? err : new Error(String(err)));
-    });
+    // Every process/stdio error is terminal for this RPC transport. Settle the
+    // client once and terminate a helper whose pipe failed; otherwise the
+    // monitor can wait forever on an unusable child. #75438 covered stdin only.
+    const failFromProcessError = (err: unknown) => {
+      if (!this.finish(err instanceof Error ? err : new Error(String(err)))) {
+        return;
+      }
+      try {
+        child.kill("SIGTERM");
+      } catch {
+        // The helper may already be gone.
+      }
+    };
+    child.on("error", failFromProcessError);
+    child.stdin.on("error", failFromProcessError);
+    child.stdout.on("error", failFromProcessError);
+    child.stderr.on("error", failFromProcessError);
 
     child.on("close", (code, signal) => {
       if (this.child === child) {
         this.flushStdoutBuffer();
       }
-      this.failAll(this.buildCloseError(code, signal));
-      this.closedResolve?.();
+      this.finish(this.buildCloseError(code, signal));
     });
   }
 
@@ -328,6 +334,17 @@ export class IMessageRpcClient {
       this.pending.delete(key);
     }
   }
+
+  private finish(err: Error): boolean {
+    const resolve = this.closedResolve;
+    if (!resolve) {
+      return false;
+    }
+    this.closedResolve = null;
+    this.failAll(err);
+    resolve();
+    return true;
+  }
 }
 
 export async function createIMessageRpcClient(