src/process/exec-runner.ts

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<SpawnResult> {
  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<SpawnResult> {
  return await runCommandWithOutputEncoding(argv, optionsOrTimeout, true);
}

export type BufferSpawnResult = Omit<SpawnResult, "stdout" | "stderr"> & {
  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<BufferSpawnResult> {
  return await runCommandWithOutputEncoding(argv, optionsOrTimeout, false, true);
}

async function runCommandWithOutputEncoding(
  argv: string[],
  optionsOrTimeout: number | CommandOptions,
  forceUtf8: boolean,
): Promise<SpawnResult>;
async function runCommandWithOutputEncoding(
  argv: string[],
  optionsOrTimeout: number | CommandOptions,
  forceUtf8: boolean,
  raw: true,
): Promise<BufferSpawnResult>;
async function runCommandWithOutputEncoding(
  argv: string[],
  optionsOrTimeout: number | CommandOptions,
  forceUtf8: boolean,
  raw = false,
): Promise<SpawnResult | BufferSpawnResult> {
  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<CommandTerminationReason, "exit">) => {
    // 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<void>((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 }),
      };
}