// CHAT-03 §6 stop, cohort proof and recovery (#1507): K1–K19. The scope
// fixtures run the fake engine as a real process under ScopeLauncher, so the
// shim, the `engine` cgroup and systemd's invocation ID are all real; they
// skip when systemd user scopes are unavailable. K2 runs on the process-group
// fallback. K6–K9 and K16–K18 use the in-process fake, whose force stop is a
// fixture stand-in (see FakeLauncher). K19 launches /bin/sleep through
// ScopeLauncher with no controller. E1–E6 (row 52) are the engine-exit stop
// on the scope fixtures. Controllers that must die run in ctrl-child.mjs.

import { test, after } from "node:test";
import assert from "node:assert/strict";
import { appendFileSync, chmodSync, copyFileSync, mkdirSync, readFileSync, renameSync, rmdirSync, writeFileSync } from "node:fs";
import { spawn, spawnSync } from "node:child_process";
import { dirname, join } from "node:path";
import { ClaimStore, FOREIGN_HOST, defaultHost, newClaimId, unitNameFor } from "../src/claim.mjs";
import { ConversationClient } from "../src/client.mjs";
import { AUTHORITY, PgroupLauncher, ScopeLauncher, engineExitCohort, forceStopCohort, releaseCohort, scopeAvailable, shimRequest, systemctlShow, systemdUnits } from "../src/cohort.mjs";
import { Controller, ELIGIBILITY, TEST_ENGINE } from "../src/controller.mjs";
import { ENGINE_ENV, ENGINE_PIN_MISMATCH, engineEnv } from "../src/pi-pin.mjs";
import { FixtureVerifier, newId } from "../src/records.mjs";
import { ControlClient, FakeLauncher } from "./fake-pi.mjs";
import { FAST, REPO, assistantEntry, claimRecords, cleanupAll, controllerFor, fixture, killChildren, killShims, liveShims, noUnits, reap, receiptState, shimsGone, spawnController, started, tick } from "./harness.mjs";

const reaped = [];
const strays = new Set();
after(async () => {
  for (const pid of strays) {
    try {
      process.kill(pid, "SIGKILL");
    } catch {
      // gone
    }
  }
  for (const fx of reaped) reap(fx);
  killShims();
  const left = await shimsGone();
  killChildren();
  cleanupAll();
  assert.deepEqual(left, [], "no shim outlives this file (#1533)");
});
const track = (fx) => (reaped.push(fx), fx);
const SCOPE = scopeAvailable();
const NEEDS_SCOPE = { skip: !SCOPE && "systemd user scopes unavailable", timeout: 60000 };
const FAKE_PI = join(import.meta.dirname, "fake-pi.mjs");

async function until(pred, ms = 4000, what = "condition") {
  const end = Date.now() + ms;
  while (!(await pred())) {
    if (Date.now() > end) throw new Error(`timed out waiting for ${what}`);
    await tick(10);
  }
}

// `p`'s value, or null if `ms` pass first (Filbert N3 on #1533).
async function within(p, ms) {
  let timer;
  const late = new Promise((r) => (timer = setTimeout(() => r(null), ms)));
  try {
    return await Promise.race([p, late]);
  } finally {
    clearTimeout(timer);
  }
}

// A zombie has exited; only its parent hasn't reaped it yet.
function alive(pid) {
  try {
    const st = readFileSync(`/proc/${pid}/stat`, "utf8");
    return st.slice(st.lastIndexOf(")") + 2)[0] !== "Z";
  } catch {
    return false;
  }
}

class SpyVerifier extends FixtureVerifier {
  constructor() {
    super({ authorities: [AUTHORITY] });
    this.posted = [];
  }
  post(p) {
    this.posted.push(structuredClone(p));
    return super.post(p);
  }
}

// Holds the controller at named barriers (as in races.test.mjs).
function gate() {
  const want = new Set(), held = new Map();
  return {
    barrier: async (name) => {
      if (!want.has(name)) return;
      want.delete(name);
      await new Promise((r) => held.set(name, r));
    },
    hold: (name) => want.add(name),
    waitHeld: (name, ms = 8000) => until(() => held.has(name), ms, `barrier ${name}`),
    release: (name) => {
      const r = held.get(name);
      held.delete(name);
      r?.();
    },
  };
}

// A controller in this process on a real engine process: the fake engine
// under ScopeLauncher ("scope") or PgroupLauncher ("pgroup").
async function live({ kind = "scope", barrier = null, verifier = new FixtureVerifier({ authorities: [AUTHORITY] }), launcher = null, timeouts = FAST } = {}) {
  const fx = track(fixture());
  const control = join(fx.base, "fake.sock");
  const ctrl = new Controller({
    fixtureRoot: fx.base, claimRoot: fx.claimRoot, socketDir: fx.socketDir, sessionFile: fx.sessionFile, seat: fx.seat,
    launcher: launcher ?? (kind === "scope" ? new ScopeLauncher() : new PgroupLauncher()),
    engine: { cwd: fx.proj }, [TEST_ENGINE]: { command: process.execPath, preArgs: [FAKE_PI], env: { ...process.env, FAKE_PI_CONTROL: control, FAKE_PI_LOG: join(fx.base, "fake.log") } },
    verifier, units: kind === "scope" ? systemdUnits : noUnits, timeouts, barrier,
  });
  await ctrl.start();
  assert.equal(ctrl.binding.state, "active", JSON.stringify(ctrl.evidence.uncertain));
  const c = new ConversationClient({ socketPath: ctrl.socketPath });
  await c.connect();
  assert.equal((await c.takeover()).outcome, "transferred");
  const fake = new ControlClient(control);
  await fake.connect();
  const close = async () => {
    c.close();
    fake.close();
    await ctrl.close({ killEngine: true });
    reap(fx);
  };
  return { fx, ctrl, c, fake, rec: () => ctrl.claim.record, close };
}

const childOf = async (h, args) => {
  const r = await h.fake.call("child", { args });
  assert.ok(r.ok, JSON.stringify(r));
  strays.add(r.result.pid);
  return r.result.pid;
};
const memberPids = async (shim) => {
  const m = await shimRequest(shim, "members");
  assert.ok(m.ok, JSON.stringify(m));
  return m.members.map((x) => x.pid);
};

// A confirmed force stop, waited to its end (`stopped` or `uncertain`).
async function forceStop(h, c = h.c, ms = 20000) {
  const fs = await c.confirmed("force-stop");
  assert.equal(fs.outcome, "force-stop-fenced", JSON.stringify(fs));
  await until(() => ["stopped", "uncertain"].includes(h.ctrl.binding.state), ms, "the force stop to end");
  return fs.stop.id;
}

// ---- scope fixtures --------------------------------------------------------

test("K1: force stop kills a tool child that called setsid; stopped with a verified proof", NEEDS_SCOPE, async () => {
  const h = await live();
  try {
    // It also ignores TERM, so only the cgroup kill ends it.
    const child = await childOf(h, { setsid: true, ignoreTerm: true });
    assert.ok((await memberPids(h.rec().shim)).includes(child), "setsid leaves the process group, not the cgroup");
    const stop = await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    const { proof } = h.ctrl.stoppedProof;
    assert.equal(proof.stop, stop);
    assert.equal(proof.membershipComplete, true);
    assert.equal(proof.membershipEpoch, h.rec().invocationId);
    assert.ok(proof.members.some((m) => m.pid === child), "the escaped child is a listed member");
    assert.ok(h.ctrl.verifier.cohort(proof, h.ctrl.stoppedProof.effects, { binding: h.ctrl.binding, stop, now: new Date(), epoch: h.rec().invocationId }));
    assert.equal(alive(child), false);
  } finally {
    await h.close();
  }
});

test("K2: K1 on the process-group fallback ends uncertain, never stopped", async () => {
  const h = await live({ kind: "pgroup" });
  try {
    const child = await childOf(h, { setsid: true });
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.equal(h.ctrl.stoppedProof ?? null, null);
    const last = h.ctrl.evidence.stops.at(-1);
    assert.equal(last.outcome, "uncertain");
    assert.match(last.reason, /process-group fallback/);
    assert.ok(alive(child), "the setsid child left the group; a stopped claim here would have been false");
    assert.equal((await h.c.prompt("after an uncertain stop")).refusal, "fenced");
  } finally {
    await h.close();
  }
});

test("K3: SIGTERM acknowledged while a member lives: stopping until the kill phase, never stopped from TERM", NEEDS_SCOPE, async () => {
  const g = gate();
  const h = await live({ barrier: g.barrier });
  try {
    const child = await childOf(h, { ignoreTerm: true });
    g.hold("phase-kill");
    const fs = await h.c.confirmed("force-stop");
    assert.equal(fs.outcome, "force-stop-fenced");
    await g.waitHeld("phase-kill");
    assert.equal(h.ctrl.binding.state, "stopping");
    assert.notEqual(h.ctrl.stops.get(fs.stop.id).state, "stopped");
    assert.equal(h.rec().state, "stopping");
    assert.equal(h.rec().proof ?? null, null);
    assert.equal(h.rec().stop.phaseStarted, "kill");
    assert.ok(alive(child), "the member ignored TERM");
    assert.equal((await shimRequest(h.rec().shim, "events")).populated, 1);
    g.release("phase-kill");
    await until(() => h.ctrl.binding.state !== "stopping", 15000, "the kill phase");
    assert.equal(h.ctrl.binding.state, "stopped");
    assert.equal(alive(child), false);
    assert.ok(h.ctrl.stoppedProof.proof.members.some((m) => m.pid === child));
  } finally {
    g.release("phase-kill");
    await h.close();
  }
});

test("K4: two engines; force stop one; the other survives by independent observation", NEEDS_SCOPE, async () => {
  const a = await live();
  const b = await live();
  try {
    assert.notEqual(a.rec().unitName, b.rec().unitName);
    await forceStop(a);
    assert.equal(a.ctrl.binding.state, "stopped");
    const ev = await shimRequest(b.rec().shim, "events");
    assert.equal(ev.populated, 1, "b's own engine cgroup is still populated");
    const st = await b.fake.call("state");
    assert.ok(st.ok);
    assert.equal(st.result.pid, b.rec().engine.pid);
    assert.equal(systemctlShow(b.rec().unitName).invocationId, b.rec().invocationId);
    const p = await b.c.prompt("still here?");
    assert.equal(p.outcome, "admitted", JSON.stringify(p));
    await receiptState(b.c, p.receipt.id, "finished", 8000);
    assert.equal(b.ctrl.binding.state, "active");
  } finally {
    await a.close();
    await b.close();
  }
});

test("K5: a stop during a tool call leaves the effect uncertain, and it is shown", NEEDS_SCOPE, async () => {
  const h = await live();
  try {
    await h.fake.call("script", { steps: [{ tool: { id: "call-k5", name: "bash", args: { command: "touch x" }, hold: true } }, { text: "never", stop: "stop" }] });
    await h.fake.call("arm", { point: "tool:call-k5" });
    const p = await h.c.prompt("run a tool");
    assert.equal(p.outcome, "admitted");
    await h.fake.call("waitPaused", { point: "tool:call-k5" });
    await until(() => [...(h.ctrl.exec?.tools.values() ?? [])].some((t) => t.start && !t.end), 4000, "the tool start");
    const stop = await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped");
    const s = h.ctrl.stops.get(stop);
    assert.equal(s.externalEffects, "uncertain");
    const inv = h.ctrl.stoppedProof.effects.invocations;
    assert.equal(inv.length, 1);
    assert.equal(inv[0].disposition, "uncertain", "killing is never a rollback");
    await until(() => h.c.pushes.some((m) => m.kind === "stop" && m.stop.id === stop && m.stop.state === "stopped" && m.stop.externalEffects === "uncertain"), 4000, "the stop push");
  } finally {
    await h.close();
  }
});

// The freeze is shown two ways that don't depend on timing: engine's
// cgroup.freeze still reads 1 after the stop (the shim holds the scope), and
// every child the loop forked after its SIGTERM, which nothing else ends
// before the kill, is in the proof's list. Mutants r2-B5 (no freeze write,
// `frozen 1` answered) and r2-B5b (freeze written, not waited on) fail here.
test("K12: a member forking in a loop: the freeze stops it, enumeration is complete, populated 0 after cgroup.kill", NEEDS_SCOPE, async () => {
  const h = await live();
  try {
    const pidLog = join(h.fx.base, "fork-pids.log");
    const forker = await childOf(h, { forkLoop: true, ignoreTerm: true, pidLog });
    await until(async () => (await memberPids(h.rec().shim)).length >= 6, 4000, "the fork loop");
    const engineDir = join("/sys/fs/cgroup", (await shimRequest(h.rec().shim, "hello")).scope, "engine");
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    const { members } = h.ctrl.stoppedProof.proof;
    assert.ok(members.some((m) => m.pid === forker));
    assert.ok(members.length >= 2, `members ${members.length}`);
    for (const m of members) assert.equal(alive(m.pid), false, `member ${m.pid}`);
    assert.equal(readFileSync(join(engineDir, "cgroup.freeze"), "utf8").trim(), "1", "the freeze was written");
    const lines = readFileSync(pidLog, "utf8").split("\n").filter(Boolean);
    assert.ok(lines.includes("term"), "the fork loop got the TERM phase");
    const listed = new Set(members.map((m) => m.pid));
    const late = lines.slice(lines.indexOf("term") + 1).map(Number);
    assert.ok(late.length > 0, "the loop forked after its TERM");
    for (const pid of late) {
      strays.add(pid);
      assert.ok(listed.has(pid), `child ${pid} forked after TERM is in the proof's list`);
    }
    const ev = await shimRequest(h.rec().shim, "events");
    assert.equal(ev.populated, 0);
    assert.deepEqual(await memberPids(h.rec().shim), []);
  } finally {
    await h.close();
  }
});

test("K13: a member writing its pid into another cgroup is refused by the namespace; the kill is complete", NEEDS_SCOPE, async () => {
  const h = await live();
  const engine = h.rec().engine.pid;
  try {
    const hello = await shimRequest(h.rec().shim, "hello");
    const scope = join("/sys/fs/cgroup", hello.scope);
    for (const target of [join(scope, "supervisor", "cgroup.procs"), join(dirname(scope), "cgroup.procs")]) {
      const r = await h.fake.call("escape", { target });
      assert.ok(r.ok, JSON.stringify(r));
      assert.equal(r.result.escaped, false, `escape into ${target}`);
    }
    assert.ok((await memberPids(h.rec().shim)).includes(engine), "the engine is still in its cgroup");
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped");
    assert.equal(alive(engine), false);
  } finally {
    strays.add(engine);
    await h.close();
  }
});

test("K15: the shim gone, engine/cgroup.events unreadable, or the engine cgroup missing: evidence unavailable, not empty; uncertain", NEEDS_SCOPE, async () => {
  {
    const h = await live();
    const engine = h.rec().engine.pid;
    try {
      const hello = await shimRequest(h.rec().shim, "hello");
      process.kill(hello.shimPid, "SIGKILL");
      await until(() => !alive(hello.shimPid), 4000, "the shim to die");
      await forceStop(h);
      assert.equal(h.ctrl.binding.state, "uncertain");
      assert.match(h.ctrl.evidence.stops.at(-1).reason, /shim/);
      assert.ok(alive(engine), "no signal reached the engine without the shim's evidence");
    } finally {
      await h.close();
    }
  }
  {
    const h = await live();
    try {
      const hello = await shimRequest(h.rec().shim, "hello");
      chmodSync(join("/sys/fs/cgroup", hello.scope, "engine", "cgroup.events"), 0o000);
      assert.equal((await shimRequest(h.rec().shim, "events")).ok, false);
      await forceStop(h);
      assert.equal(h.ctrl.binding.state, "uncertain");
      assert.match(h.ctrl.evidence.stops.at(-1).reason, /cgroup\.events unreadable/);
      assert.equal(h.ctrl.stoppedProof ?? null, null);
    } finally {
      await h.close();
    }
  }
  // The `engine` cgroup path missing: its processes moved to a sibling and
  // the directory removed. Absent, never empty. Mutant r2-B6 (ENOENT read as
  // `populated 0`) fails here.
  {
    const h = await live();
    const engine = h.rec().engine.pid;
    try {
      const scope = join("/sys/fs/cgroup", (await shimRequest(h.rec().shim, "hello")).scope);
      const aside = join(scope, "aside");
      mkdirSync(aside);
      for (const pid of readFileSync(join(scope, "engine", "cgroup.procs"), "utf8").split("\n").filter(Boolean).map(Number)) {
        strays.add(pid);
        writeFileSync(join(aside, "cgroup.procs"), String(pid));
      }
      rmdirSync(join(scope, "engine"));
      for (const op of ["events", "members"]) {
        const r = await shimRequest(h.rec().shim, op);
        assert.equal(r.ok, false, `${op}: ${JSON.stringify(r)}`);
        assert.match(r.unavailable, /ENOENT/);
      }
      await forceStop(h);
      assert.equal(h.ctrl.binding.state, "uncertain");
      assert.match(h.ctrl.evidence.stops.at(-1).reason, /ENOENT/);
      assert.equal(h.ctrl.stoppedProof ?? null, null);
      assert.ok(alive(engine), "no signal reached the engine without the cgroup's evidence");
    } finally {
      await h.close();
    }
  }
});

// ---- controller death during a force stop ----------------------------------

function childFixture() {
  const fx = track(fixture());
  return { fx, fakeEnv: { FAKE_PI_CONTROL: join(fx.base, "fake.sock"), FAKE_PI_LOG: join(fx.base, "fake.log") } };
}

async function connectTo(socketPath, client = null) {
  const c = client ?? new ConversationClient({ socketPath });
  c.socketPath = socketPath;
  await c.connect();
  return c;
}

const stopping = (fx) => claimRecords(fx).map((r) => r.record).filter((r) => r?.state === "stopping" && r.stop);

// The controller dies at `barrier` during a confirmed force stop; a tool
// child that ignores TERM keeps the cohort populated past the TERM phase.
async function crashDuringStop(barrier) {
  const { fx, fakeEnv } = childFixture();
  const a = spawnController({ fx, launcher: "scope", fakeEnv, dieAt: { [barrier]: 1 } });
  const ra = await a.next((m) => m.ready || m.error);
  assert.ok(ra.ready, JSON.stringify(ra));
  const c = await connectTo(ra.socketPath);
  assert.equal((await c.takeover()).outcome, "transferred");
  const fake = new ControlClient(fakeEnv.FAKE_PI_CONTROL);
  await fake.connect();
  const child = (await fake.call("child", { args: { ignoreTerm: true } })).result.pid;
  strays.add(child);
  fake.close();
  void c.confirmed("force-stop");
  await a.next((m) => m.dying === barrier, 15000);
  await a.exited;
  const recs = stopping(fx);
  assert.ok(recs.length > 0, "the stop was recorded before any signal");
  const last = recs.at(-1);
  return { fx, fakeEnv, c, child, stopId: last.stop.id, last, engine: last.engine.pid };
}

async function restartAndResume({ fx, fakeEnv, c, stopId }) {
  const b = spawnController({ fx, launcher: "scope", fakeEnv });
  const rb = await b.next((m) => m.ready || m.error, 15000);
  assert.ok(rb.ready, JSON.stringify(rb));
  assert.equal(rb.started.launched, false);
  assert.equal(rb.started.classified.state, "stopping");
  await connectTo(rb.socketPath, c);
  assert.ok(await c.waitFor(() => ["stopped", "uncertain"].includes(c.binding?.state), 20000), `binding ${c.binding?.state}`);
  b.send("evidence");
  const ev = (await b.next((m) => m.evidence !== undefined)).evidence;
  b.send("proof");
  const proof = (await b.next((m) => m.proof !== undefined)).proof;
  const recs = claimRecords(fx).map((r) => r.record).filter((r) => r?.stop);
  assert.ok(recs.every((r) => r.stop.id === stopId), "the restart continues the recorded stop; no new stop");
  return { b, ev, proof };
}

for (const [id, barrier, title] of [
  ["K10", "phase-kill", "controller killed between the TERM and kill phases"],
  ["K11", "force-stop-recorded", "controller killed after the confirmation is recorded, before TERM"],
]) {
  test(`${id}: ${title}: restart checks the invocation ID and re-runs from TERM for the same stop`, NEEDS_SCOPE, async () => {
    const crashed = await crashDuringStop(barrier);
    const { fx, c, child, stopId, last, engine } = crashed;
    assert.equal(last.stop.phaseStarted, barrier === "phase-kill" ? "kill" : null);
    assert.ok(!claimRecords(fx).some((r) => r.record?.state === "stopped"), "nothing recorded as stopped before the restart");
    assert.ok(alive(child), "the member is alive across the crash");
    if (barrier === "force-stop-recorded") assert.ok(alive(engine), "no TERM was sent before the crash");
    const { b, ev, proof } = await restartAndResume(crashed);
    try {
      assert.equal(c.binding.state, "stopped", JSON.stringify(ev.stops));
      const done = ev.stops.at(-1);
      assert.equal(done.stop, stopId);
      assert.equal(done.resumed, true);
      assert.equal(done.outcome, "stopped");
      assert.equal(proof.stop, stopId);
      assert.ok(proof.members.some((m) => m.pid === child), "the member observed at the freeze is listed");
      if (barrier === "phase-kill") assert.ok(!proof.members.some((m) => m.pid === engine), "the engine ended at TERM before the crash; it isn't listed as killed");
      assert.equal(alive(child), false);
    } finally {
      c.close();
      b.send("close");
      await b.exited;
      reap(fx);
    }
  });
}

test("K14: a unit with the recorded name but another invocation ID: evidence unavailable, no signals, uncertain", NEEDS_SCOPE, async () => {
  const { fx, fakeEnv } = childFixture();
  const a = spawnController({ fx, launcher: "scope", fakeEnv });
  const ra = await a.next((m) => m.ready || m.error);
  assert.ok(ra.ready, JSON.stringify(ra));
  const rec = claimRecords(fx).map((r) => r.record).filter((r) => r?.invocationId).at(-1);
  const unit = rec.unitName;
  a.proc.kill("SIGKILL");
  await a.exited;
  // The shim ignores TERM by design, so the scope goes with SIGKILL.
  spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${unit}.scope`], { stdio: "ignore", timeout: 10000 });
  await until(() => systemctlShow(unit)?.loadState === "not-found", 10000, "the original scope to go");
  let impostor = null;
  await until(() => {
    if (impostor && systemctlShow(unit)?.activeState === "active") return true;
    if (!impostor || impostor.exitCode !== null) {
      impostor = spawn("systemd-run", ["--user", "--scope", `--unit=${unit}`, "--quiet", "--", "sleep", "300"], { stdio: "ignore", detached: true });
      impostor.unref();
    }
    return false;
  }, 10000, "the impostor unit");
  strays.add(impostor.pid);
  const theirs = systemctlShow(unit).invocationId;
  assert.notEqual(theirs, rec.invocationId);
  const b = spawnController({ fx, launcher: "scope", fakeEnv });
  try {
    const rb = await b.next((m) => m.ready || m.error, 15000);
    assert.ok(rb.ready, JSON.stringify(rb));
    assert.equal(rb.started.classified.state, "uncertain");
    const c = await connectTo(rb.socketPath);
    assert.equal((await c.confirmed("acquire-recovery-control")).outcome, "recovery-control-acquired");
    const fs = await c.confirmed("force-stop");
    assert.equal(fs.outcome, "force-stop-fenced", JSON.stringify(fs));
    assert.ok(await c.waitFor(() => c.binding?.state === "uncertain" && c.pushes.some((m) => m.kind === "stop" && m.stop.id === fs.stop.id && m.stop.state === "uncertain"), 15000));
    b.send("evidence");
    const ev = (await b.next((m) => m.evidence !== undefined)).evidence;
    assert.match(ev.stops.at(-1).reason, /invocation ID mismatch/);
    assert.ok(alive(impostor.pid), "the other cohort got no signal");
    assert.equal(systemctlShow(unit).invocationId, theirs);
    c.close();
  } finally {
    b.send("close");
    await b.exited;
    spawnSync("systemctl", ["--user", "stop", `${unit}.scope`], { stdio: "ignore", timeout: 10000 });
    reap(fx);
  }
});

// ---- in-process fake: recovery, launch and the K9 fence ---------------------

async function provenStop(h, c = h.client) {
  const fs = await c.confirmed("force-stop");
  assert.equal(fs.outcome, "force-stop-fenced", JSON.stringify(fs));
  await until(() => h.ctrl.binding.state === "stopped", 4000, "the proven stop");
  return fs.stop.id;
}

function pinRootCopy(fx) {
  const root = join(fx.base, "pins");
  mkdirSync(join(root, "node_modules"), { recursive: true });
  copyFileSync(join(REPO, "package-lock.json"), join(root, "package-lock.json"));
  copyFileSync(join(REPO, "node_modules", ".package-lock.json"), join(root, "node_modules", ".package-lock.json"));
  return root;
}

test("K6: recover without proof, without confirmation, or with changed pins is refused", async () => {
  {
    const h = await started();
    try {
      const notYet = await h.client.confirmed("recover", { stop: newId("stop") });
      assert.equal(notYet.refusal, "stop-proof", "no stop at all");
    } finally {
      await h.close();
    }
  }
  {
    const h = await started({ launcher: new FakeLauncher({ stopOutcome: "unavailable" }) });
    try {
      const fs = await h.client.confirmed("force-stop");
      await until(() => h.ctrl.binding.state === "uncertain", 4000, "the uncertain stop");
      assert.equal((await h.client.confirmed("recover", { stop: fs.stop.id })).refusal, "stop-proof", "an uncertain stop is no proof");
    } finally {
      await h.close();
    }
  }
  {
    const fx = track(fixture());
    const pinRoot = pinRootCopy(fx);
    const h = await started({ fx, pinRoot });
    try {
      const stop = await provenStop(h);
      const missing = await h.client.request("recover", { stop });
      assert.equal(missing.refusal, "malformed", "no confirmation field");
      const unknown = await h.client.request("recover", { stop, confirmation: newId("confirmation") });
      assert.equal(unknown.refusal, "confirmation");
      const issued = await h.client.request("issue-confirmation", { operationToConfirm: "force-stop" });
      const id = issued.data.confirmation.id;
      await h.client.request("answer-confirmation", { confirmation: id, answer: "confirm" });
      assert.equal((await h.client.request("recover", { stop, confirmation: id })).refusal, "confirmation", "a confirmation for another operation");
      const lock = join(pinRoot, "package-lock.json");
      const original = readFileSync(lock, "utf8");
      const changed = JSON.parse(original);
      changed.packages["node_modules/@earendil-works/pi-coding-agent"].version = "0.0.0";
      writeFileSync(lock, JSON.stringify(changed));
      assert.equal((await h.client.confirmed("recover", { stop })).refusal, ENGINE_PIN_MISMATCH);
      writeFileSync(lock, original);
      const again = await h.client.request("issue-confirmation", { operationToConfirm: "recover" });
      const once = again.data.confirmation.id;
      await h.client.request("answer-confirmation", { confirmation: once, answer: "confirm" });
      assert.equal((await h.client.request("recover", { stop, confirmation: once })).outcome, "recovery-eligible", "the same request with the pins restored");
      assert.equal(h.ctrl.confirmations.get(once).state, "consumed");
      assert.equal((await h.client.request("recover", { stop, confirmation: once })).refusal, "confirmation", "a confirmation is single-use");
    } finally {
      await h.close();
    }
  }
});

test("K7: recover after proof, then launch: new claim and execution, generation +1, same leaf; the cancelled prompt is not replayed", async () => {
  const h = await started();
  try {
    h.engine.script([{ pause: "p1" }, { text: "never", stop: "stop" }]);
    h.engine.arm("p1");
    const p = await h.client.prompt("cancelled by the stop");
    assert.equal(p.outcome, "admitted");
    await receiptState(h.client, p.receipt.id, "working");
    await h.engine.waitPaused("p1");
    const before = { claim: h.ctrl.claim.claimId, execution: h.ctrl.binding.execution, generation: h.ctrl.binding.controllerGeneration };
    const stop = await provenStop(h);
    const leafAtProof = h.ctrl.claim.record.leafAtProof;
    const rec = await h.client.confirmed("recover", { stop });
    assert.equal(rec.outcome, "recovery-eligible", JSON.stringify(rec));
    assert.equal(rec.data.generation, before.generation + 1);
    const engines = h.launcher.engines.length;
    const r = await h.ctrl.launch(rec.data.eligibility);
    assert.equal(h.launcher.engines.length, engines + 1);
    assert.notEqual(r.claim, before.claim);
    assert.equal(r.claim, rec.data.claim);
    assert.notEqual(h.ctrl.binding.execution, before.execution, "a new execution (K7's incarnation)");
    assert.equal(h.ctrl.binding.controllerGeneration, before.generation + 1);
    assert.equal(h.ctrl.binding.state, "active");
    assert.equal(h.ctrl.claim.record.leaf, leafAtProof);
    assert.equal(h.ctrl.claim.record.prior.stop, stop);
    const fresh = h.launcher.last;
    assert.equal(fresh.leaf, leafAtProof, "the new engine loaded the leaf at proof");
    assert.ok(!fresh.commands.some((x) => x.type === "prompt"));
    assert.ok(!fresh.bytes().toString().includes("cancelled by the stop"));
  } finally {
    await h.close();
  }
});

test("K8: an engine that loads another leaf on resume is refused before admission; it stays claimed until a proven stop", async () => {
  const h = await started();
  try {
    const stop = await provenStop(h);
    const rec = await h.client.confirmed("recover", { stop });
    assert.equal(rec.outcome, "recovery-eligible");
    h.launcher.opts.leaf = "ffff0000";
    await h.ctrl.launch(rec.data.eligibility);
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.ok(await h.client.waitFor(() => h.client.binding?.id === h.ctrl.binding.id && h.client.binding.state === "uncertain"), "the client sees the new binding");
    assert.equal(h.ctrl.binding.admission, "closed");
    assert.ok(h.ctrl.evidence.uncertain.some((u) => u.reason === "loaded-session" && /another leaf/.test(u.detail)));
    assert.equal(h.ctrl.claim.claimId, rec.data.claim);
    assert.notEqual(h.ctrl.claim.record.state, "active", "never promoted");
    assert.ok(h.ctrl.claim.record.engine?.pid, "the engine stays recorded under the claim");
    const stops = h.launcher.stops;
    assert.equal(h.launcher.last.ended ?? false, false, "nothing killed it outside a force stop");
    assert.equal((await h.client.confirmed("acquire-recovery-control")).outcome, "recovery-control-acquired");
    assert.equal((await h.client.prompt("admitted?")).refusal, "preflight", "the loaded-session check never passed");
    await provenStop(h);
    assert.equal(h.launcher.stops, stops + 1);
    assert.equal(h.ctrl.claim.record.state, "stopped");
  } finally {
    await h.close();
  }
});

test("K9: an interrupt that never settles stays uncertain; force stop stays available; takeover is refused while fenced", async () => {
  const h = await started({ clients: 2 });
  try {
    const [c1, c2] = h.clients;
    h.engine.script([{ hang: true }]);
    const p = await c1.prompt("hangs");
    assert.equal(p.outcome, "admitted");
    await receiptState(c1, p.receipt.id, "working");
    const r = await c1.interrupt();
    const stop = r.stop?.id ?? h.ctrl.binding.stop;
    await until(() => h.ctrl.stops.get(stop)?.state === "uncertain", 10000, "the interrupt to end uncertain");
    assert.equal(h.ctrl.binding.admission, "closed");
    assert.ok(!h.ctrl.events.some((e) => e.type === "reconciled"));
    assert.equal((await c1.prompt("again")).refusal, "fenced");
    await until(() => c2.binding?.controllerGeneration === h.ctrl.binding.controllerGeneration, 2000, "c2 synced");
    assert.equal((await c2.takeover()).refusal, "fenced");
    await provenStop(h, c1);
    assert.equal(h.ctrl.binding.state, "stopped");
  } finally {
    await h.close();
  }
});

test("K16: a claim from another machine ID refuses foreign-host; no boot proof is issued", async () => {
  const fx = track(fixture());
  const verifier = new SpyVerifier();
  const { ctrl } = controllerFor(fx, { verifier });
  const foreign = new ClaimStore({ root: fx.claimRoot, host: { machineId: () => "f".repeat(32), bootId: () => "00000000-0000-4000-8000-000000000000" } });
  await foreign.acquire(ctrl.seatK, ctrl.sessionK, {
    bindingId: "binding-1", harness: "pi", conversation: ctrl.conversation, branch: "main", leaf: "b2c3d4e5",
    pins: { engineVersion: "0.85.1", enginePin: "x", argvDigest: "y" },
    owner: { pid: 1, start: "1", boot: "00000000-0000-4000-8000-000000000000", incarnation: "f".repeat(32) }, generation: 1,
  });
  await assert.rejects(ctrl.start(), { code: FOREIGN_HOST });
  assert.deepEqual(verifier.posted, [], "no proof of any kind was posted");
});

test("K17: two launcher calls with one eligibility record: one launch, the other refuses, no second engine", async () => {
  const h = await started();
  try {
    const stop = await provenStop(h);
    const rec = await h.client.confirmed("recover", { stop });
    const engines = h.launcher.engines.length;
    const [x, y] = await Promise.allSettled([h.ctrl.launch(rec.data.eligibility), h.ctrl.launch(rec.data.eligibility)]);
    const ok = [x, y].filter((v) => v.status === "fulfilled");
    const no = [x, y].filter((v) => v.status === "rejected");
    assert.equal(ok.length, 1);
    assert.equal(no.length, 1);
    assert.equal(no[0].reason.code, ELIGIBILITY);
    assert.equal(h.launcher.engines.length, engines + 1);
    await assert.rejects(h.ctrl.launch(rec.data.eligibility), { code: ELIGIBILITY });
    assert.equal(h.launcher.engines.length, engines + 1);
  } finally {
    await h.close();
  }
});

test("K18: the leaf changes after eligibility: launch refused; the reservation stays until released with proof", async () => {
  const h = await started();
  try {
    const stop = await provenStop(h);
    const rec = await h.client.confirmed("recover", { stop });
    const leaf = h.ctrl.claim.record.leafAtProof;
    appendFileSync(h.fx.sessionFile, JSON.stringify(assistantEntry("c3d4e5f6", leaf, "written after eligibility", 9)) + "\n");
    const engines = h.launcher.engines.length;
    await assert.rejects(h.ctrl.launch(rec.data.eligibility), { code: "target" });
    assert.equal(h.launcher.engines.length, engines, "no engine started");
    for (const key of [h.ctrl.seatK, h.ctrl.sessionK]) {
      const head = h.ctrl.store.head(key);
      assert.equal(head.record.claimId, rec.data.claim);
      assert.equal(head.record.state, "reserved");
      assert.equal(head.record.spawnMarker, false);
    }
    assert.equal((await h.ctrl.release(rec.data.eligibility)).released, rec.data.claim);
    for (const key of [h.ctrl.seatK, h.ctrl.sessionK]) {
      const head = h.ctrl.store.head(key);
      assert.equal(head.record.claimId, rec.data.claim);
      assert.equal(head.record.state, "stopped");
      assert.equal(head.record.proof.kind, "no-unit");
    }
    await assert.rejects(h.ctrl.launch(rec.data.eligibility), { code: ELIGIBILITY });
  } finally {
    await h.close();
  }
});

test("K19: a scope launched with only the engine environment still reaches the user manager; the engine sees no other names", NEEDS_SCOPE, async () => {
  // A real launch passes engineEnv(), which names neither variable systemd-run
  // needs; ScopeLauncher.MANAGER_ENV adds them (slice 1 S5, #1522).
  if (!ScopeLauncher.MANAGER_ENV.some((k) => typeof process.env[k] === "string")) return;
  const fx = track(fixture());
  mkdirSync(fx.socketDir, { recursive: true });
  const unitName = unitNameFor(newClaimId());
  const env = engineEnv([]);
  for (const k of ScopeLauncher.MANAGER_ENV) assert.equal(k in env, false, k);
  const socketPath = join(fx.socketDir, "k19.sock");
  const proc = await new ScopeLauncher().launch({ unitName, socketPath, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env });
  try {
    const names = readFileSync(`/proc/${proc.pid}/environ`, "utf8").split("\0").filter(Boolean).map((e) => e.slice(0, e.indexOf("=")));
    // systemd-run sets INVOCATION_ID; the shim's /bin/sh sets PWD, and SHLVL
    // and _ when it is bash. None comes from the controller's environment.
    const set = ["INVOCATION_ID", "PWD", "OLDPWD", "SHLVL", "_"];
    for (const k of names) assert.ok(ENGINE_ENV.includes(k) || ScopeLauncher.MANAGER_ENV.includes(k) || set.includes(k), k);
    for (const k of ScopeLauncher.MANAGER_ENV) if (typeof process.env[k] === "string") assert.ok(names.includes(k), k);
  } finally {
    // A failed kill or release still ends the scope here, not in the after hook.
    const killed = await shimRequest(socketPath, "kill", { timeoutMs: 5000 }, 8000);
    const released = killed.ok ? await shimRequest(socketPath, "release") : killed;
    if (!released.ok) spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${unitName}.scope`], { stdio: "ignore", timeout: 5000 });
    assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
    assert.equal(killed.ok, true, JSON.stringify(killed));
    assert.equal(released.ok, true, JSON.stringify(released));
  }
});

// ---- scope release after a force stop (#1536) ------------------------------

const unitGone = (unit) => systemdUnits.lookup({ unitName: unit }).state === "absent";
const unitActive = (unit) => systemctlShow(unit)?.activeState === "active";

test("R1: after a proven force stop the controller releases the scope; the shim and the unit are gone", NEEDS_SCOPE, async () => {
  const h = await live();
  const unit = h.rec().unitName, shim = h.rec().shim;
  try {
    const shimPid = (await shimRequest(shim, "hello")).shimPid;
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unitName, r.claim]), [["force-stop", "released", unit, h.ctrl.claim.claimId]]);
    // Observers get the same entry (Darkwing note 1 on #1536).
    await until(() => h.c.pushes.some((m) => m.kind === "evidence" && m.evidence.kind === "release"), 4000, "the pushed release");
    assert.deepEqual(h.c.pushes.filter((m) => m.kind === "evidence" && m.evidence.kind === "release").map((m) => m.evidence), h.ctrl.evidence.releases);
    await until(() => unitGone(unit), 8000, "the scope to go");
    assert.equal(alive(shimPid), false);
    assert.equal((await shimRequest(shim, "hello")).ok, false);
  } finally {
    await h.close();
  }
});

test("R2: close releases the scope of a binding proven stopped, once, when the stop's own release hasn't run", NEEDS_SCOPE, async () => {
  const g = gate();
  const h = await live({ barrier: g.barrier });
  const unit = h.rec().unitName;
  try {
    g.hold("scope-release");
    await forceStop(h);
    await g.waitHeld("scope-release");
    assert.equal(h.ctrl.binding.state, "stopped");
    assert.ok(unitActive(unit), "held before the stop's release: the scope is still up");
    assert.deepEqual(h.ctrl.evidence.releases, []);
    await h.ctrl.close();
    assert.ok(unitGone(unit), JSON.stringify(systemctlShow(unit)));
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unitName, r.unit]), [["close", "released", unit, "absent"]]);
    g.release("scope-release");
    await tick(100);
    assert.equal(h.ctrl.evidence.releases.length, 1, "the stop's own release reuses the close's");
  } finally {
    g.release("scope-release");
    h.c.close();
    h.fake.close();
    reap(h.fx);
  }
});

// The real stop runs, so the cohort is empty and the shim would accept
// `release`; only the outcome says unavailable.
class UnavailableAfterStop extends ScopeLauncher {
  async forceStop(a) {
    await forceStopCohort(a);
    return { outcome: "unavailable", reason: "R3: evidence withheld" };
  }
}
class RejectingVerifier extends FixtureVerifier {
  constructor() {
    super({ authorities: [AUTHORITY] });
  }
  cohort() {
    return false;
  }
}

test("R3: no release after an unavailable stop or an unverified proof: the scope and its shim stay as evidence", NEEDS_SCOPE, async () => {
  for (const [what, opts, reason] of [
    ["unavailable", { launcher: new UnavailableAfterStop() }, /R3: evidence withheld/],
    ["unverified", { verifier: new RejectingVerifier() }, /proof not verified/],
  ]) {
    const h = await live(opts);
    const unit = h.rec().unitName, shim = h.rec().shim;
    try {
      await forceStop(h);
      assert.equal(h.ctrl.binding.state, "uncertain", what);
      assert.match(h.ctrl.evidence.stops.at(-1).reason, reason);
      await h.ctrl.close();
      await tick(200);
      assert.deepEqual(h.ctrl.evidence.releases, [], what);
      assert.ok(unitActive(unit), `${what}: the scope stays up`);
      const hello = await shimRequest(shim, "hello");
      assert.equal(hello.ok, true, `${what}: ${JSON.stringify(hello)}`);
      assert.equal(hello.invocationId, h.rec().invocationId);
    } finally {
      h.c.close();
      h.fake.close();
      reap(h.fx);
    }
  }
});

// A normal engine exit leaves the shim and the scope: the shim exits only on
// `release`. The controller doesn't release then: the binding is uncertain
// and the scope is what a force stop reads. The proven stop releases it.
// ---- engine exit (row 52, CHAT-01 `engine-exit`) ----------------------------

// A confirmation issued and confirmed now, not yet used.
async function confirmation(c, op) {
  const issued = await c.request("issue-confirmation", { operationToConfirm: op });
  assert.equal(issued.outcome, "confirmation-issued", JSON.stringify(issued));
  const answered = await c.request("answer-confirmation", { confirmation: issued.data.confirmation.id, answer: "confirm" });
  assert.equal(answered.outcome, "confirmation-confirmed", JSON.stringify(answered));
  return issued.data.confirmation.id;
}
const exitStops = (h) => [...h.ctrl.stops.values()].filter((s) => s.mode === "engine-exit");

test("E1: a natural engine exit with no member left: an engine-exit stop, stopped on a proof naming the engine, and the scope released", NEEDS_SCOPE, async () => {
  const g = gate();
  const h = await live({ barrier: g.barrier });
  const unit = h.rec().unitName, shim = h.rec().shim, engine = h.rec().engine;
  try {
    const before = h.ctrl.binding.stop;
    const early = await confirmation(h.c, "recover");
    g.hold("scope-release");
    assert.equal((await h.fake.call("exit")).ok, true);
    await g.waitHeld("scope-release");
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    const [stop] = exitStops(h);
    assert.equal(h.ctrl.binding.stop, stop.id);
    assert.deepEqual([stop.state, stop.request, stop.supersedes], ["stopped", null, before]);
    // The proof names the engine by the shim's wait: PID, start ticks, boot
    // and reap time, read while the shim still answers.
    const exit = (await shimRequest(shim, "hello")).engineExit;
    assert.equal(exit.code, 0);
    assert.deepEqual(h.ctrl.stoppedProof.proof.members, [{ pid: engine.pid, boot: exit.boot, startTicks: Number(engine.start), terminatedAt: exit.at }]);
    assert.deepEqual(stoppedOnProof(h.fx).map((r) => [r.stop.id, r.stop.record.mode]), [[stop.id, "engine-exit"], [stop.id, "engine-exit"]]);
    g.release("scope-release");
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unit ?? null]), [["engine-exit", "released", "absent"]]);
    assert.ok(unitGone(unit));
    // Recover takes its own confirmation, bound to this stop (K6); one issued
    // before the engine exited names the old stop.
    assert.equal((await h.c.request("recover", { stop: stop.id, confirmation: early })).refusal, "confirmation");
    const rec = await h.c.confirmed("recover", { stop: stop.id });
    assert.equal(rec.outcome, "recovery-eligible", JSON.stringify(rec));
    await h.ctrl.release(rec.data.eligibility);
  } finally {
    g.release("scope-release");
    await h.close();
  }
});

test("E2: an engine exit with a tool child alive: the stop ends uncertain and nothing is released; a force stop then ends it", NEEDS_SCOPE, async () => {
  const h = await live();
  const unit = h.rec().unitName, shim = h.rec().shim;
  try {
    const child = await childOf(h, {});
    const early = await confirmation(h.c, "force-stop");
    assert.equal((await h.fake.call("exit")).ok, true);
    await until(() => exitStops(h)[0]?.state === "uncertain", 8000, "the engine-exit stop to end");
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.deepEqual(h.ctrl.evidence.stops.map((s) => [s.mode, s.outcome]), [["engine-exit", "uncertain"]]);
    assert.deepEqual(h.ctrl.evidence.releases, []);
    assert.ok(unitActive(unit));
    assert.ok(alive(child));
    assert.deepEqual(await memberPids(shim), [child]);
    assert.ok(!h.ctrl.stoppedProof);
    // A force-stop confirmation issued before the engine exited is stale
    // (H17): it names the stop the engine-exit stop superseded.
    assert.equal((await h.c.request("force-stop", { confirmation: early })).refusal, "confirmation");
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    assert.equal(h.ctrl.stops.get(h.ctrl.binding.stop).supersedes, exitStops(h)[0].id);
    await until(() => !alive(child), 4000, "the child to die");
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome]), [["force-stop", "released"]]);
    await until(() => unitGone(unit), 8000, "the scope to go");
  } finally {
    await h.close();
  }
});

test("E3: an engine exit and a force stop take one escalation slot, in either order", NEEDS_SCOPE, async () => {
  {
    const g = gate();
    const h = await live({ barrier: g.barrier });
    try {
      g.hold("engine-exit-recorded");
      assert.equal((await h.fake.call("exit")).ok, true);
      await g.waitHeld("engine-exit-recorded");
      assert.equal(h.ctrl.binding.state, "stopping");
      assert.equal((await h.c.confirmed("force-stop")).refusal, "fenced");
      g.release("engine-exit-recorded");
      await until(() => h.ctrl.binding.state === "stopped", 8000, "the engine-exit proof");
      assert.deepEqual(h.ctrl.evidence.stops.map((s) => [s.mode, s.outcome]), [["engine-exit", "stopped"]]);
    } finally {
      g.release("engine-exit-recorded");
      await h.close();
    }
  }
  {
    const g = gate();
    const h = await live({ barrier: g.barrier });
    try {
      // A force stop already running: the EOF its kill causes starts nothing.
      g.hold("force-stop-recorded");
      const fs = await h.c.confirmed("force-stop");
      assert.equal(fs.outcome, "force-stop-fenced", JSON.stringify(fs));
      await g.waitHeld("force-stop-recorded");
      assert.equal((await h.fake.call("exit")).ok, true);
      await tick(300);
      assert.deepEqual(exitStops(h), []);
      g.release("force-stop-recorded");
      await until(() => ["stopped", "uncertain"].includes(h.ctrl.binding.state), 20000, "the force stop to end");
      assert.equal(h.ctrl.binding.stop, fs.stop.id);
      assert.deepEqual(exitStops(h), []);
      assert.deepEqual(h.ctrl.evidence.stops.map((s) => s.mode), ["force-stop"]);
    } finally {
      g.release("force-stop-recorded");
      await h.close();
    }
  }
});

test("E4: a shim that doesn't answer: the engine-exit stop misses its deadline, ends uncertain and frees the slot", NEEDS_SCOPE, async () => {
  const h = await live({ timeouts: { ...FAST, exitSettle: 200, exitProof: 500 } });
  const unit = h.rec().unitName, shim = h.rec().shim;
  const shimPid = (await shimRequest(shim, "hello")).shimPid;
  try {
    process.kill(shimPid, "SIGSTOP");
    assert.equal((await h.fake.call("exit")).ok, true);
    await until(() => exitStops(h)[0]?.state === "uncertain", 8000, "the engine-exit deadline");
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.match(h.ctrl.evidence.stops.at(-1).reason, /deadline/);
    assert.equal(h.ctrl.escalating, null);
    process.kill(shimPid, "SIGCONT");
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    // The force stop's proof is unchanged by row 52: the engine had already
    // left the cohort, so it lists no member.
    assert.deepEqual(h.ctrl.stoppedProof.proof.members, []);
    await until(() => unitGone(unit), 8000, "the scope to go");
  } finally {
    try {
      process.kill(shimPid, "SIGCONT");
    } catch {
      // gone
    }
    await h.close();
  }
});

test("E5: an engine exit after a force stop ended uncertain starts no engine-exit stop; the force stop stays current", NEEDS_SCOPE, async () => {
  const h = await live();
  const advance = h.ctrl.store.advance.bind(h.ctrl.store);
  let failed = false;
  // The force stop's `stopping` claim write fails once, so it ends uncertain
  // before it closes the execution and the engine's EOF still arrives.
  h.ctrl.store.advance = async (claim, patch) => {
    if (!failed && patch?.state === "stopping") {
      failed = true;
      throw Object.assign(new Error("injected"), { code: "E5-INJECTED" });
    }
    return advance(claim, patch);
  };
  try {
    const stop = await forceStop(h);
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.match(h.ctrl.evidence.stops.at(-1).reason, /E5-INJECTED/);
    assert.equal(h.ctrl.escalating, null);
    assert.equal((await h.fake.call("exit")).ok, true);
    await until(() => h.ctrl.evidence.uncertain.some((u) => u.reason === "eof"), 4000, "the EOF");
    await tick(300);
    assert.deepEqual(exitStops(h), []);
    assert.equal(h.ctrl.binding.stop, stop);
    assert.equal(h.ctrl.binding.state, "uncertain");
    assert.deepEqual(h.ctrl.evidence.stops.map((s) => s.mode), ["force-stop"]);
  } finally {
    h.ctrl.store.advance = advance;
    await h.close();
  }
});

test("E6: the engine-exit observation refuses a reaped process that isn't the recorded engine: another PID or start ticks", NEEDS_SCOPE, async () => {
  const g = gate();
  const h = await live({ barrier: g.barrier });
  try {
    g.hold("engine-exit-recorded");
    assert.equal((await h.fake.call("exit")).ok, true);
    await g.waitHeld("engine-exit-recorded");
    const rec = h.rec(), engine = rec.engine;
    const observe = (over) => engineExitCohort({ kind: "scope", unitName: rec.unitName, invocationId: rec.invocationId, shimSocket: rec.shim, pid: engine.pid, start: engine.start, settleMs: 200, ...over });
    for (const over of [{ pid: engine.pid + 1 }, { start: String(Number(engine.start) + 1) }]) {
      const r = await observe(over);
      assert.deepEqual([r.outcome, r.reason], ["unavailable", "the shim reaped a process that isn't the recorded engine"], JSON.stringify(over));
    }
    assert.equal((await observe({})).outcome, "proven");
    g.release("engine-exit-recorded");
    await until(() => h.ctrl.binding.state === "stopped", 8000, "the engine-exit proof");
  } finally {
    g.release("engine-exit-recorded");
    await h.close();
  }
});

test("R5: the harness finds a running shim and reaps it, and never signals a shim outside a mosaic-chat- scope (Filbert N1 and N2 on #1533)", NEEDS_SCOPE, async () => {
  const fx = track(fixture());
  mkdirSync(fx.socketDir, { recursive: true });
  {
    const unitName = unitNameFor(newClaimId());
    const socketPath = join(fx.socketDir, "r5a.sock");
    const proc = await new ScopeLauncher().launch({ unitName, socketPath, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
    const hello = await shimRequest(socketPath, "hello");
    assert.deepEqual(liveShims(fx.base), [{ pid: hello.shimPid, socket: socketPath, unit: unitName }]);
    reap(fx);
    assert.deepEqual(await shimsGone(fx.base), []);
    assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
    await until(() => unitGone(unitName), 8000, "the reaped scope to go");
  }
  {
    const unitName = `r5-not-mosaic-${newClaimId().slice(0, 12)}`;
    const socketPath = join(fx.socketDir, "r5b.sock");
    const proc = await new ScopeLauncher().launch({ unitName, socketPath, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
    let released = null;
    try {
      const hello = await shimRequest(socketPath, "hello");
      assert.deepEqual(killShims(fx.base).map((s) => [s.pid, s.unit]), [[hello.shimPid, unitName]]);
      await tick(200);
      assert.ok(alive(hello.shimPid), "the refused shim got no signal");
      assert.ok(unitActive(unitName), "the refused scope got no signal");
    } finally {
      const killed = await shimRequest(socketPath, "kill", { timeoutMs: 5000 }, 8000);
      released = killed.ok ? await shimRequest(socketPath, "release") : killed;
      if (!released.ok) spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${unitName}.scope`], { stdio: "ignore", timeout: 5000 });
      assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
    }
    assert.equal(released.ok, true, JSON.stringify(released));
  }
});

test("R6: releaseCohort releases nothing while the engine runs, for another invocation ID, or for a cohort without a scope", NEEDS_SCOPE, async () => {
  const fx = track(fixture());
  mkdirSync(fx.socketDir, { recursive: true });
  const unitName = unitNameFor(newClaimId());
  const shimSocket = join(fx.socketDir, "r6.sock");
  const proc = await new ScopeLauncher().launch({ unitName, socketPath: shimSocket, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
  try {
    // The right scope while its engine runs: the shim refuses, and that is
    // never recorded as released (Filbert N1 on #1536).
    const refused = await releaseCohort({ kind: "scope", unitName, invocationId: proc.invocationId, shimSocket });
    assert.equal(refused.outcome, "unavailable", JSON.stringify(refused));
    assert.equal((await shimRequest(shimSocket, "hello")).ok, true, "the shim refused the release while its engine ran");
    assert.ok(unitActive(unitName));
    const killed = await shimRequest(shimSocket, "kill", { timeoutMs: 5000 }, 8000);
    assert.equal(killed.ok, true, JSON.stringify(killed));
    for (const [kind, invocationId] of [["scope", "another-invocation"], ["scope", null], ["pgroup", proc.invocationId], ["fake", proc.invocationId]]) {
      const r = await releaseCohort({ kind, unitName, invocationId, shimSocket });
      assert.equal(r.outcome, "unavailable", `${kind} ${invocationId}: ${JSON.stringify(r)}`);
      assert.equal((await shimRequest(shimSocket, "hello")).ok, true, `${kind} ${invocationId}: the shim got no release`);
    }
    assert.ok(unitActive(unitName));
    assert.deepEqual(await releaseCohort({ kind: "scope", unitName, invocationId: proc.invocationId, shimSocket }), { outcome: "released", unit: "absent" });
    assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
  } finally {
    reap(fx);
  }
});

// ---- crash window, retry and close after a proven stop (#1537) -------------

// A controller in this process on the scope launcher, started but not
// asserted active: a restart may find the pair stopped.
function scopeController(fx, { host, verifier = new FixtureVerifier({ authorities: [AUTHORITY] }), barrier = null, launcher = new ScopeLauncher() } = {}) {
  return new Controller({
    fixtureRoot: fx.base, claimRoot: fx.claimRoot, socketDir: fx.socketDir, sessionFile: fx.sessionFile, seat: fx.seat,
    launcher, engine: { cwd: fx.proj },
    [TEST_ENGINE]: { command: process.execPath, preArgs: [FAKE_PI], env: { ...process.env, FAKE_PI_CONTROL: join(fx.base, "fake-b.sock"), FAKE_PI_LOG: join(fx.base, "fake-b.log") } },
    verifier, units: systemdUnits, timeouts: FAST, barrier, host,
  });
}

const stoppedOnProof = (fx) => claimRecords(fx).map((r) => r.record).filter((r) => r?.state === "stopped" && r.proof?.kind === "cohortProof");

// The child controller proves a force stop and dies before its release:
// after both keys are `stopped` (start finds the pair free), or between
// the keys (classify finishes the second).
for (const [id, crash, title, launched] of [
  ["R7", { dieAt: { "scope-release": 1 } }, "after the claim is stopped, before the release", true],
  ["R8", { dieAtState: { "between-keys": "stopped" } }, "between the two keys' stopped revisions", false],
]) {
  test(`${id}: controller killed ${title}: the restart releases the scope`, NEEDS_SCOPE, async () => {
    const { fx, fakeEnv } = childFixture();
    const a = spawnController({ fx, launcher: "scope", fakeEnv, ...crash });
    const ra = await a.next((m) => m.ready || m.error);
    assert.ok(ra.ready, JSON.stringify(ra));
    const c = await connectTo(ra.socketPath);
    let b = null;
    try {
      assert.equal((await c.takeover()).outcome, "transferred");
      void c.confirmed("force-stop");
      await a.next((m) => m.dying, 20000);
      await a.exited;
      const proven = stoppedOnProof(fx);
      assert.ok(proven.length > 0, "the claim was stopped on a cohort proof before the crash");
      const { unitName: unit, shim, claimId } = proven.at(-1);
      assert.ok(unitActive(unit), "the crash left the scope");
      const hello = await shimRequest(shim, "hello");
      assert.equal(hello.ok, true, JSON.stringify(hello));
      // The release has finished when the launch begins (Filbert N1).
      let releasedAtLaunch = null;
      const launcher = new ScopeLauncher();
      const launch = launcher.launch.bind(launcher);
      launcher.launch = (o) => ((releasedAtLaunch = b.evidence.releases.length), launch(o));
      b = scopeController(fx, { launcher });
      const started = await b.start();
      assert.equal(started.launched, launched, JSON.stringify(started));
      if (launched) assert.equal(releasedAtLaunch, 1);
      if (!launched) assert.deepEqual(started.classified, { state: "stopped", proofKind: "cohortProof" });
      assert.deepEqual(b.evidence.releases.map((r) => [r.why, r.outcome, r.unit, r.unitName, r.claim]), [["start", "released", "absent", unit, claimId]]);
      assert.ok(unitGone(unit));
      assert.equal(alive(hello.shimPid), false);
    } finally {
      c.close();
      await b?.close({ killEngine: true });
      reap(fx);
    }
  });
}

test("R9: a release that ended unavailable is tried again on a confirmed recover, not a refused one", NEEDS_SCOPE, async () => {
  const g = gate();
  const h = await live({ barrier: g.barrier });
  const unit = h.rec().unitName, shim = h.rec().shim;
  try {
    g.hold("scope-release");
    const stop = await forceStop(h);
    await g.waitHeld("scope-release");
    // The shim's socket moved: its `hello` fails and nothing is released.
    renameSync(shim, `${shim}.aside`);
    g.release("scope-release");
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the first release");
    assert.equal(h.ctrl.evidence.releases[0].outcome, "unavailable", JSON.stringify(h.ctrl.evidence.releases));
    assert.ok(unitActive(unit));
    renameSync(`${shim}.aside`, shim);
    // A refused recover changes nothing: a confirmation never issued, or
    // one confirmed for another operation.
    const other = (await h.c.request("issue-confirmation", { operationToConfirm: "force-stop" })).data.confirmation.id;
    await h.c.request("answer-confirmation", { confirmation: other, answer: "confirm" });
    for (const confirmation of [newId("confirmation"), other]) {
      assert.equal((await h.c.request("recover", { stop, confirmation })).refusal, "confirmation", confirmation);
    }
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome]), [["force-stop", "unavailable"]]);
    assert.ok(unitActive(unit));
    const rec = await h.c.confirmed("recover", { stop });
    assert.equal(rec.outcome, "recovery-eligible", JSON.stringify(rec));
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unit ?? null]), [["force-stop", "unavailable", null], ["recover", "released", "absent"]]);
    assert.ok(unitGone(unit));
    await h.ctrl.release(rec.data.eligibility);
  } finally {
    g.release("scope-release");
    await h.close();
  }
});

class WithdrawnVerifier extends FixtureVerifier {
  constructor() {
    super({ authorities: [AUTHORITY] });
    this.withdrawn = false;
  }
  cohort(...a) {
    return !this.withdrawn && super.cohort(...a);
  }
}

test("R10: close releases and signals nothing for a stopped binding whose proof no longer verifies (Filbert N3 on #1536)", NEEDS_SCOPE, async () => {
  const g = gate();
  const verifier = new WithdrawnVerifier();
  const h = await live({ barrier: g.barrier, verifier });
  const unit = h.rec().unitName, kill = process.kill;
  try {
    g.hold("scope-release");
    await forceStop(h);
    await g.waitHeld("scope-release");
    assert.equal(h.ctrl.binding.state, "stopped");
    verifier.withdrawn = true;
    // Nor a signal: the engine was proven dead when the binding moved (Darkwing note 1).
    const pid = h.rec().engine.pid, signals = [];
    process.kill = (p, sig) => (signals.push([p, sig]), kill.call(process, p, sig));
    await h.ctrl.close({ killEngine: true });
    process.kill = kill;
    assert.deepEqual(signals.filter(([p]) => Math.abs(p) === pid), []);
    assert.deepEqual(h.ctrl.evidence.releases, []);
    assert.ok(unitActive(unit));
    // The stop's own release acts on the proof it verified.
    g.release("scope-release");
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the stop's release");
    assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome]), [["force-stop", "released"]]);
  } finally {
    process.kill = kill;
    g.release("scope-release");
    h.c.close();
    h.fake.close();
    reap(h.fx);
  }
});

test("R11: close({ killEngine: true }) after a proven stop sends no signal to the recorded engine PID; a restart releases nothing more", NEEDS_SCOPE, async () => {
  const h = await live();
  const pid = h.rec().engine.pid;
  const signals = [];
  const kill = process.kill;
  let b = null;
  try {
    await forceStop(h);
    assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
    await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
    process.kill = (p, sig) => (signals.push([p, sig]), kill.call(process, p, sig));
    await h.ctrl.close({ killEngine: true });
    process.kill = kill;
    // A restart finds the released unit absent and sends nothing (Darkwing note 3).
    b = scopeController(h.fx);
    assert.equal((await b.start()).launched, true);
    assert.deepEqual(b.evidence.releases, []);
  } finally {
    process.kill = kill;
    await b?.close({ killEngine: true });
    h.c.close();
    h.fake.close();
    reap(h.fx);
  }
  assert.deepEqual(signals.filter(([p]) => Math.abs(p) === pid), []);
});

test("R12: a restart onto a claim stopped on a boot proof releases nothing, though its scope is listed", NEEDS_SCOPE, async () => {
  const { fx, fakeEnv } = childFixture();
  const a = spawnController({ fx, launcher: "scope", fakeEnv });
  const ra = await a.next((m) => m.ready || m.error);
  assert.ok(ra.ready, JSON.stringify(ra));
  a.proc.kill("SIGKILL");
  await a.exited;
  const { unitName: unit, shim } = claimRecords(fx).map((r) => r.record).filter((r) => r?.shim).at(-1);
  const b = scopeController(fx, { host: { machineId: defaultHost.machineId, bootId: () => "00000000-0000-4000-8000-000000000000" } });
  try {
    assert.deepEqual(await b.start(), { launched: false, classified: { state: "stopped", proofKind: "boot" } });
    assert.ok(stoppedOnProof(fx).length === 0 && claimRecords(fx).some((r) => r.record?.proof?.kind === "boot"));
    assert.deepEqual(b.evidence.releases, []);
    assert.ok(unitActive(unit));
    assert.equal((await shimRequest(shim, "hello")).ok, true);
  } finally {
    await b.close();
    reap(fx);
  }
});

// Last in the file. A shim ignores SIGTERM and outlives its controller, so a
// test that leaves one running relies on the after hook's kill (#1533).
test("no shim from this file's tests is left running for the after hook", async () => {
  assert.deepEqual(await shimsGone(), [], "a test left its shim running");
});