stack/packages/cli/tests/tester-threads.test.mjs

// Row 184 (#1682, Q173): the CLI's thread verbs with --tester. The broker is in process on a real
// store, as in thread-cli.test.mjs, with the human and the tester each behind a stand-in transport
// that records its verbs. Tester forms run with an agent marker set, as Q173's seats run them. A
// run bound to a thread plays the assistant, and the chief plays a role outside any thread. The
// last test runs scripts/mosaic as its own process, so the tester-cli.mjs it spawns is the real one.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { BusError, Broker } from "../../bus/src/broker.mjs";
import { startBroker } from "../../bus/src/runtime.mjs";
import { Store } from "../../bus/src/store.mjs";
import { systemFor } from "../../business/tests/helpers.mjs";
import { run } from "../src/cli.mjs";
import { CliError } from "../src/errors.mjs";
import { busExit } from "../src/transport.mjs";
import { fixture, humanEnv, io, tmp } from "./helpers.mjs";

const shipped = (name) => JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), "utf8"));
const roles = {
  chief: { definition: "chief-of-staff", authority: { withinRole: ["message.send"], crossRole: [] } },
  assistant: { definition: "assistant", authority: { withinRole: ["message.send"], crossRole: [] } },
};
const definitions = { "chief-of-staff": {}, assistant: { messaging: shipped("assistant").messaging } };
const inAgent = () => humanEnv({ CLAUDECODE: "1", CLAUDE_CODE_ENTRYPOINT: "sdk-ts" });

function setup(t) {
  const root = mkdtempSync(join(tmpdir(), "mosaic-tester-threads-"));
  const store = new Store(root);
  t.after(() => {
    store.close();
    rmSync(root, { recursive: true, force: true });
  });
  const b = new Broker({
    organization: (business, verb) => {
      assert.equal(verb, "plugins");
      return { business, revision: "a".repeat(64), plugins: { channel: ["cli", "webui"] } };
    },
    store,
    businesses: { demo: { id: "demo", human: "jason", vars: { "bus.tester": true }, arbiters: { technical: "chief", delivery: "chief" }, roles, definitions, projects: { stack: { root: "/nonexistent" } } } },
  });
  const caps = {
    human: b.bindHuman({ business: "demo", human: "jason", via: "cli", outsideAgent: true }),
    tester: b.bindTester({ business: "demo" }),
  };
  const call = (who, verb, args = {}) => JSON.parse(JSON.stringify(b.request(caps[who], { verb, args })));
  const launch = (name, role, extra = {}) => {
    caps[name] = b.bindLaunch({ business: "demo", role, run: `${name}-run`, harness: "pi", ...extra });
    call(name, "role.claim");
  };
  const hooks = {};
  const calls = { human: [], tester: [] };
  const via = (who) => () => async (verb, args = {}) => {
    calls[who].push(verb);
    try {
      const result = structuredClone(b.request(caps[who], { verb, args }));
      hooks[verb]?.(result, args);
      return result;
    } catch (e) {
      if (!(e instanceof BusError)) throw e;
      const error = new CliError(e.code, busExit(e.code));
      error.code = e.code;
      throw error;
    }
  };
  const deps = { system: { dataRoot: root }, transport: via("human"), testerTransport: via("tester"), talkPollMs: 5 };
  const count = (table) => store.get(`SELECT count(*) n FROM ${table}`).n;
  // The rows a refused command must not have written.
  const writes = () => ["messages", "thread_opens", "deliveries", "message_threads", "threads", "projects"].map(count);
  const cli = (env) => async (argv, extra = {}) => {
    const x = io(env);
    const code = await run([...argv, "--business", "demo"], x, { ...deps, ...extra });
    return { code, out: x.out.text, err: x.err.text };
  };
  const human = cli(humanEnv());
  const testerCli = cli(inAgent());
  const tester = (argv, extra) => testerCli([...argv, "--tester"], extra);
  const project = call("human", "project.create", { name: "stack", repo: "stack" }).id;
  return { b, store, call, launch, hooks, calls, deps, writes, human, tester, project };
}

// Each message to the tester, by body: true once it has been marked read.
const testerRead = (s) => Object.fromEntries(s.store.all(
  "SELECT body, EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='read') AS r FROM messages m WHERE to_role='tester' ORDER BY seq",
).map((m) => [m.body, m.r === 1]));
const byBody = (s, body) => s.store.get("SELECT * FROM messages WHERE body=?", body);

test("row 184: with --tester, thread create, threads, open, talk --thread and read run on the tester's own thread", async (t) => {
  const s = setup(t);
  const made = await s.tester(["thread", "create", "stack", "probe"]);
  assert.equal(made.code, 0, made.err);
  const row = s.store.get("SELECT id, owner, instance FROM threads WHERE name='probe'");
  assert.deepEqual([row.owner, row.instance], ["tester", "assistant"]);
  assert.equal(made.out, `created stack/probe  id ${row.id}  instance assistant\n`);
  assert.equal((await s.tester(["threads"])).out, `stack/probe  assistant  ${row.id.slice(0, 8)}\n`);
  assert.deepEqual(JSON.parse((await s.tester(["threads", "--json"])).out).map((r) => r.id), [row.id]);

  const opened = await s.tester(["thread", "open", "stack/probe"]);
  assert.deepEqual([opened.code, opened.out], [0, "opened stack/probe\n"]);
  assert.deepEqual(s.store.all("SELECT thread, \"by\" FROM thread_opens").map((r) => [r.thread, r.by]), [[row.id, "tester"]]);

  s.launch("a", "assistant", { thread: row.id });
  s.hooks["thread.send"] = (sent) => s.call("a", "message.send", { to: "tester", body: "the answer", class: "RESULT", in_reply_to: sent.id });
  const talked = await s.tester(["talk", "--thread", row.id.slice(0, 8), "what is the state?"]);
  assert.equal(talked.code, 0, talked.err);
  const q = byBody(s, "what is the state?");
  const r = byBody(s, "the answer");
  assert.deepEqual([q.from_role, q.from_run, q.to_role, q.class], ["tester", "tester-cli", "assistant", "REQUEST"]);
  assert.equal(talked.out, `sent ${q.id.slice(0, 8)} to stack/probe\nassistant (RESULT, re ${q.id.slice(0, 8)}) ${r.id.slice(0, 8)}:\nthe answer\n`);
  assert.deepEqual(testerRead(s), { "the answer": true });

  const read = await s.tester(["thread", "read", "stack/probe"]);
  assert.equal(read.out, `tester (REQUEST) ${q.id.slice(0, 8)}:\nwhat is the state?\nassistant (RESULT, re ${q.id.slice(0, 8)}) ${r.id.slice(0, 8)}:\nthe answer\n`);
  assert.deepEqual(JSON.parse((await s.tester(["thread", "read", row.id, "--json"])).out).map((m) => m.id), [q.id, r.id]);
  assert.equal((await s.tester(["talk", "--thread", "stack/probe", "send only", "--wait", "0"])).out, `sent ${byBody(s, "send only").id.slice(0, 8)} to stack/probe\n`);

  // Every call went through the tester's transport, and the human sees none of it.
  assert.deepEqual(s.calls.human, []);
  assert.ok(!s.calls.tester.includes("inbox"));
  assert.equal((await s.human(["threads"])).out, "no threads\n");
  assert.deepEqual(s.call("human", "messages", { limit: 200 }).messages, []);
});

test("row 184: talk --thread --follow and follow --thread with --tester print and read only that thread's messages, from message.receive", async (t) => {
  const s = setup(t);
  const P = s.call("tester", "thread.create", { project: s.project, name: "probe" }).id;
  const O = s.call("tester", "thread.create", { project: s.project, name: "other" }).id;
  s.launch("chief", "chief");
  s.launch("a", "assistant", { thread: P });
  s.launch("o", "assistant", { thread: O });
  // Replies to the tester outside any thread and in its other thread, still undelivered.
  const ping = s.call("tester", "message.send", { to: "chief", body: "ping", class: "REQUEST" }).id;
  s.call("chief", "message.send", { to: "tester", body: "outside any thread", in_reply_to: ping });
  const asked = s.call("tester", "thread.send", { thread: O, body: "other question", class: "REQUEST" }).id;
  s.call("o", "message.send", { to: "tester", body: "other answer", in_reply_to: asked });

  const controller = new AbortController();
  let polls = 0;
  const base = s.deps.testerTransport;
  const testerTransport = (business) => {
    const inner = base(business);
    return async (verb, args) => {
      if (verb === "message.receive") {
        polls++;
        const question = byBody(s, "question")?.id;
        if (polls === 1) s.call("a", "message.send", { to: "tester", body: "answer", class: "RESULT", in_reply_to: question });
        if (polls === 2) s.call("a", "message.send", { to: "tester", body: "later in probe", class: "INFO", in_reply_to: question });
        if (polls === 3) controller.abort();
      }
      return inner(verb, args);
    };
  };
  const followed = await s.tester(["talk", "--thread", "stack/probe", "question", "--follow"], { testerTransport, signal: controller.signal });
  assert.equal(followed.code, 0, followed.err);
  assert.equal(polls, 3);
  assert.match(followed.out, /\nanswer\n/);
  assert.match(followed.out, /\nlater in probe\n$/);
  assert.doesNotMatch(followed.out, /outside any thread|other answer/);
  assert.deepEqual(testerRead(s), { "outside any thread": false, "other answer": false, answer: true, "later in probe": true });
  assert.ok(s.calls.tester.includes("message.receive"));
  assert.ok(!s.calls.tester.includes("inbox"));
  assert.deepEqual(s.calls.human, []);

  // follow --thread is the same loop without the send, from the start of the thread.
  const s2 = setup(t);
  const P2 = s2.call("tester", "thread.create", { project: s2.project, name: "probe" }).id;
  const O2 = s2.call("tester", "thread.create", { project: s2.project, name: "other" }).id;
  s2.launch("chief", "chief");
  s2.launch("a", "assistant", { thread: P2 });
  s2.launch("o", "assistant", { thread: O2 });
  const mine = s2.call("tester", "thread.send", { thread: P2, body: "earlier question", class: "REQUEST" }).id;
  s2.call("a", "message.send", { to: "tester", body: "older in probe", in_reply_to: mine });
  const theirs = s2.call("tester", "thread.send", { thread: O2, body: "other question", class: "REQUEST" }).id;
  s2.call("o", "message.send", { to: "tester", body: "other answer", in_reply_to: theirs });
  const ping2 = s2.call("tester", "message.send", { to: "chief", body: "ping", class: "REQUEST" }).id;
  s2.call("chief", "message.send", { to: "tester", body: "outside any thread", in_reply_to: ping2 });
  const c2 = new AbortController();
  let n = 0;
  const base2 = s2.deps.testerTransport;
  const only = await s2.tester(["follow", "--thread", "stack/probe"], {
    signal: c2.signal,
    testerTransport: (business) => {
      const inner = base2(business);
      return async (verb, args) => {
        if (verb === "message.receive" && ++n === 2) c2.abort();
        return inner(verb, args);
      };
    },
  });
  assert.equal(only.code, 0, only.err);
  assert.match(only.out, /\nolder in probe\n/);
  assert.doesNotMatch(only.out, /other answer|outside any thread/);
  assert.deepEqual(testerRead(s2), { "older in probe": true, "other answer": false, "outside any thread": false });
  assert.equal(s2.writes()[1], 0, "follow opens nothing");
  assert.equal(s2.call("tester", "thread.messages", { thread: P2 }).length, 2, "follow sends nothing");
  assert.deepEqual(s2.calls.human, []);
});

test("row 184: with --tester a human thread refuses with the bus's code, and nothing is written", async (t) => {
  const s = setup(t);
  const H = s.call("human", "thread.create", { project: s.project, name: "alpha" }).id;
  s.launch("chief", "chief");
  s.launch("a", "assistant", { thread: H });
  s.call("human", "thread.send", { thread: H, body: "the human's", class: "REQUEST" });
  // A reply waiting for the tester: message.receive would write a delivery row for it.
  const ping = s.call("tester", "message.send", { to: "chief", body: "ping", class: "REQUEST" }).id;
  s.call("chief", "message.send", { to: "tester", body: "waiting", in_reply_to: ping });
  const before = s.writes();
  const history = s.call("human", "thread.messages", { thread: H });
  for (const argv of [["thread", "open", H], ["thread", "read", H], ["talk", "--thread", H, "hello"], ["follow", "--thread", H], ["talk", "--thread", H, "hello", "--follow"]]) {
    const refused = await s.tester(argv, { signal: AbortSignal.timeout(500) });
    assert.deepEqual([refused.code, refused.err, refused.out], [2, "mosaic: thread-not-owned\n", ""], argv.join(" "));
  }
  // By name or prefix the human's thread isn't in the tester's list at all.
  for (const ref of ["stack/alpha", H.slice(0, 8)]) {
    const unknown = await s.tester(["thread", "read", ref]);
    assert.deepEqual([unknown.code, /^mosaic: no thread of yours matches /.test(unknown.err)], [2, true], ref);
  }
  // A thread name is unique in its project whoever owns it.
  const dup = await s.tester(["thread", "create", "stack", "alpha"]);
  assert.deepEqual([dup.code, dup.err], [2, "mosaic: thread-exists\n"]);
  assert.equal((await s.tester(["threads"])).out, "no threads\n");
  assert.deepEqual(s.writes(), before);
  assert.deepEqual(s.call("human", "thread.messages", { thread: H }), history);
  assert.deepEqual(testerRead(s), { waiting: false });
  assert.deepEqual(s.calls.human, []);

  // The other way round as well: the human doesn't reach the tester's thread.
  const T = s.call("tester", "thread.create", { project: s.project, name: "probe" }).id;
  const back = await s.human(["thread", "read", T]);
  assert.deepEqual([back.code, back.err], [2, "mosaic: thread-not-owned\n"]);
});

test("row 184: without --tester the thread verbs stay the human's and never call the tester transport", async (t) => {
  const s = setup(t);
  const made = await s.human(["thread", "create", "stack", "alpha"]);
  assert.equal(made.code, 0, made.err);
  const [row] = s.call("human", "thread.list");
  assert.equal(made.out, `created stack/alpha  id ${row.id}  instance assistant\n`);
  assert.equal((await s.human(["threads"])).out, `stack/alpha  assistant  ${row.id.slice(0, 8)}\n`);
  assert.equal((await s.human(["thread", "open", "stack/alpha"])).out, "opened stack/alpha\n");
  s.launch("a", "assistant", { thread: row.id });
  s.hooks["thread.send"] = (sent) => s.call("a", "message.send", { to: "human", body: "the answer", class: "RESULT", in_reply_to: sent.id });
  const talked = await s.human(["talk", "--thread", "stack/alpha", "hi"]);
  const q = byBody(s, "hi");
  assert.deepEqual([q.from_role, q.from_run], ["human", "human-cli"]);
  assert.equal(talked.out, `sent ${q.id.slice(0, 8)} to stack/alpha\nassistant (RESULT, re ${q.id.slice(0, 8)}) ${byBody(s, "the answer").id.slice(0, 8)}:\nthe answer\n`);
  assert.match((await s.human(["thread", "read", "stack/alpha"])).out, /^human \(REQUEST\) /);
  const c = new AbortController();
  const followed = await s.human(["follow", "--thread", "stack/alpha"], { signal: c.signal, talkPollMs: 1, transport: (business) => {
    const inner = s.deps.transport(business);
    return async (verb, args) => {
      if (verb === "message.receive") c.abort();
      return inner(verb, args);
    };
  } });
  assert.equal(followed.code, 0, followed.err);
  assert.deepEqual(s.calls.tester, []);
});

// The real tester transport: scripts/mosaic runs as its own process and spawns tester-cli.mjs,
// which reaches the runtime in this one, as tester-reads.test.mjs's W7 does.
const MOSAIC = fileURLToPath(new URL("../../../scripts/mosaic", import.meta.url));
function mosaic(args, env) {
  const clean = Object.fromEntries(Object.entries(env).filter(([k]) => !/^MOSAIC_BUS_(TESTER|CLI)_NONCE$/.test(k)));
  return new Promise((resolve, reject) => {
    const child = spawn(MOSAIC, args, { env: clean, timeout: 60000 });
    let out = "";
    let err = "";
    child.stdout.on("data", (d) => (out += d));
    child.stderr.on("data", (d) => (err += d));
    child.on("error", reject);
    child.on("close", (status) => resolve({ status, out, err }));
  });
}

test("row 184: scripts/mosaic runs thread create, threads, talk --thread and thread read as the tester through tester-cli.mjs", async (t) => {
  const root = tmp(t, "mosaic-cli-tester-threads-");
  const f = fixture(root, "acme");
  const none = { authority: { withinRole: ["message.send"], crossRole: [] } };
  const businesses = { acme: { id: "acme", human: "jason", arbiters: { technical: "coder", delivery: "coder" }, vars: { "bus.tester": true },
    roles: { coder: { definition: "coder", ...none }, assistant: { definition: "assistant", ...none } } } };
  const rt = await startBroker({ dataRoot: f.dataRoot, businesses, organization: { dir: join(root, "config"), rolesDir: f.roles, system: systemFor(root) } });
  t.after(() => rt.close());
  const human = rt.broker.bindHuman({ business: "acme", human: "jason", via: "cli", outsideAgent: true });
  const call = (verb, args = {}) => rt.broker.request(human, { verb, args });
  call("plugins.set", { revision: call("plugins").revision, plugins: { channel: ["cli", "webui"] } });
  call("project.create", { name: "stack" });
  const agent = { ...f.env, CLAUDECODE: "1", CLAUDE_CODE_ENTRYPOINT: "sdk-ts" };
  const as = (args) => mosaic([...args, "--business", "acme", "--tester"], agent);

  const made = await as(["thread", "create", "stack", "probe"]);
  assert.equal(made.status, 0, made.err);
  assert.match(made.out, /^created stack\/probe {2}id [0-9a-f-]{36} {2}instance assistant\n$/);
  const listed = await as(["threads", "--json"]);
  assert.deepEqual(JSON.parse(listed.out).map((r) => r.name), ["probe"]);
  const talked = await as(["talk", "--thread", "stack/probe", "hello", "--wait", "0"]);
  assert.equal(talked.status, 0, talked.err);
  assert.match(talked.out, /^sent [0-9a-f]{8} to stack\/probe\n$/);
  const read = await as(["thread", "read", "stack/probe", "--json"]);
  assert.equal(read.status, 0, read.err);
  assert.deepEqual(JSON.parse(read.out).map((m) => [m.from_role, m.from_run, m.body]), [["tester", "tester-cli", "hello"]]);
  // The human's list leaves the tester's thread out.
  assert.deepEqual(call("thread.list"), []);
});