bus, cli, webui: every message to the human, read state, Messages page (#1588)

Row 94, round 2. Rocko built it, Dewey approved it (comment 28104) on
manifest 53b01ea9 (16 paths, base e918a915). The bus keeps a messages
view with read state, where delivered but not read stays read:false. The
CLI gains messages and follow, and inbox leaves out messages already
read while inbox --json stays byte-compatible. The Console gets a
Messages page and an unread count in the nav. Mos gave the go under
decision 123 (public/ files). Sage gate on 645489e4 plus the patch,
12:52:14Z to 12:55:44Z: bus 115, cli 129, webui 104, all fail 0. Every
scripts/test-*.sh exited 0 (auth 15, conductor 17, config 24, discord
66, extension-package 18, foundation 44, queue 27, release 14, task 98).
Manifest 16/16 before and after.

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

packages/cli/README.md

@@ -29,7 +29,8 @@ mosaic decide <decision> <option> [--note <text>] [--yes] [--business <id>]
 mosaic tasks [--business <id>] [--json]
 mosaic agents [--business <id>] [--json]
 mosaic trail <task|decision> [--business <id>] [--json]
-mosaic talk <role> <text> [--wait <seconds>] [--business <id>]
+mosaic talk <role> <text> [--wait <seconds> | --follow] [--business <id>]
+mosaic follow [--business <id>]
 mosaic stop <run>
 mosaic launches off|on [--business <id>]
 mosaic launches list [--json]
@@ -48,7 +49,9 @@ mosaic launches list [--json]
   these commands.** The tests use a stand-in transport.
 - `inbox` lists the open decisions routed to the human, gated first, in
   broker order. For a gated decision it prints what approving authorizes:
-  the action, its target, and the one choice that authorizes it.
+  the action, its target, and the one choice that authorizes it. Text output
+  then lists all unread messages to the human, newest first, without marking
+  any delivered or read. `--json` keeps its existing decision-row array.
 - `decide` takes the full id, or a unique prefix of at least 8 characters,
   from the inbox. It prints the decision and whether the choice authorizes
   or declines, then asks `[y/N]` on a terminal. Without a terminal it needs
@@ -62,7 +65,11 @@ mosaic launches list [--json]
   inbox every 2 s for up to `--wait` seconds (default 900). It prints every
   message that arrives and marks it read, and stops at the reply to its
   REQUEST. No reply in time exits 1; the reply may still come, and the next
-  `mosaic talk` prints it. `--wait 0` only sends.
+  `mosaic talk` prints it. `--wait 0` only sends. `talk --follow` keeps
+  printing and marking messages read until Ctrl-C, including messages with
+  no `in_reply_to`. `follow` runs that same loop without sending. `--wait`
+  and `--follow` together refuse. These commands consume the same human
+  inbox; run one receiving CLI at a time. Console only reads history.
 - `launches off` and `on` are the broker's `launch.revoke` and
   `launch.restore`. While off, every `role.launch` refuses with
   `launch-revoked`; running sessions keep running. Stop them with `stop`.

packages/cli/src/cli.mjs

@@ -29,7 +29,8 @@ export const USAGE = `usage:
   mosaic tasks [--business <id>] [--json]
   mosaic agents [--business <id>] [--json]
   mosaic trail <task|decision> [--business <id>] [--json]
-  mosaic talk <role> <text> [--wait <seconds>] [--business <id>]
+  mosaic talk <role> <text> [--wait <seconds> | --follow] [--business <id>]
+  mosaic follow [--business <id>]
   mosaic stop <run>
   mosaic launches off|on [--business <id>]
   mosaic launches list [--json]
@@ -170,7 +171,19 @@ async function human(verb, rest, io, deps) {
     return print(await call("trail", { subject }), (rows) => formatTrail(subject, rows));
   }
   if (parsed.positional.length !== 0) throw usage(`${verb} takes no arguments`);
-  if (verb === "inbox") return print(await call("inbox"), formatInbox);
+  if (verb === "inbox") {
+    const decisions = await call("inbox");
+    if (json) return print(decisions, formatInbox);
+    const messages = [];
+    let after;
+    do {
+      const page = await call("messages", { limit: 200, ...(after ? { after } : {}) });
+      messages.push(...page.messages.filter(m => !m.read));
+      after = page.next;
+    } while (after);
+    return print({ decisions, messages }, value => formatInbox(value.decisions) +
+      `\nmessages: ${value.messages.length} unread\n` + value.messages.map(formatMessage).join(""));
+  }
   if (verb === "tasks") return print(await call("tasks"), formatTasks);
   return print(await call("agents"), formatAgents);
 }
@@ -184,11 +197,14 @@ function formatMessage(m) {
 }
 
 // `mosaic talk <role> <text>`: a REQUEST to the role, then the human's inbox
-// until the reply arrives or --wait runs out. Every message received on the
-// way is printed and marked read, so a late reply shows up on the next talk.
-async function talk(rest, io, deps) {
-  const parsed = parse(rest, { values: ["--business", "--wait"] });
-  if (parsed.positional.length !== 2) throw usage("talk takes a role and one quoted text");
+// until the reply arrives or --wait runs out. --follow keeps receiving after
+// that reply; follow starts the same receive loop without sending.
+async function talk(rest, io, deps, receiveOnly = false) {
+  const parsed = parse(rest, { flags: ["--follow"], 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");
+  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");
   refuseInsideAgent(io.env);
@@ -196,21 +212,40 @@ async function talk(rest, io, deps) {
   const business = businessFor(parsed, system.dataRoot);
   const call = deps.transport ? deps.transport(business) : humanTransport({ socket: socketPath(system.dataRoot), business, env: io.env });
   const [to, body] = parsed.positional;
-  const sent = await call("message.send", { to, body, class: "REQUEST" });
-  io.stdout.write(`sent ${shortId(sent.id)} to ${to}\n`);
-  if (wait === 0) return;
+  let sent;
+  if (!receiveOnly) {
+    sent = await call("message.send", { to, body, class: "REQUEST" });
+    io.stdout.write(`sent ${shortId(sent.id)} to ${to}\n`);
+    if (!follow && wait === 0) return;
+  }
   const pollMs = deps.talkPollMs ?? TALK_POLL_MS;
   const deadline = Date.now() + wait * 1000;
-  for (;;) {
-    let answered = false;
-    for (const m of await call("message.receive")) {
-      io.stdout.write(formatMessage(m));
-      await call("message.read", { id: m.id });
-      if (m.in_reply_to === sent.id) answered = true;
+  const controller = new AbortController();
+  const stop = () => controller.abort();
+  // Only continuous commands take ownership of Ctrl-C, and release it on every exit.
+  if (follow) process.on("SIGINT", stop);
+  deps.signal?.addEventListener("abort", stop, { once: true });
+  if (deps.signal?.aborted) stop();
+  try {
+    while (!controller.signal.aborted) {
+      let answered = false;
+      for (const m of await call("message.receive")) {
+        io.stdout.write(formatMessage(m));
+        await call("message.read", { id: m.id });
+        if (m.in_reply_to === sent?.id) answered = true;
+      }
+      if (!follow && answered) return;
+      if (!follow && Date.now() >= deadline) throw new CliError(`no reply from ${to} within ${wait} s; it may still come, and the next mosaic talk shows it`, 1);
+      if (controller.signal.aborted) break;
+      await new Promise(resolve => {
+        const wake = () => { clearTimeout(timer); controller.signal.removeEventListener("abort", wake); resolve(); };
+        const timer = setTimeout(wake, pollMs);
+        controller.signal.addEventListener("abort", wake, { once: true });
+      });
     }
-    if (answered) return;
-    if (Date.now() >= deadline) throw new CliError(`no reply from ${to} within ${wait} s; it may still come, and the next mosaic talk shows it`, 1);
-    await new Promise((r) => setTimeout(r, pollMs));
+  } finally {
+    if (follow) process.removeListener("SIGINT", stop);
+    deps.signal?.removeEventListener("abort", stop);
   }
 }
 
@@ -336,6 +371,7 @@ export async function main(argv, io = { env: process.env, stdin: process.stdin,
   const [verb, ...rest] = argv;
   if (["inbox", "decide", "tasks", "agents", "trail"].includes(verb)) return human(verb, rest, io, deps);
   if (verb === "talk") return talk(rest, io, deps);
+  if (verb === "follow") return talk(rest, io, deps, true);
   if (verb === "stop") return stop(rest, io, deps);
   if (verb === "launches") return launches(rest, io, deps);
   if (verb === "bus") return bus(rest, io, deps);

packages/cli/tests/cli.test.mjs

@@ -40,7 +40,7 @@ test("inbox lists only decisions routed to the human, with what approving author
   assert.match(text, new RegExp(`decide: mosaic decide ${gated.id.slice(0, 8)} <option>`));
   const json = await s.run(["inbox", "--business", "demo", "--json"]);
   assert.equal(JSON.parse(json.io.out.text)[0].id, gated.id);
-  assert.deepEqual(s.calls.map((c) => c.verb), ["inbox", "inbox"]);
+  assert.deepEqual(s.calls.map((c) => c.verb), ["inbox", "messages", "inbox"]);
 });
 
 test("decide resolves by id prefix with --yes, and the trail shows it in broker order with the task to follow", async (t) => {
@@ -139,7 +139,7 @@ test("without --business a command uses the live host's business, and a stale ho
   assert.equal(s.calls.length, 0);
   write(startTimeOf(process.pid));
   assert.equal((await s.run(["inbox"])).code, 0);
-  assert.deepEqual(s.calls.map((c) => c.business), ["demo"]);
+  assert.deepEqual(s.calls.map((c) => c.business), ["demo", "demo"]);
 });
 
 test("every human command refuses inside an agent run before it touches the bus", async (t) => {

packages/cli/tests/follow.test.mjs

@@ -0,0 +1,86 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {main} from '../src/cli.mjs';
+import {broker,io,tmp} from './helpers.mjs';
+const call=(s,cap,verb,args={})=>s.b.request(cap,{verb,args});
+for(const follow of [false,true]) test(follow?'talk --follow continues after reply and prints a later unsolicited message':'follow prints an unsolicited message after plain talk returned',async t=>{
+  const s=broker(t), x=io(), controller=new AbortController();
+  let polls=0,sent;
+  const transport=business=>{const inner=s.transport(business);return async(verb,args)=>{
+    if(verb==='message.send') {const result=await inner(verb,args);sent=result.id;return result;}
+    if(verb==='message.receive'){
+      polls++;
+      if(polls===1 && sent)call(s,s.coder,'message.send',{to:'human',body:'direct answer',in_reply_to:sent});
+      if(polls===2)call(s,s.coder,'message.send',{to:'human',body:'later unsolicited',class:'INFO'});
+      if(polls===3)controller.abort();
+    }
+    return inner(verb,args);
+  }};
+  const deps={system:{dataRoot:tmp(t)},transport,talkPollMs:5,signal:controller.signal};
+  if(!follow){await main(['talk','coder','question','--business','demo'],x,deps); assert.match(x.out.text,/direct answer/);}
+  await main(follow?['talk','coder','question','--follow','--business','demo']:['follow','--business','demo'],x,deps);
+  assert.equal(polls,3); assert.match(x.out.text,/later unsolicited/);
+  assert.equal((x.out.text.match(/later unsolicited/g)||[]).length,1);
+  assert.deepEqual(call(s,s.human,'message.receive'),[]);
+  assert.equal(s.calls.filter(c=>c.verb==='message.send').length,1);
+  assert.equal(s.calls.filter(c=>c.verb==='message.read').length,2);
+});
+test('inbox lists all unread human messages under decisions, does not deliver or read them',async t=>{
+ const s=broker(t),x=io(),deps={system:{dataRoot:tmp(t)},transport:s.transport};
+ for(let i=0;i<205;i++)call(s,s.coder,'message.send',{to:'human',body:`unsolicited-${i}`,class:'INFO'});
+ const delivered=call(s,s.human,'message.receive');
+ call(s,s.human,'message.read',{id:delivered[0].id});
+ const before=s.store.all('SELECT * FROM deliveries');
+ await main(['inbox','--business','demo'],x,deps);
+ assert.match(x.out.text,/inbox:/);assert.match(x.out.text,/messages: 204 unread/);
+ assert.doesNotMatch(x.out.text,/unsolicited-0\n/);
+ assert.match(x.out.text,/unsolicited-1\n/);assert.match(x.out.text,/unsolicited-204\n/);
+ assert.deepEqual(s.store.all('SELECT * FROM deliveries'),before);
+ const json=io(); await main(['inbox','--business','demo','--json'],json,deps);
+ const value=JSON.parse(json.out.text);assert.deepEqual(value,[]);assert.equal(json.out.text,JSON.stringify(s.read("inbox"),null,2)+"\n");
+});
+test('follow refuses agent context and bad arguments',async t=>{
+ const s=broker(t),deps={system:{dataRoot:tmp(t)},transport:s.transport};
+ for(const argv of [['follow','extra'],['follow','--wait','1'],['talk','coder','hi','--follow','--wait','1']])await assert.rejects(main(argv,io(),deps),{exitCode:4});
+ await assert.rejects(main(['follow','--business','demo'],io({...io().env,CLAUDECODE:'1'}),deps),{exitCode:3});
+});
+
+test('follow Ctrl-C wakes the poll sleep and removes its handler; transport errors also clean up', async t => {
+  const s = broker(t), x = io(), before = process.listenerCount('SIGINT'), cleanup = new AbortController();
+  let received;
+  const ready = new Promise(resolve => { received = resolve; });
+  const deps = { system: { dataRoot: tmp(t) }, talkPollMs: 60000, signal: cleanup.signal,
+    transport: () => async verb => { assert.equal(verb, 'message.receive'); received(); return []; } };
+  const running = main(['follow', '--business', 'demo'], x, deps);
+  await ready;
+  // Let the command install its sleep's abort listener before emitting Ctrl-C.
+  await new Promise(resolve => setImmediate(resolve));
+  process.emit('SIGINT');
+  try {
+    await Promise.race([running, new Promise((_, reject) => {
+      const timer = setTimeout(() => reject(Error('Ctrl-C did not wake follow')), 500);
+      timer.unref();
+    })]);
+  } finally { cleanup.abort(); await running; }
+  assert.equal(process.listenerCount('SIGINT'), before);
+  await assert.rejects(main(['follow', '--business', 'demo'], io(), {
+    ...deps, signal: undefined, transport: () => async () => { throw Error('transport failed'); },
+  }), /transport failed/);
+  assert.equal(process.listenerCount('SIGINT'), before);
+});
+
+
+test('plain talk leaves the process SIGINT handlers unchanged',async t=>{
+ const s=broker(t),before=process.listeners('SIGINT'),x=io();let sent;
+ const deps={system:{dataRoot:tmp(t)},transport:business=>{const inner=s.transport(business);return async(verb,args)=>{
+  if(verb==='message.send'){const result=await inner(verb,args);sent=result.id;return result;}
+  if(verb==='message.receive'){
+   assert.deepEqual(process.listeners('SIGINT'),before);
+   call(s,s.coder,'message.send',{to:'human',body:'answer',in_reply_to:sent});
+  }
+  return inner(verb,args);
+ }}};
+ try {await main(['talk','coder','question','--business','demo'],x,deps);}
+ finally {for(const handler of process.listeners('SIGINT'))if(!before.includes(handler))process.removeListener('SIGINT',handler);}
+ assert.deepEqual(process.listeners('SIGINT'),before);
+});