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 { projectBusCommand, projectCommand, PROJECT_BUS_USAGE, PROJECT_USAGE } from "./project.mjs";
import { resolveThread, threadCommand, threadMembers, threadsCommand, THREAD_TALK_USAGE, THREAD_USAGE, THREADS_USAGE } from "./thread.mjs";
import { sessionCommand, SESSION_USAGE } from "./session.mjs";
import { limitsCommand, LIMITS_USAGE } from "./limits.mjs";
import { pluginsCommand, PLUGINS_USAGE } from "./plugins.mjs";
import { proxyCommand } from "./model-proxy.mjs";
import { credentialsCommand, CREDENTIALS_USAGE } from "./credentials.mjs";
import { hostsCommand, HOSTS_USAGE } from "./hosts.mjs";
import { grantsCommand, GRANTS_USAGE } from "./grants.mjs";
import { instancesCommand, INSTANCES_USAGE } from "./instances.mjs";
import { accessCommand, accessContextFor } from "./access.mjs";
import { createChannelCaller } from "../../plugins/src/channel.mjs";
import { PluginError } from "../../plugins/src/kinds.mjs";
import { CliError } from "./errors.mjs";
import { DEV_OWNER, devOwnerInstance } from "./dev-local-owner.mjs";
import { bootConfig, loadSystem, socketPath } from "./config.mjs";
import { bootLine, readBootStatus, recordBootOk, recordBootRefused } from "./boot-status.mjs";
import { businessFilePath, configDir } from "../../business/src/index.mjs";
import { busExit, 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";

import { usageCommand, USAGE_USAGE } from "./usage.mjs";
import { orgCommand, ORG_USAGE } from "./org.mjs";

export const USAGE = `usage:
  ${ORG_USAGE}
  ${USAGE_USAGE}
  ${LIMITS_USAGE}
  ${PLUGINS_USAGE}
  ${CREDENTIALS_USAGE}
  mosaic proxy fetch | init [--port N] [--business ID] [--rotate-grace-minutes N] [--first-output-seconds N]
  ${HOSTS_USAGE}
  ${GRANTS_USAGE}
  ${INSTANCES_USAGE}
  ${PROJECT_USAGE}
  ${PROJECT_BUS_USAGE}
  ${THREADS_USAGE}
  ${THREAD_USAGE}
  ${SESSION_USAGE}
  mosaic inbox [--business <id>] [--json] [--tester]
  mosaic inbox --read <id>... | --read-all [--business <id>]
  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]
  ${THREAD_TALK_USAGE}
  mosaic name <instance> <name> [--business <id>] [--tester]
  mosaic names [--business <id>] [--json] [--tester]
  mosaic halt [--business <id>]
  mosaic resume [--business <id>]
  mosaic processing status [--business <id>] [--json]
  mosaic stop <run>
  mosaic launches off|on [--business <id>]
  mosaic launches list [--json]
  mosaic bus start <business> [--pm | --no-launch] [--dev-local-owner]
  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);
  parsed.business = business;
  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 });
}


// Only the converted project/thread surface uses a channel. Management stays on callFor.
function channelCallFor(parsed, io, deps) {
  const call = callFor(parsed, io, deps);
  const channel = createChannelCaller({ name: "cli", call,
    log: entry => io.stderr.write(`mosaic: channel module refused (${entry.code})\n`),
  });
  return Object.assign(async (verb, args = {}) => {
    try { return await channel(verb, args); }
    catch (error) {
      if (!(error instanceof PluginError)) throw error;
      let message = error.message;
      if (error.code === "channel-disabled" && parsed.values["--business"]) {
        message += ` Pass --business ${JSON.stringify(parsed.values["--business"])} to both commands.`;
      }
      const refused = new CliError(message, ["channel-disabled", "channel-config"].includes(error.code) ? 3 : 2);
      refused.code = error.code;
      throw refused;
    }
  }, { business: parsed.business });
}

// The human's unread messages, newest first, from every page of `messages`.
async function unreadMessages(call) {
  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 messages;
}

const MESSAGE_ID_LENGTH = 36;

// Row 150 (#1644): `mosaic inbox --read <id>...` and `--read-all` mark unread messages read
// through message.read. Every id is tried; a refusal of one is printed with its code and the
// command exits 2 at the end. A message never received refuses not-recipient, so it is received
// first, which stops a later `mosaic follow` printing it. See agents/darkwing/work/queue-150/BUILD.md.
async function markRead(parsed, call, io) {
  const unread = await unreadMessages(call);
  const all = parsed.flags.has("--read-all");
  const targets = all ? unread.slice().reverse().map(m => ({ arg: m.id, id: m.id })) : parsed.positional.map(arg => {
    // At least the 8 characters inbox prints, as decide asks of a decision reference.
    if (arg.length < 8) return { arg, refused: "too-short, use at least 8 characters of the id" };
    const matches = unread.filter(m => String(m.id).startsWith(arg));
    if (matches.length === 1) return { arg, id: matches[0].id };
    if (matches.length > 1) return { arg, refused: `ambiguous, matches ${matches.map(m => m.id).join(", ")}` };
    return arg.length === MESSAGE_ID_LENGTH ? { arg, id: arg } : { arg, refused: "no-unread-match" };
  });
  const perMessage = e => typeof e?.code === "string" && /^[a-z-]{1,64}$/.test(e.code) && busExit(e.code) === 2;
  const read = async id => {
    try {
      await call("message.read", { id });
    } catch (e) {
      if (e.code !== "not-recipient") throw e;
      // Undelivered: receive until a batch holds it or none is left, then read once more.
      let batch;
      do batch = await call("message.receive"); while (batch.length && !batch.some(m => m.id === id));
      await call("message.read", { id });
    }
  };
  let marked = 0;
  for (const t of targets) {
    if (t.refused === undefined) {
      try {
        await read(t.id);
        marked++;
        if (!all) io.stdout.write(`marked ${shortId(t.id)}\n`);
        continue;
      } catch (e) {
        if (!perMessage(e)) throw e;
        t.refused = e.code;
      }
    }
    io.stdout.write(`not marked ${t.arg}: ${t.refused}\n`);
  }
  io.stdout.write(`marked ${marked} of ${targets.length}\n`);
  if (marked < targets.length) throw new CliError(`${targets.length - marked} of ${targets.length} not marked`, 2);
}

async function human(verb, rest, io, deps) {
  const spec = verb === "decide" ? { flags: ["--yes"], values: ["--business", "--note"] }
    : { flags: ["--json", "--tester", ...(verb === "inbox" ? ["--read", "--read-all"] : [])], values: ["--business"] };
  const parsed = parse(rest, spec);
  const marking = parsed.flags.has("--read") || parsed.flags.has("--read-all");
  if (parsed.flags.has("--read") && parsed.flags.has("--read-all")) throw usage("--read and --read-all cannot be combined");
  if (marking && parsed.flags.has("--json")) throw usage("--read and --read-all print no JSON");
  // The tester's message.read reads the tester's own messages, not the human's that inbox lists.
  if (marking && parsed.flags.has("--tester")) throw usage("--read and --read-all take no --tester");
  if (parsed.flags.has("--read") && parsed.positional.length === 0) throw usage("--read takes one or more message ids");
  if (parsed.flags.has("--read-all") && parsed.positional.length !== 0) throw usage("--read-all takes no ids");
  const call = callFor(parsed, io, deps);
  if (verb === "decide") return decide(parsed, call, io);
  if (marking) return markRead(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 = await unreadMessages(call);
    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);
  const agents = await call("agents");
  // Row 277 (#1782, C1/C2): the human's text output adds the held thread claims. A tester can't read
  // thread.claims, so --tester makes the one call, and --json stays the agents array.
  if (json || parsed.flags.has("--tester")) return print(agents, formatAgents);
  // Row 277 N1 (sage-2505): `agents` stays as it was, so a refused thread.claims still prints the
  // role rows, then the refusal with its code. An error without a bus code still throws.
  let threads;
  try {
    threads = await call("thread.claims");
  } catch (e) {
    if (typeof e?.code !== "string" || !/^[a-z-]{1,64}$/.test(e.code)) throw e;
    return print(agents, (list) => `${formatAgents(list)}thread claims: refused ${e.code}\n`);
  }
  return print(agents, (list) => formatAgents(list, threads));
}

// Row 141 (#1635, REQ-MSG-2): `mosaic name` sets the name `talk` and message.send accept for an
// instance; `mosaic names` lists the current ones. Both are the human's, and since row 175
// (#1673, Q173) the tester's too.
async function names(verb, rest, io, deps) {
  const parsed = parse(rest, { flags: verb === "names" ? ["--json", "--tester"] : ["--tester"], values: ["--business"] });
  if (parsed.positional.length !== (verb === "name" ? 2 : 0)) throw usage(verb === "name" ? "name takes an instance and a name" : "names takes no arguments");
  const call = callFor(parsed, io, deps);
  if (verb === "name") {
    const [instance, name] = parsed.positional;
    await call("name.set", { instance, name });
    return io.stdout.write(`named ${instance} ${name}\n`);
  }
  const rows = await call("names");
  io.stdout.write(parsed.flags.has("--json") ? `${JSON.stringify(rows, null, 2)}\n` : rows.map((r) => `${r.instance} ${r.name}\n`).join(""));
}

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`;
}

// 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);
  // Row 152 (#1646): launch.by can launch only launch.instances. An older bus sends no
  // launchable, so only an explicit false prints this line.
  if (w.launchable === false) return `${w.role} isn't running and isn't in the launch block, so ${w.launcher} can't launch it\n`;
  return `${w.role} isn't running, ${w.launcher} has been told\n` + (w.launcherRunning ? "" : restart(w.launcher, " either"));
}

// `mosaic talk <role> <text>`: a REQUEST to the role, then the human's inbox
// until the reply to it arrives or --wait runs out; other messages stay unread.
// --follow prints and reads every message and keeps receiving after that
// reply; follow starts the same receive loop without sending.
// Row 128 (#1622): --thread names a thread (src/thread.mjs) and sends with thread.send. The
// reply is found as before; --follow prints and reads only that thread's messages, and every
// other message stays unread for mosaic inbox.
// Row 184 (#1682, Q173): with --tester the same loop runs as the tester on its own thread. The
// tester's replies arrive only through message.receive, as for the human; inbox --tester lists
// decisions only.
async function talk(rest, io, deps, receiveOnly = false) {
  const parsed = parse(rest, { flags: ["--follow", "--tester"], values: ["--business", "--thread", "--session", ...(receiveOnly ? [] : ["--wait"])] });
  const threadRef = parsed.values["--thread"];
  const session = parsed.values["--session"];
  if (session !== undefined && (threadRef === undefined || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(session))) throw usage('--session needs --thread and a full session id');
  if (parsed.positional.length !== (receiveOnly ? 0 : threadRef === undefined ? 2 : 1)) {
    throw usage(receiveOnly ? "follow takes no arguments" : threadRef === undefined ? "talk takes a role and one quoted text" : "talk --thread takes 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");
  let [to, body] = parsed.positional;
  if (threadRef !== undefined) [to, body] = [threadRef, to];
  const call = threadRef === undefined ? callFor(parsed, io, deps) : channelCallFor(parsed, io, deps);
  let sent;
  let thread = null;
  if (threadRef !== undefined) {
    thread = await resolveThread(call, threadRef);
    to = thread.label;
  }
  if (!receiveOnly) {
    try {
      sent = await call(thread ? "thread.send" : "message.send", thread ? { thread: thread.id, ...(session ? { session } : {}), body, class: "REQUEST" } : { to, body, class: "REQUEST" });
    } catch (e) {
      // The bus never picks a thread for the human, and neither does the CLI (row 128).
      if (e.code === "thread-required-use-thread-send") {
        throw new CliError(`${to} runs in threads, so a message to it names one: mosaic talk --thread <thread> <text>; mosaic threads lists yours`, 2);
      }
      // Row 170 (#1667): the bus stores no tester message to a role nobody holds. Exit 3 like the
      // other tester refusals; transport.mjs's EXIT map stays as row 164 pinned it.
      if (e.code === "tester-target-not-running" && parsed.flags.has("--tester")) {
        throw new CliError(`${to} isn't running, so the tester's message was refused and nothing was stored`, 3);
      }
      throw e;
    }
    if (sent.cancelled !== undefined) {
      if (!['session-clearing', 'session-retired'].includes(sent.cancelled)) throw new CliError('outcome unknown: invalid cancellation receipt; nothing was retried', 1);
      io.stdout.write(`stored message ${sent.id}\n`);
      throw new CliError(`message cancelled (${sent.cancelled}); it was not delivered and will not be replayed`, 3);
    }
    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;
  }
  // message.receive says nothing of a message's thread, so --follow asks the thread for its ids.
  // Reading the thread first makes `follow --thread` refuse a thread that isn't the user's before
  // anything is received: message.receive writes a delivery row for every message it returns.
  const members = thread && follow ? threadMembers(call, thread.id, session) : null;
  await members?.refresh();
  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;
      const received = await call("message.receive");
      if (members && received.length > 0) await members.refresh();
      for (const m of received) {
        // Row 142 (#1636): without --follow only the reply to the sent message is printed and marked read; the rest stays unread for mosaic inbox.
        if (!follow && m.in_reply_to !== sent.id) continue;
        // Only the thread's messages after the one sent; an earlier one stays unread.
        if (members && !(members.seq(m.id) > (members.seq(sent?.id) ?? 0))) continue;
        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 mosaic inbox 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 processingCommand(verb, rest, io, deps) {
  const parsed = parse(rest, { flags: ["--json"], values: ["--business"] });
  if (verb === 'processing' ? parsed.positional.length !== 1 || parsed.positional[0] !== 'status' : parsed.positional.length !== 0)
    throw usage('use mosaic halt, mosaic resume or mosaic processing status');
  refuseInsideAgent(io.env);
  const system = deps.system ?? loadSystem({ env: io.env });
  const call = callFor(parsed, io, { ...deps, system });
  try {
    let result;
    if (verb === 'resume') {
      const current = await call('processing.status');
      if (current?.invalid) throw new CliError(`processing state is invalid; use mosaic halt, then mosaic resume (${current.code})`, 3);
      result = await call('processing.resume', { generation: current?.generation });
    } else result = await call(verb === 'halt' ? 'processing.halt' : 'processing.status', {});
    if (!result || typeof result !== 'object' || (typeof result.halted !== 'boolean' && !result.invalid))
      throw new CliError('processing outcome unknown; use mosaic processing status before trying again', 1);
    if (verb === 'halt' && result.quiescent !== true)
      throw new CliError(`halt gate is on for instance ${system.dataRoot}; ${result.remaining ?? 'unknown'} stops unconfirmed. Use mosaic processing status`, 1);
    if (parsed.flags.has('--json')) { io.stdout.write(JSON.stringify(result, null, 2) + '\n'); return; }
    const status = verb === 'resume' ? 'resumed' : result.halted ? 'halted' : result.invalid ? 'uninitialized' : 'running';
    io.stdout.write(`processing ${status} for instance ${system.dataRoot}\n`);
    if (result.invalid) io.stdout.write('state is invalid; use mosaic halt, then mosaic resume\n');
  } catch (e) {
    if (e instanceof CliError) throw e;
    const code = typeof e.code === 'string' && /^[a-z-]{1,64}$/.test(e.code) ? e.code : 'outcome-unknown';
    if (code === 'outcome-unknown') throw new CliError('processing outcome unknown; use mosaic processing status before trying again', 1);
    throw new CliError(`processing refused: ${code}`, 3);
  }
}

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", "--no-launch", "--dev-local-owner"] });
    if (parsed.positional.length !== 1) throw usage("bus start takes one business");
    // Row 163: --no-launch is the D2 rehearsal's host. No unit passes it.
    const launches = !parsed.flags.has("--no-launch");
    if (parsed.flags.has("--no-launch") && parsed.flags.has("--pm")) throw usage("bus start takes --pm or --no-launch, not both");
    const instance = parsed.flags.has("--dev-local-owner") ? devOwnerInstance(io.env) : null;
    const businessId = parsed.positional[0];
    const system = instance ? loadSystem({ env: io.env, instance }) : (deps.system ?? loadSystem({ env: io.env }));
    // Fail closed, and say so (vikunja:19/302): a refused load is recorded where the Console and
    // `bus status` can read it, then thrown as before. No retry.
    const file = businessFilePath(businessId, configDir(io.env));
    let boot;
    try {
      boot = bootConfig({ system, businessId, env: io.env, warn: (w) => io.stderr.write(`mosaic-bus: warning: ${w}\n`) });
    } catch (e) {
      try { recordBootRefused(system.dataRoot, { business: businessId, file, reason: e.message }); } catch { /* the original refusal still throws */ }
      throw e;
    }
    if (instance) boot.consoleOwners = [...(boot.consoleOwners ?? []), DEV_OWNER];
    try { recordBootOk(system.dataRoot, { business: businessId, file }); } catch { /* status is advisory */ }
    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]), launches, ...(deps.launcher ?? {}) });
      await launcher.listen();
      if (instance) io.stdout.write(`DEV LOCAL OWNER MODE: console owner local-dev accepted (named instance ${instance})\n`);
      if (instance && parsed.flags.has("--pm")) io.stdout.write('DEV LOCAL OWNER MODE: launched agents run as this OS user and can reach the loopback Console as owner; use this instance for UI testing only\n');
      io.stdout.write(`bus host up: business ${businessId}, socket ${host.path}, notifier ${binding ?? "off"}${launches ? "" : ", launches disabled"}\n`);
      if (parsed.flags.has("--pm")) {
        try {
          const pm = await launcher.launchPm();
          io.stdout.write(`launched ${pm.instance}: run ${pm.run}, ${pm.harness} ${pm.model}\n`);
        } catch (e) {
          if (e.code !== 'processing-halted') throw e;
          io.stdout.write('processing halted: chief launch skipped; control bus remains available\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);
    s.boot = readBootStatus(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",
      bootLine(s.boot) + (s.boot && !s.boot.ok && s.host?.live ? " (a host started earlier still runs on the old file)" : ""),
    ];
    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);
  if (argv[0] !== 'org') {
    const org = argv.indexOf('--org'), business = argv.indexOf('--business');
    if (org !== -1 && business !== -1) {
      if (!argv[org + 1] || !argv[business + 1] || argv[org + 1] !== argv[business + 1]) throw usage('--org and --business conflict');
      argv = argv.filter((_, i) => i !== org && i !== org + 1);
    } else if (org !== -1) argv = argv.map((arg, i) => i === org ? '--business' : arg);
  }
  const [verb, ...rest] = argv;
  if (verb === 'org') return orgCommand(rest, io, deps);
  if (["inbox", "decide", "tasks", "agents", "trail"].includes(verb)) return human(verb, rest, io, deps);
  if (verb === "project") return ["create", "list", "archive", "unarchive"].includes(rest[0]) ? projectBusCommand(rest, io, (p) => channelCallFor(p, io, deps)) : projectCommand(rest, io, deps, (p) => callFor(p, io, deps));
  if (verb === "threads") return threadsCommand(rest, io, (p) => channelCallFor(p, io, deps));
  if (verb === "session") return sessionCommand(rest, io, (p) => channelCallFor(p, io, deps));
  if (verb === "thread") return threadCommand(rest, io, (p) => channelCallFor(p, io, deps), formatMessage);
  if (verb === "access") {
    const system = deps.system ?? loadSystem({ env: io.env });
    return accessCommand(rest, io, {
      system,
      transport: (b) => humanTransport({ socket: socketPath(system.dataRoot), business: b, env: io.env }),
      contextFor: () => deps.accessContext ?? accessContextFor(system, io.env),
      now: deps.now,
    });
  }
  if (verb === "limits") return limitsCommand(rest,io,p=>callFor(p,io,deps));
  if (verb === "usage") return usageCommand(rest, io, p => callFor(p, io, deps));
  if (verb === "plugins") return pluginsCommand(rest, io, (p) => callFor(p, io, deps));
  // Row 199 (#1697): issue and rotate read the token on stdin.
  if (verb === "proxy") return proxyCommand(rest, io, { system: () => deps.system ?? loadSystem({ env: io.env }), callFor: p => callFor(p, io, deps), prompt: deps.promptProxyKey });
  if (verb === "credentials") return credentialsCommand(rest, io, (p) => callFor(p, io, deps));
  // Row 234 (#1740, r6 E4): the host records.
  if (verb === "hosts") return hostsCommand(rest, io, (p) => callFor(p, io, deps));
  // Row 236 (#1742, #436b r5): the Gitea grant path; admin set reads the admin token on stdin.
  if (verb === "grants") return grantsCommand(rest, io, (p) => callFor(p, io, deps));
  // Row 165 (#1662): scripts/mosaic sends only `business instances`, `business instance` and (row 194) `business veto` here.
  if (verb === "business") return instancesCommand(rest, io, (p) => callFor(p, io, deps));
  if (verb === "talk") return talk(rest, io, deps);
  if (verb === "follow") return talk(rest, io, deps, true);
  if (verb === "name" || verb === "names") return names(verb, rest, io, deps);
  if (verb === "stop") return stop(rest, io, deps);
  if (["halt", "resume", "processing"].includes(verb)) return processingCommand(verb, 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();
}

// The broker's human and tester proofs accept only the human-cli.mjs and
// tester-cli.mjs of the tree the bus runs from. Since row 85 that is a
// pinned copy, so the checkout's own scripts/mosaic is refused with these
// codes (row 130). The message stays the bare code: app.mjs reads it.
export const OTHER_TREE = "the bus accepts these commands only from the tree it runs from; on a pinned host run `mosaic` on PATH";
const OTHER_TREE_CODES = new Set(["human-required", "tester-required"]);

// main, with a CliError printed to stderr. Returns the exit code.
export async function run(argv, io = { env: process.env, stdin: process.stdin, stdout: process.stdout, stderr: process.stderr }, deps = {}) {
  try {
    await main(argv, io, deps);
    return 0;
  } catch (error) {
    if (!(error instanceof CliError)) throw error;
    io.stderr.write(`mosaic: ${error.message}\n`);
    if (OTHER_TREE_CODES.has(error.code)) io.stderr.write(`mosaic: ${OTHER_TREE}\n`);
    return error.exitCode;
  }
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const code = await run(process.argv.slice(2));
  if (code !== 0) process.exitCode = code;
}