stack/packages/webui/src/bus.mjs

// Slice 1 S5 (#1522): the Console's read of the bus. It runs the S4 human
// transport (packages/bus/src/human-cli.mjs, lead decision 70) as a child
// with spawn, not spawnSync, so one slow read never stalls the HTTP server.
// Only the four verbs of the Q1 module (lead decision 56), row 57's
// `decisions`, row 60's `credentials` and row 81's `replies` reads are
// reachable as reads: no read writes, not even seen or delivered (Q4). The
// one write is row 76's human `message.send` (lead decision 95), built by
// the CLI only with --send. The bus does the proof; a server started inside
// an agent run gets `human-required` from the broker.

import { spawn } from 'node:child_process';
import { randomBytes } from 'node:crypto';
import { fileURLToPath } from 'node:url';
import { HUMAN_CLI } from '../../cli/src/transport.mjs';
import { readHostState } from '../../cli/src/host.mjs';
import { views } from '../../bus/src/views.mjs';
import { CONSOLE_VERBS } from '../../bus/src/broker.mjs';

export class BusReadError extends Error {
  constructor(code, message = code) { super(message); this.code = code; }
}

// The broker's identifier rule (packages/bus/src/broker.mjs `id`), checked
// here so a bad subject is a 400 and never reaches the bus.
export const subjectOk = s => typeof s === 'string' && s.length > 0 && s.length <= 160 && /^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$/.test(s);

// Same protocol as humanTransport in packages/cli/src/transport.mjs: the
// request on stdin, JSON on stdout, or a bus error code as the last
// `^[a-z-]{1,64}$` line of stderr.
export function humanCall({ socket, business, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024, signal, deadline }) {
  return (verb, args = {}) => transport({ cli, socket, signal, deadline, env: cli === HUMAN_CLI ? { ...env, MOSAIC_BUS_CLI_NONCE: randomBytes(32).toString('hex') } : env, input: { business, verb, args }, timeoutMs, maxBytes });
}
function transport({ cli, socket, env, input, timeoutMs, maxBytes, signal, deadline }) {
  return new Promise((resolve, reject) => {
    const remaining = deadline === undefined ? timeoutMs : Math.min(timeoutMs, deadline - Date.now());
    if (signal?.aborted || (deadline !== undefined && remaining <= 0)) return reject(new BusReadError('outcome-unknown', 'the bus transport deadline expired'));
    let child;
    try {
      child = spawn(process.execPath, [cli, socket], { env, stdio: ['pipe', 'pipe', 'pipe'] });
    } catch { return reject(new BusReadError('outcome-unknown', 'the bus transport did not start')); }
    const out = [], err = [];
    let size = 0, settled = false;
    const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); signal?.removeEventListener('abort', abort); fn(value); } };
    const abort = () => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport deadline expired')); };
    const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, remaining);
    signal?.addEventListener('abort', abort, { once: true });
    if (signal?.aborted) abort();
    child.stdout.on('data', chunk => {
      size += chunk.length;
      if (size > maxBytes) { child.kill('SIGKILL'); return done(reject, new BusReadError('response-too-large')); }
      out.push(chunk);
    });
    child.stderr.on('data', chunk => { if (err.length < 64) err.push(chunk); });
    child.on('error', () => done(reject, new BusReadError('outcome-unknown', 'the bus transport did not start')));
    child.on('close', status => {
      if (status === null) return done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish'));
      if (status !== 0) {
        const code = Buffer.concat(err).toString('utf8').trim().split('\n').reverse().find(l => /^[a-z-]{1,64}$/.test(l)) ?? 'invalid-response';
        return done(reject, new BusReadError(code));
      }
      try { done(resolve, JSON.parse(Buffer.concat(out).toString('utf8'))); }
      catch { done(reject, new BusReadError('invalid-response')); }
    });
    child.stdin.on('error', () => {}); // A child that exits early reports through 'close'.
    child.stdin.end(`${JSON.stringify(input)}\n`);
  });
}

// Row 252 (#1753, design section 2): the Console owner's answer to a grant request. Each call spawns
// a fresh console-cli with a fresh nonce, and the business and the signed-in (provider, iss, sub) go
// in its spawn environment, where the bus reads them back. Inherited console keys are dropped first,
// so only this call's values reach the wrapper. Only the two console verbs pass (N4). A resolve pins
// the business its confirmation was issued for.
export const CONSOLE_CLI = fileURLToPath(new URL('../../bus/src/console-cli.mjs', import.meta.url));
export function consoleCall({ dataRoot, socket, business = null, env = process.env, cli = CONSOLE_CLI, timeoutMs = 30000 }) {
  return async (verb, args, { subject, business: pinned = null }) => {
    if (!CONSOLE_VERBS.includes(verb)) throw new Error(`the Console does not call ${verb}`);
    const id = pinned ?? (business || hostBusiness(dataRoot));
    const own = Object.fromEntries(Object.entries(env).filter(([k]) => !k.startsWith('MOSAIC_BUS_CONSOLE_')));
    return transport({
      cli, socket, timeoutMs, maxBytes: (verb === 'request.artifact.console' ? 2 : 1) * 1024 * 1024, input: { verb, args },
      env: { ...own, MOSAIC_BUS_CONSOLE_NONCE: randomBytes(32).toString('hex'), MOSAIC_BUS_CONSOLE_BUSINESS: id,
        MOSAIC_BUS_CONSOLE_PROVIDER: subject.provider, MOSAIC_BUS_CONSOLE_ISS: subject.iss, MOSAIC_BUS_CONSOLE_SUB: subject.sub },
    });
  };
}

// The business is the --business flag, or else the live bus host's, read
// on every request so a host started after the Console is picked up.
export function busCaller({ dataRoot, socket, business = null, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, signal, deadline }) {
  return async (verb, args) => {
    if (signal?.aborted || (deadline !== undefined && Date.now() >= deadline)) throw new BusReadError('outcome-unknown', 'the bus transport deadline expired');
    return humanCall({ socket, business: business || hostBusiness(dataRoot), env, cli, timeoutMs, signal, deadline })(verb, args);
  };
}
function hostBusiness(dataRoot) {
  let state;
  try { state = readHostState(dataRoot); } catch { throw new BusReadError('no-bus-host', 'the bus host state file is unreadable'); }
  if (!state?.live) throw new BusReadError('no-bus-host', 'no bus host is running and no --business was given');
  return state.business;
}
export const busReader = options => views({ call: busCaller(options) });

// FED R6 (#1800): a granted user's bus transport, R3's user-cli.mjs (#414, design 3.6). Same
// protocol as humanCall: the request on stdin, JSON on stdout, or a bus error code as the last
// `^[a-z-]{1,64}$` line of stderr. Row 301 (#1808) moved it here from user-bus.mjs.
export const USER_CLI = fileURLToPath(new URL('../../bus/src/user-cli.mjs', import.meta.url));

// The Console sets the user nonce itself, so user-cli.mjs doesn't re-exec and the broker sees the
// Console (`cli.mjs serve`) as its parent (principal.mjs). The human nonce is dropped, so this child
// can never pass as human-cli, and it only ever runs USER_CLI.
export function userCall({ socket, business, user, env = process.env, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024, signal, deadline }) {
  const { MOSAIC_BUS_CLI_NONCE, ...rest } = env;
  return (verb, args = {}) => transport({ cli: USER_CLI, socket, env: { ...rest, MOSAIC_BUS_USER_NONCE: randomBytes(32).toString('hex') },
    input: { business, user, verb, args }, timeoutMs, maxBytes, signal, deadline });
}

// The business is the --business flag, or else the live bus host's, read on every call, as busCaller does.
export function busBusiness({ dataRoot, business = null }) {
  return () => {
    if (business) return business;
    let state;
    try { state = readHostState(dataRoot); } catch { throw new BusReadError('no-bus-host', 'the bus host state file is unreadable'); }
    if (!state?.live) throw new BusReadError('no-bus-host', 'no bus host is running and no --business was given');
    return state.business;
  };
}

// One granted user's caller, for that user's session only.
export function userCaller({ dataRoot, socket, business = null, user, env = process.env, timeoutMs = 30000, signal, deadline }) {
  const which = busBusiness({ dataRoot, business });
  return async (verb, args) => {
    if (signal?.aborted || (deadline !== undefined && Date.now() >= deadline)) throw new BusReadError('outcome-unknown', 'the bus transport deadline expired');
    return userCall({ socket, business: which(), user, env, timeoutMs, signal, deadline })(verb, args);
  };
}

// A send's error, sorted. The broker writes in one transaction, so any code it
// returns means nothing was written. A transport that lost the answer, or an
// answer that can't be read, leaves the outcome open: the message may exist.
// Row 239 (duckworth-0165 C1): so does a write that failed after its revoke.
const OPEN = ['outcome-unknown', 'invalid-response', 'response-too-large', 'outcome-unknown-revoked'];
export function sendOutcome(err) {
  const code = err instanceof BusReadError ? err.code : 'outcome-unknown';
  return OPEN.includes(code) ? { code: 'outcome-unknown', sent: 'unknown' } : { code, sent: false };
}

// HTTP status by bus error code. 403: the bus refused this caller; 503: no
// bus to ask, or no answer; 502: an answer that can't be used.
const STATUS = {
  'human-required': 403, 'peercred-required': 403, unauthenticated: 403, 'unknown-business': 403, 'read-only': 403,
  'outcome-unknown': 503, 'no-bus-host': 503, 'not-configured': 503,
  'revision-conflict':409,'organization-busy':409,'organization-uncertain':503,
  // Row 239 (#1745): a fire's or a delete's write refused after its revoke ran.
  'revision-conflict-revoked': 409, 'outcome-unknown-revoked': 503,
  'invalid-request': 400, 'unknown-role': 400, 'message-not-found': 400, 'request-too-large': 400,
  // #1873 A3: a root or project copy that can't launch.
  'credential-unbound': 409, 'copy-refused': 409, 'project-refused': 409, 'project-agent-elsewhere': 409,
  // Row 148: the spec and docs editor. A stale base, a busy or dirty repository and an unchanged
  // file are the user's to resolve by reloading; a path or content the project refuses is a 400.
  'stale-base': 409, 'project-busy': 409, 'project-dirty': 409,
  'project-name-is-repo': 409,
  'project-record-unchanged': 400, 'project-record-path': 400, 'project-record-invalid': 400,
  'project-read-path': 400, 'project-not-declared': 400, 'project-read-missing': 404, 'project-not-configured': 503,
  // Row 155 (#1649): the assistant's settings, seat names and stop. A name in use or no run to stop
  // is the user's to resolve by reading again.
  'organization-not-assistant': 400, 'organization-definition-refused': 400, 'unknown-model': 400,
  'name-taken': 409, 'name-is-instance': 400, 'name-is-role': 400, 'not-assistant': 400, 'not-held': 409,
  // Row 165 (#1662): role instances. An id already taken is the user's to resolve by reading again.
  'organization-instance-exists': 409, 'organization-unknown-instance': 400,
  'organization-credentials-out-of-scope': 400, 'organization-launch-not-configured': 400,
  // Row 194 (brief item 7, sage-1610 a): only the CLI names a Vikunja bot on add.
  'organization-tracker-required': 400, 'organization-tracker-bot-taken': 409,
  'organization-no-change': 400, 'organization-project-mismatch': 400, 'organization-unresolvable-instance': 400,
  // Row 254 (#1759): library sync. A sync already running, a held divergence and a stale accept are
  // the user's to resolve by reading library.status again; a missing credential or a failed fetch
  // leaves the remote unreachable, like no bus to ask.
  'library-sync-busy': 409, 'library-diverged': 409, 'library-accept-stale': 409, 'library-accept-invalid': 400,
  'library-credential-missing': 503, 'library-fetch-failed': 503,
  // Row 225 (C2-10b): a Revert of a commit that isn't an agent's is hers to read, like stale-base.
  'library-revert-not-agent': 409,
};
export const busStatus = code => STATUS[code] ?? 502;