morgana: row 306 red-first tests (#1813): bus revoke-window T1/T2, cli revoke-window T3 forcing test; no product code

packages/bus/tests/revoke-window.test.mjs

@@ -0,0 +1,140 @@
+// Row 306 (#1813, row 295's N1): a run whose claim #revokeLost ended answers thread-revoked on every
+// claim-needing call until the run ends, so the runner stops it instead of exiting session-invalid.
+// A claim another cause ended still answers not-holder. access-revoke.test.mjs's fixture, copied, with
+// a limit stub and a usage stub for the limit stop. 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 { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { Store } from '../src/store.mjs';
+import { Broker } from '../src/broker.mjs';
+
+const shipped = (name) =>
+  JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
+const grant = (over = {}) => ({
+  id: `g-${over.user ?? 'alice'}`, user: 'alice', business: 'demo', projects: ['*'], instances: ['assistant'], level: 'collaborator', ...over,
+});
+const BUSINESSES = {
+  demo: {
+    id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' }, vars: { 'bus.tester': true }, plugins: { harness: ['pi'] },
+    roles: {
+      chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
+      assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
+    },
+    definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging } },
+  },
+};
+const HIT = { path: ['tokens', 'instances', 'assistant'], limit: 100, count: 120, unknownTurns: 1 };
+
+// Alice and Bob each hold a collaborator grant and own a thread; the human owns thread H. Each thread
+// has a claimed run: a-run on A, b-run on B, h-run on H. `stops` are the limit checks' answers, in order.
+function setup(t, { stops = [] } = {}) {
+  const root = mkdtempSync(join(tmpdir(), 'bus-revoke-window-'));
+  const store = new Store(root);
+  t.after(() => {
+    store.close();
+    rmSync(root, { recursive: true, force: true });
+  });
+  const access = { doc: { accessVersion: 1, revision: 1, users: [], grants: [grant(), grant({ user: 'bob' })] }, rev: 1 };
+  const b = new Broker({
+    store,
+    businesses: BUSINESSES,
+    access: () => access.doc,
+    accessStamp: () => String(access.rev),
+    limits: { mustStop: async () => stops.shift() ?? { stop: false } },
+    usage: { collect: async () => ({ recorded: true }), view: () => ({}) },
+  });
+  const caps = { human: b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }) };
+  const call = (who, verb, args = {}) => b.request(caps[who], { verb, args });
+  const code = (who, verb, args = {}) => {
+    try {
+      call(who, verb, args);
+    } catch (e) {
+      return e.code;
+    }
+    return null;
+  };
+  const edit = (fn) => {
+    access.doc = { ...access.doc, revision: access.doc.revision + 1, grants: fn(access.doc.grants) };
+    access.rev++;
+  };
+  const launch = (name, thread) => {
+    caps[name] = b.bindLaunch({ business: 'demo', role: 'assistant', run: `${name}-run`, harness: 'pi', thread });
+    return call(name, 'role.claim');
+  };
+  const claim = (run) => ({ ...store.get('SELECT op, reason, by FROM thread_claims WHERE holder_run=? ORDER BY seq DESC LIMIT 1', run) });
+  // The codes of the run's refusals, as request() logs them.
+  const refused = (run) =>
+    store.all("SELECT body FROM events WHERE kind='action.refused' AND actor_run=? ORDER BY seq", run).map((e) => JSON.parse(e.body).code);
+  const P = call('human', 'project.create', { name: 'shared' }).id;
+  caps.alice = b.bindUser({ business: 'demo', user: 'alice' });
+  caps.bob = b.bindUser({ business: 'demo', user: 'bob' });
+  const A = call('alice', 'thread.create', { project: P, name: 'alice work' }).id;
+  const B = call('bob', 'thread.create', { project: P, name: 'bob work' }).id;
+  const H = call('human', 'thread.create', { project: P, name: 'owner work' }).id;
+  launch('a', A);
+  launch('b', B);
+  launch('h', H);
+  return { b, store, caps, call, code, edit, claim, refused, A, B, H };
+}
+
+const withoutAlice = (gs) => gs.filter((g) => g.user !== 'alice');
+
+// The run's claim-needing calls after the sweep: each answers thread-revoked, and each is logged.
+function lateCalls(code) {
+  return [
+    code('a', 'message.receive'),
+    code('a', 'state'),
+    code('a', 'message.read', { id: 'm-none' }),
+    code('a', 'message.send', { to: 'human', body: 'late' }),
+  ];
+}
+
+test('row 306 T1: a run the timer sweep ended answers thread-revoked on every claim-needing call, logged, and its claim still refuses run-revoked', (t) => {
+  const { b, code, edit, claim, refused } = setup(t);
+  edit(withoutAlice);
+  assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
+  assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-timer' });
+  assert.deepEqual(lateCalls(code), Array(4).fill('thread-revoked'));
+  assert.deepEqual(refused('a-run'), Array(4).fill('thread-revoked'));
+  assert.equal(code('a', 'role.claim'), 'run-revoked', 'role.claim keeps its order');
+  assert.equal(code('b', 'message.receive'), null, 'another user\'s run stays');
+  assert.equal(code('h', 'message.receive'), null, 'the human\'s run stays');
+});
+
+test('row 306 T1: a user\'s request and access.reload end the run the same way, and it answers thread-revoked', (t) => {
+  for (const trigger of ['request', 'reload']) {
+    const { code, call, edit, claim } = setup(t);
+    edit(withoutAlice);
+    if (trigger === 'request') assert.equal(code('bob', 'thread.list'), null);
+    else assert.deepEqual(call('human', 'access.reload').ended.map((h) => h.run), ['a-run']);
+    assert.equal(claim('a-run').reason, 'access-revoked', trigger);
+    assert.equal(code('a', 'message.receive'), 'thread-revoked', trigger);
+  }
+});
+
+// The control (Dewey's C2 in dewey-0918): the first call after a pending limit stop answers from
+// #limitStop and never reaches #agent, so only the run's next call checks how its claim ended.
+test('row 306 T2: after the sweep, a run the human stopped and a run the limit stop ended still answer not-holder', async (t) => {
+  const { b, call, code, edit, claim, refused, B } = setup(t, { stops: [{ stop: true, check: 'token-limit', hit: HIT }] });
+  edit(withoutAlice);
+  assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
+  // The human's stop takes the human's own thread only: H.
+  assert.deepEqual(call('human', 'assistant.stop', { instance: 'assistant' }).stopped.map((s) => s.run), ['h-run']);
+  assert.equal(code('h', 'message.receive'), 'not-holder', 'stopped by the human');
+  assert.equal(code('h', 'state'), 'not-holder', 'stopped by the human, again');
+  // Bob's turn ends at the limit, and the run's next receive ends its claim.
+  const { id } = call('bob', 'thread.send', { thread: B, body: 'look' });
+  call('b', 'message.receive');
+  await call('b', 'usage.begin', { turn: id });
+  await call('b', 'usage.end', { turn: id });
+  assert.equal(code('b', 'message.receive'), 'not-holder', 'the limit stop');
+  assert.equal(claim('b-run').by, 'limit-check');
+  assert.equal(code('b', 'message.receive'), 'not-holder', 'the limit-stopped run\'s next call');
+  assert.equal(code('b', 'state'), 'not-holder');
+  assert.deepEqual(refused('h-run'), ['not-holder', 'not-holder']);
+  assert.deepEqual(refused('b-run'), ['not-holder', 'not-holder', 'not-holder']);
+});

packages/cli/tests/revoke-window.test.mjs

@@ -0,0 +1,228 @@
+// 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 "../../business/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, SESSION_CREDENTIALS } from "../../seat/src/session.mjs";
+import { bootConfig, loadSystem } from "../src/config.mjs";
+import { createLauncher } from "../src/launcher.mjs";
+import { fakeHarnesses, fixture, tmp } 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", "[email protected]", "-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;
+  });
+  for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k];
+  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 = {
+    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)));
+});