// The S6 launcher, inside the trusted bus host: it starts managed role
// sessions (packages/seat/src/session.mjs) and serves the launch socket the
// PM's `launch` tool calls. One launch at a time.
//
// A session asks over `<dataRoot>/bus-host/launch.sock` (0600) with one line,
// `{"cap": "<its capability>", "instance": "<role instance>"}`, and gets one
// line back: `{"ok": true, "result": {run, instance, harness, model}}` or
// `{"ok": false, "error": "<code>"}`. The host checks, in order:
//
//   1. identity: the capability names a role of this business
//   2. the business file's launch.instances lists the instance
//      (instance-not-listed)
//   3. the instance isn't already running (instance-running)
//   4. it resolves, has a role contract (no-role-contract), sets the
//      harness, model and thinking vars (no-route names the missing one)
//      and its model is a launch.models key (unsupported-harness)
//   5. the route's harness (row 65): plugins.harness switches it on, and a
//      stated fallback's too (harness-off); the plug-in loader loads it
//      (harness-refused names the module and the loader's reason); it takes
//      the thinking level (thinking-unsupported); its preflight passes
//      (the module's code, e.g. no-session-credential or
//      harness-binary-missing). A failed primary preflight takes the
//      stated fallback through the same checks; there is no other retry.
//      A harness whose launches count against launch.account gets the
//      row 126 account checks.
//   6. bwrap runs here (sandbox-unavailable)
//   7. its model names an exact provider/model table entry
//   8. total, provider and provider/model have room, including the launcher
//      (capacity-full)
//   9. every switched-on limit check allows it (limit-reached,
//      limit-check-failed), asked through the bus's plug-in loader only
//  10. the broker authorizes role.launch for the caller (its own codes:
//      launch-revoked, gated, ...)
//
// Steps 2 to 9 refuse through the broker's refuse(), so every refusal is an
// action.refused event; step 10's refusals the broker records itself. Then
// the host makes sure the workspace has a clone of each project
// (packages/seat/src/checkout.mjs): one it can't make refuses
// checkout-failed, one it fetches but doesn't fast-forward is a
// checkout-fetched record in the launch log, and one it leaves alone is a
// checkout-kept record. Then it builds the bundle, spawns the runner in its
// sandbox (packages/seat/src/sandbox.mjs) with the paths it may see, binds
// the run's process identity (the outer bwrap pid) and writes the new
// capability on the runner's stdin. When the runner exits, the host ends the launch.
//
// `bus start <business> --pm` launches launch.by the same way with
// launchedBy = the business's human and no caller capability.
//
// listen() also serves the git socket (src/git-socket.mjs, row 93), which
// the sandbox binds for a role that holds git.push.working,
// review.verdict or git.push.remote (row 199), or a merge decision to land
// (landHeld, row 137);
// session.json names it as gitSocket.
//
// Thread sessions (row 121, #1615). A thread instance (an instance whose
// role definition is in THREAD_DEFINITIONS) runs once per thread, never on its own:
// a launch from a session or --pm refuses thread-required. Every tick
// (threadTickMs) the host asks the broker for the business's threads
// (launch op threadsState) and:
//
//   - starts a session for a thread with no running session that has an
//     undelivered message to its instance, or an open newer than its last
//     claim (REQ-LAUNCH-1). The start goes through check(), so it refuses
//     as any launch does, and the next tick tries again;
//   - stops a thread session once the thread has had no message or open
//     for launch.threadIdleSeconds, the way close() stops a run. The broker
//     decides (launch op threadIdle): it releases the thread only when the
//     run is between turns, with nothing delivered and unread and nothing
//     waiting, and the SIGTERM goes only after it releases. A message sent
//     after the snapshot gets a turn from the same run or waits for the
//     next one, instead of going to the stopping run. An idle thread holds
//     no session and counts against no limit.
//   - FED R4 (#1801): never starts a thread the broker marks revoked, and
//     SIGTERMs its running session at once (the broker already ended the
//     claim). bindLaunch refuses a new run of one with thread-revoked.
//
// A thread run's sessionDir is the thread's store (session.mjs
// threadStore()), so the next run for the thread resumes the harness
// session (REQ-THREAD-2).
//
// Row 269 (264b, #1774): threadsState has one row per session of each
// thread, and everything above is per session. Each session has its own
// store (session.mjs sessionStore()) and its own workspace under
// session-workspaces/ (sessionWorkspace()), which a lock beside it keeps to
// one writer from before the clones until the run's proven exit. A legacy
// session (op legacy-3l) keeps the thread's store, and its new workspace
// shows inside the sandbox at the old instance path, the cwd its history
// recorded. The broker's half of the fence is its per-session claim.
//
// Landing runs (row 156, #1651). Every launch of a non-thread instance
// reads the broker's landHolders (launch op): the approved, unspent merge
// decisions each non-thread instance raised. The instance's entries go into
// session.json as `landing: [{decision, task_ref}]`, and the same read
// decides land_work, so a landing turn always comes with the tool. The
// runner takes one turn per entry before its receive loop. After the
// thread pass, every tick also:
//
//   - starts a holder that launch.instances lists and that isn't running,
//     through check() as any launch, once per decision. A run that ended
//     without landing isn't started again, and a refusal is logged once per
//     code and isn't a start;
//   - stops a running holder whose landing doesn't name the decision (it
//     started before the resolution, so it has no land_work for it). The
//     broker decides (launch op landRelease): it releases the role only
//     between turns, with nothing delivered and unread, and leaves waiting
//     messages undelivered for the next run. The SIGTERM goes only after it
//     releases, and the next tick starts the instance.
//
// Nothing here is persisted. After a restart recover() ends the earlier
// host's runs, so the first tick starts each holder.
//
// Row 194 (#1692, D1.1): the lifecycle pass, on the same tick, carries out the
// human's and the tester's requests the broker recorded (launch, end,
// regenerate, delete). The broker never starts a process or touches a
// workspace. Under serialize, each open request is taken (host op
// lifecycleTake, which re-checks it and closes it with a code when it may not
// run), then acted on, then closed with lifecycleDone. A taken request never
// runs again, whatever happens after the take. A done that throws leaves it
// taken with no done, which the CLI reports as outcome-unknown.
//   - launch: check() and start() with `human`, so an unlisted instance runs.
//   - end: SIGTERM to each named run's runner after the start-time check, and
//     to any run of the instance in the registry that spawned by the verb's
//     time (N6, DESIGN.md:118-120). A run spawned after it is the next end's.
//     The exit handler records the end. No SIGKILL for a terminate.
//   - a fire's end (kill: true, S6): the same runs, each through row 191's
//     stopRun with reason fire, which escalates to SIGKILL after the grace.
//     A run it killed or left alive goes back in lifecycleDone's stops.
//   - regenerate: SIGTERM to every run of the instance, then a new run once
//     the old one has left the registry. Not gone by the request's 60 s mark
//     closes it regenerate-timeout and starts nothing.
//   While row 191's switch is on the tick runs only this pass (D6): a
//   terminate still ends runs, and the take closes a launch or regenerate
//   processing-halted.
//   - delete: instance-running against the registry, then lifecycleDelete
//     (the writer's commit), then the rename of workspaces/<b>/I/ to
//     workspaces/<b>/.retired/I-<transaction>/, and of
//     session-workspaces/<b>/I/ to session-workspaces/<b>/.retired/I-<transaction>/
//     (row 269). A failed rename closes it workspace-rename-failed with the
//     transaction; a later delete retries it.

import { createServer } from "node:net";
import { createHash } from 'node:crypto';
import { watchModelOutput } from './model-stall.mjs';
import { chmodSync, existsSync, lstatSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
import { dirname, join, relative } from "node:path";
import { configDir, loadBusiness, resolveInstance, systemVars } from "../../business/src/index.mjs";
import { organizationWriter } from "../../business/src/organization.mjs";
import { businessFilePath, validateBusinessDocument } from "../../business/src/business.mjs";
import { selectModelProvider } from "../../plugins/src/model-provider-selection.mjs";
import { THREAD_DEFINITIONS } from "../../business/src/vocabulary.mjs";
import { limitHit } from "../../business/src/launch.mjs";
import { BundleError, buildBundle, hasFallback, readSkills, sessionModel } from "../../harness/src/bundle.mjs";
import { createLoader } from "../../plugins/src/loader.mjs";
import {
  appendLaunchLog,
  endReason,
  gitSocketPath,
  launchSocketPath,
  newRun,
  readSessions,
  Registry,
  runnerOf,
  SESSION_CREDENTIALS,
  sessionEnv,
  isProviderVariable,
  sessionExit,
  sessionStore,
  sessionWorkspace,
  lockSessionWorkspace,
  spawnSession,
  workspaceDir,
  writeSessionFile,
} from "../../seat/src/session.mjs";
import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
import { collectCompactions, READ_BYTES, readBudget, readStreams, STREAMS } from "../../seat/src/compactions.mjs";
import { CheckoutError, ensureCheckouts } from "../../seat/src/checkout.mjs";
import { createGitSocket } from "./git-socket.mjs";
import { cmdlineOf, startTimeOf } from "../../seat/src/proc.mjs";
import { rolesDir } from "./config.mjs";
import { CliError } from "./errors.mjs";
import { assertProcessing, processingStatus, ProcessingError } from "../../processing-control/src/state.mjs";
import { registerRuntime, updateRuntime, runtime, unregisterRuntime, processIdentity, stopRun } from "../../processing-control/src/runtimes.mjs";
import { withProcessing } from '../../processing-control/src/effects.mjs';

const STOP_TIMEOUT_MS = 30000;
const LINE_MAX = 4096;
// How often the host asks for the threads' state; a poll period, not a limit.
const THREAD_TICK_MS = 2000;
// Row 194: the broker's LIFECYCLE_REQUEST_MS. A regenerate's old run must be gone by then.
const LIFECYCLE_REQUEST_MS = 60000;
// #774 R2: the session proxy key the first check of a proxy launch gives the harness preflights in
// place of the key P2 hasn't admitted yet. A fixed marker, never a credential; it never leaves check().
export const PENDING_ADMISSION_KEY = "pending-admission";

// A refusal the host decided; `code` goes back to the caller and to the broker.
class Refusal extends Error {
  constructor(code, message = code) {
    super(message);
    this.code = code;
  }
}

// A harness module runs with the host's authority, and anything it throws refuses the launch
// and names the module rather than escaping the host.
function call(name, what, fn) {
  try {
    return fn();
  } catch (e) {
    throw new Refusal("harness-failed", `harness ${name} ${what}: ${e.message}`);
  }
}

// host: startHost()'s handle. tracker: true when the broker has task verbs for
// the business. launches: false for `bus start --no-launch` (row 163, the D2
// rehearsal): start() refuses launches-disabled, the ticker runs only the
// lifecycle pass (row 194), which closes each request launches-disabled, a
// session's launch request is refused, recover() kills and ends nothing, and
// sessions.json stays as it was.
// makeLoader (the plug-in loader's factory, which tests point at a
// scratch tree), bwrap, readOnly, sessionDefaults, harnessVersion,
// stopTimeoutMs, threadTickMs, lifecycleRequestMs, git (createGitSocket
// options) and stopOptions (row 191's stopRun options, for a fire's stop)
// exist for the tests; the command line never sets them.
export function createLauncher({
  host,
  system,
  env = process.env,
  // Tests only (#774 F4): a static session proxy context {baseUrl,key}, or a
  // factory giving one per check, for a provider "proxy" route while
  // system.proxy is unset. It skips P2 admission, so createLauncher refuses it
  // together with system.proxy. P3 never reads a key store or calls the proxy.
  testProxy = null,
  tracker = false,
  launches = true,
  makeLoader = createLoader,
  bwrap = BWRAP,
  readOnly = [],
  sessionDefaults = {},
  harnessVersion = undefined,
  stopTimeoutMs = STOP_TIMEOUT_MS,
  threadTickMs = THREAD_TICK_MS,
  lifecycleRequestMs = LIFECYCLE_REQUEST_MS,
  git = {},
  stopOptions = {},
  log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`),
}) {
  if (testProxy !== null && system.proxy)
    throw Object.assign(new Error("testProxy skips P2 admission, so it is refused with system.proxy set"), { code: "test-proxy-refused" });
  const dataRoot = system.dataRoot;
  function admission(generation = null) {
    try { return assertProcessing(dataRoot, generation); }
    catch (e) {
      if (e instanceof ProcessingError) throw new Refusal(e.code);
      throw e;
    }
  }
  // Row 64 (#1557): where a role's skills are read from; scripts/skill.sh activate fills it.
  const skillsDir = join(system.instanceRoot ?? dataRoot, "skills-enabled");
  let b = loadBusiness(host.business, { dir: configDir(env), rolesDir: rolesDir(env) });
  const vars = systemVars(system);

  // Sessions an earlier host left in sessions.json: it died without ending
  // them, so their runs still hold their roles. listen() ends them. A file
  // that can't be read refuses the host rather than drop them unseen.
  let leftover;
  try {
    leftover = readSessions(dataRoot).filter((s) => s.business === b.id);
  } catch (e) {
    throw new CliError(`${e.message}; check which sessions still run, then move the file aside`, 3);
  }
  const registry = new Registry(dataRoot, { readOnly: !launches });
  // Row 269: earlier-host runs still live after recover(). check() sees each while it lives.
  const survivors = [];
  const liveRun = (s) => startTimeOf(s.pid) === s.startTime || startTimeOf(s.runnerPid) === s.runnerStartTime;
  const children = new Map();
  let chain = Promise.resolve();
  let closed = false;
  let server = null;
  let ticker = null;
  let ticking = false;
  const stopping = new Set();
  const refusedThreads = new Map();
  const startedFor = new Map();
  // Row 271 (264d): each clear's stop time (by op), and each clear refusal logged once (by op and code).
  const clearStops = new Map();
  const clearLogged = new Map();
  // Row 194 N6: run -> when its process was spawned, so an end signals only the runs spawned by its verb.
  const spawnedAt = new Map();
  // Row 156: run -> the decisions its landing names; decision -> started; decision -> last
  // refusal code; run -> last landRelease refusal code; and landHolders' last refusal.
  const landingOf = new Map();
  const landStarted = new Set();
  const refusedLand = new Map();
  const refusedRelease = new Map();
  let refusedHolders = null;
  let refusedRequests = null;
  const requestDrains = new Map();
  const sockets = new Set();
  const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git });

  const record = (entry) => {
    try {
      appendLaunchLog(dataRoot, b.id, entry);
    } catch (e) {
      log(`launch log: ${e.message}`);
    }
  };

  // Row 270 (264c, D6): count a thread session's completed compactions into its
  // store's ledger, at run start (what earlier runs and a resume left) and at
  // run end. The caller holds the session's workspace lock. A failure is
  // logged and recorded, and never stops or fails the run: the count then
  // reads unknown with its reason. Row 270 r2 (B1, mos-1352): each collection
  // reads at most the system config's compactionReadBytes, a human-only
  // setting; a missing or bad value takes the default.
  const budget = readBudget(system?.compactionReadBytes);
  if (system?.compactionReadBytes !== undefined && budget !== system.compactionReadBytes) {
    log(`system config: compactionReadBytes ${JSON.stringify(system.compactionReadBytes)} isn't a whole number of bytes from 1048576 to 1073741824; compaction collections read up to the default ${READ_BYTES}`);
  }
  // Row 270 r3 (B4, sage-3038): and opens at most compactionStreams stream
  // files, the same kind of setting.
  const streams = readStreams(system?.compactionStreams);
  if (system?.compactionStreams !== undefined && streams !== system.compactionStreams) {
    log(`system config: compactionStreams ${JSON.stringify(system.compactionStreams)} isn't a whole number from 32 to 65536; compaction collections open up to the default ${STREAMS} streams`);
  }
  // Row 307 (#1814): the route's harness module reads its own store through the kind's
  // `compactions`. The call skips call(): call() turns a throw into a launch refusal, and a
  // collection that fails must never stop or fail the run, so this catch records it instead.
  const compactions = (store, name, harness, { instance, thread, session, run }) => {
    let c;
    try {
      c = collectCompactions(store, { name, compactions: harness.compactions }, { budget, streams });
    } catch (e) {
      log(`run ${run}: compactions: ${e.message}`);
      c = { count: null, known: false, reason: "store-unreadable" };
    }
    record({ event: "compactions", instance, thread, session, run, count: c.count, known: c.known, reason: c.reason });
  };

  function serialize(fn) {
    const run = chain.then(fn);
    chain = run.catch(() => {});
    return run;
  }

  // Steps 2 to 9. Returns what the launch needs. Only `--pm` may name
  // launch.by itself. A thread instance launches only for a thread, and
  // nothing else launches for one. Row 194 (#1692, V2): `human` is the
  // human's launch request, which may name any declared instance. The listing
  // limits the chief, not the human. launch.by still takes the `pm` branch.
  async function check(instance, { pm = false, thread = null, session = null, human = false, proxyContext = null, proxyReady = false } = {}) {
    const admissionGeneration = admission().generation;
    let businessRevision;
    try {
      const snapshot = organizationWriter({dir:configDir(env),rolesDir:rolesDir(env)}).read(host.business);
      businessRevision = snapshot.revision;
      b = validateBusinessDocument(snapshot.document, businessFilePath(host.business, configDir(env)), {rolesDir:rolesDir(env)});
    } catch(e){throw new Refusal("organization-refused",e.message);}
    if (!b.launch) throw new Refusal("launch-not-configured", `business ${b.id} has no launch block`);
    if (!(pm ? instance === b.launch.by : b.launch.instances.includes(instance) || (human && Object.hasOwn(b.roles, instance)))) throw new Refusal("instance-not-listed");
    const threaded = THREAD_DEFINITIONS.includes(b.roles[instance]?.definition);
    if (threaded && !thread) throw new Refusal("thread-required", `${instance} runs only for a thread`);
    if (!threaded && thread) throw new Refusal("thread-not-allowed", `${instance} isn't a thread instance`);
    if (threaded && !session) throw new Refusal("session-required", `${instance} runs for one session of a thread`);
    if (registry.running(b.id, instance, thread, session)) throw new Refusal("instance-running");
    // Row 269 (Q2): a run an earlier host started with no session works in the instance's old
    // workspace and the thread's store, so no session of the instance starts while one lives.
    if (threaded && [...registry.sessions, ...leftover, ...survivors.filter(liveRun)].some((s) => s.business === b.id && s.instance === instance && !s.session))
      throw new Refusal("legacy-run-live", `a run of ${instance} with no session is still live`);
    let resolved;
    try {
      resolved = resolveInstance({ system: vars, business: b, instance });
    } catch (e) {
      throw new Refusal("unsupported-harness", e.message);
    }
    // Resolve the business's current provider modules with a fresh loader; a refused
    // load never inherits a previous business's authority.
    const providerNames = b.plugins?.['model-provider'] ?? [];
    const providerModules = new Map();
    if (providerNames.length) {
      const loader = makeLoader({ log: entry => record({ ...entry, instance }) });
      const loaded = await loader.load({ 'model-provider': providerNames });
      if (loaded.refused.length) throw new Refusal('model-provider-unavailable', 'model provider module refused');
      for (const { name, module } of loader.snapshot('model-provider')) providerModules.set(name, module);
    }
    const resolver = providerModules.size ? ({ selection, harness }) => {
      const selected = selectModelProvider([...providerModules].map(([name,module]) => ({name,module})), selection);
      if (!selected) return { code: 'model-provider-unavailable' };
      return selected.module.route({ selection, harness });
    } : undefined;
    // Row 65 (#1558): the route's harness, model and thinking come from the instance's vars only.
    const routeOf = (which) => {
      try {
        return sessionModel(resolved, b.launch.models, which, { proxy: Boolean(system.proxy || proxyContext), resolver });
      } catch (e) {
        const code = e instanceof BundleError && (e.code === "no-route" || e.code.startsWith("proxy-") || e.code.startsWith("model-")) ? e.code : "unsupported-harness";
        throw new Refusal(code, e.message);
      }
    };
    const primary = routeOf("primary");
    const fallback = hasFallback(resolved) ? routeOf("fallback") : null;
    if (!resolved.contract) throw new Refusal("no-role-contract", `role ${resolved.definition} has no contract file`);
    // Row 64 (#1557): every skill on the role's list is enabled and readable before a run exists.
    try {
      readSkills(resolved.skills, skillsDir);
    } catch (e) {
      if (e instanceof BundleError) throw new Refusal(e.code === "skill-not-enabled" ? "skill-not-enabled" : "skill-refused", e.message);
      throw e;
    }
    // Both of a launch's harnesses must be switched on, whichever it takes.
    const on = Array.isArray(b.plugins?.harness) ? b.plugins.harness : [];
    for (const m of [primary, fallback].filter(Boolean)) {
      if (!on.includes(m.harness)) throw new Refusal("harness-off", `harness ${m.harness} isn't switched on in plugins.harness`);
    }
    // One harness through a fresh loader, so a refusal never leaves an earlier load's module in use.
    const harnessOf = async (m) => {
      const loader = makeLoader({ log: (entry) => record({ ...entry, instance }) });
      const { refused } = await loader.load({ harness: [m.harness] });
      if (refused.length) throw new Refusal("harness-refused", `harness ${m.harness}: ${refused.map((r) => `${r.code}: ${r.message}`).join("; ")}`);
      const module = loader.modules("harness")[0];
      const levels = call(m.harness, "levels", () => module.levels());
      if (!levels.includes(m.thinking)) throw new Refusal("thinking-unsupported", `harness ${m.harness} takes thinking ${levels.join(", ")}, not ${m.thinking}`);
      return module;
    };
    let model = primary;
    let harness = await harnessOf(primary);
    let route = "primary";
    let routeReason = null;
    // A launch is checked twice. Without system.proxy both are full checks: the first before the
    // authorization, workspace and checkout work, the final one in startLocked before newRun. With
    // system.proxy the first check comes before P2 admits a key in startLocked, so it predicts the
    // route: it runs the harness preflights on a stand-in key and picks the route as the final check
    // does (#774 R2). The prediction holds because a harness preflight tests only that a proxy key
    // is present, never its value. On the predicted route it runs the account declaration
    // (credential "proxy" only), the sandbox, capacity, the limit checks and admission, so a launch
    // that can never start costs no authorization record, checkout or key (#774 F1). It skips only
    // the match of the session proxy key against launch.account.name. The final check is always
    // full, on the admitted key, and it picks the route again: if a harness binary appears or goes
    // between the two checks, its pick decides. Never reject proxy-only login using upstream host env.
    const early = Boolean(system.proxy) && !proxyReady;
    // The checks that follow the route's preflight, on the route the caller picked. In the first
    // check of a proxy launch sessionProxy is null, and the account must then name credential "proxy".
    const gate = async (model, harness, sessionProxy) => {
      const proxied = early || Boolean(sessionProxy);
      // Row 126 (#1620): only a launch on a harness whose module counts the account spends the named
      // account. The bus environment must hold the named set and no other credential variable, so the
      // session can't log in some other way. Messages name variables, never values.
      const account = call(model.harness, "countsAccount", () => harness.countsAccount()) === true ? b.launch.account?.name ?? null : null;
      if (account !== null) {
        // The usage records keep this name (bus schema 3i), so it must never be a credential's value.
        const holds = [...new Set(SESSION_CREDENTIALS.flat())].filter((k) => env[k] === account);
        if (sessionProxy?.key === account) holds.push("session proxy key");
        if (holds.length) throw new Refusal("account-is-credential", `launch.account.name is the value of ${holds.join(", ")}; it names the account, never its credential`);
        if (proxied || b.launch.account.credential === "proxy") {
          if (!proxied || b.launch.account.credential !== "proxy")
            throw new Refusal("account-credential-missing", `account ${account} needs the declared proxy credential`);
        } else {
          const named = b.launch.account.credential.split("+");
          if (!SESSION_CREDENTIALS.some((set) => set.join("+") === named.join("+")) || !named.every((k) => Boolean(env[k])))
            throw new Refusal("account-credential-missing", `account ${account} names ${named.join(" with ")}, which the bus host doesn't hold as a session login`);
          const other = [...new Set(SESSION_CREDENTIALS.flat())].filter((k) => !named.includes(k) && Boolean(env[k]));
          if (other.length) throw new Refusal("account-credential-ambiguous", `account ${account} names ${named.join(" with ")}, but the bus host also sets ${other.join(", ")}`);
        }
      }
      const unavailable = sandboxProblem({ bwrap });
      if (unavailable) throw new Refusal("sandbox-unavailable", unavailable);
      const hit=limitHit(registry.sessions,b.id,b.launch.limits,model);
      if(hit){const error=new Refusal("capacity-full",JSON.stringify(hit));error.limit=hit;throw error;}
      // Row 125 (#1619, REQ-LIMIT-4): the limit checks. An answer that isn't an allow refuses,
      // and so does a bus that can't answer (fail closed, lead decision 150).
      let may;
      try {
        may = await host.op({ op: "mayLaunch", business: b.id, instance, provider: model.provider, model: model.model, account });
      } catch (e) {
        throw new Refusal("limit-check-failed", `bus: ${e.code ?? "launch-op-refused"}`);
      }
      // Row 333: under a daily key quota the bus refuses key-quota-no-model-key for an agent P2
      // hasn't issued a key yet. Only the first check of a proxy launch lets that one answer pass:
      // startLocked's modelPreflight issues the key, and the final check asks again with it.
      const deferredKey = early && may?.allow === false && may.check === null && may.code === "key-quota-no-model-key";
      if (may?.allow !== true && !deferredKey) {
        if (may?.hit) { const error = new Refusal("limit-reached", JSON.stringify(may.hit)); error.limit = { check: may.check, ...may.hit }; throw error; }
        throw new Refusal("limit-check-failed", `${may?.check ?? "bus"}: ${may?.code ?? "malformed-answer"}`);
      }
      const selected = b.launch.models[model.modelKey];
      if (selected?.module || (proxied && providerModules.size)) {
        let checked;
        try { checked = await host.op({ op: 'modelAdmission', business: b.id, instance, modelKey: model.modelKey, harness: model.harness,
          revision: businessRevision, entry: selected, provider: model.provider, model: model.model, account, route, transport: model.proxyTransport }); }
        catch { throw new Refusal('model-catalogue-unavailable', 'model catalogue admission unavailable'); }
        if (checked?.allow !== true) {
          const code = ['model-provider-unavailable', 'model-catalogue-unavailable', 'model-harness-incompatible'].includes(checked?.code) ? checked.code : 'model-provider-unavailable';
          throw new Refusal(code, `model ${selected?.module ?? 'legacy'}/${selected?.upstream ?? selected?.provider}/${selected?.model} unavailable (${code})`);
        }
        if (!proxied) throw new Refusal('model-provider-unavailable', 'module-backed model requires a proxy');
      }
      return account;
    };
    const proxyFor = async (selected) => {
      if (!system.proxy && selected.provider !== "proxy") return null;
      const context = system.proxy ? proxyContext : typeof testProxy === "function" ? await testProxy({ business: b.id, instance }) : testProxy;
      if (!context || typeof context.key !== "string" || !context.key
        || typeof context.baseUrl !== "string" || !/^http:\/\/127\.0\.0\.1:[0-9]+$/.test(context.baseUrl)) {
        throw new Refusal("no-session-credential", "proxy model needs its own session proxy context");
      }
      // Copy only the agreed seam; never keep a store or management handle.
      return { baseUrl: context.baseUrl, key: context.key };
    };
    // The route both checks pick (#774 R2): the primary when its preflight passes, else the fallback
    // through its own preflight. proxyOf gives the session proxy for a route; the preflights see it,
    // and the helper returns the last one they saw.
    const pickRoute = async (proxyOf) => {
      let sessionProxy = await proxyOf(primary);
      const failed = call(primary.harness, "preflight", () => harness.preflight({ env: sessionProxy ? sessionEnv(env, "preflight", sessionProxy) : env, proxy: sessionProxy }));
      if (failed) {
        if (!fallback) throw new Refusal(failed.code, failed.message);
        // The fallback only on a failed primary preflight, and only at launch (row 65). Any refusal of
        // the fallback names the primary's failure too, so the reason it was tried stays visible.
        const second = await harnessOf(fallback).catch((e) => {
          if (e instanceof Refusal) e.message += ` (primary ${primary.harness}: ${failed.message})`;
          throw e;
        });
        sessionProxy = await proxyOf(fallback);
        const also = call(fallback.harness, "preflight", () => second.preflight({ env: sessionProxy ? sessionEnv(env, "preflight", sessionProxy) : env, proxy: sessionProxy }));
        if (also) throw new Refusal(also.code, `fallback ${fallback.harness}: ${also.message} (primary ${primary.harness}: ${failed.message})`);
        [model, harness, route, routeReason] = [fallback, second, "fallback", failed.code];
      }
      return sessionProxy;
    };
    if (early) {
      // The first check predicts the final check's route. A harness preflight tests only that a proxy
      // key is present, so a stand-in key gets the answer the admitted key will get, and the route
      // picked here is the one the final check picks unless a harness binary appears or goes in
      // between; then the final check's own pick decides. The stand-in reaches only the preflights:
      // pickRoute's answer is dropped, gate() gets null, and nothing records or returns it. The gate
      // then runs on the predicted route only, with the codes, messages and limits the final check uses.
      const standIn = { baseUrl: Number.isInteger(system.proxy.port) ? `http://127.0.0.1:${system.proxy.port}` : "http://127.0.0.1:0", key: PENDING_ADMISSION_KEY };
      await pickRoute(() => standIn);
      await gate(model, harness, null);
      admission(admissionGeneration);
      return { resolved, model, account: null, harness, route, routeReason, admissionGeneration, resolver };
    }
    const sessionProxy = await pickRoute(proxyFor);
    const account = await gate(model, harness, sessionProxy);
    admission(admissionGeneration);
    return { resolved, model, account, harness, route, routeReason, admissionGeneration, sessionProxy, resolver };
  }

  // Reread host state on every new-run path, including opens/clears queued before archive.
  async function threadAdmission(thread, session) {
    if (!thread) return;
    const rows = await host.op({ op: "threadsState", business: b.id });
    const row = rows.find(r => r.thread === thread && r.session === session);
    if (!row) throw new Refusal("session-not-found");
    if (row.archived) throw new Refusal("project-archived");
  }

  // Bundle, spawn, bind, register, hand over the capability.
  async function start({ instance, launchedBy, resolved, model, account = null, thread = null, session = null, legacy = false, legacyThread = legacy, human = false, admissionGeneration }) {
    admission(admissionGeneration);
    // Every launch path ends here, so a --no-launch host spawns nothing whichever path asked.
    if (!launches) throw new Refusal("launches-disabled", "this host was started with --no-launch");
    await threadAdmission(thread, session);
    // A thread session's store is checked first, so a bad id or an unsafe
    // directory launches nothing.
    let store = null;
    if (thread) {
      try {
        store = sessionStore(dataRoot, b.id, thread, session, { legacy, legacyThread });
      } catch (e) {
        throw new Refusal(e.code === "legacy-store-unmapped" ? e.code : "thread-store-refused", e.message);
      }
    }
    const credentials = await host.op({ op: "credentialStatus", business: b.id, role: instance });
    admission(admissionGeneration);
    // The clones come before the run, so one that can't be made leaves no
    // run behind. The host's git treats each clone as hostile, and its
    // checks hold only while no session writes the workspace. A non-thread
    // workspace is one instance's, check() refused instance-running, and
    // launches are serialized. A thread session's workspace is its own (row
    // 269), and its lock refuses a second writer, whichever host asks.
    let workspace;
    let workspaceDest = null;
    let lock = null;
    if (thread) {
      try {
        const live = registry.sessions.filter((s) => s.business === b.id && s.workspace).map((s) => s.workspace);
        workspace = sessionWorkspace(dataRoot, b.id, instance, session, { live });
        lock = lockSessionWorkspace(workspace, { hostPid: process.pid, hostStartTime: startTimeOf(process.pid), business: b.id, instance, thread, session });
      } catch (e) {
        if (["session-workspace-unsafe", "session-workspace-overlap", "session-workspace-busy"].includes(e.code)) throw new Refusal(e.code, e.message);
        throw e;
      }
      if (legacy) workspaceDest = workspaceDir(dataRoot, b.id, instance);
    } else {
      workspace = workspaceDir(dataRoot, b.id, instance);
      mkdirSync(workspace, { recursive: true, mode: 0o700 });
    }
    try {
      return await startLocked({ instance, launchedBy, resolved, model, account, thread, session, human, admissionGeneration, store, credentials, workspace, workspaceDest, lock });
    } catch (e) {
      // Every failure path in startLocked has proven the run's exit, or never spawned one.
      lock?.release();
      throw e;
    }
  }

  // The rest of start(): clones, bundle, spawn, bind, register. The lock, when there is one, is
  // released at the run's close.
  async function startLocked({ instance, launchedBy, resolved, model, account, thread, session, human, admissionGeneration, store, credentials, workspace, workspaceDest, lock }) {
    // What the session sees: a legacy session's workspace at the old instance path.
    const inSandbox = workspaceDest ?? workspace;
    // Lifecycle Task 2 (#797): a product the request lifecycle built under <dataRoot>/products is cloned only
    // once the bus has admitted it. The bus answers from its receipts, so a reserved or half-built product
    // (a mapping published before the seed and records finished) is left out and the launch goes on without
    // it. Ordinary mappings also recheck archive state through the same host admission.
    const codeProjects = {};
    for (const [project, entry] of Object.entries(b.projects)) {
      if (entry.kind === 'library') continue;
      let answer = null;
      try { answer = await host.op({ op: 'request.admission', business: b.id, project }); } catch (e) { answer = { ready: false, code: e.code ?? 'admission-unavailable' }; }
      if (answer?.ready !== true) { record({ event: "checkout-skipped", instance, project, reason: answer?.code ?? 'project-pending' }); continue; }
      codeProjects[project] = entry;
    }
    await threadAdmission(thread, session);
    let checkouts;
    try {
      checkouts = await withProcessing(dataRoot, () => {
        admission(admissionGeneration);
        return ensureCheckouts({ workspace, projects: codeProjects, env });
      });
    } catch (e) {
      if (e instanceof CheckoutError) throw new Refusal("checkout-failed", `${e.project}: ${e.message}`);
      throw e;
    }
    for (const c of checkouts) {
      if (c.outcome === "kept") record({ event: "checkout-kept", instance, project: c.project, reason: c.reason });
      else if (c.outcome === "fetched") record({ event: "checkout-fetched", instance, project: c.project, reason: c.reason });
    }
    // Retain P2's real admission/ownership operation; do not substitute the
    // old stand-in option or acquire again in each metadata/preflight check.
    let admittedProxy = null;
    if (system.proxy) {
      try { admittedProxy = await host.op({ op: 'modelPreflight', business: b.id, role: instance }); }
      catch (error) { throw new Refusal(error.code ?? 'proxy-down', error.code ?? 'proxy-down'); }
      if (!admittedProxy || typeof admittedProxy.key !== 'string' || !admittedProxy.key || admittedProxy.baseUrl !== `http://127.0.0.1:${system.proxy.port}`) throw new Refusal('proxy-down', 'proxy-down');
    }
    let harness, route, routeReason, sessionProxy, resolver;
    ({resolved,model,account,harness,route,routeReason,sessionProxy,resolver}=await check(instance,{pm:instance===b.launch.by,thread,session,human,proxyContext:admittedProxy,proxyReady:true}));
    admission(admissionGeneration);
    const { run, runDir } = newRun(dataRoot, b.id);
    let modelKeyId;
    if (admittedProxy) {
      modelKeyId = createHash('sha256').update(sessionProxy.key).digest('hex').slice(0, 12);
      // Internal opt-in only: no session.json, user environment or config field.
      try { writeFileSync(join(runDir, 'proxy-first-turn-enabled'), 'enabled\n', { flag: 'wx', mode: 0o600 }); }
      catch { throw new Refusal('launch-failed', 'first-turn protocol setup failed'); }
    }
    const sessionDir = store ? store.session : join(runDir, "session");
    if (!store) mkdirSync(sessionDir, { mode: 0o700 });
    if (store) compactions(store, model.harness, harness, { instance, thread, session, run });
    const toolSocket = join(runDir, "tools.sock");
    const turnMarker = join(runDir, "turn.done");
    const version = harnessVersion !== undefined ? harnessVersion : call(model.harness, "version", () => harness.version({ env }));
    // Row 137 (#1631): a role and thread holding an approved, unspent merge decision get land_work.
    // The broker checks the decision again on every land; this only decides what the bundle shows.
    // Row 156 (#1651): a non-thread instance's land and landing come from one landHolders read, so a
    // decision resolved between two reads can't give a landing turn without land_work.
    const holders = thread ? null : await host.op({ op: "landHolders", business: b.id });
    if (holders !== null && !Array.isArray(holders)) throw new Refusal("launch-failed", "landHolders: malformed answer");
    const holding = holders?.filter((h) => h.instance === instance) ?? null;
    const land = thread ? await host.op({ op: "landHeld", record: { business: b.id, role: instance, thread } }) : holding.length > 0;
    const bundle = buildBundle({
      dir: join(runDir, "bundle"),
      resolved,
      models: b.launch.models,
      proxy: sessionProxy,
      resolver,
      contract: readFileSync(resolved.contract, "utf8"),
      arbiter: Object.values(b.arbiters).includes(instance),
      tracker,
      library: Object.values(b.projects ?? {}).some(project => project.kind === "library"),
      credentials,
      session: { run, launchedBy, workspace: inSandbox, toolSocket, turnMarker, checkouts: inSandbox === workspace ? checkouts : checkouts.map((c) => ({ ...c, path: join(inSandbox, relative(workspace, c.path)) })) },
      harness,
      route,
      routeReason,
      harnessVersion: version,
      land,
      skillsDir,
      personaDir: join(configDir(env), "personas"),
    });
    const { manifest, ...files } = bundle.files;
    const adapter = call(model.harness, "adapter", () => harness.adapter());
    const turnEnd = call(model.harness, "turnEnd", () => harness.turnEnd());
    const outputPolicy = call(model.harness, "outputPolicy", () => harness.outputPolicy({ proxy: Boolean(sessionProxy) }));
    if (!["adapter", "non-newline-stdout", "stdout"].includes(outputPolicy)) throw new Refusal("harness-refused", "invalid harness output policy");
    const transcript = call(model.harness, "transcript", () => harness.transcript({ runDir, store }));
    const harnessMounts = call(model.harness, "mounts", () => harness.mounts({ env }));
    // Row 93: the git socket, for a role that pushes work or reviews it. Row 137: or lands it.
    // Row 199 (D11): or publishes it, once a reviewed commit routes git.push.remote within-role.
    const held = [...resolved.limits.authority.withinRole, ...resolved.limits.authority.crossRole];
    const git = land || ["git.push.working", "review.verdict", "git.push.remote"].some((a) => held.includes(a)) ? gitSocketPath(dataRoot) : null;
    // Secrets go only to spawn, never session.json. Keep this split at the
    // actual serialization boundary, even if a harness adds another key.
    const harnessEnv = {}, transientEnv = {};
    for (const [name, value] of Object.entries(bundle.env)) {
      (sessionProxy && isProviderVariable(name) ? transientEnv : harnessEnv)[name] = value;
    }
    writeSessionFile(runDir, {
      ...sessionDefaults,
      business: b.id,
      instance,
      ...(thread ? { thread } : {}),
      run,
      launchedBy,
      harness: model.harness,
      provider: model.provider,
      model: model.model,
      brokerSocket: host.path,
      launchSocket: launchSocketPath(dataRoot),
      gitSocket: git,
      workspace: inSandbox,
      sessionDir,
      adapter,
      bundle: files,
      harnessEnv,
      turnEnd,
      outputPolicy,
      outputPolicyVersion: 1,
      toolSocket,
      turnMarker,
      ...(holding?.length ? { landing: holding.map(({ decision, task_ref }) => ({ decision, task_ref })) } : {}),
    });

    const paths = {
      workspace,
      workspaceDest,
      adapter,
      harnessMounts,
      brokerSocket: host.path,
      launchSocket: resolved.launch ? launchSocketPath(dataRoot) : null,
      gitSocket: git,
      threadStore: store,
      readOnly,
    };
    admission(admissionGeneration);
    await threadAdmission(thread, session);
    // Archive can change while the checkout/bundle is prepared. Never spawn with a stale manifest.
    for (const project of Object.keys(codeProjects)) {
      const ready = await host.op({ op: "request.admission", business: b.id, project });
      if (ready?.ready !== true) throw new Refusal(ready?.code ?? "project-pending");
    }
    // One synchronous broker snapshot covers the thread and the entire manifest after every
    // asynchronous preparation read. Admission is the launch boundary; bindLaunch remains a fence.
    const finalAdmission = await host.op({ op: "request.admission", business: b.id, projects: Object.keys(codeProjects), thread, session });
    if (finalAdmission?.ready !== true) throw new Refusal(finalAdmission?.code ?? "project-pending");
    const managed = registerRuntime(dataRoot, { kind: 'session', business: b.id, run });
    let child;
    try {
      admission(admissionGeneration);
      spawnedAt.set(run, Date.now());
      child = spawnSession({ runDir, env: { ...sessionEnv(env, run, sessionProxy), ...transientEnv }, paths, business: b.id, role: instance, bwrap });
    } catch (e) {
      spawnedAt.delete(run);
      unregisterRuntime(dataRoot, managed.id);
      throw e;
    }
    let bound = false;
    let entry = null;
    const exited = new Promise((resolve) => child.once("close", (code, signal) => resolve(sessionExit(code, signal))));
    child.on("error", (e) => log(`run ${run}: ${e.message}`));
    child.stdin.on("error", () => {});
    try {
      if (child.pid === undefined) throw new Error("the session process did not start");
      const outerStart = startTimeOf(child.pid);
      const identity = processIdentity(child.pid);
      if (!identity) throw new Error("the session process exited at once");
      updateRuntime(dataRoot, managed.id, { ...identity, group: true, phase: 'running' });
      const runner = await runnerOf(child);
      updateRuntime(dataRoot, managed.id, { runnerPid: runner.pid, runnerStartTime: runner.startTime });
      admission(admissionGeneration);
      if (!outerStart) throw new Error("the session process exited at once");
      lock?.update({ run, runPid: child.pid, runStartTime: outerStart });
      const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, provider: model.provider, model: model.model, account, transcript, pid: child.pid, startTime: outerStart, ...(thread ? { thread, session } : {}), ...(admittedProxy ? { modelKeyId } : {}) });
      bound = true;
      admission(admissionGeneration);
      entry = {
        business: b.id, instance, ...(thread ? { thread, session, workspace } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey,
        pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(),
        ...(admittedProxy ? { modelKeyId } : {}),
      };
      registry.add(entry);
      // Row 156: one start per decision, whichever launch wrote it into landing.
      landingOf.set(run, new Set(holding?.map((h) => h.decision) ?? []));
      for (const h of holding ?? []) landStarted.add(h.decision);
      child.stdin.end(`${JSON.stringify({ cap: launch.cap })}\n`);
    } catch (e) {
      child.kill("SIGKILL");
      await exited;
      if (bound) await endRun(run, "failed", null);
      spawnedAt.delete(run);
      unregisterRuntime(dataRoot, managed.id);
      if (e instanceof Refusal) throw e;
      if (e.code?.startsWith('proxy-')) throw new Refusal(e.code, e.code);
      throw new Refusal("launch-failed", e.message);
    }
    children.set(run, exited);
    const outputWatch = admittedProxy ? watchModelOutput({ runDir, armAtFirstTurn: true, keyId: modelKeyId, seconds: system.proxy.firstOutputSeconds ?? 120,
      stop: () => host.op({ op: 'modelStall', business: b.id, role: instance, run, keyId: modelKeyId }), receipt: record }) : null;
    exited.then(async ({ code, signal }) => {
      outputWatch?.cancel();
      let receipt;
      try { receipt = runtime(dataRoot, managed.id); }
      catch { log(`run ${run}: processing receipt unavailable at exit`); }
      const state = processingStatus(dataRoot);
      const haltGeneration = receipt?.haltGeneration ?? (state.halted && !state.invalid ? state.generation : null);
      const reason = haltGeneration ? 'kill-switch' : endReason(code, signal);
      try {
        const ended = await endRun(run, reason, signal ? null : code);
        // Keep the exact launch identity for the next host if the broker
        // refused or did not confirm the end.
        if (typeof ended?.released === "boolean") registry.remove(b.id, run);
        // Row 125 (#1619, REQ-LIMIT-4): a run a limit check stopped names the limit, from the claim row.
        record({ event: "end", instance, ...(thread ? { thread, session } : {}), run, reason, exitCode: code, signal, ...(haltGeneration ? { haltGeneration } : {}), ...(ended?.limit ? { limit: ended.limit } : {}) });
        log(`run ${run} (${instance}) ended: ${reason}`);
      } catch (e) {
        log(`run ${run}: ${e.message}`);
      } finally {
        try { unregisterRuntime(dataRoot, managed.id); }
        catch { log(`run ${run}: processing receipt cleanup refused`); }
        if (store) compactions(store, model.harness, harness, { instance, thread, session, run });
        try { lock?.release(); }
        catch (e) { log(`run ${run}: workspace lock release: ${e.message}`); }
        children.delete(run);
        stopping.delete(run);
        spawnedAt.delete(run);
        landingOf.delete(run);
        refusedRelease.delete(run);
      }
    });
    record({ event: "launch", instance, ...(thread ? { thread, session } : {}), run, launchedBy, harness: model.harness, route, routeReason, model: model.model, provider: model.provider, modelKey: model.modelKey, account, pid: child.pid, runnerPid: entry.runnerPid, manifest });
    return { run, instance, harness: model.harness, model: model.model };
  }

  // Row 121: one tick of thread starts and idle stops, then (row 156) the
  // landing pass. A tick that is still running when the next is due skips it.
  // Row 194 (D6): while row 191's switch is on, only the lifecycle pass runs.
  async function tick() {
    if (ticking || closed) return;
    let halted = false;
    try { admission(); } catch { halted = true; }
    ticking = true;
    try {
      // A --no-launch host runs only the lifecycle pass, which closes each request launches-disabled.
      if (launches && !halted) {
        await threadPass();
        await landPass();
      }
      await requestPass(halted);
      await lifecyclePass();
    } finally {
      ticking = false;
    }
  }

  // Request recovery starts only the configured chief through the existing --pm path.
  // Workers always come back through that chief's ordinary launch tool/capability.
  async function requestPass(halted = false) {
    for (const [run, request] of launches && !halted ? requestDrains : []) {
      const session = registry.sessions.find(s => s.run === run);
      if (!session) { requestDrains.delete(run); continue; }
      try {
        const out = await host.op({ op: 'request.release', cap: request.cap, record: { business: b.id, run, instance: session.instance, project: request.project } });
        if (out?.released) { terminate([session]); requestDrains.delete(run); }
      } catch (e) {
        // A revoked chief capability cannot keep retiring workers.
        requestDrains.delete(run);
        await host.op({ op: 'request.launchResult', record: { business: b.id, instance: session.instance, code: e.code ?? 'launch-failed', run: null } });
      }
    }
    const pending = await host.op({ op: 'request.launches', business: b.id });
    if (!Array.isArray(pending) || !pending.length) return;
    if (!launches || halted) {
      await host.op({ op: 'request.launchResult', record: { business: b.id, instance: b.launch.by, code: !launches ? 'launches-disabled' : 'processing-halted', run: null } });
      return;
    }
    for (const i of pending) {
      const session = registry.running(b.id, i.target, null);
      if (!session || !i.project || stopping.has(session.run)) continue;
      let projects = [];
      try { projects = JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json'), 'utf8')).projects ?? []; } catch {}
      if (!projects.includes(i.project)) {
        const ready = await host.op({ op: 'request.admission', business: b.id, project: i.project });
        if (ready?.ready === true) await host.op({ op: 'request.manifest', record: { business: b.id, instance: i.target, project: i.project, run: session.run } });
      }
    }
    const chief = b.launch?.by;
    if (!chief || registry.running(b.id, chief, null) || pending.some(i => i.chief_run)) return;
    const next = pending.find(i => i.chief === chief && i.chief_attempt !== i.prerequisite);
    if (!next) return;
    try {
      await host.op({ op: 'request.chief', business: b.id });
      const out = await launchPm();
      await host.op({ op: 'request.launchResult', record: { business: b.id, instance: chief, code: null, run: out.run } });
    } catch (e) {
      await host.op({ op: 'request.launchResult', record: { business: b.id, instance: chief, code: e.code ?? 'launch-failed', run: null } });
    }
  }

  async function threadPass() {
    let threads;
    try {
      threads = await host.op({ op: "threadsState", business: b.id });
    } catch (e) {
      log(`threadsState refused: ${e.code ?? e.message}`);
      return;
    }
    const now = Date.now();
    const ms = (at) => (at ? Date.parse(at) : 0);
    // Row 269: a thread with a legacy session keeps its old store for that session only.
    const legacyThreads = new Set(threads.filter((t) => t.legacy).map((t) => t.thread));
    // Row 271 (264d): a session in a clear is the clear pass's, and nothing below starts or idles it.
    const clearing = await clearPass(now, legacyThreads);
    for (const t of threads) {
      if (closed) return;
      const key = `${t.thread}/${t.session}`;
      if (clearing.has(t.session)) continue;
      const s = registry.running(b.id, t.instance, t.thread, t.session);
      // FED R4 (#1801, design 3.7): the broker marks a thread whose user owner lost the grant, and has
      // ended its claim before it returns the row, whether the file changed or a grant expired. A
      // revoked thread never starts, and a running one gets the idle stop's start-time-checked
      // SIGTERM at once, with no threadIdle: there is no claim left to release.
      if (t.revoked) {
        if (s && !stopping.has(s.run)) {
          stopping.add(s.run);
          record({ event: "thread-revoked", instance: t.instance, thread: t.thread, session: t.session, run: s.run });
          try {
            if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
          } catch {}
        }
        continue;
      }
      if (s) {
        const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen));
        if (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) {
          let idle;
          try {
            idle = await host.op({ op: "threadIdle", record: { business: b.id, thread: t.thread, run: s.run } });
          } catch (e) {
            log(`run ${s.run}: threadIdle refused: ${e.code ?? e.message}`);
            continue;
          }
          // A turn in progress, or a message since the snapshot: the run keeps going.
          if (!idle.released) continue;
          stopping.add(s.run);
          record({ event: "thread-idle", instance: t.instance, thread: t.thread, session: t.session, run: s.run });
          try {
            if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
          } catch {}
        }
        continue;
      }
      const opened = ms(t.lastOpen) > ms(t.lastClaim);
      if (t.holder || !(t.pending > 0 || opened)) continue;
      // One start per message or open: a session that ended without taking
      // the work (it died before its claim, say) waits for the next one
      // rather than start again every tick. A refused start isn't a start.
      const signature = `${t.lastMessage}|${t.lastOpen}`;
      if (startedFor.get(key) === signature) continue;
      await serialize(async () => {
        if (closed || registry.running(b.id, t.instance, t.thread, t.session)) return;
        try {
          const where = { thread: t.thread, session: t.session };
          await start({ instance: t.instance, launchedBy: b.human, ...where, legacy: t.legacy, legacyThread: legacyThreads.has(t.thread), ...(await check(t.instance, where)) });
          startedFor.set(key, signature);
          refusedThreads.delete(key);
        } catch (e) {
          const code = e instanceof Refusal ? e.code : "launch-failed";
          // Logged once per code, not once per tick.
          if (refusedThreads.get(key) === code) return;
          refusedThreads.set(key, code);
          log(`thread ${t.thread} session ${t.session} (${t.instance}): start refused: ${code}`);
          record({ event: "refused", by: b.human, instance: t.instance, thread: t.thread, session: t.session, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
        }
      });
    }
  }

  // Row 271 (264d, #1776, D7 steps 3-5): one step of each clear the broker lists. The drain comes
  // from the run itself (its receive, or its end), so a requested clear waits here. A drained run
  // gets the idle stop's start-time-checked SIGTERM once, and 30 s without its exit proof records
  // uncertain once; nothing is signalled again or forced. The proof is the run's own close handler,
  // whose endRun the broker checks, or for no run, the session's workspace lock. Then retirement,
  // and the successor's start through the same check()/start() as any session. Returns the
  // sessions those steps own this tick.
  async function clearPass(now, legacyThreads) {
    const owned = new Set();
    let clears;
    try {
      clears = await host.op({ op: "sessionClears", business: b.id });
    } catch (e) {
      clearLog("sessionClears", e.code ?? e.message, `sessionClears refused: ${e.code ?? e.message}`);
      return owned;
    }
    for (const c of clears) {
      if (closed) break;
      const start = ["retired", "start-held", "start-refused"].includes(c.stage);
      owned.add(start ? c.successor : c.session);
      try {
        await (start ? clearStart(c, legacyThreads) : clearStop(c, now));
      } catch (e) {
        clearLog(c.op, e.code ?? e.message, `clear ${c.op} (session ${c.session}): ${e.code ?? e.message}`);
      }
    }
    return owned;
  }

  function clearLog(op, code, line) {
    if (clearLogged.get(op) === code) return;
    clearLogged.set(op, code);
    log(line);
  }

  async function clearStop(c, now) {
    const where = { business: b.id, thread: c.thread, op: c.op };
    let stage = c.stage;
    if (stage === "drained" || stage === "uncertain") {
      const s = registry.running(b.id, c.instance, c.thread, c.session);
      if (s && s.run === c.run) {
        if (!clearStops.has(c.op)) {
          // An uncertain clear was signalled before this launcher started, and isn't signalled again.
          if (stage === "uncertain") return void clearStops.set(c.op, now);
          // A cancel since the snapshot leaves the run alone; one after this read is a residual.
          const fresh = (await host.op({ op: "sessionClears", business: b.id })).find((x) => x.thread === c.thread && x.op === c.op);
          if (fresh?.stage !== "drained") return;
          clearStops.set(c.op, now);
          stopping.add(s.run);
          record({ event: "session-clear-stop", instance: c.instance, thread: c.thread, session: c.session, run: s.run, op: c.op });
          try {
            if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
          } catch {}
        } else if (stage === "drained" && now - clearStops.get(c.op) >= stopTimeoutMs) {
          await host.op({ op: "sessionClearUncertain", record: where });
          record({ event: "session-clear-uncertain", instance: c.instance, thread: c.thread, session: c.session, run: s.run, op: c.op });
        }
        return;
      }
      // Some other run of the session (one that never claimed) exits on its refused claim first.
      if (s) return;
      // An earlier host's run that outlived recover()'s kill was ended host-lost, but it still runs:
      // its exit, not that end, is the proof (row 269 Q2). It isn't signalled.
      if (survivors.some((x) => x.business === b.id && x.run === c.run && liveRun(x))) return;
      if (c.run === null) {
        // No run: the proof is the session's workspace lock, which a spawned run holds until its close.
        const live = registry.sessions.filter((x) => x.business === b.id && x.workspace).map((x) => x.workspace);
        const workspace = sessionWorkspace(dataRoot, b.id, c.instance, c.session, { live });
        lockSessionWorkspace(workspace, { hostPid: process.pid, hostStartTime: startTimeOf(process.pid), business: b.id, instance: c.instance, thread: c.thread, session: c.session }).release();
      }
      await host.op({ op: "sessionClearExited", record: { ...where, run: c.run } });
      stage = "exited";
    }
    if (stage !== "exited") return;
    const out = await host.op({ op: "sessionClearRetire", record: where });
    clearStops.delete(c.op);
    record({ event: out.state === "retired" ? "session-clear-retired" : "session-clear-failed", instance: c.instance, thread: c.thread, session: c.session, successor: c.successor, op: c.op, ...(out.refusal ? { code: out.refusal } : {}) });
  }

  // D8: capacity and limits hold a start, which spends no work attempt. Any other refusal is one.
  const CLEAR_HELD = new Set(["capacity-full", "limit-reached"]);

  async function clearStart(c, legacyThreads) {
    const where = { business: b.id, thread: c.thread, op: c.op };
    await serialize(async () => {
      if (closed) return;
      if (!registry.running(b.id, c.instance, c.thread, c.successor)) {
        const session = { thread: c.thread, session: c.successor };
        try {
          await start({ instance: c.instance, launchedBy: b.human, ...session, legacy: false, legacyThread: legacyThreads.has(c.thread), ...(await check(c.instance, session)) });
        } catch (e) {
          const code = e instanceof Refusal ? e.code : "launch-failed";
          // Reported and logged once per code, not once per tick.
          if (clearLogged.get(c.op) === code) return;
          clearLogged.set(c.op, code);
          log(`clear ${c.op}: successor ${c.successor} (${c.instance}): start ${CLEAR_HELD.has(code) ? "held" : "refused"}: ${code}`);
          record({ event: "refused", by: b.human, instance: c.instance, thread: c.thread, session: c.successor, code, why: e.message, ...(e.limit ? { limit: e.limit } : {}) });
          await host.op({ op: "sessionClearStart", record: { ...where, start: CLEAR_HELD.has(code) ? "held" : "refused", refusal: code } });
          return;
        }
      }
      clearLogged.delete(c.op);
      await host.op({ op: "sessionClearStart", record: { ...where, start: "started" } });
    });
  }

  // Row 156 (#1651): start each listed merge decision holder once, and stop a
  // running holder that started before its decision's resolution.
  async function landPass() {
    if (!b.launch) return;
    let holders;
    try {
      holders = await host.op({ op: "landHolders", business: b.id });
      if (!Array.isArray(holders)) throw new Refusal("malformed-answer");
      refusedHolders = null;
    } catch (e) {
      const code = e.code ?? e.message;
      if (refusedHolders !== code) log(`landHolders refused: ${code}`);
      refusedHolders = code;
      return;
    }
    for (const h of holders) {
      if (closed) return;
      if (!b.launch.instances.includes(h.instance)) continue;
      const s = registry.running(b.id, h.instance, null);
      if (s) {
        if (stopping.has(s.run) || landingOf.get(s.run)?.has(h.decision)) continue;
        let r;
        try {
          r = await host.op({ op: "landRelease", record: { business: b.id, run: s.run, decision: h.decision } });
        } catch (e) {
          const code = e.code ?? e.message;
          if (refusedRelease.get(s.run) !== code) log(`run ${s.run}: landRelease refused: ${code}`);
          refusedRelease.set(s.run, code);
          continue;
        }
        // A landing turn or a message turn in progress: a later tick.
        if (!r?.released) continue;
        stopping.add(s.run);
        record({ event: "land-relaunch", instance: h.instance, run: s.run, decision: h.decision });
        try {
          if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
        } catch {}
        continue;
      }
      if (landStarted.has(h.decision)) continue;
      await serialize(async () => {
        if (closed || registry.running(b.id, h.instance, null)) return;
        try {
          await start({ instance: h.instance, launchedBy: b.human, ...(await check(h.instance)) });
          refusedLand.delete(h.decision);
        } catch (e) {
          const code = e instanceof Refusal ? e.code : "launch-failed";
          if (refusedLand.get(h.decision) === code) return;
          refusedLand.set(h.decision, code);
          log(`decision ${h.decision} (${h.instance}): landing start refused: ${code}`);
          record({ event: "refused", by: b.human, instance: h.instance, decision: h.decision, code, why: e.message, ...(e.limit ? { limit: e.limit } : {}) });
        }
      });
    }
  }

  // Row 194 (#1692, D1.1): the human's and the tester's lifecycle requests, oldest first.
  async function lifecyclePass() {
    let requests;
    try {
      requests = await host.op({ op: "lifecycleRequests", business: b.id });
      if (!Array.isArray(requests)) throw new Refusal("malformed-answer");
      refusedRequests = null;
    } catch (e) {
      const code = e.code ?? e.message;
      if (refusedRequests !== code) log(`lifecycleRequests refused: ${code}`);
      refusedRequests = code;
      return;
    }
    for (const r of requests) {
      if (closed) return;
      // A taken request is past this pass: it ran, or its done failed.
      if (r.state !== "open") continue;
      await serialize(() => carry(r));
    }
  }

  // One request: the take, the act, the done. Runs under serialize, so no start() runs alongside it.
  async function carry(r) {
    if (closed) return;
    let take;
    try {
      take = await host.op({ op: "lifecycleTake", record: { business: b.id, request: r.request } });
    } catch (e) {
      log(`request ${r.request} (${r.instance}): lifecycleTake refused: ${e.code ?? e.message}`);
      return;
    }
    if (take?.go !== true) {
      if (take?.refusal && take.refusal !== "request-closed") record({ event: "refused", by: r.by, instance: r.instance, request: r.request, code: take.refusal });
      return;
    }
    let done;
    try {
      done = await act(take);
    } catch (e) {
      done = { refusal: e instanceof Refusal ? e.code : "launch-failed", why: e.message, limit: e.limit };
    }
    const { why, limit, ...closing } = done;
    if (closing.refusal) record({ event: "refused", by: r.by, instance: r.instance, request: r.request, code: closing.refusal, ...(why ? { why } : {}), ...(limit ? { limit } : {}) });
    try {
      await host.op({ op: "lifecycleDone", record: { business: b.id, request: r.request, ...closing } });
    } catch (e) {
      log(`request ${r.request} (${r.instance}): lifecycleDone refused: ${e.code ?? e.message}${closing.run ? `; run ${closing.run} started` : ""}`);
    }
  }

  // What the request did: { run } for a start, { transaction } for a delete, {} for an end, or { refusal }.
  async function act(r) {
    // Row 163: a --no-launch host starts, signals and renames nothing.
    if (!launches) return { refusal: "launches-disabled" };
    const launchedBy = r.by === "tester" ? "tester" : b.human;
    if (r.action === "lifecycle.launch") return launch(r.instance, launchedBy);
    if (r.action === "lifecycle.end") {
      const ends = ending(r);
      if (r.kill !== true) {
        terminate(ends);
        return {};
      }
      return { stops: await fireStop(r, ends) };
    }
    if (r.action === "lifecycle.regenerate") {
      terminate(registry.sessions.filter((s) => s.business === b.id && s.instance === r.instance));
      const deadline = Date.parse(r.at) + lifecycleRequestMs;
      while (registry.sessions.some((s) => s.business === b.id && s.instance === r.instance)) {
        if (closed) return { refusal: "host-closing" };
        if (Date.now() >= deadline) return { refusal: "regenerate-timeout" };
        await new Promise((resolve) => setTimeout(resolve, 50));
      }
      return launch(r.instance, launchedBy);
    }
    if (r.action === "lifecycle.delete") return remove(r);
    return { refusal: "invalid-request" };
  }

  async function launch(instance, launchedBy) {
    const pm = instance === b.launch?.by;
    const { run } = await start({ instance, launchedBy, human: true, ...(await check(instance, { pm, human: true })) });
    return { run };
  }

  // The registry runs an end stops (N6, DESIGN.md:118-120): the runs its verb named, and the runs of
  // the instance spawned by the verb's time that the verb didn't name, which hadn't bound then. A
  // run spawned after the verb is the next end's. A thread instance's end names only the runs it
  // revoked (the tester's own). A run this host didn't spawn counts from its registry start.
  function ending(r) {
    const runs = r.runs ?? [];
    const at = Date.parse(r.at);
    const threaded = THREAD_DEFINITIONS.includes(b.roles[r.instance]?.definition);
    return registry.sessions.filter((s) => s.business === b.id && (runs.includes(s.run) ||
      (!threaded && s.instance === r.instance && (spawnedAt.get(s.run) ?? Date.parse(s.startedAt)) <= at)));
  }

  // SIGTERM, after the start-time check, to each run. No SIGKILL: only a fire escalates.
  function terminate(sessions) {
    for (const s of sessions) {
      stopping.add(s.run);
      record({ event: "lifecycle-end", instance: s.instance, run: s.run });
      try {
        if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
      } catch {}
    }
  }

  // D2.2 (S6): a fire's end stops each run through row 191's stopRun, owned by the end request (fire
  // takes no decision), so a halt sweep that meets the run joins this stop. stopRun checks the
  // identity (start ticks and boot id) before each signal and escalates to SIGKILL after its
  // grace. The runs it killed or left alive go back to the broker as lifecycle.end.killed and
  // lifecycle.end.alive.
  async function fireStop(r, sessions) {
    const results = await Promise.all(sessions.map(async (s) => {
      stopping.add(s.run);
      record({ event: "lifecycle-end", instance: s.instance, run: s.run, kill: true });
      let out;
      try {
        out = await stopRun({ root: dataRoot, business: b.id, run: s.run, reason: "fire", owner: { decision: r.request } }, stopOptions);
      } catch (e) {
        log(`run ${s.run}: fire stop failed: ${e.message}`);
        out = { outcome: "unknown", pid: null, signals: [] };
      }
      record({ event: "fire-stop", instance: s.instance, run: s.run, request: r.request, outcome: out.outcome, pid: out.pid, signals: out.signals });
      return { run: s.run, pid: out.pid, outcome: out.outcome };
    }));
    // sage-2289: a stop that can't confirm the kill (unknown) reads as alive with reason unknown,
    // never as killed, so the Console flags it.
    return results.flatMap((x) => x.outcome === "unknown" ? [{ ...x, outcome: "alive", reason: "unknown" }]
      : x.outcome === "killed" || x.outcome === "alive" ? [x] : []);
  }

  // V5 steps 1 to 3. The file goes first: once committed, check() refuses the name, so no launch
  // opens the old workspace even if the rename fails.
  async function remove(r) {
    if (registry.sessions.some((s) => s.business === b.id && s.instance === r.instance)) return { refusal: "instance-running" };
    let out;
    try {
      out = await host.op({ op: "lifecycleDelete", record: { business: b.id, request: r.request } });
    } catch (e) {
      return { refusal: e.code ?? "outcome-unknown" };
    }
    const transaction = out?.transaction ?? null;
    if (transaction === null) return { refusal: "workspace-rename-failed" };
    const bot = out.bot ? { bot: out.bot } : {};
    // The old instance workspace, then (row 269) the instance's session workspaces, each moved
    // whole. A retry moves whichever is still in place.
    for (const from of [workspaceDir(dataRoot, b.id, r.instance), join(dataRoot, "session-workspaces", b.id, r.instance)]) {
      if (!existsSync(from)) continue;
      const to = join(dirname(from), ".retired", `${r.instance}-${transaction}`);
      try {
        mkdirSync(dirname(to), { recursive: true, mode: 0o700 });
        chmodSync(dirname(to), 0o700);
        if (existsSync(to)) throw new Error(`${to} exists`);
        renameSync(from, to);
        chmodSync(to, 0o700);
      } catch (e) {
        log(`request ${r.request} (${r.instance}): workspace rename failed: ${e.message}; left at ${from}`);
        return { transaction, ...bot, refusal: "workspace-rename-failed", why: `${e.message}; left at ${from}` };
      }
      record({ event: "workspace-retired", instance: r.instance, request: r.request, transaction, path: to });
    }
    return { transaction, ...bot };
  }

  // endLaunch's answer ({ released, limit? }), or null when it refused.
  async function endRun(run, reason, exitCode) {
    try {
      return await host.op({ op: "endLaunch", record: { business: b.id, run, reason, exitCode } });
    } catch (e) {
      log(`run ${run}: endLaunch refused: ${e.code ?? e.message}`);
      return null;
    }
  }

  // A launch a session asked for. Resolves to the reply line's object.
  function fromSession(cap, instance) {
    return serialize(async () => {
      if (closed) return { ok: false, error: "host-closing" };
      let who;
      try {
        who = await host.op({ op: "identity", cap });
      } catch (e) {
        return { ok: false, error: e.code ?? "unauthenticated" };
      }
      if (!who.role || who.business !== b.id) {
        await host.op({ op: "refuse", cap, code: "not-a-role" }).catch(() => {});
        return { ok: false, error: "not-a-role" };
      }
      if (!launches) {
        await host.op({ op: "refuse", cap, code: "launches-disabled" }).catch(() => {});
        record({ event: "refused", by: who.role, instance, code: "launches-disabled" });
        return { ok: false, error: "launches-disabled" };
      }
      const reply = async out => {
        if (typeof instance === 'string') await host.op({ op: 'request.launchResult', record: { business: b.id, instance, code: out.ok ? null : out.error, run: out.ok ? out.result.run : null } });
        return out;
      };
      try { b = loadBusiness(host.business, { dir: configDir(env), rolesDir: rolesDir(env) }); }
      catch { return reply({ ok: false, error: 'organization-refused' }); }
      // Product enums and clones are frozen in a session's manifest. A legitimate chief
      // launch may request a drain, never interrupt the turn or start a privileged worker.
      const running = registry.running(b.id, instance, null);
      if (running && who.role === b.launch?.by) {
        const pending = await host.op({ op: 'request.launches', business: b.id });
        const needed = pending.filter(i => i.target === instance && i.project);
        let projects = [];
        try { projects = JSON.parse(readFileSync(join(running.runDir, 'bundle/manifest.json'), 'utf8')).projects ?? []; } catch {}
        const missing = needed.find(i => !projects.includes(i.project));
        if (missing) {
          if (!b.launch?.instances.includes(instance)) return reply({ ok: false, error: 'instance-not-listed' });
          try {
            await host.op({ op: 'authorizeLaunch', cap, instance });
            const ready = await host.op({ op: 'request.admission', business: b.id, project: missing.project });
            if (ready?.ready !== true) return reply({ ok: false, error: ready?.code ?? 'project-pending' });
            const out = await host.op({ op: 'request.release', cap, record: { business: b.id, run: running.run, instance, project: missing.project } });
            if (out?.released) terminate([running]);
            else requestDrains.set(running.run, { cap, project: missing.project });
            return reply({ ok: false, error: 'request-manifest-draining' });
          } catch (e) { return reply({ ok: false, error: e.code ?? 'launch-failed' }); }
        }
      }
      let checked;
      try {
        if (typeof instance !== "string") throw new Refusal("invalid-request");
        checked = await check(instance);
      } catch (e) {
        if (!(e instanceof Refusal)) throw e;
        await host.op({ op: "refuse", cap, code: e.code }).catch(() => {});
        record({ event: "refused", by: who.role, instance, code: e.code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
        return reply({ ok: false, error: e.code });
      }
      try {
        await host.op({ op: "authorizeLaunch", cap, instance });
      } catch (e) {
        record({ event: "refused", by: who.role, instance, code: e.code ?? "not-authorized" });
        return reply({ ok: false, error: e.code ?? "not-authorized" });
      }
      try {
        return reply({ ok: true, result: await start({ instance, launchedBy: who.role, ...checked }) });
      } catch (e) {
        const code = e instanceof Refusal ? e.code : "launch-failed";
        log(`launch of ${instance} for ${who.role} failed: ${e.message}`);
        await host.op({ op: "refuse", cap, code }).catch(() => {});
        record({ event: "refused", by: who.role, instance, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
        return reply({ ok: false, error: code });
      }
    });
  }

  // `bus start --pm`: the business's launcher instance, launched by the human.
  function launchPm() {
    return serialize(async () => {
      if (!b.launch) throw new CliError(`business ${b.id} has no launch block, so --pm has nothing to launch`, 3);
      try {
        return await start({ instance: b.launch.by, launchedBy: b.human, ...(await check(b.launch.by, { pm: true })) });
      } catch (e) {
        if (!(e instanceof Refusal)) throw e;
        record({ event: "refused", by: b.human, instance: b.launch.by, code: e.code, why: e.message, ...(e.limit?{limit:e.limit}:{}) });
        const error = new CliError(`PM launch refused: ${e.code}${e.message !== e.code ? ` (${e.message})` : ""}`, 3);
        error.code = e.code;
        throw error;
      }
    });
  }

  // The earlier host's sessions. Their capabilities died with its broker, so
  // one that still runs can do nothing: kill it (the namespace takes its
  // turn with it). Then end each run in the bus, which releases its role.
  // This broker never bound those runs, so each is bound again first with
  // its identical launch record, which the broker accepts.
  async function recover() {
    // Row 163 (mos-0961): a --no-launch host never signals or ends an earlier host's session. Its
    // data root may hold a sessions.json that names live processes, so it only says what it left.
    if (!launches) {
      for (const s of leftover) log(`run ${s.run} (${s.instance}) from an earlier host: left as it is, this host was started with --no-launch`);
      leftover = [];
      return;
    }
    for (const s of leftover) {
      if (startTimeOf(s.pid) === s.startTime && (cmdlineOf(s.pid) ?? [])[0] === bwrap) {
        try {
          process.kill(s.pid, "SIGKILL");
        } catch {}
        const end = Date.now() + 5000;
        while (liveRun(s) && Date.now() < end) await new Promise((r) => setTimeout(r, 20));
      }
      // Row 269 (Q2): the session or its runner outlived the kill, or wasn't a session to kill.
      if (liveRun(s)) {
        survivors.push(s);
        log(`run ${s.run} (${s.instance}) from an earlier host still runs (pid ${s.pid}, runner ${s.runnerPid}): no session of ${s.instance} starts until it exits`);
      }
      try {
        await host.op({ op: 'recoverLaunch', record: { business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime, ...(s.thread ? { thread: s.thread } : {}), ...(s.session ? { session: s.session } : {}) } });
      } catch (e) {
        // A host that died after ending the run but before updating sessions.json.
        if (e.code === "run-ended") log(`run ${s.run} (${s.instance}) from an earlier host was already ended`);
        else log(`run ${s.run} (${s.instance}) from an earlier host: not ended, historical cleanup refused: ${e.code ?? e.message}`);
        continue;
      }
      record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), ...(s.session ? { session: s.session } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null });
      log(`run ${s.run} (${s.instance}) from an earlier host ended: host-lost`);
    }
    leftover = [];
  }

  async function listen() {
    await recover();
    const path = launchSocketPath(dataRoot);
    if (existsSync(path)) {
      if (!lstatSync(path).isSocket()) throw new CliError(`${path} exists and isn't a socket`, 3);
      rmSync(path);
    }
    server = createServer((socket) => {
      sockets.add(socket);
      socket.once("close", () => sockets.delete(socket));
      let buf = "";
      let answered = false;
      const reply = (obj) => {
        if (answered) return;
        answered = true;
        socket.end(`${JSON.stringify(obj)}\n`);
      };
      socket.setTimeout(10000, () => reply({ ok: false, error: "invalid-request" }));
      socket.on("error", () => {});
      socket.on("data", (chunk) => {
        if (answered) return;
        buf += chunk;
        if (buf.length > LINE_MAX) return reply({ ok: false, error: "invalid-request" });
        const nl = buf.indexOf("\n");
        if (nl < 0) return;
        socket.setTimeout(0);
        let req;
        try {
          req = JSON.parse(buf.slice(0, nl));
        } catch {
          return reply({ ok: false, error: "invalid-request" });
        }
        if (typeof req?.cap !== "string" || !/^[0-9a-f]{64}$/.test(req.cap)) return reply({ ok: false, error: "unauthenticated" });
        fromSession(req.cap, req.instance).then(reply, (e) => {
          log(`launch socket: ${e.message}`);
          reply({ ok: false, error: "launch-failed" });
        });
      });
    });
    // bus-host/ is 0700, so the moment before the chmod exposes nothing.
    await new Promise((resolve, reject) => {
      server.once("error", reject);
      server.listen(path, resolve);
    });
    chmodSync(path, 0o600);
    await gitSocket.listen();
    if (b.launch) ticker = setInterval(() => tick().catch((e) => log(`thread tick: ${e.message}`)), threadTickMs);
  }

  // SIGTERM every runner, wait, then end what's left. Runs before the host
  // stops the broker, so endLaunch still works.
  async function close() {
    closed = true;
    if (ticker) clearInterval(ticker);
    if (server) {
      // server.close waits for every connection, and a client that never
      // ends its side (or never sends) would hold it; so they go first.
      const stopped = new Promise((r) => server.close(r));
      for (const socket of sockets) socket.destroy();
      await stopped;
      rmSync(launchSocketPath(dataRoot), { force: true });
    }
    await gitSocket.close();
    await chain;
    // Again every second: a runner still starting has no handler yet, and as
    // pid 1 of its namespace it never sees a signal sent before it has one.
    const term = () => {
      for (const s of registry.sessions) {
        try {
          if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
        } catch {}
      }
    };
    term();
    const again = setInterval(term, 1000);
    const pending = [...children.values()];
    const timer = setTimeout(() => {
      for (const s of registry.sessions) {
        try {
          if (startTimeOf(s.pid) === s.startTime) process.kill(s.pid, "SIGKILL");
        } catch {}
      }
    }, stopTimeoutMs);
    await Promise.all(pending);
    // The exit handlers run on the same tick as the close events; let them finish.
    while (children.size) await new Promise((r) => setTimeout(r, 20));
    clearInterval(again);
    clearTimeout(timer);
  }

  host.beforeClose(close);
  return { business: b, registry, listen, launchPm, fromSession, tick, close };
}