Download src/process/supervisor/supervisor.anchored-shell.real.test.ts from SaylorTwift/openclaw: direct link, hf CLI and curl.
- Browser
- Download file 17.4 kB
-
https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/process/supervisor/supervisor.anchored-shell.real.test.ts
- Command line
-
hf download hf://SaylorTwift/openclaw/src/process/supervisor/supervisor.anchored-shell.real.test.ts
-
curl -L -o supervisor.anchored-shell.real.test.ts https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/process/supervisor/supervisor.anchored-shell.real.test.ts
17.4 kB
| 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<number>(); | |
| 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<void>) { | |
| const settled = await Promise.race([ | |
| promise.then(() => true), | |
| new Promise<false>((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<ReturnType<typeof first.supervisor.spawn>> | 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); | |
| }); | |
| }); | |