fix(flows): kill shell children on interrupt and node timeout

src/flows/executors/shell.ts

@@ -1,7 +1,16 @@
-import { spawn } from "node:child_process";
-import { TimeoutError } from "../../async-control.js";
+import { spawn, type ChildProcess, type ChildProcessWithoutNullStreams } from "node:child_process";
+import { InterruptedError, TimeoutError } from "../../async-control.js";
 import type { ShellActionExecution, ShellActionResult } from "../runtime.js";
 
+const SHELL_CHILD_KILL_GRACE_MS = 1_000;
+
+function terminateShellChild(child: ChildProcess): void {
+  child.kill("SIGTERM");
+  setTimeout(() => {
+    child.kill("SIGKILL");
+  }, SHELL_CHILD_KILL_GRACE_MS).unref();
+}
+
 export function formatShellActionSummary(spec: ShellActionExecution): string {
   return `shell: ${renderShellCommand(spec.command, spec.args ?? [])}`;
 }
@@ -40,11 +49,35 @@ function rejectIfShellFailed(
   return undefined;
 }
 
-export async function runShellAction(spec: ShellActionExecution): Promise<ShellActionResult> {
-  const cwd = spec.cwd ?? process.cwd();
-  const args = spec.args ?? [];
-  const startMs = Date.now();
-  const child = spawn(spec.command, args, {
+function settleShellResult(
+  spec: ShellActionExecution,
+  args: string[],
+  result: ShellActionResult,
+  timedOut: boolean,
+  aborted: boolean,
+): Error | undefined {
+  if (aborted && !timedOut) {
+    return new InterruptedError();
+  }
+  return rejectIfShellFailed(spec, args, result, timedOut);
+}
+
+function scheduleShellTimeout(
+  timeoutMs: number | undefined,
+  onTimeout: () => void,
+): NodeJS.Timeout | undefined {
+  if (timeoutMs == null || timeoutMs <= 0) {
+    return undefined;
+  }
+  return setTimeout(onTimeout, timeoutMs);
+}
+
+function spawnShellChild(
+  spec: ShellActionExecution,
+  cwd: string,
+  args: string[],
+): ChildProcessWithoutNullStreams {
+  return spawn(spec.command, args, {
     cwd,
     env: {
       ...process.env,
@@ -54,13 +87,56 @@ export async function runShellAction(spec: ShellActionExecution): Promise<ShellA
     stdio: ["pipe", "pipe", "pipe"],
     windowsHide: true,
   });
+}
+
+function writeShellStdin(child: ChildProcessWithoutNullStreams, stdin: string | undefined): void {
+  if (stdin != null) {
+    child.stdin.write(stdin);
+  }
+  child.stdin.end();
+}
+
+function bindShellAbort(
+  signal: AbortSignal | undefined,
+  child: ChildProcess,
+): { isAborted: () => boolean; dispose: () => void } {
+  let aborted = false;
+  const onAbort = () => {
+    if (aborted) {
+      return;
+    }
+    aborted = true;
+    terminateShellChild(child);
+  };
+  if (!signal) {
+    return {
+      isAborted: () => aborted,
+      dispose: () => undefined,
+    };
+  }
+  signal.addEventListener("abort", onAbort, { once: true });
+  if (signal.aborted) {
+    onAbort();
+  }
+  return {
+    isAborted: () => aborted,
+    dispose: () => {
+      signal.removeEventListener("abort", onAbort);
+    },
+  };
+}
 
+function waitForShellExit(
+  child: ChildProcessWithoutNullStreams,
+  spec: ShellActionExecution,
+  args: string[],
+  cwd: string,
+  startMs: number,
+  getFlags: () => { timedOut: boolean; aborted: boolean },
+): Promise<ShellActionResult> {
   let stdout = "";
   let stderr = "";
-  let timedOut = false;
-  let timeout: NodeJS.Timeout | undefined;
-
-  const finish = new Promise<ShellActionResult>((resolve, reject) => {
+  return new Promise<ShellActionResult>((resolve, reject) => {
     child.stdout.setEncoding("utf8");
     child.stderr.setEncoding("utf8");
     child.stdout.on("data", (chunk: string) => {
@@ -69,9 +145,9 @@ export async function runShellAction(spec: ShellActionExecution): Promise<ShellA
     child.stderr.on("data", (chunk: string) => {
       stderr += chunk;
     });
-
     child.once("error", reject);
-    child.once("exit", (exitCode, signal) => {
+    child.once("exit", (exitCode, exitSignal) => {
+      const flags = getFlags();
       const result: ShellActionResult = {
         command: spec.command,
         args,
@@ -80,40 +156,55 @@ export async function runShellAction(spec: ShellActionExecution): Promise<ShellA
         stderr,
         combinedOutput: `${stdout}${stderr}`,
         exitCode,
-        signal,
+        signal: exitSignal,
         durationMs: Date.now() - startMs,
       };
-
-      const error = rejectIfShellFailed(spec, args, result, timedOut);
+      const error = settleShellResult(spec, args, result, flags.timedOut, flags.aborted);
       if (error) {
         reject(error);
         return;
       }
-
       resolve(result);
     });
   });
+}
 
-  if (spec.stdin != null) {
-    child.stdin.write(spec.stdin);
-  }
-  child.stdin.end();
-
-  if (spec.timeoutMs != null && spec.timeoutMs > 0) {
-    timeout = setTimeout(() => {
-      timedOut = true;
-      child.kill("SIGTERM");
-      setTimeout(() => {
-        child.kill("SIGKILL");
-      }, 1_000).unref();
-    }, spec.timeoutMs);
-  }
+async function runSpawnedShellAction(
+  spec: ShellActionExecution,
+  signal: AbortSignal | undefined,
+): Promise<ShellActionResult> {
+  const cwd = spec.cwd ?? process.cwd();
+  const args = spec.args ?? [];
+  const startMs = Date.now();
+  const child = spawnShellChild(spec, cwd, args);
+  let timedOut = false;
+  const abort = bindShellAbort(signal, child);
+  const finish = waitForShellExit(child, spec, args, cwd, startMs, () => ({
+    timedOut,
+    aborted: abort.isAborted(),
+  }));
+  writeShellStdin(child, spec.stdin);
+  const timeout = scheduleShellTimeout(spec.timeoutMs, () => {
+    timedOut = true;
+    terminateShellChild(child);
+  });
 
   try {
     return await finish;
   } finally {
     if (timeout) {
       clearTimeout(timeout);
     }
+    abort.dispose();
+  }
+}
+
+export async function runShellAction(
+  spec: ShellActionExecution,
+  options?: { signal?: AbortSignal },
+): Promise<ShellActionResult> {
+  if (options?.signal?.aborted) {
+    throw new InterruptedError();
   }
+  return await runSpawnedShellAction(spec, options?.signal);
 }

src/flows/runtime-support.ts

@@ -28,16 +28,19 @@ export function isoNow(): string {
   return new Date().toISOString();
 }
 
+export function isRunFailureFinalized(state: FlowRunState): boolean {
+  return (
+    state.finishedAt !== undefined && (state.status === "failed" || state.status === "timed_out")
+  );
+}
+
 export function persistRunFailure(
   store: FlowRunStore,
   runDir: string,
   state: FlowRunState,
   error: unknown,
 ): Promise<void> {
-  if (
-    state.finishedAt !== undefined &&
-    (state.status === "failed" || state.status === "timed_out")
-  ) {
+  if (isRunFailureFinalized(state)) {
     return Promise.resolve();
   }