// 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, existsSync, 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 "../../organization/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", "owner@example.com"); // 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", "coder@example.com"); 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("row 237: two publishes race on one decision; the second refuses publish-busy and exactly one push reaches the fork", async (t) => { const w = await setup(t); w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01"); const d = w.approve(); // The second publish, to another branch, runs while the first holds its ticket. let second; w.hooks.checkPublish = async (m, next) => { const facts = await next(); if (m.args.branch === "feat") second = await w.publish(d, { branch: "other" }); return facts; }; const first = await w.publish(d); assert.deepEqual(second, { ok: false, error: "publish-busy" }); assert.equal(first.ok, true); assert.equal(w.gitea.requests.filter((r) => r.path.endsWith("/git-receive-pack")).length, 1); assert.equal(git(w.root, `--git-dir=${w.forkRepo}`, "for-each-ref", "--format=%(refname)", "refs/heads/coder/"), "refs/heads/coder/feat"); assert.equal(w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise").length, 1); }); 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"))); }); test("row 242: a retry after publish-timeout refuses publish-busy while the bus process's push runs; the late push is recorded and the decision pushes once", async (t) => { const w = await setup(t, { publishDeadlineMs: 150 }); w.bus.credentials.issue("demo/coder", "gitea", TOKEN, "2099-01-01"); // The fork holds the push in pre-receive until the test writes `release`; `held` says it's there. const held = join(w.root, "held"); const release = join(w.root, "release"); writeFileSync(join(w.forkRepo, "hooks", "pre-receive"), `#!/bin/sh\n: > '${held}'\nwhile [ ! -e '${release}' ]; do sleep 0.05; done\ncat > /dev/null\n`, { mode: 0o755 }); const until = async (what, ok) => { for (const end = Date.now() + 10_000; !ok(); ) { if (Date.now() > end) assert.fail(`timed out waiting for ${what}`); await new Promise((r) => setTimeout(r, 20)); } }; const allowed = () => w.events("action.allowed").filter((e) => e.body.action === "git.push.remote" && e.body.operation !== "decision.raise"); // A failed assert must still let the hook end: teardown waits for the fake Gitea, which waits for the push. try { const d = w.approve(); assert.deepEqual(await w.publish(d), { ok: false, error: "publish-timeout" }); await until("the push to reach pre-receive", () => existsSync(held)); // The socket gave up, but the push still runs: the decision is held, and nothing is filed yet. assert.deepEqual(await w.publish(d, { branch: "other" }), { ok: false, error: "publish-busy" }); assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["publish-busy"]); assert.equal(allowed().length, 0); writeFileSync(release, ""); // The push reached the fork, so it's recorded, which spends the decision. await until("the late push to be recorded", () => allowed().length === 1); assert.equal(allowed()[0].body.ref, "refs/heads/coder/feat"); assert.equal(allowed()[0].body.commit, w.work); assert.deepEqual(await w.publish(d, { branch: "other" }), { ok: false, error: "decision-consumed" }); assert.equal(w.gitea.requests.filter((r) => r.path.endsWith("/git-receive-pack")).length, 1); assert.equal(git(w.root, `--git-dir=${w.forkRepo}`, "for-each-ref", "--format=%(refname)", "refs/heads/coder/"), "refs/heads/coder/feat"); assert.deepEqual(w.events("action.refused").map((e) => e.body.code), ["publish-busy", "decision-consumed"]); } finally { writeFileSync(release, ""); } assert.equal(JSON.stringify([w.events("action.allowed"), w.events("action.refused"), w.logs]).includes(TOKEN), false); });