// Row 93 (#1587): the host's git socket inside a real bus host, driven by the
// runner's own git tools (packages/harness/src/git-tools.mjs) over real
// clones. A coder pushes, asks for a review, the reviewer fetches the commit
// and answers, all as the two roles' launches. No sandbox: the socket and the
// tools are what is under test, and each test checks that the project root is
// unchanged and that no bundle file is left in bus-host/ or either tmp.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { createHash, randomBytes } from "node:crypto";
import { connect } from "node:net";
import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, readlinkSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
import { open } from "node:fs/promises";
import { tmpdir } from "node:os";
import { Duplex } from "node:stream";
import { join } from "node:path";
import { Client } from "../../bus/src/client.mjs";
import { configDir, loadBusiness } from "../../organization/src/index.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { gitSocketPath } from "../../seat/src/session.mjs";
import { workingRepo } from "../../seat/src/working.mjs";
import { fetchReview, landWork, pushWorking } from "../../harness/src/git-tools.mjs";
import { gateEnv, isMosaicStack, runGate } from "../../seat/src/landing.mjs";
import { bootConfig, loadSystem, rolesDir } from "../src/config.mjs";
import { CLOSE_WAIT_MS, createGitSocket, LAND_GATE, letGo, send } from "../src/git-socket.mjs";
import { startHost, startTimeOf } from "../src/host.mjs";
import { fixture } from "./helpers.mjs";

const TASK = "vikunja:1/2";

function gitIn(cwd, ...args) {
  const r = spawnSync("git", ["-c", "maintenance.auto=false", "-c", "gc.auto=0", "-c", "user.name=test", "-c", "user.email=test@example.invalid", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
  assert.equal(r.status, 0, r.stderr);
  return r.stdout.trim();
}
function commit(dir, text, file = "file.txt") {
  writeFileSync(join(dir, file), `${text}\n`);
  gitIn(dir, "add", file);
  gitIn(dir, "commit", "-q", "-m", text);
  return gitIn(dir, "rev-parse", "HEAD");
}
function fingerprint(dir) {
  const h = createHash("sha256");
  const walk = (d) => {
    for (const name of readdirSync(d).sort()) {
      const p = join(d, name);
      const st = lstatSync(p);
      h.update(`${p}\0${st.mode}\0`);
      if (st.isDirectory()) walk(p);
      else if (st.isSymbolicLink()) h.update(readlinkSync(p));
      else if (st.isFile()) h.update(readFileSync(p));
    }
  };
  walk(dir);
  return h.digest("hex");
}
const bundles = (dir) => (existsSync(dir) ? readdirSync(dir).filter((n) => n.endsWith(".bundle")) : []);

// A host for business acme, its git socket, and the coder, reviewer and pm
// bound and claimed. Each role has a workspace with a clone of the project.
// The socket sees wrap(host), so a test can change one answer of the host's.
async function setup(t, { wrap = (h) => h, ...git } = {}) {
  // One teardown, in order: stop the socket and the host, check, then remove
  // the tree. (t.after hooks run first-registered first.)
  const root = mkdtempSync(join(tmpdir(), "mosaic-git-"));
  const down = [];
  t.after(async () => {
    try {
      for (const fn of down) await fn();
    } finally {
      rmSync(root, { recursive: true, force: true });
    }
  });
  const f = fixture(root, "acme");
  const project = f.doc.projects.stack.root;
  gitIn(project, "init", "-q", "-b", "main");
  const main = commit(project, "one");
  const before = fingerprint(project);
  const system = loadSystem({ env: f.env });
  const host = await startHost({ boot: bootConfig({ system, businessId: "acme", env: f.env }), business: "acme", log: () => {} });
  down.push(() => host.close(0));
  const logs = [];
  const sock = createGitSocket({ host: wrap(host), business: loadBusiness("acme", { dir: configDir(f.env), rolesDir: rolesDir(f.env) }), dataRoot: f.dataRoot, env: f.env, log: (l) => logs.push(l), ...git });
  down.unshift(() => sock.close());
  await sock.listen();
  const roles = {};
  for (const role of ["coder", "reviewer", "pm", "cto"]) {
    const { cap } = await host.bindLaunch({ business: "acme", role, run: `${role}-test`, harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
    const client = new Client({ path: host.path, cap });
    await client.call("role.claim");
    const workspace = join(root, "ws", role);
    const tmpDir = join(root, "tmp", role);
    mkdirSync(workspace, { recursive: true });
    mkdirSync(tmpDir, { recursive: true });
    gitIn(workspace, "clone", "-q", project, "stack");
    roles[role] = { cap, client, workspace, tmp: tmpDir, clone: join(workspace, "stack") };
  }
  // Checked after every test: the root as it was, and no bundle anywhere.
  down.push(() => {
    assert.equal(fingerprint(project), before, "the project root is unchanged");
    for (const dir of [join(f.dataRoot, "bus-host"), ...Object.values(roles).map((r) => r.tmp)]) assert.deepEqual(bundles(dir), [], `no bundle left in ${dir}`);
  });
  const push = (role, args = {}) => pushWorking({ path: sock.path, cap: roles[role].cap, workspace: roles[role].workspace, tmp: roles[role].tmp, task_ref: TASK, project: "stack", branch: "feat", ...args });
  const fetch = (role, request, args = {}) => fetchReview({ path: sock.path, cap: roles[role].cap, workspace: roles[role].workspace, tmp: roles[role].tmp, task_ref: TASK, request, ...args });
  const repo = workingRepo(f.dataRoot, "acme", "stack");
  const refs = () => (existsSync(repo) ? gitIn(root, `--git-dir=${repo}`, "for-each-ref", "--format=%(refname) %(objectname)") : "");
  const trail = async (kind, subject = TASK) => (await roles.cto.client.call("trail", { subject })).filter((r) => r.table === "events" && r.kind === kind);
  return { root, f, project, main, host, sock, roles, push, fetch, repo, refs, trail, logs };
}

const rejects = (p, code) => assert.rejects(p, (e) => e.code === code || assert.fail(`${e.code}: ${e.message}, wanted ${code}`));

// One raw exchange on the git socket.
function raw(path, text) {
  return new Promise((resolve, reject) => {
    const s = connect(path);
    let buf = "";
    s.on("connect", () => s.write(text));
    s.on("data", (b) => (buf += b));
    s.on("end", () => resolve(JSON.parse(buf.split("\n")[0])));
    s.on("error", reject);
  });
}

// A connection that has sent `text` and waits. hangUp() ends it; reply
// resolves with whatever the host sent once the connection is closed.
function idle(path, text = "") {
  return new Promise((resolve, reject) => {
    const s = connect(path);
    let buf = "";
    s.on("data", (b) => (buf += b));
    s.on("error", () => {});
    const reply = new Promise((r) => s.on("close", () => r(buf)));
    s.on("connect", () => {
      if (text) s.write(text);
      resolve({ reply, hangUp: () => (s.end(), reply) });
    });
    s.once("error", reject);
  });
}

// Polls fn() until it is truthy, for up to 5 s.
async function until(fn, what) {
  for (const end = Date.now() + 5000; Date.now() < end; await new Promise((r) => setTimeout(r, 20))) if (await fn()) return;
  assert.fail(`timed out waiting for ${what}`);
}

const refusedOn = async (w, subject = TASK) => (await w.trail("action.refused", subject)).map((e) => `${e.actor_role} ${e.body.code}`);
const pushReq = (cap, size) => `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size })}\n`;

test("the review loop: the coder pushes and asks, the reviewer fetches and approves, and the task's trail holds each step", async (t) => {
  const w = await setup(t);
  assert.equal(statSync(w.sock.path).mode & 0o777, 0o600);
  assert.equal(w.sock.path, gitSocketPath(w.f.dataRoot));
  const { coder, reviewer } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  const c1 = commit(coder.clone, "two");
  assert.deepEqual(await w.push("coder"), { ref: "work/vikunja-1-2", commit: c1, previous: null });
  assert.equal(gitIn(coder.clone, "rev-parse", "refs/remotes/working/work/vikunja-1-2"), c1);
  assert.equal(w.refs(), `refs/heads/main ${w.main}\nrefs/heads/root ${w.main}\nrefs/heads/work/vikunja-1-2 ${c1}`);
  // Nothing new: the clone remembers the push.
  await rejects(w.push("coder"), "nothing-to-push");

  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
  const got = await w.fetch("reviewer", req.request);
  assert.deepEqual(got, { project: "stack", ref: "working/work/vikunja-1-2", commit: c1, clone: reviewer.clone });
  assert.equal(gitIn(reviewer.clone, "rev-parse", "refs/remotes/working/work/vikunja-1-2"), c1);
  assert.equal(gitIn(reviewer.clone, "show", "-s", "--format=%s", c1), "two");
  const v = await reviewer.client.call("review.verdict", { task_ref: TASK, request: req.request, verdict: "changes", body: "one more" });
  assert.equal(v.commit, c1);

  // The coder answers with a fast-forward; the bundle carries only the new commit.
  const c2 = commit(coder.clone, "three");
  assert.deepEqual(await w.push("coder"), { ref: "work/vikunja-1-2", commit: c2, previous: c1 });
  const req2 = await coder.client.call("review.request", { task_ref: TASK, commit: c2, to: "reviewer" });
  assert.equal((await w.fetch("reviewer", req2.request)).commit, c2);
  const ok = await reviewer.client.call("review.verdict", { task_ref: TASK, request: req2.request, verdict: "approve", body: "good" });

  // A push is the action.allowed for git.push.working that names its commit.
  const pushes = (await w.trail("action.allowed")).filter((e) => e.body.action === "git.push.working");
  assert.deepEqual(pushes.map((e) => [e.actor_role, e.body.commit, e.body.previous, e.body.branch]), [["coder", c1, null, "feat"], ["coder", c2, c1, "feat"]]);
  assert.deepEqual((await w.trail("review.requested")).map((e) => e.id), [req.request, req2.request]);
  assert.deepEqual((await w.trail("review.verdict")).map((e) => [e.id, e.body.verdict, e.body.commit]), [[v.verdict, "changes", c1], [ok.verdict, "approve", c2]]);
  // The reviewer hangs up once it has the bundle, which can beat the host's
  // own end of the send. close() waits for each connection's job, so a fetch
  // filed as anything but done would be on the trail by now.
  await w.sock.close();
  assert.deepEqual(await refusedOn(w), []);
  assert.deepEqual(w.logs, []);
});

test("refusals: no action, non-fast-forward, an unknown commit or request, and a push for another task moves only that task's ref", async (t) => {
  const w = await setup(t);
  const { coder, reviewer, pm } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  const c1 = commit(coder.clone, "two");
  await w.push("coder");
  const refs = w.refs();

  // The reviewer and pm hold no git.push.working; the broker refuses before the host takes a byte.
  for (const role of ["reviewer", "pm"]) {
    gitIn(w.roles[role].clone, "checkout", "-q", "-b", "feat");
    commit(w.roles[role].clone, `${role} work`);
    await rejects(w.push(role), "decision-required");
  }
  assert.equal(w.refs(), refs);

  // A rewritten branch.
  gitIn(coder.clone, "update-ref", "-d", "refs/remotes/working/work/vikunja-1-2");
  gitIn(coder.clone, "reset", "-q", "--hard", w.main);
  commit(coder.clone, "other two");
  await rejects(w.push("coder"), "non-fast-forward");
  assert.equal(w.refs(), refs);

  // The same branch for another task lands on that task's ref and leaves this one.
  const other = await w.push("coder", { task_ref: "vikunja:1/3" });
  assert.equal(other.ref, "work/vikunja-1-3");
  assert.match(w.refs(), new RegExp(`refs/heads/work/vikunja-1-2 ${c1}`));

  await rejects(coder.client.call("review.request", { task_ref: TASK, commit: "e".repeat(40), to: "reviewer" }), "unknown-commit");
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
  await rejects(w.fetch("reviewer", "no-such-request"), "request-not-found");
  await rejects(w.fetch("reviewer", req.request, { task_ref: "vikunja:1/3" }), "request-not-found");
  // The coder and pm hold no review.verdict.
  await rejects(w.fetch("coder", req.request), "not-authorized");
  await rejects(w.fetch("pm", req.request), "not-authorized");
  assert.equal(existsSync(join(reviewer.clone, ".git", "refs", "remotes", "working")), false);

  // Each refusal is an action.refused event on the task, from the role that tried.
  const refused = (await w.trail("action.refused")).map((e) => `${e.actor_role} ${e.body.code}`);
  for (const r of ["reviewer decision-required", "pm decision-required", "coder non-fast-forward", "coder unknown-commit", "reviewer request-not-found", "coder not-authorized", "pm not-authorized"])
    assert.ok(refused.includes(r), `${r} in ${refused.join(", ")}`);
});

test("a scoped task refuses the wrong code project before the socket asks for bundle bytes", async (t) => {
  const { DatabaseSync } = await import("node:sqlite");
  let admission;
  const w = await setup(t, { wrap: host => ({ op: m => { if (m.op === "checkPush") admission = m; return host.op(m); } }) });
  const db = new DatabaseSync(join(w.f.dataRoot, "bus/bus.sqlite"));
  db.prepare("INSERT INTO events(id,at,business,kind,body) VALUES('fixture',?,'acme','human.input','{}')").run(new Date().toISOString());
  db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('scoped-create',?,'acme','task.created',?,?)")
    .run(new Date().toISOString(), TASK, JSON.stringify({ request: "fixture", requirement: "REQ-POMO-1", scope: { project_key: "timer", requirement_snapshot: "prd-fixture", requirement: "REQ-POMO-1" } }));
  db.close();
  const answer = await raw(w.sock.path, JSON.stringify({ op: "push", cap: w.roles.coder.cap, task_ref: TASK, project: "stack", branch: "feat", size: 100 }) + "\n");
  assert.equal(answer.ok, false); assert.equal(answer.error, "task-project-refused"); assert.equal(answer.send, undefined);
  assert.equal(admission.project, "stack"); assert.equal(w.refs(), "");
});

test("the size cap refuses a push before any byte", async (t) => {
  const w = await setup(t, { bundleMax: 64 });
  gitIn(w.roles.coder.clone, "checkout", "-q", "-b", "feat");
  commit(w.roles.coder.clone, "two");
  await rejects(w.push("coder"), "bundle-too-large");
  assert.equal(w.refs(), "");
  assert.deepEqual((await w.trail("action.allowed")).filter((e) => e.body.action === "git.push.working"), []);
});

test("a fetch over the cap refuses before any byte reaches the reviewer", async (t) => {
  const w = await setup(t, { bundleMax: 1024 * 1024 });
  const { coder } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  // Two pushes of 600 KiB of noise each pass a 1 MiB cap; the reviewer has
  // neither, so its bundle carries both and is over.
  writeFileSync(join(coder.clone, "big"), randomBytes(600 * 1024));
  gitIn(coder.clone, "add", "big");
  gitIn(coder.clone, "commit", "-q", "-m", "big");
  const c = (await w.push("coder")).commit;
  writeFileSync(join(coder.clone, "big"), randomBytes(600 * 1024));
  gitIn(coder.clone, "commit", "-q", "-am", "bigger");
  const c2 = (await w.push("coder")).commit;
  assert.notEqual(c, c2);
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c2, to: "reviewer" });
  await rejects(w.fetch("reviewer", req.request), "bundle-too-large");
  assert.equal(existsSync(join(w.roles.reviewer.clone, ".git", "refs", "remotes", "working")), false);
});

test("the socket refuses a request with no capability, an unknown one, and a bad request line", async (t) => {
  const w = await setup(t);
  assert.deepEqual(await raw(w.sock.path, "not json\n"), { ok: false, error: "invalid-request" });
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap: "f".repeat(64), op: "push" })}\n`), { ok: false, error: "unauthenticated" });
  const cap = w.roles.coder.cap;
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "merge" })}\n`), { ok: false, error: "invalid-request" });
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "nope", branch: "feat", size: 10 })}\n`), { ok: false, error: "invalid-request" });
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: "gitea:1", project: "stack", branch: "feat", size: 10 })}\n`), { ok: false, error: "invalid-request" });
  // Bytes before "send" are refused, not read.
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size: 10 })}\n0123456789`), { ok: false, error: "invalid-request" });
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap, op: "fetch", task_ref: TASK, request: "x", have: ["nope"] })}\n`), { ok: false, error: "invalid-request" });
  // A push whose bytes have a bundle's header and then no ref line git can read.
  const junk = await new Promise((resolve, reject) => {
    const s = connect(w.sock.path);
    let buf = "";
    s.on("connect", () => s.write(`${JSON.stringify({ cap, op: "push", task_ref: TASK, project: "stack", branch: "feat", size: 20 })}\n`));
    s.on("data", (b) => {
      buf += b;
      if (buf === `${JSON.stringify({ ok: true, send: 20 })}\n`) s.write("# v2 git bundle\nzzz\n");
    });
    s.on("end", () => resolve(buf.trim().split("\n").pop()));
    s.on("error", reject);
  });
  assert.deepEqual(JSON.parse(junk), { ok: false, error: "invalid-bundle" });
  assert.equal(w.refs(), `refs/heads/main ${w.main}\nrefs/heads/root ${w.main}`);
});

test("a rigged clone runs nothing: hooks, fsmonitor, a filter, a gitlink and a symlinked .git/config leave no marker", async (t) => {
  const w = await setup(t);
  const { coder, reviewer } = w.roles;
  const marks = join(w.root, "marks");
  mkdirSync(marks);
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  commit(coder.clone, "two");
  // A gitlink whose .gitmodules would run a command if anything cloned it.
  // (An `update = !cmd` line fsck refuses outright; the end of the test pushes one.)
  writeFileSync(join(coder.clone, ".gitmodules"), `[submodule "sub"]\n\tpath = sub\n\turl = ext::sh -c touch% ${marks}/submodule\n`);
  gitIn(coder.clone, "add", ".gitmodules");
  gitIn(coder.clone, "update-index", "--add", "--cacheinfo", `160000,${w.main},sub`);
  gitIn(coder.clone, "commit", "-q", "-m", "gitlink");
  writeFileSync(join(coder.clone, ".gitattributes"), "* filter=evil\n");
  gitIn(coder.clone, "add", ".gitattributes");
  gitIn(coder.clone, "commit", "-q", "-m", "attributes");
  const tipCommit = gitIn(coder.clone, "rev-parse", "HEAD");
  // On a second branch, a .gitmodules with a command as its update setting.
  gitIn(coder.clone, "checkout", "-q", "-b", "bad");
  writeFileSync(join(coder.clone, ".gitmodules"), `[submodule "sub"]\n\tpath = sub\n\turl = ./sub\n\tupdate = !touch ${marks}/update\n`);
  gitIn(coder.clone, "add", ".gitmodules");
  gitIn(coder.clone, "commit", "-q", "-m", "update command");
  gitIn(coder.clone, "checkout", "-q", "feat");
  // Rigged only now, so the test's own git calls above ran none of it.
  const rig = (clone, name) => {
    const hooks = join(w.root, `hooks-${name}`);
    mkdirSync(hooks);
    for (const hook of ["reference-transaction", "pre-push", "post-checkout", "post-merge", "post-rewrite", "pre-auto-gc", "fsmonitor-watchman"]) {
      writeFileSync(join(hooks, hook), `#!/bin/sh\ntouch '${marks}/${name}-${hook}'\nexit 0\n`);
      chmodSync(join(hooks, hook), 0o755);
      mkdirSync(join(clone, ".git", "hooks"), { recursive: true });
      writeFileSync(join(clone, ".git", "hooks", hook), `#!/bin/sh\ntouch '${marks}/${name}-local-${hook}'\nexit 0\n`);
      chmodSync(join(clone, ".git", "hooks", hook), 0o755);
    }
    const config = join(w.root, `config-${name}`);
    writeFileSync(
      config,
      `${readFileSync(join(clone, ".git", "config"), "utf8")}[core]\n\thooksPath = ${hooks}\n\tfsmonitor = ${hooks}/fsmonitor-watchman\n[filter "evil"]\n\tclean = touch ${marks}/${name}-clean\n\tsmudge = touch ${marks}/${name}-smudge\n\tprocess = touch ${marks}/${name}-process\n[uploadpack]\n\tpackObjectsHook = touch ${marks}/${name}-pack\n[gc]\n\tauto = 1\n`,
    );
    renameSync(join(clone, ".git", "config"), join(w.root, `config-${name}.orig`));
    symlinkSync(config, join(clone, ".git", "config"));
  };
  rig(coder.clone, "coder");
  rig(reviewer.clone, "reviewer");
  const pushed = await w.push("coder");
  assert.equal(pushed.commit, tipCommit);
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: pushed.commit, to: "reviewer" });
  assert.equal((await w.fetch("reviewer", req.request)).commit, pushed.commit);
  assert.deepEqual(readdirSync(marks), []);
  // The gitlink came across as a tree entry and nothing more.
  assert.match(gitIn(reviewer.clone, "-c", "core.hooksPath=/dev/null", "ls-tree", pushed.commit, "sub"), /^160000 commit /);
  assert.equal(existsSync(join(reviewer.clone, "sub", ".git")), false);
  assert.deepEqual(readdirSync(marks), []);
  // The update command: the host's fsck refuses the bundle and the work ref stays.
  const landed = w.refs();
  await rejects(w.push("coder", { branch: "bad" }), "invalid-bundle");
  assert.equal(w.refs(), landed);
  assert.deepEqual(readdirSync(marks), []);
});

test("an identity with no role, or of another business, refuses not-a-role and moves no ref", async (t) => {
  let who = null;
  const w = await setup(t, { wrap: (host) => ({ op: async (m) => (m.op === "identity" && who ? { ...(await host.op(m)), ...who } : host.op(m)) }) });
  const { coder } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  commit(coder.clone, "two");
  for (const answer of [{ role: null }, { business: "other" }]) {
    who = answer;
    await rejects(w.push("coder"), "not-a-role");
  }
  assert.equal(w.refs(), "");
  assert.deepEqual((await refusedOn(w)).filter((r) => r.endsWith("not-a-role")), ["coder not-a-role", "coder not-a-role"]);
  who = null;
  assert.equal((await w.push("coder")).ref, "work/vikunja-1-2");
});

test("a review of a project the business no longer has refuses unknown-project", async (t) => {
  const w = await setup(t, { wrap: (host) => ({ op: async (m) => (m.op === "reviewFetch" ? { ...(await host.op(m)), project: "gone" } : host.op(m)) }) });
  const { coder } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  const c1 = commit(coder.clone, "two");
  await w.push("coder");
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c1, to: "reviewer" });
  await rejects(w.fetch("reviewer", req.request), "unknown-project");
  assert.ok((await refusedOn(w)).includes("reviewer unknown-project"));
  assert.deepEqual(w.logs, []);
});

test("past maxActive connections the socket refuses git-busy, and a hung-up connection frees its slot", async (t) => {
  const w = await setup(t, { maxActive: 2 });
  const held = [await idle(w.sock.path), await idle(w.sock.path)];
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "git-busy" });
  await held[0].hangUp();
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
  await held[1].hangUp();
});

test("close() answers host-closing to a connection still sending its request line", async (t) => {
  const w = await setup(t);
  const waiting = await idle(w.sock.path, '{"cap":');
  // Served in order, so a full exchange after it means the host took it.
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
  await w.sock.close();
  assert.deepEqual(JSON.parse(await waiting.reply), { ok: false, error: "host-closing" });
});

test("a bundle that trickles in refuses body-timeout at its deadline and frees its slot", async (t) => {
  const w = await setup(t, { maxActive: 2, bodyDeadline: () => 400 });
  const { coder } = w.roles;
  const other = await idle(w.sock.path);
  const started = Date.now();
  // One byte every 50 ms: 1000 bytes would take 50 s.
  const trickled = new Promise((resolve, reject) => {
    const s = connect(w.sock.path);
    let buf = "";
    let drip = null;
    s.on("connect", () => s.write(pushReq(coder.cap, 1000)));
    s.on("data", (b) => {
      buf += b;
      if (!drip && buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`) drip = setInterval(() => s.writable && s.write("x"), 50);
    });
    s.on("close", () => {
      clearInterval(drip);
      resolve({ last: buf.trim().split("\n").pop(), ms: Date.now() - started });
    });
    s.on("error", () => {});
    setTimeout(() => (s.destroy(), reject(new Error("no refusal in 5 s"))), 5000).unref();
  });
  // The trickle holds the second slot while it lasts.
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "git-busy" });
  const { last, ms } = await trickled;
  assert.deepEqual(JSON.parse(last), { ok: false, error: "body-timeout" });
  assert.ok(ms < 5000, `${ms} ms`);
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ op: "push" })}\n`), { ok: false, error: "unauthenticated" });
  assert.ok((await refusedOn(w)).includes("coder body-timeout"));
  assert.equal(w.refs(), "");
  await other.hangUp();
});

test("a reviewer gone before its bundle was sent is filed fetch-aborted, not as the host's fault", async (t) => {
  const w = await setup(t);
  const { coder, reviewer } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  // Noise git can't compress, so the bundle can't all fit in the socket's buffers.
  writeFileSync(join(coder.clone, "big"), randomBytes(4 * 1024 * 1024));
  gitIn(coder.clone, "add", "big");
  gitIn(coder.clone, "commit", "-q", "-m", "big");
  const c = (await w.push("coder")).commit;
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c, to: "reviewer" });
  // Reads the result line and hangs up.
  const line = await new Promise((resolve, reject) => {
    const s = connect(w.sock.path);
    let buf = "";
    s.on("connect", () => s.write(`${JSON.stringify({ cap: reviewer.cap, op: "fetch", task_ref: TASK, request: req.request, have: [] })}\n`));
    s.on("data", (b) => {
      buf += b.toString("latin1");
      const nl = buf.indexOf("\n");
      if (nl >= 0) {
        s.destroy();
        resolve(JSON.parse(buf.slice(0, nl)));
      }
    });
    s.on("error", reject);
  });
  assert.equal(line.ok, true);
  assert.equal(line.result.commit, c);
  await until(async () => (await refusedOn(w)).includes("reviewer fetch-aborted"), "fetch-aborted on the trail");
  assert.ok(!(await refusedOn(w)).some((r) => r.endsWith("git-failed")));
  assert.deepEqual(w.logs, []);
});

test("a reviewer that hangs up the moment it has the whole bundle was served, not aborted", async (t) => {
  const w = await setup(t);
  const { coder, reviewer } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  writeFileSync(join(coder.clone, "big"), randomBytes(1024 * 1024));
  gitIn(coder.clone, "add", "big");
  gitIn(coder.clone, "commit", "-q", "-m", "big");
  const c = (await w.push("coder")).commit;
  const req = await coder.client.call("review.request", { task_ref: TASK, commit: c, to: "reviewer" });
  // Reads the result line and `size` bytes, then destroys the connection at
  // once, which can beat the host's own end of the send. Ten times, since it
  // is a race.
  const take = () =>
    new Promise((resolve, reject) => {
      const s = connect(w.sock.path);
      let buf = Buffer.alloc(0);
      let size = null;
      s.on("connect", () => s.write(`${JSON.stringify({ cap: reviewer.cap, op: "fetch", task_ref: TASK, request: req.request, have: [] })}\n`));
      s.on("data", (b) => {
        buf = Buffer.concat([buf, b]);
        if (size === null) {
          const nl = buf.indexOf(0x0a);
          if (nl < 0) return;
          size = JSON.parse(buf.subarray(0, nl).toString("utf8")).result.size;
          buf = buf.subarray(nl + 1);
        }
        if (buf.length >= size) {
          s.destroy();
          resolve(buf.length);
        }
      });
      s.on("error", reject);
    });
  for (let i = 0; i < 10; i++) assert.ok((await take()) > 0);
  // close() waits for each connection's job.
  await w.sock.close();
  assert.deepEqual(await refusedOn(w), []);
  assert.deepEqual(w.logs, []);
});

// The two closes the socket tests can't order every time, on a stand-in
// socket: the reader gone once every byte left the host but before 'finish',
// and the reader gone while the last chunk was still queued in the host.
test("send: a close after every byte left is served; one that fails a queued write is fetch-aborted", async (t) => {
  const dir = mkdtempSync(join(tmpdir(), "mosaic-send-"));
  t.after(() => rmSync(dir, { recursive: true, force: true }));
  const file = join(dir, "b.bundle");
  writeFileSync(file, "ten bytes!");

  // Takes every write at once and never finishes the end.
  const quick = new Duplex({ read() {}, write: (c, e, cb) => cb(), final() {} });
  quick.on("error", () => {});
  const served = send(quick, file);
  await until(() => quick.writableEnded, "the end");
  quick.destroy();
  await served;

  // Holds the one write, then fails it as a socket whose reader left does.
  let pending = null;
  const slow = new Duplex({ read() {}, writableHighWaterMark: 1, write: (c, e, cb) => (pending = cb), final: (cb) => cb() });
  slow.on("error", () => {});
  const gone = send(slow, file);
  await until(() => pending, "the write");
  pending(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
  await rejects(gone, "fetch-aborted");
});

test("a bundle with no bytes for bodyIdle refuses body-timeout long before its deadline", async (t) => {
  // The deadline is a minute away, so only the idle limit can end it.
  const w = await setup(t, { bodyDeadline: () => 60000, bodyIdle: 300 });
  const { coder } = w.roles;
  const started = Date.now();
  // Sends one byte of 1000 after "send", then nothing.
  const stalled = new Promise((resolve, reject) => {
    const s = connect(w.sock.path);
    let buf = "";
    s.on("connect", () => s.write(pushReq(coder.cap, 1000)));
    s.on("data", (b) => {
      const sent = buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`;
      buf += b;
      if (!sent && buf === `${JSON.stringify({ ok: true, send: 1000 })}\n`) s.write("x");
    });
    s.on("close", () => resolve({ last: buf.trim().split("\n").pop(), ms: Date.now() - started }));
    s.on("error", () => {});
    setTimeout(() => (s.destroy(), reject(new Error("no refusal in 5 s"))), 5000).unref();
  });
  const { last, ms } = await stalled;
  assert.deepEqual(JSON.parse(last), { ok: false, error: "body-timeout" });
  assert.ok(ms >= 300 && ms < 5000, `${ms} ms`);
  assert.ok((await refusedOn(w)).includes("coder body-timeout"));
  assert.equal(w.refs(), "");
});

// A FIFO as the bundle holds send()'s loop open on a read, so a test can end
// the connection between the last chunk and the loop's end: the arm after the
// loop. A plain close there was served; a failed write was fetch-aborted.
test("send: a close once the loop is past its last chunk is served, and a failed write there is fetch-aborted", async (t) => {
  const dir = mkdtempSync(join(tmpdir(), "mosaic-send-"));
  t.after(() => rmSync(dir, { recursive: true, force: true }));
  const fifo = join(dir, "b.fifo");
  assert.equal(spawnSync("mkfifo", [fifo]).status, 0);

  // `fail` is how the stand-in socket ends: destroyed as it is, or by
  // failing its held write as a socket whose reader left does.
  const run = async (fail) => {
    let pending = null;
    const sock = new Duplex({ read() {}, write: (c, e, cb) => (pending = cb), final: (cb) => cb() });
    sock.on("error", () => {});
    const sent = send(sock, fifo);
    // Settled before the FIFO ends, so neither outcome is unhandled.
    const outcome = sent.then(() => "served", (e) => e.code);
    const writer = await open(fifo, "w");
    await writer.write("ten bytes!");
    await until(() => pending, "the write");
    if (fail) pending(Object.assign(new Error("write EPIPE"), { code: "EPIPE" }));
    else sock.destroy();
    await until(() => sock.destroyed, "the socket's end");
    await writer.close();
    return outcome;
  };
  assert.equal(await run(false), "served");
  assert.equal(await run(true), "fetch-aborted");
});

// A peer that never reads leaves the host's last write unfinished; this
// stand-in never finishes one.
test("letGo waits for an answer to leave for at most its cap, then destroys the socket", async (t) => {
  const stuck = new Duplex({ read() {}, write() {}, final() {} });
  stuck.on("error", () => {});
  stuck.end("host-closing\n");
  const quick = new Duplex({ read() {}, write: (c, e, cb) => cb(), final: (cb) => cb() });
  quick.on("error", () => {});
  quick.end("host-closing\n");
  const started = Date.now();
  let timer = null;
  const hung = new Promise((r) => (timer = setTimeout(() => r("hung"), 3000)));
  t.after(() => clearTimeout(timer));
  assert.equal(await Promise.race([letGo([stuck, quick], 300).then(() => "let go"), hung]), "let go");
  const ms = Date.now() - started;
  assert.ok(ms >= 290 && ms < 3000, `${ms} ms`);
  assert.ok(stuck.destroyed && quick.destroyed);
  assert.ok(CLOSE_WAIT_MS > 0 && CLOSE_WAIT_MS <= 10000, `CLOSE_WAIT_MS is ${CLOSE_WAIT_MS}`);
});

test("a record that fails and can't be undone refuses rollback-failed, which the host logs", async (t) => {
  let lock = null;
  const w = await setup(t, {
    wrap: (host) => ({
      op: async (m) => {
        if (m.op !== "recordPush" || !lock) return host.op(m);
        writeFileSync(lock, "");
        throw Object.assign(new Error("not recorded"), { code: "store-failed" });
      },
    }),
  });
  const { coder } = w.roles;
  gitIn(coder.clone, "checkout", "-q", "-b", "feat");
  const c1 = commit(coder.clone, "two");
  lock = join(w.repo, "refs", "heads", "work", "vikunja-1-2.lock");
  await rejects(w.push("coder"), "rollback-failed");
  assert.equal(w.refs(), `refs/heads/main ${w.main}\nrefs/heads/root ${w.main}\nrefs/heads/work/vikunja-1-2 ${c1}`);
  assert.equal(w.logs.length, 1);
  assert.match(w.logs[0], /^git socket: rollback-failed: refs\/heads\/work\/vikunja-1-2 still names /);
  assert.ok((await refusedOn(w)).includes("coder rollback-failed"));
  rmSync(lock);
});

// Row 137 (#1631): the land op. The broker's side is a stand-in: checkLand
// needs a cross-role merge decision its arbiter or the human resolved
// (packages/bus/tests/land.test.mjs covers it). The
// working repository, prepareLand, finishLand and dropLand are real, and the
// gate is a stand-in the test passes as `gate`.
const TICKET = "0b1c2d3e-0000-4000-8000-000000000000";
async function landSetup(t, { gate = {}, ...opts } = {}) {
  const calls = [];
  const state = { facts: null, refuse: null, recordFails: null };
  const w = await setup(t, {
    gate: {
      classify: async (root, env) => (calls.push(["classify", root, env]), true),
      run: async () => ({ ok: true }),
      dropNetwork: async (name, env) => calls.push(["dropNetwork", name, env]),
      ...gate,
    },
    wrap: (host) => ({
      op: async (m) => {
        if (m.op === "checkLand") {
          calls.push(["checkLand", m.args]);
          if (state.refuse) throw Object.assign(new Error(state.refuse), { code: state.refuse });
          return { ...state.facts, ticket: TICKET };
        }
        if (m.op === "recordLand") {
          calls.push(["recordLand", m.ticket, m.land]);
          if (state.recordFails) throw Object.assign(new Error(state.recordFails), { code: state.recordFails });
          return { task_ref: TASK, commit: m.land.commit };
        }
        if (m.op === "landRefused") return calls.push(["landRefused", m.ticket, m.args]), { task_ref: TASK };
        return host.op(m);
      },
    }),
    ...opts,
  });
  gitIn(w.roles.coder.clone, "checkout", "-q", "-b", "feat");
  const c1 = commit(w.roles.coder.clone, "two");
  await w.push("coder");
  state.facts = {
    task_ref: TASK, work_ref: "work/vikunja-1-2", decision: "decision-1", project: "stack", ref: "work/vikunja-1-2", commit: c1,
    verdict: "verdict-1", subject: "Land the parser fix", choice: "yes", by: "jason", via: "cli",
    resolved_at: "2026-10-16T00:00:00.000Z",
  };
  const land = () => landWork({ path: w.sock.path, cap: w.roles.coder.cap, task_ref: TASK, decision: "decision-1" });
  const git = (...args) => gitIn(w.root, `--git-dir=${w.repo}`, ...args);
  // Worktrees other than the bare repository itself.
  const trees = () => git("worktree", "list", "--porcelain").split("\n").filter((l) => l.startsWith("worktree ")).length - 1;
  const of = (kind) => calls.filter(([k]) => k === kind).map(([, ...rest]) => rest);
  const ended = () => until(() => of("dropNetwork").length, "the landing's end");
  return { ...w, calls, state, c1, land, git, trees, of, ended };
}

test("land: the squash runs the gate in its own worktree, then main moves and the broker records it", async (t) => {
  const seen = [];
  const w = await landSetup(t, {
    gate: {
      run: async (a) => {
        seen.push({ ...a, head: gitIn(a.worktree, "rev-parse", "HEAD"), file: readFileSync(join(a.worktree, "file.txt"), "utf8"), link: readlinkSync(join(a.worktree, "node_modules")), status: gitIn(a.worktree, "status", "--porcelain", "--ignored=no") });
        return { ok: true };
      },
    },
  });
  const r = await w.land();
  assert.equal(r.landing, "started");
  assert.equal(r.reviewed, w.c1);
  assert.equal(r.base, w.main);
  assert.match(r.patch_id, /^[0-9a-f]{40}$/);
  await w.ended();

  // main is the squash: one parent, the old main, and the reviewed commit's tree.
  assert.equal(w.git("rev-parse", "refs/heads/main"), r.commit);
  assert.equal(w.git("rev-parse", `${r.commit}^@`), w.main);
  assert.equal(w.git("rev-parse", `${r.commit}^{tree}`), w.git("rev-parse", `${w.c1}^{tree}`));
  const msg = w.git("show", "-s", "--format=%B", r.commit);
  assert.match(msg, /^Land the parser fix\n/);
  assert.ok(msg.includes(" Resolved yes by jason via the CLI at 2026-10-16T00:00:00.000Z.\n"), msg);
  for (const line of [`Decision: decision-1`, `Verdict: verdict-1`, `Reviewed-commit: ${w.c1}`, `Base: main ${w.main}`, `Patch-id: ${r.patch_id}`]) assert.ok(msg.includes(`\n${line}\n`), line);
  // w.git trims, so the last line has no newline after it.
  assert.match(msg, /\nApplied-under: git socket land \(bus host\)$/);

  // The gate ran once, at the squash, with the root's node_modules and the trimmed environment.
  assert.equal(seen.length, 1);
  const [g] = seen;
  assert.equal(g.head, r.commit);
  assert.equal(g.file, "two\n");
  assert.equal(g.link, join(w.project, "node_modules"));
  // Nothing but the link: the fixture project has no .gitignore for it.
  assert.equal(g.status, "?? node_modules");
  assert.deepEqual(g.env, gateEnv(w.f.env));
  assert.equal(g.signal.aborted, false);
  const id = g.worktree.split("/").pop();
  assert.match(id, /^land-[0-9a-f]{12}$/);
  assert.equal(g.worktree, join(w.f.dataRoot, "bus-host", "land", id, id));
  assert.equal(g.logs, join(w.f.dataRoot, "bus-host", "land", id));
  assert.ok(existsSync(join(g.logs, "reviewed.patch")), "the landing's files stay in its log directory");

  assert.deepEqual(w.of("classify"), [[w.project, w.f.env]]);
  assert.deepEqual(w.of("checkLand"), [[{ task_ref: TASK, decision: "decision-1" }]]);
  assert.deepEqual(w.of("recordLand"), [[TICKET, { commit: r.commit, previous: w.main, reviewed: w.c1, base: w.main, patch_id: r.patch_id }]]);
  assert.deepEqual(w.of("landRefused"), []);
  assert.deepEqual(w.of("dropNetwork"), [[`${id}_default`, gateEnv(w.f.env)]]);
  assert.equal(existsSync(g.worktree), false);
  assert.equal(w.trees(), 0);
  assert.deepEqual(await refusedOn(w), []);
  assert.deepEqual(w.logs, []);
});

// Row 137 B1: the squash names the channel that resolved the decision. An arbiter's
// resolution came through the broker, not the CLI.
test("land: an arbiter-resolved landing's squash says the arbiter resolved it through the broker", async (t) => {
  const w = await landSetup(t);
  w.state.facts = { ...w.state.facts, by: "pm-run", via: "broker" };
  const r = await w.land();
  await w.ended();
  const msg = w.git("show", "-s", "--format=%B", r.commit);
  assert.ok(msg.includes(" Resolved yes by pm-run as its arbiter, via the broker at 2026-10-16T00:00:00.000Z.\n"), msg);
  assert.ok(!msg.includes("via the CLI"), msg);
});

test("land: refusals before the gate answer the role at once and leave main and no worktree", async (t) => {
  // Until `fit` goes false the gate fits; then the host's own check decides,
  // and the fixture's project is no mosaic-stack checkout.
  let fit = true;
  const w = await landSetup(t, { gate: { classify: async (root, env) => fit || isMosaicStack(root, env) } });
  assert.equal(LAND_GATE.classify, isMosaicStack);
  assert.equal(LAND_GATE.run, runGate);
  const before = w.refs();
  // The broker's own refusal: it filed it, so the host files nothing more.
  w.state.refuse = "decision-not-approved";
  await rejects(w.land(), "decision-not-approved");
  assert.deepEqual(w.of("landRefused"), []);
  w.state.refuse = null;
  // The work ref isn't where the verdict was.
  w.state.facts = { ...w.state.facts, commit: "e".repeat(40) };
  await rejects(w.land(), "ref-moved");
  w.state.facts = { ...w.state.facts, commit: w.c1 };
  // The project root isn't a mosaic-stack checkout, so it has no gate.
  fit = false;
  await rejects(w.land(), "gate-missing");
  // A project the business no longer has.
  w.state.facts = { ...w.state.facts, project: "gone" };
  await rejects(w.land(), "unknown-project");
  assert.deepEqual(
    w.of("landRefused").map(([ticket, a]) => [ticket, a.code, a.detail, a.notify]),
    [[TICKET, "ref-moved", null, false], [TICKET, "gate-missing", null, false], [TICKET, "unknown-project", null, false]],
  );
  // A request with no decision is the host's own refusal, on the task's trail.
  assert.deepEqual(await raw(w.sock.path, `${JSON.stringify({ cap: w.roles.coder.cap, op: "land", task_ref: TASK })}\n`), { ok: false, error: "invalid-request" });
  assert.ok((await refusedOn(w)).includes("coder invalid-request"));
  assert.equal(w.refs(), before);
  assert.equal(w.trees(), 0);
  assert.deepEqual(w.of("recordLand"), []);
  assert.deepEqual(w.of("dropNetwork"), []);
  assert.deepEqual(w.logs, []);
});

test("land: a failed gate is refused gate-failed with its step, told to the role, and main stays", async (t) => {
  const w = await landSetup(t, { gate: { run: async ({ logs }) => ({ ok: false, step: "pkg-bus", log: join(logs, "pkg-bus.log") }) } });
  const r = await w.land();
  await w.ended();
  const [[ticket, a]] = w.of("landRefused");
  assert.equal(ticket, TICKET);
  assert.equal(a.code, "gate-failed");
  assert.equal(a.notify, true);
  assert.match(a.detail, /^pkg-bus failed, see land\/land-[0-9a-f]{12}\/pkg-bus\.log in the bus host directory$/);
  assert.equal(w.git("rev-parse", "refs/heads/main"), w.main);
  assert.equal(w.git("cat-file", "-t", r.commit), "commit", "the squash was made, and only main didn't move");
  assert.deepEqual(w.of("recordLand"), []);
  assert.equal(w.trees(), 0);
});

// PR #1916 (org-redact5): the detail reaches the role, so it never carries the log's host
// path; a log file name outside the id pattern is dropped too.
test("land: a failed gate's detail names the land directory, never the log's host path", async (t) => {
  const SENTINEL = "sk-test-NOT-A-SECRET-1234";
  let log = null;
  const w = await landSetup(t, { gate: { run: async ({ logs }) => ({ ok: false, step: "pkg-bus", log: (log = join(logs, SENTINEL, `${SENTINEL}.log`)) }) } });
  await w.land();
  await w.ended();
  const [[, a]] = w.of("landRefused");
  assert.equal(a.code, "gate-failed");
  assert.match(a.detail, /^pkg-bus failed, see land\/land-[0-9a-f]{12} in the bus host directory$/);
  assert.ok(!a.detail.includes(SENTINEL) && !a.detail.includes(log) && !a.detail.includes(w.f.dataRoot), a.detail);
});

test("land: a record the broker refuses moves main back, and the landing is refused with the broker's code", async (t) => {
  const w = await landSetup(t);
  w.state.recordFails = "decision-consumed";
  await w.land();
  await w.ended();
  assert.equal(w.of("recordLand").length, 1);
  assert.deepEqual(w.of("landRefused"), [[TICKET, { code: "decision-consumed", detail: null, notify: true }]]);
  assert.equal(w.git("rev-parse", "refs/heads/main"), w.main);
  assert.equal(w.trees(), 0);
  assert.deepEqual(w.logs, []);
});

test("land: a second landing while one runs refuses land-busy, and close() stops the gate and refuses host-closing", async (t) => {
  let started = null;
  const running = new Promise((r) => (started = r));
  const w = await landSetup(t, {
    gate: {
      run: ({ signal }) => {
        started();
        return new Promise((resolve) => signal.addEventListener("abort", () => resolve({ ok: false, aborted: true }), { once: true }));
      },
    },
  });
  await w.land();
  await running;
  await rejects(w.land(), "land-busy");
  assert.ok((await refusedOn(w)).includes("coder land-busy"));
  assert.deepEqual(w.of("checkLand").length, 1, "land-busy refuses before the broker is asked");
  await w.sock.close();
  assert.deepEqual(
    w.of("landRefused").map(([, a]) => a),
    [{ code: "host-closing", detail: "the host closed during the gate; nothing moved", notify: true }],
  );
  assert.equal(w.of("dropNetwork").length, 1);
  assert.equal(w.git("rev-parse", "refs/heads/main"), w.main);
  assert.equal(w.trees(), 0);
});