import process from "node:process"; import { expectDefined } from "@openclaw/normalization-core"; import { toErrorObject } from "@openclaw/normalization-core/error-coercion"; import { resolveTimerTimeoutMs } from "@openclaw/normalization-core/number-coercion"; import { isPromiseLike } from "@openclaw/normalization-core/promise-like"; import { decodeWindowsOutputBuffer, resolveWindowsConsoleEncoding, } from "../infra/windows-encoding.js"; import { releaseChildProcessOutputAfterExit } from "./child-process.js"; import { appendCapturedOutput, appendPreservedOutputLines, createCapturedOutputBuffers, finalizeCapturedOutput, flushPreservedOutputLine, MAX_PRESERVED_PENDING_LINE_BYTES, resolveMaxOutputBytes, resolveOutputCapture, shouldTerminateOnOutputError, shouldTerminateOnOutputLimit, type CapturedOutputBuffers, type CommandOutputCaptureMode, type CommandOutputCaptureOption, type CommandOutputErrorOption, type CommandOutputLimitOption, type CommandOutputStream, type PreserveOutputLine, } from "./exec-output.js"; import { createSanitizedCommandError, isPlainCommandExitFailure, isPlainCommandSignalFailure, resolveProcessExitCode, TIMEOUT_EXIT_CODE, type SpawnResult, } from "./exec-result.js"; import { COMMAND_PROCESS_TREE_KILL_GRACE_MS, resolveCommandProcessSignal, spawnCommandWithInvocation, } from "./exec-spawn.js"; import { createCommandTerminationController } from "./exec-termination.js"; const WINDOWS_CLOSE_STATE_SETTLE_TIMEOUT_MS = 250; const WINDOWS_CLOSE_STATE_POLL_MS = 10; type CommandTerminationReason = SpawnResult["termination"] | "output-limit"; export type CommandOptions = { timeoutMs?: number; cwd?: string; input?: string | Uint8Array; /** Synchronous admission with the spawned PID and argv, before input is released. */ beforeInput?: (pid: number, argv?: readonly string[]) => void; baseEnv?: NodeJS.ProcessEnv; env?: NodeJS.ProcessEnv; windowsVerbatimArguments?: boolean; noOutputTimeoutMs?: number; signal?: AbortSignal; maxOutputBytes?: number | { stdout?: number; stderr?: number }; maxCombinedOutputBytes?: number; outputCapture?: CommandOutputCaptureOption; /** Observe raw output without owning child lifecycle. Return false to stop the command. */ onOutputChunk?: (chunk: Buffer, stream: CommandOutputStream) => boolean | void; /** Accept a successful exit when only the selected diagnostic output stream failed. */ tolerateOutputError?: { stdout?: boolean; stderr?: boolean }; /** Terminate when the selected output stream emits an error. */ terminateOnOutputError?: CommandOutputErrorOption; terminateOnOutputLimit?: CommandOutputLimitOption; maxPreservedOutputLines?: number; preserveOutputLine?: PreserveOutputLine; killProcessTree?: boolean; /** Join owned descendants even after a successful root exits. */ requireProcessTreeExtinction?: boolean; /** Initial signal for direct-child and graceful process-group cancellation. */ killSignal?: NodeJS.Signals | number; /** Grace between graceful termination and the force-kill fallback. */ killGraceMs?: number; }; export async function runCommandWithTimeout( argv: string[], optionsOrTimeout: number | CommandOptions, ): Promise { return await runCommandWithOutputEncoding(argv, optionsOrTimeout, false); } /** Run a command whose stdout and stderr are defined to be UTF-8 on every platform. */ export async function runUtf8CommandWithTimeout( argv: string[], optionsOrTimeout: number | CommandOptions, ): Promise { return await runCommandWithOutputEncoding(argv, optionsOrTimeout, true); } export type BufferSpawnResult = Omit & { stdout: Buffer; stderr: Buffer; windowsEncoding: string | null; }; /** Preserve the ordinary process lifecycle while deferring decoding to its consumer. */ export async function runCommandBuffersWithTimeout( argv: string[], optionsOrTimeout: number | CommandOptions, ): Promise { return await runCommandWithOutputEncoding(argv, optionsOrTimeout, false, true); } async function runCommandWithOutputEncoding( argv: string[], optionsOrTimeout: number | CommandOptions, forceUtf8: boolean, ): Promise; async function runCommandWithOutputEncoding( argv: string[], optionsOrTimeout: number | CommandOptions, forceUtf8: boolean, raw: true, ): Promise; async function runCommandWithOutputEncoding( argv: string[], optionsOrTimeout: number | CommandOptions, forceUtf8: boolean, raw = false, ): Promise { const options: CommandOptions = typeof optionsOrTimeout === "number" ? { timeoutMs: optionsOrTimeout } : optionsOrTimeout; const { timeoutMs, cwd, input, baseEnv, env, noOutputTimeoutMs, killProcessTree, killSignal, killGraceMs, } = options; const signal = resolveCommandProcessSignal(options.signal); const resolvedTimeoutMs = typeof timeoutMs === "number" ? resolveTimerTimeoutMs(timeoutMs, 1) : undefined; if (options.requireProcessTreeExtinction && !killProcessTree) { throw new Error("Process-tree extinction requires process-tree ownership"); } const hasInput = input !== undefined; if (options.beforeInput && !hasInput) { throw new Error("Child input admission requires explicit input"); } const resolvedKillGraceMs = resolveTimerTimeoutMs( killGraceMs, COMMAND_PROCESS_TREE_KILL_GRACE_MS, 0, ); if (signal?.aborted) { const interrupted = { code: null, signal: null, killed: false, termination: "signal" as const, cleanup: "normal" as const, noOutputTimedOut: false, }; return raw ? { ...interrupted, stdout: Buffer.alloc(0), stderr: Buffer.alloc(0), windowsEncoding: null } : { ...interrupted, stdout: "", stderr: "" }; } const stdoutCapture = createCapturedOutputBuffers(); const stderrCapture = createCapturedOutputBuffers(); const maxStdoutBytes = resolveMaxOutputBytes(options.maxOutputBytes, "stdout"); const maxStderrBytes = resolveMaxOutputBytes(options.maxOutputBytes, "stderr"); const maxCombinedOutputBytes = typeof options.maxCombinedOutputBytes === "number" && Number.isFinite(options.maxCombinedOutputBytes) && options.maxCombinedOutputBytes > 0 ? Math.max(1, Math.floor(options.maxCombinedOutputBytes)) : undefined; const stdoutCaptureMode = resolveOutputCapture(options.outputCapture, "stdout"); const stderrCaptureMode = resolveOutputCapture(options.outputCapture, "stderr"); if (maxCombinedOutputBytes !== undefined && stdoutCaptureMode !== stderrCaptureMode) { throw new Error("maxCombinedOutputBytes requires matching stdout and stderr capture modes"); } const usesCombinedTailCapture = maxCombinedOutputBytes !== undefined && stdoutCaptureMode === "tail" && stderrCaptureMode === "tail"; const maxPreservedPendingLineBytes = Math.min( Math.max(maxStdoutBytes, maxStderrBytes), MAX_PRESERVED_PENDING_LINE_BYTES, ); const maxPreservedOutputLines = Math.max(0, Math.floor(options.maxPreservedOutputLines ?? 16)); const windowsEncoding = forceUtf8 ? null : resolveWindowsConsoleEncoding(); const cancelController = new AbortController(); let termination: CommandTerminationReason | undefined; let childExitState: { code: number | null; signal: NodeJS.Signals | null } | undefined; let commandSettled = false; let combinedOutputBytes = 0; let combinedCapturedBytes = 0; const outputBytesByStream = { stdout: 0, stderr: 0 }; const combinedCapturedBytesByStream = { stdout: 0, stderr: 0 }; const combinedTailChunks: Array<{ stream: CommandOutputStream; buffer: Buffer }> = []; let noOutputTimer: NodeJS.Timeout | undefined; let outputObserverError: unknown; let outputErrorStream: CommandOutputStream | undefined; let terminatingOutputError: Error | undefined; const { child, invocation } = spawnCommandWithInvocation(argv, { buffer: false, cancelSignal: cancelController.signal, inheritScopeCancellation: false, cwd, detached: Boolean(killProcessTree && process.platform !== "win32"), encoding: "buffer", baseEnv, env, forceKillAfterDelay: resolvedKillGraceMs, killSignal, ...(hasInput && !options.beforeInput ? { input } : {}), reject: false, stdio: [hasInput ? "pipe" : "inherit", "pipe", "pipe"], stripFinalNewline: false, windowsVerbatimArguments: options.windowsVerbatimArguments, }); const nodeChild = child.nodeChildProcess; const ownsExitedProcessTree = Boolean(killProcessTree && process.platform !== "win32"); const shouldTrackOutputTimeout = typeof noOutputTimeoutMs === "number" && Number.isFinite(noOutputTimeoutMs) && noOutputTimeoutMs > 0; const resolvedNoOutputTimeoutMs = shouldTrackOutputTimeout ? resolveTimerTimeoutMs(noOutputTimeoutMs, 1) : undefined; const ownsOutputDeadline = ownsExitedProcessTree && (resolvedTimeoutMs !== undefined || resolvedNoOutputTimeoutMs !== undefined); let releaseOutput: (() => void) | undefined; const terminationController = createCommandTerminationController({ child: nodeChild, cancelController, baseEnv, env, processTree: killProcessTree ? { mode: "graceful" } : undefined, isChildExited: () => childExitState !== undefined, isCommandSettled: () => commandSettled, killGraceMs: resolvedKillGraceMs, killSignal, }); nodeChild.once("exit", (code, signalValue) => { childExitState = { code, signal: signalValue }; // Successful tree output belongs to its command deadline, not the diagnostic // idle cutoff. Failed, terminated, and unowned output still gets a bounded drain. if (!ownsOutputDeadline || code !== 0 || termination) { releaseOutput = releaseChildProcessOutputAfterExit(nodeChild); } // An inner timeout can become an ordinary failed exit while its descendants survive. // Retain the existing tree owner through its drain without changing that exit result. if (killProcessTree && !termination && (code !== 0 || options.requireProcessTreeExtinction)) { terminationController.terminate(); } }); const clearNoOutputTimer = () => { if (noOutputTimer) { clearTimeout(noOutputTimer); noOutputTimer = undefined; } }; const cancel = (reason: Exclude) => { // Failed roots already own a drain; later deadlines must preserve their exit result. // Successful POSIX roots retain deadline ownership of inherited descendants. // Output caps remain meaningful for bytes drained after either exit. if ( termination || commandSettled || (childExitState && reason !== "output-limit" && (!ownsExitedProcessTree || childExitState.code !== 0)) ) { return; } termination = reason; if (childExitState) { // An escaped pipe holder can survive group termination; bound its final drain. releaseOutput ??= releaseChildProcessOutputAfterExit(nodeChild); } const abortDeferred = terminationController.terminate(); if (!abortDeferred) { cancelController.abort(); } }; const armNoOutputTimer = () => { if ( resolvedNoOutputTimeoutMs === undefined || commandSettled || termination || (childExitState && !ownsExitedProcessTree) ) { return; } clearNoOutputTimer(); noOutputTimer = setTimeout(() => cancel("no-output-timeout"), resolvedNoOutputTimeoutMs); }; const timeoutTimer = resolvedTimeoutMs === undefined ? undefined : setTimeout(() => cancel("timeout"), resolvedTimeoutMs); const onAbort = () => cancel("signal"); signal?.addEventListener("abort", onAbort, { once: true }); armNoOutputTimer(); const captureOutput = ( capture: CapturedOutputBuffers, chunk: Buffer | string, maxBytes: number, stream: CommandOutputStream, captureMode: CommandOutputCaptureMode, ) => { const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); outputBytesByStream[stream] += buffer.byteLength; const streamLimitExceeded = outputBytesByStream[stream] > maxBytes; if (maxCombinedOutputBytes === undefined) { appendCapturedOutput(capture, buffer, maxBytes, captureMode); if ( streamLimitExceeded && shouldTerminateOnOutputLimit(options.terminateOnOutputLimit, stream) ) { cancel("output-limit"); } return; } const combinedBytesBeforeChunk = combinedOutputBytes; combinedOutputBytes += buffer.byteLength; const combinedLimitExceeded = combinedOutputBytes > maxCombinedOutputBytes; if (usesCombinedTailCapture) { combinedTailChunks.push({ stream, buffer }); combinedCapturedBytes += buffer.byteLength; combinedCapturedBytesByStream[stream] += buffer.byteLength; const removeCapturedBytes = (index: number, requestedBytes: number) => { const entry = expectDefined(combinedTailChunks[index], "combined tail chunk"); const removedBytes = Math.min(requestedBytes, entry.buffer.byteLength); if (removedBytes === entry.buffer.byteLength) { combinedTailChunks.splice(index, 1); } else { entry.buffer = Buffer.from(entry.buffer.subarray(removedBytes)); } combinedCapturedBytes -= removedBytes; combinedCapturedBytesByStream[entry.stream] -= removedBytes; (entry.stream === "stdout" ? stdoutCapture : stderrCapture).truncatedBytes += removedBytes; }; while (combinedCapturedBytesByStream[stream] > maxBytes) { const index = combinedTailChunks.findIndex((entry) => entry.stream === stream); if (index < 0) { break; } removeCapturedBytes(index, combinedCapturedBytesByStream[stream] - maxBytes); } let combinedOverflow = combinedCapturedBytes - maxCombinedOutputBytes; while (combinedOverflow > 0) { removeCapturedBytes(0, combinedOverflow); combinedOverflow = combinedCapturedBytes - maxCombinedOutputBytes; } } else { const remaining = Math.max(0, maxCombinedOutputBytes - combinedBytesBeforeChunk); const maxCaptureBytes = Math.min(maxBytes, capture.bytes + remaining); appendCapturedOutput(capture, buffer, maxCaptureBytes, captureMode); } if ( (combinedLimitExceeded && shouldTerminateOnOutputLimit(options.terminateOnOutputLimit, "combined")) || (streamLimitExceeded && shouldTerminateOnOutputLimit(options.terminateOnOutputLimit, stream)) ) { cancel("output-limit"); } }; const observeOutputChunk = (chunk: Buffer | string, stream: CommandOutputStream): Buffer => { const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); if (termination || !options.onOutputChunk) { return buffer; } try { if (options.onOutputChunk(buffer, stream) === false) { cancel("output-limit"); } } catch (error) { outputObserverError = error; cancel("output-limit"); } return buffer; }; const onOutputError = (error: unknown, stream: CommandOutputStream) => { outputErrorStream ??= stream; if ( termination || options.tolerateOutputError?.[stream] === true || !shouldTerminateOnOutputError(options.terminateOnOutputError, stream) ) { return; } terminatingOutputError = toErrorObject(error, `Command ${stream} stream failed`); Object.assign(terminatingOutputError, { outputErrorStream: stream }); cancel("signal"); }; child.stdout?.once("error", (error) => onOutputError(error, "stdout")); child.stderr?.once("error", (error) => onOutputError(error, "stderr")); child.stdout?.on("data", (chunk) => { const buffer = observeOutputChunk(chunk, "stdout"); appendPreservedOutputLines({ capture: stdoutCapture, chunk: buffer, stream: "stdout", preserveOutputLine: options.preserveOutputLine, maxPreservedOutputLines, maxPendingLineBytes: maxPreservedPendingLineBytes, }); captureOutput(stdoutCapture, buffer, maxStdoutBytes, "stdout", stdoutCaptureMode); armNoOutputTimer(); }); child.stderr?.on("data", (chunk) => { const buffer = observeOutputChunk(chunk, "stderr"); appendPreservedOutputLines({ capture: stderrCapture, chunk: buffer, stream: "stderr", preserveOutputLine: options.preserveOutputLine, maxPreservedOutputLines, maxPendingLineBytes: maxPreservedPendingLineBytes, }); captureOutput(stderrCapture, buffer, maxStderrBytes, "stderr", stderrCaptureMode); armNoOutputTimer(); }); let inputAdmissionError: Error | undefined; if (options.beforeInput) { nodeChild.stdin?.once("error", (cause) => { inputAdmissionError ??= toErrorObject(cause, "Command input failed"); cancel("signal"); }); try { if (nodeChild.pid === undefined || !nodeChild.stdin) { throw new Error("Child input admission has no spawned process"); } const admitted: unknown = options.beforeInput(nodeChild.pid, nodeChild.spawnargs); if (admitted !== undefined) { if (isPromiseLike(admitted)) { void Promise.resolve(admitted).catch(() => undefined); } throw new TypeError("Child input admission must complete synchronously"); } nodeChild.stdin.end(input); } catch (cause) { inputAdmissionError = toErrorObject(cause, "Child input admission failed"); nodeChild.stdin?.destroy(); cancel("signal"); } } const result = await child.finally(() => { commandSettled = true; if (timeoutTimer) { clearTimeout(timeoutTimer); } clearNoOutputTimer(); signal?.removeEventListener("abort", onAbort); releaseOutput?.(); }); let cleanup = await terminationController.settle(); const resolvedSignal = result.signal ?? childExitState?.signal ?? nodeChild.signalCode ?? null; if (cleanup !== "forced" && resolvedSignal) { cleanup = "uncertain"; } if (inputAdmissionError) { throw Object.assign(inputAdmissionError, { cleanup }); } if (terminatingOutputError) { throw Object.assign(terminatingOutputError, { cleanup }); } if (outputObserverError !== undefined) { throw Object.assign(toErrorObject(outputObserverError, "Command output observer failed"), { cleanup, }); } // Patched Node can report null/null after a cmd.exe shim exits. Execa turns // that into a cause-less failure; preserve the shim fallback only post-spawn. const isCauseLessWindowsShimResult = !termination && invocation.usesWindowsExitCodeShim && typeof nodeChild.pid === "number" && result.code === undefined && result.cause === undefined && !result.timedOut && !result.isCanceled && !result.isMaxBuffer && !result.isTerminated; if (isCauseLessWindowsShimResult) { // A patched Windows runtime can populate exitCode shortly after close. // Settle that state before the shim fallback can infer a clean exit. for ( let elapsedMs = 0; elapsedMs < WINDOWS_CLOSE_STATE_SETTLE_TIMEOUT_MS; elapsedMs += WINDOWS_CLOSE_STATE_POLL_MS ) { if ( childExitState?.code != null || childExitState?.signal != null || nodeChild.exitCode != null || nodeChild.signalCode != null ) { break; } await new Promise((resolve) => { setTimeout(resolve, WINDOWS_CLOSE_STATE_POLL_MS); }); } } if ( result.failed && !termination && !isPlainCommandExitFailure(result) && !isPlainCommandSignalFailure(result) && !isCauseLessWindowsShimResult && !( result.exitCode === 0 && outputErrorStream !== undefined && options.tolerateOutputError?.[outputErrorStream] === true ) ) { const error = createSanitizedCommandError(result); Object.assign(error, { cleanup: typeof nodeChild.pid === "number" ? cleanup === "normal" ? "uncertain" : cleanup : "normal", }); if (outputErrorStream) { Object.assign(error, { outputErrorStream }); } throw error; } const resolvedCode = resolveProcessExitCode({ explicitCode: result.exitCode ?? childExitState?.code, childExitCode: nodeChild.exitCode, resolvedSignal, usesWindowsExitCodeShim: invocation.usesWindowsExitCodeShim, timedOut: termination === "timeout", noOutputTimedOut: termination === "no-output-timeout", killIssuedByTimeout: termination === "timeout" || termination === "no-output-timeout", killIssuedByAbort: termination === "signal" || termination === "output-limit", }); termination ??= resolvedSignal != null || result.isTerminated ? "signal" : "exit"; const normalizedCode = termination === "timeout" || termination === "no-output-timeout" ? resolvedCode == null || resolvedCode === 0 ? TIMEOUT_EXIT_CODE : resolvedCode : resolvedCode; flushPreservedOutputLine({ capture: stdoutCapture, stream: "stdout", preserveOutputLine: options.preserveOutputLine, maxPreservedOutputLines, maxPendingLineBytes: maxPreservedPendingLineBytes, }); flushPreservedOutputLine({ capture: stderrCapture, stream: "stderr", preserveOutputLine: options.preserveOutputLine, maxPreservedOutputLines, maxPendingLineBytes: maxPreservedPendingLineBytes, }); if (usesCombinedTailCapture) { for (const entry of combinedTailChunks) { const capture = entry.stream === "stdout" ? stdoutCapture : stderrCapture; capture.chunks.push(entry.buffer); capture.bytes += entry.buffer.byteLength; } } const stdout = finalizeCapturedOutput(stdoutCapture, stdoutCaptureMode, forceUtf8); const stderr = finalizeCapturedOutput(stderrCapture, stderrCaptureMode, forceUtf8); const settled = { pid: nodeChild.pid, stdoutTruncatedBytes: stdoutCapture.truncatedBytes || undefined, stderrTruncatedBytes: stderrCapture.truncatedBytes || undefined, preservedStdoutLines: stdoutCapture.preservedLines.length > 0 ? stdoutCapture.preservedLines : undefined, preservedStderrLines: stderrCapture.preservedLines.length > 0 ? stderrCapture.preservedLines : undefined, code: normalizedCode, signal: resolvedSignal, killed: nodeChild.killed, cleanup, termination: termination === "output-limit" ? ("signal" as const) : termination, noOutputTimedOut: termination === "no-output-timeout", outputLimitExceeded: termination === "output-limit" || undefined, ...(outputErrorStream ? { outputErrorStream } : {}), }; return raw ? { ...settled, stdout, stderr, windowsEncoding } : { ...settled, stdout: forceUtf8 ? stdout.toString("utf8") : decodeWindowsOutputBuffer({ buffer: stdout, windowsEncoding }), stderr: forceUtf8 ? stderr.toString("utf8") : decodeWindowsOutputBuffer({ buffer: stderr, windowsEncoding }), }; }