// Row 128 (#1622): mosaic projects, threads and talk --thread against a real broker. // The broker is in process on a real store, with the shipped assistant's messaging lists; the // transport raises what humanTransport raises (a CliError carrying the bus code), so the // exit codes and the one-line refusals are the CLI's own. A run bound to a thread plays the // assistant, and the chief plays a sender outside any thread. import { test } from "node:test"; import assert from "node:assert/strict"; import { randomUUID } from "node:crypto"; 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 { Store } from "../../bus/src/store.mjs"; import { run } from "../src/cli.mjs"; import { CliError } from "../src/errors.mjs"; import { busExit } from "../src/transport.mjs"; import { humanEnv, io } from "./helpers.mjs"; // Row 264a (#1769): a thread made by hand gets the default session the 3k-3l migration gives a // legacy thread, with the thread's id. const DEFAULT_SESSION = [ "INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3l' FROM threads WHERE id=?", "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3l' FROM threads WHERE id=?", ]; 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 } }; function setup(t, { businessRoles = roles } = {}) { const root = mkdtempSync(join(tmpdir(), "mosaic-thread-cli-")); const store = new Store(root); t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); }); const managementReads = []; const b = new Broker({ organization: (business, verb) => { assert.equal(verb, "plugins"); managementReads.push([business, verb]); return { business, revision: "a".repeat(64), plugins: { channel: ["cli", "webui"] } }; }, store, businesses: { demo: { id: "demo", human: "jason", arbiters: { technical: "chief", delivery: "chief" }, roles: businessRoles, definitions, projects: { stack: { root: "/nonexistent" } } } }, }); const caps = { human: b.bindHuman({ business: "demo", human: "jason", via: "cli", outsideAgent: true }) }; // Store rows have a null prototype; the CLI prints and the tests compare plain JSON. 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 transport = () => async (verb, args = {}) => { try { const result = structuredClone(b.request(caps.human, { 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, 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 = async (argv, extra = {}) => { const x = io(); const code = await run([...argv, "--business", "demo"], x, { ...deps, ...extra }); return { code, out: x.out.text, err: x.err.text }; }; return { b, store, call, caps, launch, hooks, cli, writes, deps }; } // Two threads in the project "stack", each with an assistant run holding it, and the chief. function twoThreads(t) { const s = setup(t); const project = s.call("human", "project.create", { name: "stack", repo: "stack" }).id; const A = s.call("human", "thread.create", { project, name: "alpha" }).id; const B = s.call("human", "thread.create", { project, name: "beta" }).id; s.launch("chief", "chief"); s.launch("a", "assistant", { thread: A }); s.launch("b", "assistant", { thread: B }); return { ...s, project, A, B }; } const unread = (s) => Object.fromEntries(s.call("human", "messages", { limit: 200 }).messages.map((m) => [m.body, !m.read])); test("REQ-CLI-1 project create and list work on the bus record and refuse what the broker refuses", async (t) => { const s = setup(t); assert.deepEqual(await s.cli(["project", "list"]), { code: 0, out: "no projects\n", err: "" }); const made = await s.cli(["project", "create", "stack work", "--repo", "stack"]); assert.equal(made.code, 0, made.err); const [row] = s.call("human", "project.list"); assert.match(made.out, new RegExp(`^created project stack work id ${row.id}\n$`)); assert.equal((await s.cli(["project", "create", "loose"])).code, 0); const listed = await s.cli(["project", "list"]); assert.equal(listed.out, `stack work stack ${row.id.slice(0, 8)}\nloose - ${s.call("human", "project.list")[1].id.slice(0, 8)}\n`); const asJson = await s.cli(["project", "list", "--json"]); assert.deepEqual(JSON.parse(asJson.out), s.call("human", "project.list")); const before = s.writes(); const dup = await s.cli(["project", "create", "stack work"]); assert.deepEqual([dup.code, dup.err], [2, "mosaic: project-exists\n"]); const repo = await s.cli(["project", "create", "other", "--repo", "nowhere"]); assert.deepEqual([repo.code, repo.err], [2, "mosaic: unknown-repo\n"]); assert.deepEqual(s.writes(), before); }); test("REQ-CLI-1 thread create, list, open and read", async (t) => { const s = setup(t); await s.cli(["project", "create", "stack", "--repo", "stack"]); await s.cli(["project", "create", "docs"]); const made = await s.cli(["thread", "create", "stack", "assistant"]); assert.equal(made.code, 0, made.err); const [t1] = s.call("human", "thread.list"); assert.equal(made.out, `created stack/assistant id ${t1.id} instance assistant\n`); assert.equal((await s.cli(["thread", "create", "docs", "notes"])).code, 0); const all = await s.cli(["threads"]); const [, t2] = s.call("human", "thread.list"); assert.equal(all.out, `stack/assistant assistant ${t1.id.slice(0, 8)}\ndocs/notes assistant ${t2.id.slice(0, 8)}\n`); assert.equal((await s.cli(["threads", "--project", "docs"])).out, `docs/notes assistant ${t2.id.slice(0, 8)}\n`); assert.deepEqual(JSON.parse((await s.cli(["threads", "--project", "stack", "--json"])).out), [t1]); assert.equal((await s.cli(["threads", "--project", "nothing"])).code, 2); const dup = await s.cli(["thread", "create", "stack", "assistant"]); assert.deepEqual([dup.code, dup.err], [2, "mosaic: thread-exists\n"]); const nobody = await s.cli(["thread", "create", "nothing", "x"]); assert.equal(nobody.code, 2); assert.match(nobody.err, /no project is named or numbered "nothing"/); // open writes the open row the launcher reads; read prints the history and marks nothing read. const opens = s.writes()[1]; assert.equal((await s.cli(["thread", "open", "stack/assistant"])).out, "opened stack/assistant\n"); assert.equal(s.writes()[1], opens + 1); assert.equal((await s.cli(["thread", "read", "stack/assistant"])).out, "stack/assistant: no messages\n"); s.launch("a", "assistant", { thread: t1.id }); const q = s.call("human", "thread.send", { thread: t1.id, body: "first", class: "REQUEST" }); const r = s.call("a", "message.send", { to: "human", body: "second", in_reply_to: q.id }); const deliveries = s.writes()[2]; const read = await s.cli(["thread", "read", "stack/assistant"]); assert.match(read.out, new RegExp(`^human \\(REQUEST\\) ${q.id.slice(0, 8)}:\nfirst\nassistant \\(INFO, re ${q.id.slice(0, 8)}\\) ${r.id.slice(0, 8)}:\nsecond\n$`)); assert.deepEqual(JSON.parse((await s.cli(["thread", "read", t1.id, "--json"])).out).map((m) => m.id), [q.id, r.id]); assert.equal(s.writes()[2], deliveries, "reading wrote no delivery"); assert.deepEqual(unread(s), { second: true }); // A project is named by its id as well. const docs = s.call("human", "project.list").find((p) => p.name === "docs"); assert.match((await s.cli(["thread", "create", docs.id, "ideas"])).out, /^created docs\/ideas id /); assert.equal((await s.cli(["threads", "--project", docs.id])).out.split("\n").length, 3); for (const argv of [["thread"], ["thread", "other"], ["thread", "create", "stack"], ["thread", "open"], ["thread", "read", "a", "b"], ["threads", "extra"], ["thread", "open", "x", "--wat"]]) { assert.equal((await s.cli(argv)).code, 4, argv.join(" ")); } }); test("REQ-CLI-1 thread read pages through every message, oldest first", async (t) => { const s = twoThreads(t); const sent = []; for (let i = 0; i < 205; i++) sent.push(s.call("human", "thread.send", { thread: s.A, body: `note ${i}`, class: "INFO" }).id); const read = JSON.parse((await s.cli(["thread", "read", "stack/alpha", "--json"])).out); assert.deepEqual(read.map((m) => m.id), sent); }); test("REQ-CLI-1 the business needs exactly one threaded instance to create a thread", async (t) => { for (const [roleSet, code] of [ [{ chief: roles.chief }, "thread-instance-missing"], [{ ...roles, second: roles.assistant }, "thread-instance-ambiguous"], ]) { const s = setup(t, { businessRoles: roleSet }); assert.equal((await s.cli(["project", "create", "stack", "--repo", "stack"])).code, 0); const before = s.writes(); const refused = await s.cli(["thread", "create", "stack", "x"]); assert.deepEqual([refused.code, refused.err], [2, `mosaic: ${code}\n`]); assert.deepEqual(s.writes(), before); } }); test("REQ-CLI-1 a thread is its id, an id prefix of 8 characters or more, or /; anything else is refused before any write", async (t) => { const s = twoThreads(t); // Two threads whose ids share their first 8 characters, made by hand. const at = new Date().toISOString(); for (const n of ["1", "2"]) { s.store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", `abcdef01-0000-4000-8000-00000000000${n}`, at, "demo", s.project, "assistant", "jason", `twin${n}`); for (const sql of DEFAULT_SESSION) s.store.run(sql, `abcdef01-0000-4000-8000-00000000000${n}`); } const resolve = async (ref) => (await s.cli(["thread", "open", ref])); const opened = (r) => r.out.trim(); assert.equal(opened(await resolve(s.A)), "opened stack/alpha"); assert.equal(opened(await resolve(s.A.slice(0, 8))), "opened stack/alpha"); assert.equal(opened(await resolve(s.B.slice(0, 12))), "opened stack/beta"); assert.equal(opened(await resolve("stack/beta")), "opened stack/beta"); assert.equal(opened(await resolve("abcdef01-0000-4000-8000-000000000002")), "opened stack/twin2"); const before = s.writes(); const short = await resolve(s.A.slice(0, 7)); assert.deepEqual([short.code, /too short; use at least 8 characters/.test(short.err)], [2, true]); const twins = await resolve("abcdef01"); assert.deepEqual([twins.code, /abcdef01 matches 2 threads/.test(twins.err)], [2, true]); for (const ref of ["stack/nothing", "nothing/alpha", "alpha", "ffffffff"]) { const unknown = await resolve(ref); assert.deepEqual([unknown.code, /^mosaic: no thread of yours matches /.test(unknown.err)], [2, true], ref); } // talk checks the reference before it sends. const talked = await s.cli(["talk", "--thread", "abcdef01", "hello"]); assert.equal(talked.code, 2); assert.deepEqual(s.writes(), before); }); test("REQ-CLI-2 REQ-THREAD-2 talk --thread sends to the thread and prints only the reply to it; the rest stays unread", async (t) => { const s = twoThreads(t); s.call("a", "message.send", { to: "human", body: "older in alpha", class: "INFO" }); s.hooks["thread.send"] = (sent) => { s.call("chief", "message.send", { to: "human", body: "outside any thread", class: "INFO" }); s.call("b", "message.send", { to: "human", body: "beta speaking", class: "INFO" }); s.call("a", "message.send", { to: "human", body: "unrelated in alpha", class: "INFO" }); s.call("a", "message.send", { to: "human", body: "the answer", class: "RESULT", in_reply_to: sent.id }); }; const talked = await s.cli(["talk", "--thread", "stack/alpha", "what is the state?"]); assert.equal(talked.code, 0, talked.err); const [sent] = s.call("human", "thread.messages", { thread: s.A }).filter((m) => m.from_role === "human"); assert.equal(sent.class, "REQUEST"); assert.equal(sent.body, "what is the state?"); assert.equal(talked.out.startsWith(`sent ${sent.id.slice(0, 8)} to stack/alpha\nassistant (RESULT, re ${sent.id.slice(0, 8)})`), true, talked.out); assert.match(talked.out, /\nthe answer\n$/); assert.doesNotMatch(talked.out, /older in alpha|outside any thread|beta speaking|unrelated in alpha/); assert.deepEqual(unread(s), { "older in alpha": true, "outside any thread": true, "beta speaking": true, "unrelated in alpha": true, "the answer": false }); }); test("REQ-CLI-2 talk --thread --wait gives up with exit 1 and leaves the thread's messages unread", async (t) => { const s = twoThreads(t); s.hooks["thread.send"] = () => s.call("a", "message.send", { to: "human", body: "not a reply", class: "INFO" }); const waited = await s.cli(["talk", "--thread", s.A.slice(0, 8), "anyone?", "--wait", "1"]); assert.equal(waited.code, 1); assert.match(waited.err, /^mosaic: no reply from stack\/alpha within 1 s; /); assert.deepEqual(unread(s), { "not a reply": true }); const none = await s.cli(["talk", "--thread", "stack/alpha", "send only", "--wait", "0"]); assert.equal(none.code, 0); assert.match(none.out, /^sent [0-9a-f]{8} to stack\/alpha\n$/); }); test("REQ-CLI-2 talk --thread --follow and follow --thread print and read only that thread's messages", async (t) => { const s = twoThreads(t); const controller = new AbortController(); s.call("a", "message.send", { to: "human", body: "older in alpha", class: "INFO" }); let polls = 0; const base = s.deps.transport; const transport = (business) => { const inner = base(business); return async (verb, args) => { if (verb === "message.receive") { polls++; if (polls === 1) { s.call("a", "message.send", { to: "human", body: "answer", class: "RESULT", in_reply_to: s.call("human", "thread.messages", { thread: s.A }).at(-1).id }); s.call("b", "message.send", { to: "human", body: "beta speaking", class: "INFO" }); s.call("chief", "message.send", { to: "human", body: "outside any thread", class: "INFO" }); } if (polls === 2) s.call("a", "message.send", { to: "human", body: "later in alpha", class: "INFO" }); if (polls === 3) controller.abort(); } return inner(verb, args); }; }; const followed = await s.cli(["talk", "--thread", "stack/alpha", "question", "--follow"], { transport, signal: controller.signal }); assert.equal(followed.code, 0, followed.err); assert.equal(polls, 3); assert.match(followed.out, /answer\n/); assert.match(followed.out, /later in alpha\n/); assert.doesNotMatch(followed.out, /older in alpha|beta speaking|outside any thread/); assert.deepEqual(unread(s), { "older in alpha": true, answer: false, "beta speaking": true, "outside any thread": true, "later in alpha": false }); // follow --thread is the same loop without the send, from the start of the thread: it // prints the older message that is still undelivered, and B's is not its own. const s2 = twoThreads(t); const c2 = new AbortController(); s2.call("a", "message.send", { to: "human", body: "older in alpha", class: "INFO" }); s2.call("b", "message.send", { to: "human", body: "beta speaking", class: "INFO" }); s2.call("chief", "message.send", { to: "human", body: "outside any thread", class: "INFO" }); let n = 0; const base2 = s2.deps.transport; const only = await s2.cli(["follow", "--thread", "stack/alpha"], { signal: c2.signal, transport: (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, /older in alpha\n/); assert.doesNotMatch(only.out, /beta speaking|outside any thread/); assert.equal(s2.writes()[1], 0, "follow opens nothing"); assert.deepEqual(unread(s2), { "older in alpha": false, "beta speaking": true, "outside any thread": true }); assert.equal(s2.call("human", "thread.messages", { thread: s2.A }).length, 1, "follow sends nothing"); }); test("REQ-CLI-2 plain talk to the assistant asks for a thread and exits 2 without a write", async (t) => { const s = twoThreads(t); const before = s.writes(); const plain = await s.cli(["talk", "assistant", "hello"]); assert.equal(plain.code, 2); assert.match(plain.err, /assistant runs in threads/); assert.match(plain.err, /mosaic talk --thread /); assert.match(plain.err, /mosaic threads/); assert.deepEqual(s.writes(), before); for (const argv of [["talk", "--thread", "stack/alpha"], ["talk", "--thread", "stack/alpha", "assistant", "hello", "--wait", "0"], ["talk", "--thread", "stack/alpha", "hi", "--follow", "--wait", "1"]]) { assert.equal((await s.cli(argv)).code, 4, argv.join(" ")); } assert.deepEqual(s.writes(), before); }); test("REQ-USER-2 a thread of another user is refused for talk, open, read and follow, and nothing is written", async (t) => { const s = twoThreads(t); // Another user's project and thread, made by hand: the business has one human, so the bus // has no verb that makes one. const at = new Date().toISOString(); const theirs = randomUUID(); s.store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-theirs", at, "demo", "theirs", "stack", "someone-else"); s.store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", theirs, at, "demo", "p-theirs", "assistant", "someone-else", "private"); for (const sql of DEFAULT_SESSION) s.store.run(sql, theirs); // A message waiting for the user: message.receive would write a delivery row for it. s.call("chief", "message.send", { to: "human", body: "waiting", class: "INFO" }); const before = s.writes(); for (const argv of [["talk", "--thread", theirs, "hello"], ["thread", "open", theirs], ["thread", "read", theirs], ["follow", "--thread", theirs], ["talk", "--thread", theirs, "hello", "--follow"]]) { const refused = await s.cli(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 it isn't in the user's list at all. for (const ref of ["theirs/private", theirs.slice(0, 8)]) { assert.equal((await s.cli(["talk", "--thread", ref, "hello"])).code, 2, ref); } assert.deepEqual(s.writes(), before); // A full id that names no thread is the broker's thread-not-found. const none = await s.cli(["thread", "read", randomUUID()]); assert.deepEqual([none.code, none.err], [2, "mosaic: thread-not-found\n"]); }); test("REQ-CLI-1 the new verbs refuse inside an agent run, as every human verb does", async (t) => { const s = setup(t); const before = s.writes(); for (const argv of [["project", "list"], ["project", "create", "x"], ["threads"], ["thread", "create", "a", "b"], ["thread", "read", "a/b"], ["talk", "--thread", "a/b", "hi"], ["follow", "--thread", "a/b"]]) { assert.equal(await run([...argv, "--business", "demo"], io({ ...humanEnv(), CLAUDECODE: "1" }), s.deps), 3, argv.join(" ")); } assert.deepEqual(s.writes(), before); });