stack/packages/bus/src/process.mjs

// Started with fork() by the trusted host; boot refs/capabilities travel over IPC,
// never command arguments, stdout or service-token-bearing environment variables.
import { startBroker } from './runtime.mjs';
import { BusError } from './broker.mjs';
const LAUNCH_OPS = new Set([
  'identity',
  'authorizeLaunch',
  'refuse',
  'endLaunch',
  'credentialStatus',
  'checkPush',
  'recordPush',
  'reviewFetch',
  'checkLand',
  'recordLand',
  'landRefused',
  'landHeld',
  'threadsState',
  'threadIdle',
  'mayLaunch',
  'landHolders',
  'landRelease',
]);
function launchOp(m) {
  if (m.op === 'identity') return runtime.identity(m.cap);
  if (m.op === 'authorizeLaunch') return runtime.authorizeLaunch(m.cap, m.instance);
  if (m.op === 'refuse') return runtime.refuse(m.cap, m.code, m.subject);
  if (m.op === 'endLaunch') return runtime.endLaunch(m.record);
  // Row 93 (#1587): the git socket's ops.
  if (m.op === 'checkPush') return runtime.checkPush(m.cap, m.task_ref);
  if (m.op === 'recordPush') return runtime.recordPush(m.cap, m.push);
  if (m.op === 'reviewFetch') return runtime.reviewFetch(m.cap, m.args);
  // Row 137 (#1631): the git socket's land op, and the launcher's question whether to bind it.
  if (m.op === 'checkLand') return runtime.checkLand(m.cap, m.args);
  if (m.op === 'recordLand') return runtime.recordLand(m.ticket, m.land);
  if (m.op === 'landRefused') return runtime.landRefused(m.ticket, m.args);
  if (m.op === 'landHeld') return runtime.landHeld(m.record);
  // Row 121 (#1615): the launcher's view of the business's threads, and its idle release.
  if (m.op === 'threadsState') return runtime.threadsState(m.business);
  if (m.op === 'threadIdle') return runtime.threadIdle(m.record);
  // Row 125 (#1619, REQ-LIMIT-4): the limit checks' answer before a launch (async).
  if (m.op === 'mayLaunch') return runtime.mayLaunch(m);
  // Row 156 (#1651): the launcher's landing pass: who holds a merge decision, and the stop of a
  // role run started before its decision was resolved.
  if (m.op === 'landHolders') return runtime.landHolders(m.business);
  if (m.op === 'landRelease') return runtime.landRelease(m.record);
  if (m.op === 'credentialStatus') return runtime.credentialStatus(m.business, m.role);
  // Every LAUNCH_OPS name has its line above. A missing one used to answer as credentialStatus,
  // whose [] for an unknown role read as "no holders" for landHolders (row 156 mutant P02).
  throw new BusError('launch-op-refused');
}
// The host's request id comes back on the reply, so a late reply can't
// answer a later request (host.mjs). A message without one gets none.
const tag = (message, reply) => (Number.isSafeInteger(message?.id) ? { ...reply, id: message.id } : reply);
let runtime,
  booted = false,
  closing = false;
async function close(code) {
  if (closing) return;
  closing = true;
  try {
    await runtime?.close();
  } catch {
    code = 2;
  }
  process.exitCode = code;
  if (process.connected) process.disconnect();
}
if (!process.send) {
  process.stderr.write('trusted-host-required\n');
  process.exitCode = 2;
} else {
  const timer = setTimeout(() => close(2), 10000);
  process.on('message', async (message) => {
    try {
      if (message?.op === 'bindLaunch' && runtime) {
        try {
          process.send(tag(message, { ok: true, launch: runtime.bindLaunch(message.record) }));
        } catch (e) {
          process.send(tag(message, { ok: false, error: e instanceof BusError ? e.code : 'bind-refused' }));
        }
        return;
      }
      // S6 launcher ops, one reply each, like bindLaunch.
      if (LAUNCH_OPS.has(message?.op) && runtime) {
        try {
          process.send(tag(message, { ok: true, result: await launchOp(message) }));
        } catch (e) {
          process.send(tag(message, { ok: false, error: e instanceof BusError ? e.code : 'launch-op-refused' }));
        }
        return;
      }
      if (message?.op === 'close') {
        clearTimeout(timer);
        await close(0);
        return;
      }
      if (message?.op !== 'boot' || booted) throw new BusError('invalid-host-request');
      booted = true;
      clearTimeout(timer);
      // `trackers` is S3's plain-data boot config; the adapter is loaded only when a business has one.
      const { trackers, ...config } = message.config ?? {};
      if (trackers) {
        const { tasksAdapter } = await import('../../tasks/src/adapter.mjs');
        config.tasks = tasksAdapter({ trackers });
      }
      runtime = await startBroker(config);
      if (closing) {
        await runtime.close();
        return;
      }
      process.send({ ok: true, path: runtime.path, launches: runtime.launches, readers: runtime.readers });
    } catch (e) {
      process.send?.(tag(message, { ok: false, error: e instanceof BusError ? e.code : 'startup-refused' }), () =>
        close(2),
      );
    }
  });
  process.on('disconnect', () => close(2));
  process.on('SIGTERM', () => close(0));
  process.on('SIGINT', () => close(0));
}