Handle Deno subprocess stdin failures

.changeset/handle-deno-stdin-failures.md

@@ -0,0 +1,5 @@
+---
+"executor": patch
+---
+
+Return an execution error when a Deno subprocess closes stdin instead of emitting an unhandled write failure.

packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts

@@ -25,11 +25,13 @@ export type DenoWorkerProcessCallbacks = {
   onStdoutLine: (line: string) => void;
   onStderr: (chunk: string) => void;
   onError: (error: Error) => void;
+  onStdinError: (error: Error) => void;
   onExit: (code: number | null, signal: NodeJS.Signals | null) => void;
 };
 
 export type DenoWorkerProcess = {
-  stdin: NodeJS.WritableStream;
+  ready: Promise<void>;
+  write: (data: string) => Promise<void>;
   dispose: () => void;
 };
 
@@ -81,6 +83,20 @@ export const spawnDenoWorkerProcess = (
     throw new Error("Failed to create piped stdio for Deno worker subprocess");
   }
 
+  const ready = new Promise<void>((resolve, reject) => {
+    const onSpawn = () => {
+      child.removeListener("error", onSpawnError);
+      resolve();
+    };
+    const onSpawnError = (cause: unknown) => {
+      child.removeListener("spawn", onSpawn);
+      reject(normalizeError(cause));
+    };
+
+    child.once("spawn", onSpawn);
+    child.once("error", onSpawnError);
+  });
+
   child.stdout.setEncoding("utf8");
   child.stderr.setEncoding("utf8");
 
@@ -109,17 +125,44 @@ export const spawnDenoWorkerProcess = (
     callbacks.onError(normalizeError(cause));
   };
 
+  const onStdinError = (cause: unknown) => {
+    callbacks.onStdinError(normalizeError(cause));
+  };
+
+  const onStdinClose = () => {
+    child.stdin.removeListener("error", onStdinError);
+  };
+
   const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
     callbacks.onExit(code, signal);
   };
 
   child.stdout.on("data", onStdoutData);
   child.stderr.on("data", onStderrData);
+  child.stdin.on("error", onStdinError);
+  child.stdin.once("close", onStdinClose);
   child.on("error", onError);
   child.on("exit", onExit);
 
   let disposed = false;
 
+  const write = (data: string): Promise<void> =>
+    new Promise((resolve, reject) => {
+      if (disposed || !child.stdin.writable) {
+        reject(new Error("Deno subprocess stdin is not writable"));
+        return;
+      }
+
+      child.stdin.write(data, (cause: Error | null | undefined) => {
+        if (cause) {
+          reject(cause);
+          return;
+        }
+
+        resolve();
+      });
+    });
+
   const dispose = () => {
     if (disposed) {
       return;
@@ -130,14 +173,16 @@ export const spawnDenoWorkerProcess = (
     child.stderr!.removeListener("data", onStderrData);
     child.removeListener("error", onError);
     child.removeListener("exit", onExit);
+    child.stdin.destroy();
 
     if (!child.killed) {
       child.kill("SIGKILL");
     }
   };
 
   return {
-    stdin: child.stdin,
+    ready,
+    write,
     dispose,
   };
 };

packages/kernel/runtime-deno-subprocess/src/fixtures/close-stdin-worker.mjs

@@ -0,0 +1,37 @@
+#!/usr/bin/env node
+
+// Reads the start message with raw fd reads instead of a stdin stream: only
+// an unwatched fd can be closed here, since libuv aborts the process when
+// fd 0 is closed while a stream watcher is attached, and a stream destroy()
+// never closes the underlying stdio fd, so the host's writes would keep
+// succeeding instead of failing with EPIPE.
+
+import { closeSync, readSync } from "node:fs";
+
+let buffered = "";
+const chunk = Buffer.alloc(4096);
+while (!buffered.includes("\n")) {
+  try {
+    const bytesRead = readSync(0, chunk, 0, chunk.length);
+    if (bytesRead <= 0) break;
+    buffered += chunk.toString("utf8", 0, bytesRead);
+  } catch (error) {
+    if (error.code === "EAGAIN") continue;
+    throw error;
+  }
+}
+
+const { nonce } = JSON.parse(buffered.slice(0, buffered.indexOf("\n")));
+
+closeSync(0);
+process.stdout.write(
+  `@@executor-ipc@@${JSON.stringify({
+    type: "tool_call",
+    nonce,
+    requestId: "request-1",
+    toolPath: "test.call",
+    args: {},
+  })}\n`,
+);
+
+setTimeout(() => process.exit(0), 5_000);

packages/kernel/runtime-deno-subprocess/src/index.test.ts

@@ -1,3 +1,5 @@
+import { fileURLToPath } from "node:url";
+
 import { describe, expect, it } from "@effect/vitest";
 import * as Data from "effect/Data";
 import * as Effect from "effect/Effect";
@@ -39,6 +41,27 @@ it.effect("returns an actionable error when Deno is missing", () =>
   }),
 );
 
+it.effect.skipIf(process.platform === "win32")(
+  "returns an execution error when the subprocess closes stdin",
+  () =>
+    Effect.gen(function* () {
+      const executor = makeDenoSubprocessExecutor({
+        denoExecutable: fileURLToPath(
+          new URL("./fixtures/close-stdin-worker.mjs", import.meta.url),
+        ),
+        timeoutMs: 5_000,
+      });
+      const toolInvoker = makeTestInvoker({
+        "test.call": () => "done",
+      });
+
+      const output = yield* executor.execute("return tools.test.call({});", toolInvoker);
+
+      expect(output.result).toBeNull();
+      expect(output.error).toContain("Failed to write to Deno subprocess stdin");
+    }),
+);
+
 describe.skipIf(!isDenoAvailable())("runtime-deno-subprocess", () => {
   it.effect("executes simple code and returns result", () =>
     Effect.gen(function* () {

packages/kernel/runtime-deno-subprocess/src/index.ts

@@ -16,7 +16,11 @@ import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
 import * as Schema from "effect/Schema";
 
-import { type DenoPermissions, spawnDenoWorkerProcess } from "./deno-worker-process";
+import {
+  type DenoPermissions,
+  type DenoWorkerProcess,
+  spawnDenoWorkerProcess,
+} from "./deno-worker-process";
 
 export type { DenoPermissions };
 
@@ -49,6 +53,15 @@ class DenoSpawnError extends Data.TaggedError("DenoSpawnError")<{
   }
 }
 
+class DenoProcessWriteError extends Data.TaggedError("DenoProcessWriteError")<{
+  readonly reason: unknown;
+}> {
+  override get message() {
+    const detail = this.reason instanceof Error ? this.reason.message : String(this.reason);
+    return `Failed to write to Deno subprocess stdin: ${detail}`;
+  }
+}
+
 // ---------------------------------------------------------------------------
 // IPC schemas
 // ---------------------------------------------------------------------------
@@ -149,9 +162,14 @@ const causeMessage = (cause: Cause.Cause<unknown>): string => {
   return String(squashed);
 };
 
-const writeMessage = (stdin: NodeJS.WritableStream, message: HostToWorkerMessage): void => {
-  stdin.write(`${JSON.stringify(message)}\n`);
-};
+const writeMessage = (
+  worker: DenoWorkerProcess,
+  message: HostToWorkerMessage,
+): Effect.Effect<void, DenoProcessWriteError> =>
+  Effect.tryPromise({
+    try: () => worker.write(`${JSON.stringify(message)}\n`),
+    catch: (reason) => new DenoProcessWriteError({ reason }),
+  });
 
 // ---------------------------------------------------------------------------
 // Core execution
@@ -217,6 +235,14 @@ const executeInDeno = (
                 }),
               );
             },
+            onStdinError: (cause) => {
+              runSync(
+                completeWith({
+                  result: null,
+                  error: new DenoProcessWriteError({ reason: cause }).message,
+                }),
+              );
+            },
             onExit: (exitCode, signal) => {
               runSync(
                 completeWith({
@@ -234,8 +260,21 @@ const executeInDeno = (
     let inFlight = 0;
     let lastReturnedAt = start;
 
-    // Send code to the subprocess
-    writeMessage(worker.stdin, { type: "start", code: recoveredBody, nonce });
+    const startError = yield* Effect.gen(function* () {
+      yield* Effect.tryPromise({
+        try: () => worker.ready,
+        catch: (reason) => new DenoSpawnError({ executable: denoExecutable, reason }),
+      });
+      yield* writeMessage(worker, { type: "start", code: recoveredBody, nonce });
+    }).pipe(
+      Effect.as<Error | undefined>(undefined),
+      Effect.catch((error) => Effect.succeed(error)),
+      Effect.onInterrupt(() => Effect.sync(worker.dispose)),
+    );
+    if (startError) {
+      worker.dispose();
+      return { result: null, error: startError.message };
+    }
 
     // Measure only continuous autonomous subprocess compute. Tool dispatches
     // suspend the clock and each return grants a fresh timeout budget.
@@ -269,35 +308,42 @@ const executeInDeno = (
               // must run even if the invoke or write dies, or the watchdog's
               // in-flight guard would suspend the timeout forever.
               inFlight += 1;
-              yield* toolInvoker.invoke({ path: msg.toolPath, args: msg.args }).pipe(
-                Effect.map(
-                  (value): HostToWorkerMessage => ({
-                    type: "tool_result",
-                    nonce,
-                    requestId: msg.requestId,
-                    ok: true,
-                    value,
-                  }),
-                ),
-                Effect.catchCause((cause) =>
-                  Effect.succeed<HostToWorkerMessage>({
-                    type: "tool_result",
-                    nonce,
-                    requestId: msg.requestId,
-                    ok: false,
-                    error: causeMessage(cause),
-                  }),
-                ),
-                Effect.flatMap((toolResult) =>
-                  Effect.sync(() => writeMessage(worker.stdin, toolResult)),
-                ),
-                Effect.ensuring(
-                  Effect.sync(() => {
-                    inFlight -= 1;
-                    lastReturnedAt = Date.now();
-                  }),
-                ),
-              );
+              const writeSucceeded = yield* toolInvoker
+                .invoke({ path: msg.toolPath, args: msg.args })
+                .pipe(
+                  Effect.map(
+                    (value): HostToWorkerMessage => ({
+                      type: "tool_result",
+                      nonce,
+                      requestId: msg.requestId,
+                      ok: true,
+                      value,
+                    }),
+                  ),
+                  Effect.catchCause((cause) =>
+                    Effect.succeed<HostToWorkerMessage>({
+                      type: "tool_result",
+                      nonce,
+                      requestId: msg.requestId,
+                      ok: false,
+                      error: causeMessage(cause),
+                    }),
+                  ),
+                  Effect.flatMap((toolResult) => writeMessage(worker, toolResult)),
+                  Effect.as(true),
+                  Effect.catchTag("DenoProcessWriteError", (error) =>
+                    completeWith({ result: null, error: error.message }).pipe(Effect.as(false)),
+                  ),
+                  Effect.ensuring(
+                    Effect.sync(() => {
+                      inFlight -= 1;
+                      lastReturnedAt = Date.now();
+                    }),
+                  ),
+                );
+              if (!writeSucceeded) {
+                return;
+              }
               break;
             }