mosaicstack/stack
25
0
Fork 1
Code Issues 365 Pull Requests 13 Actions Packages Projects Releases 15 Wiki Activity

bus, webui, cli: the Console shows the reply to a web send (#1575)

Browse Source 
Row 81. The broker gains a `replies` read verb for human and reader
sessions: `{id}`, a message UUID, returns that business's messages whose
in_reply_to is the id, oldest first, at most 50, and writes no
deliveries row, so the reply still reaches the next `mosaic talk`. An
unknown id refuses message-not-found. The Console serves it through
BUS_VERBS with an id check, and the Sent receipt shows the replies to a
web send. The cli's talk timeout message and README now say the reply
may still come and the next `mosaic talk` prints it.

Dewey's round 2, candidate manifest 3ce7527f, base 1e5f5958, a rebase
after row 61. Approved by Darkwing (27783 round 1, 27821 round 2).
Sage's gate green 19:25:18Z to 19:30:39Z. Live at the next bus restart
(the broker verb) and Console restart (the view), bus first.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
jason.woltje
2026-10-12 14:33:11 -05:00
co-authored by Claude Opus 5.5
parent 9438880e67
commit c9a5650ed0
13 changed files with 455 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/webui/tests/bus.test.mjs
+48 -2
View File
@@ -150,9 +150,53 @@ test('credentials route serves the broker credential status from the reader view
assert.deepEqual(calls, [['credentials', {}]]);
});
// Row 81 (#1575): the replies to one message, for the Console's send receipt. The bus refuses an
// id that names no message; a malformed one never reaches it.
test('replies route serves the replies to one message from the reader view', async t => {
const store = new Store(tmp(t, 'webui-bus-replies-'));
t.after(() => store.close());
const b = new Broker({ store, businesses });
const coder = b.bindLaunch({ business: 'demo', role: 'coder', run: 'coder-run', harness: 'pi', address: 'coder-run' });
b.request(coder, { verb: 'role.claim' });
const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
const sent = b.request(human, { verb: 'message.send', args: { to: 'coder', body: 'Please look.' } });
const quiet = b.request(human, { verb: 'message.send', args: { to: 'coder', body: 'No answer to this one.' } });
const answer = b.request(coder, { verb: 'message.send', args: { to: 'human', body: `Done. ${HOSTILE}`, class: 'RESULT', in_reply_to: sent.id } });
const reader = b.bindReader({ business: 'demo' });
const calls = [];
// A broker refusal comes back from the transport as a BusReadError carrying its code.
const call = async (verb, args = {}) => {
calls.push([verb, args]);
try { return structuredClone(b.request(reader, { verb, args })); } catch (err) { throw new BusReadError(err.code); }
};
const bus = views({ call });
const web = await startServer({ port: 0, board: 'http://127.0.0.1:9', bus });
t.after(() => close(web));
const base = `http://127.0.0.1:${web.address().port}`;
const read = async path => { const r = await fetch(base + path); return { status: r.status, cache: r.headers.get('cache-control'), body: await r.json() }; };
const deliveries = () => store.all('SELECT * FROM deliveries ORDER BY seq');
const before = deliveries();
const r = await read('/api/bus/replies?id=' + sent.id);
assert.equal(r.status, 200); assert.equal(r.cache, 'no-store');
assert.match(r.body.at, /^\d{4}-\d\d-\d\dT/);
assert.deepEqual(r.body.rows.map(m => [m.id, m.from_role, m.class, m.in_reply_to, m.body]), [[answer.id, 'coder', 'RESULT', sent.id, `Done. ${HOSTILE}`]]);
assert.deepEqual((await read('/api/bus/replies?id=' + quiet.id)).body.rows, []);
const unknown = await read('/api/bus/replies?id=00000000-0000-4000-8000-000000000000');
assert.deepEqual([unknown.status, unknown.body.error], [400, 'message-not-found']);
assert.deepEqual(deliveries(), before, 'no delivered or read row');
assert.deepEqual(calls, [['replies', { id: sent.id }], ['replies', { id: quiet.id }], ['replies', { id: '00000000-0000-4000-8000-000000000000' }]]);
// A missing or malformed id never reaches the bus; trail's `subject` is not this read's argument.
const n = calls.length;
for (const q of ['', '?id=', '?id=' + 'a'.repeat(161), '?id=-x', '?id=a%20b', '?id=%3Cb%3E', '?subject=' + sent.id])
assert.deepEqual([q, (await read('/api/bus/replies' + q)).body.error], [q, 'invalid-request']);
assert.equal(calls.length, n);
});
test('bus routes are GET-only, have no write verb, and keep the same-origin checks', async t => {
const calls = [];
const bus = Object.fromEntries(['inbox', 'tasks', 'agents', 'trail', 'decisions', 'credentials'].map(v => [v, async () => { calls.push(v); return []; }]));
const bus = Object.fromEntries(['inbox', 'tasks', 'agents', 'trail', 'decisions', 'credentials', 'replies'].map(v => [v, async () => { calls.push(v); return []; }]));
const web = await startServer({ port: 0, board: 'http://127.0.0.1:9', bus });
t.after(() => close(web));
const base = `http://127.0.0.1:${web.address().port}`;
@@ -160,6 +204,7 @@ test('bus routes are GET-only, have no write verb, and keep the same-origin chec
assert.equal((await fetch(base + '/api/bus/inbox', { method })).status, 405, method);
assert.equal((await fetch(base + '/api/bus/decisions?list=roles', { method })).status, 405, method);
assert.equal((await fetch(base + '/api/bus/credentials', { method })).status, 405, method);
assert.equal((await fetch(base + '/api/bus/replies?id=m-1', { method })).status, 405, method);
}
for (const path of ['/api/bus/decide', '/api/bus/seen', '/api/bus/', '/api/bus/inbox/x', '/api/bus/request', '/api/bus/raise']) {
assert.equal((await fetch(base + path)).status, 404, path);
@@ -192,7 +237,7 @@ test('bus refusals map to statuses and never carry a path or a stack', async t =
const unset = await fetch(`http://127.0.0.1:${none.address().port}/api/bus/agents`);
assert.deepEqual([unset.status, (await unset.json()).error], [503, 'not-configured']);
// Row 67 (#1560): with no bus, a bad argument is still a 400; the request is checked first.
for (const [query, status, error] of [['decisions?list=bad', 400, 'invalid-request'], ['decisions?list=roles', 503, 'not-configured'], ['trail?subject=%3C', 400, 'invalid-request']]) {
for (const [query, status, error] of [['decisions?list=bad', 400, 'invalid-request'], ['decisions?list=roles', 503, 'not-configured'], ['trail?subject=%3C', 400, 'invalid-request'], ['replies?id=%3C', 400, 'invalid-request'], ['replies?id=m-1', 503, 'not-configured']]) {
const r = await fetch(`http://127.0.0.1:${none.address().port}/api/bus/${query}`);
assert.deepEqual([query, r.status, (await r.json()).error], [query, status, error]);
}
@@ -250,6 +295,7 @@ test('busReader takes --business, else the live bus host, else refuses', async t
assert.equal((await pinned.inbox()).request.business, 'pinned');
assert.deepEqual((await pinned.decisions('recent')).request, { business: 'pinned', verb: 'decisions', args: { list: 'recent' } });
assert.deepEqual((await pinned.credentials()).request, { business: 'pinned', verb: 'credentials', args: {} });
assert.deepEqual((await pinned.replies('m-1')).request, { business: 'pinned', verb: 'replies', args: { id: 'm-1' } });
const reader = busReader({ dataRoot, socket: 'sock', cli });
const code = () => reader.agents().then(() => 'resolved', e => e.code);
packages/webui/tests/send-browser.test.mjs
+135 -1
View File
@@ -13,9 +13,11 @@ import { Store } from '../../bus/src/store.mjs';
import { Broker } from '../../bus/src/broker.mjs';
import { views } from '../../bus/src/views.mjs';
import { startServer } from '../src/serve.mjs';
import { BusReadError } from '../src/bus.mjs';
import { browser } from './browser.mjs';
import { fixture, close } from './fixture.mjs';
const HOSTILE = '<script>window.injected=1</script><img src=x onerror="window.injected=2"> \u001b[31mred \u202eevil';
const businesses = {
demo: {
id: 'demo', human: 'jason', arbiters: { technical: 'cto', delivery: 'pm' },
@@ -100,7 +102,7 @@ test('the send box: held roles, the draft and caret across a refresh, each recei
await b.navigate(`${base}#/inbox`);
await wait('document.querySelector("#send-form")');
assert.equal(await text('#send-section h2'), 'Message a role');
assert.match(await text('#send-section'), /does not show yet\. The next mosaic talk in a terminal prints it, after sending its own message\./);
assert.match(await text('#send-section'), /Replies to it show under the receipt\. The Console only shows them and never marks one delivered or read, so the next mosaic talk in a terminal still prints each one\./);
assert.deepEqual(await b.evaluate('[...document.querySelectorAll("#send-to option")].map(o => [o.value, o.textContent])'), [['', 'Choose a held role'], ['coder', 'coder'], ['pm', 'pm']]);
// Nothing to send is refused in the page, without a request.
@@ -193,3 +195,135 @@ test('the send box: held roles, the draft and caret across a refresh, each recei
rmSync(root, { recursive: true, force: true });
}
});
// Row 81 (#1575): a sent receipt lists the replies to that message, read with `replies` on each
// render. The read writes no delivery or read row, so the human's next message.receive
// (mosaic talk) still gets every reply the page showed.
test('a sent receipt lists its replies as the page refreshes, oldest first and capped, and marks none delivered or read', { timeout: 120000 }, async () => {
const root = mkdtempSync(join(tmpdir(), 'webui-replies-browser-'));
const store = new Store(root), b0 = new Broker({ store, businesses });
const caps = {};
for (const role of ['pm', 'coder']) b0.request(caps[role] = b0.bindLaunch({ business: 'demo', role, run: `${role}-run`, harness: 'pi', address: `${role}-run` }), { verb: 'role.claim' });
const human = b0.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
const sent = b0.request(human, { verb: 'message.send', args: { to: 'coder', body: 'Please look at row 81.' } });
const other = b0.request(human, { verb: 'message.send', args: { to: 'pm', body: 'Another question' } });
const reader = b0.bindReader({ business: 'demo' });
// A broker refusal comes back from the transport as a BusReadError carrying its code. While
// `hold` is set, a replies read waits for it, counted in `held`.
let hold = null, held = 0;
const bus = views({
call: async (verb, args = {}) => {
if (verb === 'replies' && hold) { held++; await hold.promise; }
try { return structuredClone(b0.request(reader, { verb, args })); } catch (err) { throw new BusReadError(err.code); }
},
});
const f = await fixture(), b = await browser();
const web = await startServer({ port: 0, board: f.boardURL, bus }), base = `http://127.0.0.1:${web.address().port}/`;
const wait = (expr, tries = 100) => b.evaluate(`(async()=>{for(let i=0;i<${tries};i++){if(${expr})return true;await new Promise(r=>setTimeout(r,50))}throw new Error('Condition timed out: '+${JSON.stringify(expr)})})()`);
const text = sel => b.evaluate(`document.querySelector(${JSON.stringify(sel)})?.textContent ?? null`);
const set = (sel, v) => b.evaluate(`(()=>{const e=document.querySelector(${JSON.stringify(sel)});e.value=${JSON.stringify(v)};e.dispatchEvent(new Event('input',{bubbles:true}));return true})()`);
const send = async (id, body) => {
await set('#send-body', body);
await b.evaluate(`window.__answers.push(${JSON.stringify({ status: 200, body: { id, at: new Date().toISOString() } })}), true`);
await b.evaluate('document.querySelector("#send-form [data-send]").click(), true');
await wait('document.querySelector("#send-section .receipt") && document.querySelector("#send-form [data-send]").getAttribute("aria-disabled") === "false"');
};
const refresh = async () => {
await b.evaluate('(()=>{document.querySelector("#send-section").dataset.before="1";document.querySelector("#refresh").click();return true})()');
await wait('!document.querySelector("#send-section")?.dataset.before && !document.querySelector("#bus-view [aria-busy]")');
};
const items = () => b.evaluate('[...document.querySelectorAll("#send-section .s1-replies li")].map(li => [...li.querySelectorAll("p")].map(p => p.textContent).concat(li.querySelector("a").getAttribute("href")))');
const reply = (role, to, body) => b0.request(caps[role], { verb: 'message.send', args: { to: 'human', body, class: 'RESULT', ...(to ? { in_reply_to: to } : {}) } });
const deliveries = () => store.all('SELECT * FROM deliveries ORDER BY seq').map(d => ({ ...d }));
const NONE = 'No reply yet. One shows here when the role answers, as the page refreshes.';
try {
await b.call('Page.addScriptToEvaluateOnNewDocument', { source: STUB });
await b.viewport(1280);
await b.navigate(`${base}#/inbox`);
await wait('document.querySelector("#send-form")');
await set('#send-to', 'coder');
// Straight after the send there is no reply.
await send(sent.id, 'Please look at row 81.');
assert.match(await text('#send-section .receipt'), new RegExp(`^Sent to coder .*, message ${sent.id.slice(0, 8)}\\. Its trail\\.$`));
assert.match(await text('#send-section'), new RegExp(NONE.replace(/\./g, '\\.')));
assert.deepEqual(await items(), []);
// Replies to the sent message show on Read again, oldest first, as text. A reply to another
// message and a message with no in_reply_to don't.
const first = reply('coder', sent.id, `Done: 3 files. ${HOSTILE}`);
reply('pm', other.id, 'Another answer');
reply('pm', null, 'fyi');
const second = reply('pm', sent.id, 'Second look.');
let before = deliveries();
await refresh();
assert.deepEqual(deliveries(), before, 'no delivered or read row');
assert.equal(await text('#send-section h3'), 'Replies 2');
const two = await items();
assert.deepEqual(two.map(([meta, , href]) => [meta.split(' ')[0], / · RESULT · trail$/.test(meta), href]), [['coder', true, `#/trail/message/${first.id}`], ['pm', true, `#/trail/message/${second.id}`]]);
assert.match(two[0][1], /^Done: 3 files\. <script>window\.injected=1<\/script><img src=x onerror="window\.injected=2">/);
assert.equal(two[1][1], 'Second look.');
assert.equal(await b.evaluate('!!window.injected || !!document.querySelector("#send-section script, #send-section img")'), false);
assert.match(await text('#send-section .receipt'), /^Sent to coder /);
assert.doesNotMatch(await text('#send-section'), /The first 50 replies/);
// The page's own refresh reads them too, with no click.
const third = reply('coder', sent.id, 'Third.');
before = deliveries();
await wait('document.querySelectorAll("#send-section .s1-replies li").length === 3', 300);
assert.deepEqual((await items()).map(i => i[1]).slice(1), ['Second look.', 'Third.']);
// Showing them wrote nothing: mosaic talk's message.receive still gets each one, and once
// the human has received and read them they are still listed.
assert.deepEqual(deliveries(), before, 'no delivered or read row');
const got = b0.request(human, { verb: 'message.receive' });
assert.deepEqual(got.filter(m => m.in_reply_to === sent.id).map(m => m.id), [first.id, second.id, third.id]);
for (const m of got) b0.request(human, { verb: 'message.read', args: { id: m.id } });
await refresh();
assert.equal((await items()).length, 3);
// The broker's cap: 51 replies show the first 50 and say so.
for (let i = 0; i < 48; i++) reply(i % 2 ? 'pm' : 'coder', sent.id, `reply ${i}`);
await refresh();
const capped = await items();
assert.equal(capped.length, 50);
assert.deepEqual([capped[0][1].slice(0, 5), capped[3][1], capped[49][1]], ['Done:', 'reply 0', 'reply 46']);
assert.equal(await text('#send-section h3'), 'Replies 50');
assert.match(await text('#send-section'), /The first 50 replies show here\./);
// A sent id the bus doesn't know: its replies can't be read, which the receipt says; the
// Inbox stays fresh.
await send('00000000-0000-4000-8000-000000000000', 'Lost?');
assert.match(await text('#send-section'), new RegExp(NONE.replace(/\./g, '\\.')));
await refresh();
assert.match(await text('#send-section'), /The replies to it could not be read\. Read again to retry; mosaic talk in a terminal prints them\./);
assert.deepEqual(await items(), []);
assert.equal(await b.evaluate('!!document.querySelector("#bus-view .s1-stale, #bus-view .banner")'), false);
assert.match(await text('#send-section .receipt'), /, message 00000000\. Its trail\.$/);
// A new send starts its own list.
await send(sent.id, 'Once more.');
assert.deepEqual(await items(), []);
await refresh();
assert.equal((await items()).length, 50);
// A send that finishes while a replies read is out keeps its own list: that read's answer is
// the earlier message's replies, so it is dropped and the next render reads the new one's.
let release;
hold = { promise: new Promise(r => { release = r; }) };
await b.evaluate('(()=>{document.querySelector("#bus-view h1").dataset.before="1";document.querySelector("#refresh").click();return true})()');
for (let i = 0; i < 100 && !held; i++) await new Promise(r => setTimeout(r, 50));
assert.equal(held, 1);
await send(other.id, 'Another?');
hold = null; release();
await wait('!document.querySelector("#bus-view h1")?.dataset.before && !document.querySelector("#bus-view [aria-busy]")');
assert.match(await text('#send-section .receipt'), new RegExp(`, message ${other.id.slice(0, 8)}\\. Its trail\\.$`));
assert.deepEqual(await items(), []);
assert.match(await text('#send-section'), new RegExp(NONE.replace(/\./g, '\\.')));
await refresh();
assert.deepEqual((await items()).map(i => i[1]), ['Another answer']);
} finally {
await b.close(); await close(web); await f.close(); store.close();
rmSync(root, { recursive: true, force: true });
}
});