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..0c85f7d06 --- /dev/null +++ b/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); diff --git a/packages/kernel/runtime-deno-subprocess/src/index.test.ts b/packages/kernel/runtime-deno-subprocess/src/index.test.ts index aadcedcb0..04086cfa3 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 8ef4e78ad..4a69800f4 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({ @@ -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(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({ - 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({ + 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; }