From bf0aadbc40c25a2d5231f7736633f5fa68ebca5b Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sat, 8 Aug 2026 15:16:44 +0800 Subject: [PATCH] fix(qa): settle Unix commands after PGID cleanup (#120010) Punchcard-Session: cobalt-valley-meadow-mg --- extensions/qa-lab/src/gateway-child.ts | 105 +------ .../scenario-runtime-cli-stream-error.test.ts | 18 +- .../matrix/scenarios/scenario-runtime-cli.ts | 194 ++++-------- .../src/posix-command-settlement.test.ts | 169 +++++++++++ .../qa-lab/src/posix-command-settlement.ts | 267 ++++++++++++++++ .../qa-lab/src/posix-process-group.test.ts | 80 +++++ extensions/qa-lab/src/posix-process-group.ts | 126 ++++++++ .../src/suite-runtime-agent-process.test.ts | 59 +++- .../qa-lab/src/suite-runtime-agent-process.ts | 127 +++++--- ...st-file-scenario-command-lifecycle.test.ts | 271 ++++++++++++++++- .../test-file-scenario-command-lifecycle.ts | 284 +++++------------- 11 files changed, 1206 insertions(+), 494 deletions(-) create mode 100644 extensions/qa-lab/src/posix-command-settlement.test.ts create mode 100644 extensions/qa-lab/src/posix-command-settlement.ts create mode 100644 extensions/qa-lab/src/posix-process-group.test.ts create mode 100644 extensions/qa-lab/src/posix-process-group.ts diff --git a/extensions/qa-lab/src/gateway-child.ts b/extensions/qa-lab/src/gateway-child.ts index 800852965390..71707286d972 100644 --- a/extensions/qa-lab/src/gateway-child.ts +++ b/extensions/qa-lab/src/gateway-child.ts @@ -1,13 +1,7 @@ // Qa Lab plugin module implements gateway child behavior. import { spawn, spawnSync, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { - createWriteStream, - existsSync, - readFileSync, - readdirSync, - type WriteStream, -} from "node:fs"; +import { createWriteStream, existsSync, type WriteStream } from "node:fs"; import fs from "node:fs/promises"; import net from "node:net"; import os from "node:os"; @@ -54,6 +48,11 @@ import { import { startQaGatewayRpcClient } from "./gateway-rpc-client.js"; import { splitQaModelRef, type QaProviderMode } from "./model-selection.js"; import { resolveQaNodeExecPath } from "./node-exec.js"; +import { + inspectLinuxProcessGroup, + inspectLinuxProcessGroupStats, + type QaLinuxProcessGroupInspector, +} from "./posix-process-group.js"; import { readProcessTreeCpuMs, readProcessTreeRssBytes } from "./process-tree-cpu.js"; import { normalizeQaProviderModeEnv, @@ -89,7 +88,6 @@ const QA_GATEWAY_CHILD_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 30_000; // Loaded Docker runners can take several seconds to reap a force-killed process group. const QA_GATEWAY_CHILD_FORCE_SHUTDOWN_TIMEOUT_MS = 10_000; const QA_GATEWAY_LOG_CLOSE_TIMEOUT_MS = 5_000; -const QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS = 2_048; const QA_MOCK_OPENAI_API_KEY = ["qa", "mock", "openai", "key"].join("-"); const QA_GATEWAY_CHILD_BLOCKED_SECRET_ENV_VARS = Object.freeze([ "OPENCLAW_QA_CONVEX_SECRET_CI", @@ -794,98 +792,11 @@ function isProcessAlreadyExitedError(error: unknown): boolean { return (error as NodeJS.ErrnoException | undefined)?.code === "ESRCH"; } -function parseLinuxProcessStat(raw: string) { - const commandStart = raw.indexOf("("); - const commandEnd = raw.lastIndexOf(")"); - if (commandStart <= 0 || commandEnd <= commandStart) { - return null; - } - const pid = Number.parseInt(raw.slice(0, commandStart).trim(), 10); - const fields = raw - .slice(commandEnd + 1) - .trim() - .split(/\s+/u); - const state = fields[0]; - const processGroupId = Number.parseInt(fields[2] ?? "", 10); - if ( - !Number.isSafeInteger(pid) || - pid <= 0 || - !state || - !Number.isSafeInteger(processGroupId) || - processGroupId <= 0 - ) { - return null; - } - return { - command: raw.slice(commandStart + 1, commandEnd), - pid, - processGroupId, - state, - }; -} - function boundQaGatewayProcessTreeDiagnostics(details: string) { - if (details.length <= QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS) { + if (details.length <= 2_048) { return details; } - return `${sliceUtf16Safe(details, 0, QA_GATEWAY_PROCESS_TREE_DIAGNOSTIC_MAX_CHARS - 3)}...`; -} - -function inspectLinuxProcessGroupStats(processGroupId: number, stats: readonly string[]) { - const members = stats - .map((raw) => parseLinuxProcessStat(raw)) - .filter( - (entry): entry is NonNullable> => - entry?.processGroupId === processGroupId, - ) - .toSorted((left, right) => left.pid - right.pid); - const diagnostics = members - .map( - (member) => - `pid=${member.pid} state=${member.state} command=${JSON.stringify(member.command)}`, - ) - .join(", "); - return { - alive: - members.length === 0 - ? null - : members.some((entry) => entry.state !== "Z" && entry.state !== "X"), - diagnostics: boundQaGatewayProcessTreeDiagnostics( - `pgid=${processGroupId} members=[${diagnostics}]`, - ), - }; -} - -type QaLinuxProcessGroupInspection = ReturnType; -type QaLinuxProcessGroupInspector = ( - processGroupId: number, -) => QaLinuxProcessGroupInspection | null; - -function inspectLinuxProcessGroup(processGroupId: number): QaLinuxProcessGroupInspection | null { - if (process.platform !== "linux") { - return null; - } - let entries; - try { - entries = readdirSync("/proc", { withFileTypes: true }); - } catch { - return null; - } - const stats: string[] = []; - for (const entry of entries) { - if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) { - continue; - } - try { - stats.push(readFileSync(path.join("/proc", entry.name, "stat"), "utf8")); - } catch (error) { - // Processes can exit while /proc is being scanned. - if ((error as NodeJS.ErrnoException).code !== "ENOENT") { - return null; - } - } - } - return inspectLinuxProcessGroupStats(processGroupId, stats); + return `${sliceUtf16Safe(details, 0, 2_045)}...`; } function isQaGatewayChildProcessTreeAlive( diff --git a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli-stream-error.test.ts b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli-stream-error.test.ts index e8876e7d842b..c5d3d13eadb6 100644 --- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli-stream-error.test.ts +++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli-stream-error.test.ts @@ -115,9 +115,13 @@ describe("Matrix QA CLI runtime stream errors", () => { childProcessMocks.spawn.mockClear(); }); - it.each(["stdout", "stderr"] as const)( - "rejects after cleaning up when %s emits a stream error", - async (streamName) => { + it.each([ + ["stdout", false], + ["stderr", false], + ["stdout", true], + ] as const)( + "rejects after cleaning up when %s emits a stream error (after exit: %s)", + async (streamName, afterExit) => { const { grandchildPidPath, grandchildReadyPath, root } = await createCliRoot(); let child: ChildProcess | undefined; let grandchildPid: number | undefined; @@ -142,7 +146,13 @@ describe("Matrix QA CLI runtime stream errors", () => { grandchildPid = await waitForPidFile(grandchildPidPath, 2_000); await waitForFile(grandchildReadyPath, 2_000); - child[streamName]?.emit("error", new Error(`${streamName} pipe failed`)); + const streamError = new Error(`${streamName} pipe failed`); + if (afterExit) { + child.once("exit", () => child?.[streamName]?.emit("error", streamError)); + child.kill("SIGTERM"); + } else { + child[streamName]?.emit("error", streamError); + } await expect(session.wait()).rejects.toThrow( `${streamName} stream error: ${streamName} pipe failed`, diff --git a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli.ts b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli.ts index a3482b9b6f6c..e724bf25577a 100644 --- a/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli.ts +++ b/extensions/qa-lab/src/live-transports/matrix/scenarios/scenario-runtime-cli.ts @@ -7,6 +7,7 @@ import { setTimeout as sleep } from "node:timers/promises"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { redactSensitiveText } from "openclaw/plugin-sdk/logging-core"; import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; +import { createQaPosixCommandSettlement } from "../../../posix-command-settlement.js"; import { killMatrixQaCliChild, resolveMatrixQaOpenClawCliEntryPath, @@ -100,20 +101,6 @@ function formatMatrixQaCliTimeoutError(result: MatrixQaCliRunResult, timeoutMs: .join("\n"); } -function isMatrixQaCliChildProcessGroupRunning( - child: ReturnType, -): boolean { - if (process.platform === "win32" || !child.pid) { - return false; - } - try { - process.kill(-child.pid, 0); - return true; - } catch { - return false; - } -} - export function startMatrixQaOpenClawCli(params: { allowNonZero?: boolean; args: string[]; @@ -129,11 +116,6 @@ export function startMatrixQaOpenClawCli(params: { let closed = false; let closeError: Error | undefined; let closeResult: MatrixQaCliRunResult | undefined; - let killRequested = false; - let timedOut = false; - let forceKillTimeout: NodeJS.Timeout | undefined; - let forceSettleTimeout: NodeJS.Timeout | undefined; - let streamFailure: Error | undefined; let settleWait: | { reject: (error: Error) => void; @@ -167,120 +149,67 @@ export function startMatrixQaOpenClawCli(params: { settleWait.resolve(result); } }; - const finishTimeout = (result: MatrixQaCliRunResult) => { - finish(result, new Error(formatMatrixQaCliTimeoutError(result, params.timeoutMs))); - }; - const finishResult = (result: MatrixQaCliRunResult) => { - if (result.exitCode !== 0 && params.allowNonZero !== true) { - finish(result, new Error(formatMatrixQaCliExitError(result))); - return; - } - finish(result); - }; - const clearForcedTimeouts = () => { - if (forceKillTimeout) { - clearTimeout(forceKillTimeout); - forceKillTimeout = undefined; - } - if (forceSettleTimeout) { - clearTimeout(forceSettleTimeout); - forceSettleTimeout = undefined; - } - }; - const finishForcedCleanup = (result: MatrixQaCliRunResult) => { - if (timedOut) { - finishTimeout(result); - return; - } - if (streamFailure) { - finish(result, streamFailure); - return; - } - finishResult(result); - }; - const scheduleForcedCleanup = () => { - if (forceKillTimeout || forceSettleTimeout) { - return; - } - forceKillTimeout = setTimeout(() => { - forceKillTimeout = undefined; - killMatrixQaCliChild(child, "SIGKILL"); - forceSettleTimeout = setTimeout(() => { - forceSettleTimeout = undefined; - finishForcedCleanup( - buildMatrixQaCliResult({ - args: params.args, - exitCode: 1, - output: readOutput(), - }), - ); - }, MATRIX_QA_CLI_TIMEOUT_FORCE_SETTLE_MS); - }, MATRIX_QA_CLI_TIMEOUT_KILL_GRACE_MS); - }; - - const timeout = setTimeout(() => { - timedOut = true; - killMatrixQaCliChild(child, "SIGTERM"); - scheduleForcedCleanup(); - }, params.timeoutMs); - const handleStreamError = (stream: "stderr" | "stdout", error: Error) => { - if (closed || timedOut || killRequested) { - return; - } - clearTimeout(timeout); - killRequested = true; - streamFailure = new Error(`${stream} stream error: ${formatErrorMessage(error)}`, { - cause: error, - }); - // Keep stream failures on the normal kill path so detached descendants are gone - // before the session reports the parent-side pipe failure. - killMatrixQaCliChild(child, "SIGTERM"); - scheduleForcedCleanup(); - }; - - child.stdout.on("data", (chunk) => stdout.push(Buffer.from(chunk))); - child.stdout.on("error", (error) => handleStreamError("stdout", error)); - child.stderr.on("data", (chunk) => stderr.push(Buffer.from(chunk))); - child.stderr.on("error", (error) => handleStreamError("stderr", error)); + const isWindows = process.platform === "win32"; + const settlement = createQaPosixCommandSettlement({ + child, + settlementFailureMessage: `${formatMatrixQaCliCommand(params.args)} settlement failed`, + forceKillAfterMs: MATRIX_QA_CLI_TIMEOUT_KILL_GRACE_MS, + initialSignal: "SIGTERM", + ...(isWindows + ? { + windowsCleanup: { + closeCompletesCleanup: true, + signal: (signal: NodeJS.Signals) => { + try { + killMatrixQaCliChild(child, signal); + return undefined; + } catch (error) { + return error instanceof Error ? error : new Error(String(error)); + } + }, + }, + } + : {}), + executionTimeoutMs: params.timeoutMs, + onSettled: (outcome) => { + const primary = outcome.primary; + const result = buildMatrixQaCliResult({ + args: params.args, + exitCode: primary.type === "exit" ? (primary.exitCode ?? 1) : 1, + output: readOutput(), + }); + const primaryError = + primary.type === "spawn-error" + ? primary.error + : primary.type === "stream-error" + ? new Error(`${primary.stream} stream error: ${formatErrorMessage(primary.error)}`, { + cause: primary.error, + }) + : primary.type === "timeout" + ? new Error(formatMatrixQaCliTimeoutError(result, params.timeoutMs)) + : result.exitCode !== 0 && params.allowNonZero !== true + ? new Error(formatMatrixQaCliExitError(result)) + : undefined; + finish( + result, + outcome.settlementFailure + ? primaryError + ? new AggregateError( + [primaryError, outcome.settlementFailure], + "Matrix QA CLI command and settlement failed", + ) + : outcome.settlementFailure + : primaryError, + ); + }, + onStderrData: (chunk) => stderr.push(Buffer.from(chunk)), + onStdoutData: (chunk) => stdout.push(Buffer.from(chunk)), + processGroupId: isWindows ? undefined : child.pid, + verifyAfterMs: MATRIX_QA_CLI_TIMEOUT_FORCE_SETTLE_MS, + }); if (params.stdin !== undefined) { child.stdin.end(params.stdin); } - child.on("error", (error) => { - if (streamFailure) { - // Forced cleanup owns settlement after a stream failure. Finishing here could - // reject before detached descendants are proven gone. - return; - } - clearTimeout(timeout); - clearForcedTimeouts(); - finish( - buildMatrixQaCliResult({ - args: params.args, - exitCode: 1, - output: readOutput(), - }), - error, - ); - }); - child.on("close", (exitCode) => { - clearTimeout(timeout); - const result = buildMatrixQaCliResult({ - args: params.args, - exitCode: exitCode ?? 1, - output: readOutput(), - }); - if (timedOut || killRequested) { - // A closed parent is not proof that detached, ignored-stdio descendants are gone. - if (isMatrixQaCliChildProcessGroupRunning(child)) { - return; - } - clearForcedTimeouts(); - finishForcedCleanup(result); - return; - } - clearForcedTimeouts(); - finishResult(result); - }); return { args: params.args, @@ -335,10 +264,7 @@ export function startMatrixQaOpenClawCli(params: { }, kill: () => { if (!closed) { - clearTimeout(timeout); - killRequested = true; - killMatrixQaCliChild(child, "SIGTERM"); - scheduleForcedCleanup(); + settlement.requestCleanup(); } }, }; diff --git a/extensions/qa-lab/src/posix-command-settlement.test.ts b/extensions/qa-lab/src/posix-command-settlement.test.ts new file mode 100644 index 000000000000..81cf357d7d4a --- /dev/null +++ b/extensions/qa-lab/src/posix-command-settlement.test.ts @@ -0,0 +1,169 @@ +import type { ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createQaPosixCommandSettlement } from "./posix-command-settlement.js"; + +type TestOutcome = Parameters[0]["onSettled"]>[0]; + +function createChild() { + const child = new EventEmitter() as ChildProcess; + const childKill = vi.fn(() => true); + const stderrDestroy = vi.fn(); + const stdoutDestroy = vi.fn(); + Object.defineProperty(child, "pid", { value: 42 }); + child.stdout = Object.assign(new EventEmitter(), { destroy: stdoutDestroy }) as never; + child.stderr = Object.assign(new EventEmitter(), { destroy: stderrDestroy }) as never; + child.kill = childKill as ChildProcess["kill"]; + return { child, childKill, stderrDestroy, stdoutDestroy }; +} + +describe("POSIX command settlement", () => { + let processGroupAlive: boolean; + let processKill: ReturnType; + + beforeEach(() => { + vi.useFakeTimers(); + processGroupAlive = false; + processKill = vi.spyOn(process, "kill").mockImplementation((pid, signal) => { + if (pid !== -42) { + return true; + } + if (signal === 0 && !processGroupAlive) { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + if (signal === "SIGKILL") { + processGroupAlive = false; + } + return true; + }); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + function start( + params: { + executionTimeoutMs?: number; + onSettled?: (outcome: TestOutcome) => void; + } = {}, + ) { + const childFixture = createChild(); + const { child } = childFixture; + const settled = vi.fn(params.onSettled); + const controller = createQaPosixCommandSettlement({ + child, + settlementFailureMessage: "settlement failed", + executionTimeoutMs: params.executionTimeoutMs, + forceKillAfterMs: 20, + initialSignal: "SIGTERM", + onSettled: settled, + processGroupId: 42, + verifyAfterMs: 10, + }); + return { ...childFixture, controller, settled }; + } + + it("keeps the exit tuple, resets only idle, and disposes listeners once", async () => { + const { child, settled } = start(); + + child.emit("exit", 7, null); + await vi.advanceTimersByTimeAsync(90); + child.stdout?.emit("data", Buffer.from("tail")); + await vi.advanceTimersByTimeAsync(99); + expect(settled).not.toHaveBeenCalled(); + child.emit("close", 7, null); + + expect(settled).toHaveBeenCalledOnce(); + expect(settled).toHaveBeenCalledWith({ + primary: { type: "exit", exitCode: 7, signal: null }, + }); + expect(child.listenerCount("exit")).toBe(0); + expect(child.listenerCount("close")).toBe(0); + child.emit("close", 7, null); + await vi.advanceTimersByTimeAsync(1_000); + expect(settled).toHaveBeenCalledOnce(); + }); + + it("caps active output at one second and destroys readers only after cleanup", async () => { + processGroupAlive = true; + const { child, settled, stderrDestroy, stdoutDestroy } = start(); + child.emit("exit", 0, null); + + for (let index = 0; index < 10; index += 1) { + await vi.advanceTimersByTimeAsync(90); + child.stdout?.emit("data", Buffer.from(String(index))); + } + await vi.advanceTimersByTimeAsync(99); + expect(stdoutDestroy).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(processKill).toHaveBeenCalledWith(-42, "SIGTERM"); + expect(stdoutDestroy).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(20); + expect(processKill).toHaveBeenCalledWith(-42, "SIGKILL"); + await vi.advanceTimersByTimeAsync(10); + + expect(stdoutDestroy).toHaveBeenCalledOnce(); + expect(stderrDestroy).toHaveBeenCalledOnce(); + expect(settled).toHaveBeenCalledWith({ + settlementFailure: expect.objectContaining({ message: "stdio-drain-timeout" }), + primary: { type: "exit", exitCode: 0, signal: null }, + }); + }); + + it("preserves the timeout primary when process-group signaling fails", async () => { + processGroupAlive = true; + processKill.mockImplementation((pid: number, signal?: NodeJS.Signals | 0) => { + if (pid === -42 && signal !== 0) { + throw Object.assign(new Error(`cannot send ${String(signal)}`), { code: "EPERM" }); + } + return true; + }); + const { child, childKill, settled } = start({ executionTimeoutMs: 100 }); + + await vi.advanceTimersByTimeAsync(100); + child.emit("exit", 0, null); + child.emit("close", 0, null); + await vi.advanceTimersByTimeAsync(30); + + expect(settled).toHaveBeenCalledOnce(); + expect(settled.mock.calls[0]?.[0]).toMatchObject({ + primary: { type: "timeout" }, + settlementFailure: expect.any(Error), + }); + expect(childKill).not.toHaveBeenCalled(); + }); + + it("bounds stdio draining when a timeout cleans the group without close", async () => { + const { settled, stderrDestroy, stdoutDestroy } = start({ executionTimeoutMs: 100 }); + + await vi.advanceTimersByTimeAsync(1_099); + expect(settled).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + + expect(stdoutDestroy).toHaveBeenCalledOnce(); + expect(stderrDestroy).toHaveBeenCalledOnce(); + expect(settled).toHaveBeenCalledWith({ + settlementFailure: expect.objectContaining({ message: "stdio-drain-timeout" }), + primary: { type: "timeout" }, + }); + }); + + it("preserves a stream failure that follows a successful exit", async () => { + const { child, settled } = start(); + const streamFailure = new Error("stdout pipe failed"); + + child.emit("exit", 0, null); + child.stdout?.emit("error", streamFailure); + child.emit("close", 0, null); + + expect(settled).toHaveBeenCalledWith({ + primary: { type: "exit", exitCode: 0, signal: null }, + settlementFailure: expect.objectContaining({ + cause: streamFailure, + message: "stdout stream error: stdout pipe failed", + }), + }); + }); +}); diff --git a/extensions/qa-lab/src/posix-command-settlement.ts b/extensions/qa-lab/src/posix-command-settlement.ts new file mode 100644 index 000000000000..ec9fb73600bd --- /dev/null +++ b/extensions/qa-lab/src/posix-command-settlement.ts @@ -0,0 +1,267 @@ +import type { ChildProcess } from "node:child_process"; +import { isQaPosixProcessGroupAlive, signalQaPosixProcessGroup } from "./posix-process-group.js"; + +type QaPosixCommandPrimary = + | { type: "exit"; exitCode: number | null; signal: NodeJS.Signals | null } + | { type: "manual" | "timeout" } + | { type: "parent-signal"; signal: "SIGINT" | "SIGTERM" } + | { type: "spawn-error"; error: Error } + | { type: "stream-error"; error: Error; stream: "stderr" | "stdout" }; + +type QaPosixCommandSettlementParams = { + child: ChildProcess; + settlementFailureMessage: string; + executionTimeoutMs?: number; + forceKillAfterMs: number; + forwardParentSignals?: boolean; + initialSignal: NodeJS.Signals; + onStderrData?: (chunk: Buffer) => void; + onSettled: (outcome: { primary: QaPosixCommandPrimary; settlementFailure?: Error }) => void; + onStdoutData?: (chunk: Buffer) => void; + processGroupId: number | undefined; + verifyAfterMs: number; + windowsCleanup?: { + alive?: () => boolean; + closeCompletesCleanup?: boolean; + signal: (signal: NodeJS.Signals) => Error | undefined; + }; +}; + +export function createQaPosixCommandSettlement(params: QaPosixCommandSettlementParams) { + const timers: NodeJS.Timeout[] = []; + const errors: Error[] = []; + let cleanupDone = false; + let cleanupStarted = false; + let disposed = false; + let drainDeadline: NodeJS.Timeout | undefined; + let drainIdle: NodeJS.Timeout | undefined; + let drainTimedOut = false; + let executionTimer: NodeJS.Timeout | undefined; + let primary: QaPosixCommandPrimary | undefined; + let stdioDrained = false; + const windows = params.windowsCleanup; + + const schedule = (fn: () => void, delay: number) => { + const timer = setTimeout(() => { + fn(); + }, delay); + timers.push(timer); + return timer; + }; + const cancel = (timer: NodeJS.Timeout | undefined) => { + if (timer) { + clearTimeout(timer); + } + }; + // Cleanup owns only the original PGID; a descendant that calls setsid can escape it. + const alive = () => + windows?.alive?.() ?? + (windows + ? true + : params.processGroupId !== undefined && isQaPosixProcessGroupAlive(params.processGroupId)); + const signal = (nextSignal: NodeJS.Signals) => { + const error = windows + ? windows.signal(nextSignal) + : params.processGroupId === undefined + ? undefined + : signalQaPosixProcessGroup(params.processGroupId, nextSignal); + if (error) { + errors.push(error); + } + }; + const dispose = () => { + if (disposed) { + return; + } + disposed = true; + for (const timer of timers) { + clearTimeout(timer); + } + params.child.stdout?.removeListener("data", onStdoutData); + params.child.stdout?.removeListener("error", onStdoutError); + params.child.stderr?.removeListener("data", onStderrData); + params.child.stderr?.removeListener("error", onStderrError); + params.child.removeListener("error", onChildError); + params.child.removeListener("exit", onExit); + params.child.removeListener("close", onClose); + process.removeListener("exit", onParentExit); + process.removeListener("SIGINT", onParentSigint); + process.removeListener("SIGTERM", onParentSigterm); + }; + const settle = () => { + if (disposed || !primary || !cleanupDone || !stdioDrained) { + return; + } + const parentSignal = primary.type === "parent-signal" ? primary.signal : undefined; + const settlementFailure = + errors.length > 1 ? new AggregateError(errors, params.settlementFailureMessage) : errors[0]; + dispose(); + params.onSettled({ primary, ...(settlementFailure ? { settlementFailure } : {}) }); + if (parentSignal) { + process.kill(process.pid, parentSignal); + } + }; + const finishCleanup = () => { + cleanupDone = true; + if (drainTimedOut && !stdioDrained) { + params.child.stdout?.destroy(); + params.child.stderr?.destroy(); + stdioDrained = true; + } + settle(); + }; + const verify = () => { + if (alive() && !windows) { + errors.push( + new Error(`${params.settlementFailureMessage}: pgid=${params.processGroupId} alive`), + ); + } + if (!stdioDrained) { + drainTimedOut = true; + if (!windows && !errors.some((error) => error.message === "stdio-drain-timeout")) { + errors.push(new Error("stdio-drain-timeout")); + } + } + finishCleanup(); + }; + const forceKill = () => { + if (!alive()) { + finishCleanup(); + return; + } + signal("SIGKILL"); + schedule(verify, params.verifyAfterMs); + }; + const startCleanup = (initialSignal = params.initialSignal) => { + if (cleanupStarted) { + return; + } + cleanupStarted = true; + if (!alive()) { + finishCleanup(); + return; + } + signal(initialSignal); + schedule( + initialSignal === "SIGKILL" ? verify : forceKill, + initialSignal === "SIGKILL" ? params.verifyAfterMs : params.forceKillAfterMs, + ); + }; + const onDrainDeadline = () => { + drainTimedOut = true; + errors.push(new Error("stdio-drain-timeout")); + startCleanup(); + if (cleanupDone) { + finishCleanup(); + } + }; + const armDrainDeadline = () => { + if (!drainDeadline && !stdioDrained) { + drainDeadline = schedule(onDrainDeadline, 1_000); + } + }; + const freeze = (nextPrimary: QaPosixCommandPrimary, initialSignal = params.initialSignal) => { + primary ??= nextPrimary; + cancel(executionTimer); + // Every terminal path needs a drain bound: an escaped descendant can retain + // inherited stdio even after the original process group is gone. + armDrainDeadline(); + startCleanup(initialSignal); + settle(); + }; + const armIdle = () => { + cancel(drainIdle); + drainIdle = schedule(startCleanup, 100); + }; + const onOutput = () => { + if (primary?.type === "exit" && !drainTimedOut && !stdioDrained) { + armIdle(); + } + }; + function onStdoutData(chunk: Buffer) { + params.onStdoutData?.(chunk); + onOutput(); + } + function onStderrData(chunk: Buffer) { + params.onStderrData?.(chunk); + onOutput(); + } + const freezeStreamError = (stream: "stderr" | "stdout", error: Error) => { + if (primary) { + errors.push(new Error(`${stream} stream error: ${error.message}`, { cause: error })); + } + freeze({ type: "stream-error", stream, error }); + }; + function onStdoutError(error: Error) { + freezeStreamError("stdout", error); + } + function onStderrError(error: Error) { + freezeStreamError("stderr", error); + } + function onChildError(error: Error) { + freeze({ type: "spawn-error", error }); + } + // `exit` freezes the leader tuple; only `close` can prove stdio drained. + function onExit(exitCode: number | null, nextSignal: NodeJS.Signals | null) { + primary ??= { type: "exit", exitCode, signal: nextSignal }; + cancel(executionTimer); + armDrainDeadline(); + armIdle(); + if (cleanupStarted && !cleanupDone && !alive()) { + finishCleanup(); + } + } + function onClose() { + stdioDrained = true; + cancel(drainIdle); + cancel(drainDeadline); + if (!cleanupStarted && windows) { + cleanupDone = true; + } else if (cleanupStarted && windows?.closeCompletesCleanup) { + finishCleanup(); + } else { + startCleanup(); + if (!cleanupDone && !alive()) { + finishCleanup(); + } + } + settle(); + } + function onParentExit() { + signal("SIGKILL"); + } + const onParentSignal = (nextSignal: "SIGINT" | "SIGTERM") => { + if (windows) { + primary ??= { type: "parent-signal", signal: nextSignal }; + signal(nextSignal); + dispose(); + process.kill(process.pid, nextSignal); + return; + } + freeze({ type: "parent-signal", signal: nextSignal }, nextSignal); + }; + function onParentSigint() { + onParentSignal("SIGINT"); + } + function onParentSigterm() { + onParentSignal("SIGTERM"); + } + + params.child.stdout?.on("data", onStdoutData); + params.child.stdout?.on("error", onStdoutError); + params.child.stderr?.on("data", onStderrData); + params.child.stderr?.on("error", onStderrError); + params.child.on("error", onChildError); + params.child.on("exit", onExit); + params.child.on("close", onClose); + if (params.forwardParentSignals) { + process.on("exit", onParentExit); + process.on("SIGINT", onParentSigint); + process.on("SIGTERM", onParentSigterm); + } + if (params.executionTimeoutMs !== undefined) { + executionTimer = schedule(() => freeze({ type: "timeout" }), params.executionTimeoutMs); + } + + return { requestCleanup: () => freeze({ type: "manual" }) }; +} diff --git a/extensions/qa-lab/src/posix-process-group.test.ts b/extensions/qa-lab/src/posix-process-group.test.ts new file mode 100644 index 000000000000..54e7acedc177 --- /dev/null +++ b/extensions/qa-lab/src/posix-process-group.test.ts @@ -0,0 +1,80 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + inspectLinuxProcessGroupStats, + isQaPosixProcessGroupAlive, + signalQaPosixProcessGroup, +} from "./posix-process-group.js"; + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("POSIX process group inspection", () => { + it("treats Linux zombie and dead members as stopped", () => { + expect( + inspectLinuxProcessGroupStats(123, [ + "123 (leader) Z 1 123 123 0 -1 0", + "124 (helper (worker)) X 1 123 123 0 -1 0", + "125 (unrelated) S 1 999 999 0 -1 0", + ]), + ).toEqual({ + alive: false, + diagnostics: + 'pgid=123 members=[pid=123 state=Z command="leader", pid=124 state=X command="helper (worker)"]', + }); + }); + + it("treats runnable members as alive and empty snapshots as unknown", () => { + expect( + inspectLinuxProcessGroupStats(123, [ + "123 (leader) Z 1 123 123 0 -1 0", + "124 (worker) D 1 123 123 0 -1 0", + ]).alive, + ).toBe(true); + expect(inspectLinuxProcessGroupStats(123, ["125 (other) S 1 999 999 0 -1 0"])).toEqual({ + alive: null, + diagnostics: "pgid=123 members=[]", + }); + }); + + it("fails closed when the Linux member snapshot is unavailable", () => { + const platform = vi.spyOn(process, "platform", "get").mockReturnValue("linux"); + const processKill = vi.spyOn(process, "kill").mockImplementation(() => true); + try { + expect(isQaPosixProcessGroupAlive(123, () => null)).toBe(true); + expect(processKill).toHaveBeenCalledWith(-123, 0); + } finally { + platform.mockRestore(); + } + }); + + it("treats a kill-visible Linux group with only zombie members as stopped", () => { + const platform = vi.spyOn(process, "platform", "get").mockReturnValue("linux"); + const processKill = vi.spyOn(process, "kill").mockImplementation(() => true); + try { + expect( + isQaPosixProcessGroupAlive(123, () => ({ + alive: false, + diagnostics: 'pgid=123 members=[pid=123 state=Z command="leader"]', + })), + ).toBe(false); + expect(processKill).toHaveBeenCalledWith(-123, 0); + } finally { + platform.mockRestore(); + } + }); + + it("stops on ESRCH and never falls back to a positive pid", () => { + const processKill = vi.spyOn(process, "kill").mockImplementation((pid, signal) => { + expect(pid).toBe(-123); + if (signal === 0) { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + return true; + }); + + expect(isQaPosixProcessGroupAlive(123)).toBe(false); + expect(signalQaPosixProcessGroup(123, "SIGTERM")).toBeUndefined(); + expect(processKill).not.toHaveBeenCalledWith(123, expect.anything()); + }); +}); diff --git a/extensions/qa-lab/src/posix-process-group.ts b/extensions/qa-lab/src/posix-process-group.ts new file mode 100644 index 000000000000..7e2d328ab817 --- /dev/null +++ b/extensions/qa-lab/src/posix-process-group.ts @@ -0,0 +1,126 @@ +import { readFileSync, readdirSync } from "node:fs"; +import path from "node:path"; +import { sliceUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; + +function parseLinuxProcessStat(raw: string) { + const commandStart = raw.indexOf("("); + const commandEnd = raw.lastIndexOf(")"); + if (commandStart <= 0 || commandEnd <= commandStart) { + return null; + } + const pid = Number.parseInt(raw.slice(0, commandStart).trim(), 10); + const fields = raw + .slice(commandEnd + 1) + .trim() + .split(/\s+/u); + const state = fields[0]; + const processGroupId = Number.parseInt(fields[2] ?? "", 10); + if ( + !Number.isSafeInteger(pid) || + pid <= 0 || + !state || + !Number.isSafeInteger(processGroupId) || + processGroupId <= 0 + ) { + return null; + } + return { + command: raw.slice(commandStart + 1, commandEnd), + pid, + processGroupId, + state, + }; +} + +function boundProcessGroupDiagnostics(details: string) { + if (details.length <= 2_048) { + return details; + } + return `${sliceUtf16Safe(details, 0, 2_045)}...`; +} + +export function inspectLinuxProcessGroupStats(processGroupId: number, stats: readonly string[]) { + const members = stats + .map((raw) => parseLinuxProcessStat(raw)) + .filter( + (entry): entry is NonNullable> => + entry?.processGroupId === processGroupId, + ) + .toSorted((left, right) => left.pid - right.pid); + const diagnostics = members + .map( + (member) => + `pid=${member.pid} state=${member.state} command=${JSON.stringify(member.command)}`, + ) + .join(", "); + return { + alive: + members.length === 0 + ? null + : members.some((entry) => entry.state !== "Z" && entry.state !== "X"), + diagnostics: boundProcessGroupDiagnostics(`pgid=${processGroupId} members=[${diagnostics}]`), + }; +} + +type QaLinuxProcessGroupInspection = ReturnType; +export type QaLinuxProcessGroupInspector = ( + processGroupId: number, +) => QaLinuxProcessGroupInspection | null; + +export function inspectLinuxProcessGroup( + processGroupId: number, +): QaLinuxProcessGroupInspection | null { + if (process.platform !== "linux") { + return null; + } + let entries; + try { + entries = readdirSync("/proc", { withFileTypes: true }); + } catch { + return null; + } + const stats: string[] = []; + for (const entry of entries) { + if (!entry.isDirectory() || !/^\d+$/u.test(entry.name)) { + continue; + } + try { + stats.push(readFileSync(path.join("/proc", entry.name, "stat"), "utf8")); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + return null; + } + } + } + return inspectLinuxProcessGroupStats(processGroupId, stats); +} + +export function isQaPosixProcessGroupAlive( + processGroupId: number, + inspectLinuxProcessGroupFn: QaLinuxProcessGroupInspector = inspectLinuxProcessGroup, +) { + try { + process.kill(-processGroupId, 0); + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } + if (process.platform !== "linux") { + return true; + } + return inspectLinuxProcessGroupFn(processGroupId)?.alive ?? true; +} + +export function signalQaPosixProcessGroup( + processGroupId: number, + signal: NodeJS.Signals, +): Error | undefined { + try { + process.kill(-processGroupId, signal); + return undefined; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ESRCH") { + return undefined; + } + return error instanceof Error ? error : new Error(String(error)); + } +} diff --git a/extensions/qa-lab/src/suite-runtime-agent-process.test.ts b/extensions/qa-lab/src/suite-runtime-agent-process.test.ts index 3c487ee1db8a..64002e8eb90b 100644 --- a/extensions/qa-lab/src/suite-runtime-agent-process.test.ts +++ b/extensions/qa-lab/src/suite-runtime-agent-process.test.ts @@ -60,6 +60,17 @@ function createMockEmitter() { function createSpawnedProcess(params: { pid?: number } = {}) { const child = createMockEmitter() as MockChildProcess; + const emit = child.emit.bind(child); + let exited = false; + child.emit = (eventName, ...args) => { + if (eventName === "exit") { + exited = true; + } else if (eventName === "close" && !exited) { + exited = true; + emit("exit", ...args); + } + return emit(eventName, ...args); + }; child.pid = params.pid; child.stdout = createMockEmitter(); child.stderr = createMockEmitter(); @@ -173,7 +184,16 @@ describe("qa suite runtime agent process helpers", () => { }); it.runIf(process.platform !== "win32")("kills timed-out qa cli process groups", async () => { - const killSpy = vi.spyOn(process, "kill").mockImplementation(() => true); + let processGroupAlive = true; + const killSpy = vi.spyOn(process, "kill").mockImplementation((pid, signal) => { + if (pid === -12345 && signal === "SIGKILL") { + processGroupAlive = false; + } + if (pid === -12345 && signal === 0 && !processGroupAlive) { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + return true; + }); vi.useFakeTimers(); try { const child = createSpawnedProcess({ pid: 12345 }); @@ -198,6 +218,8 @@ describe("qa suite runtime agent process helpers", () => { ), ); await vi.advanceTimersByTimeAsync(1); + child.emit("exit", null, "SIGKILL"); + child.emit("close", null, "SIGKILL"); const error = await errorPromise; expect(error).toMatchObject({ code: "qa_cli_timeout" }); @@ -217,6 +239,41 @@ describe("qa suite runtime agent process helpers", () => { } }); + it.runIf(process.platform !== "win32")( + "preserves a nonzero qa cli failure when process-group cleanup also fails", + async () => { + const killSpy = vi.spyOn(process, "kill").mockImplementation((pid, signal) => { + if (pid === -12345 && signal === "SIGKILL") { + throw Object.assign(new Error("cleanup denied"), { code: "EPERM" }); + } + return true; + }); + vi.useFakeTimers(); + try { + const child = createSpawnedProcess({ pid: 12345 }); + const { pending } = startMockQaCli({ args: ["qa", "suite"], child }); + const errorPromise = pending.catch((value: unknown) => value); + await Promise.resolve(); + child.stderr.emit("data", Buffer.from("suite failed\n")); + child.emit("exit", 7, null); + child.emit("close", 7, null); + await vi.advanceTimersByTimeAsync(500); + + const error = await errorPromise; + expect(error).toBeInstanceOf(AggregateError); + expect(error).toMatchObject({ message: "qa cli command and settlement failed" }); + const failures = error instanceof AggregateError ? error.errors : []; + expect(failures).toEqual([ + expect.objectContaining({ message: "qa cli failed (7): suite failed" }), + expect.any(Error), + ]); + } finally { + vi.useRealTimers(); + killSpy.mockRestore(); + } + }, + ); + it("force-kills timed-out Windows qa cli process trees with taskkill", async () => { const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform"); const originalSystemRoot = process.env.SystemRoot; diff --git a/extensions/qa-lab/src/suite-runtime-agent-process.ts b/extensions/qa-lab/src/suite-runtime-agent-process.ts index f3675be4824d..1ab1db4a2ad0 100644 --- a/extensions/qa-lab/src/suite-runtime-agent-process.ts +++ b/extensions/qa-lab/src/suite-runtime-agent-process.ts @@ -19,6 +19,7 @@ import { import { QaSuiteInfraError } from "./errors.js"; import { extractGatewayMessageText } from "./gateway-log-sentinel.js"; import { resolveQaNodeExecPath } from "./node-exec.js"; +import { createQaPosixCommandSettlement } from "./posix-command-settlement.js"; import { liveTurnTimeoutMs } from "./suite-runtime-agent-common.js"; import { readSessionTranscriptSummary } from "./suite-runtime-agent-session.js"; import { waitForGatewayHealthy, waitForTransportReady } from "./suite-runtime-gateway.js"; @@ -234,37 +235,22 @@ function parseQaCliJsonOutput(text: string, args: readonly string[]) { } } -function signalQaCliProcessTree( - child: Pick, - signal: NodeJS.Signals, -) { - if (process.platform === "win32") { - if (typeof child.pid === "number") { - const result = spawnSync( - resolveQaWindowsSystem32ExePath("taskkill.exe"), - ["/PID", String(child.pid), "/T", "/F"], - { - stdio: "ignore", - windowsHide: true, - timeout: 5_000, - }, - ); - if (!result.error && result.status === 0) { - return; - } - } - child.kill(signal); - return; - } - if (typeof child.pid === "number") { - try { - process.kill(-child.pid, signal); +function killQaCliWindowsProcessTree(child: Pick) { + if (child.pid) { + const result = spawnSync( + resolveQaWindowsSystem32ExePath("taskkill.exe"), + ["/PID", String(child.pid), "/T", "/F"], + { + stdio: "ignore", + windowsHide: true, + timeout: 5_000, + }, + ); + if (!result.error && result.status === 0) { return; - } catch { - // The detached process group may already be gone; fall back to the child handle. } } - child.kill(signal); + child.kill("SIGKILL"); } async function runQaCli( @@ -291,8 +277,7 @@ async function runQaCli( stdio: ["ignore", "pipe", "pipe"], }); const timeoutMs = resolveTimerTimeoutMs(opts?.timeoutMs, 60_000); - const timeout = setTimeout(() => { - signalQaCliProcessTree(child, "SIGKILL"); + const rejectTimeout = () => { const stdoutText = formatQaChildOutputTail(stdoutTail, "qa cli stdout"); const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr"); const diagnostics = [ @@ -301,12 +286,68 @@ async function runQaCli( ] .filter(Boolean) .join("\n"); - reject( - new QaSuiteInfraError( - "qa_cli_timeout", - `qa cli timed out: openclaw ${args.join(" ")}${diagnostics ? `\n${diagnostics}` : ""}`, - ), + return new QaSuiteInfraError( + "qa_cli_timeout", + `qa cli timed out: openclaw ${args.join(" ")}${diagnostics ? `\n${diagnostics}` : ""}`, ); + }; + const getExitError = (code: number | null) => { + if (code === 0) { + if (stdout.exceeded) { + return new Error( + `qa cli stdout exceeded ${QA_CHILD_STDOUT_MAX_BYTES} bytes; refusing to parse truncated output`, + ); + } + return undefined; + } + const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr"); + return new Error(`qa cli failed (${code ?? "unknown"}): ${stderrText}`); + }; + if (process.platform !== "win32") { + createQaPosixCommandSettlement({ + child, + settlementFailureMessage: "qa cli settlement failed", + executionTimeoutMs: timeoutMs, + forceKillAfterMs: 0, + initialSignal: "SIGKILL", + onSettled: (outcome) => { + const primary = outcome.primary; + const primaryError = + primary.type === "spawn-error" || primary.type === "stream-error" + ? primary.error + : primary.type === "timeout" + ? rejectTimeout() + : getExitError(primary.type === "exit" ? primary.exitCode : 1); + if (outcome.settlementFailure) { + reject( + primaryError + ? new AggregateError( + [primaryError, outcome.settlementFailure], + "qa cli command and settlement failed", + ) + : outcome.settlementFailure, + ); + return; + } + if (primaryError) { + reject(primaryError); + return; + } + resolve(); + }, + onStderrData: (chunk) => appendQaChildOutputTail(stderr, chunk), + onStdoutData: (chunk) => { + appendQaChildOutput(stdout, chunk); + appendQaChildOutputTail(stdoutTail, chunk); + }, + processGroupId: child.pid, + verifyAfterMs: 500, + }); + return; + } + const timeout = setTimeout(() => { + killQaCliWindowsProcessTree(child); + reject(rejectTimeout()); }, timeoutMs); child.stdout.on("data", (chunk) => { appendQaChildOutput(stdout, chunk); @@ -319,20 +360,8 @@ async function runQaCli( }); child.once("close", (code) => { clearTimeout(timeout); - if (code === 0) { - if (stdout.exceeded) { - reject( - new Error( - `qa cli stdout exceeded ${QA_CHILD_STDOUT_MAX_BYTES} bytes; refusing to parse truncated output`, - ), - ); - return; - } - resolve(); - return; - } - const stderrText = formatQaChildOutputTail(stderr, "qa cli stderr"); - reject(new Error(`qa cli failed (${code ?? "unknown"}): ${stderrText}`)); + const error = getExitError(code); + return error ? reject(error) : resolve(); }); }); const text = readQaChildOutput(stdout).trim(); diff --git a/extensions/qa-lab/src/test-file-scenario-command-lifecycle.test.ts b/extensions/qa-lab/src/test-file-scenario-command-lifecycle.test.ts index 7176b9a54382..20c7e7acf4d0 100644 --- a/extensions/qa-lab/src/test-file-scenario-command-lifecycle.test.ts +++ b/extensions/qa-lab/src/test-file-scenario-command-lifecycle.test.ts @@ -1,17 +1,33 @@ import type { ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const spawnMock = vi.hoisted(() => vi.fn()); +const spawnSyncMock = vi.hoisted(() => vi.fn()); +const actualSpawn = vi.hoisted( + () => + ({ + value: undefined, + }) as { + value: typeof import("node:child_process").spawn | undefined; + }, +); vi.mock("node:child_process", async (importOriginal) => { const actual = await importOriginal(); + actualSpawn.value = actual.spawn; return { ...actual, spawn: spawnMock, + spawnSync: spawnSyncMock, }; }); +import { isQaPosixProcessGroupAlive } from "./posix-process-group.js"; import { resetQaScenarioCommandCleanupTimings, runQaScenarioCommandLifecycle, @@ -28,8 +44,8 @@ function spyOnProcessKill() { function createChild(pid = 42) { const child = new EventEmitter() as ChildProcess; Object.defineProperty(child, "pid", { value: pid }); - child.stdout = new EventEmitter() as NonNullable; - child.stderr = new EventEmitter() as NonNullable; + child.stdout = Object.assign(new EventEmitter(), { destroy: vi.fn() }) as never; + child.stderr = Object.assign(new EventEmitter(), { destroy: vi.fn() }) as never; child.kill = vi.fn(() => true) as ChildProcess["kill"]; spawnMock.mockReturnValue(child); return child; @@ -45,6 +61,207 @@ function runCommand(timeoutMs?: number) { }); } +function isProcessRunning(pid: number) { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function waitForPidFile(filePath: string, timeoutMs = 2_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = await readFile(filePath, "utf8").catch(() => ""); + if (/^[1-9]\d*$/u.test(value.trim())) { + return Number(value.trim()); + } + await sleep(10); + } + throw new Error(`timed out waiting for pid file ${filePath}`); +} + +async function waitForProcessExit(pid: number, timeoutMs = 2_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (!isProcessRunning(pid)) { + return; + } + await sleep(10); + } + throw new Error(`timed out waiting for process ${pid} to exit`); +} + +describe.skipIf(process.platform === "win32")("qa scenario command real POSIX lifecycle", () => { + afterEach(() => { + resetQaScenarioCommandCleanupTimings(); + spawnMock.mockReset(); + }); + + it("settles within a bound after the leader writes its final result with inherited stdio open", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-settlement-")); + const descendantPidPath = path.join(root, "descendant.pid"); + let descendantPid: number | undefined; + spawnMock.mockImplementation((...args: Parameters>) => { + if (!actualSpawn.value) { + throw new Error("real spawn unavailable"); + } + return actualSpawn.value(...args); + }); + setQaScenarioCommandCleanupTimings({ killGraceMs: 100, forceSettleMs: 100 }); + try { + const descendantScript = [ + "const { writeFileSync } = require('node:fs');", + "process.on('SIGTERM', () => {});", + `writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`, + "setTimeout(() => process.stdout.write('delayed descendant output\\n'), 40);", + "setInterval(() => {}, 1000);", + ].join(" "); + const leaderScript = [ + "const { spawn } = require('node:child_process');", + "const { existsSync } = require('node:fs');", + `spawn(process.execPath, ['-e', ${JSON.stringify(descendantScript)}], { stdio: ['ignore', 'inherit', 'inherit'] }).unref();`, + `const ready = setInterval(() => { if (!existsSync(${JSON.stringify(descendantPidPath)})) return; clearInterval(ready); process.stdout.write('Docker scheduling finished\\n', () => process.exit(7)); }, 5);`, + ].join("\n"); + + const pending = runQaScenarioCommandLifecycle({ + command: process.execPath, + args: ["-e", leaderScript], + cwd: root, + env: process.env, + timeoutMs: 5_000, + }); + descendantPid = await waitForPidFile(descendantPidPath); + const processGroupId = (spawnMock.mock.results[0]?.value as ChildProcess | undefined)?.pid; + if (!processGroupId) { + throw new Error("scenario command did not expose its process group id"); + } + const startedAt = Date.now(); + const deadline = new AbortController(); + const result = await Promise.race([ + pending, + sleep(1_500, undefined, { signal: deadline.signal }).then(() => { + throw new Error("command did not settle after process-group cleanup"); + }), + ]).finally(() => deadline.abort()); + + expect(Date.now() - startedAt).toBeLessThan(1_500); + expect(result).toEqual({ + exitCode: 7, + signal: null, + stdout: "Docker scheduling finished\ndelayed descendant output\n", + stderr: "", + }); + expect(isQaPosixProcessGroupAlive(processGroupId)).toBe(false); + } finally { + if (descendantPid && isProcessRunning(descendantPid)) { + process.kill(descendantPid, "SIGKILL"); + } + await rm(root, { force: true, recursive: true }); + } + }); + + it("reports that a self-detached descendant escaped and cleans it explicitly", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-setsid-escape-")); + const descendantPidPath = path.join(root, "descendant.pid"); + let descendantPid: number | undefined; + spawnMock.mockImplementation((...args: Parameters>) => { + if (!actualSpawn.value) { + throw new Error("real spawn unavailable"); + } + return actualSpawn.value(...args); + }); + try { + const descendantScript = [ + "const { writeFileSync } = require('node:fs');", + `writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`, + "process.stdout.write('escaped descendant output\\n');", + "setInterval(() => {}, 1000);", + ].join(" "); + const leaderScript = [ + "const { spawn } = require('node:child_process');", + `spawn(process.execPath, ['-e', ${JSON.stringify(descendantScript)}], { detached: true, stdio: ['ignore', 'inherit', 'inherit'] }).unref();`, + "process.exit(0);", + ].join("\n"); + + const result = await runQaScenarioCommandLifecycle({ + command: process.execPath, + args: ["-e", leaderScript], + cwd: root, + env: process.env, + timeoutMs: 5_000, + }); + + descendantPid = await waitForPidFile(descendantPidPath); + expect(result.exitCode).toBe(1); + expect(result.failureMessage).toBe("stdio-drain-timeout"); + expect(result.stdout).toContain("escaped descendant output"); + expect(isProcessRunning(descendantPid)).toBe(true); + } finally { + // A true setsid descendant is outside the original PGID by design. + if (descendantPid && isProcessRunning(descendantPid)) { + process.kill(descendantPid, "SIGKILL"); + await waitForProcessExit(descendantPid).catch(() => undefined); + } + await rm(root, { force: true, recursive: true }); + } + }); + + it("cleans the command group before re-raising a parent SIGTERM", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "qa-command-parent-signal-")); + const descendantPidPath = path.join(root, "descendant.pid"); + const moduleUrl = new URL("./test-file-scenario-command-lifecycle.ts", import.meta.url).href; + let descendantPid: number | undefined; + if (!actualSpawn.value) { + throw new Error("real spawn unavailable"); + } + const controllerScript = [ + `import { runQaScenarioCommandLifecycle, setQaScenarioCommandCleanupTimings } from ${JSON.stringify(moduleUrl)};`, + "setQaScenarioCommandCleanupTimings({ killGraceMs: 50, forceSettleMs: 50 });", + "const nested = [", + " \"const { writeFileSync } = require('node:fs');\",", + ` ${JSON.stringify(`writeFileSync(${JSON.stringify(descendantPidPath)}, String(process.pid));`)},`, + " \"process.on('SIGTERM', () => {});\",", + ' "setInterval(() => {}, 1000);",', + "].join(' ');", + "await runQaScenarioCommandLifecycle({", + " command: process.execPath,", + " args: ['-e', nested],", + ` cwd: ${JSON.stringify(root)},`, + " env: process.env,", + " timeoutMs: 5000,", + "});", + ].join("\n"); + const controller = actualSpawn.value( + process.execPath, + ["--import", "tsx", "--input-type=module", "-e", controllerScript], + { cwd: process.cwd(), env: process.env, stdio: "ignore" }, + ); + try { + descendantPid = await waitForPidFile(descendantPidPath, 10_000); + controller.kill("SIGTERM"); + const [exitCode, signal] = await new Promise<[number | null, NodeJS.Signals | null]>( + (resolve) => { + controller.once("close", (code, nextSignal) => resolve([code, nextSignal])); + }, + ); + + expect(exitCode).toBeNull(); + expect(signal).toBe("SIGTERM"); + await waitForProcessExit(descendantPid); + } finally { + if (controller.exitCode === null && controller.signalCode === null) { + controller.kill("SIGKILL"); + } + if (descendantPid && isProcessRunning(descendantPid)) { + process.kill(descendantPid, "SIGKILL"); + } + await rm(root, { force: true, recursive: true }); + } + }); +}); + describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", () => { const parentHandlers = new Map(); let processKill: ReturnType; @@ -52,10 +269,15 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", ( beforeEach(() => { vi.useFakeTimers(); spawnMock.mockReset(); + spawnSyncMock.mockReset(); vi.spyOn(process, "once").mockImplementation((event, listener) => { parentHandlers.set(event as ParentSignal | "exit", listener as ParentHandler); return process; }); + vi.spyOn(process, "on").mockImplementation((event, listener) => { + parentHandlers.set(event as ParentSignal | "exit", listener as ParentHandler); + return process; + }); vi.spyOn(process, "removeListener").mockImplementation((event, listener) => { if (parentHandlers.get(event as ParentSignal | "exit") === listener) { parentHandlers.delete(event as ParentSignal | "exit"); @@ -83,6 +305,7 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", ( child.stdout?.emit("data", Buffer.from("out\n")); child.stderr?.emit("data", Buffer.from("err\n")); + child.emit("exit", 3, null); child.emit("close", 3, null); await expect(resultPromise).resolves.toEqual({ @@ -114,6 +337,38 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", ( expect(parentHandlers.size).toBe(0); }); + it("preserves the Windows taskkill timeout lifecycle", async () => { + const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform"); + const originalSystemRoot = process.env.SystemRoot; + Object.defineProperty(process, "platform", { value: "win32", configurable: true }); + process.env.SystemRoot = "C:\\Windows"; + spawnSyncMock.mockReturnValue({ status: 0 }); + createChild(12345); + setQaScenarioCommandCleanupTimings({ killGraceMs: 20, forceSettleMs: 10 }); + try { + const resultPromise = runCommand(100); + await vi.advanceTimersByTimeAsync(130); + + await expect(resultPromise).resolves.toMatchObject({ + exitCode: 1, + failureMessage: "scenario-command timed out after 100ms", + signal: null, + }); + expect(spawnSyncMock).toHaveBeenCalledTimes(2); + expect(spawnSyncMock.mock.calls[0]?.[1]).toEqual(["/pid", "12345", "/T"]); + expect(spawnSyncMock.mock.calls[1]?.[1]).toEqual(["/pid", "12345", "/T", "/F"]); + } finally { + if (platformDescriptor) { + Object.defineProperty(process, "platform", platformDescriptor); + } + if (originalSystemRoot === undefined) { + delete process.env.SystemRoot; + } else { + process.env.SystemRoot = originalSystemRoot; + } + } + }); + it("escalates timed-out commands and preserves the timeout result", async () => { createChild(); setQaScenarioCommandCleanupTimings({ killGraceMs: 20, forceSettleMs: 10 }); @@ -129,7 +384,11 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", ( }); const resultPromise = runCommand(100); - await vi.advanceTimersByTimeAsync(130); + await vi.advanceTimersByTimeAsync(120); + const child = spawnMock.mock.results[0]?.value as ChildProcess; + child.emit("exit", null, "SIGKILL"); + child.emit("close", null, "SIGKILL"); + await vi.advanceTimersByTimeAsync(10); await expect(resultPromise).resolves.toEqual({ exitCode: 1, @@ -163,7 +422,11 @@ describe.skipIf(process.platform === "win32")("qa scenario command lifecycle", ( | undefined; expect(signalHandler).toBeDefined(); signalHandler?.("SIGTERM"); - await vi.advanceTimersByTimeAsync(30); + await vi.advanceTimersByTimeAsync(20); + const child = spawnMock.mock.results[0]?.value as ChildProcess; + child.emit("exit", null, "SIGKILL"); + child.emit("close", null, "SIGKILL"); + await vi.advanceTimersByTimeAsync(10); await expect(resultPromise).resolves.toEqual({ exitCode: 1, diff --git a/extensions/qa-lab/src/test-file-scenario-command-lifecycle.ts b/extensions/qa-lab/src/test-file-scenario-command-lifecycle.ts index 5955ecc20bad..3943cfe53c4a 100644 --- a/extensions/qa-lab/src/test-file-scenario-command-lifecycle.ts +++ b/extensions/qa-lab/src/test-file-scenario-command-lifecycle.ts @@ -1,5 +1,6 @@ -import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import path from "node:path"; +import { createQaPosixCommandSettlement } from "./posix-command-settlement.js"; import { resolveQaWindowsSystem32ExePath } from "./windows-system-tools.js"; export type QaScenarioCommandExecution = { @@ -24,14 +25,9 @@ type QaScenarioCommandTerminalResult = Pick< >; type QaScenarioTaskkillRunner = typeof spawnSync; -type QaScenarioCommandTimers = Partial< - Record<"timeout" | "forceKill" | "forceSettle", NodeJS.Timeout> ->; const QA_SCENARIO_COMMAND_TIMEOUT_KILL_GRACE_MS = 2_000; const QA_SCENARIO_COMMAND_TIMEOUT_FORCE_SETTLE_MS = 500; -const QA_SCENARIO_COMMAND_PARENT_SIGNALS = ["SIGINT", "SIGTERM"] as const; -type QaScenarioParentSignal = (typeof QA_SCENARIO_COMMAND_PARENT_SIGNALS)[number]; let timeoutKillGraceMs = QA_SCENARIO_COMMAND_TIMEOUT_KILL_GRACE_MS; let timeoutForceSettleMs = QA_SCENARIO_COMMAND_TIMEOUT_FORCE_SETTLE_MS; @@ -55,214 +51,92 @@ export function killQaScenarioWindowsProcessTree( return signal === "SIGKILL" ? run(true) : run(false) || run(true); } -// One owner keeps timers, parent handlers, child signals, and final result -// settlement symmetric across every command exit path. -class QaScenarioCommandLifecycle { - private readonly stderr: Buffer[] = []; - private readonly stdout: Buffer[] = []; - private resolve: ((result: QaScenarioCommandResult) => void) | undefined; - private settled = false; - private timers: QaScenarioCommandTimers = {}; - private timedOut = false; - - constructor( - private readonly execution: QaScenarioCommandExecution, - private readonly child: ChildProcess, - private readonly useProcessGroup: boolean, - ) {} - - start(resolve: (result: QaScenarioCommandResult) => void, reject: (reason?: unknown) => void) { - this.resolve = resolve; - this.armTimeout(); - this.child.stdout?.on("data", (chunk: Buffer) => this.stdout.push(chunk)); - this.child.stderr?.on("data", (chunk: Buffer) => this.stderr.push(chunk)); - process.once("exit", this.handleParentExit); - for (const signal of QA_SCENARIO_COMMAND_PARENT_SIGNALS) { - process.once(signal, this.handleParentSignal); - } - this.child.on("error", (error) => { - if (this.settled) { - return; - } - this.clearTimers(); - this.cleanupParentHandlers(); - reject(error); - }); - this.child.on("close", this.handleClose); - } - - private readonly handleParentExit = () => { - this.signalChild("SIGKILL"); - }; - - private readonly handleParentSignal = (signal: QaScenarioParentSignal) => { - this.removeParentSignalHandlers(); - this.signalChild(signal); - this.scheduleForcedCleanup({ - exitCode: 1, - failureMessage: `${this.commandLabel()} interrupted by ${signal}`, - signal, - }); - process.kill(process.pid, signal); - }; - - private readonly handleClose = (exitCode: number | null, signal: NodeJS.Signals | null) => { - if (this.settled) { - return; - } - if (!this.timedOut) { - this.clearTimeoutTimer(); - } - const result = { - exitCode: this.timedOut ? 1 : (exitCode ?? (signal ? 1 : 0)), - signal, - ...(this.timedOut - ? { - failureMessage: `${this.commandLabel()} timed out after ${this.execution.timeoutMs}ms`, - } - : {}), - }; - if ( - this.timedOut && - !this.useProcessGroup && - (this.timers.forceKill || this.timers.forceSettle) - ) { - return; - } - if (this.isProcessGroupRunning()) { - if (!this.timedOut) { - this.signalChild("SIGTERM"); - } - this.scheduleForcedCleanup(result); - return; - } - this.finish(result); - }; - - private armTimeout() { - const timeoutMs = this.execution.timeoutMs; - if (timeoutMs === undefined) { - return; - } - this.timers.timeout = setTimeout(() => { - delete this.timers.timeout; - this.timedOut = true; - this.signalChild("SIGTERM"); - this.scheduleForcedCleanup({ - exitCode: 1, - failureMessage: `${this.commandLabel()} timed out after ${timeoutMs}ms`, - signal: null, - }); - }, timeoutMs); - } - - private signalChild(signal: NodeJS.Signals) { - if (this.useProcessGroup && this.child.pid) { - try { - process.kill(-this.child.pid, signal); - return; - } catch { - // The process group may already be gone; fall back to the direct child. - } - } - if (!this.useProcessGroup && process.platform === "win32") { - if (killQaScenarioWindowsProcessTree(this.child.pid, signal)) { - return; - } - } - this.child.kill(signal); - } - - private isProcessGroupRunning() { - if (!this.useProcessGroup || !this.child.pid) { - return false; - } - try { - process.kill(-this.child.pid, 0); - return true; - } catch (error) { - return (error as NodeJS.ErrnoException).code === "EPERM"; - } - } - - private scheduleForcedCleanup(result: QaScenarioCommandTerminalResult) { - if (this.timers.forceKill || this.timers.forceSettle) { - return; - } - this.timers.forceKill = setTimeout(() => { - delete this.timers.forceKill; - this.signalChild("SIGKILL"); - this.timers.forceSettle = setTimeout(() => { - delete this.timers.forceSettle; - const stillRunning = this.isProcessGroupRunning(); - const failureMessage = - result.failureMessage ?? - (stillRunning ? `${this.commandLabel()} left background processes running` : undefined); - this.finish({ - exitCode: stillRunning ? 1 : result.exitCode, - signal: result.signal, - ...(failureMessage ? { failureMessage } : {}), - }); - }, timeoutForceSettleMs); - }, timeoutKillGraceMs); - } - - private finish(result: QaScenarioCommandTerminalResult) { - if (this.settled) { - return; - } - this.settled = true; - this.clearTimers(); - this.cleanupParentHandlers(); - this.resolve?.({ - ...result, - stdout: Buffer.concat(this.stdout).toString("utf8"), - stderr: Buffer.concat(this.stderr).toString("utf8"), - }); - } - - private commandLabel() { - return path.basename(this.execution.command); - } - - private clearTimeoutTimer() { - if (this.timers.timeout) { - clearTimeout(this.timers.timeout); - delete this.timers.timeout; - } - } - - private clearTimers() { - for (const timer of Object.values(this.timers)) { - clearTimeout(timer); - } - this.timers = {}; - } - - private removeParentSignalHandlers() { - for (const signal of QA_SCENARIO_COMMAND_PARENT_SIGNALS) { - process.removeListener(signal, this.handleParentSignal); - } - } - - private cleanupParentHandlers() { - this.removeParentSignalHandlers(); - process.removeListener("exit", this.handleParentExit); - } -} - export function runQaScenarioCommandLifecycle( execution: QaScenarioCommandExecution, ): Promise { return new Promise((resolve, reject) => { - const useProcessGroup = process.platform !== "win32"; + const isWindows = process.platform === "win32"; const child = spawn(execution.command, execution.args, { cwd: execution.cwd, - detached: useProcessGroup, + detached: !isWindows, env: execution.env, stdio: ["ignore", "pipe", "pipe"], }); - new QaScenarioCommandLifecycle(execution, child, useProcessGroup).start(resolve, reject); + const stdout: Buffer[] = []; + const stderr: Buffer[] = []; + const commandLabel = path.basename(execution.command); + createQaPosixCommandSettlement({ + child, + settlementFailureMessage: `${commandLabel} settlement failed`, + forceKillAfterMs: timeoutKillGraceMs, + ...(isWindows + ? { + windowsCleanup: { + alive: () => child.pid !== undefined, + signal: (signal: NodeJS.Signals) => { + try { + if (!killQaScenarioWindowsProcessTree(child.pid, signal)) { + child.kill(signal); + } + return undefined; + } catch (error) { + return error instanceof Error ? error : new Error(String(error)); + } + }, + }, + } + : {}), + executionTimeoutMs: execution.timeoutMs, + forwardParentSignals: true, + initialSignal: "SIGTERM", + onSettled: (outcome) => { + const primary = outcome.primary; + if (primary.type === "spawn-error" || primary.type === "stream-error") { + reject( + outcome.settlementFailure + ? new AggregateError( + [primary.error, outcome.settlementFailure], + `${commandLabel} command and settlement failed`, + ) + : primary.error, + ); + return; + } + const result: QaScenarioCommandTerminalResult = + primary.type === "exit" + ? { + exitCode: primary.exitCode ?? (primary.signal ? 1 : 0), + signal: primary.signal, + } + : primary.type === "parent-signal" + ? { + exitCode: 1, + failureMessage: `${commandLabel} interrupted by ${primary.signal}`, + signal: primary.signal, + } + : { + exitCode: 1, + failureMessage: `${commandLabel} timed out after ${execution.timeoutMs}ms`, + signal: null, + }; + const settlementFailure = outcome.settlementFailure?.message; + resolve({ + ...result, + ...(settlementFailure && result.exitCode === 0 ? { exitCode: 1 } : {}), + stdout: Buffer.concat(stdout).toString("utf8"), + stderr: Buffer.concat(stderr).toString("utf8"), + ...(settlementFailure + ? result.failureMessage + ? { failureMessage: `${result.failureMessage}; settlement: ${settlementFailure}` } + : { failureMessage: settlementFailure } + : {}), + }); + }, + onStderrData: (chunk) => stderr.push(Buffer.from(chunk)), + onStdoutData: (chunk) => stdout.push(Buffer.from(chunk)), + processGroupId: isWindows ? undefined : child.pid, + verifyAfterMs: timeoutForceSettleMs, + }); }); }