import { spawnSync } from "node:child_process"; import { writeFile } from "node:fs/promises"; import { createRequire } from "node:module"; import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { isProcessAlive, waitForDead, waitForPidFile } from "../../../test/helpers/process-wait.js"; import { createDeferred } from "../../../test/helpers/promise.js"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; import { createWindowsOutputDecoder } from "../../infra/windows-encoding.js"; import { getWindowsCmdExePath } from "../../infra/windows-install-roots.js"; import { killPidIfAlive } from "../../test-utils/process-tree.js"; import { createProcessSupervisor } from "./supervisor.js"; import type { ManagedRun } from "./types.js"; const activePids = new Set(); const tempDirs = useAutoCleanupTempDirTracker(afterEach); afterEach(async () => { for (const pid of activePids) { killPidIfAlive(pid); } await Promise.all([...activePids].map((pid) => waitForDead(pid, 5_000).catch(() => {}))); activePids.clear(); }); async function createDescendantScope( options: { inheritLineage?: boolean; oneShot?: boolean; ignoreTerm?: boolean } = {}, ) { const cwd = tempDirs.make("openclaw-anchored-shell-"); const descendantPath = path.join(cwd, "descendant.cjs"); const descendantPidPath = path.join(cwd, "descendant.pid"); const releasePath = path.join(cwd, "descendant.release"); const rootPath = path.join(cwd, "root.cjs"); await writeFile( descendantPath, ` const { existsSync, writeFileSync } = require("node:fs"); ${options.ignoreTerm ? 'process.on("SIGTERM", () => {});' : ""} const releaseTimer = setInterval(() => { if (existsSync(process.argv[2])) { clearInterval(releaseTimer); } }, 20); writeFileSync(process.argv[3], String(process.pid)); `, "utf8", ); if (process.platform === "win32") { const koffiPath = createRequire(import.meta.url).resolve("koffi"); await writeFile( rootPath, ` const koffi = require(${JSON.stringify(koffiPath)}); const kernel32 = koffi.load("kernel32.dll"); const handle = koffi.pointer("ANCHORED_SHELL_FIXTURE_HANDLE", koffi.opaque()); const bytes = koffi.pointer("uint8_t"); const createProcess = kernel32.func("__stdcall", "CreateProcessW", "int32_t", [ "str16", koffi.pointer("uint16_t"), "void *", "void *", "int32_t", "uint32_t", "void *", "str16", bytes, bytes, ]); const closeHandle = kernel32.func("__stdcall", "CloseHandle", "int32_t", [handle]); const getLastError = kernel32.func("__stdcall", "GetLastError", "uint32_t", []); // Production only supports x64/arm64, where these Win32 structures are 104/24 bytes. const startupInfo = Buffer.alloc(104); const processInfo = Buffer.alloc(24); startupInfo.writeUInt32LE(startupInfo.length, 0); const commandLine = Buffer.from( [ process.execPath, ${JSON.stringify(descendantPath)}, ${JSON.stringify(releasePath)}, ${JSON.stringify(descendantPidPath)}, ].map((value) => '"' + value + '"').join(" ") + String.fromCharCode(0), "utf16le", ); // Native creation avoids libuv's private Job; no inherited handles guarantees pipe EOF. if (!createProcess( process.execPath, commandLine, null, null, 0, 0x08000000, null, null, startupInfo, processInfo, )) { throw new Error("fixture CreateProcessW failed (Win32 error " + getLastError() + ")"); } for (const offset of [8, 0]) { if (!closeHandle(processInfo.readBigUInt64LE(offset))) { throw new Error("fixture CloseHandle failed (Win32 error " + getLastError() + ")"); } } ${fragmentedOutputFixture()} `, "utf8", ); } else { await writeFile( rootPath, ` const { spawn } = require("node:child_process"); const child = spawn(process.execPath, [${JSON.stringify(descendantPath)}, ${JSON.stringify(releasePath)}, ${JSON.stringify(descendantPidPath)}], { stdio: ${JSON.stringify(options.inheritLineage === false ? ["ignore", "ignore", "ignore"] : ["ignore", "ignore", "ignore", 3])}, }); child.unref(); const ready = setInterval(() => { if (!require("node:fs").existsSync(${JSON.stringify(descendantPidPath)})) return; clearInterval(ready); ${fragmentedOutputFixture()} }, 10); `, "utf8", ); } const supervisor = createProcessSupervisor(); const scopeKey = `anchored-shell:${cwd}`; const cleanup = supervisor.acquireScopeCleanup(scopeKey, { processTree: "required-all" }); const run = await supervisor.spawn({ ...(options.oneShot ? { mode: "child" as const, argv: [process.execPath, rootPath], stdinMode: "pipe-closed" as const, } : { mode: "anchored-shell" as const, command: "node root.cjs" }), scopeKey, cwd, env: process.platform === "win32" ? { ...process.env, COMSPEC: getWindowsCmdExePath(process.env), ComSpec: "Z:\\invalid-later-duplicate\\cmd.exe", } : process.env, }); return { run, supervisor, scopeKey, cleanup, readPid: () => waitForPidFile(descendantPidPath, 5_000), release: () => writeFile(releasePath, "", "utf8"), }; } function fragmentedOutputFixture(): string { return ` process.stdout.write("owned-stdout-one\\nowned-stdout-two\\n"); process.stderr.write("owned-stderr-one\\nowned-stderr-two\\n"); process.stdout.write(Buffer.from([0xf0, 0x9f])); process.stderr.write(Buffer.from([0xf0, 0x9f])); setTimeout(() => { process.stdout.write(Buffer.from([0x98, 0x80, 0xe2, 0x82])); process.stderr.write(Buffer.from([0x98, 0x80, 0xe2, 0x82])); }, 50); `; } async function expectPending(promise: Promise) { const settled = await Promise.race([ promise.then(() => true), new Promise((resolve) => { setImmediate(() => resolve(false)); }), ]); expect(settled).toBe(false); } describe("supervisor anchored shell real process ownership", () => { it.skipIf(process.platform === "win32")( "keeps the root command alive when it closes its inherited lineage descriptor", async () => { const supervisor = createProcessSupervisor(); const command = `${JSON.stringify(process.execPath)} -e ${JSON.stringify( 'require("node:fs").closeSync(3); setTimeout(() => process.stdout.write("SURVIVED\\n"), 300)', )}`; const run = await supervisor.spawn({ mode: "anchored-shell", command }); try { await expect(run.wait()).resolves.toMatchObject({ exitCode: 0, exitSignal: null, stdout: "SURVIVED\n", }); await run.waitForExtinction!(); } finally { await supervisor.shutdown(); } }, ); it.skipIf(process.platform !== "win32")( "keeps anchored Windows commands console-free", async () => { const cwd = tempDirs.make("openclaw-anchored-shell-console-"); const koffiPath = createRequire(import.meta.url).resolve("koffi"); await writeFile( path.join(cwd, "console.cjs"), ` const koffi = require(${JSON.stringify(koffiPath)}); const kernel32 = koffi.load("kernel32.dll"); const getConsoleWindow = kernel32.func("__stdcall", "GetConsoleWindow", "void *", []); process.stdout.write(JSON.stringify({ hasConsole: Boolean(getConsoleWindow()) })); process.stderr.write("owned-console-stderr"); process.exitCode = 23; `, "utf8", ); const supervisor = createProcessSupervisor(); try { const run = await supervisor.spawn({ mode: "anchored-shell", command: `"${process.execPath}" console.cjs`, cwd, }); const result = await run.wait(); await run.waitForExtinction!(); expect(result).toMatchObject({ reason: "exit", exitCode: 23, stderr: "owned-console-stderr", }); expect(JSON.parse(result.stdout), "PR138751_UNEXPECTED_CONSOLE").toEqual({ hasConsole: false, }); } finally { await supervisor.shutdown(); } }, ); it.skipIf(process.platform === "win32").each([ { oneShot: false, ignoreTerm: false }, { oneShot: true, ignoreTerm: false }, { oneShot: true, ignoreTerm: true }, ])( "observes children that closed inherited descriptors (oneShot=$oneShot, ignores TERM=$ignoreTerm)", async ({ oneShot, ignoreTerm }) => { const fixture = await createDescendantScope({ inheritLineage: false, oneShot, ignoreTerm }); const pid = await fixture.readPid(); activePids.add(pid); try { await expect(fixture.run.wait()).resolves.toMatchObject({ exitCode: 0, exitSignal: null }); const cleanup = fixture.cleanup(); await cleanup; expect(isProcessAlive(pid)).toBe(false); await waitForDead(pid, 5_000); } finally { await fixture.release(); killPidIfAlive(pid); await waitForDead(pid, 5_000); await fixture.supervisor.shutdown(); } }, ); it.each(["inherited", "replacement", "empty"] as const)( "completes an otherwise idle host with %s command environment", async (environment) => { const cwd = tempDirs.make("openclaw-anchored-shell-idle-"); const hostPath = path.join(cwd, "host.mts"); const supervisorUrl = new URL("./supervisor.ts", import.meta.url).href; let command = 'printf "%s\\n" "${OPENCLAW_TEST_PARENT_ENV-absent}" "${OPENCLAW_TEST_CHILD_ENV-absent}"'; if (process.platform === "win32") { const commandPath = path.join(cwd, "environment.cmd"); await writeFile( commandPath, "@echo off\r\nif defined OPENCLAW_TEST_PARENT_ENV (echo parent) else (echo absent)\r\nif defined OPENCLAW_TEST_CHILD_ENV (echo child) else (echo absent)\r\n", ); command = `"${commandPath}"`; } await writeFile( hostPath, ` const { createProcessSupervisor } = await import(${JSON.stringify(supervisorUrl)}); const supervisor = createProcessSupervisor(); const environment = ${JSON.stringify(environment)}; const run = await supervisor.spawn({ mode: "anchored-shell", command: ${JSON.stringify(command)}, ...(environment === "inherited" ? {} : { env: environment === "empty" ? {} : { OPENCLAW_TEST_CHILD_ENV: "child" }, }), }); try { const result = await run.wait(); await run.waitForExtinction(); console.log(JSON.stringify(result)); } finally { await supervisor.shutdown(); } `, "utf8", ); // A separate host has no Vitest timers or IPC keeping admission alive. const host = spawnSync(process.execPath, ["--import", "tsx", hostPath], { env: { ...process.env, OPENCLAW_TEST_PARENT_ENV: "parent", OPENCLAW_TEST_CHILD_ENV: undefined, }, encoding: "utf8", timeout: 10_000, }); expect(host.error).toBeUndefined(); expect(host.status, host.stderr).toBe(0); const result = JSON.parse(host.stdout); expect({ ...result, stdout: result.stdout.replaceAll("\r\n", "\n") }).toMatchObject({ reason: "exit", exitCode: 0, stdout: environment === "inherited" ? "parent\nabsent\n" : environment === "replacement" ? "absent\nchild\n" : "absent\nabsent\n", }); }, ); it.each(["exit", "cancel"] as const)( "keeps a replacement's status independent of an older live child's %s", async (completion) => { const supervisor = createProcessSupervisor(); const runs: ManagedRun[] = []; const oldOutput = createDeferred(); const spawn = async (scopeKey: string) => { const ready = createDeferred(); const run = await supervisor.spawn({ mode: "child", runId: "shared-correlation", scopeKey, argv: [ process.execPath, "-e", ` process.stdout.write("ready\\n"); process.stdin.on("data", (data) => { if (data.toString() === "emit") process.stdout.write("older-output\\n"); else process.exit(Number(data.toString())); }); `, ], stdinMode: "pipe-open", onStdout: (chunk) => { if (chunk.includes("ready")) { ready.resolve(); } if (chunk.includes("older-output")) { oldOutput.resolve(); } }, }); runs.push(run); activePids.add(run.pid!); await Promise.race([ ready.promise, run.wait().then(() => { throw new Error("child exited before readiness"); }), ]); return run; }; try { const older = await spawn("older-backend"); const replacement = await spawn("replacement-backend"); const snapshot = { ...replacement.activity }; older.stdin!.write("emit"); await oldOutput.promise; expect(replacement.activity).toEqual(snapshot); if (completion === "exit") { older.stdin!.write("23"); } else { older.cancel(); } expect(replacement.activity).toEqual(snapshot); await older.wait(); expect(isProcessAlive(replacement.pid!)).toBe(true); expect(replacement.activity).toEqual(snapshot); replacement.stdin!.write("0"); await expect(replacement.wait()).resolves.toMatchObject({ reason: "exit", exitCode: 0 }); expect(replacement.activity.resultSettled).toBe(true); } finally { for (const run of runs) { run.cancel(); } await Promise.all(runs.map((run) => run.wait())); await supervisor.shutdown(); } }, ); it("keeps a replacement supervised after an older tree with the same run ID becomes extinct", async () => { const first = await createDescendantScope(); const descendantPid = await first.readPid(); activePids.add(descendantPid); let replacement: Awaited> | undefined; try { await first.run.wait(); expect(isProcessAlive(descendantPid)).toBe(true); const ready = createDeferred(); replacement = await first.supervisor.spawn({ mode: "child", runId: first.run.runId, scopeKey: "fallback-real", argv: [process.execPath, "-e", "process.stdout.write('ready');setInterval(() => {}, 1000)"], stdinMode: "pipe-closed", onStdout: () => ready.resolve(), }); const replacementPid = replacement.pid!; activePids.add(replacementPid); await ready.promise; await first.release(); await first.run.waitForExtinction!(); expect(replacement.activity.resultSettled).toBe(false); await first.supervisor.shutdown(); expect(isProcessAlive(replacementPid)).toBe(false); await expect(replacement.wait()).resolves.toMatchObject({ reason: "manual-cancel" }); } finally { await first.release(); first.run.cancel(); replacement?.cancel(); await Promise.all([first.run.waitForExtinction!(), replacement?.wait()]); await first.supervisor.shutdown(); } }); it.each([ { name: "cancels retained descendants idempotently", cancel: true }, { name: "releases ownership after descendants exit naturally", cancel: false }, ])("$name after root settlement and fragmented output flush", async ({ cancel }) => { const { run, supervisor, scopeKey, cleanup, readPid, release } = await createDescendantScope(); const result = await run.wait(); const decoder = createWindowsOutputDecoder(); const finalTail = decoder.decode(Buffer.from([0xe2, 0x82])) + decoder.flush(); expect(result).toMatchObject({ reason: "exit", exitCode: 0, exitSignal: null }); expect(finalTail).not.toBe(""); expect(result.stdout.replaceAll("\r\n", "\n")).toBe( `owned-stdout-one\nowned-stdout-two\nšŸ˜€${finalTail}`, ); expect(result.stderr.replaceAll("\r\n", "\n")).toBe( `owned-stderr-one\nowned-stderr-two\nšŸ˜€${finalTail}`, ); const descendantPid = await readPid(); activePids.add(descendantPid); expect(descendantPid).toBeGreaterThan(0); expect(isProcessAlive(descendantPid)).toBe(true); await expectPending(run.waitForExtinction!()); if (cancel) { supervisor.cancelScope(scopeKey); supervisor.cancelScope(scopeKey); } else { await release(); } await Promise.all([run.waitForExtinction!(), cleanup(), cleanup()]); await expect(run.wait()).resolves.toMatchObject({ reason: "exit", exitCode: 0 }); await waitForDead(descendantPid, 5_000); }); });