stack/packages/cli/tests/thread-revoke.test.mjs

import { requestHostOps } from './request-host-ops.mjs';
// FED R4 (#1801, federation design 3.7): the launcher half of revocation. A thread whose user owner
// lost the grant never starts, even with a message waiting across a restart; a running one gets
// SIGTERM at once; and recover() still rebinds and ends an earlier host's run on such a thread
// (sage-3139, D1). Real runners in the bwrap sandbox, with thread-adapter.mjs as the turn.
// The bus half is packages/bus/tests/access-revoke.test.mjs; BUILD.md
// (agents/morgana/work/queue-295) names the mutant each test kills.
//
// The host is thread-launch.test.mjs's, with an access file passed to startBroker, which the live
// boot config doesn't pass yet.

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, readSessions, Registry } 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", "[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.
function prepare(t) {
  const root = tmp(t, "mosaic-revoke-");
  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 only when the test calls tick(). pollInterval is long, so a run
// with nothing to do sleeps rather than ask the bus, and only a signal ends it.
async function start(t, { f, adapter, access }) {
  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 = {
    ...requestHostOps(runtime),
    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),
    recoverLaunch: (m) => runtime.recoverLaunch(m.record),
    credentialStatus: (m) => runtime.credentialStatus(m.business, m.role),
    threadsState: (m) => runtime.threadsState(m.business),
    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 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 logs = [];
  const launcher = createLauncher({
    host,
    system,
    env: f.env,
    makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }),
    sessionDefaults: { pollInterval: 600000 },
    harnessVersion: "0.85.1",
    readOnly: [ADAPTER],
    threadTickMs: 3600000,
    log: (l) => logs.push(l),
  });
  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 });
  // Alice binds at her first call: after the hand edit her bind refuses access-revoked.
  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, logs, close };
}

const launchLog = (f) =>
  existsSync(launchLogFile(f.dataRoot, "acme"))
    ? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l))
    : [];
const events = (f, event) => launchLog(f).filter((e) => e.event === event);

// Alice's thread A in the stack project.
function thread(host) {
  const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
  return host.as("thread.create", { project, name: "alice work" }).id;
}

test("REQ-LAUNCH-1 R4: a pending message on a revoked thread starts no run after a launcher restart", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const one = await start(t, prepared);
  const A = thread(one);
  one.as("thread.send", { thread: A, body: "for A" });
  await one.close();
  // The hand edit, while no host runs.
  prepared.grant(false);
  const two = await start(t, prepared);
  for (let i = 0; i < 3; i++) await two.launcher.tick();
  const row = two.runtime.threadsState("acme").find((x) => x.thread === A);
  assert.deepEqual([row.revoked, row.pending, row.holder], [true, 1, null]);
  assert.deepEqual(launchLog(f), [], "no launch and no refused start");
  assert.deepEqual(readSessions(f.dataRoot), []);
});

test("REQ-LAUNCH-1 R4: a running thread's run gets SIGTERM at the first tick after the hand edit, with no idle check", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const host = await start(t, prepared);
  const A = thread(host);
  const sent = host.as("thread.send", { thread: A, body: "hello" });
  await host.launcher.tick();
  const reply = await settle(() => host.as("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
  assert.ok(reply, JSON.stringify(launchLog(f)));
  const [launch] = events(f, "launch");
  assert.equal(launch.thread, A);
  prepared.grant(false);
  await host.launcher.tick();
  assert.deepEqual(events(f, "thread-revoked").map((e) => [e.thread, e.run]), [[A, launch.run]]);
  // The runner sleeps for its poll interval, so only the signal ends it, and it exits 0, stopped.
  const end = await settle(() => events(f, "end").find((e) => e.run === launch.run));
  assert.deepEqual([end?.reason, end?.exitCode, end?.signal], ["stopped", 0, null], JSON.stringify(launchLog(f)));
  assert.deepEqual(events(f, "thread-idle"), []);
  assert.deepEqual(readSessions(f.dataRoot), []);
  const row = host.runtime.threadsState("acme").find((x) => x.thread === A);
  assert.deepEqual([row.revoked, row.holder], [true, null]);
  // More ticks start nothing.
  for (let i = 0; i < 3; i++) await host.launcher.tick();
  assert.equal(events(f, "launch").length, 1);
  // Dewey's B1: the broker ended the claim before the launcher's kill, so the run's end didn't release it.
  await host.close();
  const store = new Store(f.dataRoot);
  t.after(() => store.close());
  assert.deepEqual(
    { ...store.get("SELECT op, reason, by FROM thread_claims WHERE holder_run=? ORDER BY seq DESC LIMIT 1", launch.run) },
    { op: "revoke", reason: "access-revoked", by: "access-change" },
  );
});

// sage-3139, D1: the bind refusal skips a rebind, so recover() ends the earlier host's run.
test("REQ-LAUNCH-1 R4 D1: recover() rebinds an earlier host's run on a revoked thread and ends it, which ends its claim", { skip, timeout: 60000 }, async (t) => {
  const prepared = prepare(t);
  const { f } = prepared;
  const one = await start(t, prepared);
  const A = thread(one);
  const D = one.runtime.threadsState("acme").find((x) => x.thread === A).session;
  // A stand-in for the earlier host's run, with pids that aren't running.
  const record = { business: "acme", role: "assistant", run: "run-left", harness: "pi", pid: 2 ** 22 + 7, startTime: "1", thread: A, session: D };
  const left = one.runtime.bindLaunch(record);
  one.runtime.broker.request(left.cap, { verb: "role.claim" });
  await one.close();
  prepared.grant(false);
  new Registry(f.dataRoot).add({ business: "acme", instance: "assistant", thread: A, session: D, run: "run-left", harness: "pi", pid: record.pid, startTime: "1", runnerPid: 2 ** 22 + 9, runnerStartTime: "1" });
  const two = await start(t, prepared);
  assert.ok(two.logs.includes("run run-left (assistant) from an earlier host ended: host-lost"), two.logs.join("\n"));
  assert.deepEqual(events(f, "end").map((e) => [e.run, e.reason]), [["run-left", "host-lost"]]);
  const row = two.runtime.threadsState("acme").find((x) => x.thread === A);
  assert.deepEqual([row.revoked, row.holder], [true, null]);
  for (let i = 0; i < 3; i++) await two.launcher.tick();
  assert.deepEqual(events(f, "launch"), []);
});