diff --git a/packages/bus/README.md b/packages/bus/README.md index 26220494f..f31c699a4 100644 --- a/packages/bus/README.md +++ b/packages/bus/README.md @@ -144,7 +144,7 @@ an operator inspect content and delivery evidence after an uncertain result. | `decision.resolve` | `{id,choice,note?}`; option must exist, resolver must be routed role or outside-agent human CLI. Sends the raising role a DECISION message (Decision close messages, below). | | `decision.seen` | `{id,note?}`; acknowledgement by resolver, not a resolution. | | `decision.withdraw` | `{id,note?}`; raiser or human. Never rewrites the decision. A human withdraw sends the raising role a DECISION message; the raising run's own withdraw sends none. | -| `message.send` | `{to,body,class?,in_reply_to?,corrects?,decision?}`; returns `{id,request}`. Human instruction's `request` is its `human.input` event ID. | +| `message.send` | `{to,body,class?,in_reply_to?,corrects?,decision?}`; returns `{id,request}`, plus `waiting` when no run holds the target (Launch notices, below). Human instruction's `request` is its `human.input` event ID. | | `message.receive` | No args; up to 100 undelivered messages for current holder, including originating `request` for human instructions. A human decision close's message carries the close's `human.input` answer event as its `request`; a role close's has `null`. | | `message.read` | `{id}`; recipient holder's acknowledgement; no new holder can acknowledge old holder's delivery. | | `review.request` | `{task_ref,commit,to,note?,decision?}`; action `review.request` (row 93, #1587). `commit` must be the task's latest pushed commit: an older one refuses `stale-commit`, any other `unknown-commit`. `to` must be another role holding `review.verdict` (`unknown-role`, `self-review`, `not-a-reviewer`). Writes a `review.requested` event `{task_ref,project,ref,commit,to,message,held}` and sends `to` a REVIEW-REQUEST message naming it. `held` is the run id holding the `to` role when the request is routed (row 105, #1599), or `null` when no run holds it: the message then waits for a claim, and the requester should tell whoever launches roles. Returns `{request,message,commit,ref,held}`. | @@ -354,6 +354,43 @@ message the close's `human.input` answer event as its `request`, so the role's turn cites it as it cites a human instruction; a role close's `request` is `null`. The message also appears in the decision's `trail`. +### Launch notices (row 146, #1640) + +A `message.send` whose target no run holds tells the business's `launch.by` +instance, once, in the same transaction as the message. The notice is an +ordinary message: `from_role '_bus'`, `from_run 'bus'`, to `launch.by`, class +`RESULT`, `in_reply_to` the waiting message, no `decision`. It is routed like +any other, so a stopped `launch.by` gets it at its next claim. The body is + +``` +Message from to waits: has no running session. Launch within the launch block's limits, or tell why not with message.send in_reply_to . +``` + +`_bus` can't be a role: an instance name matches the business `ID_PATTERN`, +which starts with `[a-z0-9]`. So `message.send` to `_bus` refuses +`unknown-role`, and a caller can't set `from_role` or `from_run` (unknown keys +are `invalid-request`). The class is `RESULT` because the runner sends an +automatic RESULT reply to every other class, and the bus would refuse a reply +to `_bus`; the runner would log it and skip the `message.read`. On a RESULT +the runner takes one turn, sends nothing and marks it read. + +The result's `waiting` is `{role,launcher,told,launcherRunning}`: the target, +`launch.by` (`null` with no launch block), whether the notice went out, and +whether a run holds `launch.by` (`null` with no launch block). No `waiting` +field means a run held the target, or the message went to the human, the +tester, or a thread instance inside its thread (row 121's thread launcher starts +that session). `told` is `false` and no notice goes out for a message to +`launch.by` itself, a message from `launch.by` (the result tells it), or a +tester's (a PM test authorizes nothing). + +Only `message.send` sends a notice, and only for the message it stores, so a +re-route at a later claim, a broker restart or a release and claim sends none. +A run that is launched but hasn't claimed counts as no holder, so the notice +can go out while it starts; the launcher's `instance-running` refusal stops a +second launch. The Console's replies under a human's sent message include the +notice, since it is `in_reply_to` that message. `review.request`, +`thread.send` and a decision close's message send no notice. + ## Human and reader paths `src/human-cli.mjs ` is S4's transport shim, not another product CLI. diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index 703fb6373..61febf532 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -111,6 +111,10 @@ const REVIEW_KINDS = ['review.requested', 'review.verdict']; // Row 121 (#1615): an instance of this definition talks with the user in threads (REQ-THREAD-1). // Its runs claim a thread, not the role, and every message to or from it names its thread (REQ-MSG-1). const THREAD_DEFINITION = 'assistant'; +// Row 146 (#1640): the sender of the bus's notice to launch.by. A role instance name matches the +// business ID_PATTERN, which starts with [a-z0-9], so no instance can be named this. +const NOTICE_ROLE = '_bus'; +const NOTICE_RUN = 'bus'; // How many messages `thread.messages` returns by default and at most, oldest first. export const THREAD_PAGE = 50; export const THREAD_PAGE_MAX = 200; @@ -1313,7 +1317,8 @@ export class Broker { const request = s.human ? this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) : null; - return { id: mid, request }; + const waiting = this.#waiting(s, mid, a.to, thread); + return waiting ? { id: mid, request, waiting } : { id: mid, request }; } case 'message.receive': { keys(a, []); @@ -1823,6 +1828,43 @@ export class Broker { if (thread) this.#store.run('INSERT INTO message_threads(message,thread) VALUES(?,?)', mid, thread); this.#delivery(mid, 'routed', this.#recipient(business, to, thread)); } + // Row 146 (#1640): a message.send that nobody holds the target for waits for a claim. Null when + // the message was routed to a holder, or is the human's, a tester's, or in a thread to a thread + // instance (row 121's launcher starts that session). Otherwise it says whether launch.by was + // told and whether launch.by itself has a holder. A run launched but not yet claimed counts as + // no holder; the instance-running refusal stops a second launch. launch.by isn't told of a + // message to it, of its own message (this result tells it) or of a tester's (a PM test + // authorizes nothing). + #waiting(s, mid, to, thread) { + if (to === 'human' || to === 'tester' || (thread && this.#threaded(s.business, to))) return null; + if (this.#recipient(s.business, to, thread)) return null; + const by = this.#business(s.business).launch?.by ?? null; + const sender = this.#sender(s); + const told = by !== null && to !== by && sender !== by && sender !== 'tester'; + if (told) this.#notice(s.business, mid, by, to, sender); + return { role: to, launcher: by, told, launcherRunning: by === null ? null : this.#recipient(s.business, by, null) !== null }; + } + // One notice per waiting message: only message.send calls this, once for the message it stores, so + // a re-route at a claim or a broker restart never sends another. Its class is RESULT: the runner + // takes a turn on it and sends no automatic reply, which NOTICE_ROLE could not receive. + #notice(business, mid, by, to, sender) { + const nid = randomUUID(); + this.#store.run( + 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', + nid, + this.#now(), + business, + NOTICE_ROLE, + NOTICE_RUN, + by, + 'RESULT', + mid, + null, + null, + `Message ${mid} from ${sender} to ${to} waits: ${to} has no running session. Launch ${to} within the launch block's limits, or tell ${sender} why not with message.send in_reply_to ${mid}.`, + ); + this.#route(business, nid, by, null); + } #delivery(mid, op, h, transport = null) { this.#store.run( 'INSERT INTO deliveries(message,at,op,holder_run,transport,address) VALUES(?,?,?,?,?,?)', diff --git a/packages/bus/tests/launch-notice.test.mjs b/packages/bus/tests/launch-notice.test.mjs new file mode 100644 index 000000000..3ac83e702 --- /dev/null +++ b/packages/bus/tests/launch-notice.test.mjs @@ -0,0 +1,212 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { Store } from '../src/store.mjs'; +import { Broker } from '../src/broker.mjs'; +import { ID_PATTERN } from '../../business/src/vocabulary.mjs'; + +// Row 146 (#1640): a message.send to a role nobody holds tells the business's launch.by +// instance once, and the result says so. The assistant's lists are the shipped ones, as in +// threads.test.mjs, so the thread case runs under the lists the stack runs. +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: [] } }, + pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } }, + coder: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } }, +}; +const definitions = { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging }, pm: {}, coder: {} }; +const demo = (extra = {}) => ({ + demo: { + id: 'demo', + human: 'jason', + arbiters: { technical: 'pm', delivery: 'chief' }, + roles, + definitions, + projects: { stack: { root: '/nonexistent' } }, + launch: { by: 'chief', instances: ['pm', 'coder'], models: {}, limits: {} }, + vars: { 'bus.tester': true }, + ...extra, + }, +}); + +function setup(t, businesses = demo()) { + const root = mkdtempSync(join(tmpdir(), 'bus-notice-')); + let store = new Store(root); + t.after(() => { + store.close(); + rmSync(root, { recursive: true, force: true }); + }); + const ctx = { root }; + const open = () => { + ctx.b = new Broker({ store, businesses }); + ctx.caps = { human: ctx.b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }) }; + }; + open(); + ctx.store = () => store; + // A broker restart: close the store and open it again under a new Broker. + ctx.restart = () => { + store.close(); + store = new Store(root); + open(); + }; + ctx.call = (who, verb, args = {}) => ctx.b.request(ctx.caps[who], { verb, args }); + ctx.launch = (name, role, extra = {}) => { + ctx.caps[name] = ctx.b.bindLaunch({ business: 'demo', role, run: `${name}-run`, harness: 'pi', ...extra }); + return ctx.caps[name]; + }; + ctx.claim = (name) => ctx.call(name, 'role.claim'); + ctx.notices = () => store.all("SELECT * FROM messages WHERE from_role='_bus' ORDER BY seq"); + return ctx; +} + +test('a message to a stopped role sends launch.by one notice, and the result says so', (t) => { + const ctx = setup(t); + const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'is the plan current?', class: 'REQUEST' }); + assert.deepEqual(sent.waiting, { role: 'pm', launcher: 'chief', told: true, launcherRunning: false }); + const [n, ...more] = ctx.notices(); + assert.deepEqual(more, []); + assert.equal(n.from_run, 'bus'); + assert.equal(n.to_role, 'chief'); + assert.equal(n.class, 'RESULT'); + assert.equal(n.in_reply_to, sent.id); + assert.equal(n.decision, null); + assert.equal( + n.body, + `Message ${sent.id} from human to pm waits: pm has no running session. Launch pm within the launch block's limits, or tell human why not with message.send in_reply_to ${sent.id}.`, + ); + assert.equal(ctx.store().get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed'", n.id).holder_run, null); + // chief comes up and receives it; with chief running, the next result says so. + ctx.launch('chief', 'chief'); + ctx.claim('chief'); + assert.deepEqual(ctx.call('chief', 'message.receive').map((m) => m.id), [n.id]); + assert.deepEqual(ctx.call('human', 'message.send', { to: 'coder', body: 'and you?' }).waiting, { + role: 'coder', + launcher: 'chief', + told: true, + launcherRunning: true, + }); + assert.equal(ctx.notices().length, 2); + // A role's message to a stopped role tells launch.by too. + ctx.launch('coder', 'coder'); + ctx.claim('coder'); + assert.equal(ctx.call('coder', 'message.send', { to: 'pm', body: 'blocked on you' }).waiting.told, true); + assert.equal(ctx.notices().at(-1).body.startsWith('Message '), true); + assert.match(ctx.notices().at(-1).body, / from coder to pm waits: /); +}); + +test('a message to a running role sends no notice and its result is unchanged', (t) => { + const ctx = setup(t); + ctx.launch('pm', 'pm'); + ctx.claim('pm'); + const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'hello' }); + assert.deepEqual(Object.keys(sent).sort(), ['id', 'request']); + assert.deepEqual(ctx.notices(), []); +}); + +test('a re-route at a claim, a release and claim, and a broker restart send no second notice', (t) => { + const ctx = setup(t); + const sent = ctx.call('human', 'message.send', { to: 'pm', body: 'waits' }); + assert.equal(ctx.notices().length, 1); + ctx.launch('pm', 'pm'); + ctx.claim('pm'); + assert.equal(ctx.notices().length, 1, 'role.claim re-routes the waiting message only'); + assert.equal( + ctx.store().get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", sent.id).holder_run, + 'pm-run', + ); + ctx.call('pm', 'role.release'); + ctx.restart(); + ctx.launch('pm2', 'pm'); + ctx.claim('pm2'); + assert.deepEqual(ctx.call('pm2', 'message.receive').map((m) => m.id), [sent.id]); + assert.equal(ctx.notices().length, 1, 'a restarted broker and a new claim send none'); +}); + +test('a run launched but not yet claimed counts as no holder', (t) => { + const ctx = setup(t); + ctx.launch('pm', 'pm'); + assert.equal(ctx.call('human', 'message.send', { to: 'pm', body: 'early' }).waiting.told, true); + assert.equal(ctx.notices().length, 1); +}); + +test('no notice for a message to launch.by, from it, from a tester, to the human, or in a thread to a thread instance', (t) => { + const ctx = setup(t); + // To launch.by while it is stopped: the result says so, and nobody is told. + assert.deepEqual(ctx.call('human', 'message.send', { to: 'chief', body: 'are you there?' }).waiting, { + role: 'chief', + launcher: 'chief', + told: false, + launcherRunning: false, + }); + // From launch.by: the result tells it. + ctx.launch('chief', 'chief'); + ctx.claim('chief'); + assert.deepEqual(ctx.call('chief', 'message.send', { to: 'pm', body: 'start on it' }).waiting, { + role: 'pm', + launcher: 'chief', + told: false, + launcherRunning: true, + }); + // From a tester: a PM test authorizes nothing, so it wakes no launcher. + ctx.caps.tester = ctx.b.bindTester({ business: 'demo' }); + assert.equal(ctx.call('tester', 'message.send', { to: 'pm', body: '[mos test] ping' }).waiting.told, false); + // To the human or the tester, which no run holds: no waiting field at all. + ctx.launch('coder', 'coder'); + ctx.claim('coder'); + assert.equal(ctx.call('coder', 'message.send', { to: 'human', body: 'fyi' }).waiting, undefined); + const ping = ctx.call('tester', 'message.send', { to: 'coder', body: '[mos test] are you up?' }); + assert.equal(ping.waiting, undefined); + assert.equal(ctx.call('coder', 'message.send', { to: 'tester', body: 'up', in_reply_to: ping.id }).waiting, undefined); + // In a thread to a thread instance: row 121's launcher starts that session. + const project = ctx.call('human', 'project.create', { name: 'work', repo: 'stack' }).id; + const A = ctx.call('human', 'thread.create', { project, name: 'alpha' }).id; + const asked = ctx.call('human', 'thread.send', { thread: A, body: 'what is left?' }); + const answer = ctx.call('chief', 'message.send', { to: 'assistant', body: 'two rows', in_reply_to: asked.id }); + assert.equal(answer.waiting, undefined); + assert.equal(ctx.store().get('SELECT thread FROM message_threads WHERE message=?', answer.id).thread, A); + assert.deepEqual(ctx.notices(), []); +}); + +test('a business with no launch block tells nobody and says there is no launcher', (t) => { + const ctx = setup(t, demo({ launch: undefined })); + assert.deepEqual(ctx.call('human', 'message.send', { to: 'pm', body: 'hello' }).waiting, { + role: 'pm', + launcher: null, + told: false, + launcherRunning: null, + }); + assert.deepEqual(ctx.notices(), []); +}); + +test('the notice sender stays internal: no role is named _bus and no caller can send as it or to it', (t) => { + assert.equal(ID_PATTERN.test('_bus'), false, 'a business file cannot declare an instance named _bus'); + const ctx = setup(t); + ctx.launch('coder', 'coder'); + ctx.claim('coder'); + ctx.call('human', 'message.send', { to: 'pm', body: 'waits' }); + const [n] = ctx.notices(); + const refused = (who, args, code) => + assert.throws(() => ctx.call(who, 'message.send', { body: 'x', ...args }), { code }, JSON.stringify(args)); + for (const who of ['human', 'coder']) { + refused(who, { to: '_bus' }, 'unknown-role'); + refused(who, { to: 'pm', from_role: '_bus' }, 'invalid-request'); + refused(who, { to: 'pm', from_run: 'bus' }, 'invalid-request'); + refused(who, { to: 'pm', class: '_bus' }, 'invalid-class'); + refused(who, { to: 'pm', in_reply_to: '_bus' }, 'invalid-request'); + refused(who, { to: 'pm', corrects: '_bus' }, 'invalid-request'); + refused(who, { to: 'pm', decision: '_bus' }, 'decision-not-found'); + } + // Answering the notice goes to the sender it names, never back to _bus. + ctx.launch('chief', 'chief'); + ctx.claim('chief'); + refused('chief', { to: '_bus', in_reply_to: n.id }, 'unknown-role'); + const why = ctx.call('chief', 'message.send', { to: 'human', body: 'pm is over its limit', in_reply_to: n.in_reply_to }); + assert.equal(ctx.store().get('SELECT from_role FROM messages WHERE id=?', why.id).from_role, 'chief'); + assert.equal(ctx.notices().length, 1); +}); diff --git a/packages/cli/README.md b/packages/cli/README.md index da6877607..6d9756872 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -65,7 +65,18 @@ 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. `talk --follow` keeps + `mosaic talk` prints it. When no run holds the role (row 146, #1640), + a line after `sent` says so: `pm isn't running, chief has been told` + when the bus told `launch.by`; `pm isn't running, and the business names + no launch.by` with no launch block; `pm isn't running; a tester's message + tells nobody to launch it` for `--tester`. When `launch.by` itself isn't + running, talk prints ` isn't running` (or `isn't running + either` after the told line) and the command that starts it: + `systemctl --user restart mosaic-bus@`, which starts it only if + the unit runs with `--pm` (`scripts/bus-service.sh status ` + shows `pm: on`) and stops running sessions. A hand-started host takes + `mosaic bus stop`, then `mosaic bus start --pm`. No verb + launches `launch.by` on a running host. `--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 diff --git a/packages/cli/src/cli.mjs b/packages/cli/src/cli.mjs index 60a0ba29b..8db1a8b3a 100644 --- a/packages/cli/src/cli.mjs +++ b/packages/cli/src/cli.mjs @@ -212,6 +212,18 @@ function formatMessage(m) { return `${m.from_role} (${meta}) ${shortId(m.id)}:\n${m.body.trimEnd()}\n`; } +// Row 146 (#1640): what the bus said about a message whose role has no running session. Only +// a restart of a unit with --pm starts launch.by; a human verb can't on a running host. +function waitingLines(w, business) { + const restart = (role, either = "") => + `${role} isn't running${either}. systemctl --user restart mosaic-bus@${business()} starts it if the unit runs with --pm (scripts/bus-service.sh status ${business()} shows pm: on). The restart stops running sessions.\n`; + if (w.launcher === null) return `${w.role} isn't running, and the business names no launch.by\n`; + if (w.role === w.launcher) return restart(w.role); + // A human's message to another role always tells launch.by; a tester's never does. + if (!w.told) return `${w.role} isn't running; a tester's message tells nobody to launch it\n`; + return `${w.role} isn't running, ${w.launcher} has been told\n` + (w.launcherRunning ? "" : restart(w.launcher, " either")); +} + // `mosaic talk `: a REQUEST to the role, then the human's inbox // until the reply arrives or --wait runs out. --follow keeps receiving after // that reply; follow starts the same receive loop without sending. @@ -231,6 +243,7 @@ async function talk(rest, io, deps, receiveOnly = false) { if (!receiveOnly) { sent = await call("message.send", { to, body, class: "REQUEST" }); 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; } const pollMs = deps.talkPollMs ?? TALK_POLL_MS; diff --git a/packages/cli/tests/helpers.mjs b/packages/cli/tests/helpers.mjs index f8026bdfd..79f653687 100644 --- a/packages/cli/tests/helpers.mjs +++ b/packages/cli/tests/helpers.mjs @@ -77,11 +77,11 @@ export const BUSINESSES = { // An in-process broker with a claimed coder and the human's capability. // transport(business) mirrors humanTransport: errors carry the bus code. -export function broker(t) { +export function broker(t, businesses = BUSINESSES) { const root = tmp(t, "mosaic-cli-bus-"); const store = new Store(root); t.after(() => store.close()); - const b = new Broker({ store, businesses: BUSINESSES }); + 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 }); diff --git a/packages/cli/tests/tester.test.mjs b/packages/cli/tests/tester.test.mjs index 0994b12cf..5ccf7705c 100644 --- a/packages/cli/tests/tester.test.mjs +++ b/packages/cli/tests/tester.test.mjs @@ -11,10 +11,10 @@ import { BUSINESSES, humanEnv, io, tmp } from "./helpers.mjs"; // Row 110 (#1604): --tester binds as the business's tester instead of as the // human. The in-process broker here has bus.tester on; the real tester-cli.mjs // runs in packages/bus/tests/tester.test.mjs. -function setup(t) { +function setup(t, extra = {}) { const store = new Store(tmp(t, "mosaic-cli-tester-")); t.after(() => store.close()); - const businesses = { demo: { ...BUSINESSES.demo, vars: { "bus.tester": true } } }; + const businesses = { demo: { ...BUSINESSES.demo, vars: { "bus.tester": true }, ...extra } }; 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" }); @@ -126,3 +126,12 @@ test("with no stand-in transport, --tester runs tester-cli.mjs on the data root' await main(["agents", "--tester", "--business", "demo"], inAgent(), { system: { dataRoot }, testerSpawn }); assert.deepEqual(seen, [{ argv: [TESTER_CLI, join(dataRoot, "bus", "broker.sock")], input: { business: "demo", verb: "agents", args: {} } }]); }); + +// Row 146 (#1640): a tester's message to a stopped role tells launch.by nothing, and talk says so. +test("talk --tester to a stopped role says nobody is told to launch it", async (t) => { + const s = setup(t, { launch: { by: "pm", instances: ["cto", "coder"], models: {}, limits: {} } }); + const x = inAgent(); + await main(["talk", "cto", "[mos test] hello", "--tester", "--wait", "0", "--business", "demo"], x, s.deps); + assert.match(x.out.text, /^sent [0-9a-f-]{8} to cto\ncto isn't running; a tester's message tells nobody to launch it\n$/); + assert.equal(s.store.get("SELECT count(*) n FROM messages WHERE from_role='_bus'").n, 0); +}); diff --git a/packages/cli/tests/verbs.test.mjs b/packages/cli/tests/verbs.test.mjs index 6bdbc6793..cb4a295be 100644 --- a/packages/cli/tests/verbs.test.mjs +++ b/packages/cli/tests/verbs.test.mjs @@ -9,10 +9,10 @@ import { main } from "../src/cli.mjs"; import { CliError } from "../src/errors.mjs"; import { startTimeOf } from "../src/host.mjs"; import { Registry } from "../../seat/src/session.mjs"; -import { broker, io, tmp } from "./helpers.mjs"; +import { BUSINESSES, broker, io, tmp } from "./helpers.mjs"; -function setup(t, deps = {}) { - const bus = broker(t); +function setup(t, deps = {}, businesses) { + const bus = broker(t, businesses); const dataRoot = tmp(t); const run = async (argv, x = io(), more = {}) => { try { @@ -81,6 +81,29 @@ test("a late reply the Console has listed still reaches the next talk", async (t assert.deepEqual(s.calls.filter((c) => c.verb === "message.read").map((c) => c.args.id), [answer.id]); }); +// Row 146 (#1640): the lines after "sent" when the role has no running session. +test("talk says when the role isn't running, whether launch.by was told, and how to start launch.by", async (t) => { + const restart = (role, either = "") => + `${role} isn't running${either}. systemctl --user restart mosaic-bus@demo starts it if the unit runs with --pm (scripts/bus-service.sh status demo shows pm: on). The restart stops running sessions.\n`; + const text = async (s, role) => { + const r = await s.run(["talk", role, "hello", "--wait", "0", "--business", "demo"]); + assert.equal(r.code, 0, r.error); + return r.io.out.text.replace(/^sent [0-9a-f]{8} to \S+\n/, ""); + }; + // No launch block: nobody to tell. + const plain = setup(t); + assert.equal(await text(plain, "cto"), "cto isn't running, and the business names no launch.by\n"); + assert.equal(await text(plain, "coder"), "", "a running role gets the sent line only"); + // pm launches: told, and pm itself down too. + const s = setup(t, {}, { demo: { ...BUSINESSES.demo, launch: { by: "pm", instances: ["cto", "coder"], models: {}, limits: {} } } }); + assert.equal(await text(s, "cto"), "cto isn't running, pm has been told\n" + restart("pm", " either")); + assert.equal(await text(s, "pm"), restart("pm")); + assert.equal(s.store.get("SELECT count(*) n FROM messages WHERE from_role='_bus'").n, 1, "a message to launch.by tells nobody"); + call(s.b, s.b.bindLaunch({ business: "demo", role: "pm", run: "pm-run", harness: "pi" }), "role.claim"); + assert.equal(await text(s, "cto"), "cto isn't running, pm has been told\n"); + assert.equal(await text(s, "coder"), ""); +}); + test("talk, stop and launches off/on refuse inside an agent run; bad arguments are usage errors", async (t) => { const s = setup(t); const agent = () => io({ ...io().env, CLAUDECODE: "1" }); diff --git a/packages/harness/tests/runner.test.mjs b/packages/harness/tests/runner.test.mjs index 4e5da7f17..1361cf030 100644 --- a/packages/harness/tests/runner.test.mjs +++ b/packages/harness/tests/runner.test.mjs @@ -6,7 +6,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { spawn, spawnSync } from "node:child_process"; import { once } from "node:events"; -import { existsSync, mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; import { open } from "node:fs/promises"; import { createServer } from "node:net"; import { join } from "node:path"; @@ -89,10 +89,11 @@ test("stateBlock names the launch state with its time and event, and the open ta assert.match(capped, /Open tasks: 9, the first 1 of cap 1\./); }); -async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null } = {}) { +async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy } = {}) { const dir = scratch(t); const store = new Store(dir); - const broker = new Broker({ store, businesses, organization }); + const demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo; + const broker = new Broker({ store, businesses: { demo }, organization }); const brokerSocket = join(dir, "bus.sock"); const server = await serve({ broker, path: brokerSocket }); t.after(async () => { @@ -433,6 +434,32 @@ test("a RESULT gets no automatic reply; failed turns reply with the reason", asy await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", res.id)); }); +// Row 146 (#1640): the bus's notice to launch.by is a RESULT from _bus. The runner takes one turn +// on it, sends no automatic reply and marks it read. Any other class would send a reply to _bus, +// which the bus refuses; the runner would log the failure and skip the read, leaving the notice +// unread for good. +test("the bus's launch notice is one turn with no reply, and is not taken again", async (t) => { + const ctx = await setup(t, { as: "chief", launchBy: "chief" }); + ctx.inbox = []; + const r = startRunner(ctx.runDir, ctx.cap); + t.after(() => r.child.kill("SIGKILL")); + await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='chief' AND op='claim'")); + const sent = ctx.call(ctx.human, "message.send", { to: "coder", body: "is the build green?" }); + assert.equal(sent.waiting.told, true); + const notice = ctx.store.get("SELECT * FROM messages WHERE from_role='_bus' AND in_reply_to=?", sent.id); + assert.equal(notice.to_role, "chief"); + await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", notice.id)); + // A later message is one more turn, so the notice wasn't taken again in between. + const later = ctx.call(ctx.pm, "message.send", { to: "chief", body: "status?" }); + const [reply] = await until(replies(ctx)); + assert.equal(reply.in_reply_to, later.id); + assert.deepEqual(readdirSync(join(ctx.runDir, "turns")).sort(), ["0001.stderr", "0002.stderr"]); + assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='delivered'", notice.id).n, 1); + assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='read'", notice.id).n, 1); + assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE in_reply_to=? AND from_run='chief-run'", notice.id).n, 0); + assert.doesNotMatch(r.stderr(), /reply to .* failed/); +}); + test("SIGTERM during a turn kills the turn's process group and still exits 0", async (t) => { // A long wall clock, so only the stop path can end the turn in time. const ctx = await setup(t, { session: { turnTimeout: 600 } }); diff --git a/roles/chief-of-staff.md b/roles/chief-of-staff.md index 28ef82826..98794e7f8 100644 --- a/roles/chief-of-staff.md +++ b/roles/chief-of-staff.md @@ -16,11 +16,17 @@ product's code, you don't review it and you don't create tasks yourself. - Launch role sessions with the `launch` tool, for the instances and within the limits the business file's launch block lists. A launch outside them is refused, not queued. Launch only role instances. +- A message from `_bus` is the bus telling you that a message waits for + a role with no running session. It names the message, its sender and + the role. Launch that role within the launch block's limits, or tell + the sender why not with a message `in_reply_to` the waiting message. + Don't reply to `_bus`; it receives nothing. - Keep the organization's structure: which role instances exist, who holds them and what they may do. To terminate a role instance, raise a - `role.revoke` decision, naming the instance and saying why. To hire one - or change the structure, raise a `policy.change` decision until the - broker gives you a verb for the organization file. + `role.revoke` decision, naming the instance and saying why. Change the + launch limits, the model table and the model each instance runs with + the `change_organization` tool. To hire an instance or change what one + may do, raise a `policy.change` decision. - Report to the assistant what the roles answer, and every refusal with its exact text.