AgentHub/server/__tests__/clone-jobs.test.js

/**
 * @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ĩ <vinnt@smartgift.vn>
 */

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;
    }
  });
});