openclaw/extensions/acpx/src/process-reaper.ts

/**
 * ACPX process ownership checks and cleanup. The reaper only terminates
 * OpenClaw-owned wrapper trees after validating paths, packages, and lease ids.
 */
import { createRequire } from "node:module";
import path from "node:path";
import { runExec } from "openclaw/plugin-sdk/process-runtime";
import { splitCommandParts } from "./command-line.js";
import { resolveAcpxPluginRoot } from "./config.js";
import { OPENCLAW_ACPX_LEASE_ID_ARG, OPENCLAW_GATEWAY_INSTANCE_ID_ARG } from "./process-lease.js";

const requireFromHere = createRequire(import.meta.url);
const GENERATED_WRAPPER_BASENAMES = new Set([
  "codex-acp-wrapper.mjs",
  "claude-agent-acp-wrapper.mjs",
]);
const OPENCLAW_PLUGIN_DEPS_MARKER = "/plugin-runtime-deps/";
const OWNED_ACP_PACKAGE_NAMES = [
  "@zed-industries/codex-acp",
  "@zed-industries/codex-acp-darwin-arm64",
  "@zed-industries/codex-acp-darwin-x64",
  "@zed-industries/codex-acp-linux-arm64",
  "@zed-industries/codex-acp-linux-x64",
  "@zed-industries/codex-acp-win32-arm64",
  "@zed-industries/codex-acp-win32-x64",
  "@agentclientprotocol/claude-agent-acp",
  "acpx",
];
const ACP_PACKAGE_MARKERS = [
  ...OWNED_ACP_PACKAGE_NAMES.map((packageName) => `/node_modules/${packageName}/`),
  "/acpx/dist/",
];

/** Minimal process-table row used by ACPX cleanup. */
type AcpxProcessInfo = {
  pid: number;
  ppid: number;
  command: string;
};

/** Injectable process-listing and termination hooks for tests. */
export type AcpxProcessCleanupDeps = {
  listProcesses?: () => Promise<AcpxProcessInfo[]>;
  killProcess?: (pid: number, signal: NodeJS.Signals) => void;
  sleep?: (ms: number) => Promise<void>;
};

/** Result from cleaning up a single ACPX process tree. */
type AcpxProcessCleanupResult = {
  inspectedPids: number[];
  terminatedPids: number[];
  skippedReason?: "missing-root" | "not-openclaw-owned" | "unverified-root";
};

/** Result from startup orphan reaping. */
type AcpxStartupReapResult = {
  inspectedPids: number[];
  terminatedPids: number[];
  skippedReason?: "unsupported-platform" | "process-list-unavailable";
};

function normalizePathLike(value: string): string {
  return value.replaceAll("\\", "/");
}

function resolvePackageRoot(packageName: string): string | undefined {
  try {
    return normalizePathLike(path.dirname(requireFromHere.resolve(`${packageName}/package.json`)));
  } catch {
    return undefined;
  }
}

function resolveOpenClawInstallRoot(pluginRoot: string): string {
  if (
    path.basename(pluginRoot) === "acpx" &&
    path.basename(path.dirname(pluginRoot)) === "extensions"
  ) {
    const parent = path.dirname(path.dirname(pluginRoot));
    return path.basename(parent) === "dist" ? path.dirname(parent) : parent;
  }
  return path.resolve(pluginRoot, "..");
}

function resolveOwnedAcpPackageRootCandidates(packageName: string): string[] {
  const pluginRoot = resolveAcpxPluginRoot(import.meta.url);
  const openClawRoot = resolveOpenClawInstallRoot(pluginRoot);
  return [
    resolvePackageRoot(packageName),
    path.join(pluginRoot, "node_modules", packageName),
    path.join(openClawRoot, "node_modules", packageName),
  ].flatMap((root) => (root ? [normalizePathLike(root)] : []));
}

const OWNED_ACP_PACKAGE_ROOTS = Array.from(
  new Set(OWNED_ACP_PACKAGE_NAMES.flatMap(resolveOwnedAcpPackageRootCandidates)),
);

function commandBelongsToResolvedAcpPackage(command: string): boolean {
  return OWNED_ACP_PACKAGE_ROOTS.some((root) => command.includes(`${root}/`));
}

function commandMentionsGeneratedWrapper(command: string): boolean {
  return Array.from(GENERATED_WRAPPER_BASENAMES).some((basename) => command.includes(basename));
}

function commandWrapperBelongsToRoot(command: string, wrapperRoot: string | undefined): boolean {
  if (!wrapperRoot) {
    return true;
  }
  const normalizedCommand = normalizePathLike(command);
  const normalizedRoot = normalizePathLike(wrapperRoot).replace(/\/+$/, "");
  return Array.from(GENERATED_WRAPPER_BASENAMES).some((basename) =>
    normalizedCommand.includes(`${normalizedRoot}/${basename}`),
  );
}

/** Check whether a command references an OpenClaw-generated ACPX wrapper path. */
export function isOpenClawLeaseAwareAcpxProcessCommand(params: {
  command: string | undefined;
  wrapperRoot?: string;
}): boolean {
  const command = params.command?.trim();
  if (!command) {
    return false;
  }
  const normalized = normalizePathLike(command);
  return (
    commandMentionsGeneratedWrapper(normalized) &&
    commandWrapperBelongsToRoot(normalized, params.wrapperRoot)
  );
}

function commandsReferToSameRootCommand(liveCommand: string, storedCommand: string | undefined) {
  if (!storedCommand?.trim()) {
    return true;
  }
  return normalizePathLike(liveCommand).trim() === normalizePathLike(storedCommand).trim();
}

function commandOptionEquals(
  parts: string[],
  option: string,
  expected: string | undefined,
): boolean {
  if (!expected) {
    return true;
  }
  const index = parts.indexOf(option);
  return index >= 0 && parts[index + 1] === expected;
}

function liveCommandMatchesLeaseIdentity(params: {
  command: string | undefined;
  expectedLeaseId?: string;
  expectedGatewayInstanceId?: string;
}): boolean {
  if (!params.expectedLeaseId && !params.expectedGatewayInstanceId) {
    return true;
  }
  const parts = splitCommandParts(params.command ?? "");
  return (
    commandOptionEquals(parts, OPENCLAW_ACPX_LEASE_ID_ARG, params.expectedLeaseId) &&
    commandOptionEquals(parts, OPENCLAW_GATEWAY_INSTANCE_ID_ARG, params.expectedGatewayInstanceId)
  );
}

/** Check whether a command is owned by OpenClaw ACPX runtime packages or wrappers. */
function isOpenClawOwnedAcpxProcessCommand(params: {
  command: string | undefined;
  wrapperRoot?: string;
}): boolean {
  const command = params.command?.trim();
  if (!command) {
    return false;
  }
  const normalized = normalizePathLike(command);
  if (
    isOpenClawLeaseAwareAcpxProcessCommand({
      command: normalized,
      wrapperRoot: params.wrapperRoot,
    })
  ) {
    return true;
  }
  if (commandBelongsToResolvedAcpPackage(normalized)) {
    return true;
  }
  if (!normalized.includes(OPENCLAW_PLUGIN_DEPS_MARKER)) {
    return false;
  }
  return ACP_PACKAGE_MARKERS.some((marker) => normalized.includes(marker));
}

function parseProcessList(stdout: string): AcpxProcessInfo[] {
  const processes: AcpxProcessInfo[] = [];
  for (const line of stdout.split(/\r?\n/)) {
    const match = /^\s*(?<pid>\d+)\s+(?<ppid>\d+)\s+(?<command>.+?)\s*$/.exec(line);
    const pid = match?.groups?.pid;
    const ppid = match?.groups?.ppid;
    const command = match?.groups?.command;
    if (!pid || !ppid || !command) {
      continue;
    }
    processes.push({
      pid: Number.parseInt(pid, 10),
      ppid: Number.parseInt(ppid, 10),
      command,
    });
  }
  return processes;
}

/** List host processes in the compact shape needed by ACPX cleanup. */
async function listPlatformProcesses(): Promise<AcpxProcessInfo[]> {
  if (process.platform === "win32") {
    return [];
  }
  const { stdout } = await runExec("ps", ["-axo", "pid=,ppid=,command="], {
    logOutput: false,
    maxBuffer: 8 * 1024 * 1024,
  });
  return parseProcessList(stdout);
}

function collectProcessTree(processes: AcpxProcessInfo[], rootPid: number): AcpxProcessInfo[] {
  const childrenByParent = new Map<number, AcpxProcessInfo[]>();
  for (const processInfo of processes) {
    const children = childrenByParent.get(processInfo.ppid) ?? [];
    children.push(processInfo);
    childrenByParent.set(processInfo.ppid, children);
  }

  const byPid = new Map(processes.map((processInfo) => [processInfo.pid, processInfo]));
  const root = byPid.get(rootPid);
  const collected: AcpxProcessInfo[] = [];
  if (root) {
    collected.push(root);
  }

  const queue = [...(childrenByParent.get(rootPid) ?? [])];
  while (queue.length > 0) {
    const next = queue.shift();
    if (!next || collected.some((processInfo) => processInfo.pid === next.pid)) {
      continue;
    }
    collected.push(next);
    queue.push(...(childrenByParent.get(next.pid) ?? []));
  }

  return collected;
}

function uniquePids(processes: AcpxProcessInfo[]): number[] {
  return Array.from(
    new Set(
      processes
        .map((processInfo) => processInfo.pid)
        .filter((pid) => Number.isInteger(pid) && pid > 0 && pid !== process.pid),
    ),
  );
}

function isProcessAlive(pid: number): boolean {
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

async function terminatePids(
  pids: number[],
  deps: AcpxProcessCleanupDeps | undefined,
): Promise<number[]> {
  const killProcess = deps?.killProcess ?? ((pid, signal) => process.kill(pid, signal));
  const sleep =
    deps?.sleep ??
    ((ms) =>
      new Promise<void>((resolve) => {
        setTimeout(resolve, ms);
      }));
  const terminated: number[] = [];

  for (const pid of pids) {
    try {
      killProcess(pid, "SIGTERM");
      terminated.push(pid);
    } catch {
      // The process may already be gone.
    }
  }
  if (terminated.length === 0) {
    return terminated;
  }
  await sleep(750);
  for (const pid of terminated) {
    if (deps?.killProcess || isProcessAlive(pid)) {
      try {
        killProcess(pid, "SIGKILL");
      } catch {
        // Best-effort cleanup only.
      }
    }
  }
  return terminated;
}

/** Terminate one validated OpenClaw-owned ACPX wrapper process tree. */
export async function cleanupOpenClawOwnedAcpxProcessTree(params: {
  rootPid?: number;
  rootCommand?: string;
  expectedLeaseId?: string;
  expectedGatewayInstanceId?: string;
  wrapperRoot?: string;
  deps?: AcpxProcessCleanupDeps;
}): Promise<AcpxProcessCleanupResult> {
  const rootPid = params.rootPid;
  if (!rootPid || rootPid <= 0 || rootPid === process.pid) {
    return { inspectedPids: [], terminatedPids: [], skippedReason: "missing-root" };
  }

  let processes: AcpxProcessInfo[];
  try {
    processes = await (params.deps?.listProcesses ?? listPlatformProcesses)();
  } catch {
    processes = [];
  }

  const listedTree = collectProcessTree(processes, rootPid);
  // Session-store PIDs are stale data. If the live process table cannot prove
  // that this PID still belongs to an OpenClaw-owned wrapper, fail closed to
  // avoid killing an unrelated process after PID reuse.
  if (listedTree.length === 0) {
    return { inspectedPids: [], terminatedPids: [], skippedReason: "unverified-root" };
  }
  const rootCommand = listedTree[0]?.command ?? params.rootCommand;
  const liveCommandWasGeneratedWrapper = commandMentionsGeneratedWrapper(
    normalizePathLike(rootCommand ?? ""),
  );
  const storedCommandWasGeneratedWrapper = commandMentionsGeneratedWrapper(
    normalizePathLike(params.rootCommand ?? ""),
  );
  if (!liveCommandWasGeneratedWrapper && storedCommandWasGeneratedWrapper) {
    return {
      inspectedPids: listedTree.map((processInfo) => processInfo.pid),
      terminatedPids: [],
      skippedReason: "not-openclaw-owned",
    };
  }
  if (
    !liveCommandWasGeneratedWrapper &&
    !commandsReferToSameRootCommand(rootCommand ?? "", params.rootCommand)
  ) {
    return {
      inspectedPids: listedTree.map((processInfo) => processInfo.pid),
      terminatedPids: [],
      skippedReason: "not-openclaw-owned",
    };
  }
  if (
    !isOpenClawOwnedAcpxProcessCommand({
      command: rootCommand,
      wrapperRoot: params.wrapperRoot,
    })
  ) {
    return {
      inspectedPids: listedTree.map((processInfo) => processInfo.pid),
      terminatedPids: [],
      skippedReason: "not-openclaw-owned",
    };
  }
  if (
    !liveCommandMatchesLeaseIdentity({
      command: rootCommand,
      expectedLeaseId: params.expectedLeaseId,
      expectedGatewayInstanceId: params.expectedGatewayInstanceId,
    })
  ) {
    return {
      inspectedPids: listedTree.map((processInfo) => processInfo.pid),
      terminatedPids: [],
      skippedReason: "not-openclaw-owned",
    };
  }

  const pids = uniquePids(listedTree.toReversed());
  return {
    inspectedPids: uniquePids(listedTree),
    terminatedPids: await terminatePids(pids, params.deps),
  };
}

/** Reap orphaned OpenClaw-owned ACPX wrapper trees during runtime startup. */
export async function reapStaleOpenClawOwnedAcpxOrphans(params: {
  wrapperRoot: string;
  deps?: AcpxProcessCleanupDeps;
}): Promise<AcpxStartupReapResult> {
  if (process.platform === "win32") {
    return { inspectedPids: [], terminatedPids: [], skippedReason: "unsupported-platform" };
  }

  let processes: AcpxProcessInfo[];
  try {
    processes = await (params.deps?.listProcesses ?? listPlatformProcesses)();
  } catch {
    return { inspectedPids: [], terminatedPids: [], skippedReason: "process-list-unavailable" };
  }

  const orphans = processes.filter(
    (processInfo) =>
      processInfo.ppid === 1 &&
      isOpenClawOwnedAcpxProcessCommand({
        command: processInfo.command,
        wrapperRoot: params.wrapperRoot,
      }),
  );
  // Startup reaping starts from currently visible orphan roots and then expands
  // each tree, so adapter grandchildren do not survive as fresh orphans after
  // the wrapper root exits.
  const orphanTrees = orphans.map((orphan) => collectProcessTree(processes, orphan.pid));
  const inspectedPids = uniquePids(orphanTrees.flat());
  const pids = uniquePids(orphanTrees.flatMap((tree) => tree.toReversed()));
  return {
    inspectedPids,
    terminatedPids: await terminatePids(pids, params.deps),
  };
}