Download src/process/supervisor/supervisor.ts from SaylorTwift/openclaw: direct link, hf CLI and curl.
- Browser
- Download file 26.5 kB
-
https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/process/supervisor/supervisor.ts
- Command line
-
hf download hf://SaylorTwift/openclaw/src/process/supervisor/supervisor.ts
-
curl -L -o supervisor.ts https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/process/supervisor/supervisor.ts
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, | |
| }; | |
| } | |