import { FAR_DATE, launchDoc } from "../../organization/tests/helpers.mjs";
// The S6 launcher inside a real bus host: real broker child, real runners
// in the bwrap sandbox, the fake adapter from packages/harness standing in for the
// harness. The chief of staff launches a coder through its `launch` tool (row 119: the
// PM no longer launches, REQ-LAUNCH-1); every host
// refusal comes back over the launch socket and lands in the launch log.

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { createHash } from "node:crypto";
import { once } from "node:events";
import { createInterface } from "node:readline";
import { connect } from "node:net";
import { chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { Client } from "../../bus/src/client.mjs";
import { gitSocketPath, launchLogFile, launchSocketPath, readSessions, Registry, runsDir, sessionsFile, workspaceDir } from "../../seat/src/session.mjs";
import { BWRAP, sandboxProblem } from "../../seat/src/sandbox.mjs";
import { childrenOf } from "../../seat/src/proc.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { bootConfig, loadSystem } from "../src/config.mjs";
import { cmdlineOf, startHost, startTimeOf } from "../src/host.mjs";
import { createLauncher } from "../src/launcher.mjs";
import { mockAnthropic, REPO } from "../../harness/tests/helpers.mjs";
import { fakeHarnesses, fixture, HOST_CREDENTIALS, tmp, withoutHostCredentials } from "./helpers.mjs";

const FAKE = fileURLToPath(new URL("../../harness/tests/fixtures/fake-adapter.mjs", import.meta.url));
const HOST = fileURLToPath(new URL("./fixtures/launch-host.mjs", import.meta.url));
const skip = sandboxProblem() ?? false;

test('C2-04: launcher workspace clones code but never the library', { skip, timeout: 30000 }, async t => {
  const { mappingEntry } = await import('../../library/tests/helpers.mjs');
  const f = await setup(t, doc => { doc.roles.chief.credentials = { gitea: { ...doc.roles.coder.credentials.gitea } }; doc.projects = { books: mappingEntry(), ...doc.projects }; return doc; });
  await f.launcher.launchPm();
  const workspace = workspaceDir(f.f.dataRoot, 'acme', 'chief');
  assert.equal(existsSync(join(workspace, 'stack', '.git')), true);
  assert.equal(existsSync(join(workspace, 'books')), false);
  assert.equal(await f.close(), 0);
});

// chief opus, coder and reviewer sonnet, one of each family; cto is listed but
// runs the system model, which is in no family. The chief of staff is the launcher.
function edit(doc) {
  doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5", thinking: "high" };
  doc.roles.coder.vars.model = "claude-sonnet-5-5";
  doc.roles.reviewer.vars = { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" };
  doc.launch = launchDoc("chief",["coder","reviewer","cto"],{opus:1,sonnet:1});
  doc.launch.limits.models.zai["glm-5.3-flash"]=0;
  return doc;
}

async function until(fn, ms = 15000) {
  const end = Date.now() + ms;
  while (Date.now() < end) {
    const v = await fn();
    if (v) return v;
    await new Promise((r) => setTimeout(r, 50));
  }
  throw new Error("timed out waiting");
}

// git in a test repository, with nothing of the shell's git config.
function gitIn(cwd, ...args) {
  const r = spawnSync("git", ["-c", "user.name=test", "-c", "user.email=test@example.invalid", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
  assert.equal(r.status, 0, r.stderr);
  return r.stdout.trim();
}
function commit(dir, text) {
  writeFileSync(join(dir, "file.txt"), `${text}\n`);
  gitIn(dir, "add", "file.txt");
  gitIn(dir, "commit", "-q", "-m", text);
  return gitIn(dir, "rev-parse", "HEAD");
}

// A config and data root with the fake adapter; setup() starts a host on it.
// The session credential is a stand-in, and the fake adapter never uses it.
// Every credential variable of the shell running the tests is dropped, pi's own provider keys
// included (row 332 R2); a pi chief gets one made-up pi key instead.
function prepare(t, more = (doc) => doc) {
  const root = tmp(t, "mosaic-launch-");
  const f = fixture(root, "acme", (doc) => more(edit(doc)));
  withoutHostCredentials(f.env, { piKey: f.doc.roles.chief.vars.harness === "pi" || f.doc.roles.chief.vars["fallback.harness"] === "pi" });
  f.env.CLAUDE_CODE_OAUTH_TOKEN = "stand-in";
  // The project root is a repository on main, so each launch can clone it.
  const project = f.doc.projects.stack.root;
  gitIn(project, "init", "-q", "-b", "main");
  commit(project, "one");
  const adapter = join(root, "adapter.sh");
  writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`, { mode: 0o644 });
  return { root, f, adapter };
}

// options.wrap(host) is what the launcher sees, so a test can change one of the host's answers.
// options.harnesses is fakeHarnesses' map (row 65): both modules run the fake adapter by default.
async function setup(t, more = (doc) => doc, { root, f, adapter } = prepare(t, more), { wrap = (h) => h, harnesses = { pi: adapter, "claude-code": adapter }, ...options } = {}) {
  const makeLoader = await fakeHarnesses(t, harnesses);
  const system = loadSystem({ env: f.env });
  const boot = bootConfig({ system, businessId: "acme", env: f.env });
  const logs = [];
  const host = await startHost({ boot, business: "acme", log: (l) => logs.push(l) });
  let launcher;
  try {
    launcher = createLauncher({
      host: wrap(host),
      system,
      env: f.env,
      makeLoader,
      sessionDefaults: { pollInterval: 100 },
      harnessVersion: "0.85.1",
      readOnly: [FAKE],
      log: (l) => logs.push(l),
      ...options,
    });
    await launcher.listen();
  } catch (e) {
    await host.close(1);
    throw e;
  }
  let closed = false;
  const close = async () => {
    if (closed) return host.done;
    closed = true;
    return host.close(0);
  };
  t.after(close);
  // A test caller: a cto launch bound to this process, claimed.
  const cto = await host.bindLaunch({ business: "acme", role: "cto", run: "cto-test", harness: "claude-code", pid: process.pid, startTime: startTimeOf(process.pid) });
  const client = new Client({ path: host.path, cap: cto.cap });
  await client.call("role.claim");
  return { root, f, host, launcher, cto, client, logs, close };
}

// One request on the launch socket, one reply line.
function ask(path, line) {
  return new Promise((resolve, reject) => {
    const s = connect(path);
    let buf = "";
    s.on("connect", () => s.write(line));
    s.on("data", (b) => (buf += b));
    s.on("end", () => resolve(JSON.parse(buf)));
    s.on("error", reject);
  });
}

const log = (f) => readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));

// Named mutant: proxy key serialized into session.json. Keep the real launcher,
// harness env generation, session writer and runner; only the model adapter is fake.
test('P3 proxy key reaches the spawned runner but never session.json or the bundle', { skip, timeout: 30000 }, async t => {
  const proxy = { baseUrl: 'http://127.0.0.1:12345', key: 'fx332-agent-a' };
  const prepared = prepare(t, doc => {
    doc.launch.models['claude-opus-5-5'] = { provider: 'proxy', model: 'claude-fixture', route: 'anthropic' };
    return doc;
  });
  const { launcher, f, close } = await setup(t, doc => doc, prepared, { testProxy: proxy });
  const launched = await launcher.launchPm();
  const session = readSessions(f.dataRoot).find(s => s.run === launched.run);
  assert.ok(session);
  const environ = readFileSync(`/proc/${session.runnerPid}/environ`, 'utf8').split('\0');
  assert.ok(environ.includes(`ANTHROPIC_API_KEY=${proxy.key}`));
  assert.ok(!environ.some(value => value.startsWith('CLAUDE_CODE_OAUTH_TOKEN=')));
  const stored = readFileSync(join(session.runDir, 'session.json'), 'utf8');
  assert.ok(!stored.includes(proxy.key), 'session.json must never persist the proxy key');
  const bundleFiles = readdirSync(join(session.runDir, 'bundle'), { recursive: true, withFileTypes: true })
    .filter(entry => entry.isFile()).map(entry => join(entry.parentPath, entry.name));
  for (const file of bundleFiles) assert.ok(!readFileSync(file, 'utf8').includes(proxy.key), file);
  assert.equal(await close(), 0);
});

// #774 F4: testProxy skips P2 admission, so createLauncher refuses it next to system.proxy, in both
// forms, before it reads anything else.
test('#774 F4 createLauncher refuses testProxy together with system.proxy', async t => {
  const root = tmp(t);
  const system = { dataRoot: join(root, 'data'), proxy: { port: 8317, binary: '/fixture-unused', firstOutputSeconds: 1 } };
  const host = { business: 'acme', op: async () => { throw new Error('no host operation expected'); } };
  for (const testProxy of [{ baseUrl: 'http://127.0.0.1:12345', key: 'fx774-static' }, async () => ({ baseUrl: 'http://127.0.0.1:12345', key: 'fx774-factory' })])
    assert.throws(() => createLauncher({ host, system, env: { MOSAIC_CONFIG: join(root, 'config', 'config.json'), MOSAIC_ROLES_DIR: join(root, 'roles') }, testProxy, log: () => {} }),
      (e) => e.code === 'test-proxy-refused' && /system\.proxy/.test(e.message) && !e.message.includes('fx774'));
  assert.equal(existsSync(join(root, 'data')), false, 'refused before any data-root work');
});

test("bus start --pm: the launcher instance (the chief of staff) runs under its own PID namespace, registered, with a manifest", { skip, timeout: 30000 }, async (t) => {
  // Row 122: uncertain write receipts cannot prevent the window's first chief launch.
  const prepared = prepare(t);
  const history = join(prepared.root, "config", "businesses", ".history", "acme");
  mkdirSync(history, { recursive: true, mode: 0o700 });
  writeFileSync(join(history, "changes.jsonl"), JSON.stringify({
    state: "prepared", transaction: "fixture-uncertain", before: "a".repeat(64), after: "b".repeat(64),
  }) + "\n", { mode: 0o600 });
  const { f, launcher, host, cto, close } = await setup(t, doc => doc, prepared);
  const sock = launchSocketPath(f.dataRoot);
  assert.equal(statSync(sock).mode & 0o777, 0o600);

  const chief = await launcher.launchPm();
  assert.deepEqual({ ...chief, run: undefined }, { run: undefined, instance: "chief", harness: "pi", model: "claude-opus-5-5" });
  assert.match(chief.run, /^r[0-9a-f]{12}$/);

  const [s] = readSessions(f.dataRoot);
  assert.equal(s.run, chief.run);
  assert.equal(s.live, true);
  assert.equal(s.provider, "zai");
  assert.equal(s.launchedBy, "jason");
  assert.notEqual(s.pid, s.runnerPid, "the runner is the bwrap process's child");
  assert.equal(statSync(sessionsFile(f.dataRoot)).mode & 0o777, 0o600);

  const runDir = s.runDir;
  assert.equal(statSync(runDir).mode & 0o777, 0o700);
  const session = JSON.parse(readFileSync(join(runDir, "session.json"), "utf8"));
  assert.equal(session.launchedBy, "jason");
  assert.equal(session.brokerSocket, host.path);
  assert.equal(session.launchSocket, sock);
  // Row 93: the chief of staff neither pushes nor reviews, so no git socket; the launcher serves one.
  assert.equal(session.gitSocket, null);
  assert.equal(statSync(gitSocketPath(f.dataRoot)).mode & 0o777, 0o600);
  const manifest = JSON.parse(readFileSync(join(runDir, "bundle", "manifest.json"), "utf8"));
  // Row 65: one version, the route's harness, and the route the launch took.
  assert.equal(manifest.harnessVersion, "0.85.1");
  assert.deepEqual([manifest.route, manifest.routeReason], ["primary", null]);
  assert.ok(!("piVersion" in manifest) && !("claudeVersion" in manifest));
  const [launched] = log(f).filter((e) => e.event === "launch");
  assert.equal(launched.run, chief.run);
  assert.equal(launched.manifest, join(runDir, "bundle", "manifest.json"));
  // Row 126: a pi launch spends no account, so the record says null.
  assert.equal(launched.account, null);
  assert.equal(manifest.model, "claude-opus-5-5");

  // A second PM is refused before anything spawns.
  await assert.rejects(launcher.launchPm(), /PM launch refused: instance-running/);

  // The session's environment carries the run id and no capability; the
  // capability is nowhere on disk or on any session process's command line.
  await until(() => existsSync(join(runDir, "runner.log")) && readSessions(f.dataRoot)[0]?.live);
  const proc = (pid, what) => {
    try {
      return readFileSync(`/proc/${pid}/${what}`, "utf8");
    } catch {
      return "";
    }
  };
  const environ = proc(s.runnerPid, "environ").split("\0");
  assert.ok(environ.includes(`MOSAIC_RUN_ID=${chief.run}`));
  assert.ok(!environ.some((v) => v.startsWith("MOSAIC_CONFIG=")), "only allowlisted variables pass");
  const files = (dir) => readdirSync(dir, { recursive: true, withFileTypes: true }).filter((d) => d.isFile()).map((d) => join(d.parentPath, d.name));
  const capOf = cto.cap; // a known capability shape, to prove the scan finds nothing like it
  for (const file of files(f.dataRoot)) {
    const text = readFileSync(file, "latin1");
    assert.ok(!text.includes(capOf), `${file} holds no capability`);
  }
  for (const pid of [s.pid, s.runnerPid]) {
    assert.notEqual(proc(pid, "cmdline"), "");
    assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "cmdline")), "no capability in argv");
    assert.ok(!/[0-9a-f]{64}/.test(proc(pid, "environ")), "no capability in the environment");
  }

  // Closing the host stops the session and ends the launch.
  assert.equal(await close(), 0);
  assert.equal(startTimeOf(s.runnerPid), null);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.equal(existsSync(sock), false);
  const end = log(f).find((e) => e.event === "end");
  assert.equal(end.run, chief.run);
  assert.equal(end.reason, "stopped");
  assert.equal(end.exitCode, 0);
});

test("the chief of staff launches a coder through its launch tool; the coder answers; refusals name their code", { skip, timeout: 30000 }, async (t) => {
  const { f, launcher, client, cto } = await setup(t);
  const sock = launchSocketPath(f.dataRoot);
  const chief = await launcher.launchPm();

  // cto asks chief to launch a coder; chief's turn calls the launch tool.
  const sent = await client.call("message.send", { to: "chief", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
  const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
  assert.equal(reply.from_role, "chief");
  assert.equal(reply.class, "RESULT");
  const result = JSON.parse(reply.body);
  assert.equal(result.instance, "coder");
  assert.equal(result.model, "claude-sonnet-5-5");
  const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
  assert.equal(coder.run, result.run);
  assert.equal(coder.launchedBy, "chief");
  assert.equal(coder.live, true);
  // Only the role that may launch has the launch socket in its sandbox.
  const bound = (pid) => readFileSync(`/proc/${pid}/cmdline`, "utf8").split("\0").includes(sock);
  assert.equal(bound(readSessions(f.dataRoot).find((s) => s.instance === "chief").pid), true);
  assert.equal(bound(coder.pid), false);
  // Row 93: and only a role that pushes or reviews has the git socket.
  const gitBound = (pid) => readFileSync(`/proc/${pid}/cmdline`, "utf8").split("\0").includes(gitSocketPath(f.dataRoot));
  assert.equal(gitBound(coder.pid), true);
  assert.equal(gitBound(readSessions(f.dataRoot).find((s) => s.instance === "chief").pid), false);
  assert.equal(JSON.parse(readFileSync(join(coder.runDir, "session.json"), "utf8")).gitSocket, gitSocketPath(f.dataRoot));
  // Row 104 (#1598): each runner commits as its own instance, in this business.
  const identity = (pid) => readFileSync(`/proc/${pid}/environ`, "utf8").split("\0").filter((e) => e.startsWith("GIT_")).sort();
  for (const who of ["chief", "coder"]) {
    const runnerPid = readSessions(f.dataRoot).find((s) => s.instance === who).runnerPid;
    assert.deepEqual(identity(runnerPid), [`GIT_AUTHOR_EMAIL=${who}@acme.mosaic.invalid`, `GIT_AUTHOR_NAME=${who}`, `GIT_COMMITTER_EMAIL=${who}@acme.mosaic.invalid`, `GIT_COMMITTER_NAME=${who}`], who);
  }

  // The coder claimed its role and answers its own messages.
  const before = Date.now();
  const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
  const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
  assert.equal(echo.from_role, "coder");
  // Row 109 (#1603): the turn opens with the state block, then the header with the time it was sent.
  const turn = echo.body.match(new RegExp(`^echo: Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\nMessage ${hello.id} from cto, sent (\\S+), class REQUEST:\n\nhello$`));
  assert.ok(turn, echo.body);
  assert.ok(Date.parse(turn[1]) >= before - 1000 && Date.parse(turn[1]) <= Date.now(), "the sent time is when it was sent");

  // Host refusals, asked with the cto capability (steps 2 to 8 come before
  // the broker's role.launch check, which cto would fail).
  const refusals = [
    ["pm", "instance-not-listed"],
    ["nobody", "instance-not-listed"],
    ["coder", "instance-running"],
    ["reviewer", "capacity-full"],
    ["cto", "capacity-full"],
  ];
  for (const [instance, code] of refusals) {
    assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance })}\n`), { ok: false, error: code }, instance);
  }
  // A listed, free instance with room still needs role.launch: cto lacks it.
  // Stop the coder first so sonnet has room again.
  const coderRun = coder.run;
  process.kill(coder.runnerPid, "SIGTERM");
  await until(() => !readSessions(f.dataRoot).some((s) => s.run === coderRun) && log(f).some((e) => e.event === "end" && e.run === coderRun));
  const denied = await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: "reviewer" })}\n`);
  assert.equal(denied.ok, false);
  assert.notEqual(denied.error, "capacity-full");

  // Malformed requests.
  assert.deepEqual(await ask(sock, "not json\n"), { ok: false, error: "invalid-request" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "short", instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: "0".repeat(64), instance: "coder" })}\n`), { ok: false, error: "unauthenticated" });
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: 7 })}\n`), { ok: false, error: "invalid-request" });

  // The launch log has the launches, every refusal, and the coder's end.
  const entries = log(f);
  assert.deepEqual(entries.filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["chief", "jason"], ["coder", "chief"]]);
  const refused = entries.filter((e) => e.event === "refused").map((e) => [e.instance, e.code]);
  for (const r of refusals) assert.ok(refused.some(([i, c]) => i === r[0] && c === r[1]), `${r} is logged`);
  const end = entries.find((e) => e.event === "end" && e.run === coderRun);
  assert.equal(end.reason, "stopped");
  assert.ok(readSessions(f.dataRoot).some((s) => s.run === chief.run));
});

// Row 140 (#1634, REQ-COS-1): the pm asks the chief of staff for staff through request_staff, over the real
// tool socket; the request grants no launch, and the chief's launches stay inside the organization's limits.
test("a launched pm's request_staff reaches the chief of staff; no other session has the tool and the launch limits still hold", { skip, timeout: 60000 }, async (t) => {
  const { f, launcher, client, cto } = await setup(t, (doc) => {
    doc.roles.pm.vars = { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" };
    doc.launch.instances.push("pm");
    doc.launch.limits.models.zai["claude-sonnet-5-5"] = 2;
    return doc;
  });
  const sock = launchSocketPath(f.dataRoot);
  await launcher.launchPm();
  const call = async (to, body) => {
    const sent = await client.call("message.send", { to, class: "REQUEST", body });
    return until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
  };
  // The chief launches the pm within its limits, and the pm holds request_staff.
  assert.equal(JSON.parse((await call("chief", 'TOOL launch {"instance":"pm"}')).body).instance, "pm");
  const ask1 = await call("pm", `TOOL request_staff ${JSON.stringify({ task_ref: "vikunja:1/2", roles: [{ role: "coder", count: 1 }, { role: "reviewer", count: 1 }], note: "build and review" })}`);
  assert.equal(ask1.from_role, "pm");
  assert.equal(JSON.parse(ask1.body).to, "chief");
  // A refusal from the broker comes back as the tool's error, the same path any typed tool takes.
  const bad = await call("pm", `TOOL request_staff ${JSON.stringify({ task_ref: "vikunja:1/2", roles: [{ role: "coder", count: 0 }] })}`);
  assert.match(bad.body, /invalid-request/);
  // The chief of staff has no request_staff, and a pm has no launch.
  assert.equal((await call("chief", 'TOOL request_staff {"task_ref":"vikunja:1/2","roles":[{"role":"coder","count":1}]}')).body, "refused: unknown-tool");
  assert.equal((await call("pm", 'TOOL launch {"instance":"coder"}')).body, "refused: unknown-tool");
  // Nothing was launched by asking: the chief of staff's limits are the same, and sonnet has one place left.
  assert.deepEqual(readSessions(f.dataRoot).map((s) => s.instance).sort(), ["chief", "pm"]);
  const launched = await call("chief", 'TOOL launch {"instance":"coder"}');
  assert.equal(JSON.parse(launched.body).instance, "coder");
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: cto.cap, instance: "reviewer" })}\n`), { ok: false, error: "capacity-full" });
  assert.equal(log(f).filter((e) => e.event === "launch" && e.launchedBy === "pm").length, 0);
});

// Row 91 (#1585): the project clone in each workspace.
// Row 117 (#1611): and a dirty one is fetched, not merged, and logged.
test("each launch makes sure the workspace has a clone of the project: cloned, fast-forwarded, or fetched and logged", { skip, timeout: 60000 }, async (t) => {
  const given = prepare(t);
  const project = given.f.doc.projects.stack.root;
  const clone = join(workspaceDir(given.f.dataRoot, "acme", "chief"), "stack");
  const prompt = (run) => readFileSync(join(given.f.dataRoot, "launches", "acme", run, "bundle", "prompt.md"), "utf8");
  const kept = () => log(given.f).filter((e) => e.event === "checkout-kept");
  const fetched = () => log(given.f).filter((e) => e.event === "checkout-fetched");

  // The first launch clones the root's branch, with the root as origin.
  const first = await setup(t, undefined, given);
  const one = gitIn(project, "rev-parse", "HEAD");
  const pm1 = await first.launcher.launchPm();
  assert.equal(gitIn(clone, "rev-parse", "HEAD"), one);
  assert.equal(gitIn(clone, "symbolic-ref", "--short", "HEAD"), "main");
  assert.equal(gitIn(clone, "config", "--get-all", "remote.origin.url"), project);
  assert.ok(prompt(pm1.run).includes(`- Project clones: stack at ${clone}, on branch main at commit ${one} when this session started. `));

  // While the chief of staff runs, a second launch is refused before the host runs git
  // in its workspace: the root moved on, and the clone didn't.
  const two = commit(project, "two");
  await assert.rejects(first.launcher.launchPm(), /PM launch refused: instance-running/);
  assert.equal(gitIn(clone, "rev-parse", "HEAD"), one);
  await first.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
  assert.equal(await first.close(), 0);

  // The next launch fast-forwards the clean clone.
  const second = await setup(t, undefined, given);
  const pm2 = await second.launcher.launchPm();
  assert.equal(gitIn(clone, "rev-parse", "HEAD"), two);
  assert.ok(prompt(pm2.run).includes(`- Project clones: stack at ${clone}, on branch main at commit ${two} when this session started. `));
  assert.deepEqual(kept(), []);
  assert.deepEqual(fetched(), []);
  await second.client.call("role.release");
  assert.equal(await second.close(), 0);

  // A clone with the session's own work in it is fetched and not merged,
  // and the log says why.
  writeFileSync(join(clone, "notes.txt"), "mine\n");
  const three = commit(project, "three");
  const third = await setup(t, undefined, given);
  const pm3 = await third.launcher.launchPm();
  assert.equal(gitIn(clone, "rev-parse", "HEAD"), two);
  assert.equal(gitIn(clone, "rev-parse", "refs/remotes/origin/main"), three);
  assert.equal(readFileSync(join(clone, "notes.txt"), "utf8"), "mine\n");
  assert.deepEqual(fetched().map(({ instance, project: p, reason }) => ({ instance, project: p, reason })), [{ instance: "chief", project: "stack", reason: "dirty" }]);
  assert.deepEqual(kept(), []);
  assert.ok(prompt(pm3.run).includes(`- Project clones: stack at ${clone}, on branch main at commit ${two} when this session started, fetched but not fast-forwarded at this launch (dirty), with origin/main at commit ${three}. `));
  assert.equal(await third.close(), 0);

  // The root was never written: its branch, head and tree are the test's.
  assert.equal(gitIn(project, "rev-parse", "HEAD"), three);
  assert.equal(gitIn(project, "for-each-ref", "--format=%(refname)"), "refs/heads/main");
  assert.equal(gitIn(project, "status", "--porcelain"), "");
});

// Row 137 (#1631): the chief holds neither git.push.working nor
// review.verdict, so only a held merge decision can give it land_work and the
// git socket. Row 156 (#1651): for a non-thread launch that read is
// landHolders, which also gives session.json its landing. Both are the
// broker's (packages/bus/tests/land.test.mjs); here the host answers, since a
// real answer needs a merge decision the human resolved through the CLI.
test("a role the broker says holds a merge decision gets land_work, its prompt line, its landing and the git socket; without it, none", { skip, timeout: 30000 }, async (t) => {
  const asked = [];
  const held = [{ instance: "chief", decision: "d1", task_ref: "vikunja:1/2" }, { instance: "coder", decision: "d2", task_ref: "vikunja:1/3" }];
  const answering = (holders) => ({ wrap: (host) => Object.create(host, { op: { value: async (m) => (m.op === "landHolders" ? (asked.push(m), holders) : host.op(m)) } }) });
  const landing = "- Landing: once your merge decision is approved, land_work it with the task.";
  const read = (f, run, file) => readFileSync(join(f.dataRoot, "launches", "acme", run, file), "utf8");

  const one = await setup(t, undefined, undefined, answering(held));
  const pm1 = await one.launcher.launchPm();
  assert.deepEqual(asked, [{ op: "landHolders", business: "acme" }]);
  const tool = JSON.parse(read(one.f, pm1.run, "bundle/tools.json")).find((x) => x.name === "land_work");
  assert.equal(tool.verb, "working.land");
  assert.equal(tool.action, "git.merge.protected");
  assert.deepEqual(tool.parameters.required, ["task_ref", "decision"]);
  assert.ok(read(one.f, pm1.run, "bundle/prompt.md").includes(landing));
  assert.equal(JSON.parse(read(one.f, pm1.run, "session.json")).gitSocket, gitSocketPath(one.f.dataRoot));
  // Only the chief's own entry, from the same read as land_work.
  assert.deepEqual(JSON.parse(read(one.f, pm1.run, "session.json")).landing, [{ decision: "d1", task_ref: "vikunja:1/2" }]);

  // A separate fixture, so its business and data root are new. Another
  // instance's decision gives the chief nothing.
  const two = await setup(t, undefined, undefined, answering(held.slice(1)));
  const pm2 = await two.launcher.launchPm();
  assert.equal(asked.length, 2);
  assert.equal(JSON.parse(read(two.f, pm2.run, "bundle/tools.json")).some((x) => x.name === "land_work"), false);
  assert.equal(read(two.f, pm2.run, "bundle/prompt.md").includes(landing), false);
  assert.equal(JSON.parse(read(two.f, pm2.run, "session.json")).gitSocket, null);
  assert.equal("landing" in JSON.parse(read(two.f, pm2.run, "session.json")), false);
});

// Row 156 (#1651): the landing pass. The host answers landHolders and
// landRelease from `state`, as above; the broker's side of both is in
// packages/bus/tests/land.test.mjs. With no state.release, landRelease goes
// to the real host. The tick timer is an hour, so only the test's own tick()
// calls run a pass. The coder is listed in launch.instances and the chief
// isn't.
const LANDING = "- Landing: once your merge decision is approved, land_work it with the task.";
function landingHost(state) {
  const op = (host) => async (m) => {
    if (m.op === "landHolders") return state.holders;
    if (m.op === "landRelease") return (state.released.push(m.record), state.release ?? host.op(m));
    return host.op(m);
  };
  return { threadTickMs: 3600000, wrap: (host) => Object.create(host, { op: { value: op(host) } }) };
}
const runFile = (f, run, file) => readFileSync(join(f.dataRoot, "launches", "acme", run, file), "utf8");
const runnerText = (f, run) => (existsSync(join(f.dataRoot, "launches", "acme", run, "runner.log")) ? runFile(f, run, "runner.log") : "");
const landingOf = (f, run) => JSON.parse(runFile(f, run, "session.json")).landing;
const hasLand = (f, run) => JSON.parse(runFile(f, run, "bundle/tools.json")).some((x) => x.name === "land_work");
const coderRuns = (f) => log(f).filter((e) => e.event === "launch" && e.instance === "coder").map((e) => e.run);
const D1 = { instance: "coder", decision: "d1", task_ref: "vikunja:1/2" };

test("REQ-LAUNCH-1 the tick starts a listed merge decision holder once, with land_work, the Landing line and its landing turn; an unlisted holder starts nothing", { skip, timeout: 30000 }, async (t) => {
  const state = { holders: [{ instance: "chief", decision: "d0", task_ref: "vikunja:1/1" }, D1], release: { released: false, unread: 0 }, released: [] };
  const { f, launcher } = await setup(t, undefined, undefined, landingHost(state));
  await launcher.tick();
  assert.deepEqual(log(f).filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["coder", "jason"]]);
  const [run] = coderRuns(f);
  assert.equal(hasLand(f, run), true);
  assert.ok(runFile(f, run, "bundle/prompt.md").includes(LANDING));
  assert.deepEqual(landingOf(f, run), [{ decision: "d1", task_ref: "vikunja:1/2" }]);
  await until(() => runnerText(f, run).includes("runner: landing turn 1 for decision d1 (vikunja:1/2): ok"));

  // Running with the decision in its landing: no second start, no release.
  await launcher.tick();
  await launcher.tick();
  assert.deepEqual(coderRuns(f), [run]);
  assert.deepEqual(state.released, []);
  assert.equal(log(f).some((e) => e.event === "refused"), false);
});

test("a holder running since before the resolution is stopped once the broker releases it, land-relaunch, and the next tick starts it: landing turn first, then the message that waited", { skip, timeout: 45000 }, async (t) => {
  const state = { holders: [], released: [] };
  const { f, host, launcher, client, logs } = await setup(t, undefined, undefined, landingHost(state));
  const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  await new Client({ path: host.path, cap: chief.cap }).call("role.claim");
  const launched = await launcher.fromSession(chief.cap, "coder");
  assert.equal(launched.ok, true, JSON.stringify(launched));
  const old = launched.result.run;
  assert.equal(landingOf(f, old), undefined);
  assert.equal(hasLand(f, old), false);
  await until(() => runnerText(f, old).includes("claimed by run"));

  // Resolved now. The real broker holds no such decision and refuses through
  // the broker process; the launcher logs it once, and nothing stops.
  state.holders = [D1];
  await launcher.tick();
  await launcher.tick();
  assert.deepEqual(state.released, [{ business: "acme", run: old, decision: "d1" }, { business: "acme", run: old, decision: "d1" }]);
  assert.deepEqual(logs.filter((l) => l.includes("landRelease refused")), [`run ${old}: landRelease refused: decision-not-held`]);
  // The broker keeps the claim (a message delivered and unread): nothing stops.
  state.release = { released: false, unread: 1 };
  await launcher.tick();
  assert.equal(state.released.length, 3);
  assert.equal(log(f).some((e) => e.event === "land-relaunch"), false);
  assert.deepEqual(coderRuns(f), [old]);
  assert.equal(readSessions(f.dataRoot).find((s) => s.run === old)?.live, true);
  // Row 156 round 2 (F1 in 29033): the broker keeps the claim while the run is in its landing
  // turns (it hasn't asked message.receive yet). Nothing stops either; runner.test holds a real
  // landing turn open against the broker for this answer.
  state.release = { released: false, landing: true };
  await launcher.tick();
  assert.equal(state.released.length, 4);
  assert.equal(log(f).some((e) => e.event === "land-relaunch"), false);
  assert.deepEqual(coderRuns(f), [old]);
  assert.equal(readSessions(f.dataRoot).find((s) => s.run === old)?.live, true);

  // Released: one land-relaunch record, a SIGTERM, and no start while it ends.
  state.release = { released: true };
  await launcher.tick();
  assert.equal(state.released.length, 5);
  assert.deepEqual(log(f).filter((e) => e.event === "land-relaunch").map(({ event, instance, run, decision }) => ({ event, instance, run, decision })), [{ event: "land-relaunch", instance: "coder", run: old, decision: "d1" }]);
  const end = await until(() => log(f).find((e) => e.event === "end" && e.run === old));
  assert.deepEqual([end.reason, end.exitCode], ["stopped", 0]);
  assert.deepEqual(coderRuns(f), [old]);

  // A message for the coder waits with no run. The next tick starts the
  // holder with land_work and its landing; the landing turn is turn 1 and the
  // message is answered by the new run after it.
  const sent = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
  await launcher.tick();
  const [, next] = coderRuns(f);
  assert.ok(next);
  assert.equal(hasLand(f, next), true);
  assert.deepEqual(landingOf(f, next), [{ decision: "d1", task_ref: "vikunja:1/2" }]);
  const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
  assert.equal(reply.from_run, next);
  assert.match(runnerText(f, next), /runner: landing turn 1 for decision d1 \(vikunja:1\/2\): ok/);

  // Started after the resolution: left alone.
  await launcher.tick();
  assert.equal(state.released.length, 5);
  assert.deepEqual(coderRuns(f), [old, next]);
});

test("a landing run that ended isn't started again for its decision, a spent decision starts nothing, and a new decision starts the holder", { skip, timeout: 45000 }, async (t) => {
  const state = { holders: [D1], release: { released: false, unread: 0 }, released: [] };
  const { f, launcher } = await setup(t, undefined, undefined, landingHost(state));
  await launcher.tick();
  const [first] = coderRuns(f);
  await until(() => runnerText(f, first).includes("landing turn 1 for decision d1"));
  const s = readSessions(f.dataRoot).find((x) => x.run === first);
  process.kill(s.runnerPid, "SIGTERM");
  await until(() => log(f).find((e) => e.event === "end" && e.run === first));

  // Still listed, as if the run ended without landing: no second start.
  await launcher.tick();
  assert.deepEqual(coderRuns(f), [first]);
  // Spent: not listed.
  state.holders = [];
  await launcher.tick();
  assert.deepEqual(coderRuns(f), [first]);
  // A new decision for the same instance starts it once more.
  state.holders = [{ ...D1, decision: "d2" }];
  await launcher.tick();
  const runs = coderRuns(f);
  assert.equal(runs.length, 2);
  assert.deepEqual(landingOf(f, runs[1]), [{ decision: "d2", task_ref: "vikunja:1/2" }]);
  assert.deepEqual(state.released, []);
});

test("after a restart, the new host's first tick starts a holder whose decision is still held", { skip, timeout: 45000 }, async (t) => {
  const given = prepare(t);
  const state = { holders: [D1], release: { released: false, unread: 0 }, released: [] };
  const one = await setup(t, undefined, given, landingHost(state));
  await one.launcher.tick();
  const [first] = coderRuns(given.f);
  await until(() => runnerText(given.f, first).includes("landing turn 1 for decision d1"));
  await one.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
  assert.equal(await one.close(), 0);
  assert.deepEqual(readSessions(given.f.dataRoot), []);

  const two = await setup(t, undefined, given, landingHost(state));
  assert.deepEqual(coderRuns(given.f), [first]);
  await two.launcher.tick();
  const runs = coderRuns(given.f);
  assert.equal(runs.length, 2);
  assert.equal(hasLand(given.f, runs[1]), true);
  assert.deepEqual(landingOf(given.f, runs[1]), [{ decision: "d1", task_ref: "vikunja:1/2" }]);
  await until(() => runnerText(given.f, runs[1]).includes("landing turn 1 for decision d1 (vikunja:1/2): ok"));
  await two.client.call("role.release");
  assert.equal(await two.close(), 0);
});

// The reviewer and the coder share sonnet's one place, so a running coder
// makes the reviewer's landing start refuse capacity-full.
test("a refused landing start is logged once per code with its decision, isn't a start, and the next tick with room starts the holder", { skip, timeout: 45000 }, async (t) => {
  const state = { holders: [], release: { released: false, unread: 0 }, released: [] };
  const { f, host, launcher, logs } = await setup(t, undefined, undefined, landingHost(state));
  const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  await new Client({ path: host.path, cap: chief.cap }).call("role.claim");
  const coder = await launcher.fromSession(chief.cap, "coder");
  assert.equal(coder.ok, true, JSON.stringify(coder));
  await until(() => runnerText(f, coder.result.run).includes("claimed by run"));
  state.holders = [{ instance: "reviewer", decision: "d3", task_ref: "vikunja:1/4" }];
  await launcher.tick();
  await launcher.tick();
  const refused = log(f).filter((e) => e.event === "refused");
  assert.deepEqual(refused.map(({ by, instance, decision, code }) => ({ by, instance, decision, code })), [{ by: "jason", instance: "reviewer", decision: "d3", code: "capacity-full" }]);
  assert.deepEqual(logs.filter((l) => l.includes("landing start refused")), ["decision d3 (reviewer): landing start refused: capacity-full"]);
  assert.equal(log(f).some((e) => e.event === "launch" && e.instance === "reviewer"), false);

  const s = readSessions(f.dataRoot).find((x) => x.run === coder.result.run);
  process.kill(s.runnerPid, "SIGTERM");
  await until(() => !readSessions(f.dataRoot).some((x) => x.run === coder.result.run) && log(f).some((e) => e.event === "end" && e.run === coder.result.run));
  await launcher.tick();
  const [run] = log(f).filter((e) => e.event === "launch" && e.instance === "reviewer").map((e) => e.run);
  assert.ok(run);
  assert.deepEqual(landingOf(f, run), [{ decision: "d3", task_ref: "vikunja:1/4" }]);
  await until(() => runnerText(f, run).includes("runner: landing turn 1 for decision d3 (vikunja:1/4): ok"));
});

test("a project clone that can't be made refuses checkout-failed, from --chief and from a session, with no run and no clone left", { skip, timeout: 30000 }, async (t) => {
  const given = prepare(t);
  renameSync(join(given.f.doc.projects.stack.root, ".git"), join(given.root, "away.git"));
  const { f, launcher, host } = await setup(t, undefined, given);
  await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && /^PM launch refused: checkout-failed \(stack: the root .* can't be read: /.test(e.message));

  // A chief capability bound to this process asks for the coder.
  const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  await new Client({ path: host.path, cap: chief.cap }).call("role.claim");
  assert.deepEqual(await launcher.fromSession(chief.cap, "coder"), { ok: false, error: "checkout-failed" });

  const entries = log(f);
  assert.deepEqual(entries.map((e) => [e.event, e.instance, e.code]), [["refused", "chief", "checkout-failed"], ["refused", "coder", "checkout-failed"]]);
  for (const e of entries) assert.match(e.why, /^stack: the root /);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
  for (const instance of ["chief", "coder"]) assert.deepEqual(readdirSync(workspaceDir(f.dataRoot, "acme", instance)), [], `${instance}: no clone, no temp`);
});

test("a runner that stops at once ends its launch with the runner's reason", { skip, timeout: 30000 }, async (t) => {
  // The coder's tracker token has expired: the runner's founder-credential stop
  // exits 20 before role.claim, and the host records founder-credentials. The chief
  // of staff holds no credential (row 119), so a chief capability launches the coder.
  const { f, launcher, host } = await setup(t, (doc) => {
    doc.roles.coder.credentials.vikunja.expires = "2000-01-01";
    return doc;
  });
  const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) });
  await new Client({ path: host.path, cap: chief.cap }).call("role.claim");
  const launched = await launcher.fromSession(chief.cap, "coder");
  assert.equal(launched.ok, true, JSON.stringify(launched));
  const run = launched.result.run;
  const end = await until(() => existsSync(launchLogFile(f.dataRoot, "acme")) && log(f).find((e) => e.event === "end" && e.run === run));
  assert.equal(end.reason, "founder-credentials");
  assert.equal(end.exitCode, 20);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.match(readFileSync(join(f.dataRoot, "launches", "acme", run, "runner.log"), "utf8"), /vikunja/);
});

test("a Claude Code launch with no session credential refuses before it starts", { timeout: 30000 }, async (t) => {
  const given = prepare(t, (doc) => {
    doc.roles.chief.vars.harness = "claude-code";
    return doc;
  });
  delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
  given.f.env.ANTHROPIC_AUTH_TOKEN = "stand-in"; // a proxy key with no proxy
  const { f, launcher } = await setup(t, undefined, given);
  const names = "CLAUDE_CODE_OAUTH_TOKEN, ANTHROPIC_API_KEY, ANTHROPIC_AUTH_TOKEN with ANTHROPIC_BASE_URL";
  await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && e.message === `PM launch refused: no-session-credential (the bus host has none of ${names}; the bus unit's EnvironmentFile sets one when the bus starts)`);
  assert.deepEqual(log(f).map((e) => [e.event, e.code]), [["refused", "no-session-credential"]]);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
});

test("a pi launch with no own credential refuses no-session-credential", { skip, timeout: 30000 }, async (t) => {
  const given = prepare(t);
  delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
  delete given.f.env.ZAI_API_KEY;
  // No session login and none of pi's own provider keys, whatever the host exports.
  assert.deepEqual(HOST_CREDENTIALS.filter((k) => k in given.f.env), []);
  const { f, launcher } = await setup(t, undefined, given);
  await assert.rejects(launcher.launchPm(), /PM launch refused: no-session-credential/);
  assert.deepEqual(log(f).map((e) => [e.event, e.code]), [["refused", "no-session-credential"]]);
  assert.deepEqual(readSessions(f.dataRoot), []);
});

// Row 65: a harness module no core file names, switched on in the organization file, runs a real
// sandboxed session. The chief's primary harness fails its preflight, so the launch takes the
// stated fallback, and the manifest and the launch log record the route and the reason.
test("row 65 a failed primary preflight launches the fallback on a fixture harness module, recording the route and the reason", { skip, timeout: 30000 }, async (t) => {
  const given = prepare(t, (doc) => {
    doc.roles.chief.vars = { harness: "down", model: "claude-opus-5-5", thinking: "high", "fallback.harness": "fixture", "fallback.model": "claude-sonnet-5-5", "fallback.thinking": "low" };
    doc.plugins = { ...doc.plugins, harness: [...doc.plugins.harness, "down", "fixture"] };
    return doc;
  });
  const makeLoader = await fakeHarnesses(t, {}, (write) => {
    const pi = JSON.stringify(new URL("../../plugins/harness/pi/index.mjs", import.meta.url).href);
    write("harness", "down", `export * from ${pi};\nexport const preflight = () => ({ code: "fixture-down", message: "fixture preflight fails" });\n`);
    write("harness", "fixture", `export * from ${pi};\nexport const adapter = () => ${JSON.stringify(given.adapter)};\n`);
  });
  given.f.env.ZAI_API_KEY = "fx332-pi";
  const { f, launcher } = await setup(t, undefined, given, { makeLoader });
  const chief = await launcher.launchPm();
  assert.deepEqual({ ...chief, run: undefined }, { run: undefined, instance: "chief", harness: "fixture", model: "claude-sonnet-5-5" });
  const [s] = readSessions(f.dataRoot);
  const manifest = JSON.parse(readFileSync(join(s.runDir, "bundle", "manifest.json"), "utf8"));
  assert.deepEqual([manifest.harness, manifest.model, manifest.thinking, manifest.route, manifest.routeReason], ["fixture", "claude-sonnet-5-5", "low", "fallback", "fixture-down"]);
  const [launched] = log(f).filter((e) => e.event === "launch");
  assert.deepEqual([launched.harness, launched.model, launched.route, launched.routeReason], ["fixture", "claude-sonnet-5-5", "fallback", "fixture-down"]);
  // The session runs on the fixture's adapter: the fake adapter claims the run.
  const runnerLog = join(s.runDir, "runner.log");
  await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
});

// The runner is pid 1 in its namespace and ignores a SIGKILL sent from
// inside it, so the host kills it by its host pid. bwrap then exits 137,
// which the launcher reads back as the signal.
test("a runner killed from the host ends its launch as killed, not failed", { skip, timeout: 30000 }, async (t) => {
  const { f, launcher } = await setup(t);
  const chief = await launcher.launchPm();
  const runnerLog = join(f.dataRoot, "launches", "acme", chief.run, "runner.log");
  await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
  const [s] = readSessions(f.dataRoot);
  process.kill(s.runnerPid, "SIGKILL");
  const end = await until(() => log(f).find((e) => e.event === "end" && e.run === chief.run));
  assert.deepEqual([end.reason, end.exitCode, end.signal], ["killed", null, "SIGKILL"]);
  assert.deepEqual(readSessions(f.dataRoot), []);
});

test("a missing or failing bwrap refuses sandbox-unavailable before anything starts; there is no unconfined fallback",{ timeout: 30000 }, async (t) => {
  const given = prepare(t);
  const bad = join(given.root, "bwrap");
  writeFileSync(bad, "#!/bin/sh\necho 'bwrap: setting up uid map: Permission denied' >&2\nexit 1\n", { mode: 0o755 });
  const { f, launcher, cto } = await setup(t, undefined, given, { bwrap: bad });
  await assert.rejects(launcher.launchPm(), (e) => e.message === `PM launch refused: sandbox-unavailable (${bad} failed: bwrap: setting up uid map: Permission denied)`);
  rmSync(bad);
  await assert.rejects(launcher.launchPm(), (e) => e.message === `PM launch refused: sandbox-unavailable (${bad} is missing)`);
  // A session's launch request gets the same code, through refuse().
  assert.deepEqual(await launcher.fromSession(cto.cap, "coder"), { ok: false, error: "sandbox-unavailable" });
  assert.deepEqual(log(f).map((e) => e.code), ["sandbox-unavailable", "sandbox-unavailable", "sandbox-unavailable"]);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
});

test("row 64 a role skill that isn't enabled refuses skill-not-enabled, and one that isn't a plain directory skill-refused, before anything starts", { timeout: 30000 }, async (t) => {
  const given = prepare(t);
  // The chief of staff's role file names no skill; this copy gives it one, so --chief takes the check too.
  const chief = join(given.f.roles, "chief-of-staff.json");
  writeFileSync(chief, JSON.stringify({ ...JSON.parse(readFileSync(chief, "utf8")), skills: ["ms-unslop"] }, null, 2));
  const enabled = join(given.f.dataRoot, "skills-enabled");
  rmSync(join(enabled, "ms-unslop"), { recursive: true });
  const { f, launcher, cto } = await setup(t, undefined, given);
  await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && e.message === "PM launch refused: skill-not-enabled (skill ms-unslop isn't enabled; run scripts/skill.sh activate ms-unslop)");
  // The coder's list is ms-unslop, ms-sdlc, ms-file-read, ms-file-write: the first one missing is named.
  assert.deepEqual(await launcher.fromSession(cto.cap, "coder"), { ok: false, error: "skill-not-enabled" });
  // A symlink in skills-enabled is refused, not followed.
  symlinkSync(join(REPO, "skills", "ms-unslop"), join(enabled, "ms-unslop"));
  await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && e.message === `PM launch refused: skill-refused (skill ms-unslop isn't a plain directory under ${enabled})`);
  assert.deepEqual(await launcher.fromSession(cto.cap, "coder"), { ok: false, error: "skill-refused" });
  assert.deepEqual(log(f).map((e) => [e.event, e.code]), [["refused", "skill-not-enabled"], ["refused", "skill-not-enabled"], ["refused", "skill-refused"], ["refused", "skill-refused"]]);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
});

// Row 213 (C2-08, #1714), deviation D2: the launcher hands the bundle <configDir>/personas, so a
// creative instance's persona file is read from beside the business file. A file off its digest
// refuses persona-mismatch, and the launch logs that reason; with no persona directory the reason
// would be that the bundle has none.
test("row 213 a creative instance's persona is read from <configDir>/personas: off its digest the launch fails with the mismatch as its reason", { timeout: 30000 }, async (t) => {
  const text = "Warm. PERSONA-MARKER-d2\n";
  const given = prepare(t, (doc) => {
    doc.roles.bean = { definition: "companion", credentials: { gitea: { ...doc.roles.coder.credentials.gitea } }, vars: { harness: "pi", model: "claude-sonnet-5-5", thinking: "high", "persona.name": "Bean", "persona.sha256": createHash("sha256").update(text).digest("hex") } };
    doc.launch.instances.push("bean");
    return doc;
  });
  const dir = join(given.root, "config", "personas", "acme");
  mkdirSync(dir, { recursive: true, mode: 0o700 });
  writeFileSync(join(dir, "bean.md"), "Not the pinned text.\n", { mode: 0o600 });
  // Row 219: creative instances launch from their thread, not from a chief's standalone request.
  // The mismatch occurs before bindLaunch, so only the host's thread-state snapshot is a fixture.
  const thread = 'persona-thread';
  const SESSION = 'persona-session';
  const { f, launcher } = await setup(t, undefined, given, { wrap: host => ({
    ...host,
    op: message => message.op === 'threadsState'
      ? Promise.resolve([{ thread, session: SESSION, instance: 'bean', holder: null, pending: 1 }])
      : host.op(message),
  }) });
  // The thread path records row 270's run-start count before the bundle's persona refusal.
  await launcher.tick();
  const runs = readdirSync(join(f.dataRoot, "launches", "acme"));
  const entries = log(f);
  assert.equal(entries.length, 2);
  const { at, ...compactions } = entries[0];
  assert.deepEqual(compactions, { event: 'compactions', instance: 'bean', thread, session: SESSION, run: runs[0], count: 0, known: true, reason: null });
  const e = entries[1];
  assert.deepEqual([e.event, e.code], ["refused", "launch-failed"]);
  assert.equal(e.thread, thread);
  assert.equal(e.why, `instance bean: persona file ${join(dir, "bean.md")} isn't the file persona.sha256 names`);
  assert.ok(!JSON.stringify(log(f)).includes("PERSONA-MARKER-d2"));
  assert.deepEqual(readSessions(f.dataRoot), []);
  // The run directory exists by then (BUILD.md residual); the persona is read before any bundle file.
  assert.equal(runs.length, 1);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches", "acme", runs[0])), []);
  assert.equal(existsSync(join(f.dataRoot, "threads", "acme", thread, "sessions", SESSION, "session")), true);
  assert.equal(existsSync(join(f.dataRoot, "threads", "acme", thread, "session")), false);
});

test("row 64 on Claude Code a role skill that isn't enabled refuses the same way, before anything starts", { timeout: 30000 }, async (t) => {
  const given = prepare(t, (doc) => {
    doc.roles.chief.vars.harness = "claude-code";
    doc.roles.coder.vars.harness = "claude-code";
    return doc;
  });
  const chief = join(given.f.roles, "chief-of-staff.json");
  writeFileSync(chief, JSON.stringify({ ...JSON.parse(readFileSync(chief, "utf8")), skills: ["ms-unslop"] }, null, 2));
  rmSync(join(given.f.dataRoot, "skills-enabled", "ms-unslop"), { recursive: true });
  const { f, launcher, cto } = await setup(t, undefined, given);
  await assert.rejects(launcher.launchPm(), (e) => e.exitCode === 3 && e.message === "PM launch refused: skill-not-enabled (skill ms-unslop isn't enabled; run scripts/skill.sh activate ms-unslop)");
  assert.deepEqual(await launcher.fromSession(cto.cap, "coder"), { ok: false, error: "skill-not-enabled" });
  assert.deepEqual(log(f).map((e) => [e.event, e.code]), [["refused", "skill-not-enabled"], ["refused", "skill-not-enabled"]]);
  assert.deepEqual(readSessions(f.dataRoot), []);
  assert.deepEqual(readdirSync(join(f.dataRoot, "launches")), ["acme.jsonl"], "no run was made");
});

// The brief's one real Claude Code launch: the host's claude in a coder's
// sandbox, logged in with a stand-in proxy key against the scripted Messages
// API, runs one Bash call that tries the founder's stand-ins, another run
// and the checkout. No real credential and no real model.
const which = spawnSync("sh", ["-c", "command -v claude"], { encoding: "utf8" });
const CLAUDE = process.env.MOSAIC_CLAUDE_BIN ?? (which.status === 0 ? which.stdout.trim() : null);

test("a real Claude Code coder's shell can't read a stand-in secret or anything else its sandbox doesn't hold", { skip: skip || (CLAUDE ? false : "claude is not on PATH"), timeout: 180000 }, async (t) => {
  const script = [];
  const api = await mockAnthropic(t, { script, done: (r) => `ANSWER ${JSON.stringify(r.map((x) => [x.isError, x.content.slice(0, 1500)]))}` });
  const given = prepare(t, (doc) => {
    doc.roles.coder.vars.harness = "claude-code";
    doc.roles.coder.vars["limits.tools"] = ["bash"];
    // Row 126: the claude-code launch spends this account, which names the proxy login set.
    doc.launch.account = { name: "proxy-login", credential: "ANTHROPIC_AUTH_TOKEN+ANTHROPIC_BASE_URL" };
    return doc;
  });
  delete given.f.env.CLAUDE_CODE_OAUTH_TOKEN;
  Object.assign(given.f.env, { ANTHROPIC_AUTH_TOKEN: "stand-in", ANTHROPIC_BASE_URL: api.url, MOSAIC_CLAUDE_BIN: CLAUDE });
  const secret = join(given.root, "home", ".config", "mosaic-dev", "secrets", "stand-in.env");
  const mosaic = join(given.root, "home", ".mosaic", "fleet", "token");
  for (const p of [secret, mosaic]) {
    mkdirSync(join(p, ".."), { recursive: true, mode: 0o700 });
    writeFileSync(p, "FAKE_TOKEN=stand-in-value\n", { mode: 0o600 });
  }
  // The real claude-code module, so its own adapter and binary check (row 65).
  const { f, launcher, client } = await setup(t, undefined, given, { harnesses: { pi: given.adapter, "claude-code": null } });
  const chief = await launcher.launchPm();
  const chiefRun = readSessions(f.dataRoot).find((s) => s.run === chief.run).runDir;

  // [what, shell command, may the session do it]
  const probes = [
    ["read the stand-in secret", `cat '${secret}'`, false],
    ["read the stand-in ~/.mosaic", `cat '${mosaic}'`, false],
    ["read the chief's run directory", `cat '${chiefRun}/session.json'`, false],
    ["read the data root's sessions file", `cat '${sessionsFile(f.dataRoot)}'`, false],
    ["read the checkout", `cat '${join(REPO, "AGENTS.md")}'`, false],
    ["write outside", `touch '${given.root}/escaped'`, false],
    ["write and read its workspace", "echo ws > probe && cat probe", true],
    ["write and read /tmp", "echo tmp > /tmp/probe && cat /tmp/probe", true],
    ["HOME is the run's home/", `case "$HOME" in */home) test -w "$HOME";; *) false;; esac`, true],
  ];
  const command = probes.map(([what, cmd], i) => `if { ${cmd}; } >/dev/null 2>&1; then echo "${i} yes"; else echo "${i} no"; fi`).join("\n");
  script.push({ name: "Bash", input: { command, description: "probe the sandbox" } });

  const sent = await client.call("message.send", { to: "chief", class: "REQUEST", body: 'TOOL launch {"instance":"coder"}' });
  const launched = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id));
  assert.equal(JSON.parse(launched.body).instance, "coder", launched.body);
  const coder = readSessions(f.dataRoot).find((s) => s.instance === "coder");
  assert.equal(coder.harness, "claude-code");
  assert.deepEqual(log(f).filter((e) => e.event === "launch").map((e) => [e.instance, e.account]), [["chief", null], ["coder", "proxy-login"]]);

  const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "probe" });
  const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id), 150000);
  assert.equal(reply.from_role, "coder");
  const answer = reply.body.match(/ANSWER (\[.*\])/s);
  assert.ok(answer, reply.body);
  const [[isError, out]] = JSON.parse(answer[1]);
  assert.equal(isError, false, out);
  const got = Object.fromEntries(out.trim().split("\n").map((l) => l.split(" ")).map(([i, v]) => [probes[i][0], v === "yes"]));
  assert.deepEqual(got, Object.fromEntries(probes.map(([what, , may]) => [what, may])));
  assert.ok(!existsSync(join(given.root, "escaped")));
  assert.ok(!reply.body.includes("stand-in-value"), "the secret's value never comes back");
  // The session logged in with the stand-in proxy key, against the mock.
  assert.ok(api.requests.length > 0);
  assert.ok(api.requests.some((r) => r.tools.includes("Bash")), api.requests.map((r) => r.tools.join(",")).join(" | "));
  // Row 126: the bus got the account with the run, so the coder's turn records it (no usage
  // source here, so the turn is unknown); the pi chief's turns record null.
  const accounts = await until(async () => {
    const seen = (await client.call("usage")).byAccount.map((r) => r.account);
    return seen.includes("proxy-login") && seen;
  });
  assert.deepEqual([...accounts].sort(), [null, "proxy-login"]);
});

for (const exited of [false, true]) {
  test(`a host that died hard leaves its sessions to the next host, which ${exited ? "finds one already exited (23)" : "kills them"} and ends their runs so the role can be launched again`, { skip, timeout: 30000 }, async (t) => {
    const given = prepare(t);
    const { f } = given;
    const old = spawn(process.execPath, [HOST, given.adapter, FAKE], { env: f.env, stdio: ["ignore", "pipe", "inherit"] });
    t.after(() => old.kill("SIGKILL"));
    const [line] = await once(createInterface({ input: old.stdout }), "line");
    const first = JSON.parse(line);
    const [left] = readSessions(f.dataRoot);
    assert.equal(left.run, first.run);
    // Killed before the claim, the runner would exit 23 on its own.
    const runnerLog = join(f.dataRoot, "launches", "acme", first.run, "runner.log");
    await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
    assert.equal(left.live, true);
    old.kill("SIGKILL");
    await once(old, "exit");
    // The broker child exits on the lost IPC channel; the session runs on.
    await until(() => !existsSync(join(f.dataRoot, "bus", "writer.lock")));
    assert.equal(startTimeOf(left.runnerPid), left.runnerStartTime);
    if (!exited) {
      // Stopped, the leftover neither polls the dead broker nor answers
      // SIGTERM: only the next host's SIGKILL ends it.
      for (const pid of [left.pid, left.runnerPid]) process.kill(pid, "SIGSTOP");
      t.after(() => {
        for (const [pid, start] of [[left.pid, left.startTime], [left.runnerPid, left.runnerStartTime]]) {
          if (startTimeOf(pid) === start) process.kill(pid, "SIGKILL");
        }
      });
    }
    if (exited) {
      await until(() => startTimeOf(left.pid) === null);
      assert.match(readFileSync(runnerLog, "utf8"), /broker unreachable after 30 tries[\s\S]*exiting 23/);
      assert.equal(readSessions(f.dataRoot)[0].live, false);
    }

    const s = await setup(t, undefined, given);
    assert.equal(startTimeOf(left.pid), null, "the old session process is gone");
    assert.equal(startTimeOf(left.runnerPid), null, "and its runner with it");
    assert.ok(s.logs.includes(`run ${first.run} (chief) from an earlier host ended: host-lost`), s.logs.join("\n"));
    const ended = (await s.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended");
    assert.equal(ended.length, 1);
    assert.deepEqual(ended[0].body, { reason: "host-lost", exitCode: null, released: true });
    const log = readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l));
    assert.deepEqual(log.filter((e) => e.event === "end").map((e) => [e.run, e.reason]), [[first.run, "host-lost"]]);
    assert.deepEqual(readSessions(f.dataRoot), []);

    // Without the end, the old run would still hold chief and this runner would exit 21.
    const again = await s.launcher.launchPm();
    await until(async () => (await s.client.call("agents")).some((a) => a.role === "chief" && a.holder_run === again.run));
    await s.client.call("role.release"); // the test's own cto claim, so the next host's can claim again
    assert.equal(await s.close(), 0);

    // Died after the end but before sessions.json changed: the run stays ended.
    const { live, ...entry } = left;
    new Registry(f.dataRoot).add(entry);
    const third = await setup(t, undefined, given);
    assert.ok(third.logs.includes(`run ${first.run} (chief) from an earlier host was already ended`), third.logs.join("\n"));
    assert.equal((await third.client.call("trail", { subject: first.run })).filter((r) => r.kind === "session.ended").length, 1);
    assert.equal(await third.close(), 0);
  });
}

// Row 163 (mos-0961): the D2 rehearsal host. A live bwrap process the test owns stands in for an
// earlier host's session; its outer bwrap is what recover() would kill, and its child stands in
// for the runner that close() would signal.
test("bus start --no-launch leaves an earlier host's live session alone: no signal, no end, sessions.json as it was", { skip, timeout: 30000 }, async (t) => {
  const given = prepare(t);
  const { f } = given;
  const box = spawn(BWRAP, ["--ro-bind", "/", "/", "--dev", "/dev", "--proc", "/proc", "--unshare-pid", "--die-with-parent", "sleep", "60"], { stdio: "ignore" });
  t.after(() => box.kill("SIGKILL"));
  const pid = box.pid;
  const startTime = await until(() => startTimeOf(pid));
  const runnerPid = await until(() => childrenOf(pid)[0]);
  const runnerStartTime = await until(() => startTimeOf(runnerPid));
  assert.equal(cmdlineOf(pid)[0], BWRAP, "recover() would take it for a session");
  new Registry(f.dataRoot).add({ business: "acme", instance: "chief", run: "run-left", harness: "pi", pid, startTime, runnerPid, runnerStartTime });
  const before = readFileSync(sessionsFile(f.dataRoot), "utf8");

  const s = await setup(t, undefined, given, { launches: false });
  assert.ok(s.logs.includes("run run-left (chief) from an earlier host: left as it is, this host was started with --no-launch"), s.logs.join("\n"));
  assert.equal(startTimeOf(pid), startTime, "listen() sent no signal");
  assert.equal(readFileSync(sessionsFile(f.dataRoot), "utf8"), before);
  assert.deepEqual((await s.client.call("trail", { subject: "run-left" })).filter((r) => r.kind === "session.ended"), []);
  assert.equal(await s.close(), 0);
  assert.equal(startTimeOf(pid), startTime, "close() sent no signal either");
  assert.equal(startTimeOf(runnerPid), runnerStartTime);
  assert.equal(readFileSync(sessionsFile(f.dataRoot), "utf8"), before);
});

test("bus start --no-launch refuses every launch: the PM launch and a session's request, each logged, nothing spawned", { skip, timeout: 30000 }, async (t) => {
  const s = await setup(t, undefined, undefined, { launches: false });
  await assert.rejects(s.launcher.launchPm(), (e) => e.exitCode === 3 && /^PM launch refused: launches-disabled/.test(e.message));
  const sock = launchSocketPath(s.f.dataRoot);
  assert.deepEqual(await ask(sock, `${JSON.stringify({ cap: s.cto.cap, instance: "coder" })}\n`), { ok: false, error: "launches-disabled" });
  assert.deepEqual(log(s.f).map((e) => [e.event, e.instance, e.code]), [["refused", "chief", "launches-disabled"], ["refused", "coder", "launches-disabled"]]);
  assert.deepEqual(readSessions(s.f.dataRoot), []);
  const runs = runsDir(s.f.dataRoot, "acme");
  assert.deepEqual(existsSync(runs) ? readdirSync(runs) : [], [], "no run directory");
  assert.equal(await s.close(), 0);
});

test("bus start --no-launch starts no thread tick; the same host with launches runs it", { skip, timeout: 30000 }, async (t) => {
  for (const launches of [false, true]) {
    const asked = [];
    const wrap = (host) => Object.create(host, { op: { value: async (m) => (["threadsState", "landHolders"].includes(m.op) && asked.push(m.op), host.op(m)) } });
    const s = await setup(t, undefined, undefined, { launches, threadTickMs: 50, wrap });
    await new Promise((r) => setTimeout(r, 400));
    assert.equal(asked.length > 0, launches, `launches ${launches}: ${asked.join(", ")}`);
    assert.equal(await s.close(), 0);
  }
});

test("a malformed sessions.json refuses the host start with exit 3 and stays as it was", { timeout: 30000 }, async (t) => {
  const given = prepare(t);
  const file = sessionsFile(given.f.dataRoot);
  mkdirSync(join(given.f.dataRoot, "bus-host"), { mode: 0o700 });
  writeFileSync(file, "{", { mode: 0o600 });
  await assert.rejects(setup(t, undefined, given), (e) => e.exitCode === 3 && /sessions file is unreadable.*move the file aside/.test(e.message));
  assert.equal(readFileSync(file, "utf8"), "{");
  assert.equal(existsSync(join(given.f.dataRoot, "bus", "writer.lock")), false, "the host closed");
});

const within = (p, ms, what) => Promise.race([p, new Promise((_, reject) => setTimeout(() => reject(new Error(`${what} took over ${ms} ms`)), ms).unref())]);

test("an over-long launch request is refused at once, not at the 10 s idle timeout", { timeout: 30000 }, async (t) => {
  const { f } = await setup(t);
  const s = connect(launchSocketPath(f.dataRoot));
  t.after(() => s.destroy());
  let buf = "";
  s.on("data", (b) => (buf += b));
  s.write("x".repeat(5000));
  await within(once(s, "end"), 3000, "the refusal");
  assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
});

test("a launch client that never closes its side doesn't hold the host's close", { timeout: 30000 }, async (t) => {
  const { f, close } = await setup(t);
  const path = launchSocketPath(f.dataRoot);
  // One got its reply and never ends; one never sends anything.
  const answered = connect({ path, allowHalfOpen: true });
  const silent = connect({ path, allowHalfOpen: true });
  const connected = once(silent, "connect");
  const gone = Promise.all([once(answered, "close"), once(silent, "close")]);
  const drop = () => {
    answered.destroy();
    silent.destroy();
  };
  t.after(drop);
  let buf = "";
  answered.on("data", (b) => (buf += b));
  answered.write("not json\n");
  await once(answered, "end");
  assert.deepEqual(JSON.parse(buf), { ok: false, error: "invalid-request" });
  await within(connected, 2000, "connect");
  // On a failure the clients go, so the host can still close and the test
  // fails instead of hanging.
  const code = await within(close(), 5000, "close").catch((e) => {
    drop();
    throw e;
  });
  assert.equal(code, 0);
  assert.equal(existsSync(path), false);
  answered.end();
  silent.end();
  await within(gone, 2000, "the clients' close");
});

test("a runner that ignores SIGTERM is killed when the host closes", { skip, timeout: 60000 }, async (t) => {
  const { f, launcher, close } = await setup(t, undefined, undefined, { stopTimeoutMs: 1000 });
  const chief = await launcher.launchPm();
  const runnerLog = join(f.dataRoot, "launches", "acme", chief.run, "runner.log");
  await until(() => existsSync(runnerLog) && readFileSync(runnerLog, "utf8").includes("claimed by run"));
  const [s] = readSessions(f.dataRoot);
  // Stopped from outside its namespace, it can't act on SIGTERM.
  process.kill(s.runnerPid, "SIGSTOP");
  const kill = () => {
    if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGKILL");
  };
  t.after(kill);
  const started = Date.now();
  const code = await within(close(), 15000, "close").catch((e) => {
    kill();
    throw e;
  });
  assert.equal(code, 0);
  assert.ok(Date.now() - started >= 1000, "it waited for the stop timeout");
  assert.equal(startTimeOf(s.pid), null);
  assert.equal(startTimeOf(s.runnerPid), null);
  assert.deepEqual(readSessions(f.dataRoot), []);
  const ends = log(f).filter((e) => e.event === "end" && e.run === chief.run);
  assert.equal(ends.length, 1);
  assert.equal(ends[0].signal, "SIGKILL");
});

// Row 191: root-scoped processing is stronger than per-business launch.restore.
test('kill switch refuses chief and hire before launch work and a clear permits a later request', { skip, timeout: 30000 }, async t => {
  const { readControl, transition } = await import('../../processing-control/src/state.mjs');
  const s = await setup(t);
  const halted = transition(s.f.dataRoot, { business: 'acme', human: s.f.doc.human, halted: true });
  await assert.rejects(s.launcher.launchPm(), /PM launch refused: processing-halted/);
  assert.deepEqual(await s.launcher.fromSession(s.cto.cap, 'coder'), { ok: false, error: 'processing-halted' });
  assert.deepEqual(readSessions(s.f.dataRoot), []);
  transition(s.f.dataRoot, { business: 'acme', human: s.f.doc.human, halted: false, generation: halted.generation });
  const run = await s.launcher.launchPm();
  assert.equal(readControl(s.f.dataRoot).halted, false);
  assert.equal(readSessions(s.f.dataRoot)[0].run, run.run);
});

test('kill switch stops an actual managed namespace and records the halt reason without rewriting its bundle', { skip, timeout: 30000 }, async t => {
  const { transition } = await import('../../processing-control/src/state.mjs');
  const s = await setup(t);
  const run = await s.launcher.launchPm();
  const entry = readSessions(s.f.dataRoot).find(r => r.run === run.run);
  const session = readFileSync(join(entry.runDir, 'session.json'));
  const halted = transition(s.f.dataRoot, { business: 'acme', human: s.f.doc.human, halted: true });
  await until(() => !readSessions(s.f.dataRoot).some(r => r.run === run.run), 12000);
  assert.deepEqual(readFileSync(join(entry.runDir, 'session.json')), session);
  const end = log(s.f).find(r => r.event === 'end' && r.run === run.run);
  assert.equal(end.reason, 'kill-switch');
  assert.equal(end.haltGeneration, halted.generation);
  assert.equal(startTimeOf(entry.runnerPid), null);
});

test('a launch waiting across halt and resume cannot use its stale admission', { skip, timeout: 30000 }, async t => {
  const { transition } = await import('../../processing-control/src/state.mjs');
  let root, first = true;
  const prepared = prepare(t); root = prepared.f.dataRoot;
  const s = await setup(t, doc => doc, prepared, { wrap: host => ({ ...host, async op(m) {
    const answer = await host.op(m);
    if (m.op === 'mayLaunch' && first) {
      first = false;
      const state = transition(root, { business: 'acme', human: 'jason', halted: true });
      transition(root, { business: 'acme', human: 'jason', halted: false, generation: state.generation });
    }
    return answer;
  } }) });
  await assert.rejects(s.launcher.launchPm(), /processing-generation-changed/);
  assert.deepEqual(readSessions(root), []);
});