AgentHub/server/lib/ai-tool-jobs.js

/**
 * @file The one-at-a-time install/update runner behind Settings → AI Tools.
 * Every command comes from a catalog file (never a request) and runs as an
 * argv with stdin closed and the login shell's PATH first: `npm install -g
 * <pkg>@latest`, a tool's own updater, or a vendor script the server
 * downloads (https, same site as the tool's homepage, ≤ 1 MB) and runs with
 * `bash <file>`. The job lives in memory — a child of the server, so it dies
 * with it — and it reports its end only AFTER the PATH is re-read and the tool
 * re-detected, so the client's first look at a finished job is already true.
 * Each command runs in its own process group: a timeout kills the whole group
 * (SIGTERM, then SIGKILL), and a job whose child has exited ends even when a
 * leftover grandchild still holds its output pipes.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const crypto = require("node:crypto");
const { spawn } = require("node:child_process");
const aiTools = require("./ai-tools");
const versions = require("./ai-tool-versions");
const { hydrateShellPath } = require("./shell-path");

const LOG_CAP = 256 * 1024;
let jobTimeoutMs = 10 * 60 * 1000;
const SCRIPT_MAX = 1024 * 1024;
const EACCES_HINT =
  "\n[agenthub] npm could not write its global directory (EACCES): this Node is installed " +
  "system-wide. Use a user-level Node (nvm, fnm) or install the tool by hand.\n";

const EXIT_GRACE_MS = 500; // after 'exit', how long to wait for the pipes to close
const KILL_GRACE_MS = 5000; // SIGTERM → SIGKILL

let spawnImpl = spawn;
let hydrate = hydrateShellPath;
let job = null;
let starting = false;
let seq = 0;

function makeErr(code, message) {
  const err = new Error(message);
  err.code = code;
  return err;
}

const npmArgv = (pkg) => {
  const npm = versions.npmBin();
  if (!npm) throw makeErr("ENONPM", "npm was not found on the login shell's PATH");
  return [npm, "install", "-g", `${pkg}@latest`];
};

/** The steps one action on one tool would run — `{argv}` or `{script: url}`. */
async function stepsFor(id, action) {
  const tool = aiTools.get(id);
  if (!tool) throw makeErr("ENOTOOL", `unknown AI tool: ${id}`);
  const d = aiTools.detect(id);
  if (action === "install") {
    if (!tool.install) throw makeErr("ENOINSTALL", `${tool.label} has no install command`);
    if (d.installed) throw makeErr("EINSTALLED", `${tool.label} is already installed`);
    return tool.install.kind === "npm"
      ? [{ argv: npmArgv(tool.install.package) }]
      : [{ script: tool.install.url }];
  }
  if (!tool.update) throw makeErr("ENOUPDATE", `${tool.label} has no update command`);
  if (!d.installed) throw makeErr("ENOTINSTALLED", `${tool.label} is not installed`);
  if (tool.update.kind === "self") return [{ argv: [d.path, ...tool.update.args] }];
  if (tool.update.kind === "script") return [{ script: tool.install.url }];
  if (!(await versions.npmManaged(id))) {
    throw makeErr(
      "ENOTNPM",
      `${tool.label} was installed outside npm; update it the way it was installed`
    );
  }
  return [{ argv: npmArgv(tool.install.package) }];
}

function append(j, text) {
  j.log += text;
  if (j.log.length > LOG_CAP) j.log = j.log.slice(-LOG_CAP);
}

/** Download a vendor script to a 0600 temp file; the final URL must stay on its site. */
async function download(url) {
  const res = await fetch(url, { signal: AbortSignal.timeout(15_000), redirect: "follow" });
  const final = new URL(res.url || url);
  const site = aiTools.siteOf(new URL(url).hostname);
  if (final.protocol !== "https:" || aiTools.siteOf(final.hostname) !== site) {
    throw new Error(`script redirected off ${site}: ${final.href}`);
  }
  if (!res.ok) throw new Error(`script download failed: HTTP ${res.status}`);
  const body = Buffer.from(await res.arrayBuffer());
  if (body.length > SCRIPT_MAX) throw new Error("script is larger than 1 MB");
  const file = path.join(
    os.tmpdir(),
    `agenthub-install-${crypto.randomBytes(6).toString("hex")}.sh`
  );
  fs.writeFileSync(file, body, { mode: 0o600 });
  return file;
}

/** Signal the child's whole process group; the child alone when it has none. */
function killTree(child, sig) {
  try {
    if (child.pid) return void process.kill(-child.pid, sig);
  } catch {
    /* group already gone — try the child itself */
  }
  try {
    child.kill?.(sig);
  } catch {
    /* already dead */
  }
}

function runArgv(j, argv) {
  return new Promise((resolve) => {
    append(j, `$ ${argv.join(" ")}\n`);
    let child;
    let done = false;
    const end = (code) => {
      if (done) return;
      done = true;
      j.child = null;
      resolve(code == null ? 1 : code);
    };
    try {
      const { file, shell } = aiTools.spawnTarget(argv[0]);
      child = spawnImpl(file, argv.slice(1), {
        shell,
        stdio: ["ignore", "pipe", "pipe"],
        env: { ...process.env, PATH: versions.jobPath() },
        detached: true, // its own process group, so a timeout reaches grandchildren
      });
    } catch (err) {
      append(j, `${err.message}\n`);
      return resolve(127);
    }
    j.child = child;
    child.stdout?.on("data", (b) => append(j, String(b)));
    child.stderr?.on("data", (b) => append(j, String(b)));
    child.on("error", (err) => append(j, `${err.message}\n`));
    child.on("close", (code) => end(code));
    // 'close' waits for every holder of the pipes; a daemon an installer left
    // behind would hold them forever and keep the job — and EBUSY — alive.
    child.on("exit", (code) => setTimeout(() => end(code), EXIT_GRACE_MS).unref?.());
  });
}

async function runSteps(j, steps, onStepOk) {
  const timer = setTimeout(() => {
    j.timedOut = true;
    append(j, "\n[agenthub] ETIMEDOUT: the job ran past 10 minutes and was stopped.\n");
    const child = j.child;
    if (!child) return;
    killTree(child, "SIGTERM");
    setTimeout(() => killTree(child, "SIGKILL"), KILL_GRACE_MS).unref?.();
  }, jobTimeoutMs);
  timer.unref?.();
  let code = 0;
  try {
    for (const step of steps) {
      let argv = step.argv;
      let tmp = null;
      if (step.script) {
        append(j, `$ download ${step.script}\n`);
        tmp = await download(step.script);
        argv = ["bash", tmp];
      }
      try {
        code = await runArgv(j, argv);
      } finally {
        if (tmp) fs.rmSync(tmp, { force: true });
      }
      if (j.timedOut) code = code || 124;
      if (code !== 0) {
        if (/EACCES/.test(j.log) && /npm/.test(path.basename(argv[0]))) append(j, EACCES_HINT);
        break;
      }
      onStepOk?.(step);
    }
  } catch (err) {
    append(j, `${err.message}\n`);
    code = code || 1;
  } finally {
    clearTimeout(timer);
  }
  return code;
}

function newJob(tool, action) {
  return {
    id: ++seq,
    tool,
    action,
    status: "running",
    startedAt: new Date().toISOString(),
    endedAt: null,
    exitCode: null,
    log: "",
  };
}

/**
 * Re-detect, THEN report the end: the client refetches the list once, on the
 * first poll that sees a non-running status. Never throws — a failure here
 * (a read-only data dir, a tool gone from the catalog mid-job) goes to the log.
 */
async function finish(j, code) {
  try {
    await hydrate().catch(() => {});
    aiTools.refresh();
    versions.invalidate();
    if (code === 0 && j.action === "install" && aiTools.detect(j.tool).installed) {
      aiTools.setEnabled(j.tool, true);
    }
    // S3: members run the shared copy under /opt/agenthub/tools — refresh it.
    if (code === 0 && require("./linux-accounts").s3Enabled()) {
      const shared = require("./shared-tools");
      const done = j.tool ? [await shared.publish(j.tool)] : await shared.publishAll();
      for (const r of done) {
        append(
          j,
          `\n[agenthub] ${r.error ? `${r.id}: not shared — ${r.error}` : `shared for every account: ${r.path}`}\n`
        );
      }
    }
  } catch (err) {
    append(j, `\n[agenthub] ${err.message}\n`);
  } finally {
    j.exitCode = code;
    j.endedAt = new Date().toISOString();
    j.status = code === 0 ? "ok" : "failed";
  }
}

function claim() {
  if (starting || job?.status === "running")
    throw makeErr("EBUSY", "another install or update is running");
  starting = true;
}

/**
 * Start `install` or `update` on one tool. Resolves once it has STARTED with
 * the public job; the job itself runs on and is read through current().
 */
async function start(id, action) {
  claim();
  let steps;
  try {
    steps = await stepsFor(id, action);
  } finally {
    starting = false;
  }
  job = newJob(id, action);
  const j = job;
  void runSteps(j, steps)
    .then((code) => finish(j, code))
    .catch(() => finish(j, 1));
  return view(j);
}

/**
 * One job that runs every installed tool's updater — its own, or npm's when
 * npm manages it (one installed another way is skipped). No version lookup:
 * each updater knows whether it has anything to do.
 */
async function startUpdateAll() {
  claim();
  let steps = [];
  try {
    for (const t of aiTools.list()) {
      if (!aiTools.detect(t.id).installed || !t.update) continue;
      if (t.update.kind === "npm" && !(await versions.npmManaged(t.id))) continue;
      steps.push(...(await stepsFor(t.id, "update")));
    }
    if (!steps.length) throw makeErr("ENOTHING", "no AI tool has an update to run");
  } finally {
    starting = false;
  }
  job = newJob(null, "update-all");
  const j = job;
  void runSteps(j, steps)
    .then((code) => finish(j, code))
    .catch(() => finish(j, 1));
  return view(j);
}

function view(j) {
  if (!j) return null;
  const { child: _c, timedOut: _t, ...rest } = j;
  return rest;
}

const current = () => view(job);

/** Test seams. */
function __setSpawn(fn) {
  spawnImpl = fn || spawn;
}
function __setHydrate(fn) {
  hydrate = fn || hydrateShellPath;
}
function __reset({ timeoutMs = 10 * 60 * 1000 } = {}) {
  job = null;
  starting = false;
  jobTimeoutMs = timeoutMs;
}

module.exports = { start, startUpdateAll, current, stepsFor, __setSpawn, __setHydrate, __reset };