309 lines
18 KiB
JavaScript
309 lines
18 KiB
JavaScript
// 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 "../../organization/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, orgFixture } 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, f, org } = await orgFixture(t);
|
|
const none = { authority: { withinRole: ["message.send"], crossRole: [] } };
|
|
const businesses = { acme: { id: "acme", human: "jason", arbiters: { technical: 'chief', delivery: 'chief' }, vars: { "bus.tester": true },
|
|
roles: {chief: { definition: 'chief-of-staff', authority: { withinRole: ['decision.resolve.technical', 'decision.resolve.delivery'], crossRole: [] } }, coder: { definition: "coder", ...none }, assistant: { definition: "assistant", ...none } } } };
|
|
const rt = await startBroker({ dataRoot: org.root, businesses, organization: { dir: join(root, ".config", "mosaic-test"), 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"), []);
|
|
});
|