stack/packages/cli/tests/host.test.mjs

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { createServer } from "node:http";
import { once } from "node:events";
import { existsSync, readFileSync, statSync, writeFileSync, mkdirSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { Client } from "../../bus/src/client.mjs";
import { bootConfig, loadSystem, REPO } from "../src/config.mjs";
import { hostDir, hostFile, hostStatus, startHost, startTimeOf, stopHost, watchChildren } from "../src/host.mjs";
import { journalPath } from "../src/notifier.mjs";
import { IDS, makeDeployment } from "../../discord/tests/helpers.mjs";
import { fixture, notifyConfig, OPTIONS, tmp } from "./helpers.mjs";

const CLI = fileURLToPath(new URL("../src/cli.mjs", import.meta.url));
const CHANNEL = "100000000000000900";
const TOKEN = "MTAw.abcdefghijklmnopqrstuvwxyz0123456789";

// A fake Discord REST on 127.0.0.1: opens one DM channel, accepts messages.
async function fakeDiscord(t) {
  const requests = [];
  let n = 0;
  const server = createServer((req, res) => {
    let body = "";
    req.on("data", (b) => (body += b));
    req.on("end", () => {
      requests.push({ method: req.method, url: req.url, body: body ? JSON.parse(body) : null, authorized: req.headers.authorization === `Bot ${TOKEN}` });
      res.setHeader("content-type", "application/json");
      if (req.url === "/users/@me/channels") return res.end(JSON.stringify({ id: CHANNEL, type: 1 }));
      if (req.url === `/channels/${CHANNEL}/messages`) return res.end(JSON.stringify({ id: `30000000000000${String(++n).padStart(4, "0")}` }));
      res.statusCode = 404;
      res.end("{}");
    });
  });
  server.listen(0, "127.0.0.1");
  await once(server, "listening");
  t.after(() => server.close());
  return { base: `http://127.0.0.1:${server.address().port}`, requests, dms: () => requests.filter((r) => r.url.endsWith("/messages") && r.body.nonce.startsWith("dm")) };
}

async function until(fn, ms = 8000) {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    if (fn()) return;
    await new Promise((r) => setTimeout(r, 50));
  }
  throw new Error("timed out waiting");
}

const procText = (pid, what) => {
  try {
    return readFileSync(`/proc/${pid}/${what}`, "utf8");
  } catch {
    return "";
  }
};

test("the host boots the broker, binds a launch in process, and the notifier DMs a blocking decision exactly once", async (t) => {
  const root = tmp(t);
  const f = fixture(root);
  makeDeployment(root, { dmRecipient: IDS.owner });
  const discord = await fakeDiscord(t);
  const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
  assert.equal("trackers" in boot, false);
  const logs = [];
  const host = await startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: (l) => logs.push(l) });
  t.after(() => host.close(0));

  const state = JSON.parse(readFileSync(hostFile(f.dataRoot), "utf8"));
  assert.equal(statSync(hostFile(f.dataRoot)).mode & 0o777, 0o600);
  assert.deepEqual(Object.keys(state).sort(), ["business", "hostVersion", "notifier", "pid", "startTime", "startedAt"]);
  assert.equal(hostStatus(f.dataRoot).host.live, true);

  const launch = await host.bindLaunch({ business: "acme", role: "coder", run: "coder-run", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  assert.equal(launch.run, "coder-run");
  const coder = new Client({ path: host.path, cap: launch.cap });
  await coder.call("role.claim");
  const d = await coder.call("decision.raise", { action: "git.push.protected", target: "refactor", question: "Push?", options: OPTIONS, recommendation: "no", blocking: true, task_ref: "vikunja:1/7" });

  await until(() => discord.dms().length === 1);
  await new Promise((r) => setTimeout(r, 500));
  assert.equal(discord.dms().length, 1, "five more polls send nothing new");
  assert.ok(discord.requests.every((r) => r.authorized));
  assert.match(discord.dms()[0].body.content, new RegExp(`mosaic decide ${d.id.slice(0, 8)}`));

  // No capability in a child's argv or environment, or in the state file.
  for (const pid of Object.values(host.pids)) {
    assert.ok(!procText(pid, "cmdline").includes(launch.cap));
    assert.ok(!procText(pid, "environ").includes(launch.cap));
  }
  assert.ok(!readFileSync(hostFile(f.dataRoot), "utf8").includes(launch.cap));

  assert.equal(await host.close(0), 0);
  const journal = readFileSync(journalPath(f.dataRoot, "acme"), "utf8");
  for (const id of [IDS.owner, CHANNEL, TOKEN]) assert.ok(!journal.includes(id));
  assert.equal(journal.trim().split("\n").filter((l) => JSON.parse(l).kind === "dm").length, 1);
  assert.equal(existsSync(hostFile(f.dataRoot)), false);
  assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
});

test("a notifier that dies takes the host down with exit 1, so the unit restarts the pair", async (t) => {
  const root = tmp(t);
  const f = fixture(root);
  makeDeployment(root, { dmRecipient: IDS.owner });
  const discord = await fakeDiscord(t);
  const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
  const logs = [];
  const host = await startHost({ boot, business: "acme", notifier: { binding: "test-seat", base: discord.base, pollMs: 100 }, log: (l) => logs.push(l) });
  process.kill(host.pids.notifier, "SIGKILL");
  assert.equal(await host.done, 1);
  assert.match(logs.join("\n"), /notifier exited \(SIGKILL\); stopping the host/);
  assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
  assert.equal(existsSync(hostFile(f.dataRoot)), false);
});

test("a notifier that refuses stops the broker and the host refuses with exit 3", async (t) => {
  const root = tmp(t);
  const f = fixture(root);
  makeDeployment(root);
  const boot = bootConfig({ system: loadSystem({ env: f.env }), businessId: "acme", env: f.env });
  const started = startHost({ boot, business: "acme", notifier: { binding: "test-seat" }, log: () => {} });
  // If the refusal regresses, the host starts; close it so the file still ends.
  t.after(async () => (await started.catch(() => null))?.close(0));
  await assert.rejects(started, (e) => e.exitCode === 3 && /no dmRecipient/.test(e.message));
  assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
  assert.equal(existsSync(hostFile(f.dataRoot)), false);
});

test("watchChildren reports a child that died before it was called, and one that dies later", async (t) => {
  const early = spawn(process.execPath, ["-e", "process.exit(7)"], { stdio: "ignore" });
  await once(early, "exit");
  const late = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { stdio: "ignore" });
  t.after(() => late.kill("SIGKILL"));
  await once(late, "spawn");
  const deaths = [];
  watchChildren({ broker: early, notifier: late, none: null }, (...d) => deaths.push(d));
  assert.deepEqual(deaths, [["broker", 7, null]], "the exit before the watch is not lost");
  late.kill("SIGTERM");
  await once(late, "exit");
  assert.deepEqual(deaths, [["broker", 7, null], ["notifier", null, "SIGTERM"]]);
});

test("bus stop refuses to signal a live pid that is not a bus host", async (t) => {
  const dataRoot = tmp(t);
  const child = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60000)"], { stdio: "ignore" });
  t.after(() => child.kill("SIGKILL"));
  await once(child, "spawn");
  mkdirSync(hostDir(dataRoot), { recursive: true, mode: 0o700 });
  writeFileSync(hostFile(dataRoot), JSON.stringify({ pid: child.pid, startTime: startTimeOf(child.pid), business: "acme" }), { mode: 0o600 });
  await assert.rejects(stopHost(dataRoot, { timeoutMs: 1000 }), (e) => e.exitCode === 3 && /is not a bus host; refusing to signal it/.test(e.message));
  await new Promise((r) => setTimeout(r, 200));
  assert.equal(child.exitCode, null);
  assert.equal(child.signalCode, null, "the child was not signalled");
});

test("bus start refuses with exit 3 and the code when the broker refuses to boot; bus status names the lock", (t) => {
  const f = fixture(tmp(t));
  notifyConfig(f.dataRoot, "acme", null);
  mkdirSync(join(f.dataRoot, "bus"), { recursive: true, mode: 0o700 });
  writeFileSync(join(f.dataRoot, "bus", "writer.lock"), JSON.stringify({ pid: 999999999, at: "2026-10-08T00:00:00Z" }), { mode: 0o600 });
  const r = spawnSync(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, encoding: "utf8", timeout: 40000 });
  assert.equal(r.status, 3, r.stderr);
  assert.match(r.stderr, /broker refused to start: startup-refused/);
  const s = spawnSync(process.execPath, [CLI, "bus", "status"], { env: f.env, encoding: "utf8" });
  assert.equal(s.status, 0, s.stderr);
  assert.match(s.stdout, /host: none/);
  assert.match(s.stdout, /writer\.lock: pid 999999999 \(not running/);
});

test("bus start refuses with exit 3 without a notifier config", (t) => {
  const f = fixture(tmp(t));
  const r = spawnSync(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, encoding: "utf8", timeout: 40000 });
  assert.equal(r.status, 3);
  assert.match(r.stderr, /no notifier config/);
  assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
});

test("bus start runs until bus stop; status reports it while it runs", async (t) => {
  const f = fixture(tmp(t));
  notifyConfig(f.dataRoot, "acme", null);
  const child = spawn(process.execPath, [CLI, "bus", "start", "acme"], { env: f.env, stdio: ["ignore", "pipe", "pipe"] });
  t.after(() => child.exitCode === null && child.kill("SIGKILL"));
  let out = "";
  child.stdout.on("data", (b) => (out += b));
  child.stderr.on("data", (b) => (out += b));
  await until(() => /bus host up: business acme/.test(out), 30000);
  const status = spawnSync(process.execPath, [CLI, "bus", "status", "--json"], { env: f.env, encoding: "utf8" });
  const s = JSON.parse(status.stdout);
  assert.equal(s.host.live, true);
  assert.equal(s.host.pid, child.pid);
  assert.equal(s.host.notifier, null);
  assert.equal(s.socket, true);
  assert.equal(s.writerLock.live, true);
  const exit = once(child, "exit");
  const stop = spawnSync(process.execPath, [CLI, "bus", "stop"], { env: f.env, encoding: "utf8", timeout: 70000 });
  assert.equal(stop.status, 0, stop.stderr);
  assert.match(stop.stdout, new RegExp(`stopped bus host for acme \\(pid ${child.pid}\\)`));
  assert.equal((await exit)[0], 0, out);
  assert.equal(existsSync(join(f.dataRoot, "bus", "writer.lock")), false);
  const again = spawnSync(process.execPath, [CLI, "bus", "stop"], { env: f.env, encoding: "utf8" });
  assert.match(again.stdout, /no host runs/);
});

test("bus-service.sh renders the unit and installs it into a given directory", (t) => {
  const dir = tmp(t);
  const script = join(REPO, "scripts", "bus-service.sh");
  const render = spawnSync(script, ["render"], { encoding: "utf8" });
  assert.equal(render.status, 0, render.stderr);
  assert.match(render.stdout, new RegExp(`ExecStart=${REPO.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}/scripts/mosaic bus start %i`));
  assert.match(render.stdout, /RestartPreventExitStatus=2 3 4/);
  assert.doesNotMatch(render.stdout, /@REPO@|@PATH@/);
  // A user unit cannot order on a system target (Darkwing F3).
  assert.doesNotMatch(render.stdout, /network-online/);
  const first = spawnSync(script, ["install", "--dir", dir, "--no-reload"], { encoding: "utf8" });
  assert.equal(first.status, 0, first.stderr);
  assert.match(first.stdout, /written: /);
  assert.equal(readFileSync(join(dir, "[email protected]"), "utf8"), render.stdout);
  assert.match(spawnSync(script, ["install", "--dir", dir, "--no-reload"], { encoding: "utf8" }).stdout, /unchanged: /);
  assert.match(spawnSync(script, ["uninstall", "--dir", dir, "--no-reload"], { encoding: "utf8" }).stdout, /removed: /);
  assert.equal(spawnSync(script, ["bogus"], { encoding: "utf8" }).status, 4);
});