diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f14d77b..8585333 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,35 @@ jobs: npm pack --pack-destination /tmp npx -y --package /tmp/roberttlange-headless-*.tgz headless --help + windows-run-storage: + name: Windows run storage / Node ${{ matrix.node-version }} + runs-on: windows-latest + strategy: + fail-fast: false + matrix: + node-version: + - 22 + - 24 + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Node + uses: actions/setup-node@v4 + with: + node-version: ${{ matrix.node-version }} + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Build + run: npm run build + + - name: Test run storage + run: node --import tsx --test tests/run-storage.test.ts tests/run-storage-process.test.ts + python: name: Python ${{ matrix.python-version }} runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b2902e..2645a1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## TBD +- Fixed run coordination to recover stale run and node locks after owner crashes, retain async ownership for the full detached process tree, and handle Windows process probing and state replacement safely (#28). + ## 0.6.1 - 2026-08-12 - Fixed Modal runs to use the immutable, verified v0.6.0 runtime image digest instead of a stale cached `latest` tag. diff --git a/src/async-run-message.ts b/src/async-run-message.ts new file mode 100644 index 0000000..6ee2a27 --- /dev/null +++ b/src/async-run-message.ts @@ -0,0 +1,207 @@ +import { execFileSync, spawn, type ChildProcess } from "node:child_process"; + +import { windowsTaskkillPath } from "./process-tree.js"; +import { + handoffNodeStoreLockOwner, + nodeStoreLockHasLiveSuccessor, + removeRunStoreLockSignalListeners, +} from "./run-storage.js"; +import { + acquireNodeLock, + appendNodeLog, + nodeLockPath, + updateNodeStatus, +} from "./runs.js"; +import type { Env } from "./types.js"; + +export interface AsyncRunMessageTask { + command: { command: string; args: string[] }; + cwd?: string; + runId: string; + nodeId: string; +} + +export type AsyncRunMessageRequest = + | { type: "task"; task: AsyncRunMessageTask } + | { type: "start" } + | { type: "cancel" }; + +export type AsyncRunMessageResponse = + | { type: "ready" } + | { type: "started" } + | { type: "error"; message: string }; + +let task: AsyncRunMessageTask | undefined; +let releaseLock: (() => void) | undefined; +let activeChild: ChildProcess | undefined; +let started = false; +let finished = false; + +if (process.send) { + process.on("message", (message: AsyncRunMessageRequest) => { + void handleRequest(message); + }); + process.once("disconnect", () => { + if (!started && !finished) failBeforeStart("async message parent disconnected before agent startup"); + }); + for (const signal of forwardedSignals()) { + process.on(signal, () => terminateActiveChild(signal)); + } +} + +async function handleRequest(request: AsyncRunMessageRequest): Promise { + if (request.type === "task") { + await prepare(request.task); + return; + } + if (request.type === "cancel") { + if (started) { + terminateActiveChild("SIGTERM"); + } else { + finish(1); + } + return; + } + if (request.type === "start" && task && releaseLock && !started) { + started = true; + await execute(task); + } +} + +function terminateActiveChild(signal: NodeJS.Signals): void { + if (!activeChild) return; + if (process.platform === "win32" && activeChild.pid) { + try { + execFileSync(windowsTaskkillPath(), ["/pid", String(activeChild.pid), "/T", "/F"], { + stdio: "ignore", + timeout: 3_000, + windowsHide: true, + }); + return; + } catch { + return; + } + } + activeChild.kill(signal); +} + +function failBeforeStart(message: string): void { + if (finished) return; + if (task) { + try { + updateNodeStatus(process.env as Env, task.runId, task.nodeId, "failed", message); + } catch { + // Lock cleanup must still run when status persistence fails. + } + } + finish(1); +} + +async function prepare(nextTask: AsyncRunMessageTask): Promise { + if (task || finished) return; + try { + releaseLock = acquireNodeLock(process.env as Env, nextTask.runId, nextTask.nodeId, { + processTreeRootPid: process.pid, + }); + removeRunStoreLockSignalListeners(); + task = nextTask; + send({ type: "ready" }); + } catch (error) { + send({ type: "error", message: errorMessage(error) }); + finish(2); + } +} + +async function execute(currentTask: AsyncRunMessageTask): Promise { + let code = 1; + try { + code = await runChild(currentTask); + if (code !== 0) { + appendNodeLog( + process.env as Env, + currentTask.runId, + currentTask.nodeId, + "stderr", + `async child exited with code ${code}\n`, + ); + } + updateNodeStatus( + process.env as Env, + currentTask.runId, + currentTask.nodeId, + code === 0 ? "idle" : "failed", + ); + } catch (error) { + const message = errorMessage(error); + appendNodeLog(process.env as Env, currentTask.runId, currentTask.nodeId, "stderr", `${message}\n`); + updateNodeStatus(process.env as Env, currentTask.runId, currentTask.nodeId, "failed", message); + } finally { + finish(code); + } +} + +function runChild(currentTask: AsyncRunMessageTask): Promise { + return new Promise((resolve, reject) => { + const child = spawn(currentTask.command.command, currentTask.command.args, { + cwd: currentTask.cwd, + detached: process.platform !== "win32", + env: { + ...process.env, + HEADLESS_ASYNC_MESSAGE_OWNER_PID: String(process.pid), + HEADLESS_ASYNC_MESSAGE_WORKER: "1", + }, + stdio: ["ignore", "ignore", "inherit"], + }); + activeChild = child; + child.once("error", reject); + child.once("spawn", () => { + try { + if (!child.pid) throw new Error("async message CLI started without a process ID"); + handoffNodeStoreLockOwner( + nodeLockPath(process.env as Env, currentTask.runId, currentTask.nodeId), + process.pid, + { processTreeRootPid: child.pid }, + ); + send({ type: "started" }); + } catch (error) { + terminateActiveChild("SIGKILL"); + reject(error); + } + }); + child.once("close", (code, signal) => { + activeChild = undefined; + resolve(signal ? 1 : (code ?? 1)); + }); + }); +} + +function finish(code = 0): void { + if (finished) return; + finished = true; + try { + if (!task || !nodeStoreLockHasLiveSuccessor( + nodeLockPath(process.env as Env, task.runId, task.nodeId), + process.pid, + )) { + releaseLock?.(); + } + } finally { + releaseLock = undefined; + if (process.connected) process.disconnect(); + process.exitCode = code; + } +} + +function send(response: AsyncRunMessageResponse): void { + if (process.send) process.send(response); +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function forwardedSignals(): NodeJS.Signals[] { + return process.platform === "win32" + ? ["SIGINT", "SIGTERM", "SIGBREAK"] + : ["SIGHUP", "SIGINT", "SIGTERM", "SIGQUIT"]; +} diff --git a/src/cli.ts b/src/cli.ts index 149e9ff..b265e19 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -102,9 +102,11 @@ import { handleCronCommand as handleCronCommandImpl, type CronCommand } from "./ import { runCronDaemon } from "./cron.js"; import { extractRunNodeMetrics } from "./run-metrics.js"; import { createRunStatusReporter, parseRunStatusIntervalMs } from "./run-status.js"; +import { isRunStoreLockSignalListener, waitForNodeStoreLockOwner } from "./run-storage.js"; import { appendNodeLog, completeIdleRunNodes, + nodeLockPath, readRun, registerNode, runDirectory, @@ -1887,14 +1889,25 @@ async function executeCommand( result.stdoutEndsWithNewline = stdoutEndsWithNewline; resolve(result); }; - const ownsChildProcessGroup = - process.platform !== "win32" && - (options.timeoutSeconds !== undefined || options.cleanupBeforeParentSignalExit !== undefined); - const handlesParentSignals = ownsChildProcessGroup || options.cleanupBeforeParentSignalExit !== undefined; + const asyncMessageWorker = env.HEADLESS_ASYNC_MESSAGE_WORKER === "1"; + const ownsChildProcessGroup = !asyncMessageWorker && process.platform !== "win32" && ( + options.timeoutSeconds !== undefined + || options.cleanupBeforeParentSignalExit !== undefined + ); + const handlesParentSignals = asyncMessageWorker + || ownsChildProcessGroup + || options.cleanupBeforeParentSignalExit !== undefined; + waitForAsyncMessageOwnership(env, command); + let childEnv = commandEnv(env, command); + if (asyncMessageWorker) { + childEnv = { ...childEnv }; + delete childEnv.HEADLESS_ASYNC_MESSAGE_OWNER_PID; + delete childEnv.HEADLESS_ASYNC_MESSAGE_WORKER; + } const child = spawn(command.command, command.args, { cwd, detached: ownsChildProcessGroup, - env: commandEnv(env, command) as NodeJS.ProcessEnv, + env: childEnv as NodeJS.ProcessEnv, stdio, }); @@ -3225,6 +3238,23 @@ function withRunEnvironment(command: BuiltCommand, runId: string | undefined, no }; } +function waitForAsyncMessageOwnership(env: Env, command: BuiltCommand): void { + const ownerValue = env.HEADLESS_ASYNC_MESSAGE_OWNER_PID; + if (ownerValue === undefined) return; + const expectedProcessTreeRootPid = Number(ownerValue); + const runId = command.env?.HEADLESS_RUN_ID; + const nodeId = command.env?.HEADLESS_RUN_NODE; + if ( + !Number.isSafeInteger(expectedProcessTreeRootPid) + || expectedProcessTreeRootPid <= 0 + || !runId + || !nodeId + ) { + throw new Error("invalid async message ownership context"); + } + waitForNodeStoreLockOwner(nodeLockPath(env, runId, nodeId), process.pid); +} + async function executeStoredNode( node: { agent: AgentName; @@ -3370,7 +3400,11 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise [ signal, - new Set(process.listeners(signal).filter((listener) => !isDockerSessionLockSignalListener(listener))), + new Set( + process.listeners(signal).filter( + (listener) => !isDockerSessionLockSignalListener(listener) && !isRunStoreLockSignalListener(listener), + ), + ), ] as const), ); let registeredRunNode: { runId: string; nodeId: string } | undefined; diff --git a/src/run-commands.ts b/src/run-commands.ts index 8a639e8..57750d1 100644 --- a/src/run-commands.ts +++ b/src/run-commands.ts @@ -1,6 +1,7 @@ import { closeSync, openSync } from "node:fs"; -import { spawn } from "node:child_process"; +import { fork, type ChildProcess } from "node:child_process"; import { join } from "node:path"; +import { fileURLToPath } from "node:url"; import { extractFinalMessage } from "./output.js"; import { deriveNativeTranscriptActivity, nativeTranscriptKey, resolveLatestNativeTranscripts } from "./native-transcripts.js"; @@ -10,7 +11,6 @@ import { acquireNodeLock, appendNodeLog, listRuns, - nodeLockPath, readRun, recordMessage, runDirectory, @@ -22,6 +22,7 @@ import { quoteCommand } from "./shell.js"; import { renderSdkResult, type SdkFormat } from "./sdk.js"; import type { AgentName, Env } from "./types.js"; import type { RunStatus } from "./roles.js"; +import type { AsyncRunMessageRequest, AsyncRunMessageResponse, AsyncRunMessageTask } from "./async-run-message.js"; export interface RunCommandInput { command: "list" | "view" | "mark" | "message" | "wait"; @@ -195,14 +196,14 @@ async function handleRunMessage( if (input.printCommand) { const command = input.async - ? buildAsyncRunMessageCommand(handlers.env, runId, nodeId, node, prompt.prompt) + ? buildAsyncMessageCliCommand(handlers.env, runId, nodeId, prompt.prompt) : buildNodeInvocationCommand(handlers.env, runId, nodeId, node, prompt.prompt); handlers.stdout(`${quoteCommand(command)}\n`); return 0; } if (input.async) { - return startAsyncRunMessage(handlers, runId, nodeId, node, prompt.prompt); + return await startAsyncRunMessage(handlers, runId, nodeId, node, prompt.prompt); } const releaseLock = acquireNodeLock(handlers.env, runId, nodeId); @@ -301,41 +302,147 @@ function tmuxTranscriptScope(agent: AgentName, workDir: string | undefined): str return `${agent}\t${workDir ?? ""}`; } -function startAsyncRunMessage( +async function startAsyncRunMessage( handlers: RunCommandHandlers, runId: string, nodeId: string, node: RunNode, prompt: string, -): number { - const releaseLock = acquireNodeLock(handlers.env, runId, nodeId); +): Promise { const stderrLog = node.logs?.stderr ?? join(runDirectory(handlers.env, runId), "nodes", nodeId, "latest.stderr.log"); - recordMessage(handlers.env, runId, handlers.env.HEADLESS_RUN_NODE || "cli", nodeId, prompt); - updateNodeStatus(handlers.env, runId, nodeId, "busy"); - appendNodeLog(handlers.env, runId, nodeId, "stdout", `\n===== async message ${new Date().toISOString()} =====\n`); - appendNodeLog(handlers.env, runId, nodeId, "stderr", `\n===== async message ${new Date().toISOString()} =====\n`); - const command = buildAsyncRunMessageCommand(handlers.env, runId, nodeId, node, prompt); - const errFd = openSync(stderrLog, "a"); + let worker: ChildProcess; try { - const childProcess = spawn(command.command, command.args, { - cwd: node.workDir, + worker = fork(asyncRunMessageWorkerPath(), [], { env: handlers.env as NodeJS.ProcessEnv, detached: true, - stdio: ["ignore", "ignore", errFd], + stdio: ["ignore", "ignore", errFd, "ipc"], }); - childProcess.unref(); - } catch (error) { - releaseLock(); - updateNodeStatus(handlers.env, runId, nodeId, "failed", error instanceof Error ? error.message : String(error)); - throw error; } finally { closeSync(errFd); } + + const task: AsyncRunMessageTask = { + command: buildNodeInvocationCommand(handlers.env, runId, nodeId, node, prompt), + cwd: node.workDir, + runId, + nodeId, + }; + await prepareAsyncWorker(worker, task); + try { + recordMessage(handlers.env, runId, handlers.env.HEADLESS_RUN_NODE || "cli", nodeId, prompt); + updateNodeStatus(handlers.env, runId, nodeId, "busy"); + const timestamp = new Date().toISOString(); + appendNodeLog(handlers.env, runId, nodeId, "stdout", `\n===== async message ${timestamp} =====\n`); + appendNodeLog(handlers.env, runId, nodeId, "stderr", `\n===== async message ${timestamp} =====\n`); + await startPreparedAsyncWorker(worker); + } catch (error) { + try { + updateNodeStatus( + handlers.env, + runId, + nodeId, + "failed", + error instanceof Error ? error.message : String(error), + ); + } catch { + // Preserve the startup error when rollback storage also fails. + } + cancelAsyncWorker(worker); + throw error; + } + if (worker.connected) worker.disconnect(); + worker.unref(); handlers.stdout(`started: ${runId}/${nodeId}\n`); return 0; } +function prepareAsyncWorker(worker: ChildProcess, task: AsyncRunMessageTask): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => fail(new Error("async message worker did not become ready")), 5_000); + timeout.unref(); + const cleanup = () => { + clearTimeout(timeout); + worker.off("message", onMessage); + worker.off("error", fail); + worker.off("exit", onExit); + }; + const fail = (error: Error) => { + cleanup(); + cancelAsyncWorker(worker); + reject(error); + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + fail(new Error(`async message worker exited before startup (${signal ?? code ?? "unknown"})`)); + }; + const onMessage = (message: AsyncRunMessageResponse) => { + if (message.type === "error") { + fail(new Error(message.message)); + return; + } + cleanup(); + resolve(); + }; + worker.once("error", fail); + worker.once("exit", onExit); + worker.on("message", onMessage); + void sendWorkerRequest(worker, { type: "task", task }).catch(fail); + }); +} + +function sendWorkerRequest(worker: ChildProcess, request: AsyncRunMessageRequest): Promise { + return new Promise((resolve, reject) => { + worker.send(request, (error) => error ? reject(error) : resolve()); + }); +} + +function startPreparedAsyncWorker(worker: ChildProcess): Promise { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => fail(new Error("async message worker did not start its agent")), 5_000); + timeout.unref(); + const cleanup = () => { + clearTimeout(timeout); + worker.off("message", onMessage); + worker.off("error", fail); + worker.off("exit", onExit); + }; + const fail = (error: Error) => { + cleanup(); + reject(error); + }; + const onExit = (code: number | null, signal: NodeJS.Signals | null) => { + fail(new Error(`async message worker exited before agent startup (${signal ?? code ?? "unknown"})`)); + }; + const onMessage = (message: AsyncRunMessageResponse) => { + if (message.type === "error") { + fail(new Error(message.message)); + } else if (message.type === "started") { + cleanup(); + resolve(); + } + }; + worker.once("error", fail); + worker.once("exit", onExit); + worker.on("message", onMessage); + void sendWorkerRequest(worker, { type: "start" }).catch(fail); + }); +} + +function cancelAsyncWorker(worker: ChildProcess): void { + worker.once("error", () => undefined); + if (worker.connected) { + worker.send({ type: "cancel" } satisfies AsyncRunMessageRequest, () => { + if (worker.connected) worker.disconnect(); + }); + } + worker.unref(); +} + +function asyncRunMessageWorkerPath(): string { + const extension = import.meta.url.endsWith(".ts") ? "ts" : "js"; + return fileURLToPath(new URL(`./async-run-message.${extension}`, import.meta.url)); +} + function buildNodeInvocationCommand(env: Env, runId: string, nodeId: string, node: RunNode, prompt: string): { command: string; args: string[]; @@ -366,27 +473,14 @@ function buildNodeInvocationCommand(env: Env, runId: string, nodeId: string, nod }; } -function buildAsyncRunMessageCommand(env: Env, runId: string, nodeId: string, node: RunNode, prompt: string): { +function buildAsyncMessageCliCommand(env: Env, runId: string, nodeId: string, prompt: string): { command: string; args: string[]; } { - const stderrLog = node.logs?.stderr ?? join(runDirectory(env, runId), "nodes", nodeId, "latest.stderr.log"); - const child = quoteCommand(buildNodeInvocationCommand(env, runId, nodeId, node, prompt)); - const cli = headlessCli(env); - const success = quoteCommand({ command: cli, args: ["run", "mark", runId, nodeId, "--status", "idle"] }); - const failure = quoteCommand({ command: cli, args: ["run", "mark", runId, nodeId, "--status", "failed"] }); - const unlock = quoteCommand({ command: "rm", args: ["-f", nodeLockPath(env, runId, nodeId)] }); - const quotedStderrLog = quotePath(stderrLog); - const signalFailure = `${failure} >/dev/null 2>> ${quotedStderrLog}; ${unlock}; exit 143`; - const script = [ - `trap "${signalFailure}" INT TERM HUP`, - `trap "${unlock}" EXIT`, - `${child} >/dev/null 2>> ${quotedStderrLog}`, - "code=$?", - `if [ "$code" -eq 0 ]; then ${success} >/dev/null 2>> ${quotedStderrLog}; else printf '%s\\n' "async child exited with code $code" >> ${quotedStderrLog}; ${failure} >/dev/null 2>> ${quotedStderrLog}; fi`, - 'exit "$code"', - ].join("; "); - return { command: "sh", args: ["-c", script] }; + return { + command: headlessCli(env), + args: ["run", "message", runId, nodeId, "--prompt", prompt, "--async"], + }; } function headlessCli(env: Env): string { @@ -401,10 +495,6 @@ function parseDelayMs(value: string | undefined, fallback: number): number { return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback; } -function quotePath(path: string): string { - return quoteCommand({ command: path, args: [] }); -} - function requireValue(value: string | undefined, label: string): string { if (!value) { throw new Error(`invalid ${label}; use letters, numbers, dots, dashes, or underscores`); diff --git a/src/run-storage.ts b/src/run-storage.ts new file mode 100644 index 0000000..a47df86 --- /dev/null +++ b/src/run-storage.ts @@ -0,0 +1,529 @@ +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { + chmodSync, + closeSync, + fchmodSync, + lstatSync, + openSync, + readFileSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { createRequire } from "node:module"; +import { win32 } from "node:path"; + +import type { lockSync as LockSync } from "proper-lockfile"; + +const privateDirMode = 0o700; +const legacyInitializationGraceMs = 1_000; +const runLockStaleMs = 5_000; +const runLockUpdateMs = 2_500; +const nodeLockStaleMs = 20_000; +const nodeLockUpdateMs = 5_000; +const runLockTimeoutMs = 30_000; +const runLockRetryMs = 10; +const maximumProcessId = 0xffff_ffff; +const windowsProcessTreeTimeoutMs = 3_000; +const windowsProcessIdentityTimeoutMs = 5_000; +const windowsOwnerIdentityTimeoutMs = 10_000; +const windowsRenameAttempts = 40; +const windowsRenameRetryMs = 25; +const windowsRenameRetryCodes = new Set(["EACCES", "EBUSY", "EPERM"]); +const managedSignals: NodeJS.Signals[] = process.platform === "win32" + ? ["SIGINT", "SIGTERM", "SIGBREAK"] + : ["SIGHUP", "SIGINT", "SIGTERM", "SIGQUIT"]; +const runStoreLockSignalListeners = new Set(); +let lockSync: typeof LockSync | undefined; + +export interface NodeStoreLockOwner { + processTreeRootPid: number; +} + +interface StoredNodeStoreLockOwner extends NodeStoreLockOwner { + createdAtMs: number; + processStartIdentity?: string; +} + +interface StoredNodeStoreLockOwnerSnapshot { + owner: StoredNodeStoreLockOwner; +} + +export interface RunStateReplacementOptions { + platform?: NodeJS.Platform; + rename?: (source: string, destination: string) => void; + sleep?: (milliseconds: number) => void; +} + +export interface WindowsProcessTreeProbeOptions { + execute?: (command: string, args: string[]) => string; + systemRoot?: string; +} + +export interface MacosProcessStartIdentityOptions { + env?: NodeJS.ProcessEnv; + execute?: ( + command: string, + args: string[], + options: { env: NodeJS.ProcessEnv }, + ) => string; +} + +export function acquireRunStoreLock(lockPath: string, runId: string): () => void { + const deadline = Date.now() + runLockTimeoutMs; + while (true) { + try { + return acquireStoreLock(lockPath, runLockStaleMs, runLockUpdateMs); + } catch (error) { + if (!isLockContention(error)) throw error; + if (Date.now() >= deadline) throw new Error(`run is locked: ${runId}`); + sleepSync(runLockRetryMs); + } + } +} + +export function acquireNodeStoreLock(lockPath: string, nodeId: string, owner?: NodeStoreLockOwner): () => void { + try { + return acquireStoreLock(lockPath, nodeLockStaleMs, nodeLockUpdateMs, owner); + } catch (error) { + if (!isLockContention(error)) throw error; + throw new Error(`node is locked: ${nodeId}`); + } +} + +export function handoffNodeStoreLockOwner( + lockPath: string, + expectedProcessTreeRootPid: number, + owner: NodeStoreLockOwner, +): void { + if (!lstatSync(lockPath).isDirectory()) throw new Error("node lock lease is missing"); + const storedOwner = readStoredLockOwner(lockPath)?.owner; + if (storedOwner?.processTreeRootPid !== expectedProcessTreeRootPid) { + throw new Error("node lock owner changed before handoff"); + } + writeLockOwner(lockPath, owner); +} + +export function waitForNodeStoreLockOwner( + lockPath: string, + processTreeRootPid: number, + timeoutMs = 5_000, +): void { + const deadline = Date.now() + timeoutMs; + do { + if (readStoredLockOwner(lockPath)?.owner.processTreeRootPid === processTreeRootPid) return; + sleepSync(runLockRetryMs); + } while (Date.now() < deadline); + throw new Error("async message lock ownership handoff timed out"); +} + +export function nodeStoreLockHasLiveSuccessor(lockPath: string, currentProcessTreeRootPid: number): boolean { + try { + const snapshot = readStoredLockOwner(lockPath); + if (!snapshot || snapshot.owner.processTreeRootPid === currentProcessTreeRootPid) return false; + if (!storedOwnerIsTrusted(snapshot.owner)) return true; + return processTreeAlive(snapshot.owner); + } catch { + return true; + } +} + +export function isRunStoreLockSignalListener(listener: NodeJS.SignalsListener): boolean { + return runStoreLockSignalListeners.has(listener); +} + +export function removeRunStoreLockSignalListeners(): void { + for (const signal of managedSignals) { + for (const listener of runStoreLockSignalListeners) process.off(signal, listener); + } +} + +export function replaceRunStateFile( + source: string, + destination: string, + options: RunStateReplacementOptions = {}, +): void { + const platform = options.platform ?? process.platform; + const rename = options.rename ?? renameSync; + const sleep = options.sleep ?? sleepSync; + for (let attempt = 1; attempt <= windowsRenameAttempts; attempt += 1) { + try { + rename(source, destination); + return; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + const retryable = platform === "win32" && code !== undefined && windowsRenameRetryCodes.has(code); + if (!retryable || attempt === windowsRenameAttempts) throw error; + sleep(windowsRenameRetryMs); + } + } +} + +function acquireStoreLock( + lockPath: string, + stale: number, + update: number, + owner?: NodeStoreLockOwner, +): () => void { + if (leaseOwnerBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); + if (legacyLockBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); + const storedOwner = owner ? createStoredLockOwner(owner, windowsOwnerIdentityTimeoutMs) : undefined; + + let compromisedError: Error | undefined; + const release = loadLockSync()(lockPath, { + lockfilePath: lockPath, + onCompromised: (error) => { + compromisedError = error; + }, + realpath: false, + retries: 0, + stale, + update, + }); + let released = false; + const releaseLease = () => { + if (released) return; + released = true; + release(); + }; + try { + chmodSync(lockPath, privateDirMode); + if (ownerSidecarBlocksAcquisition(lockPath)) { + releaseLease(); + throw lockContentionError(lockPath); + } + rmSync(lockOwnerPath(lockPath), { force: true }); + if (storedOwner) writeStoredLockOwner(lockPath, storedOwner); + } catch (error) { + releaseLease(); + throw error; + } + + return () => { + if (owner) rmSync(lockOwnerPath(lockPath), { force: true }); + if (compromisedError) throw compromisedError; + releaseLease(); + }; +} + +function leaseOwnerBlocksAcquisition(lockPath: string): boolean { + return storedOwnerBlocksAcquisition(lockPath); +} + +function ownerSidecarBlocksAcquisition(lockPath: string): boolean { + return storedOwnerBlocksAcquisition(lockPath); +} + +function storedOwnerBlocksAcquisition(lockPath: string): boolean { + let snapshot: StoredNodeStoreLockOwnerSnapshot | undefined; + try { + snapshot = readStoredLockOwner(lockPath); + } catch (error) { + if (error instanceof SyntaxError) return false; + return true; + } + if (!snapshot || !storedOwnerIsTrusted(snapshot.owner)) return false; + return processTreeAlive(snapshot.owner); +} + +function readStoredLockOwner(lockPath: string): StoredNodeStoreLockOwnerSnapshot | undefined { + const ownerPath = lockOwnerPath(lockPath); + try { + const owner = JSON.parse(readFileSync(ownerPath, "utf8")) as StoredNodeStoreLockOwner; + return { owner }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw error; + } +} + +function storedOwnerIsTrusted(owner: StoredNodeStoreLockOwner): boolean { + return validProcessId(owner.processTreeRootPid) && Number.isFinite(owner.createdAtMs); +} + +function validProcessId(pid: number): boolean { + return Number.isSafeInteger(pid) && pid > 0 && pid <= maximumProcessId; +} + +function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { + writeStoredLockOwner(lockPath, createStoredLockOwner(owner, windowsProcessIdentityTimeoutMs)); +} + +function createStoredLockOwner( + owner: NodeStoreLockOwner, + windowsIdentityTimeoutMs: number, +): StoredNodeStoreLockOwner { + return { + ...owner, + createdAtMs: Date.now(), + processStartIdentity: processStartIdentity(owner.processTreeRootPid, windowsIdentityTimeoutMs), + }; +} + +function writeStoredLockOwner(lockPath: string, storedOwner: StoredNodeStoreLockOwner): void { + const path = lockOwnerPath(lockPath); + const temporaryPath = `${path}.tmp-${randomUUID()}`; + let descriptor: number | undefined; + try { + descriptor = openSync(temporaryPath, "wx", 0o600); + writeFileSync(descriptor, `${JSON.stringify(storedOwner)}\n`); + fchmodSync(descriptor, 0o600); + closeSync(descriptor); + descriptor = undefined; + replaceRunStateFile(temporaryPath, path); + } finally { + if (descriptor !== undefined) closeSync(descriptor); + rmSync(temporaryPath, { force: true }); + } +} + +function lockOwnerPath(lockPath: string): string { + return `${lockPath}.owner`; +} + +function loadLockSync(): typeof LockSync { + if (lockSync) return lockSync; + + const listenersBeforeImport = new Map( + managedSignals.map((signal) => [signal, new Set(process.listeners(signal))] as const), + ); + lockSync = (createRequire(import.meta.url)("proper-lockfile") as { lockSync: typeof LockSync }).lockSync; + for (const signal of managedSignals) { + for (const listener of process.listeners(signal)) { + if (!listenersBeforeImport.get(signal)?.has(listener)) runStoreLockSignalListeners.add(listener); + } + } + return lockSync; +} + +function legacyLockBlocksAcquisition(lockPath: string): boolean { + let status; + try { + status = lstatSync(lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } + if (!status.isFile()) return false; + + const pid = readLegacyLockPid(lockPath); + const malformedAndFresh = pid === undefined && Date.now() - status.mtimeMs < legacyInitializationGraceMs; + if (malformedAndFresh || (pid !== undefined && processAlive(pid))) return true; + + try { + rmSync(lockPath, { force: true }); + return false; + } catch (error) { + try { + if (lstatSync(lockPath).isDirectory()) return true; + } catch (statusError) { + if ((statusError as NodeJS.ErrnoException).code === "ENOENT") return false; + } + throw error; + } +} + +function readLegacyLockPid(lockPath: string): number | undefined { + let value: string; + try { + value = readFileSync(lockPath, "utf8").trim(); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw error; + } + if (!/^[1-9]\d*$/.test(value)) return undefined; + const pid = Number(value); + return Number.isSafeInteger(pid) ? pid : undefined; +} + +function processAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +function processTreeAlive(owner: StoredNodeStoreLockOwner): boolean { + const rootPid = owner.processTreeRootPid; + const rootAlive = processAlive(rootPid); + let identityMismatch = false; + if (rootAlive) { + const currentIdentity = processStartIdentity(rootPid); + identityMismatch = Boolean( + owner.processStartIdentity + && currentIdentity + && owner.processStartIdentity !== currentIdentity, + ); + } + return processTreeAliveFromProbes(process.platform, rootAlive, identityMismatch, () => { + if (process.platform === "win32") return windowsProcessTreeAlive(rootPid); + try { + process.kill(-rootPid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } + }); +} + +export function processTreeAliveFromProbes( + platform: NodeJS.Platform, + rootAlive: boolean, + identityMismatch: boolean, + descendantsAlive: () => boolean, +): boolean { + if (rootAlive && !identityMismatch) return true; + if (rootAlive && platform !== "win32") return false; + return descendantsAlive(); +} + +export function windowsProcessTreeAlive( + rootPid: number, + options: WindowsProcessTreeProbeOptions = {}, +): boolean { + if (!validProcessId(rootPid)) return true; + const execute = options.execute + ?? ((command, args) => executeWindowsPowerShell(command, args, windowsProcessTreeTimeoutMs)); + try { + const output = execute(windowsPowerShellPath(options.systemRoot), [ + "-NoProfile", + "-NonInteractive", + "-Command", + windowsDescendantProbe(rootPid), + ]); + return output.trim() !== "HEADLESS_PROCESS_TREE_DEAD"; + } catch { + return true; + } +} + +export function windowsProcessStartIdentity( + pid: number, + options: WindowsProcessTreeProbeOptions = {}, +): string | undefined { + return probeWindowsProcessStartIdentity(pid, options, windowsProcessIdentityTimeoutMs); +} + +function probeWindowsProcessStartIdentity( + pid: number, + options: WindowsProcessTreeProbeOptions, + timeoutMs: number, +): string | undefined { + if (!validProcessId(pid)) return undefined; + const execute = options.execute + ?? ((command, args) => executeWindowsPowerShell(command, args, timeoutMs)); + try { + const output = execute(windowsPowerShellPath(options.systemRoot), [ + "-NoProfile", + "-NonInteractive", + "-Command", + windowsProcessStartIdentityProbe(pid), + ]).trim(); + const match = /^HEADLESS_PROCESS_START:(\d+)$/.exec(output); + return match ? `win32:${match[1]}` : undefined; + } catch { + return undefined; + } +} + +function executeWindowsPowerShell(command: string, args: string[], timeout: number): string { + return execFileSync(command, args, { + encoding: "utf8", + timeout, + windowsHide: true, + }); +} + +function processStartIdentity(pid: number, windowsTimeoutMs = windowsProcessIdentityTimeoutMs): string | undefined { + if (process.platform === "linux") { + try { + const stat = readFileSync(`/proc/${pid}/stat`, "utf8"); + const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" "); + return fields[19]; + } catch { + return undefined; + } + } + if (process.platform === "darwin") { + return macosProcessStartIdentity(pid); + } + if (process.platform === "win32") { + return probeWindowsProcessStartIdentity(pid, {}, windowsTimeoutMs); + } + return undefined; +} + +export function macosProcessStartIdentity( + pid: number, + options: MacosProcessStartIdentityOptions = {}, +): string | undefined { + const execute = options.execute ?? executeMacosProcessStartIdentityProbe; + const env = { ...(options.env ?? process.env), LC_ALL: "C", TZ: "UTC" }; + try { + const startedAt = execute( + "/bin/ps", + ["-o", "lstart=", "-p", String(pid)], + { env }, + ).trim(); + return startedAt ? `darwin:${startedAt}` : undefined; + } catch { + return undefined; + } +} + +function executeMacosProcessStartIdentityProbe( + command: string, + args: string[], + options: { env: NodeJS.ProcessEnv }, +): string { + return execFileSync(command, args, { + ...options, + encoding: "utf8", + timeout: 3_000, + }); +} + +function windowsPowerShellPath(systemRoot = process.env.SystemRoot): string { + const root = systemRoot && win32.isAbsolute(systemRoot) ? systemRoot : "C:\\Windows"; + return win32.join(root, "System32", "WindowsPowerShell", "v1.0", "powershell.exe"); +} + +function windowsDescendantProbe(rootPid: number): string { + return [ + "$ErrorActionPreference = 'Stop'", + `$rootPid = [uint32]${rootPid}`, + "$processes = @(Get-CimInstance Win32_Process | Where-Object { $_.ProcessId -ne $PID })", + "$parents = @($rootPid)", + "$found = $false", + "do {", + " $children = @($processes | Where-Object { $parents -contains $_.ParentProcessId })", + " if ($children.Count -gt 0) { $found = $true }", + " $parents = @($children | ForEach-Object { $_.ProcessId })", + "} while ($parents.Count -gt 0)", + "if ($found) { 'HEADLESS_PROCESS_TREE_ALIVE' } else { 'HEADLESS_PROCESS_TREE_DEAD' }", + ].join("; "); +} + +function windowsProcessStartIdentityProbe(rootPid: number): string { + return [ + "$ErrorActionPreference = 'Stop'", + `$rootPid = [uint32]${rootPid}`, + "$process = Get-CimInstance Win32_Process -Filter \"ProcessId = $rootPid\"", + "if ($null -ne $process) { 'HEADLESS_PROCESS_START:' + $process.CreationDate.ToUniversalTime().Ticks }", + ].join("; "); +} + +function lockContentionError(lockPath: string): Error { + return Object.assign(new Error(`lock is already held: ${lockPath}`), { code: "ELOCKED" }); +} + +function isLockContention(error: unknown): boolean { + return (error as NodeJS.ErrnoException).code === "ELOCKED"; +} + +function sleepSync(milliseconds: number): void { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, milliseconds); +} diff --git a/src/runs.ts b/src/runs.ts index 19813a4..c76e430 100644 --- a/src/runs.ts +++ b/src/runs.ts @@ -3,17 +3,24 @@ import { chmodSync, closeSync, existsSync, + fchmodSync, mkdirSync, openSync, readFileSync, readdirSync, - renameSync, rmSync, writeFileSync, } from "node:fs"; +import { randomUUID } from "node:crypto"; import { basename, dirname, join } from "node:path"; import type { CoordinationMode, Role, RunStatus } from "./roles.js"; +import { + acquireNodeStoreLock, + acquireRunStoreLock, + replaceRunStateFile, + type NodeStoreLockOwner, +} from "./run-storage.js"; import type { AgentName, AllowMode, Env, ReasoningEffort } from "./types.js"; const privateDirMode = 0o700; @@ -361,21 +368,10 @@ export function recordMessage( }); } -export function acquireNodeLock(env: Env, runId: string, nodeId: string): () => void { +export function acquireNodeLock(env: Env, runId: string, nodeId: string, owner?: NodeStoreLockOwner): () => void { const lockPath = nodeLockPath(env, runId, nodeId); ensurePrivateDir(dirname(lockPath)); - let fd: number; - try { - fd = openSync(lockPath, "wx", privateFileMode); - chmodSync(lockPath, privateFileMode); - } catch { - throw new Error(`node is locked: ${nodeId}`); - } - writeFileSync(fd, `${process.pid}\n`); - return () => { - closeSync(fd); - rmSync(lockPath, { force: true }); - }; + return acquireNodeStoreLock(lockPath, nodeId, owner); } export function writeRun(env: Env, run: RunRecord): void { @@ -386,10 +382,19 @@ export function writeRun(env: Env, run: RunRecord): void { } run.updatedAt = new Date().toISOString(); const path = join(dir, "run.json"); - const tmpPath = `${path}.tmp-${process.pid}`; - writeFileSync(tmpPath, `${JSON.stringify(run, null, 2)}\n`, { mode: privateFileMode }); - chmodSync(tmpPath, privateFileMode); - renameSync(tmpPath, path); + const tmpPath = `${path}.tmp-${randomUUID()}`; + let descriptor: number | undefined; + try { + descriptor = openSync(tmpPath, "wx", privateFileMode); + writeFileSync(descriptor, `${JSON.stringify(run, null, 2)}\n`); + fchmodSync(descriptor, privateFileMode); + closeSync(descriptor); + descriptor = undefined; + replaceRunStateFile(tmpPath, path); + } finally { + if (descriptor !== undefined) closeSync(descriptor); + rmSync(tmpPath, { force: true }); + } chmodSync(path, privateFileMode); const eventsPath = join(dir, "events.jsonl"); const event = run.events.at(-1); @@ -411,29 +416,7 @@ function acquireRunLock(env: Env, runId: string): () => void { const dir = runDirectory(env, runId); ensurePrivateDir(dir); const lockPath = join(dir, "run.lock"); - const deadline = Date.now() + 30000; - let fd: number; - while (true) { - try { - fd = openSync(lockPath, "wx", privateFileMode); - chmodSync(lockPath, privateFileMode); - break; - } catch { - if (Date.now() >= deadline) { - throw new Error(`run is locked: ${runId}`); - } - sleepSync(10); - } - } - writeFileSync(fd, `${process.pid}\n`); - return () => { - closeSync(fd); - rmSync(lockPath, { force: true }); - }; -} - -function sleepSync(milliseconds: number): void { - Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, milliseconds); + return acquireRunStoreLock(lockPath, runId); } function requireRun(env: Env, runId: string): RunRecord { diff --git a/tests/async-run-message.test.ts b/tests/async-run-message.test.ts new file mode 100644 index 0000000..32b24db --- /dev/null +++ b/tests/async-run-message.test.ts @@ -0,0 +1,342 @@ +import assert from "node:assert/strict"; +import { fork } from "node:child_process"; +import { once } from "node:events"; +import { chmod, mkdir, writeFile } from "node:fs/promises"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import { runCli } from "../src/cli.ts"; +import { + acquireNodeLock, + nodeLockPath, + readRun, + registerNode, + runDirectory, + updateNodeStatus, +} from "../src/runs.ts"; +import type { AsyncRunMessageResponse, AsyncRunMessageTask } from "../src/async-run-message.ts"; +import type { Env } from "../src/types.ts"; + +interface AsyncMessageFixture { + directory: string; + env: Env; +} + +async function writeExecutable(path: string, source: string): Promise { + await mkdir(join(path, ".."), { recursive: true }); + await writeFile(path, source); + await chmod(path, 0o755); +} + +async function waitFor(assertion: () => boolean): Promise { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + if (assertion()) return; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + assert.equal(assertion(), true); +} + +function processAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function createFixture(agentSource?: string, extraEnv: Env = {}): Promise { + const directory = mkdtempSync(join(tmpdir(), "headless-async-message-test-")); + const home = join(directory, "home"); + const binDirectory = join(directory, "bin"); + await mkdir(home); + await writeExecutable( + join(binDirectory, "headless"), + [ + "#!/bin/sh", + `exec "${process.execPath}" --import tsx "${join(process.cwd(), "src", "cli.ts")}" "$@"`, + "", + ].join("\n"), + ); + if (agentSource) await writeExecutable(join(binDirectory, "codex"), agentSource); + const env = { + ...process.env, + ...extraEnv, + HEADLESS_CLI_BIN: join(binDirectory, "headless"), + HOME: home, + PATH: binDirectory, + }; + registerNode(env, { + runId: "auth", + nodeId: "worker-1", + role: "worker", + agent: "codex", + coordination: "oneshot", + status: "idle", + planned: true, + }); + return { directory, env }; +} + +async function startAsyncMessage(env: Env): Promise { + assert.equal( + await runCli(["run", "message", "auth", "worker-1", "--prompt", "continue", "--async"], { + env, + stdout: () => undefined, + }), + 0, + ); +} + +test("async run message preserves ownership while a signaled agent remains alive", { skip: process.platform === "win32" }, async () => { + const processDirectory = mkdtempSync(join(tmpdir(), "headless-async-agent-test-")); + const agentProcessFile = join(processDirectory, "agent-process.json"); + const signalFile = join(processDirectory, "agent-signal"); + const agentSource = [ + `#!${process.execPath}`, + "const fs = require('node:fs');", + "fs.writeFileSync(process.env.HEADLESS_AGENT_PROCESS, JSON.stringify({", + " pid: process.pid,", + " ppid: process.ppid,", + " ownerMarker: process.env.HEADLESS_ASYNC_MESSAGE_OWNER_PID ?? null,", + " workerMarker: process.env.HEADLESS_ASYNC_MESSAGE_WORKER ?? null,", + "}));", + "process.on('SIGTERM', () => fs.writeFileSync(process.env.HEADLESS_AGENT_SIGNAL, 'SIGTERM\\n'));", + "setInterval(() => undefined, 1000);", + "", + ].join("\n"); + const { directory, env } = await createFixture(agentSource, { + HEADLESS_AGENT_PROCESS: agentProcessFile, + HEADLESS_AGENT_SIGNAL: signalFile, + }); + let agentPid: number | undefined; + let cliPid: number | undefined; + try { + await startAsyncMessage(env); + await waitFor(() => existsSync(agentProcessFile)); + const agentProcess = JSON.parse(readFileSync(agentProcessFile, "utf8")); + ({ pid: agentPid, ppid: cliPid } = agentProcess); + assert.equal(agentProcess.ownerMarker, null); + assert.equal(agentProcess.workerMarker, null); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + const ownerPath = `${lockPath}.owner`; + await waitFor(() => JSON.parse(readFileSync(ownerPath, "utf8")).processTreeRootPid === cliPid); + + process.kill(cliPid!, "SIGTERM"); + await waitFor(() => existsSync(signalFile)); + await waitFor(() => readRun(env, "auth")?.nodes["worker-1"].status === "failed"); + assert.equal(processAlive(agentPid!), true); + assert.throws(() => acquireNodeLock(env, "auth", "worker-1"), /node is locked: worker-1/); + + process.kill(agentPid!, "SIGKILL"); + await waitFor(() => !processAlive(agentPid!)); + agentPid = undefined; + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + if (agentPid && processAlive(agentPid)) process.kill(agentPid, "SIGKILL"); + if (cliPid && processAlive(cliPid)) process.kill(cliPid, "SIGKILL"); + rmSync(directory, { force: true, recursive: true }); + rmSync(processDirectory, { force: true, recursive: true }); + } +}); + +test("async run message releases handed-off ownership after normal completion", async () => { + const agentSource = [ + `#!${process.execPath}`, + "if (process.env.HEADLESS_ASYNC_MESSAGE_OWNER_PID) process.exit(91);", + "if (process.env.HEADLESS_ASYNC_MESSAGE_WORKER) process.exit(92);", + "console.log(JSON.stringify({ type: 'agent_message', text: 'done' }));", + "", + ].join("\n"); + const { directory, env } = await createFixture(agentSource); + try { + await startAsyncMessage(env); + await waitFor(() => readRun(env, "auth")?.nodes["worker-1"].status !== "busy"); + const node = readRun(env, "auth")?.nodes["worker-1"]; + assert.equal(node?.status, "idle", readFileSync(node?.logs?.stderr ?? "", "utf8")); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + await waitFor(() => !existsSync(lockPath) && !existsSync(`${lockPath}.owner`)); + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("async run message handles a missing agent without leaking its lock", async () => { + const { directory, env } = await createFixture(); + try { + await startAsyncMessage(env); + await waitFor(() => readRun(env, "auth")?.nodes["worker-1"].status === "failed"); + const node = readRun(env, "auth")?.nodes["worker-1"]; + const stderr = readFileSync(node?.logs?.stderr ?? "", "utf8"); + assert.match(stderr, /spawn codex ENOENT/); + assert.doesNotMatch(stderr, /Unhandled 'error' event|Emitted 'error' event/); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + await waitFor(() => !existsSync(lockPath) && !existsSync(`${lockPath}.owner`)); + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("async run message rolls back when its worker cannot spawn the CLI", async () => { + const { directory, env } = await createFixture(); + env.HEADLESS_CLI_BIN = join(directory, "missing-headless"); + try { + assert.equal( + await runCli(["run", "message", "auth", "worker-1", "--prompt", "continue", "--async"], { + env, + stderr: () => undefined, + stdout: () => undefined, + }), + 2, + ); + const node = readRun(env, "auth")?.nodes["worker-1"]; + assert.equal(node?.status, "failed"); + assert.match(node?.lastMessage ?? "", /exited before agent startup/); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + await waitFor(() => !existsSync(lockPath) && !existsSync(`${lockPath}.owner`)); + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("async run message rolls back busy status when startup logging fails", async () => { + const { directory, env } = await createFixture(); + try { + const stdoutLog = readRun(env, "auth")?.nodes["worker-1"].logs?.stdout; + assert.ok(stdoutLog); + await mkdir(stdoutLog, { recursive: true }); + const stderr: string[] = []; + + assert.equal( + await runCli(["run", "message", "auth", "worker-1", "--prompt", "continue", "--async"], { + env, + stderr: (text) => stderr.push(text), + stdout: () => undefined, + }), + 2, + ); + const node = readRun(env, "auth")?.nodes["worker-1"]; + assert.equal(node?.status, "failed"); + assert.match(node?.lastMessage ?? "", /EISDIR/); + assert.match(stderr.join(""), /EISDIR/); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + await waitFor(() => !existsSync(lockPath) && !existsSync(`${lockPath}.owner`)); + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("async run message preserves failed status after a partial state write", async () => { + const { directory, env } = await createFixture(); + try { + const eventsPath = join(runDirectory(env, "auth"), "events.jsonl"); + rmSync(eventsPath, { force: true }); + await mkdir(eventsPath); + const stderr: string[] = []; + + assert.equal( + await runCli(["run", "message", "auth", "worker-1", "--prompt", "continue", "--async"], { + env, + stderr: (text) => stderr.push(text), + stdout: () => undefined, + }), + 2, + ); + const node = readRun(env, "auth")?.nodes["worker-1"]; + assert.equal(node?.status, "failed"); + assert.match(node?.lastMessage ?? "", /EISDIR/); + assert.match(stderr.join(""), /EISDIR/); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + await waitFor(() => !existsSync(lockPath) && !existsSync(`${lockPath}.owner`)); + acquireNodeLock(env, "auth", "worker-1")(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("prepared async worker repairs busy status when its parent disconnects", async () => { + const { directory, env } = await createFixture(); + const worker = fork(join(process.cwd(), "src", "async-run-message.ts"), [], { + env: env as NodeJS.ProcessEnv, + execArgv: ["--import", "tsx"], + stdio: ["ignore", "ignore", "ignore", "ipc"], + }); + try { + const task: AsyncRunMessageTask = { + command: { command: process.execPath, args: ["--eval", "process.exit(0)"] }, + runId: "auth", + nodeId: "worker-1", + }; + const ready = new Promise((resolve, reject) => { + worker.once("error", reject); + worker.on("message", (message: AsyncRunMessageResponse) => { + if (message.type === "error") reject(new Error(message.message)); + else if (message.type === "ready") resolve(); + }); + }); + worker.send({ type: "task", task }); + await ready; + updateNodeStatus(env, "auth", "worker-1", "busy"); + + worker.disconnect(); + await once(worker, "exit"); + + const node = readRun(env, "auth")?.nodes["worker-1"]; + assert.equal(node?.status, "failed"); + assert.match(node?.lastMessage ?? "", /disconnected before agent startup/); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + assert.equal(existsSync(lockPath), false); + assert.equal(existsSync(`${lockPath}.owner`), false); + } finally { + if (worker.connected) worker.disconnect(); + if (worker.exitCode === null && worker.signalCode === null) worker.kill("SIGKILL"); + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("prepared async worker leaves cancellation status repair to its parent", async () => { + const { directory, env } = await createFixture(); + const worker = fork(join(process.cwd(), "src", "async-run-message.ts"), [], { + env: env as NodeJS.ProcessEnv, + execArgv: ["--import", "tsx"], + stdio: ["ignore", "ignore", "ignore", "ipc"], + }); + try { + const task: AsyncRunMessageTask = { + command: { command: process.execPath, args: ["--eval", "process.exit(0)"] }, + runId: "auth", + nodeId: "worker-1", + }; + const ready = new Promise((resolve, reject) => { + worker.once("error", reject); + worker.on("message", (message: AsyncRunMessageResponse) => { + if (message.type === "error") reject(new Error(message.message)); + else if (message.type === "ready") resolve(); + }); + }); + worker.send({ type: "task", task }); + await ready; + updateNodeStatus(env, "auth", "worker-1", "busy"); + + worker.send({ type: "cancel" }); + await once(worker, "exit"); + + const run = readRun(env, "auth"); + assert.equal(run?.nodes["worker-1"].status, "busy"); + assert.equal(run?.events.filter((event) => event.type === "node_failed").length, 0); + const lockPath = nodeLockPath(env, "auth", "worker-1"); + assert.equal(existsSync(lockPath), false); + assert.equal(existsSync(`${lockPath}.owner`), false); + } finally { + if (worker.connected) worker.disconnect(); + if (worker.exitCode === null && worker.signalCode === null) worker.kill("SIGKILL"); + rmSync(directory, { force: true, recursive: true }); + } +}); diff --git a/tests/ci.test.ts b/tests/ci.test.ts new file mode 100644 index 0000000..9ce007d --- /dev/null +++ b/tests/ci.test.ts @@ -0,0 +1,14 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; + +const workflow = readFileSync(".github/workflows/ci.yml", "utf8"); + +test("Windows CI runs every run-storage process test", () => { + const windowsJob = workflow.match(/\n windows-run-storage:\n[\s\S]*?(?=\n [\w-]+:\n|$)/)?.[0]; + + assert.ok(windowsJob); + assert.match(windowsJob, /run: node --import tsx --test /); + assert.match(windowsJob, /tests\/run-storage\.test\.ts/); + assert.match(windowsJob, /tests\/run-storage-process\.test\.ts/); +}); diff --git a/tests/headless.test.ts b/tests/headless.test.ts index 8cbbdbe..5271399 100644 --- a/tests/headless.test.ts +++ b/tests/headless.test.ts @@ -1558,6 +1558,66 @@ test("CLI forwards parent signals to timeout-enabled agents", async () => { } }); +test("CLI forwards parent signals for async message workers without a timeout", async () => { + const dir = mkdtempSync(join(tmpdir(), "headless-test-")); + const agentPidFile = join(dir, "agent.pid"); + const signalFile = join(dir, "signal.txt"); + let agentPid: number | undefined; + try { + const binDir = join(dir, "bin"); + mkdirSync(binDir); + const binary = join(binDir, "codex"); + writeFileSync( + binary, + [ + "#!/usr/bin/env node", + "const { writeFileSync } = require('node:fs');", + "writeFileSync(process.env.HEADLESS_TEST_AGENT_PID, String(process.pid));", + "process.on('SIGTERM', () => {", + " writeFileSync(process.env.HEADLESS_TEST_SIGNAL_FILE, 'SIGTERM');", + " process.exit(143);", + "});", + "setInterval(() => {}, 1000);", + "", + ].join("\n"), + ); + chmodSync(binary, 0o755); + + const headless = spawn( + process.execPath, + ["--import", "tsx", join(repoRoot, "src", "cli.ts"), "codex", "--prompt", "hello", "--json"], + { + env: { + ...process.env, + HEADLESS_ASYNC_MESSAGE_WORKER: "1", + HEADLESS_TEST_AGENT_PID: agentPidFile, + HEADLESS_TEST_SIGNAL_FILE: signalFile, + PATH: `${binDir}:${process.env.PATH ?? ""}`, + }, + stdio: "ignore", + }, + ); + await waitFor(() => existsSync(agentPidFile)); + + headless.kill("SIGTERM"); + const exit = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => { + headless.once("exit", (code, signal) => resolve({ code, signal })); + }); + + assert.deepEqual(exit, { code: null, signal: "SIGTERM" }); + await waitFor(() => existsSync(signalFile)); + assert.equal(readFileSync(signalFile, "utf8"), "SIGTERM"); + } finally { + try { + agentPid = Number(readFileSync(agentPidFile, "utf8")); + if (Number.isInteger(agentPid) && agentPid > 0 && processIsAlive(agentPid)) process.kill(agentPid, "SIGKILL"); + } catch { + // The wrapper may fail before launching its agent. + } + rmSync(dir, { force: true, recursive: true }); + } +}); + test("CLI forwards parent signals while holding a durable Docker session lock", async () => { const dir = mkdtempSync(join(tmpdir(), "headless-test-")); const dockerPidFile = join(dir, "docker.pid"); diff --git a/tests/run-coordination.test.ts b/tests/run-coordination.test.ts index 46770a3..de4b8e9 100644 --- a/tests/run-coordination.test.ts +++ b/tests/run-coordination.test.ts @@ -1,12 +1,21 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; -import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import test from "node:test"; import { runCli } from "../src/cli.ts"; -import { acquireNodeLock, appendNodeLog, nodeLockPath, readRun, registerNode, updateNodeStatus } from "../src/runs.ts"; +import { + acquireNodeLock, + appendNodeLog, + nodeLockPath, + readRun, + registerNode, + runDirectory, + updateNodeStatus, + writeRun, +} from "../src/runs.ts"; import { expandTeamSpecs, parseTeamSpec } from "../src/teams.ts"; async function writeExecutable(path: string, source: string): Promise { @@ -185,7 +194,7 @@ test("run store writes private run files, logs, and locks", () => { const release = acquireNodeLock(env, "auth", "worker-1"); try { - assert.equal(modeOf(nodeLockPath(env, "auth", "worker-1")), 0o600); + assert.equal(modeOf(nodeLockPath(env, "auth", "worker-1")), 0o700); } finally { release(); } @@ -195,6 +204,32 @@ test("run store writes private run files, logs, and locks", () => { } }); +test("run store removes its temporary state file after replacement fails", () => { + const dir = mkdtempSync(join(tmpdir(), "headless-run-test-")); + try { + const env = { ...process.env, HOME: join(dir, "home") }; + registerNode(env, { + runId: "auth", + nodeId: "worker-1", + role: "worker", + agent: "codex", + coordination: "oneshot", + status: "idle", + planned: true, + }); + const run = readRun(env, "auth"); + assert.ok(run); + const statePath = join(runDirectory(env, "auth"), "run.json"); + rmSync(statePath); + mkdirSync(statePath); + + assert.throws(() => writeRun(env, run)); + assert.equal(readdirSync(runDirectory(env, "auth")).some((name) => name.startsWith("run.json.tmp-")), false); + } finally { + rmSync(dir, { force: true, recursive: true }); + } +}); + test("run store rejects dot segment run and node names", () => { const dir = mkdtempSync(join(tmpdir(), "headless-run-test-")); try { @@ -1321,6 +1356,7 @@ test("run message --async records busy status, logs output, and marks completion const binDir = join(dir, "bin"); const fakeHeadless = join(binDir, "headless"); const captureFile = join(dir, "headless-args.jsonl"); + const childStartedFile = join(dir, "child-started"); mkdirSync(home); await writeExecutable( fakeHeadless, @@ -1345,12 +1381,20 @@ test("run message --async records busy status, logs output, and marks completion " const dir = path.join(process.env.HOME, '.headless', 'runs', runId, 'nodes', nodeId);", " fs.mkdirSync(dir, { recursive: true });", " fs.appendFileSync(path.join(dir, 'latest.stdout.log'), 'async child output\\n');", + " fs.writeFileSync(process.env.HEADLESS_CHILD_STARTED, 'started\\n');", + " Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 1000);", "}", "console.log('async child output should not be wrapper-redirected');", "", ].join("\n"), ); - const env = { ...process.env, HEADLESS_CAPTURE: captureFile, HEADLESS_CLI_BIN: fakeHeadless, HOME: home }; + const env = { + ...process.env, + HEADLESS_CAPTURE: captureFile, + HEADLESS_CHILD_STARTED: childStartedFile, + HEADLESS_CLI_BIN: fakeHeadless, + HOME: home, + }; registerNode(env, { runId: "auth", nodeId: "worker-1", @@ -1371,6 +1415,17 @@ test("run message --async records busy status, logs output, and marks completion 0, ); assert.equal(readRun(env, "auth")?.nodes["worker-1"].status, "busy"); + await waitFor(() => existsSync(childStartedFile)); + const lockedStderr: string[] = []; + assert.equal( + await runCli(["run", "message", "auth", "worker-1", "--prompt", "overlap", "--async"], { + env, + stdout: () => undefined, + stderr: (text) => lockedStderr.push(text), + }), + 2, + ); + assert.match(lockedStderr.join(""), /node is locked: worker-1/); await waitFor(() => readRun(env, "auth")?.nodes["worker-1"].status === "idle"); const stdoutLog = readRun(env, "auth")?.nodes["worker-1"].logs?.stdout ?? ""; await waitFor(() => existsSync(stdoutLog) && readFileSync(stdoutLog, "utf8").includes("async child output")); @@ -1378,6 +1433,11 @@ test("run message --async records busy status, logs output, and marks completion assert.match(stdoutText, /previous output/); assert.doesNotMatch(stdoutText, /marked fake/); assert.doesNotMatch(stdoutText, /wrapper-redirected/); + await waitFor( + () => !existsSync(nodeLockPath(env, "auth", "worker-1")) + && !existsSync(`${nodeLockPath(env, "auth", "worker-1")}.owner`), + ); + acquireNodeLock(env, "auth", "worker-1")(); } finally { rmSync(dir, { force: true, recursive: true }); } @@ -1494,13 +1554,13 @@ test("run message --async uses HEADLESS_BIN for detached child invocations", asy const calls = readFileSync(captureFile, "utf8").trim().split("\n").map((line) => JSON.parse(line)); assert.deepEqual(calls[0].slice(0, 7), ["codex", "--role", "worker", "--coordination", "oneshot", "--run", "auth"]); assert.equal(calls[0].includes("--fast"), true); - assert.deepEqual(calls.at(-1), ["run", "mark", "auth", "worker-1", "--status", "idle"]); + assert.equal(calls.length, 1); } finally { rmSync(dir, { force: true, recursive: true }); } }); -test("run message --async print-command uses a non-login shell to preserve PATH", async () => { +test("run message --async print-command renders an executable async invocation without mutating state", async () => { const dir = mkdtempSync(join(tmpdir(), "headless-run-test-")); try { const env = { ...process.env, HOME: join(dir, "home") }; @@ -1524,9 +1584,8 @@ test("run message --async print-command uses a non-login shell to preserve PATH" 0, ); const output = stdout.join(""); - assert.match(output, /^sh -c /); - assert.match(output, /trap /); - assert.doesNotMatch(output, /^sh -lc /); + assert.match(output, /^headless run message auth explorer-1 --prompt continue --async/); + assert.doesNotMatch(output, /^sh /); assert.equal(readRun(env, "auth")?.nodes["explorer-1"].status, "idle"); assert.equal(readRun(env, "auth")?.nodes["explorer-1"].lastMessage, undefined); } finally { diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts new file mode 100644 index 0000000..91385ce --- /dev/null +++ b/tests/run-storage-process.test.ts @@ -0,0 +1,266 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import { + acquireNodeStoreLock, + macosProcessStartIdentity, + processTreeAliveFromProbes, + windowsProcessStartIdentity, + windowsProcessTreeAlive, +} from "../src/run-storage.ts"; + +function withTemporaryDirectory(callback: (directory: string) => void): void { + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-process-")); + try { + callback(directory); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +} + +function agePath(path: string): void { + const stale = new Date(Date.now() - 30_000); + utimesSync(path, stale, stale); +} + +async function captureLockOwnerWithIdentity( + lockPath: string, + processTreeRootPid: number, +): Promise & { processStartIdentity: string }> { + for (let attempt = 0; attempt < 10; attempt += 1) { + const release = acquireNodeStoreLock(lockPath, "worker-1", { processTreeRootPid }); + let owner: Record; + try { + owner = JSON.parse(readFileSync(`${lockPath}.owner`, "utf8")) as Record; + } finally { + release(); + } + if (typeof owner.processStartIdentity === "string") { + return { ...owner, processStartIdentity: owner.processStartIdentity }; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error("process start identity was not observable"); +} + +async function stopChildProcess(child: ReturnType): Promise { + const exited = once(child, "exit"); + if (child.exitCode !== null || child.signalCode !== null) return; + child.kill(); + await exited; +} + +async function waitForChildReady(child: ReturnType): Promise { + let timeout: NodeJS.Timeout | undefined; + try { + await Promise.race([ + once(child, "message").then(([message]) => assert.equal(message, "ready")), + once(child, "exit").then(([code, signal]) => { + throw new Error(`child exited before ready: code=${code}, signal=${signal}`); + }), + new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error("child readiness timed out")), 5_000); + timeout.unref(); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +} + +test("node store lock preserves a live owner beyond the identity lifetime", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + const ownerPath = `${lockPath}.owner`; + const releaseOwner = acquireNodeStoreLock(lockPath, "worker-1", { processTreeRootPid: process.pid }); + const owner = JSON.parse(readFileSync(ownerPath, "utf8")); + owner.createdAtMs = 0; + writeFileSync(ownerPath, `${JSON.stringify(owner)}\n`); + rmSync(lockPath, { recursive: true }); + + try { + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + assert.equal(existsSync(ownerPath), true); + } finally { + releaseOwner(); + } + }); +}); + +test("node store lock recovers when a live PID has a different start identity", async () => { + const reusedProcess = spawn( + process.execPath, + ["-e", "process.send?.('ready'); setInterval(() => {}, 60_000)"], + { stdio: ["ignore", "ignore", "ignore", "ipc"] }, + ); + const reusedPid = reusedProcess.pid; + assert.ok(reusedPid); + try { + await waitForChildReady(reusedProcess); + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-process-")); + try { + const lockPath = join(directory, "session.lock"); + const ownerPath = `${lockPath}.owner`; + const owner = await captureLockOwnerWithIdentity(lockPath, reusedPid); + owner.processStartIdentity = `${owner.processStartIdentity}-reused`; + writeFileSync(ownerPath, `${JSON.stringify(owner)}\n`); + let releaseReplacement: (() => void) | undefined; + + try { + releaseReplacement = acquireNodeStoreLock(lockPath, "worker-1"); + releaseReplacement(); + releaseReplacement = undefined; + assert.equal(existsSync(ownerPath), false); + } finally { + if (releaseReplacement) releaseReplacement(); + } + } finally { + rmSync(directory, { force: true, recursive: true }); + } + } finally { + await stopChildProcess(reusedProcess); + } +}); + +test("node store lock preserves an old identity-less live owner", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ + createdAtMs: 0, + processTreeRootPid: process.pid, + })}\n`); + agePath(lockPath); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + }); +}); + +test("node store lock recovers an old owner after its process tree exits", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ + createdAtMs: 0, + processTreeRootPid: 99_999_999, + })}\n`); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + release(); + assert.equal(existsSync(`${lockPath}.owner`), false); + }); +}); + +test("node store lock rejects an out-of-range owner PID", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ + createdAtMs: Date.now(), + processTreeRootPid: Number.MAX_SAFE_INTEGER, + })}\n`); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + release(); + assert.equal(existsSync(`${lockPath}.owner`), false); + }); +}); + +test("Windows process-tree probe releases only an explicit dead result", () => { + const probe = (output: string) => windowsProcessTreeAlive(123, { + execute: () => output, + systemRoot: "C:\\Windows", + }); + + assert.equal(probe("HEADLESS_PROCESS_TREE_DEAD\r\n"), false); + assert.equal(probe("HEADLESS_PROCESS_TREE_ALIVE\r\n"), true); + assert.equal(probe(""), true); + assert.equal(probe("0\r\n"), true); + assert.equal(probe("warning\r\nHEADLESS_PROCESS_TREE_DEAD\r\n"), true); +}); + +test("Windows process-tree probe fails closed on execution errors", () => { + assert.equal(windowsProcessTreeAlive(123, { + execute: () => { + throw new Error("CIM unavailable"); + }, + }), true); +}); + +test("Windows process-tree probe makes CIM failures terminating", () => { + let command = ""; + windowsProcessTreeAlive(123, { + execute: (_path, args) => { + command = args.join(" "); + return "HEADLESS_PROCESS_TREE_DEAD\n"; + }, + }); + + assert.match(command, /ErrorActionPreference.*Stop/); + assert.match(command, /rootPid = \[uint32\]123/); + assert.match(command, /ProcessId -ne \$PID/); + assert.doesNotMatch(command, /\$args/); +}); + +test("Windows process-start probe accepts only an explicit identity", () => { + let command = ""; + const probe = (output: string) => windowsProcessStartIdentity(123, { + execute: (_path, args) => { + command = args.join(" "); + return output; + }, + }); + + assert.equal(probe("HEADLESS_PROCESS_START:638920627920000000\r\n"), "win32:638920627920000000"); + assert.match(command, /rootPid = \[uint32\]123/); + assert.doesNotMatch(command, /\$args/); + assert.equal(probe(""), undefined); + assert.equal(probe("638920627920000000\r\n"), undefined); + assert.equal(probe("warning\r\nHEADLESS_PROCESS_START:638920627920000000\r\n"), undefined); +}); + +test("Windows process-start probe ignores execution errors", () => { + assert.equal(windowsProcessStartIdentity(123, { + execute: () => { + throw new Error("CIM unavailable"); + }, + }), undefined); +}); + +test("macOS process-start probe canonicalizes timezone and locale", () => { + let probeEnv: NodeJS.ProcessEnv | undefined; + const identity = macosProcessStartIdentity(123, { + env: { LC_ALL: "de_DE.UTF-8", TZ: "Pacific/Honolulu" }, + execute: (_command, _args, options) => { + probeEnv = options.env; + return "Sat Aug 29 17:00:00 2026\n"; + }, + }); + + assert.equal(identity, "darwin:Sat Aug 29 17:00:00 2026"); + assert.equal(probeEnv?.LC_ALL, "C"); + assert.equal(probeEnv?.TZ, "UTC"); +}); + +test("Windows probes descendants when a live root PID has been reused", () => { + let probes = 0; + const descendantsAlive = () => { + probes += 1; + return true; + }; + + assert.equal(processTreeAliveFromProbes("win32", true, true, descendantsAlive), true); + assert.equal(probes, 1); + assert.equal(processTreeAliveFromProbes("win32", true, true, () => false), false); + assert.equal(processTreeAliveFromProbes("linux", true, true, descendantsAlive), false); + assert.equal(probes, 1); +}); diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts new file mode 100644 index 0000000..480141e --- /dev/null +++ b/tests/run-storage.test.ts @@ -0,0 +1,549 @@ +import assert from "node:assert/strict"; +import { spawn, spawnSync } from "node:child_process"; +import { once } from "node:events"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, win32 } from "node:path"; +import { pathToFileURL } from "node:url"; +import test from "node:test"; + +import { + acquireNodeStoreLock, + acquireRunStoreLock, + isRunStoreLockSignalListener, + removeRunStoreLockSignalListeners, + replaceRunStateFile, +} from "../src/run-storage.ts"; + +function withTemporaryDirectory(callback: (directory: string) => void): void { + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-")); + try { + callback(directory); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +} + +function agePath(path: string, milliseconds = 30_000): void { + const stale = new Date(Date.now() - milliseconds); + utimesSync(path, stale, stale); +} + +test("run store lock recovers a legacy lock owned by a dead process", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "run.lock"); + writeFileSync(lockPath, "99999999\n"); + + const release = acquireRunStoreLock(lockPath, "auth"); + + assert.equal(statSync(lockPath).isDirectory(), true); + release(); + assert.equal(existsSync(lockPath), false); + }); +}); + +test("node store lock preserves a fresh malformed legacy lock", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + writeFileSync(lockPath, ""); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + assert.equal(readFileSync(lockPath, "utf8"), ""); + }); +}); + +test("node store lock recovers an aged malformed legacy lock", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + writeFileSync(lockPath, "not-a-pid\n"); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + assert.equal(statSync(lockPath).isDirectory(), true); + release(); + }); +}); + +test("node store lock recovers an aged legacy lock with an unsafe PID", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + writeFileSync(lockPath, "999999999999999999999999\n"); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + assert.equal(statSync(lockPath).isDirectory(), true); + release(); + }); +}); + +test("node store lock preserves a legacy lock owned by a live process", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + writeFileSync(lockPath, `${process.pid}\n`); + agePath(lockPath); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + }); +}); + +test("node store lock excludes a live holder and supports reacquisition", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + const releaseFirst = acquireNodeStoreLock(lockPath, "worker-1"); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + releaseFirst(); + + const releaseSecond = acquireNodeStoreLock(lockPath, "worker-1"); + releaseSecond(); + assert.equal(existsSync(lockPath), false); + }); +}); + +test("node store lock persists private owner identity while its process is alive", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + const ownerPath = `${lockPath}.owner`; + const release = acquireNodeStoreLock(lockPath, "worker-1", { processTreeRootPid: process.pid }); + const owner = JSON.parse(readFileSync(ownerPath, "utf8")); + + if (process.platform !== "win32") assert.equal(statSync(ownerPath).mode & 0o777, 0o600); + assert.equal(typeof owner.processStartIdentity, "string"); + assert.ok(owner.processStartIdentity.length > 0); + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + release(); + assert.equal(existsSync(ownerPath), false); + }); +}); + +test("node store lock preserves a live owner after lease cleanup", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + const releaseOwner = acquireNodeStoreLock(lockPath, "worker-1", { + processTreeRootPid: process.pid, + }); + let releaseReplacement: (() => void) | undefined; + rmSync(lockPath, { recursive: true }); + + try { + assert.throws(() => { + releaseReplacement = acquireNodeStoreLock(lockPath, "worker-1"); + }, /node is locked: worker-1/); + } finally { + if (releaseReplacement) releaseReplacement(); + else releaseOwner(); + } + }); +}); + +test("node store lock rechecks a live owner after lease acquisition", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + const ownerPath = `${lockPath}.owner`; + const moduleUrl = pathToFileURL(join(process.cwd(), "src", "run-storage.ts")).href; + const requirePath = join(process.cwd(), "package.json"); + const script = [ + `import { writeFileSync } from "node:fs";`, + `import { createRequire } from "node:module";`, + `const require = createRequire(${JSON.stringify(requirePath)});`, + `const properLockfile = require("proper-lockfile");`, + `const nativeLockSync = properLockfile.lockSync;`, + `properLockfile.lockSync = (...args) => {`, + ` const release = nativeLockSync(...args);`, + ` writeFileSync(${JSON.stringify(ownerPath)}, JSON.stringify({`, + ` createdAtMs: Date.now(),`, + ` processTreeRootPid: process.pid,`, + ` }) + "\\n");`, + ` return release;`, + `};`, + `const { acquireNodeStoreLock } = await import(${JSON.stringify(moduleUrl)});`, + `try {`, + ` acquireNodeStoreLock(${JSON.stringify(lockPath)}, "worker-1")();`, + ` process.stdout.write("acquired\\n");`, + `} catch (error) {`, + ` process.stdout.write(String(error.message) + "\\n");`, + `}`, + ].join("\n"); + + const result = spawnSync( + process.execPath, + ["--import", "tsx", "--input-type=module", "--eval", script], + { encoding: "utf8", timeout: 5_000 }, + ); + + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout.trim(), "node is locked: worker-1"); + }); +}); + +test("node store lock does not remove an unexpected lock directory", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + assert.equal(statSync(lockPath).isDirectory(), true); + }); +}); + +test("node store lock preserves a lease inside the stale window", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + agePath(lockPath, 15_000); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + assert.equal(statSync(lockPath).isDirectory(), true); + }); +}); + +test("node store lock recovers a stale lease directory", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + assert.equal(statSync(lockPath).isDirectory(), true); + release(); + assert.equal(existsSync(lockPath), false); + }); +}); + +test("node store lock does not trust an invalid owner PID", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ + createdAtMs: Date.now(), + processTreeRootPid: 0, + })}\n`); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + release(); + }); +}); + +test("node store lock recovers a stale lease with malformed owner identity", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, "not-json\n"); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + release(); + }); +}); + +test("node store lock recovers an aged lock owned by a dead process", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + writeFileSync(lockPath, "99999999\n"); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + assert.equal(statSync(lockPath).isDirectory(), true); + release(); + }); +}); + +test("node store lock blocks stale takeover while the owner's process group is alive", { skip: process.platform === "win32" }, async () => { + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-")); + let processGroupId: number | undefined; + try { + const lockPath = join(directory, "session.lock"); + const moduleUrl = pathToFileURL(join(process.cwd(), "src", "run-storage.ts")).href; + const script = [ + `import { spawn } from "node:child_process";`, + `import { acquireNodeStoreLock } from ${JSON.stringify(moduleUrl)};`, + `acquireNodeStoreLock(${JSON.stringify(lockPath)}, "worker-1", { processTreeRootPid: process.pid });`, + `spawn(process.execPath, ["--eval", "setInterval(() => undefined, 1000)"], { stdio: "ignore" });`, + `process.stdout.write("ready\\n");`, + `setInterval(() => undefined, 1000);`, + ].join("\n"); + const owner = spawn(process.execPath, ["--import", "tsx", "--input-type=module", "--eval", script], { + detached: true, + stdio: ["ignore", "pipe", "inherit"], + }); + processGroupId = owner.pid; + await once(owner.stdout, "data"); + owner.kill("SIGKILL"); + await once(owner, "exit"); + rmSync(lockPath, { recursive: true }); + + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + + process.kill(-processGroupId!, "SIGKILL"); + await waitForProcessGroupExit(processGroupId!); + processGroupId = undefined; + const release = acquireNodeStoreLock(lockPath, "worker-1"); + release(); + } finally { + if (processGroupId) { + try { + process.kill(-processGroupId, "SIGKILL"); + } catch { + // Process group already exited. + } + } + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("run store lock excludes a live holder and supports reacquisition", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "run.lock"); + const releaseFirst = acquireRunStoreLock(lockPath, "auth"); + + assert.equal(statSync(lockPath).isDirectory(), true); + releaseFirst(); + + const releaseSecond = acquireRunStoreLock(lockPath, "auth"); + releaseSecond(); + assert.equal(existsSync(lockPath), false); + }); +}); + +test("run store lock recovers a stale lease directory", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "run.lock"); + mkdirSync(lockPath); + agePath(lockPath, 120_000); + + const release = acquireRunStoreLock(lockPath, "auth"); + + release(); + assert.equal(existsSync(lockPath), false); + }); +}); + +test("run store lock recovers after its owner crashes", { skip: process.platform === "win32" }, async () => { + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-")); + try { + const lockPath = join(directory, "run.lock"); + const moduleUrl = pathToFileURL(join(process.cwd(), "src", "run-storage.ts")).href; + const script = [ + `import { acquireRunStoreLock } from ${JSON.stringify(moduleUrl)};`, + `acquireRunStoreLock(${JSON.stringify(lockPath)}, "auth");`, + `process.stdout.write("ready\\n");`, + `setInterval(() => undefined, 1000);`, + ].join("\n"); + const owner = spawn(process.execPath, ["--import", "tsx", "--input-type=module", "--eval", script], { + stdio: ["ignore", "pipe", "inherit"], + }); + await once(owner.stdout, "data"); + owner.kill("SIGKILL"); + await once(owner, "exit"); + + const startedAt = Date.now(); + const release = acquireRunStoreLock(lockPath, "auth"); + + assert.ok(Date.now() - startedAt < 10_000); + release(); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +test("run store identifies signal listeners installed by its lock dependency", () => { + withTemporaryDirectory((directory) => { + const release = acquireNodeStoreLock(join(directory, "session.lock"), "worker-1"); + release(); + + assert.equal( + managedTestSignals().some((signal) => process.listeners(signal).some(isRunStoreLockSignalListener)), + true, + ); + removeRunStoreLockSignalListeners(); + assert.equal( + managedTestSignals().some((signal) => process.listeners(signal).some(isRunStoreLockSignalListener)), + false, + ); + }); +}); + +function managedTestSignals(): NodeJS.Signals[] { + return process.platform === "win32" + ? ["SIGINT", "SIGTERM", "SIGBREAK"] + : ["SIGHUP", "SIGINT", "SIGTERM", "SIGQUIT"]; +} + +test("run state replacement retries transient Windows rename failures", () => { + const attempts: number[] = []; + const delays: number[] = []; + + replaceRunStateFile("run.tmp", "run.json", { + platform: "win32", + rename: () => { + attempts.push(attempts.length + 1); + if (attempts.length < 3) throw Object.assign(new Error("busy"), { code: "EPERM" }); + }, + sleep: (milliseconds) => delays.push(milliseconds), + }); + + assert.equal(attempts.length, 3); + assert.deepEqual(delays, [25, 25]); +}); + +test("run state replacement uses the native rename defaults", () => { + withTemporaryDirectory((directory) => { + const source = join(directory, "run.tmp"); + const destination = join(directory, "run.json"); + writeFileSync(source, "state\n"); + + replaceRunStateFile(source, destination); + + assert.equal(readFileSync(destination, "utf8"), "state\n"); + assert.equal(existsSync(source), false); + }); +}); + +test("run state replacement does not retry non-Windows rename failures", () => { + let attempts = 0; + const failure = Object.assign(new Error("busy"), { code: "EPERM" }); + + assert.throws( + () => replaceRunStateFile("run.tmp", "run.json", { + platform: "linux", + rename: () => { + attempts += 1; + throw failure; + }, + sleep: () => assert.fail("unexpected retry"), + }), + failure, + ); + assert.equal(attempts, 1); +}); + +test("run state replacement does not retry permanent Windows rename failures", () => { + let attempts = 0; + const failure = Object.assign(new Error("missing"), { code: "ENOENT" }); + + assert.throws( + () => replaceRunStateFile("run.tmp", "run.json", { + platform: "win32", + rename: () => { + attempts += 1; + throw failure; + }, + sleep: () => assert.fail("unexpected retry"), + }), + failure, + ); + assert.equal(attempts, 1); +}); + +test("run state replacement does not retry Windows errors without a code", () => { + let attempts = 0; + + assert.throws(() => replaceRunStateFile("run.tmp", "run.json", { + platform: "win32", + rename: () => { + attempts += 1; + throw new Error("unknown"); + }, + sleep: () => assert.fail("unexpected retry"), + }), /unknown/); + assert.equal(attempts, 1); +}); + +test("run state replacement bounds Windows retries", () => { + let attempts = 0; + const failure = Object.assign(new Error("busy"), { code: "EBUSY" }); + + assert.throws( + () => replaceRunStateFile("run.tmp", "run.json", { + platform: "win32", + rename: () => { + attempts += 1; + throw failure; + }, + sleep: () => undefined, + }), + failure, + ); + assert.equal(attempts, 40); +}); + +test("run state replacement waits for a native Windows sharing lock", { skip: process.platform !== "win32" }, async () => { + const directory = mkdtempSync(join(tmpdir(), "headless-run-storage-")); + try { + const source = join(directory, "run.tmp"); + const destination = join(directory, "run.json"); + writeFileSync(source, "new\n"); + writeFileSync(destination, "old\n"); + const systemRoot = process.env.SystemRoot ?? "C:\\Windows"; + const powershell = win32.join(systemRoot, "System32", "WindowsPowerShell", "v1.0", "powershell.exe"); + const script = [ + "$path = $env:HEADLESS_TEST_LOCK_PATH", + "$stream = [System.IO.File]::Open($path, [System.IO.FileMode]::Open, [System.IO.FileAccess]::Read, [System.IO.FileShare]::None)", + "[Console]::Out.WriteLine('ready')", + "Start-Sleep -Milliseconds 200", + "$stream.Dispose()", + ].join("; "); + const locker = spawn(powershell, ["-NoProfile", "-NonInteractive", "-Command", script], { + env: { ...process.env, HEADLESS_TEST_LOCK_PATH: destination }, + stdio: ["ignore", "pipe", "inherit"], + windowsHide: true, + }); + const lockerExit = once(locker, "exit"); + await new Promise((resolve, reject) => { + const cleanup = () => { + locker.stdout.off("data", onReady); + locker.off("error", onError); + locker.off("exit", onEarlyExit); + }; + const onReady = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + const onEarlyExit = (code: number | null) => { + cleanup(); + reject(new Error(`PowerShell lock holder exited before ready: ${code ?? "unknown"}`)); + }; + locker.stdout.once("data", onReady); + locker.once("error", onError); + locker.once("exit", onEarlyExit); + }); + + let replacementError: unknown; + try { + replaceRunStateFile(source, destination); + } catch (error) { + replacementError = error; + } + await lockerExit; + if (replacementError) throw replacementError; + + assert.equal(readFileSync(destination, "utf8"), "new\n"); + assert.equal(existsSync(source), false); + } finally { + rmSync(directory, { force: true, recursive: true }); + } +}); + +async function waitForProcessGroupExit(processGroupId: number): Promise { + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + try { + process.kill(-processGroupId, 0); + } catch { + return; + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`process group did not exit: ${processGroupId}`); +}