From 8e2f1d29988b0e1d3bc843b57020895d7c214974 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Tue, 11 Aug 2026 22:49:58 -0400
Subject: [PATCH] =?UTF-8?q?=E2=9A=97=EF=B8=8F=20Rebuild=20process=20adapte?=
 =?UTF-8?q?rs=20on=20a=20native-listener=20core?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace the fromReadable pump architecture with a single
createNativeProcess core that wires the child process entirely through
native Node listeners attached in the same synchronous continuation as
the spawn. posix and win32 reduce to SpawnStrategy objects: how to
spawn, and how to shut down.

Consequences:

- the orphan window from #236 closes: a guard teardown registers before
  the child exists and the spawn plus all listener wiring follow with no
  suspension points in between, so a halt can never land between a live
  process and an armed teardown
- close-settled means raw-output-complete by construction (#244): Node
  emits "close" only after both stdio streams have closed, and chunk
  delivery into the raw signals is synchronous with stream emission
- Stdio middleware becomes a consumer layer over the raw signals; a
  sequencer settles join()/expect() only after both consumers drain, and
  a failing handler resolves the result with Err instead of hanging
- the graceful-shutdown drain contract is preserved: the primary
  teardown registers after the consumers, so they are still alive to
  forward output produced during termination

Ports the halt-sweep regression test from #237 and the output
completeness and middleware failure tests from #245; all bite when their
respective guarantees are mutated out.
---
 process/src/exec/internal.ts |  12 +++
 process/src/exec/native.ts   | 195 +++++++++++++++++++++++++++++++++++
 process/src/exec/posix.ts    | 151 ++++-----------------------
 process/src/exec/win32.ts    | 186 +++++++--------------------------
 process/test/exec.test.ts    | 194 +++++++++++++++++++++++++++++++++-
 5 files changed, 461 insertions(+), 277 deletions(-)
 create mode 100644 process/src/exec/internal.ts
 create mode 100644 process/src/exec/native.ts

diff --git a/process/src/exec/internal.ts b/process/src/exec/internal.ts
new file mode 100644
index 00000000..35352b59
--- /dev/null
+++ b/process/src/exec/internal.ts
@@ -0,0 +1,12 @@
+import { type Operation, createContext } from "effection";
+
+/**
+ * Internal test seam, deliberately not exported from mod.ts. When set, the
+ * adapters run this operation as soon as the child-process "close" event is
+ * received, before waiting on the output pumps. This lets tests order their
+ * assertions deterministically around the close event instead of relying on
+ * scheduler timing.
+ */
+export const CloseEvent = createContext<() => Operation<void>>(
+  "@effectionx/process:close-event",
+);
diff --git a/process/src/exec/native.ts b/process/src/exec/native.ts
new file mode 100644
index 00000000..c7c41a1f
--- /dev/null
+++ b/process/src/exec/native.ts
@@ -0,0 +1,195 @@
+import type { ChildProcess } from "node:child_process";
+import {
+  type Operation,
+  type Result,
+  type Yielded,
+  Err,
+  Ok,
+  all,
+  createSignal,
+  ensure,
+  spawn,
+  withResolvers,
+} from "effection";
+import { unbox, useEvalScope } from "@effectionx/scope-eval";
+import type {
+  CreateOSProcess,
+  ExecOptions,
+  ExitStatus,
+  Process,
+  Writable,
+} from "./types.ts";
+import { Stdio } from "../api.ts";
+import { ExecError } from "./error.ts";
+import { CloseEvent } from "./internal.ts";
+
+type ProcessResultValue = [number?, string?];
+
+/**
+ * The platform-specific half of a process adapter. `spawn` must create the
+ * child process and attach any platform listeners synchronously — it must not
+ * suspend. `shutdown` requests termination and then waits on `drained`, which
+ * resolves once the output the caller cares about has been delivered.
+ */
+export interface SpawnStrategy {
+  spawn(command: string, options: ExecOptions): ChildProcess;
+  shutdown(
+    child: ChildProcess,
+    drained: () => Operation<void>,
+  ): Operation<void>;
+}
+
+export function* createNativeProcess(
+  command: string,
+  options: ExecOptions,
+  strategy: SpawnStrategy,
+): Operation<Process> {
+  let processResult = withResolvers<Result<ProcessResultValue>>();
+  const evalScope = yield* useEvalScope();
+  const result = yield* evalScope.eval(function* () {
+    let rawClose = withResolvers<Result<ProcessResultValue>>();
+    let stdoutDone = withResolvers<void>();
+    let stderrDone = withResolvers<void>();
+
+    let rawStdout = createSignal<Uint8Array, void>();
+    let rawStderr = createSignal<Uint8Array, void>();
+
+    let stdout = createSignal<Uint8Array, void>();
+    let stderr = createSignal<Uint8Array, void>();
+
+    let child: ChildProcess | undefined;
+    let teardownStarted = false;
+
+    // Guard for halts landing between here and the primary teardown at the
+    // end of acquisition: registered while `child` is still undefined, with
+    // the spawn following in the same synchronous continuation, so the
+    // process never exists without an armed teardown. It joins on the close
+    // event alone because the middleware consumers may never have been wired.
+    yield* ensure(function* () {
+      if (child && !teardownStarted) {
+        yield* strategy.shutdown(child, function* () {
+          yield* rawClose.operation;
+        });
+      }
+    });
+
+    const spawned = strategy.spawn(command, options);
+    child = spawned;
+
+    spawned.once("error", (error) => {
+      rawClose.resolve(Err(error));
+      processResult.resolve(Err(error));
+    });
+    spawned.once("close", (code, signal) => {
+      rawClose.resolve(Ok([code ?? undefined, signal ?? undefined]));
+    });
+
+    if (!spawned.stdout || !spawned.stderr) {
+      throw new Error("stdout and stderr must be available with stdio: pipe");
+    }
+
+    // Native listeners, wired in the same synchronous continuation as the
+    // spawn. Chunk delivery is synchronous with stream emission and Node
+    // emits the child "close" event only after both stdio streams have
+    // closed, so by the time rawClose resolves every chunk is already in the
+    // raw signals: close-settled means raw-output-complete by construction.
+    spawned.stdout.on("data", (chunk: Uint8Array) => rawStdout.send(chunk));
+    spawned.stdout.once("close", () => rawStdout.close());
+    spawned.stderr.on("data", (chunk: Uint8Array) => rawStderr.send(chunk));
+    spawned.stderr.once("close", () => rawStderr.close());
+
+    yield* spawn(function* () {
+      let subscription = yield* rawStdout;
+      try {
+        let next = yield* subscription.next();
+        while (!next.done) {
+          yield* Stdio.operations.stdout(next.value);
+          stdout.send(next.value);
+          next = yield* subscription.next();
+        }
+      } catch (error) {
+        // deliver Stdio middleware failures through join()/expect() so a
+        // broken handler cannot leave callers waiting on processResult
+        processResult.resolve(Err(error as Error));
+      } finally {
+        stdout.close();
+        stdoutDone.resolve();
+      }
+    });
+
+    yield* spawn(function* () {
+      let subscription = yield* rawStderr;
+      try {
+        let next = yield* subscription.next();
+        while (!next.done) {
+          yield* Stdio.operations.stderr(next.value);
+          stderr.send(next.value);
+          next = yield* subscription.next();
+        }
+      } catch (error) {
+        processResult.resolve(Err(error as Error));
+      } finally {
+        stderr.close();
+        stderrDone.resolve();
+      }
+    });
+
+    yield* spawn(function* () {
+      let result = yield* rawClose.operation;
+      let closeEvent = yield* CloseEvent.get();
+      if (closeEvent) {
+        yield* closeEvent();
+      }
+      // join() settles only after every chunk has also cleared Stdio
+      // middleware and the public signals
+      yield* all([stdoutDone.operation, stderrDone.operation]);
+      processResult.resolve(result);
+    });
+
+    function* join() {
+      let result = yield* processResult.operation;
+      if (result.ok) {
+        let [code, signal] = result.value;
+        return { command, options, code, signal } as ExitStatus;
+      }
+      throw result.error;
+    }
+
+    function* expect() {
+      let status: ExitStatus = yield* join();
+      if (status.code !== 0) {
+        throw new ExecError(status, command, options);
+      }
+      return status;
+    }
+
+    let stdin: Writable<string> = {
+      send(data: string) {
+        spawned.stdin?.write(data);
+      },
+    };
+
+    yield* ensure(function* () {
+      teardownStarted = true;
+      yield* strategy.shutdown(spawned, function* () {
+        yield* all([stdoutDone.operation, stderrDone.operation]);
+      });
+    });
+
+    return {
+      pid: spawned.pid as number,
+      *around(
+        ...args: Parameters<typeof Stdio.around>
+      ): ReturnType<typeof Stdio.around> {
+        const result = yield* evalScope.eval(() => Stdio.around(...args));
+        return unbox(result);
+      },
+      stdin,
+      stdout,
+      stderr,
+      join,
+      expect,
+    } satisfies Yielded<ReturnType<CreateOSProcess>>;
+  });
+  return unbox(result);
+}
diff --git a/process/src/exec/posix.ts b/process/src/exec/posix.ts
index 6b5bb9bc..c78552e5 100644
--- a/process/src/exec/posix.ts
+++ b/process/src/exec/posix.ts
@@ -1,39 +1,11 @@
 import { spawn as spawnProcess } from "node:child_process";
 import process from "node:process";
-import {
-  type Result,
-  type Operation,
-  type Yielded,
-  Err,
-  Ok,
-  all,
-  createSignal,
-  ensure,
-  spawn,
-  withResolvers,
-} from "effection";
-import { unbox, useEvalScope } from "@effectionx/scope-eval";
-import { once } from "@effectionx/node/events";
-import { fromReadable } from "@effectionx/node/stream";
-import type {
-  CreateOSProcess,
-  ExecOptions,
-  ExitStatus,
-  Process,
-  Writable,
-} from "./types.ts";
-import { Stdio } from "../api.ts";
-import { ExecError } from "./error.ts";
+import type { Operation } from "effection";
+import type { ExecOptions, Process } from "./types.ts";
+import { type SpawnStrategy, createNativeProcess } from "./native.ts";
 
-type ProcessResultValue = [number?, string?];
-
-export function* createPosixProcess(
-  command: string,
-  options: ExecOptions,
-): Operation<Process> {
-  let processResult = withResolvers<Result<ProcessResultValue>>();
-  const evalScope = yield* useEvalScope();
-  const result = yield* evalScope.eval(function* () {
+const posix: SpawnStrategy = {
+  spawn(command: string, options: ExecOptions) {
     // Killing all child processes started by this command is surprisingly
     // tricky. If a process spawns another processes and we kill the parent,
     // then the child process is NOT automatically killed. Instead we're using
@@ -44,111 +16,30 @@ export function* createPosixProcess(
     // process.
     //
     // More information here: https://unix.stackexchange.com/questions/14815/process-descendants
-    let childProcess = spawnProcess(command, options.arguments || [], {
+    return spawnProcess(command, options.arguments || [], {
       detached: true,
       shell: options.shell,
       env: options.env,
       cwd: options.cwd,
       stdio: "pipe",
     });
+  },
 
-    let { pid } = childProcess;
-
-    if (!childProcess.stdout || !childProcess.stderr) {
-      throw new Error("stdout and stderr must be available with stdio: pipe");
-    }
-
-    let io = {
-      stdout: yield* fromReadable(childProcess.stdout),
-      stderr: yield* fromReadable(childProcess.stderr),
-      stdoutDone: withResolvers<void>(),
-      stderrDone: withResolvers<void>(),
-    };
-
-    let stdout = createSignal<Uint8Array, void>();
-    let stderr = createSignal<Uint8Array, void>();
-
-    yield* spawn(function* () {
-      let next = yield* io.stdout.next();
-      while (!next.done) {
-        yield* Stdio.operations.stdout(next.value);
-        stdout.send(next.value);
-        next = yield* io.stdout.next();
-      }
-      stdout.close();
-      io.stdoutDone.resolve();
-    });
-
-    yield* spawn(function* () {
-      let next = yield* io.stderr.next();
-      while (!next.done) {
-        yield* Stdio.operations.stderr(next.value);
-        stderr.send(next.value);
-        next = yield* io.stderr.next();
-      }
-      stderr.close();
-      io.stderrDone.resolve();
-    });
-
-    let stdin: Writable<string> = {
-      send(data: string) {
-        childProcess.stdin.write(data);
-      },
-    };
-
-    yield* spawn(function* trapError() {
-      let [error] = yield* once<[Error]>(childProcess, "error");
-      processResult.resolve(Err(error));
-    });
-
-    yield* spawn(function* () {
-      let value = yield* once<ProcessResultValue>(childProcess, "close");
-      processResult.resolve(Ok(value));
-    });
-
-    function* join() {
-      let result = yield* processResult.operation;
-      if (result.ok) {
-        let [code, signal] = result.value;
-        return { command, options, code, signal } as ExitStatus;
-      }
-      throw result.error;
-    }
-
-    function* expect() {
-      let status: ExitStatus = yield* join();
-      if (status.code !== 0) {
-        throw new ExecError(status, command, options);
-      }
-      return status;
-    }
-
-    yield* ensure(function* () {
+  *shutdown(child, drained) {
+    if (typeof child.pid !== "undefined") {
       try {
-        if (typeof childProcess.pid === "undefined") {
-          throw new Error("no pid for childProcess");
-        }
-        process.kill(-childProcess.pid, "SIGTERM");
-        yield* all([io.stdoutDone.operation, io.stderrDone.operation]);
-      } catch (_e) {
-        // do nothing, process is probably already dead
+        process.kill(-child.pid, "SIGTERM");
+      } catch (_error) {
+        // the process group is already gone
       }
-    });
+    }
+    yield* drained();
+  },
+};
 
-    return {
-      pid: pid as number,
-      *around(
-        ...args: Parameters<typeof Stdio.around>
-      ): ReturnType<typeof Stdio.around> {
-        const result = yield* evalScope.eval(() => Stdio.around(...args));
-        return unbox(result);
-      },
-      stdin,
-      stdout,
-      stderr,
-      join,
-      expect,
-    } satisfies Yielded<ReturnType<CreateOSProcess>>;
-  });
-  return unbox(result);
+export function* createPosixProcess(
+  command: string,
+  options: ExecOptions,
+): Operation<Process> {
+  return yield* createNativeProcess(command, options, posix);
 }
diff --git a/process/src/exec/win32.ts b/process/src/exec/win32.ts
index 3a2c78b2..66f11aca 100644
--- a/process/src/exec/win32.ts
+++ b/process/src/exec/win32.ts
@@ -1,33 +1,11 @@
 import { platform } from "node:os";
 import { once } from "@effectionx/node/events";
-import { fromReadable } from "@effectionx/node/stream";
 // @ts-types="npm:@types/cross-spawn@6.0.6"
 import { spawn as spawnProcess } from "cross-spawn";
 import { ctrlc } from "ctrlc-windows";
-import {
-  type Operation,
-  type Result,
-  type Yielded,
-  Err,
-  Ok,
-  all,
-  createSignal,
-  ensure,
-  spawn,
-  withResolvers,
-} from "effection";
-import type {
-  CreateOSProcess,
-  ExecOptions,
-  ExitStatus,
-  Process,
-  Writable,
-} from "./types.ts";
-import { Stdio } from "../api.ts";
-import { ExecError } from "./error.ts";
-import { unbox, useEvalScope } from "@effectionx/scope-eval";
-
-type ProcessResultValue = [number?, string?];
+import type { Operation } from "effection";
+import type { ExecOptions, Process } from "./types.ts";
+import { type SpawnStrategy, createNativeProcess } from "./native.ts";
 
 function* killTree(pid: number) {
   try {
@@ -42,15 +20,10 @@ function* killTree(pid: number) {
   }
 }
 
-export function* createWin32Process(
-  command: string,
-  options: ExecOptions,
-): Operation<Process> {
-  let processResult = withResolvers<Result<ProcessResultValue>>();
-  const evalScope = yield* useEvalScope();
-  const result = yield* evalScope.eval(function* () {
+const win32: SpawnStrategy = {
+  spawn(command: string, options: ExecOptions) {
     // Windows-specific process spawning with different options than POSIX
-    let childProcess = spawnProcess(command, options.arguments || [], {
+    const child = spawnProcess(command, options.arguments || [], {
       // We lose exit information and events if this is detached in windows
       // and it opens a window in windows+powershell.
       detached: false,
@@ -70,102 +43,32 @@ export function* createWin32Process(
       cwd: options.cwd,
     });
 
-    let { pid } = childProcess;
-
-    if (!childProcess.stdout || !childProcess.stderr) {
-      throw new Error("stdout and stderr must be available with stdio: pipe");
-    }
-
-    let io = {
-      stdout: yield* fromReadable(childProcess.stdout),
-      stderr: yield* fromReadable(childProcess.stderr),
-      stdoutDone: withResolvers<void>(),
-      stderrDone: withResolvers<void>(),
-    };
-
-    const stdout = createSignal<Uint8Array, void>();
-    const stderr = createSignal<Uint8Array, void>();
-
-    yield* spawn(function* () {
-      let next = yield* io.stdout.next();
-      while (!next.done) {
-        yield* Stdio.operations.stdout(next.value);
-        stdout.send(next.value);
-        next = yield* io.stdout.next();
-      }
-      stdout.close();
-      io.stdoutDone.resolve();
-    });
-
-    yield* spawn(function* () {
-      let next = yield* io.stderr.next();
-      while (!next.done) {
-        yield* Stdio.operations.stderr(next.value);
-        stderr.send(next.value);
-        next = yield* io.stderr.next();
-      }
-      stderr.close();
-      io.stderrDone.resolve();
-    });
-
-    let stdin: Writable<string> = {
-      send(data: string) {
-        childProcess.stdin.write(data);
-      },
-    };
-
-    yield* spawn(function* trapError() {
-      const [error] = yield* once<Error[]>(childProcess, "error");
-      processResult.resolve(Err(error));
-    });
-
-    yield* spawn(function* () {
-      let value = yield* once<ProcessResultValue>(childProcess, "close");
-      // out of band with the finally block below compared to posix as
-      // win32 is more sensitive to graceful shutdown timing that it is
-      // worth waiting for stdout and stderr to close before resolving the process result
-      yield* all([io.stdoutDone.operation, io.stderrDone.operation]);
-      processResult.resolve(Ok(value));
-    });
-
-    function* join() {
-      let result = yield* processResult.operation;
-      if (result.ok) {
-        let [code, signal] = result.value;
-        return { command, options, code, signal } as ExitStatus;
-      }
-      throw result.error;
-    }
-
-    function* expect() {
-      let status = yield* join();
-      if (status.code !== 0) {
-        throw new ExecError(status, command, options);
-      }
-      return status;
-    }
-
     // Suppress EPIPE errors on stdin - these occur on Windows when the child
     // process exits before we finish writing to it. This is expected during
     // cleanup when we're killing the process.
-    childProcess.stdin.on("error", (err: Error & { code?: string }) => {
+    child.stdin?.on("error", (err: Error & { code?: string }) => {
       if (err.code !== "EPIPE") {
         throw err;
       }
     });
 
-    yield* ensure(function* () {
-      // If no pid is available, we have no way to kill the process,
-      //  so we skip and presume it is cleaned up.
-      if (pid) {
-        try {
-          ctrlc(pid);
-        } catch (_) {
-          // if it throws, the process probably doesn't exist anymore
-          //  as it does a process.kill(0) check which will throw if the process is not found
-        }
+    return child;
+  },
+
+  *shutdown(child, drained) {
+    // If no pid is available, we have no way to kill the process,
+    //  so we skip and presume it is cleaned up.
+    const pid = child.pid;
+    if (pid) {
+      try {
+        ctrlc(pid);
+      } catch (_) {
+        // if it throws, the process probably doesn't exist anymore
+        //  as it does a process.kill(0) check which will throw if the process is not found
+      }
 
-        let stdin = childProcess.stdin;
+      const stdin = child.stdin;
+      if (stdin) {
         if (stdin.writable) {
           try {
             //Terminate batch process (Y/N)
@@ -176,35 +79,26 @@ export function* createWin32Process(
         }
         stdin.end();
       }
+    }
 
-      // depending on how we shutdown, this may already be closed and
-      // will pass immediately over the operations
-      yield* all([io.stdoutDone.operation, io.stderrDone.operation]);
+    // depending on how we shutdown, this may already be closed and
+    // will pass immediately over the operations
+    yield* drained();
 
-      if (pid && childProcess.exitCode === null) {
-        // If the process is still around after we've waited
-        // for stdout and stderr to close,
-        // then force kill the tree.
-        yield* killTree(pid);
-      }
-    });
+    if (pid && child.exitCode === null) {
+      // If the process is still around after we've waited
+      // for stdout and stderr to close,
+      // then force kill the tree.
+      yield* killTree(pid);
+    }
+  },
+};
 
-    return {
-      pid: pid as number,
-      *around(
-        ...args: Parameters<typeof Stdio.around>
-      ): ReturnType<typeof Stdio.around> {
-        const result = yield* evalScope.eval(() => Stdio.around(...args));
-        return unbox(result);
-      },
-      stdin,
-      stdout,
-      stderr,
-      join,
-      expect,
-    } satisfies Yielded<ReturnType<CreateOSProcess>>;
-  });
-  return unbox(result);
+export function* createWin32Process(
+  command: string,
+  options: ExecOptions,
+): Operation<Process> {
+  return yield* createNativeProcess(command, options, win32);
 }
 
 export const isWin32 = (): boolean => platform() === "win32";
diff --git a/process/test/exec.test.ts b/process/test/exec.test.ts
index 3b6908fa..58e080fb 100644
--- a/process/test/exec.test.ts
+++ b/process/test/exec.test.ts
@@ -1,6 +1,15 @@
+import { execSync } from "node:child_process";
 import process from "node:process";
 import { beforeEach, describe, it } from "@effectionx/vitest";
-import { type Task, sleep, spawn, withResolvers } from "effection";
+import {
+  type Operation,
+  type Task,
+  sleep,
+  spawn,
+  suspend,
+  useScope,
+  withResolvers,
+} from "effection";
 import { expect } from "expect";
 
 import { captureError, expectMatch, fetchText } from "./helpers.ts";
@@ -8,6 +17,7 @@ import { captureError, expectMatch, fetchText } from "./helpers.ts";
 import { lines } from "@effectionx/stream-helpers";
 import { type Process, type ProcessResult, exec } from "../mod.ts";
 import { Stdio } from "../src/api.ts";
+import { CloseEvent } from "../src/exec/internal.ts";
 
 const SystemRoot = process.env.SystemRoot;
 
@@ -299,6 +309,127 @@ describe("exec", () => {
     });
   });
 
+  describe("output completeness", () => {
+    // Gate every stdout chunk behind `release`, and only resolve `release`
+    // once the adapter has received the child-process close event (observed
+    // through the internal CloseEvent seam). join() is therefore settling
+    // strictly after the close event, and must still have the full gated
+    // output forwarded through Stdio middleware by the time it settles.
+    it("delivers all stdout through Stdio middleware before join() settles", function* () {
+      const closed = withResolvers<void>();
+      const release = withResolvers<void>();
+      const procReady = withResolvers<Process>();
+      const observed: Uint8Array[] = [];
+      let stdoutAtSettle: string | undefined;
+
+      yield* CloseEvent.set(function* () {
+        closed.resolve();
+      });
+
+      yield* Stdio.around({
+        *stdout([bytes]) {
+          yield* release.operation;
+          observed.push(bytes);
+        },
+      });
+
+      // spawned before exec so that whenever this task and the adapter's
+      // stdout pump are both runnable, this one resumes first and captures
+      // what had been forwarded at the moment join() settled
+      const joined = yield* spawn(function* () {
+        const proc = yield* procReady.operation;
+        const status = yield* proc.join();
+        stdoutAtSettle = Buffer.concat(observed)
+          .toString("utf8")
+          .replace(/\r\n/g, "\n");
+        return status;
+      });
+
+      const proc = yield* exec("node './fixtures/hello-world.js'", {
+        cwd: import.meta.dirname,
+      });
+      procReady.resolve(proc);
+
+      // the child process has closed, but the gate is still held
+      yield* closed.operation;
+      release.resolve();
+
+      const status = yield* joined;
+      expect(status.code).toEqual(0);
+      expect(stdoutAtSettle).toEqual("hello\nworld\n");
+    });
+
+    it("delivers all stderr through Stdio middleware before join() settles", function* () {
+      const closed = withResolvers<void>();
+      const release = withResolvers<void>();
+      const procReady = withResolvers<Process>();
+      const observed: Uint8Array[] = [];
+      let stderrAtSettle: string | undefined;
+
+      yield* CloseEvent.set(function* () {
+        closed.resolve();
+      });
+
+      yield* Stdio.around({
+        *stderr([bytes]) {
+          yield* release.operation;
+          observed.push(bytes);
+        },
+      });
+
+      const joined = yield* spawn(function* () {
+        const proc = yield* procReady.operation;
+        const status = yield* proc.join();
+        stderrAtSettle = Buffer.concat(observed)
+          .toString("utf8")
+          .replace(/\r\n/g, "\n");
+        return status;
+      });
+
+      const proc = yield* exec("node './fixtures/hello-world.js'", {
+        cwd: import.meta.dirname,
+      });
+      procReady.resolve(proc);
+
+      yield* closed.operation;
+      release.resolve();
+
+      const status = yield* joined;
+      expect(status.code).toEqual(0);
+      expect(stderrAtSettle).toContain("boom\n");
+    });
+
+    it("fails join() instead of hanging when a Stdio handler fails", function* () {
+      yield* Stdio.around({
+        *stdout() {
+          throw new Error("stdout handler exploded");
+        },
+      });
+
+      const proc = yield* exec("node './fixtures/hello-world.js'", {
+        cwd: import.meta.dirname,
+      });
+
+      const error = yield* captureError(proc.join());
+      expect(error.message).toEqual("stdout handler exploded");
+    });
+
+    it("fails expect() instead of hanging when a Stdio handler fails", function* () {
+      yield* Stdio.around({
+        *stderr() {
+          throw new Error("stderr handler exploded");
+        },
+      });
+
+      const proc = yield* exec("node './fixtures/hello-world.js'", {
+        cwd: import.meta.dirname,
+      });
+
+      const error = yield* captureError(proc.expect());
+      expect(error.message).toEqual("stderr handler exploded");
+    });
+  });
+
   describe("io api", () => {
     it("allows redirecting stdio to array", function* () {
       let outputStdout: Uint8Array[] = [];
@@ -631,4 +762,65 @@ describe("exec", () => {
       });
     }
   });
+
+  if (process.platform !== "win32") {
+    describe("halt during acquisition", () => {
+      function deep(
+        levels: number,
+        body: () => Operation<void>,
+      ): Operation<void> {
+        if (levels === 0) {
+          return body();
+        }
+        return {
+          *[Symbol.iterator]() {
+            const inner = yield* spawn(() => deep(levels - 1, body));
+            yield* inner;
+          },
+        };
+      }
+
+      function isRunning(marker: string): boolean {
+        const commands = execSync("ps -axww -o command").toString();
+        return commands
+          .split("\n")
+          .some((line) => line.includes(marker) && !line.includes("ps -axww"));
+      }
+
+      it("never leaves the child process behind, wherever the halt lands", function* () {
+        // The scheduler interleaves same-generation routines one
+        // instruction per turn, so a halter spawned one scope deeper than
+        // the exec task runs in lockstep with the acquisition sequence.
+        // Sweeping the number of turns before the halt lands it on every
+        // suspension point of that sequence — including between the OS
+        // spawn and the registration of the kill-teardown (#236).
+        for (let turns = 0; turns < 40; turns++) {
+          const marker = `halt-sweep-236-${process.pid}-${turns}`;
+          const task = yield* spawn(function* () {
+            yield* exec("node", {
+              arguments: ["-e", "setTimeout(() => {}, 30000)", marker],
+            });
+            yield* suspend();
+          });
+          const halter = yield* spawn(() =>
+            deep(1, function* () {
+              for (let t = 0; t < turns; t++) {
+                yield* useScope();
+              }
+              yield* task.halt();
+            }),
+          );
+          yield* halter;
+          if (isRunning(marker)) {
+            try {
+              execSync(`pkill -f ${marker}`);
+            } catch (_error) {}
+            throw new Error(
+              `process orphaned when halted after ${turns} scheduler turns`,
+            );
+          }
+        }
+      }, 30000);
+    });
+  }
 });