fix(hooks): keep a hook's broken pipe from crashing the run

The payload was written to a lifecycle hook's stdin without awaiting the
write. A shell hook that handles one lifecycle kind exits at once on the
other, closing its end of the pipe; a run summary larger than the pipe
buffers, as a long session's is, then fails with EPIPE, and the
unhandled rejection took the process down at the end of a finished run.
The write is awaited, a broken pipe is recorded as the hook's outcome,
and the run ends as it did.

Author: brianjfox

src/session/hooks.test.ts

@@ -1,8 +1,13 @@
 import { describe, expect, test } from "bun:test";
+import { mkdtemp, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import type { ReactorEmittedEvent } from "@intx/inference";
 import {
+  createLifecycleHookManager,
   createTurnContextCollector,
   HOOK_PAYLOAD_TOOL_RESULT_CHARS,
+  type RunSummary,
 } from "./hooks.js";
 
 function event(type: string, data: unknown): ReactorEmittedEvent {
@@ -60,3 +65,53 @@ describe("createTurnContextCollector tool result truncation", () => {
     expect(turn?.toolResults[0]?.content).toBe(smallOutput);
   });
 });
+
+describe("lifecycle hook payload delivery", () => {
+  test("a hook that exits without reading a large payload is a hook outcome, not a crash", async () => {
+    // A shell hook that handles only postTurn exits at once on postRun; the
+    // run summary of a long session is megabytes, far past what the pipe
+    // buffers, so the write lands on a closed pipe.
+    const directory = await mkdtemp(join(tmpdir(), "corbits-hook-"));
+    const path = join(directory, "post-turn-only.sh");
+    await writeFile(path, 'case "$1" in postTurn) cat > /dev/null ;; esac\n');
+    const errors: string[] = [];
+    const manager = createLifecycleHookManager({
+      hooks: [{ id: path, name: "post-turn-only.sh", type: "shell", path }],
+      logError: (message) => errors.push(message),
+    });
+    const summary: RunSummary = {
+      task: "x".repeat(2_000_000),
+      status: "done",
+      startedAt: 0,
+      finishedAt: 1,
+      durationMs: 1,
+      turnsUsed: 0,
+      tokenUsage: {
+        input: 0,
+        output: 0,
+        cacheRead: 0,
+        cacheWrite: 0,
+        thinking: 0,
+      },
+      turns: [],
+      toolCallCount: 0,
+    };
+    let unhandled: unknown = null;
+    const onUnhandled = (reason: unknown) => {
+      unhandled = reason;
+    };
+    process.on("unhandledRejection", onUnhandled);
+    try {
+      await manager.dispatchPostRun(summary);
+      await new Promise((resolve) => setTimeout(resolve, 200));
+    } finally {
+      process.off("unhandledRejection", onUnhandled);
+    }
+    // Whether the pipe breaks before the payload is buffered varies run to
+    // run; what must not vary is that neither outcome takes the process down.
+    expect(unhandled).toBeNull();
+    expect(errors).toEqual([]);
+    const [status] = manager.getStatuses();
+    expect(status?.lastExitStatus?.code).toBe(0);
+  });
+});

src/session/hooks.ts

@@ -1,3 +1,4 @@
+import type { FileSink } from "bun";
 import { mkdir, readdir, writeFile } from "node:fs/promises";
 import { createHash } from "node:crypto";
 import { homedir, tmpdir } from "node:os";
@@ -410,19 +411,51 @@ async function runLifecycleHook(
     stdout: "ignore",
     stderr: "pipe",
   });
-  proc.stdin.write(JSON.stringify(payload));
-  proc.stdin.end();
+  const delivered = await deliverPayload(proc.stdin, JSON.stringify(payload));
   const [exitCode, stderr] = await Promise.all([
     proc.exited,
     new Response(proc.stderr).text(),
   ]);
   return {
     code: exitCode,
     signal: proc.signalCode,
-    stderr,
+    stderr: delivered
+      ? stderr
+      : `${stderr}${stderr.length > 0 && !stderr.endsWith("\n") ? "\n" : ""}hook exited without reading its payload\n`,
   };
 }
 
+/**
+ * Writes the payload to the hook and says whether the hook took it. A hook
+ * that exits before reading — a shell hook that handles one lifecycle kind
+ * and ignores the other — closes its end of the pipe, and a payload larger
+ * than the pipe buffers (a long session's run summary) then fails with
+ * EPIPE. Unawaited, that rejection was fatal to the whole process at the end
+ * of a finished run. It is the hook's outcome, not the run's.
+ */
+async function deliverPayload(
+  stdin: FileSink,
+  payload: string,
+): Promise<boolean> {
+  try {
+    await stdin.write(payload);
+    await stdin.end();
+    return true;
+  } catch (err: unknown) {
+    if (isBrokenPipe(err)) return false;
+    throw err;
+  }
+}
+
+function isBrokenPipe(err: unknown): boolean {
+  return (
+    typeof err === "object" &&
+    err !== null &&
+    "code" in err &&
+    (err as { code?: unknown }).code === "EPIPE"
+  );
+}
+
 function hookCommand(hook: LifecycleHook, kind: HookKind): string[] {
   return ["sh", hook.path, kind];
 }