🐛 Handle EPIPE on stdin in POSIX process implementation

process/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/process",
   "description": "Spawn and manage child processes with structured concurrency",
-  "version": "0.7.3",
+  "version": "0.7.4",
   "keywords": [
     "effection",
     "effectionx",

process/src/exec/posix.ts

@@ -14,6 +14,7 @@ import {
 } from "effection";
 import type { CreateOSProcess, ExitStatus, Writable } from "./api.ts";
 import { ExecError } from "./error.ts";
+import { suppressStdinEPIPE } from "./stdin.ts";
 
 type ProcessResultValue = [number?, string?];
 
@@ -81,6 +82,8 @@ export const createPosixProcess: CreateOSProcess = (command, options) => {
       },
     };
 
+    yield* spawn(() => suppressStdinEPIPE(childProcess.stdin, processResult));
+
     yield* spawn(function* trapError() {
       let [error] = yield* once<[Error]>(childProcess, "error");
       processResult.resolve(Err(error));

process/src/exec/stdin.ts

@@ -0,0 +1,19 @@
+import type { Writable } from "node:stream";
+import type { Operation, Result, WithResolvers } from "effection";
+import { Err, action } from "effection";
+
+type ProcessResultValue = [number?, string?];
+
+export function suppressStdinEPIPE(
+  stdin: Writable,
+  processResult: WithResolvers<Result<ProcessResultValue>>,
+): Operation<void> {
+  return action((_resolve, _reject) => {
+    const handler = (err: Error & { code?: string }) => {
+      if (err.code === "EPIPE") return;
+      processResult.resolve(Err(err));
+    };
+    stdin.on("error", handler);
+    return () => stdin.off("error", handler);
+  });
+}

process/src/exec/win32.ts

@@ -16,6 +16,7 @@ import {
 } from "effection";
 import type { CreateOSProcess, ExitStatus, Writable } from "./api.ts";
 import { ExecError } from "./error.ts";
+import { suppressStdinEPIPE } from "./stdin.ts";
 
 type ProcessResultValue = [number?, string?];
 
@@ -127,14 +128,7 @@ export const createWin32Process: CreateOSProcess = (command, options) => {
       return status;
     }
 
-    // Suppress EPIPE errors on stdin - these occur on Windows when the child
-    // process exits before we finish writing to it. This is expected during
-    // cleanup when we're killing the process.
-    childProcess.stdin.on("error", (err: Error & { code?: string }) => {
-      if (err.code !== "EPIPE") {
-        throw err;
-      }
-    });
+    yield* spawn(() => suppressStdinEPIPE(childProcess.stdin, processResult));
 
     try {
       yield* provide({

process/test/exec.test.ts

@@ -565,3 +565,27 @@ describe("handles env vars", () => {
 
   // Close the main "handles env vars" describe block
 });
+
+describe("stdin EPIPE handling", () => {
+  it("does not crash when writing to stdin after child exits", function* () {
+    let proc = yield* exec("node './fixtures/read-one-line.js'", {
+      cwd: import.meta.dirname,
+    });
+
+    // First write succeeds — child reads this line and exits
+    proc.stdin.send("hello\n");
+
+    // Wait for child to exit cleanly
+    let status = yield* proc.join();
+    expect(status.code).toEqual(0);
+
+    // Explicitly write to stdin after child has exited.
+    // Without the EPIPE handler, this would surface as an uncaught exception.
+    proc.stdin.send("this should not crash\n");
+
+    // If we reach here, the EPIPE was handled gracefully.
+    // The test completing is the assertion — an uncaught EPIPE would
+    // have crashed the test runner before this point.
+    expect(true).toBe(true);
+  });
+});

process/test/fixtures/read-one-line.js

@@ -0,0 +1,9 @@
+let buffer = "";
+process.stdin.setEncoding("utf8");
+process.stdin.on("data", (chunk) => {
+  buffer += chunk;
+  if (buffer.includes("\n")) {
+    process.stdout.write("got line\n");
+    process.exit(0);
+  }
+});