stack/packages/cli/tests/publish-socket.test.mjs

// Row 199 (#1697, design D9, D9b, D11, T3): the git socket's publish op end to end. The socket,
// working.mjs, the bus runtime (in this process, so its publishTls can name the scratch CA), the
// per-agent store and a TLS fake Gitea on 127.0.0.1. The host stand-in answers the socket's ops
// the way packages/bus/src/process.mjs does. Fake tokens in a scratch root only.
import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { chmodSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { connect } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { DatabaseSync } from "node:sqlite";
import { startBroker } from "../../bus/src/runtime.mjs";
import { businessDoc, rolesCopy, systemFor, writeJson } from "../../business/tests/helpers.mjs";
import { pushWorking } from "../../harness/src/git-tools.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { workRef } from "../../seat/src/working.mjs";
import { fakeGitea, scratchCa } from "../../seat/tests/fake-gitea.mjs";
import { createGitSocket } from "../src/git-socket.mjs";

const TASK = "vikunja:1/2";
const TASK2 = "vikunja:1/3";
const TOKEN = "fake_socket_token_0123456789";
const TOKEN2 = "fake_second_token_0123456789";

function git(cwd, ...args) {
  const r = spawnSync("git", args, { cwd, encoding: "utf8", env: gitEnv() });
  assert.equal(r.status, 0, `git ${args.join(" ")}: ${r.stderr}`);
  return r.stdout.trim();
}
function commit(dir, text, email) {
  writeFileSync(join(dir, "README"), `${text}\n`);
  git(dir, "add", "README");
  git(dir, "-c", `user.name=${text}`, "-c", `user.email=${email}`, "commit", "-q", "-m", text);
  return git(dir, "rev-parse", "HEAD");
}
function raw(path, req) {
  return new Promise((resolve, reject) => {
    const s = connect(path);
    let buf = "";
    s.on("connect", () => s.write(`${JSON.stringify(req)}\n`));
    s.on("data", (b) => (buf += b));
    s.on("end", () => resolve(JSON.parse(buf.split("\n")[0])));
    s.on("error", reject);
  });
}
// The socket's ops, answered as process.mjs's launchOp answers them.
const OPS = {
  identity: (r, m) => r.identity(m.cap),
  refuse: (r, m) => r.refuse(m.cap, m.code, m.subject),
  checkPush: (r, m) => r.checkPush(m.cap, m.task_ref),
  recordPush: (r, m) => r.recordPush(m.cap, m.push),
  checkPublish: (r, m) => r.checkPublish(m.cap, m.args),
  publishPush: (r, m) => r.publishPush(m.ticket),
  publishResult: (r, m) => r.publishResult(m.ticket),
  recordPublish: (r, m) => r.recordPublish(m.ticket, m.pushed),
  publishRefused: (r, m) => r.publishRefused(m.ticket, m.args),
};

async function setup(t, { publishDeadlineMs } = {}) {
  const root = mkdtempSync(join(tmpdir(), "cli-publish-"));
  const down = [];
  t.after(async () => {
    for (const fn of down) await fn();
    rmSync(root, { recursive: true, force: true });
  });
  const ca = scratchCa(join(root, "ca"));
  const gitea = await fakeGitea({ dir: join(root, "gitea"), tls: ca, users: { coder: TOKEN, second: TOKEN2 } });
  down.push(() => gitea.close());
  const project = join(root, "project");
  mkdirSync(project);
  git(project, "init", "-q", "-b", "main");
  commit(project, "one", "[email protected]");
  // Each agent's fork and the upstream, mosaic/stack, all clones of the root.
  for (const owner of ["coder", "second", "mosaic"]) {
    mkdirSync(join(gitea.repo(owner, "stack"), ".."), { recursive: true });
    git(root, "clone", "-q", "--bare", project, gitea.repo(owner, "stack"));
  }
  const forkRepo = gitea.repo("coder", "stack");

  const rolesDir = rolesCopy(root);
  writeJson(join(root, "businesses/demo.json"), { ...businessDoc(root, "demo"), plugins: { harness: ["pi"] } });
  mkdirSync(join(root, "data"), { mode: 0o700 });
  mkdirSync(join(root, "secrets", "agents"), { recursive: true, mode: 0o700 });
  chmodSync(join(root, "secrets"), 0o700);
  const agent = (user) => ({
    authority: { withinRole: ["git.push.working", "message.send"], crossRole: [] },
    identity: { name: user, email: `${user}@example.com`, gitea: { url: gitea.origin, user } },
    publish: ["stack"],
    credentials: { gitea: { store: true, file: join(root, "secrets", "agents", "demo", user, "gitea"), rotateBy: "2099-01-01" } },
  });
  const demo = {
    id: "demo", human: "jason", arbiters: { technical: "reviewer", delivery: "reviewer" },
    projects: { stack: { root: project, remote: `${gitea.origin}/mosaic/stack` } },
    roles: {
      coder: agent("coder"),
      second: agent("second"),
      reviewer: { authority: { withinRole: ["message.send"], crossRole: [] } },
    },
  };
  const dataRoot = join(root, "data");
  const bus = await startBroker({ dataRoot, businesses: { demo }, organization: { dir: root, rolesDir, system: systemFor(root) }, publishTls: ["-c", `http.sslCAInfo=${ca.ca}`] });
  down.unshift(() => bus.close());
  // A test sets hooks[op] to act around one op: hook(message, next), where next() answers it.
  const hooks = {};
  const host = { op: async (m) => (hooks[m.op] ? hooks[m.op](m, () => OPS[m.op](bus, m)) : OPS[m.op](bus, m)) };
  const logs = [];
  mkdirSync(join(dataRoot, "bus-host"), { mode: 0o700 });
  const sock = createGitSocket({ host, business: demo, dataRoot, env: process.env, log: (l) => logs.push(l), publishPollMs: 20, ...(publishDeadlineMs ? { publishDeadlineMs } : {}) });
  down.unshift(() => sock.close());
  await sock.listen();

  const caps = {};
  for (const role of ["coder", "second"]) {
    caps[role] = bus.broker.bindLaunch({ business: "demo", role, run: `${role}-run`, harness: "pi" });
    bus.broker.request(caps[role], { verb: "role.claim", args: {} });
  }
  const human = bus.broker.bindHuman({ business: "demo", human: "jason", via: "cli", outsideAgent: true });
  // Each agent's workspace: a clone of the root with one commit of its own on feat, pushed to the
  // working repo for its task.
  const work = async (role, task) => {
    const workspace = join(root, `ws-${role}`);
    const tmp = join(root, `tmp-${role}`);
    mkdirSync(workspace);
    mkdirSync(tmp);
    git(workspace, "clone", "-q", project, "stack");
    const clone = join(workspace, "stack");
    git(clone, "checkout", "-q", "-b", "feat");
    const commitId = commit(clone, `work by ${role}`, `${role}@example.com`);
    await pushWorking({ path: sock.path, cap: caps[role], workspace, tmp, task_ref: task, project: "stack", branch: "feat" });
    return commitId;
  };
  // A second commit on the coder's feat, pushed to the working repo through the socket.
  const again = async () => {
    const workspace = join(root, "ws-coder");
    const commitId = commit(join(workspace, "stack"), "more work", "[email protected]");
    await pushWorking({ path: sock.path, cap: caps.coder, workspace, tmp: join(root, "tmp-coder"), task_ref: TASK, project: "stack", branch: "feat" });
    return commitId;
  };
  const approve = (role = "coder", task = TASK) => {
    const d = bus.broker.request(caps[role], { verb: "decision.raise", args: {
      action: "git.push.remote", target: task, task_ref: task, question: "Publish?", blocking: false, recommendation: "yes",
      options: [{ key: "yes", text: "Yes" }, { key: "no", text: "No" }] } });
    bus.broker.request(human, { verb: "decision.resolve", args: { id: d.id, choice: "yes" } });
    return d.id;
  };
  const publish = (decision, extra = {}, role = "coder", task = TASK) =>
    raw(sock.path, { cap: caps[role], op: "publish", task_ref: task, project: "stack", branch: "feat", decision, ...extra });
  // The bus holds the writer lock; the test reads its events through a read-only handle.
  const events = (kind, task = TASK) => {
    const db = new DatabaseSync(join(dataRoot, "bus", "bus.sqlite"), { readOnly: true });
    try {
      return db.prepare("SELECT * FROM events WHERE kind=? AND subject=? ORDER BY seq").all(kind, task).map((e) => ({ ...e, body: JSON.parse(e.body) }));
    } finally {
      db.close();
    }
  };
  return { root, bus, gitea, forkRepo, work: await work("coder", TASK), more: work, again, hooks, caps, approve, publish, logs, events };
}

test("T3 end to end: publish under an approved decision sends the work commit to the agent's own fork and records it once", async (t) => {
  const w = await setup(t);
  // Nothing issued: the store has no token, and the refusal is filed once.
  const first = w.approve();
  assert.deepEqual(await w.publish(first), { ok: false, error: "credential-unavailable" });
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["credential-unavailable"]);
  assert.equal(w.gitea.requests.length, 0);

  w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01");
  const reply = await w.publish(first);
  assert.deepEqual(reply, { ok: true, result: { fork: `${w.gitea.origin}/coder/stack`, ref: "refs/heads/coder/feat", commit: w.work, previous: null } });
  assert.equal(git(w.root, `--git-dir=${w.forkRepo}`, "rev-parse", "refs/heads/coder/feat"), w.work);
  // Every request went to the agent's fork; the upstream, mosaic/stack, received none.
  assert.ok(w.gitea.requests.some((r) => r.path === "/coder/stack/git-receive-pack" && r.user === "coder"));
  for (const r of w.gitea.requests) assert.match(r.path, /^\/coder\/stack\//);
  const allowed = w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise");
  assert.equal(allowed.length, 1);
  assert.equal(allowed[0].body.fork, `${w.gitea.origin}/coder/stack`);
  assert.equal(allowed[0].body.commit, w.work);
  // The decision is spent.
  assert.deepEqual(await w.publish(first), { ok: false, error: "decision-consumed" });
  // T4: no token in any answer, event or log line, nor in any file under the scratch root but the
  // agent's own store file.
  assert.equal(JSON.stringify([reply, w.events("action.allowed"), w.events("action.refused"), w.logs]).includes(TOKEN), false);
  const store = join(w.root, "secrets", "agents", "demo", "coder", "gitea");
  const holding = readdirSync(w.root, { recursive: true })
    .map((name) => join(w.root, name))
    .filter((file) => lstatSync(file).isFile() && readFileSync(file).includes(TOKEN));
  assert.deepEqual(holding, [store]);
});

test("T3 end to end: a fork the token can't reach is fork-unavailable, filed once, and nothing is recorded as allowed", async (t) => {
  const w = await setup(t);
  w.bus.credentials.issue("demo/coder", "gitea", "fake_wrong_token_0123456789", "2099-01-01");
  assert.deepEqual(await w.publish(w.approve()), { ok: false, error: "fork-unavailable" });
  assert.ok(w.gitea.requests.length > 0);
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["fork-unavailable"]);
  assert.equal(w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise").length, 0);
});

test("a bad branch or project refuses before the broker; a fork, user or ref in the request is ignored", async (t) => {
  const w = await setup(t);
  const d = w.approve();
  assert.deepEqual(await w.publish(d, { branch: "../main" }), { ok: false, error: "invalid-request" });
  assert.deepEqual(await w.publish(d, { project: "nope" }), { ok: false, error: "invalid-request" });
  // Filed through the socket's refuse op, as a bad push request is; no publish ticket opened.
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["invalid-request", "invalid-request"]);
  assert.equal(w.gitea.requests.length, 0);
  // The broker builds the fork and the destination; the request's own are never read.
  w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01");
  const reply = await w.publish(d, { fork: "https://127.0.0.1:1/other/stack", user: "other", dst: "refs/heads/main", ref: "refs/heads/main" });
  assert.deepEqual(reply.result, { fork: `${w.gitea.origin}/coder/stack`, ref: "refs/heads/coder/feat", commit: w.work, previous: null });
  for (const r of w.gitea.requests) assert.match(r.path, /^\/coder\/stack\//);
  assert.equal(spawnSync("git", [`--git-dir=${w.forkRepo}`, "rev-parse", "-q", "--verify", "refs/heads/other/feat"], { env: gitEnv() }).status, 1);
});

test("T3 end to end: two agents, one project, each lands on its own fork only; the upstream receives nothing", async (t) => {
  const w = await setup(t);
  const second = await w.more("second", TASK2);
  w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01");
  w.bus.credentials.issue("demo/second", "gitea", TOKEN2, "2099-01-01");
  assert.equal((await w.publish(w.approve())).ok, true);
  const coderRequests = w.gitea.requests.length;
  for (const r of w.gitea.requests) assert.deepEqual([r.path.split("/")[1], r.user ?? "coder"], ["coder", "coder"]);
  const reply = await w.publish(w.approve("second", TASK2), {}, "second", TASK2);
  assert.deepEqual(reply.result, { fork: `${w.gitea.origin}/second/stack`, ref: "refs/heads/second/feat", commit: second, previous: null });
  for (const r of w.gitea.requests.slice(coderRequests)) assert.deepEqual([r.path.split("/")[1], r.user ?? "second"], ["second", "second"]);
  const refs = (owner) => git(w.root, `--git-dir=${w.gitea.repo(owner, "stack")}`, "for-each-ref", "--format=%(refname) %(objectname)");
  assert.equal(refs("coder").includes(`refs/heads/coder/feat ${w.work}`), true);
  assert.equal(refs("coder").includes("refs/heads/second/"), false);
  assert.equal(refs("second").includes(`refs/heads/second/feat ${second}`), true);
  assert.equal(refs("second").includes("refs/heads/coder/"), false);
  assert.equal(refs("mosaic").split("\n").length, 1, "the upstream holds only main");
});

// sage-1440 on BUILD.md 14: a push between checkPublish and the git push refuses, with one code
// for each place it can come in, and the fork never receives a commit the decision didn't bind.
test("T3: a push after checkPublish and before the lock refuses ref-moved; the fork receives nothing", async (t) => {
  const w = await setup(t);
  w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01");
  w.hooks.checkPublish = async (m, next) => {
    const facts = await next();
    await w.again();
    return facts;
  };
  assert.deepEqual(await w.publish(w.approve()), { ok: false, error: "ref-moved" });
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["ref-moved"]);
  assert.equal(w.gitea.requests.length, 0);
  assert.equal(git(w.root, `--git-dir=${w.forkRepo}`, "for-each-ref", "refs/heads/coder/"), "");
});

test("T3: a push recorded while the bus process pushes refuses decision-mismatch; only the decision's commit reached the fork", async (t) => {
  const w = await setup(t);
  w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01");
  // The working repo's lock is held through the git push, so a socket push would wait for it; the
  // broker's record of a newer push is what a push that won the lock leaves.
  w.hooks.publishPush = async (m, next) => {
    w.bus.checkPush(w.caps.coder, TASK);
    w.bus.recordPush(w.caps.coder, { task_ref: TASK, project: "stack", branch: "feat", ref: workRef(TASK), commit: "c".repeat(40), previous: w.work });
    return next();
  };
  assert.deepEqual(await w.publish(w.approve()), { ok: false, error: "decision-mismatch" });
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["decision-mismatch"]);
  assert.equal(w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise").length, 0);
  assert.equal(git(w.root, `--git-dir=${w.forkRepo}`, "rev-parse", "refs/heads/coder/feat"), w.work);
  assert.ok(w.logs.some((l) => l.includes("wasn't recorded")));
});

test("sage-1440: the publishResult poll is bounded: publish-timeout, filed once, and nothing recorded", async (t) => {
  const w = await setup(t, { publishDeadlineMs: 150 });
  let polls = 0;
  w.hooks.publishResult = async () => {
    polls++;
    return { state: "running" };
  };
  assert.deepEqual(await w.publish(w.approve()), { ok: false, error: "publish-timeout" });
  assert.ok(polls > 0);
  assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["publish-timeout"]);
  assert.equal(w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise").length, 0);
  assert.ok(w.logs.some((l) => l.includes("publish-timeout")));
});