// Row 306 (#1813, row 295's N1): the access sweep ends a lost user thread's claim in the threadsState // op, and the launcher sends SIGTERM only after the op returns. A runner call the broker answers in // between used to get not-holder, and the run ended session-invalid/22 instead of stopped/0. This test // holds the op open until a call has landed in that window and the runner has exited, so it doesn't // depend on timing. The launcher skips a tick while one is running, so nothing else signals a run // while the op is held. // // thread-revoke.test.mjs's host, with the access file, and idle-stop-race.test.mjs's short tick and // poll, with threadsState replaced by the caller. BUILD.md (agents/morgana/work/queue-306) names the // mutant each test kills. import { test } from "node:test"; import assert from "node:assert/strict"; import { spawnSync } from "node:child_process"; import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { launchDoc } from "../../organization/tests/helpers.mjs"; import { BusError } from "../../bus/src/broker.mjs"; import { startBroker } from "../../bus/src/runtime.mjs"; import { Store } from "../../bus/src/store.mjs"; import { gitEnv } from "../../seat/src/checkout.mjs"; import { sandboxProblem } from "../../seat/src/sandbox.mjs"; import { launchLogFile } from "../../seat/src/session.mjs"; import { bootConfig, loadSystem } from "../src/config.mjs"; import { createLauncher } from "../src/launcher.mjs"; import { fakeHarnesses, fixture, tmp, withoutHostCredentials } from "./helpers.mjs"; const ADAPTER = fileURLToPath(new URL("./fixtures/thread-adapter.mjs", import.meta.url)); const skip = sandboxProblem() ?? false; // A mutant's kill is an assertion failure, not a timeout (sage-2707): the wait returns what it last // saw and the assertions after it decide. async function settle(fn, ms = 20000) { const end = Date.now() + ms; let v; while (Date.now() < end) { v = await fn(); if (v) return v; await new Promise((r) => setTimeout(r, 50)); } return v; } 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); } // The assistant is the thread instance. Alice holds a collaborator grant on it in the access file; // `grant(false)` is the hand edit that removes it. The idle stop is an hour, so it never runs here. function prepare(t) { const root = tmp(t, "mosaic-revoke-window-"); const f = fixture(root, "acme", (doc) => { doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5", thinking: "high" }; doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" } }; doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet: 2 }); doc.launch.threadIdleSeconds = 3600; return doc; }); withoutHostCredentials(f.env, { piKey: true }); const project = f.doc.projects.stack.root; gitIn(project, "init", "-q", "-b", "main"); writeFileSync(join(project, "file.txt"), "one\n"); gitIn(project, "add", "file.txt"); gitIn(project, "commit", "-q", "-m", "one"); const adapter = join(root, "adapter.sh"); writeFileSync(adapter, `exec '${process.execPath}' '${ADAPTER}'\n`, { mode: 0o644 }); const file = join(root, "access.json"); let revision = 0; const grant = (on) => writeFileSync(file, JSON.stringify({ accessVersion: 1, revision: ++revision, users: [{ id: "alice", home: "local", subjects: [{ provider: "authentik", sub: "alice-sub" }] }], grants: on ? [{ id: "g-alice", user: "alice", business: "acme", projects: ["*"], instances: ["assistant"], level: "collaborator" }] : [], }), { mode: 0o600 }); grant(true); const context = { providers: ["authentik"], peers: [], businesses: { acme: { human: f.doc.human, instances: Object.keys(f.doc.roles) } }, ssoSubjects: [] }; return { root, f, adapter, access: { file, context }, grant }; } // A host and its launcher, ticking every 100 ms, with runners that poll every 100 ms, so a run's // call lands in the held window. async function start(t, { f, adapter, access }, threadsState) { const system = loadSystem({ env: f.env }); const { trackers, ...boot } = bootConfig({ system, businessId: "acme", env: f.env }); assert.equal(trackers, undefined); const runtime = await startBroker({ ...boot, access }); const ops = { 'request.admission': m => runtime.admission(m), identity: (m) => runtime.identity(m.cap), authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance), refuse: (m) => runtime.refuse(m.cap, m.code, m.subject), endLaunch: (m) => runtime.endLaunch(m.record), credentialStatus: (m) => runtime.credentialStatus(m.business, m.role), threadsState: (m) => threadsState(runtime, m), threadIdle: (m) => runtime.threadIdle(m.record), mayLaunch: (m) => runtime.mayLaunch(m), landHeld: (m) => runtime.landHeld(m.record), landHolders: (m) => runtime.landHolders(m.business), landRelease: (m) => runtime.landRelease(m.record), sessionClears: (m) => runtime.sessionClears(m.business), }; const refusing = (what, fn) => async (m) => { try { return await fn(m); } catch (e) { const code = e instanceof BusError ? e.code : `${what}-refused`; throw Object.assign(new Error(`${what} refused: ${code}`), { code }); } }; const hooks = []; let closing = null; const host = { business: "acme", path: runtime.path, dataRoot: f.dataRoot, op: refusing("launch-op", (m) => { if (!Object.hasOwn(ops, m.op)) throw new BusError("invalid-host-request"); return ops[m.op](m); }), bindLaunch: refusing("launch bind", (r) => runtime.bindLaunch(r)), beforeClose: (fn) => void hooks.push(fn), }; const launcher = createLauncher({ host, system, env: f.env, makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }), sessionDefaults: { pollInterval: 100 }, harnessVersion: "0.85.1", readOnly: [ADAPTER], threadTickMs: 100, log: () => {}, }); const close = () => (closing ??= (async () => { for (const fn of hooks) await fn(0); await runtime.close(); })()); t.after(close); await launcher.listen(); const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true }); let alice = null; const call = (verb, args = {}) => runtime.broker.request(human, { verb, args }); const as = (verb, args = {}) => runtime.broker.request((alice ??= runtime.broker.bindUser({ business: "acme", user: "alice" })), { verb, args }); return { runtime, launcher, call, as, close }; } const launchLog = (f) => existsSync(launchLogFile(f.dataRoot, "acme")) ? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l)) : []; const runnerLog = (f, run) => { const p = join(f.dataRoot, "launches", "acme", run, "runner.log"); return existsSync(p) ? readFileSync(p, "utf8") : ""; }; const ended = (f, run) => launchLog(f).find((e) => e.event === "end" && e.run === run); const launched = (f, thread) => launchLog(f).find((e) => e.event === "launch" && e.thread === thread)?.run ?? null; // The runner's own verdict on a refusal: the revoke line, the launcher-release line, or a lost session's. const verdict = (f, run) => /runner: (thread revoked by the access file|session released by the launcher|session stopped working: [a-z-]+)\n/.exec(runnerLog(f, run))?.[1] ?? null; // Send on the thread and wait for the run's RESULT, so the run holds the thread. async function ask(send, read, thread, body) { const sent = send("thread.send", { thread, body }); assert.ok(await settle(() => read("thread.messages", { thread }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT")), `no RESULT on ${thread}`); } test("row 306 REQ-LAUNCH-1 R4 a revoked run whose call lands before the SIGTERM ends stopped/0; a run the human stopped in the same window still ends session-invalid/22", { skip, timeout: 120000 }, async (t) => { const prepared = prepare(t); const { f } = prepared; let A = null; let window = null; let host = null; // The first pass that sees A revoked: the sweep in runtime.threadsState has ended A's claim. Inside // the held window the human stops their own thread's run, which no pass can signal while the op is held. const threadsState = async (runtime, m) => { const rows = runtime.threadsState(m.business); if (window || !rows.some((x) => x.thread === A && x.revoked)) return rows; window = { run: launched(f, A) }; window.stopped = host.call("assistant.stop", { instance: "assistant" }).stopped; // Both runners decide on their own before the op returns and the launcher can signal. window.revoked = await settle(() => ended(f, window.run) && verdict(f, window.run)); window.control = window.stopped[0]?.run ?? null; window.human = window.control ? await settle(() => ended(f, window.control) && verdict(f, window.control)) : null; window.done = true; return rows; }; host = await start(t, prepared, threadsState); const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; A = host.as("thread.create", { project, name: "alice work" }).id; const B = host.call("thread.create", { project, name: "owner work" }).id; await Promise.all([ask(host.as, host.as, A, "for A"), ask(host.call, host.call, B, "for B")]); // The hand edit. No user request follows, so the launcher's threadsState is what sweeps. prepared.grant(false); await settle(() => window?.done, 60000); assert.ok(window?.done, JSON.stringify(launchLog(f))); const { run, control } = window; assert.equal(run, launched(f, A)); assert.deepEqual(window.stopped.map((s) => [s.thread, s.run]), [[B, launched(f, B)]], "the human's stop took only the human's thread"); await settle(() => ended(f, run) && ended(f, control), 60000); // The revoked run refused in the window and exited by itself before the op returned. assert.equal(window.revoked, "thread revoked by the access file", runnerLog(f, run)); const log = launchLog(f); const end = (r) => log.find((e) => e.event === "end" && e.run === r); assert.deepEqual([end(run).reason, end(run).exitCode, end(run).signal], ["stopped", 0, null]); // The launcher removed the run before it recorded the end, so the pass after the op had nothing to // signal (Dewey's C1 in dewey-0918). assert.deepEqual(log.filter((e) => e.event === "thread-revoked" && e.run === run), []); assert.doesNotMatch(runnerLog(f, run), /session stopped working/); // The sweep already ended the claim, so the runner doesn't release it at exit. assert.doesNotMatch(runnerLog(f, run), /runner: role\.release:/); assert.match(runnerLog(f, run), /runner: exiting 0\n/); // The control: a real not-holder in the same window still ends 22, with no signal. assert.equal(window.human, "session stopped working: not-holder", runnerLog(f, control)); assert.deepEqual([end(control).reason, end(control).exitCode, end(control).signal], ["session-invalid", 22, null]); assert.match(runnerLog(f, control), /runner: exiting 22\n/); await host.close(); const store = new Store(f.dataRoot); t.after(() => store.close()); const codes = (r) => store.all("SELECT body FROM events WHERE kind='action.refused' AND actor_run=? ORDER BY seq", r).map((e) => JSON.parse(e.body).code); assert.ok(codes(run).includes("thread-revoked"), JSON.stringify(codes(run))); assert.ok(!codes(run).includes("not-holder"), JSON.stringify(codes(run))); assert.ok(codes(control).includes("not-holder"), JSON.stringify(codes(control))); });