diff --git a/.changeset/handle-deno-stdin-failures.md b/.changeset/handle-deno-stdin-failures.md new file mode 100644 index 000000000..c5c366e12 --- /dev/null +++ b/.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. diff --git a/packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts b/packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts index 47ab5cc9c..e90f3da71 100644 --- a/packages/kernel/runtime-deno-subprocess/src/deno-worker-process.ts +++ b/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; + write: (data: string) => Promise; 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((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 => + 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,6 +173,7 @@ 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"); @@ -137,7 +181,8 @@ export const spawnDenoWorkerProcess = ( }; return { - stdin: child.stdin, + ready, + write, dispose, }; }; diff --git a/packages/kernel/runtime-deno-subprocess/src/fixtures/close-stdin-worker.mjs b/packages/kernel/runtime-deno-subprocess/src/fixtures/close-stdin-worker.mjs new file mode 100755 index 000000000..4fe1c6f70 --- /dev/null +++ b/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); +}); diff --git a/packages/kernel/runtime-deno-subprocess/src/index.test.ts b/packages/kernel/runtime-deno-subprocess/src/index.test.ts index 92f4aba38..b551fb2fa 100644 --- a/packages/kernel/runtime-deno-subprocess/src/index.test.ts +++ b/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* () { diff --git a/packages/kernel/runtime-deno-subprocess/src/index.ts b/packages/kernel/runtime-deno-subprocess/src/index.ts index caf20ccaa..55b38cfdc 100644 --- a/packages/kernel/runtime-deno-subprocess/src/index.ts +++ b/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): 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 => + 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(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; }