feat(conversation): release a stopped cohort scope on start and recover, no kill after a proven stop (#1537)

Row 51, owner Dewey, candidate c34039dc (5 files). A claim recorded
stopped on a cohort proof whose scope is still listed is released on
start (classify's stopped branch and the free path's session and seat
heads) and on a confirmed recover, after every recover check. A release
that ended unavailable or still listed is retried; the proof check stays
in #releaseScope. close({ killEngine: true }) no longer signals the
recorded PID after a proven stop. Tests R7-R12 kill relok, closeany,
noprooffkind and nopush; R9 pins the recover order.

Reviews: Filbert approve (27070, 27074), Darkwing changes then approve
(27071, 27075). Gate green on 032b5408 plus the candidate.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
2026-10-10 02:06:20 -05:00
co-authored by Claude Opus 5.5
parent 032b54086b
commit 717e3404f4
6 changed files with 494 additions and 17 deletions
+16 -4
View File
@@ -372,15 +372,27 @@ can disagree with it.
`stopped` on the claim, and again on close of a binding proven stopped,
so a release the stop didn't reach still happens; both share one request
per claim (R1, R2). The shim must answer for the claim's recorded
invocation ID, or nothing is released (R6). Each attempt is in
`evidence.releases`. After a stop that ends `uncertain` (evidence
invocation ID, and the shim refuses `release` while its engine runs, or
nothing is released (R6). Each attempt is in `evidence.releases` and is
pushed to observers (R1). After a stop that ends `uncertain` (evidence
unavailable, or a proof the verifier rejects) nothing is released: the
scope and its shim are what a later force stop reads, and a collected
scope would be an absent observation, not proof that the cohort ended
(R3). An engine that exits on its own leaves the shim and scope up and
the binding `uncertain`; a proven force stop on the empty cohort then
releases them (R4). A controller killed between recording the stop and
releasing leaves the scope; a restart doesn't release it.
releasing leaves the scope; the restart's `start` finds the claim stopped
on a cohort proof with its unit still listed and releases it (R7, R8). A
release that ended `unavailable` or `still listed` is tried again by the
next `start`, or by a `recover` that passed every check, confirmation
included; a refused `recover` releases nothing (R9). A `start` whose
claim's unit is already absent sends no request (R11). A claim stopped on
any other proof, such as a new boot, is never released (R12), and close
releases only for a stop whose proof still verifies (R10). Once the
binding is proven `stopped`, `close({ killEngine: true })` signals no
engine PID, even if the proof stops verifying later: the proof showed
every member ended, so the recorded PID may belong to another process
(R10, R11).
- **Confirmations** bind the target, operation and stop, and are consumed on
use (K6). One issued before the stop changed is refused (H17).
- **Recovery.** A confirmed `recover` after a proven stop returns a
@@ -561,7 +573,7 @@ no prompt:
| `claim.test.mjs` | W1–W17, W20, G1–G3: the writer claim, crash barriers, the guard |
| `races.test.mjs` | H1–H4, H9–H23: takeover, Interrupt and force stop, retries, incarnations |
| `turns.test.mjs` | N1–N25, N24b: the turn tracker against the fake engine's Pi behaviors; the engine seal and environment |
| `cohort.test.mjs` | K1–K19, R1–R6: scopes, force stop, proofs, recovery, eligibility, the scope's environment, scope release (needs a systemd user manager) |
| `cohort.test.mjs` | K1–K19, R1–R12: scopes, force stop, proofs, recovery, eligibility, the scope's environment, scope release and its retry after a crash (needs a systemd user manager) |
| `flows.test.mjs` | S1–S7, P3, E1–E7, the terminal, and a CHAT-01 schema check of every record produced |
| `smoke.test.mjs` | the pinned Pi binary, as above |
+38 -6
View File
@@ -339,14 +339,24 @@ export class Controller {
});
if (classified.refusal === FOREIGN_HOST || classified.damaged) throw new ControlRefusal(classified.refusal, `the claim pair is held (${classified.refusal})`);
if (classified.refusal === ALREADY_ACTIVE || classified.state === "owned") throw new ControlRefusal(ALREADY_ACTIVE, "the writer claim for this seat or session is held by a live controller");
if (classified.state === "stopped") return { launched: false, classified: { state: "stopped", proofKind: classified.proofKind } };
if (classified.state === "stopped") {
await this.#releaseListed("start", { claimId: classified.claimId, record: classified.record });
return { launched: false, classified: { state: "stopped", proofKind: classified.proofKind } };
}
if (classified.state === "uncertain" || classified.state === "stopping") {
if (!classified.claim) throw new ControlRefusal(classified.refusal ?? UNSAFE_REPLACEMENT, `the claim pair is uncertain (${classified.needs ?? "held"})`);
await this.#serveOrphan(classified, session, pins);
return { launched: false, classified: { state: classified.state, unit: classified.unit?.state ?? null } };
}
// Free. A resume of this session checks the last proven stop.
// Free. A head stopped on a cohort proof may still have its scope: its
// controller died before the release (#1537). A resume of this session
// checks the last proven stop.
const prior = this.store.head(this.sessionK);
const seatHead = this.store.head(this.seatK);
for (const h of [prior, seatHead]) {
if (h.n === 0 || h.damaged || (h === seatHead && prior.n > 0 && !prior.damaged && h.record.claimId === prior.record.claimId)) continue;
await this.#releaseListed("start", { claimId: h.record.claimId, record: h.record });
}
let generation = 1;
let priorRef = null;
if (prior.n > 0 && !prior.damaged && prior.record.state === "stopped") {
@@ -1584,6 +1594,22 @@ export class Controller {
return pending;
}
// A claim recorded `stopped` whose scope may still be listed: its
// controller died between the proof and the release, or the release ended
// `unavailable` or `still listed` (#1537). Start and recover call it. An
// earlier result other than released and absent is tried again; the proof
// check stays in #releaseScope.
async #releaseListed(why, claim) {
const earlier = this.scopeReleases.get(claim.claimId);
if (earlier) {
const r = await earlier;
if (r.outcome === "released" && r.unit === "absent") return r;
if (this.scopeReleases.get(claim.claimId) === earlier) this.scopeReleases.delete(claim.claimId);
}
if ((await this.units.lookup(claim.record)).state === "absent") return null;
return this.#releaseScope(why, claim);
}
// check.mjs `stopped`.
#stopped(s) {
const b = this.b, p = this.stoppedProof;
@@ -1606,6 +1632,8 @@ export class Controller {
const session = this.#readSession();
if (session.leaf !== this.claim.record.leafAtProof || session.branch !== this.claim.record.branchAtProof) return refused("target");
if (!this.#checkConfirmation(r, c, "recover")) return refused("confirmation");
// After every check, so a refused recover changes nothing (K6).
await this.#releaseListed("recover", this.claim);
const execution = newId("exec");
const generation = this.claim.record.generation + 1;
const bindingId = newId("binding");
@@ -1686,13 +1714,17 @@ export class Controller {
}
// Test and shutdown hook. Never a claim release: the claim is unchanged.
// A binding proven stopped has its scope released (#1536).
// A binding proven stopped has its scope released (#1536). Its engine
// isn't signalled: the binding reaches `stopped` only on a verified proof
// that every member ended, so the recorded PID may now be another
// process's (#1537).
async close({ killEngine = false } = {}) {
const exec = this.exec;
if (exec) exec.closing = true;
if (this.b?.state === "stopped" && this.#stopped(this.stops.get(this.b.stop))) await this.#releaseScope("close");
if (killEngine && typeof exec?.proc?.kill === "function") exec.proc.kill();
else if (killEngine && exec?.proc?.pid) {
const kill = killEngine && this.b?.state !== "stopped";
if (kill && typeof exec?.proc?.kill === "function") exec.proc.kill();
else if (kill && exec?.proc?.pid) {
try {
process.kill(exec.proc.kind === "pgroup" ? -exec.proc.pid : exec.proc.pid, "SIGKILL");
} catch {
@@ -1703,6 +1735,6 @@ export class Controller {
if (this.server) await new Promise((r) => this.server.close(() => r()));
this.server = null;
await this.claimChain;
if (exec?.proc && killEngine) await Promise.race([exec.proc.exited, sleep(2000)]);
if (exec?.proc && kill) await Promise.race([exec.proc.exited, sleep(2000)]);
}
}
+198 -3
View File
@@ -9,10 +9,10 @@
import { test, after } from "node:test";
import assert from "node:assert/strict";
import { appendFileSync, chmodSync, copyFileSync, mkdirSync, readFileSync, rmdirSync, writeFileSync } from "node:fs";
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, newClaimId, unitNameFor } from "../src/claim.mjs";
import { ClaimStore, FOREIGN_HOST, defaultHost, newClaimId, unitNameFor } from "../src/claim.mjs";
import { ConversationClient } from "../src/client.mjs";
import { AUTHORITY, PgroupLauncher, ScopeLauncher, forceStopCohort, releaseCohort, scopeAvailable, shimRequest, systemctlShow, systemdUnits } from "../src/cohort.mjs";
import { Controller, ELIGIBILITY, TEST_ENGINE } from "../src/controller.mjs";
@@ -773,6 +773,9 @@ test("R1: after a proven force stop the controller releases the scope; the shim
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);
@@ -909,13 +912,19 @@ test("R5: the harness finds a running shim and reaps it, and never signals a shi
}
});
test("R6: releaseCohort releases nothing for another invocation ID or a cohort without a scope", NEEDS_SCOPE, async () => {
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]]) {
@@ -931,6 +940,192 @@ test("R6: releaseCohort releases nothing for another invocation ID or a cohort w
}
});
// ---- 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 () => {
+5 -4
View File
@@ -4,10 +4,11 @@
// runs in, and the claim's owner check sees the test's own pid as live, so
// these fixtures run the controller here.
//
// argv[2] is JSON: { fx, launcher: "pgroup" | "scope", dieAt, holdAt,
// timeouts, host, fakeEnv, verifier }. Lines on stdout are JSON:
// argv[2] is JSON: { fx, launcher: "pgroup" | "scope", dieAt, dieAtState,
// holdAt, timeouts, host, fakeEnv, verifier }. Lines on stdout are JSON:
// { ready, ... } after start, { held: name } at a held barrier,
// { dying: name } just before a SIGKILL at `dieAt`. Lines on stdin:
// { dying: name } just before a SIGKILL at `dieAt`, or at `dieAtState`'s
// barrier when the claim patch there has that state. Lines on stdin:
// "go" releases a held barrier; "close" closes and exits; "kill-engine"
// closes with the engine killed; "launch <eligibility>" launches after a
// recover; "evidence" and "proof" print the controller's evidence and its
@@ -57,7 +58,7 @@ async function barrier(name, detail) {
const n = (seen.get(name) ?? 0) + 1;
seen.set(name, n);
if (cfg.trace) out({ barrier: name, n });
if (dieAt.get(name) === n) {
if (dieAt.get(name) === n || (cfg.dieAtState?.[name] && detail?.patch?.state === cfg.dieAtState[name])) {
out({ dying: name, n, detail });
process.kill(process.pid, "SIGKILL");
await new Promise(() => {});