openclaw / src /process /supervisor /supervisor.ts
SaylorTwift's picture
SaylorTwift HF Staff
Add files using upload-large-folder tool
5cb63c1 verified
Raw History Blame Contribute Delete
26.5 kB
// Process supervisor manages long-running child and PTY process lifecycles.
import crypto from "node:crypto";
import { performance } from "node:perf_hooks";
import { expectDefined } from "@openclaw/normalization-core";
import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { sliceUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { createDeferredCore } from "../../shared/deferred.js";
import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js";
import { createChildAdapter } from "./adapters/child.js";
import { createPtyAdapter } from "./adapters/pty.js";
import { GRACEFUL_CANCEL_TIMEOUT_MS } from "./cancellation-policy.js";
import type {
ManagedRun,
ProcessSupervisor,
ProcessScopeCleanupPolicy,
RunExit,
SpawnInput,
SpawnProcessAdapter,
TerminationReason,
} from "./types.js";
type OwnedRun = {
runId: string;
scopeKey?: string;
terminationReason?: TerminationReason;
cancel: (reason: TerminationReason) => void;
pending?: Promise<ManagedRun>;
waitForExtinction?: () => Promise<void>;
cleanupOwners: ScopeCleanupOwner[];
};
type ScopeCleanupOwner = { processTree: ProcessScopeCleanupPolicy; failure?: { error: unknown } };
function requiresProcessTree(scope: ScopeCleanupOwner, external: boolean): boolean {
return scope.processTree === "required-all" || (scope.processTree === "owned-only" && !external);
}
function recordScopeCleanupFailure(owner: OwnedRun, error: unknown): void {
for (const cleanupOwner of owner.cleanupOwners) {
cleanupOwner.failure ??= { error };
}
}
type StartingScope = {
runs: Set<Promise<ManagedRun>>;
replacement?: Promise<ManagedRun>;
};
const DEFAULT_MAX_CAPTURED_OUTPUT_CHARS = 1024 * 1024;
const loadSupervisorLogRuntime = createLazyRuntimeModule(
() => import("./supervisor-log.runtime.js"),
);
function normalizeTimeoutDuration(value?: number): number | undefined {
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
return undefined;
}
return Math.max(1, Math.floor(value));
}
function clampCapturedOutputChars(value?: number): number {
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
return DEFAULT_MAX_CAPTURED_OUTPUT_CHARS;
}
return Math.max(256, Math.floor(value));
}
function appendCapturedOutput(
current: string,
chunk: string,
stream: "stdout" | "stderr",
maxChars: number,
) {
const next = current + chunk;
if (next.length <= maxChars) {
return next;
}
const marker = `[openclaw: captured ${stream} truncated to last ${maxChars} chars]\n`;
const tailChars = Math.max(0, maxChars - marker.length);
return `${marker}${sliceUtf16Safe(next, -tailChars)}`;
}
function isTimeoutReason(reason: TerminationReason) {
return reason === "overall-timeout" || reason === "no-output-timeout";
}
function resolveElapsedTimeoutReason(params: {
nowMs: number;
overallTimeoutDeadlineMs: number | null;
noOutputTimeoutDeadlineMs: number | null;
}): TerminationReason | null {
if (
params.overallTimeoutDeadlineMs !== null &&
params.nowMs >= params.overallTimeoutDeadlineMs &&
(params.noOutputTimeoutDeadlineMs === null ||
params.nowMs < params.noOutputTimeoutDeadlineMs ||
params.overallTimeoutDeadlineMs <= params.noOutputTimeoutDeadlineMs)
) {
return "overall-timeout";
}
return params.noOutputTimeoutDeadlineMs !== null &&
params.nowMs >= params.noOutputTimeoutDeadlineMs
? "no-output-timeout"
: null;
}
export function createProcessSupervisor(): ProcessSupervisor & {
shutdown: () => Promise<void>;
} {
// Retries share a run ID while an older command can still own descendants.
// Keep each admission until its own cleanup completes.
const ownedRuns = new Set<OwnedRun>();
const scopeCleanupOwners = new Map<string, Set<ScopeCleanupOwner>>();
const startingScopes = new Map<string, StartingScope>();
let shuttingDown = false;
let shutdownPromise: Promise<void> | null = null;
let cleanupFailure: { error: unknown } | undefined;
const cancel = (runId: string, reason: TerminationReason = "manual-cancel") => {
for (const current of ownedRuns) {
if (current.runId === runId) {
current.cancel(reason);
}
}
};
const cancelActiveScope = (scopeKey: string, reason: TerminationReason) => {
for (const current of ownedRuns) {
if (current.waitForExtinction && current.scopeKey === scopeKey) {
current.cancel(reason);
}
}
};
const cancelScope = (scopeKey: string, reason: TerminationReason = "manual-cancel") => {
if (!scopeKey.trim()) {
return;
}
for (const current of ownedRuns) {
if (current.scopeKey === scopeKey) {
current.cancel(reason);
}
}
};
const waitForRuns = async (
scopeKey: string | null,
ignoreStartupFailures = false,
): Promise<void> => {
let firstFailure: PromiseRejectedResult | undefined;
const observed = new Set<OwnedRun>();
while (true) {
const selected = Array.from(ownedRuns).filter(
(current) => !observed.has(current) && (scopeKey === null || current.scopeKey === scopeKey),
);
const starts = selected.flatMap((current) => (current.pending ? [current.pending] : []));
const owned = selected.flatMap((current) => {
if (!current.waitForExtinction) {
return [];
}
observed.add(current);
return [current.waitForExtinction()];
});
if (starts.length === 0 && owned.length === 0) {
if (firstFailure) {
throw firstFailure.reason;
}
return;
}
// Startup can become active while the snapshot settles; recheck admissions
// so shutdown cannot outrun an admitted command or retained descendants.
const results = await Promise.allSettled([...owned, ...starts]);
firstFailure ??= results
.slice(0, ignoreStartupFailures ? owned.length : undefined)
.find((result): result is PromiseRejectedResult => result.status === "rejected");
}
};
const acquireScopeCleanup = (
scopeKey: string,
options: { processTree: ProcessScopeCleanupPolicy },
): (() => Promise<void>) => {
const cleanupOwner: ScopeCleanupOwner = { processTree: options.processTree };
const owners = scopeCleanupOwners.get(scopeKey) ?? new Set<ScopeCleanupOwner>();
owners.add(cleanupOwner);
scopeCleanupOwners.set(scopeKey, owners);
let closing: Promise<void> | undefined;
return () =>
(closing ??= (async () => {
try {
cancelScope(scopeKey);
await waitForRuns(scopeKey);
} catch (error) {
cleanupOwner.failure ??= { error };
} finally {
owners.delete(cleanupOwner);
if (owners.size === 0) {
scopeCleanupOwners.delete(scopeKey);
}
}
if (cleanupOwner.failure) {
throw cleanupOwner.failure.error;
}
})());
};
const startRun = async (input: SpawnInput, owner: OwnedRun): Promise<ManagedRun> => {
const external = input.cleanupOwnership === "external";
const requireProcessTree = owner.cleanupOwners.some((scope) =>
requiresProcessTree(scope, external),
);
// A queued replacement must still own authority before stopping the surviving run.
if (!owner.terminationReason) {
input.assertCurrent?.();
input.beforeSpawn?.();
// Native PTY has no tree-extinction owner. Reject before spawning so exec's
// existing PTY-unavailable fallback can run once under the child anchor.
if (input.mode === "pty" && requireProcessTree) {
throw new Error("PTY is unavailable when execution requires process-tree cleanup");
}
}
const { runId, scopeKey } = owner;
const startedAtMs = Date.now();
const startingTerminationReason = owner.terminationReason;
const settleConstructionResult = (
reason: TerminationReason,
cleanup?: Promise<void>,
output?: { stdout: string; stderr: string; lastOutputAtMs: number },
): ManagedRun => {
const exit: RunExit = {
reason,
exitCode: null,
exitSignal: null,
durationMs: Date.now() - startedAtMs,
stdout: output?.stdout ?? "",
stderr: output?.stderr ?? "",
timedOut: isTimeoutReason(reason),
noOutputTimedOut: reason === "no-output-timeout",
};
return {
runId,
startedAtMs,
activity: Object.freeze({
resultSettled: true,
lastOutputAtMs: output?.lastOutputAtMs ?? startedAtMs,
}),
wait: async () => exit,
...(cleanup && { waitForExtinction: () => cleanup }),
cancel: () => undefined,
};
};
if (startingTerminationReason) {
// A replacement can be cancelled behind its scope fence. Never launch
// its command or terminate the surviving scope after that cancellation.
return settleConstructionResult(startingTerminationReason);
}
// Finish fallible argument preparation before affecting a surviving scope or arming cancellation.
if (input.mode !== "anchored-shell" && input.argv.length === 0) {
throw new Error("spawn argv cannot be empty");
}
const resolvedArgs = input.mode === "child" ? input.resolveArgs?.() : undefined;
if (owner.terminationReason) {
return settleConstructionResult(owner.terminationReason);
}
input.assertCurrent?.();
input.beforeSpawn?.();
if (input.replaceExistingScope && scopeKey) {
// Scope admission already waited for predecessor startups. Do not
// cancel this replacement or later runs reserved behind its fence.
cancelActiveScope(scopeKey, "manual-cancel");
}
let forcedReason: TerminationReason | null = owner.terminationReason ?? null;
let resultSettled = false;
let lastOutputAtMs = startedAtMs;
let cleanupSettled = false;
const captured = { stdout: "", stderr: "" };
// Forced settlement (kill-wait fallback, Windows forced close) resolves the
// result while inherited pipes stay open, and callers finalize their own
// output state from that terminal result. One fence closes every output path
// together: a late chunk reaches no listener, capture buffer, or output clock.
let outputDetached = false;
const detachOutput = () => {
outputDetached = true;
};
let forceKillTimer: NodeJS.Timeout | null = null;
let cancelRequested = false;
const captureOutput = input.captureOutput !== false;
const maxCapturedOutputChars = clampCapturedOutputChars(input.maxCapturedOutputChars);
const setForcedReason = (reason: TerminationReason) => {
if (forcedReason || resultSettled) {
return;
}
forcedReason = reason;
};
let cancelAdapter: ((reason: TerminationReason) => void) | null = null;
const constructionAbort = new AbortController();
const constructionAbortError = new Error("adapter construction aborted");
const constructionAbortPromise = new Promise<never>((_, reject) => {
const rejectConstruction = () => reject(constructionAbortError);
if (constructionAbort.signal.aborted) {
rejectConstruction();
} else {
constructionAbort.signal.addEventListener("abort", rejectConstruction, { once: true });
}
});
const requestCancel = (reason: TerminationReason) => {
setForcedReason(reason);
input.onCancel?.(reason);
cancelAdapter?.(reason);
// Any cancel must abort construction: the relay may already be spawned
// and waiting for ready, and a later deadline must not replace this reason.
if (!cancelAdapter) {
constructionAbort.abort();
}
};
owner.cancel = requestCancel;
const createDeadline = (reason: "overall-timeout" | "no-output-timeout", value?: number) => {
const durationMs = normalizeTimeoutDuration(value);
let deadlineMs: number | null = null;
let timer: NodeJS.Timeout | undefined;
// Re-arm bounded intervals: a long deadline must not overflow Node's timer cap.
const schedule = (remainingMs: number, deadline: number) => {
const intervalMs = resolveTimerTimeoutMs(remainingMs, 1);
timer = setTimeout(() => {
if (resultSettled) {
return;
}
const remaining = Math.min(remainingMs - intervalMs, deadline - performance.now());
if (remaining <= 0) {
requestCancel(reason);
} else {
schedule(remaining, deadline);
}
}, intervalMs);
};
return {
get deadlineMs() {
return deadlineMs;
},
reset: () => {
if (!durationMs || resultSettled) {
return;
}
clearTimeout(timer);
deadlineMs = performance.now() + durationMs;
schedule(durationMs, deadlineMs);
},
clear: () => clearTimeout(timer),
};
};
const overallDeadline = createDeadline("overall-timeout", input.timeoutMs);
const outputDeadline = createDeadline("no-output-timeout", input.noOutputTimeoutMs);
const touchOutput = () => {
lastOutputAtMs = Date.now();
outputDeadline.reset();
};
try {
// Reserve the join before construction: a timeout result does not release
// resources acquired later, or hide cleanup when readiness rejects after spawn.
const cleanup = createDeferredCore();
owner.waitForExtinction = () => cleanup.promise;
void cleanup.promise.catch(() => undefined);
let constructionCleanup: Promise<void> | undefined;
let ownedAdapter: SpawnProcessAdapter | undefined;
const onSpawnCleanup = (promise: Promise<void>) => {
constructionCleanup = promise;
void promise.catch(() => undefined);
};
overallDeadline.reset();
outputDeadline.reset();
const startupPromise =
input.mode === "pty"
? createPtyAdapter({
assertCurrent: input.assertCurrent,
beforeSpawn: input.beforeSpawn,
shell: expectDefined(input.argv[0], "spawn executable"),
args: input.argv.slice(1),
cwd: input.cwd,
env: input.env,
abortSignal: constructionAbort.signal,
onSpawnCleanup,
}).then((adapter) => ({ adapter, ready: Promise.resolve() }))
: input.mode === "anchored-shell"
? createChildAdapter({
assertCurrent: input.assertCurrent,
beforeSpawn: input.beforeSpawn,
anchoredShellCommand: input.command,
cwd: input.cwd,
env: input.env,
abortSignal: constructionAbort.signal,
onSpawnCleanup,
})
: createChildAdapter({
assertCurrent: input.assertCurrent,
beforeSpawn: input.beforeSpawn,
...(requireProcessTree && !external ? { ownProcessTree: true as const } : {}),
argv: resolvedArgs ? [...input.argv, ...resolvedArgs] : input.argv,
argv0: input.argv0,
cwd: input.cwd,
env: input.env,
exactEnv: input.exactEnv,
windowsVerbatimArguments: input.windowsVerbatimArguments,
input: input.input,
stdinMode: input.stdinMode,
secretInput: input.secretInput,
abortSignal: constructionAbort.signal,
onSpawnCleanup,
});
const extinctionPromise = startupPromise
.then(
async ({ adapter: started, ready }) => {
ownedAdapter = started;
if (external || !started.waitForExtinction) {
for (const scope of owner.cleanupOwners) {
if (requiresProcessTree(scope, external)) {
scope.failure ??= {
error: new Error(
"process cleanup cannot confirm owned execution-tree settlement",
),
};
}
}
}
if (constructionAbort.signal.aborted) {
started.kill("SIGKILL");
// Drain a late adapter's output without reopening the terminal result.
void started.wait().catch(() => undefined);
}
// Child close can precede a descendant's private-input consumption.
// Readiness failure is separate from the cleanup owner's outcome.
await Promise.allSettled([ready]);
await (constructionCleanup ?? started.waitForExtinction?.() ?? started.wait());
},
async () => {
await constructionCleanup;
},
)
.finally(() => {
cleanupSettled = true;
if (forceKillTimer) {
clearTimeout(forceKillTimer);
forceKillTimer = null;
}
if (resultSettled) {
ownedAdapter?.dispose();
}
});
void extinctionPromise.then(
() => {
ownedRuns.delete(owner);
cleanup.resolve();
},
(error: unknown) => {
recordScopeCleanupFailure(owner, error);
cleanupFailure ??= { error };
ownedRuns.delete(owner);
cleanup.reject(error);
},
);
const settleAbortedConstruction = (reason: TerminationReason) => {
resultSettled = true;
overallDeadline.clear();
outputDeadline.clear();
detachOutput();
if (cleanupSettled) {
ownedAdapter?.dispose();
}
return settleConstructionResult(reason, cleanup.promise, { ...captured, lastOutputAtMs });
};
let startup: Awaited<typeof startupPromise>;
try {
startup = await Promise.race([startupPromise, constructionAbortPromise]);
} catch (err) {
if (err !== constructionAbortError || !forcedReason) {
throw err;
}
return settleAbortedConstruction(forcedReason);
}
const adapter = startup.adapter;
const settleResult = () => {
resultSettled = true;
overallDeadline.clear();
outputDeadline.clear();
detachOutput();
if (cleanupSettled) {
adapter.dispose();
}
};
const withOutputFence =
<Chunk>(deliver?: (chunk: Chunk) => void, recordsOutput = true) =>
(chunk: Chunk) => {
if (outputDetached) {
return;
}
if (recordsOutput) {
touchOutput();
}
deliver?.(chunk);
};
const rawInput = input.mode === "child" ? input : undefined;
// Byte transports can flush decoded text at EOF without fresh activity.
// PTYs and Windows Job transports report only text.
for (const [stream, subscribe, onText, onRaw] of [
["stdout", adapter.onStdout, input.onStdout, rawInput?.onStdoutRaw],
["stderr", adapter.onStderr, input.onStderr, rawInput?.onStderrRaw],
] as const) {
subscribe(
withOutputFence((chunk: string) => {
if (captureOutput) {
captured[stream] = appendCapturedOutput(
captured[stream],
chunk,
stream,
maxCapturedOutputChars,
);
}
onText?.(chunk);
}, !adapter.supportsRawOutput),
withOutputFence(onRaw),
);
}
try {
await Promise.race([startup.ready, constructionAbortPromise]);
} catch (error) {
if (error === constructionAbortError && forcedReason) {
return settleAbortedConstruction(forcedReason);
}
settleResult();
throw error;
}
cancelAdapter = (reason: TerminationReason) => {
if (
cleanupSettled ||
(cancelRequested && (requireProcessTree || !(resultSettled && forceKillTimer)))
) {
return;
}
cancelRequested = true;
if (resultSettled && !requireProcessTree) {
if (forceKillTimer) {
clearTimeout(forceKillTimer);
forceKillTimer = null;
}
// Root completion closes its terminal record, not ownership of
// descendants still retained by the authoritative group or Job.
adapter.kill("SIGKILL");
return;
}
// Windows has no catchable SIGTERM equivalent: the adapter implements it
// with asynchronous taskkill, so waiting the cleanup grace only delays an
// already-expired deadline before the same forced tree termination.
if (
process.platform === "win32" &&
(reason === "overall-timeout" || reason === "no-output-timeout")
) {
adapter.kill("SIGKILL");
return;
}
adapter.kill("SIGTERM");
forceKillTimer = setTimeout(() => {
if (!cleanupSettled) {
adapter.kill("SIGKILL");
}
}, GRACEFUL_CANCEL_TIMEOUT_MS);
forceKillTimer.unref?.();
};
const waitPromise = (async (): Promise<RunExit> => {
const result = await adapter.wait();
const deadlineReason = resolveElapsedTimeoutReason({
nowMs: performance.now(),
overallTimeoutDeadlineMs: overallDeadline.deadlineMs,
noOutputTimeoutDeadlineMs: outputDeadline.deadlineMs,
});
const terminalReason = forcedReason ?? deadlineReason;
settleResult();
const reason: TerminationReason =
terminalReason ?? (result.signal != null ? ("signal" as const) : ("exit" as const));
const exit: RunExit = {
reason,
exitCode: result.code,
exitSignal: result.signal,
oomScoreWrapperSelected: adapter.oomScoreWrapperSelected === true,
durationMs: Date.now() - startedAtMs,
...captured,
timedOut: isTimeoutReason(reason),
noOutputTimedOut: terminalReason === "no-output-timeout",
};
return exit;
})().catch((err: unknown) => {
if (!resultSettled) {
settleResult();
}
throw err;
});
const managedRun: ManagedRun = {
activity: Object.freeze({
get deadlineAtMs() {
return overallDeadline.deadlineMs === null
? undefined
: Date.now() + overallDeadline.deadlineMs - performance.now();
},
get resultSettled() {
return resultSettled;
},
get lastOutputAtMs() {
return lastOutputAtMs;
},
}),
runId,
pid: adapter.pid,
startedAtMs,
stdin: adapter.stdin,
wait: async () => await waitPromise,
...(adapter.waitForExtinction && { waitForExtinction: () => cleanup.promise }),
cancel: (reason = "manual-cancel") => {
requestCancel(reason);
},
detachOutput,
};
if (forcedReason) {
managedRun.cancel(forcedReason);
}
return managedRun;
} catch (err) {
resultSettled = true;
overallDeadline.clear();
outputDeadline.clear();
detachOutput();
const { warnProcessSupervisorSpawnFailure } = await loadSupervisorLogRuntime();
warnProcessSupervisorSpawnFailure(`spawn failed: runId=${runId} reason=${String(err)}`);
throw err;
}
};
const spawn = (input: SpawnInput): Promise<ManagedRun> => {
if (shuttingDown) {
return Promise.reject(new Error("process supervisor is shut down"));
}
const scopeKey = normalizeOptionalString(input.scopeKey);
const runId = normalizeOptionalString(input.runId) ?? crypto.randomUUID();
const owner: OwnedRun = {
runId,
scopeKey,
cancel: (reason) => {
owner.terminationReason ??= reason;
input.onCancel?.(reason);
},
cleanupOwners: scopeKey ? [...(scopeCleanupOwners.get(scopeKey) ?? [])] : [],
};
// Reserve cancellation before either adapter startup or a replacement
// fence, so stopping a run cannot silently leave a late child alive.
ownedRuns.add(owner);
const starting = scopeKey
? (startingScopes.get(scopeKey) ?? { runs: new Set<Promise<ManagedRun>>() })
: undefined;
if (scopeKey && starting) {
startingScopes.set(scopeKey, starting);
}
// Ordinary runs start together, but replacements fence later arrivals so
// delayed cancellation cannot accidentally terminate a newer scoped run.
const previous = starting
? input.replaceExistingScope
? Array.from(starting.runs)
: starting.replacement
? [starting.replacement]
: []
: [];
const pending =
previous.length > 0
? Promise.allSettled(previous).then(() => startRun(input, owner))
: startRun(input, owner);
owner.pending = pending;
starting?.runs.add(pending);
if (starting && input.replaceExistingScope) {
starting.replacement = pending;
}
const clearPendingStart = () => {
delete owner.pending;
if (!owner.waitForExtinction) {
ownedRuns.delete(owner);
}
starting?.runs.delete(pending);
if (starting?.replacement === pending) {
delete starting.replacement;
}
if (scopeKey && starting?.runs.size === 0 && startingScopes.get(scopeKey) === starting) {
startingScopes.delete(scopeKey);
}
};
void pending.then(clearPendingStart, clearPendingStart);
return pending;
};
const shutdown = (): Promise<void> => {
// Publish the admission fence before cancellation can invoke owner callbacks.
shuttingDown = true;
return (shutdownPromise ??= Promise.resolve().then(async () => {
while (ownedRuns.size) {
for (const owner of ownedRuns) {
owner.cancel("manual-cancel");
}
// A failed startup owns no live process; only failed owner extinction
// must keep the process-wide supervisor fenced for operator recovery.
await waitForRuns(null, true);
}
if (cleanupFailure) {
throw cleanupFailure.error;
}
}));
};
return {
acquireScopeCleanup,
spawn,
cancel,
cancelScope,
shutdown,
};
}