From 0b53789de9ad226e4b89998eb0662d3a4e501d2c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 22:01:05 -0700 Subject: [PATCH 1/6] test(testkit): share reactor stubs, signal spawn, and process waits --- e2e/crash-finalize.test.ts | 71 ++++++---------- e2e/exec-shutdown-reap.test.ts | 71 +++++----------- e2e/signal-helpers.ts | 39 ++++++--- fixtures/exec-shutdown-reap/simulate-reap.ts | 13 +-- src/agent/director.test.ts | 71 +++++----------- src/agent/tool-execution-watchdog.test.ts | 37 ++++----- src/director-workflows.test.ts | 37 +++------ src/plugins/shell-guard-plugin.test.ts | 45 ++--------- src/session/cache-ttl-resume.test.ts | 47 ++++------- src/session/sent-messages.test.ts | 29 ++++--- src/session/stream-journal.test.ts | 80 +++++++++++-------- src/subagent/director-test-harness.ts | 31 ------- src/subagent/nudge-director.test.ts | 56 ++++++------- .../run-ask-director-continue.test.ts | 4 +- src/subagent/run-continuation-refresh.test.ts | 10 +-- testkit/process-tag.ts | 43 ++++++++++ testkit/reactor-stubs.ts | 13 ++- testkit/temporary-dirs.ts | 24 +++++- 18 files changed, 322 insertions(+), 399 deletions(-) delete mode 100644 src/subagent/director-test-harness.ts create mode 100644 testkit/process-tag.ts diff --git a/e2e/crash-finalize.test.ts b/e2e/crash-finalize.test.ts index db5cb3e05..777f180b5 100644 --- a/e2e/crash-finalize.test.ts +++ b/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"); diff --git a/e2e/exec-shutdown-reap.test.ts b/e2e/exec-shutdown-reap.test.ts index e49703326..f7ed141f9 100644 --- a/e2e/exec-shutdown-reap.test.ts +++ b/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): Promise { - 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 { - 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, diff --git a/e2e/signal-helpers.ts b/e2e/signal-helpers.ts index 7d0e211a6..765f3dc9a 100644 --- a/e2e/signal-helpers.ts +++ b/e2e/signal-helpers.ts @@ -25,14 +25,14 @@ function spawnFixtureProc( fixture: string, cwd: string, home: string, - sessionId: string, + sessionEnv: Record, ) { 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 + | ((dirs: { readonly cwd: string; readonly home: string }) => { + readonly [key: string]: string; + }); } /** @@ -67,12 +80,20 @@ export async function spawnSignalFixture( options: SpawnSignalFixtureOptions, ): Promise { 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, diff --git a/fixtures/exec-shutdown-reap/simulate-reap.ts b/fixtures/exec-shutdown-reap/simulate-reap.ts index 7ea6f4e63..6840d46dc 100644 --- a/fixtures/exec-shutdown-reap/simulate-reap.ts +++ b/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 { - 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"); diff --git a/src/agent/director.test.ts b/src/agent/director.test.ts index 1957a4329..b91742912 100644 --- a/src/agent/director.test.ts +++ b/src/agent/director.test.ts @@ -31,6 +31,7 @@ import { } from "../inference-abort.js"; import { stubReactorCapabilities, + stubReactorCapabilitiesWith, stubReactorState, stubTextTurnEvent, } from "../../testkit/reactor-stubs.js"; @@ -41,33 +42,6 @@ import { const mockState: ReactorState = { turns: [] } as unknown as ReactorState; -function makeCapabilities(): ReactorCapabilities { - return { - infer: (options) => - ({ - type: "infer", - ...(options !== undefined ? { options } : {}), - }) as ReactorAction, - executeTools: (calls, parallel, addToHistory) => - ({ - type: "execute_tools", - calls, - parallel, - addToHistory, - }) as ReactorAction, - suspend: (gate) => ({ type: "suspend", gate }) as ReactorAction, - fork: (mode, forkId) => ({ type: "fork", mode, forkId }) as ReactorAction, - emit: (eventType, data) => - ({ type: "emit", eventType, data }) as ReactorAction, - reply: (content) => ({ type: "reply", content }) as ReactorAction, - checkpoint: (message = "") => - ({ type: "checkpoint", message }) as ReactorAction, - compact: (compactor, reason) => - ({ type: "compact", compactor, reason }) as ReactorAction, - wait: () => ({ type: "wait" }) as ReactorAction, - done: () => ({ type: "done" }) as ReactorAction, - }; -} // Varied arguments per call so the fingerprint changes turn to turn, like // real tool-only orchestration. @@ -182,7 +156,7 @@ describe("ChatDirector credential recovery continuation", () => { test("consumes one matching armed continuation and rejects stale or repeated delivery", async () => { const director = createChatDirector("system", [], {}); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; expect( actionsArray( @@ -216,7 +190,7 @@ describe("ChatDirector tool-only loop protection", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; // Default family nudges at 25 consecutive tool-only turns. const actions = await runToolOnlyStreak(director, capabilities, 25); @@ -229,7 +203,7 @@ describe("ChatDirector tool-only loop protection", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; await runToolOnlyStreak(director, capabilities, 25); const nextTurn = actionsArray( @@ -246,7 +220,7 @@ describe("ChatDirector tool-only loop protection", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const actions = await runToolOnlyStreak(director, capabilities, 50); expect( @@ -281,7 +255,7 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const actions = actionsArray( await director.decide( @@ -301,7 +275,7 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const internalAbort = inferenceErrorEvent("aborted", { origin: "internal-recovery", }); @@ -332,7 +306,7 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { const director = createChatDirector("system", [], { provider: providerlessPolicy, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const internalAbort = inferenceErrorEvent("aborted", { origin: "internal-recovery", }); @@ -371,7 +345,7 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { } as unknown as WorkflowCoordinator; const director = createChatDirector("system", [], {}); director.setWorkflowCoordinator(throwingCoordinator); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const manageTasksTurn = { type: "inference.done", @@ -522,7 +496,7 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { } as unknown as WorkflowCoordinator; const director = createChatDirector("system", [], {}); director.setWorkflowCoordinator(coordinator); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; // currentStepId runs only past a terminal base action, so it throws // on a text turn; the earlier rails throw on a manage_tasks turn. @@ -545,13 +519,12 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { test("a non-coordinator mid-turn failure preserves queued task-change emits for the next turn", async () => { const failure = new Error("tool execution exploded"); const director = createChatDirector("system", [], {}); - const capabilities = makeCapabilities(); - const executing: ReactorCapabilities = { - ...capabilities, + const capabilities = stubReactorCapabilities; + const executing = stubReactorCapabilitiesWith({ executeTools: (): ReactorAction => { throw failure; }, - }; + }); await expect( director.decide(keeperManageTasksTurn(), mockState, executing), @@ -579,13 +552,12 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => { } as unknown as WorkflowCoordinator; const director = createChatDirector("system", [], {}); director.setWorkflowCoordinator(coordinator); - const capabilities = makeCapabilities(); - const executing: ReactorCapabilities = { - ...capabilities, + const capabilities = stubReactorCapabilities; + const executing = stubReactorCapabilitiesWith({ executeTools: (): ReactorAction => { throw new Error("tool execution exploded"); }, - }; + }); await expect( director.decide(keeperManageTasksPlusSubmitTurn(), mockState, executing), @@ -680,7 +652,7 @@ describe("ChatDirector live source-id tracking (CL-7973)", () => { const director = createChatDirector("system", [], { provider: { providerName: "test-provider" }, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const policy = await liveRetryPolicy(director, capabilities); // The seed id is not an xAI source, so the capacity error aborts. @@ -715,7 +687,7 @@ describe("ChatDirector live source-id tracking (CL-7973)", () => { const director = createChatDirector("system", [], { provider: { providerName: "test-provider" }, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const policy = await liveRetryPolicy(director, capabilities); expect(await isXaiStamped(policy)).toBe(false); @@ -733,7 +705,7 @@ describe("ChatDirector live source-id tracking (CL-7973)", () => { const director = createChatDirector("system", [], { provider: { providerName: "test-provider" }, }); - const capabilities = makeCapabilities(); + const capabilities = stubReactorCapabilities; const policy = await liveRetryPolicy(director, capabilities); // The harness's call-start snapshot is authoritative over the event's @@ -800,11 +772,10 @@ function messageReceived(content: string): ReactorInboundEvent { } /** Infer stub that carries the options through so tests can inspect them. */ -const capabilitiesWithInferArgs: ReactorCapabilities = { - ...stubReactorCapabilities, +const capabilitiesWithInferArgs = stubReactorCapabilitiesWith({ infer: (opts) => ({ type: "infer", options: opts }) as unknown as ReactorAction, -}; +}); function manageTasksEvent( status: "todo" | "doing" | "done", diff --git a/src/agent/tool-execution-watchdog.test.ts b/src/agent/tool-execution-watchdog.test.ts index bb63c7a0f..85bc39056 100644 --- a/src/agent/tool-execution-watchdog.test.ts +++ b/src/agent/tool-execution-watchdog.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import type { AgentTool } from "@intx/agent"; import { defined } from "../../testkit/defined.js"; +import { settleOrTimeout } from "../../testkit/settle.js"; import { createDynamicToolRunner } from "./dynamic-tool-runner.js"; import { DEFAULT_MCP_TOOL_TIMEOUT_MS, @@ -457,18 +458,14 @@ describe("tool execution watchdog", () => { { salvageGraceMs: TEST_SALVAGE_GRACE_MS, waitForApproval: true }, ); parent.abort(); - const afterAbort = await Promise.race([ + const outcome = await settleOrTimeout( pending, - new Promise((_, reject) => - setTimeout( - () => - reject( - new Error("watchdog did not settle after parent abort + grace"), - ), - TEST_SALVAGE_GRACE_MS + 500, - ), - ), - ]); + TEST_SALVAGE_GRACE_MS + 500, + ); + if (outcome.kind === "timeout") { + throw new Error("watchdog did not settle after parent abort + grace"); + } + const afterAbort = await pending; expect(afterAbort.isError).toBe(true); expect(afterAbort.content).toBe("hang aborted"); }); @@ -501,18 +498,14 @@ describe("tool execution watchdog", () => { { salvageGraceMs: TEST_SALVAGE_GRACE_MS, waitForApproval: true }, ); parent.abort(); - const afterAbort = await Promise.race([ + const outcome = await settleOrTimeout( pending, - new Promise((_, reject) => - setTimeout( - () => - reject( - new Error("watchdog did not settle after parent abort + grace"), - ), - TEST_SALVAGE_GRACE_MS + 500, - ), - ), - ]); + TEST_SALVAGE_GRACE_MS + 500, + ); + if (outcome.kind === "timeout") { + throw new Error("watchdog did not settle after parent abort + grace"); + } + const afterAbort = await pending; expect(afterAbort.isError).toBe(true); expect(afterAbort.content).toBe("wait_agents aborted"); }); diff --git a/src/director-workflows.test.ts b/src/director-workflows.test.ts index e2580d81c..a4f93bc85 100644 --- a/src/director-workflows.test.ts +++ b/src/director-workflows.test.ts @@ -2,11 +2,11 @@ import { test, expect } from "bun:test"; import type { ReactorInboundEvent, ReactorState, - ReactorCapabilities, ReactorAction, TokenUsage, ToolResult, } from "@intx/types/runtime"; +import { stubReactorCapabilities } from "../testkit/reactor-stubs.js"; import { createChatDirector } from "./agent/director.js"; import { WorkflowRuntime } from "./workflows/runtime.js"; import { WorkflowCoordinator } from "./workflows/coordinator.js"; @@ -37,21 +37,6 @@ const state: ReactorState = { sessionId: "test-session", }; -function makeCapabilities(): ReactorCapabilities { - return { - infer: (options) => - options === undefined ? { type: "infer" } : { type: "infer", options }, - executeTools: () => ({ type: "execute_tools", calls: [] }), - suspend: (gate) => ({ type: "suspend", gate }), - fork: (mode, forkId) => ({ type: "fork", mode, forkId }), - emit: (eventType, data) => ({ type: "emit", eventType, data }), - reply: (content) => ({ type: "reply", content }), - checkpoint: (message = "") => ({ type: "checkpoint", message }), - compact: (compactor, reason) => ({ type: "compact", compactor, reason }), - wait: () => ({ type: "wait" }), - done: () => ({ type: "done" }), - }; -} const flow: Workflow = { name: "flow", @@ -126,7 +111,7 @@ test("the active step directive is injected into the inferred system prompt", as content: "go", }, }; - const result = await director.decide(event, state, makeCapabilities()); + const result = await director.decide(event, state, stubReactorCapabilities); const infers = inferActions(result); expect(infers.length).toBeGreaterThan(0); @@ -146,7 +131,7 @@ test("a submit_output tool call with the current step id advances the runtime th const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; const turn: ReactorInboundEvent = { type: "inference.done", @@ -186,7 +171,7 @@ test("a stale submit_output does not skip ahead through the director", async () const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; const turn: ReactorInboundEvent = { type: "inference.done", @@ -325,7 +310,7 @@ test("a content-free workflow turn with open tasks nudges toward submit_output", const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; await director.decide(manageTasksTurn("doing"), state, caps); const result = await director.decide(emptyTurn(), state, caps); @@ -353,7 +338,7 @@ test("open tasks do not defeat the workflow stuck-cutoff after 3 idle turns", as const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; await director.decide(manageTasksTurn("doing"), state, caps); await director.decide(textTurn("text 1"), state, caps); @@ -378,7 +363,7 @@ test("seeded idle-with-fleet with no live helpers still auto-continues a workflo allowIdleWithFleet: true, }); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; const result = await director.decide( textTurn("I reviewed the diff, moving to next step."), @@ -401,7 +386,7 @@ test("live fleet plus three text waits does not declare the workflow stuck", asy }); director.setWorkflowCoordinator(coordinator); director.setAllowIdleWithFleet(true); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; await director.decide(manageTasksTurn("doing"), state, caps); await director.decide(textTurn("text 1"), state, caps); @@ -424,7 +409,7 @@ test("occupancy inbound resets the workflow idle counter", async () => { const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; await director.decide(textTurn("text 1"), state, caps); await director.decide(textTurn("text 2"), state, caps); @@ -450,7 +435,7 @@ test("auto-continuation falls back after 3 consecutive text-only turns", async ( const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; await director.decide(textTurn("text 1"), state, caps); await director.decide(textTurn("text 2"), state, caps); @@ -473,7 +458,7 @@ test("after spacer echo-cap a non-gate workflow step does not empty-settle", asy const coordinator = new WorkflowCoordinator(runtime); const director = createChatDirector("BASE", [], {}); director.setWorkflowCoordinator(coordinator); - const caps = makeCapabilities(); + const caps = stubReactorCapabilities; for (let i = 0; i < 2; i++) { const nudged = await director.decide( diff --git a/src/plugins/shell-guard-plugin.test.ts b/src/plugins/shell-guard-plugin.test.ts index 4f46324ac..c920598f9 100644 --- a/src/plugins/shell-guard-plugin.test.ts +++ b/src/plugins/shell-guard-plugin.test.ts @@ -9,6 +9,8 @@ import { EventEmitter } from "node:events"; import { spawnSync, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; +import { waitUntilGone, waitUntilTagged } from "../../testkit/process-tag.js"; + import { createBackgroundShellRegistry } from "../shell/background-shell.js"; import { createExitRecorder } from "../shell/background-shell-test-harness.js"; import { createShellOutputFeed } from "../session/shell-output-feed.js"; @@ -26,17 +28,6 @@ import { const neverAbort = () => new AbortController().signal; -/** Poll until no process carries `token`; fail instead of asserting on a pid. */ -async function waitUntilGone(token: string): Promise { - const started = Date.now(); - while (Date.now() - started < 5_000) { - const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); - if ((probe.stdout?.trim() ?? "").length === 0) return; - await new Promise((r) => setTimeout(r, 50)); - } - throw new Error(`tagged child still alive after 5s: ${token}`); -} - function toolContentTrimmed(result: ToolResult): string { const content = result.content; return (typeof content === "string" ? content : String(content)).trim(); @@ -142,7 +133,7 @@ describe("runGuardedShell", () => { const token = `ic_guard_orphan_${randomUUID()}`; const cmd = `bash -c 'IC_GUARD_TAG=${token} sleep 600 & IC_GUARD_TAG=${token} exec sleep 600'`; await runGuardedShell({ command: cmd, timeout: 150 }, neverAbort()); - await waitUntilGone(token); + expect(await waitUntilGone(token)).toBe(""); }); test("abort kills grandchildren in a shell pipeline", async () => { @@ -156,7 +147,7 @@ describe("runGuardedShell", () => { ); setTimeout(() => controller.abort(), 80); await expect(promise).rejects.toThrow(/aborted/); - await waitUntilGone(token); + expect(await waitUntilGone(token)).toBe(""); }); }); @@ -964,12 +955,7 @@ describe("shellGuardPlugin", () => { neverAbort(), ); try { - const started = Date.now(); - while (Date.now() - started < 5_000) { - const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); - if ((probe.stdout?.trim() ?? "").length > 0) break; - await new Promise((r) => setTimeout(r, 50)); - } + await waitUntilTagged(token); expect( spawnSync("pgrep", ["-f", token], { encoding: "utf8", @@ -977,7 +963,7 @@ describe("shellGuardPlugin", () => { ).not.toBe(""); expect(plugin.dispose).toBeDefined(); await defined(plugin.dispose)(); - await waitUntilGone(token); + expect(await waitUntilGone(token)).toBe(""); await defined(plugin.dispose)(); await running; } finally { @@ -1000,17 +986,7 @@ describe("shellGuardPlugin", () => { neverAbort(), ); try { - const started = Date.now(); - while (Date.now() - started < 5_000) { - const probe = spawnSync("pgrep", ["-f", token1], { encoding: "utf8" }); - if ((probe.stdout?.trim() ?? "").length > 0) break; - await new Promise((r) => setTimeout(r, 50)); - } - expect( - spawnSync("pgrep", ["-f", token1], { - encoding: "utf8", - }).stdout?.trim() ?? "", - ).not.toBe(""); + await waitUntilTagged(token1); const queued = handler( { @@ -1072,12 +1048,7 @@ describe("shellGuardPlugin", () => { neverAbort(), ); try { - const started = Date.now(); - while (Date.now() - started < 5_000) { - const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); - if ((probe.stdout?.trim() ?? "").length > 0) break; - await new Promise((r) => setTimeout(r, 50)); - } + await waitUntilTagged(token); expect(plugin.dispose).toBeDefined(); const first = defined(plugin.dispose)(); const second = defined(plugin.dispose)(); diff --git a/src/session/cache-ttl-resume.test.ts b/src/session/cache-ttl-resume.test.ts index 1e1c36685..9218e6f32 100644 --- a/src/session/cache-ttl-resume.test.ts +++ b/src/session/cache-ttl-resume.test.ts @@ -1,6 +1,3 @@ -import { mkdtemp, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import { describe, expect, test } from "bun:test"; import { createInboundMessage } from "@intx/mime"; import { createReactor, type ReactorEmittedEvent } from "@intx/inference"; @@ -23,6 +20,8 @@ import { buildZenSource, } from "../config/index.js"; import { resumeCacheWriteSeed } from "../provider/cache-ttl.js"; +import { settleOrTimeout } from "../../testkit/settle.js"; +import { withTempDir } from "../../testkit/temporary-dirs.js"; import { HANDOFF_LATEST_KEY } from "./compaction-handoff.js"; import { createPruningCompactor } from "./compactor.js"; import { createOptimizedContextStore } from "./optimized-context-store.js"; @@ -71,8 +70,7 @@ async function resumeAndInfer(args: { seed: ConversationTurn[]; }> { const seed = seedTurns(); - const dir = await mkdtemp(join(tmpdir(), "cache-ttl-resume-")); - try { + return withTempDir("cache-ttl-resume-", async (dir) => { const store = await createOptimizedContextStore(dir); await store.writeTurns(seed); await store.writeMetadata({ @@ -112,6 +110,10 @@ async function resumeAndInfer(args: { const done = new Promise((resolve) => { resolveDone = resolve; }); + let resolveStarted: (() => void) | undefined; + const startedGate = new Promise((resolve) => { + resolveStarted = resolve; + }); const reactor = createReactor({ sessionId: "cache-ttl-resume", @@ -167,6 +169,9 @@ async function resumeAndInfer(args: { }, onEvent(event) { events.push(event); + if (event.type === "reactor.start") { + resolveStarted?.(); + } if (event.type === COMPACTION_CONTINUATION_EVENT) { reactor.deliver(buildCompactionContinuationMessage()); } @@ -177,24 +182,8 @@ async function resumeAndInfer(args: { }); reactor.start(); - const started = events.some((event) => event.type === "reactor.start"); - if (!started) { - await new Promise((resolve, reject) => { - const timer = setTimeout( - () => reject(new Error("reactor did not start")), - 5_000, - ); - const poll = (): void => { - if (events.some((event) => event.type === "reactor.start")) { - clearTimeout(timer); - resolve(); - return; - } - setTimeout(poll, 10); - }; - poll(); - }); - } + const begin = await settleOrTimeout(startedGate, 5_000); + if (begin.kind === "timeout") throw new Error("reactor did not start"); reactor.deliver( createInboundMessage({ from: "user@local", @@ -202,12 +191,8 @@ async function resumeAndInfer(args: { content: "continue the task", }), ); - await Promise.race([ - done, - new Promise((_, reject) => { - setTimeout(() => reject(new Error("resume infer timed out")), 15_000); - }), - ]); + const outcome = await settleOrTimeout(done, 15_000); + if (outcome.kind === "timeout") throw new Error("resume infer timed out"); reactor.abort("admin_kill"); if (first === undefined) throw new Error("director never decided"); if (prompt === undefined || storedAtInfer === undefined) { @@ -216,9 +201,7 @@ async function resumeAndInfer(args: { ); } return { first, stored: storedAtInfer, prompt, handoff, seed }; - } finally { - await rm(dir, { recursive: true, force: true }); - } + }); } describe("resumed Anthropic cache write does not compact", () => { diff --git a/src/session/sent-messages.test.ts b/src/session/sent-messages.test.ts index 3fe606d9f..d083b826b 100644 --- a/src/session/sent-messages.test.ts +++ b/src/session/sent-messages.test.ts @@ -1,25 +1,22 @@ import { afterEach, describe, expect, test } from "bun:test"; -import { mkdtemp, rm } from "node:fs/promises"; -import { join } from "node:path"; -import { tmpdir } from "node:os"; +import { createTempDirs, type TempDirs } from "../../testkit/temporary-dirs.js"; import { generateSessionId, initSessionDir } from "./index.js"; import { appendSentMessage, loadSentMessages } from "./sent-messages.js"; describe("sent-messages", () => { - let cwd: string; - let home: string; - let sessionId: string; + let workspace: TempDirs | undefined; - afterEach(async () => { - if (cwd !== undefined) await rm(cwd, { recursive: true, force: true }); - if (home !== undefined) await rm(home, { recursive: true, force: true }); + afterEach(() => { + workspace?.cleanup(); }); test("append and load per session", async () => { - cwd = await mkdtemp(join(tmpdir(), "sent-msg-")); - home = await mkdtemp(join(tmpdir(), "sent-msg-home-")); - sessionId = generateSessionId(); + const { cwd, home } = (workspace = createTempDirs( + "sent-msg-", + "sent-msg-home-", + )); + const sessionId = generateSessionId(); await initSessionDir(cwd, sessionId, home); expect(await loadSentMessages(cwd, sessionId, home)).toEqual([]); @@ -32,9 +29,11 @@ describe("sent-messages", () => { }); test("load keeps only the last 20 messages", async () => { - cwd = await mkdtemp(join(tmpdir(), "sent-msg-")); - home = await mkdtemp(join(tmpdir(), "sent-msg-home-")); - sessionId = generateSessionId(); + const { cwd, home } = (workspace = createTempDirs( + "sent-msg-", + "sent-msg-home-", + )); + const sessionId = generateSessionId(); await initSessionDir(cwd, sessionId, home); for (let i = 0; i < 25; i++) { await appendSentMessage(cwd, sessionId, `msg-${i}`, home); diff --git a/src/session/stream-journal.test.ts b/src/session/stream-journal.test.ts index 7ff65d9d9..0e6513d29 100644 --- a/src/session/stream-journal.test.ts +++ b/src/session/stream-journal.test.ts @@ -1,10 +1,10 @@ -import { mkdtemp, readFile, rm } from "node:fs/promises"; -import { tmpdir } from "node:os"; +import { readFile } from "node:fs/promises"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import type { ReactorEmittedEvent } from "@intx/inference"; +import { createTempDir } from "../../testkit/temporary-dirs.js"; import { appendCycleText, createCycleTextRecorder, @@ -13,13 +13,16 @@ import { } from "./stream-journal.js"; let dir: string; +let cleanupDir: () => void = () => undefined; -beforeEach(async () => { - dir = await mkdtemp(join(tmpdir(), "stream-journal-")); +beforeEach(() => { + const temp = createTempDir("stream-journal-"); + dir = temp.dir; + cleanupDir = temp.cleanup; }); -afterEach(async () => { - await rm(dir, { recursive: true, force: true }); +afterEach(() => { + cleanupDir(); }); function delta(token: string): ReactorEmittedEvent { @@ -29,6 +32,23 @@ function delta(token: string): ReactorEmittedEvent { } as unknown as ReactorEmittedEvent; } +/** + * Bounded settle gate for the error-path flush: `handleEvent(inference.error)` + * fires the write off without exposing its promise, so wait for the partial + * file to exist instead of sleeping a fixed interval. + */ +async function settlePartialWritten(timeoutMs = 5_000): Promise { + const started = Date.now(); + for (;;) { + try { + return await readFile(join(dir, PARTIAL_FILE), "utf8"); + } catch (err) { + if (Date.now() - started >= timeoutMs) throw err; + await Bun.sleep(1); + } + } +} + function thinkingDelta(token: string): ReactorEmittedEvent { return { type: "inference.thinking.delta", @@ -36,29 +56,23 @@ function thinkingDelta(token: string): ReactorEmittedEvent { } as unknown as ReactorEmittedEvent; } -async function readPartialRecords(): Promise< - { - reason: string; - text: string; - thinkingText?: string; - thinkingChars?: number; - error?: { category?: string; message?: string; statusCode?: number }; - }[] -> { - const raw = await readFile(join(dir, PARTIAL_FILE), "utf8"); +type PartialRecord = { + reason: string; + text: string; + thinkingText?: string; + thinkingChars?: number; + error?: { category?: string; message?: string; statusCode?: number }; +}; + +function parsePartialRaw(raw: string): PartialRecord[] { return raw .trim() .split("\n") - .map( - (line) => - JSON.parse(line) as { - reason: string; - text: string; - thinkingText?: string; - thinkingChars?: number; - error?: { category?: string; message?: string; statusCode?: number }; - }, - ); + .map((line) => JSON.parse(line) as PartialRecord); +} + +async function readPartialRecords(): Promise { + return parsePartialRaw(await readFile(join(dir, PARTIAL_FILE), "utf8")); } describe("createCycleTextRecorder", () => { @@ -102,8 +116,8 @@ describe("createCycleTextRecorder", () => { data: { error: { category: "aborted", message: "aborted" } }, } as unknown as ReactorEmittedEvent); - await Bun.sleep(20); - const records = await readPartialRecords(); + const raw = await settlePartialWritten(); + const records = parsePartialRaw(raw); expect(records[0]?.reason).toBe("inference-error"); expect(records[0]?.text).toBe("partial before failure"); expect(records[0]?.error?.category).toBe("aborted"); @@ -125,8 +139,8 @@ describe("createCycleTextRecorder", () => { }, } as unknown as ReactorEmittedEvent); - await Bun.sleep(20); - const records = await readPartialRecords(); + const raw = await settlePartialWritten(); + const records = parsePartialRaw(raw); expect(records).toHaveLength(1); expect(records[0]?.reason).toBe("inference-error"); expect(records[0]?.text).toBe(""); @@ -174,6 +188,8 @@ describe("createCycleTextRecorder", () => { }); const disposePromise = recorder.dispose("cancelled", { drain }); + // Explicit real wait: the drain is still pending, so no partial write + // may have started -- absence over a settle window is the assertion. await Bun.sleep(10); await expect(readFile(join(dir, PARTIAL_FILE), "utf8")).rejects.toThrow(); @@ -244,8 +260,8 @@ describe("createCycleTextRecorder", () => { type: "inference.error", data: { error: { category: "aborted", message: "aborted" } }, } as unknown as ReactorEmittedEvent); - await Bun.sleep(20); - + // A closed recorder drops events synchronously, so the aborted reason + // must not append a second partial -- assert the existing record directly. const records = await readPartialRecords(); expect(records).toHaveLength(1); expect(records[0]?.reason).toBe("interrupted"); diff --git a/src/subagent/director-test-harness.ts b/src/subagent/director-test-harness.ts deleted file mode 100644 index 85c4f96f9..000000000 --- a/src/subagent/director-test-harness.ts +++ /dev/null @@ -1,31 +0,0 @@ -import type { ReactorAction, ReactorCapabilities } from "@intx/types/runtime"; - -/** Shared ReactorCapabilities stub: builders return protocol-shaped actions; - * tests assert on decide() results only. */ -export function createTestCapabilities(): ReactorCapabilities { - return { - infer: (options) => - ({ - type: "infer", - ...(options !== undefined ? { options } : {}), - }) as ReactorAction, - executeTools: (calls, parallel, addToHistory) => - ({ - type: "execute_tools", - calls, - parallel, - addToHistory, - }) as ReactorAction, - suspend: (gate) => ({ type: "suspend", gate }) as ReactorAction, - fork: (mode, forkId) => ({ type: "fork", mode, forkId }) as ReactorAction, - emit: (eventType, data) => - ({ type: "emit", eventType, data }) as ReactorAction, - reply: (content) => ({ type: "reply", content }) as ReactorAction, - checkpoint: (message = "") => - ({ type: "checkpoint", message }) as ReactorAction, - compact: (compactor, reason) => - ({ type: "compact", compactor, reason }) as ReactorAction, - wait: () => ({ type: "wait" }) as ReactorAction, - done: () => ({ type: "done" }) as ReactorAction, - }; -} diff --git a/src/subagent/nudge-director.test.ts b/src/subagent/nudge-director.test.ts index 1cddaa774..38d238255 100644 --- a/src/subagent/nudge-director.test.ts +++ b/src/subagent/nudge-director.test.ts @@ -9,7 +9,10 @@ import { createCorbitsRetryPolicy } from "../agent/retry-policy.js"; import { COMPACT_NOOP_FLOOR } from "../session/compactor.js"; import { SubAgentDirector } from "./nudge-director.js"; import type { AdmissionQueue } from "./admission.js"; -import { createTestCapabilities } from "./director-test-harness.js"; +import { + stubReactorCapabilities, + stubReactorCapabilitiesWith, +} from "../../testkit/reactor-stubs.js"; import { defined } from "../../testkit/defined.js"; import { PASS_PLAN_ENVELOPE, @@ -35,7 +38,7 @@ function makeDirector(): { caps: ReactorCapabilities; } { const director = new SubAgentDirector("system", [], undefined, 30); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; return { director, caps }; } @@ -281,7 +284,7 @@ function stallDirector(now0: number): { 1_000, () => now, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; return { director, caps, @@ -438,7 +441,7 @@ describe("SubAgentDirector tool failure recovery", () => { 30, frozenNow, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide( inferenceDone(["failed-at-threshold"], 999_999), @@ -481,8 +484,7 @@ describe("SubAgentDirector tool failure recovery", () => { test("recovery nudge appends to ephemeral turns already on the infer", async () => { const { director, caps } = makeDirector(); // Seed an ephemeral turn only when the caller supplied none. - const seeding: ReactorCapabilities = { - ...caps, + const seeding = stubReactorCapabilitiesWith({ infer: (options) => { const existing = (options as { ephemeralTurns?: unknown[] } | undefined) ?.ephemeralTurns; @@ -501,7 +503,7 @@ describe("SubAgentDirector tool failure recovery", () => { : {}), }); }, - }; + }); await director.decide(inferenceDone(["seeded-fail"]), state, seeding); const texts = ephemeralTexts( @@ -560,7 +562,7 @@ describe("SubAgentDirector tool failure recovery", () => { 30, frozenNow, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDone(["failed-then-overflow"]), state, caps); const texts = ephemeralTexts( @@ -604,7 +606,7 @@ describe("SubAgentDirector tool failure recovery", () => { 30, frozenNow, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDone(["failed-then-done"]), state, caps); const recovered = ephemeralTexts( @@ -861,7 +863,7 @@ describe("SubAgentDirector incomplete-report wiring", () => { 30, frozenNow, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const records: { id: string; class: string }[] = []; director.observeInterventions((event) => { records.push({ id: event.id, class: event.class }); @@ -973,7 +975,7 @@ describe("SubAgentDirector plan-substance wiring", () => { false, true, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const result = actions( await director.decide(inferenceDoneText(STUB_PLAN_ENVELOPE), state, caps), @@ -1005,7 +1007,7 @@ describe("SubAgentDirector plan-substance wiring", () => { false, true, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDoneText(STUB_PLAN_ENVELOPE), state, caps); @@ -1032,7 +1034,7 @@ describe("SubAgentDirector plan-substance wiring", () => { false, true, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const result = actions( await director.decide(inferenceDoneText(PASS_PLAN_ENVELOPE), state, caps), @@ -1054,7 +1056,7 @@ describe("SubAgentDirector plan-substance wiring", () => { false, true, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const wrapPlan = [ "## Summary", "Plan after reading the gate.", @@ -1094,7 +1096,7 @@ describe("SubAgentDirector post-complete terminalization (CL-7068)", () => { 1000, () => now, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await readOnce(director, state, caps); await director.decide(inferenceDoneText(REPORT_ENVELOPE), state, caps); @@ -1112,7 +1114,7 @@ describe("SubAgentDirector post-complete terminalization (CL-7068)", () => { test("a non-empty parent follow-up re-opens inference after a report reply", async () => { const director = new SubAgentDirector("system", [], undefined, 1000); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await readOnce(director, state, caps); await director.decide(inferenceDoneText(REPORT_ENVELOPE), state, caps); @@ -1130,7 +1132,7 @@ describe("SubAgentDirector post-complete terminalization (CL-7068)", () => { test("empty continuation after incomplete-report-stop salvage waits instead of re-inferring", async () => { const director = new SubAgentDirector("system", [], undefined, 1000); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const salvage = await narratingSalvage(director, state, caps); expect(salvage).toContainEqual({ @@ -1152,7 +1154,7 @@ describe("SubAgentDirector post-complete terminalization (CL-7068)", () => { }, 1000, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; // Under-threshold tooling so tool.done does not compact before the report. await readOnce(director, longState, caps); @@ -1195,7 +1197,7 @@ describe("SubAgentDirector post-complete terminalization (CL-7068)", () => { test("repeated empty continuations after a report reply keep waiting", async () => { const director = new SubAgentDirector("system", [], undefined, 1000); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await readOnce(director, state, caps); await director.decide(inferenceDoneText(REPORT_ENVELOPE), state, caps); @@ -1335,7 +1337,7 @@ describe("SubAgentDirector ask_director park wait-guard", () => { () => now, ); director.observeAskPending(() => parked); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDoneText("which file?"), state, caps); now += 60_000; @@ -1362,7 +1364,7 @@ describe("SubAgentDirector ask_director park wait-guard", () => { frozenNow, ); director.observeAskPending(() => parked); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const compact = actions( await director.decide(overflowError(), state, caps), @@ -1401,7 +1403,7 @@ describe("SubAgentDirector ask_director park wait-guard", () => { () => now, ); director.observeAskPending(() => parked); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDoneText("which file?"), state, caps); now += 60_000; @@ -1478,7 +1480,7 @@ describe("SubAgentDirector idle stall ping", () => { 60_000, () => now, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const compact = actions( await director.decide(overflowError(), state, caps), @@ -1512,7 +1514,7 @@ describe("SubAgentDirector idle stall ping", () => { 60_000, () => now, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await readOnce(director, longState, caps); const complete = actions( @@ -1557,7 +1559,7 @@ describe("SubAgentDirector idle stall ping", () => { stallTimeoutMs, () => now, ); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDoneText("working"), longState, caps); @@ -1585,7 +1587,7 @@ describe("SubAgentDirector idle stall ping", () => { test("no stall timeout waits on an unsolicited empty continuation", async () => { const director = new SubAgentDirector("system", [], undefined); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await director.decide(inferenceDoneText("working"), state, caps); const ping = actions( @@ -1633,7 +1635,7 @@ describe("SubAgentDirector infer retryPolicy", () => { await director.decide( messageReceived("go"), state, - createTestCapabilities(), + stubReactorCapabilities, ), ); const stamped = infer.options?.retryPolicy; diff --git a/src/subagent/run-ask-director-continue.test.ts b/src/subagent/run-ask-director-continue.test.ts index f942e9821..2b6806f3a 100644 --- a/src/subagent/run-ask-director-continue.test.ts +++ b/src/subagent/run-ask-director-continue.test.ts @@ -17,7 +17,7 @@ import { createPermissionGate } from "../permission/gate.js"; import type { RunSubAgentParams } from "./types.js"; import type { AskDirectorState } from "./ask-director.js"; import { SubAgentDirector } from "./nudge-director.js"; -import { createTestCapabilities } from "./director-test-harness.js"; +import { stubReactorCapabilities } from "../../testkit/reactor-stubs.js"; import { DIRECTOR_RESOLVABLE_ESCALATION } from "../../testkit/escalation.js"; const testPermissionGate = createPermissionGate({ @@ -235,7 +235,7 @@ describe("runSubAgent ask_director compact-continue wiring", () => { await director.decide( emptyContinuation(), state, - createTestCapabilities(), + stubReactorCapabilities, ), ); expect(parkedPing).toEqual([{ type: "wait" }]); diff --git a/src/subagent/run-continuation-refresh.test.ts b/src/subagent/run-continuation-refresh.test.ts index cffec981e..0bbee4108 100644 --- a/src/subagent/run-continuation-refresh.test.ts +++ b/src/subagent/run-continuation-refresh.test.ts @@ -15,7 +15,7 @@ import { peekSourceCredentialSecret, registerSourceCredentialRecord, } from "../config/source-credentials.js"; -import { createTestCapabilities } from "./director-test-harness.js"; +import { stubReactorCapabilities } from "../../testkit/reactor-stubs.js"; import { SubAgentDirector } from "./nudge-director.js"; import type { ContinuationRefreshOptions } from "./refresh-inference-source.js"; import { withContinuationOAuthRefresh } from "./refresh-inference-source.js"; @@ -120,7 +120,7 @@ describe("run continuation OAuth refresh", () => { retryPolicyStub, ); const wrapped = withContinuationOAuthRefresh(inner, options); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; return { getValid, setSourcesCalls, wrapped, caps, source }; } @@ -175,7 +175,7 @@ describe("run continuation OAuth refresh", () => { defaultSource: "codex/default", catalog: [], }); - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; await wrapped.decide(inferenceDone(["call-1"]), state, caps); const result = await wrapped.decide(toolDone("call-1"), state, caps); expect(actions(result).some((action) => action.type === "infer")).toBe( @@ -215,7 +215,7 @@ describe("run continuation OAuth refresh", () => { }; const wrapped = withContinuationOAuthRefresh(fakeInner, options); expect(withContinuationOAuthRefresh(wrapped, options)).toBe(wrapped); - await wrapped.decide(toolDone("call-1"), state, createTestCapabilities()); + await wrapped.decide(toolDone("call-1"), state, stubReactorCapabilities); expect(getValid).toHaveBeenCalledTimes(1); expect(setSources).toBe(1); }); @@ -251,7 +251,7 @@ describe("run continuation OAuth refresh", () => { capabilities: ReactorCapabilities, ) => capabilities.infer(), }; - const caps = createTestCapabilities(); + const caps = stubReactorCapabilities; const first = withContinuationOAuthRefresh(fakeInner, options); const second = withContinuationOAuthRefresh(fakeInner, options); const pending = [ diff --git a/testkit/process-tag.ts b/testkit/process-tag.ts new file mode 100644 index 000000000..234c373e7 --- /dev/null +++ b/testkit/process-tag.ts @@ -0,0 +1,43 @@ +import { spawnSync } from "node:child_process"; + +/** + * Bounded pgrep waits for tests that spawn tagged children and must observe + * their real process-tree lifecycle. The pgrep loop is the only way to see a + * process the plugin spawned outside the test's hands; keep the waits + * explicit here instead of re-rolling per test file. + */ + +function taggedPids(token: string): string { + const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); + return probe.stdout?.trim() ?? ""; +} + +/** Resolves once any process matches `token`; rejects after `timeoutMs`. */ +export async function waitUntilTagged( + token: string, + timeoutMs = 5_000, + intervalMs = 50, +): Promise { + const started = Date.now(); + while (Date.now() - started < timeoutMs) { + if (taggedPids(token).length > 0) return; + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + throw new Error(`tagged child did not appear after ${timeoutMs}ms: ${token}`); +} + +/** Resolves once no process matches `token`; returns the last observed ids. */ +export async function waitUntilGone( + token: string, + timeoutMs = 5_000, + intervalMs = 50, +): Promise { + const started = Date.now(); + let stdout = "unprobed"; + while (Date.now() - started < timeoutMs) { + stdout = taggedPids(token); + if (stdout.length === 0) return stdout; + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + return stdout; +} diff --git a/testkit/reactor-stubs.ts b/testkit/reactor-stubs.ts index 2c41d25bb..722ceeffd 100644 --- a/testkit/reactor-stubs.ts +++ b/testkit/reactor-stubs.ts @@ -13,14 +13,16 @@ import type { /** Empty reactor state for decides that never read it. */ export const stubReactorState = {} as unknown as ReactorState; -/** Pass-through capabilities: every method returns its action literal. */ +/** Pass-through capabilities: every method returns its action literal. + * Per-test overrides go through `stubReactorCapabilitiesWith`. */ export const stubReactorCapabilities: ReactorCapabilities = { infer: (options) => ({ type: "infer", ...(options !== undefined ? { options } : {}), }) as ReactorAction, - executeTools: (calls) => ({ type: "execute_tools", calls }), + executeTools: (calls, parallel, addToHistory) => + ({ type: "execute_tools", calls, parallel, addToHistory }) as ReactorAction, suspend: (gate) => ({ type: "suspend", gate }), fork: (mode, forkId) => ({ type: "fork", mode, forkId }), emit: (eventType, data) => ({ type: "emit", eventType, data }), @@ -45,3 +47,10 @@ export function stubTextTurnEvent(): ReactorInboundEvent { source: { model: "test-model" }, } as unknown as ReactorInboundEvent; } + +/** Canonical stub with per-test overrides for the methods a test observes. */ +export function stubReactorCapabilitiesWith( + overrides: Partial, +): ReactorCapabilities { + return { ...stubReactorCapabilities, ...overrides }; +} diff --git a/testkit/temporary-dirs.ts b/testkit/temporary-dirs.ts index a099deaa9..6b250c391 100644 --- a/testkit/temporary-dirs.ts +++ b/testkit/temporary-dirs.ts @@ -30,14 +30,30 @@ export function createTempDirs( }; } +export interface TempDir { + readonly dir: string; + cleanup(): void; +} + +/** Creates one temp dir with `cleanup` removal, for files under test. */ +export function createTempDir(prefix: string): TempDir { + const dir = mkdtempSync(join(tmpdir(), prefix)); + return { + dir, + cleanup(): void { + rmSync(dir, { recursive: true, force: true }); + }, + }; +} + /** Runs `body` inside a fresh temp dir, always removing it afterwards. */ -export async function withTempDir( +export async function withTempDir( prefix: string, - body: (dir: string) => Promise, -): Promise { + body: (dir: string) => Promise, +): Promise { const dir = await mkdtemp(join(tmpdir(), prefix)); try { - await body(dir); + return await body(dir); } finally { await rm(dir, { recursive: true, force: true }); } From 3642c220c09037a4210180bd99a408f47d945f4a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 22:56:13 -0700 Subject: [PATCH 2/6] style(test): format shared test stubs --- src/agent/director.test.ts | 1 - src/agent/tool-execution-watchdog.test.ts | 10 ++-------- src/director-workflows.test.ts | 1 - 3 files changed, 2 insertions(+), 10 deletions(-) diff --git a/src/agent/director.test.ts b/src/agent/director.test.ts index b91742912..040f21159 100644 --- a/src/agent/director.test.ts +++ b/src/agent/director.test.ts @@ -42,7 +42,6 @@ import { const mockState: ReactorState = { turns: [] } as unknown as ReactorState; - // Varied arguments per call so the fingerprint changes turn to turn, like // real tool-only orchestration. function toolOnlyTurn(id: string): ReactorInboundEvent { diff --git a/src/agent/tool-execution-watchdog.test.ts b/src/agent/tool-execution-watchdog.test.ts index 85bc39056..595d5c9c8 100644 --- a/src/agent/tool-execution-watchdog.test.ts +++ b/src/agent/tool-execution-watchdog.test.ts @@ -458,10 +458,7 @@ describe("tool execution watchdog", () => { { salvageGraceMs: TEST_SALVAGE_GRACE_MS, waitForApproval: true }, ); parent.abort(); - const outcome = await settleOrTimeout( - pending, - TEST_SALVAGE_GRACE_MS + 500, - ); + const outcome = await settleOrTimeout(pending, TEST_SALVAGE_GRACE_MS + 500); if (outcome.kind === "timeout") { throw new Error("watchdog did not settle after parent abort + grace"); } @@ -498,10 +495,7 @@ describe("tool execution watchdog", () => { { salvageGraceMs: TEST_SALVAGE_GRACE_MS, waitForApproval: true }, ); parent.abort(); - const outcome = await settleOrTimeout( - pending, - TEST_SALVAGE_GRACE_MS + 500, - ); + const outcome = await settleOrTimeout(pending, TEST_SALVAGE_GRACE_MS + 500); if (outcome.kind === "timeout") { throw new Error("watchdog did not settle after parent abort + grace"); } diff --git a/src/director-workflows.test.ts b/src/director-workflows.test.ts index a4f93bc85..ed06dd0d7 100644 --- a/src/director-workflows.test.ts +++ b/src/director-workflows.test.ts @@ -37,7 +37,6 @@ const state: ReactorState = { sessionId: "test-session", }; - const flow: Workflow = { name: "flow", description: "two steps", From ad429fa1518b7c5d1aa24ddf3ba9291a46c829d0 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 22:57:56 -0700 Subject: [PATCH 3/6] fix(hooks): stringify lifecycle hook signal codes Bun's signalCode is number | Signals | null. Keep the real signal as a string so HookExitStatus stays string | null without dropping numeric codes. --- src/session/hooks.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/session/hooks.ts b/src/session/hooks.ts index 9106da956..027116822 100644 --- a/src/session/hooks.ts +++ b/src/session/hooks.ts @@ -412,7 +412,8 @@ async function runLifecycleHook( ]); return { code: exitCode, - signal: proc.signalCode, + signal: + proc.signalCode === null ? null : String(proc.signalCode), stderr: delivered ? stderr : `${stderr}${stderr.length > 0 && !stderr.endsWith("\n") ? "\n" : ""}hook exited without reading its payload\n`, From 2d52b004011fff7b5247a1892ad295100ec2def5 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 23:12:30 -0700 Subject: [PATCH 4/6] style(hooks): format signalCode stringification oxfmt --check rejected the wrapped signal field in runLifecycleHook. --- src/session/hooks.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/session/hooks.ts b/src/session/hooks.ts index 027116822..7eec5f8fe 100644 --- a/src/session/hooks.ts +++ b/src/session/hooks.ts @@ -412,8 +412,7 @@ async function runLifecycleHook( ]); return { code: exitCode, - signal: - proc.signalCode === null ? null : String(proc.signalCode), + signal: proc.signalCode === null ? null : String(proc.signalCode), stderr: delivered ? stderr : `${stderr}${stderr.length > 0 && !stderr.endsWith("\n") ? "\n" : ""}hook exited without reading its payload\n`, From dfe0c870bda6a982ef2bb7f08a0b9b8472af8ac0 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 15 Oct 2026 12:06:28 -0700 Subject: [PATCH 5/6] style(e2e): drop unused spawnSync import and prefer Record Static analysis rejected an unused spawnSync import in the reap fixture and an index signature on spawnSignalFixture extraEnv. --- e2e/signal-helpers.ts | 7 ++++--- fixtures/exec-shutdown-reap/simulate-reap.ts | 1 - 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/e2e/signal-helpers.ts b/e2e/signal-helpers.ts index 765f3dc9a..adf6f7577 100644 --- a/e2e/signal-helpers.ts +++ b/e2e/signal-helpers.ts @@ -65,9 +65,10 @@ export interface SpawnSignalFixtureOptions { */ readonly extraEnv?: | Record - | ((dirs: { readonly cwd: string; readonly home: string }) => { - readonly [key: string]: string; - }); + | ((dirs: { readonly cwd: string; readonly home: string }) => Record< + string, + string + >); } /** diff --git a/fixtures/exec-shutdown-reap/simulate-reap.ts b/fixtures/exec-shutdown-reap/simulate-reap.ts index 6840d46dc..f016698f0 100644 --- a/fixtures/exec-shutdown-reap/simulate-reap.ts +++ b/fixtures/exec-shutdown-reap/simulate-reap.ts @@ -2,7 +2,6 @@ // sleep through the real shell-guard plugin, registers exec dispose as the // process dispose host, then takes the requested exit path so the parent can // assert the child was reaped. -import { spawnSync } from "node:child_process"; import { writeFileSync } from "node:fs"; import type { ToolCall, ToolResult } from "@intx/types/runtime"; From e692d99d2c18b2150cb87416bb050bb177cb6123 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 15 Oct 2026 12:15:58 -0700 Subject: [PATCH 6/6] style(e2e): oxfmt signal-helpers extraEnv union --- e2e/signal-helpers.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/e2e/signal-helpers.ts b/e2e/signal-helpers.ts index adf6f7577..c23188bf5 100644 --- a/e2e/signal-helpers.ts +++ b/e2e/signal-helpers.ts @@ -65,10 +65,10 @@ export interface SpawnSignalFixtureOptions { */ readonly extraEnv?: | Record - | ((dirs: { readonly cwd: string; readonly home: string }) => Record< - string, - string - >); + | ((dirs: { + readonly cwd: string; + readonly home: string; + }) => Record); } /**