test(testkit): share reactor stubs, signal spawn, and process waits

Author: TheGreatAxios

e2e/crash-finalize.test.ts

@@ -5,7 +5,7 @@ import { describe, expect, test } from "bun:test";
 
 import { generateSessionId, sessionDir } from "../src/session/index.js";
 import type { RunState } from "../src/session/state.js";
-import { createTempDirs } from "../testkit/temporary-dirs.js";
+import { spawnSignalFixture } from "./signal-helpers.js";
 
 const FIXTURE = join(
   import.meta.dirname,
@@ -18,28 +18,23 @@ const RUN_END_FIXTURE = join(
 
 describe("integration — crash finalizes run.json", () => {
   test("uncaughtException writes status: crashed with finishedAt, racing in-flight snapshot writes", async () => {
-    const { cwd, home, cleanup } = createTempDirs(
-      "corbits-crash-cwd-",
-      "corbits-crash-home-",
-    );
     const sessionId = generateSessionId();
+    const { proc, output, cleanup } = await spawnSignalFixture({
+      fixture: FIXTURE,
+      cwdPrefix: "corbits-crash-cwd-",
+      homePrefix: "corbits-crash-home-",
+      sessionId,
+      sessionIdEnv: "CRASH_TEST_SESSION_ID",
+    });
 
     try {
-      const proc = Bun.spawn(["bun", "run", FIXTURE], {
-        cwd,
-        env: { ...process.env, HOME: home, CRASH_TEST_SESSION_ID: sessionId },
-        stdout: "pipe",
-        stderr: "pipe",
-      });
-
       const exitCode = await proc.exited;
-      const stdout = await new Response(proc.stdout).text();
       const stderr = await new Response(proc.stderr).text();
 
       expect(exitCode).toBe(1);
       expect(stderr).toContain("uncaughtException: Error: simulated crash");
 
-      const runJsonPath = join(stdout.trim(), "run.json");
+      const runJsonPath = join(output.trim(), "run.json");
       const raw = readFileSync(runJsonPath, "utf8");
       const state = JSON.parse(raw) as RunState;
 
@@ -57,28 +52,19 @@ describe("integration — crash finalizes run.json", () => {
   }, 15_000);
 
   test("a crash after session rotation still writes crashed for the new session", async () => {
-    const { cwd, home, cleanup } = createTempDirs(
-      "corbits-crash-cwd-",
-      "corbits-crash-home-",
-    );
     const sessionId = generateSessionId();
     const rotatedSessionId = generateSessionId();
+    const { proc, cwd, home, output, cleanup } = await spawnSignalFixture({
+      fixture: FIXTURE,
+      cwdPrefix: "corbits-crash-cwd-",
+      homePrefix: "corbits-crash-home-",
+      sessionId,
+      sessionIdEnv: "CRASH_TEST_SESSION_ID",
+      extraEnv: { CRASH_TEST_ROTATED_SESSION_ID: rotatedSessionId },
+    });
 
     try {
-      const proc = Bun.spawn(["bun", "run", FIXTURE], {
-        cwd,
-        env: {
-          ...process.env,
-          HOME: home,
-          CRASH_TEST_SESSION_ID: sessionId,
-          CRASH_TEST_ROTATED_SESSION_ID: rotatedSessionId,
-        },
-        stdout: "pipe",
-        stderr: "pipe",
-      });
-
       const exitCode = await proc.exited;
-      const stdout = await new Response(proc.stdout).text();
       const stderr = await new Response(proc.stderr).text();
 
       expect(exitCode).toBe(1);
@@ -95,7 +81,7 @@ describe("integration — crash finalizes run.json", () => {
       expect(outgoingState.status).toBe("done");
       expect(outgoingState.turnsUsed).toBe(3);
 
-      const rotatedRunJsonPath = join(stdout.trim(), "run.json");
+      const rotatedRunJsonPath = join(output.trim(), "run.json");
       const rotatedState = JSON.parse(
         readFileSync(rotatedRunJsonPath, "utf8"),
       ) as RunState;
@@ -112,22 +98,17 @@ describe("integration — crash finalizes run.json", () => {
   }, 15_000);
 
   test("an unrelated crash while the run-end write is in flight does not report crashed", async () => {
-    const { cwd, home, cleanup } = createTempDirs(
-      "corbits-crash-cwd-",
-      "corbits-crash-home-",
-    );
     const sessionId = generateSessionId();
+    const { proc, output, cleanup } = await spawnSignalFixture({
+      fixture: RUN_END_FIXTURE,
+      cwdPrefix: "corbits-crash-cwd-",
+      homePrefix: "corbits-crash-home-",
+      sessionId,
+      sessionIdEnv: "RUN_END_TEST_SESSION_ID",
+    });
 
     try {
-      const proc = Bun.spawn(["bun", "run", RUN_END_FIXTURE], {
-        cwd,
-        env: { ...process.env, HOME: home, RUN_END_TEST_SESSION_ID: sessionId },
-        stdout: "pipe",
-        stderr: "pipe",
-      });
-
       const exitCode = await proc.exited;
-      const stdout = await new Response(proc.stdout).text();
       const stderr = await new Response(proc.stderr).text();
 
       expect(exitCode).toBe(1);
@@ -139,7 +120,7 @@ describe("integration — crash finalizes run.json", () => {
       // saveState write resolved; the crash handler then saw a live run and
       // wrote "crashed" via saveCrashState, clobbering the clean finish.
       // Clearing the handle before the await closes that window.
-      const runJsonPath = join(stdout.trim(), "run.json");
+      const runJsonPath = join(output.trim(), "run.json");
       const raw = readFileSync(runJsonPath, "utf8");
       const state = JSON.parse(raw) as RunState;
       expect(state.status).not.toBe("crashed");

e2e/exec-shutdown-reap.test.ts

@@ -1,41 +1,18 @@
 import { spawnSync } from "node:child_process";
-import { mkdtempSync, readFileSync, rmSync } from "node:fs";
-import { tmpdir } from "node:os";
+import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import { randomUUID } from "node:crypto";
 
 import { describe, expect, test } from "bun:test";
 
+import { waitUntilGone } from "../testkit/process-tag.js";
+import { spawnSignalFixture } from "./signal-helpers.js";
+
 const FIXTURE = join(
   import.meta.dirname,
   "../fixtures/exec-shutdown-reap/simulate-reap.ts",
 );
 
-async function readLine(stream: ReadableStream<Uint8Array>): Promise<string> {
-  const reader = stream.getReader();
-  const decoder = new TextDecoder();
-  let buffer = "";
-  while (!buffer.includes("\n")) {
-    const { value, done } = await reader.read();
-    if (done) break;
-    buffer += decoder.decode(value, { stream: true });
-  }
-  reader.releaseLock();
-  return buffer;
-}
-
-async function waitUntilGone(token: string): Promise<string> {
-  const started = Date.now();
-  let stdout = "";
-  while (Date.now() - started < 5_000) {
-    const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" });
-    stdout = probe.stdout?.trim() ?? "";
-    if (stdout.length === 0) return stdout;
-    await new Promise((resolve) => setTimeout(resolve, 50));
-  }
-  return stdout;
-}
-
 describe.skipIf(process.platform === "win32")(
   "integration — exec shutdown reaps shell-guard children",
   () => {
@@ -48,32 +25,28 @@ describe.skipIf(process.platform === "win32")(
     ] as const)(
       "%s %s reaps the tagged child and disposes once",
       async (path, signal, expectedExitCode) => {
-        const home = mkdtempSync(join(tmpdir(), "corbits-exec-reap-home-"));
         const token = `ic_reap_${randomUUID()}`;
-        const countPath = join(home, "dispose-count");
-
-        try {
-          const proc = Bun.spawn(["bun", "run", FIXTURE], {
-            cwd: home,
-            env: {
-              ...process.env,
-              HOME: home,
+        let countPath: string | undefined;
+        const { proc, output, cleanup } = await spawnSignalFixture({
+          fixture: FIXTURE,
+          cwdPrefix: "corbits-exec-reap-cwd-",
+          homePrefix: "corbits-exec-reap-home-",
+          extraEnv: ({ home }) => {
+            countPath = join(home, "dispose-count");
+            return {
               REAP_TOKEN: token,
               REAP_PATH: path,
               REAP_COUNT_PATH: countPath,
-            },
-            stdout: "pipe",
-            stderr: "pipe",
-          });
+            };
+          },
+        });
 
-          const ready = await readLine(proc.stdout);
-          if (!ready.includes("READY")) {
-            const errText = await new Response(proc.stderr).text();
-            throw new Error(
-              `fixture did not report READY: ${JSON.stringify(ready)} stderr=${errText}`,
-            );
-          }
+        const resolvedCountPath = countPath;
+        if (resolvedCountPath === undefined)
+          throw new Error("fixture env not resolved");
 
+        try {
+          expect(output).toContain("READY");
           if (
             signal === "SIGINT" ||
             signal === "SIGTERM" ||
@@ -86,10 +59,10 @@ describe.skipIf(process.platform === "win32")(
 
           const leftover = await waitUntilGone(token);
           expect(leftover).toBe("");
-          expect(readFileSync(countPath, "utf8").trim()).toBe("1");
+          expect(readFileSync(resolvedCountPath, "utf8").trim()).toBe("1");
         } finally {
           spawnSync("pkill", ["-9", "-f", token]);
-          rmSync(home, { recursive: true, force: true });
+          cleanup();
         }
       },
       15_000,

e2e/signal-helpers.ts

@@ -25,14 +25,14 @@ function spawnFixtureProc(
   fixture: string,
   cwd: string,
   home: string,
-  sessionId: string,
+  sessionEnv: Record<string, string>,
 ) {
   return Bun.spawn(["bun", "run", fixture], {
     cwd,
     env: {
       ...process.env,
       HOME: home,
-      SIGNAL_TEST_SESSION_ID: sessionId,
+      ...sessionEnv,
     },
     stdout: "pipe",
     stderr: "pipe",
@@ -54,7 +54,20 @@ export interface SpawnSignalFixtureOptions {
   readonly fixture: string;
   readonly cwdPrefix: string;
   readonly homePrefix: string;
-  readonly sessionId: string;
+  /** Optional session id, handed to the fixture child in `sessionIdEnv`. */
+  readonly sessionId?: string;
+  /** Env var name carrying `sessionId` (default `SIGNAL_TEST_SESSION_ID`). */
+  readonly sessionIdEnv?: string;
+  /**
+   * Extra child env for fixture variants (rotation ids, reap tokens). As a
+   * function it receives the spawned cwd/home so files can live inside the
+   * isolated pair; HOME is already set to `home`.
+   */
+  readonly extraEnv?:
+    | Record<string, string>
+    | ((dirs: { readonly cwd: string; readonly home: string }) => {
+        readonly [key: string]: string;
+      });
 }
 
 /**
@@ -67,12 +80,20 @@ export async function spawnSignalFixture(
   options: SpawnSignalFixtureOptions,
 ): Promise<SpawnedSignalFixture> {
   const dirs = createTempDirs(options.cwdPrefix, options.homePrefix);
-  const proc = spawnFixtureProc(
-    options.fixture,
-    dirs.cwd,
-    dirs.home,
-    options.sessionId,
-  );
+  const extra =
+    typeof options.extraEnv === "function"
+      ? options.extraEnv({ cwd: dirs.cwd, home: dirs.home })
+      : options.extraEnv;
+  const sessionEnv =
+    options.sessionId === undefined
+      ? {}
+      : {
+          [options.sessionIdEnv ?? "SIGNAL_TEST_SESSION_ID"]: options.sessionId,
+        };
+  const proc = spawnFixtureProc(options.fixture, dirs.cwd, dirs.home, {
+    ...sessionEnv,
+    ...extra,
+  });
   const output = await readLine(proc.stdout);
   return {
     proc,

fixtures/exec-shutdown-reap/simulate-reap.ts

@@ -9,6 +9,7 @@ import type { ToolCall, ToolResult } from "@intx/types/runtime";
 import { disposeExecRuntime } from "../../src/exec/dispose.js";
 import { shellGuardPlugin } from "../../src/plugins/shell-guard-plugin.js";
 import { setActiveDisposeHost } from "../../src/session/active-host.js";
+import { waitUntilTagged } from "../../testkit/process-tag.js";
 
 const token = process.env["REAP_TOKEN"];
 const path = process.env["REAP_PATH"];
@@ -79,17 +80,7 @@ void handler(
   new AbortController().signal,
 );
 
-async function waitForTag(): Promise<void> {
-  const started = Date.now();
-  while (Date.now() - started < 5_000) {
-    const probe = spawnSync("pgrep", ["-f", reapToken], { encoding: "utf8" });
-    if ((probe.stdout?.trim() ?? "").length > 0) return;
-    await new Promise((resolve) => setTimeout(resolve, 50));
-  }
-  throw new Error(`tagged child did not appear: ${reapToken}`);
-}
-
-await waitForTag();
+await waitUntilTagged(reapToken);
 setActiveDisposeHost(host);
 writeFileSync(disposeCountPath, "0");
 process.stdout.write("READY\n");