/** * @file Tests for background clones (`server/lib/clone-jobs.js` and the * `background: true` path of `POST /api/lanes/clone` and * `POST /api/lanes/:id/sources/clone`): checks still answer synchronously, * a job is answered 202 and ends `done` with the workspace / extra folder * made, a failure keeps the route's own code (ENOREPO, ECREDENTIALS with * auth), a typed login never reaches the job file, retry and dismiss, cancel * of a queued clone (never starts) and of a running one (git is killed, no * workspace made), and a restart: a live detached runner adopted (and still * cancellable), one that ended finished with its after-step run once, a dead * one `interrupted`, a queued job requeued (or interrupted when it carried a * typed login), retention; and that in server mode a job is only ever its * owner's. * @author Nguyễn Ngọc Trí Vĩ */ const { describe, it, before, after } = require("node:test"); const assert = require("node:assert/strict"); const fs = require("node:fs"); const os = require("node:os"); const path = require("node:path"); const http = require("node:http"); const { execFileSync } = require("node:child_process"); const ROOT = fs.mkdtempSync(path.join(os.tmpdir(), "clone-jobs-")); process.env.DASHBOARD_DB_PATH = path.join(ROOT, "t.db"); process.env.DASHBOARD_DATA_DIR = path.join(ROOT, "data"); process.env.CLAUDE_HOME = path.join(ROOT, ".claude"); process.env.DASHBOARD_REMOTE_SYNC_MS = "0"; process.env.DASHBOARD_LIVENESS_PROBE = "0"; process.env.GIT_CONFIG_NOSYSTEM = "1"; process.env.HOME = ROOT; const { createApp } = require("../index"); const cloneJobs = require("../lib/clone-jobs"); const queue = require("../lib/clone-queue"); let app; let remote; let base; const WORK = path.join(ROOT, "work"); function request(method, p, body) { return fetch(`${base}${p}`, { method, headers: { "Content-Type": "application/json" }, body: body === undefined ? undefined : JSON.stringify(body), }).then(async (r) => ({ status: r.status, body: await r.json().catch(() => null) })); } async function waitFor(fn, ms = 20_000) { const end = Date.now() + ms; for (;;) { const v = await fn(); if (v) return v; if (Date.now() > end) throw new Error("timed out"); await new Promise((r) => setTimeout(r, 50)); } } async function jobById(id) { const r = await request("GET", "/api/lanes/clone/jobs"); return r.body.jobs.find((j) => j.id === id); } const finished = (id) => waitFor(async () => { const j = await jobById(id); return j && !["queued", "running"].includes(j.state) ? j : null; }); before(async () => { const src = path.join(ROOT, "src"); const bare = path.join(ROOT, "bare.git"); execFileSync("git", ["init", "-q", "-b", "main", src]); fs.writeFileSync(path.join(src, "README.md"), "hi\n"); execFileSync("git", ["-C", src, "add", "."]); execFileSync("git", [ "-C", src, "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "i", ]); execFileSync("git", ["clone", "-q", "--bare", src, bare]); remote = http.createServer((_req, res) => { res.writeHead(401, { "WWW-Authenticate": 'Basic realm="t"' }); res.end(); }); await new Promise((r) => remote.listen(0, "127.0.0.1", r)); fs.writeFileSync( path.join(ROOT, ".gitconfig"), [ `[url "file://${bare}"]`, "\tinsteadOf = https://forge.test/o/r.git", `[url "file://${path.join(ROOT, "missing.git")}"]`, "\tinsteadOf = git@forge.test:o/missing.git", `[url "http://127.0.0.1:${remote.address().port}/"]`, "\tinsteadOf = https://private.test/", "", ].join("\n") ); fs.mkdirSync(WORK); app = http.createServer(createApp()); await new Promise((r) => app.listen(0, "127.0.0.1", r)); base = `http://127.0.0.1:${app.address().port}`; }); after(async () => { await new Promise((r) => app.close(r)); await new Promise((r) => remote.close(r)); }); describe("background clone", () => { it("still refuses bad input synchronously", async () => { const bad = await request("POST", "/api/lanes/clone", { url: "https://forge.test/o/r.git", parent: WORK, background: "yes", }); assert.equal(bad.status, 400); assert.equal(bad.body.error.code, "EBADFIELD"); const url = await request("POST", "/api/lanes/clone", { url: "file:///etc", parent: WORK, background: true, }); assert.equal(url.body.error.code, "EBADURL"); fs.mkdirSync(path.join(WORK, "taken")); const taken = await request("POST", "/api/lanes/clone", { url: "https://forge.test/o/r.git", parent: WORK, name: "taken", background: true, }); assert.equal(taken.status, 409); assert.equal(taken.body.error.code, "EEXISTS"); const tool = await request("POST", "/api/lanes/clone", { url: "https://forge.test/o/r.git", parent: WORK, name: "t2", tool: "nope", background: true, }); assert.equal(tool.body.error.code, "EBADTOOL"); const jobs = await request("GET", "/api/lanes/clone/jobs"); assert.deepEqual(jobs.body.jobs, []); }); it("answers 202 with a job that ends done with its workspace", async () => { const r = await request("POST", "/api/lanes/clone", { url: "https://forge.test/o/r.git", parent: WORK, name: "bg", title: "Bg", background: true, }); assert.equal(r.status, 202, JSON.stringify(r.body)); const { job } = r.body; assert.match(job.id, /^[A-Za-z0-9_-]{16,}$/); assert.equal(job.kind, "workspace"); assert.ok(["queued", "running"].includes(job.state)); assert.equal(job.target, path.join(WORK, "bg")); assert.equal(job.laneId, null); assert.equal(job.finishedAt, null); assert.ok("phase" in job && "position" in job); assert.equal(job.owner, undefined); assert.deepEqual(job.request, { url: "https://forge.test/o/r.git", parent: WORK, name: "bg", title: "Bg", }); const done = await finished(job.id); assert.equal(done.state, "done", JSON.stringify(done)); assert.equal(done.result.created, true); assert.equal(done.result.path, path.join(WORK, "bg")); assert.ok(done.finishedAt); const lane = await request("GET", `/api/lanes/${done.result.laneId}`); assert.equal(lane.body.lane?.title ?? lane.body.title, "Bg"); assert.equal(fs.readFileSync(path.join(WORK, "bg", "README.md"), "utf8"), "hi\n"); const file = cloneJobs.filePath(); assert.equal(fs.statSync(file).mode & 0o777, 0o600); const disk = JSON.parse(fs.readFileSync(file, "utf8")).jobs.find((j) => j.id === job.id); assert.equal(disk.state, "done"); const busy = await request("POST", `/api/lanes/clone/jobs/${job.id}/retry`, {}); assert.equal(busy.status, 409); assert.equal(busy.body.error.code, "EBUSY"); const dismissed = await request("DELETE", `/api/lanes/clone/jobs/${job.id}`); assert.deepEqual(dismissed.body, { ok: true }); assert.equal(await jobById(job.id), undefined); const again = await request("DELETE", `/api/lanes/clone/jobs/${job.id}`); assert.equal(again.status, 404); assert.equal(again.body.error.code, "ENOTFOUND"); }); it("records the route's own error codes, and never a typed login", async () => { const missing = await request("POST", "/api/lanes/clone", { url: "git@forge.test:o/missing.git", parent: WORK, name: "missing", background: true, }); assert.equal(missing.status, 202); const m = await finished(missing.body.job.id); assert.equal(m.state, "failed"); assert.equal(m.error.code, "ENOREPO"); assert.equal(m.result, null); const login = await request("POST", "/api/lanes/clone", { url: "https://private.test/o/r.git", parent: WORK, name: "private", credentials: { username: "u", password: "s3cret-typed" }, background: true, }); assert.equal(login.status, 202); assert.equal(login.body.job.request.credentials, undefined); const l = await finished(login.body.job.id); assert.equal(l.error.code, "ECREDENTIALS"); assert.deepEqual(l.error.auth, { protocol: "https", host: "private.test" }); assert.ok(!fs.readFileSync(cloneJobs.filePath(), "utf8").includes("s3cret-typed")); // Retry: a new job from the old request (+ a login typed now); the old goes. const retry = await request("POST", `/api/lanes/clone/jobs/${l.id}/retry`, { credentials: { username: "u", password: "again-typed" }, }); assert.equal(retry.status, 202, JSON.stringify(retry.body)); assert.notEqual(retry.body.job.id, l.id); assert.equal(retry.body.job.name, "private"); assert.equal(await jobById(l.id), undefined); const r2 = await finished(retry.body.job.id); assert.equal(r2.error.code, "ECREDENTIALS"); assert.ok(!fs.readFileSync(cloneJobs.filePath(), "utf8").includes("again-typed")); const bad = await request("POST", `/api/lanes/clone/jobs/${r2.id}/retry`, { credentials: "nope", }); assert.equal(bad.body.error.code, "EBADCRED"); assert.ok(await jobById(r2.id), "a refused retry keeps the old job"); const none = await request("POST", "/api/lanes/clone/jobs/nope-nope-nope-nope/retry", {}); assert.equal(none.status, 404); }); it("clones an extra folder beside the workspace", async () => { const home = path.join(WORK, "app"); fs.mkdirSync(home); const made = await request("POST", "/api/lanes/ensure", { cwd: home }); const id = made.body.lane.id; const r = await request("POST", `/api/lanes/${id}/sources/clone`, { url: "https://forge.test/o/r.git", name: "lib", background: true, }); assert.equal(r.status, 202, JSON.stringify(r.body)); assert.equal(r.body.job.kind, "source"); assert.equal(r.body.job.laneId, id); const done = await finished(r.body.job.id); assert.equal(done.state, "done", JSON.stringify(done)); assert.equal(done.result.laneId, id); const ctx = await request("GET", `/api/lanes/${id}/sources`); const folders = (ctx.body.sources || []).filter((s) => s.kind === "folder"); assert.deepEqual( folders.map((f) => fs.realpathSync(f.path)), [fs.realpathSync(path.join(WORK, "lib"))] ); }); it("cancels a queued clone before it starts", async () => { const saved = process.env.AGENTHUB_CLONE_CONCURRENCY; process.env.AGENTHUB_CLONE_CONCURRENCY = "1"; let open; const blocker = queue.withCloneSlot(() => new Promise((r) => (open = r))); try { const r = await request("POST", "/api/lanes/clone", { url: "https://forge.test/o/r.git", parent: WORK, name: "queued", background: true, }); assert.equal(r.status, 202); assert.equal(r.body.job.state, "queued"); assert.equal(r.body.job.position, 1); const c = await request("POST", `/api/lanes/clone/jobs/${r.body.job.id}/cancel`); assert.equal(c.status, 200, JSON.stringify(c.body)); assert.equal(c.body.job.state, "cancelled"); assert.equal(c.body.job.error.code, "ECANCELLED"); assert.equal(queue.cloneQueueState().waiting, 0); const twice = await request("POST", `/api/lanes/clone/jobs/${r.body.job.id}/cancel`); assert.equal(twice.status, 409); assert.equal(twice.body.error.code, "EDONE"); open(); await blocker; await new Promise((res) => setTimeout(res, 200)); assert.equal((await jobById(r.body.job.id)).state, "cancelled"); assert.ok(!fs.existsSync(path.join(WORK, "queued"))); } finally { open?.(); if (saved === undefined) delete process.env.AGENTHUB_CLONE_CONCURRENCY; else process.env.AGENTHUB_CLONE_CONCURRENCY = saved; } }); it("cancels a running clone: git is killed, the slot frees, no workspace", async () => { // An ssh "transport" that never answers keeps git clone running (a real // ssh exits once git is gone; this stand-in lingers for its 15 s). process.env.GIT_SSH_COMMAND = "exec sleep 15 2>/dev/null #"; try { const r = await request("POST", "/api/lanes/clone", { url: "ssh://slow.test/o/slow.git", parent: WORK, name: "slow", background: true, }); assert.equal(r.status, 202); const id = r.body.job.id; await waitFor(async () => (await jobById(id))?.state === "running"); await new Promise((res) => setTimeout(res, 300)); const c = await request("POST", `/api/lanes/clone/jobs/${id}/cancel`); assert.equal(c.body.job?.state, "cancelled", JSON.stringify([c.body, await jobById(id)])); await waitFor(() => queue.cloneQueueState().running === 0, 10_000); const j = await jobById(id); assert.equal(j.state, "cancelled"); assert.equal(j.result, null); assert.ok(!fs.existsSync(path.join(WORK, "slow"))); const lanes = await request("GET", "/api/lanes"); const list = lanes.body.lanes || lanes.body; assert.ok(!list.some((l) => l.cwd.endsWith(`${path.sep}slow`))); const retry = await request("POST", `/api/lanes/clone/jobs/${id}/retry`, {}); assert.equal(retry.status, 202, "a cancelled clone can be retried"); const c2 = await request("POST", `/api/lanes/clone/jobs/${retry.body.job.id}/cancel`); assert.equal(c2.body.job.state, "cancelled"); await waitFor(() => queue.cloneQueueState().running === 0, 10_000); } finally { delete process.env.GIT_SSH_COMMAND; } }); }); describe("clone-jobs store", () => { it("answers a job only to its owner in server mode", () => { const savedMode = process.env.AGENTHUB_MODE; process.env.AGENTHUB_MODE = "server"; try { const mk = (owner) => cloneJobs.create({ owner, kind: "workspace", url: "https://forge.test/o/r.git", name: `o${owner}`, parent: WORK, target: path.join(WORK, `o${owner}`), request: { url: "u", parent: WORK, name: "n", credentials: { password: "x" } }, }); const a = mk(101); const b = mk(102); assert.equal(a.request.credentials, undefined); assert.deepEqual( cloneJobs.list(101).map((j) => j.id), [a.id] ); assert.equal(cloneJobs.get(b.id, 101), null); assert.throws(() => cloneJobs.remove(b.id, 101), { code: "ENOTFOUND" }); assert.throws(() => cloneJobs.cancel(b.id, 101), { code: "ENOTFOUND" }); assert.throws(() => cloneJobs.remove(a.id, 101), { code: "EBUSY" }); cloneJobs.cancel(a.id, 101); cloneJobs.cancel(b.id, 102); cloneJobs.remove(a.id, 101); cloneJobs.remove(b.id, 102); } finally { if (savedMode === undefined) delete process.env.AGENTHUB_MODE; else process.env.AGENTHUB_MODE = savedMode; } }); }); describe("a restart", () => { const H = 60 * 60 * 1000; const ago = (ms) => new Date(Date.now() - ms).toISOString(); const row = (id, state, extra = {}) => ({ id, owner: null, kind: "workspace", state, url: "https://forge.test/o/r.git", name: id, parent: WORK, target: path.join(WORK, id), laneId: null, error: null, result: null, createdAt: ago(H), finishedAt: null, request: { url: "https://forge.test/o/r.git", parent: WORK, name: id }, pid: null, once: null, ...extra, }); /** Writes the job file and reloads it, as a new server process would. */ const restart = (rows) => { cloneJobs._reset(); fs.mkdirSync(path.dirname(cloneJobs.filePath()), { recursive: true }); fs.writeFileSync(cloneJobs.filePath(), JSON.stringify({ jobs: rows })); cloneJobs.resume(); }; const lanesAt = async (dir) => { const r = await request("GET", "/api/lanes"); const list = r.body.lanes || r.body; return list.filter((l) => fs.realpathSync(l.cwd) === fs.realpathSync(dir)); }; it("interrupts a dead runner, requeues a queued clone, drops expired jobs", async () => { const dead = 2 ** 22 + 12345; // past pid_max on a default kernel: never alive restart([ row("was-running-aaaaaaaa", "running", { pid: dead }), row("queued-login-aaaaaaa", "queued", { once: { used: true, remember: false, member: false, host: "forge.test" }, }), row("requeued-aaaaaaaaaaa", "queued"), row("done-fresh-aaaaaaaaa", "done", { finishedAt: ago(2 * H) }), row("done-stale-aaaaaaaaa", "done", { finishedAt: ago(25 * H) }), row("failed-fresh-aaaaaaa", "failed", { finishedAt: ago(6 * 24 * H) }), row("failed-stale-aaaaaaa", "failed", { finishedAt: ago(8 * 24 * H) }), row("cancel-stale-aaaaaaa", "cancelled", { finishedAt: ago(8 * 24 * H) }), ]); const ids = cloneJobs.list(null).map((j) => j.id); assert.deepEqual([...ids].sort(), [ "done-fresh-aaaaaaaaa", "failed-fresh-aaaaaaa", "queued-login-aaaaaaa", "requeued-aaaaaaaaaaa", "was-running-aaaaaaaa", ]); for (const id of ["was-running-aaaaaaaa", "queued-login-aaaaaaa"]) { const j = cloneJobs.get(id, null); assert.equal(j.state, "interrupted", id); assert.deepEqual(j.error, { code: "EINTERRUPTED", message: "AgentHub restarted while this clone ran", }); assert.ok(j.finishedAt); } const disk = JSON.parse(fs.readFileSync(cloneJobs.filePath(), "utf8")).jobs; assert.equal(disk.find((j) => j.id === "was-running-aaaaaaaa").state, "interrupted"); // The queued one (no typed login) simply runs now, workspace included. const done = await finished("requeued-aaaaaaaaaaa"); assert.equal(done.state, "done", JSON.stringify(done)); assert.equal((await lanesAt(path.join(WORK, "requeued-aaaaaaaaaaa"))).length, 1); }); it("finishes a runner that ended while no server ran: the after-step runs once", async () => { const id = "ended-ok-aaaaaaaaaaa"; const target = path.join(WORK, id); execFileSync("git", ["clone", "-q", path.join(ROOT, "bare.git"), target]); fs.mkdirSync(cloneJobs.jobDir(id), { recursive: true }); fs.writeFileSync(path.join(cloneJobs.jobDir(id), "exit.json"), '{"code":0,"signal":null}'); const bad = "ended-bad-aaaaaaaaaa"; fs.mkdirSync(cloneJobs.jobDir(bad), { recursive: true }); fs.writeFileSync(path.join(cloneJobs.jobDir(bad), "exit.json"), '{"code":128,"signal":null}'); fs.writeFileSync( path.join(cloneJobs.jobDir(bad), "stderr.log"), "Cloning into 'x'...\nfatal: repository 'https://forge.test/o/nope.git/' not found\n" ); restart([ row(id, "running", { pid: 2 ** 22 + 1 }), row(bad, "running", { pid: 2 ** 22 + 2, url: "https://forge.test/o/nope.git" }), ]); const ok = await finished(id); assert.equal(ok.state, "done", JSON.stringify(ok)); assert.equal(ok.result.created, true); const failed = await finished(bad); assert.equal(failed.error.code, "ENOREPO"); assert.ok(!fs.existsSync(cloneJobs.jobDir(id)), "the runner's files are cleaned up"); // A second reload finds nothing left to do: still one workspace. restart(JSON.parse(fs.readFileSync(cloneJobs.filePath(), "utf8")).jobs); await new Promise((r) => setTimeout(r, 100)); assert.equal((await lanesAt(target)).length, 1); assert.equal(cloneJobs.get(id, null).result.laneId, ok.result.laneId); }); it("adopts a runner that outlived the server, and can still cancel it", async () => { process.env.GIT_SSH_COMMAND = "exec sleep 15 2>/dev/null #"; try { const r = await request("POST", "/api/lanes/clone", { url: "ssh://slow.test/o/slow.git", parent: WORK, name: "adopted", background: true, }); assert.equal(r.status, 202); const id = r.body.job.id; const pid = await waitFor(async () => (await jobById(id))?.pid); // Its own session and group: a server restart leaves it running. const pgid = Number( execFileSync("ps", ["-o", "pgid=", "-p", String(pid)], { encoding: "utf8" }) ); assert.equal(pgid, pid); const before = queue.cloneQueueState().running; // A new server process: the old one's tail stops, the file is re-read. cloneJobs._reset(); cloneJobs.resume(); const adopted = cloneJobs.get(id, null); assert.equal(adopted.state, "running"); assert.equal(queue.cloneQueueState().running, before + 1, "counted against the limit"); const c = await request("POST", `/api/lanes/clone/jobs/${id}/cancel`); assert.equal(c.body.job.state, "cancelled"); await waitFor(() => queue.cloneQueueState().running === before, 10_000); await waitFor(() => { try { process.kill(pid, 0); return false; } catch { return true; } }, 10_000); assert.equal((await jobById(id)).state, "cancelled"); assert.ok(!fs.existsSync(path.join(WORK, "adopted"))); } finally { delete process.env.GIT_SSH_COMMAND; } }); });