From 9f35455931389ce8e1410b103b6a4907b27c8461 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:19:06 +0200 Subject: [PATCH 01/20] fix: recover stale run coordination locks --- src/async-run-message.ts | 161 ++++++++++++++++ src/cli.ts | 20 +- src/run-commands.ts | 133 ++++++++----- src/run-storage.ts | 305 ++++++++++++++++++++++++++++++ src/runs.ts | 42 +---- tests/headless.test.ts | 60 ++++++ tests/run-coordination.test.ts | 38 +++- tests/run-storage.test.ts | 332 +++++++++++++++++++++++++++++++++ 8 files changed, 998 insertions(+), 93 deletions(-) create mode 100644 src/async-run-message.ts create mode 100644 src/run-storage.ts create mode 100644 tests/run-storage.test.ts diff --git a/src/async-run-message.ts b/src/async-run-message.ts new file mode 100644 index 0000000..f8d7752 --- /dev/null +++ b/src/async-run-message.ts @@ -0,0 +1,161 @@ +import { execFileSync, spawn, type ChildProcess } from "node:child_process"; + +import { windowsTaskkillPath } from "./process-tree.js"; +import { removeRunStoreLockSignalListeners } from "./run-storage.js"; +import { acquireNodeLock, appendNodeLog, 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: "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) finish(); + }); + 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(); + } + 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); +} + +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, + env: { ...process.env, HEADLESS_ASYNC_MESSAGE_WORKER: "1" }, + stdio: ["ignore", "ignore", "inherit"], + }); + activeChild = child; + child.once("error", reject); + child.once("close", (code, signal) => { + activeChild = undefined; + resolve(signal ? 1 : (code ?? 1)); + }); + }); +} + +function finish(code = 0): void { + if (finished) return; + finished = true; + try { + 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..59d977f 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -102,6 +102,7 @@ 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 } from "./run-storage.js"; import { appendNodeLog, completeIdleRunNodes, @@ -1887,10 +1888,15 @@ 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 = process.platform !== "win32" && ( + asyncMessageWorker + || options.timeoutSeconds !== undefined + || options.cleanupBeforeParentSignalExit !== undefined + ); + const handlesParentSignals = asyncMessageWorker + || ownsChildProcessGroup + || options.cleanupBeforeParentSignalExit !== undefined; const child = spawn(command.command, command.args, { cwd, detached: ownsChildProcessGroup, @@ -3370,7 +3376,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..d9db2c7 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,104 @@ 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 sendWorkerRequest(worker, { type: "start" }); + } catch (error) { + 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 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 +430,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 +452,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..76ca6c6 --- /dev/null +++ b/src/run-storage.ts @@ -0,0 +1,305 @@ +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 = 10_000; +const nodeLockUpdateMs = 5_000; +const runLockTimeoutMs = 30_000; +const runLockRetryMs = 10; +const ownerIdentityLifetimeMs = 24 * 60 * 60 * 1_000; +const windowsProbeFailureGraceMs = 30_000; +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; +} + +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 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); + } +} + +function acquireStoreLock( + lockPath: string, + stale: number, + update: number, + owner?: NodeStoreLockOwner, +): () => void { + if (leaseOwnerBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); + if (legacyLockBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); + + let compromisedError: Error | undefined; + const release = loadLockSync()(lockPath, { + lockfilePath: lockPath, + onCompromised: (error) => { + compromisedError = error; + }, + realpath: false, + retries: 0, + stale, + update, + }); + try { + chmodSync(lockPath, privateDirMode); + rmSync(lockOwnerPath(lockPath), { force: true }); + if (owner) writeLockOwner(lockPath, owner); + } catch (error) { + release(); + throw error; + } + + return () => { + if (owner) rmSync(lockOwnerPath(lockPath), { force: true }); + if (compromisedError) throw compromisedError; + release(); + }; +} + +function leaseOwnerBlocksAcquisition(lockPath: string): boolean { + let lockMtimeMs: number; + try { + const status = lstatSync(lockPath); + if (!status.isDirectory()) return false; + lockMtimeMs = status.mtimeMs; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } + + let owner: StoredNodeStoreLockOwner; + try { + owner = JSON.parse(readFileSync(lockOwnerPath(lockPath), "utf8")) as StoredNodeStoreLockOwner; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT" || error instanceof SyntaxError) return false; + return true; + } + if ( + !Number.isSafeInteger(owner.processTreeRootPid) + || owner.processTreeRootPid <= 0 + || !Number.isFinite(owner.createdAtMs) + || Date.now() - owner.createdAtMs > ownerIdentityLifetimeMs + ) { + return false; + } + return processTreeAlive(owner, lockMtimeMs); +} + +function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { + const path = lockOwnerPath(lockPath); + const temporaryPath = `${path}.tmp-${randomUUID()}`; + const storedOwner: StoredNodeStoreLockOwner = { + ...owner, + createdAtMs: Date.now(), + processStartIdentity: processStartIdentity(owner.processTreeRootPid), + }; + let descriptor: number | undefined; + try { + descriptor = openSync(temporaryPath, "wx", 0o600); + writeFileSync(descriptor, `${JSON.stringify(storedOwner)}\n`); + fchmodSync(descriptor, 0o600); + closeSync(descriptor); + descriptor = undefined; + renameSync(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, lastHeartbeatAtMs: number): boolean { + const rootPid = owner.processTreeRootPid; + if (processAlive(rootPid)) { + const currentIdentity = processStartIdentity(rootPid); + if (owner.processStartIdentity && currentIdentity && owner.processStartIdentity !== currentIdentity) return false; + return true; + } + if (process.platform !== "win32") { + try { + process.kill(-rootPid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } + } + try { + return execFileSync(windowsPowerShellPath(), [ + "-NoProfile", + "-NonInteractive", + "-Command", + windowsDescendantProbe, + String(rootPid), + ], { + encoding: "utf8", + timeout: 3_000, + windowsHide: true, + }).trim() === "1"; + } catch { + return Date.now() - lastHeartbeatAtMs < windowsProbeFailureGraceMs; + } +} + +function processStartIdentity(pid: number): 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; + } + } + return undefined; +} + +function windowsPowerShellPath(): string { + const systemRoot = process.env.SystemRoot; + const root = systemRoot && win32.isAbsolute(systemRoot) ? systemRoot : "C:\\Windows"; + return win32.join(root, "System32", "WindowsPowerShell", "v1.0", "powershell.exe"); +} + +const windowsDescendantProbe = [ + "$rootPid = [uint32]$args[0]", + "$processes = @(Get-CimInstance Win32_Process)", + "$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) { '1' } else { '0' }", +].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..82f477d 100644 --- a/src/runs.ts +++ b/src/runs.ts @@ -1,10 +1,8 @@ import { appendFileSync, chmodSync, - closeSync, existsSync, mkdirSync, - openSync, readFileSync, readdirSync, renameSync, @@ -14,6 +12,7 @@ import { import { basename, dirname, join } from "node:path"; import type { CoordinationMode, Role, RunStatus } from "./roles.js"; +import { acquireNodeStoreLock, acquireRunStoreLock, type NodeStoreLockOwner } from "./run-storage.js"; import type { AgentName, AllowMode, Env, ReasoningEffort } from "./types.js"; const privateDirMode = 0o700; @@ -361,21 +360,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 { @@ -411,29 +399,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/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..f139fd9 100644 --- a/tests/run-coordination.test.ts +++ b/tests/run-coordination.test.ts @@ -185,7 +185,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(); } @@ -1321,6 +1321,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 +1346,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 +1380,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 +1398,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 +1519,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 +1549,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.test.ts b/tests/run-storage.test.ts new file mode 100644 index 0000000..a8f8186 --- /dev/null +++ b/tests/run-storage.test.ts @@ -0,0 +1,332 @@ +import assert from "node:assert/strict"; +import { spawn } 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 } from "node:path"; +import { pathToFileURL } from "node:url"; +import test from "node:test"; + +import { + acquireNodeStoreLock, + acquireRunStoreLock, + isRunStoreLockSignalListener, + removeRunStoreLockSignalListeners, +} 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 = 20_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 }); + + assert.equal(statSync(ownerPath).mode & 0o777, 0o600); + assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); + release(); + assert.equal(existsSync(ownerPath), false); + }); +}); + +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 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 expired owner identity", () => { + withTemporaryDirectory((directory) => { + const lockPath = join(directory, "session.lock"); + mkdirSync(lockPath); + writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ + createdAtMs: 0, + processTreeRootPid: process.pid, + })}\n`); + agePath(lockPath); + + const release = acquireNodeStoreLock(lockPath, "worker-1"); + + release(); + assert.equal(existsSync(`${lockPath}.owner`), 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"); + agePath(lockPath); + + 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"]; +} + +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}`); +} From 8132ddc9fbdb0215b6745c07a34deacd0d543faf Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:23:34 +0200 Subject: [PATCH 02/20] fix: retry Windows run state replacement --- src/run-storage.ts | 30 ++++++++++++++ src/runs.ts | 29 +++++++++++--- tests/run-coordination.test.ts | 39 +++++++++++++++++- tests/run-storage.test.ts | 72 ++++++++++++++++++++++++++++++++++ 4 files changed, 162 insertions(+), 8 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index 76ca6c6..4dfec5a 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -26,6 +26,9 @@ const runLockTimeoutMs = 30_000; const runLockRetryMs = 10; const ownerIdentityLifetimeMs = 24 * 60 * 60 * 1_000; const windowsProbeFailureGraceMs = 30_000; +const windowsRenameAttempts = 20; +const windowsRenameRetryMs = 25; +const windowsRenameRetryCodes = new Set(["EACCES", "EBUSY", "EPERM"]); const managedSignals: NodeJS.Signals[] = process.platform === "win32" ? ["SIGINT", "SIGTERM", "SIGBREAK"] : ["SIGHUP", "SIGINT", "SIGTERM", "SIGQUIT"]; @@ -41,6 +44,12 @@ interface StoredNodeStoreLockOwner extends NodeStoreLockOwner { processStartIdentity?: string; } +export interface RunStateReplacementOptions { + platform?: NodeJS.Platform; + rename?: (source: string, destination: string) => void; + sleep?: (milliseconds: number) => void; +} + export function acquireRunStoreLock(lockPath: string, runId: string): () => void { const deadline = Date.now() + runLockTimeoutMs; while (true) { @@ -73,6 +82,27 @@ export function removeRunStoreLockSignalListeners(): void { } } +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, diff --git a/src/runs.ts b/src/runs.ts index 82f477d..c76e430 100644 --- a/src/runs.ts +++ b/src/runs.ts @@ -1,18 +1,26 @@ import { appendFileSync, 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, type NodeStoreLockOwner } from "./run-storage.js"; +import { + acquireNodeStoreLock, + acquireRunStoreLock, + replaceRunStateFile, + type NodeStoreLockOwner, +} from "./run-storage.js"; import type { AgentName, AllowMode, Env, ReasoningEffort } from "./types.js"; const privateDirMode = 0o700; @@ -374,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); diff --git a/tests/run-coordination.test.ts b/tests/run-coordination.test.ts index f139fd9..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 { @@ -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 { diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index a8f8186..9d8ec84 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -12,6 +12,7 @@ import { acquireRunStoreLock, isRunStoreLockSignalListener, removeRunStoreLockSignalListeners, + replaceRunStateFile, } from "../src/run-storage.ts"; function withTemporaryDirectory(callback: (directory: string) => void): void { @@ -318,6 +319,77 @@ function managedTestSignals(): NodeJS.Signals[] { : ["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 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 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, 20); +}); + async function waitForProcessGroupExit(processGroupId: number): Promise { const deadline = Date.now() + 5_000; while (Date.now() < deadline) { From dacc0c9248e18c340e9e82cf152c93af3dab2fd9 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:27:07 +0200 Subject: [PATCH 03/20] ci: test run storage on Windows --- .github/workflows/ci.yml | 29 ++++++++++++++++++++ tests/run-storage.test.ts | 58 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f14d77b..06c4a9d 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 + python: name: Python ${{ matrix.python-version }} runs-on: ubuntu-latest diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 9d8ec84..1871cec 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -3,7 +3,7 @@ import { spawn } 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 } from "node:path"; +import { join, win32 } from "node:path"; import { pathToFileURL } from "node:url"; import test from "node:test"; @@ -108,7 +108,7 @@ test("node store lock persists private owner identity while its process is alive const ownerPath = `${lockPath}.owner`; const release = acquireNodeStoreLock(lockPath, "worker-1", { processTreeRootPid: process.pid }); - assert.equal(statSync(ownerPath).mode & 0o777, 0o600); + if (process.platform !== "win32") assert.equal(statSync(ownerPath).mode & 0o777, 0o600); assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); release(); assert.equal(existsSync(ownerPath), false); @@ -390,6 +390,60 @@ test("run state replacement bounds Windows retries", () => { assert.equal(attempts, 20); }); +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 = $args[0]", + "$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, 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); + }); + + replaceRunStateFile(source, destination); + await lockerExit; + + 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) { From d596cf3d56a2f6a487d0e8b447cf0f29dca40ef2 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:32:20 +0200 Subject: [PATCH 04/20] test: cover run state replacement defaults --- tests/run-storage.test.ts | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 1871cec..2c0466b 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -336,6 +336,19 @@ test("run state replacement retries transient Windows rename failures", () => { 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" }); @@ -372,6 +385,20 @@ test("run state replacement does not retry permanent Windows rename failures", ( 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" }); From 530a5667f60c10ab2867dff72b9b22087f1f38ad Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:41:41 +0200 Subject: [PATCH 05/20] test: pass Windows lock path through environment --- tests/run-storage.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 2c0466b..43f3c2d 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -427,13 +427,14 @@ test("run state replacement waits for a native Windows sharing lock", { skip: pr const systemRoot = process.env.SystemRoot ?? "C:\\Windows"; const powershell = win32.join(systemRoot, "System32", "WindowsPowerShell", "v1.0", "powershell.exe"); const script = [ - "$path = $args[0]", + "$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, destination], { + const locker = spawn(powershell, ["-NoProfile", "-NonInteractive", "-Command", script], { + env: { ...process.env, HEADLESS_TEST_LOCK_PATH: destination }, stdio: ["ignore", "pipe", "inherit"], windowsHide: true, }); From fc8ddb444724b46b5c69f99c89fde98e0a20877b Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:44:54 +0200 Subject: [PATCH 06/20] test: preserve Windows replacement failures --- tests/run-storage.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 43f3c2d..7c5ba49 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -462,8 +462,14 @@ test("run state replacement waits for a native Windows sharing lock", { skip: pr locker.once("exit", onEarlyExit); }); - replaceRunStateFile(source, destination); + 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); From cbf13f9abfdfc8f24486edc683c39449437d366f Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 19:48:19 +0200 Subject: [PATCH 07/20] fix: extend Windows state replacement retries --- src/run-storage.ts | 2 +- tests/run-storage.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index 4dfec5a..e133540 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -26,7 +26,7 @@ const runLockTimeoutMs = 30_000; const runLockRetryMs = 10; const ownerIdentityLifetimeMs = 24 * 60 * 60 * 1_000; const windowsProbeFailureGraceMs = 30_000; -const windowsRenameAttempts = 20; +const windowsRenameAttempts = 40; const windowsRenameRetryMs = 25; const windowsRenameRetryCodes = new Set(["EACCES", "EBUSY", "EPERM"]); const managedSignals: NodeJS.Signals[] = process.platform === "win32" diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 7c5ba49..621d221 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -414,7 +414,7 @@ test("run state replacement bounds Windows retries", () => { }), failure, ); - assert.equal(attempts, 20); + assert.equal(attempts, 40); }); test("run state replacement waits for a native Windows sharing lock", { skip: process.platform !== "win32" }, async () => { From 0cafee64629fb6b68c5cfbbc28e1591dd10a7f89 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:01:59 +0200 Subject: [PATCH 08/20] fix: preserve live node lock owners --- src/run-storage.ts | 47 ++++++++++++++++++++-------- tests/run-storage.test.ts | 64 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 96 insertions(+), 15 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index e133540..6e2a4e0 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -123,36 +123,47 @@ function acquireStoreLock( 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 (owner) writeLockOwner(lockPath, owner); } catch (error) { - release(); + releaseLease(); throw error; } return () => { if (owner) rmSync(lockOwnerPath(lockPath), { force: true }); if (compromisedError) throw compromisedError; - release(); + releaseLease(); }; } function leaseOwnerBlocksAcquisition(lockPath: string): boolean { - let lockMtimeMs: number; - try { - const status = lstatSync(lockPath); - if (!status.isDirectory()) return false; - lockMtimeMs = status.mtimeMs; - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; - throw error; - } + return storedOwnerBlocksAcquisition(lockPath, leaseHeartbeatAt(lockPath)); +} +function ownerSidecarBlocksAcquisition(lockPath: string): boolean { + return storedOwnerBlocksAcquisition(lockPath); +} + +function storedOwnerBlocksAcquisition(lockPath: string, leaseHeartbeatAtMs?: number): boolean { + const ownerPath = lockOwnerPath(lockPath); let owner: StoredNodeStoreLockOwner; + let ownerHeartbeatAtMs: number; try { - owner = JSON.parse(readFileSync(lockOwnerPath(lockPath), "utf8")) as StoredNodeStoreLockOwner; + ownerHeartbeatAtMs = lstatSync(ownerPath).mtimeMs; + owner = JSON.parse(readFileSync(ownerPath, "utf8")) as StoredNodeStoreLockOwner; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT" || error instanceof SyntaxError) return false; return true; @@ -165,7 +176,17 @@ function leaseOwnerBlocksAcquisition(lockPath: string): boolean { ) { return false; } - return processTreeAlive(owner, lockMtimeMs); + return processTreeAlive(owner, leaseHeartbeatAtMs ?? ownerHeartbeatAtMs); +} + +function leaseHeartbeatAt(lockPath: string): number | undefined { + try { + const status = lstatSync(lockPath); + return status.isDirectory() ? status.mtimeMs : undefined; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + return undefined; + } } function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index 621d221..e29f815 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { spawn } from "node:child_process"; +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"; @@ -115,6 +115,66 @@ test("node store lock persists private owner identity while its process is alive }); }); +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"); @@ -220,7 +280,7 @@ test("node store lock blocks stale takeover while the owner's process group is a await once(owner.stdout, "data"); owner.kill("SIGKILL"); await once(owner, "exit"); - agePath(lockPath); + rmSync(lockPath, { recursive: true }); assert.throws(() => acquireNodeStoreLock(lockPath, "worker-1"), /node is locked: worker-1/); From f1da717c18e00ab454e9ccacf7cdc2cc381f1b88 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:18:07 +0200 Subject: [PATCH 09/20] fix: retain async agent lock ownership --- src/async-run-message.ts | 40 +++++++- src/cli.ts | 34 ++++++- src/run-storage.ts | 81 ++++++++++++--- tests/async-run-message.test.ts | 173 ++++++++++++++++++++++++++++++++ 4 files changed, 304 insertions(+), 24 deletions(-) create mode 100644 tests/async-run-message.test.ts diff --git a/src/async-run-message.ts b/src/async-run-message.ts index f8d7752..9a95bb7 100644 --- a/src/async-run-message.ts +++ b/src/async-run-message.ts @@ -1,8 +1,17 @@ import { execFileSync, spawn, type ChildProcess } from "node:child_process"; import { windowsTaskkillPath } from "./process-tree.js"; -import { removeRunStoreLockSignalListeners } from "./run-storage.js"; -import { acquireNodeLock, appendNodeLog, updateNodeStatus } from "./runs.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 { @@ -122,11 +131,29 @@ function runChild(currentTask: AsyncRunMessageTask): Promise { return new Promise((resolve, reject) => { const child = spawn(currentTask.command.command, currentTask.command.args, { cwd: currentTask.cwd, - env: { ...process.env, HEADLESS_ASYNC_MESSAGE_WORKER: "1" }, + 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 }, + ); + } catch (error) { + terminateActiveChild("SIGKILL"); + reject(error); + } + }); child.once("close", (code, signal) => { activeChild = undefined; resolve(signal ? 1 : (code ?? 1)); @@ -138,7 +165,12 @@ function finish(code = 0): void { if (finished) return; finished = true; try { - releaseLock?.(); + if (!task || !nodeStoreLockHasLiveSuccessor( + nodeLockPath(process.env as Env, task.runId, task.nodeId), + process.pid, + )) { + releaseLock?.(); + } } finally { releaseLock = undefined; if (process.connected) process.disconnect(); diff --git a/src/cli.ts b/src/cli.ts index 59d977f..b265e19 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -102,10 +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 } from "./run-storage.js"; +import { isRunStoreLockSignalListener, waitForNodeStoreLockOwner } from "./run-storage.js"; import { appendNodeLog, completeIdleRunNodes, + nodeLockPath, readRun, registerNode, runDirectory, @@ -1889,18 +1890,24 @@ async function executeCommand( resolve(result); }; const asyncMessageWorker = env.HEADLESS_ASYNC_MESSAGE_WORKER === "1"; - const ownsChildProcessGroup = process.platform !== "win32" && ( - asyncMessageWorker - || options.timeoutSeconds !== undefined + 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, }); @@ -3231,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; diff --git a/src/run-storage.ts b/src/run-storage.ts index 6e2a4e0..ec1f8c3 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -44,6 +44,11 @@ interface StoredNodeStoreLockOwner extends NodeStoreLockOwner { processStartIdentity?: string; } +interface StoredNodeStoreLockOwnerSnapshot { + owner: StoredNodeStoreLockOwner; + heartbeatAtMs: number; +} + export interface RunStateReplacementOptions { platform?: NodeJS.Platform; rename?: (source: string, destination: string) => void; @@ -72,6 +77,43 @@ export function acquireNodeStoreLock(lockPath: string, nodeId: string, owner?: N } } +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, leaseHeartbeatAt(lockPath) ?? snapshot.heartbeatAtMs); + } catch { + return true; + } +} + export function isRunStoreLockSignalListener(listener: NodeJS.SignalsListener): boolean { return runStoreLockSignalListeners.has(listener); } @@ -158,25 +200,34 @@ function ownerSidecarBlocksAcquisition(lockPath: string): boolean { } function storedOwnerBlocksAcquisition(lockPath: string, leaseHeartbeatAtMs?: number): boolean { - const ownerPath = lockOwnerPath(lockPath); - let owner: StoredNodeStoreLockOwner; - let ownerHeartbeatAtMs: number; + let snapshot: StoredNodeStoreLockOwnerSnapshot | undefined; try { - ownerHeartbeatAtMs = lstatSync(ownerPath).mtimeMs; - owner = JSON.parse(readFileSync(ownerPath, "utf8")) as StoredNodeStoreLockOwner; + snapshot = readStoredLockOwner(lockPath); } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT" || error instanceof SyntaxError) return false; + if (error instanceof SyntaxError) return false; return true; } - if ( - !Number.isSafeInteger(owner.processTreeRootPid) - || owner.processTreeRootPid <= 0 - || !Number.isFinite(owner.createdAtMs) - || Date.now() - owner.createdAtMs > ownerIdentityLifetimeMs - ) { - return false; + if (!snapshot || !storedOwnerIsTrusted(snapshot.owner)) return false; + return processTreeAlive(snapshot.owner, leaseHeartbeatAtMs ?? snapshot.heartbeatAtMs); +} + +function readStoredLockOwner(lockPath: string): StoredNodeStoreLockOwnerSnapshot | undefined { + const ownerPath = lockOwnerPath(lockPath); + try { + const heartbeatAtMs = lstatSync(ownerPath).mtimeMs; + const owner = JSON.parse(readFileSync(ownerPath, "utf8")) as StoredNodeStoreLockOwner; + return { owner, heartbeatAtMs }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + throw error; } - return processTreeAlive(owner, leaseHeartbeatAtMs ?? ownerHeartbeatAtMs); +} + +function storedOwnerIsTrusted(owner: StoredNodeStoreLockOwner): boolean { + return Number.isSafeInteger(owner.processTreeRootPid) + && owner.processTreeRootPid > 0 + && Number.isFinite(owner.createdAtMs) + && Date.now() - owner.createdAtMs <= ownerIdentityLifetimeMs; } function leaseHeartbeatAt(lockPath: string): number | undefined { @@ -204,7 +255,7 @@ function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { fchmodSync(descriptor, 0o600); closeSync(descriptor); descriptor = undefined; - renameSync(temporaryPath, path); + replaceRunStateFile(temporaryPath, path); } finally { if (descriptor !== undefined) closeSync(descriptor); rmSync(temporaryPath, { force: true }); diff --git a/tests/async-run-message.test.ts b/tests/async-run-message.test.ts new file mode 100644 index 0000000..fa76ae2 --- /dev/null +++ b/tests/async-run-message.test.ts @@ -0,0 +1,173 @@ +import assert from "node:assert/strict"; +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 } from "../src/runs.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 }); + } +}); From 37a0d67d3fc2ae298b325f2f0987031f02aa4556 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:22:26 +0200 Subject: [PATCH 10/20] fix: roll back failed async message starts --- src/run-commands.ts | 11 +++++++ tests/async-run-message.test.ts | 58 ++++++++++++++++++++++++++++++++- 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/src/run-commands.ts b/src/run-commands.ts index d9db2c7..ab7b103 100644 --- a/src/run-commands.ts +++ b/src/run-commands.ts @@ -338,6 +338,17 @@ async function startAsyncRunMessage( await sendWorkerRequest(worker, { type: "start" }); } catch (error) { cancelAsyncWorker(worker); + try { + updateNodeStatus( + handlers.env, + runId, + nodeId, + "failed", + error instanceof Error ? error.message : String(error), + ); + } catch { + // Preserve the startup error when rollback storage also fails. + } throw error; } if (worker.connected) worker.disconnect(); diff --git a/tests/async-run-message.test.ts b/tests/async-run-message.test.ts index fa76ae2..88cf84e 100644 --- a/tests/async-run-message.test.ts +++ b/tests/async-run-message.test.ts @@ -6,7 +6,7 @@ import { join } from "node:path"; import test from "node:test"; import { runCli } from "../src/cli.ts"; -import { acquireNodeLock, nodeLockPath, readRun, registerNode } from "../src/runs.ts"; +import { acquireNodeLock, nodeLockPath, readRun, registerNode, runDirectory } from "../src/runs.ts"; import type { Env } from "../src/types.ts"; interface AsyncMessageFixture { @@ -171,3 +171,59 @@ test("async run message handles a missing agent without leaking its lock", async 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 }); + } +}); From 8df0d7524100a74f60b5af83615a9cbe2de5eb26 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:32:35 +0200 Subject: [PATCH 11/20] fix: fail closed on process probe ambiguity --- src/run-storage.ts | 141 +++++++++++++++++------ tests/run-storage-process.test.ts | 181 ++++++++++++++++++++++++++++++ tests/run-storage.test.ts | 20 +--- 3 files changed, 293 insertions(+), 49 deletions(-) create mode 100644 tests/run-storage-process.test.ts diff --git a/src/run-storage.ts b/src/run-storage.ts index ec1f8c3..ee5b067 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -24,8 +24,7 @@ const nodeLockStaleMs = 10_000; const nodeLockUpdateMs = 5_000; const runLockTimeoutMs = 30_000; const runLockRetryMs = 10; -const ownerIdentityLifetimeMs = 24 * 60 * 60 * 1_000; -const windowsProbeFailureGraceMs = 30_000; +const maximumProcessId = 0xffff_ffff; const windowsRenameAttempts = 40; const windowsRenameRetryMs = 25; const windowsRenameRetryCodes = new Set(["EACCES", "EBUSY", "EPERM"]); @@ -46,7 +45,6 @@ interface StoredNodeStoreLockOwner extends NodeStoreLockOwner { interface StoredNodeStoreLockOwnerSnapshot { owner: StoredNodeStoreLockOwner; - heartbeatAtMs: number; } export interface RunStateReplacementOptions { @@ -55,6 +53,20 @@ export interface RunStateReplacementOptions { 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) { @@ -108,7 +120,7 @@ export function nodeStoreLockHasLiveSuccessor(lockPath: string, currentProcessTr const snapshot = readStoredLockOwner(lockPath); if (!snapshot || snapshot.owner.processTreeRootPid === currentProcessTreeRootPid) return false; if (!storedOwnerIsTrusted(snapshot.owner)) return true; - return processTreeAlive(snapshot.owner, leaseHeartbeatAt(lockPath) ?? snapshot.heartbeatAtMs); + return processTreeAlive(snapshot.owner); } catch { return true; } @@ -192,14 +204,14 @@ function acquireStoreLock( } function leaseOwnerBlocksAcquisition(lockPath: string): boolean { - return storedOwnerBlocksAcquisition(lockPath, leaseHeartbeatAt(lockPath)); + return storedOwnerBlocksAcquisition(lockPath); } function ownerSidecarBlocksAcquisition(lockPath: string): boolean { return storedOwnerBlocksAcquisition(lockPath); } -function storedOwnerBlocksAcquisition(lockPath: string, leaseHeartbeatAtMs?: number): boolean { +function storedOwnerBlocksAcquisition(lockPath: string): boolean { let snapshot: StoredNodeStoreLockOwnerSnapshot | undefined; try { snapshot = readStoredLockOwner(lockPath); @@ -208,15 +220,14 @@ function storedOwnerBlocksAcquisition(lockPath: string, leaseHeartbeatAtMs?: num return true; } if (!snapshot || !storedOwnerIsTrusted(snapshot.owner)) return false; - return processTreeAlive(snapshot.owner, leaseHeartbeatAtMs ?? snapshot.heartbeatAtMs); + return processTreeAlive(snapshot.owner); } function readStoredLockOwner(lockPath: string): StoredNodeStoreLockOwnerSnapshot | undefined { const ownerPath = lockOwnerPath(lockPath); try { - const heartbeatAtMs = lstatSync(ownerPath).mtimeMs; const owner = JSON.parse(readFileSync(ownerPath, "utf8")) as StoredNodeStoreLockOwner; - return { owner, heartbeatAtMs }; + return { owner }; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; throw error; @@ -226,18 +237,8 @@ function readStoredLockOwner(lockPath: string): StoredNodeStoreLockOwnerSnapshot function storedOwnerIsTrusted(owner: StoredNodeStoreLockOwner): boolean { return Number.isSafeInteger(owner.processTreeRootPid) && owner.processTreeRootPid > 0 - && Number.isFinite(owner.createdAtMs) - && Date.now() - owner.createdAtMs <= ownerIdentityLifetimeMs; -} - -function leaseHeartbeatAt(lockPath: string): number | undefined { - try { - const status = lstatSync(lockPath); - return status.isDirectory() ? status.mtimeMs : undefined; - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; - return undefined; - } + && owner.processTreeRootPid <= maximumProcessId + && Number.isFinite(owner.createdAtMs); } function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { @@ -330,7 +331,7 @@ function processAlive(pid: number): boolean { } } -function processTreeAlive(owner: StoredNodeStoreLockOwner, lastHeartbeatAtMs: number): boolean { +function processTreeAlive(owner: StoredNodeStoreLockOwner): boolean { const rootPid = owner.processTreeRootPid; if (processAlive(rootPid)) { const currentIdentity = processStartIdentity(rootPid); @@ -345,23 +346,56 @@ function processTreeAlive(owner: StoredNodeStoreLockOwner, lastHeartbeatAtMs: nu return (error as NodeJS.ErrnoException).code === "EPERM"; } } + return windowsProcessTreeAlive(rootPid); +} + +export function windowsProcessTreeAlive( + rootPid: number, + options: WindowsProcessTreeProbeOptions = {}, +): boolean { + const execute = options.execute ?? executeWindowsPowerShell; try { - return execFileSync(windowsPowerShellPath(), [ + const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", "-NonInteractive", "-Command", windowsDescendantProbe, String(rootPid), - ], { - encoding: "utf8", - timeout: 3_000, - windowsHide: true, - }).trim() === "1"; + ]); + return output.trim() !== "HEADLESS_PROCESS_TREE_DEAD"; + } catch { + return true; + } +} + +export function windowsProcessStartIdentity( + pid: number, + options: WindowsProcessTreeProbeOptions = {}, +): string | undefined { + const execute = options.execute ?? executeWindowsPowerShell; + try { + const output = execute(windowsPowerShellPath(options.systemRoot), [ + "-NoProfile", + "-NonInteractive", + "-Command", + windowsProcessStartIdentityProbe, + String(pid), + ]).trim(); + const match = /^HEADLESS_PROCESS_START:(\d+)$/.exec(output); + return match ? `win32:${match[1]}` : undefined; } catch { - return Date.now() - lastHeartbeatAtMs < windowsProbeFailureGraceMs; + return undefined; } } +function executeWindowsPowerShell(command: string, args: string[]): string { + return execFileSync(command, args, { + encoding: "utf8", + timeout: 3_000, + windowsHide: true, + }); +} + function processStartIdentity(pid: number): string | undefined { if (process.platform === "linux") { try { @@ -372,16 +406,52 @@ function processStartIdentity(pid: number): string | undefined { return undefined; } } + if (process.platform === "darwin") { + return macosProcessStartIdentity(pid); + } + if (process.platform === "win32") { + return windowsProcessStartIdentity(pid); + } return undefined; } -function windowsPowerShellPath(): string { - const systemRoot = process.env.SystemRoot; +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"); } const windowsDescendantProbe = [ + "$ErrorActionPreference = 'Stop'", "$rootPid = [uint32]$args[0]", "$processes = @(Get-CimInstance Win32_Process)", "$parents = @($rootPid)", @@ -391,7 +461,14 @@ const windowsDescendantProbe = [ " if ($children.Count -gt 0) { $found = $true }", " $parents = @($children | ForEach-Object { $_.ProcessId })", "} while ($parents.Count -gt 0)", - "if ($found) { '1' } else { '0' }", + "if ($found) { 'HEADLESS_PROCESS_TREE_ALIVE' } else { 'HEADLESS_PROCESS_TREE_DEAD' }", +].join("; "); + +const windowsProcessStartIdentityProbe = [ + "$ErrorActionPreference = 'Stop'", + "$rootPid = [uint32]$args[0]", + "$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 { diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts new file mode 100644 index 0000000..5e09401 --- /dev/null +++ b/tests/run-storage-process.test.ts @@ -0,0 +1,181 @@ +import assert from "node:assert/strict"; +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, + 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() - 20_000); + utimesSync(path, stale, stale); +} + +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", () => { + 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.processStartIdentity = `${owner.processStartIdentity}-reused`; + writeFileSync(ownerPath, `${JSON.stringify(owner)}\n`); + rmSync(lockPath, { recursive: true }); + let releaseReplacement: (() => void) | undefined; + + try { + releaseReplacement = acquireNodeStoreLock(lockPath, "worker-1"); + releaseReplacement(); + releaseReplacement = undefined; + assert.equal(existsSync(ownerPath), false); + } finally { + if (releaseReplacement) releaseReplacement(); + else if (existsSync(lockPath)) releaseOwner(); + } + }); +}); + +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/); +}); + +test("Windows process-start probe accepts only an explicit identity", () => { + const probe = (output: string) => windowsProcessStartIdentity(123, { execute: () => output }); + + assert.equal(probe("HEADLESS_PROCESS_START:638920627920000000\r\n"), "win32:638920627920000000"); + 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"); +}); diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index e29f815..bc1f1aa 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -107,8 +107,11 @@ test("node store lock persists private owner identity while its process is alive 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); @@ -199,23 +202,6 @@ test("node store lock recovers a stale lease directory", () => { }); }); -test("node store lock does not trust an expired owner identity", () => { - withTemporaryDirectory((directory) => { - const lockPath = join(directory, "session.lock"); - mkdirSync(lockPath); - writeFileSync(`${lockPath}.owner`, `${JSON.stringify({ - createdAtMs: 0, - processTreeRootPid: process.pid, - })}\n`); - agePath(lockPath); - - const release = acquireNodeStoreLock(lockPath, "worker-1"); - - release(); - assert.equal(existsSync(`${lockPath}.owner`), false); - }); -}); - test("node store lock does not trust an invalid owner PID", () => { withTemporaryDirectory((directory) => { const lockPath = join(directory, "session.lock"); From 46cb02b8129716691983c22a5f2b04c0fc4744da Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:36:18 +0200 Subject: [PATCH 12/20] fix: preserve Windows descendants after PID reuse --- src/run-storage.ts | 28 ++++++++++++++++++++++------ tests/run-storage-process.test.ts | 15 +++++++++++++++ 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index ee5b067..df4a27c 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -333,20 +333,36 @@ function processAlive(pid: number): boolean { function processTreeAlive(owner: StoredNodeStoreLockOwner): boolean { const rootPid = owner.processTreeRootPid; - if (processAlive(rootPid)) { + const rootAlive = processAlive(rootPid); + let identityMismatch = false; + if (rootAlive) { const currentIdentity = processStartIdentity(rootPid); - if (owner.processStartIdentity && currentIdentity && owner.processStartIdentity !== currentIdentity) return false; - return true; + identityMismatch = Boolean( + owner.processStartIdentity + && currentIdentity + && owner.processStartIdentity !== currentIdentity, + ); } - if (process.platform !== "win32") { + 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"; } - } - return windowsProcessTreeAlive(rootPid); + }); +} + +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( diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts index 5e09401..d457317 100644 --- a/tests/run-storage-process.test.ts +++ b/tests/run-storage-process.test.ts @@ -7,6 +7,7 @@ import test from "node:test"; import { acquireNodeStoreLock, macosProcessStartIdentity, + processTreeAliveFromProbes, windowsProcessStartIdentity, windowsProcessTreeAlive, } from "../src/run-storage.ts"; @@ -179,3 +180,17 @@ test("macOS process-start probe canonicalizes timezone and locale", () => { 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); +}); From 70ce60e6b3ade0ecce3d4f79dd9b4ade373b49bb Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:38:12 +0200 Subject: [PATCH 13/20] ci: run process probes on Windows --- .github/workflows/ci.yml | 2 +- tests/ci.test.ts | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) create mode 100644 tests/ci.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 06c4a9d..8585333 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,7 +71,7 @@ jobs: run: npm run build - name: Test run storage - run: node --import tsx --test tests/run-storage.test.ts + run: node --import tsx --test tests/run-storage.test.ts tests/run-storage-process.test.ts python: name: Python ${{ matrix.python-version }} 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/); +}); From ea5d952e408f89fbd50e3fe15f754a8a41a9e5d3 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:44:01 +0200 Subject: [PATCH 14/20] fix: repair interrupted async message starts --- src/async-run-message.ts | 18 ++++- src/run-commands.ts | 36 +++++++++- tests/async-run-message.test.ts | 115 +++++++++++++++++++++++++++++++- 3 files changed, 164 insertions(+), 5 deletions(-) diff --git a/src/async-run-message.ts b/src/async-run-message.ts index 9a95bb7..6ee2a27 100644 --- a/src/async-run-message.ts +++ b/src/async-run-message.ts @@ -28,6 +28,7 @@ export type AsyncRunMessageRequest = export type AsyncRunMessageResponse = | { type: "ready" } + | { type: "started" } | { type: "error"; message: string }; let task: AsyncRunMessageTask | undefined; @@ -41,7 +42,7 @@ if (process.send) { void handleRequest(message); }); process.once("disconnect", () => { - if (!started) finish(); + if (!started && !finished) failBeforeStart("async message parent disconnected before agent startup"); }); for (const signal of forwardedSignals()) { process.on(signal, () => terminateActiveChild(signal)); @@ -57,7 +58,7 @@ async function handleRequest(request: AsyncRunMessageRequest): Promise { if (started) { terminateActiveChild("SIGTERM"); } else { - finish(); + finish(1); } return; } @@ -84,6 +85,18 @@ function terminateActiveChild(signal: NodeJS.Signals): void { 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 { @@ -149,6 +162,7 @@ function runChild(currentTask: AsyncRunMessageTask): Promise { process.pid, { processTreeRootPid: child.pid }, ); + send({ type: "started" }); } catch (error) { terminateActiveChild("SIGKILL"); reject(error); diff --git a/src/run-commands.ts b/src/run-commands.ts index ab7b103..57750d1 100644 --- a/src/run-commands.ts +++ b/src/run-commands.ts @@ -335,9 +335,8 @@ async function startAsyncRunMessage( 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 sendWorkerRequest(worker, { type: "start" }); + await startPreparedAsyncWorker(worker); } catch (error) { - cancelAsyncWorker(worker); try { updateNodeStatus( handlers.env, @@ -349,6 +348,7 @@ async function startAsyncRunMessage( } catch { // Preserve the startup error when rollback storage also fails. } + cancelAsyncWorker(worker); throw error; } if (worker.connected) worker.disconnect(); @@ -396,6 +396,38 @@ function sendWorkerRequest(worker: ChildProcess, request: AsyncRunMessageRequest }); } +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) { diff --git a/tests/async-run-message.test.ts b/tests/async-run-message.test.ts index 88cf84e..32b24db 100644 --- a/tests/async-run-message.test.ts +++ b/tests/async-run-message.test.ts @@ -1,4 +1,6 @@ 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"; @@ -6,7 +8,15 @@ import { join } from "node:path"; import test from "node:test"; import { runCli } from "../src/cli.ts"; -import { acquireNodeLock, nodeLockPath, readRun, registerNode, runDirectory } from "../src/runs.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 { @@ -172,6 +182,29 @@ test("async run message handles a missing agent without leaking its lock", async } }); +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 { @@ -227,3 +260,83 @@ test("async run message preserves failed status after a partial state write", as 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 }); + } +}); From 73113dbf42b56b1dea9246b73343a39d8b500921 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:50:30 +0200 Subject: [PATCH 15/20] fix: pass Windows probe PIDs reliably --- src/run-storage.ts | 61 +++++++++++++++++-------------- tests/run-storage-process.test.ts | 12 +++++- 2 files changed, 44 insertions(+), 29 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index df4a27c..81615cf 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -235,10 +235,11 @@ function readStoredLockOwner(lockPath: string): StoredNodeStoreLockOwnerSnapshot } function storedOwnerIsTrusted(owner: StoredNodeStoreLockOwner): boolean { - return Number.isSafeInteger(owner.processTreeRootPid) - && owner.processTreeRootPid > 0 - && owner.processTreeRootPid <= maximumProcessId - && Number.isFinite(owner.createdAtMs); + 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 { @@ -369,14 +370,14 @@ export function windowsProcessTreeAlive( rootPid: number, options: WindowsProcessTreeProbeOptions = {}, ): boolean { + if (!validProcessId(rootPid)) return true; const execute = options.execute ?? executeWindowsPowerShell; try { const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", "-NonInteractive", "-Command", - windowsDescendantProbe, - String(rootPid), + windowsDescendantProbe(rootPid), ]); return output.trim() !== "HEADLESS_PROCESS_TREE_DEAD"; } catch { @@ -388,14 +389,14 @@ export function windowsProcessStartIdentity( pid: number, options: WindowsProcessTreeProbeOptions = {}, ): string | undefined { + if (!validProcessId(pid)) return undefined; const execute = options.execute ?? executeWindowsPowerShell; try { const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", "-NonInteractive", "-Command", - windowsProcessStartIdentityProbe, - String(pid), + windowsProcessStartIdentityProbe(pid), ]).trim(); const match = /^HEADLESS_PROCESS_START:(\d+)$/.exec(output); return match ? `win32:${match[1]}` : undefined; @@ -466,26 +467,30 @@ function windowsPowerShellPath(systemRoot = process.env.SystemRoot): string { return win32.join(root, "System32", "WindowsPowerShell", "v1.0", "powershell.exe"); } -const windowsDescendantProbe = [ - "$ErrorActionPreference = 'Stop'", - "$rootPid = [uint32]$args[0]", - "$processes = @(Get-CimInstance Win32_Process)", - "$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("; "); - -const windowsProcessStartIdentityProbe = [ - "$ErrorActionPreference = 'Stop'", - "$rootPid = [uint32]$args[0]", - "$process = Get-CimInstance Win32_Process -Filter \"ProcessId = $rootPid\"", - "if ($null -ne $process) { 'HEADLESS_PROCESS_START:' + $process.CreationDate.ToUniversalTime().Ticks }", -].join("; "); +function windowsDescendantProbe(rootPid: number): string { + return [ + "$ErrorActionPreference = 'Stop'", + `$rootPid = [uint32]${rootPid}`, + "$processes = @(Get-CimInstance Win32_Process)", + "$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" }); diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts index d457317..903381f 100644 --- a/tests/run-storage-process.test.ts +++ b/tests/run-storage-process.test.ts @@ -147,12 +147,22 @@ test("Windows process-tree probe makes CIM failures terminating", () => { }); assert.match(command, /ErrorActionPreference.*Stop/); + assert.match(command, /rootPid = \[uint32\]123/); + assert.doesNotMatch(command, /\$args/); }); test("Windows process-start probe accepts only an explicit identity", () => { - const probe = (output: string) => windowsProcessStartIdentity(123, { execute: () => output }); + 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); From a3e84f09c9d3c76c4d0169db24da853ff6dbeea4 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 21:57:11 +0200 Subject: [PATCH 16/20] fix: ignore Windows probe process descendants --- src/run-storage.ts | 2 +- tests/run-storage-process.test.ts | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index 81615cf..707fae3 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -471,7 +471,7 @@ function windowsDescendantProbe(rootPid: number): string { return [ "$ErrorActionPreference = 'Stop'", `$rootPid = [uint32]${rootPid}`, - "$processes = @(Get-CimInstance Win32_Process)", + "$processes = @(Get-CimInstance Win32_Process | Where-Object { $_.ProcessId -ne $PID })", "$parents = @($rootPid)", "$found = $false", "do {", diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts index 903381f..c107b80 100644 --- a/tests/run-storage-process.test.ts +++ b/tests/run-storage-process.test.ts @@ -148,6 +148,7 @@ test("Windows process-tree probe makes CIM failures terminating", () => { assert.match(command, /ErrorActionPreference.*Stop/); assert.match(command, /rootPid = \[uint32\]123/); + assert.match(command, /ProcessId -ne \$PID/); assert.doesNotMatch(command, /\$args/); }); From 134eb8b611609dd98a175f61fcbf0c4bf5456512 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sat, 29 Aug 2026 22:04:25 +0200 Subject: [PATCH 17/20] test: isolate Windows PID reuse fixture --- tests/run-storage-process.test.ts | 95 +++++++++++++++++++++++++------ 1 file changed, 77 insertions(+), 18 deletions(-) diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts index c107b80..adcd394 100644 --- a/tests/run-storage-process.test.ts +++ b/tests/run-storage-process.test.ts @@ -1,4 +1,6 @@ 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"; @@ -26,6 +28,51 @@ function agePath(path: string): void { 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"); @@ -45,27 +92,39 @@ test("node store lock preserves a live owner beyond the identity lifetime", () = }); }); -test("node store lock recovers when a live PID has a different start identity", () => { - 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.processStartIdentity = `${owner.processStartIdentity}-reused`; - writeFileSync(ownerPath, `${JSON.stringify(owner)}\n`); - rmSync(lockPath, { recursive: true }); - let releaseReplacement: (() => void) | undefined; - +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 { - releaseReplacement = acquireNodeStoreLock(lockPath, "worker-1"); - releaseReplacement(); - releaseReplacement = undefined; - assert.equal(existsSync(ownerPath), false); + 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 { - if (releaseReplacement) releaseReplacement(); - else if (existsSync(lockPath)) releaseOwner(); + rmSync(directory, { force: true, recursive: true }); } - }); + } finally { + await stopChildProcess(reusedProcess); + } }); test("node store lock preserves an old identity-less live owner", () => { From a9ee571fa6583309cd3a6f9ecd56c5cc4794f4e6 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sun, 30 Aug 2026 08:29:53 +0200 Subject: [PATCH 18/20] docs: update changelog for run lock recovery --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) 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. From 8fbde30ffd24e17351ffe88b7ec91abf69894539 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sun, 30 Aug 2026 08:48:08 +0200 Subject: [PATCH 19/20] fix: tolerate slow Windows process identity probes --- src/run-storage.ts | 30 +++++++++++++++++++++--------- tests/run-storage-process.test.ts | 2 +- tests/run-storage.test.ts | 13 ++++++++++++- 3 files changed, 34 insertions(+), 11 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index 707fae3..324187a 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -20,11 +20,13 @@ const privateDirMode = 0o700; const legacyInitializationGraceMs = 1_000; const runLockStaleMs = 5_000; const runLockUpdateMs = 2_500; -const nodeLockStaleMs = 10_000; +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 windowsRenameAttempts = 40; const windowsRenameRetryMs = 25; const windowsRenameRetryCodes = new Set(["EACCES", "EBUSY", "EPERM"]); @@ -165,6 +167,7 @@ function acquireStoreLock( ): () => void { if (leaseOwnerBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); if (legacyLockBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); + const storedOwner = owner ? createStoredLockOwner(owner) : undefined; let compromisedError: Error | undefined; const release = loadLockSync()(lockPath, { @@ -190,7 +193,7 @@ function acquireStoreLock( throw lockContentionError(lockPath); } rmSync(lockOwnerPath(lockPath), { force: true }); - if (owner) writeLockOwner(lockPath, owner); + if (storedOwner) writeStoredLockOwner(lockPath, storedOwner); } catch (error) { releaseLease(); throw error; @@ -243,13 +246,20 @@ function validProcessId(pid: number): boolean { } function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { - const path = lockOwnerPath(lockPath); - const temporaryPath = `${path}.tmp-${randomUUID()}`; - const storedOwner: StoredNodeStoreLockOwner = { + writeStoredLockOwner(lockPath, createStoredLockOwner(owner)); +} + +function createStoredLockOwner(owner: NodeStoreLockOwner): StoredNodeStoreLockOwner { + return { ...owner, createdAtMs: Date.now(), processStartIdentity: processStartIdentity(owner.processTreeRootPid), }; +} + +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); @@ -371,7 +381,8 @@ export function windowsProcessTreeAlive( options: WindowsProcessTreeProbeOptions = {}, ): boolean { if (!validProcessId(rootPid)) return true; - const execute = options.execute ?? executeWindowsPowerShell; + const execute = options.execute + ?? ((command, args) => executeWindowsPowerShell(command, args, windowsProcessTreeTimeoutMs)); try { const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", @@ -390,7 +401,8 @@ export function windowsProcessStartIdentity( options: WindowsProcessTreeProbeOptions = {}, ): string | undefined { if (!validProcessId(pid)) return undefined; - const execute = options.execute ?? executeWindowsPowerShell; + const execute = options.execute + ?? ((command, args) => executeWindowsPowerShell(command, args, windowsProcessIdentityTimeoutMs)); try { const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", @@ -405,10 +417,10 @@ export function windowsProcessStartIdentity( } } -function executeWindowsPowerShell(command: string, args: string[]): string { +function executeWindowsPowerShell(command: string, args: string[], timeout: number): string { return execFileSync(command, args, { encoding: "utf8", - timeout: 3_000, + timeout, windowsHide: true, }); } diff --git a/tests/run-storage-process.test.ts b/tests/run-storage-process.test.ts index adcd394..91385ce 100644 --- a/tests/run-storage-process.test.ts +++ b/tests/run-storage-process.test.ts @@ -24,7 +24,7 @@ function withTemporaryDirectory(callback: (directory: string) => void): void { } function agePath(path: string): void { - const stale = new Date(Date.now() - 20_000); + const stale = new Date(Date.now() - 30_000); utimesSync(path, stale, stale); } diff --git a/tests/run-storage.test.ts b/tests/run-storage.test.ts index bc1f1aa..480141e 100644 --- a/tests/run-storage.test.ts +++ b/tests/run-storage.test.ts @@ -24,7 +24,7 @@ function withTemporaryDirectory(callback: (directory: string) => void): void { } } -function agePath(path: string, milliseconds = 20_000): void { +function agePath(path: string, milliseconds = 30_000): void { const stale = new Date(Date.now() - milliseconds); utimesSync(path, stale, stale); } @@ -188,6 +188,17 @@ test("node store lock does not remove an unexpected lock directory", () => { }); }); +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"); From ca588ed16655feb5ee1cbe2e706aedbd941f6bc4 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Sun, 30 Aug 2026 08:52:18 +0200 Subject: [PATCH 20/20] fix: allow slow Windows owner identity capture --- src/run-storage.ts | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/src/run-storage.ts b/src/run-storage.ts index 324187a..a47df86 100644 --- a/src/run-storage.ts +++ b/src/run-storage.ts @@ -27,6 +27,7 @@ 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"]); @@ -167,7 +168,7 @@ function acquireStoreLock( ): () => void { if (leaseOwnerBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); if (legacyLockBlocksAcquisition(lockPath)) throw lockContentionError(lockPath); - const storedOwner = owner ? createStoredLockOwner(owner) : undefined; + const storedOwner = owner ? createStoredLockOwner(owner, windowsOwnerIdentityTimeoutMs) : undefined; let compromisedError: Error | undefined; const release = loadLockSync()(lockPath, { @@ -246,14 +247,17 @@ function validProcessId(pid: number): boolean { } function writeLockOwner(lockPath: string, owner: NodeStoreLockOwner): void { - writeStoredLockOwner(lockPath, createStoredLockOwner(owner)); + writeStoredLockOwner(lockPath, createStoredLockOwner(owner, windowsProcessIdentityTimeoutMs)); } -function createStoredLockOwner(owner: NodeStoreLockOwner): StoredNodeStoreLockOwner { +function createStoredLockOwner( + owner: NodeStoreLockOwner, + windowsIdentityTimeoutMs: number, +): StoredNodeStoreLockOwner { return { ...owner, createdAtMs: Date.now(), - processStartIdentity: processStartIdentity(owner.processTreeRootPid), + processStartIdentity: processStartIdentity(owner.processTreeRootPid, windowsIdentityTimeoutMs), }; } @@ -399,10 +403,18 @@ export function windowsProcessTreeAlive( 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, windowsProcessIdentityTimeoutMs)); + ?? ((command, args) => executeWindowsPowerShell(command, args, timeoutMs)); try { const output = execute(windowsPowerShellPath(options.systemRoot), [ "-NoProfile", @@ -425,7 +437,7 @@ function executeWindowsPowerShell(command: string, args: string[], timeout: numb }); } -function processStartIdentity(pid: number): string | undefined { +function processStartIdentity(pid: number, windowsTimeoutMs = windowsProcessIdentityTimeoutMs): string | undefined { if (process.platform === "linux") { try { const stat = readFileSync(`/proc/${pid}/stat`, "utf8"); @@ -439,7 +451,7 @@ function processStartIdentity(pid: number): string | undefined { return macosProcessStartIdentity(pid); } if (process.platform === "win32") { - return windowsProcessStartIdentity(pid); + return probeWindowsProcessStartIdentity(pid, {}, windowsTimeoutMs); } return undefined; }