AgentHub/server/lib/clone-jobs.js

/**
 * @file Background clones (`POST /api/lanes/clone` with `background: true`):
 * a clone that outlives the request that started it AND the server. Each job
 * is kept in memory and in `<data dir>/runtime/clone-jobs.json` (0600, tmp +
 * rename): on every state change at once, on git's progress at most every
 * `WRITE_EVERY_MS`. Its git runs in a detached runner
 * (`scripts/clone-runner.js`, pid recorded on the job) with stderr appended
 * to `<data dir>/runtime/clone-jobs/<id>/stderr.log` — the server tails it
 * for progress and finishes the job when the runner writes `exit.json`.
 * An update restarts the server but not the runner (`KillMode=process`), so
 * the first read in a new process reconciles the file: a running job whose
 * runner is alive is ADOPTED (counted against clone-queue's limit, tailed,
 * finished when it exits), one whose `exit.json` exists is finished now, and
 * any other is `interrupted` (`EINTERRUPTED`); a queued job is queued again,
 * unless it carried a typed login (gone with the old process: interrupted,
 * so the user types it again). Finishing runs the after-step the routes
 * register (`setAfterStep`: the workspace, or the extra folder) exactly once,
 * and a cancelled job never runs it. `cancel` takes a queued job out of
 * clone-queue or SIGTERMs the runner's process group. Finished jobs are
 * dropped after 24 h (done) or 7 days (failed/interrupted/cancelled). A job
 * is answered only to its owner (`scope.ownerFor`; local mode: every job),
 * and `request` — what a retry re-sends — never holds a credential: a typed
 * login is used for its one run and never written here.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

"use strict";

const fs = require("node:fs");
const path = require("node:path");
const crypto = require("node:crypto");
const { spawn } = require("node:child_process");
const { getDataDir } = require("./claude-home");

const KEEP_DONE_MS = 24 * 60 * 60 * 1000;
const KEEP_FAILED_MS = 7 * 24 * 60 * 60 * 1000;
const WRITE_EVERY_MS = 2000;
const POLL_MS = 500;
const TAIL_MAX = 64 * 1024;
const ACTIVE = new Set(["queued", "running"]);
const REQUEST_KEYS = ["url", "parent", "name", "tool", "title", "partial"];
const RUNNER = path.join(__dirname, "..", "..", "scripts", "clone-runner.js");

/** id → job record (the CloneJob fields plus `owner`, `once`). */
let jobs = null;
let timer = null;
/** Live progress pollers, so a simulated restart (`_reset`) stops them. */
const watchers = new Set();
/** `(job) => {status, json, laneId?, created?}` — the routes' after-step. */
let afterStep = null;

const runtimeDir = () => path.join(getDataDir(), "runtime");
const filePath = () => path.join(runtimeDir(), "clone-jobs.json");
/** A job's files; its id is server-made base64url, never from a request. */
const jobDir = (id) => path.join(runtimeDir(), "clone-jobs", id);

const serverMode = () => process.env.AGENTHUB_MODE === "server";
const queue = () => require("./clone-queue");

function expired(job, now) {
  if (!job.finishedAt) return false;
  const age = now - Date.parse(job.finishedAt);
  return job.state === "done" ? age > KEEP_DONE_MS : age > KEEP_FAILED_MS;
}

/** What goes to disk (`view` drops `owner` and `once`); never a callback. */
function record(job) {
  return {
    id: job.id,
    owner: job.owner,
    kind: job.kind,
    state: job.state,
    url: job.url,
    name: job.name,
    parent: job.parent,
    target: job.target,
    laneId: job.laneId,
    phase: job.phase,
    label: job.label,
    percent: job.percent,
    detail: job.detail,
    error: job.error,
    result: job.result,
    createdAt: job.createdAt,
    finishedAt: job.finishedAt,
    request: job.request,
    pid: job.pid ?? null,
    // Whether a login was typed for this run — never the login itself.
    once: job.once ?? null,
  };
}

/** The in-memory-only parts of a job: cancel signal, queue hooks, mirror. */
function arm(job) {
  const controller = new AbortController();
  Object.defineProperties(job, {
    // A progressToken's clone-queue record, kept in step (route-set).
    mirror: { enumerable: false, writable: true, value: null },
    // An adopted runner's clone-queue slot.
    release: { enumerable: false, writable: true, value: null },
    finishing: { enumerable: false, writable: true, value: false },
    signal: { enumerable: false, value: controller.signal },
    abort: { enumerable: false, value: () => controller.abort() },
    onChange: {
      enumerable: false,
      value: (what) => {
        if (job.mirror) {
          for (const k of ["state", "phase", "label", "percent", "detail"]) {
            job.mirror[k] = job[k];
          }
        }
        if (what === "state") writeNow();
        else writeSoon();
      },
    },
  });
  return job;
}

function interrupt(job) {
  Object.assign(job, {
    state: "interrupted",
    error: { code: "EINTERRUPTED", message: "AgentHub restarted while this clone ran" },
    finishedAt: new Date().toISOString(),
  });
}

function load() {
  if (jobs) return jobs;
  jobs = new Map();
  let list = [];
  try {
    const parsed = JSON.parse(fs.readFileSync(filePath(), "utf8"));
    if (Array.isArray(parsed?.jobs)) list = parsed.jobs;
  } catch {
    // No file yet, or an unreadable one: start empty.
  }
  const now = Date.now();
  let changed = false;
  const later = [];
  for (const j of list) {
    if (!j || typeof j.id !== "string" || !/^[A-Za-z0-9_-]{16,64}$/.test(j.id)) continue;
    if (expired(j, now)) {
      rmJobDir(j.id);
      changed = true;
      continue;
    }
    arm(j);
    jobs.set(j.id, j);
    if (j.state === "running") {
      if (readExit(j.id)) later.push(() => finishFromDisk(j));
      else if (runnerAlive(j.pid)) adopt(j);
      else {
        interrupt(j);
        rmJobDir(j.id);
        changed = true;
      }
    } else if (j.state === "queued") {
      if (j.once?.used) {
        interrupt(j);
        changed = true;
      } else later.push(() => start(j, null));
    }
  }
  if (changed) writeNow();
  for (const fn of later) setImmediate(fn);
  return jobs;
}

function writeNow() {
  if (timer) {
    clearTimeout(timer);
    timer = null;
  }
  const file = filePath();
  try {
    fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });
    const tmp = `${file}.${process.pid}.tmp`;
    const body = JSON.stringify({ jobs: [...(jobs || new Map()).values()].map(record) });
    fs.writeFileSync(tmp, body, { mode: 0o600 });
    fs.chmodSync(tmp, 0o600);
    fs.renameSync(tmp, file);
  } catch (err) {
    console.warn(`[clone-jobs] could not save ${file}: ${err.message}`);
  }
}

/** Progress is cheap in memory; on disk at most every WRITE_EVERY_MS. */
function writeSoon() {
  if (timer) return;
  timer = setTimeout(writeNow, WRITE_EVERY_MS);
  timer.unref?.();
}

function rmJobDir(id) {
  try {
    fs.rmSync(jobDir(id), { recursive: true, force: true });
  } catch {
    /* AgentHub's own scratch: best effort */
  }
}

function prune() {
  const now = Date.now();
  let changed = false;
  for (const [id, j] of load()) {
    if (expired(j, now)) {
      jobs.delete(id);
      rmJobDir(id);
      changed = true;
    }
  }
  if (changed) writeNow();
}

// ── The runner ─────────────────────────────────────────────────────────────

function readExit(id) {
  try {
    return JSON.parse(fs.readFileSync(path.join(jobDir(id), "exit.json"), "utf8"));
  } catch {
    return null;
  }
}

/** Whether `pid` is still OUR runner (not a reused pid, not a zombie). */
function runnerAlive(pid) {
  if (!Number.isInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
  } catch (err) {
    if (err.code !== "EPERM") return false;
  }
  if (process.platform !== "linux") return true;
  try {
    return fs.readFileSync(`/proc/${pid}/cmdline`, "utf8").includes("clone-runner.js");
  } catch {
    return false;
  }
}

/** SIGTERMs the runner's process group: git dies, the runner records it. */
function killRunner(job) {
  if (!job.pid) return;
  try {
    process.kill(process.platform === "win32" ? job.pid : -job.pid, "SIGTERM");
  } catch {
    /* already gone */
  }
}

function spawnRunner(job, spec, env) {
  const dir = jobDir(job.id);
  fs.rmSync(dir, { recursive: true, force: true });
  fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
  const specPath = path.join(dir, "spec.json");
  fs.writeFileSync(specPath, JSON.stringify(spec), { mode: 0o600, flag: "wx" });
  const child = spawn(process.execPath, [RUNNER, specPath], {
    cwd: dir,
    env,
    detached: true,
    stdio: "ignore",
    windowsHide: true,
  });
  child.on("error", () => {
    /* the poller sees no runner and fails the job */
  });
  child.unref();
  job.pid = child.pid ?? null;
  writeNow();
}

/**
 * Tails a runner's stderr.log into `onProgress` until `exit.json` appears
 * (or the runner is gone without one: `lost`). Resolves
 * `{ok, timedOut, lost, stderr}` — `stderr` is the last TAIL_MAX of it.
 */
function watch(job, onProgress) {
  return new Promise((resolve) => {
    const dir = jobDir(job.id);
    let offset = 0;
    let tail = "";
    let misses = 0;
    const read = () => {
      let fd;
      try {
        fd = fs.openSync(path.join(dir, "stderr.log"), "r");
        const size = fs.fstatSync(fd).size;
        if (size <= offset) return;
        const from = Math.max(offset, size - TAIL_MAX);
        const buf = Buffer.alloc(size - from);
        fs.readSync(fd, buf, 0, buf.length, from);
        offset = size;
        const chunk = buf.toString("utf8");
        tail = (tail + chunk).slice(-TAIL_MAX);
        try {
          onProgress?.(chunk);
        } catch {
          /* a progress reader never fails the clone */
        }
      } catch {
        /* no log yet */
      } finally {
        if (fd !== undefined) fs.closeSync(fd);
      }
    };
    const w = {};
    const stop = (out) => {
      clearInterval(w.iv);
      watchers.delete(w);
      resolve(out);
    };
    w.iv = setInterval(() => {
      read();
      const exit = readExit(job.id);
      if (exit) {
        read();
        return stop({
          ok: exit.code === 0 && !exit.signal,
          timedOut: Boolean(exit.timedOut),
          lost: false,
          stderr: tail,
        });
      }
      if (runnerAlive(job.pid)) misses = 0;
      else if (++misses >= 2) stop({ ok: false, timedOut: false, lost: true, stderr: tail });
    }, POLL_MS);
    w.iv.unref?.();
    watchers.add(w);
  });
}

const failureMessage = (out) =>
  out.lost
    ? "the clone stopped unexpectedly"
    : out.timedOut
      ? "git clone timed out"
      : "git clone failed";

/**
 * `cloneRepo`'s `launch`: the same git command, run by a detached runner
 * instead of as this process's child. Rejects like `worktree.git()`.
 */
function launcher(job) {
  return async (cwd, args, opts) => {
    const { cmd, argv, env } = require("./worktree").gitCommand(cwd, args, {
      env: opts.env,
      background: true,
    });
    spawnRunner(job, { cmd, argv, cwd, timeoutMs: opts.timeout || 0 }, env);
    const out = await watch(job, opts.onStderr);
    if (out.ok) return { stdout: "", stderr: out.stderr };
    throw Object.assign(new Error(failureMessage(out)), { git: { stderr: out.stderr } });
  };
}

// ── Lifecycle ──────────────────────────────────────────────────────────────

/**
 * A new `queued` job. `request` is reduced to `REQUEST_KEYS` — never a
 * credential; `once` records only whether a login was typed. The record is
 * also clone-queue's job: `withCloneSlot` marks it running and `report`
 * fills its progress, each through `onChange`.
 */
function create({ owner, kind, url, name, parent, target, laneId = null, request, once = null }) {
  load();
  const req = {};
  for (const k of REQUEST_KEYS) if (request?.[k] !== undefined) req[k] = request[k];
  const job = arm({
    id: crypto.randomBytes(16).toString("base64url"),
    owner: owner ?? null,
    kind,
    state: "queued",
    url,
    name,
    parent,
    target,
    laneId,
    phase: null,
    label: null,
    percent: null,
    detail: null,
    error: null,
    result: null,
    createdAt: new Date().toISOString(),
    finishedAt: null,
    request: req,
    pid: null,
    once: once
      ? {
          used: true,
          remember: once.remember === true,
          member: once.member === true,
          host: once.host ?? null,
        }
      : null,
  });
  jobs.set(job.id, job);
  writeNow();
  return job;
}

/**
 * Runs a queued job: clone-queue, then the clone through a runner (`once`:
 * the typed login, this process only), then the after-step.
 */
function start(job, once = null) {
  void (async () => {
    let outcome;
    try {
      await require("./git-identity").cloneRepo({
        job,
        url: job.url,
        parent: job.parent,
        name: job.name,
        once,
        partial: job.request?.partial === true,
        signal: job.signal,
        launch: launcher(job),
      });
      outcome = { ok: true };
    } catch (err) {
      outcome = { ok: false, err };
    }
    finish(job, outcome);
  })();
  return job;
}

/** A runner this process did not start: counted, tailed, finished on exit. */
function adopt(job) {
  job.release = queue().occupy();
  watch(job, (chunk) => queue().report(job, chunk)).then((out) => finishFromDisk(job, out));
}

/**
 * Finishes a runner's job from its files alone (no `cloneRepo` closure: the
 * process that launched it is gone). A typed login a member's store kept for
 * this run only is taken out again, as `cloneRepo` would have.
 */
function finishFromDisk(job, out = null) {
  if (!out) {
    const exit = readExit(job.id) || {};
    let stderr = "";
    try {
      stderr = fs.readFileSync(path.join(jobDir(job.id), "stderr.log"), "utf8").slice(-TAIL_MAX);
    } catch {
      /* none */
    }
    out = {
      ok: exit.code === 0 && !exit.signal,
      timedOut: Boolean(exit.timedOut),
      lost: false,
      stderr,
    };
  }
  const identity = require("./git-identity");
  if (job.once?.used && job.once.member && !(out.ok && job.once.remember) && job.once.host) {
    try {
      identity.removeHttps(identity.accountForDir(job.parent), "https", job.once.host);
    } catch {
      /* best effort: the store is the member's own */
    }
  }
  finish(
    job,
    out.ok
      ? { ok: true }
      : { ok: false, err: identity.cloneFailure(job.url, out.stderr, failureMessage(out)) }
  );
}

/** The error body a failed job records: the route's own code and message. */
function errorBody(err) {
  return {
    code: err?.code || "ECLONE",
    message: err?.message || "clone failed",
    ...(err?.code === "ECREDENTIALS" ? { auth: { protocol: err.protocol, host: err.host } } : {}),
  };
}

/**
 * The end of a job, once: the slot back, the runner's files gone, then
 * nothing (cancelled — its half-made folder dropped), the failure, or the
 * after-step and its outcome.
 */
function finish(job, { ok, err }) {
  if (job.finishing) return;
  job.finishing = true;
  job.release?.();
  job.release = null;
  rmJobDir(job.id);
  if (cancelled(job)) {
    dropCancelledTarget(job.target);
    return;
  }
  if (!ok) return fail(job, errorBody(err));
  let out;
  try {
    out = afterStep
      ? afterStep(job)
      : { status: 500, json: { error: { code: "ECLONE", message: "no after-step registered" } } };
  } catch (e) {
    out = { status: 500, json: { error: { code: e.code || "ECLONE", message: e.message } } };
  }
  if (out.status < 300) {
    succeed(job, {
      laneId: out.laneId ?? null,
      path: job.target,
      ...(out.created !== undefined ? { created: out.created } : {}),
    });
  } else fail(job, out.json?.error);
}

/**
 * A cancelled clone of a HOST-account folder whose git left the target
 * behind (it was checked absent before the job): that exact path, never
 * through a link. A member's folder is git's own to clean (sudo relays the
 * SIGTERM) — the dashboard never deletes inside a member's home.
 */
function dropCancelledTarget(target) {
  try {
    if (require("./run-as").accountForPath(target)) return;
    if (!fs.lstatSync(target).isDirectory()) return;
    fs.rmSync(target, { recursive: true, force: true });
  } catch {
    /* nothing there, or not ours to remove */
  }
}

function settle(job, fields) {
  if (job.state === "cancelled") return;
  Object.assign(job, fields, { finishedAt: new Date().toISOString() });
  if (job.mirror) queue().finish(job.mirror, job.state === "done");
  if (jobs?.get(job.id) === job) writeNow();
}

/** The clone and its after-step worked: `result = {laneId, path, created?}`. */
function succeed(job, result) {
  settle(job, { state: "done", error: null, result });
}

/** It failed: `error = {code, message, auth?}` — what the route would answer. */
function fail(job, error) {
  const e = { code: error?.code || "ECLONE", message: error?.message || "clone failed" };
  if (error?.auth) e.auth = { protocol: error.auth.protocol, host: error.auth.host };
  settle(job, { state: "failed", error: e, result: null });
}

/**
 * Cancels one of the caller's jobs: `ENOTFOUND` when not theirs, `EDONE` once
 * it finished. Queued: out of the queue, never started. Running: the
 * runner's group is SIGTERMed (and a borrowed-login probe aborted); the slot
 * frees when git exits.
 */
function cancel(id, owner) {
  const job = get(id, owner);
  if (!job) throw Object.assign(new Error("no such clone"), { code: "ENOTFOUND" });
  if (!ACTIVE.has(job.state)) {
    throw Object.assign(new Error("the clone already finished"), { code: "EDONE" });
  }
  settle(job, {
    state: "cancelled",
    error: { code: "ECANCELLED", message: "Clone cancelled" },
    result: null,
  });
  if (!queue().dequeue(job)) {
    job.abort();
    killRunner(job);
  }
  return job;
}

/** Whether a cancel already settled this job. */
const cancelled = (job) => job.state === "cancelled";

const mine = (job, owner) => !serverMode() || job.owner === (owner ?? null);

/** The CloneJob the API answers. */
function view(job) {
  const { owner: _owner, once: _once, ...out } = record(job);
  return {
    ...out,
    position: job.state === "queued" ? queue().position(job) : null,
  };
}

/** The caller's jobs, newest first. */
function list(owner) {
  prune();
  return [...jobs.values()]
    .filter((j) => mine(j, owner))
    .sort((a, b) => (a.createdAt < b.createdAt ? 1 : a.createdAt > b.createdAt ? -1 : 0));
}

/** One of the caller's jobs, or null. */
function get(id, owner) {
  prune();
  const job = typeof id === "string" ? jobs.get(id) : null;
  return job && mine(job, owner) ? job : null;
}

/** Drops a finished job: `ENOTFOUND` when not the caller's, `EBUSY` while it runs. */
function remove(id, owner) {
  const job = get(id, owner);
  if (!job) throw Object.assign(new Error("no such clone"), { code: "ENOTFOUND" });
  if (ACTIVE.has(job.state)) {
    throw Object.assign(new Error("the clone is still running"), { code: "EBUSY" });
  }
  jobs.delete(id);
  rmJobDir(id);
  writeNow();
}

/** The routes' after-step: `(job) => {status, json, laneId?, created?}`. */
function setAfterStep(fn) {
  afterStep = fn;
}

/** Server start: reconcile the file now (adopt, finish, interrupt, requeue). */
function resume() {
  load();
}

/** Tests: forget the in-memory state and stop tailing, as a restart would. */
function _reset() {
  if (timer) clearTimeout(timer);
  timer = null;
  for (const w of watchers) clearInterval(w.iv);
  watchers.clear();
  jobs = null;
}

module.exports = {
  create,
  start,
  view,
  list,
  get,
  remove,
  cancel,
  cancelled,
  setAfterStep,
  resume,
  filePath,
  jobDir,
  ACTIVE,
  KEEP_DONE_MS,
  KEEP_FAILED_MS,
  _reset,
};