// 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;