cli: mosaic makes projects and threads and talks in a named thread (row 128, #1622)

Row 128 round 1, Researcher's candidate (agents/researcher/work/queue-128/,
row128.patch, manifest 9eed6a78, 8 paths, base 0f3eb25ea). Filbert
approved round 1 (comment 28895).

`mosaic project create` and `project list` call project.create and
project.list; `init` and `check` keep their arguments and refusals.
`mosaic threads`, `thread create`, `thread open` and `thread read` cover
the thread verbs, and `mosaic talk --thread` sends thread.send as a
REQUEST and prints only the reply to the sent message. `mosaic follow
--thread` is the same loop without the send. A thread reference is its
id, an id prefix of 8 or more characters, or <project>/<name>; unknown,
short and ambiguous references refuse with exit 2 before any send. Every
verb refuses inside an agent run. There is no default thread, so a plain
`talk assistant` still meets thread-required-use-thread-send.
Acceptance 314 step 1 now creates the project and the assistant thread.
New: packages/cli/src/thread.mjs, packages/cli/tests/thread-cli.test.mjs.

Sage's landing gate ran on 77b1a6e8a plus the candidate: 25 steps, all
rc 0 (15 packages, 10 script suites), 14:44:02Z to 14:48:42Z, network
sage-r128_default removed. The commit sits on 7de8b7270, which adds one
docs/SESSIONS.md line that no test reads.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

docs/TOOLS.md

@@ -244,6 +244,11 @@ scripts/mosaic decide <decision> <option> [--note <text>] [--yes]
 scripts/mosaic trail <task|decision> [--json]
 scripts/mosaic bus start <business> [--pm] | stop | status [--json]
 scripts/mosaic talk <role> <text> [--wait SECONDS] [--business <id>]
+scripts/mosaic project create <name> [--repo <repo>] | list [--json]
+scripts/mosaic threads [--project <project>] [--json]
+scripts/mosaic thread create <project> <name> | open <thread> | read <thread> [--json]
+scripts/mosaic talk --thread <thread> <text> [--wait SECONDS | --follow]
+scripts/mosaic follow --thread <thread>
 scripts/mosaic stop <run>
 scripts/mosaic launches off | on [--business <id>]
 scripts/mosaic launches list [--json]
@@ -623,6 +628,13 @@ Never migrate the live file from an agent session.
 
 ### Local project records (`mosaic project`)
 
+`mosaic project create` and `mosaic project list` work on the bus record of a
+business's projects; `mosaic project init` and `mosaic project check` work on
+the repository on disk. Threads (`mosaic threads`, `mosaic thread create|open|read`,
+`mosaic talk --thread`, `mosaic follow --thread`) are described in
+`packages/cli/README.md`; a plain `mosaic talk` to the threaded instance exits 2
+and asks for `--thread`.
+
 `mosaic project init <business> <project>` creates the declared project's
 separate local record repository, with layout version 1, package-validated JSON
 templates, Markdown spec/docs and its first commit authored as the business's

docs/acceptance/314.md

@@ -42,11 +42,19 @@ bus schema.
    Pass: `check` reports the layout valid. The first commit's author is
    Jason. (REQ-PROJ-1)
 
-   Then, with row 128's commands (`docs/TOOLS.md` names them once that
-   row lands), create a bus project named for `<project>` with repo
-   `<project>`, create an assistant thread in it, and open that thread.
-   Steps 3, 6 and 9 use this thread. Pass: the thread list shows the
-   thread under the project, owned by Jason.
+   Then create a bus project named for `<project>` with repo `<project>`,
+   create an assistant thread in it, and open that thread (for `stack`):
+
+   ```
+   mosaic project create stack --repo stack
+   mosaic thread create stack assistant
+   mosaic thread open stack/assistant
+   mosaic threads --project stack
+   ```
+
+   Steps 3, 6 and 9 use this thread. Pass: `mosaic threads --project stack`
+   shows `stack/assistant` under the project, and the broker made it as
+   Jason.
 
 2. Open the spec in the Console editor and write it:
 
@@ -60,10 +68,11 @@ bus schema.
    tab, which still holds the old commit: the save is refused and writes
    nothing. (REQ-PROJ-2, REQ-PROJ-3)
 
-3. Tell the assistant about the need, in the thread from step 1, with
-   row 128's talk command:
+3. Tell the assistant about the need, in the thread from step 1:
 
-   > New need: the mosaic version command in spec/north-star.md. Start it.
+   ```
+   mosaic talk --thread stack/assistant "New need: the mosaic version command in spec/north-star.md. Start it."
+   ```
 
    Pass: `talk` prints only the assistant's reply to this message.
 

packages/cli/README.md

@@ -32,6 +32,14 @@ mosaic agents [--business <id>] [--json]
 mosaic trail <task|decision> [--business <id>] [--json]
 mosaic talk <role> <text> [--wait <seconds> | --follow] [--business <id>]
 mosaic follow [--business <id>]
+mosaic project create <name> [--repo <repo>] [--business <id>]
+mosaic project list [--business <id>] [--json]
+mosaic threads [--project <project>] [--business <id>] [--json]
+mosaic thread create <project> <name> [--business <id>]
+mosaic thread open <thread> [--business <id>]
+mosaic thread read <thread> [--business <id>] [--json]
+mosaic talk --thread <thread> <text> [--wait <seconds> | --follow] [--business <id>]
+mosaic follow --thread <thread> [--business <id>]
 mosaic name <instance> <name> [--business <id>]
 mosaic names [--business <id>] [--json]
 mosaic stop <run>
@@ -76,6 +84,36 @@ mosaic launches list [--json]
   `--yes`, else it exits 4. On `outcome-unknown` it does not resend: it
   exits 1 and asks you to check `mosaic inbox` or `mosaic trail <id>`
   first.
+- Projects and threads (row 128, #1622) are the human's side of the bus's
+  `project.*` and `thread.*` ops. `project create` and `project list` work on
+  the bus record; `project init` and `project check` (`docs/TOOLS.md`, "Local
+  project records") work on the repository on disk and keep their arguments.
+  `project list` prints name, repo (`-` for none) and the first 8 characters of
+  the id. `threads` prints `<project>/<name>`, the instance and the short id of
+  the caller's threads, optionally for one project. `thread create` prints the
+  id and the instance the broker chose; `thread open` writes the open row the
+  launcher reads; `thread read` pages the thread's messages oldest first and
+  marks nothing read.
+- A project is its name or id. A thread is its id, an id prefix of at least 8
+  characters, or `<project>/<name>`. The resolver reads `project.list` and
+  `thread.list` and refuses an unknown or ambiguous reference with exit 2
+  before anything is sent. A full id that `thread.list` doesn't show goes to
+  the broker as given, which answers `thread-not-owned` for another user's
+  thread and `thread-not-found` for none. The CLI adds no ownership,
+  uniqueness or instance check of its own; `thread-not-owned`,
+  `thread-not-found`, `project-not-found`, `project-exists`, `thread-exists`,
+  `unknown-repo`, `thread-instance-missing` and `thread-instance-ambiguous`
+  print as `mosaic: <code>` and exit 2.
+- `talk --thread` sends `thread.send` as a REQUEST. Without `--follow` it
+  prints only the message whose `in_reply_to` is the sent id, like plain
+  `talk`, and leaves the rest unread. With `--follow` it prints every message
+  of that thread after the one it sent and marks read those addressed to the
+  human; other threads' messages and messages outside any thread stay unread
+  for `mosaic inbox`. `follow --thread` is the same loop without the send,
+  from the start of the thread. The loop reads the thread first, so a thread
+  that isn't the user's is refused before any message is received. A plain
+  `mosaic talk <role> <text>` to the threaded instance has no default
+  thread: it exits 2 and names `--thread` and `mosaic threads`.
 - `trail` prints rows in the broker's order (at, table, seq). A decision's
   trail ends with its `task_ref` and `follow with: mosaic trail <task>`. It
   does not pull in the task's rows.

packages/cli/src/cli.mjs

@@ -13,7 +13,8 @@ import { join } from "node:path";
 import { createInterface } from "node:readline/promises";
 import { fileURLToPath } from "node:url";
 import { BINDING_NAME } from "../../discord/src/binding.mjs";
-import { projectCommand, PROJECT_USAGE } from "./project.mjs";
+import { projectBusCommand, projectCommand, PROJECT_BUS_USAGE, PROJECT_USAGE } from "./project.mjs";
+import { resolveThread, threadCommand, threadMembers, threadsCommand, THREAD_TALK_USAGE, THREAD_USAGE, THREADS_USAGE } from "./thread.mjs";
 import { limitsCommand, LIMITS_USAGE } from "./limits.mjs";
 import { pluginsCommand, PLUGINS_USAGE } from "./plugins.mjs";
 import { CliError } from "./errors.mjs";
@@ -32,6 +33,9 @@ export const USAGE = `usage:
   ${LIMITS_USAGE}
   ${PLUGINS_USAGE}
   ${PROJECT_USAGE}
+  ${PROJECT_BUS_USAGE}
+  ${THREADS_USAGE}
+  ${THREAD_USAGE}
   mosaic inbox [--business <id>] [--json] [--tester]
   mosaic inbox --read <id>... | --read-all [--business <id>]
   mosaic decide <decision> <option> [--note <text>] [--yes] [--business <id>]
@@ -40,6 +44,7 @@ export const USAGE = `usage:
   mosaic trail <task|decision> [--business <id>] [--json] [--tester]
   mosaic talk <role> <text> [--wait <seconds> | --follow] [--business <id>] [--tester]
   mosaic follow [--business <id>] [--tester]
+  ${THREAD_TALK_USAGE}
   mosaic name <instance> <name> [--business <id>]
   mosaic names [--business <id>] [--json]
   mosaic stop <run>
@@ -314,25 +319,52 @@ function waitingLines(w, business) {
 // until the reply to it arrives or --wait runs out; other messages stay unread.
 // --follow prints and reads every message and keeps receiving after that
 // reply; follow starts the same receive loop without sending.
+// Row 128 (#1622): --thread names a thread (src/thread.mjs) and sends with thread.send. The
+// reply is found as before; --follow prints and reads only that thread's messages, and every
+// other message stays unread for mosaic inbox.
 async function talk(rest, io, deps, receiveOnly = false) {
-  const parsed = parse(rest, { flags: ["--follow", "--tester"], values: ["--business", ...(receiveOnly ? [] : ["--wait"])] });
-  if (parsed.positional.length !== (receiveOnly ? 0 : 2)) throw usage(receiveOnly ? "follow takes no arguments" : "talk takes a role and one quoted text");
+  const parsed = parse(rest, { flags: ["--follow", "--tester"], values: ["--business", "--thread", ...(receiveOnly ? [] : ["--wait"])] });
+  const threadRef = parsed.values["--thread"];
+  if (parsed.positional.length !== (receiveOnly ? 0 : threadRef === undefined ? 2 : 1)) {
+    throw usage(receiveOnly ? "follow takes no arguments" : threadRef === undefined ? "talk takes a role and one quoted text" : "talk --thread takes one quoted text");
+  }
+  // The thread verbs are the human's own: the tester has no thread.
+  if (threadRef !== undefined && parsed.flags.has("--tester")) throw usage("--thread takes no --tester");
   if (receiveOnly && parsed.flags.has("--follow")) throw usage("follow already follows");
   const follow = receiveOnly || parsed.flags.has("--follow");
   if (follow && parsed.values["--wait"] !== undefined) throw usage("--follow and --wait cannot be combined");
   const wait = parsed.values["--wait"] === undefined ? TALK_WAIT_S : Number(parsed.values["--wait"]);
   if (!Number.isInteger(wait) || wait < 0) throw usage("--wait takes whole seconds, 0 or more");
-  const [to, body] = parsed.positional;
+  let [to, body] = parsed.positional;
+  if (threadRef !== undefined) [to, body] = [threadRef, to];
   // The broker refuses it too (tester-prefix); this is the clear early message.
   if (!receiveOnly && parsed.flags.has("--tester") && !body.startsWith(TESTER_PREFIX)) throw new CliError(`a tester message starts with "${TESTER_PREFIX}"`, 2);
   const call = callFor(parsed, io, deps);
   let sent;
+  let thread = null;
+  if (threadRef !== undefined) {
+    thread = await resolveThread(call, threadRef);
+    to = thread.label;
+  }
   if (!receiveOnly) {
-    sent = await call("message.send", { to, body, class: "REQUEST" });
+    try {
+      sent = await call(thread ? "thread.send" : "message.send", thread ? { thread: thread.id, body, class: "REQUEST" } : { to, body, class: "REQUEST" });
+    } catch (e) {
+      // The bus never picks a thread for the human, and neither does the CLI (row 128).
+      if (e.code === "thread-required-use-thread-send") {
+        throw new CliError(`${to} runs in threads, so a message to it names one: mosaic talk --thread <thread> <text>; mosaic threads lists yours`, 2);
+      }
+      throw e;
+    }
     io.stdout.write(`sent ${shortId(sent.id)} to ${to}\n`);
     if (sent.waiting) io.stdout.write(waitingLines(sent.waiting, () => businessFor(parsed, (deps.system ?? loadSystem({ env: io.env })).dataRoot)));
     if (!follow && wait === 0) return;
   }
+  // message.receive says nothing of a message's thread, so --follow asks the thread for its ids.
+  // Reading the thread first makes `follow --thread` refuse a thread that isn't the user's before
+  // anything is received: message.receive writes a delivery row for every message it returns.
+  const members = thread && follow ? threadMembers(call, thread.id) : null;
+  await members?.refresh();
   const pollMs = deps.talkPollMs ?? TALK_POLL_MS;
   const deadline = Date.now() + wait * 1000;
   const controller = new AbortController();
@@ -344,9 +376,13 @@ async function talk(rest, io, deps, receiveOnly = false) {
   try {
     while (!controller.signal.aborted) {
       let answered = false;
-      for (const m of await call("message.receive")) {
+      const received = await call("message.receive");
+      if (members && received.length > 0) await members.refresh();
+      for (const m of received) {
         // Row 142 (#1636): without --follow only the reply to the sent message is printed and marked read; the rest stays unread for mosaic inbox.
         if (!follow && m.in_reply_to !== sent.id) continue;
+        // Only the thread's messages after the one sent; an earlier one stays unread.
+        if (members && !(members.seq(m.id) > (members.seq(sent?.id) ?? 0))) continue;
         io.stdout.write(formatMessage(m));
         await call("message.read", { id: m.id });
         if (m.in_reply_to === sent?.id) answered = true;
@@ -499,7 +535,9 @@ export async function main(argv, io = { env: process.env, stdin: process.stdin,
   if (argv.length === 0) return app(io, deps);
   const [verb, ...rest] = argv;
   if (["inbox", "decide", "tasks", "agents", "trail"].includes(verb)) return human(verb, rest, io, deps);
-  if (verb === "project") return projectCommand(rest, io, deps);
+  if (verb === "project") return ["create", "list"].includes(rest[0]) ? projectBusCommand(rest, io, (p) => callFor(p, io, deps)) : projectCommand(rest, io, deps);
+  if (verb === "threads") return threadsCommand(rest, io, (p) => callFor(p, io, deps));
+  if (verb === "thread") return threadCommand(rest, io, (p) => callFor(p, io, deps), formatMessage);
   if (verb === "limits") return limitsCommand(rest,io,p=>callFor(p,io,deps));
   if (verb === "plugins") return pluginsCommand(rest, io, (p) => callFor(p, io, deps));
   if (verb === "talk") return talk(rest, io, deps);

packages/cli/src/project.mjs

@@ -2,9 +2,36 @@ import { configDir, loadBusiness } from '../../business/src/index.mjs';
 import { checkProject, initProject, ProjectError } from '../../project/src/index.mjs';
 import { loadSystem, rolesDir } from './config.mjs';
 import { CliError } from './errors.mjs';
+import { shortId } from './format.mjs';
+import { parseArgs } from './thread.mjs';
 import { refuseInsideAgent } from './transport.mjs';
 
 export const PROJECT_USAGE = 'mosaic project init|check <business> <project>';
+// Row 128 (#1622): create and list work on the bus record (project.create, project.list); init
+// and check above work on the repository on disk and never contact the bus.
+export const PROJECT_BUS_USAGE = [
+  'mosaic project create <name> [--repo <repo>] [--business <id>]',
+  'mosaic project list [--business <id>] [--json]',
+].join('\n  ');
+
+export async function projectBusCommand(args, io, callFor) {
+  const [sub, ...rest] = args;
+  const parsed = parseArgs(rest, PROJECT_BUS_USAGE, sub === 'create'
+    ? { values: ['--repo', '--business'] }
+    : { flags: ['--json'], values: ['--business'] });
+  if (parsed.positional.length !== (sub === 'create' ? 1 : 0))
+    throw new CliError(`project ${sub} takes ${sub === 'create' ? 'one name' : 'no arguments'}\n${PROJECT_BUS_USAGE}`, 4);
+  const call = callFor(parsed);
+  if (sub === 'create') {
+    const [name] = parsed.positional;
+    const made = await call('project.create', { name, ...(parsed.values['--repo'] === undefined ? {} : { repo: parsed.values['--repo'] }) });
+    return io.stdout.write(`created project ${name}  id ${made.id}\n`);
+  }
+  const projects = await call('project.list');
+  if (parsed.flags.has('--json')) return io.stdout.write(`${JSON.stringify(projects, null, 2)}\n`);
+  io.stdout.write(projects.length === 0 ? 'no projects\n' : projects.map(p => `${p.name}  ${p.repo ?? '-'}  ${shortId(p.id)}\n`).join(''));
+}
+
 export async function projectCommand(args, io, deps = {}) {
   if (args.length !== 3 || !['init', 'check'].includes(args[0]) || args.some(arg => arg.startsWith('--')))
     throw new CliError(PROJECT_USAGE, 4);

packages/cli/src/thread.mjs

@@ -0,0 +1,137 @@
+// Row 128 (#1622, REQ-CLI-1, REQ-CLI-2, REQ-THREAD-2, REQ-USER-2): the human's side of the
+// bus's project and thread ops (row 121). `mosaic threads`, `mosaic thread create|open|read`,
+// and the thread resolver `mosaic talk --thread` and `mosaic follow --thread` (src/cli.mjs) use.
+// The broker owns every rule: ownership, uniqueness, the one threaded instance. The CLI adds
+// no check of those; it resolves a reference to an id and prints what the broker returns.
+
+import { CliError } from "./errors.mjs";
+import { shortId } from "./format.mjs";
+
+export const THREADS_USAGE = "mosaic threads [--project <project>] [--business <id>] [--json]";
+export const THREAD_USAGE = [
+  "mosaic thread create <project> <name> [--business <id>]",
+  "mosaic thread open <thread> [--business <id>]",
+  "mosaic thread read <thread> [--business <id>] [--json]",
+].join("\n  ");
+export const THREAD_TALK_USAGE = [
+  "mosaic talk --thread <thread> <text> [--wait <seconds> | --follow] [--business <id>]",
+  "mosaic follow --thread <thread> [--business <id>]",
+].join("\n  ");
+
+// The parse shape callFor reads (flags, values, positional). Errors carry the verb's own usage.
+export function parseArgs(args, usage, { flags = [], values = [] } = {}) {
+  const out = { positional: [], flags: new Set(), values: {} };
+  const refuse = (why) => new CliError(`${why}\n${usage}`, 4);
+  for (let i = 0; i < args.length; i++) {
+    const a = args[i];
+    if (flags.includes(a)) out.flags.add(a);
+    else if (values.includes(a)) {
+      if (i + 1 >= args.length || Object.hasOwn(out.values, a)) throw refuse(`${a} needs one value`);
+      out.values[a] = args[++i];
+    } else if (a.startsWith("--")) throw refuse(`unknown option ${a}`);
+    else out.positional.push(a);
+  }
+  return out;
+}
+
+const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+const HEX = /^[0-9a-f-]+$/i;
+
+// A project is its name or its id.
+export async function resolveProject(call, ref) {
+  const projects = await call("project.list");
+  const hit = projects.find((p) => p.id === ref) ?? projects.find((p) => p.name === ref);
+  if (hit) return hit;
+  throw new CliError(`no project is named or numbered ${JSON.stringify(ref)}; see mosaic project list`, 2);
+}
+
+const labelOf = (projects, t) => `${projects.find((p) => p.id === t.project)?.name ?? shortId(t.project)}/${t.name}`;
+
+// A thread is its id, an id prefix of at least 8 characters, or <project>/<name>. thread.list
+// holds only the caller's threads, so a full id it doesn't list is passed on: the broker says
+// thread-not-owned for another user's thread and thread-not-found for none (REQ-USER-2).
+export async function resolveThread(call, ref) {
+  const threads = await call("thread.list");
+  const projects = await call("project.list");
+  const exact = threads.find((t) => t.id === ref) ?? threads.find((t) => labelOf(projects, t) === ref);
+  if (exact) return { id: exact.id, label: labelOf(projects, exact) };
+  if (UUID.test(ref)) return { id: ref, label: ref };
+  if (HEX.test(ref)) {
+    if (ref.length < 8) throw new CliError(`thread reference ${JSON.stringify(ref)} is too short; use at least 8 characters of the id, or <project>/<name>`, 2);
+    const hits = threads.filter((t) => t.id.startsWith(ref));
+    if (hits.length === 1) return { id: hits[0].id, label: labelOf(projects, hits[0]) };
+    if (hits.length > 1) throw new CliError(`${ref} matches ${hits.length} threads; use more of the id, or <project>/<name>`, 2);
+  }
+  throw new CliError(`no thread of yours matches ${JSON.stringify(ref)}; mosaic threads lists them as <project>/<name>`, 2);
+}
+
+const PAGE = 200;
+
+// The thread's messages after `after`, oldest first, from every page of thread.messages.
+// Reading writes no delivery or read row.
+export async function threadMessages(call, thread, after) {
+  const out = [];
+  for (;;) {
+    const page = await call("thread.messages", { thread, limit: PAGE, ...(after ? { after } : {}) });
+    out.push(...page);
+    if (page.length < PAGE) return out;
+    after = page.at(-1).id;
+  }
+}
+
+// A thread's message ids with their sequence numbers, kept current by refresh(). message.receive
+// does not say which thread a message is in, so `--follow` receives first and asks the thread
+// second; the broker records a message's thread when it stores the message, so a message it
+// received is already listed.
+export function threadMembers(call, thread) {
+  const seqs = new Map();
+  let last;
+  return {
+    seq: (id) => seqs.get(id),
+    async refresh() {
+      for (const m of await threadMessages(call, thread, last)) {
+        seqs.set(m.id, m.seq);
+        last = m.id;
+      }
+    },
+  };
+}
+
+const json = (io, value) => io.stdout.write(`${JSON.stringify(value, null, 2)}\n`);
+
+export async function threadsCommand(args, io, callFor) {
+  const parsed = parseArgs(args, THREADS_USAGE, { flags: ["--json"], values: ["--business", "--project"] });
+  if (parsed.positional.length !== 0) throw new CliError(`threads takes no arguments\n${THREADS_USAGE}`, 4);
+  const call = callFor(parsed);
+  const project = parsed.values["--project"] === undefined ? null : await resolveProject(call, parsed.values["--project"]);
+  const threads = await call("thread.list", project ? { project: project.id } : {});
+  if (parsed.flags.has("--json")) return json(io, threads);
+  if (threads.length === 0) return io.stdout.write("no threads\n");
+  const projects = await call("project.list");
+  io.stdout.write(threads.map((t) => `${labelOf(projects, t)}  ${t.instance}  ${shortId(t.id)}\n`).join(""));
+}
+
+// `format` is cli.mjs's formatMessage, the one line talk and inbox print a message with.
+export async function threadCommand(args, io, callFor, format) {
+  const [sub, ...rest] = args;
+  if (!["create", "open", "read"].includes(sub)) throw new CliError(`usage:\n  ${THREAD_USAGE}`, 4);
+  const parsed = parseArgs(rest, THREAD_USAGE, { flags: sub === "read" ? ["--json"] : [], values: ["--business"] });
+  if (parsed.positional.length !== (sub === "create" ? 2 : 1)) {
+    throw new CliError(`thread ${sub} takes ${sub === "create" ? "a project and a name" : "one thread"}\n${THREAD_USAGE}`, 4);
+  }
+  const call = callFor(parsed);
+  if (sub === "create") {
+    const [ref, name] = parsed.positional;
+    const project = await resolveProject(call, ref);
+    const made = await call("thread.create", { project: project.id, name });
+    return io.stdout.write(`created ${project.name}/${name}  id ${made.id}  instance ${made.instance}\n`);
+  }
+  const thread = await resolveThread(call, parsed.positional[0]);
+  if (sub === "open") {
+    await call("thread.open", { thread: thread.id });
+    return io.stdout.write(`opened ${thread.label}\n`);
+  }
+  const messages = await threadMessages(call, thread.id);
+  if (parsed.flags.has("--json")) return json(io, messages);
+  io.stdout.write(messages.length === 0 ? `${thread.label}: no messages\n` : messages.map(format).join(""));
+}

packages/cli/tests/project.test.mjs

@@ -32,6 +32,25 @@ test('project commands validate args and refuse undeclared projects and agent in
     assert.equal(await run(['project', ...args], io(humanEnv(f.env)), {}), 4);
 });
 
+// Row 128 (#1622): create and list are bus verbs; they take their own arguments, and init and
+// check keep theirs (a name for create is not the init/check <business> <project> pair).
+test('project create and list validate their arguments before any bus call', async t => {
+  const root = tmp(t), f = fixture(root);
+  const deps = { transport: () => assert.fail('a usage error reached the bus') };
+  for (const args of [['create'], ['create', 'a', 'b'], ['create', 'a', '--repo'], ['create', 'a', '--repo', 'x', '--repo', 'y'], ['create', 'a', '--write'], ['list', 'extra'], ['list', '--repo', 'x']]) {
+    const x = io(humanEnv(f.env));
+    // --business is given, so only the arity check can produce the exit 4 (callFor would too, without it).
+    assert.equal(await run(['project', ...args, '--business', 'acme'], x, deps), 4, args.join(' '));
+    assert.match(x.err.text, /mosaic project create <name>/);
+  }
+  const listed = [];
+  const transport = () => async verb => (listed.push(verb), []);
+  const x = io(humanEnv(f.env));
+  assert.equal(await run(['project', 'list', '--business', 'acme'], x, { transport }), 0);
+  assert.deepEqual(listed, ['project.list']);
+  assert.equal(x.out.text, 'no projects\n');
+});
+
 test('project dispatcher reaches the human CLI', async t => {
   const root = tmp(t), f = fixture(root);
   const result = spawnSync(new URL('../../../scripts/mosaic', import.meta.url).pathname, ['project'], {

packages/cli/tests/thread-cli.test.mjs

@@ -0,0 +1,343 @@
+// 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";
+
+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 b = new Broker({
+    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 });
+    await s.cli(["project", "create", "stack"]);
+    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 <project>/<name>; 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}`);
+  }
+  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 <thread> <text>/);
+  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", "--tester"], ["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");
+  // 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);
+});