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,25 @@
+#!/usr/bin/env node
+
+import { closeSync } from "node:fs";
+import { createInterface } from "node:readline";
+
+const input = createInterface({ input: process.stdin });
+
+input.once("line", (line) => {
+  const { nonce } = JSON.parse(line);
+
+  input.close();
+  process.stdin.destroy();
+  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({
@@ -230,8 +256,21 @@ const executeInDeno = (
       catch: (cause) => new DenoSpawnError({ executable: denoExecutable, reason: cause }),
     });
 
-    // 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 };
+    }
 
     // Set up timeout — kills process and completes the deferred
     const timer = setTimeout(() => {
@@ -278,7 +317,15 @@ const executeInDeno = (
                   ),
                 );
 
-              writeMessage(worker.stdin, toolResult);
+              const writeSucceeded = yield* writeMessage(worker, toolResult).pipe(
+                Effect.as(true),
+                Effect.catchTag("DenoProcessWriteError", (error) =>
+                  completeWith({ result: null, error: error.message }).pipe(Effect.as(false)),
+                ),
+              );
+              if (!writeSucceeded) {
+                return;
+              }
               break;
             }