diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index 9ec4adf4..24dd8132 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -74,12 +74,37 @@ const KILL_GRACE_MS = 3000; // Windows IPC may spend 5s connecting, 2s cancelling, 2s settling, // and up to 5s releasing the entire batch of caller-owned transfers. const WINDOWS_KILL_GRACE_MS = 15_000; +// A killed child that never reports `exit` at all must still release the caller +// once the forced kill has had its chance. +const SETTLE_AFTER_KILL_SLACK_MS = 1000; +// `close` fires only once every stdio pipe has reached EOF, which needs every +// process holding a copy of the pipe handles to be gone, not just `bsk`. When +// `bsk` auto-spawns the daemon, the daemon can end up holding those handles +// (Windows `CreateProcess` inherits every inheritable handle; issue #180), so +// after `exit` we drain what the pipes still give us and then settle, rather +// than waiting for a `close` that may never come. `close` normally follows +// `exit` within the same loop turn, so the wait is only ever paid when +// something else is holding the pipes. Bytes arriving inside the window can +// still be the child's own buffered output, so every chunk restarts the window +// and EXIT_DRAIN_MAX_MS caps the total wait. +const EXIT_DRAIN_GRACE_MS = 250; +const EXIT_DRAIN_MAX_MS = 2000; const SESSION_BUSY_RETRY_DELAY_MS = 100; +/** One in-flight child plus the bounded shutdown that settles its run. */ +interface LiveRun { + tag: string | undefined; + requestKill: () => void; +} + export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner { - const live = new Map(); + const live = new Map(); const windows = process.platform === "win32"; const cancelling = new Set(); + const killGraceMs = windows ? WINDOWS_KILL_GRACE_MS : KILL_GRACE_MS; + // The kill grace and the settlement deadline stay in step: a Windows + // cancellation using its full 15s must not be cut short by a 4s fallback. + const settleAfterKillMs = killGraceMs + SETTLE_AFTER_KILL_SLACK_MS; function killChild(child: ChildProcess): void { if (child.exitCode !== null || child.signalCode !== null || cancelling.has(child)) return; @@ -88,12 +113,9 @@ export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): // send its existing cancel RPC and wait for browser reconciliation. if (windows && child.stdin) child.stdin.end(); else child.kill("SIGINT"); - const force = setTimeout( - () => { - if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); - }, - windows ? WINDOWS_KILL_GRACE_MS : KILL_GRACE_MS, - ); + const force = setTimeout(() => { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + }, killGraceMs); force.unref(); child.once("close", () => { clearTimeout(force); @@ -132,32 +154,93 @@ export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): reject(error); return; } - live.set(child, options.tag); let stdout = ""; let stderr = ""; let timedOut = false; let aborted = false; - child.stdout?.on("data", (chunk: Buffer | string) => { + const onStdout = (chunk: Buffer | string) => { stdout += chunk; - }); - child.stderr?.on("data", (chunk: Buffer | string) => { + extendDrain(); + }; + const onStderr = (chunk: Buffer | string) => { stderr += chunk; - }); + extendDrain(); + }; + child.stdout?.on("data", onStdout); + child.stderr?.on("data", onStderr); + + let settled = false; + let deadline: ReturnType | undefined; + let drainWindow: ReturnType | undefined; + let drainCap: ReturnType | undefined; + let drainCode: number | null = null; + // Dropping the listeners stops the collection, but our ends of the pipes + // stay open and keep the event loop referenced. When a grandchild holds + // the other ends, that would keep the host alive long after the run has + // settled, so close them and stop waiting on the child itself. + const release = () => { + child.stdout?.off("data", onStdout); + child.stderr?.off("data", onStderr); + child.stdout?.destroy(); + child.stderr?.destroy(); + child.stdin?.destroy(); + child.unref(); + }; + const finish = (code: number | null) => { + if (settled) return; + settled = true; + settle(); + release(); + resolve({ code, stdout, stderr, timedOut, aborted }); + }; + // Wait for `close` after a normal `exit`, but not forever. Output landing + // in the window can still be the child's own buffered bytes, so each + // chunk reopens it for another EXIT_DRAIN_GRACE_MS and the cap keeps the + // total bounded when whatever holds the pipes keeps writing. + const extendDrain = () => { + if (drainCap === undefined || settled) return; + if (drainWindow !== undefined) clearTimeout(drainWindow); + drainWindow = setTimeout(() => finish(drainCode), EXIT_DRAIN_GRACE_MS); + drainWindow.unref(); + }; + const beginDrain = (code: number | null) => { + if (drainCap !== undefined) return; + drainCode = code; + drainCap = setTimeout(() => finish(code), EXIT_DRAIN_MAX_MS); + drainCap.unref(); + extendDrain(); + }; + // Kill on our own initiative, then guarantee the promise settles even if + // the child never reports back: `exit` normally arrives promptly, and the + // deadline covers a child that reports nothing at all after SIGKILL. + const requestKill = () => { + if (child.exitCode !== null || child.signalCode !== null) { + // The process is already gone; only pipes held by a grandchild remain. + finish(child.exitCode); + return; + } + killChild(child); + if (deadline === undefined) { + deadline = setTimeout(() => finish(child.exitCode), settleAfterKillMs); + deadline.unref(); + } + }; + live.set(child, { tag: options.tag, requestKill }); const timeoutMs = options.timeoutMs; const timer = timeoutMs !== undefined && timeoutMs > 0 ? setTimeout(() => { timedOut = true; - killChild(child); + requestKill(); }, timeoutMs) : undefined; timer?.unref(); const onAbort = () => { aborted = true; - killChild(child); + requestKill(); }; if (options.signal?.aborted) { onAbort(); @@ -167,28 +250,41 @@ export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): const settle = () => { if (timer !== undefined) clearTimeout(timer); + if (deadline !== undefined) clearTimeout(deadline); + if (drainWindow !== undefined) clearTimeout(drainWindow); + if (drainCap !== undefined) clearTimeout(drainCap); options.signal?.removeEventListener("abort", onAbort); live.delete(child); }; child.on("error", (error) => { + if (settled) return; + settled = true; settle(); + release(); reject(error); }); - child.on("close", (code) => { - settle(); - resolve({ code, stdout, stderr, timedOut, aborted }); + child.on("close", (code) => finish(code)); + child.on("exit", (code, signal) => { + if (signal !== null || timedOut || aborted) { + // Killed on our initiative: nothing left worth draining. + finish(code); + return; + } + // Normal exit: keep draining while bytes are still arriving, then + // settle even if a grandchild is still holding the pipes open. + beginDrain(code); }); }); }, killAll() { - for (const child of live.keys()) killChild(child); + for (const run of live.values()) run.requestKill(); }, killFor(tag: string) { let killed = 0; - for (const [child, childTag] of live) { - if (childTag === tag) { - killChild(child); + for (const run of live.values()) { + if (run.tag === tag) { + run.requestKill(); killed += 1; } } diff --git a/packages/dsh-plugin-browserskill/tests/runner.test.ts b/packages/dsh-plugin-browserskill/tests/runner.test.ts index 219f2909..38201215 100644 --- a/packages/dsh-plugin-browserskill/tests/runner.test.ts +++ b/packages/dsh-plugin-browserskill/tests/runner.test.ts @@ -1,5 +1,8 @@ -import type { ChildProcess } from "node:child_process"; +import { type ChildProcess, spawn } from "node:child_process"; import { EventEmitter } from "node:events"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { PassThrough } from "node:stream"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -11,13 +14,27 @@ import { runWithSessionBusyRetry, } from "../src/runner"; +/** Minimal fake stdio pipe recording whether the runner closed its end. */ +class FakeStream extends EventEmitter { + destroyed = false; + + destroy(): void { + this.destroyed = true; + } +} + /** Minimal fake ChildProcess driven by the test. */ class FakeChild extends EventEmitter { - stdout = new EventEmitter(); - stderr = new EventEmitter(); + stdout = new FakeStream(); + stderr = new FakeStream(); exitCode: number | null = null; signalCode: string | null = null; killedWith: string[] = []; + unrefs = 0; + + unref(): void { + this.unrefs += 1; + } kill(signal: string): boolean { if (this.exitCode !== null || this.signalCode !== null) return false; @@ -43,6 +60,10 @@ function fakeSpawn(children: FakeChild[]) { }; } +afterEach(() => { + vi.useRealTimers(); +}); + describe("createBskRunner", () => { it("appends --json and collects stdout", async () => { const child = new FakeChild(); @@ -82,6 +103,184 @@ describe("createBskRunner", () => { expect(child.killedWith.length).toBeGreaterThan(0); }); + it("settles on timeout even when close never fires", async () => { + // When something else still holds the stdio pipes (issue #180: the daemon + // `bsk` auto-spawned), `exit` fires but `close` never follows. + const child = new FakeChild(); + child.kill = (signal: string) => { + if (child.exitCode !== null || child.signalCode !== null) return false; + child.killedWith.push(signal); + child.signalCode = signal; + queueMicrotask(() => child.emit("exit", null, signal)); + return true; + }; + const runner = createBskRunner("bsk", fakeSpawn([child])); + const result = await Promise.race([ + runner.run(["session", "start"], { timeoutMs: 5 }), + new Promise((_, reject) => + setTimeout(() => reject(new Error("run() never settled")), 2_000), + ), + ]); + expect(result).toMatchObject({ code: null, timedOut: true }); + expect(child.killedWith).toContain("SIGINT"); + }); + + it("settles on abort even when close never fires", async () => { + const child = new FakeChild(); + child.kill = (signal: string) => { + child.killedWith.push(signal); + child.signalCode = signal; + queueMicrotask(() => child.emit("exit", null, signal)); + return true; + }; + const runner = createBskRunner("bsk", fakeSpawn([child])); + const controller = new AbortController(); + const promise = runner.run(["snapshot"], { signal: controller.signal }); + controller.abort(); + const result = await Promise.race([ + promise, + new Promise((_, reject) => + setTimeout(() => reject(new Error("run() never settled")), 2_000), + ), + ]); + expect(result).toMatchObject({ code: null, aborted: true }); + }); + + it("settles after the kill grace when neither exit nor close ever fires", async () => { + // The kill lands but the child reports nothing back at all: no `exit`, no + // `close`. The caller must still be released. + vi.useFakeTimers(); + const child = new FakeChild(); + child.kill = (signal: string) => { + // The signal is delivered but the process never dies, so Node never + // populates signalCode and SIGKILL must escalate after the grace period. + child.killedWith.push(signal); + return true; // no exit, no close, ever + }; + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["session", "start"], { timeoutMs: 100 }).then((r) => { + result = r; + }); + await vi.advanceTimersByTimeAsync(100); // timeout fires -> SIGINT + expect(result).toBeUndefined(); + await vi.advanceTimersByTimeAsync(4_000); // SIGKILL at +3s, settle at +4s + expect(result).toMatchObject({ code: null, timedOut: true }); + expect(child.killedWith).toEqual(["SIGINT", "SIGKILL"]); + // Settling is not enough: our ends of the pipes have to go too, or a process + // still holding the other ends keeps the host alive. + expect([child.stdout.destroyed, child.stderr.destroyed]).toEqual([true, true]); + expect(child.unrefs).toBe(1); + }); + + it("returns the output after a normal exit even when close never fires", async () => { + // The shape of issue #180: `bsk session start` prints its JSON and exits, but + // the daemon it auto-spawned still holds the stdio pipes, so `close` never fires. + vi.useFakeTimers(); + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["session", "start"], { timeoutMs: 120_000 }).then((r) => { + result = r; + }); + child.stdout.emit("data", '{"session":"dfhj"}'); + child.exitCode = 0; + child.emit("exit", 0, null); // process gone; pipes still held open + await vi.advanceTimersByTimeAsync(200); + expect(result).toBeUndefined(); // still inside the drain grace + await vi.advanceTimersByTimeAsync(100); + expect(result).toMatchObject({ code: 0, stdout: '{"session":"dfhj"}', timedOut: false }); + }); + + it("closes its ends of the pipes after settling without close", async () => { + vi.useFakeTimers(); + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["session", "start"]).then((r) => { + result = r; + }); + child.stdout.emit("data", '{"session":"dfhj"}'); + child.exitCode = 0; + child.emit("exit", 0, null); // process gone; a grandchild still holds the pipes + await vi.advanceTimersByTimeAsync(300); + expect(result).toMatchObject({ code: 0 }); + expect([child.stdout.destroyed, child.stderr.destroyed]).toEqual([true, true]); + expect(child.unrefs).toBe(1); + }); + + it("keeps draining while the exited child's buffered output still arrives", async () => { + // Bytes that land after `exit` are not necessarily someone else's: they can + // be the child's own output, still buffered in the pipe. Each chunk reopens + // the window so a delayed tail is not cut off. + vi.useFakeTimers(); + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["session", "start"], { timeoutMs: 120_000 }).then((r) => { + result = r; + }); + child.exitCode = 0; + child.emit("exit", 0, null); + child.stdout.emit("data", '{"session":'); + await vi.advanceTimersByTimeAsync(200); + child.stdout.emit("data", '"dfhj"}'); // the tail, 200ms after the head + await vi.advanceTimersByTimeAsync(200); + expect(result).toBeUndefined(); // a fixed 250ms deadline would have cut here + await vi.advanceTimersByTimeAsync(100); // 250ms of silence closes the window + expect(result).toMatchObject({ code: 0, stdout: '{"session":"dfhj"}' }); + }); + + it("settles at the drain cap when output keeps arriving after exit", async () => { + // The other side of the boundary: output that never stops must not hold the + // caller forever, so the cap ends the drain and collection stops with it. + vi.useFakeTimers(); + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["session", "start"]).then((r) => { + result = r; + }); + child.exitCode = 0; + child.emit("exit", 0, null); + // 200ms apart, so every chunk reopens the 250ms window; the last one before + // the 2s cap lands at 1.8s. + for (let elapsed = 0; elapsed < 2_000; elapsed += 200) { + child.stdout.emit("data", "x"); + await vi.advanceTimersByTimeAsync(200); + } + expect(result).toMatchObject({ code: 0, stdout: "x".repeat(10) }); + child.stdout.emit("data", "later"); + expect(result?.stdout).toBe("x".repeat(10)); + }); + + it("settles immediately on timeout when the process already exited", async () => { + vi.useFakeTimers(); + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + let result: BskRunResult | undefined; + void runner.run(["snapshot"], { timeoutMs: 100 }).then((r) => { + result = r; + }); + child.exitCode = 0; + child.emit("exit", 0, null); + await vi.advanceTimersByTimeAsync(100); // timeout lands before the drain grace ends + expect(result).toMatchObject({ code: 0, timedOut: true }); + expect(child.killedWith).toEqual([]); // nothing to kill + }); + + it("still waits for close on a normal exit so stdout is fully drained", async () => { + const child = new FakeChild(); + const runner = createBskRunner("bsk", fakeSpawn([child])); + const promise = runner.run(["session", "list"]); + child.exitCode = 0; + child.emit("exit", 0, null); // process gone, pipes still open + child.stdout.emit("data", '{"late":true}'); + child.emit("close", 0); // now the pipes shut + const result = await promise; + expect(result).toMatchObject({ code: 0, stdout: '{"late":true}' }); + }); + it("killAll terminates in-flight children", async () => { const child = new FakeChild(); const runner = createBskRunner("bsk", fakeSpawn([child])); @@ -91,6 +290,108 @@ describe("createBskRunner", () => { expect(child.killedWith).toContain("SIGINT"); expect(result.code).toBeNull(); }); + + it("settles killAll and killFor children that never report back", async () => { + // Both kill entry points take the same bounded path as a timeout, so a child + // that answers neither the signal nor the pipes still releases its caller. + vi.useFakeTimers(); + const children = [new FakeChild(), new FakeChild()]; + for (const child of children) { + child.kill = (signal: string) => { + child.killedWith.push(signal); + return true; // no exit, no close, ever + }; + } + const runner = createBskRunner("bsk", fakeSpawn([...children])); + const results: (BskRunResult | undefined)[] = [undefined, undefined]; + void runner.run(["snapshot"], { tag: "s1" }).then((r) => { + results[0] = r; + }); + void runner.run(["snapshot"]).then((r) => { + results[1] = r; + }); + expect(runner.killFor("s1")).toBe(1); + runner.killAll(); + await vi.advanceTimersByTimeAsync(4_000); // SIGKILL at +3s, settle at +4s + expect(results[0]).toMatchObject({ code: null }); + expect(results[1]).toMatchObject({ code: null }); + expect(children.map((c) => c.stdout.destroyed)).toEqual([true, true]); + }); +}); + +describe("createBskRunner against real processes", () => { + // Node 22.18 and later strip types by default; earlier 22.x needs the flag to + // load the runner's own TypeScript source in the host process below. + const typeStripping = process.allowedNodeEnvironmentFlags.has("--experimental-strip-types") + ? ["--experimental-strip-types"] + : []; + const runnerUrl = new URL("../src/runner.ts", import.meta.url).href; + // Stands in for `bsk`: prints its JSON and exits, leaving a detached grandchild + // that inherited the stdio pipes and holds them open (issue #180), so the + // parent's `close` never fires. + const fakeBsk = [ + 'import { spawn } from "node:child_process";', + 'const grandchild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 10000)"], {', + " detached: true,", + ' stdio: ["ignore", "inherit", "inherit"],', + "});", + "grandchild.unref();", + "process.stdout.write(JSON.stringify({ ok: true, grandchild: grandchild.pid }));", + ].join("\n"); + const hostSource = (bskPath: string) => + [ + `import { createBskRunner } from ${JSON.stringify(runnerUrl)};`, + "const runner = createBskRunner(process.execPath);", + `const result = await runner.run([${JSON.stringify(bskPath)}]);`, + "process.stdout.write(JSON.stringify({ code: result.code, stdout: result.stdout }));", + ].join("\n"); + + it("settles and lets the host exit while a grandchild still holds the pipes", async () => { + const dir = await mkdtemp(join(tmpdir(), "bsk-runner-")); + let grandchild: number | undefined; + try { + const bskPath = join(dir, "bsk.mjs"); + await writeFile(bskPath, fakeBsk); + const hostPath = join(dir, "host.mts"); + await writeFile(hostPath, hostSource(bskPath)); + // A separate process, because the claim is about the host staying alive: + // it must run the command and then exit on its own. + const host = spawn(process.execPath, [...typeStripping, "--no-warnings", hostPath], { + stdio: ["ignore", "pipe", "pipe"], + }); + let out = ""; + let err = ""; + host.stdout.on("data", (chunk) => { + out += chunk; + }); + host.stderr.on("data", (chunk) => { + err += chunk; + }); + const guard = setTimeout(() => host.kill("SIGKILL"), 10_000); + const code = await new Promise((resolve) => host.on("close", resolve)); + clearTimeout(guard); + expect(err).toBe(""); + expect(code).toBe(0); // null here means the guard had to kill a live host + const settled = JSON.parse(out) as { code: number | null; stdout: string }; + expect(settled.code).toBe(0); + const reply = JSON.parse(settled.stdout) as { ok: boolean; grandchild: number }; + expect(reply.ok).toBe(true); + const pid = reply.grandchild; + grandchild = pid; + // Still running, so it still held the pipes: the host exited without + // waiting for the `close` that the grandchild was suppressing. + expect(() => process.kill(pid, 0)).not.toThrow(); + } finally { + if (grandchild !== undefined) { + try { + process.kill(grandchild, "SIGKILL"); + } catch { + // already gone + } + } + await rm(dir, { recursive: true, force: true }); + } + }, 20_000); }); describe("Windows parent cancellation", () => { diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 7faf3fa5..ab018e3a 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -1,3 +1,5 @@ +import type { ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; import { existsSync, mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -6,7 +8,12 @@ import { describe, expect, it, vi } from "vitest"; import { registerBrowserTools } from "../src/browser-tools"; import { ObservationService } from "../src/observation"; import { KeyedExecutor } from "../src/queue"; -import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; +import { + type BskRunner, + type BskRunOptions, + type BskRunResult, + createBskRunner, +} from "../src/runner"; import { SessionRegistry } from "../src/sessions"; import type { PluginConfig } from "../src/tools"; @@ -336,6 +343,53 @@ describe("session.start", () => { expect(calls.some((c) => c.args.join(" ") === "session stop s1")).toBe(true); expect(registry.current()).toBeUndefined(); }); + + it("returns the session when bsk exits but something still holds its stdio pipes", async () => { + // Issue #180: `bsk session start` printed its JSON and exited, but the daemon + // it auto-spawned kept the stdio pipes open, so the child's `close` never + // fired and the tool call hung past its own timeout. Drive the real runner + // with a child that emits `exit` and never `close`. + const pipe = () => Object.assign(new EventEmitter(), { destroy: () => {} }); + const child = Object.assign(new EventEmitter(), { + stdout: pipe(), + stderr: pipe(), + exitCode: null as number | null, + signalCode: null as string | null, + kill: () => false, + unref: () => {}, + }); + let spawned!: () => void; + const spawnedPromise = new Promise((resolve) => { + spawned = resolve; + }); + const runner = createBskRunner("bsk", () => { + spawned(); + return child as unknown as ChildProcess; + }); + const { ctx, tools } = makeCtx(); + const registry = new SessionRegistry(5); + registerBrowserTools({ + ctx: ctx as never, + runner, + registry, + config: CONFIG, + observation: disabledObservation({ ctx, runner, registry }), + queue: new KeyedExecutor(), + }); + const pending = startSession(tools); + await spawnedPromise; + child.stdout.emit("data", JSON.stringify(START_REPLY("s1"))); + child.exitCode = 0; + child.emit("exit", 0, null); // process gone; pipes still held, so no `close` + const value = await Promise.race([ + pending, + new Promise((_, reject) => + setTimeout(() => reject(new Error("browser_session start never returned")), 2_000), + ), + ]); + expect(value.sessionId).toBe("s1"); + expect(registry.current()).toBe("s1"); + }); }); describe("multi-session behavior", () => {