stack/packages/bus/src/runtime.mjs

import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from './store.mjs';
import { Broker, BusError } from './broker.mjs';
import { Credentials } from './credentials.mjs';
import { serve } from './server.mjs';
import { verifyHuman } from './human.mjs';
import { verifyTester } from './tester.mjs';
// Trusted host API. S1 supplies resolved definitions, S6 supplies launch records and ends them.
// Neither a business-file writer nor a socket-accessible configuration endpoint.
// `tasks` is S3's adapter factory: ({broker, credentials, businesses}) => {handle, timeout, close}.
export async function startBroker({
  dataRoot,
  businesses,
  launches = [],
  readers = [],
  repoRoots = [],
  tasks = null,
}) {
  let store, credentials, server, adapter;
  try {
    const references = {};
    for (const [business, b] of Object.entries(businesses)) {
      for (const [role, r] of Object.entries(b.roles))
        if (r.credentials) references[business + '/' + role] = r.credentials;
      if (b.tracker?.sync?.credentials) references[business + '/@sync'] = b.tracker.sync.credentials;
    }
    const protectedRoots = [
      fileURLToPath(new URL('../../../', import.meta.url)),
      ...repoRoots,
      ...Object.values(businesses).flatMap((b) => Object.values(b.projects ?? {}).map((p) => p.root)),
    ];
    credentials = new Credentials({ references, repoRoots: protectedRoots, dataRoot });
    store = new Store(dataRoot);
    const broker = new Broker({
      store,
      businesses,
      secretCheck: (value) => credentials.assertClean(value),
      credentialStatus: () => credentials.status(),
    });
    // Process identities must be supplied for every managed run when using the human transport.
    const launchRecords = [];
    function bindLaunch(r) {
      if (
        !Number.isSafeInteger(r.pid) ||
        r.pid < 2 ||
        typeof r.startTime !== 'string' ||
        !/^\d+$/.test(r.startTime)
      )
        throw new BusError('invalid-launch-process');
      const cap = broker.bindLaunch(r);
      launchRecords.push({ ...r });
      return { business: r.business, run: r.run, cap };
    }
    const bound = launches.map(bindLaunch);
    const readCaps = readers.map((business) => ({ business, cap: broker.bindReader({ business }) }));
    if (tasks) {
      adapter = await tasks({ broker, credentials, businesses });
      if (typeof adapter?.handle !== 'function' || !Number.isSafeInteger(adapter.timeout) || adapter.timeout < 1)
        throw new BusError('invalid-adapter');
    }
    const path = join(store.directory, 'broker.sock');
    server = await serve({
      tasks: adapter ? { handle: adapter.handle, timeout: adapter.timeout } : null,
      broker,
      path,
      authenticateHuman: (proof) => {
        const { business } = verifyHuman(proof, {
          cliPath: fileURLToPath(new URL('./human-cli.mjs', import.meta.url)),
          launches: launchRecords,
        });
        const human = businesses[business]?.human;
        if (!human) throw new BusError('unknown-business');
        return broker.bindHuman({ business, human, via: 'cli', outsideAgent: true });
      },
      // Row 110 (#1604): a tester proves its CLI process and that no managed run is above it.
      // The broker then binds only if the business sets `bus.tester` to true.
      authenticateTester: (proof) => {
        const { business } = verifyTester(proof, {
          cliPath: fileURLToPath(new URL('./tester-cli.mjs', import.meta.url)),
          launches: launchRecords,
        });
        return broker.bindTester({ business });
      },
    });
    // S6 launcher operations. The host decides capacity and the instance list; the broker
    // authorizes role.launch and keeps the evidence, refusals included.
    // A refusal is recorded against the caller, as request() records one.
    function refusing(cap, fn, subject = null) {
      try {
        return fn();
      } catch (e) {
        if (!(e instanceof BusError)) throw e;
        broker.refuse(cap, e.code, subject);
        throw e;
      }
    }
    const authorizeLaunch = (cap, instance) =>
      refusing(cap, () => broker.authorize(cap, 'role.launch', { target: instance }));
    // Row 93 (#1587): the host's git socket. A push is checked before the host takes the bundle
    // and recorded once the ref moved; a reviewer's fetch is checked against its review.requested.
    const checkPush = (cap, task_ref) => refusing(cap, () => broker.checkPush(cap, task_ref), task_ref);
    const reviewFetch = (cap, args) => refusing(cap, () => broker.reviewFetch(cap, args), args?.task_ref);
    // Metadata only (service, date, state); the token stays in Credentials.
    function credentialStatus(business, role) {
      return credentials
        .status()
        .filter((c) => c.instance === business + '/' + role)
        .map(({ service, date, state }) => ({ service, date, state }));
    }
    return {
      broker,
      credentials,
      path,
      bindLaunch,
      endLaunch: (r) => broker.endLaunch(r),
      identity: (cap) => broker.identity(cap),
      authorizeLaunch,
      checkPush,
      recordPush: (cap, push) => refusing(cap, () => broker.recordPush(cap, push), push?.task_ref),
      reviewFetch,
      refuse: (cap, code, subject) => broker.refuse(cap, code, subject),
      credentialStatus,
      launches: bound,
      readers: readCaps,
      async close() {
        await server.close();
        await adapter?.close?.();
        store.close();
        credentials.close();
      },
    };
  } catch (e) {
    await server?.close();
    await adapter?.close?.();
    store?.close();
    credentials?.close();
    throw e;
  }
}