stack/packages/cli/src/cli.mjs

#!/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 } 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]
  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 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;
}

async function human(verb, rest, io, deps) {
  const spec = verb === "decide" ? { flags: ["--yes"], values: ["--business", "--note"] } : { flags: ["--json"], values: ["--business"] };
  const parsed = parse(rest, spec);
  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 });
  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") return print(await call("inbox"), formatInbox);
  if (verb === "tasks") return print(await call("tasks"), formatTasks);
  return print(await call("agents"), formatAgents);
}

const TALK_WAIT_S = 900;
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. 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");
  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);
  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 });
  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;
  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;
    }
    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 or mosaic inbox shows it`, 1);
    await new Promise((r) => setTimeout(r, pollMs));
  }
}

// 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 === "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;
  }
}