/** * @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 `/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 `/runtime/clone-jobs//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ĩ */ "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, };