jason.woltjeandClaude Opus 5.5
179697a8a3
A tester binds with a process proof from the bus's own tester-cli.mjs when the business var bus.tester is true. It may send [mos test] messages to a role, receive replies on the tester address and use the read verbs; every other verb refuses. Messages store from_role tester and write no human.input event. turnRequest() marks them as never Jason's instruction. human.mjs is unchanged. Candidate manifest bdaf7f476abd0b2b5095ab2f7f43e38540e6ba48ee14a02a4f4995a1469b79df (Darkwing round 2, packet9f1563ec1), BASEbadfa52b9. Approved by Filbert (comment 28360) and Sage (comment 28361). Sage gate rerun passed. Co-Authored-By: Claude Opus 5.5 <[email protected]>
400 lines
20 KiB
JavaScript
400 lines
20 KiB
JavaScript
#!/usr/bin/env node
|
|
// mosaic inbox | decide | tasks | agents | trail | talk | stop | launches,
|
|
// and mosaic bus start|stop|status. See packages/cli/README.md. Exit codes
|
|
// are in src/errors.mjs.
|
|
//
|
|
// The human commands read and resolve through the broker's human transport
|
|
// (src/transport.mjs), never the SQLite file. `bus start` is the trusted
|
|
// host (src/host.mjs) the systemd unit mosaic-bus@<business> runs; with
|
|
// --pm it also launches the business's PM (src/launcher.mjs).
|
|
|
|
import { lstatSync, readFileSync } from "node:fs";
|
|
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 { CliError } from "./errors.mjs";
|
|
import { bootConfig, loadSystem, socketPath } from "./config.mjs";
|
|
import { humanTransport, refuseInsideAgent, testerTransport } from "./transport.mjs";
|
|
import { formatAgents, formatDecision, formatInbox, formatTasks, formatTrail, shortId } from "./format.mjs";
|
|
import { hostStatus, readHostState, startHost, stopHost } from "./host.mjs";
|
|
import { createLauncher } from "./launcher.mjs";
|
|
import { quitLine, runApp } from "./app.mjs";
|
|
import { readSessions, stopSession } from "../../seat/src/session.mjs";
|
|
import { SeatError } from "../../seat/src/seat.mjs";
|
|
|
|
export const USAGE = `usage:
|
|
mosaic inbox [--business <id>] [--json] [--tester]
|
|
mosaic decide <decision> <option> [--note <text>] [--yes] [--business <id>]
|
|
mosaic tasks [--business <id>] [--json] [--tester]
|
|
mosaic agents [--business <id>] [--json] [--tester]
|
|
mosaic trail <task|decision> [--business <id>] [--json] [--tester]
|
|
mosaic talk <role> <text> [--wait <seconds> | --follow] [--business <id>] [--tester]
|
|
mosaic follow [--business <id>] [--tester]
|
|
mosaic stop <run>
|
|
mosaic launches off|on [--business <id>]
|
|
mosaic launches list [--json]
|
|
mosaic bus start <business> [--pm]
|
|
mosaic bus stop
|
|
mosaic bus status [--json]`;
|
|
|
|
// packages/seat/src/cli.mjs USAGE, which that module does not export (it runs
|
|
// its main on import). tests/app.test.mjs fails if the two drift.
|
|
export const SEAT_USAGE = [
|
|
"usage: mosaic launch <seat|seat-dir> [--task TEXT] [--project NAME] [--workspace PATH] [--harness NAME] [--repo PATH] [--config PATH] [-- args...]",
|
|
" mosaic seat task <seat> <text> [--by NAME] [--layout repo|fleet|unknown] [--config PATH]",
|
|
].join("\n");
|
|
|
|
const usage = (why) => new CliError(why ? `${why}\n${USAGE}` : USAGE, 4);
|
|
|
|
function parse(args, { flags = [], values = [] }) {
|
|
const out = { positional: [], flags: new Set(), values: {} };
|
|
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 usage(`${a} needs one value`);
|
|
out.values[a] = args[++i];
|
|
} else if (a.startsWith("--")) throw usage(`unknown option ${a}`);
|
|
else out.positional.push(a);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
// The notifier's private config: `<dataRoot>/notify/<business>/notify.json`,
|
|
// 0600, `{"notifyVersion": 1, "binding": "<discord binding>" | null}`.
|
|
// Missing refuses, so a host never starts without someone deciding whether
|
|
// it DMs; null runs the host without a notifier.
|
|
export function readNotifyConfig(dataRoot, business) {
|
|
const file = join(dataRoot, "notify", business, "notify.json");
|
|
let st;
|
|
try {
|
|
st = lstatSync(file);
|
|
} catch {
|
|
throw new CliError(`no notifier config at ${file}; write {"notifyVersion": 1, "binding": "<discord binding>"} there, mode 0600, or "binding": null to run without DMs`, 3);
|
|
}
|
|
if (!st.isFile() || st.uid !== process.getuid() || (st.mode & 0o777) !== 0o600) {
|
|
throw new CliError(`notifier config must be a regular file, mode 0600, owned by this user: ${file}`, 3);
|
|
}
|
|
let doc;
|
|
try {
|
|
doc = JSON.parse(readFileSync(file, "utf8"));
|
|
} catch {
|
|
throw new CliError(`notifier config is not JSON: ${file}`, 3);
|
|
}
|
|
const keys = doc && typeof doc === "object" && !Array.isArray(doc) ? Object.keys(doc).sort().join(",") : "";
|
|
if (keys !== "binding,notifyVersion" || doc.notifyVersion !== 1 || (doc.binding !== null && !(typeof doc.binding === "string" && BINDING_NAME.test(doc.binding)))) {
|
|
throw new CliError(`notifier config must be exactly {"notifyVersion": 1, "binding": <binding name> | null}: ${file}`, 3);
|
|
}
|
|
return doc.binding;
|
|
}
|
|
|
|
// --business, else the running host's business.
|
|
function businessFor(parsed, dataRoot) {
|
|
if (parsed.values["--business"]) return parsed.values["--business"];
|
|
const state = readHostState(dataRoot);
|
|
if (state?.live) return state.business;
|
|
throw usage("no bus host runs here, so name the business with --business");
|
|
}
|
|
|
|
function findDecision(list, ref) {
|
|
const exact = list.find((d) => d.id === ref);
|
|
if (exact) return exact;
|
|
if (ref.length < 8) throw new CliError(`decision reference ${JSON.stringify(ref)} is too short; use at least 8 characters of the id`, 2);
|
|
const hits = list.filter((d) => d.id.startsWith(ref));
|
|
if (hits.length === 1) return hits[0];
|
|
if (hits.length > 1) throw new CliError(`${ref} matches ${hits.length} open decisions; use more of the id`, 2);
|
|
throw new CliError(`no open decision for you matches ${ref}; see mosaic inbox`, 2);
|
|
}
|
|
|
|
async function confirm(io, question) {
|
|
const rl = createInterface({ input: io.stdin, output: io.stdout });
|
|
try {
|
|
return /^(y|yes)$/i.test((await rl.question(question)).trim());
|
|
} finally {
|
|
rl.close();
|
|
}
|
|
}
|
|
|
|
async function decide(parsed, call, io) {
|
|
if (parsed.positional.length !== 2) throw usage("decide takes a decision and an option");
|
|
const [ref, choice] = parsed.positional;
|
|
const ask = parsed.flags.has("--yes")
|
|
? null
|
|
: (question) => {
|
|
if (!io.stdin.isTTY) throw usage("stdin is not a terminal; pass --yes to resolve without the prompt");
|
|
return confirm(io, question);
|
|
};
|
|
await resolveDecision({ call, ref, choice, note: parsed.values["--note"], write: (s) => io.stdout.write(s), ask });
|
|
}
|
|
|
|
// The decide flow, shared by `mosaic decide` and the app's Inbox (src/app.mjs):
|
|
// re-read the inbox, find the decision and the option, write the decision
|
|
// and what the choice means, ask (null skips the question, as --yes does),
|
|
// then resolve. Returns the decision.
|
|
export async function resolveDecision({ call, ref, choice, note, write, ask }) {
|
|
const d = findDecision(await call("inbox"), ref);
|
|
const option = d.options.find((o) => o.key === choice);
|
|
if (!option) throw new CliError(`decision ${shortId(d.id)} has no option ${JSON.stringify(choice)}; its options are ${d.options.map((o) => o.key).join(", ")}`, 2);
|
|
write(formatDecision(d));
|
|
const effect = d.authorization ? (choice === d.authorization.approvalChoice ? "this authorizes the action" : "this declines the action") : null;
|
|
write(`your choice: ${choice} (${option.text})${effect ? `; ${effect}` : ""}\n`);
|
|
if (ask && !(await ask(`resolve ${shortId(d.id)} with ${choice}? [y/N] `))) throw new CliError("not resolved", 1);
|
|
const args = { id: d.id, choice };
|
|
if (note !== undefined) args.note = note;
|
|
try {
|
|
await call("decision.resolve", args);
|
|
} catch (e) {
|
|
if (e.code === "outcome-unknown") {
|
|
throw new CliError(`outcome unknown: the broker may have recorded it. Check mosaic inbox or mosaic trail ${d.id} before trying again`, 1);
|
|
}
|
|
if (e.code === "decision-closed") throw new CliError(`decision ${shortId(d.id)} was closed before your answer arrived; see mosaic trail ${d.id}`, 2);
|
|
throw e;
|
|
}
|
|
write(`resolved ${d.id}: ${choice}\n`);
|
|
return d;
|
|
}
|
|
|
|
// The bus call for one command. --tester (row 110) binds as the business's
|
|
// tester through tester-cli.mjs instead of as the human, and may run inside
|
|
// an agent; the broker refuses it under a managed run or with bus.tester off.
|
|
function callFor(parsed, io, deps) {
|
|
const tester = parsed.flags.has("--tester");
|
|
if (!tester) refuseInsideAgent(io.env);
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const business = businessFor(parsed, system.dataRoot);
|
|
if (tester) return deps.testerTransport ? deps.testerTransport(business) : testerTransport({ socket: socketPath(system.dataRoot), business, env: io.env, spawn: deps.testerSpawn });
|
|
return deps.transport ? deps.transport(business) : humanTransport({ socket: socketPath(system.dataRoot), business, env: io.env });
|
|
}
|
|
|
|
async function human(verb, rest, io, deps) {
|
|
const spec = verb === "decide" ? { flags: ["--yes"], values: ["--business", "--note"] } : { flags: ["--json", "--tester"], values: ["--business"] };
|
|
const parsed = parse(rest, spec);
|
|
const call = callFor(parsed, io, deps);
|
|
if (verb === "decide") return decide(parsed, call, io);
|
|
const json = parsed.flags.has("--json");
|
|
const print = (value, text) => io.stdout.write(json ? `${JSON.stringify(value, null, 2)}\n` : text(value));
|
|
if (verb === "trail") {
|
|
if (parsed.positional.length !== 1) throw usage("trail takes one task or decision");
|
|
const subject = parsed.positional[0];
|
|
return print(await call("trail", { subject }), (rows) => formatTrail(subject, rows));
|
|
}
|
|
if (parsed.positional.length !== 0) throw usage(`${verb} takes no arguments`);
|
|
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);
|
|
}
|
|
|
|
const TALK_WAIT_S = 900;
|
|
const TESTER_PREFIX = "[mos test]";
|
|
const TALK_POLL_MS = 2000;
|
|
|
|
function formatMessage(m) {
|
|
const meta = [m.class, m.in_reply_to ? `re ${shortId(m.in_reply_to)}` : null].filter(Boolean).join(", ");
|
|
return `${m.from_role} (${meta}) ${shortId(m.id)}:\n${m.body.trimEnd()}\n`;
|
|
}
|
|
|
|
// `mosaic talk <role> <text>`: 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.
|
|
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");
|
|
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;
|
|
// 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;
|
|
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;
|
|
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 });
|
|
});
|
|
}
|
|
} finally {
|
|
if (follow) process.removeListener("SIGINT", stop);
|
|
deps.signal?.removeEventListener("abort", stop);
|
|
}
|
|
}
|
|
|
|
// Seat errors carry the seat's exit codes; here they become CLI errors.
|
|
async function seat(fn) {
|
|
try {
|
|
return await fn();
|
|
} catch (e) {
|
|
if (e instanceof SeatError) throw new CliError(e.message, e.exitCode === 2 ? 2 : 1);
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
async function stop(rest, io, deps) {
|
|
const parsed = parse(rest, {});
|
|
if (parsed.positional.length !== 1) throw usage("stop takes one run id; see mosaic launches list");
|
|
refuseInsideAgent(io.env);
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const r = await seat(() => stopSession(system.dataRoot, parsed.positional[0], deps.stopTimeoutMs ? { timeoutMs: deps.stopTimeoutMs } : {}));
|
|
io.stdout.write(r.stopped ? `stopped ${r.business}/${r.instance} run ${r.run}\n` : `${r.reason}\n`);
|
|
}
|
|
|
|
async function launches(rest, io, deps) {
|
|
const [sub, ...args] = rest;
|
|
if (sub === "off" || sub === "on") {
|
|
const parsed = parse(args, { values: ["--business"] });
|
|
if (parsed.positional.length !== 0) throw usage(`launches ${sub} takes no arguments`);
|
|
refuseInsideAgent(io.env);
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const business = businessFor(parsed, system.dataRoot);
|
|
const call = deps.transport ? deps.transport(business) : humanTransport({ socket: socketPath(system.dataRoot), business, env: io.env });
|
|
await call(sub === "off" ? "launch.revoke" : "launch.restore");
|
|
io.stdout.write(sub === "off" ? `launches off for ${business}: the PM's launch requests are refused until mosaic launches on; running sessions keep running\n` : `launches on for ${business}\n`);
|
|
return;
|
|
}
|
|
if (sub === "list") {
|
|
const parsed = parse(args, { flags: ["--json"] });
|
|
if (parsed.positional.length !== 0) throw usage("launches list takes no arguments");
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const list = await seat(() => readSessions(system.dataRoot));
|
|
if (parsed.flags.has("--json")) return io.stdout.write(`${JSON.stringify(list, null, 2)}\n`);
|
|
if (list.length === 0) return io.stdout.write("no sessions\n");
|
|
const lines = list.map((s) => `${s.run} ${s.business}/${s.instance} ${s.harness} ${s.model} by ${s.launchedBy} since ${s.startedAt} ${s.live ? "running" : "not running (stale)"}`);
|
|
return io.stdout.write(`${lines.join("\n")}\n`);
|
|
}
|
|
throw usage();
|
|
}
|
|
|
|
async function bus(rest, io, deps) {
|
|
const [sub, ...args] = rest;
|
|
if (sub === "start") {
|
|
const parsed = parse(args, { flags: ["--pm"] });
|
|
if (parsed.positional.length !== 1) throw usage("bus start takes one business");
|
|
const businessId = parsed.positional[0];
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const boot = bootConfig({ system, businessId, env: io.env, warn: (w) => io.stderr.write(`mosaic-bus: warning: ${w}\n`) });
|
|
const binding = readNotifyConfig(system.dataRoot, businessId);
|
|
const host = await startHost({ boot, business: businessId, notifier: binding ? { binding } : null });
|
|
try {
|
|
const launcher = createLauncher({ host, system, env: io.env, tracker: Boolean(boot.trackers?.[businessId]), ...(deps.launcher ?? {}) });
|
|
await launcher.listen();
|
|
io.stdout.write(`bus host up: business ${businessId}, socket ${host.path}, notifier ${binding ?? "off"}\n`);
|
|
if (parsed.flags.has("--pm")) {
|
|
const pm = await launcher.launchPm();
|
|
io.stdout.write(`launched ${pm.instance}: run ${pm.run}, ${pm.harness} ${pm.model}\n`);
|
|
}
|
|
} catch (e) {
|
|
await host.close(1);
|
|
throw e;
|
|
}
|
|
const stop = () => host.close(0);
|
|
process.on("SIGTERM", stop);
|
|
process.on("SIGINT", stop);
|
|
const code = await host.done;
|
|
process.off("SIGTERM", stop);
|
|
process.off("SIGINT", stop);
|
|
io.stdout.write(`bus host stopped (${code})\n`);
|
|
if (code !== 0) throw new CliError("bus host stopped after a child exited", code);
|
|
return;
|
|
}
|
|
if (sub === "stop") {
|
|
if (args.length !== 0) throw usage("bus stop takes no arguments");
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const r = await stopHost(system.dataRoot);
|
|
io.stdout.write(r.stopped ? `stopped bus host for ${r.business} (pid ${r.pid})\n` : `${r.reason}\n`);
|
|
return;
|
|
}
|
|
if (sub === "status") {
|
|
const parsed = parse(args, { flags: ["--json"] });
|
|
if (parsed.positional.length !== 0) throw usage("bus status takes no arguments");
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const s = hostStatus(system.dataRoot);
|
|
if (parsed.flags.has("--json")) return io.stdout.write(`${JSON.stringify(s, null, 2)}\n`);
|
|
const lines = [
|
|
s.host ? `host: ${s.host.business}, pid ${s.host.pid}, ${s.host.live ? "running" : "not running (stale state file)"}, since ${s.host.startedAt}, notifier ${s.host.notifier ?? "off"}` : "host: none",
|
|
`socket: ${s.socket ? "present" : "absent"}`,
|
|
s.writerLock
|
|
? `writer.lock: pid ${s.writerLock.pid ?? "?"}${s.writerLock.live ? "" : " (not running: a broker died hard; remove the lock by hand once you've checked, see packages/bus/README.md)"}`
|
|
: "writer.lock: absent",
|
|
];
|
|
return io.stdout.write(`${lines.join("\n")}\n`);
|
|
}
|
|
throw usage();
|
|
}
|
|
|
|
// `mosaic` with no argument. In an agent run it refuses (exit 3); without a
|
|
// terminal on both stdin and stdout it prints the usage (exit 4); on a
|
|
// terminal it opens the app (src/app.mjs) and prints one line once it quits.
|
|
async function app(io, deps) {
|
|
refuseInsideAgent(io.env);
|
|
if (!io.stdin.isTTY || !io.stdout.isTTY) throw new CliError(`the app needs a terminal on stdin and stdout; from a script, use a verb:\n${SEAT_USAGE}\n${USAGE}`, 4);
|
|
const system = deps.system ?? loadSystem({ env: io.env });
|
|
const state = readHostState(system.dataRoot);
|
|
const business = state?.live ? state.business : null;
|
|
const transport = deps.transport ?? ((b) => humanTransport({ socket: socketPath(system.dataRoot), business: b, env: io.env }));
|
|
const run = deps.runApp ?? runApp;
|
|
const result = await run({ io, business, transport, color: !io.env.NO_COLOR });
|
|
io.stdout.write(`${quitLine(result)}\n`);
|
|
}
|
|
|
|
export async function main(argv, io = { env: process.env, stdin: process.stdin, stdout: process.stdout, stderr: process.stderr }, deps = {}) {
|
|
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 === "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);
|
|
if (verb === "-h" || verb === "--help") return io.stdout.write(`${USAGE}\n`);
|
|
throw usage();
|
|
}
|
|
|
|
if (process.argv[1] === fileURLToPath(import.meta.url)) {
|
|
try {
|
|
await main(process.argv.slice(2));
|
|
} catch (error) {
|
|
if (!(error instanceof CliError)) throw error;
|
|
process.stderr.write(`mosaic: ${error.message}\n`);
|
|
process.exitCode = error.exitCode;
|
|
}
|
|
}
|